Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 23 additions & 13 deletions apps/server/src/provider/Layers/ClaudeAdapter.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
);
Expand Down Expand Up @@ -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);

Expand All @@ -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()),
Expand Down
116 changes: 80 additions & 36 deletions apps/server/src/provider/Layers/ClaudeAdapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<string, string>,
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);
Expand Down Expand Up @@ -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<string, string>;
/**
* 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<string, string>;
/**
* Last emitted workflow-member fingerprint per member slot. A coordinator
* task_progress repeats the FULL member array every tick; without a
Expand Down Expand Up @@ -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);
Comment thread
t3dotgg marked this conversation as resolved.
}
}
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);
Expand Down Expand Up @@ -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({
Expand All @@ -3754,20 +3796,20 @@ 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);
}
const model =
bufferedModel ??
trimmedString(launchInput?.model) ??
(owningAgentId ? context.taskAgents.get(owningAgentId)?.model : undefined) ??
trimmedString(context.session.model ?? undefined);
const rawLaunchEffort = launchInput?.effort;
const effort =
Expand Down Expand Up @@ -4471,6 +4513,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
const claudeTasks = new Map<string, ClaudeTaskState>();
const taskAgents = new Map<string, ClaudeTaskAgentState>();
const pendingTaskModels = new Map<string, string>();
const subagentToolParents = new Map<string, string>();
const workflowMemberFingerprints = new Map<string, string>();
const liveTaskIds = new Set<string>();

Expand Down Expand Up @@ -5078,6 +5121,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
claudeTasks,
taskAgents,
pendingTaskModels,
subagentToolParents,
workflowMemberFingerprints,
liveTaskIds,
turnState: undefined,
Expand Down
Loading