M6 FDQ-73 — expiry/terminal reconciliation sweep (grey_two)

adapters/acp-adapter/src/acpAdapter.ts

@@ -28,6 +28,7 @@ import type { AcpAdapterConfig } from './config.js';
 import { GREY_DID } from './config.js';
 import { parseRequirement } from './parseRequirement.js';
 import { createLogger, type AdapterLogger } from './logger.js';
+import type { ReputationReconciler } from './reputation/reputationReconciler.js';
 
 /** Dedup TTL — 5 minutes. */
 const DEDUP_TTL_MS = 5 * 60 * 1000;
@@ -51,6 +52,8 @@ export interface AcpAdapterOptions {
   logger?: AdapterLogger;
   /** B6 seam — null in Phase C. */
   reputationGate?: BuyerReputationGate | null;
+  /** FDQ-73 seam — null until C′-reconciliation. Swept on the poll cadence; reads status, never signs. */
+  reputationReconciler?: ReputationReconciler | null;
 }
 
 export class AcpAdapter implements ChannelIngress {
@@ -60,6 +63,7 @@ export class AcpAdapter implements ChannelIngress {
   private readonly handlers: Record<string, OfferingHandler>;
   private readonly log: AdapterLogger;
   private reputationGate: BuyerReputationGate | null;
+  private readonly reputationReconciler: ReputationReconciler | null;
 
   private agent: AcpAgentLike | null = null;
   private pollTimer: NodeJS.Timeout | null = null;
@@ -87,6 +91,7 @@ export class AcpAdapter implements ChannelIngress {
     this.handlers = opts.handlers;
     this.log = opts.logger ?? createLogger({ component: 'acp-adapter' });
     this.reputationGate = opts.reputationGate ?? null;
+    this.reputationReconciler = opts.reputationReconciler ?? null;
   }
 
   // ── ChannelIngress ─────────────────────────
@@ -149,9 +154,31 @@ export class AcpAdapter implements ChannelIngress {
     // stop() clears it for a clean exit.
     this.pollTimer = setInterval(() => {
       void this.runDeliveryPoll();
+      void this.runReconcileSweep(); // FDQ-73 — independent of the funded poll; fail-soft
     }, this.config.pollIntervalMs);
   }
 
+  /** FDQ-73 reconciliation tick — resolve stranded `submitted` jobs the SDK never event-fired as
+   *  terminal (job.expired/job.rejected). Reads job status + records reputation only; NEVER signs
+   *  (safe under OBSERVE_ONLY). Fully fail-soft. */
+  private async runReconcileSweep(): Promise<void> {
+    if (!this.agent || !this.reputationReconciler) return;
+    try {
+      await this.reputationReconciler.sweep((chainId, jobId) => this.fetchJobStatus(chainId, jobId));
+    } catch (err) {
+      this.log.warn('[reconcile] sweep failed', { error: errMsg(err) });
+    }
+  }
+
+  /** Authoritative job status (REST string) via the SDK's getJob — same convention as the funded
+   *  filter (`jobStatus ?? status`). Returns null when the agent/job is unavailable. */
+  private async fetchJobStatus(chainId: number, jobId: string): Promise<string | null> {
+    if (!this.agent) return null;
+    const full = await this.agent.getApi().getJob(chainId, jobId);
+    if (!full) return null;
+    return String((full as { jobStatus?: unknown; status?: unknown }).jobStatus ?? full.status ?? '');
+  }
+
   /** Shared dispatch-claim primitive — the first synchronous check-and-set any path performs for a
    *  job+event, before any await, so socket-vs-poll races resolve deterministically. */
   private claimDispatch(chainId: number, jobId: string, eventType: string): boolean {

adapters/acp-adapter/src/main.ts

@@ -14,6 +14,7 @@ import { createLogger } from './logger.js';
 import { PgBuyerRecordStore, PgTrackedJobsRepo, stripSslParams } from './reputation/reputationDb.js';
 import { BuyerReputationGateImpl } from './reputation/buyerReputationGate.js';
 import { makeCrossProviderFetch } from './reputation/crossProvider.js';
+import { ReputationReconciler } from './reputation/reputationReconciler.js';
 
 async function main(): Promise<void> {
   const log = createLogger({ component: 'acp-adapter' });
@@ -33,13 +34,21 @@ async function main(): Promise<void> {
     ssl: { rejectUnauthorized: false },
     max: 3,
   });
+  const trackedRepo = new PgTrackedJobsRepo(gatePool);
   const reputationGate = new BuyerReputationGateImpl({
     buyerStore: new PgBuyerRecordStore(gatePool),
-    trackedRepo: new PgTrackedJobsRepo(gatePool),
+    trackedRepo,
     gating: config.buyerGating,
     logger: log.child({ subsystem: 'reputation' }),
     crossProviderFetch: makeCrossProviderFetch(config.baseRpcUrl),
   });
+  // FDQ-73 — reconciliation sweep for stranded submitted jobs (SDK never delivers job.expired/
+  // rejected). Shares the gate's trackedRepo + idempotent onJobTerminal; runs on the poll cadence.
+  const reputationReconciler = new ReputationReconciler({
+    trackedRepo,
+    onTerminal: (jobId, chainId, terminal) => reputationGate.onJobTerminal(jobId, chainId, terminal),
+    logger: log.child({ subsystem: 'reconcile' }),
+  });
 
   const adapter = new AcpAdapter({
     config,
@@ -48,6 +57,7 @@ async function main(): Promise<void> {
     handlers: offeringHandlers,
     logger: log,
     reputationGate,
+    reputationReconciler,
   });
 
   // Register the 7 paid offerings from the single price source (invariant #20), BEFORE start() —

adapters/acp-adapter/src/reputation/reputationDb.ts

@@ -68,6 +68,13 @@ export interface BuyerRecordStore {
   writeCrossProvider(walletLowercased: string, w: CrossProviderWrite): Promise<void>;
 }
 
+/** A stranded-submitted row surfaced by the FDQ-73 reconciliation sweep. */
+export interface ExpiredSubmittedRow {
+  chainId: number;
+  jobId: string;
+  buyerAddress: string;
+}
+
 /** Tracked-job persistence surface (composite key chain_id+job_id at every site — multi-chain-safe). */
 export interface TrackedJobsRepo {
   trackSubmitted(input: TrackSubmittedInput): Promise<void>;
@@ -79,6 +86,9 @@ export interface TrackedJobsRepo {
     jobId: string,
     terminal: TrackedTerminalStatus,
   ): Promise<{ buyerAddress: string } | null>;
+  /** FDQ-73 sweep source: still-`submitted` rows whose SLA `expires_at` has passed (bounded LIMIT).
+   *  SELECT only — the reconciler discriminates the TRUE terminal via a fresh on-chain status read. */
+  listExpiredSubmitted(nowIso: string, limit: number): Promise<ExpiredSubmittedRow[]>;
 }
 
 // ── row mapping ───────────────────────────────
@@ -135,6 +145,9 @@ const RESOLVE_TRACKED = `UPDATE grey_two.tracked_jobs SET status = $3, resolved_
   WHERE chain_id = $1 AND job_id = $2 AND status = 'submitted'
   RETURNING buyer_address`;
 
+const LIST_EXPIRED_SUBMITTED = `SELECT chain_id, job_id, buyer_address FROM grey_two.tracked_jobs
+  WHERE status = 'submitted' AND expires_at < $1 ORDER BY expires_at ASC LIMIT $2`;
+
 /** Production buyer-record store — raw parameterized SQL over a pg.Pool. */
 export class PgBuyerRecordStore implements BuyerRecordStore {
   constructor(private readonly pool: PoolLike) {}
@@ -175,6 +188,14 @@ export class PgTrackedJobsRepo implements TrackedJobsRepo {
     const r = rows[0];
     return r ? { buyerAddress: String(r['buyer_address'] ?? '') } : null;
   }
+  async listExpiredSubmitted(nowIso: string, limit: number): Promise<ExpiredSubmittedRow[]> {
+    const { rows } = await this.pool.query(LIST_EXPIRED_SUBMITTED, [nowIso, limit]);
+    return rows.map((r) => ({
+      chainId: toInt(r['chain_id']),
+      jobId: String(r['job_id'] ?? ''),
+      buyerAddress: String(r['buyer_address'] ?? ''),
+    }));
+  }
 }
 
 /**

adapters/acp-adapter/src/reputation/reputationReconciler.ts

@@ -0,0 +1,107 @@
+// FDQ-73 expiry/terminal reconciliation sweep.
+//
+// WHY: the SDK never delivers job.expired / job.rejected to the adapter — acpAgent.fireHandler
+// (the sole path to the entry handler, used by both live dispatch AND startup hydrateSessions)
+// hard-returns on !session.shouldRespond(entry), and jobSession.shouldRespond's RESPONDERS map
+// omits job.expired and sets job.rejected:[] → both return false. There is no client-side expiry
+// timer, and the delivery poll is FUNDED-only. So a submitted-then-terminal job that isn't
+// `completed` strands as status='submitted' forever and no stiff is ever recorded — the ladder
+// can't advance. This sweep closes that gap; it is a hard prerequisite for flip-to-enforce.
+//
+// PORT NOTE (plugin-acp reconcileTrackedJobs, read-only under lock #5): the source TRIGGERED on
+// "tracked job dropped out of the live active set", then read getJob() to discriminate the
+// terminal. We trigger on `expires_at < now` instead (a bounded SELECT — see reputationDb) and keep
+// the source's authoritative-status discrimination. This is tighter and avoids the source's
+// "dropped from active but actually still live" ambiguity: a past-SLA row whose ON-CHAIN status is
+// still funded/open maps to null → left `submitted`, retried next tick. Only a genuinely terminal
+// on-chain status resolves the row. Marking a rejected job as expired-and-stiffed would be a false
+// strike — the failure mode this discrimination prevents.
+//
+// SAFETY: reads job status + writes reputation tables ONLY — it NEVER signs (no setBudget/submit/
+// reject), so it is safe under OBSERVE_ONLY. Fail-soft: a sweep-level error skips the tick; a
+// per-row error skips that row; nothing throws into the poll loop. Idempotent: it resolves via the
+// gate's onJobTerminal, whose resolveIfSubmitted `WHERE status='submitted'` guard makes a real
+// event (or a concurrent sweep) that also arrives a no-op for the second observer.
+import type { AdapterLogger } from '../logger.js';
+import type { JobTerminalStatus } from '../acpTypes.js';
+import type { TrackedJobsRepo, TrackedTerminalStatus } from './reputationDb.js';
+
+/** Fetch the authoritative job status (REST string) for a job, or null when unavailable. */
+export type JobStatusFetch = (chainId: number, jobId: string) => Promise<string | null>;
+
+export interface ReputationReconcilerOptions {
+  trackedRepo: TrackedJobsRepo;
+  /** The gate's onJobTerminal — idempotent resolve (+ a stiff only for genuine expiry). */
+  onTerminal: (jobId: string, chainId: number, terminal: JobTerminalStatus) => Promise<void>;
+  logger: AdapterLogger;
+  /** Max rows per sweep (bounded). Default 100. */
+  limit?: number;
+  /** Injectable clock for deterministic tests. */
+  clock?: () => Date;
+}
+
+/** Map an authoritative job-status string to the terminal we act on, or null (still live/unknown →
+ *  leave `submitted`, retry next tick). Matches the REST-string convention of the funded filter. */
+export function toTerminal(raw: string | null): TrackedTerminalStatus | null {
+  switch ((raw ?? '').toLowerCase()) {
+    case 'completed':
+      return 'completed';
+    case 'rejected':
+      return 'rejected';
+    case 'expired':
+      return 'expired';
+    default:
+      return null; // funded / open / unknown → not yet terminal
+  }
+}
+
+export class ReputationReconciler {
+  private readonly trackedRepo: TrackedJobsRepo;
+  private readonly onTerminal: (jobId: string, chainId: number, terminal: JobTerminalStatus) => Promise<void>;
+  private readonly log: AdapterLogger;
+  private readonly limit: number;
+  private readonly clock: () => Date;
+
+  constructor(opts: ReputationReconcilerOptions) {
+    this.trackedRepo = opts.trackedRepo;
+    this.onTerminal = opts.onTerminal;
+    this.log = opts.logger;
+    this.limit = opts.limit ?? 100;
+    this.clock = opts.clock ?? ((): Date => new Date());
+  }
+
+  /** One reconciliation tick. `fetchJobStatus` reads the authoritative on-chain/API status. */
+  async sweep(fetchJobStatus: JobStatusFetch): Promise<void> {
+    let rows;
+    try {
+      rows = await this.trackedRepo.listExpiredSubmitted(this.clock().toISOString(), this.limit);
+    } catch (err) {
+      this.log.warn('[reconcile] listExpiredSubmitted failed — skipping tick', { error: errMsg(err) });
+      return;
+    }
+    if (rows.length === 0) return;
+    for (const row of rows) {
+      try {
+        const terminal = toTerminal(await fetchJobStatus(row.chainId, row.jobId));
+        if (!terminal) continue; // still live/funded/unknown — leave `submitted`, retry next tick
+        // Idempotent: onJobTerminal → resolveIfSubmitted (guarded) → stiff ONLY for 'expired'.
+        await this.onTerminal(row.jobId, row.chainId, terminal);
+        this.log.info('[reconcile] resolved stranded submitted job', {
+          chainId: row.chainId,
+          jobId: row.jobId,
+          terminal,
+        });
+      } catch (err) {
+        this.log.warn('[reconcile] row failed — skipping', {
+          chainId: row.chainId,
+          jobId: row.jobId,
+          error: errMsg(err),
+        });
+      }
+    }
+  }
+}
+
+function errMsg(err: unknown): string {
+  return err instanceof Error ? err.message : String(err);
+}

adapters/acp-adapter/test/buyerReputationGate.test.ts

@@ -102,6 +102,9 @@ class MemTracked implements TrackedJobsRepo {
     j.status = terminal;
     return { buyerAddress: j.buyerAddress };
   }
+  async listExpiredSubmitted(): Promise<Array<{ chainId: number; jobId: string; buyerAddress: string }>> {
+    return []; // not exercised by the gate-logic tests (see reputationReconciler.test.ts)
+  }
 }
 
 function makeGate(store: MemBuyerStore, tracked: MemTracked, gating: BuyerGatingConfig, clock?: () => Date): BuyerReputationGateImpl {