From a284e5bb8e9e0de058ce39a1c600829048162959 Mon Sep 17 00:00:00 2001 From: root Date: Thu, 27 Aug 2026 13:25:58 +0800 Subject: [PATCH 1/2] fix(core): report teammate result settled before event bridge attach TeamManager.spawnTeammate awaits backend.spawnAgent() before attaching the event bridge, and AgentEventEmitter does not buffer events for late subscribers. If the in-process initial round settles (final round text plus IDLE) while spawnAgent() is still resolving, both events are lost: the leader never receives the initial result, and the existing idle reconciliation only flushes pending messages (#10211). At bridge-attach time, recover the last model-visible answer from the agent's message history (AgentCore appends one assistant message per ROUND_TEXT) and seed pendingFinalReports with it. When the agent already settled IDLE before attach, replay the missed IDLE STATUS_CHANGE through the existing handler so the result is reported and pending messages flush exactly once. Without pre-attach round text there is no completed round to report, so the previous flush-only behavior is kept. Co-authored-by: Qwen-Coder --- packages/core/src/agents/backends/types.ts | 8 + .../team/TeamManager.initial-result.test.ts | 172 ++++++++++++++++++ packages/core/src/agents/team/TeamManager.ts | 52 +++++- .../src/agents/team/test-utils/fake-agent.ts | 31 ++++ 4 files changed, 260 insertions(+), 3 deletions(-) create mode 100644 packages/core/src/agents/team/TeamManager.initial-result.test.ts diff --git a/packages/core/src/agents/backends/types.ts b/packages/core/src/agents/backends/types.ts index 8c3e5a80057..fdea9fa2493 100644 --- a/packages/core/src/agents/backends/types.ts +++ b/packages/core/src/agents/backends/types.ts @@ -14,6 +14,7 @@ import type { Content } from '@google/genai'; import type { AnsiOutput } from '../../utils/terminalSerializer.js'; import type { + AgentMessage, AgentStatus, PromptConfig, ModelConfig, @@ -142,6 +143,13 @@ export interface TeamAgentHandle { * handles may not. */ getError?(): string | undefined; + /** + * Conversation message history. Optional: in-process handles + * (AgentInteractive, FakeAgent) provide it; PTY handles don't. + * TeamManager reads it to recover round text emitted before its + * event bridge attached. + */ + getMessages?(): readonly AgentMessage[]; } /** diff --git a/packages/core/src/agents/team/TeamManager.initial-result.test.ts b/packages/core/src/agents/team/TeamManager.initial-result.test.ts new file mode 100644 index 00000000000..ac75e591d41 --- /dev/null +++ b/packages/core/src/agents/team/TeamManager.initial-result.test.ts @@ -0,0 +1,172 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +/** + * Regression tests for #10211: the initial teammate round can complete + * (final round text + IDLE) before TeamManager.setupEventBridge attaches, + * because spawnTeammate awaits backend.spawnAgent() first and the emitter + * does not buffer events for late subscribers. The leader must still + * receive the initial round's result exactly once. + */ + +import { describe, it, expect, vi, afterEach } from 'vitest'; +import { TeamCoordinationHarness } from './test-utils/coordination-harness.js'; +import { Storage } from '../../config/storage.js'; +import { AgentStatus } from '../runtime/agent-types.js'; +import { AgentEventType } from '../runtime/agent-events.js'; + +vi.mock('../../config/storage.js', async (importOriginal) => { + const original = + await importOriginal(); + let mockGlobalDir = ''; + return { + ...original, + Storage: { + ...original.Storage, + getGlobalQwenDir: () => mockGlobalDir, + __setMockGlobalDir: (dir: string) => { + mockGlobalDir = dir; + }, + }, + }; +}); + +function setMockDir(dir: string): void { + ( + Storage as unknown as { + __setMockGlobalDir: (d: string) => void; + } + ).__setMockGlobalDir(dir); +} + +/** Let queued fire-and-forget coordination work settle. */ +async function settleAsyncWork(): Promise { + await new Promise((resolve) => setTimeout(resolve, 50)); +} + +describe('initial teammate result before event bridge attachment (#10211)', () => { + let harness: TeamCoordinationHarness | undefined; + + afterEach(async () => { + await harness?.cleanup(); + harness = undefined; + }); + + async function createHarness(): Promise { + const h = await TeamCoordinationHarness.create(); + setMockDir(h.tmpDir); + harness = h; + return h; + } + + it('reports final text to the leader when the initial round completes before spawnAgent resolves', async () => { + const h = await createHarness(); + + // onStart runs inside FakeBackend.spawnAgent(), i.e. before + // TeamManager.setupEventBridge() subscribes. Emits the final round + // text and settles IDLE while spawnAgent() is still resolving — + // the in-process race from the issue. + await h.spawnTeammate('worker', { + onStart: (agent) => { + agent.setStatus(AgentStatus.RUNNING); + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 1, + text: 'initial result', + thoughtText: '', + timestamp: Date.now(), + }); + agent.setStatus(AgentStatus.IDLE); + }, + }); + + await vi.waitFor(async () => { + expect(await h.teamManager.getLeaderMessages()).toEqual([ + expect.objectContaining({ + from: 'worker', + text: 'initial result', + }), + ]); + }); + + // Exactly once: after coordination settles, no duplicate report + // for the same round may arrive. + await settleAsyncWork(); + expect(await h.teamManager.getLeaderMessages()).toEqual([]); + }); + + it('does not re-report the initial result when a later round completes live', async () => { + const h = await createHarness(); + + await h.spawnTeammate('worker', { + onStart: (agent) => { + agent.setStatus(AgentStatus.RUNNING); + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 1, + text: 'initial result', + thoughtText: '', + timestamp: Date.now(), + }); + agent.setStatus(AgentStatus.IDLE); + }, + onMessage: (_message, agent) => { + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 2, + text: 'follow-up result', + thoughtText: '', + timestamp: Date.now(), + }); + }, + }); + + await vi.waitFor(async () => { + expect(await h.teamManager.getLeaderMessages()).toEqual([ + expect.objectContaining({ text: 'initial result' }), + ]); + }); + + await h.teamManager.sendMessage('worker', 'next task', 'leader'); + await h.waitForStatus('worker', AgentStatus.IDLE); + + // The live second round reports its own text exactly once; the + // pre-attach initial result must not be reported again. + await vi.waitFor(async () => { + expect(await h.teamManager.getLeaderMessages()).toEqual([ + expect.objectContaining({ text: 'follow-up result' }), + ]); + }); + + await settleAsyncWork(); + expect(await h.teamManager.getLeaderMessages()).toEqual([]); + }); + + it('does not report anything at spawn when no round ran before the bridge attached', async () => { + const h = await createHarness(); + + // Plain spawn: the harness agent is IDLE at attach time but never + // emitted round text (no pre-attach round). The attach-time + // reconciliation must not invent a report. + await h.spawnTeammate('worker'); + await settleAsyncWork(); + + expect(await h.teamManager.getLeaderMessages()).toEqual([]); + + // The live path still works afterwards. + await h.teamManager.sendMessage('worker', 'task', 'leader'); + await h.waitForStatus('worker', AgentStatus.IDLE); + await vi.waitFor(async () => { + expect(await h.teamManager.getLeaderMessages()).toEqual([ + expect.objectContaining({ + text: expect.stringContaining( + 'completed a turn without a model-visible final answer', + ), + }), + ]); + }); + }); +}); diff --git a/packages/core/src/agents/team/TeamManager.ts b/packages/core/src/agents/team/TeamManager.ts index d012a7cfe2b..3293c335333 100644 --- a/packages/core/src/agents/team/TeamManager.ts +++ b/packages/core/src/agents/team/TeamManager.ts @@ -1731,10 +1731,38 @@ export class TeamManager { emitter.off(AgentEventType.TOOL_WAITING_APPROVAL, onApproval); }); - // Reconcile: if agent already reached IDLE before we - // attached, flush now. + // Reconcile state reached before we attached. The emitter does + // not buffer for late subscribers, and the in-process run loop + // can settle the initial round while spawnAgent() is still + // resolving — those events never reach the bridge. const currentStatus = agent.getStatus(); - if (currentStatus === AgentStatus.IDLE) { + + // Round text emitted before attach survives only in the agent's + // message history (AgentCore appends an assistant message per + // ROUND_TEXT). Recover the last model-visible answer — mirroring + // onRoundText's last-non-empty-text-wins semantics — so the + // settlement below reports it instead of the no-visible-answer + // fallback. Live ROUND_TEXT events after attach overwrite this + // seed as usual; RUNNING/terminal handlers clear it like any + // pending report. + const preAttachReport = this.lastVisibleAnswer(agent); + if (preAttachReport !== undefined) { + this.pendingFinalReports.set(agentId, preAttachReport); + } + + if (currentStatus === AgentStatus.IDLE && preAttachReport !== undefined) { + // The initial round already settled to IDLE before attach. + // Replay the STATUS_CHANGE through the same handler the live + // path uses so its final report and message flush happen + // exactly once. Without pre-attach round text there is no + // completed round to report — keep the flush-only behavior. + onStatusChange({ + agentId, + previousStatus: AgentStatus.RUNNING, + newStatus: AgentStatus.IDLE, + timestamp: Date.now(), + } as AgentStatusChangeEvent); + } else if (currentStatus === AgentStatus.IDLE) { this.fireAndForget( `flushNextMessage(${agentId})`, this.flushNextMessage(agentId, agentName), @@ -1754,6 +1782,24 @@ export class TeamManager { } } + /** + * The last model-visible answer in an agent handle's message + * history, or undefined when there is none. Mirrors the live + * ROUND_TEXT → pendingFinalReports semantics: the most recent + * non-empty, non-thought assistant text wins. + */ + private lastVisibleAnswer(agent: TeamAgentHandle): string | undefined { + const messages = agent.getMessages?.(); + if (!messages) return undefined; + for (let i = messages.length - 1; i >= 0; i--) { + const message = messages[i]; + if (message.role !== 'assistant' || message.thought) continue; + const text = message.content.trim(); + if (text) return text; + } + return undefined; + } + // ─── Private: Permission fallback ─────────────────────── /** diff --git a/packages/core/src/agents/team/test-utils/fake-agent.ts b/packages/core/src/agents/team/test-utils/fake-agent.ts index e802b82cb38..7ba05fa47a8 100644 --- a/packages/core/src/agents/team/test-utils/fake-agent.ts +++ b/packages/core/src/agents/team/test-utils/fake-agent.ts @@ -17,7 +17,9 @@ import { AgentEventEmitter, AgentEventType, } from '../../runtime/agent-events.js'; +import type { AgentRoundTextEvent } from '../../runtime/agent-events.js'; import { AgentStatus, isTerminalStatus } from '../../runtime/agent-types.js'; +import type { AgentMessage } from '../../runtime/agent-types.js'; import type { AgentStatsSummary } from '../../runtime/agent-statistics.js'; /** @@ -63,6 +65,7 @@ export class FakeAgent { private script: FakeAgentScript; private error: string | undefined; private lastRoundError: string | undefined; + private readonly messages: AgentMessage[] = []; /** Resolvers waiting for a specific message count. */ private messageWaiters: Array<{ @@ -92,6 +95,29 @@ export class FakeAgent { this.completionPromise = new Promise((resolve) => { this.completionResolve = resolve; }); + + // Mirror AgentCore.setupStateListeners: every ROUND_TEXT event + // appends its visible text to the message history. TeamManager + // reads getMessages() to recover round text emitted before its + // event bridge attached, so the fake must keep the same record + // the real runtime does. + this.emitter.on(AgentEventType.ROUND_TEXT, (event: AgentRoundTextEvent) => { + if (event.thoughtText) { + this.messages.push({ + role: 'assistant', + content: event.thoughtText, + timestamp: Date.now(), + thought: true, + }); + } + if (event.text) { + this.messages.push({ + role: 'assistant', + content: event.text, + timestamp: Date.now(), + }); + } + }); } // ─── Lifecycle ────────────────────────────────────────────── @@ -126,6 +152,11 @@ export class FakeAgent { return this.error; } + /** Conversation message history (mirrors AgentInteractive). */ + getMessages(): readonly AgentMessage[] { + return this.messages; + } + getLastRoundError(): string | undefined { return this.lastRoundError; } From fbd6e7696485ad8f2c05eb200b8616301ee53136 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=98=93=E8=89=AF?= <1204183885@qq.com> Date: Thu, 27 Aug 2026 17:52:50 +0800 Subject: [PATCH 2/2] fix(core): clear explicit leader-report flag when seeding pre-attach result MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Squashed replay of two checkpoint commits (git transport down; pushed via git-data API). Commit 1 — fix(core): clear explicit leader-report flag when seeding pre-attach result The pre-attach seed restores pendingFinalReports but left explicitLeaderReports untouched, while the live onRoundText handler it mirrors clears that flag on every non-empty round text. A teammate following the default initialTask prompt calls send_message(to: "leader") mid-round — sendMessage sets the flag synchronously, no event bridge needed — so when the initial round completed entirely before attach, the replayed IDLE settlement saw explicitlyReported === true and skipped the recovered answer, leaving the leader with zero automatic reports of the initial result. Clear the flag when seeding (mirroring onRoundText's semantics) and add diagnostics for the history-recovery and IDLE-replay paths so recovered pre-attach reports are distinguishable from live ones in debug logs. Adds regression coverage for the explicit pre-attach send, for last-visible-answer selection across multiple pre-attach round texts, and for thought-only trailing events (internal reasoning must never be reported as the final answer). Commit 2 — refactor(core): share last-visible-answer scan between team and arena TeamManager.lastVisibleAnswer and ArenaManager.getFinalTextFromTranscript were two implementations of the same rule — walk a message history backwards, return content.trim() of the most recent non-empty, non-thought assistant entry. Any future semantic change would have to land in both places, and missing one would make team pre-attach recovery and the arena final-text fallback silently disagree on identical histories. Extract the scan as lastVisibleAnswer() next to AgentMessage in agent-types.ts (ArenaTranscriptEntry is a field-for-field map of AgentMessage, so the helper applies to both) and delegate from both call sites. Behavior is unchanged; the new initial-result suite and the existing ArenaManager tests stay green. Co-authored-by: Qwen-Coder --- .../core/src/agents/arena/ArenaManager.ts | 17 +-- .../core/src/agents/runtime/agent-types.ts | 19 +++ .../team/TeamManager.initial-result.test.ts | 117 ++++++++++++++++++ packages/core/src/agents/team/TeamManager.ts | 30 +++-- 4 files changed, 162 insertions(+), 21 deletions(-) diff --git a/packages/core/src/agents/arena/ArenaManager.ts b/packages/core/src/agents/arena/ArenaManager.ts index 22aeae40e0b..52ebaac0687 100644 --- a/packages/core/src/agents/arena/ArenaManager.ts +++ b/packages/core/src/agents/arena/ArenaManager.ts @@ -49,6 +49,7 @@ import { isTerminalStatus, isSettledStatus, isSuccessStatus, + lastVisibleAnswer, } from '../runtime/agent-types.js'; import { logArenaSessionStarted, @@ -1674,19 +1675,9 @@ export class ArenaManager { transcript: ArenaTranscriptEntry[] | undefined, ): string | undefined { if (!transcript) return undefined; - - for (let i = transcript.length - 1; i >= 0; i--) { - const message = transcript[i]!; - if ( - message.role === 'assistant' && - !message.thought && - message.content.trim() - ) { - return message.content.trim(); - } - } - - return undefined; + // Shared with TeamManager's pre-attach recovery: the most recent + // non-empty, non-thought assistant message wins. + return lastVisibleAnswer(transcript); } private async addApproachSummaries( diff --git a/packages/core/src/agents/runtime/agent-types.ts b/packages/core/src/agents/runtime/agent-types.ts index 03c4213928c..b757ecc3a6f 100644 --- a/packages/core/src/agents/runtime/agent-types.ts +++ b/packages/core/src/agents/runtime/agent-types.ts @@ -230,6 +230,25 @@ export interface AgentMessage { metadata?: Record; } +/** + * The last model-visible answer in a message history, or undefined + * when there is none. Scans most-recent-first; the first non-empty, + * non-thought assistant message wins. Shared by the team pre-attach + * recovery (TeamManager) and the arena final-text fallback + * (ArenaManager) so both apply the same selection rule. + */ +export function lastVisibleAnswer( + messages: readonly AgentMessage[], +): string | undefined { + for (let i = messages.length - 1; i >= 0; i--) { + const message = messages[i]!; + if (message.role !== 'assistant' || message.thought) continue; + const text = message.content.trim(); + if (text) return text; + } + return undefined; +} + /** * Snapshot of in-progress streaming state for UI mid-switch handoff. * Returned by AgentInteractive.getInProgressStream(). diff --git a/packages/core/src/agents/team/TeamManager.initial-result.test.ts b/packages/core/src/agents/team/TeamManager.initial-result.test.ts index ac75e591d41..b0760161109 100644 --- a/packages/core/src/agents/team/TeamManager.initial-result.test.ts +++ b/packages/core/src/agents/team/TeamManager.initial-result.test.ts @@ -169,4 +169,121 @@ describe('initial teammate result before event bridge attachment (#10211)', () = ]); }); }); + + it('still reports the recovered result when the teammate sent an explicit leader message pre-attach', async () => { + const h = await createHarness(); + + // The default initialTask prompt instructs teammates to report via + // send_message(to: "leader"). Such a send goes through + // TeamManager.sendMessage synchronously — no event bridge needed — + // and marks the sender as having reported explicitly. The seed must + // clear that flag exactly like the live onRoundText handler does, + // or the replayed IDLE settlement skips the recovered answer and + // the leader receives zero automatic reports of the initial result. + await h.spawnTeammate('worker', { + onStart: async (agent) => { + agent.setStatus(AgentStatus.RUNNING); + await h.teamManager.sendMessage('leader', 'progress note', 'worker'); + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 1, + text: 'initial result', + thoughtText: '', + timestamp: Date.now(), + }); + agent.setStatus(AgentStatus.IDLE); + }, + }); + + await vi.waitFor(async () => { + expect(await h.teamManager.getLeaderMessages()).toEqual([ + expect.objectContaining({ from: 'worker', text: 'progress note' }), + expect.objectContaining({ from: 'worker', text: 'initial result' }), + ]); + }); + + // Exactly once: the explicit note plus one automatic forwarding. + await settleAsyncWork(); + expect(await h.teamManager.getLeaderMessages()).toEqual([]); + }); + + it('reports the last pre-attach round text when the round had multiple turns', async () => { + const h = await createHarness(); + + // A multi-turn pre-attach round emits several ROUND_TEXT events; + // the recovery scan must walk the history backwards so the most + // recent non-empty visible answer wins, not the earliest one. + await h.spawnTeammate('worker', { + onStart: (agent) => { + agent.setStatus(AgentStatus.RUNNING); + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 1, + text: 'early turn answer', + thoughtText: '', + timestamp: Date.now(), + }); + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 1, + text: 'final turn answer', + thoughtText: '', + timestamp: Date.now(), + }); + agent.setStatus(AgentStatus.IDLE); + }, + }); + + await vi.waitFor(async () => { + expect(await h.teamManager.getLeaderMessages()).toEqual([ + expect.objectContaining({ + from: 'worker', + text: 'final turn answer', + }), + ]); + }); + + // The earlier turn text must never be reported on its own. + await settleAsyncWork(); + expect(await h.teamManager.getLeaderMessages()).toEqual([]); + }); + + it('never reports thought-only round text, falling back to the earlier visible answer', async () => { + const h = await createHarness(); + + // AgentCore emits ROUND_TEXT whenever roundThoughtText is + // non-empty, so a trailing thought-only event is a reachable + // pre-attach shape. The recovery must skip thought messages and + // never surface internal reasoning as the round's final answer. + await h.spawnTeammate('worker', { + onStart: (agent) => { + agent.setStatus(AgentStatus.RUNNING); + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 1, + text: 'visible answer', + thoughtText: '', + timestamp: Date.now(), + }); + agent.getEventEmitter().emit(AgentEventType.ROUND_TEXT, { + subagentId: agent.agentId, + round: 1, + text: '', + thoughtText: 'internal reasoning about the task', + timestamp: Date.now(), + }); + agent.setStatus(AgentStatus.IDLE); + }, + }); + + await vi.waitFor(async () => { + expect(await h.teamManager.getLeaderMessages()).toEqual([ + expect.objectContaining({ from: 'worker', text: 'visible answer' }), + ]); + }); + + // Nothing else — in particular no thought content — may arrive. + await settleAsyncWork(); + expect(await h.teamManager.getLeaderMessages()).toEqual([]); + }); }); diff --git a/packages/core/src/agents/team/TeamManager.ts b/packages/core/src/agents/team/TeamManager.ts index 3293c335333..4625da2aae9 100644 --- a/packages/core/src/agents/team/TeamManager.ts +++ b/packages/core/src/agents/team/TeamManager.ts @@ -27,7 +27,11 @@ import type { TeamAgentHandle, } from '../backends/types.js'; import { PermissionMode } from '../../hooks/types.js'; -import { AgentStatus, isTerminalStatus } from '../runtime/agent-types.js'; +import { + AgentStatus, + isTerminalStatus, + lastVisibleAnswer, +} from '../runtime/agent-types.js'; import { AgentEventType } from '../runtime/agent-events.js'; import type { AgentRoundTextEvent, @@ -1748,6 +1752,19 @@ export class TeamManager { const preAttachReport = this.lastVisibleAnswer(agent); if (preAttachReport !== undefined) { this.pendingFinalReports.set(agentId, preAttachReport); + // Mirror onRoundText: visible round text supersedes any + // explicit send_message(to: leader) flag set earlier in this + // round. sendMessage sets that flag synchronously — no event + // bridge needed — so a pre-attach explicit progress note would + // otherwise survive until the replayed IDLE settlement below, + // which would then skip this recovered answer and leave the + // leader with zero automatic reports. Erring toward one extra + // delivery (when the last visible text preceded the explicit + // send) matches the "exactly once, not zero" intent. + this.explicitLeaderReports.delete(agentId); + debug.info( + `setupEventBridge: recovered pre-attach round text for "${agentName}" (${agentId}); seeding pending report (${preAttachReport.length} chars) from message history.`, + ); } if (currentStatus === AgentStatus.IDLE && preAttachReport !== undefined) { @@ -1756,6 +1773,9 @@ export class TeamManager { // path uses so its final report and message flush happen // exactly once. Without pre-attach round text there is no // completed round to report — keep the flush-only behavior. + debug.info( + `setupEventBridge: replaying missed IDLE settlement for "${agentName}" (${agentId}); the initial round settled before the event bridge attached.`, + ); onStatusChange({ agentId, previousStatus: AgentStatus.RUNNING, @@ -1791,13 +1811,7 @@ export class TeamManager { private lastVisibleAnswer(agent: TeamAgentHandle): string | undefined { const messages = agent.getMessages?.(); if (!messages) return undefined; - for (let i = messages.length - 1; i >= 0; i--) { - const message = messages[i]; - if (message.role !== 'assistant' || message.thought) continue; - const text = message.content.trim(); - if (text) return text; - } - return undefined; + return lastVisibleAnswer(messages); } // ─── Private: Permission fallback ───────────────────────