From 4824795e2852a2c7e71d820c68f75224c137bccd Mon Sep 17 00:00:00 2001 From: Alex Lavaee Date: Wed, 21 Oct 2026 13:12:11 -0700 Subject: [PATCH] Say why a run retries and keep its connection when the run ends A proxy answering for a Chopin that went away sends 502 with Content-Length 0. Bun reads that empty body as JSON null, so re-registering the machine threw a TypeError reading `error` of null instead of refusing with the status. The TypeError happened to be retried, but carries no status or code, so every "Still trying to reach Chopin" line of a run said "no answer". Registration now treats a body that is not a JSON object as empty and refuses with the HTTP status, and a failure's kind names a refusal's status ("HTTP 502") and looks for a status, server code or network code in anything the error wraps. A heartbeat still in flight when a run finished was cancelled by the run, and the MCP client rejects that as a request timeout, which counted as a failure: the terminal said the connection was lost and Chopin out of reach right after "Finished", and the healthy shared connection was dropped. A call its own caller gave up is no longer a failure: it keeps the connection and says nothing. A call that ran out of time still counts. Assistant-model: Claude Opus 5.5 (fast) Assistant-workflow: inline (fast subagents) Assistant-verification: bun test (connector suite and full suite) passed Assistant-verification: bun run types and bun run ci passed Co-authored-by: Alex Lavaee --- apps/connector/src/connection.test.ts | 19 ++- apps/connector/src/connection.ts | 25 +++- apps/connector/src/machine.ts | 4 +- apps/connector/src/reconnect.ts | 10 ++ apps/connector/src/run.test.ts | 182 ++++++++++++++++++++++++++ docs/local-experiments.md | 4 +- 6 files changed, 233 insertions(+), 11 deletions(-) diff --git a/apps/connector/src/connection.test.ts b/apps/connector/src/connection.test.ts index 9f01e4d15..0d889cf3c 100644 --- a/apps/connector/src/connection.test.ts +++ b/apps/connector/src/connection.test.ts @@ -5,9 +5,10 @@ import type { Server as NetServer } from "node:net"; import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js"; -import { failureCause, redacted, serveWork } from "./connection"; +import { StreamableHTTPError } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; +import { failureCause, failureKind, redacted, serveWork } from "./connection"; import { SELF } from "./machine"; -import { remote } from "./mcp"; +import { ConnectorError, remote } from "./mcp"; import { createOutput } from "./output"; type Tool = (args: Record) => unknown | Promise; @@ -585,6 +586,20 @@ test("a connection failure's cause never shows credentials from the error", () = expect(text).not.toContain("abcdef0123456789xyz"); }); +test("a failure's kind is its HTTP status, server code or network code, however it is wrapped", () => { + let kinds = [ + new ConnectorError("http-502", "Chopin refused the request (http-502).", 502), + new ConnectorError("unavailable", "Chopin refused the request (unavailable).", 503), + new StreamableHTTPError(502, "Error POSTing to endpoint: Bad Gateway"), + Object.assign(new TypeError("Unable to connect."), { code: "ConnectionRefused" }), + ].map(error => failureKind(new Error("Re-registering failed", { cause: error }))); + expect(kinds).toEqual(["HTTP 502", "HTTP 503 unavailable", "HTTP 502", "ConnectionRefused"]); + // Never the message, whatever it holds. + expect(failureKind(new Error("Bad gateway for Bearer chopin_r_secretsecretsecret"))).toBe( + "no answer", + ); +}); + test("machine, Planner, pairing key and build run tokens are masked wherever chopin prints text", () => { for (let prefix of ["chopin_m_", "chopin_p_", "chopin_k_", "chopin_r_"]) { let token = `${prefix}${"Zx9_-".repeat(9).slice(0, 43)}`; diff --git a/apps/connector/src/connection.ts b/apps/connector/src/connection.ts index 470ecb7ea..4379dab10 100644 --- a/apps/connector/src/connection.ts +++ b/apps/connector/src/connection.ts @@ -7,6 +7,7 @@ import type { remote } from "./mcp"; import type { Output } from "./output"; import { backoff, + cancelled, CONNECT_TIMEOUT_MS, delay, FORGET_MS, @@ -104,17 +105,26 @@ export function failureCause(error: unknown) { /** * A failure's kind alone, without any of its message, so nothing it quotes can reach the - * terminal: the HTTP status and the server's error code, else the network error's code. + * terminal: the HTTP status and the server's error code, else the network error's code, from the + * error or anything it wraps as its `cause`. */ export function failureKind(error: unknown) { let code = errorCode(error); let current = error; for (let depth = 0; current instanceof Error && depth < 5; depth++) { - let status = (current as { code?: unknown }).code; - if (current instanceof StreamableHTTPError && typeof status === "number" && status >= 100) { - return code ? `HTTP ${status} ${code}` : `HTTP ${status}`; + code ??= errorCode(current); + let own = (current as { code?: unknown }).code; + // The MCP client's HTTP error carries its status as `code`; a refused registration as `status`. + let status = current instanceof StreamableHTTPError + ? own + : current instanceof ConnectorError + ? current.status + : undefined; + if (typeof status === "number" && status >= 100) { + // A refusal without a code of its own is named `http-`, which the status says. + return code && code !== `http-${status}` ? `HTTP ${status} ${code}` : `HTTP ${status}`; } - if (!code && typeof status === "string" && /^[A-Za-z_]+$/.test(status)) return status; + if (!code && typeof own === "string" && /^[A-Za-z_]+$/.test(own)) return own; if (!code && current.name === "TimeoutError") return "timed out"; current = current.cause; } @@ -215,7 +225,8 @@ export async function serveWork(options: { }; // A call that fails transiently drops its connection, so a run's next report connects anew, // as the token of a connection the server forgot is refused for good. A call Ctrl-C cut short - // keeps it, so the stop report and the disconnect that gives back claimed work still go out. + // keeps it, so the stop report and the disconnect that gives back claimed work still go out, + // and so does one its caller gave up, as a run ending gives up its heartbeat. let link: Link = { async call(name, args, callSignal) { let api = current ?? await abandoned( @@ -228,7 +239,7 @@ export async function serveWork(options: { try { return await api.call(name, args, callSignal); } catch (error) { - if (!signal.aborted && transient(classify(error))) { + if (!signal.aborted && !cancelled(callSignal) && transient(classify(error))) { if (running && current === api) output.runConnectionLost(failureCause(error)); drop(api); } diff --git a/apps/connector/src/machine.ts b/apps/connector/src/machine.ts index 0fd312831..4d6490a6c 100644 --- a/apps/connector/src/machine.ts +++ b/apps/connector/src/machine.ts @@ -188,7 +188,9 @@ async function post( }); let refused = await refusal(response); if (refused) throw refused; - let value = await response.json().catch(() => ({})) as Record; + // Bun reads an empty body sent with Content-Length 0, as a gateway's 502 is, as JSON null. + let parsed: unknown = await response.json().catch(() => undefined); + let value = (parsed && typeof parsed === "object" ? parsed : {}) as Record; if (!response.ok) { let code = typeof value.error === "string" ? value.error : `http-${response.status}`; throw new ConnectorError( diff --git a/apps/connector/src/reconnect.ts b/apps/connector/src/reconnect.ts index b1a901c64..3693e3f37 100644 --- a/apps/connector/src/reconnect.ts +++ b/apps/connector/src/reconnect.ts @@ -20,6 +20,15 @@ export function transient(kind: ErrorClass) { return kind === "retry" || kind === "forgotten"; } +/** + * Whether a call's own caller gave it up, as a run ending abandons its heartbeat: that says + * nothing about Chopin. A call that ran out of time did not hear from Chopin, so it is no such call. + */ +export function cancelled(signal?: AbortSignal) { + return !!signal?.aborted + && (signal.reason as { name?: unknown } | undefined)?.name !== "TimeoutError"; +} + /** The message for a connection Chopin forgot: connecting again registers it anew. */ export function forgottenMessage() { return "Chopin no longer recognizes this connection (it may have restarted). " + ( @@ -136,6 +145,7 @@ export function runReports(api: Caller, options: { } return value; } catch (error) { + if (cancelled(signal)) throw error; let kind = classify(error); if (transient(kind) && lostAt === undefined && !gone.signal.aborted) { lostAt = now(); diff --git a/apps/connector/src/run.test.ts b/apps/connector/src/run.test.ts index a83c57239..280360af0 100644 --- a/apps/connector/src/run.test.ts +++ b/apps/connector/src/run.test.ts @@ -20,6 +20,7 @@ import { checkouts } from "./checkouts"; import { runWork, STOPPED } from "./run"; import { git } from "./workspace"; import { ConnectorError, remote } from "./mcp"; +import { register } from "./machine"; import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { WebStandardStreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/webStandardStreamableHttp.js"; import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js"; @@ -669,6 +670,187 @@ test("a run out of reach says it is still trying at most every 10 seconds, then expect(terminal.out() + terminal.err()).not.toContain(token); }); +/** + * A build served as `chopin connect` serves it, each connection registering this machine and + * then opening MCP, against a fake Chopin whose tools may take as long as they like. `gateway` + * answers a request instead of Chopin when it returns a response, as a proxy in front of a Chopin + * that went away does. Serving ends when `abort` aborts; work polls after the first offer go to + * `tool` and then wait. + */ +async function connected(options: { + tool: (name: string, args: Record) => unknown; + gateway?: (request: Request, name?: string) => Response | undefined; + flags: string[]; + /** More of the claim, such as `live` for a living document's first build. */ + claim?: Record; + abort: AbortController; + runOptions?: Partial[3]>; +}) { + let root = await mkdtemp(join(tmpdir(), "chopin-connected-")); + let state = await mkdtemp(join(tmpdir(), "chopin-connected-state-")); + let id = crypto.randomUUID(); + let offered = false; + let registrations = 0; + let terminal = captured(); + let server = Bun.serve({ + hostname: "127.0.0.1", + port: 0, + async fetch(request) { + let message = await request.clone().json().catch(() => undefined) as + | { params?: { name?: string } } + | undefined; + let answer = options.gateway?.(request, message?.params?.name); + if (answer) return answer; + if (new URL(request.url).pathname === "/api/connector/connections") { + return Response.json({ token: `connection-${++registrations}` }); + } + let mcp = new Server({ name: "fake-chopin", version: "1" }, { capabilities: { tools: {} } }); + mcp.setRequestHandler(ListToolsRequestSchema, () => ({ tools: [] })); + mcp.setRequestHandler(CallToolRequestSchema, async request => { + let name = request.params.name; + let args = request.params.arguments ?? {}; + // The work poll offers the build once, then hears of any later poll and waits. + let value: unknown = name === "wait_for_work" + ? offered + ? (await options.tool(name, args), await new Promise(() => {})) + : (offered = true, { id, kind: "implementation" }) + : name === "claim_implementation_build" + ? { + build: { + id, + repositoryId: "R_repo", + checkout: { repository: "org/repo", commit: git(root, "rev-parse", "HEAD") }, + }, + documentId: "document", + runToken: "run-token", + ...options.claim, + } + : await options.tool(name, args); + return { content: [{ type: "text", text: JSON.stringify(value ?? {}) }] }; + }); + let transport = new WebStandardStreamableHTTPServerTransport({ + sessionIdGenerator: undefined, + enableJsonResponse: true, + }); + await mcp.connect(transport); + return transport.handleRequest(request); + }, + }); + let url = `http://127.0.0.1:${server.port}`; + try { + git(root, "init"); + git(root, "config", "user.name", "Fixture"); + git(root, "config", "user.email", "fixture@example.test"); + git(root, "remote", "add", "origin", "git@github.com:org/repo.git"); + await writeFile(join(root, "source.txt"), "committed"); + git(root, "add", "."); + git(root, "commit", "-m", "fixture"); + let signal = options.abort.signal; + await serveWork({ + connect: async signal => { + let registration = await register( + url, + "machine-token", + { label: "box", checkouts: [] }, + signal, + ); + return remote(url, registration.token, signal); + }, + signal, + output: terminal.output, + random: () => 0.01, + run: (api, kind, claim) => + runWork(api, kind, claim, { + checkouts: checkouts({ directory: state, registered: { "org/repo": root } }), + directory: state, + command: [ + process.execPath, + new URL("./testing/fake-implementer.ts", import.meta.url).pathname, + ...options.flags, + ], + url, + signal, + output: terminal.output, + heartbeatMs: 20, + random: () => 0.01, + ...options.runOptions, + }), + }); + return { registrations, out: terminal.out }; + } finally { + await server.stop(true); + await rm(root, { recursive: true, force: true }); + await rm(state, { recursive: true, force: true }); + } +} + +test("a run behind a gateway answering 502 with no body says HTTP 502 while it retries", async () => { + let abort = new AbortController(); + // The fake clock moves four seconds with each request the gateway refuses. + let clock = 0; + let renewals = 0; + let refused = 0; + let down = false; + let { out } = await connected({ + flags: ["--hang"], + abort, + runOptions: { now: () => clock }, + tool(name) { + if (name !== "renew_implementation_build") return {}; + // Chopin answers the first heartbeat and then goes away; once it is back the run ends. + if (++renewals === 1) down = true; + else abort.abort(); + return { accepted: true }; + }, + // A proxy such as Caddy answers for a Chopin that went away with 502 and Content-Length 0, + // both the heartbeat on the open connection and each registration that follows. + gateway(request, name) { + let registering = new URL(request.url).pathname === "/api/connector/connections"; + if (!down || !registering && name !== "renew_implementation_build") return; + clock += 4000; + if (++refused === 8) down = false; + return new Response(null, { status: 502 }); + }, + }); + let lines = out().split("\n").filter(line => /Chopin/.test(line)); + expect(lines).toEqual([ + " … Lost the connection to Chopin during a run (HTTP 502). Registering this machine again; " + + "the agent keeps working.", + " … Lost contact with Chopin. The agent keeps working; reports wait until it is back.", + " … Still trying to reach Chopin (attempt 3, 12 s, HTTP 502), the run keeps going.", + " … Still trying to reach Chopin (attempt 6, 24 s, HTTP 502), the run keeps going.", + " ✓ Registered this machine with Chopin again; the run goes on.", + " ✓ Back in contact with Chopin after 28 s.", + ]); +}); + +test("a heartbeat in flight when a run finishes is cancelled without losing the connection", async () => { + let abort = new AbortController(); + let heartbeat = Promise.withResolvers(); + let { registrations, out } = await connected({ + flags: ["--live"], + claim: { live: true }, + abort, + tool(name, args) { + // Chopin is slow to answer a heartbeat, which is still in flight as the run finishes. + if (name === "renew_implementation_build") { + heartbeat.resolve(); + return new Promise(() => {}); + } + if (name === "report_implementation_build" && args.state === "stopped") { + return heartbeat.promise.then(() => ({ accepted: true })); + } + // The connector polls for its next work once the run is over. + if (name === "wait_for_work") abort.abort(); + return {}; + }, + }); + let lines = out().split("\n").filter(line => /Chopin|Finished/.test(line)); + expect(lines).toEqual([" ✓ Finished"]); + // The next poll went over the same connection rather than registering this machine again. + expect(registrations).toBe(1); +}); + test("a run out of reach for its whole grace period stops its agent", async () => { let { id, calls } = await run( {}, diff --git a/docs/local-experiments.md b/docs/local-experiments.md index 3fe8fe9b6..9af8c6946 100644 --- a/docs/local-experiments.md +++ b/docs/local-experiments.md @@ -212,7 +212,9 @@ on the way (a long poll cut by a proxy, a timeout, a 5xx answer); the terminal s Chopin again**. A build carries on through that: its next heartbeat comes on the new connection of the same paired machine and owner, and Chopin moves the build, and the agent's run credential, to it. A connection of any other machine, or of -another person, is still refused (`build-forbidden`). +another person, is still refused (`build-forbidden`). A request the connector +gives up itself, as the heartbeat still waiting when a run finishes, has not +failed: the connection stays and the terminal says nothing about it. A build also goes on through a server restart or a deploy. Its heartbeat and reports name the document (connector protocol 6, chopin 0.0.16 and later), so