From 80dc6d5995be8d29882bcf2156a2f8a94b59ad0e Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Tue, 28 Apr 2026 10:05:43 +0800 Subject: [PATCH 01/52] feat(core): event monitor tool with throttled stdout streaming (Phase C) Add a new Monitor tool that spawns a long-running shell command and streams its stdout lines back to the agent as event notifications. This is Phase C from the background task management roadmap (#3634, #3666). What changes: - New MonitorRegistry (services/monitorRegistry.ts): per-monitor entry with lifecycle (running/completed/failed/cancelled), idle timeout auto-stop, max events auto-stop, AbortController-based cancellation. Follows the same structural pattern as BackgroundTaskRegistry. - New Monitor tool (tools/monitor.ts): spawns via child_process.spawn with independent AbortController (Ctrl+C won't kill monitors), separate stdout/stderr line buffers, token-bucket throttling (burst=5, sustain=1/s). Returns immediately with monitor ID; events stream as notifications. - Sleep interception in shell.ts: detectBlockedSleepPattern() blocks foreground `sleep N` (N>=2) and guides model to use Monitor or is_background instead. - Config integration: MonitorRegistry instantiation, accessor, shutdown cleanup (abortAll), lazy tool registration. - CLI wiring: notification callbacks in useGeminiStream.ts (interactive) and nonInteractiveCli.ts (headless), including hold-back loop abort on exit and SIGINT cleanup. What this PR doesn't do (gated on #3471/#3488): - Footer pill / dialog integration - task_stop / send_message integration Test plan: - 21 MonitorRegistry unit tests (lifecycle, idle timeout, max events, XML escaping, nonexistent ID guard, callback clearing) - 20 Monitor tool unit tests (validation, spawn, line buffering, separate stdout/stderr buffers, throttling, signal-killed path, turn isolation) - 7 detectBlockedSleepPattern unit tests - 2 E2E tests (monitor invocation, sleep interception) - Full core suite: 248 files / 6151 passed Co-authored-by: Qwen-Coder --- integration-tests/cli/monitor.test.ts | 44 ++ .../cli/sleep-interception.test.ts | 65 +++ packages/cli/src/nonInteractiveCli.ts | 23 ++ packages/cli/src/ui/hooks/useGeminiStream.ts | 16 + packages/core/src/config/config.ts | 13 + packages/core/src/index.ts | 1 + .../core/src/services/monitorRegistry.test.ts | 341 ++++++++++++++++ packages/core/src/services/monitorRegistry.ts | 286 +++++++++++++ packages/core/src/tools/monitor.test.ts | 361 ++++++++++++++++ packages/core/src/tools/monitor.ts | 385 ++++++++++++++++++ packages/core/src/tools/shell.test.ts | 42 +- packages/core/src/tools/shell.ts | 35 ++ packages/core/src/tools/tool-names.ts | 2 + 13 files changed, 1613 insertions(+), 1 deletion(-) create mode 100644 integration-tests/cli/monitor.test.ts create mode 100644 integration-tests/cli/sleep-interception.test.ts create mode 100644 packages/core/src/services/monitorRegistry.test.ts create mode 100644 packages/core/src/services/monitorRegistry.ts create mode 100644 packages/core/src/tools/monitor.test.ts create mode 100644 packages/core/src/tools/monitor.ts diff --git a/integration-tests/cli/monitor.test.ts b/integration-tests/cli/monitor.test.ts new file mode 100644 index 00000000000..981d9b9fb09 --- /dev/null +++ b/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); +}); diff --git a/integration-tests/cli/sleep-interception.test.ts b/integration-tests/cli/sleep-interception.test.ts new file mode 100644 index 00000000000..5b0d38c610f --- /dev/null +++ b/integration-tests/cli/sleep-interception.test.ts @@ -0,0 +1,65 @@ +/** + * @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 tool result should contain the blocked message + const toolLogs = rig.readToolLogs(); + const shellCall = toolLogs.find( + (t) => t.toolRequest.name === 'run_shell_command', + ); + if (shellCall?.toolResult) { + const resultStr = + typeof shellCall.toolResult === 'string' + ? shellCall.toolResult + : JSON.stringify(shellCall.toolResult); + expect(resultStr.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); +}); diff --git a/packages/cli/src/nonInteractiveCli.ts b/packages/cli/src/nonInteractiveCli.ts index 1541660b2a1..8e8e58600d2 100644 --- a/packages/cli/src/nonInteractiveCli.ts +++ b/packages/cli/src/nonInteractiveCli.ts @@ -341,6 +341,23 @@ 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, + }, + }); + }, + ); + let isFirstTurn = true; let modelOverride: string | undefined; while (true) { @@ -705,6 +722,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. @@ -719,6 +737,10 @@ export async function runNonInteractive( await new Promise((r) => setTimeout(r, 100)); } + // Abort any still-running monitors before exit — monitors are designed + // to run indefinitely, so we stop them rather than wait. + monitorRegistry.abortAll(); + const metrics = uiTelemetryService.getMetrics(); const usage = computeUsageFromMetrics(metrics); // Get stats for JSON format output @@ -771,6 +793,7 @@ export async function runNonInteractive( const reg = config.getBackgroundTaskRegistry(); reg.setNotificationCallback(undefined); reg.setRegisterCallback(undefined); + config.getMonitorRegistry().setNotificationCallback(undefined); process.stdout.removeListener('error', stdoutErrorHandler); // Cleanup signal handlers diff --git a/packages/cli/src/ui/hooks/useGeminiStream.ts b/packages/cli/src/ui/hooks/useGeminiStream.ts index fd80a6ec432..bd1ed1de510 100644 --- a/packages/cli/src/ui/hooks/useGeminiStream.ts +++ b/packages/cli/src/ui/hooks/useGeminiStream.ts @@ -2090,6 +2090,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 ( diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index de4d151b1ae..b36de17e270 100644 --- a/packages/core/src/config/config.ts +++ b/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 { DEFAULT_OTLP_ENDPOINT, DEFAULT_TELEMETRY_TARGET, @@ -544,6 +545,7 @@ export class Config { private promptRegistry!: PromptRegistry; private subagentManager!: SubagentManager; private readonly backgroundTaskRegistry = new BackgroundTaskRegistry(); + private readonly monitorRegistry = new MonitorRegistry(); private extensionManager!: ExtensionManager; private skillManager: SkillManager | null = null; private permissionManager: PermissionManager | null = null; @@ -1604,6 +1606,7 @@ export class Config { } this.backgroundTaskRegistry.abortAll(); + this.monitorRegistry.abortAll(); await this.cleanupArenaRuntime(); } catch (error) { @@ -2467,6 +2470,10 @@ export class Config { return this.backgroundTaskRegistry; } + getMonitorRegistry(): MonitorRegistry { + return this.monitorRegistry; + } + /** * Whether interactive permission prompts should be auto-denied. * True for background agents that have no UI to show prompts. @@ -2708,6 +2715,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(); } diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 985dda01c8f..67cc00147ca 100644 --- a/packages/core/src/index.ts +++ b/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 './utils/bareMode.js'; // ============================================================================ diff --git a/packages/core/src/services/monitorRegistry.test.ts b/packages/core/src/services/monitorRegistry.test.ts new file mode 100644 index 00000000000..d9b77af3328 --- /dev/null +++ b/packages/core/src/services/monitorRegistry.test.ts @@ -0,0 +1,341 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { MonitorRegistry, type MonitorEntry } from './monitorRegistry.js'; + +function createEntry(overrides: Partial = {}): MonitorEntry { + return { + monitorId: 'mon-1', + command: 'tail -f /var/log/app.log', + description: 'watch app logs', + status: 'running' as const, + startTime: Date.now(), + abortController: new AbortController(), + eventCount: 0, + lastEventTime: 0, + maxEvents: 1000, + idleTimeoutMs: 300_000, + ...overrides, + }; +} + +describe('MonitorRegistry', () => { + let registry: MonitorRegistry; + + beforeEach(() => { + vi.useFakeTimers(); + registry = new MonitorRegistry(); + }); + + afterEach(() => { + // Cancel all to clear idle timers before restoring real timers + registry.abortAll(); + vi.useRealTimers(); + }); + + it('registers and retrieves a monitor', () => { + const entry = createEntry(); + registry.register(entry); + expect(registry.get('mon-1')).toBe(entry); + }); + + it('emits event notification via callback', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + registry.emitEvent('mon-1', 'hello world'); + + expect(callback).toHaveBeenCalledOnce(); + const [displayText, modelText, meta] = callback.mock.calls[0] as [ + string, + string, + { monitorId: string; status: string; eventCount: number }, + ]; + expect(displayText).toContain('watch app logs'); + expect(displayText).toContain('hello world'); + expect(modelText).toContain('monitor'); + expect(modelText).toContain('running'); + expect(modelText).toContain('1'); + expect(modelText).toContain('hello world'); + expect(meta.monitorId).toBe('mon-1'); + expect(meta.status).toBe('running'); + expect(meta.eventCount).toBe(1); + }); + + it('increments eventCount on each emitEvent', () => { + registry.register(createEntry()); + registry.emitEvent('mon-1', 'line 1'); + registry.emitEvent('mon-1', 'line 2'); + registry.emitEvent('mon-1', 'line 3'); + + expect(registry.get('mon-1')!.eventCount).toBe(3); + }); + + it('completes a monitor and emits terminal notification', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + registry.complete('mon-1', 0); + + const entry = registry.get('mon-1')!; + expect(entry.status).toBe('completed'); + expect(entry.endTime).toBeDefined(); + expect(callback).toHaveBeenCalledOnce(); + const [displayText, modelText] = callback.mock.calls[0] as [string, string]; + expect(displayText).toContain('completed'); + expect(modelText).toContain('completed'); + expect(modelText).toContain('Exited with code 0'); + }); + + it('fails a monitor and emits terminal notification', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + registry.fail('mon-1', 'spawn ENOENT'); + + const entry = registry.get('mon-1')!; + expect(entry.status).toBe('failed'); + expect(callback).toHaveBeenCalledOnce(); + const [displayText, modelText] = callback.mock.calls[0] as [string, string]; + expect(displayText).toContain('failed'); + expect(modelText).toContain('failed'); + expect(modelText).toContain('spawn ENOENT'); + }); + + it('cancels a running monitor and aborts its controller', () => { + const ac = new AbortController(); + registry.register(createEntry({ abortController: ac })); + + registry.cancel('mon-1'); + + expect(registry.get('mon-1')!.status).toBe('cancelled'); + expect(ac.signal.aborted).toBe(true); + }); + + it('no-op: complete after cancel (one-shot terminal guard)', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + registry.cancel('mon-1'); + registry.complete('mon-1', 0); + + expect(registry.get('mon-1')!.status).toBe('cancelled'); + expect(callback).toHaveBeenCalledTimes(1); // only cancel notification + }); + + it('no-op: emitEvent after cancel', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + registry.cancel('mon-1'); + registry.emitEvent('mon-1', 'late line'); + + // Only the cancel notification, no event notification + expect(callback).toHaveBeenCalledTimes(1); + expect(registry.get('mon-1')!.eventCount).toBe(0); + }); + + it('auto-stops when maxEvents is reached', () => { + const ac = new AbortController(); + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry({ maxEvents: 3, abortController: ac })); + + registry.emitEvent('mon-1', 'line 1'); + registry.emitEvent('mon-1', 'line 2'); + registry.emitEvent('mon-1', 'line 3'); // triggers auto-stop + + expect(registry.get('mon-1')!.status).toBe('completed'); + expect(ac.signal.aborted).toBe(true); + // 3 event notifications + 1 terminal notification ("Max events reached") + expect(callback).toHaveBeenCalledTimes(4); + const [, terminalModelText] = callback.mock.calls[3] as [string, string]; + expect(terminalModelText).toContain('Max events reached'); + expect(terminalModelText).toContain('completed'); + }); + + it('auto-stops on idle timeout', () => { + const ac = new AbortController(); + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register( + createEntry({ + idleTimeoutMs: 5000, + abortController: ac, + }), + ); + + // Fast-forward past the idle timeout + vi.advanceTimersByTime(5001); + + expect(registry.get('mon-1')!.status).toBe('completed'); + expect(ac.signal.aborted).toBe(true); + // Terminal notification from idle timeout + expect(callback).toHaveBeenCalledOnce(); + const [, modelText] = callback.mock.calls[0] as [string, string]; + expect(modelText).toContain('Idle timeout'); + }); + + it('resets idle timer on emitEvent', () => { + const ac = new AbortController(); + registry.register( + createEntry({ + idleTimeoutMs: 5000, + abortController: ac, + }), + ); + + // Advance 4s, emit event, advance 4s again — should NOT timeout + vi.advanceTimersByTime(4000); + registry.emitEvent('mon-1', 'keep alive'); + vi.advanceTimersByTime(4000); + + expect(registry.get('mon-1')!.status).toBe('running'); + + // Now advance past the timeout + vi.advanceTimersByTime(2000); + expect(registry.get('mon-1')!.status).toBe('completed'); + }); + + it('getRunning filters by status', () => { + registry.register(createEntry({ monitorId: 'a' })); + registry.register(createEntry({ monitorId: 'b' })); + registry.register(createEntry({ monitorId: 'c' })); + + registry.complete('a', 0); + registry.cancel('c'); + + const running = registry.getRunning(); + expect(running).toHaveLength(1); + expect(running[0].monitorId).toBe('b'); + }); + + it('abortAll cancels all running monitors', () => { + const ac1 = new AbortController(); + const ac2 = new AbortController(); + registry.register(createEntry({ monitorId: 'a', abortController: ac1 })); + registry.register(createEntry({ monitorId: 'b', abortController: ac2 })); + + registry.abortAll(); + + expect(ac1.signal.aborted).toBe(true); + expect(ac2.signal.aborted).toBe(true); + expect(registry.get('a')!.status).toBe('cancelled'); + expect(registry.get('b')!.status).toBe('cancelled'); + }); + + it('truncates long event lines', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + const longLine = 'x'.repeat(3000); + registry.emitEvent('mon-1', longLine); + + const [, modelText] = callback.mock.calls[0] as [string, string]; + expect(modelText).toContain('...[truncated]'); + expect(modelText).not.toContain('x'.repeat(3000)); + }); + + it('escapes XML metacharacters in event lines', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + registry.emitEvent('mon-1', ''); + + const [, modelText] = callback.mock.calls[0] as [string, string]; + expect(modelText).toContain('<script>'); + expect(modelText).not.toContain('