From 55bc5c1a2fdf5111230dc2c1bf1502a2d01c267c Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 12 Nov 2026 15:53:13 -0800 Subject: [PATCH] perf(server): release idle provider sessions after 2 minutes and stop idle Claude process polls Idle sessions now release after 2 minutes instead of 30. ACP-based drivers (acpRegistry, grok, kiro, fx) keep 30 minutes, since their native resume depends on what the external agent supports. Without an agent scope, the Claude query runner now polls the CLI's process tree only from launch or a prompt until the turn's result, then takes one final read. Close and CLI exit still read and stop moved jobs as before. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.ts | 61 +++++++++++---- .../ClaudeAgentSdkQueryRunner.test.ts | 76 +++++++++++++++++++ .../ProviderSessionManager.test.ts | 40 +++++++++- .../ProviderSessionManager.ts | 36 ++++++--- 4 files changed, 188 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 3e10e09906796..5785ca60624c4 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -366,7 +366,7 @@ const CLAUDE_STOP_CONFIRM_TIMEOUT_MS = 10_000; const CLAUDE_STDERR_TAIL_LENGTH = 2_048; // The process group is killed on exit, so its stderr closes promptly. const CLAUDE_STDERR_DRAIN_TIMEOUT = Duration.seconds(1); -// How often the CLI's process tree is recorded when no agent scope will stop what it leaves. +// How often the working CLI's process tree is recorded when no agent scope will stop what it leaves. const CLAUDE_DESCENDANT_POLL = "1 second"; // Claude Code's refusal of bypassPermissions as root, which it explains only on stderr. const CLAUDE_ROOT_REFUSAL = @@ -1286,14 +1286,36 @@ export const layerQueryRunner: Layer.Layer< for (const row of claudeDescendants(rows, roots)) tracked.set(identity(row), row); return running; }); - const stop = refresh(false).pipe( + const refreshRunning = refresh(true).pipe(Effect.asVoid); + let exited = false; + let polling: Fiber.Fiber | undefined; + // Jobs start only while the CLI works, so polling runs from launch or a prompt until a result. + const poll = () => { + if (exited || polling !== undefined) return; + polling = runFork( + refreshRunning.pipe(Effect.repeat(Schedule.spaced(CLAUDE_DESCENDANT_POLL))), + ); + }; + const pause = Effect.suspend(() => { + const fiber = polling; + polling = undefined; + return fiber === undefined ? Effect.void : Fiber.interrupt(fiber); + }); + const settle = pause.pipe( + Effect.andThen(Effect.suspend(() => (exited ? Effect.void : refreshRunning))), + ); + const stop = Effect.suspend(() => { + exited = true; + return pause; + }).pipe( + Effect.andThen(refresh(false)), Effect.map((running) => { for (const row of tracked.values()) { if (running.has(identity(row))) killClaudeDescendant(row); } }), ); - return { refresh: refresh(true).pipe(Effect.asVoid), stop }; + return { refresh: refreshRunning, poll, settle, stop }; }; return ClaudeAgentSdkQueryRunner.of({ @@ -1359,7 +1381,7 @@ export const layerQueryRunner: Layer.Layer< if (answered) return; stderrTail = (stderrTail + text).slice(-CLAUDE_STDERR_TAIL_LENGTH); }; - let refreshDescendants: Effect.Effect = Effect.void; + let descendants: ReturnType | undefined; const spawnProcess = options.spawnClaudeCodeProcess ?? ((spawnOptions: ClaudeSpawnOptions) => { @@ -1375,13 +1397,9 @@ export const layerQueryRunner: Layer.Layer< ); if (unit === undefined && child.pid !== undefined) { const tracker = trackDescendants(child.pid); - const polling = runFork( - tracker.refresh.pipe(Effect.repeat(Schedule.spaced(CLAUDE_DESCENDANT_POLL))), - ); - child.once("exit", () => - runFork(Fiber.interrupt(polling).pipe(Effect.andThen(tracker.stop))), - ); - refreshDescendants = tracker.refresh; + tracker.poll(); + child.once("exit", () => runFork(tracker.stop)); + descendants = tracker; } return child; } @@ -1542,6 +1560,16 @@ export const layerQueryRunner: Layer.Layer< answered = true; }), ), + // The result waits for the turn's last tree read, so a session closed on it keeps that job. + Stream.tap((message) => + descendants === undefined + ? Effect.void + : message.type === "result" + ? descendants.settle + : message.type === "assistant" + ? Effect.sync(descendants.poll) + : Effect.void, + ), Stream.tap((message) => logProtocolEvent({ direction: "incoming", @@ -1553,7 +1581,12 @@ export const layerQueryRunner: Layer.Layer< offer: (message, onQueued) => Effect.uninterruptibleMask((restore) => restore(Queue.offer(promptQueue, message)).pipe( - Effect.tap(() => Effect.sync(() => onQueued?.())), + Effect.tap(() => + Effect.sync(() => { + descendants?.poll(); + onQueued?.(); + }), + ), ), ).pipe( Effect.asVoid, @@ -1691,7 +1724,9 @@ export const layerQueryRunner: Layer.Layer< ), ), // A job started since the last poll is recorded before the CLI is told to exit. - close: Effect.suspend(() => (processExited ? Effect.void : refreshDescendants)).pipe( + close: Effect.suspend(() => + processExited || descendants === undefined ? Effect.void : descendants.refresh, + ).pipe( Effect.andThen(Queue.shutdown(promptQueue)), Effect.andThen(closeClaudeQuery(queryRuntime)), Effect.tap(() => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts index 4716ec06175a6..3823a86f231c8 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAgentSdkQueryRunner.test.ts @@ -18,6 +18,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; import * as Stream from "effect/Stream"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as TestClock from "effect/testing/TestClock"; import { vi } from "vite-plus/test"; @@ -696,6 +697,81 @@ describe.skipIf(HostProcess.Platform.defaultValue() === "win32")("Claude CLI pro }), ); + it.effect("reads the CLI's process tree once when its turn ends and not while it is idle", () => + Effect.gen(function* () { + const cli = fakeClaudeCli( + [ + 'dir=$(dirname "$0")', + `while read -r line; do case "$line" in *'"type":"user"'*) break;; esac; done`, + `perl -e 'setpgrp(0, 0); exec @ARGV' sleep 600 > "$dir/alive" &`, + 'echo $! > "$dir/leftover.pid"', + `echo '{"type":"result","subtype":"success","is_error":false,"result":"done","session_id":"native-session-fake-cli"}'`, + 'read -r _ < "$dir/exit"', + ].join("\n"), + ); + const dir = NodePath.dirname(cli); + NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "alive")]); + NodeChildProcess.execFileSync("mkfifo", [NodePath.join(dir, "exit")]); + yield* Effect.addFinalizer(() => + Effect.sync(() => { + try { + process.kill(Number(NodeFS.readFileSync(NodePath.join(dir, "leftover.pid"), "utf8"))); + } catch {} + NodeFS.rmSync(dir, { recursive: true, force: true }); + }), + ); + const alive = NodeFS.createReadStream(NodePath.join(dir, "alive")); + const leftoverStopped = new Promise((resolve) => alive.once("close", () => resolve())); + alive.resume(); + let listings = 0; + const countingSpawner = Layer.effect( + ChildProcessSpawner.ChildProcessSpawner, + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + return ChildProcessSpawner.make((command) => { + if (command._tag === "StandardCommand" && command.command === "ps") listings += 1; + return spawner.spawn(command); + }); + }), + ).pipe(Layer.provide(NodeServices.layer)); + + const session = yield* openFakeCli(cli).pipe( + Effect.provide( + layerQueryRunner.pipe( + Layer.provide(countingSpawner), + Layer.provide(PortScanner.layerTest), + Layer.provide( + Layer.succeed(ProviderEventLoggers, { native: undefined, canonical: undefined }), + ), + Layer.provide(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ), + ), + ); + const turnEnded = yield* Deferred.make(); + yield* session.messages.pipe( + Stream.tap((message) => + message.type === "result" ? Deferred.succeed(turnEnded, undefined) : Effect.void, + ), + Stream.runDrain, + Effect.ignore, + Effect.forkScoped, + ); + yield* session.offer({ + type: "user", + message: { role: "user", content: "Start a job." }, + parent_tool_use_id: null, + session_id: "native-session-fake-cli", + }); + yield* Deferred.await(turnEnded); + const listingsAtTurnEnd = listings; + yield* TestClock.adjust("1 minute"); + // The CLI exits on its own, so only the turn-end read can have recorded the job. + yield* Effect.promise(() => NodeFS.promises.writeFile(NodePath.join(dir, "exit"), "\n")); + yield* Effect.promise(() => leftoverStopped); + assert.equal(listings, listingsAtTurnEnd + 1); + }), + ); + const rootRefusal = "--dangerously-skip-permissions cannot be used with root/sudo privileges for security reasons"; diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 40c2d28a590a9..04fcf615b4705 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -466,6 +466,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( readonly archive?: Effect.Effect; readonly spawnBeforeOpen?: boolean; readonly scopeCloseReached?: Deferred.Deferred; + readonly driver?: ProviderDriverKind; } = {}, ) { const mcpSessions = yield* McpProviderSessions.McpProviderSessions; @@ -536,7 +537,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( return { instanceId: ProviderInstanceId.make("codex"), - driver: CODEX_DRIVER, + driver: options.driver ?? CODEX_DRIVER, providerSessionId: input.providerSessionId, ...(options.normalizeThreadLoadModelSelection === undefined ? {} @@ -615,7 +616,8 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( function layerTest(input: { readonly state: Ref.Ref; - readonly idleTimeoutMs: number; + readonly idleTimeoutMs?: number; + readonly driver?: ProviderDriverKind; readonly maxIdlePinMs?: number; readonly failEventStream?: boolean; readonly normalizeThreadLoadModelSelection?: (selection: ModelSelection) => ModelSelection; @@ -698,6 +700,7 @@ function layerTest(input: { ...(input.scopeCloseReached === undefined ? {} : { scopeCloseReached: input.scopeCloseReached }), + ...(input.driver === undefined ? {} : { driver: input.driver }), }).pipe(Effect.map(ProviderAdapterRegistry.layerSingle)), ); const layerConfiguredMcpRegistry = @@ -727,7 +730,7 @@ function layerTest(input: { layerConfiguredMcpRegistry, McpProviderSessions.layer, ProviderSessionManager.layerWithOptions({ - idleTimeoutMs: input.idleTimeoutMs, + ...(input.idleTimeoutMs === undefined ? {} : { idleTimeoutMs: input.idleTimeoutMs }), ...(input.maxIdlePinMs === undefined ? {} : { maxIdlePinMs: input.maxIdlePinMs }), }).pipe( Layer.provide( @@ -2014,6 +2017,37 @@ it.effect.each(["session metadata", "session status changes", "provider work"] a }), ); +it.effect.each([ + { driver: CODEX_DRIVER, minutes: 2 }, + { driver: ProviderDriverKind.make("acpRegistry"), minutes: 30 }, +])( + "ProviderSessionManagerV2 releases an idle $driver session after $minutes minutes", + ({ driver, minutes }) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const threadId = ThreadId.make("thread-default-idle-timeout"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId, now: yield* DateTime.now }), + ], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + yield* TestClock.adjust(minutes * 60_000 - 1_000); + assert.isTrue(yield* manager.isLive(providerSessionId)); + yield* TestClock.adjust("1 second"); + assert.isFalse(yield* manager.isLive(providerSessionId)); + }).pipe(Effect.provide(layerTest({ state, driver }))); + }), +); + it.effect("ProviderSessionManagerV2 interrupts event subscriptions on server shutdown", () => Effect.gen(function* () { const state = yield* Ref.make(emptyState); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index ade4668541362..ae33c8ff47528 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -62,7 +62,16 @@ import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; -const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; +const DEFAULT_IDLE_TIMEOUT_MS = 2 * 60 * 1000; +// These resume natively only when the external ACP agent supports it; otherwise a fresh session +// gets a summary of the thread, so they keep the longer timeout. +const UNVERIFIED_RESUME_IDLE_TIMEOUT_MS = 30 * 60 * 1000; +const UNVERIFIED_RESUME_DRIVERS: ReadonlySet = new Set([ + "acpRegistry", + "grok", + "kiro", + "fx", +]); const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000; /** Past adapters' own 10 s stop fallbacks, so they settle first when the provider answers at all. */ @@ -503,7 +512,14 @@ export const layerWithOptions = ( readonly providerSessionId: ProviderSessionId; readonly threadId: ThreadId; }) => `${input.providerSessionId}\u0000${input.threadId}`; - const idleTimeoutMs = Math.max(1, options.idleTimeoutMs ?? DEFAULT_IDLE_TIMEOUT_MS); + const idleTimeoutFor = (runtime: ProviderAdapter.ProviderAdapterV2SessionRuntime) => + Math.max( + 1, + options.idleTimeoutMs ?? + (UNVERIFIED_RESUME_DRIVERS.has(runtime.driver) + ? UNVERIFIED_RESUME_IDLE_TIMEOUT_MS + : DEFAULT_IDLE_TIMEOUT_MS), + ); const maxIdlePinMs = Math.max(0, options.maxIdlePinMs ?? DEFAULT_MAX_IDLE_PIN_MS); interface PreparedMcpCredential { readonly mcpCredentialId: string | undefined; @@ -1390,7 +1406,7 @@ export const layerWithOptions = ( pinnedForMs: now - pinnedSinceMs, }); // Recheck on this fiber; scheduleIdleReleaseInternal would interrupt it and self-deadlock. - yield* Effect.sleep(Duration.millis(idleTimeoutMs)); + yield* Effect.sleep(Duration.millis(idleTimeoutFor(probedRuntime))); return yield* releaseIfStillIdle(input); } yield* Effect.logWarning("orchestration-v2.driver-session.idle-release-pin-expired", { @@ -1459,7 +1475,7 @@ export const layerWithOptions = ( yield* cancelIdleFiber(entry.idleFiber); const generation = entry.idleGeneration + 1; - const idleFiber = yield* Effect.sleep(Duration.millis(idleTimeoutMs)).pipe( + const idleFiber = yield* Effect.sleep(Duration.millis(idleTimeoutFor(entry.runtime))).pipe( Effect.andThen(releaseIfStillIdle({ providerSessionId, generation })), Effect.forkIn(layerScope), ); @@ -1812,7 +1828,7 @@ export const layerWithOptions = ( if (tracked === undefined) return; const [threadId, pending] = tracked; const generation = pending.generation + 1; - const fiber = yield* Effect.sleep(Duration.millis(idleTimeoutMs)).pipe( + const fiber = yield* Effect.sleep(Duration.millis(idleTimeoutFor(entry.runtime))).pipe( Effect.andThen(unloadIdleThread({ providerSessionId, threadId, generation })), Effect.forkIn(layerScope), ); @@ -1890,10 +1906,12 @@ export const layerWithOptions = ( : yield* entry.runtime .hasPendingBackgroundWorkForThread(pending.providerThread) .pipe(Effect.catchCause(() => Effect.succeed(false))); - if (hasPendingWork) return "deferred" as const; + if (hasPendingWork) return { retryAfterMs: idleTimeoutFor(entry.runtime) }; const child = yield* delegatedChildArchive(input.threadId); // Unloading now would miss the archive once the parent takes the result. - if (child === "awaiting_delivery") return "deferred" as const; + if (child === "awaiting_delivery") { + return { retryAfterMs: idleTimeoutFor(entry.runtime) }; + } const archive = child === "ready"; if (input.archiveOnly === true && !archive) return "skipped" as const; const unloading = yield* Ref.modify(sessions, (current) => { @@ -1958,8 +1976,8 @@ export const layerWithOptions = ( }), ); // Wait outside the lock so the next attach is not blocked. - if (outcome === "deferred" && input.archiveOnly !== true) { - yield* Effect.sleep(Duration.millis(idleTimeoutMs)); + if (typeof outcome === "object" && input.archiveOnly !== true) { + yield* Effect.sleep(Duration.millis(outcome.retryAfterMs)); return yield* unloadIdleThread(input); } },