authorgravatar for 155588579+spiky02plateau@users.noreply.github.comspiky02plateau <155588579+spiky02plateau@users.noreply.github.com> 2026-08-24 12:21:43+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-08-24 03:21:43-07:00
log6a2608292dd7ba01516112c276e475e6b64e4e2c
tree6edbc27d537efed7e74216632f95e5a496c55ec8
parenta00218741f4fddb77fb2f0e37f548a798bed023a
signature Signed by PGP key B5690EEEBB952194

fix(server): keep the authoritative subagent model when snapshots race task_started (#7583)

2 files changed, 136 insertions(+), 6 deletions(-)

apps/server/src/provider/Layers/ClaudeAdapter.test.ts+78
......@@ -1997,6 +1997,84 @@ describe("ClaudeAdapterLive", () => {
19971997 );
19981998 });
19991999
2000 it.effect("a subagent snapshot that beats task_started still wins over the seed", () => {
2001 const harness = makeHarness();
2002 return Effect.gen(function* () {
2003 const adapter = yield* ClaudeAdapter;
2004
2005 const taskEventsFiber = yield* adapter.streamEvents.pipe(
2006 Stream.filter((event) => event.type.startsWith("task.")),
2007 Stream.take(2),
2008 Stream.runCollect,
2009 Effect.forkChild,
2010 );
2011
2012 const session = yield* adapter.startSession({
2013 threadId: THREAD_ID,
2014 provider: ProviderDriverKind.make("claudeAgent"),
2015 modelSelection: createModelSelection(
2016 ProviderInstanceId.make("claudeAgent"),
2017 "claude-opus-4-6",
2018 [{ id: "effort", value: "max" }],
2019 ),
2020 runtimeMode: "full-access",
2021 });
2022 yield* adapter.sendTurn({
2023 threadId: session.threadId,
2024 input: "spawn an agent",
2025 attachments: [],
2026 });
2027
2028 // The subagent streams its first assistant snapshot before the task is
2029 // registered, so there is no agent to refine yet.
2030 harness.query.emit({
2031 type: "assistant",
2032 parent_tool_use_id: "toolu_agent_early",
2033 message: {
2034 model: "claude-sonnet-5[1m]",
2035 content: [],
2036 },
2037 uuid: "early-snapshot-uuid",
2038 session_id: "sdk-session",
2039 } as unknown as SDKMessage);
2040 harness.query.emit({
2041 type: "system",
2042 subtype: "task_started",
2043 task_id: "task-early",
2044 description: "Agent E",
2045 task_type: "local_agent",
2046 tool_use_id: "toolu_agent_early",
2047 uuid: "task-early-uuid",
2048 session_id: "sdk-session",
2049 } as unknown as SDKMessage);
2050 harness.query.emit({
2051 type: "system",
2052 subtype: "task_progress",
2053 task_id: "task-early",
2054 description: "Agent E",
2055 usage: { total_tokens: 100, tool_uses: 1, duration_ms: 10 },
2056 uuid: "task-early-progress-uuid",
2057 session_id: "sdk-session",
2058 } as unknown as SDKMessage);
2059
2060 const taskEvents = Array.from(yield* Fiber.join(taskEventsFiber));
2061 const started = taskEvents[0];
2062 assert.equal(started?.type, "task.started");
2063 if (started?.type === "task.started") {
2064 assert.equal(started.payload.model, "claude-sonnet-5[1m]");
2065 assert.equal(started.payload.effort, "max");
2066 }
2067 const progress = taskEvents[1];
2068 assert.equal(progress?.type, "task.progress");
2069 if (progress?.type === "task.progress") {
2070 assert.equal(progress.payload.model, "claude-sonnet-5[1m]");
2071 }
2072 }).pipe(
2073 Effect.provideService(Random.Random, makeDeterministicRandomService()),
2074 Effect.provide(harness.layer),
2075 );
2076 });
2077
20002078 it.effect("closes the session when the Claude stream aborts after a turn starts", () => {
20012079 const harness = makeHarness();
20022080 return Effect.gen(function* () {
apps/server/src/provider/Layers/ClaudeAdapter.ts+58-6
......@@ -241,6 +241,32 @@ interface ClaudeTaskAgentState {
241241 effort: string | undefined;
242242}
243243
244/**
245 * How many racing snapshot models to buffer per session. A snapshot whose
246 * task_started never arrives would otherwise pin its entry for the session's
247 * lifetime; oldest entries evict first.
248 */
249const PENDING_TASK_MODEL_CAP = 64;
250
251/**
252 * Buffers a subagent snapshot's authoritative model under its
253 * parent_tool_use_id, for snapshots that beat their task_started to the
254 * stream. task_started consumes the entry when it registers the task.
255 */
256function rememberPendingTaskModel(
257 pending: Map<string, string>,
258 parentToolUseId: string,
259 model: string,
260): void {
261 pending.set(parentToolUseId, model);
262 if (pending.size > PENDING_TASK_MODEL_CAP) {
263 const oldest = pending.keys().next();
264 if (!oldest.done) {
265 pending.delete(oldest.value);
266 }
267 }
268}
269
244270interface ClaudeSessionContext {
245271 session: ProviderSession;
246272 readonly promptQueue: Queue.Queue<PromptQueueItem>;
......@@ -262,6 +288,12 @@ interface ClaudeSessionContext {
262288 readonly inFlightTools: Map<number, ToolInFlight>;
263289 readonly claudeTasks: Map<string, ClaudeTaskState>;
264290 readonly taskAgents: Map<string, ClaudeTaskAgentState>;
291 /**
292 * Authoritative subagent models from assistant snapshots that arrived before
293 * their task_started registered the task, keyed by parent_tool_use_id.
294 * Written through `rememberPendingTaskModel`, consumed by task_started.
295 */
296 readonly pendingTaskModels: Map<string, string>;
265297 /**
266298 * Last emitted workflow-member fingerprint per member slot. A coordinator
267299 * task_progress repeats the FULL member array every tick; without a
......@@ -2875,8 +2907,18 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
28752907 const owningTaskId = agentIdForParentToolUse(context.taskAgents, assistantParentToolUseId);
28762908 const snapshotModel = trimmedString(message.message.model);
28772909 const owningAgent = owningTaskId ? context.taskAgents.get(owningTaskId) : undefined;
2878 if (owningAgent && snapshotModel) {
2879 owningAgent.model = snapshotModel;
2910 if (snapshotModel) {
2911 if (owningAgent) {
2912 owningAgent.model = snapshotModel;
2913 } else {
2914 // The snapshot beat its task_started (or its tool_use_id was never
2915 // recorded): hold the model until the task registers.
2916 rememberPendingTaskModel(
2917 context.pendingTaskModels,
2918 assistantParentToolUseId,
2919 snapshotModel,
2920 );
2921 }
28802922 }
28812923 context.lastAssistantUuid = message.uuid;
28822924 yield* updateResumeCursor(context);
......@@ -3190,12 +3232,20 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
31903232 const owningAgentId = launchingTool?.agentId;
31913233 // Model/effort: the Agent tool's input carries explicit overrides;
31923234 // absent ones inherit the session's selection (SDK behavior).
3193 // Subagent assistant snapshots later refine model with the
3194 // authoritative API id. AgentInput.effort may be a named level or an
3195 // integer.
3235 // Subagent assistant snapshots refine model with the authoritative API
3236 // id: one that already arrived is buffered and outranks the seed here,
3237 // later ones refine the record in place. AgentInput.effort may be a
3238 // named level or an integer.
31963239 const launchInput = launchingTool?.input;
3240 const toolUseId = message.tool_use_id;
3241 const bufferedModel = toolUseId ? context.pendingTaskModels.get(toolUseId) : undefined;
3242 if (toolUseId) {
3243 context.pendingTaskModels.delete(toolUseId);
3244 }
31973245 const model =
3198 trimmedString(launchInput?.model) ?? trimmedString(context.session.model ?? undefined);
3246 bufferedModel ??
3247 trimmedString(launchInput?.model) ??
3248 trimmedString(context.session.model ?? undefined);
31993249 const rawLaunchEffort = launchInput?.effort;
32003250 const effort =
32013251 trimmedString(rawLaunchEffort) ??
......@@ -3807,6 +3857,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
38073857 const inFlightTools = new Map<number, ToolInFlight>();
38083858 const claudeTasks = new Map<string, ClaudeTaskState>();
38093859 const taskAgents = new Map<string, ClaudeTaskAgentState>();
3860 const pendingTaskModels = new Map<string, string>();
38103861 const workflowMemberFingerprints = new Map<string, string>();
38113862 const liveTaskIds = new Set<string>();
38123863
......@@ -4270,6 +4321,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
42704321 inFlightTools,
42714322 claudeTasks,
42724323 taskAgents,
4324 pendingTaskModels,
42734325 workflowMemberFingerprints,
42744326 liveTaskIds,
42754327 turnState: undefined,