M6 — FDQ-70 root-cause SDK import fix (drop tsx) + FDQ-70b setBudget idempotency guard

adapters/acp-adapter/src/acpAdapter.ts

@@ -73,6 +73,10 @@ export class AcpAdapter implements ChannelIngress {
   private readonly recentJobs = new Map<string, number>();
   // In-flight delivery guard — per `${chainId}:${jobId}` (never TTL-swept; released in the funded finally).
   private readonly inFlight = new Set<string>();
+  // In-flight ACCEPT guard (FDQ-70b) — per `${chainId}:${jobId}` (same key shape as inFlight); one
+  // accept attempt at a time, released after the attempt. Closes the job.created / requirement.message
+  // double-dispatch race (see handleEntry).
+  private readonly acceptInFlight = new Set<string>();
   // Poll log-once.
   private readonly pollSeen = new Map<string, number>();
 
@@ -167,6 +171,20 @@ export class AcpAdapter implements ChannelIngress {
     return true;
   }
 
+  /** FDQ-70b accept guard — a synchronous check-and-set (no await between `.has` and `.add`) so a
+   *  concurrent `job.created` + `requirement.message` for one job yields exactly ONE accept attempt.
+   *  Keyed `${chainId}:${jobId}` to match the funded `inFlight` claim; released by the caller once
+   *  the attempt settles. */
+  private claimAccept(chainId: number, jobId: string): boolean {
+    const key = `${chainId}:${jobId}`;
+    if (this.acceptInFlight.has(key)) return false;
+    this.acceptInFlight.add(key);
+    return true;
+  }
+  private releaseAccept(chainId: number, jobId: string): void {
+    this.acceptInFlight.delete(`${chainId}:${jobId}`);
+  }
+
   async handleEntry(session: AcpJobSession, entry: AcpRoomEntry): Promise<void> {
     // Process system lifecycle events AND the initial requirement message (arrives as a separate
     // room entry, contentType='requirement', after job.created).
@@ -187,7 +205,19 @@ export class AcpAdapter implements ChannelIngress {
       case 'requirement.message': {
         const decidedKey = `${jobId}:__decided`;
         if (this.recentJobs.has(decidedKey)) break;
-        await this.handleJobCreated(session, entry, log);
+        // FDQ-70b: `job.created` and `requirement.message` are DISTINCT eventTypes, so claimDispatch's
+        // per-event key admits BOTH into the accept path — and markDecided only fires AFTER setBudget's
+        // await, so a concurrent pair (SSE double-fire, or hydration re-firing pre-existing jobs at
+        // startup) could each pass the __decided check and each call setBudget. Claim a single accept
+        // slot per jobId SYNCHRONOUSLY here, before handleJobCreated's first await → exactly one
+        // setBudget. Released in the finally so a transient accept FAILURE can still be retried by a
+        // later event; the __decided marker makes a SUCCESSFUL accept permanent.
+        if (!this.claimAccept(session.chainId, jobId)) break;
+        try {
+          await this.handleJobCreated(session, entry, log);
+        } finally {
+          this.releaseAccept(session.chainId, jobId);
+        }
         break;
       }
       case 'job.funded':
@@ -281,6 +311,10 @@ export class AcpAdapter implements ChannelIngress {
     // Accept: propose the registered sticker price (no dynamic price resolver in the adapter).
     const price = this.offeringPrices.get(offeringId) ?? 0;
     void isPlainText;
+    // FDQ-70b INVARIANT: markDecided() must remain synchronous-adjacent to this await —
+    // do NOT insert an await or throwing statement between setBudget resolving and
+    // markDecided(). The accept claim releases in handleEntry's finally; if __decided
+    // were unset at that moment, a later event could re-accept an already-accepted job.
     try {
       await session.setBudget(this.sdk.assetUsdc(price, session.chainId));
       log.info('Job accepted via setBudget', { offeringId, price });

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

@@ -5,7 +5,7 @@ import { describe, it, expect, vi, afterEach } from 'vitest';
 import { offeringHandlers } from '@grey/core';
 import { AcpAdapter } from '../src/acpAdapter.js';
 import type { ChannelIngress } from '@grey/core';
-import type { AcpJob, OfferingHandler } from '../src/acpTypes.js';
+import type { AcpJob, AcpRoomEntry, OfferingHandler } from '../src/acpTypes.js';
 import {
   FakeSession,
   FakeAgent,
@@ -178,6 +178,140 @@ describe('AcpAdapter — accept + delivery', () => {
   });
 });
 
+describe('AcpAdapter — FDQ-70b setBudget idempotency (accept race)', () => {
+  // `job.created` and `requirement.message` are DISTINCT eventTypes that both trigger the accept
+  // path. Fired concurrently (SSE double-fire, or hydration re-firing a created-phase job at
+  // startup) they must still produce EXACTLY ONE setBudget. Deterministic: both handlers run to
+  // their first `await` synchronously, so the second observes the synchronous accept claim — no
+  // timers, no ordering luck.
+  it('concurrent job.created + requirement.message → exactly one setBudget', async () => {
+    const { adapter } = makeAdapter({ agent: new FakeAgent() });
+    adapter.registerOffering({ slug: 'legitimacy_scan', priceUsd: 0.25 });
+    const session = new FakeSession({
+      jobId: 'race-1',
+      job: fundedJob({ status: 'created' }),
+      entries: [requirementEntry({ token_address: TOKEN })],
+    });
+    const p1 = adapter.handleEntry(session, systemEntry('job.created'));
+    const p2 = adapter.handleEntry(session, requirementEntry({ token_address: TOKEN }));
+    await Promise.all([p1, p2]);
+    expect(session.budgets).toEqual([{ __usdc: 0.25, chainId: 8453 }]);
+    expect(session.rejected).toHaveLength(0);
+  });
+
+  it('hydration-at-startup re-firing the accept pair via agent.on(entry) accepts exactly once', async () => {
+    const agent = new FakeAgent();
+    const { adapter } = makeAdapter({ agent });
+    adapter.registerOffering({ slug: 'legitimacy_scan', priceUsd: 0.25 });
+    running = adapter;
+    await adapter.start(); // installs the real onEntry callback hydrateSessions() would fire
+    const session = new FakeSession({
+      jobId: 'hydra-1',
+      job: fundedJob({ status: 'created' }),
+      entries: [requirementEntry({ token_address: TOKEN })],
+    });
+    // hydrateSessions fires 'entry' (fire-and-forget) on the pre-existing created-phase job.
+    agent.onEntry!(session, systemEntry('job.created'));
+    agent.onEntry!(session, requirementEntry({ token_address: TOKEN }));
+    await new Promise((r) => setTimeout(r, 0)); // drain the fire-and-forget handler chains
+    expect(session.budgets).toHaveLength(1);
+    running = null;
+    await adapter.stop();
+  });
+
+  it('a fresh accept event AFTER a decided job does not re-budget (sequential + __decided)', async () => {
+    const { adapter } = makeAdapter({ agent: new FakeAgent() });
+    adapter.registerOffering({ slug: 'legitimacy_scan', priceUsd: 0.25 });
+    const session = new FakeSession({
+      jobId: 'seq-1',
+      job: fundedJob({ status: 'created' }),
+      entries: [requirementEntry({ token_address: TOKEN })],
+    });
+    await adapter.handleEntry(session, systemEntry('job.created'));
+    await adapter.handleEntry(session, requirementEntry({ token_address: TOKEN }));
+    expect(session.budgets).toHaveLength(1);
+  });
+
+  // Evidence that submit needs NO new guard: it is already claimed synchronously by claimDispatch
+  // (recentJobs['jobId:job.funded'] + inFlight, both set before any await) and re-checked against a
+  // fresh FUNDED status. Concurrent SSE + poll job.funded therefore delivers exactly once.
+  it('concurrent job.funded (SSE + poll) → exactly one submit (submit already single-path)', async () => {
+    const { adapter } = makeAdapter({ agent: new FakeAgent() });
+    const session = new FakeSession({
+      jobId: 'fund-1',
+      job: fundedJob(),
+      entries: [requirementEntry({ token_address: TOKEN })],
+    });
+    const p1 = adapter.handleEntry(session, systemEntry('job.funded'));
+    const p2 = adapter.handleEntry(session, systemEntry('job.funded'));
+    await Promise.all([p1, p2]);
+    expect(session.submitted).toHaveLength(1);
+  });
+
+  // Origin-independence of the ACCEPT guard. handleEntry is the SOLE entry to the accept path
+  // (setBudget @acpAdapter.ts:312 ← handleJobCreated:214 ← handleEntry, behind claimAccept), and the
+  // poll backstop routes THROUGH handleEntry too (dispatchPolledJob:551). Here a poll-SHAPED synthetic
+  // system entry (identical shape to dispatchPolledJob:544-549) carrying an accept-triggering
+  // job.created races an SSE requirement.message — a DISTINCT eventType that claimDispatch does NOT
+  // dedup, so only claimAccept stands between them. (In production poll emits job.funded only, so it
+  // cannot originate an accept; this proves the guard holds for that origin regardless.)
+  it('accept guard is origin-independent: poll-shaped job.created + SSE requirement.message → one setBudget', async () => {
+    const { adapter } = makeAdapter({ agent: new FakeAgent() });
+    adapter.registerOffering({ slug: 'legitimacy_scan', priceUsd: 0.25 });
+    const session = new FakeSession({
+      jobId: 'xorigin-1',
+      job: fundedJob({ status: 'created' }),
+      entries: [requirementEntry({ token_address: TOKEN })],
+    });
+    const pollShaped: AcpRoomEntry = {
+      kind: 'system',
+      onChainJobId: 'xorigin-1',
+      chainId: 8453,
+      event: { type: 'job.created', jobId: 'xorigin-1' },
+      timestamp: 0,
+    };
+    const p1 = adapter.handleEntry(session, pollShaped);
+    const p2 = adapter.handleEntry(session, requirementEntry({ token_address: TOKEN }));
+    await Promise.all([p1, p2]);
+    expect(session.budgets).toHaveLength(1);
+  });
+
+  // The genuinely-reachable multi-path race: the REAL poll timer dispatching the same funded job the
+  // SSE callback just fired. Both go through handleEntry → claimDispatch's synchronous job.funded
+  // claim admits exactly one. This is the #70220-class "SSE racing poll" proof on the path poll
+  // actually exercises.
+  it('cross-origin funded race: SSE job.funded + REAL poll dispatch (same job) → exactly one submit', async () => {
+    vi.useFakeTimers();
+    const agent = new FakeAgent();
+    const ourAddr = '0xa9667116b4f4e9f1bae85f93a21b4b8ea45de98f';
+    agent.activeJobs = [{ chainId: 8453, onChainJobId: 'multi-1' }];
+    agent.jobsById.set('multi-1', {
+      description: 'legitimacy_scan',
+      clientAddress: '0xbuyer',
+      providerAddress: ourAddr,
+      status: 'funded',
+      ...({ jobStatus: 'funded' } as object),
+    } as AcpJob);
+    const session = new FakeSession({
+      jobId: 'multi-1',
+      job: fundedJob(),
+      entries: [requirementEntry({ token_address: TOKEN })],
+    });
+    agent.sessions.set('8453:multi-1', session); // the session the poll's dispatchPolledJob will reuse
+
+    const { adapter } = makeAdapter({ agent, pollIntervalMs: 1000 });
+    running = adapter;
+    await adapter.start(); // installs the SSE onEntry callback + starts the poll
+    // SSE fires first (fire-and-forget, claims the synchronous job.funded slot before any await);
+    // the poll tick then dispatches the SAME job through handleEntry and is deduped.
+    agent.onEntry!(session, systemEntry('job.funded'));
+    await vi.advanceTimersByTimeAsync(1000);
+    expect(session.submitted).toHaveLength(1);
+    running = null;
+    await adapter.stop();
+  });
+});
+
 describe('AcpAdapter — OBSERVE_ONLY (FDQ-63 safety)', () => {
   it('job.created in observe-only signs NOTHING (no setBudget, no reject)', async () => {
     const { adapter } = makeAdapter({ observeOnly: true, agent: new FakeAgent() });