1import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2import type { SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
3import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
4import * as NodeServices from "@effect/platform-node/NodeServices";
5import { assert, it } from "@effect/vitest";
6import {
7 ClaudeSettings,
8 CommandId,
9 EventId,
10 MessageId,
11 NodeId,
12 ProjectId,
13 ProviderInstanceId,
14 ScheduledTaskUpsertInput,
15 ThreadId,
16 type OrchestrationV2DomainEvent,
17} from "@t3tools/contracts";
18import * as DateTime from "effect/DateTime";
19import * as Crypto from "effect/Crypto";
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 SecretRequests from "../secrets/SecretRequests.ts";
30import * as Scheduler from "../scheduling/Scheduler.ts";
31import * as SqlitePersistence from "../persistence/Sqlite.ts";
32import * as ClaudeAdapterV2 from "./Adapters/ClaudeAdapterV2.ts";
33import * as CommandReceiptStore from "./CommandReceiptStore.ts";
34import * as EffectWorker from "./EffectWorker.ts";
35import * as EventSink from "./EventSink.ts";
36import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
37import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
38import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
39import * as Orchestrator from "./Orchestrator.ts";
40import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
41import * as ThreadLaunchService from "./ThreadLaunchService.ts";
42import * as ThreadManagementService from "./ThreadManagementService.ts";
43import * as ServerSettings from "../serverSettings.ts";
44import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
45import { checkpointWorkspace } from "@t3tools/provider-testing/replayWorkspace";
46
47const sessionId = "automatic-delivery-session";
48const settings = Schema.decodeSync(ClaudeSettings)({});
49const decodeScheduledTask = Schema.decodeEffect(ScheduledTaskUpsertInput);
50const refusal =
51 "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.";
52const statusIds = ["status-1", "status-2", "status-3", "status-4"];
53const modelSelection = {
54 instanceId: ProviderInstanceId.make("claudeAgent"),
55 model: "claude-sonnet-4-6",
56};
57
58// Native frames match the captured Bash/status batch and tool_result_meta shape.
59// The external SDK boundary is replayed; the adapter and durable dispatch are real.
60function frame(value: unknown): SDKMessage {
61 return value as SDKMessage;
62}
63
64const batch = frame({
65 type: "assistant",
66 uuid: "batch",
67 session_id: sessionId,
68 parent_tool_use_id: null,
69 message: {
70 id: "tool-batch",
71 role: "assistant",
72 type: "message",
73 model: modelSelection.model,
74 content: [
75 { type: "tool_use", id: "build", name: "Bash", input: { command: "make" } },
76 ...statusIds.map((id) => ({
77 type: "tool_use",
78 id,
79 name: "mcp__t3-code__task_status",
80 input: { taskId: id },
81 })),
82 ],
83 stop_reason: "tool_use",
84 stop_sequence: null,
85 usage: { input_tokens: 1, output_tokens: 1 },
86 },
87});
88
89const toolResult = (id: string, cancelled: boolean) =>
90 frame({
91 type: "user",
92 uuid: `result-${id}`,
93 session_id: sessionId,
94 parent_tool_use_id: null,
95 message: {
96 role: "user",
97 content: [
98 {
99 type: "tool_result",
100 tool_use_id: id,
101 is_error: cancelled,
102 content: cancelled ? refusal : id === "build" ? "build OK" : "Task completed normally",
103 },
104 ],
105 },
106 ...(cancelled ? { tool_result_meta: [{ id, non_execution_kind: "cancelled" }] } : {}),
107 });
108
109const result = (uuid: string, aborted: boolean) =>
110 frame({
111 type: "result",
112 subtype: "success",
113 uuid,
114 session_id: sessionId,
115 is_error: false,
116 num_turns: 1,
117 result: "",
118 stop_reason: "end_turn",
119 terminal_reason: aborted ? "aborted_tools" : "completed",
120 permission_denials: [],
121 duration_ms: 1,
122 duration_api_ms: 1,
123 total_cost_usd: 0,
124 usage: { input_tokens: 1, output_tokens: 1 },
125 modelUsage: {},
126 });
127
128it.effect.each(["child completion", "scheduled message", "user steering"] as const)(
129 "delivers %s with Claude's native pending-tool cancellation behavior",
130 (delivery) =>
131 Effect.scoped(
132 Effect.gen(function* () {
133 const cwd = yield* checkpointWorkspace("claude-automatic-delivery");
134 const sdkMessages = yield* Queue.unbounded<SDKMessage>();
135 const offers: SDKUserMessage[] = [];
136 const nativeQueue: SDKUserMessage[] = [];
137 const batchAbort = new AbortController();
138 const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({
139 instanceId: modelSelection.instanceId,
140 settings,
141 environment: {},
142 attachmentsDir: cwd,
143 fileSystem: yield* FileSystem.FileSystem,
144 path: yield* Path.Path,
145 crypto: yield* Crypto.Crypto,
146 idAllocator: yield* IdAllocator.IdAllocatorV2,
147 queryRunner: {
148 allocateSessionId: Effect.succeed(sessionId),
149 open: () =>
150 Effect.succeed({
151 messages: Stream.fromQueue(sdkMessages),
152 offer: (message) =>
153 Effect.sync(() => {
154 offers.push(message);
155 nativeQueue.push(message);
156 // Claude Code 2.1.289's queue watcher aborts when a now-priority
157 // command arrives, including while Bash blocks pending reads.
158 if (nativeQueue.some((command) => command.priority === "now")) {
159 batchAbort.abort({ kind: "interrupt" });
160 }
161 }),
162 setModel: () => Effect.void,
163 applyFlagSettings: () => Effect.void,
164 setPermissionMode: () => Effect.void,
165 stopTask: () => Effect.void,
166 interrupt: Effect.die("automatic delivery must never interrupt"),
167 close: Effect.void,
168 }),
169 forkSession: () => Effect.die("unused"),
170 subagentLaunchToolUseId: () => Effect.succeed(null),
171 assertComplete: Effect.void,
172 },
173 });
174 yield* Effect.gen(function* () {
175 const orchestrator = yield* Orchestrator.OrchestratorV2;
176 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
177 const threadId = ThreadId.make("thread:automatic-delivery");
178 const projectId = ProjectId.make("project:automatic-delivery");
179 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
180 orchestrator.streamDomainEvents.pipe(
181 Stream.filter(predicate),
182 Stream.take(1),
183 Stream.runDrain,
184 Effect.forkScoped,
185 );
186 yield* orchestrator.dispatch({
187 type: "thread.create",
188 commandId: CommandId.make("create"),
189 threadId,
190 projectId,
191 title: "Automatic delivery",
192 modelSelection,
193 runtimeMode: "full-access",
194 interactionMode: "default",
195 branch: null,
196 worktreePath: cwd,
197 createdBy: "user",
198 creationSource: "web",
199 });
200 const running = yield* watch(
201 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
202 );
203 yield* orchestrator.dispatch({
204 type: "message.dispatch",
205 commandId: CommandId.make("first"),
206 threadId,
207 messageId: MessageId.make("first"),
208 text: "Build and check all four tasks.",
209 attachments: [],
210 dispatchMode: { type: "start_immediately" },
211 createdBy: "user",
212 creationSource: "web",
213 });
214 yield* worker.drain();
215 yield* Fiber.join(running);
216 const pending = yield* watch(
217 (event) =>
218 event.type === "turn-item.updated" &&
219 event.payload.nativeItemRef?.nativeId === "status-4",
220 );
221 yield* Queue.offer(sdkMessages, batch);
222 yield* Fiber.join(pending);
223 const before = yield* orchestrator.getThreadProjection(threadId);
224 assert.equal(before.turnItems.filter((item) => item.status === "running").length, 5);
225 const parent = before.runs[0]!;
226 if (parent.rootNodeId === null) return yield* Effect.die("parent has no root node");
227 const messageId = MessageId.make("automatic-notice");
228 if (delivery === "child completion") {
229 const taskId = NodeId.make("completed-child");
230 const now = yield* DateTime.now;
231 const sink = yield* EventSink.EventSinkV2;
232 yield* sink.write({
233 events: [
234 {
235 id: EventId.make("cohort"),
236 type: "run.updated",
237 threadId,
238 runId: parent.id,
239 occurredAt: now,
240 payload: {
241 ...parent,
242 delegatedCompletion: {
243 disposition: "open",
244 nextGeneration: 2,
245 delivery: { generation: 1, messageId, taskIds: [taskId] },
246 },
247 },
248 },
249 {
250 id: EventId.make("child"),
251 type: "subagent.updated",
252 threadId,
253 runId: parent.id,
254 nodeId: taskId,
255 occurredAt: now,
256 payload: {
257 id: taskId,
258 threadId,
259 runId: parent.id,
260 parentNodeId: parent.rootNodeId,
261 origin: "app_owned",
262 createdBy: "agent",
263 driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
264 providerInstanceId: modelSelection.instanceId,
265 providerThreadId: null,
266 childThreadId: null,
267 nativeTaskRef: null,
268 prompt: "Background work",
269 title: "Child",
270 model: null,
271 completionWake: "always",
272 completionDelivery: { state: "claimed", observedByRunId: null },
273 status: "completed",
274 result: "done",
275 startedAt: now,
276 completedAt: now,
277 updatedAt: now,
278 },
279 },
280 ],
281 });
282 const command = {
283 type: "message.dispatch" as const,
284 commandId: CommandId.make("completion"),
285 threadId,
286 messageId,
287 text: "Child completed",
288 attachments: [],
289 dispatchMode: { type: "queue_after_active" as const },
290 createdBy: "agent" as const,
291 creationSource: "server" as const,
292 delegatedCompletion: { parentRunId: parent.id, generation: 1, taskIds: [taskId] },
293 };
294 yield* orchestrator.dispatch(command);
295 yield* orchestrator.dispatch(command);
296 yield* orchestrator.recoverDelegatedTasks;
297 } else if (delivery === "scheduled message") {
298 yield* Effect.gen(function* () {
299 const service = yield* ScheduledTaskService.ScheduledTaskService;
300 const { task } = yield* service.upsert(
301 yield* decodeScheduledTask({
302 title: "Status check",
303 prompt: "Check the four tasks.",
304 enabled: true,
305 schedule: { type: "interval", everyMs: 60_000 },
306 projectId,
307 threadId,
308 workspaceStrategy: { type: "root" },
309 modelSelection,
310 runtimeMode: "full-access",
311 interactionMode: "default",
312 }),
313 );
314 const ran = yield* service.runNow({ id: task.id });
315 assert.equal(ran.task.lastRunStatus, "succeeded");
316 }).pipe(
317 Effect.provide(
318 ScheduledTaskService.layer.pipe(
319 Layer.provide(ThreadManagementService.layer),
320 Layer.provide(ServerSettings.layerTest()),
321 Layer.provide(
322 Layer.mock(LegacyV1ThreadImporter.LegacyV1ThreadImporter)({
323 ensureTranscript: () =>
324 Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }),
325 }),
326 ),
327 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
328 Layer.provide(Layer.mock(SecretRequests.SecretRequests)({})),
329 Layer.provide(
330 CommandReceiptStore.layer.pipe(Layer.provide(SqlitePersistence.layerMemory)),
331 ),
332 Layer.provide(
333 Layer.mergeAll(
334 NodeCrypto.layer,
335 Scheduler.layer,
336 SqlitePersistence.layerMemory,
337 ),
338 ),
339 ),
340 ),
341 );
342 } else {
343 yield* orchestrator.dispatch({
344 type: "message.dispatch",
345 commandId: CommandId.make("steer"),
346 threadId,
347 messageId,
348 text: "Change direction now.",
349 attachments: [],
350 dispatchMode: { type: "steer_active", targetRunId: parent.id },
351 createdBy: "user",
352 creationSource: "web",
353 });
354 }
355 yield* worker.drain();
356 const explicitSteer = delivery === "user steering";
357 assert.equal(batchAbort.signal.aborted, explicitSteer);
358 assert.equal(offers.length, explicitSteer ? 2 : 1);
359 const finished = yield* watch(
360 (event) =>
361 event.type === "run.updated" &&
362 event.payload.id === parent.id &&
363 event.payload.status === "waiting",
364 );
365 // Bash finishes before the pending status reads, as in the capture.
366 yield* Queue.offer(sdkMessages, toolResult("build", false));
367 for (const id of statusIds)
368 yield* Queue.offer(sdkMessages, toolResult(id, batchAbort.signal.aborted));
369 if (explicitSteer) yield* Queue.offer(sdkMessages, result("aborted", true));
370 yield* Queue.offer(sdkMessages, result("completed", false));
371 yield* Fiber.join(finished);
372 yield* worker.drain();
373 yield* orchestrator.resumeQueuedRuns;
374 yield* worker.drain();
375 if (!explicitSteer) {
376 const queued = (yield* orchestrator.getThreadProjection(threadId)).runs[1]!;
377 const noticeFinished = yield* watch(
378 (event) =>
379 event.type === "run.updated" &&
380 event.payload.id === queued.id &&
381 event.payload.status === "waiting",
382 );
383 const delivered =
384 delivery === "child completion"
385 ? yield* watch(
386 (event) =>
387 event.type === "subagent.updated" &&
388 event.payload.completionDelivery?.state === "delivered",
389 )
390 : null;
391 yield* Queue.offer(sdkMessages, result("notice-completed", false));
392 yield* Fiber.join(noticeFinished);
393 yield* worker.drain();
394 if (delivered !== null) yield* Fiber.join(delivered);
395 }
396 const after = yield* orchestrator.getThreadProjection(threadId);
397 const reads = after.turnItems.filter((item) =>
398 statusIds.includes(item.nativeItemRef?.nativeId ?? ""),
399 );
400 assert.equal(reads.length, 4);
401 for (const read of reads) {
402 assert.equal(read.status, explicitSteer ? "cancelled" : "completed");
403 assert.equal(read.toolNonExecutionKind, explicitSteer ? "cancelled" : undefined);
404 assert.equal(read.type === "dynamic_tool" && read.output === refusal, explicitSteer);
405 }
406 assert.equal(offers.length, 2);
407 assert.equal(
408 offers.filter((offer) => offer.priority === "now").length,
409 explicitSteer ? 1 : 0,
410 );
411 if (!explicitSteer) {
412 assert.equal(after.runs.length, 2);
413 if (delivery === "child completion") {
414 assert.equal(after.messages.filter((message) => message.id === messageId).length, 1);
415 assert.equal(after.subagents[0]?.completionDelivery?.state, "delivered");
416 } else {
417 assert.equal(
418 after.messages.filter((message) => message.scheduledTaskId !== undefined).length,
419 1,
420 );
421 }
422 }
423 yield* orchestrator.recoverDelegatedTasks;
424 yield* orchestrator.resumeQueuedRuns;
425 yield* worker.drain();
426 assert.equal(offers.length, 2);
427 }).pipe(
428 Effect.provide(
429 ProviderReplayHarness.layerWithRegistry(
430 { name: "claude-automatic-delivery" },
431 ProviderAdapterRegistry.layerSingle(adapter),
432 { runEffectWorker: false },
433 ),
434 ),
435 );
436 }).pipe(
437 Effect.provide(
438 Layer.mergeAll(
439 IdAllocator.layer,
440 McpProviderSessions.layer,
441 ScriptProgress.layerUnavailable,
442 NodeServices.layer,
443 ),
444 ),
445 ),
446 ),
447);