next-turn: report a failed hub listener and session-state write instead of swallowing them

sdk/examples/plugins/next-turn/src/hub/link.test.ts

@@ -119,6 +119,26 @@ describe("HubLink", () => {
 		expect(statuses).toEqual(["connecting", "up"]);
 	});
 
+	test("a listener that throws is reported and does not stop the others", async () => {
+		const client = fakeClient();
+		const logged: string[] = [];
+		const link = new HubLink({
+			...deps("ws://hub", [client]),
+			log: (message) => logged.push(message),
+		});
+		const seen: string[] = [];
+		link.subscribe(new Set(["run.completed"]), () => {
+			throw new Error("a listener that cannot handle it");
+		});
+		link.subscribe(new Set(["run.completed"]), (event) =>
+			seen.push(event.event),
+		);
+		await link.start();
+		client.emit({ event: "run.completed", sessionId: "s1" });
+		expect(seen).toEqual(["run.completed"]);
+		expect(logged).toEqual(["a hub listener failed"]);
+	});
+
 	test("a throwing listener does not starve the others", async () => {
 		const client = fakeClient();
 		const link = new HubLink(deps("ws://hub", [client]));

sdk/examples/plugins/next-turn/src/hub/link.ts

@@ -48,6 +48,8 @@ export type HubLinkDeps = {
 		scope: HubLinkScope,
 	): HubClientPort;
 	sleep(ms: number): Promise<void>;
+	/** A subscriber that threw: it is isolated, not silent. */
+	log?(message: string, error?: unknown): void;
 };
 
 /**
@@ -256,7 +258,9 @@ export class HubLink {
 			if (!names.has(event.event)) continue;
 			try {
 				listener(event);
-			} catch {}
+			} catch (error) {
+				this.deps.log?.("a hub listener failed", error);
+			}
 		}
 	}
 

sdk/examples/plugins/next-turn/src/runtime.ts

@@ -150,7 +150,12 @@ export function startSessionRuntime(
 	const log = pluginLog(context);
 	const pluginActive = () => options.pluginActive?.() ?? true;
 	const run = options.runCommand ?? runCommand;
-	const link = new HubLink(options.hubLink ?? nodeHubLinkDeps(), {
+	const link = new HubLink(
+		{
+			...(options.hubLink ?? nodeHubLinkDeps()),
+			log: (message, error) => log(message, error),
+		},
+		{
 		workspaceRoot,
 		cwd: workspaceRoot,
 	});
@@ -167,7 +172,12 @@ export function startSessionRuntime(
 	});
 	engine.start().catch((error: unknown) => log("the hub link stopped", error));
 	void pruneSessionStates(paths.sessions);
-	const store = new FileSessionStateStore(paths.sessions, sessionId);
+	const store = new FileSessionStateStore(
+		paths.sessions,
+		sessionId,
+		Date.now,
+		(message, error) => log(message, error),
+	);
 	const saved = store.read();
 	/** The last viewer's surface, kept past its detach so labels still name the tray it used. */
 	let rememberedSurface: PluginUiSurface | undefined =

sdk/examples/plugins/next-turn/src/store/session-state.test.ts

@@ -19,6 +19,22 @@ import {
 const tempDir = useTempDirs("next-turn-session-state-");
 
 describe("session state persistence across sandbox respawn", () => {
+	test("a write that cannot land on disk is reported and still served from memory", () => {
+		const dir = tempDir();
+		const occupied = join(dir, "sessions");
+		writeFileSync(occupied, "not a directory");
+		const logged: string[] = [];
+		const store = new FileSessionStateStore(
+			occupied,
+			"sess_1",
+			() => 1,
+			(message) => logged.push(message),
+		);
+		store.write({ seq: 3 });
+		expect(store.read()?.seq).toBe(3);
+		expect(logged).toEqual(["the session state was not written"]);
+	});
+
 	test("rehydrates the active offerId, labels the next message once, and skips through the stored pool without a planner call", async () => {
 		const dir = tempDir();
 		const labelsPath = join(dir, "labels.jsonl");

sdk/examples/plugins/next-turn/src/store/session-state.ts

@@ -82,6 +82,8 @@ export class FileSessionStateStore implements SessionStateStore {
 		dir: string,
 		private readonly sessionId: string,
 		private readonly now: () => number = Date.now,
+		/** A write that could not land: the file keeps the previous state, not the caller's. */
+		private readonly log?: (message: string, error?: unknown) => void,
 	) {
 		this.filePath = join(dir, `${sha256Hex(sessionId)}.json`);
 	}
@@ -111,7 +113,9 @@ export class FileSessionStateStore implements SessionStateStore {
 		this.cached = JSON.parse(text) as PersistedSessionState;
 		try {
 			writeFileAtomic(this.filePath, `${text}\n`);
-		} catch {}
+		} catch (error) {
+			this.log?.("the session state was not written", error);
+		}
 	}
 
 	private clear(): void {