From b74aaa4e648fd40b5e777e953c2de631721d49aa Mon Sep 17 00:00:00 2001 From: Karl Schulze <32364521+kraptor23@users.noreply.github.com> Date: Thu, 6 Aug 2026 07:45:27 +0000 Subject: [PATCH 1/2] feat(server): make provider session reaper timing configurable The idle session reaper's 30-minute inactivity threshold and 5-minute sweep interval were hardcoded; the only recourse for long-running sessions being stopped between messages was patching the bundle. Add providerSessionInactivityThreshold and providerSessionSweepInterval server settings (millis in settings.json, RPC-patchable) consumed by ProviderSessionReaper. The threshold is re-read every sweep so edits apply live; the sweep interval is fixed when the reaper starts. Explicit ProviderSessionReaperLiveOptions still override both, preserving existing test seams. Co-Authored-By: Claude Fable 5 --- .../Layers/ProviderSessionReaper.test.ts | 192 +++++++++++++++++- .../provider/Layers/ProviderSessionReaper.ts | 44 +++- apps/server/src/serverSettings.ts | 20 +- packages/contracts/src/settings.ts | 18 ++ packages/shared/src/serverSettings.ts | 6 + 5 files changed, 264 insertions(+), 16 deletions(-) diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index f3f4ca39d47..7689b144de0 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -8,6 +8,7 @@ import { } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; @@ -23,8 +24,12 @@ import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntim import { ProviderValidationError } from "../Errors.ts"; import { ProviderSessionReaper } from "../Services/ProviderSessionReaper.ts"; import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; +import { layerTest as serverSettingsLayerTest } from "../../serverSettings.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; -import { makeProviderSessionReaperLive } from "./ProviderSessionReaper.ts"; +import { + makeProviderSessionReaperLive, + type ProviderSessionReaperLiveOptions, +} from "./ProviderSessionReaper.ts"; const defaultModelSelection = { instanceId: ProviderInstanceId.make("codex"), @@ -140,6 +145,8 @@ describe("ProviderSessionReaper", () => { readonly stopSessionImplementation?: (input: { readonly threadId: ThreadId; }) => ReturnType; + readonly reaperOptions?: ProviderSessionReaperLiveOptions; + readonly settingsOverrides?: Parameters[0]; }) { const stoppedThreadIds = new Set(); const stopSession = vi.fn( @@ -183,10 +190,13 @@ describe("ProviderSessionReaper", () => { const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( Layer.provide(runtimeRepositoryLayer), ); - const layer = makeProviderSessionReaperLive({ - inactivityThresholdMs: 1_000, - sweepIntervalMs: 60_000, - }).pipe( + const layer = makeProviderSessionReaperLive( + input.reaperOptions ?? { + inactivityThresholdMs: 1_000, + sweepIntervalMs: 60_000, + }, + ).pipe( + Layer.provideMerge(serverSettingsLayerTest(input.settingsOverrides ?? {})), Layer.provideMerge(providerSessionDirectoryLayer), Layer.provideMerge(runtimeRepositoryLayer), Layer.provideMerge(Layer.succeed(ProviderService, providerService)), @@ -271,6 +281,178 @@ describe("ProviderSessionReaper", () => { expect(harness.stoppedThreadIds.has(threadId)).toBe(true); }); + // These cases seed a session idle for ~5 seconds — between the small and + // large candidate thresholds — so they fail if the settings plumbing is + // broken (hardcoded 30-minute default would not reap) or the precedence + // is inverted. + it("reads reaper timing from server settings when no options are provided", async () => { + const threadId = ThreadId.make("thread-reaper-settings"); + const now = "2026-01-01T00:00:00.000Z"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + reaperOptions: {}, + settingsOverrides: { + providerSessionInactivityThreshold: Duration.millis(1_000), + providerSessionSweepInterval: Duration.minutes(1), + }, + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + const nowMs = await Effect.runPromise(Clock.currentTimeMillis); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: new Date(nowMs - 5_000).toISOString(), + resumeCursor: { + opaque: "resume-settings", + }, + runtimePayload: null, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await Effect.runPromise(Scope.make("sequential")); + await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + + await waitFor(() => harness.stopSession.mock.calls.length === 1); + + expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ threadId }); + expect(harness.stoppedThreadIds.has(threadId)).toBe(true); + }); + + it("does not reap when the configured threshold exceeds the idle time", async () => { + const threadId = ThreadId.make("thread-reaper-settings-fresh"); + const now = "2026-01-01T00:00:00.000Z"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + reaperOptions: {}, + settingsOverrides: { + providerSessionInactivityThreshold: Duration.days(30), + providerSessionSweepInterval: Duration.minutes(1), + }, + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + const nowMs = await Effect.runPromise(Clock.currentTimeMillis); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: new Date(nowMs - 5_000).toISOString(), + resumeCursor: { + opaque: "resume-settings-fresh", + }, + runtimePayload: null, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await Effect.runPromise(Scope.make("sequential")); + await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + await Effect.runPromise(drainFibers); + + expect(harness.stopSession).not.toHaveBeenCalled(); + const remaining = await runtime!.runPromise(repository.getByThreadId({ threadId })); + expect(Option.isSome(remaining)).toBe(true); + }); + + it("prefers explicit options over server settings", async () => { + const threadId = ThreadId.make("thread-reaper-option-override"); + const now = "2026-01-01T00:00:00.000Z"; + const harness = await createHarness({ + readModel: makeReadModel([ + { + id: threadId, + session: { + threadId, + status: "ready", + providerName: "claudeAgent", + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + }, + ]), + reaperOptions: { + inactivityThresholdMs: 1_000, + sweepIntervalMs: 60_000, + }, + settingsOverrides: { + providerSessionInactivityThreshold: Duration.days(30), + providerSessionSweepInterval: Duration.hours(1), + }, + }); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); + + const nowMs = await Effect.runPromise(Clock.currentTimeMillis); + await runtime!.runPromise( + repository.upsert({ + threadId, + providerName: "claudeAgent", + providerInstanceId: null, + adapterKey: "claudeAgent", + runtimeMode: "full-access", + status: "running", + lastSeenAt: new Date(nowMs - 5_000).toISOString(), + resumeCursor: { + opaque: "resume-option-override", + }, + runtimePayload: null, + }), + ); + + const reaper = await runtime!.runPromise(Effect.service(ProviderSessionReaper)); + scope = await Effect.runPromise(Scope.make("sequential")); + await Effect.runPromise(reaper.start().pipe(Scope.provide(scope))); + + await waitFor(() => harness.stopSession.mock.calls.length === 1); + + expect(harness.stopSession.mock.calls[0]?.[0]).toEqual({ threadId }); + expect(harness.stoppedThreadIds.has(threadId)).toBe(true); + }); + it("skips stale sessions when the thread still has an active turn", async () => { const threadId = ThreadId.make("thread-reaper-active-turn"); const turnId = TurnId.make("turn-reaper-active"); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index 8eccd52fb2c..bcaa86009da 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -1,3 +1,4 @@ +import { DEFAULT_PROVIDER_SESSION_SWEEP_INTERVAL } from "@t3tools/contracts"; import * as Clock from "effect/Clock"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; @@ -6,6 +7,7 @@ import * as Option from "effect/Option"; import * as Schedule from "effect/Schedule"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { ProviderSessionReaper, @@ -14,11 +16,10 @@ import { import { forkParked } from "../../serverActivation.ts"; import { ProviderService } from "../Services/ProviderService.ts"; -const DEFAULT_INACTIVITY_THRESHOLD_MS = 30 * 60 * 1000; -const DEFAULT_SWEEP_INTERVAL_MS = 5 * 60 * 1000; - export interface ProviderSessionReaperLiveOptions { + /** Overrides the `providerSessionInactivityThreshold` server setting. */ readonly inactivityThresholdMs?: number; + /** Overrides the `providerSessionSweepInterval` server setting. */ readonly sweepIntervalMs?: number; } @@ -27,14 +28,37 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = const providerService = yield* ProviderService; const directory = yield* ProviderSessionDirectory; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; - - const inactivityThresholdMs = Math.max( - 1, - options?.inactivityThresholdMs ?? DEFAULT_INACTIVITY_THRESHOLD_MS, - ); - const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS); + const serverSettings = yield* ServerSettingsService; + + // Threshold is re-read every sweep so settings edits apply live; the + // sweep interval is fixed into the schedule when `start()` runs. A + // failed settings read fails the sweep (logged and retried on the next + // one) rather than silently reaping with a default the user overrode. + const resolveInactivityThresholdMs = + options?.inactivityThresholdMs !== undefined + ? Effect.succeed(Math.max(1, options.inactivityThresholdMs)) + : serverSettings.getSettings.pipe( + Effect.map((settings) => + Math.max(1, Duration.toMillis(settings.providerSessionInactivityThreshold)), + ), + ); + + const resolveSweepIntervalMs = + options?.sweepIntervalMs !== undefined + ? Effect.succeed(Math.max(1, options.sweepIntervalMs)) + : serverSettings.getSettings.pipe( + Effect.map((settings) => + Math.max(1, Duration.toMillis(settings.providerSessionSweepInterval)), + ), + Effect.catch((error) => + Effect.logWarning("provider.session.reaper.settings-fallback", { + error, + }).pipe(Effect.as(Duration.toMillis(DEFAULT_PROVIDER_SESSION_SWEEP_INTERVAL))), + ), + ); const sweep = Effect.gen(function* () { + const inactivityThresholdMs = yield* resolveInactivityThresholdMs; const bindings = yield* directory.listBindings(); const now = yield* Clock.currentTimeMillis; let reapedCount = 0; @@ -106,6 +130,8 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = const start: ProviderSessionReaperShape["start"] = () => Effect.gen(function* () { + const inactivityThresholdMs = yield* resolveInactivityThresholdMs; + const sweepIntervalMs = yield* resolveSweepIntervalMs; yield* forkParked( sweep.pipe( Effect.catch((error: unknown) => diff --git a/apps/server/src/serverSettings.ts b/apps/server/src/serverSettings.ts index 2798faf6f00..461eba2a2ce 100644 --- a/apps/server/src/serverSettings.ts +++ b/apps/server/src/serverSettings.ts @@ -146,8 +146,13 @@ export class ServerSettingsService extends Context.Service< const makeTest = (overrides: DeepPartial = {}) => Effect.gen(function* () { - const { automaticGitFetchInterval, providerHealthRefreshInterval, ...overridesForMerge } = - overrides; + const { + automaticGitFetchInterval, + providerHealthRefreshInterval, + providerSessionInactivityThreshold, + providerSessionSweepInterval, + ...overridesForMerge + } = overrides; const merged = deepMerge(DEFAULT_SERVER_SETTINGS, overridesForMerge); const initialSettings = yield* normalizeServerSettings({ ...merged, @@ -157,6 +162,15 @@ const makeTest = (overrides: DeepPartial = {}) => ...(providerHealthRefreshInterval !== undefined ? { providerHealthRefreshInterval: providerHealthRefreshInterval as Duration.Duration } : {}), + ...(providerSessionInactivityThreshold !== undefined + ? { + providerSessionInactivityThreshold: + providerSessionInactivityThreshold as Duration.Duration, + } + : {}), + ...(providerSessionSweepInterval !== undefined + ? { providerSessionSweepInterval: providerSessionSweepInterval as Duration.Duration } + : {}), }); const currentSettingsRef = yield* Ref.make(initialSettings); @@ -212,6 +226,8 @@ const ATOMIC_SETTINGS_KEYS: ReadonlySet = new Set([ "backgroundActivity", "automaticGitFetchInterval", "providerHealthRefreshInterval", + "providerSessionInactivityThreshold", + "providerSessionSweepInterval", "sourceControlWriterModelSelection", "textGenerationModelSelection", ]); diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index cbb547b95fb..9f7fc4e6f2c 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -494,6 +494,8 @@ export type SourceControlWritingStyleSettings = typeof SourceControlWritingStyle export const DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL = Duration.seconds(30); export const DEFAULT_PROVIDER_HEALTH_REFRESH_INTERVAL = Duration.minutes(5); +export const DEFAULT_PROVIDER_SESSION_INACTIVITY_THRESHOLD = Duration.minutes(30); +export const DEFAULT_PROVIDER_SESSION_SWEEP_INTERVAL = Duration.minutes(5); export const BackgroundActivityProfile = Schema.Literals([ "balanced", @@ -538,6 +540,20 @@ export const ServerSettings = Schema.Struct({ enableAssistantStreaming: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), enableProviderUpdateChecks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), backgroundActivity: BackgroundActivitySettings, + // How long a provider session may sit idle before the inactivity reaper + // stops it (applies from the next sweep), and how often the reaper sweeps + // (applies after a server restart). Idle sessions resume from their + // persisted cursor on the next message. + providerSessionInactivityThreshold: Schema.DurationFromMillis.pipe( + Schema.withDecodingDefault( + Effect.succeed(Duration.toMillis(DEFAULT_PROVIDER_SESSION_INACTIVITY_THRESHOLD)), + ), + ), + providerSessionSweepInterval: Schema.DurationFromMillis.pipe( + Schema.withDecodingDefault( + Effect.succeed(Duration.toMillis(DEFAULT_PROVIDER_SESSION_SWEEP_INTERVAL)), + ), + ), // Legacy flat fields retained for old settings files and old clients. New // consumers should resolve `backgroundActivity` instead. automaticGitFetchInterval: Schema.DurationFromMillis.pipe( @@ -710,6 +726,8 @@ export const ServerSettingsPatch = Schema.Struct({ ), automaticGitFetchInterval: Schema.optionalKey(Schema.DurationFromMillis), providerHealthRefreshInterval: Schema.optionalKey(Schema.DurationFromMillis), + providerSessionInactivityThreshold: Schema.optionalKey(Schema.DurationFromMillis), + providerSessionSweepInterval: Schema.optionalKey(Schema.DurationFromMillis), backgroundActivityProfile: Schema.optionalKey(BackgroundActivityProfile), defaultThreadEnvMode: Schema.optionalKey(ThreadEnvMode), newWorktreesStartFromOrigin: Schema.optionalKey(Schema.Boolean), diff --git a/packages/shared/src/serverSettings.ts b/packages/shared/src/serverSettings.ts index 21d819a1c9e..0e7e1547ada 100644 --- a/packages/shared/src/serverSettings.ts +++ b/packages/shared/src/serverSettings.ts @@ -129,6 +129,8 @@ export function applyServerSettingsPatch( const { automaticGitFetchInterval, providerHealthRefreshInterval, + providerSessionInactivityThreshold, + providerSessionSweepInterval, backgroundActivityProfile, backgroundActivity, ...patchForMerge @@ -192,6 +194,10 @@ export function applyServerSettingsPatch( : {}), ...(automaticGitFetchInterval !== undefined ? { automaticGitFetchInterval } : {}), ...(providerHealthRefreshInterval !== undefined ? { providerHealthRefreshInterval } : {}), + ...(providerSessionInactivityThreshold !== undefined + ? { providerSessionInactivityThreshold } + : {}), + ...(providerSessionSweepInterval !== undefined ? { providerSessionSweepInterval } : {}), }; const normalizedBackgroundActivity = normalizeBackgroundActivitySettings( nextWithReplacementsBase.backgroundActivity, From 47869a8e81a5ff332e40e2985ec155072e2994e3 Mon Sep 17 00:00:00 2001 From: Karl Schulze <32364521+kraptor23@users.noreply.github.com> Date: Thu, 6 Aug 2026 14:32:15 +0000 Subject: [PATCH 2/2] feat(server): keep reaper startup non-blocking when settings are unavailable Review feedback: start() yielded the failable threshold read before forking, so a failed settings read could abort the reactors.start startup phase even though the sweep re-reads the threshold anyway. Resolve it non-fatally for the started log only. Also use the module namespace import for serverSettings in the test and Effect DateTime instead of a global Date in test fixtures. Co-Authored-By: Claude Fable 5 --- .../provider/Layers/ProviderSessionReaper.test.ts | 12 ++++++------ .../src/provider/Layers/ProviderSessionReaper.ts | 6 +++++- 2 files changed, 11 insertions(+), 7 deletions(-) diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 7689b144de0..08a39effe80 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -24,7 +24,7 @@ import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntim import { ProviderValidationError } from "../Errors.ts"; import { ProviderSessionReaper } from "../Services/ProviderSessionReaper.ts"; import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; -import { layerTest as serverSettingsLayerTest } from "../../serverSettings.ts"; +import * as ServerSettings from "../../serverSettings.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; import { makeProviderSessionReaperLive, @@ -146,7 +146,7 @@ describe("ProviderSessionReaper", () => { readonly threadId: ThreadId; }) => ReturnType; readonly reaperOptions?: ProviderSessionReaperLiveOptions; - readonly settingsOverrides?: Parameters[0]; + readonly settingsOverrides?: Parameters[0]; }) { const stoppedThreadIds = new Set(); const stopSession = vi.fn( @@ -196,7 +196,7 @@ describe("ProviderSessionReaper", () => { sweepIntervalMs: 60_000, }, ).pipe( - Layer.provideMerge(serverSettingsLayerTest(input.settingsOverrides ?? {})), + Layer.provideMerge(ServerSettings.layerTest(input.settingsOverrides ?? {})), Layer.provideMerge(providerSessionDirectoryLayer), Layer.provideMerge(runtimeRepositoryLayer), Layer.provideMerge(Layer.succeed(ProviderService, providerService)), @@ -322,7 +322,7 @@ describe("ProviderSessionReaper", () => { adapterKey: "claudeAgent", runtimeMode: "full-access", status: "running", - lastSeenAt: new Date(nowMs - 5_000).toISOString(), + lastSeenAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 5_000)), resumeCursor: { opaque: "resume-settings", }, @@ -377,7 +377,7 @@ describe("ProviderSessionReaper", () => { adapterKey: "claudeAgent", runtimeMode: "full-access", status: "running", - lastSeenAt: new Date(nowMs - 5_000).toISOString(), + lastSeenAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 5_000)), resumeCursor: { opaque: "resume-settings-fresh", }, @@ -435,7 +435,7 @@ describe("ProviderSessionReaper", () => { adapterKey: "claudeAgent", runtimeMode: "full-access", status: "running", - lastSeenAt: new Date(nowMs - 5_000).toISOString(), + lastSeenAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 5_000)), resumeCursor: { opaque: "resume-option-override", }, diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.ts index bcaa86009da..e2ace2b9d76 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.ts @@ -130,7 +130,11 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = const start: ProviderSessionReaperShape["start"] = () => Effect.gen(function* () { - const inactivityThresholdMs = yield* resolveInactivityThresholdMs; + // Informational only — the sweep re-reads the threshold, so a failed + // settings read here must not abort the startup phase. + const inactivityThresholdMs = yield* resolveInactivityThresholdMs.pipe( + Effect.orElseSucceed(() => undefined), + ); const sweepIntervalMs = yield* resolveSweepIntervalMs; yield* forkParked( sweep.pipe(