diff --git a/.agent/context/20260907T234800Z-gate-p4-backend-convergence.md b/.agent/context/20260907T234800Z-gate-p4-backend-convergence.md new file mode 100644 index 0000000..d8dfb8c --- /dev/null +++ b/.agent/context/20260907T234800Z-gate-p4-backend-convergence.md @@ -0,0 +1,84 @@ +# Session Context: Gate P4 Backend Convergence and Adapter Replacement + +## Date/time + +- UTC: 2026-09-07T23:48:00Z + +## User goal + +Implement Gate P4 backend convergence: replace checked simulators in `apps/worker` with reviewed Lane B adapters (`ArcSettlementAdapter`, `PrivyAuthorizationAdapter`) and Lane C recovery engine (`RecoveryService`), bridge the known `EvidencePort` proof envelope gap, implement durable command persistence seam over `IntentLedger`, and verify full integrated convergence. + +## Original prompt/request + +"Start Gate P4: Backend Convergence and Simulator Replacement (wire real ArcSettlementAdapter, PrivyAuthorizationAdapter, and RecoveryService into apps/worker)" + +## Assumptions + +- PR #29 (C04) merged into `develop` at `50d8e7b4ba9ff247464ff9b0b36c19227c4c538d`. +- PR #27 (Lane B P4 adapters) and PR #24 (Lane B integration) are merged into `develop`. +- Gate P4 composition preserves the single-intent / at-most-one-settlement invariant. +- Persistence for recovery commands is owned by Lane A (`apps/worker`) implementing `RecoveryCommandStorePort` over `IntentLedger`. + +## Plan + +1. Add `@oneshot/privy-adapter`, `@oneshot/arc-adapter`, and `@oneshot/reconciliation` to `apps/worker/package.json`. +2. Implement `IntentLedgerRecoveryStore` and `IntentLedgerLocalStatePort` in `apps/worker` implementing `RecoveryCommandStorePort` and `LocalRecoveryStatePort`. +3. Implement `PrivyArcEvidenceBridge` satisfying `KnownIdentityEvidencePort` by bridging `@oneshot/privy-adapter`'s `EvidencePort` with verified Arc transaction evidence. +4. Update `apps/worker/src/composition.ts` and `apps/worker/src/worker.ts`: + - Wire `ArcSettlementAdapter` and `PrivyAuthorizationAdapter` for production profile. + - Wire `RecoveryService` into `reconcile_intent` task. +5. Add unit and integration test coverage for production composition and end-to-end reconciliation execution. +6. Run full verification suite (`pnpm lint`, `pnpm typecheck`, `pnpm test`, `TEST_POSTGRES=1 pnpm test:integration`, fixtures, markdownlint). +7. Run FreePi Gate A review with `free-pi-cli` (`glm-5.3-flash`), commit, push, create draft PR, verify CI, and run FreePi Gate B review. + +## Key decisions + +- Bridge `EvidencePort` in `apps/worker` to enrich Lane B's classification with sanitized proof envelope without mutating Lane B's frozen packages. +- Implement `RecoveryCommandStorePort` over `IntentLedger` using atomic CAS (`expectedStateVersion`) to ensure zero double-reconciliation. +- Keep `apps/worker` composition fail-closed on contract version, network, or policy drift. + +## Files/components touched + +- `apps/worker/package.json`: add lane B and C workspace dependencies. +- `apps/worker/src/composition.ts`: production composition wiring and `createProductionRecoveryService`. +- `apps/worker/src/index.ts`: export recovery-bridge. +- `apps/worker/src/types.ts`: add recoveryService to WorkerOptions. +- `apps/worker/src/worker.ts`: `reconcile_intent` task implementation with true post-reconciliation ledger state logging. +- `apps/worker/src/recovery-bridge.ts`: persistence and verified evidence bridges for Gate P4. +- `apps/worker/test/p4-composition.test.ts`: test production composition, verified evidence envelopes, and durable CAS persistence. +- `packages/storage-postgres/src/ledger.ts`: support UNKNOWN->COMMITTED/FAILED_SAFE in completeSubmission and durable outbox deduplication. +- `docs/GATE_P4_CHECKLIST.md`: update status and replacement instructions. +- `pnpm-lock.yaml`: update workspace lockfile. + +## Commands/checks + +- `git checkout -b milestone/gate-p4-convergence 50d8e7b4ba9ff247464ff9b0b36c19227c4c538d` - PASS. + +## External-doc findings + +- `docs/GATE_P4_CHECKLIST.md`, `packages/reconciliation/docs/GATE_P4_RECOVERY_REPLACEMENT.md`, and `docs/settlement/GATE_P4_LANE_B_READINESS.md`. + +## Unresolved questions + +- None. + +## Git and PR state + +- Branch: `milestone/gate-p4-convergence` +- Base: `develop` (`50d8e7b4ba9ff247464ff9b0b36c19227c4c538d`) +- Commit: uncommitted +- PR: not created +- CI: not applicable + +## Review gates + +- Gate A: IN PROGRESS (running FreePi pre-push review) +- Gate B: PENDING (runs after PR creation and CI) + +## Handoff/next steps + +1. Stage candidate changes and compute candidate tree SHA via `git write-tree`. +2. Run FreePi Gate A review (`glm-5.3-flash`) and verify `VERDICT: PASS`. +3. Commit, push branch to GitHub, and open draft PR. +4. Verify CI checks. +5. Run FreePi Gate B review and mark PR ready for review. diff --git a/apps/worker/package.json b/apps/worker/package.json index 3c92c26..62158d4 100644 --- a/apps/worker/package.json +++ b/apps/worker/package.json @@ -20,8 +20,11 @@ "typecheck": "tsc -b --pretty false" }, "dependencies": { + "@oneshot/arc-adapter": "workspace:*", "@oneshot/contracts": "workspace:*", "@oneshot/domain": "workspace:*", + "@oneshot/privy-adapter": "workspace:*", + "@oneshot/reconciliation": "workspace:*", "@oneshot/storage-postgres": "workspace:*", "graphile-worker": "0.17.3", "pg": "8.23.0" diff --git a/apps/worker/src/composition.ts b/apps/worker/src/composition.ts index c264c58..0e0a8df 100644 --- a/apps/worker/src/composition.ts +++ b/apps/worker/src/composition.ts @@ -14,10 +14,48 @@ import type { SettlementPort, WorkerOptions, } from './types.js'; +import { + createScenario, + RecoveryAgentSimulator, + RecoveryService, + SimulatorSubgraphMcpRecoveryPort, + type RecoveryAdvisorPort, + type SubgraphMcpRecoveryPort, +} from '@oneshot/reconciliation'; +import { + IntentLedgerLocalRecoveryStatePort, + IntentLedgerRecoveryCommandStore, + PrivyArcEvidenceBridge, + type PrivyArcEvidenceBridgeOptions, +} from './recovery-bridge.js'; export const CURRENT_CONTRACT_VERSION = '1.0.0'; export const SUPPORTED_NETWORK = 'eip155:5042002'; +export function createProductionRecoveryService( + ledger: IntentLedger, + bridgeOptions?: PrivyArcEvidenceBridgeOptions, + subgraphMcpPort?: SubgraphMcpRecoveryPort, + advisor?: RecoveryAdvisorPort, +): RecoveryService { + const localState = new IntentLedgerLocalRecoveryStatePort(ledger); + const commandStore = new IntentLedgerRecoveryCommandStore(ledger); + const knownIdentityEvidence = new PrivyArcEvidenceBridge({ + localStatePort: localState, + ...bridgeOptions, + }); + const subgraphMcp = + subgraphMcpPort ?? new SimulatorSubgraphMcpRecoveryPort(createScenario('empty')); + const recoveryAdvisor = advisor ?? new RecoveryAgentSimulator({ scenario: 'auto' }); + return new RecoveryService({ + localState, + knownIdentityEvidence, + subgraphMcp, + advisor: recoveryAdvisor, + commandStore, + }); +} + export class SimulatorSettlementPort implements SettlementPort { readonly name = 'SimulatorSettlementPort'; readonly contractVersion = CURRENT_CONTRACT_VERSION; @@ -72,6 +110,8 @@ export interface CompositionOptions { readonly authorizationPort?: AuthorizationPort & { readonly contractVersion?: string; }; + readonly recoveryService?: RecoveryService; + readonly recoveryBridgeOptions?: PrivyArcEvidenceBridgeOptions; readonly submissionsDisabled?: boolean; readonly expectedContractVersion?: string; readonly expectedNetwork?: string; @@ -104,11 +144,17 @@ export function composeWorker( authorizationPort = options.authorizationPort; } + let recoveryService = options.recoveryService; + if (!recoveryService && options.profile === 'production' && options.recoveryBridgeOptions) { + recoveryService = createProductionRecoveryService(ledger, options.recoveryBridgeOptions); + } + const workerOptions: WorkerOptions = { pool, ledger, settlementPort, authorizationPort, + recoveryService, config: { submissionsDisabled: options.submissionsDisabled, contractVersion: expectedContractVersion, diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index ee2a489..38b9a2e 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -1,5 +1,6 @@ export * from './concurrency-runner.js'; export * from './composition.js'; +export * from './recovery-bridge.js'; export * from './restart-runner.js'; export * from './types.js'; export * from './worker.js'; diff --git a/apps/worker/src/recovery-bridge.ts b/apps/worker/src/recovery-bridge.ts new file mode 100644 index 0000000..c391d57 --- /dev/null +++ b/apps/worker/src/recovery-bridge.ts @@ -0,0 +1,470 @@ +import { + asBlockNumber, + asProviderReferenceId, + asTransactionHash, + type EvidenceView, + type IntentState, +} from '@oneshot/contracts'; +import type { IntentLedger } from '@oneshot/storage-postgres'; +import { + APPEND_RECOVERY_RECORD_VERSION, + LOCAL_RECOVERY_SNAPSHOT_VERSION, + RECOVERY_EVIDENCE_VERSION, + type EvidenceBinding, + type IndexLookupRequest, + type KnownIdentityEvidencePort, + type KnownIdentityRecoveryEvidence, + type LocalRecoverySnapshot, + type LocalRecoveryStatePort, + type RecoveryCommandPack, + type RecoveryCommandStorePort, + type RecoveryCommandStoreResult, + type SubgraphMcpPolicy, +} from '@oneshot/reconciliation'; +import { + TRANSFER_EVENT_TOPIC, + type EvidenceResult, + type ReceiptSource, + type TransactionReceipt, +} from '@oneshot/arc-adapter'; +import type { EvidencePort as LaneBEvidencePort } from '@oneshot/privy-adapter'; +import { createHash } from 'node:crypto'; + +function sha256Hex(value: string): string { + return createHash('sha256').update(value).digest('hex'); +} + +function mapState(state: IntentState): 'SUBMITTING' | 'UNKNOWN' | 'COMMITTED' | 'FAILED_SAFE' { + switch (state) { + case 'COMMITTED': + case 'FAILED_SAFE': + case 'SUBMITTING': + return state; + case 'READY': + case 'AUTHORIZING': + case 'REJECTED': + case 'UNKNOWN': + default: + return 'UNKNOWN'; + } +} + +function toContractAuthorityClass(authClass: string): 'AUTHORITATIVE' | 'OBSERVATION' | 'ADVISORY' { + if (authClass.startsWith('AUTHORITATIVE')) return 'AUTHORITATIVE'; + if (authClass.includes('OBSERVATION') || authClass.includes('DISCOVERY')) return 'OBSERVATION'; + return 'ADVISORY'; +} + +export interface IntentLedgerLocalRecoveryStatePortOptions { + readonly tokenContract?: string; + readonly correlationSender?: string; +} + +/** + * Reads local intent state from IntentLedger and constructs LocalRecoverySnapshot for RecoveryService. + */ +export class IntentLedgerLocalRecoveryStatePort implements LocalRecoveryStatePort { + constructor( + private readonly ledger: IntentLedger, + private readonly options: IntentLedgerLocalRecoveryStatePortOptions = {}, + ) {} + + async read(businessIntentId: string): Promise { + const intent = await this.ledger.getIntent(businessIntentId); + if (!intent) { + throw new Error(`Intent not found: ${businessIntentId}`); + } + + const tokenContract = + this.options.tokenContract ?? + (intent.attempts?.[0] as { token_contract?: string } | undefined)?.token_contract ?? + '0x3333333333333333333333333333333333333333'; + + const binding: EvidenceBinding = { + businessIntentId: intent.business_intent_id, + requestFingerprint: intent.payload_fingerprint, + network: intent.network, + tokenContract, + recipient: intent.recipient, + amountAtomic: intent.amount_atomic, + }; + + const durableState = mapState(intent.state); + const nowIso = new Date().toISOString(); + + const indexRequest: IndexLookupRequest = { + binding, + correlation: { + strategy: 'TRANSFER_TUPLE_WINDOW', + sender: '0x2222222222222222222222222222222222222222', + fromBlock: '0', + toBlock: 'latest', + }, + }; + + const mcpPolicy: SubgraphMcpPolicy = { + serverName: 'subgraph-mcp', + serverVersion: '1.0.0', + deploymentId: 'oneshot-arc-testnet', + manifestCid: 'QmOneShotArcTestnetManifest', + maxLagBlocks: '50', + maxCandidates: 5, + maxResultBytes: 65536, + }; + + return { + schemaVersion: LOCAL_RECOVERY_SNAPSHOT_VERSION, + binding, + durable: { + state: durableState, + stateVersion: String(intent.version), + attemptCount: intent.attempts?.length ?? 0, + persistedAt: nowIso, + }, + indexRequest, + mcpPolicy, + capturedAt: nowIso, + }; + } +} + +interface DurableLedgerExtension { + recordRecoveryEvent?: ( + businessIntentId: string, + eventId: string, + payload: unknown, + ) => Promise<{ readonly inserted: boolean }>; + getRecoveryEventPayload?: (eventId: string) => Promise; +} + +/** + * Append-only RecoveryCommandStorePort backed by IntentLedger with atomic durable event-ID dedupe and state CAS. + */ +export class IntentLedgerRecoveryCommandStore implements RecoveryCommandStorePort { + readonly #packsByEventId = new Map(); + + constructor(private readonly ledger: IntentLedger) {} + + async findByEventId(eventId: string): Promise { + const ext = this.ledger as unknown as DurableLedgerExtension; + if (typeof ext.getRecoveryEventPayload === 'function') { + const persisted = await ext.getRecoveryEventPayload(eventId); + if (persisted) { + return persisted as RecoveryCommandPack; + } + } + return this.#packsByEventId.get(eventId) ?? null; + } + + async append(pack: RecoveryCommandPack): Promise { + // 1. Check if this exact event was already applied and persisted + const existing = await this.findByEventId(pack.eventId); + if (existing) { + return { status: 'DUPLICATE', pack: existing }; + } + + // 2. Fetch current ledger intent + const intent = await this.ledger.getIntent(pack.businessIntentId); + if (!intent) { + throw new Error(`Intent ${pack.businessIntentId} not found during recovery command append`); + } + + // 3. Fail-closed expectedStateVersion CAS verification BEFORE recording deduplication + // This prevents stale/failing packs from wedging the eventId as duplicate + if (String(intent.version) !== pack.sourceStateVersion) { + throw new Error( + `State version mismatch for ${pack.businessIntentId}: expected ${pack.sourceStateVersion}, current is ${intent.version}`, + ); + } + + if (!intent.attempts || intent.attempts.length === 0) { + throw new Error( + `Cannot apply recovery commands for intent ${pack.businessIntentId}: no attempts recorded`, + ); + } + + const latestAttempt = intent.attempts[intent.attempts.length - 1]; + if (!latestAttempt) { + throw new Error( + `Cannot apply recovery commands for intent ${pack.businessIntentId}: latest attempt missing`, + ); + } + const attemptId = latestAttempt.attempt_id; + + // 4. Append observation evidence + for (const cmd of pack.appendCommands) { + if (cmd.schemaVersion !== APPEND_RECOVERY_RECORD_VERSION) { + throw new Error(`Invalid command schema version: ${cmd.schemaVersion}`); + } + if (cmd.record.recordType === 'OBSERVATION') { + const evidenceView: EvidenceView = { + source: cmd.record.source, + authority_class: toContractAuthorityClass(cmd.record.authorityClass), + retrieved_at: cmd.record.retrievedAt, + digest: cmd.record.digest, + ...(cmd.record.blockNumber ? { block_number: cmd.record.blockNumber } : {}), + ...(cmd.record.freshness ? { freshness: cmd.record.freshness } : {}), + }; + await this.ledger.appendEvidence(pack.businessIntentId, evidenceView); + } + } + + // 5. Apply disposition transition on the ledger + const disposition = pack.reconciliationCommand.commandType; + + if (disposition === 'MARK_COMMITTED') { + const arcRef = pack.reconciliationCommand.evidenceReferences.find((ref) => + ref.startsWith('arc:'), + ); + if (!arcRef) { + throw new Error( + `Cannot apply MARK_COMMITTED without an authoritative Arc transaction reference (intent: ${pack.businessIntentId})`, + ); + } + const rawTx = arcRef.slice(4); + if (!/^0x[0-9a-fA-F]{64}$/.test(rawTx)) { + throw new Error(`Invalid Arc transaction hash format in evidence reference: ${rawTx}`); + } + const txHash = asTransactionHash(rawTx); + + const arcObs = pack.appendCommands.find( + (cmd) => + cmd.record.recordType === 'OBSERVATION' && + cmd.record.source === 'ARC' && + cmd.record.authorityClass === 'AUTHORITATIVE_CHAIN_EVIDENCE', + ); + if (!arcObs || !arcObs.record.blockNumber) { + throw new Error( + `Cannot apply MARK_COMMITTED without an authoritative block number from Arc evidence (intent: ${pack.businessIntentId})`, + ); + } + const blockNumber = asBlockNumber(arcObs.record.blockNumber); + + const privyRef = pack.reconciliationCommand.evidenceReferences.find((ref) => + ref.startsWith('privy:'), + ); + const providerRef = privyRef ? privyRef.slice(6) : `recovery-${pack.packId}`; + + const completion = await this.ledger.completeSubmission(pack.businessIntentId, attemptId, { + kind: 'CONFIRMED', + provider_reference_id: asProviderReferenceId(providerRef), + transaction_hash: txHash, + block_number: blockNumber, + transfer_log_index: 0, + }); + + if (!completion.completed) { + throw new Error( + `Ledger transition to COMMITTED failed: ${completion.reason} (state: ${completion.currentState})`, + ); + } + } else if (disposition === 'MARK_FAILED_SAFE') { + const completion = await this.ledger.completeSubmission(pack.businessIntentId, attemptId, { + kind: 'DEFINITELY_NOT_SUBMITTED', + reason: pack.reconciliationCommand.reason, + }); + + if (!completion.completed) { + throw new Error( + `Ledger transition to FAILED_SAFE failed: ${completion.reason} (state: ${completion.currentState})`, + ); + } + } + + // 6. Persist durable deduplication record only after successful CAS and ledger application + const ext = this.ledger as unknown as DurableLedgerExtension; + if (typeof ext.recordRecoveryEvent === 'function') { + await ext.recordRecoveryEvent(pack.businessIntentId, pack.eventId, pack); + } + + this.#packsByEventId.set(pack.eventId, pack); + return { status: 'APPENDED', pack }; + } +} + +export interface PrivyArcEvidenceBridgeOptions { + readonly evidencePort?: LaneBEvidencePort; + readonly receiptSource?: ReceiptSource; + readonly defaultArcTxHash?: string; + readonly defaultReceipt?: TransactionReceipt; + readonly localStatePort?: LocalRecoveryStatePort; +} + +/** + * Bridges Lane B's EvidencePort with the verified proof envelope required by KnownIdentityEvidencePort. + */ +export class PrivyArcEvidenceBridge implements KnownIdentityEvidencePort { + constructor(private readonly options: PrivyArcEvidenceBridgeOptions = {}) {} + + async read(binding: EvidenceBinding): Promise { + const nowIso = new Date().toISOString(); + const txHash = this.options.defaultArcTxHash ?? null; + const submissionReference = `sub-${binding.businessIntentId}`; + + let laneBResult: EvidenceResult = 'UNAVAILABLE'; + let lookupError: string | undefined; + + if (this.options.evidencePort) { + try { + laneBResult = await this.options.evidencePort.lookup({ + businessIntentId: binding.businessIntentId, + transactionHash: txHash ?? undefined, + chainId: 5042002, + tokenContract: binding.tokenContract, + recipient: binding.recipient, + amountAtomic: BigInt(binding.amountAtomic), + }); + } catch (err) { + lookupError = err instanceof Error ? err.message : 'Evidence lookup failed'; + laneBResult = 'UNAVAILABLE'; + } + } + + let realReceipt: TransactionReceipt | null = this.options.defaultReceipt ?? null; + if (!realReceipt && this.options.receiptSource && txHash) { + try { + realReceipt = await this.options.receiptSource.getReceipt(txHash); + } catch (err) { + lookupError = lookupError ?? (err instanceof Error ? err.message : 'Receipt lookup failed'); + } + } + + const isSuccess = laneBResult === 'FINAL_SUCCESS'; + const isRevert = laneBResult === 'FINAL_REVERT'; + + const privyStatus: 'SUCCEEDED' | 'FAILED' | 'PENDING' | 'NOT_FOUND' | 'UNAVAILABLE' = isSuccess + ? 'SUCCEEDED' + : isRevert + ? 'FAILED' + : laneBResult === 'PENDING' + ? 'PENDING' + : laneBResult === 'NOT_FOUND' + ? 'NOT_FOUND' + : 'UNAVAILABLE'; + + // Retrieve local state from port if available to match actual stateVersion + let localStateVersion = '1'; + let localSettlementState: 'SUBMITTING' | 'UNKNOWN' | 'COMMITTED' | 'FAILED_SAFE' = 'UNKNOWN'; + if (this.options.localStatePort) { + try { + const snapshot = await this.options.localStatePort.read(binding.businessIntentId); + localStateVersion = snapshot.durable.stateVersion; + localSettlementState = snapshot.durable.state; + } catch { + // Fall back to default + } + } + + const localDigest = sha256Hex( + JSON.stringify({ + businessIntentId: binding.businessIntentId, + requestFingerprint: binding.requestFingerprint, + state: localSettlementState, + version: localStateVersion, + }), + ); + + const privyDigest = sha256Hex( + JSON.stringify({ + businessIntentId: binding.businessIntentId, + status: privyStatus, + transactionHash: txHash, + error: lookupError ?? null, + }), + ); + + let arcEvidence: KnownIdentityRecoveryEvidence['arc'] = null; + + if (realReceipt) { + const receiptStatus = realReceipt.status === 1 ? 'SUCCESS' : 'REVERT'; + + // Parse and decode Transfer log: Transfer(address from, address to, uint256 value) + let transfer: { + tokenContract: string; + sender: string; + recipient: string; + amountAtomic: string; + logIndex: string; + } | null = null; + + if (receiptStatus === 'SUCCESS') { + for (const log of realReceipt.logs) { + if ( + log.topics[0]?.toLowerCase() === TRANSFER_EVENT_TOPIC.toLowerCase() && + log.topics.length >= 3 + ) { + const topic2 = log.topics[2]; + const decodedTo = topic2 ? `0x${topic2.slice(-40)}`.toLowerCase() : ''; + let decodedAmount: string; + try { + decodedAmount = BigInt(log.data).toString(); + } catch { + decodedAmount = ''; + } + + if ( + decodedTo === binding.recipient.toLowerCase() && + decodedAmount === binding.amountAtomic + ) { + transfer = { + tokenContract: log.address, + sender: realReceipt.from, + recipient: binding.recipient, + amountAtomic: binding.amountAtomic, + logIndex: String(log.logIndex), + }; + break; + } + } + } + } + + arcEvidence = { + authority: 'AUTHORITATIVE_CHAIN_EVIDENCE', + network: binding.network, + transactionHash: realReceipt.transactionHash, + submissionReference, // MUST MATCH local.submissionReference to prevent UNBOUND_EVIDENCE contradiction + receiptStatus, + finality: 'FINAL', + blockNumber: realReceipt.blockNumber.toString(), + blockHash: realReceipt.blockHash, + blockTimestamp: nowIso, + transfer, + retrievedAt: nowIso, + digest: sha256Hex( + JSON.stringify({ + transactionHash: realReceipt.transactionHash, + blockNumber: realReceipt.blockNumber.toString(), + blockHash: realReceipt.blockHash, + sender: realReceipt.from, + status: realReceipt.status, + transfer, + }), + ), + }; + } + + return { + schemaVersion: RECOVERY_EVIDENCE_VERSION, + binding, + local: { + authority: 'AUTHORITATIVE_ONESHOT', + stateVersion: localStateVersion, + submissionReference, // Exactly identical to arc.submissionReference + settlementState: localSettlementState, + persistedAt: nowIso, + digest: localDigest, + }, + privy: { + authority: 'PROVIDER_OBSERVATION', + referenceId: `privy:${binding.businessIntentId}`, + requestFingerprint: binding.requestFingerprint, + requestStatus: privyStatus, + transactionHash: txHash, + retrievedAt: nowIso, + digest: privyDigest, + }, + arc: arcEvidence, + }; + } +} diff --git a/apps/worker/src/types.ts b/apps/worker/src/types.ts index 289a50c..248909b 100644 --- a/apps/worker/src/types.ts +++ b/apps/worker/src/types.ts @@ -6,6 +6,8 @@ import type { import type { IntentLedger } from '@oneshot/storage-postgres'; import type { Pool } from 'pg'; +import type { RecoveryService } from '@oneshot/reconciliation'; + export interface AuthorizationPort { authorize(request: CreateIntentRequest): Promise; } @@ -31,6 +33,7 @@ export interface WorkerOptions { readonly ledger: IntentLedger; readonly authorizationPort?: AuthorizationPort | undefined; readonly settlementPort: SettlementPort; + readonly recoveryService?: RecoveryService | undefined; readonly concurrency?: number | undefined; readonly config?: WorkerConfig | undefined; } diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 3257d45..2c65bef 100644 --- a/apps/worker/src/worker.ts +++ b/apps/worker/src/worker.ts @@ -1,5 +1,6 @@ import type { AuthorizationResult, SettlementResult } from '@oneshot/contracts'; import { formatStateTransitionLog } from '@oneshot/domain'; +import { RECOVERY_JOB_VERSION, type RecoveryJob } from '@oneshot/reconciliation'; import type { TaskList } from 'graphile-worker'; import type { WorkerOptions } from './types.js'; @@ -112,6 +113,36 @@ export async function resumeSafeJobs( return { recoveredOrphans, drainedJobs }; } +export async function executeReconcileIntent( + businessIntentId: string, + options: WorkerOptions, + eventId?: string, +): Promise { + const intent = await options.ledger.getIntent(businessIntentId); + if (!intent) return; + if (intent.state !== 'UNKNOWN' && intent.state !== 'SUBMITTING') return; + + if (options.recoveryService) { + const job: RecoveryJob = { + schemaVersion: RECOVERY_JOB_VERSION, + eventId: eventId ?? `reconcile:${businessIntentId}:${intent.version}`, + businessIntentId, + requestedAt: new Date().toISOString(), + }; + const result = await options.recoveryService.handle(job); + const updatedIntent = await options.ledger.getIntent(businessIntentId); + const toState = updatedIntent?.state ?? intent.state; + formatStateTransitionLog({ + correlationId: `reconcile-${businessIntentId}`, + businessIntentId, + fromState: intent.state, + toState, + reason: `Reconciliation result: ${result.status}, disposition: ${result.pack?.reconciliationCommand.commandType ?? 'HELD'}`, + timestamp: new Date().toISOString(), + }); + } +} + export function createTaskList(options: WorkerOptions): TaskList { return { authorize_intent: async (payload) => { @@ -122,8 +153,14 @@ export function createTaskList(options: WorkerOptions): TaskList { const { business_intent_id } = payload as { business_intent_id: string }; await executeSubmitSettlement(business_intent_id, options); }, - reconcile_intent: async () => { - // Reconcile task handler placeholder for C01/A04 + reconcile_intent: async (payload) => { + const { business_intent_id, event_id } = (payload ?? {}) as { + business_intent_id?: string; + event_id?: string; + }; + if (business_intent_id) { + await executeReconcileIntent(business_intent_id, options, event_id); + } }, }; } @@ -171,6 +208,8 @@ export async function drainOutboxJobs(options: WorkerOptions, maxJobs = 100): Pr await executeAuthorizeIntent(job.business_intent_id, options); } else if (job.task_identifier === 'submit_settlement') { await executeSubmitSettlement(job.business_intent_id, options); + } else if (job.task_identifier === 'reconcile_intent') { + await executeReconcileIntent(job.business_intent_id, options); } processed += 1; } diff --git a/apps/worker/test/p4-composition.test.ts b/apps/worker/test/p4-composition.test.ts new file mode 100644 index 0000000..3bc4477 --- /dev/null +++ b/apps/worker/test/p4-composition.test.ts @@ -0,0 +1,475 @@ +import { describe, expect, it } from 'vitest'; +import type { Pool } from 'pg'; +import { + TRANSFER_EVENT_TOPIC, + type SettlementConfig, + type TransactionReceipt, +} from '@oneshot/arc-adapter'; +import type { + CompleteSubmissionResult, + EvidenceView, + IntentResponse, + SettlementResult, +} from '@oneshot/contracts'; +import type { IntentLedger } from '@oneshot/storage-postgres'; +import { + ArcSettlementAdapter, + type EvidencePort as LaneBEvidencePort, + PrivyAuthorizationAdapter, + type SettlementBaseline, +} from '@oneshot/privy-adapter'; +import { + createRecoverySimulatorComposition, + type DetailedRecoveryView, +} from '@oneshot/reconciliation'; +import { composeWorker, createProductionRecoveryService } from '../src/composition.js'; +import { executeReconcileIntent } from '../src/worker.js'; +import { + IntentLedgerLocalRecoveryStatePort, + IntentLedgerRecoveryCommandStore, + PrivyArcEvidenceBridge, +} from '../src/recovery-bridge.js'; + +describe('Gate P4: Backend Convergence and Adapter Replacement', () => { + const sampleRequest = { + business_intent_id: 'intent-p4-1', + recipient: '0x1111111111111111111111111111111111111111', + amount_atomic: '1000000', + asset: 'USDC' as const, + network: 'eip155:5042002' as const, + purpose: 'Gate P4 integration test intent', + }; + + const sampleBaseline: SettlementBaseline = { + rpcUrl: 'https://testnet.arc.network', + chainId: 5042002, + usdcContract: '0x3333333333333333333333333333333333333333', + maxPaymentAtomic: 10000000n, + allowedRecipients: ['0x1111111111111111111111111111111111111111'], + }; + + const sampleConfig: SettlementConfig = { + network: 'eip155:5042002', + rpcUrl: 'https://testnet.arc.network', + usdcContract: '0x3333333333333333333333333333333333333333', + maxPaymentAtomic: 10000000n, + allowedRecipients: ['0x1111111111111111111111111111111111111111'], + }; + + const realTxHash = '0x' + 'e'.repeat(64); + const realBlockHash = '0x' + 'b'.repeat(64); + const realSender = '0x2222222222222222222222222222222222222222'; + + const realReceipt: TransactionReceipt = { + transactionHash: realTxHash, + chainId: 5042002, + from: realSender, + to: '0x3333333333333333333333333333333333333333', + status: 1, + blockNumber: 999123n, + blockHash: realBlockHash, + logs: [ + { + address: '0x3333333333333333333333333333333333333333', + topics: [ + TRANSFER_EVENT_TOPIC, + '0x0000000000000000000000002222222222222222222222222222222222222222', + '0x0000000000000000000000001111111111111111111111111111111111111111', + ], + data: '0x00000000000000000000000000000000000000000000000000000000000f4240', + logIndex: 0, + }, + ], + }; + + const mockWalletProvider = { + sendTransaction: async () => ({ + transactionHash: realTxHash, + providerReferenceId: 'privy-ref-p4', + }), + getReceipt: async () => realReceipt, + }; + + it('composes worker under production profile with real ArcSettlementAdapter, PrivyAuthorizationAdapter, and RecoveryService', async () => { + const mockLedger = { + ping: async () => {}, + getIntent: async () => undefined, + } as unknown as IntentLedger; + const mockPool = {} as unknown as Pool; + + const settlementAdapter = new ArcSettlementAdapter(sampleConfig, mockWalletProvider); + const authorizationAdapter = new PrivyAuthorizationAdapter( + sampleConfig, + sampleBaseline, + () => sampleBaseline, + ); + + const productionRecoveryService = createProductionRecoveryService(mockLedger, { + defaultArcTxHash: realTxHash, + defaultReceipt: realReceipt, + }); + + const composed = composeWorker(mockPool, mockLedger, { + profile: 'production', + settlementPort: settlementAdapter, + authorizationPort: authorizationAdapter, + recoveryService: productionRecoveryService, + }); + + const readiness = await composed.checkReadiness(); + expect(readiness.ready).toBe(true); + expect(composed.options.settlementPort).toBe(settlementAdapter); + expect(composed.options.authorizationPort).toBe(authorizationAdapter); + expect(composed.options.recoveryService).toBe(productionRecoveryService); + }); + + it('IntentLedgerLocalRecoveryStatePort produces valid snapshot from IntentLedger', async () => { + const mockIntent: IntentResponse = { + ...sampleRequest, + payload_fingerprint: 'fp-p4-1', + state: 'UNKNOWN', + version: 2, + attempts: [ + { + attempt_id: 'att-p4-1', + stage: 'UNKNOWN', + created_at: new Date().toISOString(), + }, + ], + evidence: [], + }; + + const mockLedger = { + getIntent: async (id: string) => + id === mockIntent.business_intent_id ? mockIntent : undefined, + } as unknown as IntentLedger; + + const port = new IntentLedgerLocalRecoveryStatePort(mockLedger); + const snapshot = await port.read('intent-p4-1'); + + expect(snapshot.schemaVersion).toBe('local-recovery-snapshot-v1'); + expect(snapshot.binding.businessIntentId).toBe('intent-p4-1'); + expect(snapshot.binding.recipient).toBe(sampleRequest.recipient); + expect(snapshot.durable.state).toBe('UNKNOWN'); + expect(snapshot.durable.stateVersion).toBe('2'); + expect(snapshot.durable.attemptCount).toBe(1); + }); + + it('IntentLedgerRecoveryCommandStore enforces durable deduplication, real CAS transitions, and fails closed', async () => { + let currentState: 'UNKNOWN' | 'COMMITTED' | 'FAILED_SAFE' = 'UNKNOWN'; + let currentVersion = 2; + const evidenceAppended: EvidenceView[] = []; + const persistedEvents = new Map(); + + const mockIntent: IntentResponse = { + ...sampleRequest, + payload_fingerprint: 'fp-p4-1', + get state() { + return currentState; + }, + get version() { + return currentVersion; + }, + attempts: [ + { + attempt_id: 'att-p4-1', + stage: 'UNKNOWN', + created_at: new Date().toISOString(), + }, + ], + evidence: [], + }; + + const mockLedger = { + getIntent: async () => mockIntent, + appendEvidence: async (_id: string, ev: EvidenceView) => { + evidenceAppended.push(ev); + }, + completeSubmission: async ( + _id: string, + _att: string, + res: SettlementResult, + ): Promise => { + if (currentState !== 'UNKNOWN' && currentState !== 'SUBMITTING') { + return { completed: false, reason: 'INVALID_STATE', currentState }; + } + currentVersion += 1; + if (res.kind === 'CONFIRMED') { + currentState = 'COMMITTED'; + return { completed: true, state: 'COMMITTED', version: currentVersion }; + } + if (res.kind === 'DEFINITELY_NOT_SUBMITTED') { + currentState = 'FAILED_SAFE'; + return { completed: true, state: 'FAILED_SAFE', version: currentVersion }; + } + return { completed: true, state: 'UNKNOWN', version: currentVersion }; + }, + recordRecoveryEvent: async (_id: string, eventId: string, payload: unknown) => { + if (persistedEvents.has(eventId)) { + return { inserted: false }; + } + persistedEvents.set(eventId, payload); + return { inserted: true }; + }, + getRecoveryEventPayload: async (eventId: string) => persistedEvents.get(eventId), + } as unknown as IntentLedger; + + const store = new IntentLedgerRecoveryCommandStore(mockLedger); + + const pack = { + schemaVersion: 'recovery-command-pack-v1' as const, + packId: 'pack-p4-1', + eventId: 'evt-reconcile-p4-1', + businessIntentId: 'intent-p4-1', + sourceStateVersion: '2', + generatedAt: new Date().toISOString(), + appendCommands: [ + { + schemaVersion: 'append-recovery-record-v1' as const, + commandId: 'cmd-p4-1', + operation: 'APPEND_RECOVERY_RECORD' as const, + businessIntentId: 'intent-p4-1', + expectedStateVersion: '2', + record: { + schemaVersion: 'recovery-record-v1' as const, + recordType: 'OBSERVATION' as const, + id: 'rec-1', + source: 'ARC' as const, + authorityClass: 'AUTHORITATIVE_CHAIN_EVIDENCE' as const, + binding: { + businessIntentId: 'intent-p4-1', + requestFingerprint: 'fp-p4-1', + network: 'eip155:5042002', + tokenContract: '0x0000000000000000000000000000000000000000', + recipient: sampleRequest.recipient, + amountAtomic: '1000000', + }, + blockNumber: '999123', + retrievedAt: new Date().toISOString(), + digest: 'digest-1', + }, + }, + ], + reconciliationCommand: { + schemaVersion: 'reconciliation-command-v1' as const, + commandType: 'MARK_COMMITTED' as const, + businessIntentId: 'intent-p4-1', + requestFingerprint: 'fp-p4-1', + targetState: 'COMMITTED' as const, + reason: 'Arc proof verified', + evidenceReferences: ['arc:0x' + 'e'.repeat(64)], + disposition: 'MARK_COMMITTED', + advisoryAction: 'RETURN_EXISTING_RESULT' as const, + authoritativeProofPresent: true, + issuedAt: new Date().toISOString(), + settlementPermission: 'NEVER' as const, + }, + recoveryView: {} as DetailedRecoveryView, + externalSubmissionCount: 0 as const, + }; + + // First append succeeds and transitions UNKNOWN -> COMMITTED + const res1 = await store.append(pack); + expect(res1.status).toBe('APPENDED'); + expect(evidenceAppended).toHaveLength(1); + expect(currentState).toBe('COMMITTED'); + expect(currentVersion).toBe(3); + + // Second append with same eventId deduplicates durably + const res2 = await store.append(pack); + expect(res2.status).toBe('DUPLICATE'); + + // Fail closed: missing arc reference in MARK_COMMITTED throws + const missingArcPack = { + ...pack, + eventId: 'evt-missing-arc', + sourceStateVersion: '3', + reconciliationCommand: { + ...pack.reconciliationCommand, + evidenceReferences: [], + }, + }; + await expect(store.append(missingArcPack)).rejects.toThrow( + 'authoritative Arc transaction reference', + ); + + // Fail closed: state version mismatch throws and does NOT record/wedge the event + const stalePack = { + ...pack, + eventId: 'evt-stale', + sourceStateVersion: '99', + }; + await expect(store.append(stalePack)).rejects.toThrow('State version mismatch'); + expect(await store.findByEventId('evt-stale')).toBeNull(); + }); + + it('PrivyArcEvidenceBridge derives verified proof envelope from real receipt and avoids fabricated data', async () => { + const mockEvidencePort = { + lookup: async () => 'FINAL_SUCCESS' as const, + }; + + const bridge = new PrivyArcEvidenceBridge({ + evidencePort: mockEvidencePort as unknown as LaneBEvidencePort, + receiptSource: { + getReceipt: async () => realReceipt, + }, + defaultArcTxHash: realTxHash, + }); + + const binding = { + businessIntentId: 'intent-p4-1', + requestFingerprint: 'fp-p4-1', + network: 'eip155:5042002', + tokenContract: '0x3333333333333333333333333333333333333333', + recipient: sampleRequest.recipient, + amountAtomic: '1000000', + }; + + const evidence = await bridge.read(binding); + expect(evidence.schemaVersion).toBe('recovery-evidence-v1'); + expect(evidence.binding.businessIntentId).toBe('intent-p4-1'); + expect(evidence.local.submissionReference).toBe('sub-intent-p4-1'); + expect(evidence.privy?.requestStatus).toBe('SUCCEEDED'); + expect(evidence.arc?.receiptStatus).toBe('SUCCESS'); + expect(evidence.arc?.submissionReference).toBe('sub-intent-p4-1'); // Must match local + expect(evidence.arc?.finality).toBe('FINAL'); + expect(evidence.arc?.blockNumber).toBe('999123'); + expect(evidence.arc?.blockHash).toBe(realBlockHash); + expect(evidence.arc?.transfer?.sender).toBe(realSender); + expect(evidence.arc?.transfer?.amountAtomic).toBe('1000000'); + + // Without receipt source or verified tx, arc evidence is null (not fabricated) + const emptyBridge = new PrivyArcEvidenceBridge(); + const emptyEvidence = await emptyBridge.read(binding); + expect(emptyEvidence.arc).toBeNull(); + }); + + it('PrivyArcEvidenceBridge integrated with RecoveryService converges UNKNOWN intent to COMMITTED without UNBOUND_EVIDENCE contradiction', async () => { + let ledgerState: IntentResponse['state'] = 'UNKNOWN'; + let ledgerVersion = 2; + + const mockIntent: IntentResponse = { + ...sampleRequest, + payload_fingerprint: 'fp-p4-1', + get state() { + return ledgerState; + }, + get version() { + return ledgerVersion; + }, + attempts: [ + { + attempt_id: 'att-p4-1', + stage: 'UNKNOWN', + created_at: new Date().toISOString(), + }, + ], + evidence: [], + }; + + const mockLedger = { + getIntent: async () => mockIntent, + appendEvidence: async () => {}, + completeSubmission: async ( + _id: string, + _att: string, + res: SettlementResult, + ): Promise => { + if (ledgerState !== 'UNKNOWN' && ledgerState !== 'SUBMITTING') { + return { completed: false, reason: 'INVALID_STATE', currentState: ledgerState }; + } + ledgerVersion += 1; + ledgerState = res.kind === 'CONFIRMED' ? 'COMMITTED' : 'FAILED_SAFE'; + return { completed: true, state: ledgerState, version: ledgerVersion }; + }, + } as unknown as IntentLedger; + + const mockEvidencePort = { + lookup: async () => 'FINAL_SUCCESS' as const, + }; + + const recoveryService = createProductionRecoveryService(mockLedger, { + evidencePort: mockEvidencePort as unknown as LaneBEvidencePort, + receiptSource: { + getReceipt: async () => realReceipt, + }, + defaultArcTxHash: realTxHash, + }); + + const job = { + schemaVersion: 'recovery-job-v1' as const, + eventId: 'reconcile:intent-p4-1:2', + businessIntentId: 'intent-p4-1', + requestedAt: new Date().toISOString(), + }; + + const result = await recoveryService.handle(job); + expect(result.status).toBe('PROCESSED'); + expect(result.pack?.reconciliationCommand.commandType).toBe('MARK_COMMITTED'); + expect(result.pack?.reconciliationCommand.targetState).toBe('COMMITTED'); + expect(result.externalSubmissionCount).toBe(0); + expect(ledgerState).toBe('COMMITTED'); + expect(ledgerVersion).toBe(3); + }); + + it('executeReconcileIntent handles UNKNOWN intent through RecoveryService with zero submissions', async () => { + let completedWith: SettlementResult | null = null; + let ledgerState: IntentResponse['state'] = 'UNKNOWN'; + + const mockIntent: IntentResponse = { + ...sampleRequest, + payload_fingerprint: 'fp-p4-1', + get state() { + return ledgerState; + }, + version: 7, + attempts: [ + { + attempt_id: 'att-p4-1', + stage: 'UNKNOWN', + created_at: new Date().toISOString(), + }, + ], + evidence: [], + }; + + const mockLedger = { + getIntent: async () => mockIntent, + appendEvidence: async () => {}, + completeSubmission: async ( + _id: string, + _att: string, + res: SettlementResult, + ): Promise => { + completedWith = res; + ledgerState = res.kind === 'CONFIRMED' ? 'COMMITTED' : 'FAILED_SAFE'; + return { completed: true, state: ledgerState, version: 8 }; + }, + } as unknown as IntentLedger; + + const commandStore = new IntentLedgerRecoveryCommandStore(mockLedger); + const composition = createRecoverySimulatorComposition({ + businessIntentId: mockIntent.business_intent_id, + commandStore, + }); + + const mockPool = {} as unknown as Pool; + const worker = composeWorker(mockPool, mockLedger, { + profile: 'simulator', + recoveryService: composition.service, + }); + + await executeReconcileIntent( + mockIntent.business_intent_id, + worker.options, + composition.job.eventId, + ); + + expect(completedWith).not.toBeNull(); + expect(completedWith?.kind).toBe('CONFIRMED'); + expect(ledgerState).toBe('COMMITTED'); + const stored = await commandStore.findByEventId(composition.job.eventId); + expect(stored).not.toBeNull(); + expect(stored?.externalSubmissionCount).toBe(0); + }); +}); diff --git a/apps/worker/tsconfig.json b/apps/worker/tsconfig.json index 1d4aeba..8e0c05c 100644 --- a/apps/worker/tsconfig.json +++ b/apps/worker/tsconfig.json @@ -7,8 +7,11 @@ }, "include": ["src/**/*.ts"], "references": [ + { "path": "../../packages/arc-adapter" }, { "path": "../../packages/contracts" }, { "path": "../../packages/domain" }, + { "path": "../../packages/privy-adapter" }, + { "path": "../../packages/reconciliation" }, { "path": "../../packages/storage-postgres" } ] } diff --git a/docs/GATE_P4_CHECKLIST.md b/docs/GATE_P4_CHECKLIST.md index 868fead..65b6a1e 100644 --- a/docs/GATE_P4_CHECKLIST.md +++ b/docs/GATE_P4_CHECKLIST.md @@ -12,15 +12,15 @@ At Gate P4, checked simulators are replaced with real reviewed package versions, ## Package Version Slots -| Slot | Planned Package | Owning Lane | Current State in A04 | +| Slot | Planned Package | Owning Lane | Current State in Gate P4 | | --- | --- | --- | --- | | Core Contracts | `@oneshot/contracts@0.1.0` | Shared / Frozen | Pinned | | Domain Models | `@oneshot/domain@0.1.0` | Lane A | Pinned | | PostgreSQL Storage | `@oneshot/storage-postgres@0.1.0` | Lane A | Pinned (Schema Digest: `5d5888894ff0f4f44049579f1c8ffca2a24e0b61c3af65aabdbcd78f06020d65`) | -| Settlement Worker | `@oneshot/worker@0.1.0` | Lane A | Composed | -| Arc Settlement Adapter | `@oneshot/privy-adapter` (`ArcSettlementAdapter`) | Lane B | Simulated via `SimulatorSettlementPort` | -| Privy Authorization Adapter | `@oneshot/privy-adapter` (`PrivyAuthorizationAdapter`) | Lane B | Simulated via `SimulatorAuthorizationPort` | -| Subgraph MCP Recovery | `@oneshot/reconciliation` | Lane C | Simulated via `c01-simulator-v1` scenarios | +| Settlement Worker | `@oneshot/worker@0.1.0` | Lane A | Composed & Converged | +| Arc Settlement Adapter | `@oneshot/privy-adapter` (`ArcSettlementAdapter`) | Lane B | Integrated & Wired in Production Profile | +| Privy Authorization Adapter | `@oneshot/privy-adapter` (`PrivyAuthorizationAdapter`) | Lane B | Integrated & Wired in Production Profile | +| Subgraph MCP Recovery | `@oneshot/reconciliation` | Lane C | Integrated & Wired via `recovery-bridge` | Both lane-B adapters ship from `@oneshot/privy-adapter` rather than from separate packages: settlement is a Privy wallet action carrying an Arc @@ -31,20 +31,22 @@ readiness probing they build on. See ## Replacement Instructions for Gate P4 -1. **Replace Settlement Port**: - - In `apps/worker/src/composition.ts`, update `composeWorker`: +1. **Replace Settlement Port**: [COMPLETED] + - In `apps/worker/src/composition.ts`, updated `composeWorker`: - Set `profile: 'production'`. - - Inject instance of `ArcSettlementAdapter` conforming to `SettlementPort`. - - Verify contract version `1.0.0` and network `eip155:5042002`. - -2. **Replace Authorization Port**: - - Inject instance of `PrivyAuthorizationAdapter` conforming to `AuthorizationPort`. - - Verify contract version `1.0.0`. - -3. **Replace Recovery Engine**: - - Follow `packages/reconciliation/docs/GATE_P4_RECOVERY_REPLACEMENT.md`. - - Inject `RecoveryService` into the `reconcile_intent` task in `createTaskList`. - - Keep A-owned persistence behind `RecoveryCommandStorePort`; the recovery package does not write A tables. + - Conformance-verified and wired injection support for real `ArcSettlementAdapter` conforming to `SettlementPort`. + - Verified contract version `1.0.0` and network `eip155:5042002`. + +2. **Replace Authorization Port**: [COMPLETED] + - Conformance-verified and wired injection support for real `PrivyAuthorizationAdapter` conforming to `AuthorizationPort`. + - Verified contract version `1.0.0`. + +3. **Replace Recovery Engine**: [COMPLETED] + - Followed `packages/reconciliation/docs/GATE_P4_RECOVERY_REPLACEMENT.md`. + - Injected `RecoveryService` into the `reconcile_intent` task in `createTaskList`. + - Bridged `LocalRecoveryStatePort`, `RecoveryCommandStorePort` over `IntentLedger` (with durable `outbox_jobs` deduplication and real `UNKNOWN` CAS transitions), and `KnownIdentityEvidencePort` via `PrivyArcEvidenceBridge` in `apps/worker/src/recovery-bridge.ts`. + - Provided `createProductionRecoveryService` in `apps/worker/src/composition.ts` for full production worker recovery composition. + - Kept A-owned persistence behind `RecoveryCommandStorePort`; the recovery package does not write A tables. 4. **Freeze the frontend boundary**: - Revalidate the A01 OpenAPI v1 artifact against the composed backend. diff --git a/packages/storage-postgres/src/ledger.ts b/packages/storage-postgres/src/ledger.ts index 1600a3d..a8e93fe 100644 --- a/packages/storage-postgres/src/ledger.ts +++ b/packages/storage-postgres/src/ledger.ts @@ -443,7 +443,7 @@ export class IntentLedger { await client.query('ROLLBACK'); return { completed: false, reason: 'NOT_FOUND' }; } - if (row.state !== 'SUBMITTING') { + if (row.state !== 'SUBMITTING' && row.state !== 'UNKNOWN') { await client.query('ROLLBACK'); return { completed: false, reason: 'INVALID_STATE', currentState: row.state }; } @@ -451,10 +451,14 @@ export class IntentLedger { const now = this.#dependencies.now(); if (result.kind === 'CONFIRMED') { - await client.query( - 'UPDATE business_intents SET state = $1, version = $2, updated_at = $3 WHERE business_intent_id = $4 AND state = $5', - ['COMMITTED', newVersion, now, id, 'SUBMITTING'], + const updateRes = await client.query( + 'UPDATE business_intents SET state = $1, version = $2, updated_at = $3 WHERE business_intent_id = $4 AND (state = $5 OR state = $6)', + ['COMMITTED', newVersion, now, id, 'SUBMITTING', 'UNKNOWN'], ); + if (updateRes.rowCount !== 1) { + await client.query('ROLLBACK'); + return { completed: false, reason: 'INVALID_STATE', currentState: row.state }; + } await client.query( `INSERT INTO settlements ( business_intent_id, provider_reference_id, transaction_hash, @@ -486,10 +490,14 @@ export class IntentLedger { } if (result.kind === 'DEFINITELY_NOT_SUBMITTED') { - await client.query( - 'UPDATE business_intents SET state = $1, version = $2, updated_at = $3 WHERE business_intent_id = $4 AND state = $5', - ['FAILED_SAFE', newVersion, now, id, 'SUBMITTING'], + const updateRes = await client.query( + 'UPDATE business_intents SET state = $1, version = $2, updated_at = $3 WHERE business_intent_id = $4 AND (state = $5 OR state = $6)', + ['FAILED_SAFE', newVersion, now, id, 'SUBMITTING', 'UNKNOWN'], ); + if (updateRes.rowCount !== 1) { + await client.query('ROLLBACK'); + return { completed: false, reason: 'INVALID_STATE', currentState: row.state }; + } await client.query( 'UPDATE attempts SET stage = $1, sanitized_error = $2 WHERE attempt_id = $3', ['FAILED_SAFE', result.reason, attemptId], @@ -499,29 +507,33 @@ export class IntentLedger { } // POSSIBLY_SUBMITTED or any unexpected variant -> UNKNOWN - await client.query( - 'UPDATE business_intents SET state = $1, version = $2, updated_at = $3 WHERE business_intent_id = $4 AND state = $5', - ['UNKNOWN', newVersion, now, id, 'SUBMITTING'], - ); - await client.query( - 'UPDATE attempts SET stage = $1, sanitized_error = $2 WHERE attempt_id = $3', - [ - 'UNKNOWN', - (result as { reason?: string } | null | undefined)?.reason ?? - 'Settlement outcome uncertain', - attemptId, - ], - ); - await client.query( - `INSERT INTO outbox_jobs ( - business_intent_id, job_key, task_identifier, payload, - available_at, created_at - ) VALUES ($1, $2, 'reconcile_intent', $3::jsonb, $4, $4) - ON CONFLICT (job_key) DO NOTHING`, - [id, `reconcile:${id}:${newVersion}`, JSON.stringify({ business_intent_id: id }), now], - ); + if (row.state === 'SUBMITTING') { + await client.query( + 'UPDATE business_intents SET state = $1, version = $2, updated_at = $3 WHERE business_intent_id = $4 AND state = $5', + ['UNKNOWN', newVersion, now, id, 'SUBMITTING'], + ); + await client.query( + 'UPDATE attempts SET stage = $1, sanitized_error = $2 WHERE attempt_id = $3', + [ + 'UNKNOWN', + (result as { reason?: string } | null | undefined)?.reason ?? + 'Settlement outcome uncertain', + attemptId, + ], + ); + await client.query( + `INSERT INTO outbox_jobs ( + business_intent_id, job_key, task_identifier, payload, + available_at, created_at + ) VALUES ($1, $2, 'reconcile_intent', $3::jsonb, $4, $4) + ON CONFLICT (job_key) DO NOTHING`, + [id, `reconcile:${id}:${newVersion}`, JSON.stringify({ business_intent_id: id }), now], + ); + await client.query('COMMIT'); + return { completed: true, state: 'UNKNOWN', version: newVersion }; + } await client.query('COMMIT'); - return { completed: true, state: 'UNKNOWN', version: newVersion }; + return { completed: true, state: 'UNKNOWN', version: row.version }; } catch (error) { await client.query('ROLLBACK'); throw error; @@ -729,4 +741,31 @@ export class IntentLedger { client.release(); } } + + async recordRecoveryEvent( + businessIntentId: unknown, + eventId: string, + payload: unknown, + ): Promise<{ readonly inserted: boolean }> { + const id = asBusinessIntentId(businessIntentId); + const now = this.#dependencies.now(); + const result = await this.#pool.query( + `INSERT INTO outbox_jobs ( + business_intent_id, job_key, task_identifier, payload, + status, available_at, created_at + ) VALUES ($1, $2, 'reconcile_intent', $3::jsonb, 'DELIVERED', $4, $4) + ON CONFLICT (job_key) DO NOTHING + RETURNING outbox_job_id`, + [id, `recovery-event:${eventId}`, JSON.stringify(payload), now], + ); + return { inserted: (result.rowCount ?? 0) === 1 }; + } + + async getRecoveryEventPayload(eventId: string): Promise { + const result = await this.#pool.query<{ payload: unknown }>( + 'SELECT payload FROM outbox_jobs WHERE job_key = $1', + [`recovery-event:${eventId}`], + ); + return result.rows[0]?.payload; + } } diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3cfdd74..523e19e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -63,12 +63,21 @@ importers: apps/worker: dependencies: + '@oneshot/arc-adapter': + specifier: workspace:* + version: link:../../packages/arc-adapter '@oneshot/contracts': specifier: workspace:* version: link:../../packages/contracts '@oneshot/domain': specifier: workspace:* version: link:../../packages/domain + '@oneshot/privy-adapter': + specifier: workspace:* + version: link:../../packages/privy-adapter + '@oneshot/reconciliation': + specifier: workspace:* + version: link:../../packages/reconciliation '@oneshot/storage-postgres': specifier: workspace:* version: link:../../packages/storage-postgres diff --git a/tsconfig.json b/tsconfig.json index a8b660c..9e245df 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -17,19 +17,19 @@ "path": "./packages/testkit-domain" }, { - "path": "./apps/api" + "path": "./packages/arc-adapter" }, { - "path": "./apps/worker" + "path": "./packages/privy-adapter" }, { - "path": "./packages/arc-adapter" + "path": "./packages/testkit-settlement" }, { - "path": "./packages/privy-adapter" + "path": "./apps/api" }, { - "path": "./packages/testkit-settlement" + "path": "./apps/worker" } ] }