Skip to content

fix(channels): deliver background agent replies #7336

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wenshao merged 3 commits into from
Jul 20, 2026
Merged

fix(channels): deliver background agent replies #7336

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9b952d6
fix(channels): deliver background agent replies
qwen-code-dev-bot Jul 20, 2026
3402130
test(channels): cover background response fallback and relay error br…
qwen-code-ci-bot Jul 20, 2026
c271ff1
fix(channels): skip rewritten background responses to prevent duplica…
Jul 20, 2026
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
56 changes: 56 additions & 0 deletions packages/channels/base/src/AcpBridge.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -362,6 +362,62 @@ describe('AcpBridge', () => {
);
});

it('emits a completed background response separately from the active turn', () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
cwd: '/tmp',
}) as unknown as TestableAcpBridge;
const backgroundResponses: Array<[string, string]> = [];
bridge.on('backgroundResponse', (sessionId, text) => {
backgroundResponses.push([sessionId, text]);
});
const textChunks: Array<[string, string]> = [];
bridge.on('textChunk', (sessionId, text) => {
textChunks.push([sessionId, text]);
});

bridge.handleSessionUpdate({
sessionId: 's-1',
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'Background final answer.' },
_meta: {
source: 'background_notification_response',
qwenDiscreteMessage: true,
},
},
});

expect(backgroundResponses).toEqual([['s-1', 'Background final answer.']]);
expect(textChunks).toEqual([]);
});

it('ignores a rewritten background response to avoid duplicate delivery', () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
cwd: '/tmp',
}) as unknown as TestableAcpBridge;
const backgroundResponses: Array<[string, string]> = [];
bridge.on('backgroundResponse', (sessionId, text) => {
backgroundResponses.push([sessionId, text]);
});

bridge.handleSessionUpdate({
sessionId: 's-1',
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'Background final answer.' },
_meta: {
source: 'background_notification_response',
qwenDiscreteMessage: true,
rewritten: true,
},
},
});

expect(backgroundResponses).toEqual([]);
});
Comment on lines +391 to +419

@qwen-code-ci-bot qwen-code-ci-bot Jul 20, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] The test asserts backgroundResponse is emitted but does not assert that textChunk is NOT emitted for the same message. — Failure scenario: if the break at line 331 of AcpBridge.ts were removed, the background response text would be emitted as both backgroundResponse (delivered via Channel proactive send) and textChunk (accumulated into the active turn), causing duplicate message delivery. This test would still pass because it only checks the positive assertion.

Suggested change
expect(backgroundResponses).toEqual([['s-1', 'Background final answer.']]);
});
expect(backgroundResponses).toEqual([['s-1', 'Background final answer.']]);
// Also verify the message is NOT accumulated into the active turn
expect(textChunks).toEqual([]);
});

— qwen3.7-max via Qwen Code /review


