From 9f2a0f1be7cf795fd4500abf30b53b18e21745a2 Mon Sep 17 00:00:00 2001 From: Serhii Vecherenko Date: Fri, 24 Jul 2026 10:41:42 -0700 Subject: [PATCH] fix(qwen): preserve background subagent completion progress --- src/supervisor/agents/acp/canonicalMapping.ts | 1 + .../agents/acp/canonicalMapping/dispatch.ts | 31 ++++-- .../agents/acp/canonicalMapping/subagents.ts | 1 + .../agents/acp/subagentCoordinator.ts | 7 ++ .../agents/qwen/acpTransform.test.ts | 99 ++++++++++++++++++- src/supervisor/agents/qwen/acpTransform.ts | 7 +- 6 files changed, 133 insertions(+), 13 deletions(-) diff --git a/src/supervisor/agents/acp/canonicalMapping.ts b/src/supervisor/agents/acp/canonicalMapping.ts index 0e222f85..0063f927 100644 --- a/src/supervisor/agents/acp/canonicalMapping.ts +++ b/src/supervisor/agents/acp/canonicalMapping.ts @@ -27,6 +27,7 @@ export { getDetachedSubAgentToolCallIdForNotification, PORACODE_ACP_NEW_ASSISTANT_ITEM_META_KEY, PORACODE_ACP_PARENT_TOOL_CALL_ID_META_KEY, + PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY, PORACODE_ACP_TOP_LEVEL_TOOL_CALL_META_KEY, } from "./canonicalMapping/subagents"; export { mapAcpSessionUpdate } from "./canonicalMapping/dispatch"; diff --git a/src/supervisor/agents/acp/canonicalMapping/dispatch.ts b/src/supervisor/agents/acp/canonicalMapping/dispatch.ts index 069e6f16..cc7d9ed8 100644 --- a/src/supervisor/agents/acp/canonicalMapping/dispatch.ts +++ b/src/supervisor/agents/acp/canonicalMapping/dispatch.ts @@ -31,6 +31,7 @@ import { isUpdateTopicTool, PORACODE_ACP_DETACHED_SUBAGENT_META_KEY, PORACODE_ACP_NEW_ASSISTANT_ITEM_META_KEY, + PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY, removeActiveSubAgent, selectActiveSubAgentForToolCall, tagSubAgentChildStarts, @@ -377,6 +378,7 @@ export function mapAcpSessionUpdate( const hasOpenSubAgentContent = item.isSubAgent && isTerminal && + updateMeta?.[PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY] !== true && (state.openAssistantItemId !== undefined || state.openReasoningItemId !== undefined); const subAgentProgress = item.isSubAgent && !hasOpenSubAgentContent @@ -401,20 +403,29 @@ export function mapAcpSessionUpdate( if (isTerminal && item.isSubAgent) { events.push(...closeOpenContentItems(state)); } - events.push({ + const parentEvent: RuntimeEvent = { type: isTerminal ? "item.completed" : "item.updated", threadId, itemId: item.itemId, payload: emittedPayload, - }); - if (subAgentProgress?.text) { - events.push(...buildSubAgentProgressEvents(state, item, subAgentProgress.text, isTerminal)); - } else if (isTerminal && item.subAgentProgressItemId) { - events.push({ - type: "item.completed", - threadId, - itemId: item.subAgentProgressItemId, - }); + }; + const progressEvents = subAgentProgress?.text + ? buildSubAgentProgressEvents(state, item, subAgentProgress.text, isTerminal) + : isTerminal && item.subAgentProgressItemId + ? [ + { + type: "item.completed" as const, + threadId, + itemId: item.subAgentProgressItemId, + }, + ] + : []; + // Child transcript events must precede the terminal parent event. The + // runtime router drains buffered children when the parent completes. + if (isTerminal) { + events.push(...progressEvents, parentEvent); + } else { + events.push(parentEvent, ...progressEvents); } if (isTerminal) { state.toolCallItems.delete(toolCall.toolCallId); diff --git a/src/supervisor/agents/acp/canonicalMapping/subagents.ts b/src/supervisor/agents/acp/canonicalMapping/subagents.ts index bc9001dd..1ade6f62 100644 --- a/src/supervisor/agents/acp/canonicalMapping/subagents.ts +++ b/src/supervisor/agents/acp/canonicalMapping/subagents.ts @@ -22,6 +22,7 @@ export const PORACODE_ACP_TOP_LEVEL_TOOL_CALL_META_KEY = "poracodeTopLevelToolCa export const PORACODE_ACP_DETACHED_SUBAGENT_META_KEY = "poracodeDetachedSubAgent"; export const PORACODE_ACP_DETACHED_SUBAGENT_ACTIVITY_META_KEY = "poracodeDetachedSubAgentActivity"; export const PORACODE_ACP_NEW_ASSISTANT_ITEM_META_KEY = "poracodeNewAssistantItem"; +export const PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY = "poracodeSynthesizeSubAgentResult"; export function buildSubAgentProgress( toolCall: { diff --git a/src/supervisor/agents/acp/subagentCoordinator.ts b/src/supervisor/agents/acp/subagentCoordinator.ts index 136d5b77..0fa773fc 100644 --- a/src/supervisor/agents/acp/subagentCoordinator.ts +++ b/src/supervisor/agents/acp/subagentCoordinator.ts @@ -13,6 +13,7 @@ import { PORACODE_ACP_DETACHED_SUBAGENT_META_KEY, PORACODE_ACP_NEW_ASSISTANT_ITEM_META_KEY, PORACODE_ACP_PARENT_TOOL_CALL_ID_META_KEY, + PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY, PORACODE_ACP_TOP_LEVEL_TOOL_CALL_META_KEY, } from "./canonicalMapping/subagents"; @@ -52,6 +53,7 @@ export interface AcpSubagentCompletionInput { childOutput?: string; parentReply?: string; terminalMeta?: Record; + synthesizeResultProgress?: boolean; } export interface AcpSubagentCoordinator { @@ -170,6 +172,7 @@ export function createAcpSubagentCoordinator(): AcpSubagentCoordinator { ...(input.childOutput ? { childOutput: input.childOutput } : {}), ...(input.parentReply ? { parentReply: input.parentReply } : {}), ...(input.terminalMeta ? { terminalMeta: input.terminalMeta } : {}), + ...(input.synthesizeResultProgress ? { synthesizeResultProgress: true } : {}), }); forgetCall(toolCallId); return notifications; @@ -248,6 +251,7 @@ function createAcpSubagentCompletionNotifications(input: { childOutput?: string; parentReply?: string; terminalMeta?: Record; + synthesizeResultProgress?: boolean; }): SessionNotification[] { const notifications: SessionNotification[] = []; if (input.childOutput) { @@ -281,6 +285,9 @@ function createAcpSubagentCompletionNotifications(input: { _meta: { ...input.terminalMeta, [PORACODE_ACP_DETACHED_SUBAGENT_ACTIVITY_META_KEY]: input.toolCallId, + ...(input.synthesizeResultProgress + ? { [PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY]: true } + : {}), }, }), ); diff --git a/src/supervisor/agents/qwen/acpTransform.test.ts b/src/supervisor/agents/qwen/acpTransform.test.ts index 3e724eea..cd69d9b7 100644 --- a/src/supervisor/agents/qwen/acpTransform.test.ts +++ b/src/supervisor/agents/qwen/acpTransform.test.ts @@ -1,9 +1,12 @@ import { describe, expect, it } from "vitest"; import type { SessionNotification } from "@agentclientprotocol/sdk"; import { + createAcpMapperState, + mapAcpSessionUpdate, PORACODE_ACP_DETACHED_SUBAGENT_ACTIVITY_META_KEY, PORACODE_ACP_GOAL_META_KEY, PORACODE_ACP_PARENT_TOOL_CALL_ID_META_KEY, + PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY, PORACODE_ACP_TOP_LEVEL_TOOL_CALL_META_KEY, } from "../acp/canonicalMapping"; import { createQwenAcpSessionBridge, createQwenAcpSessionUpdateTransform } from "./acpTransform"; @@ -200,7 +203,7 @@ describe("createQwenAcpSessionUpdateTransform", () => { type: "content", content: { type: "text", - text: "Background agent launched successfully.\nagentId: Explore-abcd1234 (internal ID)", + text: "Background agent launched successfully.\ntask_id: Explore-abcd1234 (internal ID)", }, }, ], @@ -270,10 +273,104 @@ describe("createQwenAcpSessionUpdateTransform", () => { _meta: { usage: { totalTokens: 42 }, [PORACODE_ACP_DETACHED_SUBAGENT_ACTIVITY_META_KEY]: "agent-bg", + [PORACODE_ACP_SYNTHESIZE_SUBAGENT_RESULT_META_KEY]: true, }, }); }); + it("maps the Qwen 0.21 background result into a child transcript before completion", () => { + const bridge = createQwenAcpSessionBridge(); + const state = createAcpMapperState("qwen-thread"); + const mapUpdate = (update: Record) => + mapAcpSessionUpdate(bridge.sessionUpdateTransform(note(update)), state); + + const started = mapUpdate({ + sessionUpdate: "tool_call", + toolCallId: "agent-real", + title: "Agent: Inspect ACP mapping", + status: "pending", + _meta: { toolName: "agent" }, + }); + const parentItemId = ( + started.find((event) => event.type === "item.started") as { + itemId: string; + } + ).itemId; + + mapUpdate({ + sessionUpdate: "tool_call_update", + toolCallId: "agent-real", + status: "completed", + content: [ + { + type: "content", + content: { + type: "text", + text: "Background agent launched successfully.\ntask_id: Explore-real123 (internal ID)", + }, + }, + ], + rawOutput: { + status: "background", + subagentName: "Explore", + taskDescription: "Inspect ACP mapping", + }, + _meta: { toolName: "agent" }, + }); + mapUpdate({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: 'Background agent "Explore" completed.' }, + _meta: { + source: "background_notification", + backgroundTask: { taskId: "Explore-real123", status: "completed" }, + }, + }); + mapUpdate({ + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "The background agent found the mapper." }, + _meta: { + source: "background_notification_response", + backgroundTask: { taskId: "Explore-real123", status: "completed" }, + }, + }); + + const boundary = bridge.extensionSessionUpdateTransform("_qwencode/end_turn", { + sessionId: "qwen-session", + reason: "end_turn", + source: "background_notification", + }); + const terminalEvents = mapAcpSessionUpdate(bridge.sessionUpdateTransform(boundary!), state); + const childStart = terminalEvents.find( + (event) => event.type === "item.started" && event.parentItemId === parentItemId, + ); + expect(childStart).toMatchObject({ + type: "item.started", + itemType: "assistant_message", + parentItemId, + }); + expect(terminalEvents).toContainEqual( + expect.objectContaining({ + type: "content.delta", + itemId: childStart && "itemId" in childStart ? childStart.itemId : "", + stream: "assistant_text", + delta: "The background agent found the mapper.", + }), + ); + const childCompletedIndex = terminalEvents.findIndex( + (event) => + event.type === "item.completed" && + childStart && + "itemId" in childStart && + event.itemId === childStart.itemId, + ); + const parentCompletedIndex = terminalEvents.findIndex( + (event) => event.type === "item.completed" && event.itemId === parentItemId, + ); + expect(childCompletedIndex).toBeGreaterThanOrEqual(0); + expect(parentCompletedIndex).toBeGreaterThan(childCompletedIndex); + expect(state.activeSubAgents).toEqual([]); + }); + it("uses Qwen's real background end-turn extension as the terminal boundary", () => { const bridge = createQwenAcpSessionBridge(); const transform = bridge.sessionUpdateTransform; diff --git a/src/supervisor/agents/qwen/acpTransform.ts b/src/supervisor/agents/qwen/acpTransform.ts index 56139432..7dd0f2ef 100644 --- a/src/supervisor/agents/qwen/acpTransform.ts +++ b/src/supervisor/agents/qwen/acpTransform.ts @@ -22,7 +22,9 @@ import { } from "../acp/subagentCoordinator"; import type { AcpSessionUpdateTransform } from "../base"; -const BACKGROUND_AGENT_ID_RE = /\bagentId:\s*([^\s(]+)/iu; +// Qwen 0.21 renamed the launch result field from `agentId` to `task_id`. +// Accept both so resumed sessions created by older CLIs still correlate. +const BACKGROUND_TASK_ID_RE = /\b(?:agentId|task_id):\s*([^\s(]+)/iu; const QWEN_BACKGROUND_END_TURN_METHOD = "_qwencode/end_turn"; export interface QwenAcpSessionBridge { @@ -203,6 +205,7 @@ export function createQwenAcpSessionUpdateTransform(): AcpSessionUpdateTransform status: "completed", ...(result ? { result } : {}), terminalMeta: plainRecord(update._meta), + synthesizeResultProgress: true, }) .at(-1) ?? notification; const pausedGoal = pausedGoalUpdate(); @@ -244,7 +247,7 @@ function extractBackgroundTaskId(content: unknown): string | undefined { for (const entry of content) { const contentRecord = plainRecord(plainRecord(entry).content); const text = readString(contentRecord, "text"); - const taskId = text ? BACKGROUND_AGENT_ID_RE.exec(text)?.[1] : undefined; + const taskId = text ? BACKGROUND_TASK_ID_RE.exec(text)?.[1] : undefined; if (taskId) return taskId; } return undefined;