1import type { SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
2import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
3import * as NodeServices from "@effect/platform-node/NodeServices";
4import { assert, it } from "@effect/vitest";
5import {
6 ClaudeSettings,
7 CommandId,
8 EventId,
9 MessageId,
10 NodeId,
11 ProjectId,
12 ProviderInstanceId,
13 ScheduledTaskUpsertInput,
14 ThreadId,
15 type OrchestrationV2DomainEvent,
16} from "@t3tools/contracts";
17import * as DateTime from "effect/DateTime";
18import * as Deferred from "effect/Deferred";
19import * as SecureRandom from "effect/crypto/SecureRandom";
20import * as Effect from "effect/Effect";
21import * as FileSystem from "effect/FileSystem";
22import * as Fiber from "effect/Fiber";
23import * as Layer from "effect/Layer";
24import * as Path from "effect/Path";
25import * as Queue from "effect/Queue";
26import * as Schema from "effect/Schema";
27import * as Stream from "effect/Stream";
28import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
29import * as ManagedPolicy from "../policy/ManagedPolicy.ts";
30import * as SecretRequests from "../secrets/SecretRequests.ts";
31import * as ManagedProjectFolders from "../project/ManagedProjectFolders.ts";
32import * as Scheduler from "../scheduling/Scheduler.ts";
33import * as SqlitePersistence from "../persistence/Sqlite.ts";
34import * as ClaudeAdapterV2 from "./Adapters/ClaudeAdapterV2.ts";
35import * as CommandReceiptStore from "./CommandReceiptStore.ts";
36import * as EffectWorker from "./EffectWorker.ts";
37import * as EventSink from "./EventSink.ts";
38import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
39import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
40import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
41import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
42import * as Orchestrator from "./Orchestrator.ts";
43import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
44import * as ThreadLaunchService from "./ThreadLaunchService.ts";
45import * as ThreadManagementService from "./ThreadManagementService.ts";
46import * as ServerSettings from "../serverSettings.ts";
47import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
48import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
49
50const sessionId = "automatic-delivery-session";
51const settings = Schema.decodeSync(ClaudeSettings)({});
52const decodeScheduledTask = Schema.decodeEffect(ScheduledTaskUpsertInput);
53const refusal =
54 "The user doesn't want to take this action right now. STOP what you are doing and wait for the user to tell you how to proceed.";
55const statusIds = ["status-1", "status-2", "status-3", "status-4"];
56const modelSelection = {
57 instanceId: ProviderInstanceId.make("claudeAgent"),
58 model: "claude-sonnet-4-6",
59};
60
61function frame(value: unknown): SDKMessage {
62 return value as SDKMessage;
63}
64
65const batch = frame({
66 type: "assistant",
67 uuid: "batch",
68 session_id: sessionId,
69 parent_tool_use_id: null,
70 message: {
71 id: "tool-batch",
72 role: "assistant",
73 type: "message",
74 model: modelSelection.model,
75 content: [
76 { type: "tool_use", id: "build", name: "Bash", input: { command: "make" } },
77 ...statusIds.map((id) => ({
78 type: "tool_use",
79 id,
80 name: "mcp__t3-code__task_status",
81 input: { taskId: id },
82 })),
83 ],
84 stop_reason: "tool_use",
85 stop_sequence: null,
86 usage: { input_tokens: 1, output_tokens: 1 },
87 },
88});
89
90const toolResult = (id: string, cancelled: boolean) =>
91 frame({
92 type: "user",
93 uuid: `result-${id}`,
94 session_id: sessionId,
95 parent_tool_use_id: null,
96 message: {
97 role: "user",
98 content: [
99 {
100 type: "tool_result",
101 tool_use_id: id,
102 is_error: cancelled,
103 content: cancelled ? refusal : id === "build" ? "build OK" : "Task completed normally",
104 },
105 ],
106 },
107 ...(cancelled ? { tool_result_meta: [{ id, non_execution_kind: "cancelled" }] } : {}),
108 });
109
110const result = (uuid: string, aborted: boolean) =>
111 frame({
112 type: "result",
113 subtype: "success",
114 uuid,
115 session_id: sessionId,
116 is_error: false,
117 num_turns: 1,
118 result: "",
119 stop_reason: "end_turn",
120 terminal_reason: aborted ? "aborted_tools" : "completed",
121 permission_denials: [],
122 duration_ms: 1,
123 duration_api_ms: 1,
124 total_cost_usd: 0,
125 usage: { input_tokens: 1, output_tokens: 1 },
126 modelUsage: {},
127 });
128
129const deliveries = [
130 "child completion",
131 "scheduled message",
132 "user steering",
133 "promoted queued message",
134] as const;
135
136it.effect.each(deliveries)(
137 "delivers %s with Claude's native pending-tool cancellation behavior",
138 (delivery) =>
139 Effect.scoped(
140 Effect.gen(function* () {
141 const cwd = yield* checkpointWorkspace("claude-automatic-delivery");
142 const sdkMessages = yield* Queue.unbounded<SDKMessage>();
143 const offers: SDKUserMessage[] = [];
144 const nativeQueue: SDKUserMessage[] = [];
145 // @effect-diagnostics-next-line abortControllerInEffect:off - The fake SDK interrupts its batch independently of the test fiber.
146 const batchAbort = new AbortController();
147 const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({
148 instanceId: modelSelection.instanceId,
149 settings,
150 environment: {},
151 attachmentsDir: cwd,
152 fileSystem: yield* FileSystem.FileSystem,
153 path: yield* Path.Path,
154 random: yield* SecureRandom.SecureRandom,
155 idAllocator: yield* IdAllocator.IdAllocatorV2,
156 queryRunner: {
157 allocateSessionId: Effect.succeed(sessionId),
158 open: () =>
159 Effect.succeed({
160 messages: Stream.fromQueue(sdkMessages),
161 offer: (message) =>
162 Effect.sync(() => {
163 offers.push(message);
164 nativeQueue.push(message);
165 // Claude Code 2.1.289 aborts on now-priority commands even while Bash blocks reads.
166 if (nativeQueue.some((command) => command.priority === "now")) {
167 batchAbort.abort({ kind: "interrupt" });
168 }
169 }),
170 setModel: () => Effect.void,
171 applyFlagSettings: () => Effect.void,
172 setPermissionMode: () => Effect.void,
173 backgroundTasks: () => Effect.succeed(false),
174 mcpServerStatus: Effect.succeed([]),
175 readMcpResource: () => Effect.die("unexpected MCP resource read"),
176 callMcpTool: () => Effect.die("unexpected MCP tool call"),
177 listPermissionRules: Effect.die("unexpected permission rules read"),
178 getContextUsage: Effect.fail(
179 new ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError({
180 method: "getContextUsage",
181 cause: new Error("Unsupported control request subtype"),
182 }),
183 ),
184 stopTask: () => Effect.void,
185 interrupt: Effect.die("automatic delivery must never interrupt"),
186 close: Effect.void,
187 }),
188 forkSession: () => Effect.die("unused"),
189 subagentLaunchToolUseId: () => Effect.succeed(null),
190 subagentModel: () => Effect.succeed(null),
191 assertComplete: Effect.void,
192 },
193 });
194 yield* Effect.gen(function* () {
195 const orchestrator = yield* Orchestrator.OrchestratorV2;
196 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
197 const threadId = ThreadId.make("thread:automatic-delivery");
198 const projectId = ProjectId.make("project:automatic-delivery");
199 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
200 orchestrator.streamDomainEvents.pipe(
201 Stream.filter(predicate),
202 Stream.take(1),
203 Stream.runDrain,
204 Effect.forkScoped,
205 );
206 yield* orchestrator.dispatch({
207 type: "thread.create",
208 commandId: CommandId.make("create"),
209 threadId,
210 projectId,
211 title: "Automatic delivery",
212 modelSelection,
213 runtimeMode: "full-access",
214 interactionMode: "default",
215 branch: null,
216 worktreePath: cwd,
217 createdBy: "user",
218 creationSource: "web",
219 });
220 const running = yield* watch(
221 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
222 );
223 yield* orchestrator.dispatch({
224 type: "message.dispatch",
225 commandId: CommandId.make("first"),
226 threadId,
227 messageId: MessageId.make("first"),
228 text: "Build and check all four tasks.",
229 attachments: [],
230 dispatchMode: { type: "start_immediately" },
231 createdBy: "user",
232 creationSource: "web",
233 });
234 yield* worker.drain();
235 yield* Fiber.join(running);
236 const pending = yield* watch(
237 (event) =>
238 event.type === "turn-item.updated" &&
239 event.payload.nativeItemRef?.nativeId === "status-4",
240 );
241 yield* Queue.offer(sdkMessages, batch);
242 yield* Fiber.join(pending);
243 const before = yield* orchestrator.getThreadProjection(threadId);
244 assert.equal(before.turnItems.filter((item) => item.status === "running").length, 5);
245 const parent = before.runs[0]!;
246 if (parent.rootNodeId === null) return yield* Effect.die("parent has no root node");
247 const messageId = MessageId.make("automatic-notice");
248 if (delivery === "child completion") {
249 const taskId = NodeId.make("completed-child");
250 const now = yield* DateTime.now;
251 const sink = yield* EventSink.EventSinkV2;
252 yield* sink.write({
253 events: [
254 {
255 id: EventId.make("cohort"),
256 type: "run.updated",
257 threadId,
258 runId: parent.id,
259 occurredAt: now,
260 payload: {
261 ...parent,
262 delegatedCompletion: {
263 disposition: "open",
264 nextGeneration: 2,
265 delivery: { generation: 1, messageId, taskIds: [taskId] },
266 },
267 },
268 },
269 {
270 id: EventId.make("child"),
271 type: "subagent.updated",
272 threadId,
273 runId: parent.id,
274 nodeId: taskId,
275 occurredAt: now,
276 payload: {
277 id: taskId,
278 threadId,
279 runId: parent.id,
280 parentNodeId: parent.rootNodeId,
281 origin: "app_owned",
282 createdBy: "agent",
283 driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
284 providerInstanceId: modelSelection.instanceId,
285 providerThreadId: null,
286 childThreadId: null,
287 nativeTaskRef: null,
288 prompt: "Background work",
289 title: "Child",
290 model: null,
291 completionWake: "always",
292 completionDelivery: { state: "claimed", observedByRunId: null },
293 status: "completed",
294 result: "done",
295 startedAt: now,
296 completedAt: now,
297 updatedAt: now,
298 },
299 },
300 ],
301 });
302 const command = {
303 type: "message.dispatch" as const,
304 commandId: CommandId.make("completion"),
305 threadId,
306 messageId,
307 text: "Child completed",
308 attachments: [],
309 dispatchMode: { type: "queue_after_active" as const },
310 createdBy: "agent" as const,
311 creationSource: "server" as const,
312 delegatedCompletion: { parentRunId: parent.id, generation: 1, taskIds: [taskId] },
313 };
314 yield* orchestrator.dispatch(command);
315 yield* orchestrator.dispatch(command);
316 yield* orchestrator.recoverDelegatedTasks;
317 } else if (delivery === "scheduled message") {
318 yield* Effect.gen(function* () {
319 const service = yield* ScheduledTaskService.ScheduledTaskService;
320 const { task } = yield* service.upsert(
321 yield* decodeScheduledTask({
322 title: "Status check",
323 prompt: "Check the four tasks.",
324 enabled: true,
325 schedule: { type: "interval", everyMs: 60_000 },
326 projectId,
327 threadId,
328 workspaceStrategy: { type: "root" },
329 modelSelection,
330 runtimeMode: "full-access",
331 interactionMode: "default",
332 }),
333 );
334 const ran = yield* service.runNow({ id: task.id });
335 assert.equal(ran.task.lastRunStatus, "succeeded");
336 }).pipe(
337 Effect.provide(
338 ScheduledTaskService.layer.pipe(
339 Layer.provide(ThreadManagementService.layer),
340 Layer.provide(ServerSettings.layerTest()),
341 Layer.provide(
342 Layer.mock(LegacyV1ThreadImporter.LegacyV1ThreadImporter)({
343 ensureTranscript: () =>
344 Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }),
345 }),
346 ),
347 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
348 Layer.provide(ManagedPolicy.layerTest()),
349 Layer.provide(
350 Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({
351 namedProjectsRoot: "/projects",
352 isScratchProject: () => Effect.succeed(false),
353 }),
354 ),
355 Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
356 Layer.provide(
357 CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
358 ),
359 Layer.provide(
360 Layer.mergeAll(
361 NodeCrypto.layer,
362 Scheduler.layer,
363 SqlitePersistence.layerMemory,
364 ),
365 ),
366 ),
367 ),
368 );
369 } else if (delivery === "user steering") {
370 yield* orchestrator.dispatch({
371 type: "message.dispatch",
372 commandId: CommandId.make("steer"),
373 threadId,
374 messageId,
375 text: "Change direction now.",
376 attachments: [],
377 dispatchMode: { type: "steer_active", targetRunId: parent.id },
378 createdBy: "user",
379 creationSource: "web",
380 });
381 } else {
382 yield* orchestrator.dispatch({
383 type: "message.dispatch",
384 commandId: CommandId.make("queue"),
385 threadId,
386 messageId,
387 text: "Change direction now.",
388 attachments: [],
389 dispatchMode: { type: "queue_after_active" },
390 createdBy: "user",
391 creationSource: "web",
392 });
393 const queuedRun = (yield* orchestrator.getThreadProjection(threadId)).runs[1]!;
394 yield* orchestrator.dispatch({
395 type: "queued-message.promote-to-steer",
396 commandId: CommandId.make("promote"),
397 threadId,
398 queuedRunId: queuedRun.id,
399 targetRunId: parent.id,
400 });
401 const promoted = yield* orchestrator.getThreadProjection(threadId);
402 assert.equal(promoted.runs.find((run) => run.id === queuedRun.id)?.status, "cancelled");
403 assert.equal(
404 promoted.messages.find((message) => message.id === messageId)?.runId,
405 parent.id,
406 );
407 const steer = promoted.turnItems.find(
408 (item) => item.type === "user_message" && item.messageId === messageId,
409 );
410 assert.ok(steer?.type === "user_message");
411 assert.equal(steer.inputIntent, "promoted_queued_to_steer");
412 }
413 yield* worker.drain();
414 const explicitSteer =
415 delivery === "user steering" || delivery === "promoted queued message";
416 assert.equal(batchAbort.signal.aborted, explicitSteer);
417 assert.equal(offers.length, explicitSteer ? 2 : 1);
418 const finished = yield* watch(
419 (event) =>
420 event.type === "run.updated" &&
421 event.payload.id === parent.id &&
422 event.payload.status === "waiting",
423 );
424 yield* Queue.offer(sdkMessages, toolResult("build", false));
425 for (const id of statusIds)
426 yield* Queue.offer(sdkMessages, toolResult(id, batchAbort.signal.aborted));
427 if (explicitSteer) yield* Queue.offer(sdkMessages, result("aborted", true));
428 yield* Queue.offer(sdkMessages, result("completed", false));
429 yield* Fiber.join(finished);
430 yield* worker.drain();
431 yield* orchestrator.resumeQueuedRuns;
432 yield* worker.drain();
433 if (!explicitSteer) {
434 const queued = (yield* orchestrator.getThreadProjection(threadId)).runs[1]!;
435 const noticeFinished = yield* watch(
436 (event) =>
437 event.type === "run.updated" &&
438 event.payload.id === queued.id &&
439 event.payload.status === "waiting",
440 );
441 const delivered =
442 delivery === "child completion"
443 ? yield* watch(
444 (event) =>
445 event.type === "subagent.updated" &&
446 event.payload.completionDelivery?.state === "delivered",
447 )
448 : null;
449 yield* Queue.offer(sdkMessages, result("notice-completed", false));
450 yield* Fiber.join(noticeFinished);
451 yield* worker.drain();
452 if (delivered !== null) yield* Fiber.join(delivered);
453 }
454 const after = yield* orchestrator.getThreadProjection(threadId);
455 const reads = after.turnItems.filter((item) =>
456 statusIds.includes(item.nativeItemRef?.nativeId ?? ""),
457 );
458 assert.equal(reads.length, 4);
459 for (const read of reads) {
460 assert.equal(read.status, explicitSteer ? "cancelled" : "completed");
461 assert.equal(read.toolNonExecutionKind, explicitSteer ? "cancelled" : undefined);
462 assert.equal(read.type === "dynamic_tool" && read.output === refusal, explicitSteer);
463 }
464 assert.equal(offers.length, 2);
465 assert.equal(
466 offers.filter((offer) => offer.priority === "now").length,
467 explicitSteer ? 1 : 0,
468 );
469 if (!explicitSteer) {
470 assert.equal(after.runs.length, 2);
471 if (delivery === "child completion") {
472 assert.equal(after.messages.filter((message) => message.id === messageId).length, 1);
473 assert.equal(after.subagents[0]?.completionDelivery?.state, "delivered");
474 } else {
475 assert.equal(
476 after.messages.filter((message) => message.scheduledTaskId !== undefined).length,
477 1,
478 );
479 }
480 }
481 yield* orchestrator.recoverDelegatedTasks;
482 yield* orchestrator.resumeQueuedRuns;
483 yield* worker.drain();
484 assert.equal(offers.length, 2);
485 }).pipe(
486 Effect.provide(
487 ProviderReplayHarness.layerWithRegistry(
488 { name: "claude-automatic-delivery" },
489 ProviderAdapterRegistry.layerSingle(adapter),
490 { runEffectWorker: false },
491 ),
492 ),
493 );
494 }).pipe(
495 Effect.provide(
496 Layer.mergeAll(
497 IdAllocator.layer,
498 McpProviderSessions.layer,
499 NodeServices.layer,
500 NodeCrypto.layer,
501 ),
502 ),
503 ),
504 ),
505);
506
507it.effect("applies an effort change to the running Claude query without waiting for a turn", () =>
508 Effect.scoped(
509 Effect.gen(function* () {
510 const cwd = yield* checkpointWorkspace("claude-live-effort");
511 const sdkMessages = yield* Queue.unbounded<SDKMessage>();
512 const flagSettings: unknown[] = [];
513 const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({
514 instanceId: modelSelection.instanceId,
515 settings,
516 environment: {},
517 attachmentsDir: cwd,
518 fileSystem: yield* FileSystem.FileSystem,
519 path: yield* Path.Path,
520 random: yield* SecureRandom.SecureRandom,
521 idAllocator: yield* IdAllocator.IdAllocatorV2,
522 queryRunner: {
523 allocateSessionId: Effect.succeed(sessionId),
524 open: () =>
525 Effect.succeed({
526 messages: Stream.fromQueue(sdkMessages),
527 offer: () => Effect.void,
528 setModel: () => Effect.void,
529 applyFlagSettings: (flags) =>
530 Effect.sync(() => {
531 flagSettings.push(flags);
532 }),
533 setPermissionMode: () => Effect.void,
534 backgroundTasks: () => Effect.succeed(false),
535 mcpServerStatus: Effect.succeed([]),
536 readMcpResource: () => Effect.die("unexpected MCP resource read"),
537 callMcpTool: () => Effect.die("unexpected MCP tool call"),
538 listPermissionRules: Effect.die("unexpected permission rules read"),
539 getContextUsage: Effect.fail(
540 new ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError({
541 method: "getContextUsage",
542 cause: new Error("Unsupported control request subtype"),
543 }),
544 ),
545 stopTask: () => Effect.void,
546 interrupt: Effect.die("an effort change must not interrupt the turn"),
547 close: Effect.void,
548 }),
549 forkSession: () => Effect.die("unused"),
550 subagentLaunchToolUseId: () => Effect.succeed(null),
551 subagentModel: () => Effect.succeed(null),
552 assertComplete: Effect.void,
553 },
554 });
555 yield* Effect.gen(function* () {
556 const orchestrator = yield* Orchestrator.OrchestratorV2;
557 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
558 const threadId = ThreadId.make("thread:live-effort");
559 const high = { ...modelSelection, options: [{ id: "effort", value: "high" }] };
560 const running = yield* orchestrator.streamDomainEvents.pipe(
561 Stream.filter(
562 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
563 ),
564 Stream.take(1),
565 Stream.runDrain,
566 Effect.forkScoped,
567 );
568 yield* orchestrator.dispatch({
569 type: "thread.create",
570 commandId: CommandId.make("create"),
571 threadId,
572 projectId: ProjectId.make("project:live-effort"),
573 title: "Live effort",
574 modelSelection: high,
575 runtimeMode: "full-access",
576 interactionMode: "default",
577 branch: null,
578 worktreePath: cwd,
579 createdBy: "user",
580 creationSource: "web",
581 });
582 yield* orchestrator.dispatch({
583 type: "message.dispatch",
584 commandId: CommandId.make("first"),
585 threadId,
586 messageId: MessageId.make("first"),
587 text: "Keep working.",
588 attachments: [],
589 dispatchMode: { type: "start_immediately" },
590 createdBy: "user",
591 creationSource: "web",
592 });
593 yield* worker.drain();
594 yield* Fiber.join(running);
595 assert.deepEqual(flagSettings, []);
596
597 yield* orchestrator.dispatch({
598 type: "thread.model-selection.set",
599 commandId: CommandId.make("medium"),
600 threadId,
601 modelSelection: { ...modelSelection, options: [{ id: "effort", value: "medium" }] },
602 });
603 yield* worker.drain();
604
605 assert.deepEqual(flagSettings, [
606 { effortLevel: "medium", fastMode: null, ultracode: null },
607 ]);
608 }).pipe(
609 Effect.provide(
610 ProviderReplayHarness.layerWithRegistry(
611 { name: "claude-live-effort" },
612 ProviderAdapterRegistry.layerSingle(adapter),
613 { runEffectWorker: false },
614 ),
615 ),
616 );
617 }).pipe(
618 Effect.provide(
619 Layer.mergeAll(
620 IdAllocator.layer,
621 McpProviderSessions.layer,
622 NodeServices.layer,
623 NodeCrypto.layer,
624 ),
625 ),
626 ),
627 ),
628);
629
630const backgroundTaskStarted = (taskId: string) =>
631 frame({
632 type: "system",
633 subtype: "task_started",
634 task_id: taskId,
635 tool_use_id: `tool-${taskId}`,
636 description: "Background build",
637 is_backgrounded: true,
638 task_type: "local_bash",
639 uuid: `started-${taskId}`,
640 session_id: sessionId,
641 });
642
643const backgroundTaskNotification = (taskId: string) =>
644 frame({
645 type: "system",
646 subtype: "task_notification",
647 task_id: taskId,
648 tool_use_id: `tool-${taskId}`,
649 status: "completed",
650 output_file: `/tmp/claude-replay/tasks/${taskId}.output`,
651 summary: `Background build ${taskId} completed`,
652 uuid: `notification-${taskId}`,
653 session_id: sessionId,
654 });
655
656const wakeAssistant = (uuid: string) =>
657 frame({
658 type: "assistant",
659 uuid,
660 session_id: sessionId,
661 parent_tool_use_id: null,
662 message: {
663 id: `msg-${uuid}`,
664 role: "assistant",
665 type: "message",
666 model: modelSelection.model,
667 content: [{ type: "text", text: `Background build finished (${uuid}).` }],
668 stop_reason: null,
669 stop_sequence: null,
670 usage: { input_tokens: 1, output_tokens: 1 },
671 },
672 });
673
674it.effect("offers a continuation for a later wake after a queued continuation is cancelled", () =>
675 Effect.scoped(
676 Effect.gen(function* () {
677 const cwd = yield* checkpointWorkspace("claude-cancelled-continuation");
678 const sdkMessages = yield* Queue.unbounded<SDKMessage>();
679 const processed = new WeakMap<SDKMessage, Deferred.Deferred<void>>();
680 const offerAndWait = Effect.fnUntraced(function* (message: SDKMessage) {
681 const done = yield* Deferred.make<void>();
682 processed.set(message, done);
683 yield* Queue.offer(sdkMessages, message);
684 yield* Deferred.await(done);
685 });
686 const fileSystem = yield* FileSystem.FileSystem;
687 const path = yield* Path.Path;
688 const random = yield* SecureRandom.SecureRandom;
689 const idAllocator = yield* IdAllocator.IdAllocatorV2;
690 const continuations: ProviderContinuationRequests.ProviderContinuationRequest[] = [];
691 const services =
692 yield* Effect.context<
693 Effect.Services<ReturnType<typeof ClaudeAdapterV2.makeClaudeAdapterV2>>
694 >();
695 const registry = ProviderAdapterRegistry.layerFromAdaptersEffect(
696 Effect.gen(function* () {
697 const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests;
698 const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({
699 instanceId: modelSelection.instanceId,
700 settings,
701 environment: {},
702 attachmentsDir: cwd,
703 fileSystem,
704 path,
705 random,
706 idAllocator,
707 continuationRequests: {
708 offer: (request) =>
709 Effect.sync(() => continuations.push(request)).pipe(
710 Effect.andThen(requests.offer(request)),
711 ),
712 },
713 queryRunner: {
714 allocateSessionId: Effect.succeed(sessionId),
715 open: () =>
716 Effect.succeed({
717 // Resolve each frame's Deferred once the adapter pulls the next one.
718 messages: Stream.fromQueue(sdkMessages).pipe(
719 Stream.flatMap((message) =>
720 Stream.make(message).pipe(
721 Stream.concat(
722 Stream.fromEffect(
723 Effect.suspend(() => {
724 const done = processed.get(message);
725 return done === undefined
726 ? Effect.void
727 : Deferred.succeed(done, undefined);
728 }),
729 ).pipe(Stream.drain),
730 ),
731 ),
732 ),
733 ),
734 offer: () => Effect.void,
735 setModel: () => Effect.void,
736 applyFlagSettings: () => Effect.void,
737 setPermissionMode: () => Effect.void,
738 backgroundTasks: () => Effect.succeed(false),
739 mcpServerStatus: Effect.succeed([]),
740 readMcpResource: () => Effect.die("unexpected MCP resource read"),
741 callMcpTool: () => Effect.die("unexpected MCP tool call"),
742 listPermissionRules: Effect.die("unexpected permission rules read"),
743 getContextUsage: Effect.fail(
744 new ClaudeAdapterV2.ClaudeAgentSdkQueryRunnerError({
745 method: "getContextUsage",
746 cause: new Error("Unsupported control request subtype"),
747 }),
748 ),
749 stopTask: () => Effect.void,
750 interrupt: Effect.void,
751 close: Effect.void,
752 }),
753 forkSession: () => Effect.die("unused"),
754 subagentLaunchToolUseId: () => Effect.succeed(null),
755 subagentModel: () => Effect.succeed(null),
756 assertComplete: Effect.void,
757 },
758 }).pipe(Effect.provideContext(services));
759 return [adapter];
760 }),
761 );
762 yield* Effect.gen(function* () {
763 const orchestrator = yield* Orchestrator.OrchestratorV2;
764 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
765 const threadId = ThreadId.make("thread:cancelled-continuation");
766 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
767 orchestrator.streamDomainEvents.pipe(
768 Stream.filter(predicate),
769 Stream.take(1),
770 Stream.runDrain,
771 Effect.forkScoped,
772 );
773 const runUpdated = (messageId: string, status: string) =>
774 watch(
775 (event) =>
776 event.type === "run.updated" &&
777 event.payload.userMessageId === messageId &&
778 event.payload.status === status,
779 );
780 // A user turn that leaves a background task running when it ends.
781 const userTurnWithBackgroundTask = Effect.fnUntraced(function* (
782 id: string,
783 taskId: string,
784 ) {
785 const running = yield* runUpdated(id, "running");
786 yield* orchestrator.dispatch({
787 type: "message.dispatch",
788 commandId: CommandId.make(id),
789 threadId,
790 messageId: MessageId.make(id),
791 text: "Run the build in the background.",
792 attachments: [],
793 dispatchMode: { type: "start_immediately" },
794 createdBy: "user",
795 creationSource: "web",
796 });
797 yield* worker.drain();
798 yield* Fiber.join(running);
799 const waiting = yield* runUpdated(id, "waiting");
800 yield* Queue.offer(sdkMessages, backgroundTaskStarted(taskId));
801 yield* Queue.offer(sdkMessages, result(`result-${id}`, false));
802 yield* Fiber.join(waiting);
803 });
804 yield* orchestrator.dispatch({
805 type: "thread.create",
806 commandId: CommandId.make("create"),
807 threadId,
808 projectId: ProjectId.make("project:cancelled-continuation"),
809 title: "Cancelled continuation",
810 modelSelection,
811 runtimeMode: "full-access",
812 interactionMode: "default",
813 branch: null,
814 worktreePath: cwd,
815 createdBy: "user",
816 creationSource: "web",
817 });
818
819 // The first run is still finalizing when its background task wakes Claude.
820 yield* userTurnWithBackgroundTask("first", "build-1");
821 const queued = yield* watch(
822 (event) => event.type === "run.created" && event.payload.status === "queued",
823 );
824 yield* offerAndWait(backgroundTaskNotification("build-1"));
825 yield* offerAndWait(wakeAssistant("wake-1"));
826 yield* Fiber.join(queued);
827 assert.lengthOf(continuations, 1);
828 const continuationRun = (yield* orchestrator.getThreadProjection(threadId)).runs.find(
829 (run) => run.status === "queued",
830 );
831 assert.ok(continuationRun !== undefined);
832 yield* orchestrator.dispatch({
833 type: "queued-run.cancel",
834 commandId: CommandId.make("cancel"),
835 threadId,
836 runId: continuationRun.id,
837 });
838 yield* offerAndWait(result("wake-1-result", false));
839 yield* worker.drain();
840 yield* orchestrator.resumeQueuedRuns;
841 yield* worker.drain();
842 const idle = yield* orchestrator.getThreadProjection(threadId);
843 assert.equal(idle.runs.find((run) => run.id === continuationRun.id)?.status, "cancelled");
844 assert.isFalse(idle.runs.some((run) => run.status === "queued"));
845
846 yield* userTurnWithBackgroundTask("second", "build-2");
847 yield* worker.drain();
848 // The second user run has completed, so this continuation starts without queueing.
849 const isContinuationRun = (
850 event: OrchestrationV2DomainEvent,
851 type: string,
852 status: string,
853 ) =>
854 (event.type === "run.created" || event.type === "run.updated") &&
855 event.type === type &&
856 event.payload.status === status &&
857 event.payload.userMessageId !== MessageId.make("first") &&
858 event.payload.userMessageId !== MessageId.make("second");
859 const secondCreated = yield* watch((event) =>
860 isContinuationRun(event, "run.created", "starting"),
861 );
862 const secondRunning = yield* watch((event) =>
863 isContinuationRun(event, "run.updated", "running"),
864 );
865 yield* offerAndWait(backgroundTaskNotification("build-2"));
866 yield* offerAndWait(wakeAssistant("wake-2"));
867 assert.lengthOf(
868 continuations,
869 2,
870 "the wake after a cancelled queued continuation must offer a new continuation",
871 );
872 yield* Fiber.join(secondCreated);
873 yield* worker.drain();
874 yield* Fiber.join(secondRunning);
875 const secondContinuation = (yield* orchestrator.getThreadProjection(threadId)).runs.find(
876 (run) => run.status === "running",
877 );
878 assert.ok(secondContinuation !== undefined);
879 const secondWaiting = yield* watch(
880 (event) =>
881 event.type === "run.updated" &&
882 event.payload.id === secondContinuation.id &&
883 event.payload.status === "waiting",
884 );
885 yield* offerAndWait(result("wake-2-result", false));
886 yield* Fiber.join(secondWaiting);
887 yield* worker.drain();
888 const settled = yield* orchestrator.getThreadProjection(threadId);
889 assert.equal(
890 settled.runs.find((run) => run.id === secondContinuation.id)?.status,
891 "completed",
892 );
893 assert.deepEqual(
894 settled.turnItems.flatMap((item) =>
895 item.runId === secondContinuation.id && item.type === "assistant_message"
896 ? [item.text]
897 : [],
898 ),
899 ["Background build finished (wake-2)."],
900 "the continuation must carry only its own wake's output",
901 );
902 assert.lengthOf(continuations, 2, "two wakes must offer exactly two continuations");
903 }).pipe(
904 Effect.provide(
905 ProviderReplayHarness.layerWithRegistry(
906 { name: "claude-cancelled-continuation" },
907 registry,
908 { runEffectWorker: false, runContinuationWorker: true },
909 ),
910 ),
911 );
912 }).pipe(
913 Effect.provide(
914 Layer.mergeAll(
915 IdAllocator.layer,
916 McpProviderSessions.layer,
917 NodeServices.layer,
918 NodeCrypto.layer,
919 ),
920 ),
921 ),
922 ),
923);