it('returns only the final slash-command output', async () => {
const bridge = new AcpBridge({
cliEntryPath: '/tmp/qwen',
Expand Down
16 changes: 12 additions & 4 deletions packages/channels/base/src/AcpBridge.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -314,15 +314,23 @@ export class AcpBridge extends EventEmitter implements ChannelAgentBridge {
switch (type) {
case 'agent_message_chunk': {
const meta = update['_meta'] as Record<string, unknown> | undefined;
if (
typeof meta?.['parentToolCallId'] === 'string' ||
meta?.['qwenDiscreteMessage'] === true
) {
if (typeof meta?.['parentToolCallId'] === 'string') {
break;
}
const content = update['content'] as
| { type?: string; text?: string }
| undefined;
if (meta?.['qwenDiscreteMessage'] === true) {
if (
meta['source'] === 'background_notification_response' &&
meta['rewritten'] !== true &&
content?.type === 'text' &&
content.text
) {
this.emit('backgroundResponse', sessionId, content.text);
}
Comment on lines +323 to +331

@doudouOUC doudouOUC Jul 20, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] When MessageRewriteMiddleware is enabled, the same background_notification_response update is emitted through the bridge twice — once as the original (sent immediately at MessageRewriteMiddleware.ts:89) and once as the rewritten copy (flushed at line 170). The rewritten copy carries the same source: 'background_notification_response' and qwenDiscreteMessage: true metadata (confirmed by MessageRewriteMiddleware.test.ts:245–291 and captureTurnMeta at line 109, since REWRITE_META_EXCLUDED_KEYS is intentionally empty). Both satisfy this guard, so this.emit('backgroundResponse', …) fires twice — the user on the channel receives two messages for one background-task completion. — Failure scenario: enable message-rewrite middleware (production default for most sessions), run a background agent to completion after the parent turn ends → duplicate delivery. Fix: either (a) add && meta['rewritten'] !== true here and in DaemonChannelBridge.ts:656–659, or (b) exclude source === 'background_notification_response' from accumulation in the middleware (like slash_command at line 94), since a discrete background response has no meaningful rewritten form.

— qwen3.7-max via Qwen Code /review

break;
}
Comment on lines +323 to +333

@doudouOUC doudouOUC Jul 20, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] The 4-condition background response detection predicate (qwenDiscreteMessage + source === 'background_notification_response' + rewritten !== true + text-exists) is copy-pasted between AcpBridge and DaemonChannelBridge. — Failure scenario: when the detection criteria change (e.g., a new exclusion field in _meta), both copies must be updated in lockstep. A change to one bridge without the other silently causes channel-specific behavioral divergence — one bridge delivers background responses that the other drops.

Consider extracting the meta-conditions into a shared predicate (e.g., isBackgroundResponse(meta): boolean). The text-existence check can remain caller-side since it depends on each bridge's content model.

— qwen3.7-max via Qwen Code /review

if (content?.type === 'text' && content.text) {
this.emit(
meta?.['source'] === 'slash_command'
Expand Down
1 change: 1 addition & 0 deletions packages/channels/base/src/ChannelAgentBridge.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ export interface PermissionResolvedEvent {
interface ChannelAgentBridgeEventMap {
sessionDied: [SessionDiedEvent];
textChunk: [sessionId: string, chunk: string];
backgroundResponse: [sessionId: string, text: string];
responseBoundary: [sessionId: string];
toolCall: [ToolCallEvent];
permissionRequest: [PermissionRequestEvent];
Expand Down
64 changes: 64 additions & 0 deletions packages/channels/base/src/ChannelBase.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7060,6 +7060,70 @@ describe('ChannelBase', () => {
expect(router.handleSessionDied).toHaveBeenCalledWith('s-1');
});

it('proactively delivers a completed background response to the session route', async () => {
const target: SessionTarget = {
channelName: 'test-chan',
senderId: 'user1',
chatId: 'chat1',
isGroup: true,
};
const router = {
getTarget: vi.fn().mockReturnValue(target),
handleSessionDied: vi.fn(),
setBridge: vi.fn(),
};
const ch = createChannel({}, {
router,
registerBridgeEvents: true,
} as unknown as ChannelBaseOptions);
ch.proactiveSupported = true;

(bridge as unknown as EventEmitter).emit(
'backgroundResponse',
's-1',
'Background final answer.',
);

await vi.waitFor(() => {
expect(ch.proactive).toEqual([
{ chatId: 'chat1', text: 'Background final answer.' },
]);
});
expect(ch.proactiveTargets).toEqual([target]);
expect(ch.sent).toEqual([]);
});

it('falls back to sendResponseMessage when proactive send is unsupported', async () => {
const target: SessionTarget = {
channelName: 'test-chan',
senderId: 'user1',
chatId: 'chat1',
isGroup: true,
};
const router = {
getTarget: vi.fn().mockReturnValue(target),
handleSessionDied: vi.fn(),
setBridge: vi.fn(),
};
const ch = createChannel({}, {
router,
registerBridgeEvents: true,
} as unknown as ChannelBaseOptions);

(bridge as unknown as EventEmitter).emit(
'backgroundResponse',
's-1',
'Background final answer.',
);

await vi.waitFor(() => {
expect(ch.sent).toEqual([
{ chatId: 'chat1', text: 'Background final answer.' },
]);
});
expect(ch.proactive).toEqual([]);
});

it('leaves supplied router bridge events to the gateway by default', () => {
const router = {
getTarget: vi.fn(),
Expand Down
33 changes: 33 additions & 0 deletions packages/channels/base/src/ChannelBase.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -346,6 +346,18 @@ export abstract class ChannelBase {
private readonly bridgeToolCallListener = (event: ToolCallEvent): void => {
this.dispatchToolCall(event);
};
private readonly bridgeBackgroundResponseListener = (
sessionId: string,
text: string,
): void => {
void this.dispatchBackgroundResponse(sessionId, text).catch(
(err: unknown) => {
process.stderr.write(
`[${this.name}] background response delivery failed for session ${sanitizeLogText(sessionId, 128)}: ${this.lifecycleError(err)}\n`,
);
},
);
};
private readonly bridgeSessionDiedListener = (
event: SessionDiedEvent,
): void => {
Expand Down Expand Up @@ -403,6 +415,25 @@ export abstract class ChannelBase {
this.onToolCall(chatId, event);
}

async dispatchBackgroundResponse(
sessionId: string,
text: string,
): Promise<void> {
const target = this.router.getTarget(sessionId);
if (
!target ||
target.channelName !== this.name ||
text.trim().length === 0
) {
return;
}
Comment on lines +422 to +429

@doudouOUC doudouOUC Jul 20, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] dispatchBackgroundResponse has three early-return guards (null target, cross-channel name mismatch, empty/whitespace text) — none are tested. — Failure scenario: if the target.channelName !== this.name guard were removed or inverted, every bridge-listening channel would deliver backgroundResponse for sessions routed to a different channel, producing duplicate cross-channel message delivery.

Consider adding tests for each guard branch: (1) session with a different channelName target → no delivery; (2) whitespace-only text → no delivery; (3) missing route → no crash.

— qwen3.7-max via Qwen Code /review

if (this.supportsProactiveSend() && this.supportsProactiveTarget(target)) {
await this.pushProactive(target, text);
return;
}
await this.sendResponseMessage(target.chatId, text, sessionId);
Comment on lines +430 to +434

@qwen-code-ci-bot qwen-code-ci-bot Jul 20, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] The sendResponseMessage fallback path (when supportsProactiveSend() returns false) is not covered by any test. — Concrete cost: if the fallback's arguments were changed incorrectly (e.g. dropping sessionId), no test would detect the regression.

Add a test with proactiveSupported = false that asserts ch.sent contains the expected { chatId, text, sessionId } and ch.proactive remains empty.

— qwen3.7-max via Qwen Code /review

}

async dispatchPermissionRequest(
event: PermissionRequestEvent,
): Promise<void> {
Expand Down Expand Up @@ -1704,13 +1735,15 @@ export abstract class ChannelBase {

private attachBridgeEvents(bridge: ChannelAgentBridge): void {
bridge.on('toolCall', this.bridgeToolCallListener);
bridge.on('backgroundResponse', this.bridgeBackgroundResponseListener);
bridge.on('sessionDied', this.bridgeSessionDiedListener);
bridge.on('permissionRequest', this.bridgePermissionRequestListener);
bridge.on('permissionResolved', this.bridgePermissionResolvedListener);
}

private detachBridgeEvents(bridge: ChannelAgentBridge): void {
bridge.off('toolCall', this.bridgeToolCallListener);
bridge.off('backgroundResponse', this.bridgeBackgroundResponseListener);
bridge.off('sessionDied', this.bridgeSessionDiedListener);
bridge.off('permissionRequest', this.bridgePermissionRequestListener);
bridge.off('permissionResolved', this.bridgePermissionResolvedListener);
Expand Down
106 changes: 106 additions & 0 deletions packages/channels/base/src/DaemonChannelBridge.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -566,6 +566,112 @@ describe('DaemonChannelBridge', () => {
bridge.stop();
});

it('emits the completed background response without appending it to the active turn', async () => {
const events = new EventQueue();
const session = createFakeSession(events);
session.prompt.mockImplementation(async () => {
events.push({
id: 1,
v: 1,
type: 'session_update',
data: {
sessionId: 'session-1',
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'Initial answer.' },
},
},
});
events.push(turnCompleteEvent());
return { stopReason: 'end_turn' };
});
const bridge = new DaemonChannelBridge({
cwd: '/repo',
sessionFactory: vi.fn().mockResolvedValue(session),
});
const backgroundResponses: Array<[string, string]> = [];
bridge.on('backgroundResponse', (sessionId, text) => {
backgroundResponses.push([sessionId, text]);
});

await bridge.start();
await bridge.newSession('/repo');
await expect(bridge.prompt('session-1', 'investigate')).resolves.toBe(
'Initial answer.',
);

events.push({
id: 2,
v: 1,
type: 'session_update',
data: {
sessionId: 'session-1',
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'Background final answer.' },
_meta: {
source: 'background_notification_response',
qwenDiscreteMessage: true,
},
},
},
});

await vi.waitFor(() => {
expect(backgroundResponses).toEqual([
['session-1', 'Background final answer.'],
]);
});

events.close();
bridge.stop();
});

it('ignores a rewritten background response to avoid duplicate delivery', async () => {
const events = new EventQueue();
const session = createFakeSession(events);
session.prompt.mockImplementation(async () => {
events.push(turnCompleteEvent());
return { stopReason: 'end_turn' };
});
const bridge = new DaemonChannelBridge({
cwd: '/repo',
sessionFactory: vi.fn().mockResolvedValue(session),
});
const backgroundResponses: Array<[string, string]> = [];
bridge.on('backgroundResponse', (sessionId, text) => {
backgroundResponses.push([sessionId, text]);
});

await bridge.start();
await bridge.newSession('/repo');
await bridge.prompt('session-1', 'investigate');

events.push({
id: 2,
v: 1,
type: 'session_update',
data: {
sessionId: 'session-1',
update: {
sessionUpdate: 'agent_message_chunk',
content: { type: 'text', text: 'Background final answer.' },
_meta: {
source: 'background_notification_response',
qwenDiscreteMessage: true,
rewritten: true,
},
},
},
});

await new Promise((r) => setTimeout(r, 50));
expect(backgroundResponses).toEqual([]);

events.close();
bridge.stop();
});

it('returns only the final slash-command output from the daemon', async () => {
const events = new EventQueue();
const session = createFakeSession(events);
Expand Down
15 changes: 11 additions & 4 deletions packages/channels/base/src/DaemonChannelBridge.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -649,13 +649,20 @@ export class DaemonChannelBridge
switch (type) {
case 'agent_message_chunk': {
const meta = isRecord(update['_meta']) ? update['_meta'] : undefined;
if (
typeof meta?.['parentToolCallId'] === 'string' ||
meta?.['qwenDiscreteMessage'] === true
) {
if (typeof meta?.['parentToolCallId'] === 'string') {
break;
}
const text = getTextContent(update['content']);
if (meta?.['qwenDiscreteMessage'] === true) {
if (
meta['source'] === 'background_notification_response' &&
meta['rewritten'] !== true &&
text
) {
this.emit('backgroundResponse', sessionId, text);
}
break;
}
if (text) {
this.emit(
meta?.['source'] === 'slash_command'
Expand Down
7 changes: 7 additions & 0 deletions packages/cli/src/commands/channel/daemon-worker.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ const mockUpdateChannelMemoryEntry = vi.hoisted(() => vi.fn());
const mockRemoveChannelMemoryEntries = vi.hoisted(() => vi.fn());
const mockClearChannelMemory = vi.hoisted(() => vi.fn());
const mockRegisterToolCallDispatch = vi.hoisted(() => vi.fn());
const mockRegisterBackgroundResponseRelay = vi.hoisted(() => vi.fn());
const mockRegisterPermissionRelay = vi.hoisted(() => vi.fn());
const mockRegisterSessionCleanup = vi.hoisted(() => vi.fn());
const mockSessionsPath = vi.hoisted(() => vi.fn(() => '/tmp/sessions.json'));
Expand Down Expand Up @@ -174,6 +175,7 @@ vi.mock('./runtime.js', () => ({
loadChannelsConfig: mockLoadChannelsConfig,
loadChannelsFromExtensions: mockLoadChannelsFromExtensions,
parseConfiguredChannels: mockParseConfiguredChannels,
registerBackgroundResponseRelay: mockRegisterBackgroundResponseRelay,
registerPermissionRelay: mockRegisterPermissionRelay,
registerSessionCleanup: mockRegisterSessionCleanup,
registerToolCallDispatch: mockRegisterToolCallDispatch,
Expand Down Expand Up @@ -733,6 +735,11 @@ describe('runChannelDaemonWorker', () => {
mockSessionRouter.mock.results[0]!.value,
expect.any(Map),
);
expect(mockRegisterBackgroundResponseRelay).toHaveBeenCalledWith(
bridgeFacade,
mockSessionRouter.mock.results[0]!.value,
expect.any(Map),
);
expect(mockResolveProxyUrl).toHaveBeenCalledWith(
undefined,
'http://settings-proxy:8080',
Expand Down
Loading
Loading