Say why a run retries and keep its connection when the run ends

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<string, unknown>) => unknown | Promise<unknown>;
@@ -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: <html>Bad Gateway</html>"),
+		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)}`;

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-<status>`, 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);
 				}

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<string, unknown>;
+	// 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<string, unknown>;
 	if (!response.ok) {
 		let code = typeof value.error === "string" ? value.error : `http-${response.status}`;
 		throw new ConnectorError(

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();

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<string, unknown>) => 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<string, unknown>;
+	abort: AbortController;
+	runOptions?: Partial<Parameters<typeof runWork>[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<void>();
+	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(
 		{},

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