From 4f8ebedd2199f418288b10dad12e487995f4c1b4 Mon Sep 17 00:00:00 2001 From: Mayakovsky Date: Sat, 25 Jul 2026 15:46:37 -0400 Subject: [PATCH 1/2] feat(m6): FDQ-73 expiry/terminal reconciliation sweep (grey_two) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SDK never delivers job.expired/job.rejected to the adapter (acpAgent.fireHandler gates on shouldRespond, whose RESPONDERS map omits job.expired and sets job.rejected:[]), 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' and no stiff is ever recorded. The reputation ladder can't advance without this sweep; it is a hard prerequisite for flip-to-enforce. - reputationDb.ts: add listExpiredSubmitted(nowIso, limit) to TrackedJobsRepo — a bounded, oldest-first SELECT of still-`submitted` rows past expires_at. SELECT only (FDQ-65). - reputationReconciler.ts: each poll tick, pull expired-submitted rows, fetch each job's TRUE on-chain status, and resolve via the gate's idempotent onJobTerminal — a stiff only for a genuine `expired`; `rejected`/`completed` resolve with no stiff; still-live/unknown left `submitted` and retried. Ported from plugin-acp reconcileTrackedJobs but triggered on expires_at (not active-set dropout), removing the "dropped-from-active but still live" ambiguity — false-strike prevention by construction. Reads status + records only; NEVER signs (safe under OBSERVE_ONLY). Fail-soft at tick + per-row. - acpAdapter.ts: reputationReconciler seam; sweep on the existing poll cadence (cleared by stop()); fetchJobStatus via the SDK getJob (jobStatus ?? status convention). - main.ts: share the gate's trackedRepo, build + inject the reconciler. Co-Authored-By: Claude Opus 4.8 --- adapters/acp-adapter/src/acpAdapter.ts | 27 +++++ adapters/acp-adapter/src/main.ts | 12 +- .../src/reputation/reputationDb.ts | 21 ++++ .../src/reputation/reputationReconciler.ts | 107 ++++++++++++++++++ 4 files changed, 166 insertions(+), 1 deletion(-) create mode 100644 adapters/acp-adapter/src/reputation/reputationReconciler.ts diff --git a/adapters/acp-adapter/src/acpAdapter.ts b/adapters/acp-adapter/src/acpAdapter.ts index e6d2b1b..eb5d9be 100644 --- a/adapters/acp-adapter/src/acpAdapter.ts +++ b/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; 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 { + 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 { + 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 { diff --git a/adapters/acp-adapter/src/main.ts b/adapters/acp-adapter/src/main.ts index 7a7c795..11bbd57 100644 --- a/adapters/acp-adapter/src/main.ts +++ b/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 { const log = createLogger({ component: 'acp-adapter' }); @@ -33,13 +34,21 @@ async function main(): Promise { 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 { handlers: offeringHandlers, logger: log, reputationGate, + reputationReconciler, }); // Register the 7 paid offerings from the single price source (invariant #20), BEFORE start() — diff --git a/adapters/acp-adapter/src/reputation/reputationDb.ts b/adapters/acp-adapter/src/reputation/reputationDb.ts index 6f7ccca..ee2ccd0 100644 --- a/adapters/acp-adapter/src/reputation/reputationDb.ts +++ b/adapters/acp-adapter/src/reputation/reputationDb.ts @@ -68,6 +68,13 @@ export interface BuyerRecordStore { writeCrossProvider(walletLowercased: string, w: CrossProviderWrite): Promise; } +/** 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; @@ -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; } // ── 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 { + 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'] ?? ''), + })); + } } /** diff --git a/adapters/acp-adapter/src/reputation/reputationReconciler.ts b/adapters/acp-adapter/src/reputation/reputationReconciler.ts new file mode 100644 index 0000000..466ae42 --- /dev/null +++ b/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; + +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; + 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; + 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 { + 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); +} From 6d99a2a313d3977afad04532ed65396d14209bcc Mon Sep 17 00:00:00 2001 From: Mayakovsky Date: Sat, 25 Jul 2026 15:46:37 -0400 Subject: [PATCH 2/2] test(m6): FDQ-73 reconciliation sweep tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - test/reputationReconciler.test.ts: 10 deterministic tests (injected clock + fake tracked repo + injected status fetcher) — terminal discrimination (expired→one stiff clean→warned; rejected/completed→resolve no stiff; still-funded + not-yet-expired→untouched), idempotency (two sweeps → one stiff), fail-soft (list throws → tick skipped; per-row getJob throw → row skipped, loop continues), and listExpiredSubmitted grant compliance (bounded SELECT — no DELETE/TRUNCATE/UPDATE/INSERT). - test/buyerReputationGate.test.ts: interface-conformance for the fake tracked repo. vitest run 50/50, tier-1 smoke, typecheck, build, lint all green. Co-Authored-By: Claude Opus 4.8 --- .../test/buyerReputationGate.test.ts | 3 + .../test/reputationReconciler.test.ts | 222 ++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 adapters/acp-adapter/test/reputationReconciler.test.ts diff --git a/adapters/acp-adapter/test/buyerReputationGate.test.ts b/adapters/acp-adapter/test/buyerReputationGate.test.ts index a3f2f04..e191ba3 100644 --- a/adapters/acp-adapter/test/buyerReputationGate.test.ts +++ b/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> { + return []; // not exercised by the gate-logic tests (see reputationReconciler.test.ts) + } } function makeGate(store: MemBuyerStore, tracked: MemTracked, gating: BuyerGatingConfig, clock?: () => Date): BuyerReputationGateImpl { diff --git a/adapters/acp-adapter/test/reputationReconciler.test.ts b/adapters/acp-adapter/test/reputationReconciler.test.ts new file mode 100644 index 0000000..77e6210 --- /dev/null +++ b/adapters/acp-adapter/test/reputationReconciler.test.ts @@ -0,0 +1,222 @@ +// M6 FDQ-73 — ReputationReconciler sweep tests. Deterministic: injected clock, a fake tracked +// repo, an injected job-status fetcher. Covers the terminal discrimination (expired→stiff; +// rejected/completed→resolve no stiff; still-live→untouched), idempotency, fail-soft, and (via the +// Pg repo over a capturing pool) the new SELECT's grant compliance. +import { describe, it, expect } from 'vitest'; +import { ReputationReconciler, toTerminal } from '../src/reputation/reputationReconciler.js'; +import { BuyerReputationGateImpl, type BuyerGatingConfig } from '../src/reputation/buyerReputationGate.js'; +import { + PgTrackedJobsRepo, + type BuyerRecord, + type BuyerRecordStore, + type ExpiredSubmittedRow, + type PoolLike, + type StiffWrite, + type TrackSubmittedInput, + type TrackedJobsRepo, + type TrackedTerminalStatus, +} from '../src/reputation/reputationDb.js'; +import type { JobTerminalStatus } from '../src/acpTypes.js'; +import { silentLogger } from '../src/logger.js'; + +const BUYER = '0xb94182dd57798c30596f6a858802010fea0be0e1'; +const NOW = new Date('2026-07-25T18:00:00.000Z'); +const PAST = new Date('2026-07-25T17:00:00.000Z'); // expires_at in the past +const FUTURE = new Date('2026-07-25T19:00:00.000Z'); // not yet expired + +const SHADOW: BuyerGatingConfig = { blockEnabled: false, timeout1hSec: 3600, timeout12hSec: 43200, crossProviderCacheTtlSec: 3600 }; + +interface Row { + chainId: number; + jobId: string; + buyerAddress: string; + status: string; + expiresAt: Date; +} + +/** Fake tracked repo with expires-aware listExpiredSubmitted + the resolveIfSubmitted guard. */ +class FakeTracked implements TrackedJobsRepo { + rows = new Map(); + throwOnList = false; + seed(r: Row): void { + this.rows.set(`${r.chainId}:${r.jobId}`, r); + } + async trackSubmitted(i: TrackSubmittedInput): Promise { + this.seed({ chainId: i.chainId, jobId: i.jobId, buyerAddress: i.buyerAddress, status: 'submitted', expiresAt: i.expiresAt }); + } + async resolveIfSubmitted(chainId: number, jobId: string, terminal: TrackedTerminalStatus): Promise<{ buyerAddress: string } | null> { + const r = this.rows.get(`${chainId}:${jobId}`); + if (!r || r.status !== 'submitted') return null; + r.status = terminal; + return { buyerAddress: r.buyerAddress }; + } + async listExpiredSubmitted(nowIso: string, limit: number): Promise { + if (this.throwOnList) throw new Error('list failed'); + const now = new Date(nowIso).getTime(); + return [...this.rows.values()] + .filter((r) => r.status === 'submitted' && r.expiresAt.getTime() < now) + .slice(0, limit) + .map((r) => ({ chainId: r.chainId, jobId: r.jobId, buyerAddress: r.buyerAddress })); + } +} + +/** Minimal in-memory buyer store (for the end-to-end stiff assertion via a real gate). */ +class MemBuyerStore implements BuyerRecordStore { + recs = new Map(); + async get(w: string): Promise { + return this.recs.get(w) ?? null; + } + async insertStubIfAbsent(): Promise {} + async writeStiff(w: string, sw: StiffWrite): Promise { + const cur = this.recs.get(w) ?? { walletAddress: w, status: 'clean', strikes: 0, timeoutUntil: null, lastStiffAt: null, crossProviderCompletesTotal: 0, crossProviderCreatesTotal: 0, crossProviderDataCachedAt: null }; + this.recs.set(w, { ...cur, status: sw.status, strikes: sw.strikes, timeoutUntil: sw.timeoutUntil, lastStiffAt: sw.lastStiffAt }); + } + async writeCrossProvider(): Promise {} +} + +/** A status fetcher from a jobId→status map (undefined status → the fetcher throws for that job). */ +function fetcher(map: Record, throwFor: string[] = []) { + return async (_chainId: number, jobId: string): Promise => { + if (throwFor.includes(jobId)) throw new Error('getJob failed'); + return map[jobId] ?? null; + }; +} + +function makeReconciler(tracked: TrackedJobsRepo, onTerminal: (j: string, c: number, t: JobTerminalStatus) => Promise) { + return new ReputationReconciler({ trackedRepo: tracked, onTerminal, logger: silentLogger(), clock: () => NOW }); +} + +describe('ReputationReconciler — toTerminal discrimination', () => { + it('maps only genuine terminals; still-live/unknown → null', () => { + expect(toTerminal('expired')).toBe('expired'); + expect(toTerminal('COMPLETED')).toBe('completed'); + expect(toTerminal('Rejected')).toBe('rejected'); + expect(toTerminal('funded')).toBeNull(); + expect(toTerminal('open')).toBeNull(); + expect(toTerminal(null)).toBeNull(); + }); +}); + +describe('ReputationReconciler — sweep', () => { + it('expired-submitted + on-chain EXPIRED → one stiff (clean→warned), row resolved expired', async () => { + const tracked = new FakeTracked(); + tracked.seed({ chainId: 8453, jobId: '1', buyerAddress: BUYER, status: 'submitted', expiresAt: PAST }); + const buyerStore = new MemBuyerStore(); + const gate = new BuyerReputationGateImpl({ buyerStore, trackedRepo: tracked, gating: SHADOW, logger: silentLogger(), clock: () => NOW }); + const rec = makeReconciler(tracked, (j, c, t) => gate.onJobTerminal(j, c, t)); + + await rec.sweep(fetcher({ '1': 'expired' })); + + expect(tracked.rows.get('8453:1')?.status).toBe('expired'); + const buyer = await buyerStore.get(BUYER); + expect(buyer?.status).toBe('warned'); + expect(buyer?.strikes).toBe(1); + }); + + it('expired-submitted + on-chain REJECTED → resolved rejected, NO stiff', async () => { + const tracked = new FakeTracked(); + tracked.seed({ chainId: 8453, jobId: '2', buyerAddress: BUYER, status: 'submitted', expiresAt: PAST }); + const calls: TrackedTerminalStatus[] = []; + const rec = makeReconciler(tracked, async (_j, _c, t) => { + calls.push(t); + await tracked.resolveIfSubmitted(8453, '2', t as TrackedTerminalStatus); + }); + await rec.sweep(fetcher({ '2': 'rejected' })); + expect(calls).toEqual(['rejected']); + expect(tracked.rows.get('8453:2')?.status).toBe('rejected'); + }); + + it('expired-submitted + on-chain COMPLETED → resolved completed, NO stiff', async () => { + const tracked = new FakeTracked(); + tracked.seed({ chainId: 8453, jobId: '3', buyerAddress: BUYER, status: 'submitted', expiresAt: PAST }); + const calls: TrackedTerminalStatus[] = []; + const rec = makeReconciler(tracked, async (_j, _c, t) => { + calls.push(t); + await tracked.resolveIfSubmitted(8453, '3', t as TrackedTerminalStatus); + }); + await rec.sweep(fetcher({ '3': 'completed' })); + expect(calls).toEqual(['completed']); + expect(tracked.rows.get('8453:3')?.status).toBe('completed'); + }); + + it('past-SLA but on-chain STILL FUNDED → untouched (still submitted), retried next tick', async () => { + const tracked = new FakeTracked(); + tracked.seed({ chainId: 8453, jobId: '4', buyerAddress: BUYER, status: 'submitted', expiresAt: PAST }); + let onTerminalCalls = 0; + const rec = makeReconciler(tracked, async () => { + onTerminalCalls++; + }); + await rec.sweep(fetcher({ '4': 'funded' })); + expect(onTerminalCalls).toBe(0); + expect(tracked.rows.get('8453:4')?.status).toBe('submitted'); + }); + + it('not-yet-expired submitted row is never swept (expires_at in the future)', async () => { + const tracked = new FakeTracked(); + tracked.seed({ chainId: 8453, jobId: '5', buyerAddress: BUYER, status: 'submitted', expiresAt: FUTURE }); + let calls = 0; + const rec = makeReconciler(tracked, async () => { + calls++; + }); + await rec.sweep(fetcher({ '5': 'expired' })); // even if it were expired on-chain, listExpiredSubmitted excludes it + expect(calls).toBe(0); + }); + + it('idempotent: two sweeps of the same expired row → one stiff (second finds no submitted row)', async () => { + const tracked = new FakeTracked(); + tracked.seed({ chainId: 8453, jobId: '6', buyerAddress: BUYER, status: 'submitted', expiresAt: PAST }); + const buyerStore = new MemBuyerStore(); + const gate = new BuyerReputationGateImpl({ buyerStore, trackedRepo: tracked, gating: SHADOW, logger: silentLogger(), clock: () => NOW }); + const rec = makeReconciler(tracked, (j, c, t) => gate.onJobTerminal(j, c, t)); + await rec.sweep(fetcher({ '6': 'expired' })); + await rec.sweep(fetcher({ '6': 'expired' })); + expect((await buyerStore.get(BUYER))?.status).toBe('warned'); + expect((await buyerStore.get(BUYER))?.strikes).toBe(1); // not 2 + }); + + it('fail-soft: listExpiredSubmitted throws → tick skipped, no throw, onTerminal untouched', async () => { + const tracked = new FakeTracked(); + tracked.throwOnList = true; + let calls = 0; + const rec = makeReconciler(tracked, async () => { + calls++; + }); + await expect(rec.sweep(fetcher({}))).resolves.toBeUndefined(); + expect(calls).toBe(0); + }); + + it('fail-soft: a per-row getJob throw skips that row, the loop continues', async () => { + const tracked = new FakeTracked(); + tracked.seed({ chainId: 8453, jobId: 'bad', buyerAddress: BUYER, status: 'submitted', expiresAt: PAST }); + tracked.seed({ chainId: 8453, jobId: 'good', buyerAddress: BUYER, status: 'submitted', expiresAt: PAST }); + const resolved: string[] = []; + const rec = makeReconciler(tracked, async (j, c, t) => { + resolved.push(j); + await tracked.resolveIfSubmitted(c, j, t as TrackedTerminalStatus); + }); + await rec.sweep(fetcher({ good: 'expired' }, ['bad'])); + expect(resolved).toEqual(['good']); // 'bad' threw, 'good' still processed + expect(tracked.rows.get('8453:bad')?.status).toBe('submitted'); + expect(tracked.rows.get('8453:good')?.status).toBe('expired'); + }); +}); + +describe('ReputationReconciler — listExpiredSubmitted grant compliance', () => { + class CapturePool implements PoolLike { + calls: Array<{ text: string; params?: ReadonlyArray }> = []; + async query(text: string, params?: ReadonlyArray): Promise<{ rows: Array> }> { + this.calls.push({ text, params }); + return { rows: [] }; + } + } + it('the new sweep source is a bounded SELECT — no DELETE/TRUNCATE, LIMIT-bound', async () => { + const pool = new CapturePool(); + const repo = new PgTrackedJobsRepo(pool); + await repo.listExpiredSubmitted(NOW.toISOString(), 100); + const sql = pool.calls[0]?.text ?? ''; + expect(sql.trimStart()).toMatch(/^SELECT\b/i); + expect(sql).toMatch(/status = 'submitted' AND expires_at < \$1/); + expect(sql).toMatch(/LIMIT \$2/); + expect(sql).not.toMatch(/\b(DELETE|TRUNCATE|UPDATE|INSERT)\b/i); + }); +});