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); } },