From b6ba362f544da1f2056a43923b2c669fe0e948ee Mon Sep 17 00:00:00 2001 From: lukemaj Date: Thu, 8 Oct 2026 21:50:14 +0200 Subject: [PATCH 1/2] feat(server): restore v2 stream clocks and silent child notices --- .../orchestration-v2/RunExecutionService.ts | 92 +++-- .../src/orchestration-v2/runtimeLayer.ts | 22 +- .../server/src/prism/ProviderEventIngestor.ts | 28 ++ .../server/src/prism/StreamStatsStore.test.ts | 181 +++++++++ apps/server/src/prism/StreamStatsStore.ts | 162 ++++++++ .../src/prism/staleTurnDetector.test.ts | 90 +++++ apps/server/src/prism/staleTurnDetector.ts | 90 +++++ .../staleTurnMonitor.integration.test.ts | 244 ++++++++++++ apps/server/src/prism/staleTurnMonitor.ts | 123 +++++++ .../src/prism/streamClock.integration.test.ts | 346 ++++++++++++++++++ apps/server/src/prism/streamClock.test.ts | 211 +++++++++++ apps/server/src/prism/streamClock.ts | 346 ++++++++++++++++++ docs/fork-features.md | 48 +++ scripts/fork-upstream-edits.txt | 3 + 14 files changed, 1954 insertions(+), 32 deletions(-) create mode 100644 apps/server/src/prism/ProviderEventIngestor.ts create mode 100644 apps/server/src/prism/StreamStatsStore.test.ts create mode 100644 apps/server/src/prism/StreamStatsStore.ts create mode 100644 apps/server/src/prism/staleTurnDetector.test.ts create mode 100644 apps/server/src/prism/staleTurnDetector.ts create mode 100644 apps/server/src/prism/staleTurnMonitor.integration.test.ts create mode 100644 apps/server/src/prism/staleTurnMonitor.ts create mode 100644 apps/server/src/prism/streamClock.integration.test.ts create mode 100644 apps/server/src/prism/streamClock.test.ts create mode 100644 apps/server/src/prism/streamClock.ts diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 5f3e36ffd6..79964e53a6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/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,16 @@ export const layer: Layer.Layer< } return true; }); + yield* streamClock.beginAttempt({ + threadId: input.run.threadId, + runId: input.run.id, + runOrdinal: input.run.ordinal, + attemptId: input.attempt.id, + attemptOrdinal: input.attempt.attemptOrdinal, + providerThreadId: input.providerThread.id, + provider: input.session.driver, + model: input.modelSelection.model, + }); const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode); const providerEventFiber = yield* eventSubscription.events.pipe( Stream.filterEffect((event) => @@ -1203,37 +1215,39 @@ export const layer: Layer.Layer< const isRootProviderThreadUpdate = event.type === "provider_thread.updated" && event.providerThread.id === input.providerThread.id; - const storedEvents = yield* providerEventIngestor.ingestNormalized({ - analyticsContext: { - modelSelection: input.modelSelection, - runtimeMode: input.runtimePolicy.runtimeMode, - interactionMode: input.runtimePolicy.interactionMode, - }, - providerSessionId: input.providerSessionId, - providerInstanceId: input.run.providerInstanceId, - threadId: input.run.threadId, - runId: input.run.id, - nodeId: input.rootNode.id, - event: deliveredEvent, - ...(isRootProviderThreadUpdate - ? rootTerminalAlreadySeen - ? { - writeIfProviderThreadOwner: { - providerThreadId: input.providerThread.id, - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedLastRunOrdinal: input.run.ordinal, - }, - } - : { - writeIfRunCurrent: { - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedStatus: "running" as const, - }, - } - : {}), - }); + const storedEvents = yield* providerEventIngestor + .ingestNormalized({ + analyticsContext: { + modelSelection: input.modelSelection, + runtimeMode: input.runtimePolicy.runtimeMode, + interactionMode: input.runtimePolicy.interactionMode, + }, + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + threadId: input.run.threadId, + runId: input.run.id, + nodeId: input.rootNode.id, + event: deliveredEvent, + ...(isRootProviderThreadUpdate + ? rootTerminalAlreadySeen + ? { + writeIfProviderThreadOwner: { + providerThreadId: input.providerThread.id, + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedLastRunOrdinal: input.run.ordinal, + }, + } + : { + writeIfRunCurrent: { + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + }, + } + : {}), + }) + .pipe(Effect.provideService(StreamClock.StreamClockAttempt, input.attempt.id)); storedEventCount = storedEvents.length; if ( isRootProviderThreadUpdate && @@ -1245,6 +1259,19 @@ export const layer: Layer.Layer< yield* Ref.set(providerThreadOwnerLost, true); } } + if (deliveredEvent === null) { + // Presentation buffering must not turn healthy provider text into silence. + yield* streamClock.observe( + { + threadId: input.run.threadId, + runId: input.run.id, + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + event, + }, + input.attempt.id, + ); + } if (event.type === "provider_thread.updated") { if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) { yield* Ref.set(latestProviderThread, event.providerThread); @@ -1337,6 +1364,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, ); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index c73383e159..02b783219c 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/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( diff --git a/apps/server/src/prism/ProviderEventIngestor.ts b/apps/server/src/prism/ProviderEventIngestor.ts new file mode 100644 index 0000000000..d9971c5024 --- /dev/null +++ b/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, + ), + ), + }); + }), +); diff --git a/apps/server/src/prism/StreamStatsStore.test.ts b/apps/server/src/prism/StreamStatsStore.test.ts new file mode 100644 index 0000000000..a5b2371dea --- /dev/null +++ b/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 = (database: Layer.Layer) => + StreamStatsStore.layer.pipe(Layer.provide(database)); + +let turnNumber = 0; +const turn = ( + overrides: Partial = {}, +): 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)), + ); +}); diff --git a/apps/server/src/prism/StreamStatsStore.ts b/apps/server/src/prism/StreamStatsStore.ts new file mode 100644 index 0000000000..ee5b9eac14 --- /dev/null +++ b/apps/server/src/prism/StreamStatsStore.ts @@ -0,0 +1,162 @@ +import { ProviderDriverKind, ProviderTurnId, RunId, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as SqlSchema from "effect/sql/SqlSchema"; + +const Milliseconds = Schema.Number.check(Schema.isFinite()); +const Duration = Milliseconds.check(Schema.isGreaterThanOrEqualTo(0)); +const Count = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)); + +export const StreamTurnOutcome = Schema.Literals(["completed", "aborted", "error"]); +export type StreamTurnOutcome = typeof StreamTurnOutcome.Type; + +/** The final stream measurements of one finished provider turn. */ +export const TurnStreamStats = Schema.Struct({ + threadId: ThreadId, + runId: RunId, + providerTurnId: ProviderTurnId, + provider: ProviderDriverKind, + model: Schema.NullOr(Schema.String), + outcome: StreamTurnOutcome, + /** Epoch milliseconds. */ + startedAt: Milliseconds, + endedAt: Milliseconds, + /** From turn start to the first content event; null when none streamed. */ + timeToFirstTokenMs: Schema.NullOr(Duration), + /** Longest silence between two events while no tool call was open. */ + maxGapMs: Duration, + eventCount: Count, +}); +export type TurnStreamStats = typeof TurnStreamStats.Type; + +/** + * Healthy turns of one provider and model: completed, with a first token. Each such turn is one + * first-token sample and one mid-stream sample, so both phases share `count`. + */ +export const HealthyStreamSample = Schema.Struct({ + provider: ProviderDriverKind, + model: Schema.NullOr(Schema.String), + count: Count, + maxTimeToFirstTokenMs: Duration, + maxGapMs: Duration, +}); +export type HealthyStreamSample = typeof HealthyStreamSample.Type; + +export class StreamStatsWriteError extends Schema.TaggedError()( + "StreamStatsWriteError", + { + runId: Schema.String, + providerTurnId: Schema.String, + cause: Schema.Defect(), + }, +) { + override get message(): string { + return "Failed to record Prism stream statistics."; + } +} + +export class StreamStatsReadError extends Schema.TaggedError()( + "StreamStatsReadError", + { cause: Schema.Defect() }, +) { + override get message(): string { + return "Failed to read Prism stream statistics."; + } +} + +/** + * Finished-turn stream statistics in the fork-owned `fork_prism_stream_stats` table, outside + * upstream migrations. One row per provider turn of a run; per-event clocks stay in memory. + */ +export class StreamStatsStore extends Context.Service< + StreamStatsStore, + { + /** + * Records a finished turn once. Returns false when that run's provider turn is already + * recorded; a retried write never replaces the first measurements. + */ + readonly record: (stats: TurnStreamStats) => Effect.Effect; + /** Healthy-turn counts and maxima per provider and model, across all recorded history. */ + readonly healthySamples: Effect.Effect< + ReadonlyArray, + StreamStatsReadError + >; + } +>()("t3/prism/StreamStatsStore") {} + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS fork_prism_stream_stats ( + run_id TEXT NOT NULL, + provider_turn_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + provider TEXT NOT NULL, + model TEXT, + outcome TEXT NOT NULL, + started_at REAL NOT NULL, + ended_at REAL NOT NULL, + time_to_first_token_ms REAL, + max_gap_ms REAL NOT NULL, + event_count INTEGER NOT NULL, + PRIMARY KEY (run_id, provider_turn_id) + ) + `; + + const insert = SqlSchema.findAll({ + Request: TurnStreamStats, + Result: Schema.Struct({ runId: Schema.String }), + execute: (stats) => sql` + INSERT INTO fork_prism_stream_stats ( + run_id, provider_turn_id, thread_id, provider, model, outcome, + started_at, ended_at, time_to_first_token_ms, max_gap_ms, event_count + ) VALUES ( + ${stats.runId}, ${stats.providerTurnId}, ${stats.threadId}, ${stats.provider}, + ${stats.model}, ${stats.outcome}, ${stats.startedAt}, ${stats.endedAt}, + ${stats.timeToFirstTokenMs}, ${stats.maxGapMs}, ${stats.eventCount} + ) + ON CONFLICT (run_id, provider_turn_id) DO NOTHING + RETURNING run_id AS "runId" + `, + }); + + const healthy = SqlSchema.findAll({ + Request: Schema.Void, + Result: HealthyStreamSample, + execute: () => sql` + SELECT + provider, + model, + COUNT(*) AS count, + MAX(time_to_first_token_ms) AS "maxTimeToFirstTokenMs", + MAX(max_gap_ms) AS "maxGapMs" + FROM fork_prism_stream_stats + WHERE outcome = 'completed' AND time_to_first_token_ms IS NOT NULL + GROUP BY provider, model + `, + }); + + return StreamStatsStore.of({ + record: (stats) => + insert(stats).pipe( + Effect.map((rows) => rows.length > 0), + Effect.mapError( + (cause) => + new StreamStatsWriteError({ + runId: stats.runId, + providerTurnId: stats.providerTurnId, + cause, + }), + ), + ), + healthySamples: healthy(undefined).pipe( + Effect.mapError((cause) => new StreamStatsReadError({ cause })), + ), + }); +}); + +export const layer = Layer.effect(StreamStatsStore, make); diff --git a/apps/server/src/prism/staleTurnDetector.test.ts b/apps/server/src/prism/staleTurnDetector.test.ts new file mode 100644 index 0000000000..ab3134d08e --- /dev/null +++ b/apps/server/src/prism/staleTurnDetector.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, it } from "@effect/vitest"; +import { + ProviderDriverKind, + ProviderThreadId, + ProviderTurnId, + RunAttemptId, + RunId, + ThreadId, +} from "@t3tools/contracts"; +import { makeStaleTurnDetectorState } from "./staleTurnDetector.ts"; +import type { ThreadLiveness } from "./streamClock.ts"; + +const live: ThreadLiveness = { + threadId: ThreadId.make("child"), + runId: RunId.make("run"), + runOrdinal: 1, + attemptId: RunAttemptId.make("attempt"), + attemptOrdinal: 1, + providerThreadId: ProviderThreadId.make("native"), + providerTurnId: ProviderTurnId.make("turn"), + provider: ProviderDriverKind.make("codex"), + model: "fixed", + startedAt: 0, + firstTokenAt: null, + lastStreamAt: 0, + lastEventKind: "attempt.started", + openTool: null, + eventCount: 0, +}; +describe("silence detector", () => { + it("uses default confirmed thresholds and reports once per turn", () => { + const detector = makeStaleTurnDetectorState(); + expect(detector.check(live, 150_000, [])).toBeNull(); + expect(detector.check(live, 150_001, [])).toMatchObject({ + thresholdMs: 120_000, + thresholdSource: "default", + }); + detector.notified(live); + expect( + detector.check({ ...live, firstTokenAt: 160_000, lastStreamAt: 160_000 }, 400_000, []), + ).toBeNull(); + expect( + detector.check( + { ...live, providerTurnId: ProviderTurnId.make("next"), lastStreamAt: 200_000 }, + 400_000, + [], + ), + ).not.toBeNull(); + }); + it("uses persisted healthy samples only after twenty matching turns", () => { + const detector = makeStaleTurnDetectorState(); + const sample = { + provider: live.provider, + model: live.model, + count: 20, + maxTimeToFirstTokenMs: 4_000, + maxGapMs: 2_000, + }; + expect(detector.check(live, 7_500, [sample])).toBeNull(); + expect(detector.check(live, 7_501, [sample])).toMatchObject({ + thresholdMs: 6_000, + thresholdSource: "measured", + }); + expect(makeStaleTurnDetectorState().check(live, 7_501, [{ ...sample, count: 19 }])).toBeNull(); + expect( + makeStaleTurnDetectorState().check(live, 7_501, [{ ...sample, model: "other" }]), + ).toBeNull(); + expect( + makeStaleTurnDetectorState().check({ ...live, firstTokenAt: 1_000 }, 3_751, [sample]), + ).toMatchObject({ thresholdMs: 3_000 }); + }); + it("does not treat host suspension or a busy event loop as provider silence", () => { + const detector = makeStaleTurnDetectorState(); + detector.tick(0, 0, [live]); + detector.tick(300_000, 1_000, [live]); + expect(detector.check(live, 300_000, [])).toBeNull(); + detector.tick(600_000, 301_000, [live]); + expect(detector.check(live, 600_000, [])).toBeNull(); + expect(detector.check(live, 750_001, [])).not.toBeNull(); + }); + it("suspends silence during open tools and requires a fresh window afterwards", () => { + const detector = makeStaleTurnDetectorState(); + const tool = { ...live, firstTokenAt: 1_000, openTool: { itemId: "tool", startedAt: 2_000 } }; + detector.tick(300_000, 300_000, [tool]); + expect(detector.check(tool, 300_000, [])).toBeNull(); + const closed = { ...tool, openTool: null, lastStreamAt: 300_001 }; + expect(detector.check(closed, 400_000, [])).toBeNull(); + expect(detector.check(closed, 412_502, [])).not.toBeNull(); + }); +}); diff --git a/apps/server/src/prism/staleTurnDetector.ts b/apps/server/src/prism/staleTurnDetector.ts new file mode 100644 index 0000000000..b6501c1bfa --- /dev/null +++ b/apps/server/src/prism/staleTurnDetector.ts @@ -0,0 +1,90 @@ +import type { ThreadId } from "@t3tools/contracts"; +import type { HealthyStreamSample } from "./StreamStatsStore.ts"; +import type { ThreadLiveness } from "./streamClock.ts"; + +export interface StaleTurn { + readonly live: ThreadLiveness; + readonly silenceMs: number; + readonly thresholdMs: number; + readonly thresholdSource: "measured" | "default"; +} + +/** Silence only: terminal provider failures are owned by v2 and Prism's persisted recovery. */ +export function makeStaleTurnDetectorState() { + const watched = new Map< + ThreadId, + { + key: Pick; + silenceFrom: number; + lastEventAt: number; + notified: boolean; + } + >(); + let previousTick: { wall: number; monotonic: number } | undefined; + const watch = (live: ThreadLiveness) => { + let state = watched.get(live.threadId); + if ( + !state || + state.key.runId !== live.runId || + state.key.attemptId !== live.attemptId || + state.key.providerTurnId !== live.providerTurnId + ) { + state = { + key: live, + silenceFrom: live.lastStreamAt, + lastEventAt: live.lastStreamAt, + notified: false, + }; + watched.set(live.threadId, state); + } + if (live.lastStreamAt > state.lastEventAt) { + state.lastEventAt = live.lastStreamAt; + state.silenceFrom = live.lastStreamAt; + } + return state; + }; + return { + tick: (now: number, monotonic: number, lives: ReadonlyArray) => { + const wake = + previousTick !== undefined && + (Math.abs(now - previousTick.wall - (monotonic - previousTick.monotonic)) > 1_000 || + monotonic - previousTick.monotonic > 2_500 || + now < previousTick.wall); + previousTick = { wall: now, monotonic }; + const active = new Set(lives.map((live) => live.threadId)); + for (const threadId of watched.keys()) if (!active.has(threadId)) watched.delete(threadId); + for (const live of lives) { + const state = watch(live); + // Host sleep or a suspended event loop is not evidence of provider silence. + if (wake || live.openTool !== null) state.silenceFrom = now; + } + }, + pause: (live: ThreadLiveness, now: number) => { + watch(live).silenceFrom = now; + }, + notified: (live: ThreadLiveness) => { + watch(live).notified = true; + }, + check: ( + live: ThreadLiveness, + now: number, + samples: ReadonlyArray, + ): StaleTurn | null => { + const state = watch(live); + if (state.notified || live.openTool !== null) return null; + const sample = samples.find( + (sample) => sample.provider === live.provider && sample.model === live.model, + ); + const firstToken = live.firstTokenAt === null; + const max = firstToken ? sample?.maxTimeToFirstTokenMs : sample?.maxGapMs; + const measured = sample !== undefined && sample.count >= 20; + const thresholdMs = measured + ? (max ?? 0) * 1.5 + : Math.max(firstToken ? 120_000 : 90_000, max ?? 0); + const silenceMs = Math.max(0, now - state.silenceFrom); + return silenceMs > thresholdMs * 1.25 + ? { live, silenceMs, thresholdMs, thresholdSource: measured ? "measured" : "default" } + : null; + }, + }; +} diff --git a/apps/server/src/prism/staleTurnMonitor.integration.test.ts b/apps/server/src/prism/staleTurnMonitor.integration.test.ts new file mode 100644 index 0000000000..02b70f283d --- /dev/null +++ b/apps/server/src/prism/staleTurnMonitor.integration.test.ts @@ -0,0 +1,244 @@ +import { expect, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ProviderThreadId, + RunAttemptId, + RuntimeRequestId, + ThreadId, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as TestClock from "effect/testing/TestClock"; +import * as Registry from "../orchestration-v2/ProviderAdapterRegistry.ts"; +import type { ProviderAdapterV2Shape } from "../orchestration-v2/ProviderAdapter.ts"; +import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as EventSink from "../orchestration-v2/EventSink.ts"; +import * as Harness from "../orchestration-v2/testkit/ProviderReplayHarness.ts"; +import * as Threads from "../orchestration-v2/ThreadManagementService.ts"; +import * as Persistence from "../persistence/Sqlite.ts"; +import { ServerActivation } from "../serverActivation.ts"; +import * as Monitor from "./staleTurnMonitor.ts"; +import * as StreamClock from "./streamClock.ts"; +import * as Stats from "./StreamStatsStore.ts"; + +const instanceId = ProviderInstanceId.make("codex"); +const modelSelection = { + instanceId, + model: "fixed", + options: [{ id: "reasoningEffort", value: "high" }], +}; +const parentId = ThreadId.make("silence:parent"); +const database = Persistence.layerMemory; +const adapter = { + instanceId, + driver: ProviderDriverKind.make("codex"), + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" as const }), + openSession: () => Effect.die("No live provider in this durable dispatch proof"), +} as ProviderAdapterV2Shape; +const replay = Harness.layerWithRegistry( + { name: "silence" }, + Registry.layerFromAdapters([adapter]), + { databaseLayer: database, runEffectWorker: false }, +); +const clock = StreamClock.layer.pipe(Layer.provide(Stats.layer), Layer.provide(database)); +const dependencies = Layer.mergeAll( + replay, + clock, + Threads.layer.pipe(Layer.provide(replay)), + Layer.succeed(ServerActivation, Effect.never), + Layer.succeed(Monitor.StaleTurnMonotonicClock, Clock.currentTimeMillis), +); +const layer = Layer.mergeAll(dependencies, Monitor.layer.pipe(Layer.provide(dependencies))); + +const setup = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("silence:create"), + threadId: parentId, + projectId: ProjectId.make("silence:project"), + title: "Parent", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("silence:start"), + threadId: parentId, + messageId: MessageId.make("silence:message"), + text: "Delegate", + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }); + const parent = (yield* orchestrator.getThreadProjection(parentId)).runs[0]!; + yield* orchestrator.dispatch({ + type: "delegated_task.request", + commandId: CommandId.make("silence:delegate"), + parentThreadId: parentId, + parentRunId: parent.id, + parentNodeId: parent.rootNodeId!, + task: "Work", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + completionWake: "always", + createdBy: "agent", + creationSource: "mcp", + }); + const childId = (yield* orchestrator.getThreadProjection(parentId)).subagents[0]!.childThreadId!; + const run = (yield* orchestrator.getThreadProjection(childId)).runs[0]!; + const activeAttemptId = RunAttemptId.make("silence:attempt"); + const now = yield* DateTime.now; + yield* (yield* EventSink.EventSinkV2).write({ + events: [ + { + id: EventId.make("silence:running"), + type: "run.updated", + threadId: childId, + occurredAt: now, + payload: { ...run, status: "running", activeAttemptId, startedAt: now }, + }, + ], + }); + yield* (yield* StreamClock.StreamClock).beginAttempt({ + threadId: childId, + runId: run.id, + runOrdinal: run.ordinal, + attemptId: activeAttemptId, + attemptOrdinal: 1, + providerThreadId: ProviderThreadId.make("silence:native"), + provider: adapter.driver, + model: modelSelection.model, + }); + return { orchestrator, childId, run, activeAttemptId }; +}); +const advanceSilence = (epoch: number) => + Effect.gen(function* () { + const monitor = yield* Monitor.StaleTurnMonitor; + // Drive real sweep drains at its cadence; a jump would correctly be host sleep. + for (let second = 0; second <= 151; second += 1) { + yield* TestClock.setTime(epoch + second * 1_000); + yield* monitor.sweep; + } + }); + +it.effect("queues one parent notice with stored effort and leaves the silent child running", () => + Effect.gen(function* () { + const epoch = yield* Clock.currentTimeMillis; + const { orchestrator, childId, run } = yield* setup; + yield* advanceSilence(epoch); + yield* (yield* Monitor.StaleTurnMonitor).sweep; + const parent = yield* orchestrator.getThreadProjection(parentId); + const notices = parent.messages.filter((message) => message.senderThreadId === childId); + expect(notices).toHaveLength(1); + expect(notices[0]!.text).toContain("Reason: silence."); + expect(notices[0]!.text).toContain(run.id); + expect(notices[0]!.creationSource).toBe("server"); + expect( + parent.runs.find((candidate) => candidate.userMessageId === notices[0]!.id), + ).toMatchObject({ status: "queued", modelSelection }); + expect((yield* orchestrator.getThreadProjection(childId)).runs[0]).toMatchObject({ + id: run.id, + status: "running", + }); + }).pipe(Effect.provide(layer)), +); + +it.effect( + "does not notify after a propagated Stop and preserves its original retirement token", + () => + Effect.gen(function* () { + const epoch = yield* Clock.currentTimeMillis; + const { orchestrator, childId } = yield* setup; + const token = CommandId.make("silence:stop"); + yield* orchestrator.dispatch({ type: "thread.stop", commandId: token, threadId: parentId }); + yield* advanceSilence(epoch); + const parent = yield* orchestrator.getThreadProjection(parentId); + expect(parent.thread.forkRetirement?.token).toBe(token); + expect(parent.messages.filter((message) => message.senderThreadId === childId)).toEqual([]); + expect( + (yield* orchestrator.getThreadProjection(childId)).thread.forkResumedRetirements ?? [], + ).not.toContain(token); + }).pipe(Effect.provide(layer)), +); + +it.effect("pending input and a replaced attempt cannot produce a silence notice", () => + Effect.gen(function* () { + const epoch = yield* Clock.currentTimeMillis; + const { orchestrator, childId, run } = yield* setup; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const request = { + id: RuntimeRequestId.make("silence:input"), + nodeId: NodeId.make("silence:input-node"), + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input" as const, + status: "pending" as const, + responseCapability: { type: "message" as const }, + createdAt: now, + resolvedAt: null, + }; + yield* sink.write({ + events: [ + { + id: EventId.make("silence:input-event"), + type: "runtime-request.updated", + threadId: childId, + occurredAt: now, + payload: request, + }, + ], + }); + yield* advanceSilence(epoch); + expect( + (yield* orchestrator.getThreadProjection(parentId)).messages.filter( + (message) => message.senderThreadId === childId, + ), + ).toEqual([]); + const active = (yield* orchestrator.getThreadProjection(childId)).runs.find( + (candidate) => candidate.id === run.id, + )!; + yield* sink.write({ + events: [ + { + id: EventId.make("silence:resolved"), + type: "runtime-request.updated", + threadId: childId, + occurredAt: now, + payload: { ...request, status: "resolved", resolvedAt: now }, + }, + { + id: EventId.make("silence:replacement"), + type: "run.updated", + threadId: childId, + occurredAt: now, + payload: { ...active, activeAttemptId: RunAttemptId.make("silence:replacement-attempt") }, + }, + ], + }); + yield* advanceSilence(epoch + 152_000); + expect( + (yield* orchestrator.getThreadProjection(parentId)).messages.filter( + (message) => message.senderThreadId === childId, + ), + ).toEqual([]); + }).pipe(Effect.provide(layer)), +); diff --git a/apps/server/src/prism/staleTurnMonitor.ts b/apps/server/src/prism/staleTurnMonitor.ts new file mode 100644 index 0000000000..f023308ff2 --- /dev/null +++ b/apps/server/src/prism/staleTurnMonitor.ts @@ -0,0 +1,123 @@ +import { CommandId, MessageId } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Semaphore from "effect/Semaphore"; + +import { readRetirementState } from "../childThreads/retirement.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import { forkParked } from "../serverActivation.ts"; +import { makeStaleTurnDetectorState, type StaleTurn } from "./staleTurnDetector.ts"; +import * as StreamClock from "./streamClock.ts"; + +export class StaleTurnMonotonicClock extends Context.Reference>( + "t3/prism/StaleTurnMonotonicClock", + { + defaultValue: () => Effect.sync(() => performance.now()), + }, +) {} + +const notice = (stale: StaleTurn) => + `[Stale child thread ${stale.live.threadId}]\n` + + `Model: ${stale.live.model ?? "unknown"} (${stale.live.provider}).\n` + + `Run: ${stale.live.runId}; attempt: ${stale.live.attemptId}.\n` + + `Last event: ${DateTime.formatIso(DateTime.makeUnsafe(stale.live.lastStreamAt))} (${stale.live.lastEventKind}).\n` + + `Silence: ${stale.silenceMs} ms; threshold: ${stale.thresholdMs} ms (${stale.thresholdSource}).\nReason: silence.`; + +export class StaleTurnMonitor extends Context.Service< + StaleTurnMonitor, + { + readonly sweep: Effect.Effect; + } +>()("t3/prism/staleTurnMonitor") {} + +const make = Effect.gen(function* () { + const clock = yield* StreamClock.StreamClock; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threads = yield* ThreadManagement.ThreadManagementService; + const monotonic = yield* StaleTurnMonotonicClock; + const permit = yield* Semaphore.make(1); + const detector = makeStaleTurnDetectorState(); + const sweep = Effect.gen(function* () { + const now = yield* Clock.currentTimeMillis; + const lives = yield* clock.liveness; + const samples = yield* clock.healthySamples; + detector.tick(now, yield* monotonic, lives); + for (const live of lives) { + const stale = detector.check(live, now, samples); + if (stale === null) continue; + // Projection reads happen only at a suspected silence, never per token or per tick. + yield* Effect.gen(function* () { + const child = yield* projections.getThreadRecords(live.threadId, [ + "runs", + "runtimeRequests", + ]); + const run = child.runs.find((run) => run.id === live.runId); + const parentId = + child.thread.lineage.relationshipToParent === "subagent" + ? child.thread.lineage.parentThreadId + : null; + const retirement = yield* readRetirementState(live.threadId).pipe( + Effect.provideService(ProjectionStore.ProjectionStoreV2, projections), + ); + if ( + run?.status !== "running" || + run.activeAttemptId !== live.attemptId || + child.thread.archivedAt !== null || + child.thread.deletedAt !== null || + child.runtimeRequests.some((request) => request.status === "pending") || + retirement.retired || + !retirement.complete || + parentId === null + ) { + detector.pause(live, now); + return; + } + const parent = yield* projections.getThread(parentId); + if (parent.archivedAt !== null || parent.deletedAt !== null) { + detector.pause(live, now); + return; + } + const key = `server:prism-stale:${live.runId}:${live.attemptId}:${live.providerTurnId ?? "pending"}`; + yield* threads.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(key), + messageId: MessageId.make(key), + threadId: parent.id, + senderThreadId: child.thread.id, + text: notice(stale), + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "system", + creationSource: "server", + }); + detector.notified(live); + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("prism.stale-turn.notice-failed", { + threadId: live.threadId, + cause, + }), + ), + ); + } + }).pipe( + permit.withPermits(1), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("prism.stale-turn.sweep-failed", { cause }), + ), + ); + // This scoped timer needs a one-second cadence for the existing suspend detector. + yield* forkParked(sweep.pipe(Effect.andThen(Effect.sleep("1 second")), Effect.forever)); + return StaleTurnMonitor.of({ sweep }); +}); + +export const layer = Layer.effect(StaleTurnMonitor, make); diff --git a/apps/server/src/prism/streamClock.integration.test.ts b/apps/server/src/prism/streamClock.integration.test.ts new file mode 100644 index 0000000000..33a95aecb2 --- /dev/null +++ b/apps/server/src/prism/streamClock.integration.test.ts @@ -0,0 +1,346 @@ +import { expect, it } from "@effect/vitest"; +import { + CheckpointScopeId, + CommandId, + EventId, + MessageId, + NodeId, + ProviderDriverKind, + ProviderSessionId, + ProviderThreadId, + ProviderTurnId, + RunAttemptId, + type OrchestrationV2ExecutionNode, + type OrchestrationV2ProviderThread, + type OrchestrationV2RunAttempt, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as AppModelContext from "../mcpApps/McpAppModelContext.ts"; +import * as Persistence from "../persistence/Sqlite.ts"; +import * as Settings from "../serverSettings.ts"; +import * as Checkpoints from "../orchestration-v2/CheckpointService.ts"; +import * as EventStore from "../orchestration-v2/EventStore.ts"; +import * as EventSink from "../orchestration-v2/EventSink.ts"; +import * as Projection from "../orchestration-v2/ProjectionStore.ts"; +import * as IdAllocator from "../orchestration-v2/IdAllocator.ts"; +import * as Ingestor from "../orchestration-v2/ProviderEventIngestor.ts"; +import type { + ProviderAdapterV2Event, + ProviderAdapterV2SessionRuntime, +} from "../orchestration-v2/ProviderAdapter.ts"; +import * as RunExecution from "../orchestration-v2/RunExecutionService.ts"; +import * as ThreadCommands from "../orchestration-v2/ThreadCommandExecutor.ts"; +import * as Decorator from "./ProviderEventIngestor.ts"; +import * as Stats from "./StreamStatsStore.ts"; +import * as StreamClock from "./streamClock.ts"; +import { recoveryRun, threadFor } from "./recovery.testkit.ts"; + +it.effect.each([ + { mode: "paragraph", unfinished: false }, + { mode: "turn", unfinished: false }, + { mode: "paragraph", unfinished: true }, +] as const)( + "measures original text and terminal-only completion: $mode unfinished=$unfinished", + ({ mode, unfinished }) => { + const interval = unfinished ? 60_000 : 1_000; + return Effect.gen(function* () { + const epoch = yield* Clock.currentTimeMillis; + const at = DateTime.makeUnsafe(epoch); + const driver = ProviderDriverKind.make("codex"); + const nodeId = NodeId.make("clock:root"); + const providerThreadId = ProviderThreadId.make("clock:native"); + const providerTurnId = ProviderTurnId.make("clock:turn"); + const providerSessionId = ProviderSessionId.make("clock:session"); + const attemptId = RunAttemptId.make("clock:attempt"); + const run = { + ...recoveryRun, + status: "running" as const, + rootNodeId: nodeId, + activeAttemptId: attemptId, + providerThreadId, + requestedAt: at, + startedAt: at, + completedAt: null, + }; + const appThread = threadFor(run); + const rootNode: OrchestrationV2ExecutionNode = { + id: nodeId, + threadId: run.threadId, + runId: run.id, + parentNodeId: null, + rootNodeId: nodeId, + kind: "root_turn", + status: "running", + countsForRun: true, + providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: at, + completedAt: null, + }; + const attempt: OrchestrationV2RunAttempt = { + id: attemptId, + runId: run.id, + attemptOrdinal: 1, + rootNodeId: nodeId, + providerInstanceId: run.providerInstanceId, + providerThreadId, + providerTurnId: null, + reason: "initial", + status: "running", + startedAt: at, + completedAt: null, + }; + const providerThread: OrchestrationV2ProviderThread = { + id: providerThreadId, + driver, + providerInstanceId: run.providerInstanceId, + providerSessionId, + appThreadId: run.threadId, + ownerNodeId: nodeId, + nativeThreadRef: null, + nativeConversationHeadRef: null, + status: "active", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: at, + updatedAt: at, + }; + const drained = yield* Deferred.make(); + const database = Persistence.layerMemory; + const stores = Layer.mergeAll( + database, + EventStore.layer.pipe(Layer.provide(database)), + Projection.layer.pipe(Layer.provide(database)), + ); + const sink = EventSink.layer.pipe(Layer.provide(stores)); + const clockLayer = StreamClock.layer.pipe( + Layer.provide(Stats.layer), + Layer.provide(database), + ); + const ingestor = Decorator.layer.pipe( + Layer.provide( + Layer.merge( + clockLayer, + Ingestor.layer.pipe( + Layer.provide(Layer.mergeAll(sink, stores, IdAllocator.layer, ThreadCommands.layer)), + ), + ), + ), + ); + const hooks = Layer.effect( + StreamClock.StreamClockHooks, + Effect.map(StreamClock.StreamClock, (clock) => ({ + ...clock, + endAttempt: (...args: Parameters) => + clock + .endAttempt(...args) + .pipe(Effect.andThen(Deferred.succeed(drained, undefined)), Effect.asVoid), + })), + ).pipe(Layer.provide(clockLayer)); + const execution = RunExecution.layer.pipe( + Layer.provide( + Layer.mergeAll( + ingestor, + hooks, + sink, + IdAllocator.layer, + AppModelContext.layerEmpty, + Settings.layerTest({ responseStreamingMode: mode }), + Layer.mock(Checkpoints.CheckpointServiceV2)({ captureBaseline: () => Effect.void }), + ), + ), + ); + const testLayer = Layer.mergeAll( + execution, + ingestor, + clockLayer, + stores, + sink, + Stats.layer.pipe(Layer.provide(database)), + ); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const clock = yield* StreamClock.StreamClock; + const base = { threadId: run.threadId, occurredAt: at }; + yield* eventSink.write({ + events: [ + { + ...base, + id: EventId.make("clock:thread"), + type: "thread.created", + payload: appThread, + }, + { ...base, id: EventId.make("clock:run"), type: "run.created", payload: run }, + { ...base, id: EventId.make("clock:node"), type: "node.updated", payload: rootNode }, + { + ...base, + id: EventId.make("clock:attempt"), + type: "run-attempt.updated", + payload: attempt, + }, + { + ...base, + id: EventId.make("clock:provider"), + type: "provider-thread.updated", + payload: providerThread, + }, + ], + }); + const message = (text: string): ProviderAdapterV2Event => ({ + type: "message.updated", + driver, + message: { + createdBy: "agent", + creationSource: "provider", + id: MessageId.make("clock:answer"), + threadId: run.threadId, + runId: run.id, + nodeId, + role: "assistant", + text, + attachments: [], + streaming: true, + createdAt: at, + updatedAt: at, + }, + }); + const events: ReadonlyArray = [ + { + type: "provider_turn.updated", + driver, + threadId: run.threadId, + providerTurn: { + id: providerTurnId, + providerThreadId, + nodeId, + runAttemptId: attemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: at, + completedAt: null, + }, + }, + message("An unfinished paragraph"), + message( + unfinished + ? "An unfinished paragraph keeps growing" + : "First paragraph.\n\nAn unfinished paragraph", + ), + message( + unfinished + ? "An unfinished paragraph keeps growing for three minutes" + : "First paragraph.\n\nAn even longer unfinished paragraph", + ), + { + type: "turn.terminal", + driver, + providerThreadId, + providerTurnId, + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", + }, + ]; + const stream = Stream.fromIterable( + events.map((event, index) => ({ event, index })), + { chunkSize: 1 }, + ).pipe( + Stream.mapEffect(({ event, index }) => + Effect.gen(function* () { + // The next source event is a drain boundary for the preceding ingest/tap. + if (event.type === "turn.terminal") { + expect(yield* clock.liveness).toMatchObject([ + { + firstTokenAt: epoch + interval, + lastStreamAt: epoch + 3 * interval, + eventCount: 4, + }, + ]); + const normalized = yield* (yield* Ingestor.ProviderEventIngestorV2).normalize({ + threadId: run.threadId, + runId: run.id, + providerSessionId, + providerInstanceId: run.providerInstanceId, + event, + }); + expect(normalized).toEqual([]); + } + yield* TestClock.setTime(epoch + index * interval); + return event; + }), + ), + ); + // The fake is only the external provider boundary; execution, filtering, + // normalization, EventSink and SQLite are the production implementations. + const session = { + driver, + events: stream, + startTurn: () => Effect.void, + } as unknown as ProviderAdapterV2SessionRuntime; + yield* (yield* RunExecution.RunExecutionServiceV2).startRootRun({ + commandId: CommandId.make("clock:start"), + appThread, + providerSessionId, + session, + run, + rootNode, + checkpointScope: { + id: CheckpointScopeId.make("clock:scope"), + threadId: run.threadId, + runId: run.id, + nodeId, + parentScopeId: null, + providerThreadId, + kind: "root_run", + ordinalWithinParent: 0, + advancesAppRunCount: true, + cwd: process.cwd(), + createdAt: at, + }, + providerThread, + attempt, + attemptId, + providerTurnOrdinal: 1, + message: { + messageId: run.userMessageId, + text: "Work", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + modelSelection: run.modelSelection, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }, + }); + yield* Deferred.await(drained); + expect(yield* clock.liveness).toEqual([]); + const sql = yield* SqlClient.SqlClient; + expect( + yield* sql`SELECT outcome, time_to_first_token_ms AS firstToken, event_count AS eventCount FROM fork_prism_stream_stats`, + ).toEqual([{ outcome: "completed", firstToken: interval, eventCount: 5 }]); + const projection = yield* Projection.ProjectionStoreV2; + const records = yield* projection.getThreadRecords(run.threadId, ["messages"]); + expect(records.messages.map((m) => m.text)).toEqual( + mode === "turn" || unfinished ? [] : ["First paragraph.\n\n"], + ); + }).pipe(Effect.provide(testLayer)); + }); + }, +); diff --git a/apps/server/src/prism/streamClock.test.ts b/apps/server/src/prism/streamClock.test.ts new file mode 100644 index 0000000000..927331da60 --- /dev/null +++ b/apps/server/src/prism/streamClock.test.ts @@ -0,0 +1,211 @@ +import { describe, expect, it } from "@effect/vitest"; +import { + MessageId, + NodeId, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ProviderThreadId, + ProviderTurnId, + RunAttemptId, + RunId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import type { ProviderAdapterV2Event } from "../orchestration-v2/ProviderAdapter.ts"; +import { makeStreamClockState } from "./streamClock.ts"; + +const identity = { + threadId: ThreadId.make("child"), + runId: RunId.make("run"), + runOrdinal: 1, + attemptId: RunAttemptId.make("attempt"), + attemptOrdinal: 1, + providerThreadId: ProviderThreadId.make("native"), + provider: ProviderDriverKind.make("codex"), + model: "fixed", +}; +const turnId = ProviderTurnId.make("turn"); +const now = DateTime.makeUnsafe(0); +const running: ProviderAdapterV2Event = { + type: "provider_turn.updated", + driver: identity.provider, + threadId: identity.threadId, + providerTurn: { + id: turnId, + providerThreadId: identity.providerThreadId, + nodeId: NodeId.make("node"), + runAttemptId: identity.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, +}; +const text: ProviderAdapterV2Event = { + type: "message.updated", + driver: identity.provider, + message: { + createdBy: "agent", + creationSource: "provider", + id: MessageId.make("answer"), + threadId: identity.threadId, + runId: identity.runId, + nodeId: NodeId.make("node"), + role: "assistant", + text: "An unfinished paragraph", + attachments: [], + streaming: true, + createdAt: now, + updatedAt: now, + }, +}; +const terminal: ProviderAdapterV2Event = { + type: "turn.terminal", + driver: identity.provider, + providerThreadId: identity.providerThreadId, + providerTurnId: turnId, + runOrdinal: 1, + status: "completed", + failure: null, + threadDisposition: "reusable", +}; +const input = (event: ProviderAdapterV2Event) => ({ + threadId: identity.threadId, + runId: identity.runId, + providerInstanceId: ProviderInstanceId.make("codex"), + providerSessionId: ProviderSessionId.make("session"), + event, +}); + +describe("stream clock ownership and turn measurements", () => { + it("records a terminal-only completion once and releases its liveness", () => { + const clock = makeStreamClockState(); + clock.beginAttempt(identity, 1_000); + clock.observe(input(running), 1_000, identity.attemptId); + clock.observe(input(text), 1_500, identity.attemptId); + clock.observe(input(text), 2_000, identity.attemptId); + expect(clock.observe(input(terminal), 3_000, identity.attemptId)).toMatchObject({ + outcome: "completed", + timeToFirstTokenMs: 500, + maxGapMs: 1_000, + eventCount: 4, + }); + expect(clock.observe(input(terminal), 4_000, identity.attemptId)).toBeNull(); + expect( + clock.endAttempt(identity.threadId, identity.runId, identity.attemptId, 5_000), + ).toBeNull(); + expect(clock.liveness()).toEqual([]); + }); + + it("fences old attempt text and cleanup after same-run recovery", () => { + const clock = makeStreamClockState(); + clock.beginAttempt(identity, 1_000); + clock.observe(input(running), 1_000, identity.attemptId); + const next = { ...identity, attemptId: RunAttemptId.make("retry"), attemptOrdinal: 2 }; + expect(clock.beginAttempt(next, 2_000)).toMatchObject({ + runId: identity.runId, + providerTurnId: turnId, + outcome: "aborted", + endedAt: 2_000, + }); + expect(clock.beginAttempt(identity, 2_100)).toBeNull(); + clock.observe(input(text), 10_000, identity.attemptId); + clock.observe(input(running), 10_000, identity.attemptId); + expect( + clock.endAttempt(identity.threadId, identity.runId, identity.attemptId, 10_000), + ).toBeNull(); + expect(clock.liveness()).toMatchObject([ + { attemptId: next.attemptId, firstTokenAt: null, lastStreamAt: 2_000, eventCount: 0 }, + ]); + clock.observe(input(text), 2_100, next.attemptId); + expect(clock.liveness()).toMatchObject([{ firstTokenAt: 2_100, eventCount: 1 }]); + }); + + it("does not count child-provider text or another run as root activity", () => { + const clock = makeStreamClockState(); + clock.beginAttempt(identity, 1_000); + clock.observe(input(running), 1_100, identity.attemptId); + clock.observe({ ...input(text), runId: RunId.make("other") }, 9_000, identity.attemptId); + clock.observe( + input({ + ...text, + message: { ...text.message, threadId: ThreadId.make("native-child"), runId: null }, + }), + 9_000, + identity.attemptId, + ); + expect(clock.liveness()).toMatchObject([ + { firstTokenAt: null, lastStreamAt: 1_100, eventCount: 1 }, + ]); + }); + + it("measures each provider turn separately when one attempt continues", () => { + const clock = makeStreamClockState(); + clock.beginAttempt(identity, 1_000); + clock.observe(input(running), 1_000, identity.attemptId); + clock.observe(input(text), 1_500, identity.attemptId); + expect(clock.observe(input(terminal), 2_000, identity.attemptId)).toMatchObject({ + providerTurnId: turnId, + startedAt: 1_000, + timeToFirstTokenMs: 500, + }); + const nextTurnId = ProviderTurnId.make("next-turn"); + clock.observe( + input({ ...running, providerTurn: { ...running.providerTurn, id: nextTurnId, ordinal: 2 } }), + 3_000, + identity.attemptId, + ); + clock.observe(input(text), 3_100, identity.attemptId); + expect( + clock.observe(input({ ...terminal, providerTurnId: nextTurnId }), 4_000, identity.attemptId), + ).toMatchObject({ + providerTurnId: nextTurnId, + startedAt: 3_000, + timeToFirstTokenMs: 100, + }); + expect( + clock.endAttempt(identity.threadId, identity.runId, identity.attemptId, 5_000), + ).toBeNull(); + }); + + it("tracks overlapping tools and excludes their execution time from healthy gaps", () => { + const clock = makeStreamClockState(); + clock.beginAttempt(identity, 1_000); + clock.observe(input(running), 1_000, identity.attemptId); + const tool = (id: string, status: "running" | "completed"): ProviderAdapterV2Event => ({ + type: "turn_item.updated", + driver: identity.provider, + turnItem: { + id: TurnItemId.make(id), + threadId: identity.threadId, + runId: identity.runId, + nodeId: NodeId.make("node"), + providerThreadId: identity.providerThreadId, + providerTurnId: turnId, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status, + title: null, + startedAt: now, + completedAt: status === "completed" ? now : null, + updatedAt: now, + type: "command_execution", + input: "work", + }, + }); + clock.observe(input(text), 1_100, identity.attemptId); + clock.observe(input(tool("first", "running")), 1_200, identity.attemptId); + clock.observe(input(tool("second", "running")), 100_000, identity.attemptId); + clock.observe(input(tool("first", "completed")), 200_000, identity.attemptId); + expect(clock.liveness()).toMatchObject([{ openTool: { itemId: "second" } }]); + clock.observe(input(tool("second", "completed")), 300_000, identity.attemptId); + expect(clock.liveness()).toMatchObject([{ openTool: null }]); + expect(clock.observe(input(terminal), 300_100, identity.attemptId)).toMatchObject({ + maxGapMs: 100, + }); + }); +}); diff --git a/apps/server/src/prism/streamClock.ts b/apps/server/src/prism/streamClock.ts new file mode 100644 index 0000000000..c4f4337a12 --- /dev/null +++ b/apps/server/src/prism/streamClock.ts @@ -0,0 +1,346 @@ +import { + isToolLifecycleItemType, + type OrchestrationV2Run, + type OrchestrationV2RunAttempt, + type ProviderDriverKind, + type ProviderThreadId, + type ProviderTurnId, + type RunAttemptId, + type RunId, + type ThreadId, +} from "@t3tools/contracts"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import type { ProviderEventIngestInput } from "../orchestration-v2/ProviderEventIngestor.ts"; +import * as StreamStatsStore from "./StreamStatsStore.ts"; + +interface AttemptIdentity { + readonly threadId: ThreadId; + readonly runId: RunId; + readonly runOrdinal: OrchestrationV2Run["ordinal"]; + readonly attemptId: RunAttemptId; + readonly attemptOrdinal: OrchestrationV2RunAttempt["attemptOrdinal"]; + readonly providerThreadId: ProviderThreadId; + readonly provider: ProviderDriverKind; + readonly model: string | null; +} + +export interface ThreadLiveness extends AttemptIdentity { + readonly providerTurnId: ProviderTurnId | null; + readonly startedAt: number; + readonly firstTokenAt: number | null; + readonly lastStreamAt: number; + readonly lastEventKind: string; + readonly openTool: { readonly itemId: string; readonly startedAt: number } | null; + readonly eventCount: number; +} + +interface AttemptState { + readonly identity: AttemptIdentity; + lastProviderTurnOrdinal: number; + turn: { + providerTurnId: ProviderTurnId | null; + startedAt: number; + firstTokenAt: number | null; + lastStreamAt: number; + lastEventKind: string; + maxGapMs: number; + eventCount: number; + readonly tools: Map; + } | null; +} + +const newTurn = (now: number): NonNullable => ({ + providerTurnId: null, + startedAt: now, + firstTokenAt: null, + lastStreamAt: now, + lastEventKind: "attempt.started", + maxGapMs: 0, + eventCount: 0, + tools: new Map(), +}); + +/** Attempt registration fences late events and cleanup without consulting SQL per event. */ +export function makeStreamClockState() { + const attempts = new Map(); + const finish = ( + state: AttemptState, + now: number, + outcome: StreamStatsStore.StreamTurnOutcome, + ) => { + const turn = state.turn; + state.turn = null; + if (turn?.providerTurnId === null || turn === null) return null; + return { + threadId: state.identity.threadId, + runId: state.identity.runId, + providerTurnId: turn.providerTurnId, + provider: state.identity.provider, + model: state.identity.model, + outcome, + startedAt: turn.startedAt, + endedAt: now, + timeToFirstTokenMs: turn.firstTokenAt === null ? null : turn.firstTokenAt - turn.startedAt, + maxGapMs: turn.maxGapMs, + eventCount: turn.eventCount, + } satisfies StreamStatsStore.TurnStreamStats; + }; + return { + beginAttempt: (identity: AttemptIdentity, now: number) => { + const previous = attempts.get(identity.threadId); + if ( + previous && + (previous.identity.runOrdinal > identity.runOrdinal || + (previous.identity.runOrdinal === identity.runOrdinal && + previous.identity.attemptOrdinal >= identity.attemptOrdinal)) + ) + return null; + const displaced = previous === undefined ? null : finish(previous, now, "aborted"); + attempts.set(identity.threadId, { identity, lastProviderTurnOrdinal: 0, turn: newTurn(now) }); + return displaced; + }, + observe: (input: ProviderEventIngestInput, now: number, attemptId?: RunAttemptId) => { + const state = attempts.get(input.threadId); + if (!state || input.runId !== state.identity.runId || attemptId !== state.identity.attemptId) + return null; + const event = input.event; + if (event.type === "provider_turn.updated") { + const providerTurn = event.providerTurn; + if ( + providerTurn.runAttemptId !== state.identity.attemptId || + providerTurn.providerThreadId !== state.identity.providerThreadId || + providerTurn.ordinal < state.lastProviderTurnOrdinal + ) + return null; + if (state.turn === null && providerTurn.ordinal <= state.lastProviderTurnOrdinal) + return null; + if (providerTurn.ordinal > state.lastProviderTurnOrdinal) { + // One Codex run can contain several provider turns on the same attempt. + if (state.lastProviderTurnOrdinal > 0) state.turn = newTurn(now); + state.lastProviderTurnOrdinal = providerTurn.ordinal; + } + if (state.turn) state.turn.providerTurnId = providerTurn.id; + } + const turn = state.turn; + if (!turn) return null; + switch (event.type) { + case "message.updated": + if (event.message.threadId !== input.threadId || event.message.runId !== input.runId) + return null; + break; + case "turn_item.updated": + if ( + event.turnItem.threadId !== input.threadId || + event.turnItem.runId !== input.runId || + (event.turnItem.providerTurnId !== null && + event.turnItem.providerTurnId !== turn.providerTurnId) + ) + return null; + break; + case "turn.terminal": + if ( + event.providerTurnId !== turn.providerTurnId || + event.providerThreadId !== state.identity.providerThreadId + ) + return null; + break; + case "provider_turn.updated": + break; + case "node.updated": + if (event.node.threadId !== input.threadId || event.node.runId !== input.runId) + return null; + break; + case "provider_thread.updated": + if (event.providerThread.id !== state.identity.providerThreadId) return null; + break; + case "plan.updated": + if (event.plan.threadId !== input.threadId || event.plan.runId !== input.runId) + return null; + break; + case "runtime_request.updated": + if (event.runtimeRequest.providerTurnId !== turn.providerTurnId) return null; + break; + default: + return null; + } + if (turn.tools.size === 0) + turn.maxGapMs = Math.max(turn.maxGapMs, Math.max(0, now - turn.lastStreamAt)); + turn.lastStreamAt = now; + turn.lastEventKind = event.type; + turn.eventCount += 1; + if ( + (event.type === "message.updated" && + event.message.role === "assistant" && + event.message.text.length > 0) || + (event.type === "turn_item.updated" && + (event.turnItem.type === "assistant_message" || event.turnItem.type === "reasoning") && + event.turnItem.text.length > 0) + ) + turn.firstTokenAt ??= now; + if (event.type === "turn_item.updated" && isToolLifecycleItemType(event.turnItem.type)) { + if (event.turnItem.status === "running" || event.turnItem.status === "pending") { + if (!turn.tools.has(event.turnItem.id)) + turn.tools.set(event.turnItem.id, { itemId: event.turnItem.id, startedAt: now }); + } else turn.tools.delete(event.turnItem.id); + } + const terminalStatus = + event.type === "turn.terminal" + ? event.status + : event.type === "provider_turn.updated" + ? event.providerTurn.status + : undefined; + if ( + terminalStatus === "completed" || + terminalStatus === "failed" || + terminalStatus === "interrupted" || + terminalStatus === "cancelled" + ) { + return finish( + state, + now, + terminalStatus === "completed" + ? "completed" + : terminalStatus === "failed" + ? "error" + : "aborted", + ); + } + return null; + }, + endAttempt: (threadId: ThreadId, runId: RunId, attemptId: RunAttemptId, now: number) => { + const state = attempts.get(threadId); + if (!state || state.identity.runId !== runId || state.identity.attemptId !== attemptId) + return null; + const stats = finish(state, now, "aborted"); + attempts.delete(threadId); + return stats; + }, + liveness: (): ReadonlyArray => + [...attempts.values()].flatMap(({ identity, turn }) => + turn === null + ? [] + : [ + { + ...identity, + providerTurnId: turn.providerTurnId, + startedAt: turn.startedAt, + firstTokenAt: turn.firstTokenAt, + lastStreamAt: turn.lastStreamAt, + lastEventKind: turn.lastEventKind, + openTool: turn.tools.values().next().value ?? null, + eventCount: turn.eventCount, + }, + ], + ), + }; +} + +export class StreamClock extends Context.Service< + StreamClock, + { + readonly beginAttempt: (input: AttemptIdentity) => Effect.Effect; + readonly observe: ( + input: ProviderEventIngestInput, + attemptId?: RunAttemptId, + ) => Effect.Effect; + readonly endAttempt: ( + threadId: ThreadId, + runId: RunId, + attemptId: RunAttemptId, + ) => Effect.Effect; + readonly liveness: Effect.Effect>; + readonly healthySamples: Effect.Effect>; + } +>()("t3/prism/streamClock") {} + +/** Default no-op keeps upstream harnesses independent of the production fork clock. */ +export class StreamClockHooks extends Context.Reference< + Pick +>("t3/prism/StreamClockHooks", { + defaultValue: () => ({ + beginAttempt: () => Effect.void, + observe: () => Effect.void, + endAttempt: () => Effect.void, + }), +}) {} + +/** The captured execution attempt accompanies delivered events without changing the ingestor input. */ +export class StreamClockAttempt extends Context.Reference( + "t3/prism/StreamClockAttempt", + { + defaultValue: () => undefined, + }, +) {} + +const make = Effect.gen(function* () { + const store = yield* StreamStatsStore.StreamStatsStore; + const state = makeStreamClockState(); + const samples = new Map< + ProviderDriverKind, + Map + >(); + const setSample = (sample: StreamStatsStore.HealthyStreamSample) => { + let models = samples.get(sample.provider); + if (models === undefined) { + models = new Map(); + samples.set(sample.provider, models); + } + models.set(sample.model, sample); + }; + for (const sample of yield* store.healthySamples) setSample(sample); + const record = (stats: StreamStatsStore.TurnStreamStats | null) => + stats === null + ? Effect.void + : store.record(stats).pipe( + Effect.tap((inserted) => + Effect.sync(() => { + if (!inserted || stats.outcome !== "completed" || stats.timeToFirstTokenMs === null) + return; + const previous = samples.get(stats.provider)?.get(stats.model); + setSample({ + provider: stats.provider, + model: stats.model, + count: (previous?.count ?? 0) + 1, + maxTimeToFirstTokenMs: Math.max( + previous?.maxTimeToFirstTokenMs ?? 0, + stats.timeToFirstTokenMs, + ), + maxGapMs: Math.max(previous?.maxGapMs ?? 0, stats.maxGapMs), + }); + }), + ), + Effect.catchTags({ + StreamStatsWriteError: (cause) => + Effect.logWarning("prism.stream-stats.write-failed", { cause }), + }), + Effect.asVoid, + ); + return StreamClock.of({ + beginAttempt: (identity) => + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => record(state.beginAttempt(identity, now))), + ), + observe: (input, attemptId) => + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => record(state.observe(input, now, attemptId))), + ), + endAttempt: (threadId, runId, attemptId) => + Clock.currentTimeMillis.pipe( + Effect.flatMap((now) => record(state.endAttempt(threadId, runId, attemptId, now))), + ), + liveness: Effect.sync(state.liveness), + healthySamples: Effect.sync(() => + [...samples.values()].flatMap((models) => [...models.values()]), + ), + }); +}); + +const layerClock = Layer.effect(StreamClock, make); +export const layer = Layer.effect( + StreamClockHooks, + Effect.map(StreamClock, (clock) => clock), +).pipe(Layer.provideMerge(layerClock)); diff --git a/docs/fork-features.md b/docs/fork-features.md index 2644567026..515ee2698f 100644 --- a/docs/fork-features.md +++ b/docs/fork-features.md @@ -753,3 +753,51 @@ Child creation paths are owned by child-threads and shared with thread-people. ] } ``` + +## prism-stream-clock + +```json +{ + "id": "prism-stream-clock", + "purpose": "Measure attempt-owned provider activity in memory and persist one idempotent healthy sample per finished provider turn across restarts, without importing v1 activities.", + "issues": ["https://github.com/toolboxmd/chromeria/issues/172"], + "prs": ["https://github.com/toolboxmd/chromeria/compare/fork/v2...feat/172-stream-clock-v2"], + "newFiles": [ + "apps/server/src/prism/ProviderEventIngestor.ts", + "apps/server/src/prism/StreamStatsStore.ts", + "apps/server/src/prism/StreamStatsStore.test.ts", + "apps/server/src/prism/streamClock.ts", + "apps/server/src/prism/streamClock.test.ts", + "apps/server/src/prism/streamClock.integration.test.ts" + ], + "upstreamFiles": ["apps/server/src/orchestration-v2/RunExecutionService.ts"], + "sharedFiles": ["apps/server/src/orchestration-v2/runtimeLayer.ts"], + "keywords": [ + "StreamClock", + "StreamStatsStore", + "fork_prism_stream_stats", + "StreamClockAttempt", + "StreamClockHooks" + ] +} +``` + +## prism-stale-turn-detector + +```json +{ + "id": "prism-stale-turn-detector", + "purpose": "Send one advisory parent notice for a silent active child; exclude host suspension, open tools and retired or waiting work. V2 and Prism recovery retain terminal failure delivery.", + "issues": ["https://github.com/toolboxmd/chromeria/issues/172"], + "prs": ["https://github.com/toolboxmd/chromeria/compare/fork/v2...feat/172-stream-clock-v2"], + "newFiles": [ + "apps/server/src/prism/staleTurnDetector.ts", + "apps/server/src/prism/staleTurnDetector.test.ts", + "apps/server/src/prism/staleTurnMonitor.ts", + "apps/server/src/prism/staleTurnMonitor.integration.test.ts" + ], + "upstreamFiles": [], + "sharedFiles": ["apps/server/src/orchestration-v2/runtimeLayer.ts"], + "keywords": ["StaleTurnMonitor", "makeStaleTurnDetectorState", "queue_after_active", "silence"] +} +``` diff --git a/scripts/fork-upstream-edits.txt b/scripts/fork-upstream-edits.txt index 92e5d2229b..c8f19b4c3d 100644 --- a/scripts/fork-upstream-edits.txt +++ b/scripts/fork-upstream-edits.txt @@ -149,3 +149,6 @@ apps/web/src/routeTree.gen.ts packages/contracts/src/index.ts apps/web/src/components/settings/SettingsPanels.restore.test.tsx apps/server/src/orchestration-v2/testkit/ReplayTranscriptNdjson.ts + +# Attempt-owned stream observation before presentation buffering (#172) +apps/server/src/orchestration-v2/RunExecutionService.ts From 57a59b4fa1d26abe842a12827e776428dc2f2b9f Mon Sep 17 00:00:00 2001 From: lukemaj Date: Thu, 8 Oct 2026 22:04:37 +0200 Subject: [PATCH 2/2] refactor(prism): shrink execution stream clock hooks --- .../orchestration-v2/RunExecutionService.ts | 88 +++++++------------ apps/server/src/prism/streamClock.ts | 34 +++++++ docs/fork-features.md | 4 +- 3 files changed, 70 insertions(+), 56 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 79964e53a6..da244023d6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1183,16 +1183,7 @@ export const layer: Layer.Layer< } return true; }); - yield* streamClock.beginAttempt({ - threadId: input.run.threadId, - runId: input.run.id, - runOrdinal: input.run.ordinal, - attemptId: input.attempt.id, - attemptOrdinal: input.attempt.attemptOrdinal, - providerThreadId: input.providerThread.id, - provider: input.session.driver, - model: input.modelSelection.model, - }); + yield* StreamClock.beginExecutionAttempt(streamClock, input); const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode); const providerEventFiber = yield* eventSubscription.events.pipe( Stream.filterEffect((event) => @@ -1215,39 +1206,37 @@ export const layer: Layer.Layer< const isRootProviderThreadUpdate = event.type === "provider_thread.updated" && event.providerThread.id === input.providerThread.id; - const storedEvents = yield* providerEventIngestor - .ingestNormalized({ - analyticsContext: { - modelSelection: input.modelSelection, - runtimeMode: input.runtimePolicy.runtimeMode, - interactionMode: input.runtimePolicy.interactionMode, - }, - providerSessionId: input.providerSessionId, - providerInstanceId: input.run.providerInstanceId, - threadId: input.run.threadId, - runId: input.run.id, - nodeId: input.rootNode.id, - event: deliveredEvent, - ...(isRootProviderThreadUpdate - ? rootTerminalAlreadySeen - ? { - writeIfProviderThreadOwner: { - providerThreadId: input.providerThread.id, - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedLastRunOrdinal: input.run.ordinal, - }, - } - : { - writeIfRunCurrent: { - runId: input.run.id, - activeAttemptId: input.attempt.id, - expectedStatus: "running" as const, - }, - } - : {}), - }) - .pipe(Effect.provideService(StreamClock.StreamClockAttempt, input.attempt.id)); + const storedEvents = yield* providerEventIngestor.ingestNormalized({ + analyticsContext: { + modelSelection: input.modelSelection, + runtimeMode: input.runtimePolicy.runtimeMode, + interactionMode: input.runtimePolicy.interactionMode, + }, + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + threadId: input.run.threadId, + runId: input.run.id, + nodeId: input.rootNode.id, + event: deliveredEvent, + ...(isRootProviderThreadUpdate + ? rootTerminalAlreadySeen + ? { + writeIfProviderThreadOwner: { + providerThreadId: input.providerThread.id, + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedLastRunOrdinal: input.run.ordinal, + }, + } + : { + writeIfRunCurrent: { + runId: input.run.id, + activeAttemptId: input.attempt.id, + expectedStatus: "running" as const, + }, + } + : {}), + }); storedEventCount = storedEvents.length; if ( isRootProviderThreadUpdate && @@ -1260,17 +1249,7 @@ export const layer: Layer.Layer< } } if (deliveredEvent === null) { - // Presentation buffering must not turn healthy provider text into silence. - yield* streamClock.observe( - { - threadId: input.run.threadId, - runId: input.run.id, - providerSessionId: input.providerSessionId, - providerInstanceId: input.run.providerInstanceId, - event, - }, - input.attempt.id, - ); + yield* StreamClock.observeFilteredEvent(streamClock, input, event); } if (event.type === "provider_thread.updated") { if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) { @@ -1295,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(() => diff --git a/apps/server/src/prism/streamClock.ts b/apps/server/src/prism/streamClock.ts index c4f4337a12..84016c7942 100644 --- a/apps/server/src/prism/streamClock.ts +++ b/apps/server/src/prism/streamClock.ts @@ -14,6 +14,7 @@ import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import type { RunExecutionServiceV2StartRootRunInput } from "../orchestration-v2/RunExecutionService.ts"; import type { ProviderEventIngestInput } from "../orchestration-v2/ProviderEventIngestor.ts"; import * as StreamStatsStore from "./StreamStatsStore.ts"; @@ -344,3 +345,36 @@ export const layer = Layer.effect( StreamClockHooks, Effect.map(StreamClock, (clock) => clock), ).pipe(Layer.provideMerge(layerClock)); + +/** Register the captured execution owner without expanding the upstream lifecycle hook. */ +export const beginExecutionAttempt = ( + clock: Pick, + input: RunExecutionServiceV2StartRootRunInput, +) => + clock.beginAttempt({ + threadId: input.run.threadId, + runId: input.run.id, + runOrdinal: input.run.ordinal, + attemptId: input.attempt.id, + attemptOrdinal: input.attempt.attemptOrdinal, + providerThreadId: input.providerThread.id, + provider: input.session.driver, + model: input.modelSelection.model, + }); + +/** Presentation buffering must not turn healthy provider text into silence. */ +export const observeFilteredEvent = ( + clock: Pick, + input: RunExecutionServiceV2StartRootRunInput, + event: ProviderEventIngestInput["event"], +) => + clock.observe( + { + threadId: input.run.threadId, + runId: input.run.id, + providerSessionId: input.providerSessionId, + providerInstanceId: input.run.providerInstanceId, + event, + }, + input.attempt.id, + ); diff --git a/docs/fork-features.md b/docs/fork-features.md index 515ee2698f..e242f55885 100644 --- a/docs/fork-features.md +++ b/docs/fork-features.md @@ -761,7 +761,7 @@ Child creation paths are owned by child-threads and shared with thread-people. "id": "prism-stream-clock", "purpose": "Measure attempt-owned provider activity in memory and persist one idempotent healthy sample per finished provider turn across restarts, without importing v1 activities.", "issues": ["https://github.com/toolboxmd/chromeria/issues/172"], - "prs": ["https://github.com/toolboxmd/chromeria/compare/fork/v2...feat/172-stream-clock-v2"], + "prs": ["https://github.com/toolboxmd/chromeria/pull/184"], "newFiles": [ "apps/server/src/prism/ProviderEventIngestor.ts", "apps/server/src/prism/StreamStatsStore.ts", @@ -789,7 +789,7 @@ Child creation paths are owned by child-threads and shared with thread-people. "id": "prism-stale-turn-detector", "purpose": "Send one advisory parent notice for a silent active child; exclude host suspension, open tools and retired or waiting work. V2 and Prism recovery retain terminal failure delivery.", "issues": ["https://github.com/toolboxmd/chromeria/issues/172"], - "prs": ["https://github.com/toolboxmd/chromeria/compare/fork/v2...feat/172-stream-clock-v2"], + "prs": ["https://github.com/toolboxmd/chromeria/pull/184"], "newFiles": [ "apps/server/src/prism/staleTurnDetector.ts", "apps/server/src/prism/staleTurnDetector.test.ts",