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..452652019c6 --- /dev/null +++ b/integration-tests/cli/sleep-interception.test.ts @@ -0,0 +1,78 @@ +/** + * @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); + + it('should block sleep >= 2s even when followed by a trailing comment', async () => { + // The `trimTrailingShellComment` state machine strips trailing `#...` + // comments before matching the sleep pattern, so a model trying to + // route around interception with `sleep 5 # wait for db` must still + // be blocked. This test locks in that behavior end-to-end. + rig = new TestRig(); + await rig.setup('sleep-blocked-trailing-comment'); + + const result = await rig.run( + 'Run this exact shell command: sleep 5 # wait for db. ' + + 'If the command is blocked, say "BLOCKED" and explain why. ' + + 'If it succeeds, say "SUCCESS".', + ); + + validateModelOutput(result, null, 'sleep blocked with trailing comment'); + + const foundShell = await rig.waitForToolCall('run_shell_command'); + expect(foundShell).toBeTruthy(); + + // Model must report it was blocked despite the trailing comment. + expect(result.toLowerCase()).toContain('blocked'); + }, 30000); +}); diff --git a/integration-tests/terminal-capture/subagent-flicker-regression.ts b/integration-tests/terminal-capture/subagent-flicker-regression.ts index 927f2155c2b..2d76504c0f6 100644 --- a/integration-tests/terminal-capture/subagent-flicker-regression.ts +++ b/integration-tests/terminal-capture/subagent-flicker-regression.ts @@ -73,12 +73,7 @@ import { type IncomingMessage, type ServerResponse, } from 'node:http'; -import { - existsSync, - mkdirSync, - rmSync, - writeFileSync, -} from 'node:fs'; +import { existsSync, mkdirSync, rmSync, writeFileSync } from 'node:fs'; import { basename, dirname, join, resolve } from 'node:path'; import { tmpdir } from 'node:os'; import { fileURLToPath } from 'node:url'; diff --git a/packages/cli/src/config/config.test.ts b/packages/cli/src/config/config.test.ts index 751535fcd85..9d59ad707cd 100644 --- a/packages/cli/src/config/config.test.ts +++ b/packages/cli/src/config/config.test.ts @@ -1059,6 +1059,7 @@ describe('loadCliConfig telemetry', () => { describe('mergeExcludeTools', () => { const defaultExcludes = [ ToolNames.SHELL, + ToolNames.MONITOR, ToolNames.EDIT, ToolNames.WRITE_FILE, ]; @@ -1125,6 +1126,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).toContain(ToolNames.SHELL); + expect(excludedTools).toContain(ToolNames.MONITOR); expect(excludedTools).toContain(ToolNames.EDIT); expect(excludedTools).toContain(ToolNames.WRITE_FILE); }); @@ -1144,6 +1146,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).toContain(ToolNames.SHELL); + expect(excludedTools).toContain(ToolNames.MONITOR); expect(excludedTools).toContain(ToolNames.EDIT); expect(excludedTools).toContain(ToolNames.WRITE_FILE); }); @@ -1164,6 +1167,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).toContain(ToolNames.SHELL); + expect(excludedTools).toContain(ToolNames.MONITOR); expect(excludedTools).toContain(ToolNames.EDIT); expect(excludedTools).toContain(ToolNames.WRITE_FILE); }); @@ -1181,6 +1185,43 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).not.toContain(ToolNames.SHELL); + expect(excludedTools).toContain(ToolNames.MONITOR); + expect(excludedTools).toContain(ToolNames.EDIT); + expect(excludedTools).toContain(ToolNames.WRITE_FILE); + }); + + it('should not exclude monitor when explicitly allowed in tools.allowed', async () => { + process.argv = ['node', 'script.js', '-p', 'test']; + const argv = await parseArguments(); + const settings: Settings = { + tools: { + allowed: [ToolNames.MONITOR], + }, + }; + + const config = await loadCliConfig(settings, argv, undefined, []); + + const excludedTools = config.getPermissionsDeny(); + expect(excludedTools).toContain(ToolNames.SHELL); + expect(excludedTools).not.toContain(ToolNames.MONITOR); + expect(excludedTools).toContain(ToolNames.EDIT); + expect(excludedTools).toContain(ToolNames.WRITE_FILE); + }); + + it('should honor monitor aliases in tools.allowed for non-interactive exclusions', async () => { + process.argv = ['node', 'script.js', '-p', 'test']; + const argv = await parseArguments(); + const settings: Settings = { + tools: { + allowed: ['Monitor', 'Shell(git status)'], + }, + }; + + const config = await loadCliConfig(settings, argv, undefined, []); + + const excludedTools = config.getPermissionsDeny(); + expect(excludedTools).not.toContain(ToolNames.MONITOR); + expect(excludedTools).not.toContain(ToolNames.SHELL); expect(excludedTools).toContain(ToolNames.EDIT); expect(excludedTools).toContain(ToolNames.WRITE_FILE); }); @@ -1198,6 +1239,25 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).not.toContain(ToolNames.SHELL); + expect(excludedTools).toContain(ToolNames.MONITOR); + expect(excludedTools).toContain(ToolNames.EDIT); + expect(excludedTools).toContain(ToolNames.WRITE_FILE); + }); + + it('should not exclude monitor when explicitly allowed in tools.core', async () => { + process.argv = ['node', 'script.js', '-p', 'test']; + const argv = await parseArguments(); + const settings: Settings = { + tools: { + core: [ToolNames.MONITOR], + }, + }; + + const config = await loadCliConfig(settings, argv, undefined, []); + + const excludedTools = config.getPermissionsDeny(); + expect(excludedTools).toContain(ToolNames.SHELL); + expect(excludedTools).not.toContain(ToolNames.MONITOR); expect(excludedTools).toContain(ToolNames.EDIT); expect(excludedTools).toContain(ToolNames.WRITE_FILE); }); @@ -1218,6 +1278,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).toContain(ToolNames.SHELL); + expect(excludedTools).toContain(ToolNames.MONITOR); expect(excludedTools).not.toContain(ToolNames.EDIT); expect(excludedTools).not.toContain(ToolNames.WRITE_FILE); }); @@ -1238,6 +1299,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).not.toContain(ToolNames.SHELL); + expect(excludedTools).not.toContain(ToolNames.MONITOR); expect(excludedTools).not.toContain(ToolNames.EDIT); expect(excludedTools).not.toContain(ToolNames.WRITE_FILE); }); @@ -1251,6 +1313,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).not.toContain(ToolNames.SHELL); + expect(excludedTools).not.toContain(ToolNames.MONITOR); expect(excludedTools).not.toContain(ToolNames.EDIT); expect(excludedTools).not.toContain(ToolNames.WRITE_FILE); }); @@ -1276,6 +1339,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).not.toContain(ToolNames.SHELL); + expect(excludedTools).not.toContain(ToolNames.MONITOR); expect(excludedTools).not.toContain(ToolNames.EDIT); expect(excludedTools).not.toContain(ToolNames.WRITE_FILE); } @@ -1288,6 +1352,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).not.toContain(ToolNames.SHELL); + expect(excludedTools).not.toContain(ToolNames.MONITOR); expect(excludedTools).not.toContain(ToolNames.EDIT); expect(excludedTools).not.toContain(ToolNames.WRITE_FILE); }); @@ -1308,6 +1373,7 @@ describe('Approval mode tool exclusion logic', () => { const excludedTools = config.getPermissionsDeny(); expect(excludedTools).toContain('custom_tool'); // From settings expect(excludedTools).toContain(ToolNames.SHELL); // From approval mode + expect(excludedTools).toContain(ToolNames.MONITOR); // From approval mode expect(excludedTools).not.toContain(ToolNames.EDIT); // Should be allowed in auto-edit expect(excludedTools).not.toContain(ToolNames.WRITE_FILE); // Should be allowed in auto-edit }); diff --git a/packages/cli/src/config/config.ts b/packages/cli/src/config/config.ts index 80c728d624b..7fed6043af1 100755 --- a/packages/cli/src/config/config.ts +++ b/packages/cli/src/config/config.ts @@ -1028,16 +1028,8 @@ export async function loadCliConfig( // to preserve the original behaviour where "ShellTool", "Shell", and // "run_shell_command" are all accepted as the same tool. const isExplicitlyAllowed = (toolName: ToolName): boolean => { - const name = toolName as string; // 1. Check permissions.allow / allowedTools rules. - if ( - mergedAllow.some((rule) => { - const openParen = rule.indexOf('('); - const ruleName = - openParen === -1 ? rule.trim() : rule.substring(0, openParen).trim(); - return ruleName === name; - }) - ) { + if (mergedAllow.some((rule) => isToolEnabled(toolName, [rule], []))) { return true; } // 2. Check coreTools whitelist (with alias matching). @@ -1071,12 +1063,14 @@ export async function loadCliConfig( case ApprovalMode.DEFAULT: // Deny all write/execute tools unless explicitly allowed. denyUnlessAllowed(ToolNames.SHELL as ToolName); + denyUnlessAllowed(ToolNames.MONITOR as ToolName); denyUnlessAllowed(ToolNames.EDIT as ToolName); denyUnlessAllowed(ToolNames.WRITE_FILE as ToolName); break; case ApprovalMode.AUTO_EDIT: - // Only shell requires a prompt in auto-edit mode. + // Shell-like execute tools still require a prompt in auto-edit mode. denyUnlessAllowed(ToolNames.SHELL as ToolName); + denyUnlessAllowed(ToolNames.MONITOR as ToolName); break; case ApprovalMode.YOLO: // No extra denials for YOLO mode. diff --git a/packages/cli/src/nonInteractive/session.test.ts b/packages/cli/src/nonInteractive/session.test.ts index e163f516a85..6461f9fbef9 100644 --- a/packages/cli/src/nonInteractive/session.test.ts +++ b/packages/cli/src/nonInteractive/session.test.ts @@ -5,7 +5,7 @@ */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; -import type { Config } from '@qwen-code/qwen-code-core'; +import { SendMessageType, type Config } from '@qwen-code/qwen-code-core'; import { runNonInteractiveStreamJson } from './session.js'; import type { CLIUserMessage, @@ -56,6 +56,18 @@ interface ConfigOverrides { [key: string]: unknown; } +let mockMonitorRegistry: { + setNotificationCallback: ReturnType; + setRegisterCallback: ReturnType; + abortAll: ReturnType; +}; +let mockBackgroundShellRegistry: { + abortAll: ReturnType; +}; +let mockBackgroundTaskRegistry: { + abortAll: ReturnType; +}; + function createConfig(overrides: ConfigOverrides = {}): Config { const base = { getSessionId: () => 'test-session', @@ -65,6 +77,9 @@ function createConfig(overrides: ConfigOverrides = {}): Config { getApprovalMode: () => 'auto', getOutputFormat: () => 'stream-json', initialize: vi.fn(), + getMonitorRegistry: () => mockMonitorRegistry, + getBackgroundShellRegistry: () => mockBackgroundShellRegistry, + getBackgroundTaskRegistry: () => mockBackgroundTaskRegistry, }; return { ...base, ...overrides } as unknown as Config; } @@ -142,6 +157,8 @@ describe('runNonInteractiveStreamJson', () => { }; let mockOutputAdapter: { emitResult: ReturnType; + emitUserMessage: ReturnType; + emitSystemMessage: ReturnType; }; let mockDispatcher: { dispatch: ReturnType; @@ -156,14 +173,29 @@ describe('runNonInteractiveStreamJson', () => { }; }; beforeEach(() => { + mockMonitorRegistry = { + setNotificationCallback: vi.fn(), + setRegisterCallback: vi.fn(), + abortAll: vi.fn(), + }; + mockBackgroundShellRegistry = { + abortAll: vi.fn(), + }; + mockBackgroundTaskRegistry = { + abortAll: vi.fn(), + }; config = createConfig(); runNonInteractiveMock.mockReset(); // Setup mocks mockOutputAdapter = { emitResult: vi.fn(), + emitUserMessage: vi.fn(), + emitSystemMessage: vi.fn(), } as { emitResult: ReturnType; + emitUserMessage: ReturnType; + emitSystemMessage: ReturnType; [key: string]: unknown; }; ( @@ -259,6 +291,240 @@ describe('runNonInteractiveStreamJson', () => { expect(runNonInteractiveMock).toHaveBeenCalledTimes(2); }); + it('routes monitor notifications through the session queue', async () => { + const initRequest = createControlRequest('initialize'); + const userMessage = createUserMessage('Start a monitor'); + let closeInput: (() => void) | undefined; + + let registerCallback: + | ((entry: { + monitorId: string; + toolUseId?: string; + description: string; + }) => void) + | undefined; + let monitorCallback: + | (( + displayText: string, + modelText: string, + meta: { + monitorId: string; + toolUseId?: string; + status: string; + }, + ) => void) + | undefined; + mockMonitorRegistry.setRegisterCallback.mockImplementation((cb) => { + registerCallback = cb; + }); + mockMonitorRegistry.setNotificationCallback.mockImplementation((cb) => { + monitorCallback = cb; + }); + + const notificationXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'running\n' + + 'Monitor emitted event #1.\n' + + 'ready\n' + + ''; + + runNonInteractiveMock + .mockImplementationOnce(async () => { + registerCallback?.({ + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + description: 'logs', + }); + monitorCallback?.('Monitor "logs" event #1: ready', notificationXml, { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'running', + }); + }) + .mockResolvedValueOnce(undefined); + + mockInputReader.read = async function* () { + yield initRequest; + yield userMessage; + await new Promise((resolve) => { + closeInput = resolve; + }); + }; + + const sessionPromise = runNonInteractiveStreamJson(config, ''); + await vi.waitFor(() => { + expect(runNonInteractiveMock).toHaveBeenCalledTimes(2); + }); + closeInput?.(); + await sessionPromise; + + expect(runNonInteractiveMock).toHaveBeenCalledTimes(2); + expect(mockOutputAdapter.emitSystemMessage).toHaveBeenCalledWith( + 'task_started', + { + task_id: 'mon_1', + tool_use_id: 'tool_mon_1', + description: 'logs', + }, + ); + expect(mockOutputAdapter.emitUserMessage).toHaveBeenCalledWith([ + { text: 'Monitor "logs" event #1: ready' }, + ]); + expect(mockOutputAdapter.emitSystemMessage).toHaveBeenCalledWith( + 'task_notification', + { + task_id: 'mon_1', + tool_use_id: 'tool_mon_1', + status: 'running', + }, + ); + expect(runNonInteractiveMock).toHaveBeenNthCalledWith( + 2, + config, + expect.objectContaining({ merged: expect.any(Object) }), + notificationXml, + expect.stringContaining('test-session'), + expect.objectContaining({ + adapter: mockOutputAdapter, + sendMessageType: SendMessageType.Notification, + notificationDisplayText: 'Monitor "logs" event #1: ready', + captureMonitorNotifications: false, + captureMonitorRegistrations: false, + }), + ); + }); + + it('stops accepting new monitor events before EOF drain', async () => { + const initRequest = createControlRequest('initialize'); + const userMessage = createUserMessage('Start a monitor'); + let closeInput: (() => void) | undefined; + + let registerCallback: + | ((entry: { + monitorId: string; + toolUseId?: string; + description: string; + }) => void) + | undefined; + let notificationCallback: + | (( + displayText: string, + modelText: string, + meta: { + monitorId: string; + toolUseId?: string; + status: string; + }, + ) => void) + | undefined; + + mockMonitorRegistry.setRegisterCallback.mockImplementation((cb) => { + registerCallback = cb; + }); + mockMonitorRegistry.setNotificationCallback.mockImplementation((cb) => { + notificationCallback = cb; + }); + + let releaseFirstTurn: (() => void) | undefined; + runNonInteractiveMock.mockImplementationOnce(async () => { + registerCallback?.({ + monitorId: 'mon_before_eof', + toolUseId: 'tool_mon_before_eof', + description: 'before eof', + }); + notificationCallback?.( + 'Monitor "before eof" event #1: ready', + 'before-eof', + { + monitorId: 'mon_before_eof', + toolUseId: 'tool_mon_before_eof', + status: 'running', + }, + ); + await new Promise((resolve) => { + releaseFirstTurn = () => { + registerCallback?.({ + monitorId: 'mon_late', + toolUseId: 'tool_mon_late', + description: 'late monitor', + }); + notificationCallback?.( + 'Monitor "late monitor" event #1: ignored', + 'late', + { + monitorId: 'mon_late', + toolUseId: 'tool_mon_late', + status: 'running', + }, + ); + resolve(); + }; + }); + }); + + mockInputReader.read = async function* () { + yield initRequest; + yield userMessage; + await new Promise((resolve) => { + closeInput = resolve; + }); + }; + + const sessionPromise = runNonInteractiveStreamJson(config, ''); + await vi.waitFor(() => { + expect(runNonInteractiveMock).toHaveBeenCalledTimes(1); + }); + + closeInput?.(); + await vi.waitFor(() => { + expect( + mockMonitorRegistry.setNotificationCallback, + ).toHaveBeenLastCalledWith(undefined); + expect(mockMonitorRegistry.setRegisterCallback).toHaveBeenLastCalledWith( + undefined, + ); + }); + + releaseFirstTurn?.(); + await sessionPromise; + + expect(mockOutputAdapter.emitSystemMessage).toHaveBeenCalledWith( + 'task_started', + { + task_id: 'mon_before_eof', + tool_use_id: 'tool_mon_before_eof', + description: 'before eof', + }, + ); + expect(mockOutputAdapter.emitSystemMessage).not.toHaveBeenCalledWith( + 'task_started', + expect.objectContaining({ task_id: 'mon_late' }), + ); + expect(mockOutputAdapter.emitSystemMessage).toHaveBeenCalledWith( + 'task_notification', + { + task_id: 'mon_before_eof', + tool_use_id: 'tool_mon_before_eof', + status: 'running', + }, + ); + expect(mockOutputAdapter.emitSystemMessage).not.toHaveBeenCalledWith( + 'task_notification', + expect.objectContaining({ task_id: 'mon_late' }), + ); + expect(runNonInteractiveMock).toHaveBeenCalledTimes(2); + + const clearCalls = mockMonitorRegistry.setNotificationCallback.mock.calls + .map(([cb]) => cb) + .filter((cb) => cb === undefined); + expect(clearCalls).toHaveLength(1); + expect(mockMonitorRegistry.setRegisterCallback).toHaveBeenLastCalledWith( + undefined, + ); + }); + it('enqueues user messages received during processing', async () => { const initRequest = createControlRequest('initialize'); const userMessage1 = createUserMessage('First message'); @@ -567,6 +833,153 @@ describe('runNonInteractiveStreamJson', () => { expect(mockDispatcher.shutdown).toHaveBeenCalledTimes(1); }); + it('aborts background registries on stream completion shutdown', async () => { + const initRequest = createControlRequest('initialize'); + + mockInputReader.read = async function* () { + yield initRequest; + }; + + await runNonInteractiveStreamJson(config, ''); + + expect(mockMonitorRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundShellRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundTaskRegistry.abortAll).toHaveBeenCalledTimes(2); + }); + + it('aborts background registries on error shutdown', async () => { + const streamError = new Error('Stream error'); + // eslint-disable-next-line require-yield + mockInputReader.read = async function* () { + throw streamError; + } as typeof mockInputReader.read; + + await expect(runNonInteractiveStreamJson(config, '')).rejects.toThrow( + 'Stream error', + ); + + expect(mockMonitorRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundShellRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundTaskRegistry.abortAll).toHaveBeenCalledTimes(2); + }); + + it('runs final background cleanup after in-flight processing drains', async () => { + const initRequest = createControlRequest('initialize'); + const userMessage = createUserMessage('Start background work'); + let releaseProcessing: (() => void) | undefined; + const callOrder: string[] = []; + + mockMonitorRegistry.abortAll.mockImplementation(() => { + callOrder.push('monitor:abortAll'); + }); + mockBackgroundShellRegistry.abortAll.mockImplementation(() => { + callOrder.push('background:abortAll'); + }); + mockBackgroundTaskRegistry.abortAll.mockImplementation(() => { + callOrder.push('agent:abortAll'); + }); + + runNonInteractiveMock.mockImplementationOnce( + () => + new Promise((resolve) => { + callOrder.push('run:start'); + releaseProcessing = () => { + callOrder.push('run:end'); + resolve(); + }; + }), + ); + + mockInputReader.read = async function* () { + yield initRequest; + yield userMessage; + }; + + const sessionPromise = runNonInteractiveStreamJson(config, ''); + await vi.waitFor(() => { + expect(releaseProcessing).toBeDefined(); + }); + + expect(mockMonitorRegistry.abortAll).toHaveBeenCalledTimes(1); + expect(mockBackgroundShellRegistry.abortAll).toHaveBeenCalledTimes(1); + expect(mockBackgroundTaskRegistry.abortAll).toHaveBeenCalledTimes(1); + expect(callOrder).toContain('run:start'); + expect(callOrder).toContain('monitor:abortAll'); + expect(callOrder).toContain('background:abortAll'); + expect(callOrder).toContain('agent:abortAll'); + + releaseProcessing?.(); + await sessionPromise; + + expect(mockMonitorRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundShellRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundTaskRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(callOrder.slice(-4)).toEqual([ + 'run:end', + 'monitor:abortAll', + 'background:abortAll', + 'agent:abortAll', + ]); + }); + + it('runs final background cleanup after in-flight processing drains on error shutdown', async () => { + const initRequest = createControlRequest('initialize'); + const userMessage = createUserMessage('Start background work'); + let releaseProcessing: (() => void) | undefined; + const callOrder: string[] = []; + const streamError = new Error('Stream error'); + + mockMonitorRegistry.abortAll.mockImplementation(() => { + callOrder.push('monitor:abortAll'); + }); + mockBackgroundShellRegistry.abortAll.mockImplementation(() => { + callOrder.push('background:abortAll'); + }); + mockBackgroundTaskRegistry.abortAll.mockImplementation(() => { + callOrder.push('agent:abortAll'); + }); + + runNonInteractiveMock.mockImplementationOnce( + () => + new Promise((resolve) => { + callOrder.push('run:start'); + releaseProcessing = () => { + callOrder.push('run:end'); + resolve(); + }; + }), + ); + + mockInputReader.read = async function* () { + yield initRequest; + yield userMessage; + throw streamError; + } as typeof mockInputReader.read; + + const sessionPromise = runNonInteractiveStreamJson(config, ''); + await vi.waitFor(() => { + expect(releaseProcessing).toBeDefined(); + }); + + expect(mockMonitorRegistry.abortAll).toHaveBeenCalledTimes(1); + expect(mockBackgroundShellRegistry.abortAll).toHaveBeenCalledTimes(1); + expect(mockBackgroundTaskRegistry.abortAll).toHaveBeenCalledTimes(1); + expect(callOrder).toContain('run:start'); + + releaseProcessing?.(); + await expect(sessionPromise).rejects.toThrow('Stream error'); + + expect(mockMonitorRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundShellRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(mockBackgroundTaskRegistry.abortAll).toHaveBeenCalledTimes(2); + expect(callOrder.slice(-4)).toEqual([ + 'run:end', + 'monitor:abortAll', + 'background:abortAll', + 'agent:abortAll', + ]); + }); + it('handles empty stream gracefully', async () => { mockInputReader.read = async function* () { // Empty stream diff --git a/packages/cli/src/nonInteractive/session.ts b/packages/cli/src/nonInteractive/session.ts index 6b8c9b88038..3159040276c 100644 --- a/packages/cli/src/nonInteractive/session.ts +++ b/packages/cli/src/nonInteractive/session.ts @@ -8,7 +8,7 @@ import type { Config, ConfigInitializeOptions, } from '@qwen-code/qwen-code-core'; -import { createDebugLogger } from '@qwen-code/qwen-code-core'; +import { createDebugLogger, SendMessageType } from '@qwen-code/qwen-code-core'; import { StreamJsonInputReader } from './io/StreamJsonInputReader.js'; import { StreamJsonOutputAdapter } from './io/StreamJsonOutputAdapter.js'; import { ControlContext } from './control/ControlContext.js'; @@ -36,8 +36,26 @@ import { runNonInteractive } from '../nonInteractiveCli.js'; const debugLogger = createDebugLogger('NON_INTERACTIVE_SESSION'); +interface MonitorStartedQueueItem { + task_id: string; + tool_use_id?: string; + description: string; +} + +interface MonitorQueueItem { + displayText: string; + modelText: string; + sdkNotification: { + task_id: string; + tool_use_id?: string; + status: string; + }; +} + class Session { private userMessageQueue: CLIUserMessage[] = []; + private monitorStartedQueue: MonitorStartedQueueItem[] = []; + private monitorQueue: MonitorQueueItem[] = []; private abortController: AbortController; private config: Config; private sessionId: string; @@ -53,6 +71,8 @@ class Session { private processingPromise: Promise | null = null; private isShuttingDown: boolean = false; private configInitialized: boolean = false; + private monitorNotificationsRegistered: boolean = false; + private monitorRegistrationsRegistered: boolean = false; // Single initialization promise that resolves when session is ready for user messages. // Created lazily once initialization actually starts. @@ -110,12 +130,56 @@ class Session { try { await this.config.initialize(options); this.configInitialized = true; + this.registerMonitorRegistrations(); + this.registerMonitorNotifications(); } catch (error) { debugLogger.error('[Session] Failed to initialize config:', error); throw error; } } + private registerMonitorNotifications(): void { + if (this.monitorNotificationsRegistered) { + return; + } + + const registry = this.config.getMonitorRegistry(); + registry.setNotificationCallback((displayText, modelText, meta) => { + if (this.isShuttingDown || this.abortController.signal.aborted) { + return; + } + this.enqueueMonitorNotification({ + displayText, + modelText, + sdkNotification: { + task_id: meta.monitorId, + tool_use_id: meta.toolUseId, + status: meta.status, + }, + }); + }); + this.monitorNotificationsRegistered = true; + } + + private registerMonitorRegistrations(): void { + if (this.monitorRegistrationsRegistered) { + return; + } + + const registry = this.config.getMonitorRegistry(); + registry.setRegisterCallback((entry) => { + if (this.isShuttingDown || this.abortController.signal.aborted) { + return; + } + this.enqueueMonitorStarted({ + task_id: entry.monitorId, + tool_use_id: entry.toolUseId, + description: entry.description, + }); + }); + this.monitorRegistrationsRegistered = true; + } + /** * Mark initialization as complete */ @@ -339,6 +403,8 @@ class Session { abortController: this.abortController, adapter: this.outputAdapter, controlService: this.controlService ?? undefined, + captureMonitorNotifications: false, + captureMonitorRegistrations: false, }, ); } catch (error) { @@ -346,21 +412,75 @@ class Session { } } - private async processUserMessageQueue(): Promise { + private async processMonitorNotification( + notification: MonitorQueueItem, + ): Promise { + await this.waitForInitialization(); + + this.outputAdapter.emitUserMessage([{ text: notification.displayText }]); + this.outputAdapter.emitSystemMessage( + 'task_notification', + notification.sdkNotification, + ); + + const promptId = this.getNextPromptId(); + await runNonInteractive( + this.config, + createMinimalSettings(), + notification.modelText, + promptId, + { + abortController: this.abortController, + adapter: this.outputAdapter, + controlService: this.controlService ?? undefined, + sendMessageType: SendMessageType.Notification, + notificationDisplayText: notification.displayText, + captureMonitorNotifications: false, + captureMonitorRegistrations: false, + }, + ); + } + + private async processPendingWork(): Promise { if (this.isShuttingDown || this.abortController.signal.aborted) { return; } while ( - this.userMessageQueue.length > 0 && + (this.userMessageQueue.length > 0 || + this.monitorStartedQueue.length > 0 || + this.monitorQueue.length > 0) && !this.isShuttingDown && !this.abortController.signal.aborted ) { - const userMessage = this.userMessageQueue.shift()!; + if (this.userMessageQueue.length > 0) { + const userMessage = this.userMessageQueue.shift()!; + try { + await this.processUserMessage(userMessage); + } catch (error) { + debugLogger.error('[Session] Error processing user message:', error); + this.emitErrorResult(error); + } + continue; + } + + const started = this.monitorStartedQueue.shift(); + if (started) { + this.outputAdapter.emitSystemMessage('task_started', started); + continue; + } + + const notification = this.monitorQueue.shift(); + if (!notification) { + continue; + } try { - await this.processUserMessage(userMessage); + await this.processMonitorNotification(notification); } catch (error) { - debugLogger.error('[Session] Error processing user message:', error); + debugLogger.error( + '[Session] Error processing monitor notification:', + error, + ); this.emitErrorResult(error); } } @@ -371,15 +491,27 @@ class Session { this.ensureProcessingStarted(); } + private enqueueMonitorStarted(started: MonitorStartedQueueItem): void { + this.monitorStartedQueue.push(started); + this.ensureProcessingStarted(); + } + + private enqueueMonitorNotification(notification: MonitorQueueItem): void { + this.monitorQueue.push(notification); + this.ensureProcessingStarted(); + } + private ensureProcessingStarted(): void { if (this.processingPromise) { return; } - this.processingPromise = this.processUserMessageQueue().finally(() => { + this.processingPromise = this.processPendingWork().finally(() => { this.processingPromise = null; if ( - this.userMessageQueue.length > 0 && + (this.userMessageQueue.length > 0 || + this.monitorStartedQueue.length > 0 || + this.monitorQueue.length > 0) && !this.isShuttingDown && !this.abortController.signal.aborted ) { @@ -463,14 +595,59 @@ class Session { debugLogger.debug('[Session] Shutting down'); this.isShuttingDown = true; + this.abortTaskRegistries(); + this.stopMonitorCallbacks(); // Wait for all pending work await this.waitForAllPendingWork(); + this.abortTaskRegistries(); + + this.finishShutdown(); + } + + private async drainAndShutdown(): Promise { + debugLogger.debug('[Session] Draining pending work before shutdown'); + // Abort monitors and stop callbacks first, then drain anything already + // queued so EOF does not remain coupled to monitor process lifetime. + this.abortTaskRegistries(); + this.stopMonitorCallbacks(); + await this.waitForAllPendingWork(); + this.abortTaskRegistries(); + + this.finishShutdown(); + } + + private abortTaskRegistries(): void { + this.config.getMonitorRegistry().abortAll({ notify: false }); + this.config.getBackgroundShellRegistry().abortAll(); + this.config.getBackgroundTaskRegistry().abortAll(); + } + + private finishShutdown(): void { this.dispatcher?.shutdown(); this.cleanupSignalHandlers(); } + private stopMonitorCallbacks(): void { + if ( + !this.monitorNotificationsRegistered && + !this.monitorRegistrationsRegistered + ) { + return; + } + + const registry = this.config.getMonitorRegistry(); + if (this.monitorNotificationsRegistered) { + registry.setNotificationCallback(undefined); + this.monitorNotificationsRegistered = false; + } + if (this.monitorRegistrationsRegistered) { + registry.setRegisterCallback(undefined); + this.monitorRegistrationsRegistered = false; + } + } + private cleanupSignalHandlers(): void { if (this.shutdownHandler) { process.removeListener('SIGINT', this.shutdownHandler); @@ -565,9 +742,7 @@ class Session { this.dispatcher.markInputClosed(); } - // Wait for all pending work before shutdown - await this.waitForAllPendingWork(); - await this.shutdown(); + await this.drainAndShutdown(); } catch (error) { debugLogger.error('[Session] Error:', error); await this.shutdown(); diff --git a/packages/cli/src/nonInteractiveCli.test.ts b/packages/cli/src/nonInteractiveCli.test.ts index 9b550823986..3ec1cb9e94b 100644 --- a/packages/cli/src/nonInteractiveCli.test.ts +++ b/packages/cli/src/nonInteractiveCli.test.ts @@ -69,6 +69,19 @@ describe('runNonInteractive', () => { let mockConfig: Config; let mockSettings: LoadedSettings; let mockToolRegistry: ToolRegistry; + let mockBackgroundTaskRegistry: { + setNotificationCallback: ReturnType; + setRegisterCallback: ReturnType; + getAll: ReturnType; + hasUnfinalizedTasks: ReturnType; + abortAll: ReturnType; + }; + let mockMonitorRegistry: { + setNotificationCallback: ReturnType; + setRegisterCallback: ReturnType; + getRunning: ReturnType; + abortAll: ReturnType; + }; let mockCoreExecuteToolCall: Mock; let mockShutdownTelemetry: Mock; let processStdoutSpy: MockInstance; @@ -113,6 +126,21 @@ describe('runNonInteractive', () => { getAllToolNames: vi.fn().mockReturnValue([]), } as unknown as ToolRegistry; + mockBackgroundTaskRegistry = { + setNotificationCallback: vi.fn(), + setRegisterCallback: vi.fn(), + getAll: vi.fn().mockReturnValue([]), + hasUnfinalizedTasks: vi.fn().mockReturnValue(false), + abortAll: vi.fn(), + }; + + mockMonitorRegistry = { + setNotificationCallback: vi.fn(), + setRegisterCallback: vi.fn(), + getRunning: vi.fn().mockReturnValue([]), + abortAll: vi.fn(), + }; + mockGetDebugResponses = vi.fn(() => []); mockGeminiClient = { @@ -163,13 +191,10 @@ describe('runNonInteractive', () => { setModelInvocableCommandsProvider: vi.fn(), setModelInvocableCommandsExecutor: vi.fn(), getDisabledSlashCommands: vi.fn().mockReturnValue([]), - getBackgroundTaskRegistry: vi.fn().mockReturnValue({ - setNotificationCallback: vi.fn(), - setRegisterCallback: vi.fn(), - getAll: vi.fn().mockReturnValue([]), - hasUnfinalizedTasks: vi.fn().mockReturnValue(false), - abortAll: vi.fn(), - }), + getBackgroundTaskRegistry: vi + .fn() + .mockReturnValue(mockBackgroundTaskRegistry), + getMonitorRegistry: vi.fn().mockReturnValue(mockMonitorRegistry), } as unknown as Config; mockSettings = { @@ -1151,6 +1176,152 @@ describe('runNonInteractive', () => { }); }); + it('flushes terminal monitor notifications before the final headless result', async () => { + (mockConfig.getOutputFormat as Mock).mockReturnValue( + OutputFormat.STREAM_JSON, + ); + (mockConfig.getIncludePartialMessages as Mock).mockReturnValue(false); + setupMetricsMock(); + + const writes: string[] = []; + processStdoutSpy.mockImplementation((chunk: string | Uint8Array) => { + if (typeof chunk === 'string') { + writes.push(chunk); + } else { + writes.push(Buffer.from(chunk).toString('utf8')); + } + return true; + }); + + const notificationXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'running\n' + + 'Monitor emitted event #1.\n' + + 'ready\n' + + ''; + const cancelledXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'cancelled\n' + + 'Monitor was cancelled.\n' + + ''; + let monitorNotificationCallback: + | (( + displayText: string, + modelText: string, + meta: { + monitorId: string; + toolUseId?: string; + status: 'running' | 'completed' | 'failed' | 'cancelled'; + eventCount: number; + }, + ) => void) + | undefined; + + mockMonitorRegistry.setNotificationCallback.mockImplementation((cb) => { + monitorNotificationCallback = cb ?? undefined; + if (!cb) { + return; + } + cb('Monitor "logs" event #1: ready', notificationXml, { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'running', + eventCount: 1, + }); + }); + mockMonitorRegistry.abortAll.mockImplementation(() => { + monitorNotificationCallback?.( + 'Monitor "logs" was cancelled.', + cancelledXml, + { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'cancelled', + eventCount: 1, + }, + ); + }); + + mockGeminiClient.sendMessageStream + .mockReturnValueOnce( + createStreamFromEvents([ + { type: GeminiEventType.Content, value: 'Monitor launched.' }, + { + type: GeminiEventType.Finished, + value: { + reason: undefined, + usageMetadata: { totalTokenCount: 2 }, + }, + }, + ]), + ) + .mockReturnValueOnce( + createStreamFromEvents([ + { type: GeminiEventType.Content, value: 'Observed.' }, + { + type: GeminiEventType.Finished, + value: { + reason: undefined, + usageMetadata: { totalTokenCount: 1 }, + }, + }, + ]), + ); + + await runNonInteractive( + mockConfig, + mockSettings, + 'Watch the logs', + 'prompt-monitor', + ); + + expect(mockGeminiClient.sendMessageStream).toHaveBeenCalledTimes(2); + expect(mockGeminiClient.sendMessageStream).toHaveBeenNthCalledWith( + 2, + [{ text: notificationXml }], + expect.any(AbortSignal), + 'prompt-monitor', + { + type: SendMessageType.Notification, + modelOverride: undefined, + notificationDisplayText: 'Monitor "logs" event #1: ready', + }, + ); + + const envelopes = writes + .join('') + .split('\n') + .filter((line) => line.trim().length > 0) + .map((line) => JSON.parse(line)); + expect( + envelopes.some( + (env) => + env.type === 'system' && + env.subtype === 'task_notification' && + env.data?.task_id === 'mon_1', + ), + ).toBe(true); + const cancelledNotificationIndex = envelopes.findIndex( + (env) => + env.type === 'system' && + env.subtype === 'task_notification' && + env.data?.task_id === 'mon_1' && + env.data?.status === 'cancelled', + ); + const resultIndex = envelopes.findIndex((env) => env.type === 'result'); + expect(cancelledNotificationIndex).toBeGreaterThanOrEqual(0); + expect(resultIndex).toBeGreaterThan(cancelledNotificationIndex); + expect(mockMonitorRegistry.abortAll).toHaveBeenCalledTimes(1); + expect(envelopes.at(-1)).toMatchObject({ + type: 'result', + is_error: false, + }); + }); + it.skip('should emit a single user envelope when userEnvelope is provided', async () => { (mockConfig.getOutputFormat as Mock).mockReturnValue('stream-json'); (mockConfig.getIncludePartialMessages as Mock).mockReturnValue(false); @@ -1208,6 +1379,317 @@ describe('runNonInteractive', () => { expect(userEnvelopes).toHaveLength(0); }); + it('does not let late monitor output keep one-shot runs alive', async () => { + (mockConfig.getOutputFormat as Mock).mockReturnValue( + OutputFormat.STREAM_JSON, + ); + (mockConfig.getIncludePartialMessages as Mock).mockReturnValue(false); + setupMetricsMock(); + + const writes: string[] = []; + processStdoutSpy.mockImplementation((chunk: string | Uint8Array) => { + if (typeof chunk === 'string') { + writes.push(chunk); + } else { + writes.push(Buffer.from(chunk).toString('utf8')); + } + return true; + }); + + const firstNotificationXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'running\n' + + 'Monitor emitted event #1.\n' + + 'ready\n' + + ''; + const secondNotificationXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'running\n' + + 'Monitor emitted event #2.\n' + + 'still running\n' + + ''; + const cancelledXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'cancelled\n' + + 'Monitor was cancelled.\n' + + ''; + + let monitorNotificationCallback: + | (( + displayText: string, + modelText: string, + meta: { + monitorId: string; + toolUseId?: string; + status: 'running' | 'completed' | 'failed' | 'cancelled'; + eventCount: number; + }, + ) => void) + | undefined; + + mockMonitorRegistry.setNotificationCallback.mockImplementation((cb) => { + monitorNotificationCallback = cb ?? undefined; + if (!cb) { + return; + } + cb('Monitor "logs" event #1: ready', firstNotificationXml, { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'running', + eventCount: 1, + }); + }); + mockMonitorRegistry.abortAll.mockImplementation(() => { + monitorNotificationCallback?.( + 'Monitor "logs" was cancelled.', + cancelledXml, + { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'cancelled', + eventCount: 2, + }, + ); + }); + + async function* secondTurnStream(): AsyncGenerator { + yield { type: GeminiEventType.Content, value: 'Observed.' }; + monitorNotificationCallback?.( + 'Monitor "logs" event #2: still running', + secondNotificationXml, + { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'running', + eventCount: 2, + }, + ); + yield { + type: GeminiEventType.Finished, + value: { + reason: undefined, + usageMetadata: { totalTokenCount: 1 }, + }, + }; + } + + mockGeminiClient.sendMessageStream + .mockReturnValueOnce( + createStreamFromEvents([ + { type: GeminiEventType.Content, value: 'Monitor launched.' }, + { + type: GeminiEventType.Finished, + value: { + reason: undefined, + usageMetadata: { totalTokenCount: 2 }, + }, + }, + ]), + ) + .mockReturnValueOnce(secondTurnStream()); + + await runNonInteractive( + mockConfig, + mockSettings, + 'Watch the logs', + 'prompt-monitor-cutover', + ); + + expect(mockGeminiClient.sendMessageStream).toHaveBeenCalledTimes(2); + + const envelopes = writes + .join('') + .split('\n') + .filter((line) => line.trim().length > 0) + .map((line) => JSON.parse(line)); + const monitorNotifications = envelopes.filter( + (env) => + env.type === 'system' && + env.subtype === 'task_notification' && + env.data?.task_id === 'mon_1', + ); + expect( + monitorNotifications.filter((env) => env.data?.status === 'running'), + ).toHaveLength(2); + expect( + monitorNotifications.some((env) => env.data?.status === 'cancelled'), + ).toBe(true); + expect(envelopes.at(-1)).toMatchObject({ + type: 'result', + is_error: false, + }); + }); + + it('streams late monitor output to the SDK before one-shot completion', async () => { + (mockConfig.getOutputFormat as Mock).mockReturnValue( + OutputFormat.STREAM_JSON, + ); + (mockConfig.getIncludePartialMessages as Mock).mockReturnValue(false); + setupMetricsMock(); + + const writes: string[] = []; + processStdoutSpy.mockImplementation((chunk: string | Uint8Array) => { + if (typeof chunk === 'string') { + writes.push(chunk); + } else { + writes.push(Buffer.from(chunk).toString('utf8')); + } + return true; + }); + + let keepBackgroundTaskOpen = true; + let lateMonitorEventEmitted = false; + mockBackgroundTaskRegistry.hasUnfinalizedTasks.mockImplementation(() => { + if (keepBackgroundTaskOpen && !lateMonitorEventEmitted) { + lateMonitorEventEmitted = true; + monitorNotificationCallback?.( + 'Monitor "logs" event #2: still running', + secondNotificationXml, + { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'running', + eventCount: 2, + }, + ); + } + return keepBackgroundTaskOpen; + }); + + const firstNotificationXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'running\n' + + 'Monitor emitted event #1.\n' + + 'ready\n' + + ''; + const secondNotificationXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'running\n' + + 'Monitor emitted event #2.\n' + + 'still running\n' + + ''; + const cancelledXml = + '\n' + + 'mon_1\n' + + 'monitor\n' + + 'cancelled\n' + + 'Monitor was cancelled.\n' + + ''; + + let monitorNotificationCallback: + | (( + displayText: string, + modelText: string, + meta: { + monitorId: string; + toolUseId?: string; + status: 'running' | 'completed' | 'failed' | 'cancelled'; + eventCount: number; + }, + ) => void) + | undefined; + + mockMonitorRegistry.setNotificationCallback.mockImplementation((cb) => { + monitorNotificationCallback = cb ?? undefined; + if (!cb) { + return; + } + cb('Monitor "logs" event #1: ready', firstNotificationXml, { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'running', + eventCount: 1, + }); + }); + mockMonitorRegistry.abortAll.mockImplementation(() => { + monitorNotificationCallback?.( + 'Monitor "logs" was cancelled.', + cancelledXml, + { + monitorId: 'mon_1', + toolUseId: 'tool_mon_1', + status: 'cancelled', + eventCount: 2, + }, + ); + }); + + mockGeminiClient.sendMessageStream + .mockReturnValueOnce( + createStreamFromEvents([ + { type: GeminiEventType.Content, value: 'Monitor launched.' }, + { + type: GeminiEventType.Finished, + value: { + reason: undefined, + usageMetadata: { totalTokenCount: 2 }, + }, + }, + ]), + ) + .mockReturnValueOnce( + createStreamFromEvents([ + { type: GeminiEventType.Content, value: 'Observed.' }, + { + type: GeminiEventType.Finished, + value: { + reason: undefined, + usageMetadata: { totalTokenCount: 1 }, + }, + }, + ]), + ); + + const runPromise = runNonInteractive( + mockConfig, + mockSettings, + 'Watch the logs', + 'prompt-monitor-late-sdk', + ); + + await vi.waitFor(() => { + const envelopes = writes + .join('') + .split('\n') + .filter((line) => line.trim().length > 0) + .map((line) => JSON.parse(line)); + const monitorNotifications = envelopes.filter( + (env) => + env.type === 'system' && + env.subtype === 'task_notification' && + env.data?.task_id === 'mon_1', + ); + + expect( + monitorNotifications.filter((env) => env.data?.status === 'running'), + ).toHaveLength(2); + expect(envelopes.some((env) => env.type === 'result')).toBe(false); + }); + + keepBackgroundTaskOpen = false; + await runPromise; + + const envelopes = writes + .join('') + .split('\n') + .filter((line) => line.trim().length > 0) + .map((line) => JSON.parse(line)); + expect(envelopes.at(-1)).toMatchObject({ + type: 'result', + is_error: false, + }); + }); + it('should include usage metadata and API duration in stream-json result', async () => { (mockConfig.getOutputFormat as Mock).mockReturnValue('stream-json'); (mockConfig.getIncludePartialMessages as Mock).mockReturnValue(false); @@ -1226,6 +1708,7 @@ describe('runNonInteractive', () => { cached: 3, thoughts: 0, }, + bySource: {}, }, }, }); diff --git a/packages/cli/src/nonInteractiveCli.ts b/packages/cli/src/nonInteractiveCli.ts index 5232f60969d..339b1efd1c0 100644 --- a/packages/cli/src/nonInteractiveCli.ts +++ b/packages/cli/src/nonInteractiveCli.ts @@ -139,6 +139,10 @@ export interface RunNonInteractiveOptions { adapter?: JsonOutputAdapterInterface; userMessage?: CLIUserMessage; controlService?: ControlService; + sendMessageType?: SendMessageType; + notificationDisplayText?: string; + captureMonitorNotifications?: boolean; + captureMonitorRegistrations?: boolean; } /** @@ -178,6 +182,49 @@ export async function runNonInteractive( const geminiClient = config.getGeminiClient(); const abortController = options.abortController ?? new AbortController(); + interface LocalQueueItem { + displayText: string; + modelText: string; + sendMessageType: SendMessageType; + sdkNotification?: { + task_id: string; + tool_use_id?: string; + status: BackgroundTaskStatus; + usage?: { + total_tokens: number; + tool_uses: number; + duration_ms: number; + }; + }; + } + const localQueue: LocalQueueItem[] = []; + const sdkOnlyMonitorQueue: LocalQueueItem[] = []; + const emitNotificationToSdk = (item: LocalQueueItem) => { + if (item.sendMessageType !== SendMessageType.Notification) return; + adapter.emitUserMessage([{ text: item.displayText }]); + if (item.sdkNotification) { + adapter.emitSystemMessage('task_notification', item.sdkNotification); + } + }; + const flushQueuedNotificationsToSdk = (queue: LocalQueueItem[]) => { + while (queue.length > 0) { + emitNotificationToSdk(queue.shift()!); + } + }; + let captureMonitorTurnsInLocalQueue = true; + let oneShotMonitorsFinalized = false; + const finalizeOneShotMonitors = () => { + if ( + options.captureMonitorNotifications === false || + oneShotMonitorsFinalized + ) + return; + oneShotMonitorsFinalized = true; + captureMonitorTurnsInLocalQueue = false; + config.getMonitorRegistry().abortAll(); + flushQueuedNotificationsToSdk(sdkOnlyMonitorQueue); + }; + // EPIPE: don't process.exit here — that bypasses the caller's // runExitCleanup → flush() and drops queued JSONL writes. Destroy // stdout instead and let the natural return drive cleanup. (Aborting @@ -295,22 +342,6 @@ export async function runNonInteractive( // Register the callback early so background agents launched during the main // tool-call chain can push completions onto the queue. - interface LocalQueueItem { - displayText: string; - modelText: string; - sendMessageType: SendMessageType; - sdkNotification?: { - task_id: string; - tool_use_id?: string; - status: BackgroundTaskStatus; - usage?: { - total_tokens: number; - tool_uses: number; - duration_ms: number; - }; - }; - } - const localQueue: LocalQueueItem[] = []; const registry = config.getBackgroundTaskRegistry(); registry.setNotificationCallback((displayText, modelText, meta) => { localQueue.push({ @@ -341,6 +372,46 @@ export async function runNonInteractive( }); }); + const monitorRegistry = config.getMonitorRegistry(); + if (options.captureMonitorNotifications !== false) { + // One-shot headless runs capture monitor notifications locally so any + // events already emitted before exit can be surfaced to the SDK/model. + // Persistent stream-json sessions own this callback at the Session + // layer instead, so future monitor events can continue after the + // originating turn has already completed. + monitorRegistry.setNotificationCallback( + (displayText, modelText, meta) => { + const queueItem = { + displayText, + modelText, + sendMessageType: SendMessageType.Notification, + sdkNotification: { + task_id: meta.monitorId, + tool_use_id: meta.toolUseId, + status: meta.status, + }, + }; + + if (captureMonitorTurnsInLocalQueue) { + localQueue.push(queueItem); + } else { + sdkOnlyMonitorQueue.push(queueItem); + flushQueuedNotificationsToSdk(sdkOnlyMonitorQueue); + } + }, + ); + } + + if (options.captureMonitorRegistrations !== false) { + 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) { @@ -360,9 +431,13 @@ export async function runNonInteractive( prompt_id, { type: isFirstTurn - ? SendMessageType.UserQuery + ? (options.sendMessageType ?? SendMessageType.UserQuery) : SendMessageType.ToolResult, modelOverride, + ...(isFirstTurn && + options.notificationDisplayText && { + notificationDisplayText: options.notificationDisplayText, + }), }, ); isFirstTurn = false; @@ -471,19 +546,6 @@ export async function runNonInteractive( } currentMessages = [{ role: 'user', parts: toolResponseParts }]; } else { - // Shared between the normal drain and the cancellation flush so stream-json - // consumers always see a terminal task_notification paired with task_started. - const emitNotificationToSdk = (item: LocalQueueItem) => { - if (item.sendMessageType !== SendMessageType.Notification) return; - adapter.emitUserMessage([{ text: item.displayText }]); - if (item.sdkNotification) { - adapter.emitSystemMessage( - 'task_notification', - item.sdkNotification, - ); - } - }; - // Drain-turns count toward getMaxSessionTurns() for symmetry with the main // loop — otherwise a looping cron or a model that keeps replying to // notifications could exceed the cap silently in headless runs. @@ -708,21 +770,29 @@ export async function runNonInteractive( // Flush queued terminal notifications before handleCancellationError // exits so stream-json consumers always see a task_notification paired // with every task_started. - while (localQueue.length > 0) { - emitNotificationToSdk(localQueue.shift()!); - } + flushQueuedNotificationsToSdk(localQueue); + finalizeOneShotMonitors(); await handleCancellationError(config); } + // Once we enter the final holdback loop, monitor events should no + // longer extend one-shot runtime. Already-queued events still drain + // through the model, but later monitor output is SDK-only. + captureMonitorTurnsInLocalQueue = false; await drainLocalQueue(); - // Wait for every task's terminal notification, not just the - // 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. + // Wait for every background task's terminal notification, not + // just the 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. Monitors are different: they intentionally + // continue in the background, so final result emission is not + // gated on monitor lifetime. if (!registry.hasUnfinalizedTasks() && localQueue.length === 0) break; await new Promise((r) => setTimeout(r, 100)); } + finalizeOneShotMonitors(); + const metrics = uiTelemetryService.getMetrics(); const usage = computeUsageFromMetrics(metrics); // Get stats for JSON format output @@ -752,6 +822,9 @@ export async function runNonInteractive( // Expected when no message was started or already finalized } + flushQueuedNotificationsToSdk(localQueue); + finalizeOneShotMonitors(); + // For JSON and STREAM_JSON modes, compute usage from metrics const message = error instanceof Error ? error.message : String(error); const metrics = uiTelemetryService.getMetrics(); @@ -775,6 +848,19 @@ export async function runNonInteractive( const reg = config.getBackgroundTaskRegistry(); reg.setNotificationCallback(undefined); reg.setRegisterCallback(undefined); + const monReg = config.getMonitorRegistry(); + // In one-shot (non-Session) runs, abort all running monitors so their + // piped stdio refs don't keep the Node event loop alive after the result + // is emitted. Session runs manage monitor lifecycle independently. + if (options.captureMonitorNotifications !== false) { + if (!oneShotMonitorsFinalized) { + monReg.abortAll({ notify: false }); + } + monReg.setNotificationCallback(undefined); + } + if (options.captureMonitorRegistrations !== false) { + monReg.setRegisterCallback(undefined); + } process.stdout.removeListener('error', stdoutErrorHandler); // Cleanup signal handlers diff --git a/packages/cli/src/ui/commands/clearCommand.test.ts b/packages/cli/src/ui/commands/clearCommand.test.ts index 1898368d2f7..1145e08c110 100644 --- a/packages/cli/src/ui/commands/clearCommand.test.ts +++ b/packages/cli/src/ui/commands/clearCommand.test.ts @@ -33,6 +33,12 @@ describe('clearCommand', () => { let mockFireSessionEndEvent: ReturnType; let mockFireSessionStartEvent: ReturnType; let mockGetHookSystem: ReturnType; + let mockAbortBackgroundTasks: ReturnType; + let mockAbortMonitors: ReturnType; + let mockAbortBackgroundShells: ReturnType; + let mockResetBackgroundTasks: ReturnType; + let mockResetMonitors: ReturnType; + let mockResetBackgroundShells: ReturnType; beforeEach(() => { mockResetChat = vi.fn().mockResolvedValue(undefined); @@ -43,6 +49,12 @@ describe('clearCommand', () => { fireSessionEndEvent: mockFireSessionEndEvent, fireSessionStartEvent: mockFireSessionStartEvent, }); + mockAbortBackgroundTasks = vi.fn(); + mockAbortMonitors = vi.fn(); + mockAbortBackgroundShells = vi.fn(); + mockResetBackgroundTasks = vi.fn(); + mockResetMonitors = vi.fn(); + mockResetBackgroundShells = vi.fn(); vi.clearAllMocks(); mockContext = createMockCommandContext({ @@ -54,11 +66,13 @@ describe('clearCommand', () => { }) as unknown as GeminiClient, getBackgroundTaskRegistry: vi.fn().mockReturnValue({ hasUnfinalizedTasks: vi.fn().mockReturnValue(false), - reset: vi.fn(), + reset: mockResetBackgroundTasks, + abortAll: mockAbortBackgroundTasks, }), getBackgroundShellRegistry: vi.fn().mockReturnValue({ getAll: vi.fn().mockReturnValue([]), - reset: vi.fn(), + reset: mockResetBackgroundShells, + abortAll: mockAbortBackgroundShells, }), startNewSession: mockStartNewSession, getHookSystem: mockGetHookSystem, @@ -68,6 +82,11 @@ describe('clearCommand', () => { getModel: () => 'test-model', getToolRegistry: () => undefined, getApprovalMode: () => 'default', + getMonitorRegistry: () => ({ + getRunning: vi.fn().mockReturnValue([]), + abortAll: mockAbortMonitors, + reset: mockResetMonitors, + }), }, }, session: { @@ -128,6 +147,42 @@ describe('clearCommand', () => { expect(sessionEndCallOrder).toBeLessThan(sessionStartCallOrder); }); + it('aborts old background work before starting a new session', async () => { + if (!clearCommand.action) { + throw new Error('clearCommand must have an action.'); + } + + await clearCommand.action(mockContext, ''); + + expect(mockAbortBackgroundTasks).toHaveBeenCalledWith({ notify: false }); + expect(mockAbortMonitors).toHaveBeenCalledWith({ notify: false }); + expect(mockAbortBackgroundShells).toHaveBeenCalledTimes(1); + expect(mockAbortBackgroundTasks.mock.invocationCallOrder[0]).toBeLessThan( + mockStartNewSession.mock.invocationCallOrder[0], + ); + expect(mockAbortMonitors.mock.invocationCallOrder[0]).toBeLessThan( + mockStartNewSession.mock.invocationCallOrder[0], + ); + expect(mockAbortBackgroundShells.mock.invocationCallOrder[0]).toBeLessThan( + mockResetBackgroundShells.mock.invocationCallOrder[0], + ); + expect(mockAbortBackgroundTasks.mock.invocationCallOrder[0]).toBeLessThan( + mockResetBackgroundTasks.mock.invocationCallOrder[0], + ); + expect(mockAbortMonitors.mock.invocationCallOrder[0]).toBeLessThan( + mockResetMonitors.mock.invocationCallOrder[0], + ); + expect(mockResetBackgroundShells.mock.invocationCallOrder[0]).toBeLessThan( + mockStartNewSession.mock.invocationCallOrder[0], + ); + expect(mockResetBackgroundTasks.mock.invocationCallOrder[0]).toBeLessThan( + mockStartNewSession.mock.invocationCallOrder[0], + ); + expect(mockResetMonitors.mock.invocationCallOrder[0]).toBeLessThan( + mockStartNewSession.mock.invocationCallOrder[0], + ); + }); + it('should handle hook errors gracefully and continue execution', async () => { if (!clearCommand.action) { throw new Error('clearCommand must have an action.'); @@ -247,11 +302,13 @@ describe('clearCommand', () => { getHookSystem: mockGetHookSystem, getBackgroundTaskRegistry: vi.fn().mockReturnValue({ hasUnfinalizedTasks: vi.fn().mockReturnValue(false), - reset: vi.fn(), + reset: mockResetBackgroundTasks, + abortAll: mockAbortBackgroundTasks, }), getBackgroundShellRegistry: vi.fn().mockReturnValue({ getAll: vi.fn().mockReturnValue([]), - reset: vi.fn(), + reset: mockResetBackgroundShells, + abortAll: mockAbortBackgroundShells, }), startNewSession: mockStartNewSession, getGeminiClient: vi.fn().mockReturnValue({ @@ -262,6 +319,11 @@ describe('clearCommand', () => { getToolRegistry: vi.fn().mockReturnValue({ getAllTools: vi.fn().mockReturnValue([]), }), + getMonitorRegistry: () => ({ + getRunning: vi.fn().mockReturnValue([]), + abortAll: mockAbortMonitors, + reset: mockResetMonitors, + }), }, }, session: { @@ -320,6 +382,65 @@ describe('clearCommand', () => { getAll: vi.fn().mockReturnValue([]), reset: vi.fn(), }), + getMonitorRegistry: vi.fn().mockReturnValue({ + getRunning: vi.fn().mockReturnValue([]), + reset: vi.fn(), + }), + getHookSystem: mockGetHookSystem, + startNewSession: mockStartNewSession, + getGeminiClient: vi.fn().mockReturnValue({ + resetChat: mockResetChat, + } as unknown as GeminiClient), + getModel: vi.fn().mockReturnValue('test-model'), + getApprovalMode: vi.fn().mockReturnValue('default'), + getToolRegistry: vi.fn().mockReturnValue({ + getAllTools: vi.fn().mockReturnValue([]), + }), + getDebugLogger: vi.fn().mockReturnValue({ warn: vi.fn() }), + }, + }, + session: { + startNewSession: vi.fn(), + }, + }); + + const result = await clearCommand.action(blockedContext, ''); + + expect(result).toEqual({ + type: 'message', + messageType: 'error', + content: + "Stop the current session's running background tasks before starting a new session.", + }); + expect(mockStartNewSession).not.toHaveBeenCalled(); + expect(mockResetChat).not.toHaveBeenCalled(); + }); + + it('blocks session clearing while a monitor is still running', async () => { + if (!clearCommand.action) + throw new Error('clearCommand must have an action.'); + + const blockedContext = createMockCommandContext({ + executionMode: 'non_interactive', + services: { + config: { + getBackgroundTaskRegistry: vi.fn().mockReturnValue({ + hasUnfinalizedTasks: vi.fn().mockReturnValue(false), + reset: vi.fn(), + }), + getBackgroundShellRegistry: vi.fn().mockReturnValue({ + getAll: vi.fn().mockReturnValue([]), + reset: vi.fn(), + }), + getMonitorRegistry: vi.fn().mockReturnValue({ + getRunning: vi.fn().mockReturnValue([ + { + monitorId: 'mon_123', + status: 'running', + }, + ]), + reset: vi.fn(), + }), getHookSystem: mockGetHookSystem, startNewSession: mockStartNewSession, getGeminiClient: vi.fn().mockReturnValue({ diff --git a/packages/cli/src/ui/commands/clearCommand.ts b/packages/cli/src/ui/commands/clearCommand.ts index 655db3b5e05..1a954d06721 100644 --- a/packages/cli/src/ui/commands/clearCommand.ts +++ b/packages/cli/src/ui/commands/clearCommand.ts @@ -19,6 +19,7 @@ import { function hasBlockingBackgroundWork(config: Config): boolean { return ( config.getBackgroundTaskRegistry().hasUnfinalizedTasks() || + config.getMonitorRegistry().getRunning().length > 0 || config .getBackgroundShellRegistry() .getAll() @@ -27,8 +28,9 @@ function hasBlockingBackgroundWork(config: Config): boolean { } function resetBackgroundStateForSessionSwitch(config: Config): void { - (config.getBackgroundTaskRegistry() as unknown as { reset(): void }).reset(); - (config.getBackgroundShellRegistry() as unknown as { reset(): void }).reset(); + config.getBackgroundTaskRegistry().reset(); + config.getMonitorRegistry().reset(); + config.getBackgroundShellRegistry().reset(); } export const clearCommand: SlashCommand = { @@ -65,7 +67,13 @@ export const clearCommand: SlashCommand = { config.getDebugLogger().warn(`SessionEnd hook failed: ${err}`); }); + // Abort old-session async work before creating the new session so + // cancellation notifications cannot leak across the reset boundary. + config.getBackgroundTaskRegistry().abortAll({ notify: false }); + config.getMonitorRegistry().abortAll({ notify: false }); + config.getBackgroundShellRegistry().abortAll(); resetBackgroundStateForSessionSwitch(config); + const newSessionId = config.startNewSession(); // Reset UI telemetry metrics for the new session diff --git a/packages/cli/src/ui/components/subagents/runtime/AgentExecutionDisplay.test.tsx b/packages/cli/src/ui/components/subagents/runtime/AgentExecutionDisplay.test.tsx index 7429b86862f..b5a148bc027 100644 --- a/packages/cli/src/ui/components/subagents/runtime/AgentExecutionDisplay.test.tsx +++ b/packages/cli/src/ui/components/subagents/runtime/AgentExecutionDisplay.test.tsx @@ -178,9 +178,7 @@ describe('', () => { ); const completedFrame = lastFrame() ?? ''; - expect(visualRowCount(completedFrame)).toBeLessThanOrEqual( - availableHeight, - ); + expect(visualRowCount(completedFrame)).toBeLessThanOrEqual(availableHeight); // useKeypress is now `{ isActive: false }`; ctrl+e on the completed // instance must not toggle anything. The mock unsets keypressHandler diff --git a/packages/cli/src/ui/hooks/useGeminiStream.test.tsx b/packages/cli/src/ui/hooks/useGeminiStream.test.tsx index fc24188896d..f878739f7b2 100644 --- a/packages/cli/src/ui/hooks/useGeminiStream.test.tsx +++ b/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); @@ -2917,7 +2920,7 @@ describe('useGeminiStream', () => { type: ServerGeminiEventType.Retry, }; yield { - type: ServerGeminiEventType.Text, + type: ServerGeminiEventType.Content, value: 'Success after retry', }; yield { @@ -3223,7 +3226,7 @@ describe('useGeminiStream', () => { mockSendMessageStream.mockReturnValueOnce( (async function* () { yield { - type: ServerGeminiEventType.Text, + type: ServerGeminiEventType.Content, value: 'Success response', }; })(), diff --git a/packages/cli/src/ui/hooks/useGeminiStream.ts b/packages/cli/src/ui/hooks/useGeminiStream.ts index 6d9ce9128ca..0bc9179efb2 100644 --- a/packages/cli/src/ui/hooks/useGeminiStream.ts +++ b/packages/cli/src/ui/hooks/useGeminiStream.ts @@ -2265,6 +2265,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/cli/src/ui/hooks/useResumeCommand.test.ts b/packages/cli/src/ui/hooks/useResumeCommand.test.ts index d4e49024916..afad6b21789 100644 --- a/packages/cli/src/ui/hooks/useResumeCommand.test.ts +++ b/packages/cli/src/ui/hooks/useResumeCommand.test.ts @@ -151,6 +151,7 @@ describe('useResumeCommand', () => { const geminiClient = { initialize: vi.fn(), }; + const resetMonitorRegistry = vi.fn(); const config = { getTargetDir: () => '/tmp', @@ -164,6 +165,10 @@ describe('useResumeCommand', () => { getAll: vi.fn().mockReturnValue([]), reset: vi.fn(), }), + getMonitorRegistry: () => ({ + getRunning: vi.fn().mockReturnValue([]), + reset: resetMonitorRegistry, + }), loadPausedBackgroundAgents: vi.fn().mockResolvedValue([]), getBackgroundAgentResumeService: () => ({ buildRecoveredBackgroundAgentsNotice: vi.fn(), @@ -216,6 +221,7 @@ describe('useResumeCommand', () => { expect(geminiClient.initialize).toHaveBeenCalledTimes(1); expect(historyManager.clearItems).toHaveBeenCalledTimes(1); expect(historyManager.loadHistory).toHaveBeenCalledTimes(1); + expect(resetMonitorRegistry).toHaveBeenCalledTimes(1); }); it('adds a recovered-background-agents notice when paused agents are restored', async () => { @@ -244,6 +250,10 @@ describe('useResumeCommand', () => { getAll: vi.fn().mockReturnValue([]), reset: vi.fn(), }), + getMonitorRegistry: () => ({ + getRunning: vi.fn().mockReturnValue([]), + reset: vi.fn(), + }), loadPausedBackgroundAgents: vi .fn() .mockResolvedValue([{ agentId: 'a' }, { agentId: 'b' }]), @@ -298,6 +308,73 @@ describe('useResumeCommand', () => { getAll: vi.fn().mockReturnValue([]), reset: vi.fn(), }), + getMonitorRegistry: () => ({ + getRunning: vi.fn().mockReturnValue([]), + reset: vi.fn(), + }), + getTargetDir: () => '/tmp', + getDebugLogger: () => ({ + warn: vi.fn(), + debug: vi.fn(), + error: vi.fn(), + }), + } as unknown as import('@qwen-code/qwen-code-core').Config; + + const { result } = renderHook(() => + useResumeCommand({ + config, + historyManager, + startNewSession, + }), + ); + + act(() => { + result.current.openResumeDialog(); + }); + + await act(async () => { + await result.current.handleResume('session-blocked'); + }); + + expect(result.current.isResumeDialogOpen).toBe(false); + expect(startNewSession).not.toHaveBeenCalled(); + expect(historyManager.clearItems).not.toHaveBeenCalled(); + expect(historyManager.loadHistory).not.toHaveBeenCalled(); + expect(historyManager.addItem).toHaveBeenCalledWith( + expect.objectContaining({ + type: 'error', + text: BACKGROUND_WORK_SWITCH_BLOCKED_MESSAGE, + }), + expect.any(Number), + ); + }); + + it('blocks resume when the current session still has a running monitor', async () => { + const historyManager = { + addItem: vi.fn(), + clearItems: vi.fn(), + loadHistory: vi.fn(), + }; + const startNewSession = vi.fn(); + + const config = { + getBackgroundTaskRegistry: () => ({ + hasUnfinalizedTasks: vi.fn().mockReturnValue(false), + reset: vi.fn(), + }), + getBackgroundShellRegistry: () => ({ + getAll: vi.fn().mockReturnValue([]), + reset: vi.fn(), + }), + getMonitorRegistry: () => ({ + getRunning: vi.fn().mockReturnValue([ + { + monitorId: 'mon_123', + status: 'running', + }, + ]), + reset: vi.fn(), + }), getTargetDir: () => '/tmp', getDebugLogger: () => ({ warn: vi.fn(), diff --git a/packages/cli/src/ui/hooks/useResumeCommand.ts b/packages/cli/src/ui/hooks/useResumeCommand.ts index 9541f305429..0a5e9e6a2e1 100644 --- a/packages/cli/src/ui/hooks/useResumeCommand.ts +++ b/packages/cli/src/ui/hooks/useResumeCommand.ts @@ -48,6 +48,7 @@ const BACKGROUND_WORK_SWITCH_BLOCKED_MESSAGE = function hasBlockingBackgroundWork(config: Config): boolean { return ( config.getBackgroundTaskRegistry().hasUnfinalizedTasks() || + config.getMonitorRegistry().getRunning().length > 0 || config .getBackgroundShellRegistry() .getAll() @@ -56,8 +57,9 @@ function hasBlockingBackgroundWork(config: Config): boolean { } function resetBackgroundStateForSessionSwitch(config: Config): void { - (config.getBackgroundTaskRegistry() as unknown as { reset(): void }).reset(); - (config.getBackgroundShellRegistry() as unknown as { reset(): void }).reset(); + config.getBackgroundTaskRegistry().reset(); + config.getMonitorRegistry().reset(); + config.getBackgroundShellRegistry().reset(); } export function useResumeCommand( diff --git a/packages/core/src/agents/background-tasks.test.ts b/packages/core/src/agents/background-tasks.test.ts index 7da05305d98..44f92025c57 100644 --- a/packages/core/src/agents/background-tasks.test.ts +++ b/packages/core/src/agents/background-tasks.test.ts @@ -345,31 +345,54 @@ describe('BackgroundTaskRegistry', () => { expect(callback).toHaveBeenCalledTimes(2); }); - it('persists shutdown interruption as running sidecar state', () => { - const patchSpy = vi - .spyOn(transcript, 'patchAgentMeta') - .mockImplementation(() => undefined); + it('abortAll({ notify: false }) suppresses terminal notifications from old tasks', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + + registry.register({ + agentId: 'a', + description: 'agent a', + status: 'running', + startTime: Date.now(), + abortController: new AbortController(), + }); + + registry.abortAll({ notify: false }); + + expect(registry.get('a')!.status).toBe('cancelled'); + expect(registry.hasUnfinalizedTasks()).toBe(false); + expect(callback).not.toHaveBeenCalled(); + + registry.complete('a', 'late result'); + registry.finalizeCancelled('a', 'late partial'); + + expect(callback).not.toHaveBeenCalled(); + expect(registry.get('a')!.status).toBe('cancelled'); + expect(registry.get('a')!.result).toBeUndefined(); + }); + + it('abortAll({ notify: false }) suppresses pending fallback notifications', () => { + vi.useFakeTimers(); try { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register({ agentId: 'a', description: 'agent a', status: 'running', startTime: Date.now(), abortController: new AbortController(), - metaPath: '/tmp/a.meta.json', }); - registry.abortAll(); + registry.cancel('a'); + registry.abortAll({ notify: false }); + vi.runAllTimers(); - expect(patchSpy).toHaveBeenCalledWith( - '/tmp/a.meta.json', - expect.objectContaining({ - status: 'running', - lastError: undefined, - }), - ); + expect(callback).not.toHaveBeenCalled(); + expect(registry.hasUnfinalizedTasks()).toBe(false); } finally { - patchSpy.mockRestore(); + vi.useRealTimers(); } }); diff --git a/packages/core/src/agents/background-tasks.ts b/packages/core/src/agents/background-tasks.ts index 77015700269..0aa55677860 100644 --- a/packages/core/src/agents/background-tasks.ts +++ b/packages/core/src/agents/background-tasks.ts @@ -156,10 +156,6 @@ export interface BackgroundTaskEntry { >; } -interface CancelOptions { - persistedStatus?: Extract; -} - export interface NotificationMeta { agentId: string; status: BackgroundTaskStatus; @@ -175,6 +171,11 @@ export type BackgroundNotificationCallback = ( export type BackgroundRegisterCallback = (entry: BackgroundTaskEntry) => void; +interface BackgroundTaskCancelOptions { + notify?: boolean; + persistedStatus?: Extract; +} + /** * Fires on entry status transitions — register, complete, fail, cancel. * Intentionally does NOT fire on `appendActivity` so consumers that only @@ -265,7 +266,7 @@ export class BackgroundTaskRegistry { // case where a tool ignores AbortSignal and bgBody never settles — the // timeout lands on finalizeCancellationIfPending(), which is a no-op // once the natural handler has already emitted. - cancel(agentId: string, options: CancelOptions = {}): void { + cancel(agentId: string, options: BackgroundTaskCancelOptions = {}): void { const entry = this.agents.get(agentId); if (!entry || entry.status !== 'running') return; const persistedStatus = options.persistedStatus ?? 'cancelled'; @@ -284,6 +285,13 @@ export class BackgroundTaskRegistry { debugLogger.info(`Background agent cancelled: ${agentId}`); this.emitStatusChange(entry); + if (options.notify === false) { + // Session reset paths intentionally suppress the old task's terminal + // notification so it cannot leak into a new conversation. + entry.notified = true; + return; + } + const timer = setTimeout(() => { this.finalizeCancellationIfPending(agentId); }, CANCEL_GRACE_MS); @@ -458,9 +466,21 @@ export class BackgroundTaskRegistry { this.activityChangeCallback = cb; } - abortAll(): void { + abortAll(options: BackgroundTaskCancelOptions = {}): void { + const cancelOptions: BackgroundTaskCancelOptions = { + persistedStatus: 'running', + ...options, + }; for (const entry of Array.from(this.agents.values())) { - this.cancel(entry.agentId, { persistedStatus: 'running' }); + if (entry.status === 'running') { + this.cancel(entry.agentId, cancelOptions); + } + + if (cancelOptions.notify === false) { + entry.notified = true; + continue; + } + // Shutdown path: no natural handler will run, so emit the cancelled // notification here to honour the one-notification-per-agent contract. this.finalizeCancellationIfPending(entry.agentId); diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index 0f80ca1120d..bcf31ae36f6 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 { BackgroundAgentResumeService } from '../agents/background-agent-resume.js'; import { BackgroundShellRegistry } from '../services/backgroundShellRegistry.js'; import { FileReadCache } from '../services/fileReadCache.js'; @@ -556,6 +557,7 @@ export class Config { private promptRegistry!: PromptRegistry; private subagentManager!: SubagentManager; private readonly backgroundTaskRegistry = new BackgroundTaskRegistry(); + private readonly monitorRegistry = new MonitorRegistry(); private backgroundAgentResumeService?: BackgroundAgentResumeService; private readonly backgroundShellRegistry = new BackgroundShellRegistry(); // Field initializer runs once on the parent Config; child Configs @@ -1657,6 +1659,7 @@ export class Config { } this.backgroundTaskRegistry.abortAll(); + this.monitorRegistry.abortAll({ notify: false }); this.backgroundShellRegistry.abortAll(); await this.cleanupArenaRuntime(); @@ -2537,6 +2540,10 @@ export class Config { return this.backgroundTaskRegistry; } + getMonitorRegistry(): MonitorRegistry { + return this.monitorRegistry; + } + getBackgroundAgentResumeService(): BackgroundAgentResumeService { if (!this.backgroundAgentResumeService) { this.backgroundAgentResumeService = new BackgroundAgentResumeService( @@ -2868,6 +2875,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/core/permission-helpers.test.ts b/packages/core/src/core/permission-helpers.test.ts new file mode 100644 index 00000000000..6197c1b1702 --- /dev/null +++ b/packages/core/src/core/permission-helpers.test.ts @@ -0,0 +1,77 @@ +/** + * @license + * Copyright 2025 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import path from 'node:path'; +import { describe, expect, it } from 'vitest'; +import { buildPermissionCheckContext } from './permission-helpers.js'; + +describe('buildPermissionCheckContext', () => { + it('uses an absolute directory as the permission cwd', () => { + expect( + buildPermissionCheckContext( + 'run_shell_command', + { + command: 'cat ./secret.txt', + directory: '/project/subdir', + }, + '/project', + ), + ).toMatchObject({ + toolName: 'run_shell_command', + command: 'cat ./secret.txt', + cwd: '/project/subdir', + }); + }); + + it('resolves a relative directory against the target dir', () => { + expect( + buildPermissionCheckContext( + 'run_shell_command', + { + command: 'cat ./secret.txt', + directory: 'subdir', + }, + '/project', + ), + ).toMatchObject({ + toolName: 'run_shell_command', + command: 'cat ./secret.txt', + cwd: path.resolve('/project', 'subdir'), + }); + }); + + it('returns raw monitor command — normalization is PM responsibility', () => { + expect( + buildPermissionCheckContext( + 'monitor', + { + command: String.raw`FOO="bar baz" /bin/bash --noprofile -c 'tail -f ./app.log &'`, + directory: '/project/subdir', + }, + '/project', + ), + ).toMatchObject({ + toolName: 'monitor', + command: String.raw`FOO="bar baz" /bin/bash --noprofile -c 'tail -f ./app.log &'`, + cwd: '/project/subdir', + }); + }); + + it('returns raw monitor command with suffix — normalization is PM responsibility', () => { + expect( + buildPermissionCheckContext( + 'monitor', + { + command: `/bin/bash -c 'tail -f ./app.log' && rm -rf /tmp/owned`, + }, + '/project', + ), + ).toMatchObject({ + toolName: 'monitor', + command: `/bin/bash -c 'tail -f ./app.log' && rm -rf /tmp/owned`, + }); + }); +}); diff --git a/packages/core/src/core/permission-helpers.ts b/packages/core/src/core/permission-helpers.ts index 6d537e7e6b1..c5246d6231c 100644 --- a/packages/core/src/core/permission-helpers.ts +++ b/packages/core/src/core/permission-helpers.ts @@ -37,8 +37,17 @@ export function buildPermissionCheckContext( toolParams: Record, targetDir: string, ): PermissionCheckContext { - const command = + const rawCommand = 'command' in toolParams ? String(toolParams['command']) : undefined; + // Monitor command normalization is handled by + // PermissionManager.normalizePermissionContext — single point of truth. + const command = rawCommand; + const cwd = + typeof toolParams['directory'] === 'string' + ? path.isAbsolute(toolParams['directory']) + ? toolParams['directory'] + : path.resolve(targetDir, toolParams['directory']) + : undefined; // Extract file path — tools use 'file_path' or 'path' (LS / grep / glob). let filePath = @@ -69,7 +78,7 @@ export function buildPermissionCheckContext( ? toolParams['subagent_type'] : undefined; - return { toolName, command, filePath, domain, specifier }; + return { toolName, command, cwd, filePath, domain, specifier }; } // ───────────────────────────────────────────────────────────────────────────── diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index ed751c93177..329cbb6fef8 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -149,6 +149,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'; diff --git a/packages/core/src/permissions/permission-manager.test.ts b/packages/core/src/permissions/permission-manager.test.ts index 7196615ba2d..9e5a031b2c1 100644 --- a/packages/core/src/permissions/permission-manager.test.ts +++ b/packages/core/src/permissions/permission-manager.test.ts @@ -112,6 +112,14 @@ describe('toolMatchesRuleToolName', () => { expect(toolMatchesRuleToolName('edit', 'write_file')).toBe(true); }); + it('"Bash" (run_shell_command) covers monitor', async () => { + expect(toolMatchesRuleToolName('run_shell_command', 'monitor')).toBe(true); + }); + + it('monitor rules do not cover run_shell_command', async () => { + expect(toolMatchesRuleToolName('monitor', 'run_shell_command')).toBe(false); + }); + it('does not cross categories', async () => { expect(toolMatchesRuleToolName('read_file', 'edit')).toBe(false); expect(toolMatchesRuleToolName('edit', 'read_file')).toBe(false); @@ -137,6 +145,11 @@ describe('parseRule', () => { expect(r.toolName).toBe('run_shell_command'); }); + it('parses Monitor alias', async () => { + const r = parseRule('Monitor'); + expect(r.toolName).toBe('monitor'); + }); + it('parses a shell tool with a specifier', async () => { const r = parseRule('Bash(git *)'); expect(r.toolName).toBe('run_shell_command'); @@ -144,6 +157,13 @@ describe('parseRule', () => { expect(r.specifierKind).toBe('command'); }); + it('parses Monitor with command specifier', async () => { + const r = parseRule('Monitor(tail -f *)'); + expect(r.toolName).toBe('monitor'); + expect(r.specifier).toBe('tail -f *'); + expect(r.specifierKind).toBe('command'); + }); + it('parses Read with path specifier', async () => { const r = parseRule('Read(./secrets/**)'); expect(r.toolName).toBe('read_file'); @@ -645,6 +665,27 @@ describe('matchesRule', () => { expect(matchesRule(rule, 'run_shell_command', 'echo hello')).toBe(false); }); + // Monitor command specifier + it('Monitor rule matches monitor invocations with command specifier', async () => { + const rule = parseRule('Monitor(tail -f *)'); + expect(matchesRule(rule, 'monitor')).toBe(false); // no command + expect(matchesRule(rule, 'monitor', 'tail -f /var/log/app.log')).toBe(true); + expect(matchesRule(rule, 'monitor', 'echo hello')).toBe(false); + }); + + it('Monitor rule does not match run_shell_command', async () => { + const rule = parseRule('Monitor(tail -f *)'); + expect( + matchesRule(rule, 'run_shell_command', 'tail -f /var/log/app.log'), + ).toBe(false); + }); + + it('Bash rule also covers monitor (shell deny rules block monitor)', async () => { + const rule = parseRule('Bash(tail -f *)'); + expect(matchesRule(rule, 'monitor', 'tail -f /var/log/app.log')).toBe(true); + expect(matchesRule(rule, 'monitor', 'echo hello')).toBe(false); + }); + it('matchesRule checks individual simple commands (compound splitting is at PM level)', async () => { const rule = parseRule('Bash(git *)'); // matchesRule receives a simple command (already split by PM) @@ -931,6 +972,224 @@ describe('PermissionManager', () => { // 'ls' is readonly, resolves to 'allow' when no rule matches expect(await pm.isCommandAllowed('ls')).toBe('allow'); }); + + it('resolves shell virtual file operations relative to the explicit cwd', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Read(./subdir/secret.txt)'], + projectRoot: '/project', + cwd: '/project', + }), + ); + pm2.initialize(); + + expect( + await pm2.evaluate({ + toolName: 'run_shell_command', + command: 'cat ./secret.txt', + cwd: '/project/subdir', + }), + ).toBe('allow'); + }); + + it('applies relative virtual file rules using the shell invocation cwd', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsDeny: ['Read(./secret.txt)'], + projectRoot: '/project', + cwd: '/project', + }), + ); + pm2.initialize(); + + expect( + await pm2.evaluate({ + toolName: 'run_shell_command', + command: 'cat ./secret.txt', + cwd: '/project/subdir', + }), + ).toBe('deny'); + }); + }); + + describe('monitor command-level evaluation', () => { + it('Monitor(...) allow rule matches monitor invocations', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Monitor(tail -f *)'], + }), + ); + pm2.initialize(); + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: 'tail -f /var/log/app.log', + }), + ).toBe('allow'); + }); + + it('Monitor(...) allow rule matches wrapped monitor invocations', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Monitor(tail -f *)'], + }), + ); + pm2.initialize(); + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: `/bin/bash --noprofile -c 'tail -f /var/log/app.log &'`, + }), + ).toBe('allow'); + }); + + it('exact Monitor(...) allow rule matches wrapped fallback commands', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Monitor(FOO="bar baz" tail -f /var/log/app.log)'], + }), + ); + pm2.initialize(); + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: String.raw`FOO="bar baz" /bin/bash --noprofile -c 'tail -f /var/log/app.log &'`, + }), + ).toBe('allow'); + }); + + it('Monitor(...) deny rule sees shell wrapper suffix commands', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Monitor(tail -f *)'], + permissionsDeny: ['Monitor(rm *)'], + }), + ); + pm2.initialize(); + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: `/bin/bash -c 'tail -f /var/log/app.log' && rm -rf /tmp/owned`, + }), + ).toBe('deny'); + }); + + it('Monitor(...) deny rule blocks monitor invocations', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsDeny: ['Monitor(rm *)'], + }), + ); + pm2.initialize(); + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: 'rm -rf /', + }), + ).toBe('deny'); + }); + + it('Monitor approval does NOT allow run_shell_command', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Monitor(npm *)'], + }), + ); + pm2.initialize(); + // Same command via shell tool should NOT be allowed by Monitor rule + expect( + await pm2.evaluate({ + toolName: 'run_shell_command', + command: 'npm install', + }), + ).not.toBe('allow'); + }); + + it('Bash approval also allows monitor (shell rules cover monitor)', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Bash(npm *)'], + }), + ); + pm2.initialize(); + // Same command via monitor tool should be allowed by Bash rule + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: 'npm install', + }), + ).toBe('allow'); + }); + + it('Bash deny rule blocks equivalent monitor command', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsDeny: ['Bash(rm *)'], + }), + ); + pm2.initialize(); + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: 'rm -rf /', + }), + ).toBe('deny'); + }); + + it('resolves default to allow for readonly monitor commands', async () => { + const pm2 = new PermissionManager(makeConfig({})); + pm2.initialize(); + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: 'echo hello', + }), + ).toBe('allow'); + }); + + it('applies relative virtual file deny rules using the monitor cwd', async () => { + // Mirrors the run_shell_command coverage above: when a monitor is + // started with an explicit `directory` (forwarded as `cwd` via + // buildPermissionCheckContext), relative-path deny rules must resolve + // against that directory rather than the global config cwd. Without + // this propagation a `Read(./secret.txt)` rule could be silently + // bypassed by switching the monitor's working directory. + const pm2 = new PermissionManager( + makeConfig({ + permissionsDeny: ['Read(./secret.txt)'], + projectRoot: '/project', + cwd: '/project', + }), + ); + pm2.initialize(); + + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: 'cat ./secret.txt', + cwd: '/project/subdir', + }), + ).toBe('deny'); + }); + + it('resolves monitor virtual file allow rules relative to the explicit cwd', async () => { + const pm2 = new PermissionManager( + makeConfig({ + permissionsAllow: ['Read(./subdir/secret.txt)'], + projectRoot: '/project', + cwd: '/project', + }), + ); + pm2.initialize(); + + expect( + await pm2.evaluate({ + toolName: 'monitor', + command: 'cat ./secret.txt', + cwd: '/project/subdir', + }), + ).toBe('allow'); + }); }); describe('compound command evaluation', () => { @@ -1691,6 +1950,19 @@ describe('buildPermissionRules', () => { const rules = buildPermissionRules({ toolName: 'run_shell_command' }); expect(rules).toEqual(['Bash']); }); + + it('generates Monitor rule with command specifier', async () => { + const rules = buildPermissionRules({ + toolName: 'monitor', + command: 'tail -f /var/log/app.log', + }); + expect(rules).toEqual(['Monitor(tail -f /var/log/app.log)']); + }); + + it('falls back to bare Monitor display name when no command', async () => { + const rules = buildPermissionRules({ toolName: 'monitor' }); + expect(rules).toEqual(['Monitor']); + }); }); describe('literal-specifier tools', () => { @@ -1741,6 +2013,10 @@ describe('buildHumanReadableRuleLabel', () => { expect(buildHumanReadableRuleLabel(['Bash'])).toBe('run commands'); }); + it('converts bare Monitor rule to "monitor commands"', () => { + expect(buildHumanReadableRuleLabel(['Monitor'])).toBe('monitor commands'); + }); + it('converts Read with absolute path specifier', () => { const label = buildHumanReadableRuleLabel(['Read(//Users/mochi/.qwen/**)']); expect(label).toBe('read files in /Users/mochi/.qwen/'); @@ -1761,6 +2037,11 @@ describe('buildHumanReadableRuleLabel', () => { expect(label).toBe("run 'git *' commands"); }); + it('converts Monitor with command specifier', () => { + const label = buildHumanReadableRuleLabel(['Monitor(tail -f *)']); + expect(label).toBe("monitor 'tail -f *' commands"); + }); + it('converts WebFetch with domain specifier', () => { const label = buildHumanReadableRuleLabel(['WebFetch(github.com)']); expect(label).toBe('fetch from github.com'); diff --git a/packages/core/src/permissions/permission-manager.ts b/packages/core/src/permissions/permission-manager.ts index 34303fc5d12..fd71399890d 100644 --- a/packages/core/src/permissions/permission-manager.ts +++ b/packages/core/src/permissions/permission-manager.ts @@ -15,7 +15,10 @@ import type { PathMatchContext } from './rule-parser.js'; import { extractShellOperations } from './shell-semantics.js'; import type { ShellOperation } from './shell-semantics.js'; import { isShellCommandReadOnlyAST } from '../utils/shellAstParser.js'; -import { detectCommandSubstitution } from '../utils/shell-utils.js'; +import { + detectCommandSubstitution, + normalizeMonitorCommand, +} from '../utils/shell-utils.js'; import { createDebugLogger } from '../utils/debugLogger.js'; import type { PermissionCheckContext, @@ -29,6 +32,13 @@ import type { const debugLogger = createDebugLogger('PERMISSIONS'); +/** + * Tools that spawn shell commands and share the same permission evaluation + * semantics: compound-command splitting, AST read-only analysis, and + * virtual file/network operation matching. + */ +const SHELL_LIKE_TOOLS = new Set(['run_shell_command', 'monitor']); + /** * Numeric priority for each PermissionDecision. * Higher number = more restrictive. Used to combine decisions by taking @@ -159,6 +169,7 @@ export class PermissionManager { * @returns A PermissionDecision indicating how to handle this tool call. */ async evaluate(ctx: PermissionCheckContext): Promise { + ctx = this.normalizePermissionContext(ctx); const { command, toolName } = ctx; // For shell commands, split compound commands and evaluate each @@ -178,7 +189,7 @@ export class PermissionManager { // a concrete permission (deny/ask/allow) based on the command's readonly status. if ( decision === 'default' && - toolName === 'run_shell_command' && + SHELL_LIKE_TOOLS.has(toolName) && command !== undefined ) { return this.resolveDefaultPermission(command); @@ -198,14 +209,14 @@ export class PermissionManager { * to match equivalent shell commands (e.g. `cat` → Read, `curl` → WebFetch). */ private evaluateSingle(ctx: PermissionCheckContext): PermissionDecision { - const { toolName, command, filePath, domain, specifier } = ctx; + const { toolName, command, cwd, filePath, domain, specifier } = ctx; // Build path context for resolving relative path patterns const pathCtx: PathMatchContext | undefined = this.config.getProjectRoot && this.config.getCwd ? { projectRoot: this.config.getProjectRoot(), - cwd: this.config.getCwd(), + cwd: cwd ?? this.config.getCwd(), } : undefined; @@ -256,7 +267,7 @@ export class PermissionManager { // must never downgrade an explicit 'allow' decision from a Bash rule. // Example: `git status` has no file ops; an allow rule for `Bash(git *)` // should return 'allow', not be downgraded to 'default'. - if (toolName === 'run_shell_command' && command !== undefined) { + if (SHELL_LIKE_TOOLS.has(toolName) && command !== undefined) { const cwd = pathCtx?.cwd ?? process.cwd(); const virtualDecision = this.evaluateShellVirtualOps( extractShellOperations(command, cwd), @@ -284,7 +295,7 @@ export class PermissionManager { */ private evaluateShellVirtualOps( ops: ShellOperation[], - _pathCtx: PathMatchContext | undefined, + pathCtx: PathMatchContext | undefined, ): PermissionDecision { if (ops.length === 0) return 'default'; @@ -296,6 +307,7 @@ export class PermissionManager { // into the shell-semantics branch. const opDecision = this.evaluateSingle({ toolName: op.virtualTool, + cwd: pathCtx?.cwd, filePath: op.filePath, domain: op.domain, }); @@ -395,6 +407,26 @@ export class PermissionManager { return 'ask'; } + private normalizePermissionContext( + ctx: PermissionCheckContext, + ): PermissionCheckContext { + if (ctx.toolName !== 'monitor' || ctx.command === undefined) { + return ctx; + } + + // Note on cwd: callers wired through `buildPermissionCheckContext` + // already populate `ctx.cwd` from the monitor's `directory` parameter + // (see permission-helpers.ts), and the spread below preserves it. That + // is what makes relative-path rules — including those derived from + // virtual shell ops in evaluateSingle() — resolve against the monitor's + // working directory rather than the global config cwd. Direct callers + // of `evaluate()` that bypass that helper must pass `cwd` themselves. + return { + ...ctx, + command: normalizeMonitorCommand(ctx.command).safetyCommand, + }; + } + // --------------------------------------------------------------------------- // Registry-level helper // --------------------------------------------------------------------------- @@ -418,6 +450,7 @@ export class PermissionManager { 'cron_create', 'cron_list', 'cron_delete', + 'monitor', ]); /** @@ -470,13 +503,14 @@ export class PermissionManager { * Useful for providing user-visible feedback about which rule caused a denial. */ findMatchingDenyRule(ctx: PermissionCheckContext): string | undefined { - const { toolName, command, filePath, domain, specifier } = ctx; + ctx = this.normalizePermissionContext(ctx); + const { toolName, command, cwd, filePath, domain, specifier } = ctx; const pathCtx: PathMatchContext | undefined = this.config.getProjectRoot && this.config.getCwd ? { projectRoot: this.config.getProjectRoot(), - cwd: this.config.getCwd(), + cwd: cwd ?? this.config.getCwd(), } : undefined; @@ -510,10 +544,14 @@ export class PermissionManager { * @param command - The shell command to evaluate. * @returns The PermissionDecision for this command. */ - async isCommandAllowed(command: string): Promise { + async isCommandAllowed( + command: string, + cwd?: string, + ): Promise { return this.evaluate({ toolName: 'run_shell_command', command, + cwd, }); } @@ -543,9 +581,10 @@ export class PermissionManager { * @returns true if at least one rule matches. */ hasRelevantRules(ctx: PermissionCheckContext): boolean { - const { toolName, command, filePath, domain, specifier } = ctx; + ctx = this.normalizePermissionContext(ctx); + const { toolName, command, cwd, filePath, domain, specifier } = ctx; - if (ctx.toolName === 'run_shell_command' && command !== undefined) { + if (SHELL_LIKE_TOOLS.has(ctx.toolName) && command !== undefined) { const subCommands = splitCompoundCommand(command); if (subCommands.length > 1) { return subCommands.some((subCmd) => @@ -558,7 +597,7 @@ export class PermissionManager { this.config.getProjectRoot && this.config.getCwd ? { projectRoot: this.config.getProjectRoot(), - cwd: this.config.getCwd(), + cwd: cwd ?? this.config.getCwd(), } : undefined; @@ -586,7 +625,7 @@ export class PermissionManager { // extracted from the command has a relevant rule. This ensures the PM is // consulted (and the confirmation dialog shown) when Read/Edit/etc. rules // would match equivalent shell commands. - if (ctx.toolName === 'run_shell_command' && ctx.command !== undefined) { + if (SHELL_LIKE_TOOLS.has(ctx.toolName) && ctx.command !== undefined) { const cwd = pathCtx?.cwd ?? process.cwd(); const ops = extractShellOperations(ctx.command, cwd); if ( @@ -619,9 +658,10 @@ export class PermissionManager { * real ask rule matched. */ hasMatchingAskRule(ctx: PermissionCheckContext): boolean { - const { toolName, command, filePath, domain, specifier } = ctx; + ctx = this.normalizePermissionContext(ctx); + const { toolName, command, cwd, filePath, domain, specifier } = ctx; - if (ctx.toolName === 'run_shell_command' && command !== undefined) { + if (SHELL_LIKE_TOOLS.has(ctx.toolName) && command !== undefined) { const subCommands = splitCompoundCommand(command); if (subCommands.length > 1) { return subCommands.some((subCmd) => @@ -634,7 +674,7 @@ export class PermissionManager { this.config.getProjectRoot && this.config.getCwd ? { projectRoot: this.config.getProjectRoot(), - cwd: this.config.getCwd(), + cwd: cwd ?? this.config.getCwd(), } : undefined; @@ -653,7 +693,7 @@ export class PermissionManager { return true; } - if (ctx.toolName === 'run_shell_command' && ctx.command !== undefined) { + if (SHELL_LIKE_TOOLS.has(ctx.toolName) && ctx.command !== undefined) { const cwd = pathCtx?.cwd ?? process.cwd(); const ops = extractShellOperations(ctx.command, cwd); return ops.some((op) => { diff --git a/packages/core/src/permissions/rule-parser.ts b/packages/core/src/permissions/rule-parser.ts index b034069c57c..bf3d3534cdf 100644 --- a/packages/core/src/permissions/rule-parser.ts +++ b/packages/core/src/permissions/rule-parser.ts @@ -118,14 +118,19 @@ export const TOOL_NAME_ALIASES: Readonly> = { Lsp: 'lsp', LspTool: 'lsp', + // Monitor tool + monitor: 'monitor', + Monitor: 'monitor', + MonitorTool: 'monitor', + // Legacy edit tool name replace: 'edit', }; /** - * Shell tool canonical names. + * Shell tool canonical names. These use command-style rule specifiers. */ -const SHELL_TOOL_NAMES = new Set(['run_shell_command']); +const SHELL_TOOL_NAMES = new Set(['run_shell_command', 'monitor']); /** * File-reading tools — "Read" rules apply to all of these (best-effort). @@ -188,6 +193,8 @@ export function getSpecifierKind(canonicalToolName: string): SpecifierKind { * * "Read" → resolves to "read_file", but also covers grep_search, glob, list_directory * "Edit" → resolves to "edit", but also covers write_file + * "Bash" → resolves to "run_shell_command", but also covers monitor + * "Monitor" → resolves to "monitor" only; it does not cover shell */ export function toolMatchesRuleToolName( ruleToolName: string, @@ -204,6 +211,12 @@ export function toolMatchesRuleToolName( if (ruleToolName === 'edit' && EDIT_TOOLS.has(contextToolName)) { return true; } + // "Bash" (run_shell_command) → also covers monitor so that existing + // `Bash(...)` allow rules are not silently bypassed by switching to + // the monitor tool. Monitor-only rules do NOT cover shell. + if (ruleToolName === 'run_shell_command' && contextToolName === 'monitor') { + return true; + } return false; } @@ -305,6 +318,8 @@ const CANONICAL_TO_RULE_DISPLAY: Readonly> = { write_file: 'Edit', // Shell run_shell_command: 'Bash', + // Monitor + monitor: 'Monitor', // Web web_fetch: 'WebFetch', // Agent / Skill @@ -419,6 +434,7 @@ const DISPLAY_NAME_TO_VERB: Readonly> = { Read: 'read files', Edit: 'edit files', Bash: 'run commands', + Monitor: 'monitor commands', WebFetch: 'fetch from', Agent: 'use agent', Skill: 'use skill', @@ -494,9 +510,13 @@ export function buildHumanReadableRuleLabel(rules: string[]): string { parts.push(`${verb} in ${cleanPath}`); break; } - case 'command': - parts.push(`run '${specifier}' commands`); + case 'command': { + const cmdVerb = DISPLAY_NAME_TO_VERB[displayName] ?? 'run'; + // Extract just the verb word (e.g. "run commands" → "run", "monitor commands" → "monitor") + const verbWord = cmdVerb.split(' ')[0]!; + parts.push(`${verbWord} '${specifier}' commands`); break; + } case 'domain': parts.push(`${verb} ${specifier}`); break; diff --git a/packages/core/src/permissions/types.ts b/packages/core/src/permissions/types.ts index d7c2ff296aa..3106c8e4d04 100644 --- a/packages/core/src/permissions/types.ts +++ b/packages/core/src/permissions/types.ts @@ -86,6 +86,11 @@ export interface PermissionCheckContext { * The shell command being executed (only for Bash / run_shell_command). */ command?: string; + /** + * Effective working directory for shell-like tools when resolving + * relative virtual file operations extracted from the command. + */ + cwd?: string; /** * The file path being accessed (only for Read / Edit / Write tools). * Should be an absolute path for matching against path patterns. diff --git a/packages/core/src/services/monitorRegistry.test.ts b/packages/core/src/services/monitorRegistry.test.ts new file mode 100644 index 00000000000..81bf8b7d97f --- /dev/null +++ b/packages/core/src/services/monitorRegistry.test.ts @@ -0,0 +1,587 @@ +/** + * @license + * Copyright 2025 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { + MAX_RETAINED_TERMINAL_MONITORS, + 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, + droppedLines: 0, + ...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('lets cancel abort handlers flush partial output before settling', () => { + const ac = new AbortController(); + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry({ abortController: ac })); + ac.signal.addEventListener( + 'abort', + () => { + registry.emitEvent('mon-1', 'last partial line'); + }, + { once: true }, + ); + + registry.cancel('mon-1'); + + const entry = registry.get('mon-1')!; + expect(entry.status).toBe('cancelled'); + expect(entry.eventCount).toBe(1); + expect(callback).toHaveBeenCalledTimes(2); + const [, eventModelText] = callback.mock.calls[0] as [string, string]; + const [, terminalModelText] = callback.mock.calls[1] as [string, string]; + expect(eventModelText).toContain('last partial line'); + expect(terminalModelText).toContain('cancelled'); + }); + + it('supports silent cancellation without terminal notification', () => { + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry()); + + registry.cancel('mon-1', { notify: false }); + + expect(registry.get('mon-1')!.status).toBe('cancelled'); + expect(callback).not.toHaveBeenCalled(); + }); + + 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-stop is re-entrancy safe: abort-driven flush cannot overshoot maxEvents or double-emit terminal', () => { + // Simulate the Monitor tool's abort listener flushing a buffered line + // back into emitEvent() synchronously. Before the fix, this re-entrant + // call would find status === 'running', increment eventCount past + // maxEvents, and emit a second "Max events reached" terminal + // notification. After the fix, settle() runs before abort() so the + // re-entrant emitEvent() short-circuits on the status guard. + const ac = new AbortController(); + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register(createEntry({ maxEvents: 2, abortController: ac })); + + // Abort listener that attempts to flush a buffered partial line. + ac.signal.addEventListener( + 'abort', + () => { + registry.emitEvent('mon-1', 'flushed-after-abort'); + }, + { once: true }, + ); + + registry.emitEvent('mon-1', 'line 1'); + registry.emitEvent('mon-1', 'line 2'); // triggers auto-stop + abort flush + + const entry = registry.get('mon-1')!; + expect(entry.status).toBe('completed'); + // eventCount must NOT exceed maxEvents; the flush must be dropped. + expect(entry.eventCount).toBe(2); + expect(ac.signal.aborted).toBe(true); + // Exactly 2 event notifications + 1 terminal notification. Neither the + // flushed line nor a second "Max events reached" notification fires. + expect(callback).toHaveBeenCalledTimes(3); + const terminalCalls = callback.mock.calls.filter( + (args) => + typeof args[1] === 'string' && + (args[1] as string).includes('Max events reached'), + ); + expect(terminalCalls).toHaveLength(1); + // And no flushed content leaked into any notification. + for (const args of callback.mock.calls) { + expect(args[0]).not.toContain('flushed-after-abort'); + expect(args[1]).not.toContain('flushed-after-abort'); + } + }); + + it('truncateDescription caps output at MAX_DESCRIPTION_LENGTH including ellipsis', () => { + // MAX_DESCRIPTION_LENGTH is a private constant but its value (80) is + // documented in the tool schema and mirrored by the Monitor tool. Verify + // that descriptions longer than the cap are truncated to exactly 80 + // chars total (ellipsis included), not 83. + const callback = vi.fn(); + registry.setNotificationCallback(callback); + const longDesc = 'A'.repeat(200); + registry.register(createEntry({ description: longDesc })); + + registry.emitEvent('mon-1', 'evt'); + + const [displayText, modelText] = callback.mock.calls[0] as [string, string]; + const displayMatch = displayText.match(/Monitor "([^"]*)"/); + expect(displayMatch).not.toBeNull(); + expect(displayMatch![1]!.length).toBeLessThanOrEqual(80); + expect(displayMatch![1]!.endsWith('...')).toBe(true); + + // The surrounding `"` chars in the template are literal; + // only the description itself flows through escapeXml. + const modelMatch = modelText.match(/Monitor "([^"]*)"/); + expect(modelMatch).not.toBeNull(); + expect(modelMatch![1]!.length).toBeLessThanOrEqual(80); + expect(modelMatch![1]!.endsWith('...')).toBe(true); + }); + + 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('lets idle-timeout abort handlers flush partial output before settling', () => { + const ac = new AbortController(); + const callback = vi.fn(); + registry.setNotificationCallback(callback); + registry.register( + createEntry({ + idleTimeoutMs: 5000, + abortController: ac, + }), + ); + ac.signal.addEventListener( + 'abort', + () => { + registry.emitEvent('mon-1', 'idle partial line'); + }, + { once: true }, + ); + + vi.advanceTimersByTime(5001); + + const entry = registry.get('mon-1')!; + expect(entry.status).toBe('completed'); + expect(entry.eventCount).toBe(1); + expect(callback).toHaveBeenCalledTimes(2); + const [, eventModelText] = callback.mock.calls[0] as [string, string]; + const [, terminalModelText] = callback.mock.calls[1] as [string, string]; + expect(eventModelText).toContain('idle partial line'); + expect(terminalModelText).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('