import * as DateTime from "effect/DateTime";import * as Duration from "effect/Duration";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import * as Schedule from "effect/Schedule"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts";import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts";import {  ProviderSessionReaper,  type ProviderSessionReaperShape,} from "../Services/ProviderSessionReaper.ts";import { ProviderService } from "../Services/ProviderService.ts"; const DEFAULT_INACTIVITY_THRESHOLD = Duration.minutes(30);const DEFAULT_SWEEP_INTERVAL = Duration.minutes(5); export interface ProviderSessionReaperLiveOptions {  readonly inactivityThreshold?: Duration.Input;  readonly sweepInterval?: Duration.Input;} const normalizePositiveDuration = (  input: Duration.Input | undefined,  fallback: Duration.Duration,) => {  const duration = input === undefined ? fallback : Duration.fromInputUnsafe(input);  return Duration.millis(Math.max(1, Duration.toMillis(duration)));}; const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =>  Effect.gen(function* () {    const providerService = yield* ProviderService;    const directory = yield* ProviderSessionDirectory;    const projectionSnapshotQuery = yield* ProjectionSnapshotQuery;     const inactivityThreshold = normalizePositiveDuration(      options?.inactivityThreshold,      DEFAULT_INACTIVITY_THRESHOLD,    );    const sweepInterval = normalizePositiveDuration(options?.sweepInterval, DEFAULT_SWEEP_INTERVAL);    const inactivityThresholdMs = Duration.toMillis(inactivityThreshold);    const sweepIntervalMs = Duration.toMillis(sweepInterval);     const sweep = Effect.gen(function* () {      const bindings = yield* directory.listBindings();      const now = yield* DateTime.now;      let reapedCount = 0;       for (const binding of bindings) {        if (binding.status === "stopped") {          continue;        }         const lastSeen = DateTime.make(binding.lastSeenAt);        if (Option.isNone(lastSeen)) {          yield* Effect.logWarning("provider.session.reaper.invalid-last-seen", {            threadId: binding.threadId,            provider: binding.provider,            lastSeenAt: binding.lastSeenAt,          });          continue;        }         const idleDuration = Duration.millis(          DateTime.toEpochMillis(now) - DateTime.toEpochMillis(lastSeen.value),        );        const idleDurationMs = Duration.toMillis(idleDuration);        if (idleDurationMs < inactivityThresholdMs) {          continue;        }         const thread = yield* projectionSnapshotQuery          .getThreadShellById(binding.threadId)          .pipe(Effect.map(Option.getOrUndefined));        if (thread?.session?.activeTurnId != null) {          yield* Effect.logDebug("provider.session.reaper.skipped-active-turn", {            threadId: binding.threadId,            activeTurnId: thread.session.activeTurnId,            idleDurationMs,          });          continue;        }         const reaped = yield* providerService.stopSession({ threadId: binding.threadId }).pipe(          Effect.tap(() =>            Effect.logInfo("provider.session.reaped", {              threadId: binding.threadId,              provider: binding.provider,              idleDurationMs,              reason: "inactivity_threshold",            }),          ),          Effect.as(true),          Effect.catchCause((cause) =>            Effect.logWarning("provider.session.reaper.stop-failed", {              threadId: binding.threadId,              provider: binding.provider,              idleDurationMs,              cause,            }).pipe(Effect.as(false)),          ),        );         if (reaped) {          reapedCount += 1;        }      }       if (reapedCount > 0) {        yield* Effect.logInfo("provider.session.reaper.sweep-complete", {          reapedCount,          totalBindings: bindings.length,        });      }    });     const start: ProviderSessionReaperShape["start"] = () =>      Effect.gen(function* () {        yield* Effect.forkScoped(          sweep.pipe(            Effect.catch((error: unknown) =>              Effect.logWarning("provider.session.reaper.sweep-failed", {                error,              }),            ),            Effect.catchDefect((defect: unknown) =>              Effect.logWarning("provider.session.reaper.sweep-defect", {                defect,              }),            ),            Effect.repeat(Schedule.spaced(sweepInterval)),          ),        );         yield* Effect.logInfo("provider.session.reaper.started", {          inactivityThresholdMs,          sweepIntervalMs,        });      });     return {      start,    } satisfies ProviderSessionReaperShape;  }); export const makeProviderSessionReaperLive = (options?: ProviderSessionReaperLiveOptions) =>  Layer.effect(ProviderSessionReaper, makeProviderSessionReaper(options)); export const ProviderSessionReaperLive = makeProviderSessionReaperLive();