fix(server): keep provider sessions alive during background agent work

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<void, ProviderSessionRuntimeRepositoryError>;
+
     /**
      * 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"];
 });

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();

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

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-"));

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<Name extends keyof ProviderService.ProviderService["Service"]> =
@@ -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<ProviderRuntimeEvent>();
+  // 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<ThreadId, number>();
   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<void> =>
+    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)),
+        ),
       ),
     );