1import type {
2 OrchestrationV2PendingBackgroundTask,
3 OrchestrationV2ProviderTurn,
4 OrchestrationV2RestartCancelledBackgroundWork,
5 OrchestrationV2Run,
6 OrchestrationV2RunAttempt,
7 OrchestrationV2TurnItem,
8} from "@t3tools/contracts";
9import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
10
11type Work = OrchestrationV2RestartCancelledBackgroundWork;
12type Attempt = Pick<OrchestrationV2RunAttempt, "id" | "runId">;
13
14const MAX_LABEL_LENGTH = 160;
15
16function compactLabel(value: unknown): string | undefined {
17 if (typeof value !== "string") return undefined;
18 const text = value.replaceAll(/\s+/g, " ").trim();
19 if (text.length === 0) return undefined;
20 return text.length > MAX_LABEL_LENGTH ? `${text.slice(0, MAX_LABEL_LENGTH - 1)}…` : text;
21}
22
23/** Describes a background-capable turn item that restart recovery cancels. */
24export function cancelledTurnItemWork(item: OrchestrationV2TurnItem): Work | undefined {
25 switch (item.type) {
26 case "subagent":
27 return {
28 kind: "subagent",
29 label: compactLabel(item.title) ?? compactLabel(item.prompt) ?? "subagent",
30 id: item.id,
31 };
32 case "command_execution":
33 return {
34 kind: "shell",
35 label: compactLabel(item.input) ?? compactLabel(item.title) ?? "background command",
36 id: item.id,
37 };
38 case "dynamic_tool": {
39 const monitor =
40 item.input !== null &&
41 typeof item.input === "object" &&
42 Reflect.get(item.input, "persistent") === true;
43 return {
44 kind: monitor ? "monitor" : "task",
45 label: compactLabel(item.title) ?? compactLabel(item.toolName) ?? "background tool",
46 id: item.id,
47 };
48 }
49 default:
50 return undefined;
51 }
52}
53
54function restartWorkKind(task: OrchestrationV2PendingBackgroundTask): Work["kind"] {
55 switch (task.kind) {
56 case "command":
57 return "shell";
58 case "subagent":
59 return "subagent";
60 case "monitor":
61 return "monitor";
62 case "background_task":
63 return "task";
64 default:
65 task satisfies never;
66 return "task";
67 }
68}
69
70/** Describes a provider-reported background task (the provider-thread roster). */
71export function cancelledRosterTaskWork(task: OrchestrationV2PendingBackgroundTask): Work {
72 const kind = restartWorkKind(task);
73 const description = compactLabel(task.description);
74 return {
75 kind,
76 label:
77 compactLabel(
78 description === undefined ? task.taskId : `${description} (id ${task.taskId})`,
79 ) ?? "background task",
80 id: task.taskId,
81 };
82}
83
84const MAX_NOTE_ENTRIES = 10;
85
86/**
87 * Provider-facing text for work the model still expects to hear back from.
88 * Bounded (entries and label length) so it cannot crowd out the turn's context.
89 */
90export function restartCancelledBackgroundWorkNote(work: ReadonlyArray<Work>): string {
91 const omitted = work.length - MAX_NOTE_ENTRIES;
92 return [
93 "Note: the T3 server restarted, and this background work was cancelled before it finished. It will not report back:",
94 ...work.slice(0, MAX_NOTE_ENTRIES).map((entry) => `- ${entry.kind}: ${entry.label}`),
95 ...(omitted > 0 ? [`- and ${omitted} more`] : []),
96 ].join("\n");
97}
98
99type ProviderTurnState = Pick<
100 OrchestrationV2ProviderTurn,
101 "runAttemptId" | "providerThreadId" | "status"
102>;
103
104/**
105 * A run whose turn had settled before the restart cancelled its background
106 * work. Continuing it prompts the provider with the note. A run cut mid-turn
107 * instead resumes that turn, which Codex does natively without a prompt.
108 */
109export function isRestartNoteSource(
110 source: Pick<OrchestrationV2Run, "status" | "activeAttemptId" | "restartCancelledBackgroundWork">,
111 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
112): boolean {
113 return (
114 // A failed or interrupted turn is never continued. Recovery cancels a
115 // waiting run it cannot checkpoint, so a cancelled one still qualifies.
116 (source.status === "completed" ||
117 source.status === "waiting" ||
118 source.status === "cancelled") &&
119 (source.restartCancelledBackgroundWork?.length ?? 0) > 0 &&
120 !providerTurns.some(
121 (turn) => turn.runAttemptId === source.activeAttemptId && turn.status === "cancelled",
122 )
123 );
124}
125
126/**
127 * Work an agent ends its turn to wait for: a monitor, a subagent, or a
128 * background task. A restart that cancels it wakes the settled turn, because
129 * nothing else will. A shell does not wake it: most are dev servers that
130 * nobody waits on, so their note waits for the next user message.
131 */
132export function isAwaitedWork(work: Pick<Work, "kind">): boolean {
133 return work.kind !== "shell";
134}
135
136/**
137 * A run's status as its waiter should read it: a delegating parent,
138 * `task_status`, or `t3_thread_wait`. A turn that ended to wait on work the
139 * restart then cancelled did not finish, so it is "interrupted", not
140 * "completed", and its last message is not a final result.
141 */
142export function runOutcome(
143 run: Pick<OrchestrationV2Run, "status" | "restartCancelledBackgroundWork">,
144): OrchestrationV2Run["status"] {
145 return (run.status === "completed" || run.status === "cancelled") &&
146 (run.restartCancelledBackgroundWork ?? []).some(isAwaitedWork)
147 ? "interrupted"
148 : run.status;
149}
150
151/**
152 * Whether a restart continuation may prompt from `source`: a turn the restart
153 * cut mid-way, or a settled turn that lost awaited work in the restart.
154 */
155export function isRestartContinuationSource(
156 source: Pick<OrchestrationV2Run, "status" | "activeAttemptId" | "restartCancelledBackgroundWork">,
157 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
158): boolean {
159 return isRestartNoteSource(source, providerTurns)
160 ? (source.restartCancelledBackgroundWork ?? []).some(isAwaitedWork)
161 : source.status === "cancelled";
162}
163
164/**
165 * The note a restart continuation of `source` carries, and whether the turn it
166 * continues had settled (the note is then the whole prompt). A continuation cut
167 * before its provider turn completed may not have delivered its own note:
168 * adapters can announce a running turn before accepting the prompt. Carry the
169 * note forward until a completed turn proves delivery.
170 */
171export function restartContinuationNote(
172 source: OrchestrationV2Run,
173 runs: ReadonlyArray<OrchestrationV2Run>,
174 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
175 attempts: ReadonlyArray<Attempt>,
176): { readonly work: ReadonlyArray<Work>; readonly settled: boolean } {
177 const completedAttempts = new Set(
178 providerTurns.filter((turn) => turn.status === "completed").map((turn) => turn.runAttemptId),
179 );
180 const completedRuns = new Set(
181 attempts.filter((attempt) => completedAttempts.has(attempt.id)).map((attempt) => attempt.runId),
182 );
183 let current = source;
184 let work = current.restartCancelledBackgroundWork ?? [];
185 const visited = new Set([current.id]);
186 while (
187 current.restartContinuationOfRunId !== undefined &&
188 !completedRuns.has(current.id) &&
189 !(current.activeAttemptId !== null && completedAttempts.has(current.activeAttemptId))
190 ) {
191 const previous = runs.find((candidate) => candidate.id === current.restartContinuationOfRunId);
192 if (previous === undefined || visited.has(previous.id)) break;
193 visited.add(previous.id);
194 current = previous;
195 work = mergeRestartCancelledBackgroundWork(current.restartCancelledBackgroundWork ?? [], work);
196 }
197 return { work, settled: isRestartNoteSource(current, providerTurns) };
198}
199
200/**
201 * A restart continuation prompted with the note rather than a native resume.
202 * A turn cut mid-way that lost background work is prompted too, so the note
203 * is delivered with it; only a continuation without a note resumes natively.
204 */
205export function isRestartNoteContinuation(
206 run: Pick<OrchestrationV2Run, "restartContinuationOfRunId">,
207 runs: ReadonlyArray<OrchestrationV2Run>,
208 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
209 attempts: ReadonlyArray<Attempt>,
210): boolean {
211 const source =
212 run.restartContinuationOfRunId === undefined
213 ? undefined
214 : runs.find((candidate) => candidate.id === run.restartContinuationOfRunId);
215 return (
216 source !== undefined &&
217 restartContinuationNote(source, runs, providerTurns, attempts).work.length > 0
218 );
219}
220
221/**
222 * Work cancelled by a restart that the run's provider thread has not been told
223 * about yet. The note belongs to the provider thread that lost the work: turns
224 * on another provider (after a switch) neither owe it nor deliver it. A later
225 * completed turn on the same provider thread proves delivery, so the pending
226 * set is derived rather than cleared. Compactions and
227 * resumed turns carry no note, and a rolled-back run left native history, so
228 * none of them counts as delivery. A note continuation's own prompt is the note.
229 */
230export function pendingRestartCancelledBackgroundWork(input: {
231 readonly runs: ReadonlyArray<OrchestrationV2Run>;
232 readonly providerTurns: ReadonlyArray<ProviderTurnState>;
233 readonly compactionMessageIds: ReadonlySet<string>;
234 readonly run: Pick<
235 OrchestrationV2Run,
236 | "id"
237 | "ordinal"
238 | "completedAt"
239 | "userMessageId"
240 | "providerThreadId"
241 | "restartContinuationOfRunId"
242 | "activeAttemptId"
243 >;
244 /** Include earlier attempts: a steer replaces the attempt but not the run. */
245 readonly attempts: ReadonlyArray<Attempt>;
246}): ReadonlyArray<Work> {
247 const isCompaction = (run: typeof input.run) => input.compactionMessageIds.has(run.userMessageId);
248 // The current run prepends the note unless it is a compaction or a
249 // continuation (whose own prompt is the note, or which resumes natively).
250 if (
251 input.run.providerThreadId === null ||
252 input.run.restartContinuationOfRunId !== undefined ||
253 isCompaction(input.run)
254 )
255 return [];
256 const providerThreadId = input.run.providerThreadId;
257 const deliveredAttemptIds = new Set(
258 input.providerTurns
259 .filter((turn) => turn.providerThreadId === providerThreadId && turn.status === "completed")
260 .map((turn) => turn.runAttemptId),
261 );
262 const sameThread = input.runs.filter((run) => run.providerThreadId === providerThreadId);
263 const prompted = sameThread.filter(
264 (candidate) =>
265 candidate.id !== input.run.id &&
266 candidate.activeAttemptId !== null &&
267 candidate.status !== "rolled_back" &&
268 (deliveredAttemptIds.has(candidate.activeAttemptId) ||
269 input.attempts.some(
270 (attempt) => attempt.runId === candidate.id && deliveredAttemptIds.has(attempt.id),
271 )) &&
272 !isCompaction(candidate) &&
273 (candidate.restartContinuationOfRunId === undefined ||
274 isRestartNoteContinuation(candidate, input.runs, input.providerTurns, input.attempts)),
275 );
276 // A steer restarts this run on a new attempt: an earlier attempt that already
277 // reached the provider delivered the note, so the replacement must not repeat it.
278 const alreadyDelivered = input.attempts.some(
279 (attempt) =>
280 attempt.runId === input.run.id &&
281 attempt.id !== input.run.activeAttemptId &&
282 deliveredAttemptIds.has(attempt.id),
283 );
284 if (alreadyDelivered) return [];
285 return sameThread
286 .filter(
287 (source) =>
288 runRanAfter(input.run, source) &&
289 (source.restartCancelledBackgroundWork?.length ?? 0) > 0 &&
290 !prompted.some((later) => runRanAfter(later, source)),
291 )
292 .reduce<ReadonlyArray<Work>>(
293 (work, source) =>
294 mergeRestartCancelledBackgroundWork(work, source.restartCancelledBackgroundWork ?? []),
295 [],
296 );
297}
298
299export function mergeRestartCancelledBackgroundWork(
300 current: ReadonlyArray<Work>,
301 added: ReadonlyArray<Work>,
302): ReadonlyArray<Work> {
303 // Rows recorded before ids existed fall back to kind + label.
304 const identity = (entry: Work) => entry.id ?? `${entry.kind}\u0000${entry.label}`;
305 const seen = new Set(current.map(identity));
306 const merged = [...current];
307 for (const entry of added) {
308 const key = identity(entry);
309 if (seen.has(key)) continue;
310 seen.add(key);
311 merged.push(entry);
312 }
313 return merged;
314}