From e578131f84f2def55bdd5224488483df695b4713 Mon Sep 17 00:00:00 2001 From: Zeus-Deus Date: Thu, 30 Jul 2026 16:43:41 +0200 Subject: [PATCH] fix(server): idle reaper no longer kills in-flight background agent work ProviderSessionReaper stopped any provider session after 30 min of last_seen_at inactivity, with session.activeTurnId as its only busy guard. Background work (dynamic workflows, subagents, background shells) outlives the foreground turn, and nothing refreshed last_seen_at from runtime activity - not even later foreground or synthetic follow-up turns - so live sessions were torn down mid-run and their background orchestration silently lost. - Refresh last_seen_at from provider runtime events via a targeted touchLastSeen (throttled per thread, armed only on a successful write of a live row, never resurrecting stopped rows, bounded by a timeout so a stalled write cannot block the event pump). - Track started-but-not-terminal background tasks in memory and have the reaper skip threads with outstanding tasks, even if a task is temporarily quiet; a staleness ceiling (3x the reap threshold of total task-event silence) settles presumed-dead entries so a stranded entry cannot pin a thread forever. - On stopSession/runStopAll/session.exited, publish synthetic terminal task.completed(status: "stopped") events for whatever is still outstanding, so a torn-down task records a terminal activity instead of leaving an orphaned spinner. The activeTurnId guard is preserved unchanged. Regression tests cover the reaper guard (not reaped while a task is outstanding, reapable after it settles), touch semantics, throttling, and the backstop. Closes #4198 Co-authored-by: Chamaru Amasara --- .../Layers/CheckpointReactor.test.ts | 1 + .../Layers/ProviderCommandReactor.test.ts | 1 + .../Layers/ProviderRuntimeIngestion.test.ts | 1 + .../src/persistence/ProviderSessionRuntime.ts | 50 ++ .../src/provider/Layers/CodexAdapter.test.ts | 1 + .../provider/Layers/OpenCodeAdapter.test.ts | 1 + .../provider/Layers/ProviderService.test.ts | 484 ++++++++++++++++++ .../src/provider/Layers/ProviderService.ts | 294 ++++++++++- .../Layers/ProviderSessionDirectory.test.ts | 71 +++ .../Layers/ProviderSessionDirectory.ts | 9 + .../Layers/ProviderSessionReaper.test.ts | 94 +++- .../provider/Layers/ProviderSessionReaper.ts | 10 + .../src/provider/Services/ProviderService.ts | 10 + .../Services/ProviderSessionDirectory.ts | 11 + 14 files changed, 1030 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index 707c87c43c99..86156074604b 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -123,6 +123,7 @@ function createProviderServiceHarness( }, }), rollbackConversation, + hasOutstandingBackgroundTasks: () => Effect.succeed(false), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index e4661061b236..74ff2d915d3c 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -338,6 +338,7 @@ describe("ProviderCommandReactor", () => { }); }, rollbackConversation: () => unsupported(), + hasOutstandingBackgroundTasks: () => Effect.succeed(false), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 74ece50cd318..c766f0aded63 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -121,6 +121,7 @@ function createProviderServiceHarness() { }); }, rollbackConversation: () => unsupported(), + hasOutstandingBackgroundTasks: () => Effect.succeed(false), get streamEvents() { return Stream.fromPubSub(runtimeEventPubSub); }, diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 2ccdd862522f..385b05655834 100644 --- a/apps/server/src/persistence/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -58,6 +58,12 @@ export type GetProviderSessionRuntimeInput = typeof GetProviderSessionRuntimeInp export const DeleteProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId }); export type DeleteProviderSessionRuntimeInput = typeof DeleteProviderSessionRuntimeInput.Type; +export const TouchLastSeenInput = Schema.Struct({ + threadId: ThreadId, + lastSeenAt: IsoDateTime, +}); +export type TouchLastSeenInput = typeof TouchLastSeenInput.Type; + /** * ProviderSessionRuntimeRepository - Service tag for provider runtime persistence. */ @@ -93,6 +99,21 @@ export class ProviderSessionRuntimeRepository extends Context.Service< ProviderSessionRuntimeRepositoryError >; + /** + * Bump only `last_seen_at` for an existing, non-stopped row. + * + * A targeted update used to keep a session's inactivity clock fresh from + * background runtime activity (e.g. a running dynamic workflow) without + * rewriting the full runtime payload. Rows in `stopped` status are left + * untouched so a reaped session is never resurrected. + * + * Returns whether a live row was actually updated, so callers can tell a + * real refresh from a no-op against an absent or stopped row. + */ + readonly touchLastSeen: ( + input: TouchLastSeenInput, + ) => Effect.Effect; + /** * Delete provider runtime state by canonical thread id. */ @@ -226,6 +247,22 @@ export const make = Effect.gen(function* () { `, }); + // `RETURNING` reports which rows the update actually matched: an absent or + // already-stopped row yields none, and the caller needs that distinction to + // avoid treating a no-op as a successful refresh. + const touchLastSeenRow = SqlSchema.findAll({ + Request: TouchLastSeenInput, + Result: Schema.Struct({ threadId: Schema.String }), + execute: ({ threadId, lastSeenAt }) => + sql` + UPDATE provider_session_runtime + SET last_seen_at = ${lastSeenAt} + WHERE thread_id = ${threadId} + AND status != 'stopped' + RETURNING thread_id AS "threadId" + `, + }); + const deleteRuntimeByThreadId = SqlSchema.void({ Request: DeleteRuntimeRequestSchema, execute: ({ threadId }) => @@ -308,6 +345,18 @@ export const make = Effect.gen(function* () { ), ); + const touchLastSeen: ProviderSessionRuntimeRepository["Service"]["touchLastSeen"] = (input) => + touchLastSeenRow(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProviderSessionRuntimeRepository.touchLastSeen:query", + "ProviderSessionRuntimeRepository.touchLastSeen:encodeRequest", + { threadId: input.threadId }, + ), + ), + Effect.map((rows) => rows.length > 0), + ); + const deleteByThreadId: ProviderSessionRuntimeRepository["Service"]["deleteByThreadId"] = ( input, ) => @@ -326,6 +375,7 @@ export const make = Effect.gen(function* () { upsert, getByThreadId, list, + touchLastSeen, deleteByThreadId, } satisfies ProviderSessionRuntimeRepository["Service"]; }); diff --git a/apps/server/src/provider/Layers/CodexAdapter.test.ts b/apps/server/src/provider/Layers/CodexAdapter.test.ts index 7b8fbec56660..5ea3dee45b74 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.test.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.test.ts @@ -218,6 +218,7 @@ const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory getBinding: () => Effect.succeed(Option.none()), listThreadIds: () => Effect.succeed([]), listBindings: () => Effect.succeed([]), + touchLastSeen: () => Effect.succeed(true), }); const validationRuntimeFactory = makeRuntimeFactory(); diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 1385ccbaabec..a79841ad71ae 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -238,6 +238,7 @@ const providerSessionDirectoryTestLayer = Layer.succeed(ProviderSessionDirectory getBinding: () => Effect.succeed(Option.none()), listThreadIds: () => Effect.succeed([]), listBindings: () => Effect.succeed([]), + touchLastSeen: () => Effect.succeed(true), }); // The adapter now receives its settings as a plain argument (the old design diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index ccbbce1759f0..4ed30aca2735 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -16,6 +16,7 @@ import { ProviderDriverKind, ProviderInstanceId, ProviderSessionStartInput, + RuntimeTaskId, ThreadId, TurnId, } from "@t3tools/contracts"; @@ -38,6 +39,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { ProviderAdapterRequestError, ProviderAdapterSessionNotFoundError, + ProviderSessionDirectoryPersistenceError, ProviderUnsupportedError, ProviderValidationError, type ProviderAdapterError, @@ -642,6 +644,488 @@ it.effect("ProviderServiceLive writes canonical events to the emitting thread se }).pipe(Effect.provide(NodeServices.layer)), ); +it.effect("ProviderServiceLive touches lastSeenAt on background runtime activity, throttled", () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + }); + + // Spy directory: records every touchLastSeen call so we assert the wiring + // directly (a runtime event triggers a touch) rather than inferring it from + // DB timestamp noise, and confirms throttling collapses a burst. + const touched: string[] = []; + const spyDirectoryLayer = Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, { + upsert: () => Effect.void, + getProvider: () => Effect.die(new Error("getProvider unused in test")), + getBinding: () => Effect.succeed(Option.none()), + listThreadIds: () => Effect.succeed([]), + listBindings: () => Effect.succeed([]), + touchLastSeen: (threadId) => + Effect.sync(() => { + touched.push(threadId); + return true; + }), + }); + + const threadId = asThreadId("thread-runtime-activity-touch"); + + const providerLayer = makeProviderServiceLive({ + // Tight throttle window so the test can prove both "touches" and + // "throttles a rapid burst" deterministically with the test clock. + runtimeActivityTouchThrottleMs: 1_000, + }).pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), + Layer.provide(spyDirectoryLayer), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ); + + const emitTaskProgress = (id: string) => + codex.emit({ + eventId: asEventId(id), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:05:00.000Z", + type: "task.progress", + payload: { + taskId: RuntimeTaskId.make("task-touch"), + description: "background work", + }, + }); + + yield* Effect.gen(function* () { + yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + // First background event → one touch. + emitTaskProgress("evt-task-progress-1"); + yield* advanceTestClock(20); + assert.deepEqual(touched, [threadId]); + + // Rapid second event within the throttle window → collapsed (no new touch). + emitTaskProgress("evt-task-progress-2"); + yield* advanceTestClock(20); + assert.deepEqual(touched, [threadId]); + + // After the throttle window elapses, a further event touches again. + yield* advanceTestClock(1_000); + emitTaskProgress("evt-task-progress-3"); + yield* advanceTestClock(20); + assert.deepEqual(touched, [threadId, threadId]); + }).pipe(Effect.provide(providerLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect( + "ProviderServiceLive retries lastSeenAt touch after a failed write (throttle not armed on failure)", + () => + Effect.gen(function* () { + const codex = makeFakeCodexAdapter(); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + }); + + // Directory whose first touch fails, then succeeds. A failed touch must not + // arm the throttle window — otherwise a stale last_seen_at would persist and + // the reaper could reap a live session. + const attempts: string[] = []; + let shouldFail = true; + const flakyDirectoryLayer = Layer.succeed(ProviderSessionDirectory.ProviderSessionDirectory, { + upsert: () => Effect.void, + getProvider: () => Effect.die(new Error("getProvider unused in test")), + getBinding: () => Effect.succeed(Option.none()), + listThreadIds: () => Effect.succeed([]), + listBindings: () => Effect.succeed([]), + touchLastSeen: (threadId) => + Effect.suspend(() => { + attempts.push(threadId); + if (shouldFail) { + shouldFail = false; + return Effect.fail( + new ProviderSessionDirectoryPersistenceError({ + operation: "test.touchLastSeen", + detail: "simulated transient touch failure", + }), + ); + } + return Effect.succeed(true); + }), + }); + + const threadId = asThreadId("thread-touch-retry"); + + const providerLayer = makeProviderServiceLive({ + runtimeActivityTouchThrottleMs: 1_000, + }).pipe( + Layer.provide(Layer.succeed(ProviderAdapterRegistry.ProviderAdapterRegistry, registry)), + Layer.provide(flakyDirectoryLayer), + Layer.provide(defaultServerSettingsLayer), + Layer.provide(AnalyticsService.layerTest), + Layer.provide( + Layer.succeed( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + ), + ); + + const emitTaskProgress = (id: string) => + codex.emit({ + eventId: asEventId(id), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:05:00.000Z", + type: "task.progress", + payload: { + taskId: RuntimeTaskId.make("task-retry"), + description: "background work", + }, + }); + + yield* Effect.gen(function* () { + yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + // First event: touch attempted, but the write fails (swallowed). + emitTaskProgress("evt-retry-1"); + yield* advanceTestClock(20); + assert.deepEqual(attempts, [threadId]); + + // Second event *within the throttle window*: because the prior touch + // failed, the window was never armed, so this event retries the touch + // instead of being throttled away. + emitTaskProgress("evt-retry-2"); + yield* advanceTestClock(20); + assert.deepEqual(attempts, [threadId, threadId]); + + // That retry succeeded and armed the window, so a third rapid event is + // now correctly throttled. + emitTaskProgress("evt-retry-3"); + yield* advanceTestClock(20); + assert.deepEqual(attempts, [threadId, threadId]); + }).pipe(Effect.provide(providerLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +// Outstanding-background-task tracking runs against the real directory so +// `stopSession` can resolve the binding `startSession` persisted, and captures +// every published runtime event through the canonical logger. +function makeOutstandingTaskHarness(options?: { readonly outstandingTaskStalenessMs?: number }) { + const codex = makeFakeCodexAdapter(); + // A second instance, so tests can emit an event carrying a provider instance + // id that differs from the one owning a task. + const claude = makeFakeCodexAdapter(CLAUDE_AGENT_DRIVER); + const registry = makeAdapterRegistryMock({ + [ProviderDriverKind.make("codex")]: codex.adapter, + [ProviderDriverKind.make("claudeAgent")]: claude.adapter, + }); + const publishedEvents: ProviderRuntimeEvent[] = []; + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( + Layer.provide(SqlitePersistenceMemory), + ); + const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); + const providerLayer = makeProviderServiceLive({ + ...(options?.outstandingTaskStalenessMs !== undefined + ? { outstandingTaskStalenessMs: options.outstandingTaskStalenessMs } + : {}), + canonicalEventLogger: { + filePath: "memory://provider-outstanding-tasks", + write: (event) => + Effect.sync(() => { + publishedEvents.push(event as ProviderRuntimeEvent); + }), + close: () => Effect.void, + }, + }).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 emitTaskEvent = ( + threadId: ThreadId, + eventId: string, + type: "task.started" | "task.progress" | "task.completed", + payload: Record, + ) => + codex.emit({ + eventId: asEventId(eventId), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:05:00.000Z", + type, + payload, + }); + + const startSession = (threadId: ThreadId) => + Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + return yield* provider.startSession(threadId, { + provider: ProviderDriverKind.make("codex"), + providerInstanceId: codexInstanceId, + threadId, + runtimeMode: "full-access", + }); + }); + + return { codex, claude, providerLayer, publishedEvents, emitTaskEvent, startSession }; +} + +const settledTaskEvents = (events: ReadonlyArray) => + events.filter( + (event) => event.type === "task.completed" && event.payload.status === "stopped", + ) as ReadonlyArray>; + +it.effect("ProviderServiceLive tracks background tasks from started to terminal", () => + Effect.gen(function* () { + const harness = makeOutstandingTaskHarness(); + const threadId = asThreadId("thread-outstanding-lifecycle"); + const taskId = RuntimeTaskId.make("task-lifecycle"); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), false); + + harness.emitTaskEvent(threadId, "evt-task-started", "task.started", { + taskId, + description: "dynamic workflow", + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), true); + + harness.emitTaskEvent(threadId, "evt-task-completed", "task.completed", { + taskId, + status: "completed", + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), false); + }).pipe(Effect.provide(harness.providerLayer)); + + // A real terminal event needs no synthetic stand-in. + assert.equal(settledTaskEvents(harness.publishedEvents).length, 0); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("ProviderServiceLive adopts a background task first seen via task.progress", () => + Effect.gen(function* () { + const harness = makeOutstandingTaskHarness(); + const threadId = asThreadId("thread-outstanding-progress-only"); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + // No task.started was seen — this is what a subscription that attached + // mid-task observes, and it must still pin the session. + harness.emitTaskEvent(threadId, "evt-task-progress-only", "task.progress", { + taskId: RuntimeTaskId.make("task-progress-only"), + description: "already running", + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), true); + }).pipe(Effect.provide(harness.providerLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect( + "ProviderServiceLive settles outstanding background tasks when a session is stopped", + () => + Effect.gen(function* () { + const harness = makeOutstandingTaskHarness(); + const threadId = asThreadId("thread-outstanding-stop"); + const taskId = RuntimeTaskId.make("task-orphaned-by-stop"); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + yield* harness.startSession(threadId); + + harness.emitTaskEvent(threadId, "evt-stop-task-started", "task.started", { + taskId, + description: "background shell", + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), true); + + yield* provider.stopSession({ threadId }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), false); + }).pipe(Effect.provide(harness.providerLayer)); + + const settled = settledTaskEvents(harness.publishedEvents); + assert.equal(settled.length, 1); + assert.equal(settled[0]?.threadId, threadId); + assert.equal(settled[0]?.payload.taskId, taskId); + assert.equal(settled[0]?.payload.summary, "Session stopped before the task finished"); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("ProviderServiceLive settles outstanding background tasks when the session exits", () => + Effect.gen(function* () { + const harness = makeOutstandingTaskHarness(); + const threadId = asThreadId("thread-outstanding-exit"); + const taskId = RuntimeTaskId.make("task-orphaned-by-exit"); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + harness.emitTaskEvent(threadId, "evt-exit-task-started", "task.started", { + taskId, + description: "subagent", + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), true); + + harness.codex.emit({ + eventId: asEventId("evt-session-exited"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:06:00.000Z", + type: "session.exited", + payload: { reason: "provider process exited" }, + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), false); + }).pipe(Effect.provide(harness.providerLayer)); + + const settled = settledTaskEvents(harness.publishedEvents); + assert.equal(settled.length, 1); + assert.equal(settled[0]?.payload.taskId, taskId); + assert.equal(settled[0]?.payload.summary, "Provider session exited before the task finished"); + // The task's terminal event lands before the exit that caused it. + const settledIndex = harness.publishedEvents.indexOf(settled[0]!); + const exitIndex = harness.publishedEvents.findIndex((event) => event.type === "session.exited"); + assert.equal(settledIndex < exitIndex, true); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("ProviderServiceLive tracks outstanding background tasks per thread", () => + Effect.gen(function* () { + const harness = makeOutstandingTaskHarness(); + const busyThreadId = asThreadId("thread-outstanding-busy"); + const idleThreadId = asThreadId("thread-outstanding-idle"); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + harness.emitTaskEvent(busyThreadId, "evt-isolation-started", "task.started", { + taskId: RuntimeTaskId.make("task-isolation"), + description: "long workflow", + }); + yield* advanceTestClock(20); + + // One thread's background work must never pin an unrelated thread. + assert.equal(yield* provider.hasOutstandingBackgroundTasks(busyThreadId), true); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(idleThreadId), false); + }).pipe(Effect.provide(harness.providerLayer)); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("ProviderServiceLive only settles tasks owned by the exiting provider instance", () => + Effect.gen(function* () { + const harness = makeOutstandingTaskHarness(); + const threadId = asThreadId("thread-outstanding-cross-instance"); + const taskId = RuntimeTaskId.make("task-cross-instance"); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + // Task belongs to the codex instance. + harness.emitTaskEvent(threadId, "evt-cross-task-started", "task.started", { + taskId, + description: "codex workflow", + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), true); + + // A different instance exits on the same thread — not this task's owner. + harness.claude.emit({ + eventId: asEventId("evt-cross-exit-other-instance"), + provider: CLAUDE_AGENT_DRIVER, + threadId, + createdAt: "2026-01-01T00:06:00.000Z", + type: "session.exited", + payload: { reason: "other instance exited" }, + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), true); + assert.equal(settledTaskEvents(harness.publishedEvents).length, 0); + + // The owning instance exits — now it settles. + harness.codex.emit({ + eventId: asEventId("evt-cross-exit-owning-instance"), + provider: ProviderDriverKind.make("codex"), + threadId, + createdAt: "2026-01-01T00:07:00.000Z", + type: "session.exited", + payload: { reason: "owning instance exited" }, + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), false); + }).pipe(Effect.provide(harness.providerLayer)); + + const settled = settledTaskEvents(harness.publishedEvents); + assert.equal(settled.length, 1); + assert.equal(settled[0]?.payload.taskId, taskId); + }).pipe(Effect.provide(NodeServices.layer)), +); + +it.effect("ProviderServiceLive settles background tasks that go silent for too long", () => + Effect.gen(function* () { + const harness = makeOutstandingTaskHarness({ outstandingTaskStalenessMs: 60_000 }); + const threadId = asThreadId("thread-outstanding-abandoned"); + const taskId = RuntimeTaskId.make("task-abandoned"); + + yield* Effect.gen(function* () { + const provider = yield* ProviderService.ProviderService; + yield* advanceTestClock(10); + + harness.emitTaskEvent(threadId, "evt-abandoned-started", "task.started", { + taskId, + description: "workflow whose owner vanished", + }); + yield* advanceTestClock(20); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), true); + + // No terminal event will ever arrive (the owning process is gone). Once + // the task has been silent past the ceiling it is presumed abandoned. + yield* advanceTestClock(60_001); + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), false); + + // The entry is gone, so a second call neither reports busy nor settles again. + assert.equal(yield* provider.hasOutstandingBackgroundTasks(threadId), false); + }).pipe(Effect.provide(harness.providerLayer)); + + const settled = settledTaskEvents(harness.publishedEvents); + assert.equal(settled.length, 1); + assert.equal(settled[0]?.threadId, threadId); + assert.equal(settled[0]?.payload.taskId, taskId); + assert.equal( + settled[0]?.payload.summary, + "No activity from the task for an extended period; presumed abandoned", + ); + }).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 ecf26a914c13..b69451ebe0bc 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -10,6 +10,7 @@ * @module ProviderServiceLive */ import { + EventId, ModelSelection, NonNegativeInt, ThreadId, @@ -23,8 +24,10 @@ import { type ProviderDriverKind, type ProviderRuntimeEvent, type ProviderSession, + type RuntimeTaskId, } 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"; @@ -57,6 +60,14 @@ import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import * as McpSessionRegistry from "../../mcp/McpSessionRegistry.ts"; const isModelSelection = Schema.is(ModelSelection); +// Summaries carried by the synthetic terminal events we publish for background +// tasks that were still running when their session went away. They surface in +// the activity feed, so they have to read as an explanation to the user. +const TASK_SETTLED_BY_STOP_SUMMARY = "Session stopped before the task finished"; +const TASK_SETTLED_BY_EXIT_SUMMARY = "Provider session exited before the task finished"; +const TASK_SETTLED_BY_ABANDONMENT_SUMMARY = + "No activity from the task for an extended period; presumed abandoned"; + /** * Hook for tests that want to override the canonical event logger pulled * from `ProviderEventLoggers`. Production wiring leaves this undefined and @@ -64,6 +75,18 @@ const isModelSelection = Schema.is(ModelSelection); */ export interface ProviderServiceLiveOptions { readonly canonicalEventLogger?: EventNdjsonLogger; + /** + * Minimum interval between `lastSeenAt` refreshes triggered by runtime + * activity, per thread. Defaults to 60s. Exposed for tests. + */ + readonly runtimeActivityTouchThrottleMs?: number; + /** + * How long a tracked background task may stay silent before it is presumed + * abandoned and settled. Defaults to 90 minutes: 3x the session reaper's + * default 30-minute inactivity threshold, so a merely slow task is never + * mistaken for a dead one. Exposed for tests. + */ + readonly outstandingTaskStalenessMs?: number; } type ProviderServiceMethod = @@ -199,6 +222,30 @@ const correlateRuntimeEventWithInstance = ( return { ...event, providerInstanceId: source.instanceId }; }; +/** + * A background task we have seen start but not finish. + * + * The owning instance is recorded so a `session.exited` from one provider + * instance cannot settle tasks belonging to a different instance. It does not + * separate successive sessions *within* one instance — `ProviderInstanceId` is + * a stable user-authored config slug, not a per-session id. Ordering within an + * instance comes from its FIFO runtime event queue instead. + */ +interface OutstandingBackgroundTask { + readonly provider: ProviderDriverKind; + readonly providerInstanceId?: ProviderInstanceId; +} + +/** + * Per-thread background-task state. `lastEventAtMs` is the clock reading of the + * most recent `task.*` event on the thread and backs the staleness ceiling in + * `hasOutstandingBackgroundTasks`. + */ +interface OutstandingThreadTasks { + readonly tasks: Map; + lastEventAtMs: number; +} + const makeProviderService = Effect.fn("makeProviderService")(function* ( options?: ProviderServiceLiveOptions, ) { @@ -213,6 +260,30 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( const registry = yield* ProviderAdapterRegistry.ProviderAdapterRegistry; const directory = yield* ProviderSessionDirectory.ProviderSessionDirectory; const runtimeEventPubSub = yield* PubSub.unbounded(); + // Throttle `lastSeenAt` refreshes keyed by thread. A running dynamic workflow + // emits many runtime events per second (`task.progress`, token-usage updates, + // …); we only need to bump the inactivity clock often enough that the session + // reaper never mistakes an active background session for an idle one. One + // write per thread per window is plenty and keeps the DB churn negligible. + const lastSeenTouchMs = options?.runtimeActivityTouchThrottleMs ?? 60_000; + const lastSeenTouchByThread = new Map(); + // Background tasks (dynamic workflows, subagents, background shells) that + // have started but not reached a terminal event, keyed by thread. They + // outlive the foreground turn, so `session.activeTurnId` says nothing about + // them and the reaper needs this as its hard busy guard. + // + // Deliberately in-memory: in-flight background work never survives a server + // restart (the provider child process dies with it), so a persisted entry + // could only ever pin a session that is already gone. + const outstandingTasksByThread = new Map(); + // Ceiling on how long a tracked task may go without emitting anything before + // we presume it dead. This guard fails open by construction — an entry only + // leaves on a terminal event — and there are real paths where that event + // never arrives (an instance rebuilt by a settings edit stops its sessions + // without emitting an exit; events queued behind a stop can re-add a task + // after teardown). Without a ceiling one stranded entry exempts the thread + // from reaping for the lifetime of the process. + const outstandingTaskStalenessMs = options?.outstandingTaskStalenessMs ?? 90 * 60_000; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => McpSessionRegistry.issueActiveMcpCredential({ threadId, providerInstanceId }).pipe( @@ -281,6 +352,176 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( }); }); + // Keep a live session's inactivity clock fresh from runtime activity. This + // matters for work that runs *after* the foreground turn settles — a + // background dynamic workflow / subagents keep emitting runtime events (e.g. + // `task.progress`) while the adapter session has already gone `ready` with no + // active turn. Without this, the session reaper sees a stale `lastSeenAt` and + // tears the session down mid-workflow. Throttled per thread so a burst of + // events is at most one lightweight `last_seen_at` write per window. + const refreshLastSeenForActivity = (event: ProviderRuntimeEvent): Effect.Effect => + Effect.gen(function* () { + const threadId = event.threadId; + const now = yield* Clock.currentTimeMillis; + const previous = lastSeenTouchByThread.get(threadId); + if (previous !== undefined && now - previous < lastSeenTouchMs) return; + // Best-effort: a failed touch must never break event processing (the row + // may be absent/stopped, where the touch is a no-op, or the write may + // transiently fail). Arm the throttle window only after a touch that + // actually updated a live row, so neither a failure nor a no-op leaves + // `last_seen_at` stale behind an armed window (which could lead to + // premature reaping) or accumulates entries for threads that have none. + // + // Bounded because this runs inline in the strictly-sequential per-instance + // event pump: a stalled write would otherwise stall streaming for every + // thread on that instance. A timeout lands in the error channel, so it is + // swallowed below without arming the throttle. `catch` handles only the + // error channel, so fiber interrupts still propagate. + yield* directory.touchLastSeen(threadId).pipe( + Effect.timeout("5 seconds"), + Effect.tap((touched) => + touched ? Effect.sync(() => lastSeenTouchByThread.set(threadId, now)) : Effect.void, + ), + Effect.catch(() => Effect.void), + ); + }); + + const rememberOutstandingTask = ( + threadId: ThreadId, + taskId: RuntimeTaskId, + task: OutstandingBackgroundTask, + nowMs: number, + ): void => { + const existing = outstandingTasksByThread.get(threadId); + if (existing) { + existing.tasks.set(taskId, task); + existing.lastEventAtMs = nowMs; + return; + } + outstandingTasksByThread.set(threadId, { + tasks: new Map([[taskId, task]]), + lastEventAtMs: nowMs, + }); + }; + + const forgetOutstandingTask = ( + threadId: ThreadId, + taskId: RuntimeTaskId, + nowMs: number, + ): void => { + const entry = outstandingTasksByThread.get(threadId); + if (!entry) return; + entry.tasks.delete(taskId); + entry.lastEventAtMs = nowMs; + if (entry.tasks.size === 0) outstandingTasksByThread.delete(threadId); + }; + + // Remove and return the thread's outstanding tasks, optionally restricted by + // `select`. Draining and settling is one step so a task can never be counted + // twice, and callers get exactly the set they are responsible for announcing. + const drainOutstandingTasks = ( + threadId: ThreadId, + select?: (task: OutstandingBackgroundTask) => boolean, + ): ReadonlyArray => { + const entry = outstandingTasksByThread.get(threadId); + if (!entry) return []; + const drained: Array = []; + for (const [taskId, task] of entry.tasks) { + if (select && !select(task)) continue; + entry.tasks.delete(taskId); + drained.push([taskId, task]); + } + if (entry.tasks.size === 0) outstandingTasksByThread.delete(threadId); + return drained; + }; + + // Announce a terminal state for background tasks whose session went away + // before they finished. Without this the task simply stops emitting and the + // activity feed keeps a spinner alive forever; a duplicate terminal activity + // is far cheaper than a silently orphaned one. + const settleOutstandingTasks = ( + threadId: ThreadId, + summary: string, + select?: (task: OutstandingBackgroundTask) => boolean, + ): Effect.Effect => + Effect.gen(function* () { + const drained = drainOutstandingTasks(threadId, select); + if (drained.length === 0) return; + const createdAt = yield* nowIso; + yield* Effect.forEach( + drained, + ([taskId, task]) => + // Counted like any other runtime event: these are real canonical + // events on the bus, not bookkeeping. + increment(providerRuntimeEventsTotal, { + provider: task.provider, + eventType: "task.completed", + }).pipe( + Effect.andThen( + publishRuntimeEvent({ + eventId: EventId.make(`settled:${threadId}:${taskId}:${createdAt}`), + provider: task.provider, + ...(task.providerInstanceId !== undefined + ? { providerInstanceId: task.providerInstanceId } + : {}), + threadId, + createdAt, + type: "task.completed", + payload: { taskId, status: "stopped", summary }, + }), + ), + ), + { discard: true }, + ); + }); + + // Everything we hold per thread is scoped to a live session, so both maps are + // dropped together when the session ends. + const forgetThreadState = (threadId: ThreadId): void => { + lastSeenTouchByThread.delete(threadId); + outstandingTasksByThread.delete(threadId); + }; + + const trackBackgroundTasks = (event: ProviderRuntimeEvent): Effect.Effect => { + switch (event.type) { + case "task.started": + // `task.progress` also adds: a subscription that attaches mid-task never + // saw the `task.started`, and an untracked running task is exactly the + // case the reaper must not miss. + case "task.progress": + return Effect.map(Clock.currentTimeMillis, (nowMs) => + rememberOutstandingTask( + event.threadId, + event.payload.taskId, + { + provider: event.provider, + ...(event.providerInstanceId !== undefined + ? { providerInstanceId: event.providerInstanceId } + : {}), + }, + nowMs, + ), + ); + // The only terminal task event, for every status it can carry. + case "task.completed": + return Effect.map(Clock.currentTimeMillis, (nowMs) => + forgetOutstandingTask(event.threadId, event.payload.taskId, nowMs), + ); + case "session.exited": + return settleOutstandingTasks( + event.threadId, + TASK_SETTLED_BY_EXIT_SUMMARY, + // Fail closed: only the instance that owns a task may settle it, so + // an exit from another instance leaves it alone. This does not order + // successive sessions of the *same* instance (the id is a stable + // config slug); that ordering comes from the instance's FIFO queue. + (task) => task.providerInstanceId === event.providerInstanceId, + ); + default: + return Effect.void; + } + }; + const processRuntimeEvent = ( source: { readonly instanceId: ProviderInstanceId; @@ -293,10 +534,39 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( increment(providerRuntimeEventsTotal, { provider: canonicalEvent.provider, eventType: canonicalEvent.type, - }).pipe(Effect.andThen(publishRuntimeEvent(canonicalEvent))), + }).pipe( + Effect.andThen(refreshLastSeenForActivity(canonicalEvent)), + // Settling runs before the triggering event is published so that a + // task's terminal activity lands ahead of the session exit that + // caused it. + Effect.andThen(trackBackgroundTasks(canonicalEvent)), + Effect.andThen(publishRuntimeEvent(canonicalEvent)), + ), ), ); + const hasOutstandingBackgroundTasks: ProviderServiceMethod<"hasOutstandingBackgroundTasks"> = ( + threadId, + ) => + Effect.gen(function* () { + const entry = outstandingTasksByThread.get(threadId); + if (entry === undefined || entry.tasks.size === 0) return false; + const silentDurationMs = (yield* Clock.currentTimeMillis) - entry.lastEventAtMs; + if (silentDurationMs <= outstandingTaskStalenessMs) return true; + // A live background workflow keeps emitting task events (which also keep + // `lastSeenAt` fresh), so total silence this long means the owning process + // is gone and no terminal event is ever coming. Settle the tasks rather + // than pin the thread forever, and report the session as reapable. + const taskCount = entry.tasks.size; + yield* Effect.logWarning("provider.session.tasks.presumed-abandoned", { + threadId, + taskCount, + silentDurationMs, + }); + yield* settleOutstandingTasks(threadId, TASK_SETTLED_BY_ABANDONMENT_SUMMARY); + return false; + }); + // `subscribedAdapters` is our source-of-truth for "which instance adapters // are currently wired into the runtime event bus". It both tracks the set // of live subscriptions (so `reconcileInstanceSubscriptions` can diff and @@ -855,6 +1125,15 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( if (routed.isActive) { yield* routed.adapter.stopSession(routed.threadId); } + // Settled after the adapter stop so that any terminal event the adapter + // had already handed us wins. This is not a complete drain: adapter + // events reach us through a forked per-instance pump, so `stopSession` + // returning does not mean the queue is empty, and a `task.started` + // still in flight can re-add an entry right after `forgetThreadState`. + // The staleness ceiling in `hasOutstandingBackgroundTasks` is what + // bounds those leftovers. + yield* settleOutstandingTasks(input.threadId, TASK_SETTLED_BY_STOP_SUMMARY); + forgetThreadState(input.threadId); yield* clearMcpSession(input.threadId); yield* directory.upsert({ threadId: input.threadId, @@ -1036,6 +1315,18 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ), ).pipe(Effect.asVoid); yield* Effect.forEach(currentAdapters, ([, adapter]) => adapter.stopAll()).pipe(Effect.asVoid); + // Best-effort by design: `runStopAll` runs from the scope finalizer, and + // ProviderRuntimeIngestion sits above this layer, so by now it has already + // been torn down. These events reach the canonical NDJSON log but not live + // ingestion — no activity row is written. Kept because the log is the + // forensic record, and because it stays correct if layer order changes. + yield* Effect.forEach( + Array.from(outstandingTasksByThread.keys()), + (threadId) => settleOutstandingTasks(threadId, TASK_SETTLED_BY_STOP_SUMMARY), + { discard: true }, + ); + outstandingTasksByThread.clear(); + lastSeenTouchByThread.clear(); yield* McpSessionRegistry.revokeAllActiveMcpCredentials(); McpProviderSession.clearAllMcpProviderSessions(); const bindings = yield* directory.listBindings().pipe(Effect.orElseSucceed(() => [])); @@ -1085,6 +1376,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( getCapabilities, getInstanceInfo, rollbackConversation, + hasOutstandingBackgroundTasks, // Each access creates a fresh PubSub subscription so that multiple // consumers (ProviderRuntimeIngestion, CheckpointReactor, etc.) each // independently receive all runtime events. diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index 079b7f10ebfd..e79d8a39eeb2 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -196,6 +196,77 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL ]); })); + it("touchLastSeen bumps last_seen_at for a live binding", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const threadId = ThreadId.make("thread-touch-live"); + + yield* runtimeRepository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: "2026-01-01T00:00:00.000Z", + resumeCursor: null, + runtimePayload: null, + }); + + const touched = yield* directory.touchLastSeen(threadId); + assert.equal(touched, true); + + const runtime = yield* runtimeRepository.getByThreadId({ threadId }); + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + // A fresh timestamp replaces the stale one; other fields are preserved. + assert.notEqual(runtime.value.lastSeenAt, "2026-01-01T00:00:00.000Z"); + assert.equal(runtime.value.status, "running"); + assert.equal(runtime.value.providerName, "claudeAgent"); + } + })); + + it("touchLastSeen does not resurrect a stopped binding", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const threadId = ThreadId.make("thread-touch-stopped"); + const stoppedAt = "2026-01-01T00:00:00.000Z"; + + yield* runtimeRepository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "stopped", + lastSeenAt: stoppedAt, + resumeCursor: null, + runtimePayload: null, + }); + + // Reported as a no-op so callers do not mistake it for a live refresh. + const touched = yield* directory.touchLastSeen(threadId); + assert.equal(touched, false); + + const runtime = yield* runtimeRepository.getByThreadId({ threadId }); + assert.equal(Option.isSome(runtime), true); + if (Option.isSome(runtime)) { + // Stopped rows are left untouched — the reaper already killed them. + assert.equal(runtime.value.lastSeenAt, stoppedAt); + assert.equal(runtime.value.status, "stopped"); + } + })); + + it("touchLastSeen reports no-op for an absent binding", () => + Effect.gen(function* () { + const directory = yield* ProviderSessionDirectory; + + const touched = yield* directory.touchLastSeen(ThreadId.make("thread-touch-absent")); + assert.equal(touched, false); + })); + it("resets adapterKey to the new provider when provider changes without an explicit adapter key", () => Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 23075bd9a06e..d3aa5dcbaa20 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -182,12 +182,21 @@ const makeProviderSessionDirectory = Effect.gen(function* () { ), ); + const touchLastSeen: ProviderSessionDirectoryShape["touchLastSeen"] = (threadId) => + Effect.gen(function* () { + const now = DateTime.formatIso(yield* DateTime.now); + return yield* repository + .touchLastSeen({ threadId, lastSeenAt: now }) + .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.touchLastSeen"))); + }); + return { upsert, getProvider, getBinding, listThreadIds, listBindings, + touchLastSeen, } satisfies ProviderSessionDirectoryShape; }); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index f3f4ca39d477..b999061dcab8 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -140,8 +140,15 @@ describe("ProviderSessionReaper", () => { readonly stopSessionImplementation?: (input: { readonly threadId: ThreadId; }) => ReturnType; + readonly outstandingTaskThreadIds?: ReadonlyArray; + readonly sweepIntervalMs?: number; }) { const stoppedThreadIds = new Set(); + // Threads the fake provider reports as still running background work. + // Tests mutate this between sweeps to model a task reaching a terminal + // state without touching the reaper's clock. + const threadsWithOutstandingTasks = new Set(input.outstandingTaskThreadIds ?? []); + const threadShellReadsByThread = new Map(); const stopSession = vi.fn( (request) => (input.stopSessionImplementation @@ -174,6 +181,8 @@ describe("ProviderSessionReaper", () => { }); }, rollbackConversation: () => unsupported(), + hasOutstandingBackgroundTasks: (threadId) => + Effect.sync(() => threadsWithOutstandingTasks.has(threadId)), streamEvents: Stream.empty, }; @@ -185,7 +194,7 @@ describe("ProviderSessionReaper", () => { ); const layer = makeProviderSessionReaperLive({ inactivityThresholdMs: 1_000, - sweepIntervalMs: 60_000, + sweepIntervalMs: input.sweepIntervalMs ?? 60_000, }).pipe( Layer.provideMerge(providerSessionDirectoryLayer), Layer.provideMerge(runtimeRepositoryLayer), @@ -205,11 +214,17 @@ describe("ProviderSessionReaper", () => { getThreadCheckpointContext: () => Effect.die("unused"), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: (threadId) => - Effect.succeed( - input.readModel.threads.find((thread) => thread.id === threadId) - ? Option.some(input.readModel.threads.find((thread) => thread.id === threadId)!) - : Option.none(), - ), + Effect.sync(() => { + // The sweep reads the thread shell for every stale binding before + // it consults the outstanding-task guard, so counting these calls + // proves a sweep actually examined the thread. + threadShellReadsByThread.set( + threadId, + (threadShellReadsByThread.get(threadId) ?? 0) + 1, + ); + const thread = input.readModel.threads.find((entry) => entry.id === threadId); + return thread ? Option.some(thread) : Option.none(); + }), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), searchThreads: () => Effect.succeed({ matches: [] }), @@ -219,7 +234,7 @@ describe("ProviderSessionReaper", () => { ); runtime = ManagedRuntime.make(layer); - return { stopSession, stoppedThreadIds }; + return { stopSession, stoppedThreadIds, threadsWithOutstandingTasks, threadShellReadsByThread }; } it("reaps stale persisted sessions without active turns", async () => { @@ -370,6 +385,71 @@ describe("ProviderSessionReaper", () => { expect(Option.isSome(remaining)).toBe(true); }); + it("skips stale sessions that still have outstanding background tasks", async () => { + const threadId = ThreadId.make("thread-reaper-background-tasks"); + const now = "2026-01-01T00:00:00.000Z"; + // Idle past the threshold with no foreground turn — the only thing keeping + // this session alive is background work the projection cannot see. + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + outstandingTaskThreadIds: [threadId], + sweepIntervalMs: 5, + }); + 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-tasks", + }, + 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))); + // Wait until a sweep has actually examined this thread, otherwise the + // negative assertion below would also hold with the guard removed. + await waitFor(() => (harness.threadShellReadsByThread.get(threadId) ?? 0) >= 1); + await runtime!.runPromise(drainFibers); + + // Holds for every sweep that runs while the task is outstanding, so the + // number of sweeps that actually elapsed cannot make this flaky. + expect(harness.stopSession).not.toHaveBeenCalled(); + const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); + expect(Option.isSome(remaining)).toBe(true); + + // The task reaches a terminal state; the session is still idle, so the next + // sweep is free to reap it. + harness.threadsWithOutstandingTasks.delete(threadId); + + await waitFor(() => harness.stopSession.mock.calls.length >= 1); + expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ threadId }); + }); + it("skips persisted sessions that are already marked stopped", async () => { const threadId = ThreadId.make("thread-reaper-stopped"); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index ca396b405969..8f65458a4c70 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -70,6 +70,16 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = continue; } + // Checked immediately before the stop so the window in which a task can + // start and still be killed is as small as we can make it. + if (yield* providerService.hasOutstandingBackgroundTasks(binding.threadId)) { + yield* Effect.logDebug("provider.session.reaper.skipped-outstanding-tasks", { + threadId: binding.threadId, + idleDurationMs, + }); + continue; + } + const reaped = yield* providerService.stopSession({ threadId: binding.threadId }).pipe( Effect.tap(() => Effect.logInfo("provider.session.reaped", { diff --git a/apps/server/src/provider/Services/ProviderService.ts b/apps/server/src/provider/Services/ProviderService.ts index 4d4cb4fa01a7..a94be21ef359 100644 --- a/apps/server/src/provider/Services/ProviderService.ts +++ b/apps/server/src/provider/Services/ProviderService.ts @@ -105,6 +105,16 @@ export interface ProviderServiceShape { readonly numTurns: number; }) => Effect.Effect; + /** + * Whether the thread has background tasks (dynamic workflows, subagents, + * background shells) that started but have not reached a terminal event. + * + * Such work outlives the foreground turn, so `session.activeTurnId` does not + * see it. Callers that tear a session down — the session reaper above all — + * must treat this as busy. + */ + readonly hasOutstandingBackgroundTasks: (threadId: ThreadId) => Effect.Effect; + /** * Canonical provider runtime event stream. * diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index f2dd4323f7a3..1335c99bb910 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -62,6 +62,17 @@ export interface ProviderSessionDirectoryShape { ReadonlyArray, ProviderSessionDirectoryPersistenceError >; + + /** + * Bump only `last_seen_at` for a live (non-stopped) binding. + * + * Used to keep a session's inactivity clock fresh from background runtime + * activity without rewriting the full binding. Resolves `false` when the row + * is absent or already stopped, where the touch is a no-op. + */ + readonly touchLastSeen: ( + threadId: ThreadId, + ) => Effect.Effect; } export class ProviderSessionDirectory extends Context.Service<