1import type {
2 NodeId,
3 OrchestrationV2PendingBackgroundTask,
4 OrchestrationV2ProviderTurn,
5 OrchestrationV2RestartCancelledBackgroundWork,
6 OrchestrationV2Run,
7 OrchestrationV2RunAttempt,
8 OrchestrationV2TurnItem,
9} from "@t3tools/contracts";
10import { latestRootProviderFailure, runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
11
12type Work = OrchestrationV2RestartCancelledBackgroundWork;
13type Attempt = Pick<OrchestrationV2RunAttempt, "id" | "runId">;
14
15const MAX_LABEL_LENGTH = 160;
16
17function compactLabel(value: unknown, maxLength = MAX_LABEL_LENGTH): string | undefined {
18 if (typeof value !== "string") return undefined;
19 const text = value.replaceAll(/\s+/g, " ").trim();
20 if (text.length === 0) return undefined;
21 if (text.length <= maxLength) return text;
22 // Half an emoji serializes to a lone surrogate escape, which provider JSON parsers reject.
23 const last = text.charCodeAt(maxLength - 2);
24 const end = last >= 0xd800 && last <= 0xdbff ? maxLength - 2 : maxLength - 1;
25 return `${text.slice(0, end)}…`;
26}
27
28/** A bounded label that keeps its id suffix whole whenever it fits. */
29function labelWithId(name: unknown, fallback: string, suffix: string): string {
30 const budget = MAX_LABEL_LENGTH - suffix.length;
31 if (budget < 2) return compactLabel(`${fallback}${suffix}`) ?? fallback;
32 return `${compactLabel(name, budget) ?? compactLabel(fallback, budget)}${suffix}`;
33}
34
35/** resumeId overrides the item id for native resume, such as Codex's child provider-thread id. */
36export function cancelledTurnItemWork(
37 item: OrchestrationV2TurnItem,
38 resumeId?: string,
39): Work | undefined {
40 switch (item.type) {
41 case "subagent": {
42 const nativeId =
43 item.origin === "provider_native" ? (resumeId ?? item.nativeItemRef?.nativeId) : undefined;
44 return {
45 kind: "subagent",
46 label: labelWithId(
47 compactLabel(item.title) === undefined ? item.prompt : item.title,
48 "subagent",
49 nativeId === undefined ? "" : ` (id ${nativeId})`,
50 ),
51 id: item.id,
52 };
53 }
54 case "command_execution":
55 return {
56 kind: item.monitor === true ? "monitor" : "shell",
57 label: compactLabel(item.input) ?? compactLabel(item.title) ?? "background command",
58 id: item.id,
59 ...(item.completionWake === "always" ? { awaited: true as const } : {}),
60 };
61 case "dynamic_tool": {
62 const monitor =
63 item.input !== null &&
64 typeof item.input === "object" &&
65 Reflect.get(item.input, "persistent") === true;
66 return {
67 kind: monitor ? "monitor" : "task",
68 label: compactLabel(item.title) ?? compactLabel(item.toolName) ?? "background tool",
69 id: item.id,
70 };
71 }
72 default:
73 return undefined;
74 }
75}
76
77/** Returns the awaiting provider thread; native subagents can run on children while their launching turn awaits them. */
78export function awaitingProviderThreadId(
79 item: OrchestrationV2TurnItem,
80 runs: ReadonlyArray<Pick<OrchestrationV2Run, "id" | "providerThreadId">>,
81): OrchestrationV2Run["providerThreadId"] | undefined {
82 const runProviderThreadId = runs.find((run) => run.id === item.runId)?.providerThreadId;
83 return item.type === "subagent" && item.origin === "provider_native"
84 ? (runProviderThreadId ?? item.providerThreadId)
85 : (item.providerThreadId ?? runProviderThreadId);
86}
87
88function restartWorkKind(task: OrchestrationV2PendingBackgroundTask): Work["kind"] {
89 switch (task.kind) {
90 case "command":
91 return "shell";
92 case "subagent":
93 return "subagent";
94 case "monitor":
95 return "monitor";
96 case "background_task":
97 return "task";
98 default:
99 task satisfies never;
100 return "task";
101 }
102}
103
104export function cancelledRosterTaskWork(task: OrchestrationV2PendingBackgroundTask): Work {
105 if (task.parentTaskId !== undefined) {
106 return {
107 kind: "subagent",
108 label: labelWithId(task.description, "background task", ` (subagent ${task.parentTaskId})`),
109 id: task.parentTaskId,
110 };
111 }
112 const kind = restartWorkKind(task);
113 const description = compactLabel(task.description);
114 return {
115 kind,
116 label:
117 compactLabel(
118 description === undefined ? task.taskId : `${description} (id ${task.taskId})`,
119 ) ?? "background task",
120 id: task.taskId,
121 ...(task.awaited === true ? { awaited: true as const } : {}),
122 };
123}
124
125const MAX_NOTE_ENTRIES = 10;
126
127/** Bounds entries and labels to protect the context budget. */
128export function restartCancelledBackgroundWorkNote(work: ReadonlyArray<Work>): string {
129 const omitted = work.length - MAX_NOTE_ENTRIES;
130 return [
131 "Note: the T3 server restarted, and this background work was cancelled before it finished. It will not report back on its own, so resume or restart what you still need and wait for its result:",
132 ...work.slice(0, MAX_NOTE_ENTRIES).map((entry) => `- ${entry.kind}: ${entry.label}`),
133 ...(omitted > 0 ? [`- and ${omitted} more`] : []),
134 ].join("\n");
135}
136
137type ProviderTurnState = Pick<
138 OrchestrationV2ProviderTurn,
139 "runAttemptId" | "providerThreadId" | "status"
140>;
141
142export function isRestartNoteSource(
143 source: Pick<OrchestrationV2Run, "status" | "activeAttemptId" | "restartCancelledBackgroundWork">,
144 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
145): boolean {
146 return (
147 // Failed and interrupted turns never continue; recovery-cancelled waiting runs still qualify.
148 (source.status === "completed" ||
149 source.status === "waiting" ||
150 source.status === "cancelled") &&
151 (source.restartCancelledBackgroundWork?.length ?? 0) > 0 &&
152 !providerTurns.some(
153 (turn) => turn.runAttemptId === source.activeAttemptId && turn.status === "cancelled",
154 )
155 );
156}
157
158/**
159 * Matches the Working-section hold: a shell or monitor wakes it only when the agent said it
160 * waits on it. Most are dev servers and their watchdogs, so their note waits for the next
161 * user message.
162 */
163export function isAwaitedWork(work: Pick<Work, "kind" | "awaited">): boolean {
164 return work.awaited === true || (work.kind !== "shell" && work.kind !== "monitor");
165}
166
167export function runOutcome(
168 run: Pick<OrchestrationV2Run, "status" | "restartCancelledBackgroundWork">,
169): OrchestrationV2Run["status"] {
170 return (run.status === "completed" || run.status === "cancelled") &&
171 (run.restartCancelledBackgroundWork ?? []).some(isAwaitedWork)
172 ? "interrupted"
173 : run.status;
174}
175
176/** Unstarted runs have no lost work to continue, except prepared continuations carrying a source run. */
177export function neverStarted(
178 run: Pick<OrchestrationV2Run, "startedAt" | "restartContinuationOfRunId">,
179): boolean {
180 return run.startedAt === null && run.restartContinuationOfRunId === undefined;
181}
182
183/** Restart-cancelled unstarted runs supersede nothing; user-stopped unstarted runs do. */
184export function cancelledBeforeStart(
185 run: Pick<OrchestrationV2Run, "status" | "startedAt" | "restartContinuationOfRunId">,
186): boolean {
187 return run.status === "cancelled" && neverStarted(run);
188}
189
190/** A follow-up task's latest run, passing over queued work it no longer waits on, as the original task does. */
191export function delegatedTaskResultRun(
192 runs: ReadonlyArray<OrchestrationV2Run>,
193 taskId: NodeId,
194): OrchestrationV2Run | undefined {
195 const taskRuns = runs
196 .filter((run) => run.delegatedTaskId === taskId)
197 .toSorted((a, b) => b.ordinal - a.ordinal);
198 const awaited = taskRuns.filter(
199 (run) => !cancelledBeforeStart(run) && !(run.status === "queued" && run.queueHeld === true),
200 );
201 // Executing work reports the task's state; a reordered queue can finish a newer run before an older one.
202 const executing = awaited.find((run) =>
203 ["preparing", "starting", "running", "waiting"].includes(run.status),
204 );
205 return executing ?? awaited.find((run) => run.status === "queued") ?? awaited[0] ?? taskRuns[0];
206}
207
208/** The provider never accepted this continuation's prompt, so starting it again repeats nothing. */
209export function continuationFailedBeforePrompt(
210 run: Pick<OrchestrationV2Run, "status" | "activeAttemptId" | "restartContinuationOfRunId">,
211 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId">>,
212): boolean {
213 return (
214 run.status === "failed" &&
215 run.restartContinuationOfRunId !== undefined &&
216 !providerTurns.some((turn) => turn.runAttemptId === run.activeAttemptId)
217 );
218}
219
220const CONTINUATION_START_RETRY_DELAYS_MS = [60_000, 180_000];
221
222/** Backoff before restarting a continuation that failed to start, or undefined to let it fail. */
223export function continuationStartRetryDelayMs(
224 run: OrchestrationV2Run,
225 projection: {
226 readonly runs: ReadonlyArray<OrchestrationV2Run>;
227 readonly providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId">>;
228 readonly turnItems: Parameters<typeof latestRootProviderFailure>[1];
229 },
230): number | undefined {
231 if (!continuationFailedBeforePrompt(run, projection.providerTurns)) return;
232 // These fail the same way however often they start.
233 const failureClass = latestRootProviderFailure(run, projection.turnItems)?.class;
234 if (
235 failureClass === "usage_limit" ||
236 failureClass === "permission_error" ||
237 failureClass === "validation_error"
238 )
239 return;
240 let failedStarts = 0;
241 for (
242 let current: OrchestrationV2Run | undefined = run;
243 current !== undefined && continuationFailedBeforePrompt(current, projection.providerTurns);
244 current = projection.runs.find(
245 (candidate) => candidate.id === current?.restartContinuationOfRunId,
246 )
247 )
248 failedStarts += 1;
249 return CONTINUATION_START_RETRY_DELAYS_MS[failedStarts - 1];
250}
251
252export function isRestartContinuationSource(
253 source: Pick<OrchestrationV2Run, "status" | "activeAttemptId" | "restartCancelledBackgroundWork">,
254 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
255): boolean {
256 return isRestartNoteSource(source, providerTurns)
257 ? (source.restartCancelledBackgroundWork ?? []).some(isAwaitedWork)
258 : source.status === "cancelled";
259}
260
261/** Running does not prove prompt acceptance; carry continuation notes until a completed turn proves delivery. */
262export function restartContinuationNote(
263 source: OrchestrationV2Run,
264 runs: ReadonlyArray<OrchestrationV2Run>,
265 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
266 attempts: ReadonlyArray<Attempt>,
267): { readonly work: ReadonlyArray<Work>; readonly settled: boolean } {
268 const completedAttempts = new Set(
269 providerTurns.filter((turn) => turn.status === "completed").map((turn) => turn.runAttemptId),
270 );
271 const completedRuns = new Set(
272 attempts.filter((attempt) => completedAttempts.has(attempt.id)).map((attempt) => attempt.runId),
273 );
274 let current = source;
275 let work = current.restartCancelledBackgroundWork ?? [];
276 const visited = new Set([current.id]);
277 while (
278 current.restartContinuationOfRunId !== undefined &&
279 !completedRuns.has(current.id) &&
280 !(current.activeAttemptId !== null && completedAttempts.has(current.activeAttemptId))
281 ) {
282 const previous = runs.find((candidate) => candidate.id === current.restartContinuationOfRunId);
283 if (previous === undefined || visited.has(previous.id)) break;
284 visited.add(previous.id);
285 current = previous;
286 work = mergeRestartCancelledBackgroundWork(current.restartCancelledBackgroundWork ?? [], work);
287 }
288 return { work, settled: isRestartNoteSource(current, providerTurns) };
289}
290
291export function isRestartNoteContinuation(
292 run: Pick<OrchestrationV2Run, "restartContinuationOfRunId">,
293 runs: ReadonlyArray<OrchestrationV2Run>,
294 providerTurns: ReadonlyArray<Pick<ProviderTurnState, "runAttemptId" | "status">>,
295 attempts: ReadonlyArray<Attempt>,
296): boolean {
297 const source =
298 run.restartContinuationOfRunId === undefined
299 ? undefined
300 : runs.find((candidate) => candidate.id === run.restartContinuationOfRunId);
301 return (
302 source !== undefined &&
303 restartContinuationNote(source, runs, providerTurns, attempts).work.length > 0
304 );
305}
306
307/** Derive delivery from later completed turns on the same provider thread, excluding compactions, resumes, and rolled-back runs. */
308export function pendingRestartCancelledBackgroundWork(input: {
309 readonly runs: ReadonlyArray<OrchestrationV2Run>;
310 readonly providerTurns: ReadonlyArray<ProviderTurnState>;
311 readonly compactionMessageIds: ReadonlySet<string>;
312 readonly run: Pick<
313 OrchestrationV2Run,
314 | "id"
315 | "ordinal"
316 | "completedAt"
317 | "userMessageId"
318 | "providerThreadId"
319 | "restartContinuationOfRunId"
320 | "activeAttemptId"
321 >;
322
323 readonly attempts: ReadonlyArray<Attempt>;
324}): ReadonlyArray<Work> {
325 const isCompaction = (run: typeof input.run) => input.compactionMessageIds.has(run.userMessageId);
326 if (
327 input.run.providerThreadId === null ||
328 input.run.restartContinuationOfRunId !== undefined ||
329 isCompaction(input.run)
330 )
331 return [];
332 const providerThreadId = input.run.providerThreadId;
333 const deliveredAttemptIds = new Set(
334 input.providerTurns
335 .filter((turn) => turn.providerThreadId === providerThreadId && turn.status === "completed")
336 .map((turn) => turn.runAttemptId),
337 );
338 const sameThread = input.runs.filter((run) => run.providerThreadId === providerThreadId);
339 const prompted = sameThread.filter(
340 (candidate) =>
341 candidate.id !== input.run.id &&
342 candidate.activeAttemptId !== null &&
343 candidate.status !== "rolled_back" &&
344 (deliveredAttemptIds.has(candidate.activeAttemptId) ||
345 input.attempts.some(
346 (attempt) => attempt.runId === candidate.id && deliveredAttemptIds.has(attempt.id),
347 )) &&
348 !isCompaction(candidate) &&
349 (candidate.restartContinuationOfRunId === undefined ||
350 isRestartNoteContinuation(candidate, input.runs, input.providerTurns, input.attempts)),
351 );
352 // Earlier attempts may already have delivered the note; a steer's replacement attempt must not repeat it.
353 const alreadyDelivered = input.attempts.some(
354 (attempt) =>
355 attempt.runId === input.run.id &&
356 attempt.id !== input.run.activeAttemptId &&
357 deliveredAttemptIds.has(attempt.id),
358 );
359 if (alreadyDelivered) return [];
360 return sameThread
361 .filter(
362 (source) =>
363 runRanAfter(input.run, source) &&
364 (source.restartCancelledBackgroundWork?.length ?? 0) > 0 &&
365 !prompted.some((later) => runRanAfter(later, source)),
366 )
367 .reduce<ReadonlyArray<Work>>(
368 (work, source) =>
369 mergeRestartCancelledBackgroundWork(work, source.restartCancelledBackgroundWork ?? []),
370 [],
371 );
372}
373
374export function mergeRestartCancelledBackgroundWork(
375 current: ReadonlyArray<Work>,
376 added: ReadonlyArray<Work>,
377): ReadonlyArray<Work> {
378 const identity = (entry: Work) => entry.id ?? `${entry.kind}\u0000${entry.label}`;
379 const seen = new Set(current.map(identity));
380 const merged = [...current];
381 for (const entry of added) {
382 const key = identity(entry);
383 if (seen.has(key)) continue;
384 seen.add(key);
385 merged.push(entry);
386 }
387 return merged;
388}