fix: keep monitor output flowing until stdout closes and let print mode exit with monitors running

Author: chengluyu

.changeset/monitor-tool.md

@@ -2,4 +2,4 @@
 "@moonshot-ai/kimi-code": minor
 ---
 
-Add an experimental Monitor tool that runs a background command and delivers each new output line to the agent as it appears. Enable it with `KIMI_CODE_EXPERIMENTAL_MONITOR=1` or `/experiments`.
+Add an experimental Monitor tool, enabled with `KIMI_CODE_EXPERIMENTAL_MONITOR=1` or `/experiments`, that runs a background command and delivers each new output line to the agent as it appears.

apps/kimi-code/src/cli/v2/run-v2-print.ts

@@ -75,6 +75,7 @@ import {
   shouldEnableTelemetry,
   shutdownTelemetry,
 } from '@moonshot-ai/kimi-telemetry';
+import { isMonitorTaskId } from '@moonshot-ai/agent-core-v2/agent/tools/task/monitor/monitor';
 import type { GoalUpdated } from '@moonshot-ai/agent-core-v2/features/goal/goalOps';
 import type { TurnEnded } from '@moonshot-ai/agent-core-v2/agent/loop/turnOps';
 import type {
@@ -987,7 +988,7 @@ function countPendingBackgroundTasks(session: ISessionScopeHandle): number {
   for (const agent of agentManager.list()) {
     const handle = agentManager.handleOf(agent.agentId);
     if (handle === undefined) continue;
-    count += handle.accessor.get(IAgentTaskService).list(true).length;
+    count += handle.accessor.get(IAgentTaskService).list(true).filter((task) => !isMonitorTaskId(task.taskId)).length;
   }
   return count;
 }
@@ -1114,6 +1115,7 @@ async function drainBackgroundTasks(
       if (handle === undefined) continue;
       const taskService = handle.accessor.get(IAgentTaskService);
       for (const task of taskService.list(true)) {
+        if (isMonitorTaskId(task.taskId)) continue;
         activeCount++;
         if (seen.has(task.taskId)) continue;
         seen.add(task.taskId);

apps/kimi-code/test/cli/v2-run-print.test.ts

@@ -360,6 +360,35 @@ describe('runV2Print', () => {
     expect(app.dispose).toHaveBeenCalled();
   });
 
+  it.each(['drain', 'steer'])('does not wait for a running monitor in %s mode before shutting down', async (mode) => {
+    const { app, agentServices, sessionServices, appServices } = makeFakeHarness();
+    const config = appServices.get(IConfigService) as { get: ReturnType<typeof vi.fn> };
+    config.get.mockImplementation((section: string) =>
+      section === 'defaultModel' ? 'k2' : section === 'background' ? { printBackgroundMode: mode } : undefined,
+    );
+    const lifecycle = sessionServices.get(IAgentLifecycleService) as { list: ReturnType<typeof vi.fn> };
+    lifecycle.list.mockReturnValue([{ agentId: 'main' }]);
+    const taskService = agentServices.get(IAgentTaskService) as {
+      list: ReturnType<typeof vi.fn>;
+      stopAllOnExit: ReturnType<typeof vi.fn>;
+      wait?: ReturnType<typeof vi.fn>;
+      suppressTerminalNotification?: ReturnType<typeof vi.fn>;
+    };
+    taskService.list.mockReturnValue([{ taskId: 'monitor-abc12345', status: 'running' }]);
+    taskService.wait = vi.fn(() => new Promise(() => {}));
+    taskService.suppressTerminalNotification = vi.fn(async () => {});
+    (agentServices.get(IAgentGoalService) as { getGoal: ReturnType<typeof vi.fn> }).getGoal.mockReturnValue({});
+    mocks.bootstrap.mockReturnValue({ app });
+    mocks.ensureMainAgent.mockResolvedValue({ agentId: 'main', generation: 1 });
+
+    const stderr = writer();
+    await runV2Print(opts() as never, '1.2.3-test', { stdout: writer(), stderr });
+
+    expect(stderr.text()).not.toContain('Warning');
+    expect(taskService.wait).not.toHaveBeenCalled();
+    expect(taskService.stopAllOnExit).toHaveBeenCalled();
+  }, 10_000);
+
   it('passes explicit skill dirs from --skillsDir into bootstrap args', async () => {
     const stdout = writer();
     const stderr = writer();

packages/agent-core-v2/src/agent/tools/os/bash/process-task.ts

@@ -68,6 +68,7 @@ export class ProcessTask implements AgentTask {
     let settlement: AgentTaskSettlement;
     try {
       const exitCode = await this.proc.wait();
+      if (this.stdoutEvents) await waitForStreamEndOrAbort(streamDrained, sink.signal);
       await waitForStreamDrain(streamDrained);
       this.exitCode = exitCode;
       settlement = {
@@ -134,6 +135,22 @@ async function waitForStreamDrain(streamDrained: Promise<void>): Promise<void> {
   }
 }
 
+async function waitForStreamEndOrAbort(streamDrained: Promise<void>, signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return;
+  let onAbort: (() => void) | undefined;
+  try {
+    await Promise.race([
+      streamDrained.catch(() => {}),
+      new Promise<void>((resolve) => {
+        onAbort = resolve;
+        signal.addEventListener('abort', onAbort, { once: true });
+      }),
+    ]);
+  } finally {
+    if (onAbort !== undefined) signal.removeEventListener('abort', onAbort);
+  }
+}
+
 async function waitForStreamDrainSettled(streamDrained: Promise<void>): Promise<void> {
   try {
     await waitForStreamDrain(streamDrained);

packages/agent-core-v2/test/agent/task/taskManager.test.ts

@@ -1645,6 +1645,47 @@ describe('AgentTaskService monitor events', () => {
     await fixture.ctx.dispose();
   });
 
+  it('keeps delivering monitor lines after the shell exits until stdout closes', async () => {
+    const fixture = createAgentTaskService();
+    const notes = captureNotifications(fixture);
+    const { proc, stdout, stderr } = controllableProcess();
+    const shellExit = Promise.withResolvers<number>();
+
+    const taskId = fixture.manager.registerTask(
+      new MonitorProcessTask({ ...proc, wait: () => shellExit.promise }, 'tail -F app.log &', 'watch app log'),
+    );
+    shellExit.resolve(0);
+    await new Promise((resolve) => setTimeout(resolve, 500));
+    expect(fixture.manager.getTask(taskId)?.status).toBe('running');
+
+    stdout.end('late line\n');
+    stderr.end();
+    await waitForTerminal(fixture.manager, taskId);
+    await vi.waitFor(() => {
+      expect(notes).toHaveLength(2);
+    });
+    expect(noteText(notes[0]!)).toContain('late line');
+    expect(noteText(notes[1]!)).toContain('type="task.completed"');
+    await fixture.ctx.dispose();
+  });
+
+  it('stops a monitor whose shell exited while its stdout is still open', async () => {
+    const fixture = createAgentTaskService();
+    captureNotifications(fixture);
+    const { proc } = controllableProcess();
+    const shellExit = Promise.withResolvers<number>();
+
+    const taskId = fixture.manager.registerTask(
+      new MonitorProcessTask({ ...proc, wait: () => shellExit.promise }, 'tail -F app.log &', 'watch app log'),
+    );
+    shellExit.resolve(0);
+    await new Promise((resolve) => setTimeout(resolve, 50));
+
+    expect(await fixture.manager.stop(taskId)).toMatchObject({ status: 'killed' });
+    expect(proc.kill).toHaveBeenCalledWith('SIGTERM');
+    await fixture.ctx.dispose();
+  });
+
   it('does not turn stdout of an ordinary background command into events', async () => {
     const fixture = createAgentTaskService();
     const notes = captureNotifications(fixture);