fix(rpc): publish close after registry removal and join watcher shutdown

packages/coding-agent/CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Fixed
 
+- RPC `close_session` acknowledgements and `session_closed` events, including worker-failure terminals, are published only after the session registry has removed the entry, so an immediate `list_sessions` never returns the closed session. Filesystem watchers are cancelled atomically with shutdown, every disposer is joined before process exit, reentrant RPC shutdown shares that join and keeps a failure exit code, and nonpersistent RPC probes do not start watchers ([#1656](https://github.com/code-yeongyu/senpi/issues/1656)).
+
 ### Removed
 
 ## [2026.9.15] - 2026-09-15

packages/coding-agent/src/core/extensions/builtin/config-reload/changes.md

@@ -1,5 +1,25 @@
 # config-reload Extension Changes
 
+## 2026-09-14 - Join watcher disposal and skip nonpersistent RPC probes (#1656)
+
+### What changed
+
+- Watch-worker registration checks a shared cancellation flag before and after `fs.watch`, so a shutdown that wins the post-load/pre-registration interleaving never retains a native watcher.
+- `ConfigReloadWatchEngine.close()` cancels synchronously and joins returned disposers; repeated close shares that join and surfaces `AggregateError` if any disposer fails.
+- `session_shutdown` awaits those joins. Nonpersistent RPC sessions (`getSessionFile() === undefined`) do not start OS watches.
+
+### Why
+
+- Fire-and-forget unsubscribe during exit left FSEvents streams running into process teardown (`pthread_join` hang). Snapshot-only RPC probes never needed live watches.
+
+### Why an extension could not handle it
+
+- The event source and watch engine are internal to this builtin; process shutdown must observe their disposal.
+
+### Expected merge conflict zones
+
+- MEDIUM: `watch-event-source.ts` worker source and unsubscribe join; `watch-engine.ts` `close()`; `index.ts` `session_shutdown` / `rebuildWatchers`.
+
 ## 2026-09-11 - Keep per-source changelog acknowledgements routine (senpi#1583)
 
 ### What changed

packages/coding-agent/src/core/extensions/builtin/config-reload/index.ts

@@ -173,6 +173,7 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt
 	let activeTargets: ActiveTarget[] = [];
 	let currentContext: ExtensionContext | undefined;
 	let started = false;
+	const watcherClosures: Array<Promise<PromiseSettledResult<void>[]>> = [];
 	let reloadInFlight = false;
 	let deferredNoticeShown = false;
 	let unavailableReloadLogged = false;
@@ -181,12 +182,10 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt
 	const vetoDeferral = new ReloadVetoDeferral();
 	let changeChain: Promise<void> = Promise.resolve();
 
-	// The engine goes inert the moment close() is called; its unsubscribe loop can
-	// take seconds per watcher, and session_shutdown is awaited by the reload flow.
+	// Cancel registrations synchronously; session_shutdown joins every disposer.
 	const closeWatchers = (): void => {
-		engine?.close().catch((error: unknown) => {
-			logger.error("watcher_error", { path: "watcher teardown", message: errorMessage(error) });
-		});
+		if (!engine) return;
+		watcherClosures.push(Promise.allSettled([engine.close()]));
 		engine = undefined;
 		activeTargets = [];
 	};
@@ -261,7 +260,8 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt
 	};
 
 	const processChange = async (change: RealChange): Promise<void> => {
-		if (reloadInFlight || !currentContext) return;
+		if (reloadInFlight || !currentContext || !started) return;
+		const changeContext = currentContext;
 		// Suppression state (self-write consumption, routine-diff base) is per path,
 		// so it must be resolved before grouping: a path watched by several
 		// registrations would otherwise be classified once per group and reach the
@@ -291,6 +291,7 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt
 		);
 		for (const [registrationId, paths] of groups) {
 			const errors = await validateChangedPaths(registrationId, paths, registrations, agentDir, currentContext.cwd);
+			if (!started || currentContext !== changeContext) return;
 			if (errors.length > 0) {
 				rejectChange(currentContext, registrationId, paths, errors, logger, pi);
 				continue;
@@ -319,11 +320,18 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt
 	};
 
 	const rebuildWatchers = (ctx: ExtensionContext): void => {
+		if (!started || currentContext !== ctx) return;
 		closeWatchers();
 		clearCompactionRecheck();
 		const settingsManager = SettingsManager.create(ctx.cwd, agentDir, { projectTrusted: ctx.isProjectTrusted() });
 		const settings = resolveConfigReloadSettings(settingsManager);
-		if (!settings.enabled || ctx.mode === "print" || ctx.mode === "json") {
+		// Nonpersistent RPC probes need a configuration snapshot, not live OS watches.
+		if (
+			!settings.enabled ||
+			ctx.mode === "print" ||
+			ctx.mode === "json" ||
+			(ctx.mode === "rpc" && ctx.sessionManager.getSessionFile() === undefined)
+		) {
 			pi.events.emit(CONFIG_WATCH_READY, { enabled: false });
 			return;
 		}
@@ -463,18 +471,20 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt
 	});
 
 	pi.on("agent_end", async (_event, ctx) => {
+		if (!started) return;
 		currentContext = ctx;
 		await flushPending();
 	});
 	pi.on("agent_settled", async (_event, ctx) => {
+		if (!started) return;
 		currentContext = ctx;
 		await flushPending();
 	});
 	pi.on("project_trust", () => {
 		if (currentContext) rebuildWatchers(currentContext);
 		return { trusted: "undecided" };
 	});
-	pi.on("session_shutdown", (event) => {
+	pi.on("session_shutdown", async (event) => {
 		const closingContext = currentContext;
 		started = false;
 		currentContext = undefined;
@@ -485,6 +495,9 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt
 		cleanupEventListeners();
 		pending.clear();
 		if (event.reason !== "reload" && closingContext) reloadHandoffs.delete(handoffKey(closingContext));
+		const results = (await Promise.all(watcherClosures.splice(0))).flat();
+		const errors = results.filter((result) => result.status === "rejected").map((result) => result.reason);
+		if (errors.length > 0) throw new AggregateError(errors, "Config watcher shutdown failed");
 	});
 
 	function canRequestReload(ctx: ExtensionContext): boolean {

packages/coding-agent/src/core/extensions/builtin/config-reload/watch-engine.ts

@@ -77,7 +77,8 @@ const DEFAULT_CLOCK: WatchClock = {
  */
 export class ConfigReloadWatchEngine {
 	readonly #states: TargetState[];
-	readonly #unsubscribes: (() => void)[] = [];
+	readonly #unsubscribes: Array<() => void> = [];
+	#closeCompletion: Promise<void> | undefined;
 	readonly #subscribe: WatchEventSource;
 	readonly #onRealChange: (change: RealChange) => void;
 	readonly #onError: (error: unknown, path: string) => void;
@@ -168,34 +169,32 @@ export class ConfigReloadWatchEngine {
 	}
 
 	/**
-	 * Marks the engine inert synchronously, then drains the unsubscribe loop off
-	 * the caller's stack. A single `fs.watch` unsubscribe can block for seconds on
-	 * a loaded machine, and a reload awaits this call; every dispatch path already
-	 * checks `#closed`, so the still-attached subscriptions are silent while the
-	 * returned promise settles. Await it only to observe teardown completion.
+	 * Marks the engine inert and cancels every subscription synchronously, then
+	 * joins whatever native disposal those unsubscribers return. Repeated close()
+	 * callers share that join. Dispatch already checks `#closed`, so stale events
+	 * cannot start reload work while disposal is outstanding.
 	 */
 	close(): Promise<void> {
 		if (this.#closed) {
-			return Promise.resolve();
+			return this.#closeCompletion ?? Promise.resolve();
 		}
 		this.#closed = true;
 		if (this.#timer) {
 			this.#clock.clearTimeout(this.#timer);
 			this.#timer = undefined;
 		}
 		const unsubscribes = this.#unsubscribes.splice(0);
-		return new Promise<void>((settle) => {
-			this.#clock.setTimeout(() => {
-				for (const unsubscribe of unsubscribes) {
-					try {
-						unsubscribe();
-					} catch (error) {
-						this.#reportError(error, "watch subscription");
-					}
+		this.#closeCompletion = Promise.allSettled(unsubscribes.map(async (unsubscribe) => unsubscribe())).then(
+			(results) => {
+				const errors = results
+					.filter((result): result is PromiseRejectedResult => result.status === "rejected")
+					.map((result) => result.reason);
+				if (errors.length > 0) {
+					throw new AggregateError(errors, "Config watcher teardown failed");
 				}
-				settle();
-			}, 0);
-		});
+			},
+		);
+		return this.#closeCompletion;
 	}
 
 	getBaselineSnapshot(): ReadonlyMap<string, string> {

packages/coding-agent/src/core/extensions/builtin/config-reload/watch-event-source.ts

@@ -27,18 +27,24 @@ const { parentPort } = require("node:worker_threads");
 if (!parentPort) throw new Error("Recursive watch worker requires a parent port");
 
 const watchers = new Map();
+const cancelled = new Set();
+const isCancelled = (message) =>
+	cancelled.has(message.id) || (message.active !== undefined && Atomics.load(message.active, 0) === 0);
 parentPort.on("message", (message) => {
 	if (message.kind === "unwatch") {
+		cancelled.add(message.id);
 		watchers.get(message.id)?.close();
 		watchers.delete(message.id);
 		return;
 	}
 	if (message.kind !== "watch") return;
+	if (isCancelled(message)) return;
 	try {
 		const watcher = watch(
 			message.path,
 			{ recursive: message.recursive !== false, encoding: "utf8" },
 			(eventType, filename) => {
+				if (isCancelled(message)) return;
 				parentPort.postMessage({
 					kind: "event",
 					id: message.id,
@@ -47,6 +53,10 @@ parentPort.on("message", (message) => {
 				});
 			},
 		);
+		if (isCancelled(message)) {
+			watcher.close();
+			return;
+		}
 		watcher.on("error", (error) => {
 			parentPort.postMessage({
 				kind: "error",
@@ -100,7 +110,12 @@ export function createFsWatchEventSource(
 ): WatchEventSource {
 	const recursiveSubscriptions = new Map<
 		number,
-		{ readonly path: string; readonly listener: WatchEventListener; readonly recursive: boolean }
+		{
+			readonly path: string;
+			readonly listener: WatchEventListener;
+			readonly recursive: boolean;
+			readonly active: Int32Array;
+		}
 	>();
 	let recursiveWorker: RecursiveWatchWorker | undefined;
 	let nextSubscriptionId = 1;
@@ -127,7 +142,13 @@ export function createFsWatchEventSource(
 			if (recursiveSubscriptions.size === 0) return;
 			const replacement = ensureRecursiveWorker();
 			for (const [id, subscription] of recursiveSubscriptions) {
-				replacement.postMessage({ kind: "watch", id, path: subscription.path, recursive: subscription.recursive });
+				replacement.postMessage({
+					kind: "watch",
+					id,
+					path: subscription.path,
+					recursive: subscription.recursive,
+					active: subscription.active,
+				});
 			}
 		});
 		recursiveWorker = worker;
@@ -138,19 +159,28 @@ export function createFsWatchEventSource(
 		if (WORKER_OFFLOADED_WATCH_PLATFORMS.has(options.platform ?? process.platform)) {
 			const id = nextSubscriptionId++;
 			const recursive = watchOptions?.recursive ?? false;
-			ensureRecursiveWorker().postMessage({ kind: "watch", id, path, recursive });
-			recursiveSubscriptions.set(id, { path, listener, recursive });
+			const active = new Int32Array(new SharedArrayBuffer(4));
+			Atomics.store(active, 0, 1);
+			ensureRecursiveWorker().postMessage({ kind: "watch", id, path, recursive, active });
+			recursiveSubscriptions.set(id, { path, listener, recursive, active });
+			let closing: Promise<void> | undefined;
 			return () => {
-				if (!recursiveSubscriptions.delete(id)) return;
+				if (!recursiveSubscriptions.delete(id)) return closing;
+				Atomics.store(active, 0, 0);
 				// Resolve at unsubscribe time: the worker may have been replaced after a crash.
 				const worker = recursiveWorker;
-				if (!worker) return;
-				if (recursiveSubscriptions.size > 0) {
-					worker.postMessage({ kind: "unwatch", id });
-					return;
-				}
+				if (!worker) return closing;
+				worker.postMessage({ kind: "unwatch", id });
+				if (recursiveSubscriptions.size > 0) return;
 				recursiveWorker = undefined;
-				void worker.terminate().catch((error: unknown) => onError(error, path));
+				closing = worker.terminate().then(
+					() => undefined,
+					(error: unknown) => {
+						onError(error, path);
+						throw error;
+					},
+				);
+				return closing;
 			};
 		}
 

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

@@ -1,5 +1,26 @@
 # changes
 
+## 2026-09-14 - Publish RPC close only after registry removal (#1656)
+
+### What changed
+
+- `closeMarked()` still replies on the close-grace deadline and keeps the entry until native exit, so a worker stuck in a syscall cannot hang cancel/close. The router waits for that exit callback before emitting `session_closed` or the close acknowledgement.
+- `session-worker-client.ts` defers worker-failure terminal records until after the same exit callback, so error and failure frames observe an empty registry too.
+- `shutdown.ts` makes reentrant `shutdown()` join the in-flight disposer and preserve a non-zero exit code (serializer-error overlapping stdin EOF).
+
+### Why
+
+- An immediate `list_sessions` after close must never return the closed session, including when the worker fails instead of a clean `close_session`.
+- A second shutdown caller must not `process.exit` while watcher disposal is still outstanding.
+
+### Why an extension could not handle it
+
+- Session registry ownership and process exit are host lifecycle, outside session extensions.
+
+### Expected merge conflict zones
+
+- LOW: `closeMarked()` in `worker-session-registry.ts`, `fail()` in `session-worker-client.ts`, and the stdio `shutdown()` wrapper in `rpc-mode.ts`. Does not touch `host-lifecycle.ts` / `host-ensure.ts`.
+
 ## 2026-09-14 - Keep bundled workers out of supervisor entry dispatch
 
 ### What changed

packages/coding-agent/src/modes/rpc/rpc-mode.ts

@@ -82,6 +82,7 @@ import { toJsonEvent } from "../json-event.ts";
 import { createRpcConnectionHandler, type RpcConnectionSink } from "./connection-handler.ts";
 import { parseClientCapabilities } from "./custom-capability.ts";
 import { attachJsonlLineReader, MAX_RPC_LINE_CHARACTERS, serializeJsonLine } from "./jsonl.ts";
+import { createRpcShutdown } from "./shutdown.ts";
 
 // Re-export types for consumers
 export type {
@@ -121,7 +122,6 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 	const capabilities = parseClientCapabilities(envValue("RPC_CLIENT_CAPABILITIES"));
 	const handler = createRpcConnectionHandler(runtimeHost, sink, { capabilities });
 
-	let shuttingDown = false;
 	const signalCleanupHandlers: Array<() => void> = [];
 
 	const registerSignalHandlers = (): void => {
@@ -144,22 +144,20 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise<neve
 
 	let detachInput = () => {};
 
-	async function shutdown(exitCode = 0, signal?: NodeJS.Signals): Promise<never> {
-		if (shuttingDown) {
-			process.exit(exitCode);
-		}
-		shuttingDown = true;
-		for (const cleanup of signalCleanupHandlers) {
-			cleanup();
-		}
-		await handler.dispose();
-		detachInput();
-		process.stdin.pause();
-		if (signal !== "SIGTERM") {
-			await flushRawStdout();
-		}
-		process.exit(exitCode);
-	}
+	const shutdown = createRpcShutdown(
+		async (signal) => {
+			for (const cleanup of signalCleanupHandlers) {
+				cleanup();
+			}
+			await handler.dispose();
+			detachInput();
+			process.stdin.pause();
+			if (signal !== "SIGTERM") {
+				await flushRawStdout();
+			}
+		},
+		(exitCode) => process.exit(exitCode),
+	);
 
 	const handleInputLine = async (line: string): Promise<void> => {
 		await handler.handleInputLine(line);

packages/coding-agent/src/modes/rpc/session-command-router.ts

@@ -594,6 +594,9 @@ export class SessionCommandRouter {
 			} catch (cause) {
 				process.stderr.write(`senpi rpc close for session ${sessionId} failed: ${String(cause)}\n`);
 			}
+			// closeMarked may return at the grace deadline while ownership remains;
+			// terminal records must still observe the exit callback's registry removal.
+			await this.registry.peek(sessionId)?.closeCompletion;
 			terminal?.();
 		} finally {
 			finalization.resolve();