diff --git a/packages/cli/src/lib/__tests__/background-job-manager.test.ts b/packages/cli/src/lib/__tests__/background-job-manager.test.ts index 5db4b24ca5..adb3419d1c 100644 --- a/packages/cli/src/lib/__tests__/background-job-manager.test.ts +++ b/packages/cli/src/lib/__tests__/background-job-manager.test.ts @@ -129,4 +129,37 @@ describe("BackgroundJobManager", () => { await rm(outputDir, { recursive: true, force: true }); } }); + + it("uses monitor ids and emits only stdout as monitor events", async () => { + const outputDir = await mkdtemp(join(tmpdir(), "pochi-monitor-test-")); + try { + const manager = new BackgroundJobManager({ + taskId: "task-test", + outputDir, + }); + const eventPromise = new Promise< + Parameters[0]>[0] + >((resolve) => manager.onDidFinish(resolve)); + const { backgroundJobId, outputFile } = manager.start( + "printf 'stdout-event\\n'; printf 'stderr-output\\n' >&2", + ".", + undefined, + { description: "test monitor" }, + ); + + expect(backgroundJobId).toMatch(/^bgjob-monitor-/); + await eventPromise; + + const batches = manager.drainMonitorEvents(); + expect(batches.flatMap((batch) => batch.lines)).toEqual([ + "stdout-event", + ]); + expect(batches.at(-1)?.ended?.reason).toBe("exited with code 0"); + const output = await readFile(outputFile, "utf8"); + expect(output).toContain("stdout-event"); + expect(output).toContain("stderr-output"); + } finally { + await rm(outputDir, { recursive: true, force: true }); + } + }); }); diff --git a/packages/cli/src/lib/background-job-manager.ts b/packages/cli/src/lib/background-job-manager.ts index 6a8c2f77d7..669a9bdd94 100644 --- a/packages/cli/src/lib/background-job-manager.ts +++ b/packages/cli/src/lib/background-job-manager.ts @@ -1,8 +1,13 @@ import { type ChildProcess, spawn } from "node:child_process"; import { tmpdir } from "node:os"; import path from "node:path"; -import type { BackgroundJobTerminalEvent } from "@getpochi/common"; -import { assertBackgroundJobReadInterval } from "@getpochi/common"; +import { + type BackgroundJobTerminalEvent, + type MonitorEventBatch, + MonitorRateLimitedReason, + MonitorWatcher, + assertBackgroundJobReadInterval, +} from "@getpochi/common"; import { getTerminalEnv } from "@getpochi/common/env-utils"; import { BackgroundJobOutputFile, @@ -24,6 +29,17 @@ export interface BackgroundJob { lastReadAt?: number; stopRequested?: boolean; finalizing?: boolean; + monitor?: { + description: string; + watcher: MonitorWatcher; + timedOut?: boolean; + rateLimited?: boolean; + }; +} + +export interface MonitorJobOptions { + description: string; + timeoutMs?: number; } export interface BackgroundJobStartResult { @@ -39,9 +55,10 @@ export interface BackgroundJobManagerOptions { type FinishListener = (event: BackgroundJobTerminalEvent) => void; export class BackgroundJobManager { - private jobs: Map = new Map(); - private maxOutputSize = 1024 * 1024; // compatibility buffer only + private jobs = new Map(); + private maxOutputSize = 1024 * 1024; private readonly finishListeners = new Set(); + private pendingMonitorEvents: MonitorEventBatch[] = []; constructor(private readonly options: BackgroundJobManagerOptions = {}) {} @@ -49,18 +66,17 @@ export class BackgroundJobManager { command: string, cwd: string, envs?: Record, + monitor?: MonitorJobOptions, ): BackgroundJobStartResult { - const id = createBackgroundJobId("command"); + const id = createBackgroundJobId(monitor ? "monitor" : "command"); const outputFile = this.options.outputDir ? path.join(this.options.outputDir, `${id}.log`) : this.options.taskId ? getBackgroundJobOutputPath(this.options.taskId, id) : path.join(tmpdir(), "pochi-background-jobs", `${id}.log`); const outputWriter = new BackgroundJobOutputFile(outputFile); - - const shell = getShellPath(); const child = spawn(command, { - shell, + shell: getShellPath(), cwd, env: { ...process.env, ...getTerminalEnv(), ...envs }, stdio: ["ignore", "pipe", "pipe"], @@ -76,39 +92,68 @@ export class BackgroundJobManager { startTime: Date.now(), status: "running", }; - this.jobs.set(id, job); + if (monitor) { + const watcher = new MonitorWatcher({ + onEvents: (lines) => { + this.pendingMonitorEvents.push({ + backgroundJobId: id, + description: monitor.description, + lines, + }); + }, + onTimeout: () => { + if (job.monitor) job.monitor.timedOut = true; + this.kill(id); + }, + onRateLimitExceeded: () => { + if (job.monitor) job.monitor.rateLimited = true; + this.kill(id); + }, + timeoutMs: monitor.timeoutMs, + }); + job.monitor = { description: monitor.description, watcher }; + } + const appendOutput = async (chunk: string) => { if (chunk.length === 0) return; await outputWriter.append(chunk); if (job.output.length + chunk.length > this.maxOutputSize) { const keep = this.maxOutputSize - chunk.length; - if (keep > 0) { - job.output = job.output.slice(-keep) + chunk; - } else { - job.output = chunk.slice(-this.maxOutputSize); - } + job.output = + keep > 0 + ? job.output.slice(-keep) + chunk + : chunk.slice(-this.maxOutputSize); } else { job.output += chunk; } }; let outputError: unknown; - const outputFinished = Promise.all( - [child.stdout, child.stderr] - .filter((stream) => stream !== null) - .map(async (stream) => { - const sanitizer = new PlainOutputSanitizer(); - // setEncoding uses Node's streaming decoder, so a multi-byte UTF-8 - // character split between Buffer chunks is not replaced with U+FFFD. - stream.setEncoding("utf8"); - for await (const chunk of stream) { - await appendOutput(sanitizer.write(chunk)); - } - await appendOutput(sanitizer.end()); - }), - ).catch((error) => { + const processStream = async ( + stream: NodeJS.ReadableStream, + isStdout: boolean, + ) => { + const sanitizer = new PlainOutputSanitizer(); + stream.setEncoding("utf8"); + for await (const chunk of stream) { + const plainText = sanitizer.write(chunk as string); + await appendOutput(plainText); + if (isStdout && plainText.length > 0) { + job.monitor?.watcher.ingest(plainText); + } + } + const remainder = sanitizer.end(); + await appendOutput(remainder); + if (isStdout && remainder.length > 0) { + job.monitor?.watcher.ingest(remainder); + } + }; + const outputFinished = Promise.all([ + ...(child.stdout ? [processStream(child.stdout, true)] : []), + ...(child.stderr ? [processStream(child.stderr, false)] : []), + ]).catch((error) => { outputError = error; child.kill(); }); @@ -167,6 +212,16 @@ export class BackgroundJobManager { job.status = finalStatus; job.finalizing = false; + let monitorReason = + finalError ?? + (exitCode === undefined ? finalStatus : `exited with code ${exitCode}`); + if (job.monitor?.rateLimited) { + monitorReason = MonitorRateLimitedReason; + } else if (job.monitor?.timedOut) { + monitorReason = "killed after timeout"; + } + this.endMonitor(job, monitorReason); + if (!this.options.taskId) return; const event: BackgroundJobTerminalEvent = { taskId: this.options.taskId, @@ -181,6 +236,36 @@ export class BackgroundJobManager { for (const listener of this.finishListeners) listener(event); } + private endMonitor(job: BackgroundJob, reason: string): void { + if (!job.monitor) return; + job.monitor.watcher.end(); + this.pendingMonitorEvents.push({ + backgroundJobId: job.id, + description: job.monitor.description, + lines: [], + ended: { reason }, + }); + job.monitor = undefined; + } + + drainMonitorEvents(): MonitorEventBatch[] { + const events = this.pendingMonitorEvents; + this.pendingMonitorEvents = []; + return events; + } + + hasPendingMonitorEvents(): boolean { + return this.pendingMonitorEvents.length > 0; + } + + getActiveMonitors(): Array<{ backgroundJobId: string; description: string }> { + return Array.from(this.jobs.values()).flatMap((job) => + job.status === "running" && job.monitor + ? [{ backgroundJobId: job.id, description: job.monitor.description }] + : [], + ); + } + readOutput(id: string): { output: string; status: "running" | "completed" | "failed" | "stopped" | "idle"; @@ -194,30 +279,31 @@ export class BackgroundJobManager { previousReadAt: job.lastReadAt, status: job.status === "running" ? "running" : "completed", }); - - const outputToReturn = job.output; + const output = job.output; job.output = ""; job.lastReadAt = now; - - return { output: outputToReturn, status: job.status }; + return { output, status: job.status }; } kill(id: string): boolean { const job = this.jobs.get(id); if (!job) return false; if (job.status !== "running" || job.finalizing) return true; - job.stopRequested = true; - return job.process.kill(); + job.process.kill(); + return true; } - killAll() { + killAll(): void { for (const job of this.jobs.values()) { if (job.status === "running" && !job.finalizing) { job.stopRequested = true; job.process.kill(); } + job.monitor?.watcher.dispose(); + job.monitor = undefined; } + this.pendingMonitorEvents = []; } hasPendingJobs(): boolean { @@ -235,17 +321,21 @@ export class BackgroundJobManager { async waitForAllJobs( timeoutMs: number, abortSignal?: AbortSignal, - ): Promise<"completed" | "timeout" | "aborted"> { + wakeOnMonitorEvents = false, + ): Promise<"completed" | "timeout" | "aborted" | "monitor-events"> { const startTime = Date.now(); - const pollInterval = 50; - while (this.hasPendingJobs()) { + if (wakeOnMonitorEvents && this.hasPendingMonitorEvents()) { + return "monitor-events"; + } if (abortSignal?.aborted) return "aborted"; if (timeoutMs > 0 && Date.now() - startTime >= timeoutMs) return "timeout"; - await new Promise((resolve) => setTimeout(resolve, pollInterval)); + await new Promise((resolve) => setTimeout(resolve, 50)); + } + if (wakeOnMonitorEvents && this.hasPendingMonitorEvents()) { + return "monitor-events"; } - return "completed"; } } diff --git a/packages/cli/src/task-runner.ts b/packages/cli/src/task-runner.ts index 115d66471a..5274ebfa86 100644 --- a/packages/cli/src/task-runner.ts +++ b/packages/cli/src/task-runner.ts @@ -329,14 +329,33 @@ export class TaskRunner { getLLM: () => options.llm, getEffectiveContextWindow: () => pochiConfig.value.effectiveContextWindow, - getEnvironment: async () => ({ - ...(await readEnvironment({ + getEnvironment: async () => { + const environment = await readEnvironment({ cwd: options.cwd, omitCustomRules: options.isSubTask && options.customAgent?.omitAgentsMd === true, - })), - todos: this.todos, - }), + }); + const monitors = this.backgroundJobManager.getActiveMonitors(); + return { + ...environment, + workspace: { + ...environment.workspace, + ...(monitors.length > 0 + ? { + terminals: monitors.map( + ({ backgroundJobId, description }) => ({ + name: description, + isActive: false, + backgroundJobId, + monitor: description, + }), + ), + } + : {}), + }, + todos: this.todos, + }; + }, getCustomAgents: () => this.toolCallOptions.customAgents || [], getSkills: () => this.toolCallOptions.skills || [], ...(options.getAutoMemory @@ -422,23 +441,25 @@ export class TaskRunner { } /** - * Wait for all background jobs to complete. - * Respects the configured asyncWaitTimeoutInMs and abort signal. - * @returns One structured notification message emitted after output files - * flush. All notifications available at this drain point are included as - * separate data parts in the same user message. + * Wait for all background jobs to complete, waking early when monitor + * events arrive. Respects the configured timeout and abort signal. */ - private async waitForAsyncWork(): Promise { + private async waitForAsyncWork(): Promise< + "notifications" | "monitor-events" | undefined + > { const spinner = createSpinner( `Waiting for background jobs to complete (timeout: ${this.asyncWaitTimeoutInMs}ms)...`, ).start(); - const jobStatus = await this.backgroundJobManager.waitForAllJobs( this.asyncWaitTimeoutInMs, this.abortSignal, + true, ); - // Handle timeout or abort - return undefined to finish without feeding back to LLM + if (jobStatus === "monitor-events") { + spinner.succeed("Monitor events arrived."); + return "monitor-events"; + } if (jobStatus === "timeout") { const remainingJobs = this.backgroundJobManager.getPendingJobIds(); spinner.fail( @@ -446,17 +467,18 @@ export class TaskRunner { ); this.backgroundJobManager.killAll(); await this.backgroundJobManager.waitForAllJobs(5000, this.abortSignal); - return this.takePendingBackgroundJobNotifications(); + return this.pendingBackgroundJobNotifications.length > 0 + ? "notifications" + : undefined; } - if (jobStatus === "aborted") { spinner.fail("Async work wait was aborted."); return undefined; } - spinner.succeed("All background jobs completed."); - - return this.takePendingBackgroundJobNotifications(); + return this.pendingBackgroundJobNotifications.length > 0 + ? "notifications" + : undefined; } private takePendingBackgroundJobNotifications(): Message | undefined { @@ -466,6 +488,17 @@ export class TaskRunner { : undefined; } + private injectPendingMonitorEvents(): boolean { + const batches = this.backgroundJobManager.drainMonitorEvents(); + if (batches.length === 0) return false; + this.chat.appendOrReplaceMessage({ + id: crypto.randomUUID(), + role: "user", + parts: [{ type: "data-monitor-events", data: { batches } }], + }); + return true; + } + /** * @returns * - "finished" if the task is finished and no more steps are needed. @@ -481,14 +514,26 @@ export class TaskRunner { const result = await this.process(lastMessage); if (result === "finished") { + // Monitor events captured during the last round are fed back before + // the task is allowed to complete. + if (this.injectPendingMonitorEvents()) { + return "next"; + } + // Check for pending background jobs const hasPendingJobs = this.backgroundJobManager.hasPendingJobs(); if (this.asyncWaitTimeoutInMs > 0 && hasPendingJobs) { - const notificationMessage = await this.waitForAsyncWork(); - if (notificationMessage) { - this.chat.appendOrReplaceMessage(notificationMessage); - return "next"; + const asyncResult = await this.waitForAsyncWork(); + if (asyncResult === "monitor-events") { + if (this.injectPendingMonitorEvents()) return "next"; + } else if (asyncResult === "notifications") { + const notificationMessage = + this.takePendingBackgroundJobNotifications(); + if (notificationMessage) { + this.chat.appendOrReplaceMessage(notificationMessage); + return "next"; + } } } @@ -533,6 +578,10 @@ export class TaskRunner { if (result === "next") { this.stepCount.throwIfReachedMaxSteps(); + // Deliver monitor events between rounds so the model sees them in + // the upcoming inference. Retry rounds are skipped: they resend a + // prepared message and an interleaved user message would break that. + this.injectPendingMonitorEvents(); } if (result === "retry") { this.stepCount.throwIfReachedMaxRetries(); diff --git a/packages/cli/src/tools/index.ts b/packages/cli/src/tools/index.ts index b934baa1e9..dc58bfdf5c 100644 --- a/packages/cli/src/tools/index.ts +++ b/packages/cli/src/tools/index.ts @@ -9,6 +9,7 @@ import { ExecuteCommandError, executeCommand } from "./execute-command"; import { globFiles } from "./glob-files"; import { killBackgroundJob } from "./kill-background-job"; import { listFiles } from "./list-files"; +import { startMonitor } from "./monitor"; import { newTask } from "./new-task"; import { readFile } from "./read-file"; @@ -32,6 +33,7 @@ const ToolMap: Record< searchFiles, executeCommand, killBackgroundJob, + startMonitor, useSkill, }; diff --git a/packages/cli/src/tools/monitor.ts b/packages/cli/src/tools/monitor.ts new file mode 100644 index 0000000000..444918f4d7 --- /dev/null +++ b/packages/cli/src/tools/monitor.ts @@ -0,0 +1,41 @@ +import * as path from "node:path"; +import { MonitorDefaultTimeoutMs } from "@getpochi/common"; +import type { ClientTools, ToolFunctionType } from "@getpochi/tools"; +import type { ToolCallOptions } from "../types"; + +export const startMonitor = + (context: ToolCallOptions): ToolFunctionType => + async ( + { command, description, cwd = ".", timeoutMs, persistent }, + { cwd: workspaceDir, envs }, + ) => { + const { backgroundJobManager } = context; + if (!backgroundJobManager) { + throw new Error("Background job manager not available."); + } + + if (!command) { + throw new Error("Command is required to execute."); + } + + let resolvedCwd: string; + if (path.isAbsolute(cwd)) { + resolvedCwd = path.normalize(cwd); + } else { + resolvedCwd = path.normalize(path.join(workspaceDir, cwd)); + } + + const { backgroundJobId } = backgroundJobManager.start( + command, + resolvedCwd, + envs, + { + description, + timeoutMs: persistent + ? undefined + : (timeoutMs ?? MonitorDefaultTimeoutMs), + }, + ); + + return { backgroundJobId }; + }; diff --git a/packages/common/src/base/environment.ts b/packages/common/src/base/environment.ts index 76b3e1acdc..c61bb94c6a 100644 --- a/packages/common/src/base/environment.ts +++ b/packages/common/src/base/environment.ts @@ -59,7 +59,7 @@ export const Environment = z.object({ .string() .optional() .describe( - 'A stable terminal id. "bgjob-cmd-" identifies a managed command job; "term-" identifies a read-only user terminal.', + 'A stable terminal id. "bgjob-cmd-" identifies a managed command job, "bgjob-monitor-" identifies a monitor, and "term-" identifies a read-only user terminal.', ), outputFile: z .string() @@ -67,6 +67,12 @@ export const Environment = z.object({ .describe( "Absolute path to the terminal transcript. Read it with readFile using offset/limit.", ), + monitor: z + .string() + .optional() + .describe( + "Present when this terminal is an active monitor; the value is the monitor description.", + ), }), ) .optional() diff --git a/packages/common/src/base/formatters.ts b/packages/common/src/base/formatters.ts index a088a557b3..b85485a79a 100644 --- a/packages/common/src/base/formatters.ts +++ b/packages/common/src/base/formatters.ts @@ -103,6 +103,7 @@ function removeSystemReminder(messages: UIMessage[]): UIMessage[] { x.type === "data-reviews" || x.type === "data-bash-outputs" || x.type === "data-background-job-notification" || + x.type === "data-monitor-events" || isStaticToolUIPart(x), ) ) { diff --git a/packages/common/src/base/index.ts b/packages/common/src/base/index.ts index 6afa8bfba4..06910eaa5c 100644 --- a/packages/common/src/base/index.ts +++ b/packages/common/src/base/index.ts @@ -13,6 +13,19 @@ export { parseEnvironmentInfo, } from "./prompts"; +export { + MonitorWatcher, + type MonitorWatcherOptions, + type MonitorEventBatch, + type MonitorEventEnvelope, + formatMonitorNotifications, + MonitorBatchIntervalMs, + MonitorDefaultTimeoutMs, + MonitorMaxLinesPerBatch, + MonitorMaxBatchesPerMinute, + MonitorRateLimitedReason, +} from "./monitor"; + export { SocialLinks } from "./social"; export * as constants from "./constants"; diff --git a/packages/common/src/base/monitor/__tests__/monitor-watcher.test.ts b/packages/common/src/base/monitor/__tests__/monitor-watcher.test.ts new file mode 100644 index 0000000000..5907272284 --- /dev/null +++ b/packages/common/src/base/monitor/__tests__/monitor-watcher.test.ts @@ -0,0 +1,210 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { + MonitorMaxBatchesPerMinute, + MonitorMaxLinesPerBatch, + MonitorWatcher, + formatMonitorNotifications, +} from ".."; + +describe("MonitorWatcher", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("batches lines arriving within the batch interval", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + watcher.ingest("line 1\nline 2\n"); + watcher.ingest("line 3\n"); + expect(onEvents).not.toHaveBeenCalled(); + + vi.advanceTimersByTime(200); + expect(onEvents).toHaveBeenCalledTimes(1); + expect(onEvents).toHaveBeenCalledWith(["line 1", "line 2", "line 3"]); + }); + + it("delivers separate batches for lines beyond the interval", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + watcher.ingest("first\n"); + vi.advanceTimersByTime(200); + watcher.ingest("second\n"); + vi.advanceTimersByTime(200); + + expect(onEvents).toHaveBeenNthCalledWith(1, ["first"]); + expect(onEvents).toHaveBeenNthCalledWith(2, ["second"]); + }); + + it("buffers partial lines across chunks", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + watcher.ingest("hel"); + watcher.ingest("lo\n"); + vi.advanceTimersByTime(200); + + expect(onEvents).toHaveBeenCalledWith(["hello"]); + }); + + it("strips ANSI escape sequences", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + watcher.ingest("\x1b[31mERROR\x1b[0m something\n"); + watcher.ingest("\x1b]633;C\x07visible\n"); + vi.advanceTimersByTime(200); + + expect(onEvents).toHaveBeenCalledWith(["ERROR something", "visible"]); + }); + + it("skips blank lines", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + watcher.ingest("\n\n \na\n\n"); + vi.advanceTimersByTime(200); + + expect(onEvents).toHaveBeenCalledWith(["a"]); + }); + + it("treats lone carriage returns as line breaks", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + watcher.ingest("progress 10%\rprogress 20%\n"); + vi.advanceTimersByTime(200); + + expect(onEvents).toHaveBeenCalledWith(["progress 10%", "progress 20%"]); + }); + + it("caps lines per batch and reports the omission", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + const lines = Array.from( + { length: MonitorMaxLinesPerBatch + 10 }, + (_, i) => `line ${i}`, + ); + watcher.ingest(`${lines.join("\n")}\n`); + vi.advanceTimersByTime(200); + + const delivered = onEvents.mock.calls[0][0] as string[]; + expect(delivered).toHaveLength(MonitorMaxLinesPerBatch + 1); + expect(delivered.at(-1)).toContain("10 more monitor events omitted"); + expect(delivered.at(-1)).toContain("narrow the monitor command's output filter"); + }); + + it("flushes buffered content synchronously on end", () => { + const onEvents = vi.fn(); + const watcher = new MonitorWatcher({ onEvents }); + + watcher.ingest("complete line\nno trailing newline"); + watcher.end(); + + expect(onEvents).toHaveBeenCalledWith([ + "complete line", + "no trailing newline", + ]); + + // No duplicate flush from the pending timer, no ingestion after end. + watcher.ingest("late\n"); + vi.advanceTimersByTime(200); + expect(onEvents).toHaveBeenCalledTimes(1); + }); + + it("fires onTimeout after the deadline", () => { + const onEvents = vi.fn(); + const onTimeout = vi.fn(); + new MonitorWatcher({ onEvents, onTimeout, timeoutMs: 1000 }); + + vi.advanceTimersByTime(999); + expect(onTimeout).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); + expect(onTimeout).toHaveBeenCalledTimes(1); + }); + + it("stops ingesting and fires onRateLimitExceeded when batches flood", () => { + const onEvents = vi.fn(); + const onRateLimitExceeded = vi.fn(); + const watcher = new MonitorWatcher({ onEvents, onRateLimitExceeded }); + + // One batch every 2 seconds: exceeds the per-minute cap on batch N+1. + for (let i = 0; i <= MonitorMaxBatchesPerMinute; i++) { + watcher.ingest(`line ${i}\n`); + vi.advanceTimersByTime(2000); + } + + expect(onRateLimitExceeded).toHaveBeenCalledTimes(1); + expect(onEvents).toHaveBeenCalledTimes(MonitorMaxBatchesPerMinute + 1); + + // Rate-limited: further chunks are ignored. + watcher.ingest("late\n"); + vi.advanceTimersByTime(2000); + expect(onEvents).toHaveBeenCalledTimes(MonitorMaxBatchesPerMinute + 1); + }); + + it("does not rate limit slow event streams", () => { + const onEvents = vi.fn(); + const onRateLimitExceeded = vi.fn(); + const watcher = new MonitorWatcher({ onEvents, onRateLimitExceeded }); + + // One batch every 10 seconds stays under the cap indefinitely. + for (let i = 0; i < MonitorMaxBatchesPerMinute * 3; i++) { + watcher.ingest(`line ${i}\n`); + vi.advanceTimersByTime(10_000); + } + + expect(onRateLimitExceeded).not.toHaveBeenCalled(); + expect(onEvents).toHaveBeenCalledTimes(MonitorMaxBatchesPerMinute * 3); + }); + + it("does not set a timeout when timeoutMs is undefined", () => { + const onEvents = vi.fn(); + const onTimeout = vi.fn(); + new MonitorWatcher({ onEvents, onTimeout }); + + vi.advanceTimersByTime(3_600_000); + expect(onTimeout).not.toHaveBeenCalled(); + }); +}); + +describe("formatMonitorNotifications", () => { + it("wraps batches in a system reminder", () => { + const text = formatMonitorNotifications([ + { + backgroundJobId: "bgjob-1", + description: "errors in dev.log", + lines: ["ERROR boom"], + }, + ]); + + expect(text.startsWith("")).toBe(true); + expect(text.endsWith("")).toBe(true); + expect(text).toContain("bgjob-1"); + expect(text).toContain('"errors in dev.log"'); + expect(text).toContain("ERROR boom"); + expect(text).toContain("not user input"); + }); + + it("merges multiple batches into one reminder", () => { + const text = formatMonitorNotifications([ + { backgroundJobId: "bgjob-1", description: "a", lines: ["x"] }, + { + backgroundJobId: "bgjob-2", + description: "b", + lines: [], + ended: { reason: "exited with code 0" }, + }, + ]); + + expect(text.match(//g)).toHaveLength(1); + expect(text).toContain("bgjob-1"); + expect(text).toContain("[monitor ended: exited with code 0]"); + }); +}); diff --git a/packages/common/src/base/monitor/index.ts b/packages/common/src/base/monitor/index.ts new file mode 100644 index 0000000000..ee4011e4e6 --- /dev/null +++ b/packages/common/src/base/monitor/index.ts @@ -0,0 +1,211 @@ +/** + * Host-agnostic event extraction layer for the startMonitor tool. + * + * A MonitorWatcher taps the raw output chunks of a background job + * (VSCode TerminalJob / CLI BackgroundJobManager), turns them into + * line events, and batches them before delivery: + * + * chunk -> strip ANSI -> partial-line buffer -> split lines + * -> batch (BatchIntervalMs) -> onEvents(lines) + */ + +import { prompts } from "../prompts"; + +/** Lines arriving within this window are delivered as one batch. */ +export const MonitorBatchIntervalMs = 200; + +/** Default watch deadline when `persistent` is not set. */ +export const MonitorDefaultTimeoutMs = 300_000; + +/** Hard cap of lines per delivered batch; the rest is summarized. */ +export const MonitorMaxLinesPerBatch = 50; + +/** + * A monitor delivering more batches than this within a rolling minute is + * stopped automatically: each batch becomes a conversation message, so a + * noisy monitor floods the context. The model is told to restart with a + * stricter filter. + */ +export const MonitorMaxBatchesPerMinute = 10; + +/** Ended reason used when a monitor is stopped for exceeding the rate limit. */ +export const MonitorRateLimitedReason = `stopped automatically: more than ${MonitorMaxBatchesPerMinute} event batches per minute. Restart the monitor with a stricter output filter that emits only the lines you would act on.`; + +/** + * A single delivery of monitor events, ready to be injected into the + * conversation between inference rounds. + */ +export interface MonitorEventBatch { + backgroundJobId: string; + description: string; + lines: string[]; + /** + * Present when the watch ended (job exit, timeout, kill). A batch with + * `ended` may still carry final lines flushed from the buffer. + */ + ended?: { reason: string }; +} + +/** + * A MonitorEventBatch with a monotonically increasing sequence number, + * used by the VSCode host <-> webview delivery channel so the webview can + * acknowledge consumed batches across reloads. + */ +export interface MonitorEventEnvelope extends MonitorEventBatch { + seq: number; +} + +export interface MonitorWatcherOptions { + /** Deliver a batch of event lines. Never called with an empty array. */ + onEvents: (lines: string[]) => void; + /** + * Called when `timeoutMs` elapses. The host is expected to kill the + * underlying job, which in turn triggers `end()`. + */ + onTimeout?: () => void; + /** + * Called once when the batch rate exceeds MonitorMaxBatchesPerMinute. + * The host is expected to kill the underlying job; the watcher stops + * ingesting further chunks on its own. + */ + onRateLimitExceeded?: () => void; + /** Watch deadline. `undefined` means no timeout (persistent monitor). */ + timeoutMs?: number; + batchIntervalMs?: number; +} + +// CSI sequences (colors, cursor movement) and OSC sequences (titles, +// hyperlinks) emitted by shells with terminal integration. +const AnsiEscapePattern = + // biome-ignore lint/suspicious/noControlCharactersInRegex: matching terminal escape sequences requires control chars + /\x1b\[[0-9;?]*[0-9A-Za-z]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)/g; + +function stripAnsi(text: string): string { + return text.replace(AnsiEscapePattern, ""); +} + +export class MonitorWatcher { + private partialLine = ""; + private pendingLines: string[] = []; + private droppedLines = 0; + private flushTimer: ReturnType | undefined; + private timeoutTimer: ReturnType | undefined; + private ended = false; + private rateLimited = false; + private flushTimestamps: number[] = []; + + constructor(private readonly options: MonitorWatcherOptions) { + if (options.timeoutMs !== undefined && options.onTimeout) { + this.timeoutTimer = setTimeout(() => { + this.options.onTimeout?.(); + }, options.timeoutMs); + } + } + + /** Feed a raw output chunk. Chunks may split lines at any position. */ + ingest(chunk: string): void { + if (this.ended || this.rateLimited) return; + + const text = this.partialLine + stripAnsi(chunk); + // Lone \r is treated as a line break so progress-bar style rewrites + // don't accumulate into one endless partial line. + const segments = text.split(/\r\n|\n|\r/); + this.partialLine = segments.pop() ?? ""; + + for (const line of segments) { + if (line.trim().length === 0) continue; + if (this.pendingLines.length >= MonitorMaxLinesPerBatch) { + this.droppedLines++; + continue; + } + this.pendingLines.push(line); + } + + if (this.pendingLines.length > 0 && this.flushTimer === undefined) { + this.flushTimer = setTimeout(() => { + this.flushTimer = undefined; + this.flush(); + }, this.options.batchIntervalMs ?? MonitorBatchIntervalMs); + } + } + + /** + * The watch ended (job exit, kill, or timeout enforcement). Flushes any + * buffered lines synchronously. Idempotent. + */ + end(): void { + if (this.ended) return; + this.ended = true; + + if (this.partialLine.trim().length > 0) { + if (this.pendingLines.length < MonitorMaxLinesPerBatch) { + this.pendingLines.push(this.partialLine); + } else { + this.droppedLines++; + } + } + this.partialLine = ""; + this.flush(); + this.dispose(); + } + + dispose(): void { + if (this.flushTimer !== undefined) { + clearTimeout(this.flushTimer); + this.flushTimer = undefined; + } + if (this.timeoutTimer !== undefined) { + clearTimeout(this.timeoutTimer); + this.timeoutTimer = undefined; + } + } + + private flush(): void { + if (this.pendingLines.length === 0) return; + const lines = this.pendingLines; + if (this.droppedLines > 0) { + lines.push( + `[${this.droppedLines} more monitor events omitted; narrow the monitor command's output filter]`, + ); + } + this.pendingLines = []; + this.droppedLines = 0; + this.options.onEvents(lines); + this.checkRateLimit(); + } + + private checkRateLimit(): void { + if (this.ended || this.rateLimited) return; + const now = Date.now(); + this.flushTimestamps.push(now); + this.flushTimestamps = this.flushTimestamps.filter((t) => t > now - 60_000); + if (this.flushTimestamps.length > MonitorMaxBatchesPerMinute) { + this.rateLimited = true; + this.options.onRateLimitExceeded?.(); + } + } +} + +function renderMonitorEventBatch(batch: MonitorEventBatch): string { + const header = `Monitor "${batch.description}" (backgroundJobId: ${batch.backgroundJobId}):`; + const lines = [...batch.lines]; + if (batch.ended) { + lines.push(`[monitor ended: ${batch.ended.reason}]`); + } + return `${header}\n${lines.join("\n")}`; +} + +/** + * Renders one delivery of monitor event batches as the system-reminder user + * message injected into the conversation. Shared by the CLI task runner and + * the VSCode webview so both hosts speak the same protocol. System reminders + * are kept on the LLM path but hidden from the chat UI. + */ +export function formatMonitorNotifications( + batches: MonitorEventBatch[], +): string { + const body = batches.map(renderMonitorEventBatch).join("\n\n"); + return prompts.createSystemReminder( + `The following events were captured by background monitors started with startMonitor. This is an automated notification, not user input:\n${body}`, + ); +} diff --git a/packages/common/src/base/prompts/__tests__/__snapshots__/prompt.test.ts.snap b/packages/common/src/base/prompts/__tests__/__snapshots__/prompt.test.ts.snap index be06d9b0cb..ed32586edb 100644 --- a/packages/common/src/base/prompts/__tests__/__snapshots__/prompt.test.ts.snap +++ b/packages/common/src/base/prompts/__tests__/__snapshots__/prompt.test.ts.snap @@ -21,7 +21,9 @@ package.json # Opened Terminals in Editor Read terminal output from its output file with \`readFile\` (use \`offset\` and \`limit\` for growing files). -- Ids prefixed with "bgjob-cmd-" are Pochi-started background jobs and can be killed with \`killBackgroundJob\`. +- Ids prefixed with "bgjob-cmd-" are Pochi-started command jobs. +- Ids prefixed with "bgjob-monitor-" are active monitors. +- Both managed job types can be killed with \`killBackgroundJob\`. - Ids prefixed with "term-" are user-opened terminals and are read-only. - A terminal without an output file has no output available to read. * Terminal 1 (selected) diff --git a/packages/common/src/base/prompts/environment.ts b/packages/common/src/base/prompts/environment.ts index 8354eba162..286359b9c5 100644 --- a/packages/common/src/base/prompts/environment.ts +++ b/packages/common/src/base/prompts/environment.ts @@ -156,11 +156,11 @@ function getVisibleTerminals(workspace: Environment["workspace"]) { return ""; } const header = - '# Opened Terminals in Editor\nRead terminal output from its output file with `readFile` (use `offset` and `limit` for growing files).\n- Ids prefixed with "bgjob-cmd-" are Pochi-started background jobs and can be killed with `killBackgroundJob`.\n- Ids prefixed with "term-" are user-opened terminals and are read-only.\n- A terminal without an output file has no output available to read.'; + '# Opened Terminals in Editor\nRead terminal output from its output file with `readFile` (use `offset` and `limit` for growing files).\n- Ids prefixed with "bgjob-cmd-" are Pochi-started command jobs.\n- Ids prefixed with "bgjob-monitor-" are active monitors.\n- Both managed job types can be killed with `killBackgroundJob`.\n- Ids prefixed with "term-" are user-opened terminals and are read-only.\n- A terminal without an output file has no output available to read.'; return `${header}\n${terminals .map( (t) => - `${t.isActive ? "* " : " "}${t.name}${t.isActive ? " (selected)" : ""}${t.backgroundJobId ? ` (id: ${t.backgroundJobId})` : ""}${t.outputFile ? ` (output: ${t.outputFile})` : ""}`, + `${t.isActive ? "* " : " "}${t.name}${t.isActive ? " (selected)" : ""}${t.backgroundJobId ? ` (id: ${t.backgroundJobId})` : ""}${t.outputFile ? ` (output: ${t.outputFile})` : ""}${t.monitor ? ` (monitoring: ${t.monitor})` : ""}`, ) .join("\n")}`; } diff --git a/packages/common/src/vscode-webui-bridge/webview-stub.ts b/packages/common/src/vscode-webui-bridge/webview-stub.ts index 74b0293954..d01a8199ac 100644 --- a/packages/common/src/vscode-webui-bridge/webview-stub.ts +++ b/packages/common/src/vscode-webui-bridge/webview-stub.ts @@ -9,6 +9,7 @@ import type { BackgroundTaskState, ContextWindowUsage, Environment, + MonitorEventEnvelope, TaskMemoryState, } from "../base"; import type { BrowserSession } from "../browser/types"; @@ -283,6 +284,19 @@ const VSCodeHostStub = { }, }); }, + readMonitorEvents: async ( + _taskId: string, + ): Promise> => { + return Promise.resolve( + {} as ThreadSignalSerialization, + ); + }, + ackMonitorEvents: async ( + _taskId: string, + _upToSeq: number, + ): Promise => { + return Promise.resolve(); + }, readModelList: async () => { return Promise.resolve( {} as { diff --git a/packages/common/src/vscode-webui-bridge/webview.ts b/packages/common/src/vscode-webui-bridge/webview.ts index 56f29d6ecc..b790151ff0 100644 --- a/packages/common/src/vscode-webui-bridge/webview.ts +++ b/packages/common/src/vscode-webui-bridge/webview.ts @@ -9,6 +9,7 @@ import type { BackgroundTaskState, ContextWindowUsage, Environment, + MonitorEventEnvelope, TaskMemoryState, TerminalTextSelection, } from "../base"; @@ -193,6 +194,18 @@ export interface VSCodeHostApi { acknowledge: (notificationId: string) => Promise; }>; + /** + * Undelivered monitor event batches of a task (startMonitor tool), as a + * live signal. The webview injects them into the conversation and then + * acknowledges via {@link ackMonitorEvents}. + */ + readMonitorEvents( + taskId: string, + ): Promise>; + + /** Drops delivered monitor event batches with seq <= upToSeq. */ + ackMonitorEvents(taskId: string, upToSeq: number): Promise; + /** * Opens a file at the specified file path. * diff --git a/packages/livekit/src/chat/flexible-chat-transport.ts b/packages/livekit/src/chat/flexible-chat-transport.ts index d416cbf946..8f9c4ad663 100644 --- a/packages/livekit/src/chat/flexible-chat-transport.ts +++ b/packages/livekit/src/chat/flexible-chat-transport.ts @@ -7,7 +7,11 @@ import type { PochiProviderOptions, PochiRequestUseCase, } from "@getpochi/common"; -import { formatters, prompts } from "@getpochi/common"; +import { + formatMonitorNotifications, + formatters, + prompts, +} from "@getpochi/common"; import { hasActiveTodos } from "@getpochi/common/message-utils"; import * as R from "remeda"; @@ -545,6 +549,12 @@ export function convertDataPartToText( text: prompts.renderBackgroundJobNotification(part.data), }; } + if (part.type === "data-monitor-events") { + return { + type: "text" as const, + text: formatMonitorNotifications(part.data.batches), + }; + } return part; } diff --git a/packages/livekit/src/types.ts b/packages/livekit/src/types.ts index 3ab9cb8ff7..e2bb906748 100644 --- a/packages/livekit/src/types.ts +++ b/packages/livekit/src/types.ts @@ -4,6 +4,7 @@ import type { BackgroundJobNotification, BashOutputs, MessageMetadata, + MonitorEventBatch, Review, TerminalTextSelection, UserEdits, @@ -37,6 +38,9 @@ export type DataParts = { bashOutputs: BashOutputs; }; "background-job-notification": BackgroundJobNotification; + "monitor-events": { + batches: MonitorEventBatch[]; + }; }; /** diff --git a/packages/tools/src/__test__/select-agent-tools.test.ts b/packages/tools/src/__test__/select-agent-tools.test.ts index b5f6aeb5bb..fd9c80c595 100644 --- a/packages/tools/src/__test__/select-agent-tools.test.ts +++ b/packages/tools/src/__test__/select-agent-tools.test.ts @@ -16,6 +16,7 @@ const ClientToolNames = [ "readFile", "searchFiles", "renderWidget", + "startMonitor", "useSkill", "writeToFile", ].sort(); diff --git a/packages/tools/src/constants.ts b/packages/tools/src/constants.ts index 634bf67e15..1e6c75bb2c 100644 --- a/packages/tools/src/constants.ts +++ b/packages/tools/src/constants.ts @@ -75,7 +75,12 @@ export const ToolsByPermission = { "webSearch", ] as string[], write: ["writeToFile", "applyDiff", "editNotebook"] as string[], - execute: ["executeCommand", "killBackgroundJob", "newTask"] as string[], + execute: [ + "executeCommand", + "killBackgroundJob", + "startMonitor", + "newTask", + ] as string[], default: ["renderWidget"] as string[], }; diff --git a/packages/tools/src/index.ts b/packages/tools/src/index.ts index e4ef612878..6213db0c2a 100644 --- a/packages/tools/src/index.ts +++ b/packages/tools/src/index.ts @@ -50,6 +50,7 @@ export type { export { QuestionSchema } from "./ask-followup-question"; import { editNotebook } from "./edit-notebook"; import { killBackgroundJob } from "./kill-background-job"; +import { startMonitor } from "./monitor"; import { createReadFileTool } from "./read-file"; import { type Skill, createSkillTool } from "./use-skill"; import { parseToolSpec } from "./utils/tool-spec"; @@ -170,6 +171,7 @@ export const createClientTools = (options?: CreateClientToolOptions) => { return { ...createCliTools(options), killBackgroundJob, + startMonitor, renderWidget, }; }; diff --git a/packages/tools/src/kill-background-job.ts b/packages/tools/src/kill-background-job.ts index 214e7e0f4f..c8c1b4dfe4 100644 --- a/packages/tools/src/kill-background-job.ts +++ b/packages/tools/src/kill-background-job.ts @@ -5,7 +5,8 @@ const toolDef = { description: `- Kills a running background job by its ID - Takes a backgroundJobId parameter identifying the job to kill - Returns a success or failure status -- Use this tool when you need to terminate a long-running background job`.trim(), +- Use this tool when you need to terminate a long-running background job +- Also stops a monitor started with startMonitor (monitors are background jobs)`.trim(), inputSchema: z.object({ backgroundJobId: z .string() diff --git a/packages/tools/src/monitor.ts b/packages/tools/src/monitor.ts new file mode 100644 index 0000000000..950ae72ea5 --- /dev/null +++ b/packages/tools/src/monitor.ts @@ -0,0 +1,68 @@ +import { z } from "zod"; +import { defineClientTool } from "./types"; + +const toolDef = { + description: + `Start a background monitor that streams events from a long-running command. Each stdout line is an event delivered proactively between steps, so you can keep working while notifications arrive. Events arrive on their own schedule and are automated notifications, not replies from the user, even if one arrives while you are waiting for the user to answer. + +Use a monitor for repeated occurrences or a sequence of results: +- Ongoing occurrences ("tell me every time an ERROR appears"): use an unbounded command such as \`tail -f\`, \`fswatch\`, or a polling loop. +- Occurrences with a known end ("report each CI check until the run completes"): emit each result and exit after the terminal state. +- A command that exits after one event is valid, but do not use an unbounded command when only one notification is needed. It remains armed after the event until timeout or cancellation. + +Your command's stdout is the event stream. Each line becomes an event; lines produced within 200ms may be delivered as one batch. Command exit ends the monitor and its exit status is reported. + +Examples: +- Each matching log line is an event: \`tail -f app.log | grep -E --line-buffered "ERROR|FAILED|Killed|OOM"\` +- Each file change is an event: \`fswatch /watched/dir\` +- Poll a PR, emit one line for each newly completed check, and exit when all checks reach a terminal state. + +Script quality: +- Every pipe stage must flush per line or matches may remain buffered: grep needs \`--line-buffered\`; awk needs \`fflush()\`. Avoid \`head\`, which can delay output until enough matches accumulate. +- Only stdout is the event stream. Stderr is captured but does not trigger events. Merge with \`2>&1\` when failures written to stderr should reach your filter. +- In polling loops, tolerate transient request failures and use suitable intervals: 30s or more for remote APIs, 0.5-1s for local checks. +- Write a specific description because it appears in every notification. + +Coverage and volume: +- Silence is not success. When watching for an outcome, emit every terminal state you would act on, including failure, cancellation, timeout, crashes, and the expected success state. +- Filter selectively to actionable signals; never stream raw logs. Excessive event volume causes the monitor to be stopped, in which case restart it with a tighter filter. + +After starting a monitor, continue with other work. If there is nothing else to do, yield the current turn without calling attemptCompletion and wait for the event notification to begin the next turn; do not keep the turn active by repeatedly checking process output. Keep the task open until you have handled the event the user requested. + +The default timeout is 5 minutes. Use persistent only for an explicitly requested session-length watch. Use killBackgroundJob to stop the monitor early.`.trim(), + inputSchema: z.object({ + command: z + .string() + .describe( + "Shell command or script. Each stdout line is an event; exit ends the monitor.", + ), + description: z + .string() + .describe( + 'Short, specific human-readable description of what is being monitored, shown with every event notification (e.g. "errors in dev.log").', + ), + cwd: z + .string() + .optional() + .describe("The working directory to execute the command in."), + timeoutMs: z + .number() + .min(1_000) + .max(3_600_000) + .optional() + .describe( + "Kill the monitor after this deadline. Default 300000ms (5 minutes), minimum 1000ms, maximum 3600000ms. Ignored when persistent is true.", + ), + persistent: z + .boolean() + .optional() + .describe( + "Run for the lifetime of the task with no timeout. Use only for explicitly requested session-length watches such as PR monitoring or log tails; stop with killBackgroundJob.", + ), + }), + outputSchema: z.object({ + backgroundJobId: z.string().optional(), + }), +}; + +export const startMonitor = defineClientTool(toolDef); diff --git a/packages/tools/src/utils/tool-batch.ts b/packages/tools/src/utils/tool-batch.ts index e7fd77f6a6..b43f512939 100644 --- a/packages/tools/src/utils/tool-batch.ts +++ b/packages/tools/src/utils/tool-batch.ts @@ -88,6 +88,8 @@ export function isSafeToBatchToolCall( if (isReadonlyToolCall(toolName, input)) return true; + if (toolName === "startMonitor") return true; + return false; } diff --git a/packages/vscode-webui/src/components/message/message-list.tsx b/packages/vscode-webui/src/components/message/message-list.tsx index 52a34dc567..f25d1901e7 100644 --- a/packages/vscode-webui/src/components/message/message-list.tsx +++ b/packages/vscode-webui/src/components/message/message-list.tsx @@ -31,6 +31,7 @@ import { BackgroundJobNotifications } from "./background-job-notifications"; import { MessageMarkdown } from "./markdown"; import type { MermaidContext } from "./mermaid-context"; import { MermaidContextProvider } from "./mermaid-context"; +import { MonitorEventsPart } from "./monitor-events"; import { Reviews } from "./reviews"; import { UserEditsPart } from "./user-edits"; @@ -415,6 +416,10 @@ function Part({ return null; } + if (part.type === "data-monitor-events") { + return ; + } + if (part.type === "data-terminal-context") { return null; } diff --git a/packages/vscode-webui/src/components/message/monitor-events.tsx b/packages/vscode-webui/src/components/message/monitor-events.tsx new file mode 100644 index 0000000000..e80c8c7ec4 --- /dev/null +++ b/packages/vscode-webui/src/components/message/monitor-events.tsx @@ -0,0 +1,49 @@ +import { CollapsibleSection } from "@/components/ui/collapsible-section"; +import type { MonitorEventBatch } from "@getpochi/common"; +import { Activity } from "lucide-react"; +import { useTranslation } from "react-i18next"; + +/** + * Visible record of monitor events (startMonitor tool) delivered to the + * model. The LLM receives the same content as a system-reminder text; this + * component keeps the event visible in the chat history. + */ +export const MonitorEventsPart: React.FC<{ + batches: MonitorEventBatch[]; +}> = ({ batches }) => { + const { t } = useTranslation(); + + if (batches.length === 0) return null; + const description = batches[0].description; + const lines = batches.flatMap((batch) => { + const rendered = [...batch.lines]; + if (batch.ended) { + rendered.push(`[monitor ended: ${batch.ended.reason}]`); + } + return rendered; + }); + + return ( + + + {description} + + } + actions={ + + {t("monitorEvents.eventCount", { count: lines.length })} + + } + > +
+ {lines.map((line, i) => ( +
+ {line} +
+ ))} +
+ + ); +}; diff --git a/packages/vscode-webui/src/features/chat/components/chat-toolbar.test.tsx b/packages/vscode-webui/src/features/chat/components/chat-toolbar.test.tsx index 9e89d62411..a9e0546613 100644 --- a/packages/vscode-webui/src/features/chat/components/chat-toolbar.test.tsx +++ b/packages/vscode-webui/src/features/chat/components/chat-toolbar.test.tsx @@ -128,6 +128,7 @@ vi.mock("@/lib/use-default-store", () => ({ })); vi.mock("@/lib/vscode", () => ({ vscodeHost: {}, + isVSCodeEnvironment: () => false, })); vi.mock("../hooks/use-chat-input-state", () => ({ useChatInputState: () => ({ diff --git a/packages/vscode-webui/src/features/chat/components/chat-toolbar.tsx b/packages/vscode-webui/src/features/chat/components/chat-toolbar.tsx index e12f4c241d..ebf0f5a1ed 100644 --- a/packages/vscode-webui/src/features/chat/components/chat-toolbar.tsx +++ b/packages/vscode-webui/src/features/chat/components/chat-toolbar.tsx @@ -28,7 +28,11 @@ import { useTaskChangedFiles } from "@/lib/hooks/use-task-changed-files"; import { useUserEdits } from "@/lib/hooks/use-user-edits"; import { cn, tw } from "@/lib/utils"; import type { UseChatHelpers } from "@ai-sdk/react"; -import { constants, type BackgroundJobNotification } from "@getpochi/common"; +import { + constants, + type BackgroundJobNotification, + type MonitorEventEnvelope, +} from "@getpochi/common"; import { hasActiveTodos } from "@getpochi/common/message-utils"; import type { DisplayModel, @@ -53,6 +57,7 @@ import { useChatInputState } from "../hooks/use-chat-input-state"; import { useChatStatus } from "../hooks/use-chat-status"; import { type DraftMessage, useChatSubmit } from "../hooks/use-chat-submit"; import { useInlineCompactTask } from "../hooks/use-inline-compact-task"; +import { useMonitorEvents } from "../hooks/use-monitor-events"; import { useNewCompactTask } from "../hooks/use-new-compact-task"; import { useShowCompleteSubtaskButton } from "../hooks/use-subtask-completed"; import type { SubtaskInfo } from "../hooks/use-subtask-info"; @@ -167,6 +172,50 @@ export const ChatToolbar: React.FC = ({ ); } }, [acknowledge, backgroundJobNotifications, messages]); + + // Monitor events (startMonitor tool) enter the conversation through the + // queued-messages pipeline: enqueue here, and the auto-dequeue effect + // below delivers them as soon as the chat is idle. Events arriving while + // a monitor draft is still queued are merged into it, so a burst becomes + // one message (and one inference round) instead of many. + const onMonitorEvents = useCallback((envelopes: MonitorEventEnvelope[]) => { + setQueuedMessages((prev) => { + const first = envelopes[0]; + if (!first) return prev; + + const last = prev.at(-1); + const queuedMonitor = last?.raw.monitor; + const queuedEnvelopes = + queuedMonitor?.backgroundJobId === first.backgroundJobId + ? queuedMonitor.envelopes + : undefined; + const merged = queuedEnvelopes + ? [...queuedEnvelopes, ...envelopes] + : envelopes; + const eventCount = merged.reduce((n, e) => n + e.lines.length, 0); + const summary = [ + eventCount > 0 ? `${eventCount} event(s)` : "", + merged.some((e) => e.ended) ? "ended" : "", + ] + .filter(Boolean) + .join(" · "); + + const draft: DraftMessage = { + parts: [{ type: "data-monitor-events", data: { batches: merged } }], + raw: { + text: `Monitor [${first.description}]: ${summary}`, + nonRemovable: true, + monitor: { + backgroundJobId: first.backgroundJobId, + description: first.description, + envelopes: merged, + }, + }, + }; + return queuedEnvelopes ? [...prev.slice(0, -1), draft] : [...prev, draft]; + }); + }, []); + useMonitorEvents(taskId, onMonitorEvents); const [excludedUserEditsContext, setExcludedUserEditsContext] = useState(); const lastCheckpointHash = task?.lastCheckpointHash ?? undefined; diff --git a/packages/vscode-webui/src/features/chat/components/queued-messages.tsx b/packages/vscode-webui/src/features/chat/components/queued-messages.tsx index befedd3041..d7924e4661 100644 --- a/packages/vscode-webui/src/features/chat/components/queued-messages.tsx +++ b/packages/vscode-webui/src/features/chat/components/queued-messages.tsx @@ -15,6 +15,7 @@ import { isVSCodeEnvironment, vscodeHost } from "@/lib/vscode"; import { parseTitle } from "@getpochi/common/message-utils"; import type { ActiveSelection } from "@getpochi/common/vscode-webui-bridge"; import { + Activity, CornerDownRight, FileCode, ListEnd, @@ -36,6 +37,7 @@ interface RenderMessage { title: string; details: string; isTodoMode?: boolean; + isMonitor?: boolean; activeSelection?: ActiveSelection; nonRemovable?: boolean; } @@ -56,6 +58,7 @@ export const QueuedMessages: React.FC = ({ userEditsCount = 0, terminalContextCount = 0, isTodoMode, + monitor, activeSelection, } = raw; const title = text.trim() ? parseTitle(text) : t("chat.noMessage"); @@ -74,6 +77,7 @@ export const QueuedMessages: React.FC = ({ title, details: details.join(" · "), isTodoMode, + isMonitor: !!monitor, activeSelection, nonRemovable: raw.nonRemovable, }; @@ -89,6 +93,8 @@ export const QueuedMessages: React.FC = ({ > {message.isTodoMode ? ( + ) : message.isMonitor ? ( + ) : ( )} diff --git a/packages/vscode-webui/src/features/chat/hooks/use-chat-submit.ts b/packages/vscode-webui/src/features/chat/hooks/use-chat-submit.ts index 2015c78b84..a76d2877fc 100644 --- a/packages/vscode-webui/src/features/chat/hooks/use-chat-submit.ts +++ b/packages/vscode-webui/src/features/chat/hooks/use-chat-submit.ts @@ -4,6 +4,7 @@ import { prepareMessageParts } from "@/lib/message-utils"; import { vscodeHost } from "@/lib/vscode"; import type { UseChatHelpers } from "@ai-sdk/react"; import { getLogger } from "@getpochi/common"; +import type { MonitorEventEnvelope } from "@getpochi/common"; import type { Message } from "@getpochi/livekit"; import { useActiveSelection } from "@/lib/hooks/use-active-selection"; @@ -43,6 +44,12 @@ export interface DraftMessage { activeSelection?: ActiveSelection; backgroundJobNotificationIds?: string[]; nonRemovable?: boolean; + /** Present when this draft was generated from monitor events. */ + monitor?: { + backgroundJobId: string; + description: string; + envelopes: MonitorEventEnvelope[]; + }; }; } diff --git a/packages/vscode-webui/src/features/chat/hooks/use-monitor-events.ts b/packages/vscode-webui/src/features/chat/hooks/use-monitor-events.ts new file mode 100644 index 0000000000..7f5707efd1 --- /dev/null +++ b/packages/vscode-webui/src/features/chat/hooks/use-monitor-events.ts @@ -0,0 +1,52 @@ +import { isVSCodeEnvironment, vscodeHost } from "@/lib/vscode"; +import { getLogger } from "@getpochi/common"; +import type { MonitorEventEnvelope } from "@getpochi/common"; +import { effect } from "@preact/signals-core"; +import { threadSignal } from "@quilted/threads/signals"; +import { useEffect, useRef } from "react"; + +const logger = getLogger("UseMonitorEvents"); + +/** + * Subscribes to the host's undelivered monitor event batches for a task + * (startMonitor tool). Fresh batches are handed to `onEvents` exactly once + * and acknowledged to the host so a webview reload doesn't redeliver them. + */ +export function useMonitorEvents( + taskId: string, + onEvents: (envelopes: MonitorEventEnvelope[]) => void, +) { + const onEventsRef = useRef(onEvents); + onEventsRef.current = onEvents; + + // Guards against redelivery while an ack roundtrip is in flight. + const lastSeqRef = useRef(0); + + useEffect(() => { + if (!isVSCodeEnvironment()) return; + + let disposed = false; + let disposeEffect: (() => void) | undefined; + + (async () => { + const serialized = await vscodeHost.readMonitorEvents(taskId); + if (disposed) return; + const events = threadSignal(serialized); + disposeEffect = effect(() => { + const envelopes = events.value; + const fresh = envelopes.filter((e) => e.seq > lastSeqRef.current); + if (fresh.length === 0) return; + lastSeqRef.current = Math.max(...fresh.map((e) => e.seq)); + onEventsRef.current(fresh); + vscodeHost + .ackMonitorEvents(taskId, lastSeqRef.current) + .catch((e) => logger.warn("Failed to ack monitor events", e)); + }); + })().catch((e) => logger.warn("Failed to read monitor events", e)); + + return () => { + disposed = true; + disposeEffect?.(); + }; + }, [taskId]); +} diff --git a/packages/vscode-webui/src/features/chat/lib/use-background-job-display.tsx b/packages/vscode-webui/src/features/chat/lib/use-background-job-display.tsx index 5d604436b1..f335632820 100644 --- a/packages/vscode-webui/src/features/chat/lib/use-background-job-display.tsx +++ b/packages/vscode-webui/src/features/chat/lib/use-background-job-display.tsx @@ -19,6 +19,12 @@ export const useBackgroundJobDisplay = (messages: Message[]) => { p.output?._meta?.backgroundJobId ) { ids.add(p.output._meta.backgroundJobId); + } else if ( + p.type === "tool-startMonitor" && + p.state !== "input-streaming" && + p.output?.backgroundJobId + ) { + ids.add(p.output.backgroundJobId); } } return Array.from(ids).toString(); @@ -40,6 +46,16 @@ export const useBackgroundJobDisplay = (messages: Message[]) => { displayId: `%${map.size + 1}`, command: p.input.command, }); + } else if ( + p.type === "tool-startMonitor" && + p.state !== "input-streaming" && + p.input?.command && + p.output?.backgroundJobId + ) { + map.set(p.output.backgroundJobId, { + displayId: `%${map.size + 1}`, + command: p.input.command, + }); } } diff --git a/packages/vscode-webui/src/features/settings/components/sections/tools-section.tsx b/packages/vscode-webui/src/features/settings/components/sections/tools-section.tsx index 3975d2c34f..2a08eb4dcc 100644 --- a/packages/vscode-webui/src/features/settings/components/sections/tools-section.tsx +++ b/packages/vscode-webui/src/features/settings/components/sections/tools-section.tsx @@ -53,6 +53,8 @@ const ToolDescriptions: Record = { "The renderWidget tool renders a local HTML/SVG widget in the VSCode chat. It is used for streaming diagrams, mockups, simple charts, art, and local interactive UI. Widgets store JSON state on a top-level element; they cannot use external APIs or load external resources.", killBackgroundJob: "When a background process is no longer needed or is causing issues, Pochi uses the killBackgroundJob tool to terminate it cleanly. This is important for resource management and ensuring that processes don't continue running unnecessarily after their purpose has been fulfilled.\n\nPochi might use this tool to stop development servers after completing work, cancel long-running builds, or terminate processes that are consuming too many resources. It gives Pochi full control over the lifecycle of background processes, ensuring efficient and clean task completion.", + startMonitor: + "The startMonitor tool lets Pochi watch something in the background and react when it changes, without pausing the conversation. Pochi runs a command whose output lines become events - for example tailing a log file for errors, polling a CI job for status changes, or watching a directory for file changes.\n\nMonitor events and end status are delivered proactively between steps, so Pochi can continue working and react when something happens. The monitor can be stopped with killBackgroundJob when it is no longer needed.", newTask: "The newTask tool allows Pochi to create a new task with a dedicated agent. This is useful for tasks that require a dedicated agent with specific capabilities or configurations. By using this tool, Pochi can tailor the agent's behavior to better suit the needs of the task at hand.", editNotebook: diff --git a/packages/vscode-webui/src/features/tools/components/command-execution-panel.tsx b/packages/vscode-webui/src/features/tools/components/command-execution-panel.tsx index a842a3cdac..27ee3eaf6b 100644 --- a/packages/vscode-webui/src/features/tools/components/command-execution-panel.tsx +++ b/packages/vscode-webui/src/features/tools/components/command-execution-panel.tsx @@ -232,11 +232,11 @@ export const BackgroundJobPanel: FC<{ backgroundJobId: string; output?: string; appearance?: "default" | "notification"; - /** Command fallback for persisted notification messages. */ command?: string; status?: "completed" | "failed" | "stopped"; exitCode?: number; outputFile?: string; + icon?: React.ReactNode; /** Terminal name snapshot from the tool output (term- ids only). */ terminalName?: string; /** Last command run in the terminal, from the tool output (term- ids only). */ @@ -249,6 +249,7 @@ export const BackgroundJobPanel: FC<{ status, exitCode, outputFile, + icon, terminalName, lastCommand, }) => { @@ -301,8 +302,9 @@ export const BackgroundJobPanel: FC<{ return ( + {icon} {isNotification && status && ( >> = { startBackgroundJob: StartBackgroundJobTool, readBackgroundJobOutput: ReadBackgroundJobOutputTool, killBackgroundJob: KillBackgroundJobTool, + startMonitor: StartMonitorTool, searchFiles: searchFilesTool, listFiles: listFilesTool, globFiles: globFilesTool, diff --git a/packages/vscode-webui/src/features/tools/components/monitor.tsx b/packages/vscode-webui/src/features/tools/components/monitor.tsx new file mode 100644 index 0000000000..dce77e8141 --- /dev/null +++ b/packages/vscode-webui/src/features/tools/components/monitor.tsx @@ -0,0 +1,56 @@ +import { Activity } from "lucide-react"; +import { useTranslation } from "react-i18next"; +import { + BackgroundJobPanel, + CommandPanelContainer, + CopyCommandButton, +} from "./command-execution-panel"; +import { HighlightedText } from "./highlight-text"; +import { StatusIcon } from "./status-icon"; +import { ExpandableToolContainer } from "./tool-container"; +import type { ToolProps } from "./types"; + +export const StartMonitorTool: React.FC> = ({ + tool, + isExecuting, +}) => { + const { t } = useTranslation(); + + const { command, description } = + tool.state === "input-available" || tool.state === "output-available" + ? tool.input + : { command: undefined, description: undefined }; + + const backgroundJobId = + tool.state === "output-available" ? tool.output.backgroundJobId : undefined; + + const title = ( + <> + + + {t("toolInvocation.startMonitor")}{" "} + {description} + + + ); + + return ( + } + /> + ) : command ? ( + } + title={command} + actions={} + /> + ) : null + } + /> + ); +}; diff --git a/packages/vscode-webui/src/features/tools/components/tool-call-lite.tsx b/packages/vscode-webui/src/features/tools/components/tool-call-lite.tsx index 6f0b7e2880..9601f8ccef 100644 --- a/packages/vscode-webui/src/features/tools/components/tool-call-lite.tsx +++ b/packages/vscode-webui/src/features/tools/components/tool-call-lite.tsx @@ -59,6 +59,9 @@ export function ToolCallLite({ case "tool-killBackgroundJob": detail = ; break; + case "tool-startMonitor": + detail = ; + break; case "tool-searchFiles": detail = ; break; @@ -196,6 +199,19 @@ const KillBackgroundJobTool = () => { ); }; +const StartMonitorLiteTool = ({ + tool, +}: ToolCallLiteViewProps<"startMonitor">) => { + const { t } = useTranslation(); + const { description } = tool.input || {}; + return ( + + {t("toolInvocation.monitoring")}{" "} + {description} + + ); +}; + const SearchFilesTool = ({ tool }: ToolCallLiteViewProps<"searchFiles">) => { const { t } = useTranslation(); const { path, regex, filePattern } = tool.input || {}; diff --git a/packages/vscode-webui/src/i18n/locales/en.json b/packages/vscode-webui/src/i18n/locales/en.json index bc3ccb2787..2fd5a3bff3 100644 --- a/packages/vscode-webui/src/i18n/locales/en.json +++ b/packages/vscode-webui/src/i18n/locales/en.json @@ -377,7 +377,9 @@ "next": "Next", "askingQuestion_one": "Asking Question", "askingQuestion_other": "Asking Questions", - "skipped": "Skipped" + "skipped": "Skipped", + "startMonitor": "I will monitor", + "monitoring": "Monitoring" }, "reasoning": { "thinking": "Thinking ...", @@ -570,6 +572,10 @@ "viewChanges": "View Changes", "openFile": "Open File" }, + "monitorEvents": { + "eventCount_one": "{{count}} event", + "eventCount_other": "{{count}} events" + }, "userEdits": { "title": "Edits", "filesEdited_one": "{{count}} file edited", diff --git a/packages/vscode-webui/src/i18n/locales/jp.json b/packages/vscode-webui/src/i18n/locales/jp.json index 81ecb317bf..35434233a8 100644 --- a/packages/vscode-webui/src/i18n/locales/jp.json +++ b/packages/vscode-webui/src/i18n/locales/jp.json @@ -371,7 +371,9 @@ "next": "次へ", "askingQuestion_one": "質問中", "askingQuestion_other": "質問中", - "skipped": "スキップ済み" + "skipped": "スキップ済み", + "startMonitor": "監視します：", + "monitoring": "監視中" }, "reasoning": { "thinking": "考え中 ...", @@ -569,6 +571,10 @@ "viewChanges": "変更を表示", "openFile": "ファイルを開く" }, + "monitorEvents": { + "eventCount_one": "{{count}} 件のイベント", + "eventCount_other": "{{count}} 件のイベント" + }, "userEdits": { "title": "編集", "filesEdited_one": "{{count}} ファイルを編集", diff --git a/packages/vscode-webui/src/i18n/locales/ko.json b/packages/vscode-webui/src/i18n/locales/ko.json index eec6f21e79..e08c6a79c4 100644 --- a/packages/vscode-webui/src/i18n/locales/ko.json +++ b/packages/vscode-webui/src/i18n/locales/ko.json @@ -369,7 +369,9 @@ "next": "다음", "askingQuestion_one": "질문 중", "askingQuestion_other": "질문 중", - "skipped": "건너뜀" + "skipped": "건너뜀", + "startMonitor": "모니터링합니다:", + "monitoring": "모니터링 중" }, "reasoning": { "thinking": "생각 중 ...", @@ -562,6 +564,10 @@ "viewChanges": "변경 사항 보기", "openFile": "파일 열기" }, + "monitorEvents": { + "eventCount_one": "이벤트 {{count}}건", + "eventCount_other": "이벤트 {{count}}건" + }, "userEdits": { "title": "편집", "filesEdited_one": "{{count}}개 파일 편집됨", diff --git a/packages/vscode-webui/src/i18n/locales/zh.json b/packages/vscode-webui/src/i18n/locales/zh.json index 248e756a82..55473c7a12 100644 --- a/packages/vscode-webui/src/i18n/locales/zh.json +++ b/packages/vscode-webui/src/i18n/locales/zh.json @@ -369,7 +369,9 @@ "next": "下一步", "askingQuestion_one": "提问中", "askingQuestion_other": "提问中", - "skipped": "已跳过" + "skipped": "已跳过", + "startMonitor": "我将监听：", + "monitoring": "正在监听" }, "reasoning": { "thinking": "思考中 ...", @@ -567,6 +569,10 @@ "viewChanges": "查看更改", "openFile": "打开文件" }, + "monitorEvents": { + "eventCount_one": "{{count}} 个事件", + "eventCount_other": "{{count}} 个事件" + }, "userEdits": { "title": "编辑", "filesEdited_one": "已编辑 {{count}} 个文件", diff --git a/packages/vscode-webui/src/lib/vscode.ts b/packages/vscode-webui/src/lib/vscode.ts index aeeaaa49cb..951ecf982c 100644 --- a/packages/vscode-webui/src/lib/vscode.ts +++ b/packages/vscode-webui/src/lib/vscode.ts @@ -131,6 +131,8 @@ function createVSCodeHost(): VSCodeHostApi { "showInformationMessage", "showWarningMessage", "readVisibleTerminals", + "readMonitorEvents", + "ackMonitorEvents", "readModelList", "readUserStorage", "readCustomAgents", diff --git a/packages/vscode/src/integrations/monitor/monitor-registry.ts b/packages/vscode/src/integrations/monitor/monitor-registry.ts new file mode 100644 index 0000000000..446c0815c8 --- /dev/null +++ b/packages/vscode/src/integrations/monitor/monitor-registry.ts @@ -0,0 +1,147 @@ +import { getLogger } from "@/lib/logger"; +import { + type MonitorEventEnvelope, + MonitorRateLimitedReason, + MonitorWatcher, +} from "@getpochi/common"; +import { type Signal, signal } from "@preact/signals-core"; +import type { TerminalJobMonitorHooks } from "../terminal/terminal-job"; +import { TerminalJob } from "../terminal/terminal-job"; + +const logger = getLogger("MonitorRegistry"); + +export interface CreateMonitorOptions { + taskId: string; + description: string; + /** Watch deadline. `undefined` means no timeout (persistent monitor). */ + timeoutMs?: number; +} + +export interface MonitorHandle { + /** Hooks to pass into TerminalJob.create as `monitor`. */ + hooks: TerminalJobMonitorHooks; + /** + * Associates the monitor with its background job id. Must be called right + * after TerminalJob.create; no chunk can arrive before that because the + * job first awaits shell integration. + */ + attach(backgroundJobId: string): void; +} + +/** + * Per-task registry of undelivered monitor event batches. + * + * Module-level state like TerminalJob/OutputManager's static registries, + * because a VSCodeHostImpl exists per webview (sidebar and panels) while + * monitors are process-wide. + */ +const taskSignals = new Map>(); +/** Active (not yet ended) monitors: backgroundJobId -> description. */ +const activeMonitors = new Map(); +const changeListeners = new Set<() => void>(); +let eventSeq = 0; + +function emitChange(): void { + for (const listener of changeListeners) listener(); +} + +/** The undelivered event batches of a task, as a live signal. */ +function monitorEvents(taskId: string): Signal { + let events = taskSignals.get(taskId); + if (!events) { + events = signal([]); + taskSignals.set(taskId, events); + } + return events; +} + +/** Drops delivered batches with seq <= upToSeq. */ +function ackMonitorEvents(taskId: string, upToSeq: number): void { + const events = taskSignals.get(taskId); + if (!events) return; + events.value = events.value.filter((e) => e.seq > upToSeq); +} + +function deleteMonitorEvents(taskId: string): void { + taskSignals.delete(taskId); +} + +function createMonitor(options: CreateMonitorOptions): MonitorHandle { + const { taskId, description, timeoutMs } = options; + let backgroundJobId: string | undefined; + let ended = false; + // The kill initiated by timeout / rate limiting surfaces in TerminalJob as + // a generic "terminal closed" error; this override preserves the real cause. + let endReasonOverride: string | undefined; + + const push = (envelope: Omit) => { + const events = monitorEvents(taskId); + events.value = [...events.value, { ...envelope, seq: ++eventSeq }]; + }; + + const kill = () => { + if (backgroundJobId) { + TerminalJob.get(backgroundJobId)?.kill(); + } + }; + + const watcher = new MonitorWatcher({ + onEvents: (lines) => { + push({ + backgroundJobId: backgroundJobId ?? "unknown", + description, + lines, + }); + }, + onTimeout: () => { + logger.debug(`Monitor timeout, killing job ${backgroundJobId}`); + endReasonOverride = "killed after timeout"; + kill(); + }, + onRateLimitExceeded: () => { + logger.debug(`Monitor rate limited, killing job ${backgroundJobId}`); + endReasonOverride = MonitorRateLimitedReason; + kill(); + }, + timeoutMs, + }); + + return { + hooks: { + ingest: (chunk) => watcher.ingest(chunk), + end: (reason) => { + if (ended) return; + ended = true; + if (backgroundJobId) { + activeMonitors.delete(backgroundJobId); + emitChange(); + } + watcher.end(); + push({ + backgroundJobId: backgroundJobId ?? "unknown", + description, + lines: [], + ended: { reason: endReasonOverride ?? reason }, + }); + }, + }, + attach: (id) => { + backgroundJobId = id; + activeMonitors.set(id, description); + emitChange(); + }, + }; +} + +export const MonitorRegistry = { + events: monitorEvents, + ack: ackMonitorEvents, + delete: deleteMonitorEvents, + createMonitor, + onDidChange: (listener: () => void): { dispose(): void } => { + changeListeners.add(listener); + return { dispose: () => changeListeners.delete(listener) }; + }, + descriptionFor: (backgroundJobId: string): string | undefined => + activeMonitors.get(backgroundJobId), +}; diff --git a/packages/vscode/src/integrations/terminal/__test__/terminal-job.test.ts b/packages/vscode/src/integrations/terminal/__test__/terminal-job.test.ts index 14585bef4d..8c551dbd00 100644 --- a/packages/vscode/src/integrations/terminal/__test__/terminal-job.test.ts +++ b/packages/vscode/src/integrations/terminal/__test__/terminal-job.test.ts @@ -43,7 +43,10 @@ async function flushPromises(): Promise { await new Promise((resolve) => setImmediate(resolve)); } -function createHarness(options?: { read?: () => AsyncIterable }) { +function createHarness(options?: { + read?: () => AsyncIterable; + jobType?: "command" | "monitor"; +}) { const closeEmitter = new TestEventEmitter(); const shellIntegrationEmitter = new TestEventEmitter<{ terminal: FakeTerminal; @@ -118,7 +121,8 @@ function createHarness(options?: { read?: () => AsyncIterable }) { return ""; } }, - createBackgroundJobId: () => "bgjob-cmd-test", + createBackgroundJobId: (jobType: "command" | "monitor") => + `bgjob-${jobType === "monitor" ? "monitor" : "cmd"}-test`, getBackgroundJobOutputPath: () => "/tmp/bgjob-cmd-test.log", getShellPath: () => "/bin/sh", }, @@ -137,6 +141,7 @@ function createHarness(options?: { read?: () => AsyncIterable }) { command: "sleep 10", cwd: "/tmp", taskId: "task-test", + ...(options?.jobType ? { jobType: options.jobType } : {}), }); const finishEvents: BackgroundJobTerminalEvent[] = []; TerminalJob.onDidFinish((event) => { @@ -211,6 +216,21 @@ describe("TerminalJob", () => { assert.strictEqual(finalizeCalls.length, 1); }); + it("does not emit background command notifications for monitors", async () => { + const harness = createHarness({ jobType: "monitor" }); + + await flushPromises(); + harness.executionEndEmitter.fire({ + execution: harness.execution, + exitCode: 0, + }); + await flushPromises(); + + assert.strictEqual(harness.job.id, "bgjob-monitor-test"); + assert.deepStrictEqual(harness.lifecycle, ["file-closed"]); + assert.deepStrictEqual(harness.finishEvents, []); + }); + it("finalizes a running job when its terminal closes", async () => { const { TerminalJob, finalizeCalls, job, terminal } = createHarness(); diff --git a/packages/vscode/src/integrations/terminal/terminal-job.ts b/packages/vscode/src/integrations/terminal/terminal-job.ts index 95dcb8b00b..9f5a03a0c3 100644 --- a/packages/vscode/src/integrations/terminal/terminal-job.ts +++ b/packages/vscode/src/integrations/terminal/terminal-job.ts @@ -2,6 +2,7 @@ import { getLogger } from "@/lib/logger"; import type { BackgroundJobTerminalEvent } from "@getpochi/common"; import { getTerminalEnv } from "@getpochi/common/env-utils"; import { + type BackgroundJobIdType, BackgroundJobOutputFile, PlainOutputSanitizer, createBackgroundJobId, @@ -15,6 +16,17 @@ import { ExecutionError } from "./utils"; const logger = getLogger("TerminalJob"); +/** + * Hooks for a monitor watching this job's output. Implemented by the + * MonitorRegistry; TerminalJob only feeds it raw chunks and signals the end + * of execution. + */ +export interface TerminalJobMonitorHooks { + ingest(chunk: string): void; + /** Called once when command execution ends (exit, abort, terminal closed). */ + end(reason: string): void; +} + /** * Configuration options for creating a TerminalJob */ @@ -31,6 +43,10 @@ export interface TerminalJobConfig { abortSignal?: AbortSignal; /** Task that owns the job and receives its terminal notification. */ taskId: string; + /** The ID namespace used for the job. */ + jobType?: Extract; + /** Monitor tapping this job's output stream. */ + monitor?: TerminalJobMonitorHooks; } /** @@ -73,7 +89,7 @@ export class TerminalJob implements vscode.Disposable { } private constructor(private readonly config: TerminalJobConfig) { - this.id = createBackgroundJobId("command"); + this.id = createBackgroundJobId(config.jobType ?? "command"); this.outputFile = getBackgroundJobOutputPath(config.taskId, this.id); this.outputWriter = new BackgroundJobOutputFile(this.outputFile); this.outputManager = OutputManager.create({ @@ -167,6 +183,11 @@ export class TerminalJob implements vscode.Disposable { } this.outputManager.finalize(executionError); await this.finish(executionError); + this.config.monitor?.end( + executionError + ? executionError.message + : `exited with code ${this.exitCode ?? 0}`, + ); // Only tear down the execution-scoped listeners here. The job itself // stays registered until the terminal is closed (see closeListener), // so the UI can still highlight and reopen the terminal. @@ -234,11 +255,13 @@ export class TerminalJob implements vscode.Disposable { if (plainText.length === 0) continue; await this.outputWriter.append(plainText); this.outputManager.addChunk(plainText); + this.config.monitor?.ingest(plainText); } const remainder = sanitizer.end(); if (remainder.length > 0) { await this.outputWriter.append(remainder); this.outputManager.addChunk(remainder); + this.config.monitor?.ingest(remainder); } } @@ -368,6 +391,10 @@ export class TerminalJob implements vscode.Disposable { ); } + if (this.config.jobType === "monitor") { + return; + } + const status = this.stopRequested || finalError?.aborted ? "stopped" diff --git a/packages/vscode/src/integrations/terminal/terminal-state.ts b/packages/vscode/src/integrations/terminal/terminal-state.ts index 41d04f7332..4683f4624c 100644 --- a/packages/vscode/src/integrations/terminal/terminal-state.ts +++ b/packages/vscode/src/integrations/terminal/terminal-state.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { MonitorRegistry } from "@/integrations/monitor/monitor-registry"; import { getLogger } from "@/lib/logger"; // biome-ignore lint/style/useImportType: needed for dependency injection import { TaskDataStore } from "@/lib/task-data-store"; @@ -23,13 +24,16 @@ export interface TerminalInfo { * A stable id associated with the terminal's output file. * * The prefix encodes the terminal's origin: - * - `bgjob-cmd-` — a Pochi-started background job. Can be read and killed. - * - `term-` — a user-opened terminal. Read-only; `killBackgroundJob` refuses + * - `bgjob-cmd-` — a Pochi-started command job. Can be read and killed. + * - `bgjob-monitor-` — a monitor started by `startMonitor`. Can be read and killed. + * - `term-` — a user-opened terminal. Read-only; `killBackgroundJob` refuses * these because they are not tracked by the `TerminalJob` registry. */ backgroundJobId?: string; /** Absolute transcript path readable with readFile. */ outputFile?: string; + /** The monitor's description when this terminal is an active monitor. */ + monitor?: string; } @injectable() @@ -88,6 +92,7 @@ export class TerminalState implements vscode.Disposable { vscode.window.onDidCloseTerminal(this.onTerminalClosed), ); this.disposables.push(TerminalJob.onDidDispose(this.onTerminalChanged)); + this.disposables.push(MonitorRegistry.onDidChange(this.onTerminalChanged)); this.disposables.push( TerminalJob.onDidFinish((event) => { void this.taskDataStore.addBackgroundJobNotification( @@ -225,10 +230,11 @@ export class TerminalState implements vscode.Disposable { TerminalHistoryManager.getOrCreate(id).terminalName = t.name; } return { - name: t.name, + name: t.name || "Unnamed Terminal", isActive: t === vscode.window.activeTerminal, backgroundJobId: id, outputFile: this.getTerminalOutputFile(t), + monitor: MonitorRegistry.descriptionFor(id), }; }); } diff --git a/packages/vscode/src/integrations/webview/vscode-host-impl.ts b/packages/vscode/src/integrations/webview/vscode-host-impl.ts index 4b2dfb0262..7cdf3f8cc4 100644 --- a/packages/vscode/src/integrations/webview/vscode-host-impl.ts +++ b/packages/vscode/src/integrations/webview/vscode-host-impl.ts @@ -1,5 +1,6 @@ import * as os from "node:os"; import path from "node:path"; +import { MonitorRegistry } from "@/integrations/monitor/monitor-registry"; import { executeCommandWithPty } from "@/integrations/terminal/execute-command-with-pty"; // biome-ignore lint/style/useImportType: needed for dependency injection import { AuthEvents } from "@/lib/auth-events"; @@ -50,6 +51,7 @@ import { executeCommand } from "@/tools/execute-command"; import { globFiles } from "@/tools/glob-files"; import { killBackgroundJob } from "@/tools/kill-background-job"; import { listFiles as listFilesTool } from "@/tools/list-files"; +import { startMonitor } from "@/tools/monitor"; import { readFile } from "@/tools/read-file"; import { renderWidget } from "@/tools/render-widget"; import { searchFiles } from "@/tools/search-files"; @@ -61,6 +63,7 @@ import { type ContextWindowUsage, type Environment, type GitStatus, + type MonitorEventEnvelope, type TaskMemoryState, toErrorMessage, } from "@getpochi/common"; @@ -446,6 +449,7 @@ export class VSCodeHostImpl implements VSCodeHostApi, vscode.Disposable { deleteFileStateCache(taskId: string): void { this.fileStateCacheRegistry.delete(taskId); + MonitorRegistry.delete(taskId); } readActiveSelection = async (): Promise< @@ -474,6 +478,16 @@ export class VSCodeHostImpl implements VSCodeHostApi, vscode.Disposable { ), }); + readMonitorEvents = async ( + taskId: string, + ): Promise> => { + return ThreadSignal.serialize(MonitorRegistry.events(taskId)); + }; + + ackMonitorEvents = async (taskId: string, upToSeq: number): Promise => { + MonitorRegistry.ack(taskId, upToSeq); + }; + readCurrentWorkspace = async (): Promise<{ cwd: string | null; workspacePath: string | null; @@ -1634,6 +1648,7 @@ const ToolMap: Record< readFile, executeCommand, killBackgroundJob, + startMonitor, searchFiles, listFiles: listFilesTool, globFiles, diff --git a/packages/vscode/src/tools/kill-background-job.ts b/packages/vscode/src/tools/kill-background-job.ts index 7dbec8e4a3..4515076f55 100644 --- a/packages/vscode/src/tools/kill-background-job.ts +++ b/packages/vscode/src/tools/kill-background-job.ts @@ -8,7 +8,7 @@ export const killBackgroundJob: ToolFunctionType< if (!job) { if (backgroundJobId.startsWith("term-")) { throw new Error( - `"${backgroundJobId}" is a user-opened terminal and cannot be killed. Only background commands started by executeCommand (ids prefixed with "bgjob-cmd-") can be killed.`, + `"${backgroundJobId}" is a user-opened terminal and cannot be killed. Managed commands ("bgjob-cmd-") and monitors ("bgjob-monitor-") can be killed.`, ); } throw new Error(`Background job with ID "${backgroundJobId}" not found.`); diff --git a/packages/vscode/src/tools/monitor.ts b/packages/vscode/src/tools/monitor.ts new file mode 100644 index 0000000000..0c723aff01 --- /dev/null +++ b/packages/vscode/src/tools/monitor.ts @@ -0,0 +1,53 @@ +import * as path from "node:path"; +import { getViewColumnForTerminal } from "@/integrations/layout"; +import { MonitorRegistry } from "@/integrations/monitor/monitor-registry"; +import { TerminalJob } from "@/integrations/terminal/terminal-job"; +import { getBackgroundJobTerminalName } from "@/lib/background-job-terminal-name"; +import { MonitorDefaultTimeoutMs } from "@getpochi/common"; +import type { ClientTools, ToolFunctionType } from "@getpochi/tools"; + +export const startMonitor: ToolFunctionType< + ClientTools["startMonitor"] +> = async ( + { command, description, cwd = ".", timeoutMs, persistent }, + { abortSignal, cwd: workspaceDir, taskId }, +) => { + if (!command) { + throw new Error("Command is required to execute."); + } + + if (!taskId) { + throw new Error("Monitor requires a task context."); + } + + if (path.isAbsolute(cwd)) { + cwd = path.normalize(cwd); + } else { + cwd = path.normalize(path.join(workspaceDir, cwd)); + } + + const viewColumn = getViewColumnForTerminal(); + const location = viewColumn ? { viewColumn } : undefined; + + const monitor = MonitorRegistry.createMonitor({ + taskId, + description, + timeoutMs: persistent ? undefined : (timeoutMs ?? MonitorDefaultTimeoutMs), + }); + + const job = TerminalJob.create({ + name: getBackgroundJobTerminalName(command), + command, + cwd, + location, + abortSignal, + taskId, + jobType: "monitor", + monitor: monitor.hooks, + }); + monitor.attach(job.id); + + return { + backgroundJobId: job.id, + }; +};