diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index f3f4ca39d47..08a39effe80 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 * as ServerSettings 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(ServerSettings.layerTest(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: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 5_000)), + 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: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 5_000)), + 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: DateTime.formatIso(DateTime.makeUnsafe(nowMs - 5_000)), + 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..e2ace2b9d76 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,12 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) = const start: ProviderSessionReaperShape["start"] = () => Effect.gen(function* () { + // 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( 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,