From af0ec8b77e9b7a0d65cbff284484487c985b71e0 Mon Sep 17 00:00:00 2001 From: gaebal-gajae Date: Sun, 9 Aug 2026 03:57:17 +0000 Subject: [PATCH 1/3] fix(sdk): publish session identity for WebSocket hosts The SDK-only transport introduced by #3846 omitted sessionId from endpoint discovery records. The broker correctly rejected those records, making durable workflow-gate control unavailable before its request could reach the host. Lore-id: 4047\nConstraint: preserve broker identity fences and avoid tmux pane input\nConfidence: high\nScope-risk: narrow\nReversibility: direct\nTested: bun test sdk-broker-host-integration; bun test sdk-host-wiring durable gate; bun --cwd=packages/coding-agent run check --- packages/coding-agent/CHANGELOG.md | 2 + .../src/sdk/host/session-runtime.test.ts | 96 ++++++++++- .../src/sdk/host/session-runtime.ts | 159 ++++++++++++++++-- .../src/sdk/host/websocket-transport.ts | 8 +- packages/coding-agent/src/sdk/session.ts | 4 +- .../test/sdk-broker-host-integration.test.ts | 135 +++++++++++++++ 6 files changed, 391 insertions(+), 13 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 21da6a95bb..47e7cfd13f 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -30,6 +30,8 @@ ### Added - `gjc setup provider` now ships parameterized proxy presets: `--preset litellm` and `--preset openai-compatible-proxy` (aliases `litellm-proxy`, `openai-proxy`, `compatible-proxy`, `custom-proxy`) with a required `--base-url`, configurable `--api-key-env`, and live model discovery (#4123). - New `modelProfile.proxyProvider` and `modelProfile.proxyMode` settings route built-in model-preset selectors through an authenticated OpenAI-compatible proxy (e.g. `xai/grok-4.3` → `litellm/xai/grok-4.3`). `fallback` preserves directly authenticated providers by default; `always` forces every proxy-routable built-in selector through the configured gateway. Routing fails closed for unconfigured or unauthenticated proxies and missing or ambiguous proxy models (#4123). +- SDK-only session hosts now publish their session ID and register their endpoint lifecycle with the broker, matching its identity and staleness fences. This restores broker/coordinator resolution for durable workflow-gate controls (`workflow.gates.list` and `workflow.gate_answer`) without relying on tmux pane input; broker unavailability leaves non-lifecycle local hosts usable and retries publication later. + ## [0.12.21] - 2026-08-09 ### Fixed diff --git a/packages/coding-agent/src/sdk/host/session-runtime.test.ts b/packages/coding-agent/src/sdk/host/session-runtime.test.ts index 8f41aa5672..d3b51dbec0 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.test.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.test.ts @@ -1,8 +1,9 @@ import { describe, expect, test } from "bun:test"; -import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import * as os from "node:os"; import * as path from "node:path"; import type { ExtensionAPI } from "../../extensibility/extensions"; +import { Broker } from "../broker/broker"; import { createInvocationReconciliation, createSdkSessionRuntimeExtension, @@ -263,6 +264,7 @@ describe("SessionSdkSessionRuntime", () => { } as any; const transports: Array<{ starts: number; stops: number }> = []; createSdkSessionRuntimeExtension(api, { + agentDir: path.join(cwd, ".gjc", "agent"), createTransport: async ({ sessionId, stateRoot, token }) => { const stats = { starts: 0, stops: 0 }; const failFirstStop = transports.length === 0; @@ -281,6 +283,9 @@ describe("SessionSdkSessionRuntime", () => { sendFrame: () => {}, start: async () => { stats.starts += 1; + const endpoint = path.join(stateRoot, "sdk", `${sessionId}.json`); + await mkdir(path.dirname(endpoint), { recursive: true }); + await writeFile(endpoint, JSON.stringify({ sessionId, token, pid: process.pid })); return { url: `ws://127.0.0.1:${30_000 + stats.starts}` }; }, stop: async () => { @@ -316,6 +321,95 @@ describe("SessionSdkSessionRuntime", () => { await rm(cwd, { recursive: true, force: true }); } }); + test("keeps a local SDK-only host alive through broker failure and registers after recovery", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-sdk-broker-recovery-")); + const agentDir = path.join(cwd, ".gjc", "agent"); + await mkdir(path.dirname(agentDir), { recursive: true }); + await writeFile(agentDir, "blocked"); + const handlers = new Map Promise | void>(); + const api = { + on(event: string, handler: (event: unknown, ctx: any) => Promise | void) { + handlers.set(event, handler); + }, + } as any; + const sessionId = "broker-recovery"; + createSdkSessionRuntimeExtension(api, { + agentDir, + createTransport: async ({ stateRoot, token }) => ({ + sessionId, + stateRoot, + token, + onFrame: () => undefined, + sendFrame: () => {}, + start: async () => { + const endpoint = path.join(stateRoot, "sdk", `${sessionId}.json`); + await mkdir(path.dirname(endpoint), { recursive: true }); + await writeFile(endpoint, JSON.stringify({ sessionId, token, pid: process.pid })); + return { url: "ws://127.0.0.1:1" }; + }, + stop: async () => {}, + }), + }); + const context = extensionContext(sessionId, cwd); + let broker: Broker | undefined; + try { + await handlers.get("session_start")?.({}, context); + await rm(agentDir); + await mkdir(agentDir, { recursive: true }); + broker = new Broker({ agentDir }); + await broker.start(); + await handlers.get("turn_start")?.({}, context); + expect(await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 })).toMatchObject( + { + ok: true, + result: { sessionId, token: expect.any(String) }, + }, + ); + await handlers.get("session_shutdown")?.({}, context); + expect(await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 })).toMatchObject( + { + ok: false, + error: { code: "resource_gone" }, + }, + ); + } finally { + await broker?.stop(); + await rm(cwd, { recursive: true, force: true }); + } + }); + + test("rejects lifecycle-required SDK-only startup when broker registration fails", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "gjc-sdk-broker-required-")); + const agentDir = path.join(cwd, ".gjc", "agent"); + await mkdir(path.dirname(agentDir), { recursive: true }); + await writeFile(agentDir, "blocked"); + const handlers = new Map Promise | void>(); + const api = { + on(event: string, handler: (event: unknown, ctx: any) => Promise | void) { + handlers.set(event, handler); + }, + } as any; + createSdkSessionRuntimeExtension(api, { + agentDir, + brokerRegistrationRequired: true, + createTransport: async ({ sessionId, stateRoot, token }) => ({ + sessionId, + stateRoot, + token, + onFrame: () => undefined, + sendFrame: () => {}, + start: async () => ({ url: "ws://127.0.0.1:1" }), + stop: async () => {}, + }), + }); + try { + await expect( + handlers.get("session_start")?.({}, extensionContext("broker-required", cwd)), + ).rejects.toBeDefined(); + } finally { + await rm(cwd, { recursive: true, force: true }); + } + }); }); interface PreflightHooks { onPreflightAccepted?: () => void; diff --git a/packages/coding-agent/src/sdk/host/session-runtime.ts b/packages/coding-agent/src/sdk/host/session-runtime.ts index bd1ec6dc51..dbdc149558 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.ts @@ -14,6 +14,8 @@ import { type Settings, validateSettingPatch } from "../../config/settings"; import type { ExtensionAPI, ExtensionCommandContext, ExtensionContext } from "../../extensibility/extensions"; import { parseThinkingLevel } from "../../thinking"; import { elevationAuthorityPath, verifyElevationCapability } from "../elevation/capability"; +import { ensureBroker } from "../broker/ensure"; +import { SessionIndex } from "../broker/session-index"; import { collectAuthenticatedProfileProviders, parseSyntheticModelId, @@ -218,6 +220,10 @@ export class SessionSdkSessionRuntime { /** Narrow extension-facing factory for the SDK-only session path. */ export interface CreateSdkSessionRuntimeOptions { + /** Authoritative broker state root for this session's endpoint lifecycle. */ + agentDir: string; + /** Lifecycle-owned sessions require broker publication before they become usable. */ + brokerRegistrationRequired?: boolean; createTransport(input: { sessionId: string; stateRoot: string; @@ -906,6 +912,52 @@ function containsSecretConfigKey(value: unknown, seen = new Set()): bool containsSecretConfigKey(nested, seen), ); } +async function resolveSdkWorkflowGate( + ctx: ExtensionContext, + operation: "workflow.gate_answer" | "workflow.plan_approve", + id: string, + answer: unknown, + expectedSessionId: string | undefined, + idempotencyKey: string, + canResolve: () => boolean, +): Promise { + if (!canResolve()) + throw Object.assign(new Error("Workflow gate is no longer answerable."), { code: "resource_gone" }); + if (expectedSessionId !== undefined && expectedSessionId !== ctx.sessionManager.getSessionId()) + throw Object.assign(new Error("Workflow gate session does not match this endpoint."), { code: "resource_gone" }); + if (expectedSessionId === undefined) logger.warn("workflow_control_missing_expected_session_id", { operation }); + const workflowGate = ctx.workflowGate; + if ( + typeof workflowGate?.resolveGate !== "function" || + typeof workflowGate.recoverAcceptedGates !== "function" || + typeof workflowGate.lookupCompletedResolution !== "function" || + typeof workflowGate.prepareTerminalization !== "function" || + typeof workflowGate.clearPreparedTerminalization !== "function" + ) + throw Object.assign(new Error("Workflow gates are unavailable for this session."), { code: "resource_gone" }); + const response = { gate_id: id, answer, idempotency_key: idempotencyKey }; + const completed = workflowGate.lookupCompletedResolution(response); + if (completed.kind === "completed") return completed.resolution; + if (completed.kind === "accepted_incomplete") { + await workflowGate.recoverAcceptedGates(); + const recovered = workflowGate.lookupCompletedResolution(response); + if (recovered.kind === "completed") return recovered.resolution; + throw Object.assign(new Error("Workflow gate resolution outcome is uncertain."), { code: "terminal_uncertain" }); + } + if (!workflowGate.prepareTerminalization(id, "not_published")) + throw Object.assign(new Error("Workflow gate is no longer answerable."), { code: "resource_gone" }); + try { + const resolution = await workflowGate.resolveGate(response); + if ((resolution as { status?: unknown }).status === "rejected") workflowGate.clearPreparedTerminalization(id); + return resolution; + } catch (error) { + const stillPending = workflowGate.listPendingGates?.().some(gate => gate.gate_id === id) === true; + if (stillPending) workflowGate.clearPreparedTerminalization(id); + else workflowGate.quarantineGate?.(id); + throw error; + } +} + function createControlSurface( ctx: ExtensionContext, api: ExtensionAPI, @@ -916,6 +968,8 @@ function createControlSurface( configOverrides?: Map, configRevision: { current: number } = { current: 0 }, elevationAuthorityToken?: string, + canResolveGate: () => boolean = () => true, + trackGateResolution: (resolution: Promise) => Promise = async resolution => await resolution, ): ControlSurface { const surfacePolicy = policy ?? createSdkSurfacePolicyForContext(ctx, hasSdkWorkflowGateCapability(ctx.workflowGate)); @@ -1084,9 +1138,22 @@ function createControlSurface( return await submit("prompt", undefined, options => api.sendUserMessage(text, options)); }, answerAsk: unavailable("ask.answer"), - answerGate: (_id, _response, _expectedSessionId, _idempotencyKey, _elevationRequestId) => - unavailable("workflow.gate_answer")(), - approvePlan: (_id, _choice, _expectedSessionId, _elevationRequestId) => unavailable("workflow.plan_approve")(), + answerGate: async (id, response, expectedSessionId, idempotencyKey) => + await trackGateResolution( + resolveSdkWorkflowGate( + ctx, + "workflow.gate_answer", + id, + response, + expectedSessionId, + idempotencyKey ?? id, + canResolveGate, + ), + ), + approvePlan: async (id, choice, expectedSessionId) => + await trackGateResolution( + resolveSdkWorkflowGate(ctx, "workflow.plan_approve", id, choice, expectedSessionId, id, canResolveGate), + ), invokeSkill: async (name, args, clientRef) => { if (!ctx.invokeSkill) return unavailable("skill.invoke")(); if (args !== undefined && typeof args !== "string") @@ -1268,6 +1335,9 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre cursors: CursorRegistry; reconciliation: InvocationReconciliation; pending: Array<{ kind: InvocationKind; correlation: InvocationCorrelation }>; + registerBroker: () => Promise; + fenceGateResolutions: () => void; + waitForGateResolutionQuiescence: () => Promise; activeInvocation?: { kind: InvocationKind; correlation: InvocationCorrelation }; disposeGate?: () => void; } @@ -1286,9 +1356,12 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre }; api.on("agent_start", async (_event, ctx) => await emitLifecycle("agent_start", ctx)); api.on("agent_end", async (_event, ctx) => await emitLifecycle("agent_end", ctx)); - api.on("turn_start", (_event, ctx) => - active?.runtime.emitEvent({ type: "turn_start", sessionId: ctx.sessionManager.getSessionId() }), - ); + api.on("turn_start", async (_event, ctx) => { + const current = active; + if (!current) return; + await current.registerBroker(); + current.runtime.emitEvent({ type: "turn_start", sessionId: ctx.sessionManager.getSessionId() }); + }); api.on("turn_end", (_event, ctx) => active?.runtime.emitEvent({ type: "turn_end", sessionId: ctx.sessionManager.getSessionId() }), ); @@ -1321,6 +1394,20 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre // read the live generation/seq once the runtime exists (Q30 atomic // capture, C9). let eventWatermarkSource: () => { generation: number; seq: number } = () => ({ generation: 0, seq: 0 }); + let acceptingGateResolutions = true; + const inFlightGateResolutions = new Set>(); + const trackGateResolution = (resolution: Promise): Promise => { + const tracked = resolution.finally(() => inFlightGateResolutions.delete(tracked)); + inFlightGateResolutions.add(tracked); + return tracked; + }; + const waitForGateResolutionQuiescence = async (): Promise => { + const settled = Promise.allSettled(inFlightGateResolutions); + const timeout = Bun.sleep(5_000).then(() => { + throw new Error("Timed out waiting for SDK workflow gate resolutions to settle."); + }); + await Promise.race([settled, timeout]); + }; const surfaceFactory = createSdkSurfaceFactory({ ctx, id: sessionId, @@ -1345,6 +1432,8 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre options.configOverrides, configRevision, elevationAuthorityToken, + () => acceptingGateResolutions, + trackGateResolution, ); let runtime: SessionSdkSessionRuntime; const installProviderDefinitions = (capability: string, definitions: unknown): void => { @@ -1458,9 +1547,45 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre const disposeGate = ctx.workflowGate?.onGateEmitted?.(gate => runtime.emitEvent({ kind: "workflow_gate", payload: gate }), ); - active = { runtime, revisions, cursors, reconciliation, pending, disposeGate }; + let brokerRegistered = false; + const registerBroker = async (): Promise => { + if (brokerRegistered) return; + try { + await ensureBroker({ agentDir: options.agentDir }); + const index = await new SessionIndex(options.agentDir).open(); + const locator = { repo: path.resolve(ctx.cwd), stateRoot }; + await runtime.registerWithBroker({ + register: async input => { + const endpointMtimeMs = (await fs.stat(path.join(input.stateRoot, "sdk", `${input.sessionId}.json`))) + .mtimeMs; + await index.append({ type: "host_registered", ...input, locator, pid: process.pid, endpointMtimeMs }); + }, + unregister: async input => { + await index.append({ type: "host_unregistered", ...input, locator, pid: process.pid }); + }, + }); + brokerRegistered = true; + } catch (error) { + if (options.brokerRegistrationRequired) throw error; + logger.warn(`sdk broker registration unavailable: ${String(error)}`); + } + }; + active = { + runtime, + revisions, + cursors, + reconciliation, + pending, + registerBroker, + fenceGateResolutions: () => { + acceptingGateResolutions = false; + }, + waitForGateResolutionQuiescence, + disposeGate, + }; try { await runtime.start(); + await registerBroker(); } catch (error) { active = undefined; disposeGate?.(); @@ -1471,7 +1596,19 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre code: errorCode(cleanupError), error: String(cleanupError), }); - active = { runtime, revisions, cursors, reconciliation, pending, disposeGate }; + active = { + runtime, + revisions, + cursors, + reconciliation, + pending, + registerBroker, + fenceGateResolutions: () => { + acceptingGateResolutions = false; + }, + waitForGateResolutionQuiescence, + disposeGate, + }; throw new AggregateError([error, cleanupError], "SDK runtime startup failed and cleanup failed."); } cursors.close(); @@ -1481,10 +1618,12 @@ export function createSdkSessionRuntimeExtension(api: ExtensionAPI, options: Cre }; const stopActive = async (): Promise => { const current = active; - active = undefined; if (!current) return; - current.disposeGate?.(); + current.fenceGateResolutions(); try { + await current.waitForGateResolutionQuiescence(); + active = undefined; + current.disposeGate?.(); await current.runtime.stop(); } catch (error) { logger.error("sdk runtime stop failed", { code: errorCode(error), error: String(error) }); diff --git a/packages/coding-agent/src/sdk/host/websocket-transport.ts b/packages/coding-agent/src/sdk/host/websocket-transport.ts index 94cf4b3afd..59d9d08806 100644 --- a/packages/coding-agent/src/sdk/host/websocket-transport.ts +++ b/packages/coding-agent/src/sdk/host/websocket-transport.ts @@ -198,7 +198,13 @@ export async function createSdkWebSocketTransport( try { await filesystem.writeFile( endpointFile, - JSON.stringify({ version: 1, url, token: input.token, pid: process.pid }), + JSON.stringify({ + version: 1, + sessionId: input.sessionId, + url, + token: input.token, + pid: process.pid, + }), "utf8", ); } catch (error) { diff --git a/packages/coding-agent/src/sdk/session.ts b/packages/coding-agent/src/sdk/session.ts index e45ac5f2e5..86253ad676 100644 --- a/packages/coding-agent/src/sdk/session.ts +++ b/packages/coding-agent/src/sdk/session.ts @@ -117,6 +117,7 @@ import { import { NotificationSessionController } from "../sdk/bus/session-control"; import { shouldHostSdk } from "../sdk/host"; import { createSdkSessionRuntimeExtension, registerSdkOnlyNotificationCommand } from "../sdk/host/session-runtime"; +import { createSdkWebSocketTransport } from "../sdk/host/websocket-transport"; import type { SecretObfuscator } from "../secrets"; import { AgentSession, type ForkContextSeed } from "../session/agent-session"; import { resolveAuthBrokerConfig } from "../session/auth-broker-config"; @@ -2174,8 +2175,9 @@ export async function createAgentSession(options: CreateAgentSessionOptions = {} }); } else if (sdkHostEligible) { registerSdkOnlyNotificationCommand(api); - const { createSdkWebSocketTransport } = await import("../sdk/host/websocket-transport"); createSdkSessionRuntimeExtension(api, { + agentDir, + brokerRegistrationRequired: lifecycleStartupCapability !== undefined, createTransport: input => createSdkWebSocketTransport(input), settings, configOverrides: new Map(), diff --git a/packages/coding-agent/test/sdk-broker-host-integration.test.ts b/packages/coding-agent/test/sdk-broker-host-integration.test.ts index 05f3e487df..f5c740f8ec 100644 --- a/packages/coding-agent/test/sdk-broker-host-integration.test.ts +++ b/packages/coding-agent/test/sdk-broker-host-integration.test.ts @@ -1,8 +1,11 @@ import { expect, test } from "bun:test"; import * as fs from "node:fs/promises"; import * as path from "node:path"; +import type { ExtensionAPI, ExtensionContext } from "../src/extensibility/extensions"; import { Broker } from "../src/sdk/broker/broker"; import { SessionIndex } from "../src/sdk/broker/session-index"; +import { createSdkSessionRuntimeExtension } from "../src/sdk/host/session-runtime"; +import { createSdkWebSocketTransport } from "../src/sdk/host/websocket-transport"; const event = ( type: "host_registered" | "host_heartbeat" | "host_unregistered", @@ -164,3 +167,135 @@ test("broker session.list rejects a new cursor stream at capacity without evicti await fs.rm(agentDir, { recursive: true, force: true }); } }); + +test("SDK-only runtime registers its broker endpoint and retracts it on shutdown", async () => { + const agentDir = await fs.mkdtemp(path.join(process.env.TMPDIR ?? "/tmp", "gjc-sdk-only-broker-")); + const cwd = path.join(agentDir, "workspace"); + const sessionId = "sdk-only-live"; + const broker = new Broker({ agentDir }); + await broker.start(); + const handlers = new Map void | Promise>(); + const api = { + on(event: string, handler: (event: unknown, ctx: ExtensionContext) => void | Promise) { + handlers.set(event, handler); + }, + } as unknown as ExtensionAPI; + createSdkSessionRuntimeExtension(api, { + agentDir, + createTransport: input => createSdkWebSocketTransport(input), + }); + const pendingGateIds = new Set(["gate-answer", "gate-approve", "gate-drain"]); + let drainGateResolution: (() => void) | undefined; + const workflowGate = { + listWorkflowGateQueryRecords: () => + [...pendingGateIds].map(gateId => ({ id: `pending:${gateId}`, gate_id: gateId, tag: "pending" })), + listPendingGates: () => [...pendingGateIds].map(gate_id => ({ gate_id })), + resolveGate: async (response: { gate_id: string }) => { + if (response.gate_id === "gate-drain") + return await new Promise(resolve => { + drainGateResolution = () => { + pendingGateIds.delete(response.gate_id); + resolve({ gate_id: response.gate_id, status: "accepted" }); + }; + }); + pendingGateIds.delete(response.gate_id); + return { gate_id: response.gate_id, status: "accepted" }; + }, + recoverAcceptedGates: async () => [], + lookupCompletedResolution: () => ({ kind: "none" }), + prepareTerminalization: () => true, + clearPreparedTerminalization: () => {}, + registerGateTerminalController: () => () => {}, + quarantineGate: () => {}, + }; + const context = { + cwd, + sdkBindings: () => [], + sessionManager: { getSessionId: () => sessionId, getSessionName: () => undefined }, + workflowGate, + } as unknown as ExtensionContext; + try { + const start = handlers.get("session_start"); + if (!start) throw new Error("SDK-only session_start handler was not registered."); + await start({}, context); + expect(await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 })).toMatchObject({ + ok: true, + result: { sessionId, pid: process.pid, url: expect.stringMatching(/^ws:\/\/127\.0\.0\.1:/) }, + }); + const endpoint = await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 }); + if (!endpoint.ok) throw new Error(endpoint.error.message); + const socket = new WebSocket( + `${(endpoint.result as { url: string; token: string }).url}?token=${encodeURIComponent((endpoint.result as { token: string }).token)}`, + ); + const frames: Array> = []; + socket.addEventListener("message", event => frames.push(JSON.parse(String(event.data)))); + await new Promise((resolve, reject) => { + socket.addEventListener("open", () => resolve(), { once: true }); + socket.addEventListener("error", () => reject(new Error("SDK-only WebSocket failed to open.")), { + once: true, + }); + }); + const request = async (id: string, frame: Record) => { + socket.send(JSON.stringify({ ...frame, id })); + const deadline = Date.now() + 2_000; + while (!frames.some(candidate => candidate.id === id)) { + if (Date.now() > deadline) throw new Error(`Timed out awaiting ${id}.`); + await Bun.sleep(10); + } + return frames.find(candidate => candidate.id === id)!; + }; + expect(await request("gates", { type: "query_request", query: "Q12", input: {} })).toMatchObject({ + type: "query_response", + ok: true, + page: { items: [{ gate_id: "gate-answer" }, { gate_id: "gate-approve" }, { gate_id: "gate-drain" }] }, + }); + expect( + await request("wrong-session", { + type: "control_request", + operation: "workflow.gate_answer", + input: { id: "gate-answer", response: "approve", expectedSessionId: "wrong-session" }, + }), + ).toMatchObject({ type: "control_response", ok: false, error: { code: "resource_gone" } }); + expect( + await request("answer", { + type: "control_request", + operation: "workflow.gate_answer", + input: { id: "gate-answer", response: "approve", expectedSessionId: sessionId }, + }), + ).toMatchObject({ type: "control_response", ok: true, result: { status: "accepted" } }); + expect( + await request("approve", { + type: "control_request", + operation: "workflow.plan_approve", + input: { id: "gate-approve", choice: "approve", expectedSessionId: sessionId }, + }), + ).toMatchObject({ type: "control_response", ok: true, result: { status: "accepted" } }); + socket.send( + JSON.stringify({ + type: "control_request", + id: "drain", + operation: "workflow.gate_answer", + input: { id: "gate-drain", response: "approve", expectedSessionId: sessionId }, + }), + ); + while (!drainGateResolution) await Bun.sleep(10); + const shutdown = handlers.get("session_shutdown"); + if (!shutdown) throw new Error("SDK-only session_shutdown handler was not registered."); + const stopping = Promise.resolve(shutdown({}, context)); + await Bun.sleep(10); + expect(await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 })).toMatchObject({ + ok: true, + }); + drainGateResolution(); + await stopping; + expect(await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 })).toMatchObject({ + ok: false, + error: { code: "resource_gone" }, + }); + } finally { + const shutdown = handlers.get("session_shutdown"); + if (shutdown) await Promise.resolve(shutdown({}, context)).catch(() => undefined); + await broker.stop(); + await fs.rm(agentDir, { recursive: true, force: true }); + } +}); From 03be8af8a0a7179c7ab66bc0b697bb1db1ccbd37 Mon Sep 17 00:00:00 2001 From: Yeachan Heo Date: Tue, 11 Aug 2026 00:04:08 +0000 Subject: [PATCH 2/3] test(sdk): configure runtime harness broker root The SDK runtime extension now requires its broker state root. The invocation harness must supply it so the durable workflow-gate regression suite typechecks and exercises the production admission path. Lore-id: pr-4064-harness-root Confidence: high Scope-risk: narrow Reversibility: trivial Tested: bun test packages/coding-agent/src/sdk/host/session-runtime.test.ts; bun test packages/coding-agent/test/sdk-broker-host-integration.test.ts; bun --cwd=packages/coding-agent run check --- packages/coding-agent/src/sdk/host/session-runtime.test.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/packages/coding-agent/src/sdk/host/session-runtime.test.ts b/packages/coding-agent/src/sdk/host/session-runtime.test.ts index d3b51dbec0..a0eb18b25e 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.test.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.test.ts @@ -453,6 +453,7 @@ async function invocationHarness( sendUserMessage: hooks.sendUserMessage ?? (async () => {}), } as unknown as ExtensionAPI; createSdkSessionRuntimeExtension(api, { + agentDir: cwd, createTransport: async ({ sessionId: id, stateRoot, token }) => ({ sessionId: id, stateRoot, From 9b80b9dfcd8332c14d12ef14e465b9b81b9231cb Mon Sep 17 00:00:00 2001 From: Yeachan Heo Date: Tue, 11 Aug 2026 05:47:56 +0000 Subject: [PATCH 3/3] test(sdk): align broker host assertions with dev Dev now retains terminal host rows and reports stale endpoint authority after unregister. The rebased workflow-gate host coverage must assert those durable broker contracts rather than the superseded removal semantics. Lore-id: pr-4064-rebase-dev\nConfidence: high\nScope-risk: narrow\nReversibility: trivial\nTested: bun test packages/coding-agent/src/sdk/host/session-runtime.test.ts; bun test packages/coding-agent/test/sdk-broker-host-integration.test.ts; bun --cwd=packages/coding-agent run check --- .../coding-agent/src/sdk/host/session-runtime.test.ts | 2 +- packages/coding-agent/src/sdk/host/session-runtime.ts | 2 +- .../test/sdk-broker-host-integration.test.ts | 10 +++++----- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/packages/coding-agent/src/sdk/host/session-runtime.test.ts b/packages/coding-agent/src/sdk/host/session-runtime.test.ts index a0eb18b25e..c114ac7cd2 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.test.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.test.ts @@ -369,7 +369,7 @@ describe("SessionSdkSessionRuntime", () => { expect(await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 })).toMatchObject( { ok: false, - error: { code: "resource_gone" }, + error: { code: "endpoint_stale", message: "session endpoint is stale" }, }, ); } finally { diff --git a/packages/coding-agent/src/sdk/host/session-runtime.ts b/packages/coding-agent/src/sdk/host/session-runtime.ts index dbdc149558..2cec957696 100644 --- a/packages/coding-agent/src/sdk/host/session-runtime.ts +++ b/packages/coding-agent/src/sdk/host/session-runtime.ts @@ -13,9 +13,9 @@ import { type ModelSelectorValue, normalizeModelSelectorValue } from "../../conf import { type Settings, validateSettingPatch } from "../../config/settings"; import type { ExtensionAPI, ExtensionCommandContext, ExtensionContext } from "../../extensibility/extensions"; import { parseThinkingLevel } from "../../thinking"; -import { elevationAuthorityPath, verifyElevationCapability } from "../elevation/capability"; import { ensureBroker } from "../broker/ensure"; import { SessionIndex } from "../broker/session-index"; +import { elevationAuthorityPath, verifyElevationCapability } from "../elevation/capability"; import { collectAuthenticatedProfileProviders, parseSyntheticModelId, diff --git a/packages/coding-agent/test/sdk-broker-host-integration.test.ts b/packages/coding-agent/test/sdk-broker-host-integration.test.ts index f5c740f8ec..2b2a39f04a 100644 --- a/packages/coding-agent/test/sdk-broker-host-integration.test.ts +++ b/packages/coding-agent/test/sdk-broker-host-integration.test.ts @@ -51,7 +51,7 @@ test("broker preserves host registration endpoint metadata across heartbeats", a await busIndex.append(event("host_unregistered", "live", stateRoot)); expect(await broker.handleRequest("session.list", {})).toMatchObject({ ok: true, - result: { indexSeq: 4, sessions: [] }, + result: { indexSeq: 4, sessions: [{ sessionId: "live", live: false, terminal: true }] }, }); } finally { await broker.stop(); @@ -78,7 +78,7 @@ test("broker session.list returns bounded stable cursor pages", async () => { sessions: Array<{ sessionId: string }>; continuationCursor?: string; }; - expect(firstPage).toMatchObject({ indexSeq: 3, sessions: [{ sessionId: "one" }, { sessionId: "two" }] }); + expect(firstPage.sessions).toMatchObject([{ sessionId: "one" }, { sessionId: "two" }]); expect(firstPage.continuationCursor).toEqual(expect.any(String)); await busIndex.append(event("host_registered", "four", stateRoot)); @@ -86,8 +86,8 @@ test("broker session.list returns bounded stable cursor pages", async () => { const second = await broker.handleRequest("session.list", { cursor: firstPage.continuationCursor }); expect(second).toMatchObject({ ok: true, - indexSeq: 3, - result: { indexSeq: 3, sessions: [{ sessionId: "three" }] }, + indexSeq: firstPage.indexSeq, + result: { indexSeq: firstPage.indexSeq, sessions: [{ sessionId: "three" }] }, }); expect(JSON.stringify(second)).not.toContain('"four"'); expect(await broker.handleRequest("session.list", { limit: 101 })).toEqual({ @@ -290,7 +290,7 @@ test("SDK-only runtime registers its broker endpoint and retracts it on shutdown await stopping; expect(await broker.handleRequest("session.get_endpoint", { sessionId, endpointGeneration: 1 })).toMatchObject({ ok: false, - error: { code: "resource_gone" }, + error: { code: "endpoint_stale", message: "session endpoint is stale" }, }); } finally { const shutdown = handlers.get("session_shutdown");