1import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
2import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
3import {
4 CommandId,
5 latestProviderTurnForAttempt,
6 MessageId,
7 type OrchestrationV2Run,
8 type RunId,
9 type ThreadId,
10} from "@t3tools/contracts";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import type { ProjectionRuntimeRecoveryState } from "./ProjectionStore.ts";
14
15import * as ServerSettings from "../serverSettings.ts";
16import { isNativeMaintenanceCommand } from "./Orchestrator.ts";
17import * as ThreadManagementService from "./ThreadManagementService.ts";
18import {
19 cancelledRosterTaskWork,
20 cancelledTurnItemWork,
21 isAwaitedWork,
22 isRestartContinuationSource,
23 isRestartNoteSource,
24 restartCancelledBackgroundWorkNote,
25 restartContinuationNote,
26} from "./RestartBackgroundNote.ts";
27
28const CONTINUE_PROMPT = "Continue where you left off.";
29
30/** Consecutive automatic continuations allowed before a person has to step in. */
31const MAX_CONTINUATION_CHAIN = 3;
32
33/**
34 * Whether a restart must leave this run for the user: the thread is put away,
35 * the provider turn already ended, or the user owes an answer. Checked when
36 * intent is recorded and again when the continuation is delivered. A
37 * `settled` run is woken for the awaited work it lost, so its ended provider
38 * turn does not block it.
39 */
40function restartContinuationBlocked(
41 projection: Pick<
42 ProjectionRuntimeRecoveryState,
43 "thread" | "runtimeRequests" | "providerTurns" | "turnItems"
44 >,
45 source: OrchestrationV2Run,
46 now: DateTime.Utc,
47 settled: boolean,
48): boolean {
49 const { thread } = projection;
50 if (
51 thread.archivedAt !== null ||
52 thread.deletedAt !== null ||
53 thread.settledOverride === "settled" ||
54 (thread.snoozedUntil != null && DateTime.isGreaterThan(thread.snoozedUntil, now))
55 )
56 return true;
57 const turn = latestProviderTurnForAttempt(projection.providerTurns, source.activeAttemptId);
58 if (
59 !settled &&
60 turn !== undefined &&
61 (turn.status === "completed" || turn.status === "failed" || turn.status === "interrupted")
62 )
63 return true;
64 const waitingRequest = projection.runtimeRequests.some(
65 (request) =>
66 // Tool calls and auth refreshes do not wait on a person, and a
67 // message-capable question stays answerable after the restart.
68 request.kind !== "dynamic_tool_call" &&
69 request.kind !== "auth_refresh" &&
70 request.responseCapability.type !== "message" &&
71 (request.status === "pending" ||
72 // Recovery closes live callbacks on the cut turn; it does not answer them.
73 ((request.status === "expired" || request.status === "cancelled") &&
74 request.responseCapability.type === "not_resumable" &&
75 turn !== undefined &&
76 turn.id === request.providerTurnId)),
77 );
78 if (waitingRequest) return true;
79 return projection.turnItems.some(
80 (item) =>
81 item.type === "secret_request" &&
82 item.runId === source.id &&
83 (item.secretStatus === "pending" ||
84 // Recovery cancels an open secret card and its run at the same instant.
85 (item.secretStatus === "cancelled" &&
86 item.completedAt !== null &&
87 source.completedAt !== null &&
88 DateTime.Equivalence(item.completedAt, source.completedAt))),
89 );
90}
91
92/**
93 * Open awaited work (see `isAwaitedWork`) on a provider thread. App-owned
94 * delegations are not counted: their own threads continue and report back.
95 */
96function hasOpenAwaitedWork(
97 projection: Pick<ProjectionRuntimeRecoveryState, "runs" | "turnItems">,
98 providerThread: ProjectionRuntimeRecoveryState["providerThreads"][number],
99): boolean {
100 const openItem = projection.turnItems.some((item) => {
101 if (item.status !== "pending" && item.status !== "running" && item.status !== "waiting")
102 return false;
103 if (item.type === "subagent" && item.origin === "app_owned" && item.childThreadId !== null)
104 return false;
105 const work = cancelledTurnItemWork(item);
106 const providerThreadId =
107 item.providerThreadId ??
108 projection.runs.find((run) => run.id === item.runId)?.providerThreadId;
109 return work !== undefined && isAwaitedWork(work) && providerThreadId === providerThread.id;
110 });
111 return (
112 openItem ||
113 (providerThread.ownerNodeId === null &&
114 (providerThread.pendingBackgroundTasks ?? []).some((task) =>
115 isAwaitedWork(cancelledRosterTaskWork(task)),
116 ))
117 );
118}
119
120/**
121 * The run a restart continues: an unfinished root turn, or a settled one
122 * waiting on awaited work that the restart cancels. A settled run needs a
123 * live session, so stale records from an older crash never wake it. Other
124 * leftover work is cleaned up and reported on the next user turn.
125 */
126export function restartContinuationRun(
127 projection: Pick<
128 ProjectionRuntimeRecoveryState,
129 | "thread"
130 | "runs"
131 | "providerThreads"
132 | "providerSessions"
133 | "providerTurns"
134 | "runtimeRequests"
135 | "turnItems"
136 >,
137 now: DateTime.Utc,
138): OrchestrationV2Run | undefined {
139 // Queued runs never started; recovery holds them behind the cut run.
140 const run = projection.runs.reduce<OrchestrationV2Run | undefined>(
141 (latest, candidate) =>
142 candidate.status !== "queued" && (!latest || runRanAfter(candidate, latest))
143 ? candidate
144 : latest,
145 undefined,
146 );
147 if (!run) return;
148 const preparedContinuation =
149 run.status === "starting" && run.restartContinuationOfRunId !== undefined;
150 const settled = run.status === "completed" || run.status === "waiting";
151 if (run.status !== "running" && !preparedContinuation && !settled) return;
152 if (restartContinuationBlocked(projection, run, now, settled)) return;
153 const liveTurnRequired = !preparedContinuation && !settled;
154 if (projection.thread.providerInstanceId !== run.providerInstanceId) return;
155 const providerThread = projection.providerThreads.find(
156 (thread) => thread.id === run.providerThreadId,
157 );
158 if (
159 !providerThread ||
160 providerThread.appThreadId !== projection.thread.id ||
161 providerThread.ownerNodeId !== null ||
162 providerThread.providerInstanceId !== run.providerInstanceId ||
163 providerThread.nativeThreadRef?.nativeId == null ||
164 providerThread.nativeThreadRef.strength !== "strong" ||
165 providerThread.nativeThreadRef.driver !== providerThread.driver ||
166 (liveTurnRequired && providerThread.status !== "active") ||
167 providerThread.status === "closed" ||
168 providerThread.status === "archived"
169 )
170 return;
171 const session = projection.providerSessions.find(
172 (candidate) => candidate.id === providerThread.providerSessionId,
173 );
174 // Most adapters keep a live session "ready" through its turns, so only a
175 // stopped or failed session rules out a live turn.
176 if (
177 session === undefined ||
178 session.providerInstanceId !== run.providerInstanceId ||
179 session.driver !== providerThread.driver ||
180 (!preparedContinuation && (session.status === "stopped" || session.status === "error"))
181 )
182 return;
183 if (settled) return hasOpenAwaitedWork(projection, providerThread) ? run : undefined;
184 if (
185 liveTurnRequired &&
186 !projection.providerTurns.some(
187 (turn) =>
188 turn.providerThreadId === providerThread.id &&
189 turn.runAttemptId === run.activeAttemptId &&
190 turn.status === "running",
191 )
192 )
193 return;
194 return run;
195}
196
197export const continueRestartedRun = Effect.fn("RestartContinuation.continueRestartedRun")(
198 function* (input: { readonly threadId: ThreadId; readonly sourceRunId: RunId }) {
199 const settings = yield* ServerSettings.ServerSettingsService;
200 const enabled = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null));
201 if (!enabled) return;
202 const threads = yield* ThreadManagementService.ThreadManagementService;
203 const messageId = MessageId.make(`message:restart-continuation:${input.sourceRunId}`);
204 const projection = yield* threads.getThreadRecords(
205 input.threadId,
206 ["messages", "runs", "providerTurns", "attempts", "runtimeRequests", "turnItems"],
207 { messageIds: [messageId], turnItemTypes: ["secret_request"] },
208 );
209 if (
210 !resolveProjectSettings(enabled, projection.thread.projectId).settings
211 .continueThreadsAfterServerUpdate
212 )
213 return;
214 if (projection.messages.some((message) => message.id === messageId)) return;
215 const source = projection.runs.find((run) => run.id === input.sourceRunId);
216 // Reconciliation cancels a waiting run it cannot checkpoint, so a settled
217 // source can be cancelled too. It continues only for awaited work.
218 if (
219 !source ||
220 !isRestartContinuationSource(source, projection.providerTurns) ||
221 restartContinuationBlocked(
222 projection,
223 source,
224 yield* DateTime.now,
225 isRestartNoteSource(source, projection.providerTurns),
226 )
227 )
228 return;
229 // A user submission after reconciliation takes precedence over an automatic
230 // prompt. Queued runs never started and stay held behind this one.
231 if (
232 projection.runs.some(
233 (run) => run.id !== source.id && run.status !== "queued" && runRanAfter(run, source),
234 )
235 )
236 return;
237 if (projection.thread.providerInstanceId !== source.providerInstanceId) return;
238 // A turn that keeps crashing the server would otherwise restart it forever.
239 let chainLength = 0;
240 for (
241 let run: OrchestrationV2Run | undefined = source;
242 run?.restartContinuationOfRunId !== undefined && chainLength < MAX_CONTINUATION_CHAIN;
243 run = projection.runs.find((candidate) => candidate.id === run?.restartContinuationOfRunId)
244 ) {
245 chainLength += 1;
246 }
247 if (chainLength >= MAX_CONTINUATION_CHAIN) {
248 yield* Effect.logWarning("Not continuing a run restarted too many times in a row", {
249 threadId: input.threadId,
250 sourceRunId: input.sourceRunId,
251 });
252 return;
253 }
254 const sourceRecords = yield* threads.getThreadRecords(
255 input.threadId,
256 ["messages", "turnItems"],
257 {
258 messageIds: [source.userMessageId],
259 turnItemRunIds: [source.id],
260 turnItemTypes: ["run_interrupt_request"],
261 },
262 );
263 // The user asked this run to stop before the restart cut it.
264 if (
265 sourceRecords.turnItems.some(
266 (item) => item.runId === source.id && item.type === "run_interrupt_request",
267 )
268 )
269 return;
270 const sourceMessage = sourceRecords.messages.find(
271 (message) => message.id === source.userMessageId,
272 );
273 if (sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) return;
274 const note = restartContinuationNote(
275 source,
276 projection.runs,
277 projection.providerTurns,
278 projection.attempts,
279 );
280 const noteText =
281 note.work.length === 0 ? undefined : restartCancelledBackgroundWorkNote(note.work);
282 const text =
283 noteText === undefined
284 ? CONTINUE_PROMPT
285 : note.settled
286 ? noteText
287 : `${noteText}\n\n${CONTINUE_PROMPT}`;
288 yield* threads
289 .dispatch({
290 type: "message.dispatch",
291 commandId: CommandId.make(`command:restart-continuation:${input.sourceRunId}`),
292 threadId: input.threadId,
293 messageId,
294 text,
295 // Shown as a work log row; the prompt the agent got is its detail.
296 notification: {
297 source: { kind: "system" },
298 outcome: "updated",
299 summary: note.settled
300 ? "T3 Code restarted and stopped background work"
301 : "T3 Code restarted and resumed this turn",
302 detail: text,
303 },
304 attachments: [],
305 modelSelection: source.modelSelection,
306 dispatchMode: { type: "start_immediately" },
307 createdBy: "agent",
308 creationSource: "server",
309 restartContinuationOfRunId: input.sourceRunId,
310 })
311 .pipe(
312 // The project turned this instance off since the restart. No turn
313 // resumes, and the success path below still runs so a delegated
314 // child waiting on this run is recovered immediately instead of
315 // after EffectWorker exhausts its retries on a refusal that will
316 // not change.
317 Effect.catchTags({
318 ThreadManagementProviderInstanceDisabledError: (error) =>
319 Effect.logInfo("orchestration-v2.restart-continuation.instance-disabled", {
320 threadId: input.threadId,
321 sourceRunId: input.sourceRunId,
322 projectId: error.projectId,
323 instanceId: error.instanceId,
324 }),
325 }),
326 );
327 },
328 // A delegated child this declined to continue still owes its parent a
329 // result. Once a continuation run exists this is a no-op; that run settles it.
330 (effect, input) =>
331 effect.pipe(
332 Effect.andThen(
333 Effect.gen(function* () {
334 const threads = yield* ThreadManagementService.ThreadManagementService;
335 yield* threads.recoverDelegatedTask(input.threadId, input.sourceRunId);
336 }),
337 ),
338 ),
339);