feat(core): event monitor tool with throttled stdout streaming (Phase C)

integration-tests/cli/monitor.test.ts

@@ -0,0 +1,44 @@
+/**
+ * @license
+ * Copyright 2025 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { describe, it, expect, afterEach } from 'vitest';
+import { TestRig, validateModelOutput } from '../test-helper.js';
+
+describe('monitor-tool', () => {
+  let rig: TestRig;
+
+  afterEach(async () => {
+    if (rig) {
+      await rig.cleanup();
+    }
+  });
+
+  it('should have monitor tool registered', async () => {
+    rig = new TestRig();
+    await rig.setup('monitor-tool-registered');
+
+    const result = await rig.run(
+      'Do you have access to a tool called "monitor"? Reply with just "yes" or "no".',
+    );
+
+    validateModelOutput(result, null, 'monitor tool registered');
+    expect(result.toLowerCase()).toContain('yes');
+  });
+
+  it('should call monitor tool when asked to watch a command', async () => {
+    rig = new TestRig();
+    await rig.setup('monitor-tool-call');
+
+    const result = await rig.run(
+      'Use the monitor tool to watch this command: for i in 1 2 3; do echo "EVENT_$i"; sleep 0.3; done. ' +
+        'Set description to "test events". After starting the monitor, just say "Monitor launched."',
+    );
+
+    const foundMonitor = await rig.waitForToolCall('monitor');
+    expect(foundMonitor).toBeTruthy();
+    validateModelOutput(result, null, 'monitor tool call');
+  }, 30000);
+});

integration-tests/cli/sleep-interception.test.ts

@@ -0,0 +1,55 @@
+/**
+ * @license
+ * Copyright 2025 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import { describe, it, expect, afterEach } from 'vitest';
+import { TestRig, validateModelOutput } from '../test-helper.js';
+
+describe('sleep-interception', () => {
+  let rig: TestRig;
+
+  afterEach(async () => {
+    if (rig) {
+      await rig.cleanup();
+    }
+  });
+
+  it('should block sleep >= 2s and mention Monitor in guidance', async () => {
+    rig = new TestRig();
+    await rig.setup('sleep-blocked');
+
+    const result = await rig.run(
+      'Run this exact shell command: sleep 5. ' +
+        'If the command is blocked, say "BLOCKED" and explain why. ' +
+        'If it succeeds, say "SUCCESS".',
+    );
+
+    validateModelOutput(result, null, 'sleep blocked');
+
+    // The model should report being blocked, since sleep 5 triggers interception
+    const foundShell = await rig.waitForToolCall('run_shell_command');
+    expect(foundShell).toBeTruthy();
+
+    // The model's output should mention it was blocked
+    expect(result.toLowerCase()).toContain('blocked');
+  }, 30000);
+
+  it('should allow sleep < 2s', async () => {
+    rig = new TestRig();
+    await rig.setup('sleep-allowed');
+
+    const result = await rig.run(
+      'Run this exact shell command: sleep 1. Then say "DONE".',
+    );
+
+    validateModelOutput(result, null, 'sleep allowed');
+
+    const foundShell = await rig.waitForToolCall('run_shell_command');
+    expect(foundShell).toBeTruthy();
+
+    // Should not be blocked — model should complete successfully
+    expect(result.toLowerCase()).not.toContain('blocked');
+  }, 30000);
+});

packages/cli/src/nonInteractiveCli.test.ts

@@ -170,6 +170,12 @@ describe('runNonInteractive', () => {
         hasUnfinalizedTasks: vi.fn().mockReturnValue(false),
         abortAll: vi.fn(),
       }),
+      getMonitorRegistry: vi.fn().mockReturnValue({
+        setNotificationCallback: vi.fn(),
+        setRegisterCallback: vi.fn(),
+        getRunning: vi.fn().mockReturnValue([]),
+        abortAll: vi.fn(),
+      }),
     } as unknown as Config;
 
     mockSettings = {

packages/cli/src/nonInteractiveCli.ts

@@ -341,6 +341,31 @@ export async function runNonInteractive(
         });
       });
 
+      // Register monitor notification callback onto the shared queue.
+      const monitorRegistry = config.getMonitorRegistry();
+      monitorRegistry.setNotificationCallback(
+        (displayText, modelText, meta) => {
+          localQueue.push({
+            displayText,
+            modelText,
+            sendMessageType: SendMessageType.Notification,
+            sdkNotification: {
+              task_id: meta.monitorId,
+              tool_use_id: meta.toolUseId,
+              status: meta.status,
+            },
+          });
+        },
+      );
+
+      monitorRegistry.setRegisterCallback((entry) => {
+        adapter.emitSystemMessage('task_started', {
+          task_id: entry.monitorId,
+          tool_use_id: entry.toolUseId,
+          description: entry.description,
+        });
+      });
+
       let isFirstTurn = true;
       let modelOverride: string | undefined;
       while (true) {
@@ -705,6 +730,7 @@ export async function runNonInteractive(
           while (true) {
             if (abortController.signal.aborted) {
               registry.abortAll();
+              monitorRegistry.abortAll();
               // Flush queued terminal notifications before handleCancellationError
               // exits so stream-json consumers always see a task_notification paired
               // with every task_started.
@@ -718,7 +744,12 @@ export async function runNonInteractive(
             // running ones: cancel() marks status 'cancelled' synchronously
             // but the notification is emitted later by the natural handler,
             // and SDK consumers need every task_started paired with one.
-            if (!registry.hasUnfinalizedTasks() && localQueue.length === 0)
+            // Also wait for running monitors to finish streaming events.
+            if (
+              !registry.hasUnfinalizedTasks() &&
+              monitorRegistry.getRunning().length === 0 &&
+              localQueue.length === 0
+            )
               break;
             await new Promise((r) => setTimeout(r, 100));
           }
@@ -775,6 +806,9 @@ export async function runNonInteractive(
       const reg = config.getBackgroundTaskRegistry();
       reg.setNotificationCallback(undefined);
       reg.setRegisterCallback(undefined);
+      const monReg = config.getMonitorRegistry();
+      monReg.setNotificationCallback(undefined);
+      monReg.setRegisterCallback(undefined);
 
       process.stdout.removeListener('error', stdoutErrorHandler);
       // Cleanup signal handlers

packages/cli/src/ui/commands/tasksCommand.ts

@@ -53,10 +53,7 @@ export const tasksCommand: SlashCommand = {
     }
 
     const now = Date.now();
-    const lines: string[] = [
-      `Background shells (${entries.length} total)`,
-      '',
-    ];
+    const lines: string[] = [`Background shells (${entries.length} total)`, ''];
     for (const entry of entries) {
       const endTime = entry.endTime ?? now;
       const runtime = formatDuration(endTime - entry.startTime, {

packages/cli/src/ui/hooks/useGeminiStream.test.tsx

@@ -212,6 +212,9 @@ describe('useGeminiStream', () => {
       getBackgroundTaskRegistry: vi.fn(() => ({
         setNotificationCallback: vi.fn(),
       })),
+      getMonitorRegistry: vi.fn(() => ({
+        setNotificationCallback: vi.fn(),
+      })),
     } as unknown as Config;
     mockOnDebugMessage = vi.fn();
     mockHandleSlashCommand = vi.fn().mockResolvedValue(false);

packages/cli/src/ui/hooks/useGeminiStream.ts

@@ -2253,6 +2253,22 @@ export const useGeminiStream = (
     };
   }, [config]);
 
+  // Register monitor notification callback onto the shared queue.
+  useEffect(() => {
+    const registry = config.getMonitorRegistry();
+    registry.setNotificationCallback((displayText, modelText) => {
+      notificationQueueRef.current.push({
+        displayText,
+        modelText,
+        sendMessageType: SendMessageType.Notification,
+      });
+      setNotificationTrigger((n) => n + 1);
+    });
+    return () => {
+      registry.setNotificationCallback(undefined);
+    };
+  }, [config]);
+
   // When idle, drain the unified queue one item at a time.
   useEffect(() => {
     if (

packages/core/src/config/config.ts

@@ -61,6 +61,7 @@ import { PermissionManager } from '../permissions/permission-manager.js';
 import { SubagentManager } from '../subagents/subagent-manager.js';
 import type { SubagentConfig } from '../subagents/types.js';
 import { BackgroundTaskRegistry } from '../agents/background-tasks.js';
+import { MonitorRegistry } from '../services/monitorRegistry.js';
 import { BackgroundShellRegistry } from '../services/backgroundShellRegistry.js';
 import {
   DEFAULT_OTLP_ENDPOINT,
@@ -546,6 +547,7 @@ export class Config {
   private promptRegistry!: PromptRegistry;
   private subagentManager!: SubagentManager;
   private readonly backgroundTaskRegistry = new BackgroundTaskRegistry();
+  private readonly monitorRegistry = new MonitorRegistry();
   private readonly backgroundShellRegistry = new BackgroundShellRegistry();
   private extensionManager!: ExtensionManager;
   private skillManager: SkillManager | null = null;
@@ -1612,6 +1614,7 @@ export class Config {
       }
 
       this.backgroundTaskRegistry.abortAll();
+      this.monitorRegistry.abortAll();
       this.backgroundShellRegistry.abortAll();
 
       await this.cleanupArenaRuntime();
@@ -2492,6 +2495,10 @@ export class Config {
     return this.backgroundTaskRegistry;
   }
 
+  getMonitorRegistry(): MonitorRegistry {
+    return this.monitorRegistry;
+  }
+
   getBackgroundShellRegistry(): BackgroundShellRegistry {
     return this.backgroundShellRegistry;
   }
@@ -2745,6 +2752,12 @@ export class Config {
       });
     }
 
+    // Register monitor tool
+    await registerLazy(ToolNames.MONITOR, async () => {
+      const { MonitorTool } = await import('../tools/monitor.js');
+      return new MonitorTool(this);
+    });
+
     if (!options?.skipDiscovery) {
       await registry.discoverAllTools();
     }

packages/core/src/index.ts

@@ -147,6 +147,7 @@ export * from './services/sessionService.js';
 export * from './services/sessionTitle.js';
 export { stripTerminalControlSequences } from './utils/terminalSafe.js';
 export * from './services/shellExecutionService.js';
+export * from './services/monitorRegistry.js';
 export * from './services/backgroundShellRegistry.js';
 export * from './services/toolUseSummary.js';
 export * from './utils/bareMode.js';

packages/core/src/permissions/permission-manager.ts

@@ -418,6 +418,7 @@ export class PermissionManager {
     'cron_create',
     'cron_list',
     'cron_delete',
+    'monitor',
   ]);
 
   /**