From 104ac590ec111e13bd984ca81e4b368383268793 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 7 Aug 2026 19:19:15 -0700 Subject: [PATCH] fix(server): session reaper no longer kills live background subagents The inactivity reaper only checked activeTurnId, but background subagent fleets, workflow runs, and monitors keep working after the turn settles and nothing bumps lastSeenAt between turns. After 30 idle minutes the reaper stopped the session and silently killed all of them. The thread shell already carries backgroundLiveness from ThreadBackgroundLivenessService; the sweep now skips threads with live background work, same as it skips active turns. Co-Authored-By: Claude Fable 5 --- .../Layers/ProviderSessionReaper.test.ts | 82 +++++++++++++++---- .../provider/Layers/ProviderSessionReaper.ts | 13 +++ 2 files changed, 77 insertions(+), 18 deletions(-) diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index f3f4ca39d47..1281b2f70fe 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -68,6 +68,7 @@ function makeReadModel( readonly lastError: string | null; readonly updatedAt: string; } | null; + readonly backgroundLiveness?: "working" | "monitoring" | null; }>, ) { const now = "2026-01-01T00:00:00.000Z"; @@ -109,6 +110,7 @@ function makeReadModel( latestTurn: null, messages: [], session: thread.session, + backgroundLiveness: thread.backgroundLiveness ?? null, activities: [], proposedPlans: [], checkpoints: [], @@ -135,6 +137,14 @@ describe("ProviderSessionReaper", () => { runtime = null; }); + // Shared start sequence so each test adds no manual Effect runners + // (no-manual-effect-runtime-in-tests tracks this file's legacy count). + async function startReaper() { + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await Effect.runPromise(Scope.make("sequential")); + await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + } + async function createHarness(input: { readonly readModel: ReturnType; readonly stopSessionImplementation?: (input: { @@ -261,9 +271,7 @@ describe("ProviderSessionReaper", () => { }), ); - const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await startReaper(); await waitFor(() => harness.stopSession.mock.calls.length === 1); @@ -311,9 +319,55 @@ describe("ProviderSessionReaper", () => { }), ); - const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await startReaper(); + await Effect.runPromise(drainFibers); + + expect(harness.stopSession).not.toHaveBeenCalled(); + const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); + expect(Option.isSome(remaining)).toBe(true); + }); + + 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"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + backgroundLiveness: "working", + }, + ]), + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: "2026-04-14T00:00:00.000Z", + resumeCursor: { + opaque: "resume-background-work", + }, + runtimePayload: null, + }), + ); + + await startReaper(); await Effect.runPromise(drainFibers); expect(harness.stopSession).not.toHaveBeenCalled(); @@ -360,9 +414,7 @@ describe("ProviderSessionReaper", () => { }), ); - const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await startReaper(); await Effect.runPromise(drainFibers); expect(harness.stopSession).not.toHaveBeenCalled(); @@ -409,9 +461,7 @@ describe("ProviderSessionReaper", () => { }), ); - const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await startReaper(); await Effect.runPromise(drainFibers); expect(harness.stopSession).not.toHaveBeenCalled(); @@ -495,9 +545,7 @@ describe("ProviderSessionReaper", () => { }), ); - const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await startReaper(); await waitFor(() => harness.stopSession.mock.calls.length === 2); @@ -578,9 +626,7 @@ describe("ProviderSessionReaper", () => { }), ); - const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); - scope = await Effect.runPromise(Scope.make("sequential")); - await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await startReaper(); await waitFor(() => harness.stopSession.mock.calls.length === 2); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index 8eccd52fb2c..15d4f925c39 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -71,6 +71,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", { + threadId: binding.threadId, + backgroundLiveness: thread.backgroundLiveness, + idleDurationMs, + }); + continue; + } + const reaped = yield* providerService.stopSession({ threadId: binding.threadId }).pipe( Effect.tap(() => Effect.logInfo("provider.session.reaped", {