fix(jobs): recover a delivery stranded in PENDING

.agent/context/20260913T170000Z-pending-delivery-refresh.md

@@ -44,8 +44,15 @@ Follow-up: skip Gate A and Gate B, open the pull request as a draft.
   after a report of unwanted traffic. The bound and the pending-only condition
   are what keep both reports satisfied; the resume control it removed stays
   removed.
-- A delivery stranded in `PENDING` server-side is out of scope here and is
-  recorded below as an unresolved risk.
+- A delivery stranded in `PENDING` is now recoverable. `PENDING` alone is not
+  treated as resumable: the deciding evidence is whether a `fulfill_supplier_order`
+  row is still queued (`status = 'PENDING'`) for that job. A row a worker is
+  currently holding is still queued, so an in-flight retrieval is never
+  duplicated; only a job with nothing left to move it is re-queued.
+- Selected `.agent/TEST_MATRIX.md` cases: downstream failure after payment (the
+  committed payment is preserved and no payment work is created on the recovery
+  path) and parallel/duplicate claim (the compare-and-set against the locked
+  row's state means two concurrent resumes cannot both claim one delivery).
 
 ## Files/components touched
 
@@ -55,13 +62,19 @@ Follow-up: skip Gate A and Gate B, open the pull request as a draft.
   stops once nothing is pending, and gives up on a delivery that stays pending.
 - `apps/web/browser/p5.spec.ts` - the read count is now a lower bound, because
   an exact count would flake once a pending delivery is re-read on a timer.
+- `packages/storage-postgres/src/jobs.ts` - `resumeDelivery` now also recovers a
+  delivery stranded in `PENDING` with no queued fulfilment row, claiming it
+  against the state read under the job's row lock.
+- `packages/storage-postgres/test/jobs.test.ts` - a queued or in-flight
+  fulfilment row is still left alone; a stranded one is re-queued under a fresh
+  `delivery_attempt`, with no payment work created.
 
 ## Commands/checks
 
 - `pnpm format:check` - PASS
 - `pnpm lint` - PASS
 - `pnpm typecheck` - PASS
-- `pnpm test` - PASS, 81 files / 1057 tests, includes build. The root vitest
+- `pnpm test` - PASS, 81 files / 1059 tests, includes build. The root vitest
   config covers no `.tsx` file, so the command below is the one that exercises
   these components.
 - `pnpm --filter @oneshot/web test` - PASS, 17 files / 97 tests
@@ -75,11 +88,11 @@ Follow-up: skip Gate A and Gate B, open the pull request as a draft.
 
 ## Unresolved questions
 
-- A delivery that is stranded in `PENDING` cannot be recovered: `resumeDelivery`
-  re-queues only `NOT_REQUESTED` or `RETRIEVAL_FAILED`, and the web resume
-  control was removed, so a `PENDING` job whose outbox row was already consumed
-  has no path forward. The bounded re-reads give up on such a job rather than
-  fixing it. This needs a separate backend change.
+- Nothing in the web UI calls `POST /v1/jobs/:jobId/resume`, so recovering a
+  stranded delivery still needs an API call. The control that used to do it was
+  removed deliberately; re-adding one is a product decision, not a defect fix.
+- Whether any currently stranded job exists in the user's environment is
+  unverified here: it is inferred from the state machine, not from their data.
 
 ## Git and PR state
 

packages/storage-postgres/src/jobs.ts

@@ -424,17 +424,39 @@ export class JobLedger {
         await client.query('COMMIT');
         return undefined;
       }
+      // A delivery is only PENDING legitimately while its fulfilment row is
+      // still queued or being worked. If no such row is left, nothing will ever
+      // move this job again: the worker no-ops a payload whose delivery_attempt
+      // no longer matches, marks the row DELIVERED, and PENDING was not
+      // resumable, so the job was stranded with the payment already committed.
+      // The queued-row check runs under the job's FOR UPDATE lock, so a
+      // concurrent resume cannot also claim it, and a row a worker currently
+      // holds is still status PENDING and so still counts as queued.
+      const queuedDelivery = await client.query(
+        `SELECT 1 FROM outbox_jobs
+         WHERE task_identifier = 'fulfill_supplier_order'
+           AND status = 'PENDING'
+           AND payload->>'job_id' = $1
+         LIMIT 1`,
+        [jobId],
+      );
+      const strandedDelivery = job.delivery_state === 'PENDING' && queuedDelivery.rowCount === 0;
       if (
         job.payment_state === 'COMMITTED' &&
-        (job.delivery_state === 'NOT_REQUESTED' || job.delivery_state === 'RETRIEVAL_FAILED')
+        (job.delivery_state === 'NOT_REQUESTED' ||
+          job.delivery_state === 'RETRIEVAL_FAILED' ||
+          strandedDelivery)
       ) {
         const now = this.#dependencies.now();
+        // Compare and set against the state just read under the row lock, so a
+        // stranded PENDING is claimed exactly once and the retrieval that a new
+        // delivery_attempt fences off can never complete under the old one.
         const resumed = await client.query<{ delivery_attempt: number }>(
           `UPDATE resumable_jobs
            SET delivery_state = 'PENDING', delivery_attempt = delivery_attempt + 1, updated_at = $1
-           WHERE job_id = $2 AND delivery_state IN ('NOT_REQUESTED', 'RETRIEVAL_FAILED')
+           WHERE job_id = $2 AND delivery_state = $3
            RETURNING delivery_attempt`,
-          [now, jobId],
+          [now, jobId, job.delivery_state],
         );
         const deliveryAttempt = resumed.rows[0]?.delivery_attempt;
         if (deliveryAttempt === undefined) {

packages/storage-postgres/test/jobs.test.ts

@@ -248,7 +248,7 @@ describe('JobLedger delivery recovery', () => {
     const resumed = await ledger.resumeDelivery('workspace-unit', failedJob.job_id);
 
     expect(resumed).toMatchObject({ delivery_state: 'PENDING', payment_state: 'COMMITTED' });
-    const outbox = calls.find((call) => call.sql.includes("'fulfill_supplier_order'"));
+    const outbox = calls.find((call) => call.sql.includes('INSERT INTO outbox_jobs'));
     expect(outbox?.values?.[1]).toBe(`fulfill:${failedJob.job_id}:team_report_order_unit:2`);
     expect(outbox?.values?.[2]).toBe(
       JSON.stringify({ job_id: failedJob.job_id, delivery_attempt: 2 }),
@@ -257,7 +257,7 @@ describe('JobLedger delivery recovery', () => {
     expect(calls.some((call) => call.sql.includes('attempts'))).toBe(false);
   });
 
-  it('does not enqueue a duplicate delivery while an attempt is already pending', async () => {
+  it('does not enqueue a duplicate delivery while an attempt is already queued', async () => {
     const calls: string[] = [];
     const pendingJob = { ...failedJob, delivery_state: 'PENDING' as const };
     const client = {
@@ -266,7 +266,10 @@ describe('JobLedger delivery recovery', () => {
         if (sql.includes('FOR UPDATE OF j') || sql.includes('WHERE j.workspace_id')) {
           return { rows: [pendingJob] };
         }
-        return { rows: [] };
+        // The fulfilment row is still queued, or a worker is holding it: either
+        // way something will still move this delivery.
+        if (sql.includes('FROM outbox_jobs')) return { rows: [{ '?column?': 1 }], rowCount: 1 };
+        return { rows: [], rowCount: 0 };
       },
       release() {},
     };
@@ -278,6 +281,51 @@ describe('JobLedger delivery recovery', () => {
     await ledger.resumeDelivery('workspace-unit', pendingJob.job_id);
 
     expect(calls.some((sql) => sql.includes('RETURNING delivery_attempt'))).toBe(false);
-    expect(calls.some((sql) => sql.includes("'fulfill_supplier_order'"))).toBe(false);
+    expect(calls.some((sql) => sql.includes('INSERT INTO outbox_jobs'))).toBe(false);
+  });
+
+  /**
+   * The payment is committed and the result was never retrieved, but nothing is
+   * queued to retrieve it: the worker no-ops a payload whose delivery_attempt no
+   * longer matches and then marks the row DELIVERED. PENDING was not resumable,
+   * so such a job could never move again.
+   */
+  it('re-queues a pending delivery that has no fulfilment work left', async () => {
+    const calls: Array<{ sql: string; values?: readonly unknown[] }> = [];
+    const strandedJob = { ...failedJob, delivery_state: 'PENDING' as const };
+    const client = {
+      async query(sql: string, values?: readonly unknown[]) {
+        calls.push({ sql, values });
+        if (sql.includes('FOR UPDATE OF j')) return { rows: [strandedJob], rowCount: 1 };
+        if (sql.includes('RETURNING delivery_attempt')) {
+          return { rows: [{ delivery_attempt: 2 }], rowCount: 1 };
+        }
+        if (sql.includes('WHERE j.workspace_id')) {
+          return { rows: [{ ...strandedJob, delivery_attempt: 2 }], rowCount: 1 };
+        }
+        // Nothing queued and nothing in flight.
+        return { rows: [], rowCount: 0 };
+      },
+      release() {},
+    };
+    const ledger = new JobLedger({ connect: async () => client } as never, {
+      now: () => new Date('2026-09-07T12:02:00.000Z'),
+      nextAttemptId: () => 'unused',
+    });
+
+    await ledger.resumeDelivery('workspace-unit', strandedJob.job_id);
+
+    // Claimed against the state read under the row lock, so two concurrent
+    // resumes cannot both take it.
+    const claim = calls.find((call) => call.sql.includes('RETURNING delivery_attempt'));
+    expect(claim?.values?.[2]).toBe('PENDING');
+    // A fresh attempt fences the retrieval, and no payment work is created.
+    const outbox = calls.find((call) => call.sql.includes('INSERT INTO outbox_jobs'));
+    expect(outbox?.values?.[1]).toBe(`fulfill:${strandedJob.job_id}:team_report_order_unit:2`);
+    expect(outbox?.values?.[2]).toBe(
+      JSON.stringify({ job_id: strandedJob.job_id, delivery_attempt: 2 }),
+    );
+    expect(calls.some((call) => call.sql.includes('submit_settlement'))).toBe(false);
+    expect(calls.some((call) => call.sql.includes('attempts'))).toBe(false);
   });
 });