diff --git a/packages/coding-agent/test/sdk-host-steer-integration.test.ts b/packages/coding-agent/test/sdk-host-steer-integration.test.ts index 1dd649449f..49c1ff3c5a 100644 --- a/packages/coding-agent/test/sdk-host-steer-integration.test.ts +++ b/packages/coding-agent/test/sdk-host-steer-integration.test.ts @@ -5,26 +5,78 @@ import * as path from "node:path"; import type { ExtensionAPI, ExtensionContext } from "../src/extensibility/extensions"; import { createSdkSessionRuntimeExtension, type SessionSdkTransport } from "../src/sdk/host/session-runtime"; +/** + * Safety bound for a single emission to resolve. Far above any real host + * dispatch time, but finite so a lost response fails the test instead of + * hanging the file to the harness timeout. + */ +const RESPONSE_TIMEOUT_MS = 5_000; + +interface HarnessOptions { + /** Artificial transport delivery delay; models a host that responds slowly. */ + responseDelayMs?: number; + /** Safety timeout for one emission; small values are for race-contract tests. */ + responseTimeoutMs?: number; +} + interface Harness { emit(frame: Record): Promise>; + /** Response frames observed after the emission they belong to already ended. */ + readonly lateResponses: ReadonlyArray>; + /** Adjust the artificial transport delivery delay between emissions. */ + setResponseDelay(ms: number): void; start(): Promise; stop(): Promise; dispatches: number; persistedAtDispatch?: string; } -function createHarness(cwd: string, sessionId: string, sessionFile: string | undefined): Harness { +function createHarness( + cwd: string, + sessionId: string, + sessionFile: string | undefined, + options: HarnessOptions = {}, +): Harness { const handlers = new Map unknown>(); let receive: ((connectionId: string, frame: never) => void) | undefined; - let response: Record | undefined; let dispatches = 0; let persistedAtDispatch: string | undefined; + let responseDelayMs = options.responseDelayMs ?? 0; + const responseTimeoutMs = options.responseTimeoutMs ?? RESPONSE_TIMEOUT_MS; + // Exactly one pending emission at a time (the tests emit sequentially). The + // handshake resolves with the frame correlated to the live emission only; + // any frame arriving before or after it is fenced as stale and recorded. + let pending: { id: string; resolve: (frame: Record) => void } | undefined; + const lateResponses: Array> = []; + const delayedDeliveries = new Set>(); const transport: SessionSdkTransport = { sessionId, stateRoot: path.join(cwd, ".gjc", "state"), token: "test-token", sendFrame: (_connectionId, frame) => { - response = frame as Record; + const response = frame as Record; + const deliver = () => { + const current = pending; + // A response only satisfies the emission whose request id it + // carries. Anything else (a late frame from an already ended + // emission, or an unsolicited frame) must never resolve a later + // await. + if (current && response.id === current.id) { + pending = undefined; + current.resolve(response); + } else { + lateResponses.push(response); + } + }; + if (responseDelayMs <= 0) { + deliver(); + return; + } + const timer = setTimeout(() => { + delayedDeliveries.delete(timer); + deliver(); + }, responseDelayMs); + delayedDeliveries.add(timer); }, onFrame: handler => { receive = handler; @@ -33,7 +85,10 @@ function createHarness(cwd: string, sessionId: string, sessionFile: string | und }; }, start: async () => ({ url: "memory://host-steer" }), - stop: async () => {}, + stop: async () => { + for (const timer of delayedDeliveries) clearTimeout(timer); + delayedDeliveries.clear(); + }, }; const api = { on: (event: string, handler: (event: unknown, context: ExtensionContext) => unknown) => @@ -79,11 +134,22 @@ function createHarness(cwd: string, sessionId: string, sessionFile: string | und await handlers.get("session_shutdown")?.({ type: "session_shutdown" }, base); }, emit: async frame => { - response = undefined; - receive?.("client", frame as never); - for (let attempts = 0; response === undefined && attempts < 100; attempts++) await Bun.sleep(1); - if (!response) throw new Error("host did not respond"); - return response; + const { promise, resolve } = Promise.withResolvers>(); + pending = { id: String(frame.id), resolve }; + try { + receive?.("client", frame as never); + // Deterministic handshake: resolved by the correlated response + // frame itself, with a bounded safety timeout. No wall-clock poll + // loop and no fixed sleep budget. + const response = await Promise.race([promise, Bun.sleep(responseTimeoutMs).then(() => undefined)]); + if (!response) throw new Error("host did not respond"); + return response; + } finally { + pending = undefined; + } + }, + setResponseDelay: ms => { + responseDelayMs = ms; }, get dispatches() { return dispatches; @@ -91,6 +157,9 @@ function createHarness(cwd: string, sessionId: string, sessionFile: string | und get persistedAtDispatch() { return persistedAtDispatch; }, + get lateResponses() { + return lateResponses; + }, }; } @@ -102,6 +171,52 @@ async function query(harness: Harness, id: string, input: Record { + const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "gjc-host-steer-handshake-")); + try { + const harness = createHarness(cwd, "handshake-delay", undefined, { responseDelayMs: 250 }); + await harness.start(); + // The response arrives well after the retired 100 x Bun.sleep(1) budget + // would have expired; the correlated-id handshake still resolves it. + const delayed = await control(harness, "delayed", "delayed steer", "delayed-ref"); + expect(delayed).toMatchObject({ ok: true, result: { accepted: true, clientRef: "delayed-ref" } }); + expect(harness.dispatches).toBe(1); + expect(harness.lateResponses).toEqual([]); + await harness.stop(); + } finally { + await fs.rm(cwd, { recursive: true, force: true }); + } +}); + +test("harness handshake times out, and a late response never satisfies a later emission", async () => { + const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "gjc-host-steer-timeout-")); + try { + const harness = createHarness(cwd, "handshake-timeout", undefined, { + responseDelayMs: 150, + responseTimeoutMs: 50, + }); + await harness.start(); + // A response slower than the safety bound must fail the emission + // deterministically instead of hanging, and must leave the harness usable. + await expect(control(harness, "slow", "slow steer", "slow-ref")).rejects.toThrow("host did not respond"); + // The stale frame for the timed-out id lands after that emission ended: + // it is fenced into lateResponses, never resolving a live await. + await Bun.sleep(200); + expect(harness.lateResponses.length).toBe(1); + expect(harness.lateResponses[0]).toMatchObject({ id: "slow", type: "control_response" }); + // A fresh emission under a different id resolves only against its own + // correlated response. + harness.setResponseDelay(0); + const accepted = await control(harness, "recovered", "recovering steer", "recovered-ref"); + expect(accepted).toMatchObject({ ok: true, result: { accepted: true, clientRef: "recovered-ref" } }); + expect(harness.dispatches).toBe(2); + expect(harness.lateResponses.length).toBe(1); + await harness.stop(); + } finally { + await fs.rm(cwd, { recursive: true, force: true }); + } +}); + test("production SDK host correlates durable steer replay and restart without redispatch", async () => { const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "gjc-host-steer-")); try { @@ -136,6 +251,9 @@ test("production SDK host correlates durable steer replay and restart without re ok: true, result: { clientRef: "caller-ref" }, }); + // Sequential controls each resolved against their OWN correlated + // response: no stale or unsolicited frame satisfied any later emission. + expect(first.lateResponses).toEqual([]); const persisted = await fs.readFile( path.join(path.dirname(sessionFile), ".sdk-reconciliation", `${sessionId}.json`), "utf8",