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 CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts";
26import * as EffectWorker from "../orchestration-v2/EffectWorker.ts";
27import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
28import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
29import * as ProviderAdapterRegistry from "../orchestration-v2/ProviderAdapterRegistry.ts";
30import * as ThreadLaunchService from "../orchestration-v2/ThreadLaunchService.ts";
31import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
32import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts";
33import * as ServerSettings from "../serverSettings.ts";
34import * as ProviderReplayHarness from "../orchestration-v2/testkit/ProviderReplayHarness.ts";
35import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
36import * as SqlitePersistence from "../persistence/Sqlite.ts";
37import * as ProjectService from "../project/ProjectService.ts";
38import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
39import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
40import * as SecretRequests from "../secrets/SecretRequests.ts";
41import type { McpInvocationScope } from "./McpInvocationContext.ts";
42import * as OrchestratorMcpService from "./OrchestratorMcpService.ts";
43
44const layerOrchestratorService = OrchestratorMcpService.layer.pipe(
45 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
46 Layer.provide(Layer.mock(GitVcsDriver.GitVcsDriver)({})),
47);
48
49const driver = ProviderDriverKind.make("claudeAgent");
50const instanceId = ProviderInstanceId.make("claudeAgent");
51const modelSelection = { instanceId, model: "claude-test" };
52
53// A delegated Claude child ends its turn while a background native subagent it
54// started keeps running (#17154). Cancelling the task stops the child, which
55// ends that subagent's process, so the task must end instead of waiting on it.
56it.effect("task_cancel ends a delegated child's background native subagent", () =>
57 Effect.scoped(
58 Effect.gen(function* () {
59 const cwd = yield* checkpointWorkspace("task-cancel-native-subagent");
60 const turns: Array<{
61 readonly turn: ProviderAdapter.ProviderAdapterV2TurnInput;
62 readonly events: Queue.Queue<ProviderAdapter.ProviderAdapterV2Event>;
63 }> = [];
64 const adapter: ProviderAdapter.ProviderAdapterV2["Service"] = {
65 instanceId,
66 driver,
67 getCapabilities: () => Effect.succeed(ClaudeProviderCapabilitiesV2),
68 planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
69 openSession: (input) =>
70 Effect.gen(function* () {
71 const now = yield* DateTime.now;
72 const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event>();
73 return {
74 instanceId,
75 driver,
76 providerSessionId: input.providerSessionId,
77 providerSession: {
78 id: input.providerSessionId,
79 driver,
80 providerInstanceId: instanceId,
81 status: "ready",
82 cwd,
83 model: modelSelection.model,
84 capabilities: ClaudeProviderCapabilitiesV2,
85 createdAt: now,
86 updatedAt: now,
87 lastError: null,
88 },
89 events: Stream.fromQueue(events),
90 ensureThread: ({ threadId }) =>
91 Effect.succeed({
92 id: ProviderThreadId.make(`provider-thread:${threadId}`),
93 driver,
94 providerInstanceId: instanceId,
95 providerSessionId: input.providerSessionId,
96 appThreadId: threadId,
97 ownerNodeId: null,
98 nativeThreadRef: { driver, nativeId: `native:${threadId}`, strength: "strong" },
99 nativeConversationHeadRef: null,
100 status: "idle",
101 firstRunOrdinal: null,
102 lastRunOrdinal: null,
103 handoffIds: [],
104 forkedFrom: null,
105 createdAt: now,
106 updatedAt: now,
107 }),
108 resumeThread: ({ providerThread }) => Effect.succeed(providerThread),
109 startTurn: (turn) =>
110 Effect.gen(function* () {
111 turns.push({ turn, events });
112 yield* Queue.offer(events, {
113 type: "provider_turn.updated",
114 driver,
115 providerTurn: {
116 id: ProviderTurnId.make(`provider-turn:${turn.attemptId}`),
117 providerThreadId: turn.providerThread.id,
118 nodeId: turn.rootNodeId,
119 runAttemptId: turn.attemptId,
120 nativeTurnRef: null,
121 ordinal: turn.providerTurnOrdinal,
122 status: "running",
123 startedAt: now,
124 completedAt: null,
125 },
126 });
127 }),
128 steerTurn: () => Effect.die("unused"),
129 interruptTurn: () => Effect.void,
130 respondToRuntimeRequest: () => Effect.die("unused"),
131 readThreadSnapshot: () => Effect.die("unused"),
132 rollbackThread: () => Effect.die("unused"),
133 forkThread: () => Effect.die("unused"),
134 };
135 }),
136 };
137 const layerAdapters = ProviderAdapterRegistry.layerSingle(adapter);
138 const layerOrchestrator = ProviderReplayHarness.layerWithRegistry(
139 { name: "task-cancel-native-subagent" },
140 layerAdapters,
141 { runEffectWorker: false },
142 );
143 const layerOrchestration = Layer.merge(
144 layerOrchestrator,
145 ThreadManagementService.layer.pipe(
146 Layer.provide(layerOrchestrator),
147 Layer.provide(ServerSettings.layerTest()),
148 Layer.provide(
149 CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
150 ),
151 ),
152 );
153 const layerTest = layerOrchestratorService.pipe(
154 Layer.provideMerge(layerOrchestration),
155 Layer.provide(
156 Layer.mergeAll(
157 layerAdapters,
158 Layer.mock(ProviderRegistry.ProviderRegistry)({ getProviders: Effect.succeed([]) }),
159 Layer.mock(ProjectService.ProjectService)({}),
160 Layer.mock(SecretRequests.SecretRequests)({}),
161 ServerSettings.layerTest(),
162 Layer.mock(ScheduledTaskService.ScheduledTaskService)({}),
163 NodeServices.layer,
164 ),
165 ),
166 );
167
168 yield* Effect.gen(function* () {
169 const orchestrator = yield* Orchestrator.OrchestratorV2;
170 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
171 const service = yield* OrchestratorMcpService.OrchestratorMcpService;
172 const parentThreadId = ThreadId.make("thread:task-cancel-parent");
173 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
174 orchestrator.streamDomainEvents.pipe(
175 Stream.filter(predicate),
176 Stream.take(1),
177 Stream.runDrain,
178 Effect.forkScoped,
179 );
180 const turnRunning = () =>
181 watch(
182 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
183 );
184 yield* orchestrator.dispatch({
185 type: "thread.create",
186 commandId: CommandId.make("create-parent"),
187 threadId: parentThreadId,
188 projectId: ProjectId.make("project:task-cancel"),
189 title: "Parent",
190 modelSelection,
191 runtimeMode: "full-access",
192 interactionMode: "default",
193 branch: null,
194 worktreePath: cwd,
195 createdBy: "user",
196 creationSource: "web",
197 });
198 const parentRunning = yield* turnRunning();
199 yield* orchestrator.dispatch({
200 type: "message.dispatch",
201 commandId: CommandId.make("start-parent"),
202 threadId: parentThreadId,
203 messageId: MessageId.make("message:start-parent"),
204 text: "Delegate a review",
205 attachments: [],
206 dispatchMode: { type: "start_immediately" },
207 createdBy: "user",
208 creationSource: "web",
209 });
210 yield* worker.drain();
211 yield* Fiber.join(parentRunning);
212
213 const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs[0]!;
214 const childRunning = yield* turnRunning();
215 yield* orchestrator.dispatch({
216 type: "delegated_task.request",
217 commandId: CommandId.make("delegate"),
218 parentThreadId,
219 parentRunId: parentRun.id,
220 parentNodeId: parentRun.rootNodeId!,
221 task: "Review the change",
222 modelSelection,
223 runtimeMode: "full-access",
224 interactionMode: "default",
225 completionWake: "always",
226 createdBy: "agent",
227 creationSource: "mcp",
228 });
229 yield* worker.drain();
230 yield* Fiber.join(childRunning);
231 const task = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents[0]!;
232 const child = turns.find((entry) => entry.turn.threadId === task.childThreadId)!;
233 const childTurn = (yield* orchestrator.getThreadProjection(task.childThreadId!))
234 .providerTurns[0]!;
235
236 // The child starts a background native subagent, then ends its turn.
237 const now = yield* DateTime.now;
238 const subagentId = NodeId.make("node:native-subagent");
239 const native = {
240 threadId: task.childThreadId!,
241 runId: child.turn.runId,
242 origin: "provider_native",
243 driver,
244 providerInstanceId: instanceId,
245 providerThreadId: childTurn.providerThreadId,
246 childThreadId: null,
247 prompt: "Check the tests",
248 title: null,
249 status: "running",
250 result: null,
251 startedAt: now,
252 completedAt: null,
253 updatedAt: now,
254 } as const;
255 const waiting = yield* watch(
256 (event) =>
257 event.type === "run.updated" &&
258 event.payload.id === child.turn.runId &&
259 event.payload.status === "waiting",
260 );
261 for (const event of [
262 {
263 type: "node.updated",
264 driver,
265 node: {
266 id: subagentId,
267 threadId: native.threadId,
268 runId: native.runId,
269 parentNodeId: child.turn.rootNodeId,
270 rootNodeId: child.turn.rootNodeId,
271 kind: "subagent",
272 status: "running",
273 countsForRun: false,
274 providerThreadId: childTurn.providerThreadId,
275 providerTurnId: childTurn.id,
276 nativeItemRef: null,
277 runtimeRequestId: null,
278 checkpointScopeId: null,
279 startedAt: now,
280 completedAt: null,
281 },
282 },
283 {
284 type: "subagent.updated",
285 driver,
286 subagent: {
287 ...native,
288 id: subagentId,
289 parentNodeId: child.turn.rootNodeId,
290 createdBy: "agent",
291 nativeTaskRef: null,
292 model: null,
293 },
294 },
295 {
296 type: "turn_item.updated",
297 driver,
298 turnItem: {
299 ...native,
300 id: TurnItemId.make("turn-item:native-subagent"),
301 nodeId: subagentId,
302 providerTurnId: childTurn.id,
303 nativeItemRef: null,
304 parentItemId: null,
305 ordinal: 1,
306 type: "subagent",
307 subagentId,
308 },
309 },
310 {
311 type: "provider_turn.updated",
312 driver,
313 providerTurn: { ...childTurn, status: "completed", completedAt: now },
314 },
315 {
316 type: "turn.terminal",
317 driver,
318 providerThreadId: childTurn.providerThreadId,
319 providerTurnId: childTurn.id,
320 runOrdinal: child.turn.runOrdinal,
321 status: "completed",
322 failure: null,
323 threadDisposition: "reusable",
324 },
325 ] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>) {
326 yield* Queue.offer(child.events, event);
327 }
328 yield* Fiber.join(waiting);
329 yield* worker.drain();
330
331 const scope: McpInvocationScope = {
332 environmentId: EnvironmentId.make("environment:task-cancel"),
333 requestNamespace: "provider-session:task-cancel",
334 thread: {
335 threadId: parentThreadId,
336 providerSessionId: "provider-session:task-cancel",
337 providerInstanceId: instanceId,
338 },
339 client: undefined,
340 capabilities: new Set(["orchestration"]),
341 issuedAt: 1,
342 };
343 const before = yield* service.taskStatus(scope, task.id);
344 assert.equal(before.status, "running");
345 assert.equal(before.workState, "waiting_for_children");
346
347 const cancelled = yield* service.cancelTask(scope, {
348 taskId: task.id,
349 clientRequestId: "cancel-task",
350 });
351 assert.equal(cancelled.status, "cancel_requested");
352 yield* worker.drain();
353
354 const childAfter = yield* orchestrator.getThreadProjection(task.childThreadId!);
355 assert.equal(
356 childAfter.subagents.find((entry) => entry.id === subagentId)?.status,
357 "interrupted",
358 );
359 assert.equal(
360 childAfter.nodes.find((entry) => entry.id === subagentId)?.status,
361 "interrupted",
362 );
363 assert.equal(
364 childAfter.turnItems.find((item) => item.type === "subagent")?.status,
365 "interrupted",
366 );
367 // The child's own turn had completed, so the task keeps that result.
368 const after = yield* service.taskStatus(scope, task.id);
369 assert.equal(after.status, "completed");
370 assert.notEqual(after.workState, "waiting_for_children");
371 }).pipe(Effect.provide(layerTest));
372 }),
373 ),
374);