Restart the server without anyone noticing: ready and live checks, a quiet reconnect, and a reload after a deploy only when it is safe

AGENTS.md

@@ -424,6 +424,16 @@ so merge ranges from all mounted editors before replacing a registry entry.
   after `send()` loses edits on disconnect.
 - Rebuilds and reconnects must replay unacknowledged updates only when the epoch
   is still compatible.
+- Shutdown closes every socket with 1012 (`OpenSockets.stopForRestart()`) before
+  `server.stop(true)`, whose own 1001 reads as a loss. The browser takes 1012 as a
+  quiet restart and a probe's 4xx as a refusal only with `x-chopin-refusal: 1`
+  (or a 401); answer any new refusal of `/ws` or `/ws/sidebar` through
+  `refused()` so a proxy's 4xx cannot be mistaken for it. A request waiting when
+  the 1012 arrives rejects as `WireLost` with `restarting` set; only a caller whose
+  request carries its own identity, as `chat:send` does, may send it again quietly.
+  For `RESTART_QUIET_MS` after a 1012 `Wire` raises no `onOutage`, so the document's one
+  offline notice stays away, and every other surface that speaks of a lost connection
+  waits `useConnectionNotice(lost, wire.restarting)`; a new one must honour both.
 - Room opening and closing share the document lifecycle lock. Opening waits for
   an active close; eviction forgets the room only after runtime shutdown and
   persistence complete. Preserve that ordering.

apps/server/src/client.test.ts

