1import type {
2 OrchestrationV2PendingBackgroundTask,
3 OrchestrationV2ProviderThread,
4 OrchestrationV2Run,
5 OrchestrationV2ThreadShell,
6 OrchestrationV2TurnItem,
7 ThreadId,
8 ThreadPullRequestLink,
9} from "@t3tools/contracts";
10import { isOrchestrationV2WorkActive } from "@t3tools/contracts";
11import * as Predicate from "effect/Predicate";
12
13import { resolveT3McpToolId } from "./t3McpToolPresentation.ts";
14
15const BACKGROUND_TURN_ITEM_TYPES = new Set<OrchestrationV2TurnItem["type"]>([
16 "command_execution",
17 "dynamic_tool",
18 "subagent",
19]);
20
21// Plans and text streams can stay open across tool calls without executing work.
22const EXECUTION_TURN_ITEM_TYPES = new Set<OrchestrationV2TurnItem["type"]>([
23 ...BACKGROUND_TURN_ITEM_TYPES,
24 "file_change",
25 "file_search",
26 "web_search",
27 "compaction",
28]);
29
30const TERMINAL_RUN_STATUSES = new Set<OrchestrationV2Run["status"]>([
31 "cancelled",
32 "completed",
33 "failed",
34 "interrupted",
35 "rolled_back",
36]);
37
38// waiting is completed work awaiting checkpoints; exclude rolled_back entirely because provider threads are not run-tied.
39const SETTLED_FOR_BACKGROUND_WAIT_RUN_STATUSES = new Set<OrchestrationV2Run["status"]>([
40 "cancelled",
41 "completed",
42 "failed",
43 "interrupted",
44 "waiting",
45]);
46
47void TERMINAL_RUN_STATUSES;
48
49export type PendingBackgroundWorkTask = OrchestrationV2PendingBackgroundTask;
50
51export function turnItemUpdateCanEndBackgroundWork(
52 item: Pick<OrchestrationV2TurnItem, "type" | "status">,
53): boolean {
54 return BACKGROUND_TURN_ITEM_TYPES.has(item.type) && !isOrchestrationV2WorkActive(item.status);
55}
56
57export function deriveBlockedOnThreadWait(input: {
58 readonly activeRun: Pick<OrchestrationV2Run, "id" | "status"> | null;
59 readonly turnItems: ReadonlyArray<OrchestrationV2TurnItem>;
60}): boolean {
61 if (input.activeRun?.status !== "running") return false;
62 let hasWait = false;
63 for (const item of input.turnItems) {
64 if (
65 item.runId !== input.activeRun.id ||
66 !EXECUTION_TURN_ITEM_TYPES.has(item.type) ||
67 (item.status !== "running" && item.status !== "waiting") ||
68 // App-owned rows track the child, including the child a blocking delegate is waiting for.
69 (item.type === "subagent" && item.origin === "app_owned")
70 ) {
71 continue;
72 }
73 if (item.type !== "dynamic_tool") return false;
74 const tool = resolveT3McpToolId(item.toolName);
75 if (
76 tool !== "t3_thread_wait" &&
77 !(
78 tool === "delegate_task" &&
79 Predicate.isObject(item.input) &&
80 Reflect.get(item.input, "mode") === "wait"
81 )
82 ) {
83 return false;
84 }
85 hasWait = true;
86 }
87 return hasWait;
88}
89
90export function backgroundWorkHoldsCompletion(
91 tasks: ReadonlyArray<Pick<PendingBackgroundWorkTask, "kind" | "awaited">>,
92): boolean {
93 return tasks.some(
94 (task) => task.awaited === true || backgroundWorkKindHoldsCompletion(task.kind),
95 );
96}
97
98// Commands and monitors are as often a dev server or its watchdog as a wait, so they hold
99// only when the agent marks them awaited.
100function backgroundWorkKindHoldsCompletion(kind: PendingBackgroundWorkTask["kind"]): boolean {
101 switch (kind) {
102 case "command":
103 case "monitor":
104 return false;
105 case "subagent":
106 case "background_task":
107 return true;
108 }
109}
110
111const PULL_REQUEST_WATCH_TASK_ID = "pull-request-watch:";
112
113export const isPullRequestWatchTask = (task: Pick<PendingBackgroundWorkTask, "taskId">) =>
114 task.taskId.startsWith(PULL_REQUEST_WATCH_TASK_ID);
115
116/**
117 * A monitor wakes the agent once per line, so a progress wake run's completion waits for the
118 * monitor to end. A pull request watch wakes only on news and holds while its checks run.
119 */
120export function monitorHoldsWakeCompletion(
121 thread: Pick<OrchestrationV2ThreadShell, "latestRunBackgroundWake" | "pendingBackgroundTasks">,
122): boolean {
123 return (
124 thread.latestRunBackgroundWake === true &&
125 (thread.pendingBackgroundTasks ?? []).some(
126 (task) => task.kind === "monitor" && !isPullRequestWatchTask(task),
127 )
128 );
129}
130
131type PendingBackgroundWorkRun = Pick<OrchestrationV2Run, "id" | "ordinal" | "status">;
132
133type PendingBackgroundWorkPullRequest = Pick<
134 ThreadPullRequestLink,
135 "host" | "repository" | "number" | "url" | "source" | "watch"
136> & {
137 readonly snapshot: Pick<NonNullable<ThreadPullRequestLink["snapshot"]>, "state"> | null;
138};
139
140type PendingBackgroundWorkProviderThread = Pick<
141 OrchestrationV2ProviderThread,
142 "id" | "pendingBackgroundTasks"
143> & { readonly providerInstanceId?: OrchestrationV2ProviderThread["providerInstanceId"] };
144
145type PendingBackgroundWorkTurnItem = {
146 readonly id: OrchestrationV2TurnItem["id"] | string;
147 readonly type: OrchestrationV2TurnItem["type"];
148 readonly status: OrchestrationV2TurnItem["status"];
149 readonly title: string | null;
150 /** Shell readers join the owner, even when its provider thread belongs to a child app thread. */
151 readonly owningProviderInstanceId?: OrchestrationV2ProviderThread["providerInstanceId"];
152 readonly runId?: OrchestrationV2Run["id"] | string | null;
153 readonly nativeItemRef?: {
154 readonly nativeId: string | null;
155 } | null;
156 readonly input?: unknown;
157 readonly monitor?: boolean | undefined;
158 readonly prompt?: string | undefined;
159 readonly childThreadId?: ThreadId | null;
160};
161
162function isLatestRunSettledForBackgroundWait(
163 latestRun: PendingBackgroundWorkRun | null | undefined,
164): boolean {
165 if (latestRun === undefined || latestRun === null) {
166 return false;
167 }
168 return SETTLED_FOR_BACKGROUND_WAIT_RUN_STATUSES.has(latestRun.status);
169}
170
171function isPersistentDynamicToolInput(input: unknown): boolean {
172 if (input === null || typeof input !== "object" || Array.isArray(input)) {
173 return false;
174 }
175 return Reflect.get(input, "persistent") === true;
176}
177
178function descriptionFromTurnItem(item: PendingBackgroundWorkTurnItem): string | undefined {
179 if (typeof item.title === "string" && item.title.trim().length > 0) {
180 return item.title.trim();
181 }
182 if (item.type === "command_execution" && typeof item.input === "string") {
183 const command = item.input.trim();
184 return command.length > 0 ? command : undefined;
185 }
186 if (item.type === "dynamic_tool") {
187 const toolName = Reflect.get(item, "toolName");
188 if (typeof toolName === "string" && toolName.trim().length > 0) {
189 return toolName.trim();
190 }
191 }
192 if (item.type === "subagent" && typeof item.prompt === "string") {
193 const prompt = item.prompt.trim();
194 return prompt.length > 0 ? prompt : undefined;
195 }
196 return undefined;
197}
198
199function pendingTaskFromTurnItem(
200 taskId: string,
201 item: PendingBackgroundWorkTurnItem,
202): PendingBackgroundWorkTask {
203 const description = descriptionFromTurnItem(item);
204 const named = { taskId, ...(description === undefined ? {} : { description }) };
205 switch (item.type) {
206 case "subagent":
207 return {
208 ...named,
209 kind: "subagent",
210 ...(item.owningProviderInstanceId === undefined
211 ? {}
212 : { providerInstanceId: item.owningProviderInstanceId }),
213 ...(item.childThreadId == null ? {} : { childThreadId: item.childThreadId }),
214 };
215 case "command_execution":
216 return { ...named, kind: item.monitor === true ? "monitor" : "command" };
217 default:
218 return { ...named, kind: "background_task" };
219 }
220}
221
222function nativeTaskIdFromTurnItem(item: PendingBackgroundWorkTurnItem): string {
223 const nativeId = item.nativeItemRef?.nativeId;
224 if (typeof nativeId === "string" && nativeId.length > 0) {
225 return nativeId;
226 }
227 return String(item.id);
228}
229
230/** Items Stop ends, without the pending list's settled-run gate. */
231export function pendingBackgroundTurnItems<Item extends PendingBackgroundWorkTurnItem>(input: {
232 readonly turnItems: ReadonlyArray<Item>;
233 readonly runs?: ReadonlyArray<PendingBackgroundWorkRun>;
234}): ReadonlyArray<Item> {
235 const rolledBackRunIds = new Set(
236 (input.runs ?? []).filter((run) => run.status === "rolled_back").map((run) => String(run.id)),
237 );
238 return input.turnItems.filter(
239 (item) =>
240 BACKGROUND_TURN_ITEM_TYPES.has(item.type) &&
241 isOrchestrationV2WorkActive(item.status) &&
242 !(item.type === "dynamic_tool" && isPersistentDynamicToolInput(item.input)) &&
243 (item.runId === undefined ||
244 item.runId === null ||
245 !rolledBackRunIds.has(String(item.runId))),
246 );
247}
248
249export function derivePendingBackgroundWork(input: {
250 readonly latestRun: PendingBackgroundWorkRun | null | undefined;
251 readonly providerThreads: ReadonlyArray<PendingBackgroundWorkProviderThread>;
252 readonly turnItems: ReadonlyArray<PendingBackgroundWorkTurnItem>;
253 readonly activeProviderThreadId?: string | null;
254 readonly hasActiveRun?: boolean;
255 /** Excludes rolled-back runs; optional only when the caller already filtered them, as SQL shell reads do. */
256 readonly runs?: ReadonlyArray<PendingBackgroundWorkRun>;
257 readonly pullRequests?: ReadonlyArray<PendingBackgroundWorkPullRequest> | undefined;
258}): ReadonlyArray<PendingBackgroundWorkTask> {
259 const hasActiveRun =
260 input.hasActiveRun ??
261 input.runs?.some(
262 (run) => run.status === "preparing" || run.status === "starting" || run.status === "running",
263 ) ??
264 false;
265 if (hasActiveRun) {
266 return [];
267 }
268 if (input.latestRun == null) {
269 return pullRequestWatchTasks(input.pullRequests);
270 }
271 if (!isLatestRunSettledForBackgroundWait(input.latestRun)) {
272 return [];
273 }
274
275 const byTaskId = new Map<string, PendingBackgroundWorkTask>();
276
277 const providerThreads =
278 input.activeProviderThreadId === undefined || input.activeProviderThreadId === null
279 ? input.providerThreads
280 : input.providerThreads.filter((thread) => thread.id === input.activeProviderThreadId);
281
282 for (const providerThread of providerThreads) {
283 for (const task of providerThread.pendingBackgroundTasks ?? []) {
284 if (task.taskId.length === 0 || byTaskId.has(task.taskId)) {
285 continue;
286 }
287 const { description: rawDescription, ...named } = task;
288 const description = rawDescription?.trim();
289 byTaskId.set(task.taskId, {
290 ...named,
291 ...(task.kind === "subagent"
292 ? { providerInstanceId: providerThread.providerInstanceId }
293 : {}),
294 ...(description === undefined || description.length === 0 ? {} : { description }),
295 });
296 }
297 }
298
299 for (const item of pendingBackgroundTurnItems(input)) {
300 const taskId = nativeTaskIdFromTurnItem(item);
301 if (byTaskId.has(taskId)) {
302 continue;
303 }
304
305 byTaskId.set(taskId, pendingTaskFromTurnItem(taskId, item));
306 }
307
308 for (const task of pullRequestWatchTasks(input.pullRequests)) byTaskId.set(task.taskId, task);
309
310 return Array.from(byTaskId.values());
311}
312
313function pullRequestWatchTasks(
314 pullRequests: ReadonlyArray<PendingBackgroundWorkPullRequest> | undefined,
315): Array<PendingBackgroundWorkTask> {
316 return (pullRequests ?? []).flatMap((link) =>
317 // The watch reactor ends a merged PR's watch without waking the agent, so it holds nothing.
318 link.watch === undefined ||
319 link.source === "stack-dismissed" ||
320 link.snapshot?.state === "merged"
321 ? []
322 : [
323 {
324 // Links store their normalized identity, and this id only has to be unique within the
325 // thread, so it reads those fields instead of parsing the URL.
326 taskId: `${PULL_REQUEST_WATCH_TASK_ID}${link.host.toLowerCase()}/${link.repository.toLowerCase()}#${link.number}`,
327 description: `Watching pull request #${link.number}`,
328 kind: "monitor" as const,
329 // Once every check finished, a person moves the PR next: the watch stays on but holds nothing.
330 ...(link.watch.checksPending === false ? {} : { awaited: true as const }),
331 },
332 ],
333 );
334}