diff --git a/.changeset/calm-wires-persist.md b/.changeset/calm-wires-persist.md new file mode 100644 index 0000000000..3a8fe8f9a5 --- /dev/null +++ b/.changeset/calm-wires-persist.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Prevent recent ACP assistant responses from being lost when a session closes. diff --git a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts index 4ccbbdd6f5..5547db1e36 100644 --- a/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts +++ b/packages/agent-core-v2/src/session/agentLifecycle/agentLifecycleService.ts @@ -9,8 +9,9 @@ * Agent-scoped telemetry view. New logs receive a metadata * envelope while non-empty unversioned logs are rejected. Removal awaits the * agent task manager's graceful exit policy before draining turns and full - * compaction, then disposing the child scope. Fans session-level - * permission-mode switches out to every live agent. Bound at Session scope. + * compaction, flushing the replayable wire, then disposing the child scope. + * Fans session-level permission-mode switches out to every live agent. Bound + * at Session scope. * * No agent id is special here: the main agent is simply the agent created * with the conventional `MAIN_AGENT_ID`, and `fork` requires its source to @@ -268,8 +269,12 @@ export class AgentLifecycleService extends Disposable implements IAgentLifecycle compaction.abortController.abort(reason); } await Promise.all([loop.settled(), compactionSettled]); - handle.dispose(); - this.onDidDisposeEmitter.fire(agentId); + try { + await handle.accessor.get(IWireService).flush(); + } finally { + handle.dispose(); + this.onDidDisposeEmitter.fire(agentId); + } } } diff --git a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts index 628b5af27d..0642de13bf 100644 --- a/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts +++ b/packages/agent-core-v2/test/session/agentLifecycle/agentLifecycle.test.ts @@ -36,6 +36,7 @@ import { createMcpOAuthStore } from '#/app/mcpConfig/oauthStore'; import { ISessionSubagentService } from '#/session/subagent/subagent'; import { SessionSubagentService } from '#/session/subagent/subagentService'; import '#/agent/mcp/mcpService'; +import { IWireService } from '#/wire/wire'; import '#/wire/wireService'; import { IAgentTaskService } from '#/agent/task/task'; import { ISessionCronService } from '#/session/cron/sessionCronService'; @@ -455,6 +456,52 @@ describe('AgentLifecycleService', () => { expect(removed).toBe(true); }); + it('remove waits for the agent wire to flush before completing', async () => { + const svc = ix.get(IAgentLifecycleService); + const handle = await svc.create({ agentId: 'main' }); + let markFlushStarted!: () => void; + const flushStarted = new Promise((resolve) => { + markFlushStarted = resolve; + }); + let releaseFlush!: () => void; + const flushReleased = new Promise((resolve) => { + releaseFlush = resolve; + }); + const flush = vi + .spyOn(handle.accessor.get(IWireService), 'flush') + .mockImplementation(async () => { + markFlushStarted(); + await flushReleased; + }); + + let removed = false; + const removal = svc.remove('main').then(() => { + removed = true; + }); + await flushStarted; + expect(removed).toBe(false); + + releaseFlush(); + await removal; + expect(flush).toHaveBeenCalledOnce(); + expect(removed).toBe(true); + }); + + it('remove still disposes the agent when the wire flush rejects', async () => { + const svc = ix.get(IAgentLifecycleService); + const handle = await svc.create({ agentId: 'main' }); + const flushError = new Error('wire flush failed'); + vi.spyOn(handle.accessor.get(IWireService), 'flush').mockRejectedValueOnce(flushError); + const disposed: string[] = []; + disposables.add(svc.onDidDispose((id) => disposed.push(id))); + + await expect(svc.remove('main')).rejects.toBe(flushError); + + expect(svc.get('main')).toBeUndefined(); + expect(disposed).toEqual(['main']); + expect(() => handle.accessor.get(IWireService)).toThrow(); + }); + it('ignites the self-wiring toolDedupe plugin so its listeners exist before the first turn', async () => { const svc = ix.get(IAgentLifecycleService); await svc.create({ agentId: 'main' });