diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index d3d768b5384..5b148366e93 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -1828,6 +1828,153 @@ describe("ClaudeAdapterLive", () => { ); }); + it.effect("reports lost background agents when the stream ends cleanly", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEvents: Array = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: THREAD_ID, + input: "spawn agents", + attachments: [], + }); + + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-lost-a", + description: "Pass 5: fix-wave regression", + task_type: "local_agent", + tool_use_id: "toolu_lost_a", + uuid: "task-lost-a-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-lost-b", + description: "Pass 5: blind feature sweep", + task_type: "local_agent", + tool_use_id: "toolu_lost_b", + uuid: "task-lost-b-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + + // The claude child process exiting ends the SDK iterable cleanly. + // This is the incident shape: background agents still live, no error. + harness.query.finish(); + + yield* Effect.yieldNow; + yield* Effect.yieldNow; + yield* Effect.yieldNow; + runtimeEventsFiber.interruptUnsafe(); + + const runtimeError = runtimeEvents.find((event) => event.type === "runtime.error"); + assert.equal(runtimeError?.type, "runtime.error"); + if (runtimeError?.type === "runtime.error") { + assert.match(runtimeError.payload.message, /2 background agents were still running/); + assert.match(runtimeError.payload.message, /Pass 5: fix-wave regression/); + assert.match(runtimeError.payload.message, /Pass 5: blind feature sweep/); + assert.match(runtimeError.payload.message, /sending a new message resumes the session/i); + } + + const stoppedTasks = runtimeEvents.filter( + (event) => event.type === "task.completed" && event.payload.status === "stopped", + ); + assert.deepEqual( + stoppedTasks + .map((event) => (event.type === "task.completed" ? String(event.payload.taskId) : "")) + .sort(), + ["task-lost-a", "task-lost-b"], + ); + + assert.equal( + runtimeEvents.some((event) => event.type === "session.exited"), + true, + ); + assert.equal(yield* adapter.hasSession(THREAD_ID), false); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("reports lost background agents when the stream fails", () => { + const harness = makeHarness(); + return Effect.gen(function* () { + const adapter = yield* ClaudeAdapter; + const runtimeEvents: Array = []; + const runtimeEventsFiber = yield* Stream.runForEach(adapter.streamEvents, (event) => + Effect.sync(() => { + runtimeEvents.push(event); + }), + ).pipe(Effect.forkChild); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: THREAD_ID, + input: "spawn agents", + attachments: [], + }); + + harness.query.emit({ + type: "system", + subtype: "task_started", + task_id: "task-lost-on-failure", + description: "Pass 5: concurrency and consistency", + task_type: "local_agent", + tool_use_id: "toolu_lost_failure", + uuid: "task-lost-failure-uuid", + session_id: "sdk-session", + } as unknown as SDKMessage); + + harness.query.fail(new Error("stream transport died")); + + yield* Effect.yieldNow; + yield* Effect.yieldNow; + yield* Effect.yieldNow; + runtimeEventsFiber.interruptUnsafe(); + + // Trigger-agnostic: a failure exit with live tasks must be as loud as + // a clean one. + const lossReport = runtimeEvents.find( + (event) => + event.type === "runtime.error" && + /background agent was still running/.test(event.payload.message), + ); + assert.equal(lossReport?.type, "runtime.error"); + if (lossReport?.type === "runtime.error") { + assert.match(lossReport.payload.message, /Pass 5: concurrency and consistency/); + } + + const stoppedTask = runtimeEvents.find( + (event) => event.type === "task.completed" && event.payload.status === "stopped", + ); + assert.equal(stoppedTask?.type, "task.completed"); + if (stoppedTask?.type === "task.completed") { + assert.equal(String(stoppedTask.payload.taskId), "task-lost-on-failure"); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + it.effect("keeps Claude stream failure events structural", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 92445522cc4..b133b8d3829 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -3567,6 +3567,34 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( return; } + // Background agents run inside the claude child process; the stream + // ending means that process is gone, and every still-tracked task with + // it. Report the loss before teardown — the previous behavior was a + // silent kill the user only discovered on their next message. The + // structured log captures the exit shape so the upstream trigger (what + // ends the stream minutes after a turn settles) stays measurable. + const lostTasks = Array.from(context.liveTaskIds, (taskId) => ({ + taskId, + description: context.taskAgents.get(taskId)?.description, + })); + if (lostTasks.length > 0) { + yield* Effect.logWarning("claude.session.stream-ended-with-live-tasks", { + threadId: context.session.threadId, + exitKind: Exit.isFailure(exit) ? "failure" : "clean-end", + liveTaskCount: lostTasks.length, + taskDescriptions: lostTasks.map((task) => task.description ?? task.taskId), + sessionStartedAt: context.startedAt, + hadActiveTurn: context.turnState !== undefined, + }); + const taskNames = lostTasks.map((task) => task.description ?? task.taskId).join(", "); + yield* emitRuntimeError( + context, + lostTasks.length === 1 + ? `Claude runtime exited while a background agent was still running: ${taskNames}. The agent was stopped; sending a new message resumes the session.` + : `Claude runtime exited while ${lostTasks.length} background agents were still running: ${taskNames}. The agents were stopped; sending a new message resumes the session.`, + ); + } + if (Exit.isFailure(exit)) { if (isClaudeInterruptedCause(exit.cause)) { if (context.turnState) { @@ -3620,6 +3648,39 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( } context.pendingApprovals.clear(); + // Background agents live inside the claude child process, so no task + // survives its session. Emit the terminal event stopTask would have + // produced (see interruptTurn) so durable UI state settles at teardown + // instead of showing phantom running agents until the next resume. + const drainLiveTasks = Effect.gen(function* () { + for (const taskId of Array.from(context.liveTaskIds)) { + // A task_notification handler suspended mid-yield can resume between + // the snapshot above and this iteration and emit the task's real + // terminal event. `delete` returning false means exactly that — the + // task already settled, and emitting a second, contradictory + // `stopped` row here would overwrite its true final status. + if (!context.liveTaskIds.delete(taskId)) { + continue; + } + const taskStamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "task.completed", + eventId: taskStamp.eventId, + provider: PROVIDER, + createdAt: taskStamp.createdAt, + threadId: context.session.threadId, + ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), + payload: { + taskId: RuntimeTaskId.make(taskId), + status: "stopped", + ...taskLinkageFor(context.taskAgents, taskId), + }, + providerRefs: nativeProviderRefs(context), + }); + } + }); + yield* drainLiveTasks; + if (context.turnState) { yield* completeTurn(context, "interrupted", "Session stopped."); } @@ -3632,6 +3693,12 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( yield* Fiber.interrupt(streamFiber); } + // A task_started handler suspended mid-yield on the stream fiber can + // resume between the first drain and the interrupt above, re-adding a + // task after it was drained. The fiber is dead now, so one more drain + // makes "zero live tasks after teardown" an invariant, not a race. + yield* drainLiveTasks; + yield* Effect.try({ try: () => context.query.close(), catch: (cause) => diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index ccbbce1759f..e163b6a9452 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -642,6 +642,411 @@ it.effect("ProviderServiceLive writes canonical events to the emitting thread se }).pipe(Effect.provide(NodeServices.layer)), ); +it.effect("marks the persisted binding stopped when session.exited flows through the pump", () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + }); + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( + Layer.provide(SqlitePersistenceMemory), + ); + const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); + const providerLayer = makeProviderServiceLive().pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), + Layer.provide(directoryLayer), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ); + + const threadId = asThreadId("thread-session-exit-binding"); + + // Single provide so every reference to the directory/repository layers + // resolves to the same in-memory database instance. + const testLayer = providerLayer.pipe( + Layer.provideMerge(directoryLayer), + Layer.provideMerge(runtimeRepositoryLayer), + ); + + // Adapter-internal exits (stream end, replace, adapter stopAll) never + // pass through ProviderService.stopSession; the pump must still leave + // the persisted binding stopped instead of a ghost `running` row. + yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + yield* ProviderService.ProviderService; + yield* directory.upsert({ + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + threadId, + status: "running", + }); + yield* advanceTestClock(10); + codex.emit({ + eventId: asEventId("evt-session-exited-binding"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:00:00.000Z", + type: "session.exited", + payload: { + reason: "Session stopped", + exitKind: "graceful", + }, + }); + yield* advanceTestClock(20); + + // The pump processes the event on a forked fiber; poll briefly. + let runtime = yield* repository.getByThreadId({ threadId }); + for (let attempt = 0; attempt < 100; attempt++) { + if (Option.isSome(runtime) && runtime.value.status === "stopped") { + break; + } + yield* Effect.yieldNow; + runtime = yield* repository.getByThreadId({ threadId }); + } + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + assert.equal(runtime.value.status, "stopped"); + } + }).pipe(Effect.provide(testLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect( + "does not mark the binding stopped for a stale session.exited while a live session exists", + () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + }); + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( + Layer.provide(SqlitePersistenceMemory), + ); + const directoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(runtimeRepositoryLayer), + ); + const providerLayer = makeProviderServiceLive().pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), + Layer.provide(directoryLayer), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ); + + const threadId = asThreadId("thread-stale-exit-guard"); + + const testLayer = providerLayer.pipe( + Layer.provideMerge(directoryLayer), + Layer.provideMerge(runtimeRepositoryLayer), + ); + + // Replacement race: a new session is already live in the adapter when + // the torn-down session's exit event drains through the pump. The + // binding must stay running — marking it stopped would hide a live + // child process from the reaper and diagnosis tooling. The guard is + // identity-based (session createdAt newer than the exit event): a + // crash-exit for the CURRENT session — which some adapters emit + // without tearing down their session map — must still mark the + // binding stopped. + yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + yield* ProviderService.ProviderService; + yield* codex.adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + runtimeMode: "full-access", + }); + yield* directory.upsert({ + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + threadId, + status: "running", + }); + yield* advanceTestClock(10); + // The fake adapter stamps sessions createdAt 2026-01-01T00:00:00Z; + // an exit event created BEFORE that models the replaced (old) + // session's teardown emission. + codex.emit({ + eventId: asEventId("evt-stale-session-exited"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2025-12-31T23:59:00.000Z", + type: "session.exited", + payload: { + reason: "Session stopped", + exitKind: "graceful", + }, + }); + yield* advanceTestClock(20); + for (let attempt = 0; attempt < 50; attempt++) { + yield* Effect.yieldNow; + } + + const runtime = yield* repository.getByThreadId({ threadId }); + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + assert.equal(runtime.value.status, "running"); + } + }).pipe(Effect.provide(testLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect( + "does not mark the binding stopped when the exit comes from a different provider instance", + () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + }); + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( + Layer.provide(SqlitePersistenceMemory), + ); + const directoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(runtimeRepositoryLayer), + ); + const providerLayer = makeProviderServiceLive().pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), + Layer.provide(directoryLayer), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ); + + const threadId = asThreadId("thread-cross-instance-exit-guard"); + + const testLayer = providerLayer.pipe( + Layer.provideMerge(directoryLayer), + Layer.provideMerge(runtimeRepositoryLayer), + ); + + // Cross-instance replacement: the thread has moved to a session on a + // DIFFERENT provider instance, then the old instance's queued exit + // drains through the pump. The old adapter's listSessions cannot see + // the replacement, so only the instance-id comparison protects the + // fresh binding — it must stay running. + yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + yield* ProviderService.ProviderService; + yield* directory.upsert({ + provider: ProviderDriverKind.make("codex"), + providerInstanceId: ProviderInstanceId.make("codex-replacement"), + threadId, + status: "running", + }); + yield* advanceTestClock(10); + // No session exists in this adapter for the thread (the replacement + // lives elsewhere), and the exit event is stamped with this + // adapter's instance id ("codex") by the pump. + codex.emit({ + eventId: asEventId("evt-cross-instance-session-exited"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:00:10.000Z", + type: "session.exited", + payload: { + reason: "Session stopped", + exitKind: "graceful", + }, + }); + yield* advanceTestClock(20); + for (let attempt = 0; attempt < 50; attempt++) { + yield* Effect.yieldNow; + } + + const runtime = yield* repository.getByThreadId({ threadId }); + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + assert.equal(runtime.value.status, "running"); + } + }).pipe(Effect.provide(testLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect( + "marks the binding stopped for a crash exit even when the adapter still reports the session", + () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + }); + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( + Layer.provide(SqlitePersistenceMemory), + ); + const directoryLayer = ProviderSessionDirectoryLive.pipe( + Layer.provide(runtimeRepositoryLayer), + ); + const providerLayer = makeProviderServiceLive().pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), + Layer.provide(directoryLayer), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ); + + const threadId = asThreadId("thread-crash-exit-binding"); + + const testLayer = providerLayer.pipe( + Layer.provideMerge(directoryLayer), + Layer.provideMerge(runtimeRepositoryLayer), + ); + + // Codex-style crash: the child process dies, the adapter emits + // session.exited but never removes its session map entry. The exit + // event is NEWER than the session, so the identity guard must not + // treat the dead session as a live replacement. + yield* Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + yield* ProviderService.ProviderService; + yield* codex.adapter.startSession({ + threadId, + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + runtimeMode: "full-access", + }); + yield* directory.upsert({ + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + threadId, + status: "running", + }); + yield* advanceTestClock(10); + codex.emit({ + eventId: asEventId("evt-crash-session-exited"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:00:05.000Z", + type: "session.exited", + payload: { + reason: "codex app-server exited unexpectedly", + exitKind: "crash", + }, + }); + yield* advanceTestClock(20); + + let runtime = yield* repository.getByThreadId({ threadId }); + for (let attempt = 0; attempt < 100; attempt++) { + if (Option.isSome(runtime) && runtime.value.status === "stopped") { + break; + } + yield* Effect.yieldNow; + runtime = yield* repository.getByThreadId({ threadId }); + } + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + assert.equal(runtime.value.status, "stopped"); + } + }).pipe(Effect.provide(testLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("refreshes the binding lastSeenAt when turn activity flows through the pump", () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + }); + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( + Layer.provide(SqlitePersistenceMemory), + ); + const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); + const providerLayer = makeProviderServiceLive().pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), + Layer.provide(directoryLayer), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ); + + const threadId = asThreadId("thread-turn-activity-touch"); + const staleLastSeenAt = "2026-01-01T00:00:00.000Z"; + + const testLayer = providerLayer.pipe( + Layer.provideMerge(directoryLayer), + Layer.provideMerge(runtimeRepositoryLayer), + ); + + // Provider-initiated turns (background task notifications waking the + // agent) never route through sendTurn, so the pump must be the thing + // that keeps lastSeenAt honest — otherwise the reaper reads a thread + // doing autonomous work as idle since the user's last message. + yield* Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + yield* ProviderService.ProviderService; + yield* repository.upsert({ + threadId, + providerName: "codex", + providerInstanceId: codexInstanceId, + adapterKey: "codex", + runtimeMode: "full-access", + status: "running", + lastSeenAt: staleLastSeenAt, + resumeCursor: null, + runtimePayload: null, + }); + yield* advanceTestClock(10); + codex.emit({ + eventId: asEventId("evt-turn-activity-touch"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:00:00.000Z", + type: "turn.completed", + payload: { + state: "completed", + }, + }); + yield* advanceTestClock(20); + + let runtime = yield* repository.getByThreadId({ threadId }); + for (let attempt = 0; attempt < 100; attempt++) { + if (Option.isSome(runtime) && runtime.value.lastSeenAt !== staleLastSeenAt) { + break; + } + yield* Effect.yieldNow; + runtime = yield* repository.getByThreadId({ threadId }); + } + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + assert.notEqual(runtime.value.lastSeenAt, staleLastSeenAt); + // Turn activity must not change the binding status. + assert.equal(runtime.value.status, "running"); + } + }).pipe(Effect.provide(testLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", () => Effect.gen(function* () { const tempDir = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-provider-service-")); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index d0acc1039c3..3b57fa60bef 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -25,6 +25,7 @@ import { type ProviderSession, } from "@t3tools/contracts"; import { causeErrorTag } from "@t3tools/shared/observability"; +import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; @@ -281,10 +282,142 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }); }); + // The event pump is the one funnel every adapter shares, so two binding + // invariants live here. Both are best-effort: a missing binding is a + // no-op and failures must never break event delivery. + // + // 1. `session.exited`: adapter-internal exits (stream end, session + // replace, adapter stopAll) never pass through `stopSession`, which + // left the persisted binding `running` — a ghost row the reaper keeps + // sweeping and diagnosis tooling misreads. Guarded against stale + // exits: if the adapter still holds a live session for the thread + // (a replacement started before the old session's exit event drained), + // the binding must stay `running`. + // 2. Turn and task lifecycle events: `lastSeenAt` previously moved only + // on user-initiated `sendTurn`. Turns the provider starts on its own + // (background task notifications waking the agent) and long-running + // background tasks that heartbeat via `task.progress` never routed + // through it, so a thread doing autonomous work read as idle from the + // moment the user last typed — and got reaped mid-work. Touching the + // binding on these events (throttled — task.progress ticks every few + // seconds) makes the reaper's clock measure actual inactivity, and + // turns the reaper's wedge cap into "no heartbeat at all for the cap + // window", which is the definition of wedged. + const BINDING_TOUCH_THROTTLE_MS = 60_000; + const lastBindingTouchAtMs = new Map(); + + const BINDING_TOUCH_EVENT_TYPES: ReadonlySet = new Set([ + "turn.started", + "turn.completed", + "task.started", + "task.progress", + "task.updated", + "task.completed", + ]); + + const syncBindingOnRuntimeEvent = ( + adapter: ProviderAdapterShape, + event: ProviderRuntimeEvent, + ): Effect.Effect => { + const isSessionExit = event.type === "session.exited"; + const isActivityTouch = BINDING_TOUCH_EVENT_TYPES.has(event.type); + if (!isSessionExit && !isActivityTouch) { + return Effect.void; + } + return Effect.gen(function* () { + const threadKey = String(event.threadId); + if (!isSessionExit) { + const nowMs = yield* Clock.currentTimeMillis; + const lastTouch = lastBindingTouchAtMs.get(threadKey); + if (lastTouch !== undefined && nowMs - lastTouch < BINDING_TOUCH_THROTTLE_MS) { + return; + } + } else { + lastBindingTouchAtMs.delete(threadKey); + } + const binding = Option.getOrUndefined(yield* directory.getBinding(event.threadId)); + const providerInstanceId = binding?.providerInstanceId; + if ( + binding === undefined || + binding.status === "stopped" || + providerInstanceId === undefined + ) { + return; + } + if (isSessionExit) { + // A session.exited that raced a replacement session must not stomp + // the fresh binding. + // + // Cross-instance replacement: the pump stamps every event with the + // instance that emitted it, so a binding owned by a DIFFERENT + // instance means the thread has already moved on — this exit belongs + // to a superseded session and the fresh binding must survive. + // (The emitting adapter's own listSessions cannot see a replacement + // living on another instance, so this check has to come first.) + if ( + event.providerInstanceId !== undefined && + event.providerInstanceId !== providerInstanceId + ) { + return; + } + // Same-instance replacement: thread-level `hasSession` is not enough + // here — a Codex child-process crash emits session.exited without + // tearing down the adapter's map entry, so the dead session itself + // still reports live. Skip only when the adapter holds a session + // created strictly after this exit event — that can only be a + // replacement. NaN-safe on purpose: an unparseable timestamp must + // not count as a replacement, or the ghost `running` row this sync + // exists to prevent comes back. + const activeSessions = yield* adapter.listSessions(); + const current = activeSessions.find((session) => session.threadId === event.threadId); + if (current !== undefined) { + const currentCreatedAtMs = Date.parse(current.createdAt); + const exitCreatedAtMs = Date.parse(event.createdAt); + const replacementIsLive = + Number.isFinite(currentCreatedAtMs) && + Number.isFinite(exitCreatedAtMs) && + currentCreatedAtMs > exitCreatedAtMs; + if (replacementIsLive) { + return; + } + } + } + const lastRuntimeEventAt = yield* nowIso; + // On activity touches `status` is omitted so the directory preserves + // the existing value; the upsert itself refreshes `lastSeenAt`. + yield* directory.upsert({ + threadId: event.threadId, + provider: binding.provider, + providerInstanceId, + ...(isSessionExit ? { status: "stopped" as const } : {}), + runtimePayload: { + ...(isSessionExit ? { activeTurnId: null } : {}), + lastRuntimeEvent: event.type, + lastRuntimeEventAt, + }, + }); + if (!isSessionExit) { + // Spend the throttle only after the touch actually landed — a missing + // binding or a failed upsert must not suppress the next 60s of + // touches for a refresh that never happened. + lastBindingTouchAtMs.set(threadKey, yield* Clock.currentTimeMillis); + } + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("provider.session.binding-sync-failed", { + threadId: event.threadId, + eventType: event.type, + cause, + }), + ), + ); + }; + const processRuntimeEvent = ( source: { readonly instanceId: ProviderInstanceId; readonly provider: ProviderDriverKind; + readonly adapter: ProviderAdapterShape; }, event: ProviderRuntimeEvent, ): Effect.Effect => @@ -293,7 +426,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( increment(providerRuntimeEventsTotal, { provider: canonicalEvent.provider, eventType: canonicalEvent.type, - }).pipe(Effect.andThen(publishRuntimeEvent(canonicalEvent))), + }).pipe( + Effect.andThen(publishRuntimeEvent(canonicalEvent)), + Effect.andThen(syncBindingOnRuntimeEvent(source.adapter, canonicalEvent)), + ), ), ); @@ -336,6 +472,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( { instanceId: id, provider: adapter.provider, + adapter, }, event, ), diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 1281b2f70fe..c957f7bc606 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -18,6 +18,8 @@ import * as Stream from "effect/Stream"; import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import * as ThreadBackgroundLiveness from "../../orchestration/ThreadBackgroundLiveness.ts"; +import { ThreadBackgroundLivenessService } from "../../orchestration/ThreadBackgroundLiveness.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; import { ProviderValidationError } from "../Errors.ts"; @@ -150,6 +152,8 @@ describe("ProviderSessionReaper", () => { readonly stopSessionImplementation?: (input: { readonly threadId: ThreadId; }) => ReturnType; + readonly backgroundLiveness?: ThreadBackgroundLivenessService["Service"]; + readonly backgroundWorkMaxIdleMs?: number; }) { const stoppedThreadIds = new Set(); const stopSession = vi.fn( @@ -196,10 +200,18 @@ describe("ProviderSessionReaper", () => { const layer = makeProviderSessionReaperLive({ inactivityThresholdMs: 1_000, sweepIntervalMs: 60_000, + ...(input.backgroundWorkMaxIdleMs !== undefined + ? { backgroundWorkMaxIdleMs: input.backgroundWorkMaxIdleMs } + : {}), }).pipe( Layer.provideMerge(providerSessionDirectoryLayer), Layer.provideMerge(runtimeRepositoryLayer), Layer.provideMerge(Layer.succeed(ProviderService, providerService)), + Layer.provideMerge( + input.backgroundLiveness !== undefined + ? Layer.succeed(ThreadBackgroundLivenessService, input.backgroundLiveness) + : ThreadBackgroundLiveness.layer, + ), Layer.provideMerge( Layer.succeed(ProjectionSnapshotQuery, { getCommandReadModel: () => Effect.die("unused"), @@ -330,6 +342,10 @@ describe("ProviderSessionReaper", () => { it("skips stale sessions while background work is still live", async () => { const threadId = ThreadId.make("thread-reaper-background-work"); const now = "2026-01-01T00:00:00.000Z"; + // The sweep reads ThreadBackgroundLivenessService directly rather than + // the projected thread shell: the shell's backgroundLiveness is computed + // from the same service at snapshot-build time, so the direct read is + // the same signal without the snapshot staleness. const harness = await createHarness({ readModel: makeReadModel([ { @@ -343,14 +359,23 @@ describe("ProviderSessionReaper", () => { lastError: null, updatedAt: now, }, - backgroundLiveness: "working", }, ]), + backgroundLiveness: { + recordTaskLiveness: () => {}, + clearThreadLiveness: () => {}, + getThreadBackgroundLiveness: () => "working", + }, }); const repository = await runtime!.runPromise( Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), ); + // Stale past the inactivity threshold but inside the wedge cap: live + // background work defers the reap. A session idle beyond the cap is + // reaped even with "live" work — that case is pinned by the wedge-cap + // test below. + const nowMs = await runtime!.runPromise(Clock.currentTimeMillis); await runtime!.runPromise( repository.upsert({ threadId, @@ -359,7 +384,7 @@ describe("ProviderSessionReaper", () => { adapterKey: "claudeAgent", runtimeMode: "full-access", status: "running", - lastSeenAt: "2026-04-14T00:00:00.000Z", + lastSeenAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 10_000)), resumeCursor: { opaque: "resume-background-work", }, @@ -375,6 +400,116 @@ describe("ProviderSessionReaper", () => { expect(Option.isSome(remaining)).toBe(true); }); + it("skips idle sessions while background work is live", async () => { + const threadId = ThreadId.make("thread-reaper-background-work"); + const now = "2026-01-01T00:00:00.000Z"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + backgroundLiveness: { + recordTaskLiveness: () => {}, + clearThreadLiveness: () => {}, + getThreadBackgroundLiveness: (id) => (id === threadId ? "working" : null), + }, + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + // Past the inactivity threshold (1s) but inside the default wedge cap. + const nowMs = await runtime!.runPromise(Clock.currentTimeMillis); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 10_000)), + resumeCursor: { + opaque: "resume-background-work", + }, + runtimePayload: null, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await runtime!.runPromise(Scope.make("sequential")); + await runtime!.runPromise(reaper.start().pipe(Scope.provide(scope))); + await runtime!.runPromise(drainFibers); + + expect(harness.stopSession).not.toHaveBeenCalled(); + }); + + it("reaps sessions with live background work once past the wedge cap", async () => { + const threadId = ThreadId.make("thread-reaper-wedged-background-work"); + const now = "2026-01-01T00:00:00.000Z"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + backgroundLiveness: { + recordTaskLiveness: () => {}, + clearThreadLiveness: () => {}, + getThreadBackgroundLiveness: () => "working", + }, + backgroundWorkMaxIdleMs: 5_000, + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + // Idle longer than the wedge cap: "live" background work that never + // reaches a terminal state must not pin the session forever. + const nowMs = await runtime!.runPromise(Clock.currentTimeMillis); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 60_000)), + resumeCursor: { + opaque: "resume-wedged-background-work", + }, + runtimePayload: null, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await runtime!.runPromise(Scope.make("sequential")); + await runtime!.runPromise(reaper.start().pipe(Scope.provide(scope))); + + await waitFor(() => harness.stopSession.mock.calls.length === 1); + expect(harness.stoppedThreadIds.has(threadId)).toBe(true); + }); + it("does not reap sessions that are still within the inactivity threshold", async () => { const threadId = ThreadId.make("thread-reaper-fresh"); const now = DateTime.formatIso(await Effect.runPromise(DateTime.now)); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index 15d4f925c39..6ed4b28001d 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -6,6 +6,7 @@ import * as Option from "effect/Option"; import * as Schedule from "effect/Schedule"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { ThreadBackgroundLivenessService } from "../../orchestration/ThreadBackgroundLiveness.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { ProviderSessionReaper, @@ -16,10 +17,17 @@ import { ProviderService } from "../Services/ProviderService.ts"; const DEFAULT_INACTIVITY_THRESHOLD_MS = 30 * 60 * 1000; const DEFAULT_SWEEP_INTERVAL_MS = 5 * 60 * 1000; +// Live background work (subagent fleets, monitors) runs outside a turn, so +// `activeTurnId` alone reads those sessions as idle. Defer reaping while the +// liveness registry reports work — but only up to this cap: a task that has +// been "live" this long without reaching a terminal state is wedged, and +// holding its session (and child process) forever is worse than reaping it. +const DEFAULT_BACKGROUND_WORK_MAX_IDLE_MS = 4 * 60 * 60 * 1000; export interface ProviderSessionReaperLiveOptions { readonly inactivityThresholdMs?: number; readonly sweepIntervalMs?: number; + readonly backgroundWorkMaxIdleMs?: number; } const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) => @@ -27,12 +35,24 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = const providerService = yield* ProviderService; const directory = yield* ProviderSessionDirectory; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + const threadBackgroundLiveness = yield* ThreadBackgroundLivenessService; const inactivityThresholdMs = Math.max( 1, options?.inactivityThresholdMs ?? DEFAULT_INACTIVITY_THRESHOLD_MS, ); const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS); + // Floor STRICTLY above the inactivity threshold: the deferral branch only + // runs once idleDurationMs >= inactivityThresholdMs, so a cap equal to + // the threshold makes `idleDurationMs < backgroundWorkMaxIdleMs` + // unreachable and silently disables background-work deferral — the exact + // bug this reaper change exists to prevent. The floor is minimal (+1ms) + // on purpose: whether a sweep actually lands inside the deferral window + // is the caller's cadence choice, not something to silently rewrite. + const backgroundWorkMaxIdleMs = Math.max( + inactivityThresholdMs + 1, + options?.backgroundWorkMaxIdleMs ?? DEFAULT_BACKGROUND_WORK_MAX_IDLE_MS, + ); const sweep = Effect.gen(function* () { const bindings = yield* directory.listBindings(); @@ -71,14 +91,19 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = continue; } - // The turn can settle while background work runs on (subagent - // fleets, workflow runs, Monitor watch loops). Those live inside the - // provider process, so stopping the session would kill them silently, - // and nothing bumps lastSeenAt between turns. - if (thread?.backgroundLiveness != null) { - yield* Effect.logDebug("provider.session.reaper.skipped-background-work", { + // Background agents and monitors run on after the turn settles, with + // no activeTurnId and no lastSeenAt refresh. The liveness registry is + // the decaying signal for that work (entries clear on terminal task + // status, session exit, and server restart), so defer reaping while + // it reports work — up to the wedge cap. Info-level on purpose: rare + // and decision-bearing. + const backgroundLiveness = threadBackgroundLiveness.getThreadBackgroundLiveness( + binding.threadId, + ); + if (backgroundLiveness !== null && idleDurationMs < backgroundWorkMaxIdleMs) { + yield* Effect.logInfo("provider.session.reaper.skipped-live-background-work", { threadId: binding.threadId, - backgroundLiveness: thread.backgroundLiveness, + backgroundLiveness, idleDurationMs, }); continue; @@ -138,6 +163,7 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = yield* Effect.logInfo("provider.session.reaper.started", { inactivityThresholdMs, sweepIntervalMs, + backgroundWorkMaxIdleMs, }); }); diff --git a/docs/internals/providers.md b/docs/internals/providers.md index a309d70f03d..7b019cb4a4c 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -66,6 +66,33 @@ synchronization. 3. [`CheckpointReactor`][checkpoint] captures workspace checkpoints on turn start and completion, and performs reverts. +## Session teardown and background work + +Background tasks (subagent fleets, monitors, background shells) live inside the provider's child +process, so no task survives its session. Three rules keep that honest: + +- **Teardown reports the loss.** When a Claude session tears down with tasks still tracked as live + (`liveTaskIds`), the adapter emits a terminal `task.completed(status: "stopped")` per task, and — + when the teardown was caused by the SDK stream ending rather than a user action — a + `runtime.error` naming the lost agents plus a structured `claude.session.stream-ended-with-live-tasks` + log capturing the exit shape. The session still tears down; resume happens through the normal + recovery path on the next message. +- **The binding follows the session.** Every `session.exited` event that flows through + [`ProviderService`][service]'s event pump marks the persisted runtime binding `stopped`, so + adapter-internal exits (stream end, session replace, adapter stopAll) cannot leave ghost + `running` rows behind. A stale exit cannot stomp a replacement session's binding: the pump + skips the sync when the binding is owned by a different provider instance than the one that + emitted the exit, or when the emitting adapter holds a session created after the exit event. + The same pump touches the binding's `lastSeenAt` on turn lifecycle (`turn.started`, + `turn.completed`) and task lifecycle (`task.started`, `task.progress`, `task.updated`, + `task.completed`) events, throttled per thread: turns the provider starts on its own + (background task notifications waking the agent) never route through `sendTurn`, and without + the touch a thread doing autonomous work reads as idle from the user's last message. +- **The reaper respects background work.** `ProviderSessionReaper` skips idle sessions while + `ThreadBackgroundLivenessService` reports live work for the thread, up to a wedge cap + (`backgroundWorkMaxIdleMs`, default 4h) after which a session is reaped anyway — a task that + never reaches a terminal state must not pin its child process forever. + ### Buffered assistant delivery A thread in `buffered` assistant delivery mode accumulates assistant text instead of streaming each