diff --git a/.changeset/quiet-metadata-admission.md b/.changeset/quiet-metadata-admission.md new file mode 100644 index 000000000..4959d49b9 --- /dev/null +++ b/.changeset/quiet-metadata-admission.md @@ -0,0 +1,5 @@ +--- +"@schlessera/brain-ui-server": minor +--- + +Cancel queued module metadata when its HTTP caller disconnects or as soon as the app starts closing, so `/api/status` no longer holds `close()` open behind a busy output lock. Admitted metadata drains within the close grace period, like other shared runs, with database ownership retained until handlers settle. Add an optional admission-cancellation signal to `BrainClient.modules`. diff --git a/api-report/ui-server.txt b/api-report/ui-server.txt index 4d8f6abaa..5b4a0f464 100644 --- a/api-report/ui-server.txt +++ b/api-report/ui-server.txt @@ -252,7 +252,9 @@ signatures (every public export and the types it is made of) export interface BrainClient { hygiene?(args: readonly string[]): Promise; cliCommand(): string[]; - modules?(): Promise<{ + modules?(opts?: { + signal?: AbortSignal; + }): Promise<{ enabled: ConfiguredModule[]; available: unknown[]; }>; diff --git a/docs/decisions/claude-code-runtime.md b/docs/decisions/claude-code-runtime.md index ed16832e4..cc6168238 100644 --- a/docs/decisions/claude-code-runtime.md +++ b/docs/decisions/claude-code-runtime.md @@ -63,8 +63,8 @@ next to it. - **Updated.** Nothing in the tree installs, updates, pins or reads the version of this binary. The only version probe in the server is for the `brain` CLI (`Probe the selected content CLI`, - `packages/ui-server/src/brain/client.ts:121-187`, called at - `probeBrainCliVersion(config.brainPath`, `packages/ui-server/src/app.ts:413`). + `packages/ui-server/src/brain/client.ts:121-189`, called at + `probeBrainCliVersion(config.brainPath`, `packages/ui-server/src/app.ts:415`). `brain doctor` runs `claude mcp list` from `PATH` (`which("claude")`, `packages/core/src/cli/commands/doctor.ts:495-499`) — the user's own Claude Code on their own machine, to check the MCP registration, not the server's @@ -212,7 +212,7 @@ unnecessary. `node_modules/@anthropic-ai/claude-agent-sdk/sdk.mjs`. The boot probe below turns either into a refusal to start, for the same reason a missing backend refuses to boot (`A missing (or unrecognized) agent backend`, - `packages/ui-server/src/app.ts:358-363`). + `packages/ui-server/src/app.ts:360-365`). - **The published range is still a caret, and that bounds what this repo can guarantee.** `@schlessera/brain-backend-claude` depends on `^0.3.241`, and a host resolves it in its own lockfile. A host can bump the SDK — and so the diff --git a/docs/decisions/governed-writable-workers.md b/docs/decisions/governed-writable-workers.md index a1fc0cb7c..ce611ea46 100644 --- a/docs/decisions/governed-writable-workers.md +++ b/docs/decisions/governed-writable-workers.md @@ -2073,7 +2073,7 @@ files, `.claude/`, `.agents/`, `.pi/`, `.mcp.json`, and - The core CLI loads `brain.config.ts` or `brain.config.json` at the canonical brain cwd (`loadUserConfig`, `packages/core/src/lib/config.ts:590-624`). Host commands bind that cwd to the selected brain (`function hostSpawn`, - `packages/ui-server/src/brain/host-boundary.ts:430-441`). + `packages/ui-server/src/brain/host-boundary.ts:433-444`). - Bun startup uses root `bunfig.toml` and `package.json`; retained literal preload/module inputs can identify a nested module's nearest package scope. The inventory walks parent directories only within the originating journal diff --git a/docs/decisions/http-api-boundary.md b/docs/decisions/http-api-boundary.md index 82970ec9c..f176aa6bc 100644 --- a/docs/decisions/http-api-boundary.md +++ b/docs/decisions/http-api-boundary.md @@ -46,11 +46,11 @@ explicit supported rows, not silently excluded by this distinction. The source audit on 2026-09-30 used `e0875269b21b707482545be9e94901acb897893b`. The public health mount precedes auth -(`createHealthRoutes`, `packages/ui-server/src/app.ts:766`), the API guard follows -public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:787`), and the -WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:969`). +(`createHealthRoutes`, `packages/ui-server/src/app.ts:770`), the API guard follows +public ceremonies (`authGuard`, `packages/ui-server/src/app.ts:791`), and the +WebSocket admission is separate (`"/ws"`, `packages/ui-server/src/app.ts:973`). Static serving is conditional (`if (options.staticRoot)`, -`packages/ui-server/src/app.ts:989-1010`). That order establishes access; +`packages/ui-server/src/app.ts:993-1014`). That order establishes access; it does not select compatibility status. A hermetic real-app launch, with temporary brain root, in-memory database, @@ -80,6 +80,16 @@ at response creation: existing SSE and journal owners still govern streaming response producers. A stalled client or producer cancellation cannot make shutdown wait for unbounded additional request bytes. +Module metadata for status and the modules list opts into caller and app-lifetime +admission cancellation. A queued lookup settles without launching after its +caller disconnects or as soon as shutdown begins. An admitted lookup is a shared +run: it keeps its publication and cleanup through shutdown's grace period and is +then cancelled with the other shared runs (#2473). Status continues to omit +unavailable optional renderer metadata, while already completed results retain +their ordinary successful response. Other HTTP work, streaming producers and +governed refresh retain their existing lifetimes; the dispatch join still +precedes state disposal. + ## Versioning and changes Supported HTTP methods, paths, authorization requirements, accepted inputs, diff --git a/docs/decisions/runtime-requirements.md b/docs/decisions/runtime-requirements.md index 511bcd5c0..906a74002 100644 --- a/docs/decisions/runtime-requirements.md +++ b/docs/decisions/runtime-requirements.md @@ -271,7 +271,7 @@ reciprocal host-version negotiation is involved. | Existing boundary | How this design uses it | | --- | --- | -| `probeBrainCliVersion` / `brainCliCommand` (`export async function probeBrainCliVersion`, `packages/ui-server/src/brain/client.ts:122-187`) | Reuse executable selection, wrapper, deadline and cleanup; add host floor composition, strict comparison and invocation verification without duplicating the 0.33.0 constant. | +| `probeBrainCliVersion` / `brainCliCommand` (`export async function probeBrainCliVersion`, `packages/ui-server/src/brain/client.ts:122-189`) | Reuse executable selection, wrapper, deadline and cleanup; add host floor composition, strict comparison and invocation verification without duplicating the 0.33.0 constant. | | `installedAgentSdkVersion` (`export function installedAgentSdkVersion`, `packages/ui-backend-claude/src/runtime-probe.ts:55-61`) | Retain importer-relative resolution; validate package name/version and the owner's manifest range. | | `selectedSpawn` / `probeClaudeRuntime` (`export function selectedSpawn`, `packages/ui-backend-claude/src/runtime-probe.ts:69-96`; `export async function probeClaudeRuntime`, `packages/ui-backend-claude/src/runtime-probe.ts:127-147`) | Keep SDK-selected command capture, native/JS override argv, environment, timeout and group cleanup; add strict identity validation and composed runtime floors. | | `probeBackendRuntimes` (`export async function probeBackendRuntimes`, `packages/ui-server/src/agent/backend.ts:392-439`) | Forward validated host requirements; retain active-backend selection and separate unmeasured-pair logging. | diff --git a/docs/decisions/scheduled-tasks.md b/docs/decisions/scheduled-tasks.md index 28727a464..9cb6e5dc1 100644 --- a/docs/decisions/scheduled-tasks.md +++ b/docs/decisions/scheduled-tasks.md @@ -23,7 +23,7 @@ Source inspected at `b5c30c180df5584584f261b446c4d440380d2441`: (`else if (item.status === "scheduled"`, `packages/ui-server/src/inbox/actions.ts:188-189`). The runtime ticks every minute (`INBOX_TICK_MS`, `packages/ui-server/src/inbox/runtime.ts:11-13`). - Production app wiring supplies recovery and budgets, without a dispatcher - (`// Recovery/heartbeat only.`, `packages/ui-server/src/app.ts:430-446`). + (`// Recovery/heartbeat only.`, `packages/ui-server/src/app.ts:432-448`). A scripted nonempty ready-item test still reports no claim, attempt or Activity root. That proves disabled wiring, not safe execution of a schedule. - Headless plumbing checks backend support, a usable principal, lifetime and diff --git a/docs/decisions/session-principals.md b/docs/decisions/session-principals.md index 1e6a6ba1d..5ef373b87 100644 --- a/docs/decisions/session-principals.md +++ b/docs/decisions/session-principals.md @@ -96,9 +96,9 @@ Both reviewers verified every row below against the source. | Cookie verify | `auth.ts:265-296` | strict `^(\d+)\.(\d+)$`, safe integers, future/expired rejection, `cookieEpoch === sessionsEpoch(db)` | | Global epoch | `auth.ts:302-318` | `settings` row; missing = 0; non-integer **throws** | | Global revoke | `auth.ts:321-330` | `setSetting` (overflow-guarded) + `clients.closeAll(1008, …)` | -| Auth routes | `authRoutes`, `app.ts:772` | mounted **before** the guard at `authGuard(authMode`, `app.ts:787`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:808`) | -| Passkey management | `passkeyManagementRoutes`, `app.ts:797` | mounted **after** the guard ("Mount AFTER the auth guard") | -| WS guard | `"/ws"`, `app.ts:969-986` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade | +| Auth routes | `authRoutes`, `app.ts:776` | mounted **before** the guard at `authGuard(authMode`, `app.ts:791`; logout guards itself (`resolveCookiePrincipal(c, auth, deps.db)`, `auth.ts:808`) | +| Passkey management | `passkeyManagementRoutes`, `app.ts:801` | mounted **after** the guard ("Mount AFTER the auth guard") | +| WS guard | `"/ws"`, `app.ts:973-990` | origin → `isWsAuthorized` (returns a boolean, `auth.ts:221-237`) → capacity → upgrade | | WS upgrade | `ws/connection.ts:262-264` | `createWsUpgrade` ignores the request context | | WS admission | `ws/connection.ts:84` | `clients.add(ws)` — the socket's identity is unknown | | Turn record | `ws/run-session.ts:134-142` | recorder built from `{turnId, sessionId, billing}`; `RunningTurn` (`ws/turns.ts:55-77`) has no actor | diff --git a/docs/integration-contract/package-api.md b/docs/integration-contract/package-api.md index 2f697c393..e29a7c5a8 100644 --- a/docs/integration-contract/package-api.md +++ b/docs/integration-contract/package-api.md @@ -3609,3 +3609,24 @@ the same identity is allowed; changing an established identity or patching a terminal span returns false without modifying the row. The recorder persists this identity in both the queryable session column and the existing attributes. This additive API ships in a minor release. + +## Module metadata admission cancellation (additive, #2457) + +`BrainClient.modules({ signal })` accepts an optional abort signal. It cancels +metadata waiting for execution admission, including a required CLI version +probe. Once an execution owns its admission locks, its worker, output +publication and cleanup drain normally. Cancellation between a version probe +and the module lookup prevents the latter from starting. Calls without an +option preserve their existing behavior and result shape. + +The internal host boundary carries a separate admission signal so existing +execution cancellation, including search worker termination and output drain, +keeps its semantics. The optional `BrainClient.modules` input is an additive +public TypeScript change and ships in a minor. + +`createApp()` passes each `/api/status` and `/api/modules` request's signal, +so a caller that disconnects cancels its queued metadata. `close()` cancels +every metadata request still waiting for admission as soon as it begins. +Admitted metadata is a shared run: it drains within `close()`'s 10 s grace +period and is then cancelled with the other shared runs (#2473). The response +shapes and status codes are unchanged. diff --git a/docs/plans/async-collaboration-requirements.md b/docs/plans/async-collaboration-requirements.md index 196541378..5c4f69f3e 100644 --- a/docs/plans/async-collaboration-requirements.md +++ b/docs/plans/async-collaboration-requirements.md @@ -226,7 +226,7 @@ Stated before the requirements because five of them derive from it. that is the shape to copy, including its `close()` lifecycle. - R19. **The cron backstop has independent authorization before the general guard.** Mount it on the existing listener before - (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:787`). + (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:791`). Authorize a boot-minted ephemeral token, rotated each boot and stored in a 0600 runtime file, with the actual socket address as an additional check. Proxy headers cannot authorize it. The poke succeeds in every auth mode diff --git a/docs/plans/async-collaboration.md b/docs/plans/async-collaboration.md index 3bfe9a444..436886fe4 100644 --- a/docs/plans/async-collaboration.md +++ b/docs/plans/async-collaboration.md @@ -131,7 +131,7 @@ system gate. It is not a prerequisite to rebuild or a new independent package. owns manual triggers/history; it does not tick the inbox. - **Two connections:** (`export function createUiDb(`, `packages/ui-server/src/db/client.ts:25-37`) sets WAL, foreign keys and a 5-second busy timeout. Claims are immediate transactions. -- **Auth mounting:** (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:787`) follows public routes; +- **Auth mounting:** (`app.use("/api/*", authGuard(`, `packages/ui-server/src/app.ts:791`) follows public routes; (`export function authGuard(`, `packages/ui-server/src/middleware/auth.ts:189-249`) binds principals in each auth mode. An internal poke needs independent token authorization before this guard. - **Permission parking:** (`requestPermission: (req) => {`, `packages/ui-server/src/ws/bridge.ts:281-375`). Timeout unwind is diff --git a/docs/plans/navigation-recovery.md b/docs/plans/navigation-recovery.md index bfad9d858..5577225bc 100644 --- a/docs/plans/navigation-recovery.md +++ b/docs/plans/navigation-recovery.md @@ -397,7 +397,7 @@ implementation. Any PR moving these lines owns their citation repairs. - Activity resolution: (`.get("/activity/runs", (c) => {`, `packages/ui-server/src/routes/activity.ts:143-281`). -- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:787`). +- API authorization: (`app.use("/api/*", authGuard(authMode, auth, db));`, `packages/ui-server/src/app.ts:791`). - Queued acceptance: (`// Queue it as the session's next turn; report queued immediately.`, `packages/ui-server/src/ws/run-session.ts:933-978`). diff --git a/docs/plans/storage-accounting.md b/docs/plans/storage-accounting.md index 2794e8010..7934d9991 100644 --- a/docs/plans/storage-accounting.md +++ b/docs/plans/storage-accounting.md @@ -67,7 +67,7 @@ Source anchors, checked against the reviewed commit: The completed #94 and #398 provide existing stats/maintenance foundations. - C: snapshot fields (`snapshotOf`, `packages/core/src/lib/stats-history.ts:132-172`), thinning (`thin`, `packages/core/src/lib/stats-history.ts:232-244`). -- D: effective path/journal resolution (`createApp`, `packages/ui-server/src/app.ts:279-315`), +- D: effective path/journal resolution (`createApp`, `packages/ui-server/src/app.ts:281-317`), WAL store (`createUiDb`, `packages/ui-server/src/db/client.ts:25-37`), and private locations (`statePaths`, `packages/ui-server/src/brain/operational-state.ts:107-113`). - E: live/archive paths (`transcriptPaths`, `packages/ui-backend-claude/src/transcript-archive.ts:29-38`). diff --git a/docs/plans/storage-lifecycle.md b/docs/plans/storage-lifecycle.md index 608fbb7fb..c7d8eccbe 100644 --- a/docs/plans/storage-lifecycle.md +++ b/docs/plans/storage-lifecycle.md @@ -31,7 +31,7 @@ Use maintenance workers for repository code, the CLI and Git. Host routes may perform narrowly validated descriptor operations, as the journal/revert do; they must not evaluate brain configuration, hooks, filters or executables. The existing brain client supplies boundary execution -(`async function execBrain`, `packages/ui-server/src/brain/client.ts:231-283`). +(`async function execBrain`, `packages/ui-server/src/brain/client.ts:240-294`). Do not turn the new surface into a host command runner or an extension seam. Reuse existing authorization. A requested reversible, fully journaled edit diff --git a/docs/process/evidence/test-wait-review-partition.json b/docs/process/evidence/test-wait-review-partition.json index 4f112bd63..c36eccc10 100644 --- a/docs/process/evidence/test-wait-review-partition.json +++ b/docs/process/evidence/test-wait-review-partition.json @@ -1999,7 +1999,7 @@ "basis": "Embedded `brain` CLI body written by brainWithCli or an inline fake bin; the test spawns it through the brain client and never waits for the literal. The hold exists so the cancellation path has a live child to kill. No completed elapsed minimum is implied by the literal.", "noElapsedMinimumImplied": true, "raceWinner": "controller.abort() after the host PID is seen wins; the 60000 ms sleep never completes in a passing run", - "holdEnd": "SIGKILL: the client's execBrain passes killSignal SIGKILL with the abort signal (packages/ui-server/src/brain/client.ts:263-265). The test asserts the rejection arrives only after the child stops answering signal 0." + "holdEnd": "SIGKILL: the client's execBrain passes killSignal SIGKILL with the abort signal (packages/ui-server/src/brain/client.ts:274-276). The test asserts the rejection arrives only after the child stops answering signal 0." }, "observation": null, "followUp": null diff --git a/packages/ui-react/tests/external-speech-runtime.test.ts b/packages/ui-react/tests/external-speech-runtime.test.ts index 9c17ea6f0..c0b52c8bf 100644 --- a/packages/ui-react/tests/external-speech-runtime.test.ts +++ b/packages/ui-react/tests/external-speech-runtime.test.ts @@ -80,7 +80,14 @@ if (worker) { browser = await chromium.launch({ executablePath, headless: true, args: ["--no-sandbox", "--disable-dev-shm-usage", "--use-fake-device-for-media-stream", "--use-fake-ui-for-media-stream"] }); }, 180_000); - afterAll(async () => { await browser?.close(); server?.stop(true); speechServer?.stop(true); await app?.close(); }); + afterAll(async () => { + const started = performance.now(); + await browser?.close(); server?.stop(true); speechServer?.stop(true); + // The #2457 receipt: app.close() must not wait on queued status metadata. + const closing = performance.now(); + await app?.close(); + process.stderr.write(`[external-speech] teardown: browser ${(closing - started).toFixed(1)} ms, app.close() ${(performance.now() - closing).toFixed(1)} ms\n`); + }); } async function open(proveBackend = false): Promise { diff --git a/packages/ui-server/src/app.ts b/packages/ui-server/src/app.ts index 92db3d4c3..2f9a3835a 100644 --- a/packages/ui-server/src/app.ts +++ b/packages/ui-server/src/app.ts @@ -250,8 +250,10 @@ export interface BrainUiApp { /** * Refuse new HTTP requests and drain admitted handlers before disposing of * their operational state. Stalled request bodies are cancelled on close. - * In-flight brain CLI reads and plain index runs get a 10 s grace period and - * are then cancelled; every other run keeps its existing shutdown behaviour. + * Module metadata still waiting for execution admission is cancelled at + * once. In-flight brain CLI reads, metadata and plain index runs get a 10 s + * grace period and are then cancelled; every other run keeps its existing + * shutdown behaviour. */ close(): Promise; } @@ -449,9 +451,11 @@ export async function createApp(options: CreateAppOptions = {}): Promise intake.close()); await inbox.ready; await intake.reconcile(); - // Aborted when close()'s grace period ends: cancels in-flight shared CLI runs. - const shutdown = new AbortController(); - const brain = createBrainClient({ brainPath: config.brainPath, exec: config.exec, minimumVersion: brainCliMinimum, log: brainLog, boundary, shutdown: shutdown.signal }); + // Aborted when close() begins: cancels module metadata still waiting for + // admission. Aborted when close()'s grace period ends: cancels in-flight + // shared CLI runs, admitted metadata included. + const closeBegun = new AbortController(), shutdown = new AbortController(); + const brain = createBrainClient({ brainPath: config.brainPath, exec: config.exec, minimumVersion: brainCliMinimum, log: brainLog, boundary, closing: closeBegun.signal, shutdown: shutdown.signal }); // After a curated application, a credential-free worker indexed the change // without provider keys (#1648). The credentialed reindex that fills in its // vectors waits for the writer gate; one at a time, and one more if more @@ -810,7 +814,7 @@ export async function createApp(options: CreateAppOptions = {}): Promise brain.modules ? declaredContentRenderers((await brain.modules()).enabled) : undefined, + getContentRenderers: async signal => brain.modules ? declaredContentRenderers((await brain.modules({ signal })).enabled) : undefined, getCronStatus: () => cron.getCronStatus(), isTurnActive: () => host.coordinator.isTurnActive(), // Undefined when the injected consumer cannot be read back (a real OTel @@ -1079,8 +1083,10 @@ export async function createApp(options: CreateAppOptions = {}): Promise shutdown.abort(new Error("Server shutting down")), CLOSE_GRACE_MS); const failures: unknown[] = []; const attempt = async (close: () => void | Promise) => { try { await close(); } catch (error) { failures.push(error); } }; diff --git a/packages/ui-server/src/brain/client.ts b/packages/ui-server/src/brain/client.ts index d0bf97f83..e977c0ee6 100644 --- a/packages/ui-server/src/brain/client.ts +++ b/packages/ui-server/src/brain/client.ts @@ -52,8 +52,8 @@ export interface BrainClient { hygiene?(args: readonly string[]): Promise; /** argv prefix for invoking the brain CLI inside the repo. */ cliCommand(): string[]; - /** Additive settings support; older injected clients may omit it. */ - modules?(): Promise<{ enabled: ConfiguredModule[]; available: unknown[] }>; + /** Additive settings support; older injected clients may omit it. Its signal cancels only before execution admission. */ + modules?(opts?: { signal?: AbortSignal }): Promise<{ enabled: ConfiguredModule[]; available: unknown[] }>; moduleSettings?(name: string, operation: "get" | "save" | "validate" | "preview" | "migrate" | "enable" | "disable" | "action", options?: { values?: Record; revision?: string; action?: string }): Promise; search( query: string, @@ -121,7 +121,7 @@ export function brainCliCommand(brainPath: string): string[] { /** Probe the selected content CLI, keeping unknown versions warning-only by default. */ export async function probeBrainCliVersion(brainPath: string, log: Logger, options: { exec: ExecWrapperConfig; minimumVersion?: string; phase?: string; command?: readonly string[]; signal?: AbortSignal; /** @internal Where the CLI runs: in a maintenance worker while the host execution boundary holds (#1648). */ - boundary?: HostBoundary; outputBytes?: number; assertAuthorized?: () => void }): Promise { + boundary?: HostBoundary; admissionSignal?: AbortSignal; outputBytes?: number; assertAuthorized?: () => void }): Promise { const phase = options.phase ?? "startup"; const requirements = [{ owner: "@schlessera/brain-ui-server", kind: "minimum" as const, declaration: MIN_BRAIN_CLI_VERSION }]; if (options.minimumVersion !== undefined) { @@ -139,10 +139,11 @@ export async function probeBrainCliVersion(brainPath: string, log: Logger, optio let result: VersionProbeResult; const boundary = options.boundary ?? createHostBoundary({ brainPath, exec: options.exec }); try { + options.admissionSignal?.throwIfAborted(); if (boundary.isolated()) { // The CLI is the brain's own code: answered from a maintenance worker. const answer = await boundary.run({ argv: [...command, "--version"], env: subprocessEnv("brainCli", { NO_COLOR: "1" }), - credentials: false, gate: "shared", purpose: "brain CLI version", ...(options.signal ? { signal: options.signal } : {}), + credentials: false, gate: "shared", purpose: "brain CLI version", admissionSignal: options.admissionSignal, ...(options.signal ? { signal: options.signal } : {}), timeoutMs: VERSION_PROBE_TIMEOUT_MS, ...(options.outputBytes ? { outputBytes: options.outputBytes } : {}), ...(options.assertAuthorized ? { assertAuthorized: options.assertAuthorized } : {}) }); result = { stdout: answer.stdout, stderr: answer.stderr, exitCode: answer.timedOut ? null : answer.exitCode, timedOut: answer.timedOut === true, cleanupWarnings: [] }; @@ -156,6 +157,7 @@ export async function probeBrainCliVersion(brainPath: string, log: Logger, optio } } catch (error) { options.signal?.throwIfAborted(); + options.admissionSignal?.throwIfAborted(); unknown(error instanceof Error ? error.message : String(error), "brain CLI version probe failed; continuing"); return; } @@ -214,6 +216,13 @@ export interface BrainClientOptions { * it once `close()`'s grace period ends (#2473). Writers are never cancelled. */ shutdown?: AbortSignal; + /** + * @internal Cancels `modules()` metadata still waiting for execution + * admission when it aborts (#2457). `createApp()` aborts it as soon as + * `close()` begins, so queued metadata never holds close open. An admitted + * metadata run keeps running until `shutdown` cancels it. + */ + closing?: AbortSignal; } export function createBrainClient(opts: BrainClientOptions): BrainClient { @@ -228,16 +237,17 @@ export function createBrainClient(opts: BrainClientOptions): BrainClient { throw new Error("searchTimeoutMs must be a positive finite number"); } - async function execBrain(args: string[], signal?: AbortSignal, input?: string, lock?: import("@schlessera/brain-ui-sdk/server").KeyedLock, maintenance?: MaintenanceRequest & { minimumVersion?: string }): Promise { + async function execBrain(args: string[], signal?: AbortSignal, input?: string, lock?: import("@schlessera/brain-ui-sdk/server").KeyedLock, maintenance?: MaintenanceRequest & { minimumVersion?: string; admissionSignal?: AbortSignal }): Promise { const { credentials, gate } = brainCommandClass(args); if (opts.shutdown && gate === "shared" && !maintenance?.bounded) signal = signal ? AbortSignal.any([signal, opts.shutdown]) : opts.shutdown; signal?.throwIfAborted(); + maintenance?.admissionSignal?.throwIfAborted(); if (maintenance?.bounded && !boundary.isolated()) throw new BrainMaintenanceUnavailable("Storage maintenance requires a qualified worker before probing repository code."); const command = brainCliCommand(brainPath); const requiredVersion = maintenance?.minimumVersion ?? minimumVersion; if (requiredVersion !== undefined) { - try { await probeBrainCliVersion(brainPath, log, { exec, minimumVersion: requiredVersion, phase: "invocation", command, signal, boundary, ...(maintenance?.bounded ? { outputBytes: 1024 * 1024 } : {}), ...(maintenance?.assertAuthorized ? { assertAuthorized: maintenance.assertAuthorized } : {}) }); } - catch (error) { signal?.throwIfAborted(); if (maintenance?.minimumVersion) throw new BrainMaintenanceUnavailable("Storage packing requires a qualified core CLI at " + maintenance.minimumVersion + " or newer.", "core_upgrade_or_probe_required"); throw error; } + try { await probeBrainCliVersion(brainPath, log, { exec, minimumVersion: requiredVersion, phase: "invocation", command, signal, boundary, admissionSignal: maintenance?.admissionSignal, ...(maintenance?.bounded ? { outputBytes: 1024 * 1024 } : {}), ...(maintenance?.assertAuthorized ? { assertAuthorized: maintenance.assertAuthorized } : {}) }); } + catch (error) { signal?.throwIfAborted(); maintenance?.admissionSignal?.throwIfAborted(); if (maintenance?.minimumVersion) throw new BrainMaintenanceUnavailable("Storage packing requires a qualified core CLI at " + maintenance.minimumVersion + " or newer.", "core_upgrade_or_probe_required"); throw error; } signal?.throwIfAborted(); } // Every brain CLI launch goes through the host boundary, not only the @@ -255,6 +265,7 @@ export function createBrainClient(opts: BrainClientOptions): BrainClient { credentials, gate, purpose: `brain ${args.slice(0, 2).join(" ")}`, + admissionSignal: maintenance?.admissionSignal, ...(args[0] === "sync" ? { sync: true } : {}), ...(lock ? { lock: { keyed: lock, key: "brain-docs" } } : {}), ...(maintenance?.assertAuthorized ? { assertAuthorized: maintenance.assertAuthorized } : {}), @@ -288,8 +299,9 @@ export function createBrainClient(opts: BrainClientOptions): BrainClient { return brainCliCommand(brainPath); }, - async modules() { - return parseJsonOutput(await execBrain(["module", "list", "--json"])); + async modules(options) { + const admissionSignal = options?.signal && opts.closing ? AbortSignal.any([options.signal, opts.closing]) : options?.signal ?? opts.closing; + return parseJsonOutput(await execBrain(["module", "list", "--json"], undefined, undefined, undefined, { admissionSignal })); }, async moduleSettings(name, operation, options = {}) { diff --git a/packages/ui-server/src/brain/host-boundary.ts b/packages/ui-server/src/brain/host-boundary.ts index c5de2c5ce..3a96aec76 100644 --- a/packages/ui-server/src/brain/host-boundary.ts +++ b/packages/ui-server/src/brain/host-boundary.ts @@ -184,7 +184,10 @@ export interface HostBoundary { /** Both, as `/api/status` reports them. */ status(): GovernedStatus; /** Runs one command to completion, inside a maintenance worker while the boundary is in force. */ - run(execution: BrainExecution): Promise; + run(execution: BrainExecution & { + /** Cancel only before execution admission; a granted owner drains publication and cleanup. */ + admissionSignal?: AbortSignal; + }): Promise; /** The same, with live output, for a streamed sync or a cron job. */ launch(execution: BrainExecution): Promise; /** A short synchronous query (git `check-ignore`). Takes no gate. */ @@ -497,13 +500,17 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary { }, withGate, async run(execution) { - execution.signal?.throwIfAborted(); + const admissionSignal = execution.admissionSignal + ? execution.signal ? AbortSignal.any([execution.signal, execution.admissionSignal]) : execution.admissionSignal + : execution.signal; + const admission = { ...execution, signal: admissionSignal }; + admissionSignal?.throwIfAborted(); execution.assertAuthorized?.(); - return gated(execution, async hold => { - execution.signal?.throwIfAborted(); + return gated(admission, async hold => { + admissionSignal?.throwIfAborted(); if (!isolated()) { - const result = await credentialed(execution, async () => { - execution.signal?.throwIfAborted(); + const result = await credentialed(admission, async () => { + admissionSignal?.throwIfAborted(); // Another shared turn can establish sticky worker history while // credential exclusion waits. Release credentials before rerouting. if (isolated()) return undefined; @@ -528,10 +535,11 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary { }); if (result) return result; } - return credentialed(execution, async () => { + return credentialed(admission, async () => { if (execution.sync) return outputLock.withLock(brain, () => runMirrorSync({ log: options.log, nativeManager: options.nativeManager, folderPaths: hold?.folderPaths, command: command(execution), transportEnv: execution.env, mirror: mirror ??= syncMirrorPath(brain, options.dbPath), - ...(execution.signal ? { signal: execution.signal } : {}), ...(execution.timeoutMs ? { timeoutMs: execution.timeoutMs } : {}) }), { signal: execution.signal }); + ...(execution.signal ? { signal: execution.signal } : {}), ...(execution.timeoutMs ? { timeoutMs: execution.timeoutMs } : {}) }), { signal: admissionSignal }); const run = async () => { + admissionSignal?.throwIfAborted(); execution.assertAuthorized?.(); const view = await prepareInput(execution.signal); let outputs: Awaited> | undefined; @@ -553,7 +561,7 @@ export function createHostBoundary(options: HostBoundaryOptions): HostBoundary { } finally { try { await outputs?.cleanup(); } finally { await view.cleanup(); } } }; return outputLock.withLock(brain, () => execution.lock - ? execution.lock.keyed.withLock(execution.lock.key, run, { signal: execution.signal }) : run(), { signal: execution.signal }); + ? execution.lock.keyed.withLock(execution.lock.key, run, { signal: admissionSignal }) : run(), { signal: admissionSignal }); }, execution.sync === true); }); }, diff --git a/packages/ui-server/src/routes/health.ts b/packages/ui-server/src/routes/health.ts index 978d55980..8490110ba 100644 --- a/packages/ui-server/src/routes/health.ts +++ b/packages/ui-server/src/routes/health.ts @@ -39,7 +39,7 @@ export function createHealthRoutes(deps: { db: Database }): Hono { export interface StatusDeps { /** Trusted metadata from configured modules; absent with an older CLI. */ - getContentRenderers?(): Promise; + getContentRenderers?(signal?: AbortSignal): Promise; /** Git SHA baked at build time (SOURCE_COMMIT), "dev" when unset. */ sourceCommit: string; getCronStatus(): unknown; @@ -60,7 +60,7 @@ export interface StatusDeps { export function createStatusRoutes(deps: StatusDeps): Hono { return new Hono().get("/status", async (c) => { const subscription = deps.getSubscription ? await deps.getSubscription() : undefined; - const contentRenderers = await deps.getContentRenderers?.().catch(() => undefined); + const contentRenderers = await deps.getContentRenderers?.(c.req.raw.signal).catch(() => undefined); return c.json({ ...(contentRenderers !== undefined ? { contentRenderers } : {}), healthy: true, diff --git a/packages/ui-server/src/routes/modules.ts b/packages/ui-server/src/routes/modules.ts index dde18eff7..513d003f8 100644 --- a/packages/ui-server/src/routes/modules.ts +++ b/packages/ui-server/src/routes/modules.ts @@ -9,7 +9,7 @@ export function createModuleRoutes(brain: BrainClient): Hono { const valid = (name: string) => /^[a-z][a-z0-9-]{0,30}$/.test(name); app.get("/modules", async (c) => { if (!brain.modules) return c.json({ error: "Module settings are unavailable with this brain CLI" }, 501); - try { return c.json(await brain.modules()); } + try { return c.json(await brain.modules({ signal: c.req.raw.signal })); } catch (error) { return c.json({ error: (error as Error).message }, 500); } }); app.get("/modules/:name/settings", async (c) => { diff --git a/packages/ui-server/tests/app-metadata-cancellation.test.ts b/packages/ui-server/tests/app-metadata-cancellation.test.ts new file mode 100644 index 000000000..d83efe286 --- /dev/null +++ b/packages/ui-server/tests/app-metadata-cancellation.test.ts @@ -0,0 +1,199 @@ +import { expect, spyOn, test } from "bun:test"; +import { existsSync, mkdtempSync, readFileSync, realpathSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { workerIdentityHelper } from "@schlessera/brain-ui-sdk/internal"; +import type { KeyedLock } from "@schlessera/brain-ui-sdk/server"; +import * as maintenanceOutput from "../src/brain/maintenance-output.js"; +import * as maintenanceWorker from "../../ui-sdk/src/server/maintenance-worker.js"; +import { CLI_LOG, httpContractApp } from "./helpers/http-contract-app.js"; + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise(yes => { resolve = yes; }); + return { promise, resolve }; +} + +test("admitted metadata drains its real worker, publication and cleanup within the close grace period", async () => { + const fixture = await httpContractApp(); + const symbol = Symbol.for("@schlessera/brain-ui-server/maintenance-output-lock"); + const lock = (globalThis as unknown as Record)[symbol]!; + const key = realpathSync(fixture.brainPath); + const launched = deferred(), inputRelease = deferred(), publishing = deferred(), publishRelease = deferred(); + const cleaning = deferred(), cleanupRelease = deferred(); + const events: string[] = []; + let pending: Promise | undefined, closing: Promise | undefined; + let pid = 0, start = ""; + const originalLaunch = maintenanceWorker.launchMaintenanceWorker; + const launch = spyOn(maintenanceWorker, "launchMaintenanceWorker").mockImplementation(input => { + expect(workerIdentityHelper(), "ACTIVE_CONTROL_USES_ORDINARY_NATIVE_LAUNCHER").toBeUndefined(); + const worker = originalLaunch(input); + pid = worker.pid; + const stat = readFileSync(`/proc/${pid}/stat`, "utf8"); + start = stat.slice(stat.lastIndexOf(")") + 2).split(" ")[19]!; + events.push("launched"); + const eof = (name: string) => new TransformStream({ + transform(chunk, controller) { controller.enqueue(chunk); }, flush() { events.push(name); }, + }); + launched.resolve(); + return { ...worker, + stdout: worker.stdout.pipeThrough(eof("stdout EOF")), + stderr: worker.stderr.pipeThrough(eof("stderr EOF")), + exited: worker.exited.then(code => { events.push(`exit ${code}`); return code; }), + async input(data) { await inputRelease.promise; await worker.input(data); events.push("stdin complete"); }, + }; + }); + const originalOutput = maintenanceOutput.prepareMaintenanceOutput; + const output = spyOn(maintenanceOutput, "prepareMaintenanceOutput").mockImplementation(async (...args) => { + const prepared = await originalOutput(...args); + return { ...prepared, + async publish() { + events.push("publication entered"); publishing.resolve(); await publishRelease.promise; + await prepared.publish(); events.push("published"); + }, + async cleanup() { + events.push("cleanup entered"); cleaning.resolve(); await cleanupRelease.promise; + await prepared.cleanup(); events.push("cleaned"); + }, + }; + }); + try { + pending = Promise.resolve(fixture.fetch("/api/status")); + await launched.promise; + expect(pid).toBeGreaterThan(0); + expect(() => process.kill(pid, 0), "ORDINARY_NATIVE_WORKER_EXISTS_AT_CLOSE").not.toThrow(); + expect(events).not.toContain("stdin complete"); + closing = fixture.app.close(); + expect(lock.heldKeys, "ACTIVE_OWNER_RETAINS_OUTPUT_LOCK").toContain(key); + expect(() => fixture.app.db.query("SELECT 1").get()).not.toThrow(); + inputRelease.resolve(); + const publishes = await Promise.race([publishing.promise.then(() => true), cleaning.promise.then(() => false)]); + expect(publishes, "ADMITTED_METADATA_REACHES_PUBLICATION_AFTER_CLOSE").toBe(true); + for (const event of ["exit 0", "stdout EOF", "stderr EOF", "stdin complete"]) expect(events).toContain(event); + expect(lock.heldKeys, "LOCK_HELD_THROUGH_PUBLICATION").toContain(key); + expect(() => fixture.app.db.query("SELECT 1").get()).not.toThrow(); + expect(existsSync(join(fixture.brainPath, CLI_LOG))).toBe(false); + publishRelease.resolve(); + await cleaning.promise; + expect(readFileSync(join(fixture.brainPath, CLI_LOG), "utf8")).toContain('"module","list"'); + expect(lock.heldKeys, "LOCK_HELD_THROUGH_CLEANUP").toContain(key); + expect(() => fixture.app.db.query("SELECT 1").get()).not.toThrow(); + cleanupRelease.resolve(); + const response = await pending; + expect(response.status).toBe(200); + expect(await response.json()).toHaveProperty("contentRenderers", []); + await closing; + expect(events).toContain("cleaned"); + expect(lock.heldKeys).not.toContain(key); + expect(() => fixture.app.db.query("SELECT 1").get()).toThrow(); + console.info("METADATA_ACTIVE_OWNER", JSON.stringify({ pid, start, events })); + } finally { + inputRelease.resolve(); publishRelease.resolve(); cleanupRelease.resolve(); + await pending?.catch(() => {}); await closing?.catch(() => {}); + launch.mockRestore(); output.mockRestore(); + await fixture.close(); + } +}); + +for (const path of ["/api/status", "/api/modules"] as const) { +for (const trigger of ["caller abort", "app close", "loopback abort", "pre-aborted"] as const) { + test(`queued metadata ${path} cancels before output-lock release on ${trigger}`, async () => { + const state = mkdtempSync(join(tmpdir(), "odysseus-metadata-cancel-")); + let fixture: Awaited> | undefined; + let release: (() => void) | undefined; + let restore: (() => void) | undefined; + let pending: Promise | undefined, closing: Promise | undefined; + let timer: ReturnType | undefined; + let server: ReturnType | undefined; + let network: Promise | undefined; + const disconnected = deferred(); + let serverAborted = false; + const arrived = deferred(), caller = new AbortController(); + let bodyEntries = 0, holderReleased = false; + try { + fixture = await httpContractApp({ env: { DB_PATH: join(state, "ui.sqlite") } }); + const key = realpathSync(fixture.brainPath); + const symbol = Symbol.for("@schlessera/brain-ui-server/maintenance-output-lock"); + const lock = (globalThis as unknown as Record)[symbol]!; + expect(lock).toBeDefined(); + release = await lock.acquire(key); + const descriptor = Object.getOwnPropertyDescriptor(lock, "withLock")!; + const original = lock.withLock; + lock.withLock = function(target: string, fn: () => T | Promise, opts?: Parameters[1]) { + const result = original.call(this, target, target === key ? () => { + bodyEntries++; + return fn(); + } : fn, opts) as Promise; + if (target === key) { + console.info("METADATA_LOCK_ARRIVAL", JSON.stringify({ path, trigger, signal: Boolean(opts?.signal), aborted: opts?.signal?.aborted ?? false })); + arrived.resolve(); + } + return result; + }; + restore = () => Object.defineProperty(lock, "withLock", descriptor); + if (trigger === "pre-aborted") caller.abort(); + if (trigger === "loopback abort") { + server = Bun.serve({ hostname: "127.0.0.1", port: 0, fetch(request) { + request.signal.addEventListener("abort", () => { serverAborted = true; disconnected.resolve(); }, { once: true }); + pending = Promise.resolve(fixture!.app.fetch(request)); + return pending; + } }); + network = fetch(`http://127.0.0.1:${server.port}${path}`, { signal: caller.signal }).then( + () => { throw Error("Cancelled loopback request unexpectedly completed"); }, + error => { expect(error.name).toBe("AbortError"); }, + ); + } else pending = Promise.resolve(fixture.fetch(path, { signal: caller.signal })); + if (trigger !== "pre-aborted") await arrived.promise; + expect(bodyEntries, "QUEUED_STATUS_HAS_NOT_ENTERED_OUTPUT_BODY").toBe(0); + expect(() => fixture!.app.db.query("SELECT 1").get(), "HTTP_OWNER_ALIVE_UNTIL_HANDLER_SETTLES").not.toThrow(); + if (trigger === "caller abort" || trigger === "loopback abort") caller.abort(); + else if (trigger === "app close") { + closing = fixture.app.close(); + expect((await fixture.fetch("/api/health")).status, "CLOSING_APP_REFUSES_NEW_HTTP_DISPATCH").toBe(503); + } + const outcome = await Promise.race([ + pending!.then(response => response.status), + new Promise<"still queued">((resolve) => { timer = setTimeout(() => resolve("still queued"), 1500); }), + ]); + clearTimeout(timer); + console.info("METADATA_QUEUE_ORACLE", JSON.stringify({ path, trigger, outcome, holderReleased, bodyEntries })); + expect(outcome, "QUEUED_STATUS_SETTLES_BEFORE_OUTPUT_LOCK_RELEASE").not.toBe("still queued"); + expect(outcome).toBe(path === "/api/status" ? 200 : 500); + // The loopback server has already streamed its response to the client. + if (trigger !== "loopback abort") { + const body = await (await pending!).json() as Record; + if (path === "/api/status") expect(body, "CANCELLED_STATUS_OMITS_RENDERERS").not.toHaveProperty("contentRenderers"); + else expect(body).toHaveProperty("error"); + } + if (trigger === "loopback abort") { + expect(serverAborted, "ACTUAL_BUN_SERVER_REQUEST_ABORT_OBSERVED").toBe(true); + await disconnected.promise; + } + expect(holderReleased).toBe(false); + expect(lock.heldKeys).toContain(key); + expect(bodyEntries).toBe(0); + release(); holderReleased = true; + // A FIFO sentinel proves no abandoned operation can start behind us. + await original.call(lock, key, () => {}); + await pending; + await closing; + expect(bodyEntries, "CANCELLED_STATUS_NEVER_STARTS_AFTER_LOCK_RELEASE").toBe(0); + const log = join(fixture.brainPath, CLI_LOG); + const commands = existsSync(log) ? readFileSync(log, "utf8") : ""; + expect(commands).not.toContain('"module","list"'); + } finally { + clearTimeout(timer); + release?.(); holderReleased = true; + await pending?.catch(() => {}); + await closing?.catch(() => {}); + await network; + server?.stop(true); + restore?.(); + await fixture?.close(); + rmSync(state, { recursive: true, force: true }); + console.info("METADATA_QUEUE_CLEANUP_JOINED", trigger); + } + }); +} + +} diff --git a/packages/ui-server/tests/close-cancels-shared-runs.test.ts b/packages/ui-server/tests/close-cancels-shared-runs.test.ts index 2898f7315..8c3d85a87 100644 --- a/packages/ui-server/tests/close-cancels-shared-runs.test.ts +++ b/packages/ui-server/tests/close-cancels-shared-runs.test.ts @@ -1,6 +1,7 @@ /** * `app.close()` gives in-flight brain CLI runs a short grace period, then - * cancels the shared ones (reads, plain index) and never a writer (#2473). + * cancels the shared ones (reads, metadata, plain index) and never a writer + * (#2473). Metadata still waiting for admission is cancelled at once (#2457). * Each case waits until the fixture CLI is actually running, so none depends * on how long sandbox setup takes. */ @@ -132,3 +133,27 @@ test("the client's shutdown signal never cancels a writer", async () => { shutdown.abort(new Error("Server shutting down")); expect(await add).toMatchObject({ path: ADDED.path }); }, 45_000); + +test("the client's closing signal lets admitted metadata finish", async () => { + const cli = brainWithCli("module", { enabled: [], available: [] }, 1_000); + const closing = new AbortController(), shutdown = new AbortController(); + const brain = createBrainClient({ brainPath: cli.root, closing: closing.signal, shutdown: shutdown.signal }); + const modules = brain.modules!().catch(error => error); + await cli.started(); + // closing cancels only metadata still waiting for admission (#2457). + closing.abort(new Error("Server shutting down")); + expect(await modules, "ADMITTED_METADATA_SURVIVES_CLOSING").toEqual({ enabled: [], available: [] }); +}, 45_000); + +test("the client's shutdown signal cancels admitted metadata with its exact reason", async () => { + const cli = brainWithCli("module", { enabled: [], available: [] }, null); + const closing = new AbortController(), shutdown = new AbortController(); + const brain = createBrainClient({ brainPath: cli.root, closing: closing.signal, shutdown: shutdown.signal }); + const modules = brain.modules!().then(() => "resolved", error => error); + await cli.started(); + closing.abort(new Error("Server shutting down")); + const reason = new Error("Server shutting down"); + shutdown.abort(reason); + const outcome = await Promise.race([modules, Bun.sleep(10_000).then(() => "still running")]); + expect(outcome, "SHUTDOWN_CANCELS_ADMITTED_METADATA").toBe(reason); +}, 45_000); diff --git a/scripts/native-proof/owned-prune-variant.ts b/scripts/native-proof/owned-prune-variant.ts index 1e1aeddfa..553831690 100644 --- a/scripts/native-proof/owned-prune-variant.ts +++ b/scripts/native-proof/owned-prune-variant.ts @@ -20,7 +20,7 @@ export function ownedPruneVariant(cronSource: string, boundarySource: string) { const runStart = boundary.indexOf(" async run(execution) {"), runEnd = boundary.indexOf(" async launch(execution) {", runStart); if (runStart < 0 || runEnd < 0) throw Error("Owned prune fixture run boundary changed"); let run = boundary.slice(runStart, runEnd); - run = replace("boundary", run, " return gated(execution, async hold => {", ' return gated(execution, async hold => {\n fixtureStage("gate:admitted");'); + run = replace("boundary", run, " return gated(admission, async hold => {", ' return gated(admission, async hold => {\n fixtureStage("gate:admitted");'); run = replace("boundary", run, " const run = async () => {", ' const run = async () => {\n fixtureStage("output-lock:admitted");\n fixtureStage("view:start");'); const mark = (before: string, start: string, done: string) => { run = replace("boundary", run, before, ` fixtureStage("${start}");\n${before}\n fixtureStage("${done}");`); }; run = replace("boundary", run, " const view = await prepareInput(execution.signal);", ' const view = await prepareInput(execution.signal);\n fixtureStage("view:done");');