diff --git a/db/migrations/20260830000000_create_settlement_operations.cjs b/db/migrations/20260830000000_create_settlement_operations.cjs
new file mode 100644
index 00000000..128851e7
--- /dev/null
+++ b/db/migrations/20260830000000_create_settlement_operations.cjs
@@ -0,0 +1,54 @@
+/**
+ * Durable state for milestone release/redirect operations.
+ *
+ * The operation identity is supplied by the caller and is unique per
+ * milestone. This lets a retried request recover the original operation
+ * instead of creating a second payout attempt.
+ */
+exports.up = async function up(knex) {
+  await knex.schema.createTable('settlement_operations', (table) => {
+    table.uuid('id').primary().defaultTo(knex.raw('gen_random_uuid()'))
+    table.string('milestone_id', 64).notNullable()
+      .references('id').inTable('milestones').onDelete('CASCADE')
+    table.string('operation_key', 255).notNullable()
+    table
+      .enu('operation_type', ['release', 'redirect'], {
+        useNative: true,
+        enumName: 'settlement_operation_type',
+      })
+      .notNullable()
+    table
+      .enu('status', ['pending', 'submitted', 'confirmed', 'failed'], {
+        useNative: true,
+        enumName: 'settlement_operation_status',
+      })
+      .notNullable()
+      .defaultTo('pending')
+    table.integer('attempt_count').notNullable().defaultTo(0)
+    table.string('transaction_hash', 128).nullable()
+    table.string('failure_code', 64).nullable()
+    table.text('failure_message').nullable()
+    table.string('requested_by', 255).notNullable()
+    table.jsonb('request_fingerprint').notNullable()
+    table.timestamp('created_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+    table.timestamp('updated_at', { useTz: true }).notNullable().defaultTo(knex.fn.now())
+    table.timestamp('submitted_at', { useTz: true }).nullable()
+    table.timestamp('confirmed_at', { useTz: true }).nullable()
+
+    table.unique(['milestone_id', 'operation_key'], {
+      indexName: 'idx_settlement_operations_identity',
+    })
+    table.unique(['transaction_hash'], {
+      indexName: 'idx_settlement_operations_transaction_hash',
+    })
+    table.index(['milestone_id', 'status'], 'idx_settlement_operations_milestone_status')
+    table.index(['status', 'updated_at'], 'idx_settlement_operations_recovery')
+  })
+}
+
+exports.down = async function down(knex) {
+  await knex.schema.dropTableIfExists('settlement_operations')
+  await knex.raw('DROP TYPE IF EXISTS settlement_operation_status')
+  await knex.raw('DROP TYPE IF EXISTS settlement_operation_type')
+}
+
diff --git a/docs/settlement-operations.md b/docs/settlement-operations.md
new file mode 100644
index 00000000..adea3c21
--- /dev/null
+++ b/docs/settlement-operations.md
@@ -0,0 +1,88 @@
+# Atomic milestone settlement operations
+
+Milestone releases and redirects are represented by a durable settlement
+operation. The operation is the unit of retry, not the HTTP request and not a
+new blockchain transaction on every retry.
+
+## State machine
+
+```text
+pending ──submit──> submitted ──confirm──> confirmed
+   ▲                    │
+   │                    └──fail──> failed ──submit──> submitted
+   └──────────────────────────────────────────────────
+```
+
+`confirmed` is terminal. A release or redirect cannot become complete merely
+because a worker constructed a transaction or because the submission endpoint
+returned. Only a callback carrying the exact submitted transaction hash can
+advance `submitted` to `confirmed`.
+
+## Operation identity
+
+Each operation has:
+
+- `milestone_id` and caller-provided `operation_key` as a unique logical
+  identity;
+- `operation_type`, either `release` or `redirect`;
+- a SHA-256 request fingerprint covering the milestone, type, destination,
+  amount, and asset;
+- `attempt_count`, submitted/confirmed timestamps, and operator-visible
+  failure details.
+
+Reusing an operation key with the same request returns the existing operation.
+Reusing it with a changed payout destination, amount, or type returns an
+identity conflict. This prevents an accidental retry from becoming a second
+settlement or from changing the beneficiary under an old key.
+
+## Atomic database behavior
+
+`settlement_operations` has a unique `(milestone_id, operation_key)` index.
+Submission, failure recording, and confirmation each run in a transaction,
+lock the operation row, and use a conditional status update. Concurrent
+workers therefore observe one of these safe outcomes:
+
+| Concurrent action | Result |
+| --- | --- |
+| same pending operation submitted twice | one attempt, same submitted record returned |
+| same submitted operation confirmed twice | one confirmation, same record returned |
+| different transaction hash | conflict; original hash is preserved |
+| timeout after broadcast | `failed`, then retry with the same operation id |
+| process restart | row is recovered by operation id and key |
+
+The final confirmation update is guarded by both operation id, status, and
+transaction hash. A late callback for a previous attempt cannot confirm a new
+attempt accidentally.
+
+## Recovery procedure
+
+1. Find `pending`, `submitted`, or `failed` rows for the milestone.
+2. For `confirmed`, do not submit or retry; the settlement is complete.
+3. For `submitted`, reconcile the stored transaction hash with the chain
+   before deciding whether to record confirmation or failure.
+4. For `failed`, retry `submit` using the existing operation id and logical
+   key. Do not create a new key to bypass an unknown outcome.
+5. Record the operator-visible failure code and a redacted message when the
+   chain or RPC is unavailable.
+
+The `SettlementOperationLedger` mirrors the production transitions and can
+snapshot/restore records for deterministic worker and restart tests. The
+database adapter is the source of truth in deployed environments.
+
+## Failure and observability trade-offs
+
+The design deliberately preserves failed rows rather than deleting them. This
+uses a small amount of storage but gives operators a complete attempt history,
+supports safe recovery after a timeout, and prevents duplicate payouts. Error
+messages are bounded and must not contain secrets, signing material, or full
+request payloads. A monitoring job should alert on old `submitted` operations
+and high retry counts; it must not auto-create replacement operations.
+
+## Test coverage
+
+The state-machine tests cover malformed requests, release/redirect identity,
+same-key replay, changed-payload conflicts, duplicate submission, transaction
+hash conflicts, timeout/rejected-transaction recovery, confirmation gating,
+duplicate callbacks, terminal behavior, and restart restoration for every
+non-terminal state.
+
diff --git a/src/services/settlementOperations.test.ts b/src/services/settlementOperations.test.ts
new file mode 100644
index 00000000..34758b05
--- /dev/null
+++ b/src/services/settlementOperations.test.ts
@@ -0,0 +1,230 @@
+import {
+  assertSettlementConfirmed,
+  SettlementOperationError,
+  SettlementOperationLedger,
+} from './settlementOperations.js'
+import { describe, expect, it } from '@jest/globals'
+
+const request = (overrides: Record<string, unknown> = {}) => ({
+  milestoneId: 'ms-1499-test',
+  operationKey: 'release-2026-08-30',
+  operationType: 'release' as const,
+  requestedBy: 'creator-1',
+  destination: 'GDESTINATION',
+  amount: '100.0000000',
+  assetCode: 'USDC',
+  ...overrides,
+})
+
+const tx = 'tx-hash-0001'
+
+describe('settlement operation identity', () => {
+  it('creates a pending operation with no side effect status', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+
+    expect(operation.id).toBeTruthy()
+    expect(operation.status).toBe('pending')
+    expect(operation.attemptCount).toBe(0)
+    expect(operation.operationType).toBe('release')
+    expect(operation.transactionHash).toBeNull()
+    expect(operation.submittedAt).toBeNull()
+    expect(operation.confirmedAt).toBeNull()
+  })
+
+  it('returns the same operation for an identical retry', () => {
+    const ledger = new SettlementOperationLedger()
+    const first = ledger.create(request())
+    const retry = ledger.create(request())
+
+    expect(retry).toEqual(first)
+    expect(ledger.snapshot()).toHaveLength(1)
+  })
+
+  it.each([
+    ['operationType', { operationType: 'unknown' }],
+    ['operationKey', { operationKey: 'contains spaces' }],
+    ['amount', { amount: '0' }],
+    ['destination', { destination: '' }],
+    ['requestedBy', { requestedBy: '' }],
+  ])('rejects malformed %s before allocation', (_field, overrides) => {
+    const ledger = new SettlementOperationLedger()
+    expect(() => ledger.create(request(overrides))).toThrow(SettlementOperationError)
+    expect(ledger.snapshot()).toEqual([])
+  })
+
+  it('rejects an idempotency-key reuse with a changed destination', () => {
+    const ledger = new SettlementOperationLedger()
+    ledger.create(request())
+
+    expect(() => ledger.create(request({ destination: 'GOTHERDESTINATION' }))).toThrow(/different settlement request/i)
+    expect(ledger.snapshot()).toHaveLength(1)
+  })
+
+  it('allows release and redirect to share no identity', () => {
+    const ledger = new SettlementOperationLedger()
+    const release = ledger.create(request())
+    const redirect = ledger.create(request({ operationType: 'redirect', operationKey: 'redirect-2026-08-30' }))
+
+    expect(redirect.id).not.toBe(release.id)
+    expect(ledger.snapshot()).toHaveLength(2)
+  })
+})
+
+describe('submission and retry state machine', () => {
+  it('moves pending to submitted exactly once', () => {
+    const ledger = new SettlementOperationLedger()
+    const pending = ledger.create(request())
+    const submitted = ledger.submit(pending.id, tx)
+    const retry = ledger.submit(pending.id, tx)
+
+    expect(submitted.status).toBe('submitted')
+    expect(submitted.attemptCount).toBe(1)
+    expect(retry).toEqual(submitted)
+    expect(ledger.snapshot()[0]?.attemptCount).toBe(1)
+  })
+
+  it('rejects a second transaction hash for a submitted operation', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+
+    expect(() => ledger.submit(operation.id, 'tx-hash-0002')).toThrow(/another transaction/i)
+    expect(ledger.get(operation.id)?.attemptCount).toBe(1)
+  })
+
+  it.each(['short', '', 'contains spaces'])('rejects malformed transaction hashes (%s)', (hash) => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    expect(() => ledger.submit(operation.id, hash)).toThrow(/transactionHash/i)
+    expect(ledger.get(operation.id)?.status).toBe('pending')
+  })
+
+  it('records a failure as recoverable and increments attempts on retry', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    const failed = ledger.fail(operation.id, 'RPC_TIMEOUT', 'Soroban RPC timed out')
+    const retry = ledger.submit(operation.id, 'tx-hash-0002')
+
+    expect(failed.status).toBe('failed')
+    expect(failed.failureCode).toBe('RPC_TIMEOUT')
+    expect(failed.failureMessage).toBe('Soroban RPC timed out')
+    expect(retry.status).toBe('submitted')
+    expect(retry.attemptCount).toBe(2)
+    expect(retry.failureCode).toBeNull()
+  })
+
+  it('does not allow failed or pending operations to be confirmed', () => {
+    const ledger = new SettlementOperationLedger()
+    const pending = ledger.create(request())
+    expect(() => ledger.confirm(pending.id, tx)).toThrow(/submitted/i)
+    ledger.submit(pending.id, tx)
+    ledger.fail(pending.id, 'REJECTED_TX', 'Transaction rejected')
+    expect(() => ledger.confirm(pending.id, tx)).toThrow(/submitted/i)
+  })
+})
+
+describe('confirmation gate', () => {
+  it('exposes a completion guard that rejects every non-confirmed state', () => {
+    for (const status of ['pending', 'submitted', 'failed'] as const) {
+      expect(() => assertSettlementConfirmed({ status })).toThrow(/not confirmed/i)
+    }
+    expect(() => assertSettlementConfirmed({ status: 'confirmed' })).not.toThrow()
+  })
+
+  it('confirms only the exact submitted transaction', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+
+    expect(() => ledger.confirm(operation.id, 'tx-hash-0002')).toThrow(/does not match/i)
+    expect(ledger.get(operation.id)?.status).toBe('submitted')
+
+    const confirmed = ledger.confirm(operation.id, tx)
+    expect(confirmed.status).toBe('confirmed')
+    expect(confirmed.confirmedAt).toBeTruthy()
+  })
+
+  it('makes the confirmation callback idempotent', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    const first = ledger.confirm(operation.id, tx)
+    const replay = ledger.confirm(operation.id, tx)
+
+    expect(replay).toEqual(first)
+    expect(ledger.snapshot()).toHaveLength(1)
+  })
+
+  it('rejects a different callback after confirmation', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    ledger.confirm(operation.id, tx)
+
+    expect(() => ledger.confirm(operation.id, 'tx-hash-0002')).toThrow(/another transaction/i)
+  })
+
+  it('does not permit a confirmed operation to be resubmitted', () => {
+    const ledger = new SettlementOperationLedger()
+    const operation = ledger.create(request())
+    ledger.submit(operation.id, tx)
+    ledger.confirm(operation.id, tx)
+
+    const result = ledger.submit(operation.id, 'tx-hash-0002')
+    expect(result.status).toBe('confirmed')
+    expect(result.transactionHash).toBe(tx)
+    expect(result.attemptCount).toBe(1)
+  })
+})
+
+describe('restart recovery', () => {
+  it('restores pending, submitted, failed, and confirmed operations', () => {
+    const firstProcess = new SettlementOperationLedger()
+    const pending = firstProcess.create(request({ operationKey: 'pending' }))
+    const submitted = firstProcess.create(request({ operationKey: 'submitted' }))
+    firstProcess.submit(submitted.id, 'tx-hash-0003')
+    const failed = firstProcess.create(request({ operationKey: 'failed' }))
+    firstProcess.submit(failed.id, 'tx-hash-0004')
+    firstProcess.fail(failed.id, 'NETWORK_ERROR', 'Network unavailable')
+    const confirmed = firstProcess.create(request({ operationKey: 'confirmed' }))
+    firstProcess.submit(confirmed.id, 'tx-hash-0005')
+    firstProcess.confirm(confirmed.id, 'tx-hash-0005')
+
+    const secondProcess = new SettlementOperationLedger()
+    secondProcess.restore(firstProcess.snapshot())
+
+    expect(secondProcess.get(pending.id)?.status).toBe('pending')
+    expect(secondProcess.get(submitted.id)?.status).toBe('submitted')
+    expect(secondProcess.get(failed.id)?.status).toBe('failed')
+    expect(secondProcess.get(confirmed.id)?.status).toBe('confirmed')
+    expect(secondProcess.create(request({ operationKey: 'pending' })).id).toBe(pending.id)
+  })
+
+  it('retries the same failed logical operation after restoration', () => {
+    const firstProcess = new SettlementOperationLedger()
+    const operation = firstProcess.create(request())
+    firstProcess.submit(operation.id, tx)
+    firstProcess.fail(operation.id, 'TIMEOUT', 'Submission timed out')
+
+    const secondProcess = new SettlementOperationLedger()
+    secondProcess.restore(firstProcess.snapshot())
+    const retry = secondProcess.submit(operation.id, 'tx-hash-0006')
+
+    expect(retry.id).toBe(operation.id)
+    expect(retry.attemptCount).toBe(2)
+    expect(retry.status).toBe('submitted')
+  })
+
+  it('does not silently replace restored records with a new operation', () => {
+    const source = new SettlementOperationLedger()
+    const original = source.create(request())
+    const restored = new SettlementOperationLedger()
+    restored.restore(source.snapshot())
+
+    const replay = restored.create(request())
+    expect(replay.id).toBe(original.id)
+    expect(restored.snapshot()).toHaveLength(1)
+  })
+})
diff --git a/src/services/settlementOperations.ts b/src/services/settlementOperations.ts
new file mode 100644
index 00000000..aa89557e
--- /dev/null
+++ b/src/services/settlementOperations.ts
@@ -0,0 +1,463 @@
+import { createHash, randomUUID } from 'node:crypto'
+import type { Knex } from 'knex'
+import db from '../db/index.js'
+
+export type SettlementOperationType = 'release' | 'redirect'
+export type SettlementOperationStatus = 'pending' | 'submitted' | 'confirmed' | 'failed'
+
+export interface SettlementOperation {
+  id: string
+  milestoneId: string
+  operationKey: string
+  operationType: SettlementOperationType
+  status: SettlementOperationStatus
+  attemptCount: number
+  transactionHash: string | null
+  failureCode: string | null
+  failureMessage: string | null
+  requestedBy: string
+  requestFingerprint: string
+  createdAt: string
+  updatedAt: string
+  submittedAt: string | null
+  confirmedAt: string | null
+}
+
+export interface CreateSettlementOperationInput {
+  milestoneId: string
+  operationKey: string
+  operationType: SettlementOperationType
+  requestedBy: string
+  destination: string
+  amount: string
+  assetCode?: string | null
+}
+
+/**
+ * Completion consumers must call this guard before marking a milestone or
+ * vault complete. Submission is deliberately not sufficient evidence of a
+ * completed payout because the chain may reject or never include the tx.
+ */
+export const assertSettlementConfirmed = (operation: Pick<SettlementOperation, 'status'>): void => {
+  if (operation.status !== 'confirmed') {
+    throw new SettlementOperationError(
+      'INVALID_STATE',
+      `Settlement is not confirmed (current state: ${operation.status})`,
+    )
+  }
+}
+
+export class SettlementOperationError extends Error {
+  constructor(
+    public readonly code:
+    | 'INVALID_INPUT'
+    | 'IDENTITY_CONFLICT'
+    | 'NOT_FOUND'
+    | 'INVALID_STATE'
+    | 'TRANSACTION_CONFLICT',
+    message: string,
+  ) {
+    super(message)
+    this.name = 'SettlementOperationError'
+  }
+}
+
+const nowIso = (): string => new Date().toISOString()
+
+/**
+ * Small deterministic model of the operation contract.
+ *
+ * The database repository below is the production adapter. This model is
+ * intentionally exported so worker code and tests can exercise retries and
+ * process-restart recovery without requiring a live PostgreSQL instance. Its
+ * snapshot format is also the shape used by recovery tooling.
+ */
+export class SettlementOperationLedger {
+  private readonly operations = new Map<string, SettlementOperation>()
+  private readonly identities = new Map<string, string>()
+
+  create(input: CreateSettlementOperationInput): SettlementOperation {
+    assertInput(input)
+    const fingerprint = fingerprintFor(input)
+    const identity = `${input.milestoneId.trim()}:${input.operationKey}`
+    const existingId = this.identities.get(identity)
+    if (existingId) {
+      const existing = this.operations.get(existingId)!
+      if (existing.requestFingerprint !== fingerprint) {
+        throw new SettlementOperationError('IDENTITY_CONFLICT', 'operationKey is already bound to a different settlement request')
+      }
+      return { ...existing }
+    }
+
+    const timestamp = nowIso()
+    const operation: SettlementOperation = {
+      id: randomUUID(),
+      milestoneId: input.milestoneId.trim(),
+      operationKey: input.operationKey,
+      operationType: input.operationType,
+      status: 'pending',
+      attemptCount: 0,
+      transactionHash: null,
+      failureCode: null,
+      failureMessage: null,
+      requestedBy: input.requestedBy.trim(),
+      requestFingerprint: fingerprint,
+      createdAt: timestamp,
+      updatedAt: timestamp,
+      submittedAt: null,
+      confirmedAt: null,
+    }
+    this.identities.set(identity, operation.id)
+    this.operations.set(operation.id, operation)
+    return { ...operation }
+  }
+
+  get(operationId: string): SettlementOperation | null {
+    const operation = this.operations.get(operationId)
+    return operation ? { ...operation } : null
+  }
+
+  submit(operationId: string, transactionHash: string): SettlementOperation {
+    this.assertTransactionHash(transactionHash)
+    const current = this.require(operationId)
+    if (current.status === 'confirmed') return { ...current }
+    if (current.status === 'submitted') {
+      if (current.transactionHash !== transactionHash) {
+        throw new SettlementOperationError('TRANSACTION_CONFLICT', 'Operation is already submitted with another transaction')
+      }
+      return { ...current }
+    }
+    if (current.status !== 'pending' && current.status !== 'failed') {
+      throw new SettlementOperationError('INVALID_STATE', `Cannot submit operation in ${current.status} state`)
+    }
+    const updated = {
+      ...current,
+      status: 'submitted' as const,
+      attemptCount: current.attemptCount + 1,
+      transactionHash,
+      failureCode: null,
+      failureMessage: null,
+      submittedAt: nowIso(),
+      updatedAt: nowIso(),
+    }
+    this.operations.set(operationId, updated)
+    return { ...updated }
+  }
+
+  confirm(operationId: string, transactionHash: string): SettlementOperation {
+    const current = this.require(operationId)
+    if (current.status === 'confirmed') {
+      if (current.transactionHash !== transactionHash) {
+        throw new SettlementOperationError('TRANSACTION_CONFLICT', 'Confirmed operation has another transaction')
+      }
+      return { ...current }
+    }
+    if (current.status !== 'submitted') {
+      throw new SettlementOperationError('INVALID_STATE', 'Only a submitted operation can be confirmed')
+    }
+    if (current.transactionHash !== transactionHash) {
+      throw new SettlementOperationError('TRANSACTION_CONFLICT', 'Confirmation does not match submitted transaction')
+    }
+    const updated = { ...current, status: 'confirmed' as const, confirmedAt: nowIso(), updatedAt: nowIso() }
+    this.operations.set(operationId, updated)
+    return { ...updated }
+  }
+
+  fail(operationId: string, failureCode: string, failureMessage: string): SettlementOperation {
+    if (!/^[A-Z][A-Z0-9_:-]{1,63}$/.test(failureCode)) {
+      throw new SettlementOperationError('INVALID_INPUT', 'failureCode has an invalid format')
+    }
+    if (!failureMessage.trim() || failureMessage.length > 2000) {
+      throw new SettlementOperationError('INVALID_INPUT', 'failureMessage must contain 1–2000 characters')
+    }
+    const current = this.require(operationId)
+    if (current.status === 'failed') return { ...current }
+    if (current.status !== 'submitted') {
+      throw new SettlementOperationError('INVALID_STATE', 'Only a submitted operation can fail')
+    }
+    const updated = {
+      ...current,
+      status: 'failed' as const,
+      failureCode,
+      failureMessage: failureMessage.trim(),
+      updatedAt: nowIso(),
+    }
+    this.operations.set(operationId, updated)
+    return { ...updated }
+  }
+
+  snapshot(): SettlementOperation[] {
+    return [...this.operations.values()].map((operation) => ({ ...operation }))
+  }
+
+  restore(snapshot: SettlementOperation[]): void {
+    this.operations.clear()
+    this.identities.clear()
+    for (const operation of snapshot) {
+      this.operations.set(operation.id, { ...operation })
+      this.identities.set(`${operation.milestoneId}:${operation.operationKey}`, operation.id)
+    }
+  }
+
+  private require(operationId: string): SettlementOperation {
+    const operation = this.operations.get(operationId)
+    if (!operation) throw new SettlementOperationError('NOT_FOUND', 'Settlement operation not found')
+    return operation
+  }
+
+  private assertTransactionHash(transactionHash: string): void {
+    if (!TX_HASH_RE.test(transactionHash)) {
+      throw new SettlementOperationError('INVALID_INPUT', 'transactionHash has an invalid format')
+    }
+  }
+}
+
+const OPERATION_KEY_RE = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,254}$/
+const TX_HASH_RE = /^[A-Za-z0-9_-]{8,128}$/
+
+const assertInput = (input: CreateSettlementOperationInput): void => {
+  if (!input || typeof input !== 'object') {
+    throw new SettlementOperationError('INVALID_INPUT', 'Settlement operation input is required')
+  }
+  for (const [name, value] of Object.entries({
+    milestoneId: input.milestoneId,
+    operationKey: input.operationKey,
+    requestedBy: input.requestedBy,
+    destination: input.destination,
+    amount: input.amount,
+  })) {
+    if (typeof value !== 'string' || value.trim() === '') {
+      throw new SettlementOperationError('INVALID_INPUT', `${name} must be a non-empty string`)
+    }
+  }
+  if (!OPERATION_KEY_RE.test(input.operationKey)) {
+    throw new SettlementOperationError('INVALID_INPUT', 'operationKey has an invalid format')
+  }
+  if (input.operationType !== 'release' && input.operationType !== 'redirect') {
+    throw new SettlementOperationError('INVALID_INPUT', 'operationType must be release or redirect')
+  }
+  if (!Number.isFinite(Number(input.amount)) || Number(input.amount) <= 0) {
+    throw new SettlementOperationError('INVALID_INPUT', 'amount must be a positive number')
+  }
+}
+
+const fingerprintFor = (input: CreateSettlementOperationInput): string =>
+  createHash('sha256')
+    .update(JSON.stringify({
+      milestoneId: input.milestoneId.trim(),
+      operationType: input.operationType,
+      destination: input.destination.trim(),
+      amount: input.amount.trim(),
+      assetCode: input.assetCode?.trim() ?? null,
+    }))
+    .digest('hex')
+
+type SettlementOperationRow = {
+  id: string
+  milestone_id: string
+  operation_key: string
+  operation_type: SettlementOperationType
+  status: SettlementOperationStatus
+  attempt_count: number
+  transaction_hash: string | null
+  failure_code: string | null
+  failure_message: string | null
+  requested_by: string
+  request_fingerprint: string
+  created_at: Date | string
+  updated_at: Date | string
+  submitted_at: Date | string | null
+  confirmed_at: Date | string | null
+}
+
+const iso = (value: Date | string | null): string | null =>
+  value === null ? null : value instanceof Date ? value.toISOString() : new Date(value).toISOString()
+
+const mapRow = (row: SettlementOperationRow): SettlementOperation => ({
+  id: row.id,
+  milestoneId: row.milestone_id,
+  operationKey: row.operation_key,
+  operationType: row.operation_type,
+  status: row.status,
+  attemptCount: Number(row.attempt_count),
+  transactionHash: row.transaction_hash,
+  failureCode: row.failure_code,
+  failureMessage: row.failure_message,
+  requestedBy: row.requested_by,
+  requestFingerprint: typeof row.request_fingerprint === 'string'
+    ? row.request_fingerprint
+    : JSON.stringify(row.request_fingerprint),
+  createdAt: iso(row.created_at)!,
+  updatedAt: iso(row.updated_at)!,
+  submittedAt: iso(row.submitted_at),
+  confirmedAt: iso(row.confirmed_at),
+})
+
+const clientFor = (trx?: Knex.Transaction): Knex => (trx ?? db) as Knex
+
+const fetchByIdentity = async (
+  client: Knex,
+  milestoneId: string,
+  operationKey: string,
+): Promise<SettlementOperationRow | undefined> => client('settlement_operations')
+  .where({ milestone_id: milestoneId, operation_key: operationKey })
+  .first()
+
+const fetchById = async (
+  client: Knex,
+  operationId: string,
+  lock = false,
+): Promise<SettlementOperationRow | undefined> => {
+  let query = client('settlement_operations').where({ id: operationId })
+  if (lock) query = query.forUpdate()
+  return query.first()
+}
+
+/**
+ * Create once or return the existing operation for the same logical key.
+ * A reused key is safe only when its request fingerprint is identical.
+ */
+export const createSettlementOperation = async (
+  input: CreateSettlementOperationInput,
+  trx?: Knex.Transaction,
+): Promise<SettlementOperation> => {
+  assertInput(input)
+  const client = clientFor(trx)
+  const fingerprint = fingerprintFor(input)
+
+  await client('settlement_operations')
+    .insert({
+      milestone_id: input.milestoneId.trim(),
+      operation_key: input.operationKey,
+      operation_type: input.operationType,
+      requested_by: input.requestedBy.trim(),
+      request_fingerprint: fingerprint,
+    })
+    .onConflict(['milestone_id', 'operation_key'])
+    .ignore()
+
+  const row = await fetchByIdentity(client, input.milestoneId.trim(), input.operationKey)
+  if (!row) throw new SettlementOperationError('NOT_FOUND', 'Settlement operation was not created')
+  if (row.request_fingerprint !== fingerprint) {
+    throw new SettlementOperationError(
+      'IDENTITY_CONFLICT',
+      'operationKey is already bound to a different settlement request',
+    )
+  }
+  return mapRow(row)
+}
+
+export const getSettlementOperation = async (operationId: string): Promise<SettlementOperation | null> => {
+  if (!operationId.trim()) throw new SettlementOperationError('INVALID_INPUT', 'operationId is required')
+  const row = await fetchById(db, operationId)
+  return row ? mapRow(row) : null
+}
+
+/**
+ * Claim a pending or failed operation for submission. The row lock and
+ * conditional status update make concurrent workers converge on one attempt.
+ * Confirmed operations are returned as-is so a client retry cannot submit a
+ * second release or redirect.
+ */
+export const markSettlementSubmitted = async (
+  operationId: string,
+  transactionHash: string,
+): Promise<SettlementOperation> => {
+  if (!TX_HASH_RE.test(transactionHash)) {
+    throw new SettlementOperationError('INVALID_INPUT', 'transactionHash has an invalid format')
+  }
+
+  return db.transaction(async (trx) => {
+    const row = await fetchById(trx, operationId, true)
+    if (!row) throw new SettlementOperationError('NOT_FOUND', 'Settlement operation not found')
+    if (row.status === 'confirmed') return mapRow(row)
+    if (row.status === 'submitted') {
+      if (row.transaction_hash !== transactionHash) {
+        throw new SettlementOperationError('TRANSACTION_CONFLICT', 'Operation is already submitted with another transaction')
+      }
+      return mapRow(row)
+    }
+    if (row.status !== 'pending' && row.status !== 'failed') {
+      throw new SettlementOperationError('INVALID_STATE', `Cannot submit operation in ${row.status} state`)
+    }
+
+    const now = new Date()
+    const [updated] = await trx('settlement_operations')
+      .where({ id: operationId, status: row.status })
+      .update({
+        status: 'submitted',
+        attempt_count: Number(row.attempt_count) + 1,
+        transaction_hash: transactionHash,
+        failure_code: null,
+        failure_message: null,
+        submitted_at: now,
+        updated_at: now,
+      })
+      .returning('*')
+    if (!updated) throw new SettlementOperationError('INVALID_STATE', 'Operation changed during submission')
+    return mapRow(updated)
+  })
+}
+
+/** A callback may confirm only the exact submitted transaction. */
+export const confirmSettlementOperation = async (
+  operationId: string,
+  transactionHash: string,
+): Promise<SettlementOperation> => {
+  return db.transaction(async (trx) => {
+    const row = await fetchById(trx, operationId, true)
+    if (!row) throw new SettlementOperationError('NOT_FOUND', 'Settlement operation not found')
+    if (row.status === 'confirmed') {
+      if (row.transaction_hash !== transactionHash) {
+        throw new SettlementOperationError('TRANSACTION_CONFLICT', 'Confirmed operation has another transaction')
+      }
+      return mapRow(row)
+    }
+    if (row.status !== 'submitted') {
+      throw new SettlementOperationError('INVALID_STATE', 'Only a submitted operation can be confirmed')
+    }
+    if (row.transaction_hash !== transactionHash) {
+      throw new SettlementOperationError('TRANSACTION_CONFLICT', 'Confirmation does not match submitted transaction')
+    }
+    const now = new Date()
+    const [updated] = await trx('settlement_operations')
+      .where({ id: operationId, status: 'submitted', transaction_hash: transactionHash })
+      .update({ status: 'confirmed', confirmed_at: now, updated_at: now })
+      .returning('*')
+    if (!updated) throw new SettlementOperationError('INVALID_STATE', 'Operation changed during confirmation')
+    return mapRow(updated)
+  })
+}
+
+/** Record a recoverable failure without losing the operation identity. */
+export const failSettlementOperation = async (
+  operationId: string,
+  failureCode: string,
+  failureMessage: string,
+): Promise<SettlementOperation> => {
+  if (!/^[A-Z][A-Z0-9_:-]{1,63}$/.test(failureCode)) {
+    throw new SettlementOperationError('INVALID_INPUT', 'failureCode has an invalid format')
+  }
+  if (!failureMessage.trim() || failureMessage.length > 2000) {
+    throw new SettlementOperationError('INVALID_INPUT', 'failureMessage must contain 1–2000 characters')
+  }
+  return db.transaction(async (trx) => {
+    const row = await fetchById(trx, operationId, true)
+    if (!row) throw new SettlementOperationError('NOT_FOUND', 'Settlement operation not found')
+    if (row.status === 'failed') return mapRow(row)
+    if (row.status !== 'submitted') {
+      throw new SettlementOperationError('INVALID_STATE', 'Only a submitted operation can fail')
+    }
+    const [updated] = await trx('settlement_operations')
+      .where({ id: operationId, status: 'submitted' })
+      .update({
+        status: 'failed',
+        failure_code: failureCode,
+        failure_message: failureMessage.trim(),
+        updated_at: new Date(),
+      })
+      .returning('*')
+    if (!updated) throw new SettlementOperationError('INVALID_STATE', 'Operation changed during failure recording')
+    return mapRow(updated)
+  })
+}