From aacae356e6d59131cb68fa67e8ad44c7ba1b8f12 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:02:46 +0000 Subject: [PATCH 01/10] feat(protocol,agent-core): add 'monitor' background task kind Co-Authored-By: Kimi K2.7 Code --- packages/agent-core/src/agent/background/task.ts | 8 +++++++- packages/agent-core/src/services/task/task.ts | 3 +++ packages/protocol/src/events.ts | 14 +++++++++++++- packages/protocol/src/task.ts | 2 +- 4 files changed, 24 insertions(+), 3 deletions(-) diff --git a/packages/agent-core/src/agent/background/task.ts b/packages/agent-core/src/agent/background/task.ts index a053d4062e..2162f28a9c 100644 --- a/packages/agent-core/src/agent/background/task.ts +++ b/packages/agent-core/src/agent/background/task.ts @@ -2,6 +2,11 @@ import type { AgentBackgroundTaskInfo } from './agent-task'; import type { ProcessBackgroundTaskInfo } from './process-task'; import type { QuestionBackgroundTaskInfo } from './question-task'; +export interface MonitorBackgroundTaskInfo extends BackgroundTaskInfoBase { + readonly kind: 'monitor'; + readonly command: string; +} + export type BackgroundTaskStatus = | 'running' | 'completed' @@ -47,7 +52,8 @@ export interface BackgroundTaskInfoBase { export type BackgroundTaskInfo = | ProcessBackgroundTaskInfo | AgentBackgroundTaskInfo - | QuestionBackgroundTaskInfo; + | QuestionBackgroundTaskInfo + | MonitorBackgroundTaskInfo; export interface BackgroundTaskSink { readonly signal: AbortSignal; diff --git a/packages/agent-core/src/services/task/task.ts b/packages/agent-core/src/services/task/task.ts index 1eca2b2de2..0b428ce2bc 100644 --- a/packages/agent-core/src/services/task/task.ts +++ b/packages/agent-core/src/services/task/task.ts @@ -57,6 +57,9 @@ function mapKind(k: BackgroundTaskInfo['kind']): BackgroundTaskKind { // tool-spawned flows (Loop runs them as part of `Question` tool // execution), so 'tool' is the closest spec literal. return 'tool'; + case 'monitor': + // Monitors are tool-spawned persistent background tasks. + return 'tool'; } } diff --git a/packages/protocol/src/events.ts b/packages/protocol/src/events.ts index 3b2eceaafb..18b3200945 100644 --- a/packages/protocol/src/events.ts +++ b/packages/protocol/src/events.ts @@ -361,10 +361,16 @@ export interface QuestionTaskInfo extends TaskInfoBase { readonly toolCallId?: string; } +export interface MonitorTaskInfo extends TaskInfoBase { + readonly kind: 'monitor'; + readonly command: string; +} + export type TaskInfo = | ProcessTaskInfo | AgentTaskInfo - | QuestionTaskInfo; + | QuestionTaskInfo + | MonitorTaskInfo; export interface CompactionResult { readonly summary: string; @@ -1264,10 +1270,16 @@ export const questionTaskInfoSchema = taskInfoBaseSchema.extend({ toolCallId: z.string().optional(), }) satisfies z.ZodType; +export const monitorTaskInfoSchema = taskInfoBaseSchema.extend({ + kind: z.literal('monitor'), + command: z.string(), +}) satisfies z.ZodType; + export const taskInfoSchema = z.discriminatedUnion('kind', [ processTaskInfoSchema, agentTaskInfoSchema, questionTaskInfoSchema, + monitorTaskInfoSchema, ]) satisfies z.ZodType; export const compactionResultSchema = z.object({ diff --git a/packages/protocol/src/task.ts b/packages/protocol/src/task.ts index bc08eec71f..a8f8dae4b3 100644 --- a/packages/protocol/src/task.ts +++ b/packages/protocol/src/task.ts @@ -2,7 +2,7 @@ import { z } from 'zod'; import { isoDateTimeSchema } from './time'; -export const taskKindSchema = z.enum(['subagent', 'bash', 'tool']); +export const taskKindSchema = z.enum(['subagent', 'bash', 'tool', 'monitor']); export type TaskKind = z.infer; export const taskStatusSchema = z.enum([ From bdda88efc045f8b6d008eb897ea9fcd38c49e8ea Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:05:33 +0000 Subject: [PATCH 02/10] feat(agent-core): add MonitorBackgroundTask with line batching and volume cap Co-Authored-By: Kimi K2.7 Code --- .../src/agent/background/monitor-task.ts | 270 ++++++++++++++++++ .../agent/background/monitor-task.test.ts | 206 +++++++++++++ 2 files changed, 476 insertions(+) create mode 100644 packages/agent-core/src/agent/background/monitor-task.ts create mode 100644 packages/agent-core/test/agent/background/monitor-task.test.ts diff --git a/packages/agent-core/src/agent/background/monitor-task.ts b/packages/agent-core/src/agent/background/monitor-task.ts new file mode 100644 index 0000000000..96db3bf0d3 --- /dev/null +++ b/packages/agent-core/src/agent/background/monitor-task.ts @@ -0,0 +1,270 @@ +import type { KaosProcess } from '@moonshot-ai/kaos'; +import type { Readable } from 'node:stream'; + +import { errorMessage } from '../../loop/errors'; +import type { + BackgroundTask, + BackgroundTaskInfoBase, + BackgroundTaskSink, + BackgroundTaskSettlement, + MonitorBackgroundTaskInfo, +} from './task'; + +export type MonitorEmit = (lines: string[], severity?: 'info' | 'warning') => void; + +export interface MonitorOptions { + /** Debounce window for batching lines before emitting. */ + readonly batchMs?: number; + /** Maximum complete lines allowed inside the volume window. */ + readonly maxLinesPerWindow?: number; + /** Length of the sliding volume window in milliseconds. */ + readonly volumeWindowMs?: number; +} + +const DEFAULT_BATCH_MS = 200; +const DEFAULT_MAX_LINES_PER_WINDOW = 200; +const DEFAULT_VOLUME_WINDOW_MS = 5000; +const STREAM_DRAIN_GRACE_MS = 250; + +export class MonitorBackgroundTask implements BackgroundTask { + readonly kind = 'monitor' as const; + readonly idPrefix = 'monitor'; + + private exitCode: number | null = null; + private pending = ''; + private batch: string[] = []; + private batchTimer: ReturnType | undefined; + private volumeWindowStart = 0; + private volumeCount = 0; + private capped = false; + + constructor( + readonly proc: KaosProcess, + readonly command: string, + readonly description: string, + private readonly emit: MonitorEmit, + private readonly options: MonitorOptions = {}, + readonly timeoutMs?: number, + ) {} + + async start(sink: BackgroundTaskSink): Promise { + const stdoutDrained = observeMonitorStream(this.proc.stdout, sink, (chunk) => + this.feed(chunk), + ); + const streamDrained = Promise.all([ + stdoutDrained, + observeMonitorStream(this.proc.stderr, sink), + ]).then(() => undefined); + // Attach a rejection handler immediately; start() still awaits the same + // promise after proc.wait() so stream errors keep failing the task. + void streamDrained.catch(() => {}); + + const requestStop = (): void => { + void this.proc.kill('SIGTERM').catch(() => {}); + }; + if (sink.signal.aborted) { + requestStop(); + } else { + sink.signal.addEventListener('abort', requestStop, { once: true }); + } + + let settlement: BackgroundTaskSettlement; + try { + const exitCode = await this.proc.wait(); + await waitForStreamDrain(streamDrained); + this.exitCode = exitCode; + settlement = { + status: sink.signal.aborted ? 'killed' : exitCode === 0 ? 'completed' : 'failed', + }; + } catch (error: unknown) { + await waitForStreamDrainSettled(streamDrained); + this.exitCode = this.proc.exitCode; + settlement = { + status: sink.signal.aborted ? 'killed' : 'failed', + stopReason: sink.signal.aborted ? undefined : errorMessage(error), + }; + } finally { + sink.signal.removeEventListener('abort', requestStop); + this.flushBatch(true); + this.clearBatchTimer(); + await this.disposeProcess(); + } + await sink.settle(settlement); + } + + async forceStop(): Promise { + try { + if (this.proc.exitCode === null) { + await this.proc.kill('SIGKILL'); + } + } finally { + this.clearBatchTimer(); + await this.disposeProcess(); + } + } + + toInfo(base: BackgroundTaskInfoBase): MonitorBackgroundTaskInfo { + return { + ...base, + kind: 'monitor', + command: this.command, + }; + } + + private feed(chunk: string): void { + if (this.capped) return; + this.pending += chunk; + const parts = this.pending.split('\n'); + this.pending = parts.pop() ?? ''; + for (const line of parts) { + this.batch.push(line); + this.checkVolume(line); + if (this.capped) break; + } + this.scheduleBatchFlush(); + } + + private checkVolume(line: string): void { + const now = Date.now(); + const windowMs = this.options.volumeWindowMs ?? DEFAULT_VOLUME_WINDOW_MS; + if (now - this.volumeWindowStart > windowMs) { + this.volumeWindowStart = now; + this.volumeCount = 0; + } + this.volumeCount += 1; + const maxLines = this.options.maxLinesPerWindow ?? DEFAULT_MAX_LINES_PER_WINDOW; + if (this.volumeCount > maxLines) { + this.capped = true; + void this.forceStop().catch(() => {}); + this.emit( + [ + `Monitor auto-stopped: too noisy (>${String(maxLines)} lines / ${String(windowMs / 1000)}s). Restart with a tighter filter.`, + ], + 'warning', + ); + } + } + + private scheduleBatchFlush(): void { + this.clearBatchTimer(); + if (this.batch.length === 0) return; + this.batchTimer = setTimeout( + () => this.flushBatch(false), + this.options.batchMs ?? DEFAULT_BATCH_MS, + ); + this.batchTimer.unref?.(); + } + + private flushBatch(terminal: boolean): void { + this.clearBatchTimer(); + if (this.batch.length > 0) { + this.emit(this.batch); + this.batch = []; + } + if (terminal && this.pending.length > 0 && !this.capped) { + this.emit([this.pending]); + this.pending = ''; + } + } + + private clearBatchTimer(): void { + if (this.batchTimer !== undefined) { + clearTimeout(this.batchTimer); + this.batchTimer = undefined; + } + } + + private async disposeProcess(): Promise { + try { + await this.proc.dispose(); + } catch { + /* best-effort cleanup */ + } + } +} + +async function waitForStreamDrain(streamDrained: Promise): Promise { + let timeout: ReturnType | undefined; + try { + await Promise.race([ + streamDrained, + new Promise((resolve) => { + timeout = setTimeout(resolve, STREAM_DRAIN_GRACE_MS); + timeout.unref?.(); + }), + ]); + } finally { + if (timeout !== undefined) clearTimeout(timeout); + } +} + +async function waitForStreamDrainSettled(streamDrained: Promise): Promise { + try { + await waitForStreamDrain(streamDrained); + } catch { + /* original process/stream error wins */ + } +} + +function observeMonitorStream( + stream: Readable, + sink: BackgroundTaskSink, + onOutput?: (chunk: string) => void, +): Promise { + stream.setEncoding('utf8'); + const onData = (chunk: string): void => { + if (chunk.length === 0) return; + sink.appendOutput(chunk); + onOutput?.(chunk); + }; + stream.on('data', onData); + + return new Promise((resolve, reject) => { + let ended = false; + const settle = (callback: () => void): void => { + cleanup(); + callback(); + }; + const done = (): void => { + settle(resolve); + }; + const fail = (error: unknown): void => { + settle(() => reject(error)); + }; + const onEnd = (): void => { + ended = true; + done(); + }; + const onClose = (): void => { + if (ended || sink.signal.aborted) { + done(); + return; + } + fail(createPrematureCloseError()); + }; + const onError = (error: Error): void => { + // When the task is aborted we intentionally destroy the streams, which + // can emit errors. Swallow those expected errors; surface anything else. + if (sink.signal.aborted) { + done(); + } else { + fail(error); + } + }; + const cleanup = (): void => { + stream.removeListener('data', onData); + stream.removeListener('end', onEnd); + stream.removeListener('close', onClose); + stream.removeListener('error', onError); + }; + stream.once('end', onEnd); + stream.once('close', onClose); + stream.once('error', onError); + }); +} + +function createPrematureCloseError(): Error { + const error = new Error('Premature close') as NodeJS.ErrnoException; + error.code = 'ERR_STREAM_PREMATURE_CLOSE'; + return error; +} diff --git a/packages/agent-core/test/agent/background/monitor-task.test.ts b/packages/agent-core/test/agent/background/monitor-task.test.ts new file mode 100644 index 0000000000..690309bf7c --- /dev/null +++ b/packages/agent-core/test/agent/background/monitor-task.test.ts @@ -0,0 +1,206 @@ +/** + * Covers: MonitorBackgroundTask line buffering, batching, terminal flush, + * and volume cap. + */ + +import { PassThrough, Readable } from 'node:stream'; +import type { Writable } from 'node:stream'; + +import type { KaosProcess } from '@moonshot-ai/kaos'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { + MonitorBackgroundTask, + type MonitorEmit, +} from '../../../src/agent/background/monitor-task'; +import type { BackgroundTaskSink } from '../../../src/agent/background/task'; + +function createMonitorProcess(): { + proc: KaosProcess; + stdout: PassThrough; + resolveWait: (exitCode: number) => void; + killSpy: ReturnType; + disposeSpy: ReturnType; +} { + const stdout = new PassThrough(); + const stderr = Readable.from([]); + let resolveWait!: (exitCode: number) => void; + const waitPromise = new Promise((resolve) => { + resolveWait = resolve; + }); + const killSpy = vi.fn().mockResolvedValue(undefined); + const disposeSpy = vi.fn().mockResolvedValue(undefined); + const proc: KaosProcess = { + stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable, + stdout, + stderr, + pid: 12345, + exitCode: null, + wait: () => waitPromise, + kill: killSpy as unknown as KaosProcess['kill'], + dispose: disposeSpy as unknown as KaosProcess['dispose'], + }; + return { proc, stdout, resolveWait, killSpy, disposeSpy }; +} + +function createSink(controller?: AbortController): BackgroundTaskSink & { appendOutput: ReturnType; settle: ReturnType } { + const ctrl = controller ?? new AbortController(); + return { + signal: ctrl.signal, + appendOutput: vi.fn<(chunk: string) => void>(), + settle: vi.fn().mockResolvedValue(true), + }; +} + +describe('MonitorBackgroundTask', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('batches complete lines and flushes after batchMs', async () => { + const { proc, stdout, resolveWait } = createMonitorProcess(); + const emits: { lines: string[]; sev?: string }[] = []; + const emit: MonitorEmit = (lines, sev) => emits.push({ lines: [...lines], sev }); + const task = new MonitorBackgroundTask(proc, 'echo a b', 'test', emit, { batchMs: 200 }); + const sink = createSink(); + + const startPromise = task.start(sink); + await Promise.resolve(); + stdout.write('a\nb\n'); + stdout.end(); + await vi.advanceTimersByTimeAsync(200); + resolveWait(0); + await startPromise; + + expect(emits).toHaveLength(1); + expect(emits[0]).toEqual({ lines: ['a', 'b'] }); + }); + + it('holds a partial line until a newline arrives', async () => { + const { proc, stdout, resolveWait } = createMonitorProcess(); + const emits: { lines: string[]; sev?: string }[] = []; + const emit: MonitorEmit = (lines, sev) => emits.push({ lines: [...lines], sev }); + const task = new MonitorBackgroundTask(proc, 'echo partial', 'test', emit, { batchMs: 200 }); + const sink = createSink(); + + const startPromise = task.start(sink); + await Promise.resolve(); + stdout.write('part'); + await vi.advanceTimersByTimeAsync(200); + expect(emits).toHaveLength(0); + + stdout.write('ial\n'); + await vi.advanceTimersByTimeAsync(200); + stdout.end(); + resolveWait(0); + await startPromise; + + expect(emits).toHaveLength(1); + expect(emits[0]).toEqual({ lines: ['partial'] }); + }); + + it('flushes a trailing partial line when the process ends', async () => { + const { proc, stdout, resolveWait } = createMonitorProcess(); + const emits: { lines: string[]; sev?: string }[] = []; + const emit: MonitorEmit = (lines, sev) => emits.push({ lines: [...lines], sev }); + const task = new MonitorBackgroundTask(proc, 'echo trailing', 'test', emit, { batchMs: 200 }); + const sink = createSink(); + + const startPromise = task.start(sink); + await Promise.resolve(); + stdout.write('a\npartial'); + stdout.end(); + resolveWait(0); + await startPromise; + + expect(emits.map((e) => e.lines)).toEqual([['a'], ['partial']]); + }); + + it('flushes a single unterminated line when the process ends', async () => { + const { proc, stdout, resolveWait } = createMonitorProcess(); + const emits: { lines: string[]; sev?: string }[] = []; + const emit: MonitorEmit = (lines, sev) => emits.push({ lines: [...lines], sev }); + const task = new MonitorBackgroundTask(proc, 'echo only', 'test', emit, { batchMs: 200 }); + const sink = createSink(); + + const startPromise = task.start(sink); + await Promise.resolve(); + stdout.write('only'); + stdout.end(); + resolveWait(0); + await startPromise; + + expect(emits.map((e) => e.lines)).toEqual([['only']]); + }); + + it('auto-stops and emits a warning when line volume exceeds the cap', async () => { + const { proc, stdout, resolveWait, killSpy } = createMonitorProcess(); + const emits: { lines: string[]; sev?: string }[] = []; + const emit: MonitorEmit = (lines, sev) => emits.push({ lines: [...lines], sev }); + const task = new MonitorBackgroundTask( + proc, + 'flood', + 'test', + emit, + { batchMs: 200, maxLinesPerWindow: 200, volumeWindowMs: 5000 }, + ); + const sink = createSink(); + + const startPromise = task.start(sink); + await Promise.resolve(); + const data = Array.from({ length: 201 }, (_, i) => `line-${String(i)}`).join('\n') + '\n'; + stdout.write(data); + stdout.end(); + resolveWait(0); + await startPromise; + + expect(killSpy).toHaveBeenCalledWith('SIGKILL'); + expect(emits.some((e) => e.sev === 'warning' && e.lines[0]?.includes('too noisy'))).toBe(true); + }); + + it('tees every stdout chunk to the sink', async () => { + const { proc, stdout, resolveWait } = createMonitorProcess(); + const task = new MonitorBackgroundTask(proc, 'echo tee', 'test', () => {}, { batchMs: 200 }); + const sink = createSink(); + + const startPromise = task.start(sink); + await Promise.resolve(); + stdout.write('a\n'); + stdout.write('b\n'); + stdout.end(); + await vi.advanceTimersByTimeAsync(200); + resolveWait(0); + await startPromise; + + expect(sink.appendOutput).toHaveBeenCalledWith('a\n'); + expect(sink.appendOutput).toHaveBeenCalledWith('b\n'); + }); + + it('reports monitor info via toInfo', () => { + const { proc } = createMonitorProcess(); + const task = new MonitorBackgroundTask(proc, 'echo info', 'desc', () => {}); + const info = task.toInfo({ + taskId: 'monitor-abc12345', + description: 'desc', + status: 'running', + detached: true, + startedAt: 1, + endedAt: null, + }); + + expect(info).toEqual({ + taskId: 'monitor-abc12345', + description: 'desc', + status: 'running', + detached: true, + startedAt: 1, + endedAt: null, + kind: 'monitor', + command: 'echo info', + }); + }); +}); From adfd70acec687ad6d3f25d4f94267587fb57091e Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:10:29 +0000 Subject: [PATCH 03/10] feat(agent-core): emit per-line monitor notifications via BackgroundManager Co-Authored-By: Kimi K2.7 Code --- .../agent-core/src/agent/background/index.ts | 59 +++++++++++++ .../agent/background/monitor-manager.test.ts | 84 +++++++++++++++++++ 2 files changed, 143 insertions(+) create mode 100644 packages/agent-core/test/agent/background/monitor-manager.test.ts diff --git a/packages/agent-core/src/agent/background/index.ts b/packages/agent-core/src/agent/background/index.ts index c9cb88fc7b..2fe15ae06e 100644 --- a/packages/agent-core/src/agent/background/index.ts +++ b/packages/agent-core/src/agent/background/index.ts @@ -14,6 +14,7 @@ import { randomBytes } from 'node:crypto'; import { createControlledPromise, type ControlledPromise } from '@antfu/utils'; import type { ContentPart } from '@moonshot-ai/kosong'; +import type { KaosProcess } from '@moonshot-ai/kaos'; import type { Agent } from '../..'; import { errorMessage } from '../../loop/errors'; @@ -22,6 +23,8 @@ import { escapeXml, escapeXmlAttr } from '../../utils/xml-escape'; import type { BackgroundTaskOrigin } from '../context'; import { renderNotificationXml } from '../context/notification-xml'; import { type BackgroundTaskPersistence } from './persist'; +import { MonitorBackgroundTask } from './monitor-task'; +import type { MonitorEmit } from './monitor-task'; import { TERMINAL_STATUSES, type BackgroundTask, @@ -48,6 +51,8 @@ export { ProcessBackgroundTask } from './process-task'; export type { ProcessBackgroundTaskInfo } from './process-task'; export { QuestionBackgroundTask } from './question-task'; export type { QuestionBackgroundTaskInfo } from './question-task'; +export { MonitorBackgroundTask } from './monitor-task'; +export type { MonitorBackgroundTaskInfo } from './task'; export { BackgroundTaskPersistence } from './persist'; export type { BackgroundTaskInfo, @@ -252,6 +257,7 @@ export class BackgroundManager { private readonly scheduledNotificationKeys = new Set(); private readonly deliveredNotificationKeys = new Set(); + private readonly monitorSeqCounters = new Map(); constructor( private readonly agent: Agent, @@ -313,6 +319,24 @@ export class BackgroundManager { return entry.options.autoBackgroundOnTimeout === true && !this.isDetached(entry); } + registerMonitorTask( + proc: KaosProcess, + command: string, + description: string, + options: RegisterBackgroundTaskOptions = {}, + ): string { + const taskIdHolder: { taskId: string } = { taskId: '' }; + const emit: MonitorEmit = (lines, severity) => { + this.monitorNotify(taskIdHolder.taskId, description, lines, severity); + }; + const taskId = this.registerTask( + new MonitorBackgroundTask(proc, command, description, emit, {}, options.timeoutMs), + options, + ); + taskIdHolder.taskId = taskId; + return taskId; + } + registerTask(task: BackgroundTask, options: RegisterBackgroundTaskOptions = {}): string { const detached = options.detached ?? true; const timeoutMs = options.timeoutMs ?? task.timeoutMs; @@ -777,6 +801,41 @@ export class BackgroundManager { this.fireNotificationHook(context.notification); } + private nextMonitorSeq(taskId: string): number { + const next = (this.monitorSeqCounters.get(taskId) ?? 0) + 1; + this.monitorSeqCounters.set(taskId, next); + return next; + } + + private monitorNotify( + taskId: string, + description: string, + lines: string[], + severity: 'info' | 'warning' = 'info', + ): void { + const body = lines.join('\n'); + const notification = { + id: `monitor:${taskId}:${this.nextMonitorSeq(taskId)}`, + category: 'task' as const, + type: 'monitor_line', + source_kind: 'monitor' as const, + source_id: taskId, + title: description, + severity, + body, + tail_output: '', + }; + const text = renderNotificationXml(notification); + const origin: BackgroundTaskOrigin = { + kind: 'background_task', + taskId, + status: 'running', + notificationId: `monitor:${taskId}`, + }; + this.agent.turn.steer([{ type: 'text', text }], origin); + this.fireNotificationHook(notification as unknown as BackgroundTaskNotification); + } + private async restoreBackgroundTaskNotification(info: BackgroundTaskInfo): Promise { const context = await this.buildBackgroundTaskNotificationContext(info); if (context === undefined) return; diff --git a/packages/agent-core/test/agent/background/monitor-manager.test.ts b/packages/agent-core/test/agent/background/monitor-manager.test.ts new file mode 100644 index 0000000000..2714c20c67 --- /dev/null +++ b/packages/agent-core/test/agent/background/monitor-manager.test.ts @@ -0,0 +1,84 @@ +/** + * Covers: BackgroundManager monitor notification wiring. + */ + +import { PassThrough, Readable } from 'node:stream'; +import type { Writable } from 'node:stream'; + +import type { KaosProcess } from '@moonshot-ai/kaos'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { createBackgroundManager } from './helpers'; + +function createMonitorProcess(): { + proc: KaosProcess; + stdout: PassThrough; + resolveWait: (exitCode: number) => void; + killSpy: ReturnType; +} { + const stdout = new PassThrough(); + const stderr = Readable.from([]); + let resolveWait!: (exitCode: number) => void; + const waitPromise = new Promise((resolve) => { + resolveWait = resolve; + }); + const killSpy = vi.fn().mockResolvedValue(undefined); + const proc: KaosProcess = { + stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable, + stdout, + stderr, + pid: 12345, + exitCode: null, + wait: () => waitPromise, + kill: killSpy as unknown as KaosProcess['kill'], + dispose: vi.fn().mockResolvedValue(undefined) as KaosProcess['dispose'], + }; + return { proc, stdout, resolveWait, killSpy }; +} + +describe('BackgroundManager monitor wiring', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it('emits per-line monitor notifications and still tees output', async () => { + const { agent, manager } = createBackgroundManager(); + const { proc, stdout, resolveWait } = createMonitorProcess(); + + const taskId = manager.registerMonitorTask( + proc, + 'tail -f log | grep --line-buffered x', + 'watch x', + { detached: true }, + ); + expect(taskId).toMatch(/^monitor-[0-9a-z]{8}$/); + + await Promise.resolve(); + stdout.write('x\ny\n'); + await vi.advanceTimersByTimeAsync(200); + + const monitorCall = agent.turn.steer.mock.calls.find((call) => { + const text = (call[0] as { text: string }[] | undefined)?.[0]?.text; + return typeof text === 'string' && text.includes('type="monitor_line"'); + }); + expect(monitorCall).toBeDefined(); + const monitorText = (monitorCall![0] as { text: string }[])[0]!.text; + expect(monitorText).toContain('x\ny'); + + stdout.end(); + resolveWait(0); + await vi.waitFor(() => expect(manager.getTask(taskId)?.status).not.toBe('running')); + + const terminalCall = agent.turn.steer.mock.calls.find((call) => { + const text = (call[0] as { text: string }[] | undefined)?.[0]?.text; + return typeof text === 'string' && text.includes('type="task.completed"'); + }); + expect(terminalCall).toBeDefined(); + + expect(await manager.readOutput(taskId)).toContain('x\ny'); + }); +}); From 8f87dad2a5aa7b0fa52017e71cc7fe0e9e7bb7ba Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:12:10 +0000 Subject: [PATCH 04/10] feat(agent-core): add Monitor tool Co-Authored-By: Kimi K2.7 Code --- .../src/tools/background/monitor.md | 20 +++ .../src/tools/background/monitor.ts | 143 ++++++++++++++++++ .../test/tools/monitor-tool.test.ts | 104 +++++++++++++ 3 files changed, 267 insertions(+) create mode 100644 packages/agent-core/src/tools/background/monitor.md create mode 100644 packages/agent-core/src/tools/background/monitor.ts create mode 100644 packages/agent-core/test/tools/monitor-tool.test.ts diff --git a/packages/agent-core/src/tools/background/monitor.md b/packages/agent-core/src/tools/background/monitor.md new file mode 100644 index 0000000000..eda6cb553b --- /dev/null +++ b/packages/agent-core/src/tools/background/monitor.md @@ -0,0 +1,20 @@ +Run a self-filtering shell command in the background and receive each new stdout line as a real-time notification. + +Use `Monitor` to watch long-running processes, log files, or build/test output without blocking the conversation. The command should produce one line per event you care about. + +**How to design a good monitor command:** +- Self-filter on the shell side so stdout only contains meaningful events. Examples: + - `tail -F app.log | grep --line-buffered "ERROR"` + - `python -u server.py 2>&1 | grep --line-buffered "Traceback\|Error"` + - `awk '/pattern/ {print; fflush()}' <(tail -F log.txt)` — always flush per line. +- Ensure line buffering. Tools like `grep` default to line buffering only when writing to a terminal; force it with `--line-buffered`. In `awk`, call `fflush()` after each `print`. +- Silence is not success. If the filter stops matching, the monitor simply emits nothing. Include failure signatures in the filter when relevant, e.g. `Traceback|Error|FAILED|Killed|OOM`. +- Keep output sparse. The monitor auto-stops with a warning if it emits more than 200 lines in a 5-second window. If that happens, tighten the filter and start again. + +**Lifecycle:** +- `timeout_ms` defaults to 5 minutes and is ignored when `persistent=true`. +- `persistent=true` runs until the session ends or you call `TaskStop`. +- Stop any monitor early with `TaskStop(task_id="...")`. + +**Output:** +The tool returns immediately with a `task_id`. Each matching stdout line arrives later as a `` injection. diff --git a/packages/agent-core/src/tools/background/monitor.ts b/packages/agent-core/src/tools/background/monitor.ts new file mode 100644 index 0000000000..22df263a8f --- /dev/null +++ b/packages/agent-core/src/tools/background/monitor.ts @@ -0,0 +1,143 @@ +/** + * MonitorTool — run a self-filtering shell command in the background and + * receive each new stdout line as a notification. + */ + +import type { Kaos, KaosProcess } from '@moonshot-ai/kaos'; +import { z } from 'zod'; + +import type { BackgroundManager } from '../../agent/background'; +import type { BuiltinTool } from '../../agent/tool'; +import type { ExecutableToolResult, ToolExecution } from '../../loop/types'; +import { toInputJsonSchema } from '../support/input-schema'; +import { literalRulePattern, matchesGlobRuleSubject } from '../support/rule-match'; +import MONITOR_DESCRIPTION from './monitor.md?raw'; + +export const MonitorInputSchema = z.object({ + command: z.string().describe('Shell command to monitor. Each stdout line is an event; self-filter (e.g. grep --line-buffered).'), + description: z.string().describe('Short description shown in every notification.'), + timeout_ms: z + .number() + .int() + .min(1000) + .max(3600000) + .default(300000) + .describe('Kill the monitor after this deadline. Ignored when persistent=true.'), + persistent: z + .boolean() + .default(false) + .describe('Run for the lifetime of the session (no timeout). Stop with TaskStop.'), +}); + +export type MonitorInput = z.Infer; + +export class MonitorTool implements BuiltinTool { + readonly name = 'Monitor' as const; + readonly description = MONITOR_DESCRIPTION; + readonly parameters: Record = toInputJsonSchema(MonitorInputSchema); + + private readonly isWindowsBash: boolean; + + constructor( + private readonly kaos: Kaos, + private readonly cwd: string, + private readonly background: BackgroundManager, + ) { + this.isWindowsBash = this.kaos.osEnv.osKind === 'Windows'; + } + + resolveExecution(args: MonitorInput): ToolExecution { + return { + description: `Monitoring: ${args.command}`, + approvalRule: literalRulePattern(this.name, args.command), + matchesRule: (ruleArgs) => matchesGlobRuleSubject(ruleArgs, args.command), + execute: async (_ctx) => this.execution(args), + }; + } + + private async execution(args: MonitorInput): Promise { + const effectiveCwd = this.cwd; + const command = this.isWindowsBash ? rewriteWindowsNullRedirect(args.command) : args.command; + + let proc: KaosProcess; + try { + proc = await this.spawn(effectiveCwd, command); + } catch (error) { + return { + isError: true, + output: error instanceof Error ? error.message : String(error), + }; + } + closeProcessStdin(proc); + + const timeoutMs = args.persistent ? undefined : args.timeout_ms; + const taskId = this.background.registerMonitorTask(proc, command, args.description, { + detached: true, + timeoutMs, + }); + + return { + isError: false, + message: 'Monitor started.', + output: + `task_id: ${taskId}\n` + + `persistent: ${args.persistent}\n` + + 'Each matching stdout line arrives as a notification. Stop with TaskStop.', + }; + } + + private spawn(effectiveCwd: string, command: string): Promise { + const shellCwd = this.isWindowsBash ? windowsPathToPosixPath(effectiveCwd) : effectiveCwd; + const shellArgs = [ + this.kaos.osEnv.shellPath, + '-c', + `cd ${shellQuote(shellCwd)} && ${command}`, + ]; + + const noninteractiveEnv: Record = { + NO_COLOR: '1', + TERM: 'dumb', + GIT_TERMINAL_PROMPT: process.env['GIT_TERMINAL_PROMPT'] ?? '0', + SHELL: this.kaos.osEnv.shellPath, + }; + + const mergedEnv: Record = { + ...(process.env as Record), + ...noninteractiveEnv, + }; + return this.kaos.execWithEnv(shellArgs, mergedEnv); + } +} + +function closeProcessStdin(proc: KaosProcess): void { + try { + proc.stdin.end(); + } catch { + /* process already gone */ + } +} + +function shellQuote(s: string): string { + return `'${s.replaceAll("'", "'\\''")}'`; +} + +function windowsPathToPosixPath(path: string): string { + if (path.startsWith('\\\\')) { + return path.replaceAll('\\', '/'); + } + + const driveMatch = /^([A-Za-z]):(?:[\\/]|$)/.exec(path); + if (driveMatch !== null) { + const drive = driveMatch[1]!.toLowerCase(); + const rest = path.slice(2).replaceAll('\\', '/'); + return `/${drive}${rest.startsWith('/') ? rest : `/${rest}`}`; + } + + return path.replaceAll('\\', '/'); +} + +const WINDOWS_NUL_REDIRECT = /(\d?&?>+\s*)[Nn][Uu][Ll](?=\s|$|[|&;)\n])/g; + +function rewriteWindowsNullRedirect(command: string): string { + return command.replace(WINDOWS_NUL_REDIRECT, '$1/dev/null'); +} diff --git a/packages/agent-core/test/tools/monitor-tool.test.ts b/packages/agent-core/test/tools/monitor-tool.test.ts new file mode 100644 index 0000000000..b74b461c77 --- /dev/null +++ b/packages/agent-core/test/tools/monitor-tool.test.ts @@ -0,0 +1,104 @@ +/** + * Covers: MonitorTool. + */ + +import { Readable } from 'node:stream'; +import type { Writable } from 'node:stream'; + +import type { Kaos, KaosProcess } from '@moonshot-ai/kaos'; +import { describe, expect, it, vi } from 'vitest'; + +import type { BackgroundManager } from '../../src/agent/background'; +import type { RunnableToolExecution } from '../../src/loop/types'; +import { MonitorTool } from '../../src/tools/background/monitor'; + +function fakeKaos(proc: KaosProcess): Kaos { + return { + osEnv: { + shellPath: '/bin/bash', + shellName: 'bash', + osKind: 'Linux', + }, + execWithEnv: vi.fn().mockResolvedValue(proc), + } as unknown as Kaos; +} + +function fakeProcess(): KaosProcess { + return { + stdin: { write: vi.fn(), end: vi.fn() } as unknown as Writable, + stdout: Readable.from([]), + stderr: Readable.from([]), + pid: 12345, + exitCode: null, + wait: vi.fn().mockResolvedValue(0), + kill: vi.fn().mockResolvedValue(undefined), + dispose: vi.fn().mockResolvedValue(undefined), + }; +} + +function fakeBackgroundManager(taskId = 'monitor-test123'): BackgroundManager { + return { + registerMonitorTask: vi.fn().mockReturnValue(taskId), + } as unknown as BackgroundManager; +} + +describe('MonitorTool', () => { + it('registers a detached monitor task and returns its task id', async () => { + const proc = fakeProcess(); + const kaos = fakeKaos(proc); + const background = fakeBackgroundManager(); + const tool = new MonitorTool(kaos, '/workspace', background); + + const execution = tool.resolveExecution({ + command: 'tail -f log.txt | grep --line-buffered ERROR', + description: 'watch errors', + timeout_ms: 300000, + persistent: false, + }); + + const result = await (execution as RunnableToolExecution).execute({ + turnId: 'turn-1', + toolCallId: 'call-1', + signal: new AbortController().signal, + }); + + expect(result.isError).toBe(false); + expect(result.output).toContain('task_id: monitor-test123'); + expect(result.output).toContain('persistent: false'); + expect(background.registerMonitorTask).toHaveBeenCalledWith( + proc, + 'tail -f log.txt | grep --line-buffered ERROR', + 'watch errors', + { detached: true, timeoutMs: 300000 }, + ); + }); + + it('passes undefined timeout for persistent monitors', async () => { + const proc = fakeProcess(); + const kaos = fakeKaos(proc); + const background = fakeBackgroundManager('monitor-persist'); + const tool = new MonitorTool(kaos, '/workspace', background); + + const execution = tool.resolveExecution({ + command: 'tail -F app.log', + description: 'watch log', + timeout_ms: 300000, + persistent: true, + }); + + const result = await (execution as RunnableToolExecution).execute({ + turnId: 'turn-1', + toolCallId: 'call-1', + signal: new AbortController().signal, + }); + + expect(result.isError).toBe(false); + expect(result.output).toContain('persistent: true'); + expect(background.registerMonitorTask).toHaveBeenCalledWith( + proc, + 'tail -F app.log', + 'watch log', + { detached: true, timeoutMs: undefined }, + ); + }); +}); From 15fdf1d3578f56418856b58e1a603b5693d45c61 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:13:19 +0000 Subject: [PATCH 05/10] feat(agent-core): register and enable the Monitor tool Co-Authored-By: Kimi K2.7 Code --- packages/agent-core/src/agent/tool/index.ts | 1 + packages/agent-core/src/profile/default/agent.yaml | 1 + packages/agent-core/src/tools/builtin/index.ts | 1 + 3 files changed, 3 insertions(+) diff --git a/packages/agent-core/src/agent/tool/index.ts b/packages/agent-core/src/agent/tool/index.ts index 13a26eeba5..f0792cb299 100644 --- a/packages/agent-core/src/agent/tool/index.ts +++ b/packages/agent-core/src/agent/tool/index.ts @@ -742,6 +742,7 @@ export class ToolManager { new b.TaskListTool(background), new b.TaskOutputTool(background), new b.TaskStopTool(background), + allowBackground && new b.MonitorTool(kaos, cwd, background), this.agent.cron && new b.CronCreateTool(this.agent.cron), this.agent.cron && new b.CronListTool(this.agent.cron), this.agent.cron && new b.CronDeleteTool(this.agent.cron), diff --git a/packages/agent-core/src/profile/default/agent.yaml b/packages/agent-core/src/profile/default/agent.yaml index b85e327d40..f28bc63f87 100644 --- a/packages/agent-core/src/profile/default/agent.yaml +++ b/packages/agent-core/src/profile/default/agent.yaml @@ -15,6 +15,7 @@ tools: - TaskList - TaskOutput - TaskStop + - Monitor - CronCreate - CronList - CronDelete diff --git a/packages/agent-core/src/tools/builtin/index.ts b/packages/agent-core/src/tools/builtin/index.ts index b5c8e3bbdc..ad62539a08 100644 --- a/packages/agent-core/src/tools/builtin/index.ts +++ b/packages/agent-core/src/tools/builtin/index.ts @@ -1,6 +1,7 @@ export * from '../background/task-list'; export * from '../background/task-output'; export * from '../background/task-stop'; +export * from '../background/monitor'; export * from '../cron/cron-create'; export * from '../cron/cron-delete'; export * from '../cron/cron-list'; From 2ba66d995bfe540b1a4c458c7e5398cbf6863d98 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:20:19 +0000 Subject: [PATCH 06/10] docs(agent-core): document Monitor tool + changeset Co-Authored-By: Kimi K2.7 Code --- .changeset/monitor-tool.md | 6 ++++++ docs/en/reference/tools.md | 5 ++++- docs/zh/reference/tools.md | 5 ++++- 3 files changed, 14 insertions(+), 2 deletions(-) create mode 100644 .changeset/monitor-tool.md diff --git a/.changeset/monitor-tool.md b/.changeset/monitor-tool.md new file mode 100644 index 0000000000..1c1d85bcc0 --- /dev/null +++ b/.changeset/monitor-tool.md @@ -0,0 +1,6 @@ +--- +"@moonshot-ai/agent-core": minor +"@moonshot-ai/protocol": patch +--- + +Add a `Monitor` tool: run a self-filtering shell command in the background and receive each new stdout line as a notification (with batching and an auto-stop volume cap). Stop with `TaskStop` or let it time out; `persistent` watches run until the session ends. diff --git a/docs/en/reference/tools.md b/docs/en/reference/tools.md index 8854caefc6..da055b4b3f 100644 --- a/docs/en/reference/tools.md +++ b/docs/en/reference/tools.md @@ -99,13 +99,14 @@ Collaboration tools handle inter-Agent coordination, user interaction, and Skill ## Background Tasks -Background task tools manage tasks started via `Bash`, `Agent`, or `AskUserQuestion`. When a task reaches a terminal state, its status and saved output path are automatically delivered back to the Agent; use `TaskOutput` to check progress early. +Background task tools manage tasks started via `Bash`, `Agent`, `AskUserQuestion`, or `Monitor`. When a task reaches a terminal state, its status and saved output path are automatically delivered back to the Agent; use `TaskOutput` to check progress early. | Tool | Default Approval | Description | | --- | --- | --- | | `TaskList` | Auto-allow | List background tasks | | `TaskOutput` | Auto-allow | View the output of a background task | | `TaskStop` | Requires approval | Stop a running background task | +| `Monitor` | Requires approval | Run a self-filtering shell command and receive each stdout line as a notification | **`TaskList`** returns the list of background tasks. Optional parameters: `active_only` (defaults to true; lists only running tasks) and `limit` (defaults to 20; range 1–100). @@ -113,6 +114,8 @@ Background task tools manage tasks started via `Bash`, `Agent`, or `AskUserQuest **`TaskStop`** accepts a `task_id` and optional `reason` (defaults to `Stopped by TaskStop`). Safe to call on tasks that are already in a terminal state. +**`Monitor`** runs a self-filtering shell command in the background and delivers each matching stdout line as a real-time ``. Parameters: `command` (required; prefer line-buffered filters such as `grep --line-buffered` or `awk` with `fflush()`), `description` (shown in every notification), `timeout_ms` (defaults to 5 minutes; ignored when `persistent=true`), and `persistent` (run until the session ends or `TaskStop` is called). Output is batched over a short window and auto-stopped if it exceeds 200 lines per 5-second window; tighten the filter and restart if this happens. + ## Scheduled Tasks Scheduled task tools allow the Agent to re-inject a prompt into the current session at a future time — either as a one-time reminder or as a recurring cron-triggered task (periodic checks, daily reports, deployment monitoring, etc.). Schedules are bound to the session and remain active after `kimi resume`, but are not carried into a brand-new session. A single session can hold at most 50 active scheduled tasks. Set `KIMI_DISABLE_CRON=1` to disable them entirely; see [Environment Variables](../configuration/env-vars.md#运行时开关). diff --git a/docs/zh/reference/tools.md b/docs/zh/reference/tools.md index 810ffc6b97..d920080233 100644 --- a/docs/zh/reference/tools.md +++ b/docs/zh/reference/tools.md @@ -99,13 +99,14 @@ Plan 模式是一种受约束的工作状态：进入后 `Write` 与 `Edit` 只 ## 后台任务 -后台任务工具用于管理通过 `Bash`、`Agent` 或 `AskUserQuestion` 启动的后台任务。任务进入终止状态时会自动把状态和已保存的输出路径送回 Agent；如需提前检查进度，使用 `TaskOutput`。 +后台任务工具用于管理通过 `Bash`、`Agent`、`AskUserQuestion` 或 `Monitor` 启动的后台任务。任务进入终止状态时会自动把状态和已保存的输出路径送回 Agent；如需提前检查进度，使用 `TaskOutput`。 | 工具 | 默认审批 | 说明 | | --- | --- | --- | | `TaskList` | 自动放行 | 列出后台任务 | | `TaskOutput` | 自动放行 | 查看后台任务的输出 | | `TaskStop` | 需审批 | 停止正在运行的后台任务 | +| `Monitor` | 需审批 | 运行自过滤 shell 命令，将每条 stdout 作为通知推送 | **`TaskList`** 返回后台任务列表。可选参数 `active_only`（默认 true，仅列出运行中的任务）和 `limit`（默认 20，取值范围 1–100）。 @@ -113,6 +114,8 @@ Plan 模式是一种受约束的工作状态：进入后 `Write` 与 `Edit` 只 **`TaskStop`** 接受 `task_id` 和可选的 `reason`（默认 `Stopped by TaskStop`）。对已处于终止状态的任务也能安全调用。 +**`Monitor`** 在后台运行一条自过滤的 shell 命令，将每条匹配的 stdout 行作为实时的 `` 推送。参数：`command`（必填；建议使用行缓冲过滤器，如 `grep --line-buffered` 或带 `fflush()` 的 `awk`）、`description`（每条通知中显示）、`timeout_ms`（默认 5 分钟；`persistent=true` 时忽略）、`persistent`（持续到会话结束或调用 `TaskStop`）。输出会在一个短窗口内批量推送，若 5 秒内超过 200 行会自动停止；出现这种情况时请收紧过滤条件后重新启动。 + ## 定时任务 定时任务工具允许 Agent 把一段 prompt 在未来某个时间重新注入到当前会话——既可以是一次性提醒，也可以是按 cron 周期触发的任务（定期巡检、每日报表、部署监控等）。计划绑定到会话，执行 `kimi resume` 后仍然有效，但不会带入全新的会话。单个会话最多保留 50 个生效中的定时任务。设置 `KIMI_DISABLE_CRON=1` 可整体禁用，详见[环境变量](../configuration/env-vars.md#运行时开关)。 From cffc1f700e7fbbf970f9b3d85ba38ff5296895cd Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:24:57 +0000 Subject: [PATCH 07/10] refactor(agent-core): type monitor notifications cleanly; drop unused param Widen BackgroundTaskNotification.source_kind to 'background_task' | 'monitor' so monitorNotify builds a properly-typed notification and fires the hook without an 'as unknown as' cast. Also drop the now-unused parameter on MonitorBackgroundTask.checkVolume. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/agent-core/src/agent/background/index.ts | 10 +++++----- .../agent-core/src/agent/background/monitor-task.ts | 4 ++-- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/packages/agent-core/src/agent/background/index.ts b/packages/agent-core/src/agent/background/index.ts index 2fe15ae06e..89f35fbfb6 100644 --- a/packages/agent-core/src/agent/background/index.ts +++ b/packages/agent-core/src/agent/background/index.ts @@ -192,7 +192,7 @@ type BackgroundTaskNotification = Record & { readonly id: string; readonly category: 'task'; readonly type: string; - readonly source_kind: 'background_task'; + readonly source_kind: 'background_task' | 'monitor'; readonly source_id: string; /** Subagent id accepted by Agent(resume=...). Omitted for process tasks. */ readonly agent_id?: string | undefined; @@ -814,11 +814,11 @@ export class BackgroundManager { severity: 'info' | 'warning' = 'info', ): void { const body = lines.join('\n'); - const notification = { + const notification: BackgroundTaskNotification = { id: `monitor:${taskId}:${this.nextMonitorSeq(taskId)}`, - category: 'task' as const, + category: 'task', type: 'monitor_line', - source_kind: 'monitor' as const, + source_kind: 'monitor', source_id: taskId, title: description, severity, @@ -833,7 +833,7 @@ export class BackgroundManager { notificationId: `monitor:${taskId}`, }; this.agent.turn.steer([{ type: 'text', text }], origin); - this.fireNotificationHook(notification as unknown as BackgroundTaskNotification); + this.fireNotificationHook(notification); } private async restoreBackgroundTaskNotification(info: BackgroundTaskInfo): Promise { diff --git a/packages/agent-core/src/agent/background/monitor-task.ts b/packages/agent-core/src/agent/background/monitor-task.ts index 96db3bf0d3..252edd31e2 100644 --- a/packages/agent-core/src/agent/background/monitor-task.ts +++ b/packages/agent-core/src/agent/background/monitor-task.ts @@ -118,13 +118,13 @@ export class MonitorBackgroundTask implements BackgroundTask { this.pending = parts.pop() ?? ''; for (const line of parts) { this.batch.push(line); - this.checkVolume(line); + this.checkVolume(); if (this.capped) break; } this.scheduleBatchFlush(); } - private checkVolume(line: string): void { + private checkVolume(): void { const now = Date.now(); const windowMs = this.options.volumeWindowMs ?? DEFAULT_VOLUME_WINDOW_MS; if (now - this.volumeWindowStart > windowMs) { From 59f4f32ee799d6e43e62dc23773038a46d1d34ea Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:26:44 +0000 Subject: [PATCH 08/10] fix(agent-core): target publishable @moonshot-ai/kimi-code in changeset agent-core and protocol are private (ignored) packages per .changeset/README.md; the changeset must declare the bump on the publishable CLI package that ships the feature. Co-Authored-By: Claude Opus 4.8 (1M context) --- .changeset/monitor-tool.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/.changeset/monitor-tool.md b/.changeset/monitor-tool.md index 1c1d85bcc0..c01f713eb0 100644 --- a/.changeset/monitor-tool.md +++ b/.changeset/monitor-tool.md @@ -1,6 +1,5 @@ --- -"@moonshot-ai/agent-core": minor -"@moonshot-ai/protocol": patch +"@moonshot-ai/kimi-code": minor --- Add a `Monitor` tool: run a self-filtering shell command in the background and receive each new stdout line as a notification (with batching and an auto-stop volume cap). Stop with `TaskStop` or let it time out; `persistent` watches run until the session ends. From ac4561f462187090d078a0e702e4cf05e8b13183 Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Mon, 22 Jun 2026 22:34:50 +0000 Subject: [PATCH 09/10] fix(agent-core): clean up orphaned monitor proc on register failure; emit 'monitor' wire kind Address Codex review on #987: - MonitorTool now kills+disposes the spawned process if registerMonitorTask throws (e.g. maxRunningTasks reached), mirroring the Bash path, so a long-running command isn't orphaned outside the manager. - mapKind returns the first-class 'monitor' wire kind (added to the protocol enum) instead of collapsing to 'tool', so REST clients can distinguish Monitor tasks. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/agent-core/src/services/task/task.ts | 6 ++-- .../src/tools/background/monitor.ts | 34 ++++++++++++++++--- 2 files changed, 34 insertions(+), 6 deletions(-) diff --git a/packages/agent-core/src/services/task/task.ts b/packages/agent-core/src/services/task/task.ts index 0b428ce2bc..fc0284c027 100644 --- a/packages/agent-core/src/services/task/task.ts +++ b/packages/agent-core/src/services/task/task.ts @@ -58,8 +58,10 @@ function mapKind(k: BackgroundTaskInfo['kind']): BackgroundTaskKind { // execution), so 'tool' is the closest spec literal. return 'tool'; case 'monitor': - // Monitors are tool-spawned persistent background tasks. - return 'tool'; + // 'monitor' is a first-class wire kind (added to backgroundTaskKindSchema), + // so emit it directly — REST clients can then distinguish Monitor tasks + // from generic tool/question tasks. + return 'monitor'; } } diff --git a/packages/agent-core/src/tools/background/monitor.ts b/packages/agent-core/src/tools/background/monitor.ts index 22df263a8f..20a25a04b7 100644 --- a/packages/agent-core/src/tools/background/monitor.ts +++ b/packages/agent-core/src/tools/background/monitor.ts @@ -71,10 +71,22 @@ export class MonitorTool implements BuiltinTool { closeProcessStdin(proc); const timeoutMs = args.persistent ? undefined : args.timeout_ms; - const taskId = this.background.registerMonitorTask(proc, command, args.description, { - detached: true, - timeoutMs, - }); + let taskId: string; + try { + taskId = this.background.registerMonitorTask(proc, command, args.description, { + detached: true, + timeoutMs, + }); + } catch (error) { + // Registration can throw (e.g. maxRunningTasks reached). The process is + // already spawned and not yet owned by the manager, so clean it up here + // to avoid orphaning a long-running command — mirrors the Bash path. + await killSpawnedProcess(proc); + return { + isError: true, + output: error instanceof Error ? error.message : String(error), + }; + } return { isError: false, @@ -117,6 +129,20 @@ function closeProcessStdin(proc: KaosProcess): void { } } +async function killSpawnedProcess(proc: KaosProcess): Promise { + try { + await proc.kill('SIGTERM'); + } catch { + /* process already gone */ + } finally { + try { + await proc.dispose(); + } catch { + /* best-effort cleanup */ + } + } +} + function shellQuote(s: string): string { return `'${s.replaceAll("'", "'\\''")}'`; } From fb3cccb72462fad949a46a5d0d817be7db685f9b Mon Sep 17 00:00:00 2001 From: Nitjsefnie Date: Thu, 16 Jul 2026 18:34:44 +0000 Subject: [PATCH 10/10] feat(klient): mirror 'monitor' task kind in the wire contract The klient facade landed after this branch was cut and pins its wire `agentTaskInfoSchema` to the protocol `TaskInfo` union via the `AssertWire` parity check in `test/contract-parity.ts`. Adding the `monitor` variant to `TaskInfo` therefore broke that assertion. Mirror the new variant in the wire contract so the facade keeps parity with the protocol. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/klient/src/contract/agent/rpc.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/klient/src/contract/agent/rpc.ts b/packages/klient/src/contract/agent/rpc.ts index 1e68532e67..3bbf262eb2 100644 --- a/packages/klient/src/contract/agent/rpc.ts +++ b/packages/klient/src/contract/agent/rpc.ts @@ -169,6 +169,11 @@ export const agentTaskInfoSchema = z.discriminatedUnion('kind', [ toolCallId: z.string().optional(), ...taskInfoBaseFields, }), + z.object({ + kind: z.literal('monitor'), + command: z.string(), + ...taskInfoBaseFields, + }), ]); export const stopTaskPayloadSchema = z.object({