1import type {
2 OrchestrationV2ProviderFailure,
3 OrchestrationV2Run,
4 OrchestrationV2TurnItem,
5} from "@t3tools/contracts";
6import * as DateTime from "effect/DateTime";
7
8type ErrorTurnItem = Extract<OrchestrationV2TurnItem, { type: "error" }>;
9/** Only error items are read, so callers may pass items stripped of other fields. */
10type FailureScanTurnItem =
11 | ErrorTurnItem
12 | { readonly type: Exclude<OrchestrationV2TurnItem["type"], "error"> };
13
14export function latestRootProviderFailure(
15 run: OrchestrationV2Run | null,
16 turnItems: ReadonlyArray<FailureScanTurnItem>,
17): OrchestrationV2ProviderFailure | null {
18 if (run?.status !== "failed") return null;
19 let latest: ErrorTurnItem | null = null;
20 for (const item of turnItems) {
21 if (
22 item.type !== "error" ||
23 item.status !== "failed" ||
24 item.runId !== run.id ||
25 item.nodeId !== run.rootNodeId
26 )
27 continue;
28 if (
29 latest === null ||
30 DateTime.toEpochMillis(item.updatedAt) > DateTime.toEpochMillis(latest.updatedAt) ||
31 (DateTime.toEpochMillis(item.updatedAt) === DateTime.toEpochMillis(latest.updatedAt) &&
32 (item.ordinal > latest.ordinal || (item.ordinal === latest.ordinal && item.id > latest.id)))
33 ) {
34 latest = item;
35 }
36 }
37 return latest?.failure ?? null;
38}
39
40export function threadErrorSummary(
41 failure: OrchestrationV2ProviderFailure | null,
42 sessionError: string | null,
43) {
44 const currentFailure =
45 sessionError !== null && sessionError !== failure?.message ? null : failure;
46 return {
47 usageLimitResetAt:
48 currentFailure?.class === "usage_limit" ? (currentFailure.resetAt ?? null) : null,
49 lastError: sessionError ?? failure?.message ?? null,
50 lastErrorClass: currentFailure?.class ?? null,
51 lastErrorCode: currentFailure?.code ?? null,
52 };
53}
54
55export function latestExecutedRun(
56 runs: ReadonlyArray<OrchestrationV2Run>,
57): OrchestrationV2Run | null {
58 let latest: OrchestrationV2Run | null = null;
59 for (const run of runs) {
60 if (run.status === "queued") continue;
61 if (run.status === "cancelled" && run.startedAt === null) continue;
62 if (latest === null || runRanAfter(run, latest)) latest = run;
63 }
64 return latest;
65}
66
67/**
68 * Work started now would run in the project checkout, not the worktree the thread asked for. A
69 * failed or interrupted setup leaves the request standing until a worktree exists or later work
70 * has run past it.
71 */
72export function orchestrationV2WorktreePending(
73 worktreePath: string | null,
74 runs: ReadonlyArray<OrchestrationV2Run>,
75): boolean {
76 const latest = latestExecutedRun(runs);
77 return (
78 worktreePath === null &&
79 (latest?.status === "preparing" ||
80 latest?.status === "failed" ||
81 latest?.status === "interrupted") &&
82 latest.workspacePreparation?.type === "worktree"
83 );
84}
85
86/** Held queues can execute out of ordinal order; unfinished runs count as latest. */
87export function runRanAfter(
88 run: Pick<OrchestrationV2Run, "ordinal" | "completedAt">,
89 other: Pick<OrchestrationV2Run, "ordinal" | "completedAt">,
90): boolean {
91 const end = (candidate: typeof run) =>
92 !candidate.completedAt
93 ? Number.POSITIVE_INFINITY
94 : DateTime.toEpochMillis(candidate.completedAt);
95 return end(run) === end(other) ? run.ordinal > other.ordinal : end(run) > end(other);
96}
97
98export function usageLimitBlockedRun(
99 runs: ReadonlyArray<OrchestrationV2Run>,
100 turnItems: ReadonlyArray<FailureScanTurnItem>,
101 sessionError: string | null,
102): OrchestrationV2Run | null {
103 const executed = latestExecutedRun(runs);
104 if (executed?.status !== "failed") return null;
105 const summary = threadErrorSummary(latestRootProviderFailure(executed, turnItems), sessionError);
106 return summary.lastErrorClass === "usage_limit" ? executed : null;
107}
108
109/** Retains a limited run's outcome when higher-ordinal runs never started. */
110export function usageLimitRunPresentedAsLatest(
111 runs: ReadonlyArray<OrchestrationV2Run>,
112 turnItems: ReadonlyArray<FailureScanTurnItem>,
113 sessionError: string | null,
114): OrchestrationV2Run | null {
115 const blocked = usageLimitBlockedRun(runs, turnItems, sessionError);
116 return blocked !== null && runs.some((run) => run.ordinal > blocked.ordinal) ? blocked : null;
117}
118
119export function latestUnheldRun(
120 runs: ReadonlyArray<OrchestrationV2Run>,
121): OrchestrationV2Run | null {
122 let latest: OrchestrationV2Run | null = null;
123 for (const run of runs) {
124 if (run.status === "queued" && run.queueHeld === true) continue;
125 if (outlivedBeforeStart(run, runs)) continue;
126 if (latest === null || run.ordinal > latest.ordinal) latest = run;
127 }
128 return latest;
129}
130
131// Steering a queued message cancels its run before it starts and hands the message to the
132// earlier run, which outlives the cancellation; that earlier run then presents as latest.
133function outlivedBeforeStart(
134 run: OrchestrationV2Run,
135 runs: ReadonlyArray<OrchestrationV2Run>,
136): boolean {
137 if (run.status !== "cancelled" || run.startedAt !== null || !run.completedAt) return false;
138 const cancelledAt = DateTime.toEpochMillis(run.completedAt);
139 return runs.some(
140 (earlier) =>
141 earlier.ordinal < run.ordinal &&
142 earlier.status !== "queued" &&
143 (!earlier.completedAt || DateTime.toEpochMillis(earlier.completedAt) >= cancelledAt),
144 );
145}