1import { assert, describe, it } from "@effect/vitest";
2import {
3 EventId,
4 MessageId,
5 ProjectId,
6 ProviderInstanceId,
7 RunId,
8 ThreadId,
9 type OrchestrationV2AppThread,
10 type OrchestrationV2DomainEvent,
11 type OrchestrationV2Run,
12 type OrchestrationV2StoredEvent,
13} from "@t3tools/contracts";
14import * as DateTime from "effect/DateTime";
15import * as Deferred from "effect/Deferred";
16import * as Effect from "effect/Effect";
17import * as Fiber from "effect/Fiber";
18import * as Layer from "effect/Layer";
19import * as Metric from "effect/Metric";
20import * as Option from "effect/Option";
21import * as SqlClient from "effect/sql/SqlClient";
22
23import * as SqlitePersistence from "../persistence/Sqlite.ts";
24import * as EventSink from "./EventSink.ts";
25import * as EventStore from "./EventStore.ts";
26import * as ProjectionStore from "./ProjectionStore.ts";
27import { makeRunOutcomeRecorder } from "./RunMetrics.ts";
28
29const at = (iso: string) => DateTime.makeUnsafe(iso);
30const threadId = ThreadId.make("thread:metrics");
31
32// Each test labels its runs with its own provider, since the metric registry is shared.
33const runUpdated = (
34 sequence: number,
35 provider: string,
36 run: Pick<OrchestrationV2Run, "id" | "status" | "startedAt" | "completedAt"> &
37 Partial<OrchestrationV2Run>,
38): OrchestrationV2StoredEvent => ({
39 sequence,
40 commandId: null,
41 event: {
42 type: "run.updated",
43 id: EventId.make(`event:${provider}:${sequence}`),
44 threadId,
45 occurredAt: at("2026-10-19T12:00:00.000Z"),
46 payload: {
47 threadId,
48 ordinal: 1,
49 providerInstanceId: ProviderInstanceId.make(provider),
50 modelSelection: { instanceId: ProviderInstanceId.make(provider), model: "claude-opus-5-5" },
51 requestedAt: at("2026-10-19T12:00:00.000Z"),
52 ...run,
53 } as OrchestrationV2Run,
54 },
55});
56
57// Prepare inside the transaction, record after its commit, as the EventSink does.
58const commitWith =
59 (recorder: ReturnType<typeof makeRunOutcomeRecorder<unknown>>) =>
60 (events: ReadonlyArray<OrchestrationV2StoredEvent>) =>
61 recorder.prepare(events).pipe(Effect.flatMap(recorder.record));
62
63const runSnapshots = (provider: string) =>
64 Metric.snapshot.pipe(
65 Effect.map((snapshots) =>
66 snapshots.filter(
67 (snapshot) =>
68 snapshot.id.startsWith("t3_run") && snapshot.attributes?.["provider"] === provider,
69 ),
70 ),
71 );
72
73describe("makeRunOutcomeRecorder", () => {
74 it.effect("records a finished run once, labelled, with its queue wait and duration", () =>
75 Effect.gen(function* () {
76 // A subagent thread's first run carries no delegatedTaskId; only its lineage says delegated.
77 const recorder = makeRunOutcomeRecorder(() => Effect.succeed(true));
78 const record = commitWith(recorder);
79 const finished = {
80 id: RunId.make("run:metrics:once"),
81 status: "completed" as const,
82 startedAt: at("2026-10-19T12:00:30.000Z"),
83 completedAt: at("2026-10-19T12:10:30.000Z"),
84 };
85
86 yield* record([runUpdated(1, "p1", { ...finished, status: "running", completedAt: null })]);
87 // A transaction that rolls back prepares and never records; it must not reserve the run.
88 yield* recorder.prepare([runUpdated(2, "p1", finished)]);
89 yield* record([runUpdated(2, "p1", finished)]);
90 // Later updates to a finished run, such as its checkpoint, are not another run.
91 yield* record([runUpdated(3, "p1", finished)]);
92
93 const snapshots = yield* runSnapshots("p1");
94 const byId = (id: string) => snapshots.find((snapshot) => snapshot.id === id);
95 const counter = byId("t3_runs_total");
96 assert.strictEqual(counter?.type === "Counter" ? counter.state.count : undefined, 1);
97 assert.deepStrictEqual(counter?.attributes, {
98 provider: "p1",
99 modelFamily: "claude",
100 outcome: "completed",
101 delegated: "true",
102 });
103 const histogramSum = (id: string) => {
104 const snapshot = byId(id);
105 return snapshot?.type === "Histogram" ? snapshot.state.sum : undefined;
106 };
107 assert.strictEqual(histogramSum("t3_run_queue_wait"), 30_000);
108 assert.strictEqual(histogramSum("t3_run_duration"), 600_000);
109 }),
110 );
111
112 it.effect("does not count a rollback of a run that already finished", () =>
113 Effect.gen(function* () {
114 // A fresh recorder stands in for a restart, which forgets the runs already counted.
115 const record = commitWith(makeRunOutcomeRecorder(() => Effect.succeed(false)));
116 yield* record([
117 runUpdated(1, "p2", {
118 id: RunId.make("run:metrics:rolled-back"),
119 status: "rolled_back",
120 startedAt: at("2026-10-19T12:00:30.000Z"),
121 completedAt: at("2026-10-20T09:00:00.000Z"),
122 }),
123 ]);
124
125 assert.deepStrictEqual(yield* runSnapshots("p2"), []);
126 }),
127 );
128});
129
130describe("EventSink run outcomes", () => {
131 const parentId = ThreadId.make("thread:metrics:sink:parent");
132 const childId = ThreadId.make("thread:metrics:sink:child");
133 const provider = ProviderInstanceId.make("p-sink");
134 const now = at("2026-10-19T12:00:00.000Z");
135 const thread = (id: ThreadId, lineage: OrchestrationV2AppThread["lineage"]) =>
136 ({
137 type: "thread.created",
138 id: EventId.make(`event:${id}:created`),
139 threadId: id,
140 occurredAt: now,
141 payload: {
142 createdBy: "user",
143 creationSource: "web",
144 id,
145 projectId: ProjectId.make("project:metrics"),
146 title: id,
147 providerInstanceId: provider,
148 modelSelection: { instanceId: provider, model: "gpt-6.1" },
149 runtimeMode: "full-access",
150 interactionMode: "default",
151 branch: null,
152 worktreePath: null,
153 activeProviderThreadId: null,
154 lineage,
155 forkedFrom: null,
156 createdAt: now,
157 updatedAt: now,
158 archivedAt: null,
159 settledOverride: null,
160 settledAt: null,
161 lastVisitedAt: null,
162 deletedAt: null,
163 },
164 }) satisfies OrchestrationV2DomainEvent;
165 const runId = RunId.make("run:metrics:sink:child");
166 const run = (status: OrchestrationV2Run["status"]): OrchestrationV2Run => ({
167 id: runId,
168 threadId: childId,
169 ordinal: 1,
170 providerInstanceId: provider,
171 modelSelection: { instanceId: provider, model: "gpt-6.1" },
172 providerThreadId: null,
173 userMessageId: MessageId.make("message:metrics:sink"),
174 rootNodeId: null,
175 activeAttemptId: null,
176 status,
177 queuePosition: null,
178 requestedAt: now,
179 startedAt: at("2026-10-19T12:00:10.000Z"),
180 completedAt: status === "running" ? null : at("2026-10-19T12:01:10.000Z"),
181 checkpointId: null,
182 contextHandoffId: null,
183 });
184 const runEvent = (type: "run.created" | "run.updated", status: OrchestrationV2Run["status"]) =>
185 ({
186 type,
187 id: EventId.make(`event:${runId}:${status}`),
188 threadId: childId,
189 runId,
190 occurredAt: now,
191 payload: run(status),
192 }) satisfies OrchestrationV2DomainEvent;
193
194 it.effect("reads lineage inside the transaction, so a concurrent writer cannot deadlock it", () =>
195 Effect.gen(function* () {
196 const lookupEntered = yield* Deferred.make<boolean>();
197 const releaseLookup = yield* Deferred.make<void>();
198 const writerStarted = yield* Deferred.make<void>();
199 const pausedLookup = Layer.effect(
200 ProjectionStore.ProjectionStoreV2,
201 Effect.gen(function* () {
202 const store = yield* ProjectionStore.ProjectionStoreV2;
203 const sql = yield* SqlClient.SqlClient;
204 let paused = false;
205 return {
206 ...store,
207 getThread: (id) =>
208 Effect.gen(function* () {
209 if (id === childId && !paused) {
210 paused = true;
211 const inTransaction = Option.isSome(
212 yield* Effect.serviceOption(sql.transactionService),
213 );
214 yield* Deferred.succeed(lookupEntered, inTransaction);
215 // Outside the transaction this read would wait on the writer it is racing.
216 if (inTransaction) yield* Deferred.await(releaseLookup);
217 }
218 return yield* store.getThread(id);
219 }),
220 };
221 }),
222 );
223 const layer = EventSink.layer.pipe(
224 Layer.provideMerge(pausedLookup),
225 Layer.provideMerge(
226 Layer.mergeAll(EventStore.layer, ProjectionStore.layer).pipe(
227 Layer.provideMerge(SqlitePersistence.layerMemory),
228 ),
229 ),
230 );
231
232 yield* Effect.gen(function* () {
233 const sink = yield* EventSink.EventSinkV2;
234 yield* sink.write({
235 events: [
236 thread(parentId, {
237 parentThreadId: null,
238 relationshipToParent: null,
239 rootThreadId: parentId,
240 }),
241 thread(childId, {
242 parentThreadId: parentId,
243 relationshipToParent: "subagent",
244 rootThreadId: parentId,
245 }),
246 runEvent("run.created", "running"),
247 ],
248 });
249 const terminal = yield* sink
250 .write({ events: [runEvent("run.updated", "completed")] })
251 .pipe(Effect.forkScoped);
252 assert.isTrue(yield* Deferred.await(lookupEntered), "lineage read outside the transaction");
253 const unrelated = yield* Deferred.succeed(writerStarted, undefined).pipe(
254 Effect.andThen(
255 sink.write({
256 events: [
257 thread(ThreadId.make("thread:metrics:sink:other"), {
258 parentThreadId: null,
259 relationshipToParent: null,
260 rootThreadId: ThreadId.make("thread:metrics:sink:other"),
261 }),
262 ],
263 }),
264 ),
265 Effect.forkScoped,
266 );
267 yield* Deferred.await(writerStarted);
268 yield* Deferred.succeed(releaseLookup, undefined);
269
270 const [terminalEvents, unrelatedEvents] = [
271 yield* Fiber.join(terminal),
272 yield* Fiber.join(unrelated),
273 ];
274 assert.strictEqual(terminalEvents.length, 1);
275 assert.strictEqual(unrelatedEvents.length, 1);
276 const counter = (yield* runSnapshots("p-sink")).find(
277 (snapshot) => snapshot.id === "t3_runs_total",
278 );
279 assert.strictEqual(counter?.type === "Counter" ? counter.state.count : undefined, 1);
280 assert.strictEqual(counter?.attributes?.["delegated"], "true");
281 }).pipe(Effect.provide(layer));
282 }),
283 );
284});