diff --git a/src/handlers/dispatch.ts b/src/handlers/dispatch.ts index 8209a8f..82f762f 100644 --- a/src/handlers/dispatch.ts +++ b/src/handlers/dispatch.ts @@ -99,9 +99,18 @@ async function dispatchConfigChanged( // events routed through a registered turn loop always have it. const zcodeSid = server.resolveSid(acpSid) ?? null; const options = await buildConfigOptions(server, zcodeSid); - if (ev.model) options[0].currentValue = formatModelValue(ev.model.providerId, ev.model.modelId); - if (ev.mode !== undefined) options[1].currentValue = ev.mode; - if (ev.thought !== undefined) options[2].currentValue = ev.thought; + // Find by id — the array order buildConfigOptions returns is not a + // contract; index-based writes would silently hit the wrong option if + // that order ever changed (emitModeViaConfigOption already does this). + const setById = (id: string, value: string) => { + const opt = options.find((o) => o.id === id); + if (opt) opt.currentValue = value; + }; + if (ev.model) { + setById("model", formatModelValue(ev.model.providerId, ev.model.modelId)); + } + if (ev.mode !== undefined) setById("mode", ev.mode); + if (ev.thought !== undefined) setById("thought", ev.thought); await sendSessionUpdate(cx, acpSid, { sessionUpdate: "config_option_update", configOptions: options, diff --git a/src/handlers/extensions.ts b/src/handlers/extensions.ts index 4a9d091..fb9b95a 100644 --- a/src/handlers/extensions.ts +++ b/src/handlers/extensions.ts @@ -264,6 +264,12 @@ export async function setMode( * the lock isn't held yet, so a probe succeeds immediately (false "released"). * With expectLock=true, we first require observing "prompt is running" once * (proving the turn truly started) before trusting a later success. + * + * `graceMs` bounds that lock-watching phase: if the lock is still unseen past + * the grace, a successful probe counts as released. Without it, a turn that + * finishes between two probes — or a backend whose lock error message drifted + * away from "prompt is running" (version drift) — spins the full timeout and + * reports a false failure. */ export async function waitForTurnIdle( server: ZcodeAcpServer, @@ -271,6 +277,7 @@ export async function waitForTurnIdle( timeoutMs: number, probeMethod: string, expectLock: boolean, + graceMs = 30_000, ): Promise { const backend = server.ensureBackend(); const t0 = Date.now(); @@ -309,6 +316,12 @@ export async function waitForTurnIdle( ); return true; } + if (Date.now() - t0 >= graceMs) { + log( + ` [probe] #${probeCount} @${elapsed}s: NON-LOCK error, grace expired → released (err="${errMsg.slice(0, 50)}")`, + ); + return true; + } log( ` [probe] #${probeCount} @${elapsed}s: NON-LOCK error, lockSeen=false → wait for lock (err="${errMsg.slice(0, 50)}")`, ); @@ -319,6 +332,10 @@ export async function waitForTurnIdle( log(` [probe] #${probeCount} @${elapsed}s: probe success after lock → released`); return true; } + if (Date.now() - t0 >= graceMs) { + log(` [probe] #${probeCount} @${elapsed}s: probe success, grace expired → released`); + return true; + } log( ` [probe] #${probeCount} @${elapsed}s: probe success, lockSeen=false → wait for lock (still in startup window)`, ); diff --git a/src/handlers/server-requests.ts b/src/handlers/server-requests.ts index 4f3768b..3122d12 100644 --- a/src/handlers/server-requests.ts +++ b/src/handlers/server-requests.ts @@ -377,17 +377,27 @@ async function handleOne( } if (dedupKey) pending.set(dedupKey, { zcodeIds: [zcodeReqId] }); + // Settle-once: whatever happens during the forward, zcode must get exactly + // one reply and the dedup entry must resolve. An unanswered request makes + // the backend reannounce forever, and every reannounce refreshes the turn + // loop's no-progress timer — the 120s timeout never fires and the turn + // hangs. Any throw degrades to decline instead of propagating. let zcodeResp: ZcodeInteractionResponse; - if (ask) { - zcodeResp = await handleAskUserQuestion( - server, - cx, - acpSid, - params as ZcodeInteractionUserInputParams, - turn, - ); - } else { - zcodeResp = await handleSinglePermission(server, cx, acpSid, params, epm, perm, turn); + try { + if (ask) { + zcodeResp = await handleAskUserQuestion( + server, + cx, + acpSid, + params as ZcodeInteractionUserInputParams, + turn, + ); + } else { + zcodeResp = await handleSinglePermission(server, cx, acpSid, params, epm, perm, turn); + } + } catch (e) { + warn(` ⚠ interaction forward threw, declining: ${e instanceof Error ? e.message : String(e)}`); + zcodeResp = { action: "decline", reason: "bridge error during forward" }; } // Reply to the first zcode id + all reannounced ones, and cache for late reannounces. diff --git a/src/handlers/session.ts b/src/handlers/session.ts index 96d90e3..49b91d2 100644 --- a/src/handlers/session.ts +++ b/src/handlers/session.ts @@ -470,7 +470,7 @@ export async function prompt( server: ZcodeAcpServer, params: acp.PromptRequest, cx: acp.AgentContext, - requestId: number, + requestId: number | string, ): Promise { const backend = server.ensureBackend(); @@ -829,8 +829,9 @@ export async function cancel( // Cancel ALL matching turns for this session (not just the first). While a // prior turn is still finalising, pendingTurns holds both it and any newer // prompt waiting on the backend's prompt lock; breaking on the first match - // could leave the live one running. The stopSent guard dedupes the backend - // stop call across turns and repeated cancels. + // could leave the live one running. Each turn guards its own stopSent, so + // multiple matching turns may each fire session/stop once — the backend + // treats stop as idempotent, so the duplicate is harmless. for (const [, turn] of server.pendingTurns) { if (turn.zcodeSid === zcodeSid) { turn.cancelled = true; @@ -944,14 +945,14 @@ function withPreemptLock( export function preemptInFlightTurn( server: ZcodeAcpServer, zcodeSid: string, - selfRequestId: number, + selfRequestId: number | string, ): boolean { // Cancel ALL matching turns (mirrors cancel()): pendingTurns can hold more // than one entry for this session — e.g. an already-cancelled turn still // finalising plus the live one. Breaking on the first match could hit the // stale entry and leave the live turn running, so the new prompt's send - // would retry against a busy backend for 30s and fail. The stopSent guard - // dedupes the backend stop call across turns. + // would retry against a busy backend for 30s and fail. Each turn guards its + // own stopSent; duplicate stops are idempotent on the backend. let found = false; for (const [reqId, turn] of server.pendingTurns) { if (turn.zcodeSid !== zcodeSid || reqId === selfRequestId) continue; @@ -1201,7 +1202,15 @@ async function runEventTurn( // Drain + handle server→client requests (interaction/*). Refreshes the // no-progress timer when any are handled. Pass `turn` so interaction // requests become turn-cancel aware (user stop aborts pending popups). - if (await handleServerRequests(server, backend, cx, acpSid, turn)) { + // Best-effort containment: a throw here would kill the turn loop (and the + // prompt response with it); warn and keep draining instead. + let handled = false; + try { + handled = await handleServerRequests(server, backend, cx, acpSid, turn); + } catch (e) { + warn(`handleServerRequests threw: ${e instanceof Error ? e.message : String(e)}`); + } + if (handled) { lastProgress = Date.now(); } diff --git a/src/index.ts b/src/index.ts index 9d161e3..a0a5fa9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -157,7 +157,14 @@ async function main(): Promise { // starts — the prompting client renders it locally, the others only // ever see the agent's output. echoUserPromptToOthers(server, ctx.client, ctx.params); - return prompt(server, ctx.params, server.clients.broadcast(), ctx.requestId as number); + // JSON-RPC requests always carry a non-null id; the SDK types it as the + // wider JsonRpcId, hence the narrowing cast. + return prompt( + server, + ctx.params, + server.clients.broadcast(), + ctx.requestId as number | string, + ); }) .onRequest("session/set_config_option", (ctx) => setConfigOptionHandler(server, ctx.params, server.clients.broadcast()), diff --git a/src/remote/hub-server.ts b/src/remote/hub-server.ts index a1e0a6d..5ec0fed 100644 --- a/src/remote/hub-server.ts +++ b/src/remote/hub-server.ts @@ -97,6 +97,9 @@ function setCors(res: ServerResponse): void { res.setHeader("Access-Control-Allow-Origin", "*"); res.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS"); res.setHeader("Access-Control-Allow-Headers", "Authorization, Content-Type"); + // Custom response headers JS may read cross-origin; without this the file + // viewer's line-window fetches cannot see X-Zcode-First-Line at all. + res.setHeader("Access-Control-Expose-Headers", "X-Zcode-First-Line"); } async function readJson(req: IncomingMessage): Promise | null> { @@ -246,7 +249,19 @@ export function startHub(options: HubOptions & { onIdleExit?: () => void }): Pro // Dial the bridge's loopback endpoint before accepting the client side, // so a dead bridge fails the upgrade instead of half-opening a pipe. const bridge = new WebSocket(`ws://127.0.0.1:${entry.port}/acp`); + // The client socket can die while we dial; without this guard + // handleUpgrade would run against a dead socket. + let clientGone = false; + const onClientGone = () => { + clientGone = true; + bridge.terminate(); + }; + socket.once("close", onClientGone); + socket.once("error", onClientGone); bridge.once("open", () => { + if (clientGone) return; // terminated above; "open" can no longer fire + socket.removeListener("close", onClientGone); + socket.removeListener("error", onClientGone); wss.handleUpgrade(req, socket, head, (client) => startProxy(client, bridge)); }); bridge.once("error", (e) => { diff --git a/src/server.ts b/src/server.ts index f6cb5b7..a62e112 100644 --- a/src/server.ts +++ b/src/server.ts @@ -89,8 +89,12 @@ export class ZcodeAcpServer { * are deleted on first use. */ readonly sessionCwds = new Map(); - /** Currently running turns, keyed by the ACP request id. */ - readonly pendingTurns = new Map(); + /** + * Currently running turns, keyed by the ACP request id (JSON-RPC ids may be + * numbers or strings; set/delete always use the same value, so the wider + * key type is only for honesty). + */ + readonly pendingTurns = new Map(); /** * Per-session (zcodeSid) preempt lock: a promise chain that serializes the * "register self + preempt others" critical section in prompt(). Prevents diff --git a/tests/settle-once.test.ts b/tests/settle-once.test.ts new file mode 100644 index 0000000..66abc9d --- /dev/null +++ b/tests/settle-once.test.ts @@ -0,0 +1,105 @@ +/** + * Tests for interaction forward settle-once. + * + * Bug: handleOne registered the reannounce dedup entry BEFORE forwarding to + * the client, with no try/catch around the forward. Any throw (adapter on + * malformed params, a rejecting notification send) skipped + * sendInteractionReply: the zcode request was never answered, the dedup entry + * leaked (its 30s cleanup timer is only armed inside sendInteractionReply), + * and the backend's ~1s reannounces kept refreshing the turn loop's + * no-progress timer — the 120s timeout never fired and the turn hung. + * + * The fix: the forward degrades to a decline reply instead of propagating, so + * zcode always gets exactly one answer and the entry always resolves. + */ + +import type * as acp from "@agentclientprotocol/sdk"; +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import type { ServerRequest, ZcodeBackend } from "../src/backend/client.js"; +import type { ZcodeAcpServer } from "../src/server.js"; + +vi.mock("../src/handlers/io.js", () => ({ + sendSessionUpdate: vi.fn().mockResolvedValue(undefined), +})); + +import { handleServerRequests } from "../src/handlers/server-requests.js"; +import { sendSessionUpdate } from "../src/handlers/io.js"; + +/** Minimal server stub (only nextId/resolveSid are touched on this path). */ +function makeServer(): ZcodeAcpServer { + return { + nextId: () => 1, + resolveSid: () => undefined, + } as unknown as ZcodeAcpServer; +} + +/** Fake backend draining a mutable request queue, recording replies. */ +function makeBackend(queue: ServerRequest[]) { + return { + pollServerRequests: () => queue.splice(0, queue.length), + requeueServerRequests: (reqs: ServerRequest[]) => queue.unshift(...reqs), + sendReply: vi.fn(), + sendError: vi.fn(), + } as unknown as ZcodeBackend; +} + +function permissionRequest(zcodeId: number): ServerRequest { + return { + id: zcodeId, + method: "interaction/requestPermission", + params: { + requestId: "r1", + sessionId: "zs1", + toolCallId: "tc1", + toolName: "Bash", + input: { command: "ls" }, + options: [{ optionId: "allow", kind: "allow_once", name: "Allow" }], + }, + }; +} + +describe("interaction forward settle-once", () => { + beforeEach(() => { + vi.mocked(sendSessionUpdate).mockReset(); + vi.mocked(sendSessionUpdate).mockResolvedValue(undefined); + }); + + it("a throwing forward still replies decline and resolves the dedup entry", async () => { + const server = makeServer(); + const queue = [permissionRequest(101)]; + const backend = makeBackend(queue); + vi.mocked(sendSessionUpdate).mockRejectedValue(new Error("client gone")); + const cx = { request: vi.fn() } as unknown as acp.AgentContext; + + const handled = await handleServerRequests(server, backend, cx, "s1"); + expect(handled).toBe(true); + expect(backend.sendReply).toHaveBeenCalledWith(101, { + action: "decline", + reason: "bridge error during forward", + }); + + // The entry resolved: a reannounce of the same key gets the CACHED + // decline immediately (no second forward, no leak). + queue.push(permissionRequest(102)); + await handleServerRequests(server, backend, cx, "s1"); + expect(backend.sendReply).toHaveBeenCalledWith(102, { + action: "decline", + reason: "bridge error during forward", + }); + expect(cx.request).not.toHaveBeenCalled(); + }); + + it("normal path unaffected: allow answer reaches the backend", async () => { + const server = makeServer(); + const queue = [permissionRequest(201)]; + const backend = makeBackend(queue); + const request = vi.fn().mockResolvedValue({ + outcome: { outcome: "selected", optionId: "allow_once" }, + }); + const cx = { request } as unknown as acp.AgentContext; + + await handleServerRequests(server, backend, cx, "s1"); + expect(backend.sendReply).toHaveBeenCalledWith(201, { decision: "allow" }); + }); +}); diff --git a/tests/wait-turn-idle.test.ts b/tests/wait-turn-idle.test.ts new file mode 100644 index 0000000..7609ac3 --- /dev/null +++ b/tests/wait-turn-idle.test.ts @@ -0,0 +1,74 @@ +/** + * Tests for waitForTurnIdle's lock-watching grace. + * + * expectLock exists to skip the "internal turn not started yet" window right + * after session/compact — a probe that succeeds immediately is a false + * "released". Previously, if the lock was NEVER observed (a turn that + * finished between two probes, or a backend whose lock error message drifted + * away from "prompt is running"), the loop spun the full 300s timeout and + * reported a false failure. The grace bounds the lock-watching phase: past + * graceMs, a successful probe (or a non-lock error) counts as released. + */ + +import { describe, expect, it } from "vitest"; + +import type { ZcodeAcpServer } from "../src/server.js"; +import { waitForTurnIdle } from "../src/handlers/extensions.js"; + +/** Probe response shape: { error: { message } } or {} for success. */ +type Probe = { error?: { message: string } }; + +function makeServer(probes: Probe[]): ZcodeAcpServer { + let i = 0; + const backend = { + request: () => Promise.resolve(probes[i++] ?? ({} as Probe)), + }; + return { + nextId: (() => { + let n = 0; + return () => ++n; + })(), + ensureBackend: () => backend, + } as unknown as ZcodeAcpServer; +} + +const LOCK_HELD = { error: { message: "session goal: prompt is running" } }; + +describe("waitForTurnIdle lock-watching grace", () => { + it("classic path unchanged: lock observed once, later probe success releases", async () => { + const server = makeServer([LOCK_HELD, {}]); + const released = await waitForTurnIdle(server, "zs1", 10_000, "session/goal", true); + expect(released).toBe(true); + }); + + it("non-lock error after lock seen releases (existing behaviour)", async () => { + const server = makeServer([LOCK_HELD, { error: { message: "something else" } }]); + const released = await waitForTurnIdle(server, "zs1", 10_000, "session/goal", true); + expect(released).toBe(true); + }); + + it("lock never seen + grace expired → probe success counts as released", async () => { + // Every probe succeeds (turn already done between probes, or too fast to + // observe). Real timers: probe #1 inside the 50ms grace waits 500ms; + // probe #2 is past the grace and releases. + const server = makeServer([{}, {}, {}]); + const released = await waitForTurnIdle(server, "zs1", 10_000, "session/goal", true, 50); + expect(released).toBe(true); + }); + + it("lock never seen + drifted error message + grace expired → released", async () => { + // Backend drift: the lock error no longer says "prompt is running". + const drifted = { error: { message: "turn in progress (renamed)" } }; + const server = makeServer([drifted, drifted, drifted]); + const released = await waitForTurnIdle(server, "zs1", 10_000, "session/goal", true, 50); + expect(released).toBe(true); + }); + + it("lock never seen + grace not expired + timeout hit → false (no false success)", async () => { + // Grace longer than the timeout: still waiting for the lock when the + // timeout expires → false, preserving the expectLock guarantee. + const server = makeServer([{}, {}, {}]); + const released = await waitForTurnIdle(server, "zs1", 50, "session/goal", true, 10_000); + expect(released).toBe(false); + }); +});