@@ -61,7 +61,7 @@ test("unfingerprinted public files are not cached as immutable", async () => {
 test("HTML and SPA fallbacks are revalidated", async () => {
 	let client = await createClient({ "index.html": "<!doctype html><main>app</main>" });
 	try {
-		for (let path of ["/index.html", "/rooms/room-123", "/assets/missing-AjK9nqi6.js"]) {
+		for (let path of ["/index.html", "/rooms/room-123", "/assets", "/assetsx/app.js"]) {
 			let response = await client.get(path);
 			expect(response.status).toBe(200);
 			expect(response.headers.get("cache-control")).toBe("no-cache");
@@ -73,6 +73,35 @@ test("HTML and SPA fallbacks are revalidated", async () => {
 	}
 });
 
+test("a missing file under /assets answers 404, where the app's HTML would fail as a script", async () => {
+	let client = await createClient({
+		"index.html": "<!doctype html><main>app</main>",
+		"assets/index-AjK9nqi6.js": "console.log('build');",
+	});
+	try {
+		for (
+			let path of [
+				"/assets/old-chunk-Zy8wXq1p.js",
+				"/assets/old-style-Zy8wXq1p.css",
+				"/assets/unhashed.js",
+				"/assets/nested/dir/chunk-Zy8wXq1p.js",
+				"/assets/",
+			]
+		) {
+			let response = await client.get(path);
+			expect([path, response.status]).toEqual([path, 404]);
+			expect(response.headers.get("content-type")).not.toContain("text/html");
+			expect(response.headers.get("cache-control")).toBe("no-store");
+			expect(await response.text()).not.toContain("<main>app</main>");
+		}
+		let present = await client.get("/assets/index-AjK9nqi6.js");
+		expect(present.status).toBe(200);
+		expect(present.headers.get("cache-control")).toBe("public, max-age=31536000, immutable");
+	} finally {
+		await client.close();
+	}
+});
+
 test("a missing built client keeps its existing 404 response", async () => {
 	let directory = await mkdtemp(join(tmpdir(), "chopin-client-"));
 	try {

apps/server/src/client.ts

@@ -32,6 +32,7 @@ const PRIVATE = new Set(["authorization", "cookie"]);
 
 /** Vite's default build names use an eight-character hash before the extension. */
 const FINGERPRINTED_ASSET_PATH = /^\/assets\/.+-[A-Za-z0-9_-]{8}\.[^/]+$/;
+const ASSETS_PATH = "/assets/";
 const IMMUTABLE_CACHE_CONTROL = "public, max-age=31536000, immutable";
 
 function forwardable(headers: Headers): Headers {
@@ -76,7 +77,10 @@ export async function proxy(req: Request, url: URL, origin: string): Promise<Res
  * Serve the built client.
  *
  * Anything that is not a real file is the single-page app, so a room URL
- * survives a reload and a deep link opens the room it names.
+ * survives a reload and a deep link opens the room it names. Under `/assets/` it
+ * is a 404 instead: a page left open across a deploy asks for the chunks of the
+ * build it came from, and HTML answering for a script is an error that names
+ * nothing, where a 404 tells the page its build is gone.
  */
 export async function serve(url: URL, dir: string): Promise<Response> {
 	if (!existsSync(dir)) {
@@ -96,6 +100,12 @@ export async function serve(url: URL, dir: string): Promise<Response> {
 		}
 		return new Response(file, { headers });
 	}
+	if (url.pathname.startsWith(ASSETS_PATH)) {
+		return new Response("not found", {
+			status: 404,
+			headers: { "cache-control": "no-store", "content-type": "text/plain; charset=utf-8" },
+		});
+	}
 	return new Response(Bun.file(join(dir, "index.html")), {
 		headers: { "cache-control": "no-cache" },
 	});

apps/server/src/http/health.test.ts

@@ -0,0 +1,80 @@
+import { expect, test } from "bun:test";
+
+import { Readiness, registerHealthRoutes } from "./health";
+import { Router } from "./router";
+
+function service(holdsLease = () => true) {
+	let readiness = new Readiness(holdsLease);
+	let router = new Router();
+	registerHealthRoutes(router, readiness);
+	let get = async (path: string, method = "GET") => {
+		let response = (await router.handle(new Request(`http://chopin.test${path}`, { method })))!;
+		return { status: response.status, body: await response.text(), response };
+	};
+	return {
+		readiness,
+		live: () => get("/api/health/live"),
+		ready: () => get("/api/health/ready"),
+		get,
+	};
+}
+
+test("readiness is false until startup recovery has finished", async () => {
+	let { readiness, ready } = service();
+	expect(await ready()).toMatchObject({ status: 503, body: "starting" });
+	readiness.started();
+	expect(await ready()).toMatchObject({ status: 200, body: "ready" });
+});
+
+test("readiness needs the writer lease as well as finished recovery", async () => {
+	let held = false;
+	let { readiness, ready } = service(() => held);
+	readiness.started();
+	expect(await ready()).toMatchObject({ status: 503, body: "starting" });
+	held = true;
+	expect(await ready()).toMatchObject({ status: 200, body: "ready" });
+});
+
+test("readiness is false from the moment a drain starts, and stays false", async () => {
+	let { readiness, ready } = service();
+	readiness.started();
+	expect((await ready()).status).toBe(200);
+	readiness.draining();
+	expect(await ready()).toMatchObject({ status: 503, body: "draining" });
+	readiness.started();
+	expect(await ready()).toMatchObject({ status: 503, body: "draining" });
+});
+
+test("a drain that begins during startup keeps the process from ever being ready", async () => {
+	let { readiness, ready } = service();
+	readiness.draining();
+	readiness.started();
+	expect(await ready()).toMatchObject({ status: 503, body: "draining" });
+});
+
+test("liveness is true while the process runs, whatever readiness says", async () => {
+	let { readiness, live } = service(() => false);
+	expect(await live()).toMatchObject({ status: 200, body: "live" });
+	readiness.started();
+	readiness.draining();
+	expect(await live()).toMatchObject({ status: 200, body: "live" });
+});
+
+test("both answer one word to a caller with no credentials, and are never cached", async () => {
+	let { readiness, live, ready } = service();
+	readiness.started();
+	for (let { status, body, response } of [await live(), await ready()]) {
+		expect(status).toBe(200);
+		expect(body).toMatch(/^[a-z]+$/);
+		expect(response.headers.get("cache-control")).toBe("no-store");
+		expect(response.headers.get("set-cookie")).toBeNull();
+	}
+});
+
+test("a probe may ask with HEAD and gets the status without a body", async () => {
+	let { readiness, get } = service();
+	expect(await get("/api/health/ready", "HEAD")).toMatchObject({ status: 503, body: "" });
+	readiness.started();
+	expect(await get("/api/health/ready", "HEAD")).toMatchObject({ status: 200, body: "" });
+	expect(await get("/api/health/live", "HEAD")).toMatchObject({ status: 200, body: "" });
+});

apps/server/src/http/health.ts

@@ -0,0 +1,57 @@
+import type { Router } from "./router";
+
+export type ReadinessStatus = "starting" | "ready" | "draining";
+
+/**
+ * Whether this process may be sent traffic.
+ *
+ * Ready only once startup recovery has finished and the process holds the writer lease, and
+ * never again once a drain has begun: a replacement waits for exactly this process to let go.
+ */
+export class Readiness {
+	readonly #holdsLease: () => boolean;
+	#started = false;
+	#draining = false;
+
+	constructor(holdsLease: () => boolean) {
+		this.#holdsLease = holdsLease;
+	}
+
+	/** Startup recovery has finished. */
+	started(): void {
+		this.#started = true;
+	}
+
+	/** A drain has begun. Irreversible, because the process is on its way out. */
+	draining(): void {
+		this.#draining = true;
+	}
+
+	status(): ReadinessStatus {
+		if (this.#draining) return "draining";
+		return this.#started && this.#holdsLease() ? "ready" : "starting";
+	}
+}
+
+function answer(request: Request, status: number, word: string): Response {
+	return new Response(request.method === "HEAD" ? null : word, {
+		status,
+		headers: { "cache-control": "no-store", "content-type": "text/plain; charset=utf-8" },
+	});
+}
+
+/**
+ * `GET /api/health/live` answers while the process runs, and `GET /api/health/ready` says
+ * whether it may be sent traffic. Both are unauthenticated and say one word, nothing else.
+ */
+export function registerHealthRoutes(router: Router, readiness: Readiness): void {
+	let live = (request: Request) => answer(request, 200, "live");
+	let ready = (request: Request) => {
+		let status = readiness.status();
+		return answer(request, status === "ready" ? 200 : 503, status);
+	};
+	for (let method of ["GET", "HEAD"]) {
+		router.on(method, "/api/health/live", live);
+		router.on(method, "/api/health/ready", ready);
+	}
+}

apps/server/src/main.ts

@@ -65,6 +65,7 @@ import { describe, load } from "./config";
 import { GitHubClient, GitHubError } from "./github/client";
 import { GitHubReferences } from "./github/references";
 import { registerConnectorScript } from "./http/connector-script";
+import { Readiness, registerHealthRoutes } from "./http/health";
 import { registerMachinesPath } from "./http/machines-path";
 import { Router } from "./http/router";
 import { JobExecutionError, JobRegistry } from "./jobs/registry";
@@ -111,7 +112,7 @@ import {
 } from "./research/service";
 import { placeResearchReference as placeResearch } from "./research/placement";
 import * as Rooms from "./rooms";
-import { admit, admitSidebar } from "./socket/admission";
+import { admit, admitSidebar, refused } from "./socket/admission";
 import { refreshAuthorization } from "./socket/authorization";
 import {
 	recheckDecisionWatch,
@@ -122,6 +123,7 @@ import {
 	watchDecisions,
 	watchRequest,
 } from "./socket/decision-watch";
+import { OpenSockets } from "./socket/open-sockets";
 import { SessionSockets } from "./socket/session-sockets";
 import { StorageError } from "./storage/errors";
 import { createStorage } from "./storage/registry";
@@ -138,6 +140,7 @@ import {
 } from "./wire";
 
 import { AGENT_PATH, LINK_CLOSE } from "@chopin/agent-link";
+import { SERVICE_RESTART } from "@chopin/protocol/restart";
 
 import type { Server, ServerWebSocket } from "bun";
 import type { AgentSocketData } from "./harness/remote/endpoint";
@@ -192,6 +195,7 @@ const writerLease = new WriterLease(storage.leases, {
 	safetyMs: LEASE_SAFETY_MS,
 	retryMs: LEASE_RETRY_MS,
 }, () => signal());
+const readiness = new Readiness(() => writerLease.held !== undefined);
 let experiments: ExperimentRuntime | undefined;
 let implementations: ReturnType<typeof registerImplementationRoutes> | undefined;
 let experimentTimer: ReturnType<typeof setInterval> | undefined;
@@ -203,6 +207,7 @@ let plannerAttachments: PlannerAttachments | undefined;
 let systemTurns = new SystemTurns();
 /** Browser sockets by login, so a login that ends closes them before another broadcast. */
 let sessionSockets = new SessionSockets(sessionId => hostedAuth.sessions.isLive(sessionId));
+const openSockets = new OpenSockets();
 let jobRunner: JobRunner | undefined;
 let researchService: ResearchWorkspaceService | undefined;
 let researchRecoveryTimer: ReturnType<typeof setTimeout> | undefined;
@@ -1783,9 +1788,7 @@ function listen(): Server<SocketData> {
 					deleting: channelId => deletingChannels.has(channelId),
 					readOnly: channelId => archivingChannels.has(channelId),
 				});
-				if ("status" in outcome) {
-					return new Response(outcome.reason, { status: outcome.status });
-				}
+				if ("status" in outcome) return refused(outcome);
 				return upgraded(req, self, outcome.data);
 			}
 			if (url.pathname === AGENT_PATH) {
@@ -1800,9 +1803,7 @@ function listen(): Server<SocketData> {
 			}
 			if (url.pathname === "/ws/sidebar") {
 				let outcome = await admitSidebar(req, hostedAuth);
-				if ("status" in outcome) {
-					return new Response(outcome.reason, { status: outcome.status });
-				}
+				if ("status" in outcome) return refused(outcome);
 				return upgraded(req, self, outcome.data);
 			}
 			let routed = await router.handle(req, url);
@@ -1817,6 +1818,7 @@ function listen(): Server<SocketData> {
 
 		websocket: {
 			open(ws: AnySocket) {
+				openSockets.opened(ws);
 				if (isAgentSocket(ws)) {
 					if (!plannerAttachments) return ws.close(LINK_CLOSE.failed, "unavailable");
 					openAgentSocket(ws, hostedAuth, {
@@ -1852,6 +1854,7 @@ function listen(): Server<SocketData> {
 			},
 
 			close(ws: AnySocket) {
+				openSockets.closed(ws);
 				if (isAgentSocket(ws)) {
 					ws.data.alive?.stop();
 					return ws.data.link?.disconnected();
@@ -1882,9 +1885,10 @@ function drain(): Promise<void> {
 				record(err);
 			}
 		};
+		readiness.draining();
 		// Before the sockets close, so no link withdraws its questions on the way out.
 		plannerAttachments?.shutdown();
-		if (server) await attempt(() => server.stop(true));
+		if (server) await attempt(() => openSockets.stopForRestart(server));
 		if (researchRecoveryTimer) clearTimeout(researchRecoveryTimer);
 		if (experimentTimer) clearInterval(experimentTimer);
 		if (recoveringResearch) await attempt(() => recoveringResearch!);
@@ -2142,7 +2146,7 @@ async function deleteChannelLocked(channelId: string): Promise<boolean> {
 			channel = await storage.channels.get(channelId);
 		} catch {
 			for (let ws of room?.members.values() ?? []) {
-				ws.close(1012, "document state is temporarily unavailable");
+				ws.close(SERVICE_RESTART, "document state is temporarily unavailable");
 			}
 			throw err;
 		}
@@ -2153,7 +2157,7 @@ async function deleteChannelLocked(channelId: string): Promise<boolean> {
 					await plan(survivingRoom, server);
 				} catch {
 					for (let ws of survivingRoom.members.values()) {
-						ws.close(1012, "document reload required");
+						ws.close(SERVICE_RESTART, "document reload required");
 					}
 				}
 			}
@@ -2892,6 +2896,7 @@ registerConnectorScript(router, {
 	source: join(import.meta.dir, "../../connector/src/main.ts"),
 });
 registerMachinesPath(router);
+registerHealthRoutes(router, readiness);
 
 try {
 	await storage.health();
@@ -2957,6 +2962,7 @@ try {
 	scheduleResearchRecovery(recovery.deferred);
 	await researchService.recoverTerminalPlannerInline();
 	jobRunner.start();
+	readiness.started();
 } catch (err) {
 	if (listening) await listening.stop(true).catch(() => {});
 	if (researchRecoveryTimer) clearTimeout(researchRecoveryTimer);

apps/server/src/socket/admission.test.ts

@@ -1,11 +1,12 @@
+import { REFUSAL_HEADER } from "@chopin/protocol/restart";
 import { Machines } from "../auth/machines";
 import { describe, expect, it } from "bun:test";
 
 import { Sessions } from "../auth/session";
 import { Admission } from "../auth/admission";
 import { GitHubError } from "../github/client";
 import { MemoryStorage } from "../storage/memory/adapter";
-import { admit } from "./admission";
+import { admit, refused } from "./admission";
 
 import type { HostedAuth } from "../auth/routes";
 import type {
@@ -129,6 +130,13 @@ describe("socket admission", () => {
 			auth,
 		);
 		expect(wrongOrigin).toEqual({ status: 403, reason: "origin is not allowed" });
+		if (!("status" in wrongOrigin)) throw new Error("expected a refusal");
+		let answer = refused(wrongOrigin);
+		expect([answer.status, await answer.text(), answer.headers.get(REFUSAL_HEADER)]).toEqual([
+			403,
+			"origin is not allowed",
+			"1",
+		]);
 		// A page still open on the name the deployment moved from keeps its socket.
 		let previousOrigin = await admit(
 			new Request(url, {