1import type {
2 OrchestrationV2Run,
3 OrchestrationV2StoredEvent,
4 RunId,
5 ThreadId,
6} from "@t3tools/contracts";
7import * as DateTime from "effect/DateTime";
8import type * as Duration from "effect/Duration";
9import * as Effect from "effect/Effect";
10import * as Metric from "effect/Metric";
11
12import { normalizeModelMetricLabel } from "../observability/Attributes.ts";
13import {
14 metricAttributes,
15 runDuration,
16 runQueueWait,
17 runsTotal,
18} from "../observability/Metrics.ts";
19
20const REMEMBERED_RUNS = 4096;
21// A rollback rewrites runs that already finished, stamping the rollback time as their completion,
22// so it is not counted as an outcome. Declared here because ThreadManagementService imports
23// EventSink.
24const OUTCOMES: ReadonlySet<OrchestrationV2Run["status"]> = new Set([
25 "completed",
26 "failed",
27 "cancelled",
28 "interrupted",
29]);
30
31export interface RunOutcome {
32 readonly runId: RunId;
33 readonly attributes: ReturnType<typeof metricAttributes>;
34 readonly queueWait: Duration.Duration | undefined;
35 readonly duration: Duration.Duration | undefined;
36}
37
38/**
39 * Finished runs can change again; deduplicate outcomes and use lineage for first subagent runs.
40 * A failed lineage lookup must not block commit or publication.
41 * `prepare` must read inside the write transaction; `record` must do no I/O under the publish lane,
42 * or it can deadlock with a writer holding the connection while waiting for that lane.
43 */
44export const makeRunOutcomeRecorder = <E>(
45 isSubagentThread: (threadId: ThreadId) => Effect.Effect<boolean, E>,
46) => {
47 const counted = new Set<string>();
48 // Untraced: both run for every committed batch inside the EventSink operation's span.
49 // Never reads or writes `counted`, so a transaction that rolls back reserves nothing.
50 const prepare = Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2StoredEvent>) {
51 const outcomes: Array<RunOutcome> = [];
52 // Every finished run is prepared, in order, even one already counted: an earlier outcome in
53 // this batch may evict it before `record` reaches it, so only `record` decides what counts.
54 // Only the lineage read is shared.
55 const delegatedThreads = new Map<ThreadId, boolean>();
56 for (const { event } of events) {
57 if (event.type !== "run.updated") continue;
58 const run = event.payload;
59 if (!OUTCOMES.has(run.status)) continue;
60 let delegated = run.delegatedTaskId !== undefined || delegatedThreads.get(run.threadId);
61 if (delegated === undefined) {
62 delegated = yield* isSubagentThread(run.threadId).pipe(Effect.orElseSucceed(() => false));
63 delegatedThreads.set(run.threadId, delegated);
64 }
65 outcomes.push({
66 runId: run.id,
67 attributes: metricAttributes({
68 provider: run.providerInstanceId,
69 modelFamily: normalizeModelMetricLabel(run.modelSelection.model),
70 outcome: run.status,
71 delegated,
72 }),
73 queueWait:
74 run.startedAt === null ? undefined : DateTime.distance(run.requestedAt, run.startedAt),
75 duration:
76 run.startedAt === null || run.completedAt === null
77 ? undefined
78 : DateTime.distance(run.startedAt, run.completedAt),
79 });
80 }
81 return outcomes;
82 });
83 // Uninterruptible so a run is never marked counted with only part of its measurements.
84 const record = Effect.fnUntraced(function* (outcomes: ReadonlyArray<RunOutcome>) {
85 for (const outcome of outcomes) {
86 // Concurrent writers can prepare the same run; the first to publish counts it.
87 if (counted.has(outcome.runId)) continue;
88 counted.add(outcome.runId);
89 if (counted.size > REMEMBERED_RUNS) counted.delete(counted.values().next().value!);
90 yield* Metric.update(Metric.withAttributes(runsTotal, outcome.attributes), 1);
91 if (outcome.queueWait !== undefined) {
92 yield* Metric.update(
93 Metric.withAttributes(runQueueWait, outcome.attributes),
94 outcome.queueWait,
95 );
96 }
97 if (outcome.duration !== undefined) {
98 yield* Metric.update(
99 Metric.withAttributes(runDuration, outcome.attributes),
100 outcome.duration,
101 );
102 }
103 }
104 }, Effect.uninterruptible);
105 return { prepare, record };
106};