feat(server): restore stream liveness on v2

apps/server/src/orchestration-v2/RunExecutionService.ts

@@ -1,4 +1,5 @@
 import { makeAssistantStreamingFilter } from "./assistantStreaming.ts";
+import * as StreamClock from "../prism/streamClock.ts";
 import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
 import {
   isOrchestrationV2WorkActive,
@@ -556,6 +557,7 @@ export const layer: Layer.Layer<
     const serverSettings = yield* ServerSettings.ServerSettingsService;
     const mcpAppModelContext = yield* McpAppModelContext.McpAppModelContext;
     const finalizationObserver = yield* RunFinalizationService.RunFinalizationObserver;
+    const streamClock = yield* StreamClock.StreamClockHooks;
 
     const writeFinalRunEvents = (input: {
       readonly run: OrchestrationV2Run;
@@ -1181,6 +1183,7 @@ export const layer: Layer.Layer<
             }
             return true;
           });
+          yield* StreamClock.beginExecutionAttempt(streamClock, input);
           const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode);
           const providerEventFiber = yield* eventSubscription.events.pipe(
             Stream.filterEffect((event) =>
@@ -1245,6 +1248,9 @@ export const layer: Layer.Layer<
                     yield* Ref.set(providerThreadOwnerLost, true);
                   }
                 }
+                if (deliveredEvent === null) {
+                  yield* StreamClock.observeFilteredEvent(streamClock, input, event);
+                }
                 if (event.type === "provider_thread.updated") {
                   if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) {
                     yield* Ref.set(latestProviderThread, event.providerThread);
@@ -1268,6 +1274,7 @@ export const layer: Layer.Layer<
               }),
             ),
             Stream.takeUntilEffect(() => shouldStopProviderEventIngestion),
+            Stream.provideService(StreamClock.StreamClockAttempt, input.attempt.id),
             Stream.runDrain,
             Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })),
             Effect.flatMap(() =>
@@ -1337,6 +1344,9 @@ export const layer: Layer.Layer<
               ),
             ),
             Effect.ensuring(eventSubscription.close),
+            Effect.ensuring(
+              streamClock.endAttempt(input.run.threadId, input.run.id, input.attempt.id),
+            ),
             Effect.forkDetach,
           );
 

apps/server/src/orchestration-v2/runtimeLayer.ts

@@ -1,4 +1,8 @@
 import * as PrismRecovery from "../prism/RecoveryCoordinator.ts";
+import * as PrismStreamClock from "../prism/streamClock.ts";
+import * as PrismStreamStats from "../prism/StreamStatsStore.ts";
+import * as PrismProviderEventIngestor from "../prism/ProviderEventIngestor.ts";
+import * as PrismStaleTurnMonitor from "../prism/staleTurnMonitor.ts";
 import * as PrismRecoveryHooks from "../prism/RecoveryHooks.ts";
 import * as PrismRecoveryReactor from "../prism/RecoveryReactor.ts";
 import * as UsageLimitRecoveryWorker from "./UsageLimitRecoveryWorker.ts";
@@ -103,7 +107,10 @@ export const layerProjectService = ProjectService.layer.pipe(
   ),
 );
 
