Replace provider session reaper polling with deadline-driven scheduling

apps/server/integration/OrchestrationEngineHarness.integration.ts

@@ -39,6 +39,7 @@ import { ProjectionPendingApprovalRepository } from "../src/persistence/Services
 import { ProviderUnsupportedError } from "../src/provider/Errors.ts";
 import { ProviderAdapterRegistry } from "../src/provider/Services/ProviderAdapterRegistry.ts";
 import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts";
+import { ProviderSessionDirectoryEventsLive } from "../src/provider/Layers/ProviderSessionDirectoryEvents.ts";
 import { ServerSettingsService } from "../src/serverSettings.ts";
 import { makeProviderServiceLive } from "../src/provider/Layers/ProviderService.ts";
 import { makeCodexAdapterLive } from "../src/provider/Layers/CodexAdapter.ts";
@@ -258,8 +259,10 @@ export const makeOrchestrationIntegrationHarness = (
       Layer.provide(OrchestrationEventStoreLive),
       Layer.provide(OrchestrationCommandReceiptRepositoryLive),
     );
+    const providerSessionDirectoryEventsLayer = ProviderSessionDirectoryEventsLive;
     const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe(
       Layer.provide(ProviderSessionRuntimeRepositoryLive),
+      Layer.provide(providerSessionDirectoryEventsLayer),
     );
     const realCodexRegistry = Layer.effect(
       ProviderAdapterRegistry,
@@ -277,16 +280,19 @@ export const makeOrchestrationIntegrationHarness = (
       Layer.provide(makeCodexAdapterLive()),
       Layer.provideMerge(ServerConfig.layerTest(workspaceDir, rootDir)),
       Layer.provideMerge(NodeServices.layer),
+      Layer.provideMerge(providerSessionDirectoryEventsLayer),
       Layer.provideMerge(providerSessionDirectoryLayer),
     );
     const providerLayer = useRealCodex
       ? makeProviderServiceLive().pipe(
           Layer.provide(providerSessionDirectoryLayer),
+          Layer.provide(providerSessionDirectoryEventsLayer),
           Layer.provide(realCodexRegistry),
           Layer.provide(AnalyticsService.layerTest),
         )
       : makeProviderServiceLive().pipe(
           Layer.provide(providerSessionDirectoryLayer),
+          Layer.provide(providerSessionDirectoryEventsLayer),
           Layer.provide(fakeRegistry!),
           Layer.provide(AnalyticsService.layerTest),
         );

apps/server/integration/providerService.integration.test.ts

@@ -8,6 +8,7 @@ import { Effect, FileSystem, Layer, Path, Queue, Stream } from "effect";
 import { ProviderUnsupportedError } from "../src/provider/Errors.ts";
 import { ProviderAdapterRegistry } from "../src/provider/Services/ProviderAdapterRegistry.ts";
 import { ProviderSessionDirectoryLive } from "../src/provider/Layers/ProviderSessionDirectory.ts";
+import { ProviderSessionDirectoryEventsLive } from "../src/provider/Layers/ProviderSessionDirectoryEvents.ts";
 import { makeProviderServiceLive } from "../src/provider/Layers/ProviderService.ts";
 import {
   ProviderService,
@@ -55,12 +56,15 @@ const makeIntegrationFixture = Effect.gen(function* () {
     listProviders: () => Effect.succeed(["codex"]),
   };
 
+  const directoryEventsLayer = ProviderSessionDirectoryEventsLive;
   const directoryLayer = ProviderSessionDirectoryLive.pipe(
     Layer.provide(ProviderSessionRuntimeRepositoryLive),
+    Layer.provide(directoryEventsLayer),
   );
 
   const shared = Layer.mergeAll(
     directoryLayer,
+    directoryEventsLayer,
     Layer.succeed(ProviderAdapterRegistry, registry),
     ServerSettingsService.layerTest(DEFAULT_SERVER_SETTINGS),
     AnalyticsService.layerTest,

apps/server/src/cli-config.test.ts

@@ -7,6 +7,10 @@ import { NetService } from "@t3tools/shared/Net";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { deriveServerPaths } from "./config.ts";
 import { resolveServerConfig } from "./cli.ts";
+import {
+  DEFAULT_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS,
+  DEFAULT_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS,
+} from "./provider/Services/ProviderSessionReaper.ts";
 
 it.layer(NodeServices.layer)("cli config resolution", (it) => {
   const defaultObservabilityConfig = {
@@ -19,6 +23,10 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => {
     otlpMetricsUrl: undefined,
     otlpExportIntervalMs: 10_000,
     otlpServiceName: "t3-server",
+    providerSessionReaperInactivityThresholdMs:
+      DEFAULT_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS,
+    providerSessionReaperFallbackReconcileIntervalMs:
+      DEFAULT_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS,
   } as const;
 
   const openBootstrapFd = Effect.fn(function* (payload: Record<string, unknown>) {
@@ -91,6 +99,63 @@ it.layer(NodeServices.layer)("cli config resolution", (it) => {
     }),
   );
 
+  it.effect("reads provider session reaper tuning env vars", () =>
+    Effect.gen(function* () {
+      const fs = yield* FileSystem.FileSystem;
+      const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-cli-config-reaper-" });
+      const derivedPaths = yield* deriveServerPaths(baseDir, undefined);
+      const resolved = yield* resolveServerConfig(
+        {
+          mode: Option.some("desktop"),
+          port: Option.some(4888),
+          host: Option.none(),
+          baseDir: Option.some(baseDir),
+          cwd: Option.none(),
+          devUrl: Option.none(),
+          noBrowser: Option.none(),
+          bootstrapFd: Option.none(),
+          autoBootstrapProjectFromCwd: Option.none(),
+          logWebSocketEvents: Option.none(),
+        },
+        Option.none(),
+      ).pipe(
+        Effect.provide(
+          Layer.mergeAll(
+            ConfigProvider.layer(
+              ConfigProvider.fromEnv({
+                env: {
+                  T3CODE_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS: "1500",
+                  T3CODE_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS: "2500",
+                },
+              }),
+            ),
+            NetService.layer,
+          ),
+        ),
+      );
+
+      expect(resolved).toEqual({
+        logLevel: "Info",
+        ...defaultObservabilityConfig,
+        providerSessionReaperInactivityThresholdMs: 1500,
+        providerSessionReaperFallbackReconcileIntervalMs: 2500,
+        mode: "desktop",
+        port: 4888,
+        cwd: process.cwd(),
+        baseDir,
+        ...derivedPaths,
+        host: "127.0.0.1",
+        staticDir: resolved.staticDir,
+        devUrl: undefined,
+        noBrowser: true,
+        startupPresentation: "browser",
+        desktopBootstrapToken: undefined,
+        autoBootstrapProjectFromCwd: false,
+        logWebSocketEvents: false,
+      });
+    }),
+  );
+
   it.effect("uses CLI flags when provided", () =>
     Effect.gen(function* () {
       const { join } = yield* Path.Path;

apps/server/src/cli.test.ts

@@ -63,6 +63,8 @@ const makeCliTestServerConfig = (baseDir: string) =>
       otlpMetricsUrl: undefined,
       otlpExportIntervalMs: 10_000,
       otlpServiceName: "t3-server",
+      providerSessionReaperInactivityThresholdMs: 30 * 60 * 1000,
+      providerSessionReaperFallbackReconcileIntervalMs: 30 * 60 * 1000,
       mode: "web",
       port: 0,
       host: "127.0.0.1",

apps/server/src/cli.ts

@@ -57,6 +57,10 @@ import { OrchestrationEngineService } from "./orchestration/Services/Orchestrati
 import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery.ts";
 import { OrchestrationLayerLive } from "./orchestration/runtimeLayer.ts";
 import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/Sqlite.ts";
+import {
+  DEFAULT_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS,
+  DEFAULT_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS,
+} from "./provider/Services/ProviderSessionReaper.ts";
 import { RepositoryIdentityResolverLive } from "./project/Layers/RepositoryIdentityResolver.ts";
 import { getAutoBootstrapDefaultModelSelection } from "./serverRuntimeStartup.ts";
 import {
@@ -150,6 +154,12 @@ const EnvServerConfig = Config.all({
     Config.withDefault(10_000),
   ),
   otlpServiceName: Config.string("T3CODE_OTLP_SERVICE_NAME").pipe(Config.withDefault("t3-server")),
+  providerSessionReaperInactivityThresholdMs: Config.int(
+    "T3CODE_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS",
+  ).pipe(Config.withDefault(DEFAULT_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS)),
+  providerSessionReaperFallbackReconcileIntervalMs: Config.int(
+    "T3CODE_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS",
+  ).pipe(Config.withDefault(DEFAULT_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS)),
   mode: Config.schema(RuntimeMode, "T3CODE_MODE").pipe(
     Config.option,
     Config.map(Option.getOrUndefined),
@@ -351,6 +361,9 @@ export const resolveServerConfig = (
         persistedObservabilitySettings.otlpMetricsUrl,
       otlpExportIntervalMs: env.otlpExportIntervalMs,
       otlpServiceName: env.otlpServiceName,
+      providerSessionReaperInactivityThresholdMs: env.providerSessionReaperInactivityThresholdMs,
+      providerSessionReaperFallbackReconcileIntervalMs:
+        env.providerSessionReaperFallbackReconcileIntervalMs,
       mode,
       port,
       cwd,

apps/server/src/config.ts

@@ -8,6 +8,11 @@
  */
 import { Effect, FileSystem, Layer, LogLevel, Path, Schema, Context } from "effect";
 
+import {
+  DEFAULT_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS,
+  DEFAULT_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS,
+} from "./provider/Services/ProviderSessionReaper.ts";
+
 export const DEFAULT_PORT = 3773;
 
 export const RuntimeMode = Schema.Literals(["web", "desktop"]);
@@ -53,6 +58,8 @@ export interface ServerConfigShape extends ServerDerivedPaths {
   readonly otlpMetricsUrl: string | undefined;
   readonly otlpExportIntervalMs: number;
   readonly otlpServiceName: string;
+  readonly providerSessionReaperInactivityThresholdMs: number;
+  readonly providerSessionReaperFallbackReconcileIntervalMs: number;
   readonly mode: RuntimeMode;
   readonly port: number;
   readonly host: string | undefined;
@@ -152,6 +159,10 @@ export class ServerConfig extends Context.Service<ServerConfig, ServerConfigShap
           otlpMetricsUrl: undefined,
           otlpExportIntervalMs: 10_000,
           otlpServiceName: "t3-server",
+          providerSessionReaperInactivityThresholdMs:
+            DEFAULT_PROVIDER_SESSION_REAPER_INACTIVITY_THRESHOLD_MS,
+          providerSessionReaperFallbackReconcileIntervalMs:
+            DEFAULT_PROVIDER_SESSION_REAPER_FALLBACK_RECONCILE_INTERVAL_MS,
           cwd,
           baseDir,
           ...derivedPaths,

apps/server/src/environment/Layers/ServerEnvironment.test.ts

@@ -25,6 +25,8 @@ const makeServerConfig = Effect.fn(function* (baseDir: string) {
     otlpMetricsUrl: undefined,
     otlpExportIntervalMs: 10_000,
     otlpServiceName: "t3-server",
+    providerSessionReaperInactivityThresholdMs: 30 * 60 * 1000,
+    providerSessionReaperFallbackReconcileIntervalMs: 30 * 60 * 1000,
     cwd: process.cwd(),
     baseDir,
     mode: "web",

apps/server/src/observability/Metrics.ts

@@ -54,6 +54,60 @@ export const providerRuntimeEventsTotal = Metric.counter("t3_provider_runtime_ev
   description: "Total canonical provider runtime events processed.",
 });
 
+export const providerSessionReaperWakeupsTotal = Metric.counter(
+  "t3_provider_session_reaper_wakeups_total",
+  {
+    description: "Total provider session reaper wakeups by mode and wake reason.",
+  },
+);
+
+export const providerSessionReaperDueCandidatesTotal = Metric.counter(
+  "t3_provider_session_reaper_due_candidates_total",
+  {
+    description: "Total provider session reaper due candidates evaluated.",
+  },
+);
+
+export const providerSessionReaperReapedTotal = Metric.counter(
+  "t3_provider_session_reaper_reaped_total",
+  {
+    description: "Total provider sessions reaped for inactivity.",
+  },
+);
+
+export const providerSessionReaperReapLag = Metric.timer("t3_provider_session_reaper_reap_lag", {
+  description: "Observed inactivity reap lag relative to the computed deadline.",
+});
+
+export const providerSessionReaperScheduleSize = Metric.gauge(
+  "t3_provider_session_reaper_schedule_size",
+  {
+    description: "Current number of future inactivity deadlines tracked by the reaper.",
+  },
+);
+
+export const providerSessionReaperReconcileDuration = Metric.timer(
+  "t3_provider_session_reaper_reconcile_duration",
+  {
+    description: "Provider session reaper authoritative reconcile duration.",
+  },
+);
+
+export const providerSessionReaperWakeCoalescedTotal = Metric.counter(
+  "t3_provider_session_reaper_wake_coalesced_total",
+  {
+    description:
+      "Total provider session reaper wake hints coalesced into an existing pending wake.",
+  },
+);
+
+export const providerSessionReaperSignalFeedRestartsTotal = Metric.counter(
+  "t3_provider_session_reaper_signal_feed_restarts_total",
+  {
+    description: "Total provider session reaper signal feed restarts after transient failure.",
+  },
+);
+
 export const gitCommandsTotal = Metric.counter("t3_git_commands_total", {
   description: "Total git commands executed by the server runtime.",
 });