fix(rpc): enforce durable_session_id on the worker session registry and write a chosen id to disk at open

.agents/skills/senpi-qa/scripts/scenarios/durable-session-id-qa.mjs

@@ -0,0 +1,79 @@
+#!/usr/bin/env node
+// C1 real-surface proof for senpi #1951: drive the REAL `senpi --mode rpc --multi-session`
+// process over its stdio JSONL protocol and check the durable-session-id contract end to end.
+import { existsSync, readFileSync, writeFileSync } from "node:fs";
+import { join } from "node:path";
+import { createChecks, installCleanupHooks, makeSandbox, repoRoot } from "../lib/common.mjs";
+import { hermeticEnv } from "../lib/mock-loop-support.mjs";
+import { TargetRpcClient } from "../lib/target-rpc-client.mjs";
+
+const CHOSEN = "0199f0d4-1c3a-7bb1-9d2e-0a1b2c3d4e5f";
+const ON_DISK = "019ffe53-4359-7618-a61b-bc85786c94ef";
+
+const headerIdOf = (path) => JSON.parse(readFileSync(path, "utf8").split("\n")[0]).id;
+
+async function main() {
+  installCleanupHooks();
+  const checks = createChecks("durable session id (C1/C2/C3 real surface)");
+  const box = makeSandbox("durable-id-qa");
+  const env = hermeticEnv(box.env);
+  const client = new TargetRpcClient({ env, cwd: box.cwd, targetRoot: repoRoot(), extraArgs: ["--multi-session"] });
+  const out = [];
+  const record = (label, value) => { const line = `${label}: ${JSON.stringify(value)}`; out.push(line); process.stdout.write(line + "\n"); };
+
+  // C3: the host advertises the capability.
+  const info = await client.send({ type: "get_protocol_info" });
+  record("get_protocol_info.capabilities", info.data?.capabilities);
+  checks.ok("advertises durable_session_id", (info.data?.capabilities ?? []).includes("durable_session_id"));
+
+  // C1: create under a chosen id; the reply names it.
+  const createdPath = join(box.sessionDir, "created.jsonl");
+  const created = await client.send({ type: "open_session", cwd: box.cwd, sessionPath: createdPath, durableSessionId: CHOSEN });
+  record("open_session(create, durableSessionId)", { success: created.success, error: created.error, durable: created.data?.state?.sessionId });
+  checks.ok("create reply state.sessionId === chosen id", created.data?.state?.sessionId === CHOSEN, String(created.data?.state?.sessionId));
+  const createdHandle = created.data?.sessionId;
+
+  // C2: a second LIVE session under the same id is refused.
+  const dup = await client.send({ type: "open_session", cwd: box.cwd, sessionPath: join(box.sessionDir, "dup.jsonl"), durableSessionId: CHOSEN });
+  record("open_session(duplicate live id)", { success: dup.success, error: dup.error });
+  checks.ok("duplicate live id -> session_id_in_use", dup.success === false && String(dup.error).includes("session_id_in_use"), String(dup.error));
+
+  // C2: a malformed id is refused at the boundary.
+  const bad = await client.send({ type: "open_session", cwd: box.cwd, sessionPath: join(box.sessionDir, "bad.jsonl"), durableSessionId: "no spaces allowed" });
+  record("open_session(malformed id)", { success: bad.success, error: bad.error });
+  checks.ok("malformed id -> invalid_session_id", bad.success === false && String(bad.error).includes("invalid_session_id"), String(bad.error));
+
+  // C2: an EXISTING file keeps its header id even when a different id is offered.
+  const existingPath = join(box.sessionDir, "existing.jsonl");
+  writeFileSync(existingPath, JSON.stringify({ type: "session", version: 3, id: ON_DISK, timestamp: new Date().toISOString(), cwd: box.cwd }) + "\n");
+  const resumed = await client.send({ type: "open_session", cwd: box.cwd, sessionPath: existingPath, durableSessionId: CHOSEN.replace("0a1b", "ffff") });
+  record("open_session(existing file, different id offered)", { success: resumed.success, durable: resumed.data?.state?.sessionId });
+  checks.ok("resume keeps the header id", resumed.data?.state?.sessionId === ON_DISK, String(resumed.data?.state?.sessionId));
+  checks.ok("resume did not rewrite the header on disk", headerIdOf(existingPath) === ON_DISK, headerIdOf(existingPath));
+
+  // C1 on disk: close the created session, reopen the same path with NO id, and the host
+  // must read the chosen id back from the file header it wrote.
+  if (createdHandle) await client.send({ type: "close_session", sessionId: createdHandle });
+  const reopened = await client.send({ type: "open_session", cwd: box.cwd, sessionPath: createdPath });
+  record("open_session(reopen created path, no id)", { success: reopened.success, durable: reopened.data?.state?.sessionId, fileExists: existsSync(createdPath) });
+  checks.ok("reopen reads the chosen id back", reopened.data?.state?.sessionId === CHOSEN, String(reopened.data?.state?.sessionId));
+  if (existsSync(createdPath)) {
+    record("created.jsonl header id", headerIdOf(createdPath));
+    checks.ok("chosen id is in the JSONL header on disk", headerIdOf(createdPath) === CHOSEN, headerIdOf(createdPath));
+  } else {
+    record("created.jsonl", "not materialized (no assistant message yet) - identity proven via reopen consistency");
+  }
+
+  // C2: once the holder is closed the id is free again.
+  if (reopened.data?.sessionId) await client.send({ type: "close_session", sessionId: reopened.data.sessionId });
+  const reused = await client.send({ type: "open_session", cwd: box.cwd, sessionPath: join(box.sessionDir, "reused.jsonl"), durableSessionId: CHOSEN });
+  record("open_session(reuse id after close)", { success: reused.success, durable: reused.data?.state?.sessionId });
+  checks.ok("id is reusable once its holder closed", reused.data?.state?.sessionId === CHOSEN, String(reused.data?.state?.sessionId));
+
+  client.close();
+  const outPath = process.argv[2] ?? "/tmp/senpi-1951-c1-proof.txt";
+  writeFileSync(outPath, out.join("\n") + "\n");
+  process.stdout.write(`evidence: ${outPath}\n`);
+  checks.finish();
+}
+main().catch((error) => { console.error(error); process.exit(1); });