-const layerProviderEventIngestorProvided = ProviderEventIngestor.layer.pipe(
+const layerPrismStreamClockProvided = PrismStreamClock.layer.pipe(
+  Layer.provide(PrismStreamStats.layer),
+);
+const layerProviderEventIngestorBase = ProviderEventIngestor.layer.pipe(
   Layer.provide(
     Layer.mergeAll(
       layerEventSinkProvided,
@@ -113,6 +120,9 @@ const layerProviderEventIngestorProvided = ProviderEventIngestor.layer.pipe(
     ),
   ),
 );
+const layerProviderEventIngestorProvided = PrismProviderEventIngestor.layer.pipe(
+  Layer.provide(Layer.mergeAll(layerProviderEventIngestorBase, layerPrismStreamClockProvided)),
+);
 
 const layerCheckpointServiceProvided = CheckpointService.layer.pipe(
   Layer.provide(IdAllocator.layer),
@@ -151,6 +161,7 @@ const layerRunExecutionServiceProvided = RunExecutionService.layer.pipe(
       layerEventSinkProvided,
       IdAllocator.layer,
       layerProviderEventIngestorProvided,
+      layerPrismStreamClockProvided,
     ),
   ),
 );
@@ -358,6 +369,15 @@ export const layerProduction = Layer.mergeAll(
   layerThreadLifecycleProvided,
   layerScheduledTaskProvided,
   layerSecretRequestsProvided,
+  PrismStaleTurnMonitor.layer.pipe(
+    Layer.provide(
+      Layer.mergeAll(
+        layerPrismStreamClockProvided,
+        ProjectionStore.layer,
+        layerThreadManagementProvided,
+      ),
+    ),
+  ),
   PrismRecoveryReactor.layer.pipe(
     Layer.provide(
       Layer.mergeAll(

apps/server/src/prism/ProviderEventIngestor.ts

@@ -0,0 +1,28 @@
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as ProviderEventIngestor from "../orchestration-v2/ProviderEventIngestor.ts";
+import * as StreamClock from "./streamClock.ts";
+
+/** Accepted writes refresh liveness. Routed terminals may intentionally produce no writes. */
+export const layer = Layer.effect(
+  ProviderEventIngestor.ProviderEventIngestorV2,
+  Effect.gen(function* () {
+    const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
+    const clock = yield* StreamClock.StreamClock;
+    return ProviderEventIngestor.ProviderEventIngestorV2.of({
+      normalize: ingestor.normalize,
+      ingestNormalized: (input) =>
+        ingestor
+          .ingestNormalized(input)
+          .pipe(
+            Effect.tap((stored) =>
+              stored.length > 0 || input.event.type === "turn.terminal"
+                ? Effect.flatMap(StreamClock.StreamClockAttempt, (attemptId) =>
+                    clock.observe(input, attemptId),
+                  )
+                : Effect.void,
+            ),
+          ),
+    });
+  }),
+);

apps/server/src/prism/StreamStatsStore.test.ts

@@ -0,0 +1,181 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { ProviderDriverKind, ProviderTurnId, RunId, ThreadId } from "@t3tools/contracts";
+import { describe, expect, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
+import * as Layer from "effect/Layer";
+import * as Path from "effect/Path";
+import * as SqlClient from "effect/sql/SqlClient";
+
+import * as SqlitePersistence from "../persistence/Sqlite.ts";
+import * as StreamStatsStore from "./StreamStatsStore.ts";
+
+const storeOver = <E, R>(database: Layer.Layer<SqlClient.SqlClient, E, R>) =>
+  StreamStatsStore.layer.pipe(Layer.provide(database));
+
+let turnNumber = 0;
+const turn = (
+  overrides: Partial<StreamStatsStore.TurnStreamStats> = {},
+): StreamStatsStore.TurnStreamStats => {
+  turnNumber += 1;
+  return {
+    threadId: ThreadId.make("thread-1"),
+    runId: RunId.make(`run-${turnNumber}`),
+    providerTurnId: ProviderTurnId.make("provider-turn-1"),
+    provider: ProviderDriverKind.make("codex"),
+    model: "gpt-5",
+    outcome: "completed",
+    startedAt: 1_000,
+    endedAt: 9_000,
+    timeToFirstTokenMs: 1_500,
+    maxGapMs: 2_000,
+    eventCount: 40,
+    ...overrides,
+  };
+};
+
+const samples = Effect.gen(function* () {
+  const store = yield* StreamStatsStore.StreamStatsStore;
+  const rows = yield* store.healthySamples;
+  return [...rows].sort((a, b) =>
+    `${a.provider}/${a.model}`.localeCompare(`${b.provider}/${b.model}`),
+  );
+});
+
+describe("StreamStatsStore", () => {
+  it.effect("keeps the first measurements of a provider turn when a write is retried", () =>
+    Effect.gen(function* () {
+      const store = yield* StreamStatsStore.StreamStatsStore;
+      const first = turn({ timeToFirstTokenMs: 1_000, maxGapMs: 3_000 });
+
+      expect(yield* store.record(first)).toBe(true);
+      expect(yield* store.record({ ...first, timeToFirstTokenMs: 90_000, maxGapMs: 80_000 })).toBe(
+        false,
+      );
+      // The same run's next provider turn is a separate measurement.
+      expect(
+        yield* store.record({ ...first, providerTurnId: ProviderTurnId.make("provider-turn-2") }),
+      ).toBe(true);
+
+      expect(yield* samples).toEqual([
+        {
+          provider: "codex",
+          model: "gpt-5",
+          count: 2,
+          maxTimeToFirstTokenMs: 1_000,
+          maxGapMs: 3_000,
+        },
+      ]);
+    }).pipe(Effect.provide(storeOver(SqlitePersistence.layerMemory))),
+  );
+
+  it.effect("counts only completed turns that streamed a first token", () =>
+    Effect.gen(function* () {
+      const store = yield* StreamStatsStore.StreamStatsStore;
+      yield* store.record(turn({ timeToFirstTokenMs: 2_000, maxGapMs: 4_000 }));
+      yield* store.record(turn({ timeToFirstTokenMs: 5_000, maxGapMs: 1_000 }));
+      yield* store.record(turn({ outcome: "error", timeToFirstTokenMs: 60_000, maxGapMs: 60_000 }));
+      yield* store.record(
+        turn({ outcome: "aborted", timeToFirstTokenMs: 70_000, maxGapMs: 70_000 }),
+      );
+      yield* store.record(turn({ timeToFirstTokenMs: null, maxGapMs: 50_000 }));
+
+      expect(yield* samples).toEqual([
+        {
+          provider: "codex",
+          model: "gpt-5",
+          count: 2,
+          maxTimeToFirstTokenMs: 5_000,
+          maxGapMs: 4_000,
+        },
+      ]);
+    }).pipe(Effect.provide(storeOver(SqlitePersistence.layerMemory))),
+  );
+
+  it.effect("keeps samples apart by provider and model", () =>
+    Effect.gen(function* () {
+      const store = yield* StreamStatsStore.StreamStatsStore;
+      const claude = ProviderDriverKind.make("claudeAgent");
+      yield* store.record(turn({ model: "gpt-5", timeToFirstTokenMs: 1_000, maxGapMs: 1_100 }));
+      yield* store.record(
+        turn({ model: "gpt-5-mini", timeToFirstTokenMs: 2_000, maxGapMs: 2_200 }),
+      );
+      yield* store.record(turn({ model: null, timeToFirstTokenMs: 3_000, maxGapMs: 3_300 }));
+      yield* store.record(turn({ model: null, timeToFirstTokenMs: 3_500, maxGapMs: 3_000 }));
+      yield* store.record(
+        turn({ provider: claude, model: "gpt-5", timeToFirstTokenMs: 4_000, maxGapMs: 4_400 }),
+      );
+
+      expect(yield* samples).toEqual([
+        {
+          provider: "claudeAgent",
+          model: "gpt-5",
+          count: 1,
+          maxTimeToFirstTokenMs: 4_000,
+          maxGapMs: 4_400,
+        },
+        {
+          provider: "codex",
+          model: "gpt-5",
+          count: 1,
+          maxTimeToFirstTokenMs: 1_000,
+          maxGapMs: 1_100,
+        },
+        {
+          provider: "codex",
+          model: "gpt-5-mini",
+          count: 1,
+          maxTimeToFirstTokenMs: 2_000,
+          maxGapMs: 2_200,
+        },
+        { provider: "codex", model: null, count: 2, maxTimeToFirstTokenMs: 3_500, maxGapMs: 3_300 },
+      ]);
+    }).pipe(Effect.provide(storeOver(SqlitePersistence.layerMemory))),
+  );
+
+  it.effect("refuses measurements that are not finite and stores nothing", () =>
+    Effect.gen(function* () {
+      const store = yield* StreamStatsStore.StreamStatsStore;
+      const error = yield* Effect.flip(store.record(turn({ maxGapMs: Number.NaN })));
+
+      expect(error._tag).toBe("StreamStatsWriteError");
+      expect(yield* samples).toEqual([]);
+    }).pipe(Effect.provide(storeOver(SqlitePersistence.layerMemory))),
+  );
+
+  it.effect("serves recorded samples again after a restart", () =>
+    Effect.gen(function* () {
+      const fs = yield* FileSystem.FileSystem;
+      const path = yield* Path.Path;
+      const dbPath = path.join(yield* fs.makeTempDirectoryScoped(), "chromeria-v2.sqlite");
+      const persistence = SqlitePersistence.layerFromPath(dbPath);
+      const stats = turn({ timeToFirstTokenMs: 1_200, maxGapMs: 6_000 });
+
+      yield* StreamStatsStore.StreamStatsStore.pipe(
+        Effect.flatMap((store) => store.record(stats)),
+        Effect.provide(storeOver(persistence)),
+      );
+
+      const afterRestart = yield* Effect.gen(function* () {
+        const store = yield* StreamStatsStore.StreamStatsStore;
+        return {
+          retried: yield* store.record({ ...stats, maxGapMs: 99_000 }),
+          samples: yield* samples,
+        };
+      }).pipe(Effect.provide(storeOver(persistence)));
+
+      expect(afterRestart).toEqual({
+        retried: false,
+        samples: [
+          {
+            provider: "codex",
+            model: "gpt-5",
+            count: 1,
+            maxTimeToFirstTokenMs: 1_200,
+            maxGapMs: 6_000,
+          },
+        ],
+      });
+    }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
+  );
+});