diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index f00c58d9635341b30be7eb656121734b11bfd415..a91d82cef9e0b076d2731509825e5a21d0c1596a 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -3991,7 +3991,7 @@ describe("ClaudeAdapterLive", () => { const taskEventsFiber = yield* adapter.streamEvents.pipe( Stream.filter((event) => event.type.startsWith("task.")), - Stream.take(2), + Stream.take(3), Stream.runCollect, Effect.forkChild, ); @@ -4025,24 +4025,27 @@ describe("ClaudeAdapterLive", () => { session_id: "sdk-session", } as unknown as SDKMessage); // The subagent's assistant snapshot carries the authoritative API - // model id, which refines the linkage on later rows. + // model id. The correction is pushed at once, not on the next task row. + // Its tool calls arrive only here, so a task one of them launches must + // still resolve to this subagent as owner. harness.query.emit({ type: "assistant", parent_tool_use_id: "toolu_agent_m", message: { model: SYNTHETIC_SUBAGENT_MODEL, - content: [], + content: [{ type: "tool_use", id: "toolu_nested", name: "Skill", input: {} }], }, uuid: "subagent-snapshot-uuid", session_id: "sdk-session", } as unknown as SDKMessage); harness.query.emit({ type: "system", - subtype: "task_progress", - task_id: "task-model", - description: "Agent M", - usage: { total_tokens: 100, tool_uses: 1, duration_ms: 10 }, - uuid: "task-model-progress-uuid", + subtype: "task_started", + task_id: "task-nested", + description: "Nested", + task_type: "local_agent", + tool_use_id: "toolu_nested", + uuid: "task-nested-uuid", session_id: "sdk-session", } as unknown as SDKMessage); @@ -4053,11 +4056,18 @@ describe("ClaudeAdapterLive", () => { assert.equal(started.payload.model, SYNTHETIC_CLAUDE_CAPABLE_MODEL); assert.equal(started.payload.effort, "max"); } - const progress = taskEvents[1]; - assert.equal(progress?.type, "task.progress"); - if (progress?.type === "task.progress") { - assert.equal(progress.payload.model, SYNTHETIC_SUBAGENT_MODEL); - assert.equal(progress.payload.effort, "max"); + const refined = taskEvents[1]; + assert.equal(refined?.type, "task.updated"); + if (refined?.type === "task.updated") { + assert.equal(refined.payload.taskId, "task-model"); + assert.equal(refined.payload.model, SYNTHETIC_SUBAGENT_MODEL); + assert.equal(refined.payload.status, undefined); + } + const nested = taskEvents[2]; + assert.equal(nested?.type, "task.started"); + if (nested?.type === "task.started") { + assert.equal(nested.payload.agentId, "task-model"); + assert.equal(nested.payload.model, SYNTHETIC_SUBAGENT_MODEL); } }).pipe( Effect.provideService(Random.Random, makeDeterministicRandomService()), diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index d971566217e60c677b49648b4f1fab2db79d0388..e4b4bea1d02fc876f80860b36e4db4c09a28ec68 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -380,26 +380,26 @@ interface ClaudeTaskAgentState { } /** - * How many racing snapshot models to buffer per session. A snapshot whose - * task_started never arrives would otherwise pin its entry for the session's + * How many entries each task_started lookup map buffers per session. An entry + * whose task_started never arrives would otherwise pin it for the session's * lifetime; oldest entries evict first. */ -const PENDING_TASK_MODEL_CAP = 64; +const PENDING_TASK_ENTRY_CAP = 256; /** How long Stop waits for Claude to abort a turn before killing the process. */ const CLAUDE_INTERRUPT_GRACE = "3 seconds"; /** - * Buffers a subagent snapshot's authoritative model under its - * parent_tool_use_id, for snapshots that beat their task_started to the - * stream. task_started consumes the entry when it registers the task. + * Buffers a value that a later task_started reads by tool_use_id (a racing + * snapshot model, or a subagent's tool_use owner). task_started consumes the + * entry when it registers the task. */ -function rememberPendingTaskModel( +function rememberPendingTaskEntry( pending: Map, - parentToolUseId: string, - model: string, + toolUseId: string, + value: string, ): void { - pending.set(parentToolUseId, model); - if (pending.size > PENDING_TASK_MODEL_CAP) { + pending.set(toolUseId, value); + if (pending.size > PENDING_TASK_ENTRY_CAP) { const oldest = pending.keys().next(); if (!oldest.done) { pending.delete(oldest.value); @@ -434,9 +434,16 @@ interface ClaudeSessionContext { /** * Authoritative subagent models from assistant snapshots that arrived before * their task_started registered the task, keyed by parent_tool_use_id. - * Written through `rememberPendingTaskModel`, consumed by task_started. + * Written through `rememberPendingTaskEntry`, consumed by task_started. */ readonly pendingTaskModels: Map; + /** + * tool_use_id → parent_tool_use_id for tool calls made inside a subagent. + * The SDK forwards those calls only as assistant snapshots, never as stream + * events, so task_started reads this to find the owner of a task launched + * from inside a subagent. + */ + readonly subagentToolParents: Map; /** * Last emitted workflow-member fingerprint per member slot. A coordinator * task_progress repeats the FULL member array every tick; without a @@ -3370,18 +3377,45 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const owningTaskId = agentIdForParentToolUse(context.taskAgents, assistantParentToolUseId); const snapshotModel = trimmedString(message.message.model); const owningAgent = owningTaskId ? context.taskAgents.get(owningTaskId) : undefined; - if (snapshotModel) { - if (owningAgent) { + const snapshotContent = Array.isArray(message.message.content) ? message.message.content : []; + for (const block of snapshotContent) { + if (block.type === "tool_use") { + rememberPendingTaskEntry(context.subagentToolParents, block.id, assistantParentToolUseId); + } + } + if (snapshotModel && owningAgent) { + if (owningAgent.model !== snapshotModel) { owningAgent.model = snapshotModel; - } else { - // The snapshot beat its task_started (or its tool_use_id was never - // recorded): hold the model until the task registers. - rememberPendingTaskModel( - context.pendingTaskModels, - assistantParentToolUseId, - snapshotModel, - ); + // Push the correction now. The next task row can be minutes away + // when the subagent blocks on a long foreground tool. + const stamp = yield* makeEventStamp(); + yield* offerRuntimeEvent({ + type: "task.updated", + eventId: stamp.eventId, + provider: PROVIDER, + createdAt: stamp.createdAt, + threadId: context.session.threadId, + ...(context.turnState ? { turnId: asCanonicalTurnId(context.turnState.turnId) } : {}), + payload: { + taskId: RuntimeTaskId.make(owningAgent.taskId), + ...taskLinkageFor(context.taskAgents, owningAgent.taskId), + }, + providerRefs: nativeProviderRefs(context), + raw: { + source: "claude.sdk.message", + method: "claude/assistant", + payload: message, + }, + }); } + } else if (snapshotModel) { + // The snapshot beat its task_started (or its tool_use_id was never + // recorded): hold the model until the task registers. + rememberPendingTaskEntry( + context.pendingTaskModels, + assistantParentToolUseId, + snapshotModel, + ); } context.lastAssistantUuid = message.uuid; yield* updateResumeCursor(context); @@ -3735,15 +3769,23 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( }); return; case "task_started": { - // A task launched by a tool that itself ran inside a subagent (the - // in-flight tool carries agentId from parent_tool_use_id) is - // agent-internal: a subagent's background shell, not parent work. - const launchingTool = message.tool_use_id - ? Array.from(context.inFlightTools.values()).find( - (tool) => tool.itemId === message.tool_use_id, - ) + // A task launched by a tool that itself ran inside a subagent is + // agent-internal: a subagent's background shell, not parent work. A + // streamed tool carries agentId from parent_tool_use_id; a tool seen + // only in a subagent snapshot resolves through subagentToolParents. + const toolUseId = message.tool_use_id; + const launchingTool = toolUseId + ? Array.from(context.inFlightTools.values()).find((tool) => tool.itemId === toolUseId) : undefined; - const owningAgentId = launchingTool?.agentId; + const launchParentToolUseId = toolUseId + ? context.subagentToolParents.get(toolUseId) + : undefined; + if (toolUseId) { + context.subagentToolParents.delete(toolUseId); + } + const owningAgentId = + launchingTool?.agentId ?? + agentIdForParentToolUse(context.taskAgents, launchParentToolUseId); if ( context.turnState && classifyTaskAgentKind({ @@ -3754,13 +3796,12 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( context.turnState.hasSubagents = true; } // Model/effort: the Agent tool's input carries explicit overrides; - // absent ones inherit the session's selection (SDK behavior). - // Subagent assistant snapshots refine model with the authoritative API - // id: one that already arrived is buffered and outranks the seed here, - // later ones refine the record in place. AgentInput.effort may be a - // named level or an integer. + // absent ones inherit the owning subagent's model, else the session's + // selection (SDK behavior). Subagent assistant snapshots refine model + // with the authoritative API id: one that already arrived is buffered + // and outranks the seed here, later ones refine the record in place. + // AgentInput.effort may be a named level or an integer. const launchInput = launchingTool?.input; - const toolUseId = message.tool_use_id; const bufferedModel = toolUseId ? context.pendingTaskModels.get(toolUseId) : undefined; if (toolUseId) { context.pendingTaskModels.delete(toolUseId); @@ -3768,6 +3809,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const model = bufferedModel ?? trimmedString(launchInput?.model) ?? + (owningAgentId ? context.taskAgents.get(owningAgentId)?.model : undefined) ?? trimmedString(context.session.model ?? undefined); const rawLaunchEffort = launchInput?.effort; const effort = @@ -4471,6 +4513,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( const claudeTasks = new Map(); const taskAgents = new Map(); const pendingTaskModels = new Map(); + const subagentToolParents = new Map(); const workflowMemberFingerprints = new Map(); const liveTaskIds = new Set(); @@ -5078,6 +5121,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( claudeTasks, taskAgents, pendingTaskModels, + subagentToolParents, workflowMemberFingerprints, liveTaskIds, turnState: undefined,