diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 785c4b1c9ec19..480b810c38dc9 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -6765,6 +6765,125 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("does not keep the thread busy for Claude's ambient artifact watch", () => + Effect.scoped( + Effect.gen(function* () { + const WATCH_TASK_ID = "task-artifact-watch"; + const description = "live updates for artifact https://claude.ai/artifact/abc"; + const harness = yield* makeWakeHarness; + const now = yield* DateTime.now; + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-artifact-watch"), + text: "Publish the page.", + attachments: [], + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "background_tasks_changed", + tasks: [ + { task_id: WATCH_TASK_ID, task_type: "monitor_ws", description, ambient: true }, + ], + uuid: "00000000-0000-4000-8000-000000000401", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: WATCH_TASK_ID, + tool_use_id: "toolu-artifact-publish", + description, + task_type: "monitor_ws", + ambient: true, + uuid: "00000000-0000-4000-8000-000000000402", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000403", result: "Published." }), + ); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + + // The watcher outlives its turn: its later frames, which carry only the task id, can + // arrive during the next one and must not reach routing. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("attempt-claude-artifact-watch-2"), + text: "Thanks.", + attachments: [], + providerTurnOrdinal: 2, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_progress", + task_id: WATCH_TASK_ID, + tool_use_id: "toolu-artifact-publish", + description: "comment received", + usage: { total_tokens: 0, tool_uses: 0, duration_ms: 0 }, + uuid: "00000000-0000-4000-8000-000000000407", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_updated", + task_id: WATCH_TASK_ID, + patch: { status: "killed", end_time: 1793289036381 }, + uuid: "00000000-0000-4000-8000-000000000404", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: WATCH_TASK_ID, + tool_use_id: "toolu-artifact-publish", + status: "stopped", + output_file: "/tmp/task-artifact-watch.output", + summary: description, + uuid: "00000000-0000-4000-8000-000000000405", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000406", result: "Done." }), + ); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + + assert.isFalse(harness.events.some((event) => event.type === "subagent.updated")); + assert.equal(harness.continuationRequests.length, 0); + assert.isFalse(yield* harness.hasPendingBackgroundWork); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ), + ); + it.effect("re-opens a resumed subagent whose task_started races past settle", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 81f9abc001a37..5fe48f6194b30 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1821,6 +1821,27 @@ function claudeTaskTypeFromSdkMessage(message: SDKMessage): string | null { return typeof taskType === "string" ? taskType : null; } +// Claude's own session-long socket watchers (e.g. the live-updates socket it +// arms on an Artifact publish) run until the session ends. They are neither +// subagents nor work the thread waits on. Ambient subagents stay subagents. +function isClaudeAmbientTaskStarted(message: SDKMessage): boolean { + return ( + message.type === "system" && + message.subtype === "task_started" && + Reflect.get(message, "ambient") === true && + claudeTaskTypeFromSdkMessage(message) === "monitor_ws" + ); +} + +// Task frames after task_started carry only the task id. +function claudeTaskIdFromSdkMessage(message: SDKMessage): string | null { + if (message.type !== "system") { + return null; + } + const taskId = Reflect.get(message, "task_id"); + return typeof taskId === "string" ? taskId : null; +} + function isClaudeNonSubagentTask(message: SDKMessage): boolean { return isClaudeOpaqueBackgroundTaskType(claudeTaskTypeFromSdkMessage(message)); } @@ -3160,6 +3181,8 @@ export function makeClaudeAdapterV2( // Subagents another subagent started (spawn_depth above 1). Claude // reports their end to the owning subagent, so it never wakes the root. const nestedSubagentTaskIds = yield* Ref.make>(new Set()); + // Ambient tasks (see isClaudeAmbientTaskStarted); their frames are dropped. + const ambientTaskIds = yield* Ref.make>(new Set()); const isNestedSubagentTask = (taskId: string) => Ref.get(nestedSubagentTaskIds).pipe(Effect.map((ids) => ids.has(taskId))); const recordWakeReport = ( @@ -6593,6 +6616,25 @@ export function makeClaudeAdapterV2( readonly message: SDKMessage; }) { const message = input.message; + if (isClaudeAmbientTaskStarted(message)) { + const taskId = claudeTaskIdFromSdkMessage(message); + if (taskId !== null) { + yield* Ref.update(ambientTaskIds, (current) => new Set(current).add(taskId)); + } + return; + } + const taskId = claudeTaskIdFromSdkMessage(message); + if (taskId !== null && (yield* Ref.get(ambientTaskIds)).has(taskId)) { + // task_notification is a task's last frame. + if (message.type === "system" && message.subtype === "task_notification") { + yield* Ref.update(ambientTaskIds, (current) => { + const next = new Set(current); + next.delete(taskId); + return next; + }); + } + return; + } const context = yield* Ref.get(activeTurn); const liveQuery = yield* Ref.get(queryContext); if (