authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-09-08 09:27:35+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-08 00:27:35-07:00
log430fbd1ffc98378f6a6a41027ddc6004a79b2e7b
tree9dce8b6e05e09efe2fe4b59be2402c0a46079a18
parent991526383f95eefdd66ffd2853136c6dd34ae008
signature Signed by PGP key B5690EEEBB952194

fix(server): give completed turns a full session idle window (#10689)

2 files changed, 139 insertions(+), 43 deletions(-)

apps/server/src/provider/Layers/ProviderSessionReaper.test.ts+127-41
...@@ -7,7 +7,7 @@ import {...@@ -7,7 +7,7 @@ import {
7 ProviderInstanceId,7 ProviderInstanceId,
8} from "@t3tools/contracts";8} from "@t3tools/contracts";
9import * as Clock from "effect/Clock";9import * as Clock from "effect/Clock";
10import * as DateTime from "effect/DateTime";10import * as Deferred from "effect/Deferred";
11import * as Effect from "effect/Effect";11import * as Effect from "effect/Effect";
12import * as Exit from "effect/Exit";12import * as Exit from "effect/Exit";
13import * as Layer from "effect/Layer";13import * as Layer from "effect/Layer";
...@@ -140,9 +140,36 @@ describe("ProviderSessionReaper", () => {...@@ -140,9 +140,36 @@ describe("ProviderSessionReaper", () => {
140 // Shared start sequence so each test adds no manual Effect runners140 // Shared start sequence so each test adds no manual Effect runners
141 // (no-manual-effect-runtime-in-tests tracks this file's legacy count).141 // (no-manual-effect-runtime-in-tests tracks this file's legacy count).
142 async function startReaper() {142 async function startReaper() {
143 const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper));143 await runtime!.runPromise(
144 scope = await Effect.runPromise(Scope.make("sequential"));144 Effect.gen(function* () {
145 await Effect.runPromise(reaper.start().pipe(Scope.provide(scope)));145 const reaper = yield* ProviderSessionReaper;
146 scope = yield* Scope.make("sequential");
147 yield* reaper.start().pipe(Scope.provide(scope));
148 }),
149 );
150 }
151
152 async function sweepAt(nowMs: number) {
153 await runtime!.runPromise(
154 Effect.gen(function* () {
155 const reaper = yield* ProviderSessionReaper;
156 const clock = yield* Clock.Clock;
157 const swept = yield* Deferred.make<void>();
158 yield* reaper.start().pipe(
159 Effect.provideService(Clock.Clock, {
160 currentTimeMillis: Effect.succeed(nowMs),
161 currentTimeMillisUnsafe: () => nowMs,
162 currentTimeNanos: Effect.succeed(BigInt(nowMs) * 1_000_000n),
163 currentTimeNanosUnsafe: () => BigInt(nowMs) * 1_000_000n,
164 monotonicTimeNanos: clock.monotonicTimeNanos,
165 monotonicTimeNanosUnsafe: () => clock.monotonicTimeNanosUnsafe(),
166 // Reaching the next scheduled sleep proves this sweep has finished.
167 sleep: () => Deferred.succeed(swept, undefined).pipe(Effect.andThen(Effect.never)),
168 }),
169 );
170 yield* Deferred.await(swept);
171 }).pipe(Effect.scoped),
172 );
146 }173 }
147174
148 async function createHarness(input: {175 async function createHarness(input: {
...@@ -383,52 +410,111 @@ describe("ProviderSessionReaper", () => {...@@ -383,52 +410,111 @@ describe("ProviderSessionReaper", () => {
383 expect(Option.isSome(remaining)).toBe(true);410 expect(Option.isSome(remaining)).toBe(true);
384 });411 });
385412
386 it("does not reap sessions that are still within the inactivity threshold", async () => {413 it.each(["ready", "interrupted", "error"] as const)(
387 const threadId = ThreadId.make("thread-reaper-fresh");414 "gives a long turn a full idle window after becoming %s",
388 const now = DateTime.formatIso(await Effect.runPromise(DateTime.now));415 async (status) => {
389 const harness = await createHarness({416 const threadId = ThreadId.make(`thread-reaper-long-turn-${status}`);
390 readModel: makeReadModel([417 const startedAt = "2026-04-14T00:00:00.000Z";
418 const completedAt = "2026-04-14T01:00:00.000Z";
419 const completedAtMs = Date.parse(completedAt);
420 const readModel = makeReadModel([
391 {421 {
392 id: threadId,422 id: threadId,
393 session: {423 session: {
394 threadId,424 threadId,
395 status: "ready",425 status: "running",
396 providerName: "claudeAgent",426 providerName: "claudeAgent",
397 runtimeMode: "full-access",427 runtimeMode: "full-access",
398 activeTurnId: null,428 activeTurnId: TurnId.make("turn-reaper-long"),
399 lastError: null,429 lastError: null,
400 updatedAt: now,430 updatedAt: startedAt,
401 },431 },
402 },432 },
403 ]),433 ]);
404 });434 const harness = await createHarness({ readModel });
405 const repository = await runtime!.runPromise(435 const repository = await runtime!.runPromise(
406 Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),436 Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),
407 );437 );
408438 await runtime!.runPromise(
409 await runtime!.runPromise(439 repository.upsert({
410 repository.upsert({440 threadId,
411 threadId,441 providerName: "claudeAgent",
412 providerName: "claudeAgent",442 providerInstanceId: null,
413 providerInstanceId: null,443 adapterKey: "claudeAgent",
414 adapterKey: "claudeAgent",444 runtimeMode: "full-access",
415 runtimeMode: "full-access",445 status: "running",
416 status: "running",446 lastSeenAt: startedAt,
417 lastSeenAt: now,447 resumeCursor: { opaque: "resume-long-turn" },
418 resumeCursor: {448 runtimePayload: null,
419 opaque: "resume-fresh",449 }),
420 },450 );
421 runtimePayload: null,451
422 }),452 await sweepAt(completedAtMs);
423 );453 expect(harness.stopSession).not.toHaveBeenCalled();
424454
425 await startReaper();455 // Ingestion changes the timestamp and clears the active turn in the same session row.
426 await Effect.runPromise(drainFibers);456 readModel.threads[0]!.session = {
427457 ...readModel.threads[0]!.session!,
428 expect(harness.stopSession).not.toHaveBeenCalled();458 status,
429 const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId }));459 activeTurnId: null,
430 expect(Option.isSome(remaining)).toBe(true);460 updatedAt: completedAt,
431 });461 };
462 await sweepAt(completedAtMs);
463 expect(harness.stopSession).not.toHaveBeenCalled();
464 await sweepAt(completedAtMs + 999);
465 expect(harness.stopSession).not.toHaveBeenCalled();
466 await sweepAt(completedAtMs + 1_000);
467 expect(harness.stopSession).toHaveBeenCalledExactlyOnceWith({ threadId });
468 },
469 );
470
471 it.each([true, false])(
472 "uses the binding idle window when the session is older or missing, hasSession=%s",
473 async (hasSession) => {
474 const threadId = ThreadId.make("thread-reaper-fresh");
475 const now = "2026-04-14T01:00:00.000Z";
476 const nowMs = Date.parse(now);
477 const harness = await createHarness({
478 readModel: makeReadModel([
479 {
480 id: threadId,
481 session: hasSession
482 ? {
483 threadId,
484 status: "ready",
485 providerName: "claudeAgent",
486 runtimeMode: "full-access",
487 activeTurnId: null,
488 lastError: null,
489 updatedAt: "2026-04-14T00:00:00.000Z",
490 }
491 : null,
492 },
493 ]),
494 });
495 const repository = await runtime!.runPromise(
496 Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository),
497 );
498 await runtime!.runPromise(
499 repository.upsert({
500 threadId,
501 providerName: "claudeAgent",
502 providerInstanceId: null,
503 adapterKey: "claudeAgent",
504 runtimeMode: "full-access",
505 status: "running",
506 lastSeenAt: now,
507 resumeCursor: { opaque: "resume-fresh" },
508 runtimePayload: null,
509 }),
510 );
511
512 await sweepAt(nowMs + 999);
513 expect(harness.stopSession).not.toHaveBeenCalled();
514 await sweepAt(nowMs + 1_000);
515 expect(harness.stopSession).toHaveBeenCalledExactlyOnceWith({ threadId });
516 },
517 );
432518
433 it("skips persisted sessions that are already marked stopped", async () => {519 it("skips persisted sessions that are already marked stopped", async () => {
434 const threadId = ThreadId.make("thread-reaper-stopped");520 const threadId = ThreadId.make("thread-reaper-stopped");
apps/server/src/provider/Layers/ProviderSessionReaper.ts+12-2
...@@ -54,14 +54,24 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =...@@ -54,14 +54,24 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =
54 continue;54 continue;
55 }55 }
5656
57 const idleDurationMs = now - lastSeenMs;57 if (now - lastSeenMs < inactivityThresholdMs) {
58 if (idleDurationMs < inactivityThresholdMs) {
59 continue;58 continue;
60 }59 }
6160
62 const thread = yield* projectionSnapshotQuery61 const thread = yield* projectionSnapshotQuery
63 .getThreadShellById(binding.threadId)62 .getThreadShellById(binding.threadId)
64 .pipe(Effect.map(Option.getOrUndefined));63 .pipe(Effect.map(Option.getOrUndefined));
64 // Ingestion updates this timestamp alongside activeTurnId when a turn
65 // settles. Long turns must get a full idle window after that transition,
66 // even though the binding was last touched when the turn was sent.
67 const lastActivityMs = Math.max(
68 lastSeenMs,
69 Date.parse(thread?.session?.updatedAt ?? binding.lastSeenAt),
70 );
71 const idleDurationMs = now - lastActivityMs;
72 if (idleDurationMs < inactivityThresholdMs) {
73 continue;
74 }
65 if (thread?.session?.activeTurnId != null) {75 if (thread?.session?.activeTurnId != null) {
66 yield* Effect.logDebug("provider.session.reaper.skipped-active-turn", {76 yield* Effect.logDebug("provider.session.reaper.skipped-active-turn", {
67 threadId: binding.threadId,77 threadId: binding.threadId,