From 08bffe72215db056527c13b53c73cfd64a528526 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Tue, 15 Sep 2026 18:10:14 +0900 Subject: [PATCH 1/3] fix(rpc): publish close after registry removal and join watcher shutdown Close acknowledgements and session_closed, including worker-failure terminals, wait for the session registry to drop the handle. closeMarked still returns at the close-grace deadline so a worker stuck in a syscall cannot wedge cancel. Watcher registration checks cancellation before and after fs.watch; close() joins disposers; reentrant RPC shutdown shares that join and keeps a failure exit code; nonpersistent RPC probes start no watchers. Plan: .omo/plans/omo-dependency-diet.md --- packages/coding-agent/CHANGELOG.md | 2 + .../builtin/config-reload/changes.md | 20 + .../extensions/builtin/config-reload/index.ts | 29 +- .../builtin/config-reload/watch-engine.ts | 35 +- .../config-reload/watch-event-source.ts | 52 ++- .../coding-agent/src/modes/rpc/changes.md | 21 ++ .../coding-agent/src/modes/rpc/rpc-mode.ts | 32 +- .../src/modes/rpc/session-command-router.ts | 3 + .../src/modes/rpc/session-worker-client.ts | 24 +- .../coding-agent/src/modes/rpc/shutdown.ts | 17 + .../suite/config-reload-extension.test.ts | 42 ++- .../suite/config-reload-lazy-teardown.test.ts | 354 ++++++++---------- .../config-reload-worker-shutdown.test.ts | 100 +++++ ...-recursive-watch-main-thread-stall.test.ts | 28 +- ...-recursive-watch-main-thread-stall.test.ts | 28 +- .../test/suite/rpc-close-backpressure.test.ts | 48 ++- .../test/suite/rpc-close-ordering.test.ts | 190 ++++++++++ .../test/suite/rpc-shutdown.test.ts | 50 +++ 18 files changed, 757 insertions(+), 318 deletions(-) create mode 100644 packages/coding-agent/src/modes/rpc/shutdown.ts create mode 100644 packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts create mode 100644 packages/coding-agent/test/suite/rpc-close-ordering.test.ts create mode 100644 packages/coding-agent/test/suite/rpc-shutdown.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 8a5e3d7cc5..6398dd4636 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/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 diff --git a/packages/coding-agent/src/core/extensions/builtin/config-reload/changes.md b/packages/coding-agent/src/core/extensions/builtin/config-reload/changes.md index 28802b3579..339c1f227a 100644 --- a/packages/coding-agent/src/core/extensions/builtin/config-reload/changes.md +++ b/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 diff --git a/packages/coding-agent/src/core/extensions/builtin/config-reload/index.ts b/packages/coding-agent/src/core/extensions/builtin/config-reload/index.ts index 9a1f3fbd19..ce5f04a409 100644 --- a/packages/coding-agent/src/core/extensions/builtin/config-reload/index.ts +++ b/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[]>> = []; 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 = 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 => { - 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,10 +471,12 @@ 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(); }); @@ -474,7 +484,7 @@ export function configReloadExtension(pi: ExtensionAPI, options: ConfigReloadExt 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 { diff --git a/packages/coding-agent/src/core/extensions/builtin/config-reload/watch-engine.ts b/packages/coding-agent/src/core/extensions/builtin/config-reload/watch-engine.ts index e87e762e93..bde0297952 100644 --- a/packages/coding-agent/src/core/extensions/builtin/config-reload/watch-engine.ts +++ b/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 | undefined; readonly #subscribe: WatchEventSource; readonly #onRealChange: (change: RealChange) => void; readonly #onError: (error: unknown, path: string) => void; @@ -168,15 +169,14 @@ 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 { if (this.#closed) { - return Promise.resolve(); + return this.#closeCompletion ?? Promise.resolve(); } this.#closed = true; if (this.#timer) { @@ -184,18 +184,17 @@ export class ConfigReloadWatchEngine { this.#timer = undefined; } const unsubscribes = this.#unsubscribes.splice(0); - return new Promise((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 { diff --git a/packages/coding-agent/src/core/extensions/builtin/config-reload/watch-event-source.ts b/packages/coding-agent/src/core/extensions/builtin/config-reload/watch-event-source.ts index 2ff4dc2898..c63139af8f 100644 --- a/packages/coding-agent/src/core/extensions/builtin/config-reload/watch-event-source.ts +++ b/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 | 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; }; } diff --git a/packages/coding-agent/src/modes/rpc/changes.md b/packages/coding-agent/src/modes/rpc/changes.md index f1b85146e1..bff5705a44 100644 --- a/packages/coding-agent/src/modes/rpc/changes.md +++ b/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 diff --git a/packages/coding-agent/src/modes/rpc/rpc-mode.ts b/packages/coding-agent/src/modes/rpc/rpc-mode.ts index 16e99eb184..d5ae733dea 100644 --- a/packages/coding-agent/src/modes/rpc/rpc-mode.ts +++ b/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 void> = []; const registerSignalHandlers = (): void => { @@ -144,22 +144,20 @@ export async function runRpcMode(runtimeHost: AgentSessionRuntime): Promise {}; - async function shutdown(exitCode = 0, signal?: NodeJS.Signals): Promise { - 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 => { await handler.handleInputLine(line); diff --git a/packages/coding-agent/src/modes/rpc/session-command-router.ts b/packages/coding-agent/src/modes/rpc/session-command-router.ts index 9a78db1ae0..440aaa6165 100644 --- a/packages/coding-agent/src/modes/rpc/session-command-router.ts +++ b/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(); diff --git a/packages/coding-agent/src/modes/rpc/session-worker-client.ts b/packages/coding-agent/src/modes/rpc/session-worker-client.ts index 1e0324e1b3..df30f375d7 100644 --- a/packages/coding-agent/src/modes/rpc/session-worker-client.ts +++ b/packages/coding-agent/src/modes/rpc/session-worker-client.ts @@ -57,6 +57,7 @@ export class SessionWorkerClient { private readonly listeners = new Set<() => void>(); private readonly controls = new Set<"display" | "cancel_ui">(); private latestDisplay?: Extract; + private terminalFailure?: string; private readonly callbacks: SessionWorkerCallbacks; @@ -70,6 +71,7 @@ export class SessionWorkerClient { this.requests.close(new Error("session_worker_exited")); this.listeners.clear(); callbacks.exit(); + this.publishTerminalFailure(); resolve(); }); }); @@ -253,15 +255,25 @@ export class SessionWorkerClient { private fail(error: string): void { if (this.stopped) return; this.callbacks.failure(error); + this.terminalFailure = error; if (this.writer && this.sessionId) { this.writer.enqueue(this.sessionId, { type: "session_error", error }); - this.writer.closeSession(this.sessionId, { - type: "response", - command: "close_session", - success: false, - error, - }); } this.quarantine(); } + + /** Terminal close records observe completed registry removal, including worker failure. */ + private publishTerminalFailure(): void { + const error = this.terminalFailure; + const writer = this.writer; + const sessionId = this.sessionId; + if (error === undefined || !writer || !sessionId) return; + this.terminalFailure = undefined; + writer.closeSession(sessionId, { + type: "response", + command: "close_session", + success: false, + error, + }); + } } diff --git a/packages/coding-agent/src/modes/rpc/shutdown.ts b/packages/coding-agent/src/modes/rpc/shutdown.ts new file mode 100644 index 0000000000..11c6195e8c --- /dev/null +++ b/packages/coding-agent/src/modes/rpc/shutdown.ts @@ -0,0 +1,17 @@ +/** Process shutdown entry shared by EOF, signals and transport failures. */ +export function createRpcShutdown( + dispose: (signal?: NodeJS.Signals) => Promise, + exit: (code: number) => never, +): (exitCode?: number, signal?: NodeJS.Signals) => Promise { + let pending: Promise | undefined; + let code = 0; + return (exitCode = 0, signal?: NodeJS.Signals): Promise => { + if (code === 0) code = exitCode; + if (pending) return pending; + pending = (async () => { + await dispose(signal); + return exit(code); + })(); + return pending; + }; +} diff --git a/packages/coding-agent/test/suite/config-reload-extension.test.ts b/packages/coding-agent/test/suite/config-reload-extension.test.ts index a0ec15ed94..4cf261c144 100644 --- a/packages/coding-agent/test/suite/config-reload-extension.test.ts +++ b/packages/coding-agent/test/suite/config-reload-extension.test.ts @@ -1632,18 +1632,22 @@ describe("macOS recursive watch offload", () => { // Then: setup went to the worker, events route back, and teardown waits for the last subscription expect(createRecursiveWorker).toHaveBeenCalledTimes(1); - expect(worker.postMessage).toHaveBeenCalledWith({ - kind: "watch", - id: 1, - path: "/Users/dev/large-workspace", - recursive: true, - }); - expect(worker.postMessage).toHaveBeenCalledWith({ - kind: "watch", - id: 2, - path: "/Users/dev/another-config-root", - recursive: true, - }); + expect(worker.postMessage).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "watch", + id: 1, + path: "/Users/dev/large-workspace", + recursive: true, + }), + ); + expect(worker.postMessage).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "watch", + id: 2, + path: "/Users/dev/another-config-root", + recursive: true, + }), + ); expect(listener).toHaveBeenCalledWith("change", ".omo/omo.json"); expect(onError).not.toHaveBeenCalled(); @@ -1669,12 +1673,14 @@ describe("macOS recursive watch offload", () => { const unsubscribe = source(agentDir, vi.fn(), { recursive: false }); // Then: setup went to the worker with the non-recursive flag - expect(worker.postMessage).toHaveBeenCalledWith({ - kind: "watch", - id: 1, - path: agentDir, - recursive: false, - }); + expect(worker.postMessage).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "watch", + id: 1, + path: agentDir, + recursive: false, + }), + ); unsubscribe(); expect(worker.terminate).toHaveBeenCalledTimes(1); diff --git a/packages/coding-agent/test/suite/config-reload-lazy-teardown.test.ts b/packages/coding-agent/test/suite/config-reload-lazy-teardown.test.ts index 382fd42aa1..478a3d3887 100644 --- a/packages/coding-agent/test/suite/config-reload-lazy-teardown.test.ts +++ b/packages/coding-agent/test/suite/config-reload-lazy-teardown.test.ts @@ -1,228 +1,174 @@ -import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { mkdtempDisposable, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { afterEach, describe, expect, it, vi } from "vitest"; -import { createEventBus } from "../../src/core/event-bus.ts"; import configReloadExtension from "../../src/core/extensions/builtin/config-reload/index.ts"; -import type { ConfigReloadLogger } from "../../src/core/extensions/builtin/config-reload/log.ts"; import { ConfigReloadWatchEngine, type WatchEventListener, } from "../../src/core/extensions/builtin/config-reload/watch-engine.ts"; -import type { - ExtensionAPI, - ExtensionContext, - ExtensionUIContext, - SessionShutdownEvent, - SessionStartEvent, -} from "../../src/core/extensions/types.ts"; +import { createHarness } from "./harness.ts"; -type RecordedHandler = (event: unknown, ctx: ExtensionContext) => unknown | Promise; +afterEach(() => vi.useRealTimers()); -type ManualExtension = { - readonly api: ExtensionAPI; - readonly handlers: Map; -}; - -/** - * A subscribe seam whose unsubscribes are slow: each one records the teardown - * marker observed at the moment it runs, so a test can prove close() returned - * before the loop drained. - */ -type SlowTeardownProbe = { - readonly subscribe: ( - path: string, - listener: WatchEventListener, - options?: { readonly recursive: boolean }, - ) => () => void; - /** Marker value each unsubscribe saw when it ran. */ - readonly unsubscribeMarkers: string[]; - marker: string; - emit(path: string, filename: string | null): void; - activeListenerCount(path: string): number; -}; - -function createSlowTeardownProbe(): SlowTeardownProbe { - const listeners = new Map>(); - const probe: SlowTeardownProbe = { - marker: "before-close", - unsubscribeMarkers: [], - subscribe: (path, listener) => { - const set = listeners.get(path) ?? new Set(); - set.add(listener); - listeners.set(path, set); - return () => { - probe.unsubscribeMarkers.push(probe.marker); - set.delete(listener); - }; - }, - emit: (path, filename) => { - for (const listener of [...(listeners.get(path) ?? [])]) listener("change", filename); - }, - activeListenerCount: (path) => listeners.get(path)?.size ?? 0, - }; - return probe; -} - -function createManualExtension(): ManualExtension { - const handlers = new Map(); - const api = { - events: createEventBus(), - on: (event: string, handler: RecordedHandler) => { - const registered = handlers.get(event) ?? []; - registered.push(handler); - handlers.set(event, registered); - }, - } as unknown as ExtensionAPI; - return { api, handlers }; -} - -async function invoke( - handlers: ReadonlyMap, - eventName: string, - event: unknown, - ctx: ExtensionContext, -): Promise { - const handler = handlers.get(eventName)?.at(-1); - if (!handler) throw new Error(`Missing ${eventName} handler`); - await handler(event, ctx); -} - -function fakeContext(cwd: string): ExtensionContext { - return { - cwd, - mode: "tui", - ui: { notify: () => {} } as unknown as ExtensionUIContext, - isIdle: () => true, - hasPendingMessages: () => false, - isProjectTrusted: () => true, - isCompacting: () => false, - } as unknown as ExtensionContext; -} - -function silentLogger(): ConfigReloadLogger { - return { - debug: vi.fn(), - info: vi.fn(), - warn: vi.fn(), - error: vi.fn(), - } as unknown as ConfigReloadLogger; -} - -const tempDirs: string[] = []; - -function createTempDir(prefix: string): string { - const directory = mkdtempSync(join(tmpdir(), prefix)); - tempDirs.push(directory); - return directory; -} - -afterEach(() => { - vi.useRealTimers(); - vi.restoreAllMocks(); - for (const directory of tempDirs.splice(0)) rmSync(directory, { recursive: true, force: true }); -}); - -describe("config reload watcher teardown is non-blocking", () => { - it("returns from close() before the unsubscribe loop runs", async () => { - // Given: an engine watching several directories through a probe that - // records the teardown marker each unsubscribe observes. - const rootDir = createTempDir("senpi-config-reload-lazy-engine-"); - const watchedDirs = ["one", "two", "three"].map((name) => { - const directory = join(rootDir, name); - mkdirSync(directory); - writeFileSync(join(directory, "config.json"), '{"a":1}\n', "utf-8"); - return directory; - }); - const probe = createSlowTeardownProbe(); - const onRealChange = vi.fn(); +// #1656: replaces the unsafe fire-and-forget contract with cancellation plus joined disposal. +describe("config reload shutdown", () => { + it("cancels synchronously and joins asynchronous disposal on repeated close", async () => { + // Given: every disposer shares an explicitly gated completion. + await using root = await mkdtempDisposable(join(tmpdir(), "config-close-")); + const released = Promise.withResolvers(); + const unsubscribed: string[] = []; const engine = new ConfigReloadWatchEngine({ - targets: watchedDirs.map((path, index) => ({ id: `target-${index}`, kind: "dir" as const, path })), - subscribe: probe.subscribe, - onRealChange, + targets: ["one", "two"].map((id) => ({ id, kind: "dir", path: root.path })), + subscribe: () => () => { + unsubscribed.push("cancelled"); + return released.promise; + }, + onRealChange: () => {}, }); - expect(watchedDirs.every((path) => probe.activeListenerCount(path) === 1)).toBe(true); - - // When: the engine is closed and the caller immediately marks that - // control returned to it. - const teardown = engine.close(); - probe.marker = "after-close-returned"; - - // Then: no unsubscribe ran before close() returned, and every one of them - // ran afterwards. - expect(probe.unsubscribeMarkers).toEqual([]); - await teardown; - expect(probe.unsubscribeMarkers).toEqual([ - "after-close-returned", - "after-close-returned", - "after-close-returned", - ]); - expect(watchedDirs.every((path) => probe.activeListenerCount(path) === 0)).toBe(true); + let completed = false; + try { + // When: shutdown is requested twice before native disposal completes. + const first = engine.close(); + const second = engine.close(); + void Promise.all([first, second]).then(() => { + completed = true; + }); + await Promise.resolve(); + // Then: cancellation already ran, but both callers still own the same join. + expect(unsubscribed).toHaveLength(2); + expect(second).toBe(first); + expect(completed).toBe(false); + released.resolve(); + await first; + } finally { + released.resolve(); + await engine.close(); + } }); - it("drops events delivered after close() while teardown is still pending", async () => { - // Given: a closed engine whose subscriptions are still live because the - // deferred unsubscribe loop has not drained yet. - vi.useFakeTimers(); - const rootDir = createTempDir("senpi-config-reload-lazy-drop-"); - const watchedDir = join(rootDir, "watched"); - mkdirSync(watchedDir); - const configPath = join(watchedDir, "config.json"); - writeFileSync(configPath, '{"a":1}\n', "utf-8"); - const probe = createSlowTeardownProbe(); - const onRealChange = vi.fn(); + it("ignores stale events while asynchronous teardown is outstanding", async () => { + // Given: a saved callback can still arrive after cancellation. + await using root = await mkdtempDisposable(join(tmpdir(), "config-stale-")); + const path = join(root.path, "config.json"); + await writeFile(path, "{}"); + const released = Promise.withResolvers(); + let listener: WatchEventListener = () => {}; + const changed = vi.fn(); const engine = new ConfigReloadWatchEngine({ - targets: [{ id: "target", kind: "dir", path: watchedDir }], - subscribe: probe.subscribe, - onRealChange, - debounceMs: 200, + targets: [{ id: "config", kind: "dir", path: root.path }], + subscribe: (_path, callback) => { + listener = callback; + return () => released.promise; + }, + onRealChange: changed, }); - - // When: a real content change is delivered to the still-attached listener - // after close() returned. - const teardown = engine.close(); - expect(probe.activeListenerCount(watchedDir)).toBe(1); - writeFileSync(configPath, '{"a":2}\n', "utf-8"); - probe.emit(watchedDir, "config.json"); - await vi.advanceTimersByTimeAsync(200); - - // Then: the closed engine reported nothing, and teardown still completes. - expect(onRealChange).not.toHaveBeenCalled(); - await teardown; - expect(probe.activeListenerCount(watchedDir)).toBe(0); + vi.useFakeTimers(); + try { + // When: a stale content event arrives during shutdown. + const closing = engine.close(); + await writeFile(path, '{"changed":true}'); + listener("change", "config.json"); + await vi.runAllTimersAsync(); + // Then: inert subscriptions cannot produce reload work. + expect(changed).not.toHaveBeenCalled(); + released.resolve(); + await closing; + } finally { + released.resolve(); + vi.useRealTimers(); + await engine.close(); + } }); - it("returns from session_shutdown without waiting for the unsubscribe loop", async () => { - // Given: a started extension whose watcher unsubscribes record the - // teardown marker they observe. - const agentDir = createTempDir("senpi-config-reload-lazy-shutdown-"); - writeFileSync(join(agentDir, "settings.json"), '{"theme":"dark"}\n', "utf-8"); - const probe = createSlowTeardownProbe(); - const extension = createManualExtension(); - configReloadExtension(extension.api, { - agentDir, - subscribe: probe.subscribe, - logger: silentLogger(), + it("waits for all disposers before surfacing teardown failures", async () => { + // Given: one failed disposer and one independently gated disposer. + await using root = await mkdtempDisposable(join(tmpdir(), "config-failure-")); + const released = Promise.withResolvers(); + const failure = new Error("disposer failed"); + let subscriptions = 0; + const engine = new ConfigReloadWatchEngine({ + targets: ["one", "two"].map((id) => ({ id, kind: "dir", path: root.path })), + subscribe: () => + ++subscriptions === 1 + ? () => { + throw failure; + } + : () => released.promise, + onRealChange: () => {}, }); - const context = fakeContext(agentDir); - await invoke( - extension.handlers, - "session_start", - { type: "session_start", reason: "startup" } satisfies SessionStartEvent, - context, + // When: shutdown encounters the failure before the other disposer settles. + const closing = engine.close(); + const outcome = closing.then( + () => "resolved", + (error: unknown) => error, ); - expect(probe.activeListenerCount(agentDir)).toBeGreaterThan(0); + released.resolve(); + // Then: the caller receives the collected failure rather than false success. + expect(await outcome).toMatchObject({ errors: [failure] }); + }); - // When: the session shuts down for a reload. - await invoke( - extension.handlers, - "session_shutdown", - { type: "session_shutdown", reason: "reload" } satisfies SessionShutdownEvent, - context, - ); - probe.marker = "after-shutdown-returned"; + it("awaits watchers before the real extension shutdown dispatch completes", async () => { + // Given: the real extension runner with a gated event-source disposer. + await using root = await mkdtempDisposable(join(tmpdir(), "config-session-close-")); + const released = Promise.withResolvers(); + const cancelled = Promise.withResolvers(); + const harness = await createHarness({ + extensionFactories: [ + (pi) => + configReloadExtension(pi, { + agentDir: root.path, + subscribe: () => () => { + cancelled.resolve(); + return released.promise; + }, + }), + ], + }); + let completed = false; + try { + await harness.session.bindExtensions({ mode: "tui" }); + // When: session_shutdown traverses the real runner. + const closing = harness + .getExtensionRunner() + .emit({ type: "session_shutdown", reason: "quit" }) + .then(() => { + completed = true; + }); + await cancelled.promise; + await Promise.resolve(); + // Then: process teardown cannot overtake pending watcher disposal. + expect(completed).toBe(false); + released.resolve(); + await closing; + } finally { + released.resolve(); + harness.cleanup(); + } + }); - // Then: the shutdown handler resolved before any unsubscribe ran. - expect(probe.unsubscribeMarkers).toEqual([]); + it.each([false, true])("starts RPC watchers only for persistent sessions (persistent=%s)", async (persistent) => { + // Given: actual in-memory or persistent session-manager ownership. + await using root = await mkdtempDisposable(join(tmpdir(), "config-rpc-probe-")); + const subscribe = vi.fn(() => () => {}); + const harness = await createHarness({ + persistSession: persistent, + extensionFactories: [ + (pi) => + configReloadExtension(pi, { + agentDir: root.path, + subscribe, + }), + ], + }); + try { + // When: the RPC session binds its extensions. + await harness.session.bindExtensions({ mode: "rpc" }); + // Then: snapshot-only probes avoid watches without disabling durable sessions. + expect(subscribe.mock.calls.length > 0).toBe(persistent); + } finally { + await harness.getExtensionRunner().emit({ type: "session_shutdown", reason: "quit" }); + harness.cleanup(); + } }); }); diff --git a/packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts b/packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts new file mode 100644 index 0000000000..1e3dfb76fb --- /dev/null +++ b/packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts @@ -0,0 +1,100 @@ +import { EventEmitter } from "node:events"; +import { readFile } from "node:fs/promises"; +import { runInNewContext } from "node:vm"; +import { describe, expect, it } from "vitest"; +import { createFsWatchEventSource } from "../../src/core/extensions/builtin/config-reload/watch-event-source.ts"; + +class GatedWorker extends EventEmitter { + readonly commands: unknown[] = []; + readonly exit = Promise.withResolvers(); + postMessage(command: unknown): void { + this.commands.push(command); + } + terminate(): Promise { + return this.exit.promise; + } +} + +// #1656: the worker source itself executes; only IPC delivery and native fs.watch are controlled. +describe("config watch worker shutdown", () => { + it.each(["queued", "admitted"] as const)( + "leaves no watcher when cancellation overtakes registration (%s)", + async (phase) => { + // Given: the production worker handler, paused before IPC delivery. + const source = await readFile( + new URL("../../src/core/extensions/builtin/config-reload/watch-event-source.ts", import.meta.url), + "utf8", + ); + const executable = source.match(/const RECURSIVE_WATCH_WORKER_SOURCE = `([\s\S]*?)`;/)?.[1]; + if (!executable) throw new Error("Worker entry source unavailable"); + const port = new EventEmitter(); + let registrations = 0; + let cancel = () => {}; + runInNewContext(executable, { + Atomics, + require: (specifier: string) => { + switch (specifier) { + case "node:fs": + return { + watch: () => { + if (phase === "admitted") cancel(); + registrations++; + return Object.assign(new EventEmitter(), { + close: () => { + registrations--; + }, + }); + }, + }; + case "node:worker_threads": + return { parentPort: port }; + default: + throw new Error(`Unexpected worker import: ${specifier}`); + } + }, + }); + const worker = new GatedWorker(); + const subscribe = createFsWatchEventSource(undefined, { + platform: "darwin", + createRecursiveWorker: () => worker, + }); + const unsubscribe = subscribe("/queued-watch", () => {}); + cancel = () => { + void unsubscribe(); + }; + try { + // When: cancellation precedes dispatch or lands after admission inside fs.watch. + if (phase === "queued") cancel(); + for (const command of worker.commands) port.emit("message", command); + worker.exit.resolve(0); + await unsubscribe(); + // Then: late registration is immediately disposed, never retained for delivery. + expect(registrations).toBe(0); + } finally { + worker.exit.resolve(0); + } + }, + ); + + it("returns the native termination join on repeated final unsubscribe", async () => { + // Given: termination cannot finish until the explicit release. + const worker = new GatedWorker(); + const subscribe = createFsWatchEventSource(undefined, { + platform: "darwin", + createRecursiveWorker: () => worker, + }); + const unsubscribe = subscribe("/queued-watch", () => {}); + try { + // When: the final unsubscribe is requested repeatedly. + const first = unsubscribe(); + const second = unsubscribe(); + // Then: both callers own the same pending native teardown. + expect(first).toBeInstanceOf(Promise); + expect(second).toBe(first); + worker.exit.resolve(0); + await first; + } finally { + worker.exit.resolve(0); + } + }); +}); diff --git a/packages/coding-agent/test/suite/regressions/477-recursive-watch-main-thread-stall.test.ts b/packages/coding-agent/test/suite/regressions/477-recursive-watch-main-thread-stall.test.ts index 6f91e9dee9..58aaf340f3 100644 --- a/packages/coding-agent/test/suite/regressions/477-recursive-watch-main-thread-stall.test.ts +++ b/packages/coding-agent/test/suite/regressions/477-recursive-watch-main-thread-stall.test.ts @@ -40,18 +40,22 @@ describe("issue #477 recursive watch main-thread stall", () => { worker.emit("message", { kind: "event", id: 1, eventType: "change", filename: ".omo/omo.json" }); expect(createRecursiveWorker).toHaveBeenCalledTimes(1); - expect(worker.postMessage).toHaveBeenCalledWith({ - kind: "watch", - id: 1, - path: "/large-workspace-mount", - recursive: true, - }); - expect(worker.postMessage).toHaveBeenCalledWith({ - kind: "watch", - id: 2, - path: "/another-config-root", - recursive: true, - }); + expect(worker.postMessage).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "watch", + id: 1, + path: "/large-workspace-mount", + recursive: true, + }), + ); + expect(worker.postMessage).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "watch", + id: 2, + path: "/another-config-root", + recursive: true, + }), + ); expect(mocks.fsWatch).not.toHaveBeenCalled(); expect(listener).toHaveBeenCalledWith("change", ".omo/omo.json"); diff --git a/packages/coding-agent/test/suite/regressions/non-recursive-watch-main-thread-stall.test.ts b/packages/coding-agent/test/suite/regressions/non-recursive-watch-main-thread-stall.test.ts index 4deee511d8..a8652e66d3 100644 --- a/packages/coding-agent/test/suite/regressions/non-recursive-watch-main-thread-stall.test.ts +++ b/packages/coding-agent/test/suite/regressions/non-recursive-watch-main-thread-stall.test.ts @@ -42,12 +42,14 @@ describe("non-recursive watch main-thread stall", () => { worker.emit("message", { kind: "event", id: 1, eventType: "change", filename: "ext.ts" }); expect(createRecursiveWorker).toHaveBeenCalledTimes(1); - expect(worker.postMessage).toHaveBeenCalledWith({ - kind: "watch", - id: 1, - path: "/agent/extensions", - recursive: false, - }); + expect(worker.postMessage).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "watch", + id: 1, + path: "/agent/extensions", + recursive: false, + }), + ); expect(mocks.fsWatch).not.toHaveBeenCalled(); expect(listener).toHaveBeenCalledWith("change", "ext.ts"); @@ -72,12 +74,14 @@ describe("non-recursive watch main-thread stall", () => { // The dead worker is dropped and live subscriptions land on a fresh one. expect(onError).toHaveBeenCalledWith(expect.any(Error), "/agent/extensions"); expect(createRecursiveWorker).toHaveBeenCalledTimes(2); - expect(workers[1]?.postMessage).toHaveBeenCalledWith({ - kind: "watch", - id: 1, - path: "/agent/extensions", - recursive: false, - }); + expect(workers[1]?.postMessage).toHaveBeenCalledWith( + expect.objectContaining({ + kind: "watch", + id: 1, + path: "/agent/extensions", + recursive: false, + }), + ); // Events from the replacement worker still reach the original listener. workers[1]?.emit("message", { kind: "event", id: 1, eventType: "change", filename: "ext.ts" }); diff --git a/packages/coding-agent/test/suite/rpc-close-backpressure.test.ts b/packages/coding-agent/test/suite/rpc-close-backpressure.test.ts index 495636139e..b3fbd61408 100644 --- a/packages/coding-agent/test/suite/rpc-close-backpressure.test.ts +++ b/packages/coding-agent/test/suite/rpc-close-backpressure.test.ts @@ -181,15 +181,33 @@ it.each(["quarantined", "finalizing-records", "finalizing-bytes"])( const router = new SessionCommandRouter(host.registry, writer, { cwd }); const gate = Promise.withResolvers(); const entered = Promise.withResolvers(); + const markedDone = Promise.withResolvers(); const closeMarked = host.registry.closeMarked.bind(host.registry); const finalizing = state !== "quarantined"; const teardown = vi.spyOn(host.registry, "closeMarked").mockImplementation(async (handle) => { entered.resolve(); if (finalizing) await gate.promise; - await closeMarked(handle); + try { + await closeMarked(handle); + } finally { + markedDone.resolve(); + } }); let reader: Awaited> | undefined; const pending: Array> = []; + const unstickWorker = async (): Promise => { + if (!reader) return; + const rescue = await open(fifo, "r+"); + try { + await unlink(fifo); + const header = `${JSON.stringify({ type: "session", version: 3, id: "bound-durable", timestamp: new Date(0).toISOString(), cwd })}\n`; + await writeFile(fifo, header); + await rescue.write(header); + } finally { + await Promise.all([reader.close(), rescue.close()]); + reader = undefined; + } + }; try { const opening = host.send("opening", { type: "open_session", cwd, sessionPath: fifo }); pending.push(opening); @@ -224,11 +242,25 @@ it.each(["quarantined", "finalizing-records", "finalizing-bytes"])( error: expect.stringContaining("session_path_in_use"), }); gate.resolve(); + if (finalizing) { + await phase("close-grace", markedDone.promise); + expect(host.exited.has(worker)).toBe(false); + expect(host.registry.peek(entry.sessionId)?.state).toBe("quarantined"); + await writer.flush(); + expect(records.filter((record) => record.type === "session_closed")).toEqual([]); + expect(records.filter((record) => record.type === "overflow")).toEqual([overflow]); + await unstickWorker(); + } await phase("close-replies-settled", Promise.all(pending)); expect(writer.pendingCloseRecordCount).toBe(0); expect(writer.pendingCloseByteLength).toBe(0); - expect(host.exited.has(worker)).toBe(false); - expect(host.registry.peek(entry.sessionId)?.state).toBe("quarantined"); + if (finalizing) { + expect(host.exited.has(worker)).toBe(true); + expect(host.registry.peek(entry.sessionId)).toBeUndefined(); + } else { + expect(host.exited.has(worker)).toBe(false); + expect(host.registry.peek(entry.sessionId)?.state).toBe("quarantined"); + } await writer.flush(); expect(records.filter((record) => record.type === "overflow")).toEqual([overflow]); const replies = records.filter((record) => record.sessionId === entry.sessionId); @@ -251,15 +283,7 @@ it.each(["quarantined", "finalizing-records", "finalizing-bytes"])( }); } finally { gate.resolve(); - const rescue = await open(fifo, "r+"); - try { - await unlink(fifo); - const header = `${JSON.stringify({ type: "session", version: 3, id: "bound-durable", timestamp: new Date(0).toISOString(), cwd })}\n`; - await writeFile(fifo, header); - await rescue.write(header); - } finally { - await Promise.all([reader?.close(), rescue.close()]); - } + await unstickWorker(); await host.dispose(); await Promise.all(pending); await router.dispose(); diff --git a/packages/coding-agent/test/suite/rpc-close-ordering.test.ts b/packages/coding-agent/test/suite/rpc-close-ordering.test.ts new file mode 100644 index 0000000000..a6f43ec824 --- /dev/null +++ b/packages/coding-agent/test/suite/rpc-close-ordering.test.ts @@ -0,0 +1,190 @@ +import type { EventEmitter } from "node:events"; +import { Worker } from "node:worker_threads"; +import { afterEach, expect, it, vi } from "vitest"; +import { parseArgs } from "../../src/cli/args.ts"; +import { buildRpcSessionState } from "../../src/modes/rpc/connection-handler.ts"; +import type { RpcResponse } from "../../src/modes/rpc/rpc-types.ts"; +import { SessionCommandRouter } from "../../src/modes/rpc/session-command-router.ts"; +import { SessionEventWriter } from "../../src/modes/rpc/session-event-writer.ts"; +import type { HostToSessionWorker } from "../../src/modes/rpc/session-worker-protocol.ts"; +import { WorkerSessionRegistry } from "../../src/modes/rpc/worker-session-registry.ts"; +import { createHarness } from "./harness.ts"; + +vi.mock("node:worker_threads", async () => { + const { EventEmitter } = await import("node:events"); + return { + Worker: class extends EventEmitter { + postMessage(): void {} + terminate(): Promise { + return Promise.resolve(0); + } + }, + }; +}); + +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +async function closeFixture() { + const harness = await createHarness(); + const state = buildRpcSessionState(harness.session); + const sessionPath = `${harness.tempDir}/session.jsonl`; + vi.spyOn(Worker.prototype, "postMessage").mockImplementation(function ( + this: EventEmitter, + message: HostToSessionWorker, + ) { + switch (message.type) { + case "prepare": + queueMicrotask(() => this.emit("message", { type: "prepared", request: message.request, sessionPath })); + break; + case "commit": + queueMicrotask(() => + this.emit("message", { + type: "ready", + request: message.request, + snapshot: { state, sessionPath, liveSessionPaths: [sessionPath], busy: false, streaming: false }, + }), + ); + break; + case "bind": + case "command": + queueMicrotask(() => this.emit("message", { type: "result", request: message.request })); + break; + case "close": + case "cancel_ui": + case "display": + break; + default: { + const exhaustive: never = message; + throw new Error(`Unexpected message ${exhaustive}`); + } + } + }); + const registry = new WorkerSessionRegistry({ + configuration: { + parsed: parseArgs(["--mode", "rpc"]), + cwd: harness.tempDir, + agentDir: harness.tempDir, + appMode: "rpc", + }, + closeGraceMs: 100, + now: () => 0, + }); + const records: unknown[] = []; + const observations: Array> = []; + const writer = new SessionEventWriter((line) => { + const record: unknown = JSON.parse(line); + records.push(record); + if ( + typeof record === "object" && + record !== null && + (("type" in record && record.type === "session_closed") || + ("command" in record && record.command === "close_session")) + ) { + observations.push(router.handle({ type: "list_sessions", id: "immediate" })); + } + }); + const router = new SessionCommandRouter(registry, writer, { cwd: harness.tempDir }); + await router.handle({ type: "open_session", cwd: harness.tempDir }); + await writer.flush(); + const entry = registry.list()[0]; + if (!entry) throw new Error("Expected open session"); + const client = registry.peek(entry.sessionId)?.worker; + if (!client) throw new Error("Expected session worker client"); + records.length = 0; + return { + registry, + router, + writer, + records, + observations, + client, + sessionId: entry.sessionId, + async [Symbol.asyncDispose]() { + client.quarantine(); + client.worker.emit("exit", 0); + await client.exited; + await router.dispose(); + harness.cleanup(); + }, + }; +} + +// #1656: only worker transport is controlled; real requests, exit callback, registry, router and writer execute. +it("publishes successful close only after native exit removes ownership", async () => { + // Given: the worker transport holds exit beyond the grace deadline. + await using host = await closeFixture(); + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + // When: close reaches its deadline before the worker exit is delivered. + const closing = host.router.handle({ type: "close_session", id: "close", sessionId: host.sessionId }); + await vi.advanceTimersByTimeAsync(100); + await host.writer.flush(); + const beforeExit = [...host.records]; + const retained = host.registry.size; + host.client.worker.emit("exit", 0); + await closing; + await host.writer.flush(); + // Then: both terminal records observe the real removal, never the deadline. + expect(beforeExit).toEqual([]); + expect(retained).toBe(1); + expect(host.records).toEqual([ + { type: "session_closed", sessionId: host.sessionId }, + { id: "close", type: "response", command: "close_session", success: true, data: {}, sessionId: host.sessionId }, + ]); + expect(await Promise.all(host.observations)).toEqual( + Array(2).fill({ + id: "immediate", + type: "response", + command: "list_sessions", + success: true, + data: { sessions: [] }, + }), + ); +}); + +it.each(["error", "failure"] as const)( + "defers terminal failure publication until ownership removal (%s)", + async (kind) => { + // Given: a bound worker whose termination acknowledgement is held. + await using host = await closeFixture(); + // When: an error event or failure frame initiates quarantine before exit. + switch (kind) { + case "error": + host.client.worker.emit("error", new Error("worker-failed")); + break; + case "failure": + host.client.worker.emit("message", { type: "failure", error: "worker-failed" }); + break; + default: { + const exhaustive: never = kind; + throw new Error(`Unexpected failure ${exhaustive}`); + } + } + await host.writer.flush(); + const beforeExit = [...host.observations]; + host.client.worker.emit("exit", 1); + await host.client.exited; + await host.writer.flush(); + // Then: error identity is retained but no terminal record precedes removal. + expect(beforeExit).toEqual([]); + expect(host.records).toContainEqual({ type: "session_closed", sessionId: host.sessionId }); + expect(host.records).toContainEqual({ + type: "response", + command: "close_session", + success: false, + error: "worker-failed", + sessionId: host.sessionId, + }); + expect(await Promise.all(host.observations)).toEqual( + Array(2).fill({ + id: "immediate", + type: "response", + command: "list_sessions", + success: true, + data: { sessions: [] }, + }), + ); + }, +); diff --git a/packages/coding-agent/test/suite/rpc-shutdown.test.ts b/packages/coding-agent/test/suite/rpc-shutdown.test.ts new file mode 100644 index 0000000000..cc3a2e458e --- /dev/null +++ b/packages/coding-agent/test/suite/rpc-shutdown.test.ts @@ -0,0 +1,50 @@ +import { expect, it } from "vitest"; +import { createRpcShutdown } from "../../src/modes/rpc/shutdown.ts"; + +class ObservedExit extends Error { + readonly code: number; + constructor(code: number) { + super(`exit ${code}`); + this.code = code; + } +} + +// #1656: serializer failure and EOF must join, not bypass watcher disposal. +it.each([ + [1, 0], + [0, 1], +])("joins reentrant shutdown and preserves failure (first=%s, second=%s)", async (firstCode, secondCode) => { + // Given: the production RPC shutdown entry with disposal and process exit observed. + const started = Promise.withResolvers(); + const released = Promise.withResolvers(); + let disposals = 0; + let disposed = false; + const exits: Array<{ code: number; disposed: boolean }> = []; + const shutdown = createRpcShutdown( + async () => { + disposals++; + started.resolve(); + await released.promise; + disposed = true; + }, + (code) => { + exits.push({ code, disposed }); + throw new ObservedExit(code); + }, + ); + const first = shutdown(firstCode); + const firstOutcome = first.catch((error: unknown) => error); + await started.promise; + // When: the other shutdown source arrives before disposal completes. + const second = shutdown(secondCode); + const secondOutcome = second.catch((error: unknown) => error); + const earlyExits = [...exits]; + released.resolve(); + const outcomes = await Promise.all([firstOutcome, secondOutcome]); + // Then: both calls share one join and exactly one failure exit after disposal. + expect(earlyExits).toEqual([]); + expect(second).toBe(first); + expect(disposals).toBe(1); + expect(exits).toEqual([{ code: 1, disposed: true }]); + for (const outcome of outcomes) expect(outcome).toMatchObject({ code: 1 }); +}); From 79c637dcb07583087bc3383cec835af9963e9b56 Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Tue, 15 Sep 2026 18:30:11 +0900 Subject: [PATCH 2/3] test(coding-agent): isolate post-watch cancellation from later unwatch The admitted interleaving posts unwatch during fs.watch. Delivering that unwatch in the same IPC drain hid a missing post-watch close. Snapshot the queued commands so only the watch message runs. Plan: .omo/plans/omo-dependency-diet.md --- .../test/suite/config-reload-worker-shutdown.test.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts b/packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts index 1e3dfb76fb..48d307f547 100644 --- a/packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts +++ b/packages/coding-agent/test/suite/config-reload-worker-shutdown.test.ts @@ -64,8 +64,11 @@ describe("config watch worker shutdown", () => { }; try { // When: cancellation precedes dispatch or lands after admission inside fs.watch. + // Snapshot IPC so an unwatch posted from inside fs.watch is not also delivered; + // the post-watch cancellation check must dispose that handle itself. if (phase === "queued") cancel(); - for (const command of worker.commands) port.emit("message", command); + const dispatched = worker.commands.splice(0); + for (const command of dispatched) port.emit("message", command); worker.exit.resolve(0); await unsubscribe(); // Then: late registration is immediately disposed, never retained for delivery. From dcc0dd2db66ef2a00559144650f741abf0cf735d Mon Sep 17 00:00:00 2001 From: YeonGyu-Kim Date: Tue, 15 Sep 2026 18:43:58 +0900 Subject: [PATCH 3/3] test(rpc): stub peek on multi-session close registry fakes finalizeClose waits on registry.peek().closeCompletion. The joined-close fakes omitted peek, so close_session threw TypeError before publishing. Plan: .omo/plans/omo-dependency-diet.md --- packages/coding-agent/test/rpc-multi-session.test.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/packages/coding-agent/test/rpc-multi-session.test.ts b/packages/coding-agent/test/rpc-multi-session.test.ts index 8b40b7fddd..01012706d7 100644 --- a/packages/coding-agent/test/rpc-multi-session.test.ts +++ b/packages/coding-agent/test/rpc-multi-session.test.ts @@ -77,6 +77,7 @@ describe("multi-session RPC routing", () => { list: () => [], beginClose: () => entry, closeMarked: async () => {}, + peek: () => undefined, } as never; const createBinding = vi.fn(async () => ({ handle: async () => {}, @@ -125,6 +126,7 @@ describe("multi-session RPC routing", () => { list: () => [], beginClose: () => entry, closeMarked: async () => {}, + peek: () => undefined, } as never; const chunks: string[] = []; const writer = new SessionEventWriter( @@ -217,6 +219,7 @@ describe("multi-session RPC routing", () => { list: () => [], beginClose: () => entryFor("closing"), closeMarked: async () => {}, + peek: () => undefined, } as never; const alphaHandle = vi.fn(async () => {}); const betaHandle = vi.fn(async () => {}); @@ -286,6 +289,7 @@ describe("multi-session RPC routing", () => { return entry; }, closeMarked: async () => closeCompletion, + peek: () => ({ closeCompletion }), } as never; const dispose = vi.fn(() => disposing); const records: Array> = []; @@ -329,6 +333,7 @@ describe("multi-session RPC routing", () => { return entry; }, closeMarked, + peek: () => undefined, } as never; const records: Array> = []; const writer = new SessionEventWriter( @@ -378,6 +383,7 @@ describe("multi-session RPC routing", () => { return entry; }, closeMarked: async () => {}, + peek: () => undefined, } as never; const writer = new SessionEventWriter(() => {}); writer.registerConnection("owner", { writeRaw: () => {}, waitForBackpressure: async () => {} });