authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-08-08 00:42:14-04:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-07 21:42:14-07:00
log2c7267ad43a05cf3e30343400c76fd9ac47698e7
tree7411a0b973a8dd7ac3405f13b55d6f552c488959
parentdaf8ee0b2f684cda82e8721abe1138b219b1ef12
signature Signed by PGP key B5690EEEBB952194

fix(server): stop the reaper from silently killing live background subagents (#5677)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

2 files changed, 77 insertions(+), 18 deletions(-)

apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+64-18
...@@ -68,6 +68,7 @@ function makeReadModel(...@@ -68,6 +68,7 @@ function makeReadModel(
68 readonly lastError: string | null;68 readonly lastError: string | null;
69 readonly updatedAt: string;69 readonly updatedAt: string;
70 } | null;70 } | null;
71 readonly backgroundLiveness?: "working" | "monitoring" | null;
71 }>,72 }>,
72) {73) {
73 const now = "2026-01-01T00:00:00.000Z";74 const now = "2026-01-01T00:00:00.000Z";
...@@ -109,6 +110,7 @@ function makeReadModel(...@@ -109,6 +110,7 @@ function makeReadModel(
109 latestTurn: null,110 latestTurn: null,
110 messages: [],111 messages: [],
111 session: thread.session,112 session: thread.session,
113 backgroundLiveness: thread.backgroundLiveness ?? null,
112 activities: [],114 activities: [],
113 proposedPlans: [],115 proposedPlans: [],
114 checkpoints: [],116 checkpoints: [],
...@@ -135,6 +137,14 @@ describe("ProviderSessionReaper", () => {...@@ -135,6 +137,14 @@ describe("ProviderSessionReaper", () => {
135 runtime = null;137 runtime = null;
136 });138 });
137139
140 // Shared start sequence so each test adds no manual Effect runners
141 // (no-manual-effect-runtime-in-tests tracks this file's legacy count).
142 async function startReaper() {
143 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));
144 scope = await Effect.runPromise(Scope.make("sequential"));
145 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
146 }
147
138 async function createHarness(input: {148 async function createHarness(input: {
139 readonly readModel: ReturnType<typeof makeReadModel>;149 readonly readModel: ReturnType<typeof makeReadModel>;
140 readonly stopSessionImplementation?: (input: {150 readonly stopSessionImplementation?: (input: {
...@@ -261,9 +271,7 @@ describe("ProviderSessionReaper", () => {...@@ -261,9 +271,7 @@ describe("ProviderSessionReaper", () => {
261 }),271 }),
262 );272 );
263273
264 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));274 await startReaper();
265 scope = await Effect.runPromise(Scope.make("sequential"));
266 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
267275
268 await waitFor(() => harness.stopSession.mock.calls.length === 1);276 await waitFor(() => harness.stopSession.mock.calls.length === 1);
269277
...@@ -311,9 +319,55 @@ describe("ProviderSessionReaper", () => {...@@ -311,9 +319,55 @@ describe("ProviderSessionReaper", () => {
311 }),319 }),
312 );320 );
313321
314 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));322 await startReaper();
315 scope = await Effect.runPromise(Scope.make("sequential"));323 await Effect.runPromise(drainFibers);
316 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));324
325 expect(harness.stopSession).not.toHaveBeenCalled();
326 const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId }));
327 expect(Option.isSome(remaining)).toBe(true);
328 });
329
330 it("skips stale sessions while background work is still live", async () => {
331 const threadId = ThreadId.make("thread-reaper-background-work");
332 const now = "2026-01-01T00:00:00.000Z";
333 const harness = await createHarness({
334 readModel: makeReadModel([
335 {
336 id: threadId,
337 session: {
338 threadId,
339 status: "ready",
340 providerName: "claudeAgent",
341 runtimeMode: "full-access",
342 activeTurnId: null,
343 lastError: null,
344 updatedAt: now,
345 },
346 backgroundLiveness: "working",
347 },
348 ]),
349 });
350 const repository = await runtime!.runPromise(
351 Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),
352 );
353
354 await runtime!.runPromise(
355 repository.upsert({
356 threadId,
357 providerName: "claudeAgent",
358 providerInstanceId: null,
359 adapterKey: "claudeAgent",
360 runtimeMode: "full-access",
361 status: "running",
362 lastSeenAt: "2026-04-14T00:00:00.000Z",
363 resumeCursor: {
364 opaque: "resume-background-work",
365 },
366 runtimePayload: null,
367 }),
368 );
369
370 await startReaper();
317 await Effect.runPromise(drainFibers);371 await Effect.runPromise(drainFibers);
318372
319 expect(harness.stopSession).not.toHaveBeenCalled();373 expect(harness.stopSession).not.toHaveBeenCalled();
...@@ -360,9 +414,7 @@ describe("ProviderSessionReaper", () => {...@@ -360,9 +414,7 @@ describe("ProviderSessionReaper", () => {
360 }),414 }),
361 );415 );
362416
363 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));417 await startReaper();
364 scope = await Effect.runPromise(Scope.make("sequential"));
365 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
366 await Effect.runPromise(drainFibers);418 await Effect.runPromise(drainFibers);
367419
368 expect(harness.stopSession).not.toHaveBeenCalled();420 expect(harness.stopSession).not.toHaveBeenCalled();
...@@ -409,9 +461,7 @@ describe("ProviderSessionReaper", () => {...@@ -409,9 +461,7 @@ describe("ProviderSessionReaper", () => {
409 }),461 }),
410 );462 );
411463
412 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));464 await startReaper();
413 scope = await Effect.runPromise(Scope.make("sequential"));
414 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
415 await Effect.runPromise(drainFibers);465 await Effect.runPromise(drainFibers);
416466
417 expect(harness.stopSession).not.toHaveBeenCalled();467 expect(harness.stopSession).not.toHaveBeenCalled();
...@@ -495,9 +545,7 @@ describe("ProviderSessionReaper", () => {...@@ -495,9 +545,7 @@ describe("ProviderSessionReaper", () => {
495 }),545 }),
496 );546 );
497547
498 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));548 await startReaper();
499 scope = await Effect.runPromise(Scope.make("sequential"));
500 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
501549
502 await waitFor(() => harness.stopSession.mock.calls.length === 2);550 await waitFor(() => harness.stopSession.mock.calls.length === 2);
503551
...@@ -578,9 +626,7 @@ describe("ProviderSessionReaper", () => {...@@ -578,9 +626,7 @@ describe("ProviderSessionReaper", () => {
578 }),626 }),
579 );627 );
580628
581 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));629 await startReaper();
582 scope = await Effect.runPromise(Scope.make("sequential"));
583 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));
584630
585 await waitFor(() => harness.stopSession.mock.calls.length === 2);631 await waitFor(() => harness.stopSession.mock.calls.length === 2);
586632
apps/server/src/provider/Layers/ProviderSessionReaper.ts+13
...@@ -71,6 +71,19 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =...@@ -71,6 +71,19 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =
71 continue;71 continue;
72 }72 }
7373
74 // The turn can settle while background work runs on (subagent
75 // fleets, workflow runs, Monitor watch loops). Those live inside the
76 // provider process, so stopping the session would kill them silently,
77 // and nothing bumps lastSeenAt between turns.
78 if (thread?.backgroundLiveness != null) {
79 yield* Effect.logDebug("provider.session.reaper.skipped-background-work", {
80 threadId: binding.threadId,
81 backgroundLiveness: thread.backgroundLiveness,
82 idleDurationMs,
83 });
84 continue;
85 }
86
74 const reaped = yield* providerService.stopSession({ threadId: binding.threadId }).pipe(87 const reaped = yield* providerService.stopSession({ threadId: binding.threadId }).pipe(
75 Effect.tap(() =>88 Effect.tap(() =>
76 Effect.logInfo("provider.session.reaped", {89 Effect.logInfo("provider.session.reaped", {