From 55099ddd89d96d4966ed581b9eb8f928fa81be0a Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 14 Jul 2026 18:01:23 +0800 Subject: [PATCH 1/3] fix(server): synthesize session status only from the main agent's turn A sub-agent runs its own turn on the shared session channel, and the broadcaster synthesized event.session.status_changed from every agent's turn.started/turn.ended. A foreground sub-agent finishing mid-turn thus emitted a bogus idle transition that clients read as 'turn finished' (notifications, sounds, unread dots, queued-message drain), while the real main-agent turn end was swallowed by dedup. Gate the status synthesis on MAIN_AGENT_ID, matching the existing main-only rule in InFlightTurnTracker. --- .../fix-subagent-turn-status-notifications.md | 5 ++ .../ws/v1/sessionEventBroadcaster.ts | 11 +++- .../test/sessionEventBroadcaster.test.ts | 50 +++++++++++++++++-- 3 files changed, 60 insertions(+), 6 deletions(-) create mode 100644 .changeset/fix-subagent-turn-status-notifications.md diff --git a/.changeset/fix-subagent-turn-status-notifications.md b/.changeset/fix-subagent-turn-status-notifications.md new file mode 100644 index 0000000000..f2818c8dca --- /dev/null +++ b/.changeset/fix-subagent-turn-status-notifications.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Fix sub-agent completions being signaled as session turn completions, which fired premature completion notifications, sounds, and unread markers while the main turn was still running. diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts index 6025c4d1f2..b7b443f0d1 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts @@ -539,7 +539,14 @@ export class SessionEventBroadcaster { // `DomainEventMap` payload types are deliberately wider than the protocol // contract, hence the assertion via `unknown`. const wireEvent = { ...event, agentId, sessionId } as unknown as Event; - if (event.type === 'turn.started') { + // Session status transitions are synthesized only from the MAIN agent's turn + // boundaries. Subagents share the session channel (their frames carry their + // own agentId and clients route them to the task view), so a subagent's + // turn.ended would otherwise emit a bogus `status_changed(idle)` mid-turn: + // the client reads that as "the turn finished" (driving notifications, + // sounds, unread dots and the queued-message drain), and the real turn end + // is then swallowed by dedup. Same main-only rule as InFlightTurnTracker. + if (agentId === MAIN_AGENT_ID && event.type === 'turn.started') { // Status is authoritative protocol state: emit it before the turn event so // clients enter running first. A pending interaction remains higher // priority than running. @@ -549,7 +556,7 @@ export class SessionEventBroadcaster { state.queue = state.queue .then(() => this.dispatch(state, wireEvent, volatile)) .catch(() => {}); - if (event.type === 'turn.ended') { + if (agentId === MAIN_AGENT_ID && event.type === 'turn.ended') { // Emit completion after the turn event. Pending interactions remain higher // priority; otherwise failed/cancelled/blocked turns abort and all others // become idle. diff --git a/packages/kap-server/test/sessionEventBroadcaster.test.ts b/packages/kap-server/test/sessionEventBroadcaster.test.ts index e96e9c133f..0233a9de62 100644 --- a/packages/kap-server/test/sessionEventBroadcaster.test.ts +++ b/packages/kap-server/test/sessionEventBroadcaster.test.ts @@ -503,6 +503,44 @@ describe('SessionEventBroadcaster', () => { ]); }); + it('does not synthesize session status from sub-agent turn boundaries', async () => { + // Regression: a sub-agent's turn.started/turn.ended stream over the same + // session channel with their own agentId. Synthesizing status transitions + // from them emitted a bogus `status_changed(idle)` the moment a foreground + // sub-agent finished — mid main turn — which kimi-web reads as "the turn + // finished" (browser notification, completion sound, unread dot, queued + // message drain), while the real main-agent turn end was then swallowed by + // dedup and never notified. + const lc = new FakeLifecycle(); + const main = lc.addAgent('main'); + const sub = lc.addAgent('agent-0'); + sessions.set('s1', lc); + const { target, envelopes } = collectingTarget(); + await bc.subscribe('s1', target); + + main.bus.emit(agentEvent('turn.started', { turnId: 1 })); + // A foreground sub-agent runs and completes while the main turn is in flight. + sub.bus.emit(agentEvent('turn.started', { turnId: 10 })); + sub.bus.emit(agentEvent('turn.ended', { turnId: 10, reason: 'completed' })); + main.bus.emit(agentEvent('turn.ended', { turnId: 1, reason: 'completed' })); + await bc.getCursor('s1'); + + // The sub-agent's turn events are still fanned out (clients render them in + // the task view), but they produce no status transitions. + expect( + envelopes + .filter((e) => e.type === 'turn.started' || e.type === 'turn.ended') + .map((e) => (e.payload as { agentId: string }).agentId), + ).toEqual(['main', 'agent-0', 'agent-0', 'main']); + const statusEnvs = envelopes.filter((e) => e.type === 'event.session.status_changed'); + expect(statusEnvs.map((e) => e.payload)).toMatchObject([ + { status: 'running', previous_status: 'idle' }, + { status: 'idle', previous_status: 'running' }, + ]); + // The idle transition fires exactly once, after the main agent's turn end. + expect(envelopes.at(-1)!.type).toBe('event.session.status_changed'); + }); + it('broadcasts question requested / answered as durable v1 events', async () => { const lc = new FakeLifecycle(); lc.addAgent('main'); @@ -795,7 +833,10 @@ describe('SessionEventBroadcaster', () => { agentEnvs.every((e) => (e.payload as { agentId: string }).agentId === 'main'), ).toBe(true); // `event.session.status_changed` is global (`event.session.*`) and bypasses - // the agent filter. The redundant idle transition from the sub-agent is deduped. + // the agent filter. The sub-agent's turn.ended synthesizes no status change + // at all (main-only rule — see the "does not synthesize session status from + // sub-agent turn boundaries" test), so only the main agent's two transitions + // are delivered. const statusEnvs = envelopes.filter((e) => e.type === 'event.session.status_changed'); expect(statusEnvs).toHaveLength(2); }); @@ -870,9 +911,10 @@ describe('SessionEventBroadcaster', () => { const result = await bc2.getBufferedSince('s1', { seq: 0 }, new Set(['main'])); expect(result.resyncRequired).toBe(false); - // The sub-agent's turn events are cropped, while global status transitions - // retain their original positions in the session sequence. - expect(result.events.map((e) => e.seq)).toEqual([1, 2, 3, 4, 5, 8, 9, 10, 11, 12]); + // The sub-agent's turn events are cropped (seq 5/6 — they synthesize no + // status change), while the main agent's turns and the global status + // transitions retain their original positions in the session sequence. + expect(result.events.map((e) => e.seq)).toEqual([1, 2, 3, 4, 7, 8, 9, 10]); expect( result.events.every((e) => (e.envelope.payload as { agentId: string }).agentId === 'main'), ).toBe(true); From 7918b45fea7e7c6d51faef0f0a8852d374e36003 Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 14 Jul 2026 18:18:42 +0800 Subject: [PATCH 2/3] fix(server): emit idle when a sub-agent turn ends as the last active agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Defensive follow-up to the main-only status synthesis: ensureState seeds lastStatus from ISessionActivity, which counts any agent's turn, so a session first activated while only a sub-agent was active would stay running for subscribers until the next main turn. A sub-agent's own lease and loop state are already released when its turn.ended is published, so a session-wide idle activity read cannot be a mid-main-turn foreground sub-agent — emit idle in that case only. Dedup keeps it a no-op everywhere else. --- .../ws/v1/sessionEventBroadcaster.ts | 13 +++++++ .../test/sessionEventBroadcaster.test.ts | 36 +++++++++++++++++++ 2 files changed, 49 insertions(+) diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts index b7b443f0d1..7ca968de08 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts @@ -564,6 +564,19 @@ export class SessionEventBroadcaster { const terminalStatus = reason === 'cancelled' || reason === 'failed' || reason === 'blocked' ? 'aborted' : 'idle'; this.enqueueStatusChanged(state, pendingStatus(state.activity?.status()) ?? terminalStatus); + } else if (event.type === 'turn.ended' && state.activity?.status() === 'idle') { + // Defensive: a sub-agent's own lease and loop state are already released + // when its turn.ended is published (see loopService), so this activity + // read reflects only the OTHER agents. Session-wide idle here therefore + // cannot be a mid-main-turn foreground sub-agent (the main lease would + // report running) — it means the last active agent finished. Emit idle + // to clear a stale `running` seed: ensureState seeds lastStatus from + // ISessionActivity, which counts any agent's turn, so a session first + // activated while only a sub-agent was active (e.g. driven by a REST-only + // client that never triggered activation during the main turn) would + // otherwise stay `running` for its subscribers until the next main turn. + // Dedup makes this a no-op in every normally-tracked case. + this.enqueueStatusChanged(state, 'idle'); } // v1 wire compat: fan the legacy `background.task.*` spelling out next to // the native `task.*` event (see `legacyTaskEvent`) so unchanged v1 clients diff --git a/packages/kap-server/test/sessionEventBroadcaster.test.ts b/packages/kap-server/test/sessionEventBroadcaster.test.ts index 0233a9de62..0b6b1517cb 100644 --- a/packages/kap-server/test/sessionEventBroadcaster.test.ts +++ b/packages/kap-server/test/sessionEventBroadcaster.test.ts @@ -519,9 +519,14 @@ describe('SessionEventBroadcaster', () => { await bc.subscribe('s1', target); main.bus.emit(agentEvent('turn.started', { turnId: 1 })); + // Model the main turn's activity lease: while it is in flight the session + // activity reports running, so the defensive idle emit for sub-agent turns + // must stay silent. + lc.baseStatus = 'running'; // A foreground sub-agent runs and completes while the main turn is in flight. sub.bus.emit(agentEvent('turn.started', { turnId: 10 })); sub.bus.emit(agentEvent('turn.ended', { turnId: 10, reason: 'completed' })); + lc.baseStatus = 'idle'; main.bus.emit(agentEvent('turn.ended', { turnId: 1, reason: 'completed' })); await bc.getCursor('s1'); @@ -541,6 +546,37 @@ describe('SessionEventBroadcaster', () => { expect(envelopes.at(-1)!.type).toBe('event.session.status_changed'); }); + it('clears a stale running seed when a sub-agent turn ends as the last active agent', async () => { + // Defensive path: a session first activated while only a sub-agent turn is + // active seeds lastStatus='running' from ISessionActivity (which counts any + // agent's turn). When that sub-agent finishes, its own lease/loop are + // already released, so an idle session activity means it was the last + // active agent — the session is genuinely idle and subscribers must be told, + // otherwise they stay `running` until the next main turn. + const lc = new FakeLifecycle(); + lc.baseStatus = 'running'; // a background sub-agent turn is active + const sub = lc.addAgent('agent-0'); + sessions.set('s1', lc); + const { target, envelopes } = collectingTarget(); + await bc.subscribe('s1', target); // activates; seeds lastStatus='running' + + // The sub-agent's turn ends; nothing else is running anymore. + lc.baseStatus = 'idle'; + sub.bus.emit(agentEvent('turn.ended', { turnId: 10, reason: 'completed' })); + await bc.getCursor('s1'); + + expect(envelopes.map((e) => e.type)).toEqual([ + 'turn.ended', + 'event.session.status_changed', + ]); + expect(envelopes[1]).toMatchObject({ + type: 'event.session.status_changed', + session_id: 's1', + payload: { status: 'idle', previous_status: 'running' }, + }); + expect(envelopes[1]!.volatile).toBeUndefined(); + }); + it('broadcasts question requested / answered as durable v1 events', async () => { const lc = new FakeLifecycle(); lc.addAgent('main'); From 3f9ff0018ac9ef89a4bed33e3deac371e75b1a8e Mon Sep 17 00:00:00 2001 From: qer Date: Tue, 14 Jul 2026 18:31:21 +0800 Subject: [PATCH 3/3] revert: drop the defensive idle emit for sub-agent-only turn endings This reverts 7918b45fe. The stale-seed scenario it guarded is effectively unreachable in production: sessions are activated at creation / first prompt and stay active for the process lifetime, and persisted detached tasks are reconciled as lost on restart, never resumed. The guard also raised a new behavioral question of its own (failed/cancelled sub-agent endings would emit a success-style idle). Keep the PR to the minimal main-only status synthesis; the theoretical corner self-corrects on the next REST status pull or main turn. --- .../ws/v1/sessionEventBroadcaster.ts | 13 ------- .../test/sessionEventBroadcaster.test.ts | 36 ------------------- 2 files changed, 49 deletions(-) diff --git a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts index 7ca968de08..b7b443f0d1 100644 --- a/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts +++ b/packages/kap-server/src/transport/ws/v1/sessionEventBroadcaster.ts @@ -564,19 +564,6 @@ export class SessionEventBroadcaster { const terminalStatus = reason === 'cancelled' || reason === 'failed' || reason === 'blocked' ? 'aborted' : 'idle'; this.enqueueStatusChanged(state, pendingStatus(state.activity?.status()) ?? terminalStatus); - } else if (event.type === 'turn.ended' && state.activity?.status() === 'idle') { - // Defensive: a sub-agent's own lease and loop state are already released - // when its turn.ended is published (see loopService), so this activity - // read reflects only the OTHER agents. Session-wide idle here therefore - // cannot be a mid-main-turn foreground sub-agent (the main lease would - // report running) — it means the last active agent finished. Emit idle - // to clear a stale `running` seed: ensureState seeds lastStatus from - // ISessionActivity, which counts any agent's turn, so a session first - // activated while only a sub-agent was active (e.g. driven by a REST-only - // client that never triggered activation during the main turn) would - // otherwise stay `running` for its subscribers until the next main turn. - // Dedup makes this a no-op in every normally-tracked case. - this.enqueueStatusChanged(state, 'idle'); } // v1 wire compat: fan the legacy `background.task.*` spelling out next to // the native `task.*` event (see `legacyTaskEvent`) so unchanged v1 clients diff --git a/packages/kap-server/test/sessionEventBroadcaster.test.ts b/packages/kap-server/test/sessionEventBroadcaster.test.ts index 0b6b1517cb..0233a9de62 100644 --- a/packages/kap-server/test/sessionEventBroadcaster.test.ts +++ b/packages/kap-server/test/sessionEventBroadcaster.test.ts @@ -519,14 +519,9 @@ describe('SessionEventBroadcaster', () => { await bc.subscribe('s1', target); main.bus.emit(agentEvent('turn.started', { turnId: 1 })); - // Model the main turn's activity lease: while it is in flight the session - // activity reports running, so the defensive idle emit for sub-agent turns - // must stay silent. - lc.baseStatus = 'running'; // A foreground sub-agent runs and completes while the main turn is in flight. sub.bus.emit(agentEvent('turn.started', { turnId: 10 })); sub.bus.emit(agentEvent('turn.ended', { turnId: 10, reason: 'completed' })); - lc.baseStatus = 'idle'; main.bus.emit(agentEvent('turn.ended', { turnId: 1, reason: 'completed' })); await bc.getCursor('s1'); @@ -546,37 +541,6 @@ describe('SessionEventBroadcaster', () => { expect(envelopes.at(-1)!.type).toBe('event.session.status_changed'); }); - it('clears a stale running seed when a sub-agent turn ends as the last active agent', async () => { - // Defensive path: a session first activated while only a sub-agent turn is - // active seeds lastStatus='running' from ISessionActivity (which counts any - // agent's turn). When that sub-agent finishes, its own lease/loop are - // already released, so an idle session activity means it was the last - // active agent — the session is genuinely idle and subscribers must be told, - // otherwise they stay `running` until the next main turn. - const lc = new FakeLifecycle(); - lc.baseStatus = 'running'; // a background sub-agent turn is active - const sub = lc.addAgent('agent-0'); - sessions.set('s1', lc); - const { target, envelopes } = collectingTarget(); - await bc.subscribe('s1', target); // activates; seeds lastStatus='running' - - // The sub-agent's turn ends; nothing else is running anymore. - lc.baseStatus = 'idle'; - sub.bus.emit(agentEvent('turn.ended', { turnId: 10, reason: 'completed' })); - await bc.getCursor('s1'); - - expect(envelopes.map((e) => e.type)).toEqual([ - 'turn.ended', - 'event.session.status_changed', - ]); - expect(envelopes[1]).toMatchObject({ - type: 'event.session.status_changed', - session_id: 's1', - payload: { status: 'idle', previous_status: 'running' }, - }); - expect(envelopes[1]!.volatile).toBeUndefined(); - }); - it('broadcasts question requested / answered as durable v1 events', async () => { const lc = new FakeLifecycle(); lc.addAgent('main');