authorgravatar for saiansh2525@gmail.comYash Singh <saiansh2525@gmail.com> 2026-10-04 23:31:33-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-04 23:31:33-05:00
log1beb0355d05bbe4a780b238e3138f0a8ed99df5f
tree2128a2f4f6ff7f053634d2aa6ae5c9672962936e
parenta1d9d72aefc2a388ea3915956e9d84d279c87da7
signature Signed by PGP key B5690EEEBB952194

fix(server): queue background notifications during active tools (#15892)

8 files changed, 581 insertions(+), 4 deletions(-)

apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts+85
...@@ -871,6 +871,20 @@ describe("ClaudeAdapterV2 context usage", () => {...@@ -871,6 +871,20 @@ describe("ClaudeAdapterV2 context usage", () => {
871});871});
872872
873describe("ClaudeAdapterV2 session permissions", () => {873describe("ClaudeAdapterV2 session permissions", () => {
874 it("keeps explicit user refusals classified as user_reject", () => {
875 const result = ClaudeAdapterV2.permissionResultFromDecision({
876 toolName: "Bash",
877 decision: "decline",
878 toolInput: { command: "make" },
879 toolUseID: "denied-build",
880 });
881 assert.equal(result.behavior, "deny");
882 if (result.behavior !== "deny") return;
883 assert.equal(result.decisionClassification, "user_reject");
884 assert.equal(result.message, "User declined tool execution.");
885 assert.equal(result.interrupt, undefined);
886 });
887
874 it("forces suggested permission updates to session scope", () => {888 it("forces suggested permission updates to session scope", () => {
875 const result = ClaudeAdapterV2.permissionResultFromDecision({889 const result = ClaudeAdapterV2.permissionResultFromDecision({
876 toolName: "Bash",890 toolName: "Bash",
...@@ -2480,6 +2494,77 @@ describe("ClaudeAdapterV2 background wake turns", () => {...@@ -2480,6 +2494,77 @@ describe("ClaudeAdapterV2 background wake turns", () => {
2480 }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),2494 }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
2481 );2495 );
24822496
2497 it.effect.each(["cancelled", "denied", "permission_denied", undefined])(
2498 "preserves native tool non-execution metadata %s without inferring a denial from text",
2499 (kind) =>
2500 Effect.scoped(
2501 Effect.gen(function* () {
2502 const harness = yield* makeWakeHarness;
2503 yield* harness.runtime.startTurn(
2504 makeClaudeTestTurnInput({
2505 threadId: harness.threadId,
2506 providerThread: harness.providerThread,
2507 now: yield* DateTime.now,
2508 attemptId: RunAttemptId.make("attempt-tool-non-execution"),
2509 text: "Run the tool.",
2510 attachments: [],
2511 }),
2512 );
2513 // Same error text can describe a cancellation or a real refusal.
2514 // Each result must use its own metadata, even in a multi-result frame.
2515 yield* harness.offerAndWait(
2516 claudeSdkFrame({
2517 type: "user",
2518 uuid: "tool-non-execution",
2519 session_id: WAKE_NATIVE_SESSION,
2520 parent_tool_use_id: null,
2521 message: {
2522 role: "user",
2523 content: [
2524 {
2525 type: "tool_result",
2526 tool_use_id: "tool-error",
2527 is_error: true,
2528 content: "STOP and wait for the user.",
2529 },
2530 { type: "tool_result", tool_use_id: "tool-ok", is_error: false, content: "OK" },
2531 ],
2532 },
2533 ...(kind === undefined
2534 ? {}
2535 : {
2536 tool_result_meta: [
2537 { id: "tool-error", non_execution_kind: kind },
2538 { id: "tool-ok", non_execution_kind: null },
2539 ],
2540 }),
2541 }),
2542 );
2543 yield* Queue.offer(
2544 harness.sdkMessages,
2545 makeResultFrame({ uuid: "result-non-execution", result: "Done" }),
2546 );
2547 yield* Queue.take(harness.terminalReceipts);
2548 const items = harness.events.flatMap((event) =>
2549 event.type === "turn_item.updated" && event.turnItem.type === "dynamic_tool"
2550 ? [event.turnItem]
2551 : [],
2552 );
2553 const failed = items.findLast((item) => item.nativeItemRef?.nativeId === "tool-error")!;
2554 assert.equal(failed.status, kind === "cancelled" ? "cancelled" : "failed");
2555 assert.equal(failed.toolNonExecutionKind, kind);
2556 const ok = items.findLast((item) => item.nativeItemRef?.nativeId === "tool-ok")!;
2557 assert.equal(ok.status, "completed");
2558 assert.equal(ok.toolNonExecutionKind, undefined);
2559 const node = harness.events.findLast(
2560 (event) =>
2561 event.type === "node.updated" && event.node.nativeItemRef?.nativeId === "tool-error",
2562 );
2563 assert.equal(node?.type === "node.updated" ? node.node.status : undefined, failed.status);
2564 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
2565 ),
2566 );
2567
2483 it.effect.each(2568 it.effect.each(
2484 (["aborted_tools", "aborted_streaming"] as const).flatMap((terminalReason) =>2569 (["aborted_tools", "aborted_streaming"] as const).flatMap((terminalReason) =>
2485 [true, false].map((steered) => ({ terminalReason, steered })),2570 [true, false].map((steered) => ({ terminalReason, steered })),
apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts+34-1
...@@ -198,6 +198,7 @@ export const ClaudeProviderCapabilitiesV2 = {...@@ -198,6 +198,7 @@ export const ClaudeProviderCapabilitiesV2 = {
198 emitsTurnCompleted: true,198 emitsTurnCompleted: true,
199 supportsInterrupt: true,199 supportsInterrupt: true,
200 supportsActiveSteering: true,200 supportsActiveSteering: true,
201 activeSteeringInterruptsTools: true,
201 supportsSteeringByInterruptRestart: false,202 supportsSteeringByInterruptRestart: false,
202 supportsQueuedMessages: true,203 supportsQueuedMessages: true,
203 terminalStatusQuality: "strong",204 terminalStatusQuality: "strong",
...@@ -2163,6 +2164,24 @@ function claudeToolResultEntriesFromMessage(message: SDKMessage): ReadonlyArray<...@@ -2163,6 +2164,24 @@ function claudeToolResultEntriesFromMessage(message: SDKMessage): ReadonlyArray<
2163 ];2164 ];
2164}2165}
21652166
2167// These wire fields are not yet declared by the SDK's SDKUserMessage type.
2168const ClaudeToolResultMetadata = Schema.Struct({
2169 tool_result_meta: Schema.Array(
2170 Schema.Struct({
2171 id: Schema.String,
2172 non_execution_kind: Schema.optional(Schema.NullOr(Schema.String)),
2173 }),
2174 ),
2175});
2176const isClaudeToolResultMetadata = Schema.is(ClaudeToolResultMetadata);
2177
2178function claudeToolNonExecutionKind(message: SDKMessage, toolUseId: string) {
2179 return isClaudeToolResultMetadata(message)
2180 ? (message.tool_result_meta.find((meta) => meta.id === toolUseId)?.non_execution_kind ??
2181 undefined)
2182 : undefined;
2183}
2184
2166function parentToolUseIdFromSdkMessage(message: SDKMessage): string | null {2185function parentToolUseIdFromSdkMessage(message: SDKMessage): string | null {
2167 return message.type === "assistant" || message.type === "user"2186 return message.type === "assistant" || message.type === "user"
2168 ? message.parent_tool_use_id2187 ? message.parent_tool_use_id
...@@ -3705,6 +3724,7 @@ export function makeClaudeAdapterV2(...@@ -3705,6 +3724,7 @@ export function makeClaudeAdapterV2(
3705 readonly startedAt: DateTime.Utc;3724 readonly startedAt: DateTime.Utc;
3706 readonly updatedAt: DateTime.Utc;3725 readonly updatedAt: DateTime.Utc;
3707 readonly presentation: ClaudeToolPresentation | undefined;3726 readonly presentation: ClaudeToolPresentation | undefined;
3727 readonly toolNonExecutionKind?: string;
3708 }) => {3728 }) => {
3709 const completedAt = input.status === "running" ? null : input.updatedAt;3729 const completedAt = input.status === "running" ? null : input.updatedAt;
3710 const nodeId = idAllocator.derive.nodeFromProviderItem({3730 const nodeId = idAllocator.derive.nodeFromProviderItem({
...@@ -3752,6 +3772,9 @@ export function makeClaudeAdapterV2(...@@ -3752,6 +3772,9 @@ export function makeClaudeAdapterV2(
3752 })3772 })
3753 : undefined;3773 : undefined;
3754 const itemBase = {3774 const itemBase = {
3775 ...(input.toolNonExecutionKind === undefined
3776 ? {}
3777 : { toolNonExecutionKind: input.toolNonExecutionKind }),
3755 id: turnItemId,3778 id: turnItemId,
3756 threadId: input.threadId,3779 threadId: input.threadId,
3757 runId: input.runId,3780 runId: input.runId,
...@@ -6080,6 +6103,10 @@ export function makeClaudeAdapterV2(...@@ -6080,6 +6103,10 @@ export function makeClaudeAdapterV2(
6080 parentToolUseId,6103 parentToolUseId,
6081 }));6104 }));
6082 const completedAt = yield* DateTime.now;6105 const completedAt = yield* DateTime.now;
6106 const toolNonExecutionKind = claudeToolNonExecutionKind(
6107 message,
6108 toolResult.tool_use_id,
6109 );
6083 const artifacts = buildToolCallArtifacts({6110 const artifacts = buildToolCallArtifacts({
6084 context,6111 context,
6085 nativeItemId: toolCall.nativeItemId,6112 nativeItemId: toolCall.nativeItemId,
...@@ -6092,7 +6119,13 @@ export function makeClaudeAdapterV2(...@@ -6092,7 +6119,13 @@ export function makeClaudeAdapterV2(
6092 parentNodeId: toolCall.parentNodeId,6119 parentNodeId: toolCall.parentNodeId,
6093 ordinal: toolCall.ordinal,6120 ordinal: toolCall.ordinal,
6094 output,6121 output,
6095 status: isClaudeToolResultError(toolResult) ? "failed" : "completed",6122 status:
6123 toolNonExecutionKind === "cancelled"
6124 ? "cancelled"
6125 : isClaudeToolResultError(toolResult)
6126 ? "failed"
6127 : "completed",
6128 ...(toolNonExecutionKind === undefined ? {} : { toolNonExecutionKind }),
6096 startedAt: toolCall.startedAt,6129 startedAt: toolCall.startedAt,
6097 updatedAt: completedAt,6130 updatedAt: completedAt,
6098 presentation: toolCall.presentation,6131 presentation: toolCall.presentation,
apps/server/src/orchestration-v2/ClaudeAutomaticDelivery.integration.test.ts created+420
...@@ -0,0 +1,420 @@
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 Effect from "effect/Effect";
19import * as FileSystem from "effect/FileSystem";
20import * as Fiber from "effect/Fiber";
21import * as Layer from "effect/Layer";
22import * as Path from "effect/Path";
23import * as Queue from "effect/Queue";
24import * as Schema from "effect/Schema";
25import * as Stream from "effect/Stream";
26import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts";
27import * as Scheduler from "../scheduling/Scheduler.ts";
28import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts";
29import * as ClaudeAdapterV2 from "./Adapters/ClaudeAdapterV2.ts";
30import * as EffectWorker from "./EffectWorker.ts";
31import * as EventSink from "./EventSink.ts";
32import * as IdAllocator from "./IdAllocator.ts";
33import * as LegacyV1ThreadImporter from "./legacy/LegacyV1ThreadImporter.ts";
34import * as Orchestrator from "./Orchestrator.ts";
35import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
36import * as ThreadLaunchService from "./ThreadLaunchService.ts";
37import * as ThreadManagementService from "./ThreadManagementService.ts";
38import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts";
39import { checkpointWorkspace } from "./testkit/ReplayFixtureWorkspace.ts";
40
41const sessionId = "automatic-delivery-session";
42const settings = Schema.decodeSync(ClaudeSettings)({});
43const decodeScheduledTask = Schema.decodeEffect(ScheduledTaskUpsertInput);
44const refusal =
45 "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.";
46const statusIds = ["status-1", "status-2", "status-3", "status-4"];
47const modelSelection = {
48 instanceId: ProviderInstanceId.make("claudeAgent"),
49 model: "claude-sonnet-4-6",
50};
51
52// Native frames match the captured Bash/status batch and tool_result_meta shape.
53// The external SDK boundary is replayed; the adapter and durable dispatch are real.
54function frame(value: unknown): SDKMessage {
55 return value as SDKMessage;
56}
57
58const batch = frame({
59 type: "assistant",
60 uuid: "batch",
61 session_id: sessionId,
62 parent_tool_use_id: null,
63 message: {
64 id: "tool-batch",
65 role: "assistant",
66 type: "message",
67 model: modelSelection.model,
68 content: [
69 { type: "tool_use", id: "build", name: "Bash", input: { command: "make" } },
70 ...statusIds.map((id) => ({
71 type: "tool_use",
72 id,
73 name: "mcp__t3-code__task_status",
74 input: { taskId: id },
75 })),
76 ],
77 stop_reason: "tool_use",
78 stop_sequence: null,
79 usage: { input_tokens: 1, output_tokens: 1 },
80 },
81});
82
83const toolResult = (id: string, cancelled: boolean) =>
84 frame({
85 type: "user",
86 uuid: `result-${id}`,
87 session_id: sessionId,
88 parent_tool_use_id: null,
89 message: {
90 role: "user",
91 content: [
92 {
93 type: "tool_result",
94 tool_use_id: id,
95 is_error: cancelled,
96 content: cancelled ? refusal : id === "build" ? "build OK" : "Task completed normally",
97 },
98 ],
99 },
100 ...(cancelled ? { tool_result_meta: [{ id, non_execution_kind: "cancelled" }] } : {}),
101 });
102
103const result = (uuid: string, aborted: boolean) =>
104 frame({
105 type: "result",
106 subtype: "success",
107 uuid,
108 session_id: sessionId,
109 is_error: false,
110 num_turns: 1,
111 result: "",
112 stop_reason: "end_turn",
113 terminal_reason: aborted ? "aborted_tools" : "completed",
114 permission_denials: [],
115 duration_ms: 1,
116 duration_api_ms: 1,
117 total_cost_usd: 0,
118 usage: { input_tokens: 1, output_tokens: 1 },
119 modelUsage: {},
120 });
121
122it.effect.each(["child completion", "scheduled message", "user steering"] as const)(
123 "delivers %s with Claude's native pending-tool cancellation behavior",
124 (delivery) =>
125 Effect.scoped(
126 Effect.gen(function* () {
127 const cwd = yield* checkpointWorkspace("claude-automatic-delivery");
128 const sdkMessages = yield* Queue.unbounded<SDKMessage>();
129 const offers: SDKUserMessage[] = [];
130 const nativeQueue: SDKUserMessage[] = [];
131 const batchAbort = new AbortController();
132 const adapter = ClaudeAdapterV2.makeClaudeAdapterV2({
133 instanceId: modelSelection.instanceId,
134 settings,
135 environment: {},
136 attachmentsDir: cwd,
137 fileSystem: yield* FileSystem.FileSystem,
138 path: yield* Path.Path,
139 idAllocator: yield* IdAllocator.IdAllocatorV2,
140 queryRunner: {
141 allocateSessionId: Effect.succeed(sessionId),
142 open: () =>
143 Effect.succeed({
144 messages: Stream.fromQueue(sdkMessages),
145 offer: (message) =>
146 Effect.sync(() => {
147 offers.push(message);
148 nativeQueue.push(message);
149 // Claude Code 2.1.289's queue watcher aborts when a now-priority
150 // command arrives, including while Bash blocks pending reads.
151 if (nativeQueue.some((command) => command.priority === "now")) {
152 batchAbort.abort({ kind: "interrupt" });
153 }
154 }),
155 setModel: () => Effect.void,
156 setPermissionMode: () => Effect.void,
157 interrupt: Effect.die("automatic delivery must never interrupt"),
158 close: Effect.void,
159 }),
160 forkSession: () => Effect.die("unused"),
161 subagentLaunchToolUseId: () => Effect.succeed(null),
162 assertComplete: Effect.void,
163 },
164 });
165 yield* Effect.gen(function* () {
166 const orchestrator = yield* Orchestrator.OrchestratorV2;
167 const worker = yield* EffectWorker.OrchestrationEffectWorkerV2;
168 const threadId = ThreadId.make("thread:automatic-delivery");
169 const projectId = ProjectId.make("project:automatic-delivery");
170 const watch = (predicate: (event: OrchestrationV2DomainEvent) => boolean) =>
171 orchestrator.streamDomainEvents.pipe(
172 Stream.filter(predicate),
173 Stream.take(1),
174 Stream.runDrain,
175 Effect.forkScoped,
176 );
177 yield* orchestrator.dispatch({
178 type: "thread.create",
179 commandId: CommandId.make("create"),
180 threadId,
181 projectId,
182 title: "Automatic delivery",
183 modelSelection,
184 runtimeMode: "full-access",
185 interactionMode: "default",
186 branch: null,
187 worktreePath: cwd,
188 createdBy: "user",
189 creationSource: "web",
190 });
191 const running = yield* watch(
192 (event) => event.type === "provider-turn.updated" && event.payload.status === "running",
193 );
194 yield* orchestrator.dispatch({
195 type: "message.dispatch",
196 commandId: CommandId.make("first"),
197 threadId,
198 messageId: MessageId.make("first"),
199 text: "Build and check all four tasks.",
200 attachments: [],
201 dispatchMode: { type: "start_immediately" },
202 createdBy: "user",
203 creationSource: "web",
204 });
205 yield* worker.drain();
206 yield* Fiber.join(running);
207 const pending = yield* watch(
208 (event) =>
209 event.type === "turn-item.updated" &&
210 event.payload.nativeItemRef?.nativeId === "status-4",
211 );
212 yield* Queue.offer(sdkMessages, batch);
213 yield* Fiber.join(pending);
214 const before = yield* orchestrator.getThreadProjection(threadId);
215 assert.equal(before.turnItems.filter((item) => item.status === "running").length, 5);
216 const parent = before.runs[0]!;
217 if (parent.rootNodeId === null) return yield* Effect.die("parent has no root node");
218 const messageId = MessageId.make("automatic-notice");
219 if (delivery === "child completion") {
220 const taskId = NodeId.make("completed-child");
221 const now = yield* DateTime.now;
222 const sink = yield* EventSink.EventSinkV2;
223 yield* sink.write({
224 events: [
225 {
226 id: EventId.make("cohort"),
227 type: "run.updated",
228 threadId,
229 runId: parent.id,
230 occurredAt: now,
231 payload: {
232 ...parent,
233 delegatedCompletion: {
234 disposition: "open",
235 nextGeneration: 2,
236 delivery: { generation: 1, messageId, taskIds: [taskId] },
237 },
238 },
239 },
240 {
241 id: EventId.make("child"),
242 type: "subagent.updated",
243 threadId,
244 runId: parent.id,
245 nodeId: taskId,
246 occurredAt: now,
247 payload: {
248 id: taskId,
249 threadId,
250 runId: parent.id,
251 parentNodeId: parent.rootNodeId,
252 origin: "app_owned",
253 createdBy: "agent",
254 driver: ClaudeAdapterV2.CLAUDE_PROVIDER,
255 providerInstanceId: modelSelection.instanceId,
256 providerThreadId: null,
257 childThreadId: null,
258 nativeTaskRef: null,
259 prompt: "Background work",
260 title: "Child",
261 model: null,
262 completionWake: "always",
263 completionDelivery: { state: "claimed", observedByRunId: null },
264 status: "completed",
265 result: "done",
266 startedAt: now,
267 completedAt: now,
268 updatedAt: now,
269 },
270 },
271 ],
272 });
273 const command = {
274 type: "message.dispatch" as const,
275 commandId: CommandId.make("completion"),
276 threadId,
277 messageId,
278 text: "Child completed",
279 attachments: [],
280 dispatchMode: { type: "queue_after_active" as const },
281 createdBy: "agent" as const,
282 creationSource: "server" as const,
283 delegatedCompletion: { parentRunId: parent.id, generation: 1, taskIds: [taskId] },
284 };
285 yield* orchestrator.dispatch(command);
286 yield* orchestrator.dispatch(command);
287 yield* orchestrator.recoverDelegatedTasks;
288 } else if (delivery === "scheduled message") {
289 yield* Effect.gen(function* () {
290 const service = yield* ScheduledTaskService.ScheduledTaskService;
291 const { task } = yield* service.upsert(
292 yield* decodeScheduledTask({
293 title: "Status check",
294 prompt: "Check the four tasks.",
295 enabled: true,
296 schedule: { type: "interval", everyMs: 60_000 },
297 projectId,
298 threadId,
299 workspaceStrategy: { type: "root" },
300 modelSelection,
301 runtimeMode: "full-access",
302 interactionMode: "default",
303 }),
304 );
305 const ran = yield* service.runNow({ id: task.id });
306 assert.equal(ran.task.lastRunStatus, "succeeded");
307 }).pipe(
308 Effect.provide(
309 ScheduledTaskService.layer.pipe(
310 Layer.provide(ThreadManagementService.layer),
311 Layer.provide(
312 Layer.mock(LegacyV1ThreadImporter.LegacyV1ThreadImporter)({
313 ensureTranscript: () =>
314 Effect.succeed({ importedThreadCount: 0, importedMessageCount: 0 }),
315 }),
316 ),
317 Layer.provide(Layer.mock(ThreadLaunchService.ThreadLaunchService)({})),
318 Layer.provide(
319 Layer.mergeAll(NodeCrypto.layer, Scheduler.layer, SqlitePersistenceMemory),
320 ),
321 ),
322 ),
323 );
324 } else {
325 yield* orchestrator.dispatch({
326 type: "message.dispatch",
327 commandId: CommandId.make("steer"),
328 threadId,
329 messageId,
330 text: "Change direction now.",
331 attachments: [],
332 dispatchMode: { type: "steer_active", targetRunId: parent.id },
333 createdBy: "user",
334 creationSource: "web",
335 });
336 }
337 yield* worker.drain();
338 const explicitSteer = delivery === "user steering";
339 assert.equal(batchAbort.signal.aborted, explicitSteer);
340 assert.equal(offers.length, explicitSteer ? 2 : 1);
341 const finished = yield* watch(
342 (event) =>
343 event.type === "run.updated" &&
344 event.payload.id === parent.id &&
345 event.payload.status === "waiting",
346 );
347 // Bash finishes before the pending status reads, as in the capture.
348 yield* Queue.offer(sdkMessages, toolResult("build", false));
349 for (const id of statusIds)
350 yield* Queue.offer(sdkMessages, toolResult(id, batchAbort.signal.aborted));
351 if (explicitSteer) yield* Queue.offer(sdkMessages, result("aborted", true));
352 yield* Queue.offer(sdkMessages, result("completed", false));
353 yield* Fiber.join(finished);
354 yield* worker.drain();
355 yield* orchestrator.resumeQueuedRuns;
356 yield* worker.drain();
357 if (!explicitSteer) {
358 const queued = (yield* orchestrator.getThreadProjection(threadId)).runs[1]!;
359 const noticeFinished = yield* watch(
360 (event) =>
361 event.type === "run.updated" &&
362 event.payload.id === queued.id &&
363 event.payload.status === "waiting",
364 );
365 const delivered =
366 delivery === "child completion"
367 ? yield* watch(
368 (event) =>
369 event.type === "subagent.updated" &&
370 event.payload.completionDelivery?.state === "delivered",
371 )
372 : null;
373 yield* Queue.offer(sdkMessages, result("notice-completed", false));
374 yield* Fiber.join(noticeFinished);
375 yield* worker.drain();
376 if (delivered !== null) yield* Fiber.join(delivered);
377 }
378 const after = yield* orchestrator.getThreadProjection(threadId);
379 const reads = after.turnItems.filter((item) =>
380 statusIds.includes(item.nativeItemRef?.nativeId ?? ""),
381 );
382 assert.equal(reads.length, 4);
383 for (const read of reads) {
384 assert.equal(read.status, explicitSteer ? "cancelled" : "completed");
385 assert.equal(read.toolNonExecutionKind, explicitSteer ? "cancelled" : undefined);
386 assert.equal(read.type === "dynamic_tool" && read.output === refusal, explicitSteer);
387 }
388 assert.equal(offers.length, 2);
389 assert.equal(
390 offers.filter((offer) => offer.priority === "now").length,
391 explicitSteer ? 1 : 0,
392 );
393 if (!explicitSteer) {
394 assert.equal(after.runs.length, 2);
395 if (delivery === "child completion") {
396 assert.equal(after.messages.filter((message) => message.id === messageId).length, 1);
397 assert.equal(after.subagents[0]?.completionDelivery?.state, "delivered");
398 } else {
399 assert.equal(
400 after.messages.filter((message) => message.scheduledTaskId !== undefined).length,
401 1,
402 );
403 }
404 }
405 yield* orchestrator.recoverDelegatedTasks;
406 yield* orchestrator.resumeQueuedRuns;
407 yield* worker.drain();
408 assert.equal(offers.length, 2);
409 }).pipe(
410 Effect.provide(
411 makeOrchestratorV2ReplayLayerWithRegistry(
412 { name: "claude-automatic-delivery" },
413 ProviderAdapterRegistry.makeSingleLayer(adapter),
414 { runEffectWorker: false },
415 ),
416 ),
417 );
418 }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
419 ),
420);
apps/server/src/orchestration-v2/Orchestrator.ts+2-1
...@@ -4596,7 +4596,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio...@@ -4596,7 +4596,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
4596 .pipe(Effect.orElseSucceed(() => Option.none()));4596 .pipe(Effect.orElseSucceed(() => Option.none()));
4597 if (4597 if (
4598 Option.isSome(session) &&4598 Option.isSome(session) &&
4599 session.value.providerSession.capabilities.turns.supportsActiveSteering4599 session.value.providerSession.capabilities.turns.supportsActiveSteering &&
4600 session.value.providerSession.capabilities.turns.activeSteeringInterruptsTools !== true
4600 ) {4601 ) {
4601 dispatchMode = { type: "steer_active", targetRunId: active.id };4602 dispatchMode = { type: "steer_active", targetRunId: active.id };
4602 }4603 }
apps/server/src/orchestration-v2/SteeringCompletion.integration.test.ts+6-1
...@@ -47,6 +47,7 @@ it.effect.each(...@@ -47,6 +47,7 @@ it.effect.each(
47 "before dispatch",47 "before dispatch",
48 "after delivery",48 "after delivery",
49 "without native steering",49 "without native steering",
50 "with interrupting native steering",
50 "settled only",51 "settled only",
51 ] as const52 ] as const
52 ).map((timing) => ({53 ).map((timing) => ({
...@@ -57,7 +58,10 @@ it.effect.each(...@@ -57,7 +58,10 @@ it.effect.each(
57 )58 )
58 .filter(59 .filter(
59 ({ mailbox, timing }) =>60 ({ mailbox, timing }) =>
60 mailbox || (timing !== "without native steering" && timing !== "settled only"),61 mailbox ||
62 (timing !== "without native steering" &&
63 timing !== "with interrupting native steering" &&
64 timing !== "settled only"),
61 ),65 ),
62)("delivers $label when completion wins $timing", ({ mailbox, timing }) =>66)("delivers $label when completion wins $timing", ({ mailbox, timing }) =>
63 Effect.scoped(67 Effect.scoped(
...@@ -73,6 +77,7 @@ it.effect.each(...@@ -73,6 +77,7 @@ it.effect.each(
73 turns: {77 turns: {
74 ...CodexProviderCapabilitiesV2.turns,78 ...CodexProviderCapabilitiesV2.turns,
75 supportsActiveSteering: timing !== "without native steering",79 supportsActiveSteering: timing !== "without native steering",
80 activeSteeringInterruptsTools: timing === "with interrupting native steering",
76 },81 },
77 };82 };
78 const adapter: ProviderAdapterV2Shape = {83 const adapter: ProviderAdapterV2Shape = {
apps/server/src/scheduledTasks/ScheduledTaskService.ts+2-1
...@@ -545,7 +545,8 @@ export const layer = Layer.effect(...@@ -545,7 +545,8 @@ export const layer = Layer.effect(
545 text: prompt,545 text: prompt,
546 attachments: [],546 attachments: [],
547 modelSelection: active.modelSelection,547 modelSelection: active.modelSelection,
548 mode: "auto",548 // Scheduled prompts must not interrupt tools in the bound thread.
549 mode: "queue",
549 createdBy: active.createdBy,550 createdBy: active.createdBy,
550 creationSource: active.creationSource,551 creationSource: active.creationSource,
551 }),552 }),
packages/contracts/src/orchestrationV2.test.ts+29
...@@ -1038,6 +1038,35 @@ describe("orchestration V2 contracts", () => {...@@ -1038,6 +1038,35 @@ describe("orchestration V2 contracts", () => {
1038 });1038 });
1039});1039});
10401040
1041it("preserves tool cancellation and denial metadata through persisted and wire schemas", () => {
1042 for (const kind of ["cancelled", "denied", undefined]) {
1043 const item = decodeOrchestrationV2TurnItem({
1044 id: "tool-result",
1045 threadId: "thread",
1046 runId: null,
1047 nodeId: null,
1048 providerThreadId: null,
1049 providerTurnId: null,
1050 nativeItemRef: null,
1051 parentItemId: null,
1052 ordinal: 1,
1053 type: "dynamic_tool",
1054 toolName: "task_status",
1055 input: { taskId: "child" },
1056 status: kind === "cancelled" ? "cancelled" : "failed",
1057 ...(kind === undefined ? {} : { toolNonExecutionKind: kind }),
1058 title: null,
1059 startedAt: now,
1060 completedAt: now,
1061 updatedAt: now,
1062 output: "Tool did not execute",
1063 });
1064 const wire = encodeOrchestrationV2TurnItemJson(item);
1065 expect(wire.toolNonExecutionKind).toBe(kind);
1066 expect(decodeOrchestrationV2TurnItemJson(wire)).toEqual(item);
1067 }
1068});
1069
1041it("round-trips typed notifications and keeps work outcome separate from item status", () => {1070it("round-trips typed notifications and keeps work outcome separate from item status", () => {
1042 const now = DateTime.makeUnsafe("2026-09-09T00:00:00Z");1071 const now = DateTime.makeUnsafe("2026-09-09T00:00:00Z");
1043 const base = {1072 const base = {
packages/contracts/src/orchestrationV2.ts+3
...@@ -211,6 +211,8 @@ export const OrchestrationV2TurnCapabilities = Schema.Struct({...@@ -211,6 +211,8 @@ export const OrchestrationV2TurnCapabilities = Schema.Struct({
211 emitsTurnCompleted: Schema.Boolean,211 emitsTurnCompleted: Schema.Boolean,
212 supportsInterrupt: Schema.Boolean,212 supportsInterrupt: Schema.Boolean,
213 supportsActiveSteering: Schema.Boolean,213 supportsActiveSteering: Schema.Boolean,
214 // Some native steering mechanisms cancel pending tools before consuming the message.
215 activeSteeringInterruptsTools: Schema.optional(Schema.Boolean),
214 supportsSteeringByInterruptRestart: Schema.Boolean,216 supportsSteeringByInterruptRestart: Schema.Boolean,
215 supportsQueuedMessages: Schema.Boolean,217 supportsQueuedMessages: Schema.Boolean,
216 terminalStatusQuality: Schema.Literals(["strong", "weak", "none"]),218 terminalStatusQuality: Schema.Literals(["strong", "weak", "none"]),
...@@ -1253,6 +1255,7 @@ export type OrchestrationV2UserMessageInputIntent =...@@ -1253,6 +1255,7 @@ export type OrchestrationV2UserMessageInputIntent =
1253 typeof OrchestrationV2UserMessageInputIntent.Type;1255 typeof OrchestrationV2UserMessageInputIntent.Type;
12541256
1255const OrchestrationV2TurnItemBaseFields = {1257const OrchestrationV2TurnItemBaseFields = {
1258 toolNonExecutionKind: Schema.optional(Schema.String),
1256 toolSurface: Schema.optional(ToolActivitySurface),1259 toolSurface: Schema.optional(ToolActivitySurface),
1257 toolIcon: Schema.optional(ToolActivityIcon),1260 toolIcon: Schema.optional(ToolActivityIcon),
1258 toolSource: Schema.optional(ToolActivitySource),1261 toolSource: Schema.optional(ToolActivitySource),