Skip to content

Commit a82a11a

Browse files
yiliang114Qwen-Coder
andauthored
fix(core): report broadcast delivery failures in send_message (#10081)
send_message(to: "*") unconditionally returned "Message broadcast to all teammates." even when TeamManager.broadcast() had rejected deliveries: broadcast() collected the per-recipient failures but returned Promise<void>, discarding them. Make broadcast() return a BroadcastResult (attempted total + failed recipient names) derived from the failures it already computes, and let the send_message broadcast branch distinguish complete success, partial failure (naming the unreachable recipients), and total failure (returned as a tool error). Co-authored-by: Qwen-Coder <qwen-coder @alibabacloud.com>
1 parent 3a46420 commit a82a11a

5 files changed

Lines changed: 209 additions & 13 deletions

File tree

packages/core/src/agents/team/TeamManager.ts

Lines changed: 24 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,14 @@ const debug = createDebugLogger('AGENTS_TEAM_MANAGER');
8888
// imported it from this module keep compiling.
8989
export type { TeamAgentHandle };
9090

91+
/** Delivery outcome of a {@link TeamManager.broadcast} call. */
92+
export interface BroadcastResult {
93+
/** Number of recipients the broadcast attempted (sender excluded). */
94+
total: number;
95+
/** Names of recipients whose delivery was rejected. */
96+
failedRecipients: string[];
97+
}
98+
9199
/** Configuration for spawning a teammate. */
92100
export interface TeammateSpawnConfig {
93101
/** Human-readable name (will be sanitized). */
@@ -670,31 +678,38 @@ export class TeamManager {
670678
/**
671679
* Broadcast a message to all teammates and the leader
672680
* (except the sender).
681+
*
682+
* Returns the delivery outcome so the caller can distinguish complete
683+
* success from partial/total failure instead of assuming every
684+
* delivery landed.
673685
*/
674-
async broadcast(message: string, fromName: string): Promise<void> {
675-
const promises = this.teamFile.members
686+
async broadcast(message: string, fromName: string): Promise<BroadcastResult> {
687+
const recipients = this.teamFile.members
676688
.filter((m) => m.name.toLowerCase() !== fromName.toLowerCase())
677-
.map((m) => this.sendMessage(m.name, message, fromName));
689+
.map((m) => m.name);
678690

679691
// Also deliver to leader inbox if sender is not the leader.
680692
if (fromName.toLowerCase() !== LEADER_NAME) {
681-
promises.push(this.sendMessage(LEADER_NAME, message, fromName));
693+
recipients.push(LEADER_NAME);
682694
}
683695

684696
// allSettled, not all: a single recipient that terminated between
685697
// the member snapshot and the send throws (its queue is gone), and
686698
// Promise.all would reject the whole broadcast — making the leader
687699
// think every recipient failed when the rest were delivered fine.
688-
const results = await Promise.allSettled(promises);
689-
const failures = results.filter(
690-
(r): r is PromiseRejectedResult => r.status === 'rejected',
700+
const results = await Promise.allSettled(
701+
recipients.map((name) => this.sendMessage(name, message, fromName)),
702+
);
703+
const failedRecipients = recipients.filter(
704+
(_, i) => results[i]?.status === 'rejected',
691705
);
692-
if (failures.length > 0) {
706+
if (failedRecipients.length > 0) {
693707
debug.warn(
694-
`Broadcast: ${failures.length}/${results.length} send(s) failed ` +
708+
`Broadcast: ${failedRecipients.length}/${results.length} send(s) failed ` +
695709
`(recipient likely terminated).`,
696710
);
697711
}
712+
return { total: recipients.length, failedRecipients };
698713
}
699714

700715
/**

packages/core/src/agents/team/test-utils/coordination-harness.test.ts

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2049,6 +2049,33 @@ describe('TeamCoordinationHarness', () => {
20492049
expect(w3.getReceivedMessages()).toHaveLength(1);
20502050
expectTeamMessage(w3.getReceivedMessages()[0], 'w2', 'hello all');
20512051
});
2052+
2053+
it('reports zero failures when every delivery lands (#10072)', async () => {
2054+
const h = await createHarness();
2055+
await h.spawnTeammate('w1');
2056+
await h.spawnTeammate('w2');
2057+
2058+
// Recipients: w2 (member) + leader inbox.
2059+
const result = await h.teamManager.broadcast('hello all', 'w1');
2060+
2061+
expect(result).toEqual({ total: 2, failedRecipients: [] });
2062+
await h.waitForMessages('w2', 1);
2063+
});
2064+
2065+
it('reports the recipients whose delivery was rejected (#10072)', async () => {
2066+
const h = await createHarness();
2067+
await h.spawnTeammate('w1');
2068+
const w2 = await h.spawnTeammate('w2');
2069+
2070+
// w2 terminates between the member snapshot and the send: its
2071+
// queue is dropped, so its delivery rejects while the leader
2072+
// inbox write still lands.
2073+
await w2.shutdown();
2074+
2075+
const result = await h.teamManager.broadcast('status update', 'w1');
2076+
2077+
expect(result).toEqual({ total: 2, failedRecipients: ['w2'] });
2078+
});
20522079
});
20532080

20542081
// ─── 6. Concurrent task claiming ──────────────────────────
Lines changed: 130 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,130 @@
1+
/**
2+
* @license
3+
* Copyright 2025 Qwen
4+
* SPDX-License-Identifier: Apache-2.0
5+
*/
6+
7+
/**
8+
* Broadcast delivery-outcome contract for send_message(to: "*") — #10072.
9+
*
10+
* Uses the real TeamManager (via TeamCoordinationHarness) so a delivery
11+
* rejects the same way it does in production: a recipient terminates
12+
* between the member snapshot and the send, its per-agent queue is
13+
* dropped, and sendMessage refuses the delivery.
14+
*/
15+
16+
import { describe, it, expect, vi, afterEach } from 'vitest';
17+
import { SendMessageTool } from './send-message.js';
18+
import { BackgroundTaskRegistry } from '../agents/background-tasks.js';
19+
import type { ApprovalMode, Config } from '../config/config.js';
20+
import type { TeamManager } from '../agents/team/TeamManager.js';
21+
import { TeamCoordinationHarness } from '../agents/team/test-utils/coordination-harness.js';
22+
23+
// Mock Storage so all file I/O uses the harness's temp dir.
24+
vi.mock('../config/storage.js', async (importOriginal) => {
25+
const original =
26+
await importOriginal<typeof import('../config/storage.js')>();
27+
let mockGlobalDir = '';
28+
return {
29+
...original,
30+
Storage: {
31+
...original.Storage,
32+
getGlobalQwenDir: () => mockGlobalDir,
33+
__setMockGlobalDir: (dir: string) => {
34+
mockGlobalDir = dir;
35+
},
36+
},
37+
};
38+
});
39+
40+
import { Storage } from '../config/storage.js';
41+
42+
function setMockDir(dir: string): void {
43+
(
44+
Storage as unknown as {
45+
__setMockGlobalDir: (d: string) => void;
46+
}
47+
).__setMockGlobalDir(dir);
48+
}
49+
50+
function makeConfig(teamManager: TeamManager): Config {
51+
return {
52+
getTeamManager: () => teamManager,
53+
getBackgroundTaskRegistry: () => new BackgroundTaskRegistry(),
54+
getApprovalMode: () => 'default' as ApprovalMode,
55+
} as unknown as Config;
56+
}
57+
58+
describe('SendMessageTool — broadcast delivery outcomes (#10072)', () => {
59+
let harness: TeamCoordinationHarness | undefined;
60+
61+
afterEach(async () => {
62+
await harness?.cleanup();
63+
harness = undefined;
64+
});
65+
66+
async function createHarness(): Promise<TeamCoordinationHarness> {
67+
const h = await TeamCoordinationHarness.create();
68+
setMockDir(h.tmpDir);
69+
harness = h;
70+
return h;
71+
}
72+
73+
function broadcastInvocation(h: TeamCoordinationHarness) {
74+
const tool = new SendMessageTool(makeConfig(h.teamManager));
75+
return tool.build({ to: '*', message: 'sync for everyone' });
76+
}
77+
78+
it('does not claim complete success when a delivery is rejected', async () => {
79+
const h = await createHarness();
80+
await h.spawnTeammate('alice');
81+
const bob = await h.spawnTeammate('bob');
82+
83+
// bob terminates between the member snapshot and the send: its
84+
// queue is dropped, so its delivery rejects while alice's lands.
85+
await bob.shutdown();
86+
87+
const result = await broadcastInvocation(h).execute(
88+
new AbortController().signal,
89+
);
90+
91+
expect(result.error).toBeUndefined();
92+
// Must not claim that every teammate received the message…
93+
expect(result.llmContent).not.toBe('Message broadcast to all teammates.');
94+
// …and must name the recipient that was not reached.
95+
expect(String(result.llmContent)).toContain('bob');
96+
});
97+
98+
it('still reports complete success when every delivery lands', async () => {
99+
const h = await createHarness();
100+
const alice = await h.spawnTeammate('alice');
101+
await h.spawnTeammate('bob');
102+
103+
const result = await broadcastInvocation(h).execute(
104+
new AbortController().signal,
105+
);
106+
107+
expect(result.error).toBeUndefined();
108+
expect(result.llmContent).toBe('Message broadcast to all teammates.');
109+
await h.waitForMessages('alice', 1);
110+
await h.waitForMessages('bob', 1);
111+
expect(alice.getReceivedMessages()).toHaveLength(1);
112+
});
113+
114+
it('reports failure when no delivery lands', async () => {
115+
const h = await createHarness();
116+
const alice = await h.spawnTeammate('alice');
117+
const bob = await h.spawnTeammate('bob');
118+
await alice.shutdown();
119+
await bob.shutdown();
120+
121+
const result = await broadcastInvocation(h).execute(
122+
new AbortController().signal,
123+
);
124+
125+
expect(result.error).toBeDefined();
126+
expect(result.llmContent).not.toBe('Message broadcast to all teammates.');
127+
expect(String(result.llmContent)).toContain('alice');
128+
expect(String(result.llmContent)).toContain('bob');
129+
});
130+
});

packages/core/src/tools/send-message.test.ts

Lines changed: 5 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,14 +10,15 @@ import { BackgroundTaskRegistry } from '../agents/background-tasks.js';
1010
import { ToolErrorType } from './tool-error.js';
1111
import type { ApprovalMode, Config } from '../config/config.js';
1212
import { runWithTeammateIdentity } from '../agents/team/identity.js';
13+
import type { BroadcastResult } from '../agents/team/TeamManager.js';
1314

1415
const DEFAULT_MODE = 'default' as ApprovalMode;
1516
const PLAN_MODE = 'plan' as ApprovalMode;
1617

1718
function makeTeamConfig(opts?: {
1819
teamManager?: {
1920
sendMessage: (...args: unknown[]) => Promise<void>;
20-
broadcast: (...args: unknown[]) => Promise<void>;
21+
broadcast: (...args: unknown[]) => Promise<BroadcastResult>;
2122
} | null;
2223
approvalMode?: ApprovalMode;
2324
}) {
@@ -69,7 +70,9 @@ describe('SendMessageTool — team mode', () => {
6970
});
7071

7172
it('broadcasts with "*"', async () => {
72-
const broadcast = vi.fn().mockResolvedValue(undefined);
73+
const broadcast = vi
74+
.fn()
75+
.mockResolvedValue({ total: 2, failedRecipients: [] });
7376
const tool = new SendMessageTool(
7477
makeTeamConfig({
7578
teamManager: {

packages/core/src/tools/send-message.ts

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -242,8 +242,29 @@ class SendMessageInvocation extends BaseToolInvocation<
242242
try {
243243
if (to === '*') {
244244
const sender = getAgentName() ?? LEADER_NAME;
245-
await teamManager.broadcast(this.params.message, sender);
246-
const msg = 'Message broadcast to all teammates.';
245+
const { total, failedRecipients } = await teamManager.broadcast(
246+
this.params.message,
247+
sender,
248+
);
249+
if (failedRecipients.length === 0) {
250+
const msg = 'Message broadcast to all teammates.';
251+
return { llmContent: msg, returnDisplay: msg };
252+
}
253+
const reached = total - failedRecipients.length;
254+
if (reached === 0) {
255+
const msg =
256+
`Broadcast failed: delivery was rejected for all ${total} ` +
257+
`recipient(s): ${failedRecipients.join(', ')}.`;
258+
return {
259+
llmContent: msg,
260+
returnDisplay: msg,
261+
error: { message: msg },
262+
};
263+
}
264+
const msg =
265+
`Message broadcast delivered to ${reached} of ${total} ` +
266+
`recipient(s); delivery failed for: ${failedRecipients.join(', ')}. ` +
267+
`The listed recipients did not receive the message.`;
247268
return { llmContent: msg, returnDisplay: msg };
248269
}
249270

0 commit comments

Comments
 (0)