authorgravatar for 228381532+im-kvijay@users.noreply.github.comKrishna Vijay <228381532+im-kvijay@users.noreply.github.com> 2026-10-10 08:32:35-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 10:32:35-05:00
logafd98d7d1840ed76f57d3ece556257aebc00f1bc
tree886f2413269f1b5b2f6e1187184031e98aa2bcf0
parent8c777fbfade340f2092c011483c7fb0cde44c675
signature Signed by PGP key B5690EEEBB952194

fix(server): stopped native subagents no longer read as Running (#17223)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Yash Singh <saiansh2525@gmail.com>

12 files changed, 973 insertions(+), 124 deletions(-)

apps/server/src/mcp/TaskCancelNativeSubagent.integration.test.ts created+357
......@@ -0,0 +1,357 @@
1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it } from "@effect/vitest";
3import {
4 CommandId,
5 EnvironmentId,
6 MessageId,
7 NodeId,
8 ProjectId,
9 ProviderDriverKind,
10 ProviderInstanceId,
11 ProviderThreadId,
12 ProviderTurnId,
13 ThreadId,
14 TurnItemId,
15 type OrchestrationV2DomainEvent,
16} from "@t3tools/contracts";
17import * as DateTime from "effect/DateTime";
18import * as Effect from "effect/Effect";
19import * as Fiber from "effect/Fiber";
20import * as Layer from "effect/Layer";
21import * as Queue from "effect/Queue";
22import * as Stream from "effect/Stream";
23
24import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";
25import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
26import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
27import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
28import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
29import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
30import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
31import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
32import * as ProjectService from "../project/ProjectService.ts";
33import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
34import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
35import * as SecretRequests from "../secrets/SecretRequests.ts";
36import type { McpInvocationScope } from "./McpInvocationContext.ts";
37import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
38
39const driver = ProviderDriverKind.make("claudeAgent");
40const instanceId = ProviderInstanceId.make("claudeAgent");
41const modelSelection = { instanceId, model: "claude-test" };
42
43// A delegated Claude child ends its turn while a background native subagent it
44// started keeps running (#17154). Cancelling the task stops the child, which
45// ends that subagent's process, so the task must end instead of waiting on it.
46it.effect("task_cancel ends a delegated child's background native subagent", () =>
47 Effect.scoped(
48 Effect.gen(function* () {
49 const cwd = yield* checkpointWorkspace("task-cancel-native-subagent");
50 const turns: Array<{
51 readonly turn: ProviderAdapter.ProviderAdapterV2TurnInput;
52 readonly events: Queue.Queue<ProviderAdapter.ProviderAdapterV2Event>;
53 }> = [];
54 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
55 instanceId,
56 driver,
57 getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2),
58 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
59 openSession: (input) =>
60 Effect.gen(function* () {
61 const now = yield* DateTime.now;
62 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
63 return {
64 instanceId,
65 driver,
66 providerSessionId: input.providerSessionId,
67 providerSession: {
68 id: input.providerSessionId,
69 driver,
70 providerInstanceId: instanceId,
71 status: "ready",
72 cwd,
73 model: modelSelection.model,
74 capabilities: ClaudeProviderCapabilitiesV2,
75 createdAt: now,
76 updatedAt: now,
77 lastError: null,
78 },
79 events: Stream.fromQueue(events),
80 ensureThread: ({ threadId }) =>
81 Effect.succeed({
82 id: ProviderThreadId.make(`provider-thread:${threadId}`),
83 driver,
84 providerInstanceId: instanceId,
85 providerSessionId: input.providerSessionId,
86 appThreadId: threadId,
87 ownerNodeId: null,
88 nativeThreadRef: { driver, nativeId: `native:${threadId}`, strength: "strong" },
89 nativeConversationHeadRef: null,
90 status: "idle",
91 firstRunOrdinal: null,
92 lastRunOrdinal: null,
93 handoffIds: [],
94 forkedFrom: null,
95 createdAt: now,
96 updatedAt: now,
97 }),
98 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
99 startTurn: (turn) =>
100 Effect.gen(function* () {
101 turns.push({ turn, events });
102 yield* Queue.offer(events, {
103 type: "provider_turn.updated",
104 driver,
105 providerTurn: {
106 id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
107 providerThreadId: turn.providerThread.id,
108 nodeId: turn.rootNodeId,
109 runAttemptId: turn.attemptId,
110 nativeTurnRef: null,
111 ordinal: turn.providerTurnOrdinal,
112 status: "running",
113 startedAt: now,
114 completedAt: null,
115 },
116 });
117 }),
118 steerTurn: () => Effect.die("unused"),
119 interruptTurn: () => Effect.void,
120 respondToRuntimeRequest: () => Effect.die("unused"),
121 readThreadSnapshot: () => Effect.die("unused"),
122 rollbackThread: () => Effect.die("unused"),
123 forkThread: () => Effect.die("unused"),
124 };
125 }),
126 };
127 const layerAdapters = ProviderAdapterRegistry.layerSingle(adapter);
128 const layerOrchestrator = ProviderReplayHarness.layerWithRegistry(
129 { name: "task-cancel-native-subagent" },
130 layerAdapters,
131 { runEffectWorker: false },
132 );
133 const layerOrchestration = Layer.merge(
134 layerOrchestrator,
135 ThreadManagementService.layer.pipe(Layer.provide(layerOrchestrator)),
136 );
137 const layerTest = OrchestratorMcpService.layer.pipe(
138 Layer.provideMerge(layerOrchestration),
139 Layer.provide(
140 Layer.mergeAll(
141 layerAdapters,
142 Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }),
143 Layer.mock(ProjectService.ProjectService)({}),
144 Layer.mock(SecretRequests.SecretRequests)({}),
145 Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
146 NodeServices.layer,
147 ),
148 ),
149 );
150
151 yield* Effect.gen(function* () {
152 const orchestrator = yield* Orchestrator.OrchestratorV2;
153 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
154 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
155 const parentThreadId = ThreadId.make("thread:task-cancel-parent");
156 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
157 orchestrator.streamDomainEvents.pipe(
158 Stream.filter(predicate),
159 Stream.take(1),
160 Stream.runDrain,
161 Effect.forkScoped,
162 );
163 const turnRunning = () =>
164 watch(
165 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
166 );
167 yield* orchestrator.dispatch({
168 type: "thread.create",
169 commandId: CommandId.make("create-parent"),
170 threadId: parentThreadId,
171 projectId: ProjectId.make("project:task-cancel"),
172 title: "Parent",
173 modelSelection,
174 runtimeMode: "full-access",
175 interactionMode: "default",
176 branch: null,
177 worktreePath: cwd,
178 createdBy: "user",
179 creationSource: "web",
180 });
181 const parentRunning = yield* turnRunning();
182 yield* orchestrator.dispatch({
183 type: "message.dispatch",
184 commandId: CommandId.make("start-parent"),
185 threadId: parentThreadId,
186 messageId: MessageId.make("message:start-parent"),
187 text: "Delegate a review",
188 attachments: [],
189 dispatchMode: { type: "start_immediately" },
190 createdBy: "user",
191 creationSource: "web",
192 });
193 yield* worker.drain();
194 yield* Fiber.join(parentRunning);
195
196 const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!;
197 const childRunning = yield* turnRunning();
198 yield* orchestrator.dispatch({
199 type: "delegated_task.request",
200 commandId: CommandId.make("delegate"),
201 parentThreadId,
202 parentRunId: parentRun.id,
203 parentNodeId: parentRun.rootNodeId!,
204 task: "Review the change",
205 modelSelection,
206 runtimeMode: "full-access",
207 interactionMode: "default",
208 completionWake: "always",
209 createdBy: "agent",
210 creationSource: "mcp",
211 });
212 yield* worker.drain();
213 yield* Fiber.join(childRunning);
214 const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!;
215 const child = turns.find((entry) => entry.turn.threadId === task.childThreadId)!;
216 const childTurn = (yield* orchestrator.getThreadProjection(task.childThreadId!))
217 .providerTurns[0]!;
218
219 // The child starts a background native subagent, then ends its turn.
220 const now = yield* DateTime.now;
221 const subagentId = NodeId.make("node:native-subagent");
222 const native = {
223 threadId: task.childThreadId!,
224 runId: child.turn.runId,
225 origin: "provider_native",
226 driver,
227 providerInstanceId: instanceId,
228 providerThreadId: childTurn.providerThreadId,
229 childThreadId: null,
230 prompt: "Check the tests",
231 title: null,
232 status: "running",
233 result: null,
234 startedAt: now,
235 completedAt: null,
236 updatedAt: now,
237 } as const;
238 const waiting = yield* watch(
239 (event) =>
240 event.type === "run.updated" &&
241 event.payload.id === child.turn.runId &&
242 event.payload.status === "waiting",
243 );
244 for (const event of [
245 {
246 type: "node.updated",
247 driver,
248 node: {
249 id: subagentId,
250 threadId: native.threadId,
251 runId: native.runId,
252 parentNodeId: child.turn.rootNodeId,
253 rootNodeId: child.turn.rootNodeId,
254 kind: "subagent",
255 status: "running",
256 countsForRun: false,
257 providerThreadId: childTurn.providerThreadId,
258 providerTurnId: childTurn.id,
259 nativeItemRef: null,
260 runtimeRequestId: null,
261 checkpointScopeId: null,
262 startedAt: now,
263 completedAt: null,
264 },
265 },
266 {
267 type: "subagent.updated",
268 driver,
269 subagent: {
270 ...native,
271 id: subagentId,
272 parentNodeId: child.turn.rootNodeId,
273 createdBy: "agent",
274 nativeTaskRef: null,
275 model: null,
276 },
277 },
278 {
279 type: "turn_item.updated",
280 driver,
281 turnItem: {
282 ...native,
283 id: TurnItemId.make("turn-item:native-subagent"),
284 nodeId: subagentId,
285 providerTurnId: childTurn.id,
286 nativeItemRef: null,
287 parentItemId: null,
288 ordinal: 1,
289 type: "subagent",
290 subagentId,
291 },
292 },
293 {
294 type: "provider_turn.updated",
295 driver,
296 providerTurn: { ...childTurn, status: "completed", completedAt: now },
297 },
298 {
299 type: "turn.terminal",
300 driver,
301 providerThreadId: childTurn.providerThreadId,
302 providerTurnId: childTurn.id,
303 runOrdinal: child.turn.runOrdinal,
304 status: "completed",
305 failure: null,
306 threadDisposition: "reusable",
307 },
308 ] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>) {
309 yield* Queue.offer(child.events, event);
310 }
311 yield* Fiber.join(waiting);
312 yield* worker.drain();
313
314 const scope: McpInvocationScope = {
315 environmentId: EnvironmentId.make("environment:task-cancel"),
316 requestNamespace: "provider-session:task-cancel",
317 thread: {
318 threadId: parentThreadId,
319 providerSessionId: "provider-session:task-cancel",
320 providerInstanceId: instanceId,
321 },
322 client: undefined,
323 capabilities: new Set(["orchestration"]),
324 issuedAt: 1,
325 };
326 const before = yield* service.taskStatus(scope, task.id);
327 assert.equal(before.status, "running");
328 assert.equal(before.workState, "waiting_for_children");
329
330 const cancelled = yield* service.cancelTask(scope, {
331 taskId: task.id,
332 clientRequestId: "cancel-task",
333 });
334 assert.equal(cancelled.status, "cancel_requested");
335 yield* worker.drain();
336
337 const childAfter = yield* orchestrator.getThreadProjection(task.childThreadId!);
338 assert.equal(
339 childAfter.subagents.find((entry) => entry.id === subagentId)?.status,
340 "interrupted",
341 );
342 assert.equal(
343 childAfter.nodes.find((entry) => entry.id === subagentId)?.status,
344 "interrupted",
345 );
346 assert.equal(
347 childAfter.turnItems.find((item) => item.type === "subagent")?.status,
348 "interrupted",
349 );
350 // The child's own turn had completed, so the task keeps that result.
351 const after = yield* service.taskStatus(scope, task.id);
352 assert.equal(after.status, "completed");
353 assert.notEqual(after.workState, "waiting_for_children");
354 }).pipe(Effect.provide(layerTest));
355 }),
356 ),
357);
apps/server/src/orchestration-v2/BackgroundWorkStop.integration.test.ts+143-1
......@@ -414,6 +414,113 @@ const stopEarlierBackgroundWork = ({
414414
415415 const codexProviderThread = (yield* orchestrator.getThreadProjection(threadId))
416416 .providerThreads[0]!;
417 // A native subagent's own thread, still showing its runless root turn as working.
418 const subagentThread = (name: string) =>
419 Effect.gen(function* () {
420 const subagentThreadId = ThreadId.make(`thread:${name}`);
421 const rootId = NodeId.make(`node:${name}-root`);
422 yield* orchestrator.dispatch({
423 type: "thread.create",
424 commandId: CommandId.make(`create-${name}`),
425 threadId: subagentThreadId,
426 projectId: ProjectId.make("project:background-work-stop"),
427 title: name,
428 modelSelection,
429 runtimeMode: "full-access",
430 interactionMode: "default",
431 branch: null,
432 worktreePath: cwd,
433 createdBy: "agent",
434 creationSource: "provider",
435 });
436 yield* sink.write({
437 events: [
438 {
439 id: EventId.make(`${name}-root`),
440 type: "node.updated",
441 threadId: subagentThreadId,
442 occurredAt: now,
443 payload: {
444 id: rootId,
445 threadId: subagentThreadId,
446 runId: null,
447 parentNodeId: null,
448 rootNodeId: rootId,
449 kind: "root_turn",
450 status: "running",
451 countsForRun: false,
452 providerThreadId: null,
453 providerTurnId: null,
454 nativeItemRef: null,
455 runtimeRequestId: null,
456 checkpointScopeId: null,
457 startedAt: now,
458 completedAt: null,
459 },
460 },
461 ],
462 });
463 return { threadId: subagentThreadId, rootId };
464 });
465 const { threadId: reviewerThreadId, rootId: reviewerRootId } =
466 yield* subagentThread("reviewer");
467 // The reviewer starts a tester, which runs on a thread of its own.
468 const tester = yield* subagentThread("tester");
469 const nestedSubagentId = NodeId.make("subagent:nested");
470 const nestedItemId = TurnItemId.make("turn-item:nested-subagent");
471 const nested = {
472 threadId: reviewerThreadId,
473 runId: null,
474 origin: "provider_native",
475 driver,
476 providerInstanceId: instanceId,
477 providerThreadId: null,
478 childThreadId: tester.threadId,
479 prompt: "Check the tests",
480 title: null,
481 status: "running",
482 result: null,
483 startedAt: now,
484 completedAt: null,
485 updatedAt: now,
486 } as const;
487 yield* sink.write({
488 events: [
489 // A subagent the reviewer started, which the provider records on the
490 // reviewer's own thread.
491 {
492 id: EventId.make("nested-subagent-item"),
493 type: "turn-item.updated",
494 threadId: reviewerThreadId,
495 occurredAt: now,
496 payload: {
497 ...nested,
498 id: nestedItemId,
499 nodeId: nestedSubagentId,
500 providerTurnId: null,
501 nativeItemRef: null,
502 parentItemId: null,
503 ordinal: 1,
504 type: "subagent",
505 subagentId: nestedSubagentId,
506 },
507 },
508 {
509 id: EventId.make("nested-subagent"),
510 type: "subagent.updated",
511 threadId: reviewerThreadId,
512 occurredAt: now,
513 payload: {
514 ...nested,
515 id: nestedSubagentId,
516 parentNodeId: reviewerRootId,
517 createdBy: "agent",
518 nativeTaskRef: null,
519 model: null,
520 },
521 },
522 ],
523 });
417524 // A settled run with its root node, attempt, provider turn and,
418525 // optionally, a command or native subagent it left running.
419526 const settledRun = (input: {
......@@ -550,11 +657,35 @@ const stopEarlierBackgroundWork = ({
550657 origin: "provider_native",
551658 driver,
552659 providerInstanceId: instanceId,
553 childThreadId: null,
660 childThreadId: reviewerThreadId,
554661 prompt: "Review the change",
555662 result: null,
556663 },
557664 });
665 if (item.kind === "subagent") {
666 events.push({
667 id: EventId.make(`subagent:${input.ordinal}`),
668 type: "subagent.updated",
669 threadId,
670 runId,
671 occurredAt: now,
672 payload: {
673 ...base,
674 id: NodeId.make(`subagent:${input.ordinal}`),
675 parentNodeId: nodeId,
676 origin: "provider_native",
677 createdBy: "agent",
678 driver,
679 providerInstanceId: instanceId,
680 providerThreadId: null,
681 childThreadId: reviewerThreadId,
682 nativeTaskRef: null,
683 prompt: "Review the change",
684 model: null,
685 result: null,
686 },
687 });
688 }
558689 }
559690 return { runId, providerTurnId, events };
560691 };
......@@ -747,6 +878,17 @@ const stopEarlierBackgroundWork = ({
747878 ? ["interrupted", "interrupted"]
748879 : ["interrupted", "interrupted", "interrupted"],
749880 );
881 // The reviewer died with its provider process, so it no longer reads as running.
882 assert.equal(after.subagents[0]?.status, "interrupted");
883 const reviewerThread = yield* orchestrator.getThreadProjection(reviewerThreadId);
884 assert.equal(reviewerThread.nodes[0]?.status, "interrupted");
885 assert.equal(reviewerThread.subagents[0]?.status, "interrupted");
886 assert.equal(
887 reviewerThread.turnItems.find((item) => item.id === nestedItemId)?.status,
888 "interrupted",
889 );
890 const testerThread = yield* orchestrator.getThreadProjection(tester.threadId);
891 assert.equal(testerThread.nodes[0]?.status, "interrupted");
750892 }).pipe(
751893 Effect.provide(
752894 ProviderReplayHarness.layerWithRegistry(
apps/server/src/orchestration-v2/FoundationPersistence.test.ts+8-4
......@@ -3052,9 +3052,9 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
30523052 completedAt: null,
30533053 });
30543054 // The parent run settled while its background subagent kept working,
3055 // then the server died. Recovery already cancels the parent's subagent
3056 // item, entity, and node; the child's runless root turn lives on another
3057 // thread and must be settled too.
3055 // Stop ended the subagent's item, then the server died. Recovery cancels
3056 // the parent's subagent entity and node; the child's runless root turn
3057 // lives on another thread and must be settled too.
30583058 yield* eventSink.commitCommand({
30593059 commandId: CommandId.make("command:foundation-native-subagent"),
30603060 threadId: parentId,
......@@ -3186,7 +3186,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
31863186 parentItemId: null,
31873187 ordinal: 1,
31883188 type: "subagent",
3189 status: "running",
3189 status: "interrupted",
31903190 title: null,
31913191 startedAt: now,
31923192 completedAt: null,
......@@ -3222,6 +3222,10 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => {
32223222
32233223 const parentProjection = yield* projectionStore.getThreadProjection(parentId);
32243224 assert.equal(parentProjection.subagents[0]?.status, "cancelled");
3225 assert.equal(
3226 parentProjection.nodes.find((node) => node.id === subagentId)?.status,
3227 "cancelled",
3228 );
32253229 const childProjection = yield* projectionStore.getThreadProjection(childId);
32263230 const childRoot = childProjection.nodes.find((candidate) => candidate.id === childRootId);
32273231 assert.equal(childRoot?.status, "cancelled");
apps/server/src/orchestration-v2/Orchestrator.ts+64-1
......@@ -117,6 +117,8 @@ import { ProviderSwitchServiceV2 } from "./ProviderSwitchService.ts";
117117import { isAutomaticCompletionRun, queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts";
118118import { RuntimePolicyV2 } from "./RuntimePolicy.ts";
119119import {
120 endOrphanedNativeSubagent,
121 endRunlessRootTurns,
120122 makeSubagentChildThread,
121123 subagentResultForRun,
122124 delegatedTaskProgress,
......@@ -8308,7 +8310,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83088310 readonly events: Ref.Ref<Array<OrchestrationV2DomainEvent>>;
83098311 readonly projection: Pick<
83108312 OrchestrationV2ThreadProjection,
8311 "runs" | "turnItems" | "providerThreads"
8313 "runs" | "nodes" | "subagents" | "turnItems" | "providerThreads"
83128314 >;
83138315 readonly stoppedProviderThreadId: OrchestrationV2ProviderThread["id"] | null;
83148316 readonly throughRunOrdinal: number;
......@@ -8317,6 +8319,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83178319 Effect.gen(function* () {
83188320 const emitEvent = emit(input.events, input.command);
83198321 const runOrdinals = new Map(input.projection.runs.map((run) => [run.id, run.ordinal]));
8322 const allocateEventId = () =>
8323 idAllocator.allocate.event({
8324 threadId: input.command.threadId,
8325 commandId: input.command.commandId,
8326 });
8327 const childThreadIds = new Set<ThreadId>();
83208328 const liveness = new Map<string, boolean>();
83218329 const hasLiveSession = (providerThreadId: OrchestrationV2ProviderThread["id"]) =>
83228330 Effect.gen(function* () {
......@@ -8366,6 +8374,61 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
83668374 updatedAt: input.now,
83678375 },
83688376 });
8377 if (item.type !== "subagent") continue;
8378 // A native subagent died with the process too: end its record, its own
8379 // thread's runless root turn, and the subagents it started there, which
8380 // a provider can record on that thread rather than this one.
8381 const dying: Array<{
8382 readonly projection: Pick<
8383 OrchestrationV2ThreadProjection,
8384 "nodes" | "subagents" | "turnItems"
8385 >;
8386 readonly subagent: Extract<OrchestrationV2TurnItem, { readonly type: "subagent" }>;
8387 }> = [{ projection: input.projection, subagent: item }];
8388 for (const { projection, subagent } of dying) {
8389 const ended = yield* endOrphanedNativeSubagent({
8390 projection,
8391 subagentId: subagent.subagentId,
8392 status: "interrupted",
8393 now: input.now,
8394 emitted: yield* Ref.get(input.events),
8395 allocateEventId,
8396 }).pipe(mapDispatchError(input.command));
8397 yield* Ref.update(input.events, (events) => [...events, ...ended]);
8398 const childThreadId = subagent.childThreadId;
8399 if (
8400 subagent.origin !== "provider_native" ||
8401 childThreadId === null ||
8402 childThreadIds.has(childThreadId)
8403 ) {
8404 continue;
8405 }
8406 childThreadIds.add(childThreadId);
8407 const child = yield* projectionStore
8408 .getThreadRecords(childThreadId, ["nodes", "subagents", "turnItems"], {
8409 // Runless items are the subagent's; any run there is the user's own.
8410 turnItemRunIds: [null],
8411 turnItemStatuses: ["pending", "running", "waiting"],
8412 })
8413 .pipe(
8414 Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }),
8415 mapDispatchError(input.command),
8416 );
8417 if (child === null) continue;
8418 const childEnded = yield* endRunlessRootTurns({
8419 threadId: childThreadId,
8420 providerInstanceId: subagent.providerInstanceId,
8421 projection: child,
8422 status: "interrupted",
8423 now: input.now,
8424 emitted: yield* Ref.get(input.events),
8425 allocateEventId,
8426 }).pipe(mapDispatchError(input.command));
8427 yield* Ref.update(input.events, (events) => [...events, ...childEnded]);
8428 for (const nested of child.turnItems) {
8429 if (nested.type === "subagent") dying.push({ projection: child, subagent: nested });
8430 }
8431 }
83698432 }
83708433 for (const providerThread of input.projection.providerThreads) {
83718434 // A live process owns its roster and reports clearing it.
apps/server/src/orchestration-v2/ProjectionStore.ts+17-1
......@@ -589,6 +589,11 @@ function needsRecovery(
589589 (run.status === "queued" && run.queueHeld !== true),
590590 ) ||
591591 projection.runtimeRequests.some((request) => request.status === "pending") ||
592 projection.subagents.some(
593 (subagent) =>
594 subagent.origin === "provider_native" &&
595 ["pending", "running", "waiting"].includes(subagent.status),
596 ) ||
592597 projection.providerSessions.some(
593598 (session) => session.status !== "stopped" && session.status !== "error",
594599 ) ||
......@@ -3664,6 +3669,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
36643669 AND node.status IN ('pending', 'running', 'waiting')
36653670 )
36663671 UNION
3672 -- No native subagent outlives the provider process that ran it.
3673 SELECT thread_id FROM orchestration_v2_projection_subagents
3674 WHERE origin = 'provider_native'
3675 AND status IN ('pending', 'running', 'waiting')
3676 UNION
36673677 SELECT item.thread_id FROM orchestration_v2_projection_turn_items AS item
36683678 WHERE NOT EXISTS (
36693679 SELECT 1 FROM orchestration_v2_projection_runs AS run
......@@ -4067,6 +4077,11 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
40674077 WHERE item.thread_id = ${threadId} AND item.type = 'subagent'
40684078 AND item.status IN ('pending', 'running', 'waiting')
40694079 )
4080 OR node.node_id IN (
4081 SELECT subagent_id FROM orchestration_v2_projection_subagents
4082 WHERE thread_id = ${threadId} AND origin = 'provider_native'
4083 AND status IN ('pending', 'running', 'waiting')
4084 )
40704085 )
40714086 ORDER BY COALESCE(node.started_at, ''), node.node_id ASC
40724087 `,
......@@ -4075,7 +4090,8 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
40754090 WHERE subagent.thread_id = ${threadId}
40764091 AND subagent.status IN ('pending', 'starting', 'running', 'waiting')
40774092 AND (
4078 subagent.run_id IN (
4093 subagent.origin = 'provider_native'
4094 OR subagent.run_id IN (
40794095 SELECT run_id FROM orchestration_v2_projection_runs
40804096 WHERE thread_id = ${threadId}
40814097 AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting')
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts+2
......@@ -1227,6 +1227,7 @@ it.effect(
12271227 {
12281228 id: staleSubagentNodeId,
12291229 runId: settledRunId,
1230 origin: "provider_native",
12301231 driver: ProviderDriverKind.make("claude"),
12311232 providerInstanceId: claudeInstanceId,
12321233 status: "running",
......@@ -1235,6 +1236,7 @@ it.effect(
12351236 // Already finished with a real result: must never be overwritten.
12361237 id: doneSubagentNodeId,
12371238 runId: settledRunId,
1239 origin: "provider_native",
12381240 driver: ProviderDriverKind.make("claude"),
12391241 providerInstanceId: claudeInstanceId,
12401242 status: "completed",
apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts+27-116
......@@ -22,6 +22,10 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
2222import * as ProjectionStore from "./ProjectionStore.ts";
2323import * as ServerSettings from "../serverSettings.ts";
2424import { restartContinuationRun } from "./RestartContinuation.ts";
25import {
26 endOrphanedNativeSubagent,
27 endRunlessRootTurns,
28} from "@t3tools/provider-core/server/subagentProjection";
2529import {
2630 cancelledRosterTaskWork,
2731 cancelledTurnItemWork,
......@@ -88,10 +92,6 @@ function isNonterminalTurnItemStatus(status: string): boolean {
8892 return status === "pending" || status === "running" || status === "waiting";
8993}
9094
91function isNonterminalSubagentStatus(status: string): boolean {
92 return status === "pending" || status === "running" || status === "waiting";
93}
94
9595function isNonterminalNodeStatus(status: string): boolean {
9696 return status === "pending" || status === "running" || status === "waiting";
9797}
......@@ -362,26 +362,6 @@ export const make = Effect.gen(function* () {
362362 payload: { ...node, status: "cancelled", completedAt: now },
363363 });
364364 }
365 for (const subagent of projection.subagents.filter(
366 (candidate) =>
367 candidate.runId === run.id &&
368 !isAppOwnedDelegation(candidate) &&
369 (candidate.status === "pending" ||
370 candidate.status === "running" ||
371 candidate.status === "waiting"),
372 )) {
373 events.push({
374 id: yield* allocateEventId(),
375 type: "subagent.updated",
376 threadId: projection.thread.id,
377 runId: run.id,
378 nodeId: subagent.id,
379 driver: subagent.driver,
380 providerInstanceId: subagent.providerInstanceId,
381 occurredAt: now,
382 payload: { ...subagent, status: "cancelled", completedAt: now, updatedAt: now },
383 });
384 }
385365 for (const providerTurn of projection.providerTurns.filter(
386366 (candidate) =>
387367 candidate.runAttemptId !== null &&
......@@ -486,103 +466,34 @@ export const make = Effect.gen(function* () {
486466 });
487467 }
488468 }
489 if (item.type !== "subagent") {
490 continue;
491 }
492 // Cancelling only the turn item would leave the linked subagent entity
493 // non-terminal forever, since the dead provider process can no longer
494 // emit its terminal event. Match the exact linked id so a subagent
495 // that already finished is never overwritten.
496 const staleSubagent = projection.subagents.find(
497 (candidate) =>
498 candidate.id === item.subagentId && isNonterminalSubagentStatus(candidate.status),
499 );
500 if (staleSubagent !== undefined) {
501 events.push({
502 id: yield* allocateEventId(),
503 type: "subagent.updated",
504 threadId: projection.thread.id,
505 ...(item.runId === null ? {} : { runId: item.runId }),
506 nodeId: staleSubagent.id,
507 driver: staleSubagent.driver,
508 providerInstanceId: staleSubagent.providerInstanceId,
509 occurredAt: now,
510 payload: { ...staleSubagent, status: "cancelled", completedAt: now, updatedAt: now },
511 });
512 }
513 const staleSubagentNode = projection.nodes.find(
514 (candidate) =>
515 candidate.id === item.subagentId && isNonterminalNodeStatus(candidate.status),
469 }
470 // No provider process outlives the server, so a native subagent still open
471 // here never reports its end, whatever state its run and item are in.
472 for (const subagent of projection.subagents ?? []) {
473 events.push(
474 ...(yield* endOrphanedNativeSubagent({
475 projection,
476 subagentId: subagent.id,
477 status: "cancelled",
478 now,
479 emitted: events,
480 allocateEventId,
481 })),
516482 );
517 if (staleSubagentNode !== undefined && !cancelledStaleNodeIds.has(staleSubagentNode.id)) {
518 cancelledStaleNodeIds.add(staleSubagentNode.id);
519 events.push({
520 id: yield* allocateEventId(),
521 type: "node.updated",
522 threadId: projection.thread.id,
523 ...(item.runId === null ? {} : { runId: item.runId }),
524 nodeId: staleSubagentNode.id,
525 providerInstanceId,
526 occurredAt: now,
527 payload: { ...staleSubagentNode, status: "cancelled", completedAt: now },
528 });
529 }
530483 }
531 // A provider-native subagent thread has no runs: its work is a runless
532 // root turn, plus items under it (Claude's live progress item), that
533 // only the dead provider process could settle. Left running, the child
484 // A provider-native subagent thread has no runs; left running, the child
534485 // would show as working forever.
535 const cancelledStaleItemIds = new Set(
536 events.flatMap((event) => (event.type === "turn-item.updated" ? [event.payload.id] : [])),
537 );
538 for (const node of projection.nodes) {
539 if (
540 node.kind !== "root_turn" ||
541 node.runId !== null ||
542 !isNonterminalNodeStatus(node.status) ||
543 cancelledStaleNodeIds.has(node.id)
544 ) {
545 continue;
546 }
547 cancelledStaleNodeIds.add(node.id);
548 events.push({
549 id: yield* allocateEventId(),
550 type: "node.updated",
486 events.push(
487 ...(yield* endRunlessRootTurns({
551488 threadId: projection.thread.id,
552 nodeId: node.id,
553489 providerInstanceId: projection.thread.providerInstanceId,
554 occurredAt: now,
555 payload: { ...node, status: "cancelled", completedAt: now },
556 });
557 for (const item of projection.turnItems) {
558 if (
559 item.nodeId !== node.id ||
560 item.runId !== null ||
561 !isNonterminalTurnItemStatus(item.status) ||
562 cancelledStaleItemIds.has(item.id)
563 ) {
564 continue;
565 }
566 cancelledStaleItemIds.add(item.id);
567 events.push({
568 id: yield* allocateEventId(),
569 type: "turn-item.updated",
570 threadId: projection.thread.id,
571 nodeId: node.id,
572 providerInstanceId: projection.thread.providerInstanceId,
573 occurredAt: now,
574 payload: {
575 ...item,
576 status: "cancelled",
577 completedAt: now,
578 updatedAt: now,
579 ...(item.type === "reasoning" || item.type === "assistant_message"
580 ? { streaming: false }
581 : {}),
582 },
583 });
584 }
585 }
490 projection,
491 status: "cancelled",
492 now,
493 emitted: events,
494 allocateEventId,
495 })),
496 );
586497 // All provider processes are gone on startup/shutdown: clear any
587498 // persisted Waiting roster (including idle threads from settled roots)
588499 // and idle active threads without resurrecting active status.
apps/server/src/orchestration-v2/RestartContinuation.test.ts+1
......@@ -481,6 +481,7 @@ it.effect.each([
481481 {
482482 id: "subagent:background",
483483 runId,
484 origin: "provider_native",
484485 driver,
485486 providerInstanceId: instanceId,
486487 status: "running",
apps/server/src/orchestration-v2/RunExecutionService.test.ts+131
......@@ -1432,6 +1432,137 @@ it.effect("keeps ingesting owned child events after the root turn terminalizes",
14321432 }),
14331433);
14341434
1435it.effect("ends a nested native subagent and its thread when the root run is interrupted", () =>
1436 Effect.gen(function* () {
1437 const key = "run-execution-nested-native";
1438 const threadId = ThreadId.make(`thread:${key}`);
1439 const childThreadId = ThreadId.make(`thread:${key}:child`);
1440 const grandchildThreadId = ThreadId.make(`thread:${key}:grandchild`);
1441 const runId = RunId.make(`run:${key}`);
1442 const attemptId = RunAttemptId.make(`attempt:${key}`);
1443 const providerInstanceId = ProviderInstanceId.make("codex");
1444 const providerThreadId = ProviderThreadId.make(`provider-thread:${key}`);
1445 const rootProviderTurnId = ProviderTurnId.make(`provider-turn:${key}`);
1446 const nestedId = NodeId.make(`node:${key}:nested`);
1447 const grandchildRootId = NodeId.make(`node:${key}:grandchild-root`);
1448 const finalized = yield* Deferred.make<ReadonlyArray<OrchestrationV2DomainEvent>>();
1449 const layerTest = RunExecutionService.layer.pipe(
1450 Layer.provide(
1451 Layer.mergeAll(
1452 McpAppModelContext.layerEmpty,
1453 Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
1454 Layer.mock(EventSink.EventSinkV2)({
1455 write: () => Effect.succeed([]),
1456 writeWithEffects: (input) =>
1457 Effect.as(
1458 input.events.some((event) => event.type === "run.updated")
1459 ? Deferred.succeed(finalized, input.events)
1460 : Effect.void,
1461 [],
1462 ),
1463 }),
1464 IdAllocator.layer,
1465 Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
1466 ingestNormalized: () => Effect.succeed([]),
1467 }),
1468 ServerSettings.layerTest(),
1469 ),
1470 ),
1471 );
1472 const subagentThread = (id: ThreadId, parentThreadId: ThreadId) =>
1473 ({
1474 type: "app_thread.created",
1475 driver,
1476 appThread: {
1477 id,
1478 lineage: { parentThreadId, relationshipToParent: "subagent", rootThreadId: threadId },
1479 },
1480 }) as ProviderAdapter.ProviderAdapterV2Event;
1481 const events: ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event> = [
1482 subagentThread(childThreadId, threadId),
1483 subagentThread(grandchildThreadId, childThreadId),
1484 {
1485 type: "subagent.updated",
1486 driver,
1487 subagent: {
1488 id: nestedId,
1489 threadId: childThreadId,
1490 runId: null,
1491 origin: "provider_native",
1492 childThreadId: grandchildThreadId,
1493 status: "running",
1494 },
1495 } as ProviderAdapter.ProviderAdapterV2Event,
1496 {
1497 type: "node.updated",
1498 driver,
1499 node: {
1500 id: grandchildRootId,
1501 threadId: grandchildThreadId,
1502 runId: null,
1503 kind: "root_turn",
1504 status: "running",
1505 },
1506 } as ProviderAdapter.ProviderAdapterV2Event,
1507 {
1508 type: "turn.terminal",
1509 driver,
1510 providerThreadId,
1511 providerTurnId: rootProviderTurnId,
1512 runOrdinal: 1,
1513 status: "interrupted",
1514 failure: null,
1515 threadDisposition: "reusable",
1516 },
1517 ];
1518
1519 yield* Effect.gen(function* () {
1520 const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
1521 yield* runExecution.startRootRun({
1522 commandId: CommandId.make(`command:${key}`),
1523 appThread: { id: threadId } as OrchestrationV2AppThread,
1524 providerSessionId: ProviderSessionId.make(`session:${key}`),
1525 session: {
1526 events: Stream.fromIterable(events),
1527 startTurn: () => Effect.void,
1528 } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
1529 run: { id: runId, threadId, ordinal: 1, providerInstanceId } as OrchestrationV2Run,
1530 rootNode: { id: NodeId.make(`node:${key}`) } as OrchestrationV2ExecutionNode,
1531 checkpointScope: {
1532 id: CheckpointScopeId.make(`checkpoint-scope:${key}`),
1533 } as OrchestrationV2CheckpointScope,
1534 providerThread: { id: providerThreadId, driver } as OrchestrationV2ProviderThread,
1535 attempt: { id: attemptId, providerTurnId: rootProviderTurnId } as OrchestrationV2RunAttempt,
1536 attemptId,
1537 providerTurnOrdinal: 1,
1538 message: {
1539 messageId: MessageId.make(`message:${key}:user`),
1540 text: "Spawn a nested subagent.",
1541 attachments: [],
1542 createdBy: "user",
1543 creationSource: "web",
1544 },
1545 modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
1546 runtimePolicy: {
1547 runtimeMode: "full-access",
1548 interactionMode: "default",
1549 cwd: process.cwd(),
1550 approvalPolicy: "never",
1551 sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false },
1552 },
1553 });
1554 }).pipe(Effect.provide(layerTest));
1555
1556 const ended = (yield* Deferred.await(finalized)).flatMap((event) =>
1557 event.type === "subagent.updated" || event.type === "node.updated"
1558 ? [[event.type, event.payload.id, event.payload.status]]
1559 : [],
1560 );
1561 assert.deepInclude(ended, ["subagent.updated", nestedId, "interrupted"]);
1562 assert.deepInclude(ended, ["node.updated", grandchildRootId, "interrupted"]);
1563 }),
1564);
1565
14351566it.effect("keeps ingesting a late background command item completion after root terminal", () =>
14361567 Effect.gen(function* () {
14371568 const observed = yield* runBackgroundItemScenario("bg-command", (ids) => [
apps/server/src/orchestration-v2/RunExecutionService.ts+8-1
......@@ -1044,7 +1044,14 @@ export const layer: Layer.Layer<
10441044 // Preserve childThreadId linkage for the root-run lifetime even
10451045 // after the subagent row terminalizes, so open child-thread
10461046 // nodes can still be proven linked on a later root interrupt.
1047 if (belongsToRootRun) {
1047 // A provider records nested native subagents on the child
1048 // thread, runless; they die with this run too.
1049 if (
1050 belongsToRootRun ||
1051 (belongsToOwnedChildThread &&
1052 event.subagent.runId === null &&
1053 event.subagent.origin === "provider_native")
1054 ) {
10481055 yield* Ref.update(openRunOwnedSubagents, (current) => {
10491056 const withLink = withLinkedChildThreadId(current, event.subagent.childThreadId);
10501057 const subagents = new Map(withLink.subagents);
apps/server/src/orchestration-v2/SubagentProjection.test.ts+64
......@@ -15,12 +15,15 @@ import {
1515 ThreadId,
1616} from "@t3tools/contracts";
1717import * as DateTime from "effect/DateTime";
18import * as Effect from "effect/Effect";
1819
1920import {
2021 makeSubagentChildThread,
2122 delegatedTaskProgress,
2223 subagentResultForRun,
2324 makeSubagentConversationArtifacts,
25 endOrphanedNativeSubagent,
26 endRunlessRootTurns,
2427} from "@t3tools/provider-core/server/subagentProjection";
2528
2629import { emptyProjection } from "./ProjectionStore.ts";
......@@ -314,3 +317,64 @@ it("exposes the provider failure rather than a progress message from the failed
314317 assert.equal(result.turnItemId, artifacts.turnItem.id);
315318 assert.isNull(result.messageId);
316319});
320
321it.effect("ends an orphaned native subagent's node when only its record already ended", () =>
322 Effect.gen(function* () {
323 const id = NodeId.make("node:orphaned-native-subagent");
324 const subagent = { id, origin: "provider_native", status: "running", runId: null };
325 const events = yield* endOrphanedNativeSubagent({
326 projection: {
327 subagents: [subagent],
328 nodes: [{ id, status: "running", runId: null }],
329 turnItems: [],
330 } as never,
331 subagentId: id,
332 status: "cancelled",
333 now: yield* DateTime.now,
334 emitted: [
335 { type: "subagent.updated", payload: { ...subagent, status: "cancelled" } },
336 ] as never,
337 allocateEventId: () => Effect.succeed(EventId.make("event:orphaned-native-subagent")),
338 });
339 assert.deepEqual(
340 events.map((event) => [event.type, event.type === "node.updated" && event.payload.status]),
341 [["node.updated", "cancelled"]],
342 );
343 }),
344);
345
346it.effect("ends a runless root turn's items when its node already ended", () =>
347 Effect.gen(function* () {
348 const root = {
349 id: NodeId.make("node:runless-root"),
350 kind: "root_turn",
351 status: "running",
352 runId: null,
353 };
354 const item = (id: string) => ({ id, nodeId: root.id, runId: null, status: "running" });
355 const events = yield* endRunlessRootTurns({
356 threadId: childThreadId,
357 providerInstanceId: parentProviderInstanceId,
358 projection: {
359 nodes: [root],
360 turnItems: [item("turn-item:command"), item("turn-item:reasoning")],
361 } as never,
362 status: "cancelled",
363 now: yield* DateTime.now,
364 emitted: [
365 {
366 type: "turn-item.updated",
367 payload: { ...item("turn-item:command"), status: "cancelled" },
368 },
369 { type: "node.updated", payload: { ...root, status: "cancelled" } },
370 ] as never,
371 allocateEventId: () => Effect.succeed(EventId.make("event:runless-root")),
372 });
373 assert.deepEqual(
374 events.map(
375 (event) => `${event.type} ${event.type === "turn-item.updated" && event.payload.id}`,
376 ),
377 ["turn-item.updated turn-item:reasoning"],
378 );
379 }),
380);
packages/provider-core/src/server/subagentProjection.ts+151
......@@ -1,4 +1,5 @@
11import type {
2 EventId,
23 MessageId,
34 ModelSelection,
45 NodeId,
......@@ -6,6 +7,7 @@ import type {
67 OrchestrationV2Actor,
78 OrchestrationV2ConversationMessage,
89 OrchestrationV2CreationSource,
10 OrchestrationV2DomainEvent,
911 OrchestrationV2ProviderRef,
1012 OrchestrationV2Run,
1113 OrchestrationV2ThreadProjection,
......@@ -18,6 +20,7 @@ import type {
1820} from "@t3tools/contracts";
1921import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
2022import * as DateTime from "effect/DateTime";
23import * as Effect from "effect/Effect";
2124import { isOrchestrationV2WorkActive } from "@t3tools/contracts";
2225
2326function trimmed(value: string | null | undefined): string | undefined {
......@@ -257,3 +260,151 @@ export function delegatedTaskProgress(projection: {
257260 resultRun,
258261 };
259262}
263
264/** Ids that `emitted` already wrote with an event of `type`. */
265function writtenIds(
266 emitted: ReadonlyArray<OrchestrationV2DomainEvent>,
267 type: "subagent.updated" | "node.updated" | "turn-item.updated",
268): ReadonlySet<string> {
269 return new Set(emitted.flatMap((event) => (event.type === type ? [event.payload.id] : [])));
270}
271
272/**
273 * Ends a provider-native subagent, its node and its turn item once the provider
274 * process that ran it is gone, since no terminal event will come from it. Each
275 * record is skipped if already settled or already ended by `emitted`.
276 */
277export const endOrphanedNativeSubagent = <E>(input: {
278 readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "subagents" | "turnItems">;
279 readonly subagentId: NodeId;
280 readonly status: "interrupted" | "cancelled";
281 readonly now: DateTime.Utc;
282 readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>;
283 readonly allocateEventId: () => Effect.Effect<EventId, E>;
284}) =>
285 Effect.gen(function* () {
286 const { now, status, subagentId } = input;
287 const subagent = input.projection.subagents.find((entry) => entry.id === subagentId);
288 if (subagent?.origin !== "provider_native") return [];
289 const node = input.projection.nodes.find((entry) => entry.id === subagentId);
290 const item = input.projection.turnItems.find(
291 (entry) => entry.type === "subagent" && entry.subagentId === subagentId,
292 );
293 const scope = {
294 threadId: subagent.threadId,
295 ...(subagent.runId === null ? {} : { runId: subagent.runId }),
296 nodeId: subagentId,
297 providerInstanceId: subagent.providerInstanceId,
298 occurredAt: now,
299 };
300 const events: Array<OrchestrationV2DomainEvent> = [];
301 if (
302 isOrchestrationV2WorkActive(subagent.status) &&
303 !writtenIds(input.emitted, "subagent.updated").has(subagentId)
304 ) {
305 events.push({
306 ...scope,
307 id: yield* input.allocateEventId(),
308 type: "subagent.updated",
309 driver: subagent.driver,
310 payload: { ...subagent, status, completedAt: now, updatedAt: now },
311 });
312 }
313 if (
314 node !== undefined &&
315 isOrchestrationV2WorkActive(node.status) &&
316 !writtenIds(input.emitted, "node.updated").has(subagentId)
317 ) {
318 events.push({
319 ...scope,
320 id: yield* input.allocateEventId(),
321 type: "node.updated",
322 payload: { ...node, status, completedAt: now },
323 });
324 }
325 if (
326 item !== undefined &&
327 isOrchestrationV2WorkActive(item.status) &&
328 !writtenIds(input.emitted, "turn-item.updated").has(item.id)
329 ) {
330 events.push({
331 ...scope,
332 id: yield* input.allocateEventId(),
333 type: "turn-item.updated",
334 payload: { ...item, status, completedAt: now, updatedAt: now },
335 });
336 }
337 return events;
338 });
339
340/**
341 * Ends a provider-native subagent thread's work, a runless root turn plus the
342 * items under it (Claude's live progress), which only the provider process that
343 * ran it could settle. Skips records `emitted` already ended.
344 */
345export const endRunlessRootTurns = <E>(input: {
346 readonly threadId: ThreadId;
347 readonly providerInstanceId: ProviderInstanceId;
348 readonly projection: Pick<OrchestrationV2ThreadProjection, "nodes" | "turnItems">;
349 readonly status: "interrupted" | "cancelled";
350 readonly now: DateTime.Utc;
351 readonly emitted: ReadonlyArray<OrchestrationV2DomainEvent>;
352 readonly allocateEventId: () => Effect.Effect<EventId, E>;
353}) =>
354 Effect.gen(function* () {
355 const { now, status } = input;
356 const endedNodeIds = writtenIds(input.emitted, "node.updated");
357 const endedItemIds = writtenIds(input.emitted, "turn-item.updated");
358 const scope = {
359 threadId: input.threadId,
360 providerInstanceId: input.providerInstanceId,
361 occurredAt: now,
362 };
363 const events: Array<OrchestrationV2DomainEvent> = [];
364 for (const node of input.projection.nodes) {
365 if (
366 node.kind !== "root_turn" ||
367 node.runId !== null ||
368 !isOrchestrationV2WorkActive(node.status)
369 ) {
370 continue;
371 }
372 // Recovery may have ended this node through one of its items; its other
373 // items still need ending.
374 if (!endedNodeIds.has(node.id)) {
375 events.push({
376 ...scope,
377 id: yield* input.allocateEventId(),
378 type: "node.updated",
379 nodeId: node.id,
380 payload: { ...node, status, completedAt: now },
381 });
382 }
383 for (const item of input.projection.turnItems) {
384 if (
385 item.nodeId !== node.id ||
386 item.runId !== null ||
387 !isOrchestrationV2WorkActive(item.status) ||
388 endedItemIds.has(item.id)
389 ) {
390 continue;
391 }
392 events.push({
393 ...scope,
394 id: yield* input.allocateEventId(),
395 type: "turn-item.updated",
396 nodeId: node.id,
397 payload: {
398 ...item,
399 status,
400 completedAt: now,
401 updatedAt: now,
402 ...(item.type === "reasoning" || item.type === "assistant_message"
403 ? { streaming: false }
404 : {}),
405 },
406 });
407 }
408 }
409 return events;
410 });