diff --git a/.agent/context/20260913T050212Z-always-on-graph-evidence.md b/.agent/context/20260913T050212Z-always-on-graph-evidence.md new file mode 100644 index 0000000..3978175 --- /dev/null +++ b/.agent/context/20260913T050212Z-always-on-graph-evidence.md @@ -0,0 +1,112 @@ +# Session Context: always-on-graph-evidence + +## Date/time + +- UTC: 2026-09-13T05:02:12Z + +## User goal + +Make The Graph evidence appear for every confirmed transaction, then commit, +push, and open a draft PR. The PR must state that Gate A was not started. + +## Original prompt/request + +"right know we use the graph only if payment failed or something got wrong. We +want that the graph evidence will apear always, on every transaction. Implement +this feature. Also make commit and push after you finished. And alos make a +draft pr without starting gate A, but write in pr msg that GATE a wasn't +started" + +## Assumptions + +- “Every transaction” means every confirmed OneShot settlement, including + server-wallet and user-wallet commits; failed-safe attempts are not + transactions. +- Graph evidence remains non-authoritative and cannot change settlement or + retry permission. +- A durable outbox task is preferable to an inline post-commit call so a worker + restart cannot permanently lose the evidence capture. +- The existing modified `.agent/context/20260912T-user-wallet-payment.md` is + unrelated user work and must remain unstaged. + +## Plan + +1. Add an idempotent post-commit Graph evidence outbox task and worker port. +2. Wire production recovery configuration to capture Graph observations and + record `UNAVAILABLE` evidence on Graph boundary failure. +3. Add focused worker/recovery coverage and update migration/integration + expectations. +4. Run local checks, inspect the exact staged tree, commit, push, and create a + draft PR without starting Gate A. + +## Key decisions + +- Use a separate `capture_graph_evidence` task instead of running the recovery + LLM for successful payments. This keeps normal execution read-only and + avoids turning evidence capture into a retry/reconciliation decision. +- Make `(business_intent_id, source, digest)` unique and use `ON CONFLICT DO +NOTHING`, so redelivery after a crash does not duplicate evidence. +- Query the durable payer wallet for user-wallet jobs when constructing the + Graph correlation request; server-wallet intents retain the configured + sender fallback. + +## Files/components touched + +- `packages/storage-postgres/migrations/012_graph_evidence.sql` +- `packages/storage-postgres/src/ledger.ts` +- `apps/worker/src/types.ts` +- `apps/worker/src/recovery-bridge.ts` +- `apps/worker/src/composition.ts` +- `apps/worker/src/worker.ts` +- `apps/worker/README.md` and `apps/worker/FAILURE_CATALOG.md` +- focused worker tests and storage integration expectations + +## Commands/checks + +- Repository policy and routed documents read: `.agent/AGENTS.md`, project + context, security invariants, sponsor requirements, test matrix, + implementation loop, and `oneshot-idempotency/SKILL.md`. +- Branch created from current `develop`: `feature/always-on-graph-evidence`. +- `pnpm.cmd --filter @oneshot/worker test` - 7 files / 50 tests passed. +- `pnpm.cmd --filter @oneshot/reconciliation test` - 8 files / 89 tests passed. +- `pnpm.cmd --filter @oneshot/storage-postgres test` - 4 files / 15 tests passed + (PostgreSQL-gated tests skipped without a container runtime). +- `pnpm.cmd --filter @oneshot/settlement-ui test` - 5 files / 211 tests passed. +- `pnpm.cmd test` - 80 files / 1,055 tests passed. +- Worker, reconciliation, and storage typechecks plus root lint/build passed. +- `git diff --check` passed; targeted Prettier checks passed after formatting. + +## External-doc findings + +- Repository policy defines OneShot as authoritative for settlement and The + Graph as non-authoritative candidate discovery; missing/delayed index data + cannot authorize payment. +- Test matrix requires Graph boundary failures to fail closed and durable + evidence metadata to be retained. + +## Unresolved questions + +- None; Graph evidence capture may be unavailable, but the observation must + still be recorded with `UNAVAILABLE` freshness. + +## Git and PR state + +- Branch: `feature/always-on-graph-evidence` +- Base: `develop` at `62920523c5a323cfc0e38d57c632f498b4d921d5` +- Feature commit: `4fea7e66d6187760052d24508d47e73fa2b8daca` +- Feature tree: `86f39915dbab92450483c367f9fc0df67bae6172` +- Remote branch: pushed to `origin/feature/always-on-graph-evidence` +- Draft PR: [#124](https://github.com/SWOFART/OneShot/pull/124) +- PR head at creation: feature commit/tree above +- CI: GitHub checks are pending/queued; local validation passed + +## Review gates + +- Gate A: NOT RUN (explicitly requested to skip) +- Gate B: NOT RUN + +## Handoff/next steps + +1. Keep the unrelated `.agent/context/20260912T-user-wallet-payment.md` + modification unstaged. +2. Human review and CI follow-up remain; do not start Gate A or Gate B. diff --git a/apps/worker/FAILURE_CATALOG.md b/apps/worker/FAILURE_CATALOG.md index 83fef0f..1a50fc6 100644 --- a/apps/worker/FAILURE_CATALOG.md +++ b/apps/worker/FAILURE_CATALOG.md @@ -11,6 +11,7 @@ This catalog documents the external-boundary failure points, expected state tran | **FP-03: Provider error / timeout** | Port throws network exception or timeout | `SUBMITTING` | 1 | Transition to `UNKNOWN`, persist sanitized error, enqueue reconciliation | No retry without proof | | **FP-04: Definitive rejection** | Port returns `DEFINITELY_NOT_SUBMITTED` | `SUBMITTING` | 1 | Transition to `FAILED_SAFE`, persist failure reason; a policy layer may schedule a fresh authorization attempt | No retry without authoritative no-effect proof | | **FP-05: Downstream failure after commit** | Failure after `COMMITTED` state and settlement persisted | `COMMITTED` | 1 | Settlement remains permanently recorded; no replacement payment | Settlement identity is immutable | +| **FP-05a: Graph evidence read failure** | Graph lookup fails after the committed settlement transaction | `COMMITTED` | 1 | Durable evidence task records `THE_GRAPH` with `UNAVAILABLE`; settlement remains committed | Index evidence never controls payment | | **FP-06: 10 Parallel workers storm** | 10 workers race on same `READY` intent | `READY` | 1 (winner only) | Exactly 1 worker wins CAS to `SUBMITTING`; 9 workers exit without calling port | Exactly 1 committed settlement | | **FP-07: 10 Sequential deliveries** | Same intent job delivered 10 times in sequence | `AUTHORIZING` $\rightarrow$ `COMMITTED` | 1 | First delivery commits settlement; subsequent deliveries find `COMMITTED` and exit | At most 1 settlement | diff --git a/apps/worker/README.md b/apps/worker/README.md index 7549bdb..29cfb42 100644 --- a/apps/worker/README.md +++ b/apps/worker/README.md @@ -14,6 +14,10 @@ Atomic at-most-once execution worker for OneShot Business Intents. - `authorize_intent`: Validates intent against corporate spending and policy rules, advancing state to `READY` (or `REJECTED`). - `submit_settlement`: Atomically claims submission right and executes settlement via configured settlement port. - `reconcile_intent`: Reconciles ambiguous intent state against evidence observations. +- `capture_graph_evidence`: Reads the pinned Graph source after a confirmed + settlement and appends a non-authoritative observation. Graph failure is + recorded as `UNAVAILABLE`; it never changes settlement state or grants a + retry. ## Architecture and Dispatch diff --git a/apps/worker/src/composition.ts b/apps/worker/src/composition.ts index b2aafc0..2ce4cfb 100644 --- a/apps/worker/src/composition.ts +++ b/apps/worker/src/composition.ts @@ -13,6 +13,7 @@ import { TeamReportSupplier } from '@oneshot/supplier-adapter'; import type { Pool } from 'pg'; import type { AuthorizationPort, + GraphEvidenceCapturePort, SettlementContext, SettlementPort, WorkerOptions, @@ -27,6 +28,7 @@ import { IntentLedgerLocalRecoveryStatePort, IntentLedgerRecoveryCommandStore, PrivyArcEvidenceBridge, + IntentLedgerGraphEvidenceCapturePort, type IntentLedgerLocalRecoveryStatePortOptions, type PrivyArcEvidenceBridgeOptions, } from './recovery-bridge.js'; @@ -121,6 +123,7 @@ export interface CompositionOptions { readonly contractVersion?: string; }; readonly recoveryService?: RecoveryService; + readonly graphEvidence?: GraphEvidenceCapturePort; readonly recovery?: ProductionRecoveryServiceOptions; readonly submissionsDisabled?: boolean; readonly expectedContractVersion?: string; @@ -164,12 +167,23 @@ export function composeWorker( } let recoveryService = options.recoveryService; + let graphEvidence = options.graphEvidence; if (!recoveryService && options.profile === 'production' && options.recovery) { recoveryService = createProductionRecoveryService(ledger, options.recovery); } + if (!graphEvidence && options.profile === 'production' && options.recovery) { + const localState = new IntentLedgerLocalRecoveryStatePort(ledger, options.recovery.localState); + graphEvidence = new IntentLedgerGraphEvidenceCapturePort( + localState, + options.recovery.subgraphMcp, + ); + } if (options.profile === 'production' && !recoveryService) { throw new Error('Production composition profile requires an injected recoveryService'); } + if (options.profile === 'production' && !graphEvidence) { + throw new Error('Production composition profile requires an injected graphEvidence port'); + } const workerOptions: WorkerOptions = { pool, @@ -177,6 +191,7 @@ export function composeWorker( settlementPort, authorizationPort, recoveryService, + graphEvidence, jobLedger: new JobLedger(pool, { now: () => new Date(), nextAttemptId: randomUUID }), supplier: options.supplier ?? new TeamReportSupplier(), config: { diff --git a/apps/worker/src/recovery-bridge.ts b/apps/worker/src/recovery-bridge.ts index 760431b..4260fc4 100644 --- a/apps/worker/src/recovery-bridge.ts +++ b/apps/worker/src/recovery-bridge.ts @@ -21,6 +21,7 @@ import { type RecoveryCommandPack, type RecoveryCommandStorePort, type RecoveryCommandStoreResult, + type SubgraphMcpRecoveryPort, type SubgraphMcpPolicy, } from '@oneshot/reconciliation'; import { @@ -30,6 +31,7 @@ import { type TransactionReceipt, } from '@oneshot/arc-adapter'; import type { EvidencePort as LaneBEvidencePort } from '@oneshot/privy-adapter'; +import type { GraphEvidenceCapturePort, GraphEvidenceCaptureRequest } from './types.js'; import { createHash } from 'node:crypto'; function sha256Hex(value: string): string { @@ -92,6 +94,12 @@ export class IntentLedgerLocalRecoveryStatePort implements LocalRecoveryStatePor const durableState = mapState(intent.state); const nowIso = new Date().toISOString(); + const ledgerWithPayer = this.ledger as IntentLedger & { + getPaymentPayerWallet?: (businessIntentId: string) => Promise; + }; + const payerWallet = ledgerWithPayer.getPaymentPayerWallet + ? await ledgerWithPayer.getPaymentPayerWallet(businessIntentId) + : undefined; const toBlock = this.options.getToBlock ? await this.options.getToBlock() : this.options.toBlock; @@ -104,7 +112,7 @@ export class IntentLedgerLocalRecoveryStatePort implements LocalRecoveryStatePor binding, correlation: { strategy: 'TRANSFER_TUPLE_WINDOW', - sender: this.options.correlationSender, + sender: payerWallet ?? this.options.correlationSender, fromBlock: this.options.fromBlock, toBlock, }, @@ -146,6 +154,66 @@ export class IntentLedgerLocalRecoveryStatePort implements LocalRecoveryStatePor } } +/** + * Captures non-authoritative Graph evidence for a confirmed settlement. This + * port never reads or writes settlement authority; it only produces a bounded + * observation for the durable evidence timeline. + */ +export class IntentLedgerGraphEvidenceCapturePort implements GraphEvidenceCapturePort { + constructor( + private readonly localState: LocalRecoveryStatePort, + private readonly subgraphMcp: SubgraphMcpRecoveryPort, + private readonly now: () => string = () => new Date().toISOString(), + ) {} + + async capture(request: GraphEvidenceCaptureRequest): Promise { + const retrievedAt = this.now(); + try { + const snapshot = await this.localState.read(request.businessIntentId); + const outcome = await this.subgraphMcp.lookup(snapshot.indexRequest, snapshot.mcpPolicy); + const view = outcome.view; + const digest = sha256Hex( + JSON.stringify({ + businessIntentId: request.businessIntentId, + transactionHash: request.transactionHash, + blockNumber: request.blockNumber, + graph: view.graph, + observedThrough: view.observedThrough, + health: view.health, + candidates: view.candidates.map((candidate) => ({ + id: candidate.id, + transactionHash: candidate.transactionHash, + logIndex: candidate.logIndex, + blockNumber: candidate.blockNumber, + bindingStatus: candidate.bindingStatus, + })), + diagnostics: view.diagnostics, + }), + ); + return { + source: 'THE_GRAPH', + authority_class: 'OBSERVATION', + retrieved_at: view.retrievedAt, + digest: `graph-capture:${digest}`, + ...(view.observedThrough?.blockNumber + ? { block_number: view.observedThrough.blockNumber } + : {}), + freshness: view.health, + }; + } catch { + return { + source: 'THE_GRAPH', + authority_class: 'OBSERVATION', + retrieved_at: retrievedAt, + digest: `graph-capture:${sha256Hex( + `graph-unavailable:${request.businessIntentId}:${request.transactionHash}:${request.blockNumber}`, + )}`, + freshness: 'UNAVAILABLE', + }; + } + } +} + interface DurableLedgerExtension { recordRecoveryEvent?: ( businessIntentId: string, diff --git a/apps/worker/src/types.ts b/apps/worker/src/types.ts index aeee07f..5350842 100644 --- a/apps/worker/src/types.ts +++ b/apps/worker/src/types.ts @@ -1,6 +1,7 @@ import type { AuthorizationResult, CreateIntentRequest, + EvidenceView, SettlementResult, } from '@oneshot/contracts'; import type { IntentLedger } from '@oneshot/storage-postgres'; @@ -34,6 +35,16 @@ export interface SettlementPort { submit(request: CreateIntentRequest, context: SettlementContext): Promise; } +export interface GraphEvidenceCaptureRequest { + readonly businessIntentId: string; + readonly transactionHash: string; + readonly blockNumber: string; +} + +export interface GraphEvidenceCapturePort { + capture(request: GraphEvidenceCaptureRequest): Promise; +} + export interface WorkerConfig { readonly submissionsDisabled?: boolean | undefined; readonly authorizationRetryDelayMs?: number | undefined; @@ -48,6 +59,7 @@ export interface WorkerOptions { readonly authorizationPort?: AuthorizationPort | undefined; readonly settlementPort: SettlementPort; readonly recoveryService?: RecoveryService | undefined; + readonly graphEvidence?: GraphEvidenceCapturePort | undefined; readonly jobLedger?: JobLedger | undefined; readonly supplier?: SupplierPort | undefined; readonly concurrency?: number | undefined; diff --git a/apps/worker/src/worker.ts b/apps/worker/src/worker.ts index 00a53c2..b68a282 100644 --- a/apps/worker/src/worker.ts +++ b/apps/worker/src/worker.ts @@ -1,8 +1,35 @@ -import type { AuthorizationResult, SettlementResult } from '@oneshot/contracts'; +import { + asBlockNumber, + asTransactionHash, + type AuthorizationResult, + type 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'; +import type { GraphEvidenceCaptureRequest, WorkerOptions } from './types.js'; + +function graphEvidenceRequest( + businessIntentId: string, + payload: unknown, +): GraphEvidenceCaptureRequest { + if (payload === null || typeof payload !== 'object' || Array.isArray(payload)) { + throw new Error('Graph evidence task payload is invalid'); + } + const record = payload as Record; + if ( + record.business_intent_id !== businessIntentId || + typeof record.transaction_hash !== 'string' || + typeof record.block_number !== 'string' + ) { + throw new Error('Graph evidence task payload does not match the outbox identity'); + } + return { + businessIntentId, + transactionHash: asTransactionHash(record.transaction_hash), + blockNumber: asBlockNumber(record.block_number), + }; +} export async function executeAuthorizeIntent( businessIntentId: string, @@ -181,6 +208,15 @@ export async function executeReconcileIntent( } } +export async function executeCaptureGraphEvidence( + request: GraphEvidenceCaptureRequest, + options: WorkerOptions, +): Promise { + if (!options.graphEvidence) return; + const observation = await options.graphEvidence.capture(request); + await options.ledger.appendEvidence(request.businessIntentId, observation); +} + /** Supplier fulfillment is deliberately reachable only from a committed job. * It cannot change payment state or construct a replacement settlement. */ export async function executeFulfillSupplierOrder( @@ -225,6 +261,16 @@ export function createTaskList(options: WorkerOptions): TaskList { await executeReconcileIntent(business_intent_id, options, event_id); } }, + capture_graph_evidence: async (payload) => { + const record = payload as { business_intent_id?: unknown }; + if (typeof record.business_intent_id !== 'string') { + throw new Error('Graph evidence task is missing business_intent_id'); + } + await executeCaptureGraphEvidence( + graphEvidenceRequest(record.business_intent_id, payload), + options, + ); + }, fulfill_supplier_order: async (payload) => { const { job_id, delivery_attempt } = payload as { job_id?: string; @@ -285,6 +331,15 @@ export async function drainOutboxJobs(options: WorkerOptions, maxJobs = 100): Pr await executeSubmitSettlement(job.business_intent_id, options); } else if (job.task_identifier === 'reconcile_intent') { await executeReconcileIntent(job.business_intent_id, options, job.job_key); + } else if (job.task_identifier === 'capture_graph_evidence') { + const payload = await client.query<{ payload: unknown }>( + 'SELECT payload FROM outbox_jobs WHERE outbox_job_id = $1', + [job.outbox_job_id], + ); + await executeCaptureGraphEvidence( + graphEvidenceRequest(job.business_intent_id, payload.rows[0]?.payload), + options, + ); } else if (job.task_identifier === 'fulfill_supplier_order') { const payload = await client.query<{ payload: { job_id?: string; delivery_attempt?: number }; diff --git a/apps/worker/test/composition.test.ts b/apps/worker/test/composition.test.ts index a0a41ba..7ded75e 100644 --- a/apps/worker/test/composition.test.ts +++ b/apps/worker/test/composition.test.ts @@ -97,6 +97,7 @@ describe('Worker composition and simulator profile (A04.4)', () => { settlementPort: incompatiblePort, authorizationPort: new SimulatorAuthorizationPort(), recoveryService: { handle: async () => ({}) } as never, + graphEvidence: { capture: async () => ({}) } as never, }); const readiness = await composed.checkReadiness(); @@ -152,6 +153,7 @@ describe('Worker composition and simulator profile (A04.4)', () => { settlementPort: wrongNetworkPort, authorizationPort: new SimulatorAuthorizationPort(), recoveryService: { handle: async () => ({}) } as never, + graphEvidence: { capture: async () => ({}) } as never, }); const readiness = await composed.checkReadiness(); diff --git a/apps/worker/test/graph-evidence.test.ts b/apps/worker/test/graph-evidence.test.ts new file mode 100644 index 0000000..921fe19 --- /dev/null +++ b/apps/worker/test/graph-evidence.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it, vi } from 'vitest'; +import { + createRecoverySimulatorComposition, + type SubgraphMcpRecoveryPort, +} from '@oneshot/reconciliation'; +import { IntentLedgerGraphEvidenceCapturePort } from '../src/recovery-bridge.js'; + +describe('post-commit Graph evidence capture', () => { + it('records a bounded Graph observation for a confirmed transaction', async () => { + const composition = createRecoverySimulatorComposition(); + const port = new IntentLedgerGraphEvidenceCapturePort( + composition.localState, + composition.subgraphMcp, + () => '2026-09-13T10:00:00.000Z', + ); + + const evidence = await port.capture({ + businessIntentId: composition.job.businessIntentId, + transactionHash: `0x${'a'.repeat(64)}`, + blockNumber: '110', + }); + + expect(evidence).toMatchObject({ + source: 'THE_GRAPH', + authority_class: 'OBSERVATION', + freshness: 'FRESH', + block_number: '112', + }); + expect(evidence.digest).toMatch(/^graph-capture:[0-9a-f]{64}$/u); + }); + + it('records unavailable evidence when the Graph boundary fails', async () => { + const composition = createRecoverySimulatorComposition(); + const unavailable: SubgraphMcpRecoveryPort = { + lookup: vi.fn().mockRejectedValue(new Error('Graph timeout')), + }; + const port = new IntentLedgerGraphEvidenceCapturePort( + composition.localState, + unavailable, + () => '2026-09-13T10:00:00.000Z', + ); + + await expect( + port.capture({ + businessIntentId: composition.job.businessIntentId, + transactionHash: `0x${'b'.repeat(64)}`, + blockNumber: '111', + }), + ).resolves.toMatchObject({ + source: 'THE_GRAPH', + authority_class: 'OBSERVATION', + freshness: 'UNAVAILABLE', + }); + }); +}); diff --git a/apps/worker/test/production-runtime.integration.test.ts b/apps/worker/test/production-runtime.integration.test.ts index cc895a2..8725101 100644 --- a/apps/worker/test/production-runtime.integration.test.ts +++ b/apps/worker/test/production-runtime.integration.test.ts @@ -93,6 +93,15 @@ describePostgres('production worker API to adapter path', () => { }, }, recoveryService: { handle: async () => ({}) } as never, + graphEvidence: { + capture: async () => ({ + source: 'THE_GRAPH' as const, + authority_class: 'OBSERVATION' as const, + retrieved_at: '2026-09-07T12:03:00.000Z', + digest: 'graph-capture:production-runtime-fixture', + freshness: 'FRESH' as const, + }), + } as never, }), }); let attemptCounter = 0; diff --git a/apps/worker/test/worker.integration.test.ts b/apps/worker/test/worker.integration.test.ts index 4161525..24d3ec1 100644 --- a/apps/worker/test/worker.integration.test.ts +++ b/apps/worker/test/worker.integration.test.ts @@ -93,6 +93,13 @@ describePostgres('Atomic at-most-once worker (A03)', () => { [sampleRequest.business_intent_id], ); expect(counts.rows[0]?.settlements).toBe('1'); + + const graphJobs = await pool.query<{ count: string }>( + `SELECT count(*)::text AS count FROM outbox_jobs + WHERE business_intent_id = $1 AND task_identifier = 'capture_graph_evidence'`, + [sampleRequest.business_intent_id], + ); + expect(graphJobs.rows[0]?.count).toBe('1'); }); it('concurrency storm: 10 parallel workers converge on exactly 1 submission and 1 settlement', async () => { @@ -287,9 +294,11 @@ describePostgres('Atomic at-most-once worker (A03)', () => { expect(committedIntent?.state).toBe('COMMITTED'); expect(portCalls).toBe(1); - // Third drain confirms all outbox jobs are drained + // Third drain delivers the durable Graph evidence task. This test profile + // does not inject a Graph port, so the handler safely becomes a no-op. const processedThird = await drainOutboxJobs(workerOptions, 1); - expect(processedThird).toBe(0); + expect(processedThird).toBe(1); + expect(await drainOutboxJobs(workerOptions, 1)).toBe(0); }); it('safe disable leaves submission work pending and re-enable settles exactly once', async () => { diff --git a/apps/worker/test/worker.test.ts b/apps/worker/test/worker.test.ts index 98e660c..a97a865 100644 --- a/apps/worker/test/worker.test.ts +++ b/apps/worker/test/worker.test.ts @@ -17,6 +17,7 @@ import { createTaskList, drainOutboxJobs, executeAuthorizeIntent, + executeCaptureGraphEvidence, executeFulfillSupplierOrder, executeSubmitSettlement, } from '../src/index.js'; @@ -94,6 +95,7 @@ describe('Worker Unit Logic', () => { }); expect(Object.keys(tasks).sort()).toEqual([ 'authorize_intent', + 'capture_graph_evidence', 'fulfill_supplier_order', 'reconcile_intent', 'submit_settlement', @@ -155,6 +157,45 @@ describe('Worker Unit Logic', () => { expect(completedState).toBe('CONFIRMED'); }); + it('persists Graph evidence from the durable post-commit task', async () => { + const appended: unknown[] = []; + const ledger = createMockLedger({ + async appendEvidence(_id, evidence) { + appended.push(evidence); + }, + }); + + await executeCaptureGraphEvidence( + { + businessIntentId: sampleRequest.business_intent_id, + transactionHash: `0x${'a'.repeat(64)}`, + blockNumber: '500', + }, + { + pool: {} as never, + ledger, + settlementPort: {} as never, + graphEvidence: { + async capture(request) { + expect(request.transactionHash).toBe(`0x${'a'.repeat(64)}`); + return { + source: 'THE_GRAPH', + authority_class: 'OBSERVATION', + retrieved_at: '2026-09-13T10:00:00.000Z', + digest: 'graph-digest', + block_number: '500', + freshness: 'FRESH', + }; + }, + }, + }, + ); + + expect(appended).toEqual([ + expect.objectContaining({ source: 'THE_GRAPH', freshness: 'FRESH' }), + ]); + }); + it('passes provider request identity into the atomic claim before calling the settlement port', async () => { const order: string[] = []; let claimedIdentity: unknown; diff --git a/packages/storage-postgres/MIGRATIONS.md b/packages/storage-postgres/MIGRATIONS.md index 84c923f..64ba966 100644 --- a/packages/storage-postgres/MIGRATIONS.md +++ b/packages/storage-postgres/MIGRATIONS.md @@ -12,11 +12,12 @@ silently edited or automatically reversed. ## Current schema digest -The append-only ledger, resumable-jobs, provider-identity, and personal MCP -credential migration set (`001` through `011`) has SHA-256 digest: +The append-only ledger, resumable-jobs, provider-identity, personal MCP +credential, and Graph evidence capture migration set (`001` through `012`) has +SHA-256 digest: ```text -c38fa0d21cd675b385b01304d64c328d57bf742500af61ec12117280e8a3245b +cfbda4ad89e7cb2bc88e4c0c0603ba274f20bbd40fa713bd6cf535a670b6ba9f ``` ## Containerized Testing Command diff --git a/packages/storage-postgres/migrations/012_graph_evidence.sql b/packages/storage-postgres/migrations/012_graph_evidence.sql new file mode 100644 index 0000000..4f1587c --- /dev/null +++ b/packages/storage-postgres/migrations/012_graph_evidence.sql @@ -0,0 +1,16 @@ +ALTER TABLE outbox_jobs + DROP CONSTRAINT outbox_jobs_task_identifier_check; + +ALTER TABLE outbox_jobs + ADD CONSTRAINT outbox_jobs_task_identifier_check + CHECK (task_identifier IN ( + 'authorize_intent', + 'submit_settlement', + 'reconcile_intent', + 'fulfill_supplier_order', + 'capture_graph_evidence' + )); + +CREATE UNIQUE INDEX evidence_observations_intent_source_digest_idx + ON evidence_observations (business_intent_id, source, digest) + WHERE source = 'THE_GRAPH' AND digest LIKE 'graph-capture:%'; diff --git a/packages/storage-postgres/src/ledger.ts b/packages/storage-postgres/src/ledger.ts index faa3ca7..88e3d70 100644 --- a/packages/storage-postgres/src/ledger.ts +++ b/packages/storage-postgres/src/ledger.ts @@ -497,6 +497,21 @@ export class IntentLedger { } } + async getPaymentPayerWallet(idValue: unknown): Promise { + const id = asBusinessIntentId(idValue); + const result = await this.#pool.query<{ + payment_mode: PaymentMode; + payer_wallet: string | null; + }>( + `SELECT payment_mode, payer_wallet + FROM resumable_jobs + WHERE business_intent_id = $1`, + [id], + ); + const row = result.rows[0]; + return row?.payment_mode === 'USER_WALLET' ? (row.payer_wallet ?? undefined) : undefined; + } + async getRecoveryView(idValue: unknown): Promise { const intent = await this.getIntent(idValue); if (!intent) return undefined; @@ -583,7 +598,8 @@ export class IntentLedger { `INSERT INTO evidence_observations ( business_intent_id, source, authority_class, retrieved_at, digest, block_number, freshness - ) VALUES ($1, $2, $3, $4, $5, $6, $7)`, + ) VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT DO NOTHING`, [ id, evidence.source, @@ -1164,6 +1180,22 @@ export class IntentLedger { ], ); } + await client.query( + `INSERT INTO outbox_jobs ( + business_intent_id, job_key, task_identifier, payload, available_at, created_at + ) VALUES ($1, $2, 'capture_graph_evidence', $3::jsonb, $4, $4) + ON CONFLICT (job_key) DO NOTHING`, + [ + id, + `graph-evidence:${id}:${newVersion}`, + JSON.stringify({ + business_intent_id: id, + transaction_hash: result.transaction_hash, + block_number: result.block_number, + }), + now, + ], + ); await client.query('COMMIT'); return { completed: true, state: 'COMMITTED', version: newVersion }; } diff --git a/packages/storage-postgres/src/migrations.ts b/packages/storage-postgres/src/migrations.ts index ef8f5a6..2d80633 100644 --- a/packages/storage-postgres/src/migrations.ts +++ b/packages/storage-postgres/src/migrations.ts @@ -41,7 +41,7 @@ async function migrationFiles(directory: string): Promise { const files = await migrationFiles(directory); diff --git a/packages/storage-postgres/test/ledger.integration.test.ts b/packages/storage-postgres/test/ledger.integration.test.ts index 8c9e7fe..0edb1ec 100644 --- a/packages/storage-postgres/test/ledger.integration.test.ts +++ b/packages/storage-postgres/test/ledger.integration.test.ts @@ -51,7 +51,9 @@ describePostgres('PostgreSQL intent ledger', () => { const versions = await pool.query<{ version: number }>( 'SELECT version FROM schema_versions ORDER BY version', ); - expect(versions.rows.map((row) => row.version)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]); + expect(versions.rows.map((row) => row.version)).toEqual([ + 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, + ]); expect(await migrationDigest()).toMatch(/^[0-9a-f]{64}$/u); }); @@ -59,7 +61,7 @@ describePostgres('PostgreSQL intent ledger', () => { const directory = await mkdtemp(join(tmpdir(), 'oneshot-migration-')); try { await writeFile( - join(directory, '012_broken.sql'), + join(directory, '013_broken.sql'), 'CREATE TABLE must_rollback (id integer); SELECT missing_function();', 'utf8', ); @@ -68,7 +70,7 @@ describePostgres('PostgreSQL intent ledger', () => { "SELECT to_regclass('public.must_rollback')::text AS name", ); expect(table.rows[0]?.name).toBeNull(); - const version = await pool.query('SELECT 1 FROM schema_versions WHERE version = 12'); + const version = await pool.query('SELECT 1 FROM schema_versions WHERE version = 13'); expect(version.rowCount).toBe(0); } finally { await rm(directory, { recursive: true, force: true }); @@ -612,6 +614,25 @@ describePostgres('PostgreSQL intent ledger', () => { } }); + it('deduplicates redelivered post-commit Graph evidence', async () => { + const ledger = newLedger(); + await ledger.createOrReplay(request, 'correlation-graph-evidence'); + const observation = { + source: 'THE_GRAPH' as const, + authority_class: 'OBSERVATION' as const, + retrieved_at: '2026-09-07T12:03:00.000Z', + digest: 'graph-capture:duplicate-safe', + block_number: '12345', + freshness: 'FRESH' as const, + }; + + await ledger.appendEvidence(request.business_intent_id, observation); + await ledger.appendEvidence(request.business_intent_id, observation); + + const intent = await ledger.getIntent(request.business_intent_id); + expect(intent?.evidence.filter((entry) => entry.source === 'THE_GRAPH')).toHaveLength(1); + }); + it('returns the persisted Recovery Agent and deterministic-core decision', async () => { const ledger = newLedger(); await ledger.createOrReplay(request, 'correlation-recovery-view');