packages/coding-agent/CHANGELOG.md

@@ -10,6 +10,7 @@
 
 - Reusing a transcript container after teardown re-arms progressive hydration instead of leaving it permanently halted, so `clear()` and `detachAll()` reset the hydration halt alongside the watermark rather than only the watermark. Teardown still halts hydration; only reuse re-arms it. Latent today, since the chat container is constructed once and never disposed ([#2002](https://github.com/code-yeongyu/senpi/pull/2002)).
 - Daemon status reads no longer spawn a `ps` process per request. `senpi host status` and the generations rows now read the process table through the kernel (`sysctl` on macOS, `/proc` on Linux), so a client polling daemon status costs zero child processes and nothing can accumulate as unreaped `<defunct>` children on a runtime whose `execFile` does not reap them - the leak class that filled the macOS process table from a long-lived RPC host ([code-yeongyu/omo-desktop-app#594](https://github.com/code-yeongyu/omo-desktop-app/issues/594), [#1507](https://github.com/code-yeongyu/senpi/issues/1507)). On runtimes without the kernel bindings (plain Node) the memory, descriptor and zombie fields report `null` instead of spawning a probe.
+- RPC `open_session.durableSessionId` is now enforced on the worker-backed multi-session host too: a duplicate live id is refused with `session_id_in_use` and a malformed id with `invalid_session_id` at the registry boundary. A session created under a caller-chosen id writes its header at open, so reopening that path before the first assistant message keeps the chosen identity. ([#2010](https://github.com/code-yeongyu/senpi/issues/2010))
 
 ### Added
 

packages/coding-agent/src/core/changes.md

@@ -40,6 +40,24 @@ CLI parsing and the interactive login command are core surfaces that run before
 ### Expected merge conflict zones
 
 
+
+## 2026-09-22 - A caller-chosen session id is written to disk at open (#2010)
+
+### What changed
+
+- `packages/coding-agent/src/core/session-manager.ts`: in `_setSessionFile`'s missing-file branch, when `NewSessionOptions.id` is present the header is written immediately (`_rewriteFile()`, `flushed = true`) instead of waiting for the first assistant message.
+
+### Why
+
+- Deferring the file until an assistant message exists is the right default for a HOST-minted id: nothing references it yet. It is the wrong default for a CALLER-chosen id, which the caller already holds in its own records. Without this, a create under a chosen id followed by a close before the first reply left no file, and a reopen of that path minted a different identity - the id the caller stored pointed at nothing. Measured on the real host before the fix: reopen returned a fresh uuidv7 and `existsSync(path)` was false.
+
+### Why an extension could not handle it
+
+- Session identity and its first persistence happen inside `SessionManager` before any extension is bound to the session.
+
+### Expected merge conflict zones
+
+- The `else` branch of `_setSessionFile` that handles a not-yet-existing path.
 ## 2026-09-22 - settings.json provider-key migration (senpi#1989)
 
 ### What changed

packages/coding-agent/src/core/session-manager.ts

@@ -1017,6 +1017,14 @@ export class SessionManager {
 			// id is the session's identity from here on. An EXISTING file never reaches this
 			// branch, which is why a supplied id can never overwrite a header id.
 			this._resetToNewSession(newSessionOptions);
+			// A host-minted id is referenced by nothing yet, so its file may wait for the first
+			// assistant message. A CALLER-chosen id is already held in the caller's own records:
+			// the file has to answer to it now, or a reopen before the first reply would mint a
+			// different identity and the caller's record would point at nothing (#2010).
+			if (newSessionOptions?.id !== undefined) {
+				this._rewriteFile();
+				this.flushed = true;
+			}
 		}
 	}
 

packages/coding-agent/src/modes/rpc/changes.md

@@ -20,6 +20,26 @@ The read happens inside the package's daemon-control surface (`readHostStatus`,
 
 - `packages/coding-agent/src/modes/rpc/host-process-metrics.ts`, against any change to the metrics fields or their null semantics.
 
+## 2026-09-22 - durable_session_id on the worker registry, and a chosen id that is on disk at once (#2010)
+
+### What changed
+
+- `packages/coding-agent/src/modes/rpc/worker-session-registry.ts`: `openSession` validates `durableSessionId` with `assertValidSessionId` (refusing with `invalid_session_id`) and refuses an id any non-closed entry already holds with `session_id_in_use`, both synchronously before its first await; the requested id is recorded on the entry before that await so a concurrent open sees it, and `snapshot.state.sessionId` still overwrites it after commit. Re-opening the same path stays an attach.
+- `.agents/skills/senpi-qa/scripts/scenarios/durable-session-id-qa.mjs` (new): drives the REAL `senpi --mode rpc --multi-session` process over stdio and asserts the whole contract - capability advertised, create under a chosen id, duplicate live id refused, malformed id refused, resume keeps its header id, reopen of the created path reads the chosen id back from disk, id reusable once its holder closed.
+
+### Why
+
+- #1956 added the collision guard to `RpcSessionRegistry` only. `multi-session-host.ts` instantiates `WorkerSessionRegistry` whenever a worker configuration is present - the real host's normal shape - and that registry forwarded the profile straight to `worker.prepare`, so on the real host two live sessions could share one durable id. The unit suite drove `RpcSessionRegistry` directly and stayed green while the host bypassed the guard; the real-surface scenario is what caught it.
+- The format refusal did fire on the real host, but as `open_failed: invalid_session_id` from the worker's own `SessionManager` after the worker had already died. Validating at the registry boundary gives the caller the stable code before any worker is spawned.
+
+### Why an extension could not handle it
+
+- Both checks need the registry's view of every live session and run before any runtime or extension exists for the new session.
+
+### Expected merge conflict zones
+
+- `worker-session-registry.ts`: the prologue of `openSession` (after the `invalid_path` check) and the entry literal.
+
 ## 2026-09-22 - chatgpt-subscription provider id on the RPC surface (senpi#1989)
 
 ### What changed

packages/coding-agent/src/modes/rpc/worker-session-registry.ts

@@ -1,5 +1,6 @@
 import { isAbsolute } from "node:path";
 import { ProviderScope } from "@earendil-works/pi-ai/node/provider-scope";
+import { assertValidSessionId } from "../../core/session-manager.ts";
 import type { CliRuntimeConfiguration } from "../../main.ts";
 import {
 	type LiveWorkerPaths,
@@ -60,6 +61,27 @@ export class WorkerSessionRegistry {
 	async openSession(profile: RpcSessionLaunchProfile, options?: RpcSessionOpenOptions): Promise<OpenRpcSession> {
 		if (!isAbsolute(profile.cwd) || (profile.sessionPath !== undefined && !isAbsolute(profile.sessionPath)))
 			throw new RpcSessionRegistryError("invalid_path");
+		// Same contract as RpcSessionRegistry.openSession (#1951), on the registry the multi-session
+		// host actually instantiates (#2010). Both checks run SYNCHRONOUSLY before the first await:
+		// the format check so a bad id is refused with its own code instead of surfacing as the
+		// worker's death (`session_closing`), and the collision scan so a concurrent open naming the
+		// same durable id finds this one already recorded. Re-opening the SAME path is an attach,
+		// not a collision: the id is the file's own.
+		const requestedDurableId = profile.durableSessionId;
+		if (requestedDurableId !== undefined) {
+			try {
+				assertValidSessionId(requestedDurableId);
+			} catch (cause) {
+				throw new RpcSessionRegistryError("invalid_session_id", String(cause));
+			}
+			const requestedKey = profile.sessionPath ? this.knownReservationKey(profile.sessionPath) : undefined;
+			for (const entry of this.entries.values()) {
+				if (entry.state === "closed") continue;
+				if (entry.durableSessionId !== requestedDurableId) continue;
+				if (requestedKey !== undefined && entry.reservationKey === requestedKey) continue;
+				throw new RpcSessionRegistryError("session_id_in_use");
+			}
+		}
 		if (profile.sessionPath) {
 			const key = this.knownReservationKey(profile.sessionPath);
 			const owner = key ? this.reservations.owner(key) : undefined;
@@ -78,6 +100,10 @@ export class WorkerSessionRegistry {
 			retainOnDisconnect: options?.retainOnDisconnect === true,
 			lastCommandAt: this.now(),
 			lifecycleMutex: Promise.resolve(),
+			// Recorded before the first await so the collision scan above sees an open still being
+			// built; `snapshot.state.sessionId` overwrites it with the authoritative value after
+			// commit, which on a resume is the header's id rather than the requested one.
+			...(requestedDurableId !== undefined ? { durableSessionId: requestedDurableId } : {}),
 		};
 		let workerFailure: string | undefined;
 		const worker = this.createWorker({

packages/coding-agent/test/suite/rpc-worker-durable-session-id.test.ts

@@ -0,0 +1,94 @@
+import { existsSync, readFileSync } from "node:fs";
+import { join } from "node:path";
+import { expect, it } from "vitest";
+import { parseArgs } from "../../src/cli/args.ts";
+import { WorkerSessionRegistry } from "../../src/modes/rpc/worker-session-registry.ts";
+import { startWorkerHost } from "./rpc-worker-host-support.ts";
+
+/**
+ * #2010: the durable-session-id contract on the REAL worker-backed registry, which is what the
+ * multi-session host instantiates. #1956 guarded only RpcSessionRegistry.
+ */
+
+const CHOSEN = "0199f0d4-1c3a-7bb1-9d2e-0a1b2c3d4e5f";
+
+function makeRegistry(host: Awaited<ReturnType<typeof startWorkerHost>>) {
+	return new WorkerSessionRegistry({
+		configuration: {
+			parsed: parseArgs(["--mode", "rpc", "--no-extensions", "--no-skills", "--no-context-files"]),
+			cwd: host.cwd,
+			agentDir: join(host.scratch, "agent"),
+			appMode: "rpc",
+		},
+		closeGraceMs: 1000,
+		now: Date.now,
+	});
+}
+
+it("worker registry refuses a durable id a LIVE session already holds", async () => {
+	const host = await startWorkerHost();
+	const registry = makeRegistry(host);
+	const handles: string[] = [];
+	try {
+		const first = await registry.openSession({
+			cwd: host.cwd,
+			sessionPath: join(host.scratch, "one.jsonl"),
+			durableSessionId: CHOSEN,
+		});
+		handles.push(first.sessionId);
+		expect(first.durableSessionId).toBe(CHOSEN);
+
+		await expect(
+			registry.openSession({
+				cwd: host.cwd,
+				sessionPath: join(host.scratch, "two.jsonl"),
+				durableSessionId: CHOSEN,
+			}),
+		).rejects.toMatchObject({ code: "session_id_in_use" });
+	} finally {
+		for (const handle of handles) await registry.close(handle).catch(() => undefined);
+		await host.dispose();
+	}
+}, 60_000);
+
+it("worker registry refuses a malformed durable id at its own boundary", async () => {
+	const host = await startWorkerHost();
+	const registry = makeRegistry(host);
+	try {
+		await expect(
+			registry.openSession({
+				cwd: host.cwd,
+				sessionPath: join(host.scratch, "bad.jsonl"),
+				durableSessionId: "no spaces",
+			}),
+		).rejects.toMatchObject({ code: "invalid_session_id" });
+		expect(registry.list()).toHaveLength(0);
+	} finally {
+		await host.dispose();
+	}
+}, 60_000);
+
+it("a caller-chosen id is on disk immediately, so a reopen with no id reads it back", async () => {
+	const host = await startWorkerHost();
+	const registry = makeRegistry(host);
+	const sessionPath = join(host.scratch, "chosen.jsonl");
+	let handle: string | undefined;
+	try {
+		const created = await registry.openSession({ cwd: host.cwd, sessionPath, durableSessionId: CHOSEN });
+		handle = created.sessionId;
+		expect(created.durableSessionId).toBe(CHOSEN);
+		// No assistant message was ever appended. The caller already holds this id in its own
+		// records, so the file must answer to it now, not after the first reply.
+		expect(existsSync(sessionPath)).toBe(true);
+		expect(JSON.parse(readFileSync(sessionPath, "utf8").split("\n")[0] ?? "{}").id).toBe(CHOSEN);
+
+		await registry.close(handle);
+		handle = undefined;
+		const reopened = await registry.openSession({ cwd: host.cwd, sessionPath });
+		handle = reopened.sessionId;
+		expect(reopened.durableSessionId).toBe(CHOSEN);
+	} finally {
+		if (handle) await registry.close(handle).catch(() => undefined);
+		await host.dispose();
+	}
+}, 60_000);