1import type {
2 OrchestrationV2ProviderFailure,
3 OrchestrationV2Run,
4 OrchestrationV2TurnItem,
5} from "@t3tools/contracts";
6import * as DateTime from "effect/DateTime";
7
8/** Only a failed root turn of the current run owns the thread's failure state. */
9export function latestRootProviderFailure(
10 run: OrchestrationV2Run | null,
11 turnItems: ReadonlyArray<OrchestrationV2TurnItem>,
12): OrchestrationV2ProviderFailure | null {
13 if (run?.status !== "failed") return null;
14 let latest: Extract<OrchestrationV2TurnItem, { type: "error" }> | null = null;
15 for (const item of turnItems) {
16 if (
17 item.type !== "error" ||
18 item.status !== "failed" ||
19 item.runId !== run.id ||
20 item.nodeId !== run.rootNodeId
21 )
22 continue;
23 if (
24 latest === null ||
25 DateTime.toEpochMillis(item.updatedAt) > DateTime.toEpochMillis(latest.updatedAt) ||
26 (DateTime.toEpochMillis(item.updatedAt) === DateTime.toEpochMillis(latest.updatedAt) &&
27 (item.ordinal > latest.ordinal || (item.ordinal === latest.ordinal && item.id > latest.id)))
28 ) {
29 latest = item;
30 }
31 }
32 return latest?.failure ?? null;
33}
34
35/** A distinct session failure supersedes the turn's classification. */
36export function threadErrorSummary(
37 failure: OrchestrationV2ProviderFailure | null,
38 sessionError: string | null,
39) {
40 const currentFailure =
41 sessionError !== null && sessionError !== failure?.message ? null : failure;
42 return {
43 usageLimitResetAt:
44 currentFailure?.class === "usage_limit" ? (currentFailure.resetAt ?? null) : null,
45 lastError: sessionError ?? failure?.message ?? null,
46 lastErrorClass:
47 sessionError !== null && sessionError !== failure?.message ? null : (failure?.class ?? null),
48 };
49}
50
51export function latestExecutedRun(
52 runs: ReadonlyArray<OrchestrationV2Run>,
53): OrchestrationV2Run | null {
54 let latest: OrchestrationV2Run | null = null;
55 for (const run of runs) {
56 if (run.status === "queued") continue;
57 if (run.status === "cancelled" && run.startedAt === null) continue;
58 if (latest === null || runRanAfter(run, latest)) latest = run;
59 }
60 return latest;
61}
62
63/**
64 * Whether started `run` ran after `other`. Ordinals follow submission, but a
65 * run can start ahead of a held queue (a restart continuation, or a message
66 * sent while the queue is held), so a queued run resumed later can have a
67 * lower ordinal than one that already ended. An unfinished run is the latest.
68 */
69export function runRanAfter(
70 run: Pick<OrchestrationV2Run, "ordinal" | "completedAt">,
71 other: Pick<OrchestrationV2Run, "ordinal" | "completedAt">,
72): boolean {
73 const end = (candidate: typeof run) =>
74 !candidate.completedAt
75 ? Number.POSITIVE_INFINITY
76 : DateTime.toEpochMillis(candidate.completedAt);
77 return end(run) === end(other) ? run.ordinal > other.ordinal : end(run) > end(other);
78}
79
80/**
81 * The latest run that actually started, when it stopped because the
82 * subscription limit was reached. Queued messages after that run must stay
83 * queued instead of being sent into the same limit.
84 */
85export function usageLimitBlockedRun(
86 runs: ReadonlyArray<OrchestrationV2Run>,
87 turnItems: ReadonlyArray<OrchestrationV2TurnItem>,
88 sessionError: string | null,
89): OrchestrationV2Run | null {
90 const executed = latestExecutedRun(runs);
91 if (executed?.status !== "failed") return null;
92 const summary = threadErrorSummary(latestRootProviderFailure(executed, turnItems), sessionError);
93 return summary.lastErrorClass === "usage_limit" ? executed : null;
94}
95
96/**
97 * That limited run, when newer runs were queued or cancelled before starting.
98 * Callers that treat the highest-ordinal run as the thread outcome would
99 * otherwise hide the limit.
100 */
101export function usageLimitRunPresentedAsLatest(
102 runs: ReadonlyArray<OrchestrationV2Run>,
103 turnItems: ReadonlyArray<OrchestrationV2TurnItem>,
104 sessionError: string | null,
105): OrchestrationV2Run | null {
106 const blocked = usageLimitBlockedRun(runs, turnItems, sessionError);
107 return blocked !== null && runs.some((run) => run.ordinal > blocked.ordinal) ? blocked : null;
108}
109
110/**
111 * The newest run that is not waiting in a held queue, or null when only held
112 * runs exist. A held queue waits for the user, so its runs never stand for the
113 * thread's outcome. The SQL thread shell selects the same run.
114 */
115export function latestUnheldRun(
116 runs: ReadonlyArray<OrchestrationV2Run>,
117): OrchestrationV2Run | null {
118 let latest: OrchestrationV2Run | null = null;
119 for (const run of runs) {
120 if (run.status === "queued" && run.queueHeld === true) continue;
121 if (latest === null || run.ordinal > latest.ordinal) latest = run;
122 }
123 return latest;
124}