diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 2ccdd862522..c2be2c046f1 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,18 @@ 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. + */ + readonly touchLastSeen: ( + input: TouchLastSeenInput, + ) => Effect.Effect; + /** * Delete provider runtime state by canonical thread id. */ @@ -235,6 +253,17 @@ export const make = Effect.gen(function* () { `, }); + const touchLastSeenRow = SqlSchema.void({ + Request: TouchLastSeenInput, + execute: ({ threadId, lastSeenAt }) => + sql` + UPDATE provider_session_runtime + SET last_seen_at = ${lastSeenAt} + WHERE thread_id = ${threadId} + AND status != 'stopped' + `, + }); + const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime) => upsertRuntimeRow(runtime).pipe( Effect.mapError( @@ -308,6 +337,17 @@ 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 }, + ), + ), + ); + const deleteByThreadId: ProviderSessionRuntimeRepository["Service"]["deleteByThreadId"] = ( input, ) => @@ -326,6 +366,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 4ae654a5187..4d151ebe058 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.void, }); const validationRuntimeFactory = makeRuntimeFactory(); diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 1385ccbaabe..6ec56d819b0 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.void, }); // 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 ccbbce1759f..cef92f5c36f 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -38,6 +38,7 @@ import * as SqlClient from "effect/unstable/sql/SqlClient"; import { ProviderAdapterRequestError, ProviderAdapterSessionNotFoundError, + ProviderSessionDirectoryPersistenceError, ProviderUnsupportedError, ProviderValidationError, type ProviderAdapterError, @@ -642,6 +643,169 @@ 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); + }), + }); + + 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: {}, + }); + + 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.void; + }), + }); + + 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: {}, + }); + + 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)), +); + 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 2eaaeb8ce3c..b4898acd2aa 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"; @@ -64,6 +65,11 @@ 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; } type ProviderServiceMethod = @@ -213,6 +219,13 @@ 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(); const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => McpSessionRegistry.issueActiveMcpCredential({ threadId, providerInstanceId }).pipe( @@ -281,6 +294,32 @@ 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; + if (threadId === undefined) return; + 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 successful + // write, so a failed touch doesn't suppress the next refresh and leave + // `last_seen_at` stale (which could lead to premature reaping). `catch` + // handles only the error channel, so fiber interrupts still propagate. + yield* directory.touchLastSeen(threadId).pipe( + Effect.tap(() => Effect.sync(() => lastSeenTouchByThread.set(threadId, now))), + Effect.catch(() => Effect.void), + ); + }); + const processRuntimeEvent = ( source: { readonly instanceId: ProviderInstanceId; @@ -293,7 +332,10 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( increment(providerRuntimeEventsTotal, { provider: canonicalEvent.provider, eventType: canonicalEvent.type, - }).pipe(Effect.andThen(publishRuntimeEvent(canonicalEvent))), + }).pipe( + Effect.andThen(refreshLastSeenForActivity(canonicalEvent)), + Effect.andThen(publishRuntimeEvent(canonicalEvent)), + ), ), ); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index 079b7f10ebf..ca4e4eaa8c4 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -196,6 +196,66 @@ 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, + }); + + yield* directory.touchLastSeen(threadId); + + 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, + }); + + yield* directory.touchLastSeen(threadId); + + 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("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 23075bd9a06..9c99263c23d 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); + 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/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index f2dd4323f7a..a9be9d04cd3 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. No-op if the row is absent + * or already stopped. + */ + readonly touchLastSeen: ( + threadId: ThreadId, + ) => Effect.Effect; } export class ProviderSessionDirectory extends Context.Service<