diff --git a/packages/agent-core-v2/src/agent/task/taskService.ts b/packages/agent-core-v2/src/agent/task/taskService.ts index 3b72188f8..699bf559c 100644 --- a/packages/agent-core-v2/src/agent/task/taskService.ts +++ b/packages/agent-core-v2/src/agent/task/taskService.ts @@ -93,6 +93,10 @@ interface ManagedTask { readonly outputChunks: string[]; outputSizeBytes: number; retainedOutputBytes: number; + /** + * True once a command has crossed `MAX_TASK_OUTPUT_BYTES` and termination has + * been requested. One-shot guard so the ceiling fires exactly once. + */ outputLimitTripped: boolean; status: AgentTaskStatus; options: RegisterAgentTaskOptions & { description?: string }; @@ -116,8 +120,30 @@ interface ManagedTask { handleSubscription?: { dispose(): void }; } -const MAX_OUTPUT_BYTES = 1024 * 1024; -const MAX_TASK_OUTPUT_BYTES = 16 * 1024 * 1024; +const MAX_OUTPUT_BYTES = 1024 * 1024; // 1 MiB + +/** + * Hard ceiling on the combined output a single shell command may stream before + * it is force-terminated (SIGTERM → grace → SIGKILL). It guards both the + * live-forward path and the on-disk `output.log` write chain from a runaway + * command (e.g. `b3sum --length `) whose output would otherwise grow + * without bound — filling the disk, or retaining each pending-write chunk until + * Node aborts with an out-of-memory crash. Scoped to process tasks (foreground + * and background); subagent and user-question results are appended once and must + * always be persisted, so they are intentionally not capped here. + */ +const MAX_TASK_OUTPUT_BYTES = 16 * 1024 * 1024; // 16 MiB + +/** Terminal `stopReason` recorded when a command trips the output ceiling. */ +function outputLimitReason(): string { + const mib = Math.floor(MAX_TASK_OUTPUT_BYTES / (1024 * 1024)); + return ( + `Output limit exceeded: the command produced more than ${mib} MiB and was ` + + 'terminated. Redirect large output to a file (e.g. `command > out.txt`) and ' + + 'inspect it in slices instead.' + ); +} + const SIGTERM_GRACE_MS = 5_000; const TASK_ID_ALPHABET = '0123456789abcdefghijklmnopqrstuvwxyz'; const USER_INTERRUPT_REASON = 'Interrupted by user'; @@ -774,6 +800,13 @@ export class AgentTaskService extends Disposable implements IAgentTaskService { entry.outputSizeBytes += chunkBytes; this.appendRetainedOutput(entry, chunk, chunkBytes); + // Output ceiling: a single shell command must not grow the (unbounded) + // live-forward buffer or the on-disk write chain until the process runs out + // of memory or fills the disk. Trip once, then request graceful termination + // through the shared stop path (SIGTERM → grace → SIGKILL). Scoped to + // process tasks (foreground and background): subagent and user-question tasks + // append their bounded result in one shot and must always persist it, so they + // are intentionally not capped here. if ( !entry.outputLimitTripped && entry.task?.kind === 'process' && @@ -783,6 +816,11 @@ export class AgentTaskService extends Disposable implements IAgentTaskService { void this.stop(entry.taskId, outputLimitReason()); } + // Once the cap has tripped the task is being terminated: keep only the + // bounded in-memory ring buffer above and stop feeding the (unbounded) disk + // write chain. A producer that ignores SIGTERM could otherwise keep the + // chain — and the chunk strings each pending write retains — growing through + // the grace window until SIGKILL, re-introducing the OOM this cap prevents. if (entry.outputLimitTripped) return; if (!entry.outputPersistStarted) { @@ -1045,15 +1083,6 @@ function emptyOutputSnapshot(): AgentTaskOutputSnapshot { }; } -function outputLimitReason(): string { - const mib = Math.floor(MAX_TASK_OUTPUT_BYTES / (1024 * 1024)); - return ( - `Output limit exceeded: the command produced more than ${String(mib)} MiB and was ` + - 'terminated. Redirect large output to a file (e.g. `command > out.txt`) and ' + - 'inspect it in slices instead.' - ); -} - function agentTaskNotificationChildren( output: AgentTaskOutputSnapshot, ): readonly string[] | undefined { diff --git a/packages/agent-core-v2/src/agent/tool/result-builder.ts b/packages/agent-core-v2/src/agent/tool/result-builder.ts index 85ddef6cb..34cf147db 100644 --- a/packages/agent-core-v2/src/agent/tool/result-builder.ts +++ b/packages/agent-core-v2/src/agent/tool/result-builder.ts @@ -27,6 +27,7 @@ export type ExecutableToolResultBuilderResult = ( readonly output: string; readonly message: string; readonly truncated: boolean; + readonly brief?: string; }; export class ToolResultBuilder { @@ -103,7 +104,7 @@ export class ToolResultBuilder { return charsWritten; } - ok(message = ''): ExecutableToolResultBuilderResult { + ok(message = '', options: { readonly brief?: string } = {}): ExecutableToolResultBuilderResult { let finalMessage = message; if (finalMessage.length > 0 && !finalMessage.endsWith('.')) { finalMessage += '.'; @@ -127,10 +128,14 @@ export class ToolResultBuilder { : output, message: finalMessage, truncated: this.truncationHappened, + brief: options.brief, }; } - error(message: string): ExecutableToolResultBuilderResult { + error( + message: string, + options: { readonly brief?: string } = {}, + ): ExecutableToolResultBuilderResult { const finalMessage = this.truncationHappened ? message.length === 0 ? TRUNCATION_MESSAGE @@ -149,6 +154,7 @@ export class ToolResultBuilder { : `${output}\n${finalMessage}`, message: finalMessage, truncated: this.truncationHappened, + brief: options.brief, }; } } diff --git a/packages/agent-core-v2/src/os/backends/node-local/tools/bash.ts b/packages/agent-core-v2/src/os/backends/node-local/tools/bash.ts index c8ac5b920..dd3d08ecc 100644 --- a/packages/agent-core-v2/src/os/backends/node-local/tools/bash.ts +++ b/packages/agent-core-v2/src/os/backends/node-local/tools/bash.ts @@ -38,7 +38,10 @@ import { ISessionContext } from '#/session/sessionContext/sessionContext'; import { ISessionProcessRunner, type IProcess } from '#/session/process/processRunner'; import { IAgentProfileService } from '#/agent/profile/profile'; import type { BuiltinTool, ExecutableToolResult, ToolExecution, ToolUpdate } from '#/agent/tool/toolContract'; -import { ToolResultBuilder } from '#/agent/tool/result-builder'; +import { + type ExecutableToolResultBuilderResult, + ToolResultBuilder, +} from '#/agent/tool/result-builder'; import { registerTool } from '#/agent/toolRegistry/toolContribution'; import { toInputJsonSchema } from '#/_base/tools/support/input-schema'; import { literalRulePattern, matchesGlobRuleSubject } from '#/_base/tools/support/rule-match'; @@ -146,7 +149,7 @@ function renderBashDescription(shellName: string): string { function withoutBackgroundDescription(description: string): string { return description .replace( - /\n\nIf `run_in_background=true`,[\s\S]*?point them to the `\/tasks` command, which opens an interactive panel; it has no subcommands\./, + /\r?\n\r?\nIf `run_in_background=true`,[\s\S]*?point them to the `\/tasks` command, which opens an interactive panel; it has no subcommands\./, '\n\nBackground execution is disabled for this agent. Do not set `run_in_background=true`.', ) .replace( @@ -154,7 +157,7 @@ function withoutBackgroundDescription(description: string): string { ` For possibly long-running commands, set the \`timeout\` argument in seconds. The default is ${String(DEFAULT_TIMEOUT_S)}s; foreground commands allow up to ${String(MAX_TIMEOUT_S)}s.`, ) .replace( - /\n- Prefer `run_in_background=true`[\s\S]*?conversation to continue before the command finishes\./, + /\r?\n- Prefer `run_in_background=true`[\s\S]*?conversation to continue before the command finishes\./, '\n- Do not set `run_in_background=true`; background task management tools are not available.', ); } @@ -179,7 +182,11 @@ export class BashTool implements BuiltinTool { } private allowBackground(): boolean { - return this.profile.isToolActive('TaskOutput') && this.profile.isToolActive('TaskStop'); + return ( + this.profile.isToolActive('TaskList') && + this.profile.isToolActive('TaskOutput') && + this.profile.isToolActive('TaskStop') + ); } get description(): string { @@ -286,6 +293,9 @@ export class BashTool implements BuiltinTool { { detached: startsInBackground, timeoutMs, + // Detaching (ctrl+b) moves a foreground command to the background; + // give it the background timeout so it is not still bounded by the + // shorter foreground deadline. detachTimeoutMs: DEFAULT_BACKGROUND_TIMEOUT_S * MS_PER_SECOND, signal: startsInBackground ? undefined : signal, }, @@ -300,11 +310,14 @@ export class BashTool implements BuiltinTool { }; } + // Foreground `!` shell commands surface their task id so the TUI can detach + // (ctrl+b) this exact task. Background runs are already detached. if (!startsInBackground) onForegroundTaskStart?.(taskId); if (startsInBackground) { - return this.detachedTaskResult(taskId, proc, description, { - title: 'Task started in background', + return this.backgroundStartedResult(taskId, proc, description, { + title: 'Background task started', + brief: `Started ${taskId}`, }); } @@ -312,19 +325,20 @@ export class BashTool implements BuiltinTool { const release = await this.tasks.waitForForegroundRelease(taskId); if (release === 'detached') { collectForegroundOutput = false; - return this.detachedTaskResult( + return this.backgroundStartedResult( taskId, proc, description, { title: 'Task moved to background', + brief: `Backgrounded ${taskId}`, }, builder, 'foreground_detached', ); } - return this.foregroundCompletionResult(taskId, proc, builder, foregroundTimeoutMs); + return await this.foregroundCompletionResult(taskId, proc, builder, foregroundTimeoutMs); } finally { collectForegroundOutput = false; } @@ -353,46 +367,65 @@ export class BashTool implements BuiltinTool { return undefined; } - private foregroundCompletionResult( + private async foregroundCompletionResult( taskId: string, proc: IProcess, builder: ToolResultBuilder, foregroundTimeoutMs: number, - ): ExecutableToolResult { + ): Promise { const current = this.tasks.getTask(taskId); const exitCode = current?.kind === 'process' ? current.exitCode : proc.exitCode; + let result: ExecutableToolResultBuilderResult; if (current?.status === 'timed_out') { const timeoutLabel = formatTimeoutLabel(foregroundTimeoutMs); - return builder.error(`Command killed by timeout (${timeoutLabel})`); - } - if (current?.status === 'killed' && current.stopReason === USER_INTERRUPT_REASON) { - return builder.error(USER_INTERRUPT_REASON); - } - if ( + result = builder.error(`Command killed by timeout (${timeoutLabel})`, { + brief: `Killed by timeout (${timeoutLabel})`, + }); + } else if (current?.status === 'killed' && current.stopReason === USER_INTERRUPT_REASON) { + result = builder.error(USER_INTERRUPT_REASON, { brief: USER_INTERRUPT_REASON }); + } else if ( (current?.status === 'failed' || current?.status === 'killed') && current.stopReason !== undefined ) { - return builder.error(current.stopReason); + result = builder.error(current.stopReason, { brief: current.stopReason }); + } else if (exitCode === 0) { + result = builder.ok('Command executed successfully.'); + } else { + if (builder.nChars === 0) builder.write(`Process exited with code ${String(exitCode)}`); + result = builder.error(`Command failed with exit code: ${String(exitCode)}.`, { + brief: `Failed with exit code: ${String(exitCode)}`, + }); } - - const isError = exitCode !== 0; - if (isError && builder.nChars === 0) { - builder.write(`Process exited with code ${String(exitCode)}`); - } - - if (!isError) { - return builder.ok('Command executed successfully.'); - } - return builder.error(`Command failed with exit code: ${String(exitCode)}.`); + return this.addForegroundOutputReference(taskId, result); } - private detachedTaskResult( + private async addForegroundOutputReference( + taskId: string, + result: ExecutableToolResultBuilderResult, + ): Promise { + if (!result.truncated) return result; + const output = await this.tasks.getOutputSnapshot(taskId, 0); + if (!output.fullOutputAvailable || output.outputPath === undefined) return result; + + const taskOutputHint = this.allowBackground() + ? `, or TaskOutput(task_id="${taskId}", block=false)` + : ''; + const reference = + `\n\n[Full output saved]\n` + + `task_id: ${taskId}\n` + + `output_path: ${output.outputPath}\n` + + `output_size_bytes: ${String(output.outputSizeBytes)}\n` + + `next_step: Use Read with output_path to page through the full log${taskOutputHint}.`; + return { ...result, output: `${result.output}${reference}` }; + } + + private backgroundStartedResult( taskId: string, proc: IProcess, description: string, - labels: { title: string }, + labels: { title: string; brief: string }, builder = new ToolResultBuilder(), - scenario: 'detached_started' | 'foreground_detached' = 'detached_started', + scenario: 'background_started' | 'foreground_detached' = 'background_started', ): ExecutableToolResult { const status = this.tasks.getTask(taskId)?.status ?? 'running'; const metadata = @@ -402,23 +435,31 @@ export class BashTool implements BuiltinTool { `status: ${status}\n` + `automatic_notification: true\n` + this.nextStepLines(scenario) + - 'human_shell_hint: Tell the human to run /tasks to open the interactive task panel.'; + 'human_shell_hint: Tell the human to run /tasks to open the interactive background-task panel.'; const foregroundResult = builder.ok(''); const foregroundOutput = foregroundResult.output.length > 0 ? foregroundResult.output : ''; - const message = taskResultMessage(labels.title, foregroundResult.message); - return { + const message = backgroundResultMessage(labels.title, foregroundResult.message); + const result: ExecutableToolResult & { + readonly message: string; + readonly brief: string; + readonly truncated: boolean; + } = { isError: false, output: foregroundOutput.length === 0 ? metadata : `${metadata}\n\nforeground_output:\n${foregroundOutput}`, message, + brief: labels.brief, truncated: foregroundResult.truncated, }; + return result; } - private nextStepLines(scenario: 'detached_started' | 'foreground_detached'): string { + private nextStepLines( + scenario: 'background_started' | 'foreground_detached', + ): string { if (scenario === 'foreground_detached') { // The user explicitly moved a foreground call to the background to avoid // blocking the current turn. Steer the model away from waiting on it. @@ -431,7 +472,9 @@ export class BashTool implements BuiltinTool { `when it completes — ${avoid}; continue with your current work.\n` ); } - // detached_started: the model chose to launch in the background. + // background_started: the model chose to launch in the background. Same anti-wait + // stance — immediately waiting on a background task is just a blocked turn, so do + // not invite a TaskOutput peek here. if (!this.allowBackground()) { return 'next_step: You will be automatically notified when it completes.\n'; } @@ -445,7 +488,7 @@ export class BashTool implements BuiltinTool { registerTool(BashTool); -function taskResultMessage(title: string, suffix: string): string { +function backgroundResultMessage(title: string, suffix: string): string { const normalized = title.endsWith('.') ? title : `${title}.`; if (suffix.length === 0) return normalized; return suffix.endsWith('.') ? `${normalized} ${suffix}` : `${normalized} ${suffix}.`; diff --git a/packages/agent-core-v2/src/os/backends/node-local/tools/process-task.ts b/packages/agent-core-v2/src/os/backends/node-local/tools/process-task.ts index 81134c254..3f2559a8a 100644 --- a/packages/agent-core-v2/src/os/backends/node-local/tools/process-task.ts +++ b/packages/agent-core-v2/src/os/backends/node-local/tools/process-task.ts @@ -145,6 +145,11 @@ function observeProcessStream( const onData = (chunk: string): void => { if (chunk.length === 0) return; sink.appendOutput(chunk); + // Once the manager has begun terminating the task — an output-limit trip + // (see MAX_TASK_OUTPUT_BYTES), a user interrupt, or a timeout — + // `appendOutput` above may synchronously abort the signal. Stop forwarding + // live output from that point so the unbounded forward buffer cannot keep + // growing while the process is being killed. if (sink.signal.aborted) return; onOutput?.(kind, chunk); }; diff --git a/packages/agent-core-v2/test/shellTools/bash.test.ts b/packages/agent-core-v2/test/shellTools/bash.test.ts index 2aecac3ca..a2a1290a3 100644 --- a/packages/agent-core-v2/test/shellTools/bash.test.ts +++ b/packages/agent-core-v2/test/shellTools/bash.test.ts @@ -10,9 +10,6 @@ * semantics match production. * * Deviations from v1: - * - The `brief` result field does not exist on v2's `ExecutableToolResult`, - * so v1 assertions on `result.brief` are dropped; the `output` / `message` - * assertions are kept. * - v1's `execWithEnv(args, env)` is now `runner.exec(args, { env })`, so * spawn-call assertions read `options.env` from the second argument. */ @@ -364,8 +361,10 @@ function errorMessage(error: unknown): string { function createFakeTaskService(options: { maxRunningTasks?: number } = {}): { readonly service: IAgentTaskService; readonly tasks: Map; + readonly persisted: Set; } { const tasks = new Map(); + const persisted = new Set(); let counter = 0; const nextId = (prefix: string): string => { @@ -541,18 +540,20 @@ function createFakeTaskService(options: { maxRunningTasks?: number } = {}): { return result; }, - persistOutput(): void { - /* no-op in the fake */ + persistOutput(taskId: string): void { + persisted.add(taskId); }, async getOutputSnapshot(taskId: string): Promise { const entry = tasks.get(taskId); const preview = entry === undefined ? '' : entry.outputChunks.join(''); + const fullOutputAvailable = persisted.has(taskId); return { + outputPath: fullOutputAvailable ? `/fake/tasks/${taskId}/output.log` : undefined, outputSizeBytes: preview.length, previewBytes: preview.length, truncated: false, - fullOutputAvailable: false, + fullOutputAvailable, preview, }; }, @@ -655,7 +656,7 @@ function createFakeTaskService(options: { maxRunningTasks?: number } = {}): { }, }; - return { service, tasks }; + return { service, tasks, persisted }; } // ── Test execution helper ──────────────────────────────────────────── @@ -860,6 +861,7 @@ describe('BashTool', () => { expect(result).toMatchObject({ isError: true, message: 'Command failed with exit code: 2.', + brief: 'Failed with exit code: 2', }); expect(result.output).toContain('boom\n'); expect(result.output).toContain('Command failed with exit code: 2.'); @@ -889,6 +891,7 @@ describe('BashTool', () => { expect(result).toMatchObject({ isError: true, message: 'Command failed with exit code: 2.', + brief: 'Failed with exit code: 2', }); expect(result.output).toContain('partial\nboom\n'); expect(result.output).toContain('Command failed with exit code: 2.'); @@ -911,6 +914,7 @@ describe('BashTool', () => { expect(result).toMatchObject({ isError: true, message: 'wait failed', + brief: 'wait failed', }); expect(result.output).toContain('partial output\nwait failed'); expect(result.output).not.toContain('exit code: null'); @@ -992,7 +996,7 @@ describe('BashTool', () => { await vi.advanceTimersByTimeAsync(250); const result = await running; - expect(result).toMatchObject({ isError: true }); + expect(result).toMatchObject({ isError: true, brief: 'Killed by timeout (1s)' }); expect(result.output).toContain('Command killed by timeout (1s)'); } finally { vi.useRealTimers(); @@ -1012,7 +1016,7 @@ describe('BashTool', () => { const result = await running; expect(proc.kill).toHaveBeenCalledWith('SIGTERM'); - expect(result).toMatchObject({ isError: true }); + expect(result).toMatchObject({ isError: true, brief: 'Killed by timeout (1s)' }); expect(result.output).toContain('Command killed by timeout (1s)'); expect(result.output).not.toContain('Premature close'); } finally { @@ -1111,6 +1115,40 @@ describe('BashTool', () => { expect(output).toContain('Output is truncated'); }); + it('saves full foreground output when the inline result is truncated', async () => { + const fullOutput = `${'short line\n'.repeat(6_000)}tail survives\n`; + const { runner } = createTestRunner(processWithOutput({ stdout: fullOutput })); + const { service, persisted } = createFakeTaskService(); + const tool = bashTool(runner, createTestEnv(), createTestCtx(), service); + + const result = await executeTool(tool, context({ command: 'flood', timeout: 60 })); + const output = result.output as string; + const taskId = /^task_id: (bash-[0-9a-z]{8})$/m.exec(output)?.[1]; + + expect(output).toContain('[...truncated]'); + expect(output).toContain('[Full output saved]'); + expect(taskId).toBeTruthy(); + // The inline truncation must have started early persistence of the full log. + expect(persisted.has(taskId!)).toBe(true); + expect(output).toContain(`output_path: /fake/tasks/${taskId}/output.log`); + expect(output).toContain('Use Read with output_path'); + expect(output).toContain(`TaskOutput(task_id="${taskId}", block=false)`); + }); + + it('omits the TaskOutput hint from the saved-output reference when background tools are disabled', async () => { + const fullOutput = 'short line\n'.repeat(6_000); + const { runner } = createTestRunner(processWithOutput({ stdout: fullOutput })); + const { service } = createFakeTaskService(); + const tool = bashTool(runner, createTestEnv(), createTestCtx(), service, stubProfile(() => false)); + + const result = await executeTool(tool, context({ command: 'flood', timeout: 60 })); + const output = result.output as string; + + expect(output).toContain('[Full output saved]'); + expect(output).toContain('Use Read with output_path'); + expect(output).not.toContain('TaskOutput'); + }); + it('rejects empty-string commands at the schema layer', () => { expect(BashInputSchema.safeParse({ command: '' }).success).toBe(false); }); @@ -1163,6 +1201,33 @@ describe('BashTool', () => { expect(description).toContain('**Guidelines for efficiency:**'); expect(description).toContain('run_in_background=true'); expect(description).toContain('automatically notified'); + // Moved here from system.md: the "don't block on a background task" nudge belongs in + // the background-enabled Bash description, the only place that documents it. + expect(description).toContain('returning control to the user'); + }); + + it('disables background execution when TaskList is inactive even if TaskOutput/TaskStop are active', async () => { + const { runner, exec } = createTestRunner(processWithOutput()); + const tool = bashTool( + runner, + createTestEnv(), + createTestCtx(), + createFakeTaskService().service, + stubProfile((name) => name !== 'TaskList'), + ); + + // Background management needs TaskList, TaskOutput, and TaskStop; without + // TaskList the description must fall back to the disabled variant. + expect(tool.description).toContain('Background execution is disabled for this agent'); + + const result = await executeTool( + tool, + context({ command: 'sleep 10', run_in_background: true, description: 'watch' }), + ); + + expect(result).toMatchObject({ isError: true }); + expect(result.output).toContain('Background execution is not available'); + expect(exec).not.toHaveBeenCalled(); }); }); @@ -1199,6 +1264,8 @@ describe('BashTool background mode', () => { expect(result.output).toContain(`task_id: ${task.taskId}`); expect(result.output).toContain('automatic_notification: true'); expect(result.output).toContain('do NOT wait, poll, or call TaskOutput'); + expect(result).toMatchObject({ message: 'Task moved to background.' }); + expect((result as { brief?: string }).brief).toBe(`Backgrounded ${task.taskId}`); expect(service.getTask(task.taskId)).toMatchObject({ detached: true }); await vi.waitFor(async () => { await expect(service.readOutput(task.taskId)).resolves.toContain('after detach\n'); @@ -1367,6 +1434,11 @@ describe('BashTool background mode', () => { expect(result.output).toMatch(/task_id: bash-[0-9a-z]{8}/); expect(result.output).toContain('automatic_notification: true'); + expect(result).toMatchObject({ message: 'Background task started.' }); + expect((result as { brief?: string }).brief).toMatch(/^Started bash-[0-9a-z]{8}$/); + // The launch message must steer away from waiting, not invite a TaskOutput peek. + expect(result.output).toContain('do NOT wait, poll, or call TaskOutput on it'); + expect(result.output).not.toContain('block=false'); expect(service.list(false)).toHaveLength(1); }); diff --git a/packages/agent-core-v2/test/task/service.test.ts b/packages/agent-core-v2/test/task/service.test.ts index 333b7580a..d64748f30 100644 --- a/packages/agent-core-v2/test/task/service.test.ts +++ b/packages/agent-core-v2/test/task/service.test.ts @@ -1,11 +1,19 @@ -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { Readable, type Writable } from 'node:stream'; + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { SyncDescriptor } from '#/_base/di/descriptors'; import { DisposableStore, toDisposable } from '#/_base/di/lifecycle'; import { TestInstantiationService } from '#/_base/di/test'; -import { IAgentTaskService, type AgentTask } from '#/agent/task/task'; +import { + IAgentTaskService, + type AgentTask, + type AgentTaskInfo, +} from '#/agent/task/task'; import { renderNotificationXml } from '#/agent/task/notificationXml'; import { AgentTaskService } from '#/agent/task/taskService'; +import { ProcessTask } from '#/os/backends/node-local/tools/process-task'; +import type { IProcess } from '#/session/process/processRunner'; import { IConfigRegistry, IConfigService } from '#/app/config/config'; import { IAgentContextMemoryService } from '#/agent/contextMemory/contextMemory'; import { IAgentPromptService } from '#/agent/prompt/prompt'; @@ -119,6 +127,254 @@ describe('AgentTaskService', () => { expect(await svc.readOutput(id)).toBe(''); await svc.stop(id); }); + + // ── Output ceiling for shell (process) tasks ───────────────────────── + // + // A single shell command that streams more output than the per-command limit + // must be force-terminated instead of growing the (unbounded) live-forward + // buffer or the on-disk write chain until the process runs out of memory or + // fills the disk. The ceiling applies to process tasks, foreground and + // background alike. Subagent and user-question tasks append their bounded + // result in one shot and must always be persisted, so they are not capped. + + const MiB = 1024 * 1024; + const LIMIT_BYTES = 16 * MiB; + + /** + * A process that streams `chunks` of stdout, then exits 0 on its own — unless + * it is killed first, in which case `wait()` resolves with the signal's exit + * code and the stream is destroyed (simulating the child dying on SIGTERM). + */ + function streamingProcess(chunks: string[]): { + proc: IProcess; + kill: ReturnType; + } { + const stdout = Readable.from(chunks); + const stderr = Readable.from([]); + let resolveWait!: (code: number) => void; + const waitP = new Promise((resolve) => { + resolveWait = resolve; + }); + stdout.on('end', () => { + resolveWait(0); + }); + const kill = vi.fn(async (signal: string) => { + stdout.destroy(); + resolveWait(signal === 'SIGKILL' ? 137 : 143); + }); + const proc = { + stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable, + stdout, + stderr, + pid: 4242, + exitCode: null, + wait: () => waitP, + kill, + dispose: vi.fn().mockResolvedValue(undefined), + } as unknown as IProcess; + return { proc, kill }; + } + + /** + * A process that keeps streaming all of `chunks` regardless of SIGTERM (only + * SIGKILL stops it) — simulating a producer that ignores the graceful stop + * and keeps writing through the SIGTERM grace window. + */ + function sigtermIgnoringProcess(chunks: string[]): { + proc: IProcess; + kill: ReturnType; + } { + const stdout = Readable.from(chunks); + const stderr = Readable.from([]); + let resolveWait!: (code: number) => void; + const waitP = new Promise((resolve) => { + resolveWait = resolve; + }); + stdout.on('end', () => { + resolveWait(0); + }); + const kill = vi.fn(async (signal: string) => { + if (signal === 'SIGKILL') { + stdout.destroy(); + resolveWait(137); + } + // SIGTERM is intentionally ignored. + }); + const proc = { + stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable, + stdout, + stderr, + pid: 4243, + exitCode: null, + wait: () => waitP, + kill, + dispose: vi.fn().mockResolvedValue(undefined), + } as unknown as IProcess; + return { proc, kill }; + } + + /** One-shot non-process task appending its full result at once, like a subagent. */ + function agentLikeTask(result: string, description: string): AgentTask { + return { + idPrefix: 'agent', + kind: 'agent', + description, + start: async (sink) => { + sink.appendOutput(result); + await sink.settle({ status: 'completed' }); + }, + toInfo: (base) => ({ ...base, kind: 'agent' }), + }; + } + + async function waitForTerminal( + svc: IAgentTaskService, + taskId: string, + timeoutMs = 30_000, + ): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() <= deadline) { + const info = await svc.wait(taskId, 5); + if ( + info?.status === 'completed' || + info?.status === 'failed' || + info?.status === 'timed_out' || + info?.status === 'killed' || + info?.status === 'lost' + ) { + return info; + } + await new Promise((resolve) => setTimeout(resolve, 1)); + } + return svc.getTask(taskId); + } + + /** Re-stub the byte store so `output.log` appends are counted, then build the service. */ + function serviceWithAppendCounter(): { + svc: IAgentTaskService; + persistedChars: () => number; + } { + let persistedChars = 0; + ix.stub(IFileSystemStorageService, { + read: async () => undefined, + readStream: async function* () {}, + write: async () => {}, + append: async (_scope: string, _key: string, chunk: Uint8Array) => { + persistedChars += chunk.byteLength; + }, + list: async () => [], + delete: async () => {}, + flush: async () => {}, + close: async () => {}, + }); + return { svc: ix.get(IAgentTaskService), persistedChars: () => persistedChars }; + } + + it('terminates a foreground command that exceeds the output limit and stops forwarding', async () => { + const svc = ix.get(IAgentTaskService); + // 20 MiB total, well past the 16 MiB ceiling. + const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB)); + const { proc, kill } = streamingProcess(chunks); + + let forwardedChars = 0; + const onOutput = vi.fn((_kind: 'stdout' | 'stderr', text: string) => { + forwardedChars += text.length; + }); + + const taskId = svc.registerTask( + new ProcessTask(proc, 'b3sum --length 18446744073709551615', 'hash', onOutput), + { detached: false, signal: new AbortController().signal, timeoutMs: 60_000 }, + ); + + const info = await waitForTerminal(svc, taskId); + + expect(info?.status).toBe('killed'); + expect(info?.stopReason ?? '').toMatch(/output limit/i); + expect(kill).toHaveBeenCalledWith('SIGTERM'); + // The live-forward path is capped at the ceiling rather than draining the + // full 20 MiB into the (unbounded) transcript/stderr buffer. + expect(forwardedChars).toBeLessThanOrEqual(LIMIT_BYTES); + }); + + it('also terminates a detached (background) task that exceeds the output limit', async () => { + const svc = ix.get(IAgentTaskService); + const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB)); + const { proc, kill } = streamingProcess(chunks); + + const taskId = svc.registerTask(new ProcessTask(proc, 'producer', 'bg'), { + detached: true, + timeoutMs: 60_000, + }); + + const info = await waitForTerminal(svc, taskId); + + expect(info?.status).toBe('killed'); + expect(info?.stopReason ?? '').toMatch(/output limit/i); + expect(kill).toHaveBeenCalledWith('SIGTERM'); + }); + + it('stops enqueuing output to disk once the foreground cap trips', async () => { + const { svc, persistedChars } = serviceWithAppendCounter(); + + // 20 MiB, and the producer ignores SIGTERM so it keeps writing through + // the whole grace window. + const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB)); + const { proc } = sigtermIgnoringProcess(chunks); + + const taskId = svc.registerTask(new ProcessTask(proc, 'runaway', 'hash', () => {}), { + detached: false, + signal: new AbortController().signal, + timeoutMs: 60_000, + }); + + const info = await waitForTerminal(svc, taskId); + + expect(info?.status).toBe('killed'); + // Before the fix every chunk of the 20 MiB is enqueued into the disk + // write chain (retaining each string until its write drains); afterwards + // enqueuing stops at the ceiling so the chain cannot grow unbounded. + expect(persistedChars()).toBeLessThanOrEqual(17 * MiB); + }); + + it('stops enqueuing output to disk once the cap trips for a background task', async () => { + const { svc, persistedChars } = serviceWithAppendCounter(); + + // 20 MiB, and the producer ignores SIGTERM so it keeps writing through + // the whole grace window. Background tasks share the same ceiling. + const chunks = Array.from({ length: 20 }, () => 'x'.repeat(MiB)); + const { proc } = sigtermIgnoringProcess(chunks); + + const taskId = svc.registerTask(new ProcessTask(proc, 'runaway', 'bg', () => {}), { + detached: true, + timeoutMs: 60_000, + }); + + const info = await waitForTerminal(svc, taskId); + + expect(info?.status).toBe('killed'); + // Same guarantee as the foreground case: once the cap trips, subsequent + // chunks are dropped before they reach the disk write chain. + expect(persistedChars()).toBeLessThanOrEqual(17 * MiB); + }); + + it('does not cap or drop a detached subagent result larger than the limit', async () => { + const { svc, persistedChars } = serviceWithAppendCounter(); + + // 20 MiB result — well past the 16 MiB ceiling — delivered in one shot, + // exactly how a subagent appends its completed result. + const bigResult = 'y'.repeat(20 * MiB); + const taskId = svc.registerTask(agentLikeTask(bigResult, 'big subagent result'), { + detached: true, + timeoutMs: 60_000, + }); + + const info = await waitForTerminal(svc, taskId); + + // Non-process tasks must complete normally and have their full result + // persisted; the shell-output ceiling must not drop it. + expect(info?.status).toBe('completed'); + expect(persistedChars()).toBeGreaterThanOrEqual(bigResult.length); + }); }); describe('Agent task notification XML', () => {