1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import * as NodeServices from "@effect/platform-node/NodeServices";
3import { assert, it } from "@effect/vitest";
4import {
5 CommandId,
6 EnvironmentId,
7 EventId,
8 MessageId,
9 NodeId,
10 ProjectId,
11 ProviderDriverKind,
12 ProviderInstanceId,
13 ProviderThreadId,
14 ProviderTurnId,
15 RunId,
16 RuntimeRequestId,
17 ThreadId,
18 TurnItemId,
19 type OrchestrationV2DomainEvent,
20} from "@t3tools/contracts";
21import * as DateTime from "effect/DateTime";
22import * as Effect from "effect/Effect";
23import * as Fiber from "effect/Fiber";
24import * as FileSystem from "effect/FileSystem";
25import * as Layer from "effect/Layer";
26import * as Path from "effect/Path";
27import * as Queue from "effect/Queue";
28import * as Stream from "effect/Stream";
29
30import { ClaudeProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/ClaudeAdapterV2.ts";
31import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
32import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
33import * as EventSink from "../orchestration-v2/EventSink.ts";
34import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
35import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
36import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
37import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
38import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
39import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
40import * as ServerSettings from "../serverSettings.ts";
41import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
42import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
43import * as SqlitePersistence from "../persistence/Sqlite.ts";
44import * as ProjectService from "../project/ProjectService.ts";
45import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
46import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
47import * as SecretRequests from "../secrets/SecretRequests.ts";
48import type { McpInvocationScope } from "./McpInvocationContext.ts";
49import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
50import * as RemoteDelegation from "../peer/RemoteDelegation.ts";
51
52const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
53 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
54 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
55);
56
57const driver = ProviderDriverKind.make("claudeAgent");
58const instanceId = ProviderInstanceId.make("claudeAgent");
59const modelSelection = { instanceId, model: "claude-test" };
60
61// Cancellation races with turn completion and checkpoint capture; "recorded" restarts before the Stop outbox runs.
62const cancelBackgroundOnlyTask = (
63 background: "subagent" | "command",
64 {
65 cancelAt = "settled",
66 remote = false,
67 }: {
68 readonly remote?: boolean;
69 readonly cancelAt?: "settled" | "checkpoint" | "turn" | "recorded" | "monitor" | "refused";
70 } = {},
71) =>
72 Effect.scoped(
73 Effect.gen(function* () {
74 const cwd = yield* checkpointWorkspace(`task-cancel-background-${background}`);
75 const turns: Array<{
76 readonly turn: ProviderAdapter.ProviderAdapterV2TurnInput;
77 readonly events: Queue.Queue<ProviderAdapter.ProviderAdapterV2Event>;
78 }> = [];
79 const interrupted: Array<ReadonlyArray<string> | undefined> = [];
80 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
81 instanceId,
82 driver,
83 getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2),
84 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
85 openSession: (input) =>
86 Effect.gen(function* () {
87 const now = yield* DateTime.now;
88 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
89 return {
90 instanceId,
91 driver,
92 providerSessionId: input.providerSessionId,
93 providerSession: {
94 id: input.providerSessionId,
95 driver,
96 providerInstanceId: instanceId,
97 status: "ready",
98 cwd,
99 model: modelSelection.model,
100 capabilities: ClaudeProviderCapabilitiesV2,
101 createdAt: now,
102 updatedAt: now,
103 lastError: null,
104 },
105 events: Stream.fromQueue(events),
106 ensureThread: ({ threadId }) =>
107 Effect.succeed({
108 id: ProviderThreadId.make(`provider-thread:${threadId}`),
109 driver,
110 providerInstanceId: instanceId,
111 providerSessionId: input.providerSessionId,
112 appThreadId: threadId,
113 ownerNodeId: null,
114 nativeThreadRef: { driver, nativeId: `native:${threadId}`, strength: "strong" },
115 nativeConversationHeadRef: null,
116 status: "idle",
117 firstRunOrdinal: null,
118 lastRunOrdinal: null,
119 handoffIds: [],
120 forkedFrom: null,
121 createdAt: now,
122 updatedAt: now,
123 }),
124 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
125 startTurn: (turn) =>
126 Effect.gen(function* () {
127 turns.push({ turn, events });
128 yield* Queue.offer(events, {
129 type: "provider_turn.updated",
130 driver,
131 providerTurn: {
132 id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
133 providerThreadId: turn.providerThread.id,
134 nodeId: turn.rootNodeId,
135 runAttemptId: turn.attemptId,
136 nativeTurnRef: null,
137 ordinal: turn.providerTurnOrdinal,
138 status: "running",
139 startedAt: now,
140 completedAt: null,
141 },
142 });
143 }),
144 steerTurn: () => Effect.die("unused"),
145 interruptTurn: ({ providerThread }) =>
146 Effect.sync(() => {
147 interrupted.push(providerThread.pendingBackgroundTasks?.map((t) => t.taskId));
148 }),
149 respondToRuntimeRequest: () => Effect.die("unused"),
150 readThreadSnapshot: () => Effect.die("unused"),
151 rollbackThread: () => Effect.die("unused"),
152 forkThread: () => Effect.die("unused"),
153 };
154 }),
155 };
156 const layerAdapters = ProviderAdapterRegistry.layerSingle(adapter);
157 const fs = yield* FileSystem.FileSystem;
158 const path = yield* Path.Path;
159 const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-task-cancel-" });
160 const databaseLayer = SqlitePersistence.layerFromPath(
161 path.join(directory, "state.sqlite"),
162 ).pipe(Layer.provide(NodeServices.layer));
163 // Migrate the file first, as an existing install's database already is.
164 yield* Effect.provide(Effect.void, databaseLayer);
165 // Each build is a server on the same database; the second runs startup recovery.
166 let restarted = false;
167 const layerOrchestrator = Layer.suspend(() =>
168 ProviderReplayHarness.layerWithRegistry(
169 { name: "task-cancel-native-subagent" },
170 layerAdapters,
171 restarted
172 ? { databaseLayer, recoverOnStartup: true }
173 : { databaseLayer, runEffectWorker: false },
174 ),
175 );
176 const layerOrchestration = Layer.merge(
177 layerOrchestrator,
178 Layer.effect(
179 ThreadManagementService.ThreadManagementService,
180 Effect.map(ThreadManagementService.ThreadManagementService, (service) =>
181 cancelAt === "refused"
182 ? {
183 ...service,
184 stopDelegatedTasks: () =>
185 Effect.fail(
186 new Orchestrator.OrchestratorDispatchError({
187 commandId: CommandId.make("refused-descendant-stop"),
188 commandType: "thread.stop",
189 cause: "descendant refused stop",
190 }),
191 ),
192 }
193 : service,
194 ),
195 ).pipe(
196 Layer.provide(
197 ThreadManagementService.layer.pipe(
198 Layer.provide(layerOrchestrator),
199 Layer.provide(ServerSettings.layerTest()),
200 Layer.provide(
201 CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
202 ),
203 ),
204 ),
205 Layer.provide(layerOrchestrator),
206 Layer.provide(ServerSettings.layerTest()),
207 Layer.provide(
208 CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
209 ),
210 ),
211 );
212 const layerTest = layerOrchestratorService.pipe(
213 Layer.provideMerge(layerOrchestration),
214 Layer.provide(
215 Layer.mergeAll(
216 layerAdapters,
217 Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }),
218 Layer.mock(ProjectService.ProjectService)({}),
219 Layer.mock(SecretRequests.SecretRequests)({}),
220 Layer.mock(RemoteDelegation.RemoteDelegation)({}),
221 ServerSettings.layerTest(),
222 Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
223 NodeServices.layer,
224 NodeCrypto.layer,
225 ),
226 ),
227 );
228
229 const stopped = yield* Effect.gen(function* () {
230 const orchestrator = yield* Orchestrator.OrchestratorV2;
231 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
232 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
233 const eventSink = yield* EventSink.EventSinkV2;
234 const parentThreadId = ThreadId.make("thread:task-cancel-parent");
235 // Read the cursor before forking: provider events offered right after a watch can
236 // commit before a live-tail subscriber has read its starting sequence.
237 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
238 eventSink.latestSequence().pipe(
239 Effect.flatMap((afterSequence) =>
240 eventSink.stream({ afterSequence }).pipe(
241 Stream.map((stored) => stored.event),
242 Stream.filter(predicate),
243 Stream.take(1),
244 Stream.runDrain,
245 Effect.forkScoped,
246 ),
247 ),
248 );
249 const turnRunning = () =>
250 watch(
251 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
252 );
253 yield* orchestrator.dispatch({
254 type: "thread.create",
255 commandId: CommandId.make("create-parent"),
256 threadId: parentThreadId,
257 projectId: ProjectId.make("project:task-cancel"),
258 title: "Parent",
259 modelSelection,
260 runtimeMode: "full-access",
261 interactionMode: "default",
262 branch: null,
263 worktreePath: cwd,
264 createdBy: "user",
265 creationSource: "web",
266 });
267 const parentRunning = yield* turnRunning();
268 yield* orchestrator.dispatch({
269 type: "message.dispatch",
270 commandId: CommandId.make("start-parent"),
271 threadId: parentThreadId,
272 messageId: MessageId.make("message:start-parent"),
273 text: "Delegate a review",
274 attachments: [],
275 dispatchMode: { type: "start_immediately" },
276 createdBy: "user",
277 creationSource: "web",
278 });
279 yield* worker.drain();
280 yield* Fiber.join(parentRunning);
281
282 const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!;
283 const childRunning = yield* turnRunning();
284 yield* orchestrator.dispatch({
285 type: "delegated_task.request",
286 commandId: CommandId.make("delegate"),
287 parentThreadId,
288 parentRunId: parentRun.id,
289 parentNodeId: parentRun.rootNodeId!,
290 task: "Review the change",
291 modelSelection,
292 runtimeMode: "full-access",
293 interactionMode: "default",
294 completionWake: "always",
295 createdBy: "agent",
296 creationSource: "mcp",
297 });
298 if (cancelAt === "turn") {
299 const pendingTask = (yield* orchestrator.getThreadProjection(parentThreadId))
300 .subagents[0]!;
301 const { runs } = yield* orchestrator.getThreadProjection(pendingTask.childThreadId!);
302 const now = yield* DateTime.now;
303 const current = runs[0]!;
304 const previous = {
305 ...current,
306 id: RunId.make("run:before-continuation"),
307 ordinal: 1,
308 status: "completed" as const,
309 startedAt: now,
310 completedAt: now,
311 };
312 const at = { threadId: current.threadId, occurredAt: now };
313 yield* eventSink.write({
314 commandId: CommandId.make("command:runtime-reconcile:continuation"),
315 events: [
316 {
317 ...at,
318 runId: current.id,
319 id: EventId.make("event:continuation"),
320 type: "run.updated",
321 payload: { ...current, ordinal: 2, restartContinuationOfRunId: previous.id },
322 },
323 {
324 ...at,
325 runId: previous.id,
326 id: EventId.make("event:previous"),
327 type: "run.updated",
328 payload: previous,
329 },
330 ],
331 });
332 }
333 yield* worker.drain();
334 yield* Fiber.join(childRunning);
335 const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!;
336 const child = turns.find((entry) => entry.turn.threadId === task.childThreadId)!;
337 const childTurn = (yield* orchestrator.getThreadProjection(task.childThreadId!))
338 .providerTurns[0]!;
339
340 const now = yield* DateTime.now;
341 const subagentId = NodeId.make("node:native-subagent");
342 const native = {
343 threadId: task.childThreadId!,
344 runId: child.turn.runId,
345 origin: "provider_native",
346 driver,
347 providerInstanceId: instanceId,
348 providerThreadId: childTurn.providerThreadId,
349 childThreadId: null,
350 prompt: "Check the tests",
351 title: null,
352 status: "running",
353 result: null,
354 startedAt: now,
355 completedAt: null,
356 updatedAt: now,
357 } as const;
358 const waiting = yield* watch(
359 (event) =>
360 event.type === "run.updated" &&
361 event.payload.id === child.turn.runId &&
362 event.payload.status === "waiting",
363 );
364 const nativeSubagentEvents = [
365 {
366 type: "node.updated",
367 driver,
368 node: {
369 id: subagentId,
370 threadId: native.threadId,
371 runId: native.runId,
372 parentNodeId: child.turn.rootNodeId,
373 rootNodeId: child.turn.rootNodeId,
374 kind: "subagent",
375 status: "running",
376 countsForRun: false,
377 providerThreadId: childTurn.providerThreadId,
378 providerTurnId: childTurn.id,
379 nativeItemRef: null,
380 runtimeRequestId: null,
381 checkpointScopeId: null,
382 startedAt: now,
383 completedAt: null,
384 },
385 },
386 {
387 type: "subagent.updated",
388 driver,
389 subagent: {
390 ...native,
391 id: subagentId,
392 parentNodeId: child.turn.rootNodeId,
393 createdBy: "agent",
394 nativeTaskRef: null,
395 model: null,
396 },
397 },
398 {
399 type: "turn_item.updated",
400 driver,
401 turnItem: {
402 ...native,
403 id: TurnItemId.make("turn-item:native-subagent"),
404 nodeId: subagentId,
405 providerTurnId: childTurn.id,
406 nativeItemRef: null,
407 parentItemId: null,
408 ordinal: 1,
409 type: "subagent",
410 subagentId,
411 },
412 },
413 ] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>;
414 const shellEvents = [
415 {
416 type: "provider_thread.updated",
417 driver,
418 providerThread: {
419 ...child.turn.providerThread,
420 status: "active",
421 pendingBackgroundTasks: [
422 { taskId: "bg-sleep", description: "sleep 300", kind: "command", awaited: true },
423 ],
424 updatedAt: now,
425 },
426 },
427 ] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>;
428 const started = yield* watch((event) =>
429 background === "subagent"
430 ? event.type === "subagent.updated" && event.payload.id === subagentId
431 : event.type === "provider-thread.updated" &&
432 event.threadId === task.childThreadId &&
433 (event.payload.pendingBackgroundTasks?.length ?? 0) > 0,
434 );
435 // An async question the child asked during its turn, and one no run of this task owns.
436 const question = (id: string, nodeId: NodeId) =>
437 ({
438 type: "runtime_request.updated",
439 driver,
440 runtimeRequest: {
441 id: RuntimeRequestId.make(id),
442 nodeId,
443 providerTurnId: childTurn.id,
444 nativeRequestRef: null,
445 kind: "user_input",
446 status: "pending",
447 responseCapability: { type: "message" },
448 createdAt: now,
449 resolvedAt: null,
450 },
451 }) satisfies ProviderAdapter.ProviderAdapterV2Event;
452 for (const event of [
453 question("request:owned", child.turn.rootNodeId),
454 question("request:unrelated", NodeId.make("node:unrelated")),
455 ...(background === "subagent" ? nativeSubagentEvents : shellEvents),
456 ]) {
457 yield* Queue.offer(child.events, event);
458 }
459 yield* Fiber.join(started);
460 if (remote) {
461 yield* orchestrator.dispatch({
462 type: "delegated_task.remote.request",
463 commandId: CommandId.make("remote-descendant"),
464 parentThreadId: task.childThreadId!,
465 parentRunId: child.turn.runId,
466 parentNodeId: child.turn.rootNodeId,
467 task: "Work in a linked environment",
468 driver,
469 modelSelection,
470 remoteChild: {
471 environmentId: EnvironmentId.make("environment:remote-descendant"),
472 threadId: ThreadId.make("thread:remote-descendant"),
473 label: "Remote",
474 },
475 });
476 }
477 const turnEndEvents = [
478 {
479 type: "provider_turn.updated",
480 driver,
481 providerTurn: { ...childTurn, status: "completed", completedAt: now },
482 },
483 {
484 type: "turn.terminal",
485 driver,
486 providerThreadId: childTurn.providerThreadId,
487 providerTurnId: childTurn.id,
488 runOrdinal: child.turn.runOrdinal,
489 status: "completed",
490 failure: null,
491 threadDisposition: "reusable",
492 },
493 ] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>;
494 const endTurn = Effect.gen(function* () {
495 for (const event of turnEndEvents) yield* Queue.offer(child.events, event);
496 yield* Fiber.join(waiting);
497 });
498 if (cancelAt !== "turn") yield* endTurn;
499 const turnSettled =
500 cancelAt !== "checkpoint" && cancelAt !== "turn" && cancelAt !== "recorded";
501 if (turnSettled) yield* worker.drain();
502 if (cancelAt === "monitor") {
503 // A monitor notification then ran and completed while the shell kept running.
504 const { runs, messages } = yield* orchestrator.getThreadProjection(task.childThreadId!);
505 const run = { ...runs[0]!, id: RunId.make("run:monitor"), ordinal: 2 };
506 const source = { kind: "monitor" } as const;
507 const notification = { source, outcome: "updated", summary: "Ran" } as const;
508 const user = { ...messages[0]!, id: MessageId.make("message:monitor"), runId: run.id };
509 const message = { ...user, notification };
510 const at = { threadId: run.threadId, runId: run.id, occurredAt: now };
511 yield* (yield* EventSink.EventSinkV2).write({
512 commandId: CommandId.make("command:runtime-reconcile:monitor"),
513 events: [
514 { ...at, id: EventId.make("event:m1"), type: "message.updated", payload: message },
515 { ...at, id: EventId.make("event:m2"), type: "run.updated", payload: run },
516 ],
517 });
518 }
519
520 const scope: McpInvocationScope = {
521 environmentId: EnvironmentId.make("environment:task-cancel"),
522 requestNamespace: "provider-session:task-cancel",
523 thread: {
524 threadId: parentThreadId,
525 providerSessionId: "provider-session:task-cancel",
526 providerInstanceId: instanceId,
527 },
528 client: undefined,
529 capabilities: new Set(["orchestration"]),
530 issuedAt: 1,
531 };
532 const before = yield* service.taskStatus(scope, task.id);
533 assert.equal(before.status, "running");
534 assert.equal(before.workState, turnSettled ? "waiting_for_children" : "working");
535
536 const parentTask = Effect.map(orchestrator.getThreadProjection(parentThreadId), (parent) =>
537 parent.subagents.find((entry) => entry.id === task.id),
538 );
539 if (cancelAt === "recorded") {
540 const commandId = CommandId.make("stop-recorded");
541 yield* orchestrator.cancelDelegatedTask({
542 parentThreadId,
543 taskId: task.id,
544 command: {
545 type: "thread.stop",
546 commandId,
547 threadId: task.childThreadId!,
548 stopDelegatedTasks: false,
549 },
550 stopDescendants: () => Effect.void,
551 });
552 const committed = yield* eventSink.readByCommandId({ commandId }).pipe(Stream.runCollect);
553 assert.isTrue(
554 committed.some(
555 ({ event }) =>
556 event.type === "subagent.updated" && event.payload.cancelledRunId !== undefined,
557 ),
558 );
559 assert.isTrue(
560 committed.some(
561 ({ event }) =>
562 event.type === "turn-item.updated" &&
563 event.payload.type === "run_interrupt_request",
564 ),
565 );
566 assert.equal((yield* parentTask)?.status, "running");
567 return { scope, taskId: task.id, settled: undefined };
568 }
569 if (cancelAt === "refused") {
570 yield* Effect.flip(
571 service.cancelTask(scope, { taskId: task.id, clientRequestId: "refused-cancel" }),
572 );
573 const after = yield* parentTask;
574 assert.equal(after?.status, "running");
575 assert.equal(after?.result, null);
576 assert.isUndefined(after?.cancelledRunId);
577 return undefined;
578 }
579 // Settling the task ends the async question it left, after its own row.
580 const settled = yield* watch(
581 (event) =>
582 event.type === "runtime-request.updated" &&
583 event.payload.id === "request:owned" &&
584 event.payload.status === "resolved",
585 );
586 const cancelled = yield* service.cancelTask(scope, {
587 taskId: task.id,
588 clientRequestId: "cancel-task",
589 });
590 assert.equal(cancelled.status, "cancel_requested");
591 assert.equal((yield* parentTask)?.cancelledRunId, child.turn.runId);
592 // The turn completes normally after Stop, before its interrupt reaches the provider.
593 if (cancelAt === "turn") yield* endTurn;
594 yield* worker.drain();
595 const owned = Effect.map(orchestrator.getThreadProjection(task.childThreadId!), (child) =>
596 child.runtimeRequests.find((request) => request.id === "request:owned"),
597 );
598 // task_cancel settles a finished turn itself; a running one settles when it ends.
599 if ((yield* owned)?.status !== "resolved") yield* Fiber.join(settled);
600
601 const childAfter = yield* orchestrator.getThreadProjection(task.childThreadId!);
602 const nativeStatus = background === "subagent" ? "interrupted" : undefined;
603 assert.equal(
604 childAfter.subagents.find((entry) => entry.id === subagentId)?.status,
605 nativeStatus,
606 );
607 assert.equal(
608 childAfter.nodes.find((entry) => entry.id === subagentId)?.status,
609 nativeStatus,
610 );
611 assert.equal(
612 childAfter.turnItems.find(
613 (item) => item.type === "subagent" && item.subagentId === subagentId,
614 )?.status,
615 nativeStatus,
616 );
617 if (remote) {
618 const descendant = childAfter.subagents.find((entry) => entry.remoteChild !== undefined);
619 assert.equal(descendant?.status, "interrupted");
620 assert.notEqual(descendant?.remoteChild?.stoppedThere, true);
621 }
622 // Stop reached the provider turn that owns the shell.
623 if (background === "command") assert.deepEqual(interrupted, [["bg-sleep"]]);
624 assert.equal(childAfter.runs.length, cancelAt === "monitor" || cancelAt === "turn" ? 2 : 1);
625 assert.equal(childAfter.runs[0]!.status, "completed");
626 assert.deepEqual(
627 childAfter.runtimeRequests.map((request) => [request.id, request.status]),
628 [
629 ["request:owned", "resolved"],
630 ["request:unrelated", "pending"],
631 ],
632 );
633
634 const after = yield* service.taskStatus(scope, task.id);
635 assert.equal(after.status, "interrupted");
636 assert.equal(after.workState, "result_available");
637 const settledTask = yield* parentTask;
638 assert.equal(settledTask?.status, "interrupted");
639 assert.notEqual(settledTask?.completedAt, null);
640 assert.equal(settledTask?.cancelledRunId, child.turn.runId);
641
642 return { scope, taskId: task.id, settled: settledTask };
643 }).pipe(Effect.provide(layerTest));
644 if (stopped === undefined) return;
645 restarted = true;
646
647 // A server restarted on the same database runs startup recovery and keeps the outcome.
648 yield* Effect.gen(function* () {
649 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
650 const orchestrator = yield* Orchestrator.OrchestratorV2;
651 yield* (yield* EffectWorker.OrchestrationEffectWorkerV2).drain();
652 yield* orchestrator.recoverDelegatedTasks;
653 const task = (yield* orchestrator.getThreadProjection(
654 stopped.scope.thread!.threadId,
655 )).subagents.find((entry) => entry.id === stopped.taskId);
656 const status = yield* service.taskStatus(stopped.scope, stopped.taskId);
657 assert.equal(status.status, "interrupted");
658 assert.equal(status.summary, task?.result);
659 assert.equal(task?.status, "interrupted");
660 assert.notEqual(task?.completedAt, null);
661 if (stopped.settled !== undefined) assert.deepEqual(task, stopped.settled);
662 }).pipe(Effect.provide(layerTest));
663 }),
664 ).pipe(Effect.provide(NodeServices.layer));
665
666it.effect("task_cancel ends a delegated child's background native subagent", () =>
667 cancelBackgroundOnlyTask("subagent"),
668);
669
670it.effect("task_cancel ends a delegated child's background shell command", () =>
671 cancelBackgroundOnlyTask("command"),
672);
673
674it.effect("task_cancel ends background work while the child's run captures its checkpoint", () =>
675 cancelBackgroundOnlyTask("command", { cancelAt: "checkpoint" }),
676);
677
678it.effect("task_cancel ends background work left by a turn that finishes after Stop", () =>
679 cancelBackgroundOnlyTask("command", { cancelAt: "turn" }),
680);
681
682it.effect("task_cancel ends the shell's question though a monitor ran after the turn", () =>
683 cancelBackgroundOnlyTask("command", { cancelAt: "monitor" }),
684);
685
686it.effect("cancellation and Stop commit together and recover before the outbox ran", () =>
687 cancelBackgroundOnlyTask("subagent", { cancelAt: "recorded" }),
688);
689
690it.effect("a refused descendant stop does not settle or dispose its parent task", () =>
691 cancelBackgroundOnlyTask("command", { cancelAt: "refused" }),
692);
693
694it.effect("task_cancel stops a remote descendant without reacquiring its parent lock", () =>
695 cancelBackgroundOnlyTask("command", { remote: true }),
696);