Open History to see the last commit
1import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
2import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
3import {
4 CommandId,
5 isOrchestrationV2WorkActive,
6 latestProviderTurnForAttempt,
7 MessageId,
8 type OrchestrationV2Run,
9 type RunId,
10 type ThreadId,
11} from "@t3tools/contracts";
12import * as DateTime from "effect/DateTime";
13import * as Effect from "effect/Effect";
14import type { ProjectionRuntimeRecoveryState } from "./ProjectionStore.ts";
15
16import * as ServerSettings from "../serverSettings.ts";
17import { isNativeMaintenanceCommand } from "./CommandPolicy.ts";
18import * as ThreadManagementService from "./ThreadManagementService.ts";
19import {
20 awaitingProviderThreadId,
21 cancelledRosterTaskWork,
22 cancelledTurnItemWork,
23 continuationFailedBeforePrompt,
24 isAwaitedWork,
25 isRestartContinuationSource,
26 cancelledBeforeStart,
27 isRestartNoteSource,
28 neverStarted,
29 restartCancelledBackgroundWorkNote,
30 restartContinuationNote,
31} from "./RestartBackgroundNote.ts";
32
33const CONTINUE_PROMPT =
34 "Continue where you left off. Your last turn was cut short, so a command or other action it started may have finished or may still be running. Before you repeat one that changes state, confirm it did not take effect and is not still running; if you cannot tell, say so instead of repeating it.";
35
36const MAX_CONTINUATION_CHAIN = 3;
37const CONTINUATION_LOOP_WINDOW_MS = 5 * 60 * 1_000;
38
39function restartContinuationBlocked(
40 projection: Pick<
41 ProjectionRuntimeRecoveryState,
42 "thread" | "runtimeRequests" | "providerTurns" | "turnItems"
43 >,
44 source: OrchestrationV2Run,
45 now: DateTime.Utc,
46 settled: boolean,
47): boolean {
48 const { thread } = projection;
49 if (
50 thread.archivedAt !== null ||
51 thread.deletedAt !== null ||
52 thread.settledOverride === "settled" ||
53 (thread.snoozedUntil != null && DateTime.isGreaterThan(thread.snoozedUntil, now))
54 )
55 return true;
56 const turn = latestProviderTurnForAttempt(projection.providerTurns, source.activeAttemptId);
57 if (
58 !settled &&
59 turn !== undefined &&
60 (turn.status === "completed" || turn.status === "failed" || turn.status === "interrupted")
61 )
62 return true;
63 const waitingRequest = projection.runtimeRequests.some(
64 (request) =>
65 request.kind !== "dynamic_tool_call" &&
66 request.kind !== "auth_refresh" &&
67 request.responseCapability.type !== "message" &&
68 (request.status === "pending" ||
69 ((request.status === "expired" || request.status === "cancelled") &&
70 request.responseCapability.type === "not_resumable" &&
71 turn !== undefined &&
72 turn.id === request.providerTurnId)),
73 );
74 if (waitingRequest) return true;
75 return projection.turnItems.some(
76 (item) =>
77 item.type === "secret_request" &&
78 item.runId === source.id &&
79 (item.secretStatus === "pending" ||
80 (item.secretStatus === "cancelled" &&
81 item.completedAt !== null &&
82 source.completedAt !== null &&
83 DateTime.Equivalence(item.completedAt, source.completedAt))),
84 );
85}
86
87/** App-owned delegations continue in their own threads and report back. */
88function hasOpenAwaitedWork(
89 projection: Pick<ProjectionRuntimeRecoveryState, "runs" | "turnItems">,
90 providerThread: ProjectionRuntimeRecoveryState["providerThreads"][number],
91): boolean {
92 const openItem = projection.turnItems.some((item) => {
93 if (item.status !== "pending" && item.status !== "running" && item.status !== "waiting")
94 return false;
95 if (item.type === "subagent" && item.origin === "app_owned" && item.childThreadId !== null)
96 return false;
97 const work = cancelledTurnItemWork(item);
98 return (
99 work !== undefined &&
100 isAwaitedWork(work) &&
101 awaitingProviderThreadId(item, projection.runs) === providerThread.id
102 );
103 });
104 return (
105 openItem ||
106 (providerThread.ownerNodeId === null &&
107 (providerThread.pendingBackgroundTasks ?? []).some((task) =>
108 isAwaitedWork(cancelledRosterTaskWork(task)),
109 ))
110 );
111}
112
113/** Settled runs require a live session so stale crash records cannot wake them. */
114export function restartContinuationRun(
115 projection: Pick<
116 ProjectionRuntimeRecoveryState,
117 | "thread"
118 | "runs"
119 | "providerThreads"
120 | "providerSessions"
121 | "providerTurns"
122 | "runtimeRequests"
123 | "turnItems"
124 >,
125 now: DateTime.Utc,
126): OrchestrationV2Run | undefined {
127 const run = projection.runs.reduce<OrchestrationV2Run | undefined>(
128 (latest, candidate) =>
129 candidate.status !== "queued" &&
130 !neverStarted(candidate) &&
131 (!latest || runRanAfter(candidate, latest))
132 ? candidate
133 : latest,
134 undefined,
135 );
136 if (!run) return;
137 const preparedContinuation =
138 run.status === "starting" && run.restartContinuationOfRunId !== undefined;
139 const settled = run.status === "completed" || run.status === "waiting";
140 if (run.status !== "running" && !preparedContinuation && !settled) return;
141 if (restartContinuationBlocked(projection, run, now, settled)) return;
142 const liveTurnRequired = !preparedContinuation && !settled;
143 if (projection.thread.providerInstanceId !== run.providerInstanceId) return;
144 const providerThread = projection.providerThreads.find(
145 (thread) => thread.id === run.providerThreadId,
146 );
147 if (
148 !providerThread ||
149 providerThread.appThreadId !== projection.thread.id ||
150 providerThread.ownerNodeId !== null ||
151 providerThread.providerInstanceId !== run.providerInstanceId ||
152 providerThread.nativeThreadRef?.nativeId == null ||
153 providerThread.nativeThreadRef.strength !== "strong" ||
154 providerThread.nativeThreadRef.driver !== providerThread.driver ||
155 (liveTurnRequired && providerThread.status !== "active") ||
156 providerThread.status === "closed" ||
157 providerThread.status === "archived"
158 )
159 return;
160 const session = projection.providerSessions.find(
161 (candidate) => candidate.id === providerThread.providerSessionId,
162 );
163 // Adapters remain ready during turns; only stopped or failed sessions rule out liveness.
164 if (
165 session === undefined ||
166 session.providerInstanceId !== run.providerInstanceId ||
167 session.driver !== providerThread.driver ||
168 (!preparedContinuation && (session.status === "stopped" || session.status === "error"))
169 )
170 return;
171 if (settled) return hasOpenAwaitedWork(projection, providerThread) ? run : undefined;
172 if (
173 liveTurnRequired &&
174 !projection.providerTurns.some(
175 (turn) =>
176 turn.providerThreadId === providerThread.id &&
177 turn.runAttemptId === run.activeAttemptId &&
178 turn.status === "running",
179 )
180 )
181 return;
182 return run;
183}
184
185export const continueRestartedRun = Effect.fn("RestartContinuation.continueRestartedRun")(
186 function* (input: { readonly threadId: ThreadId; readonly sourceRunId: RunId }) {
187 const settings = yield* ServerSettings.ServerSettingsService;
188 const enabled = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null));
189 if (!enabled) return;
190 const threads = yield* ThreadManagementService.ThreadManagementService;
191 const messageId = MessageId.make(`message:restart-continuation:${input.sourceRunId}`);
192 const projection = yield* threads.getThreadRecords(
193 input.threadId,
194 ["messages", "runs", "providerTurns", "attempts", "runtimeRequests", "turnItems"],
195 { messageIds: [messageId], turnItemTypes: ["secret_request"] },
196 );
197 if (
198 !resolveProjectSettings(enabled, projection.thread.projectId).settings
199 .continueThreadsAfterServerUpdate
200 )
201 return;
202 if (projection.messages.some((message) => message.id === messageId)) return;
203 const source = projection.runs.find((run) => run.id === input.sourceRunId);
204 if (
205 !source ||
206 !(
207 isRestartContinuationSource(source, projection.providerTurns) ||
208 continuationFailedBeforePrompt(source, projection.providerTurns)
209 ) ||
210 restartContinuationBlocked(
211 projection,
212 source,
213 yield* DateTime.now,
214 isRestartNoteSource(source, projection.providerTurns),
215 )
216 )
217 return;
218 if (
219 projection.runs.some(
220 (run) =>
221 run.id !== source.id &&
222 run.status !== "queued" &&
223 !cancelledBeforeStart(run) &&
224 runRanAfter(run, source),
225 )
226 )
227 return;
228 if (projection.thread.providerInstanceId !== source.providerInstanceId) return;
229 // A turn that keeps crashing the server would otherwise restart it forever.
230 let chainLength = 0;
231 const loopWindowStart =
232 DateTime.toEpochMillis(source.completedAt ?? source.requestedAt) -
233 CONTINUATION_LOOP_WINDOW_MS;
234 for (
235 let run: OrchestrationV2Run | undefined = source;
236 run?.restartContinuationOfRunId !== undefined && chainLength < MAX_CONTINUATION_CHAIN;
237 run = projection.runs.find((candidate) => candidate.id === run?.restartContinuationOfRunId)
238 ) {
239 if (DateTime.toEpochMillis(run.requestedAt) < loopWindowStart) break;
240 chainLength += 1;
241 }
242 if (chainLength >= MAX_CONTINUATION_CHAIN) {
243 yield* Effect.logWarning("Not continuing a run restarted too many times in a row", {
244 threadId: input.threadId,
245 sourceRunId: input.sourceRunId,
246 });
247 return;
248 }
249 const sourceRecords = yield* threads.getThreadRecords(
250 input.threadId,
251 ["messages", "turnItems"],
252 {
253 messageIds: [source.userMessageId],
254 turnItemRunIds: [source.id],
255 turnItemTypes: ["run_interrupt_request"],
256 },
257 );
258 if (
259 sourceRecords.turnItems.some(
260 (item) => item.runId === source.id && item.type === "run_interrupt_request",
261 )
262 )
263 return;
264 // A parent's Stop reaches its open delegated tasks through an effect a restart can cut short.
265 const { lineage, forkedFrom } = projection.thread;
266 const taskId =
267 source.delegatedTaskId ?? (forkedFrom?.type === "node" ? forkedFrom.nodeId : undefined);
268 if (
269 lineage.relationshipToParent === "subagent" &&
270 lineage.parentThreadId !== null &&
271 taskId !== undefined
272 ) {
273 const task = (yield* threads.getThreadRecords(lineage.parentThreadId, ["subagents"], {
274 subagentIds: [taskId],
275 })).subagents.find((candidate) => candidate.id === taskId);
276 // A settled task no longer owns later work in its thread.
277 const parentRunId =
278 task !== undefined && task.result === null && isOrchestrationV2WorkActive(task.status)
279 ? task.runId
280 : null;
281 if (
282 parentRunId !== null &&
283 (yield* threads.getThreadRecords(lineage.parentThreadId, ["turnItems"], {
284 turnItemTypes: ["run_interrupt_request"],
285 turnItemRunIds: [parentRunId],
286 })).turnItems.length > 0
287 )
288 return;
289 }
290 const sourceMessage = sourceRecords.messages.find(
291 (message) => message.id === source.userMessageId,
292 );
293 if (sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) return;
294 const note = restartContinuationNote(
295 source,
296 projection.runs,
297 projection.providerTurns,
298 projection.attempts,
299 );
300 const noteText =
301 note.work.length === 0 ? undefined : restartCancelledBackgroundWorkNote(note.work);
302 const text =
303 noteText === undefined
304 ? CONTINUE_PROMPT
305 : note.settled
306 ? noteText
307 : `${noteText}\n\n${CONTINUE_PROMPT}`;
308 yield* threads
309 .dispatch({
310 type: "message.dispatch",
311 commandId: CommandId.make(`command:restart-continuation:${input.sourceRunId}`),
312 threadId: input.threadId,
313 messageId,
314 text,
315 notification: {
316 source: { kind: "system" },
317 outcome: "updated",
318 summary: note.settled
319 ? "T3 Code restarted and stopped background work"
320 : "T3 Code restarted and resumed this turn",
321 detail: text,
322 },
323 attachments: [],
324 modelSelection: source.modelSelection,
325 dispatchMode: { type: "start_immediately" },
326 createdBy: "agent",
327 creationSource: "server",
328 restartContinuationOfRunId: input.sourceRunId,
329 })
330 .pipe(
331 // Disabled instances, and threads above a policy ceiling, cannot resume; still run success
332 // cleanup so delegated children recover without futile retries.
333 Effect.catchTags({
334 ThreadManagementProviderInstanceDisabledError: (error) =>
335 Effect.logInfo("orchestration-v2.restart-continuation.instance-disabled", {
336 threadId: input.threadId,
337 sourceRunId: input.sourceRunId,
338 projectId: error.projectId,
339 instanceId: error.instanceId,
340 }),
341 OrchestratorRuntimeModeAboveCeilingError: () =>
342 Effect.logInfo("orchestration-v2.restart-continuation.above-ceiling", {
343 threadId: input.threadId,
344 sourceRunId: input.sourceRunId,
345 }),
346 }),
347 );
348 },
349 // Declined child continuations still owe parent results; existing continuation runs settle them instead.
350 (effect, input) =>
351 effect.pipe(
352 Effect.andThen(
353 Effect.gen(function* () {
354 const threads = yield* ThreadManagementService.ThreadManagementService;
355 yield* threads.recoverDelegatedTask(input.threadId, { sourceRunId: input.sourceRunId });
356 }),
357 ),
358 ),
359);