persistent-monitor

1type Watch = {
2  id: string
3  description: string
4  stop: (reason: string) => void
5}
6
7type Io = {
8  spawn: (r: { argv: string[] }) => any
9  after: (ms: number, fn: () => void) => any
10  submit: (text: string) => Promise<unknown>
11}
12
13const DEFAULT_TIMEOUT_MS = 300_000
14const BATCH_MS = 200
15const MAX_BATCH_CHARS = 8_000
16// Models tend to poll or block after starting a watch, which only delays delivery
17const DELIVERY_NOTE =
18  'Events arrive between turns as new messages from this plugin, never as tool results. ' +
19  'After starting a watch, continue with other work or end your turn. Do not poll for events (ReadNotifications, sleep, tail or wait loops) ' +
20  'and do not wait inside the turn: waiting does not speed delivery up and only delays it.'
21const RESULT_NOTE = 'Events arrive as messages after this turn; do not poll or wait for them, just continue or end your turn.'
22
23export function register(on: any) {
24  const watches = new Map<string, Watch>()
25  let nextId = 1
26
27  on('session.start', async ($: any, e: any, next: any) => {
28    await $.tool.register({
29      name: 'monitor',
30      description:
31        'Start a background monitor that streams events from a long-running shell command. Each stdout line is an event: ' +
32        'you keep working and notifications arrive as messages from this plugin. ' + DELIVERY_NOTE + ' Unlike the built-in Monitor, a watch with ' +
33        '`persistent: true` has no deadline and runs until `monitor_stop` or the session ends, so you never need to re-arm it.\n\n' +
34        'Pick by how many notifications you need:\n' +
35        '- One ("tell me when the build finishes") -> use Bash with run_in_background and a command that exits when the condition is true.\n' +
36        '- One per occurrence -> this tool, with an unbounded command (`tail -f`, `inotifywait -m`, `while true`).\n\n' +
37        'Script quality: every pipe stage must flush per line (`grep --line-buffered`, awk `fflush()`); handle transient failures in poll loops ' +
38        '(`curl ... || true`); poll 30s+ for remote APIs. Only stdout is the event stream; merge stderr with `2>&1` if failures should reach you. ' +
39        'Silence is not success: the filter must match failure/terminal states too. Keep output selective, since every line is a message; ' +
40        'a monitor that floods is stopped automatically. Write a specific `description`: it appears in every notification. ' +
41        'The command runs via `bash -c` in the session working directory.',
42      inputSchema: {
43        type: 'object',
44        properties: {
45          command: { type: 'string', description: 'Shell command or script. Each stdout line is an event; exit ends the watch.' },
46          description: { type: 'string', description: 'Short human-readable description of what you are monitoring (shown in notifications).' },
47          persistent: {
48            type: 'boolean',
49            default: false,
50            description: 'Run for the whole session with no deadline, until monitor_stop. Ignores timeout_ms.',
51          },
52          timeout_ms: {
53            type: 'number',
54            default: DEFAULT_TIMEOUT_MS,
55            minimum: 1000,
56            description: 'Kill the monitor after this deadline (default 300000). Ignored when persistent is true.',
57          },
58        },
59        required: ['command', 'description'],
60        additionalProperties: false,
61      },
62    })
63    await $.tool.register({
64      name: 'monitor_stop',
65      description: 'Stop a monitor started with the monitor tool, by its id. With no id, lists the running monitors.',
66      inputSchema: {
67        type: 'object',
68        properties: { id: { type: 'string', description: 'Monitor id, e.g. "pm-1".' } },
69        additionalProperties: false,
70      },
71    })
72    await $.tool.register({
73      name: 'waitpid',
74      description:
75        'Wait for a process to exit. Returns immediately; you get one message when process `pid` ends (no deadline). ' +
76        'Use this instead of polling `ps`. Cancel with monitor_stop. ' + DELIVERY_NOTE,
77      inputSchema: {
78        type: 'object',
79        properties: {
80          pid: { type: 'integer', description: 'Process id to wait for.' },
81          description: { type: 'string', description: 'What the process is, shown in the notification.' },
82        },
83        required: ['pid'],
84        additionalProperties: false,
85      },
86    })
87    await $.tool.register({
88      name: 'waitfile',
89      description:
90        'Follow a file (`tail -F`) and get a message for each line appended to it, optionally only lines matching a regex. ' +
91        'The file may not exist yet. Set `once` to stop after the first (matching) line. Runs with no deadline by default; cancel with monitor_stop. ' + DELIVERY_NOTE,
92      inputSchema: {
93        type: 'object',
94        properties: {
95          path: { type: 'string', description: 'File to follow.' },
96          pattern: { type: 'string', description: 'JavaScript regex; only matching lines are events.' },
97          once: { type: 'boolean', default: false, description: 'Stop after the first event.' },
98          from_start: { type: 'boolean', default: false, description: 'Also report lines already in the file.' },
99          persistent: { type: 'boolean', default: true, description: 'No deadline. Set false to use timeout_ms.' },
100          timeout_ms: { type: 'number', default: DEFAULT_TIMEOUT_MS, minimum: 1000, description: 'Deadline when persistent is false.' },
101          description: { type: 'string', description: 'Shown in notifications.' },
102        },
103        required: ['path'],
104        additionalProperties: false,
105      },
106    })
107    return next(e)
108  })
109
110  type WatchOpts = {
111    argv: string[]
112    description: string
113    persistent: boolean
114    timeoutMs: number
115    kind?: string
116    // Keep only lines this accepts
117    filter?: (line: string) => boolean
118    // Stop after the first event
119    once?: boolean
120    // Replaces the generic "ended" text on a natural exit
121    endText?: (exit: { code: number | null; signal: string | null } | undefined) => string
122  }
123
124  function startWatch(io: Io, o: WatchOpts) {
125    const { description, persistent, timeoutMs } = o
126    const id = 'pm-' + nextId++
127    const stream = io.spawn({ argv: o.argv })
128    let stopReason: string | null = null
129    let timer: any = null
130    let events = 0
131
132    const watch: Watch = {
133      id,
134      description,
135      stop: (reason: string) => {
136        if (stopReason === null) stopReason = reason
137        // Leaving the loop kills the child
138        void stream.return?.(undefined as any).catch(() => {})
139      },
140    }
141    watches.set(id, watch)
142    if (!persistent) timer = io.after(timeoutMs, () => watch.stop(`timed out after ${Math.round(timeoutMs / 1000)}s`))
143
144    const notify = (text: string) => {
145      io.submit(text).catch(() => {})
146    }
147
148    void (async () => {
149      let buf = ''
150      let pending: string[] = []
151      let flushTimer: any = null
152      let size = 0
153      const flush = () => {
154        flushTimer = null
155        if (!pending.length) return
156        const lines = pending
157        pending = []
158        size = 0
159        events += lines.length
160        notify(`[${o.kind ?? 'monitor'} ${id}: ${description}]\n` + lines.join('\n'))
161      }
162      const push = (line: string) => {
163        if (o.filter && !o.filter(line)) return
164        if (size > MAX_BATCH_CHARS) return
165        size += line.length
166        pending.push(size > MAX_BATCH_CHARS ? '[output truncated]' : line)
167        if (!flushTimer) flushTimer = io.after(BATCH_MS, flush)
168        if (o.once) watch.stop('first event received')
169      }
170      let exit: { code: number | null; signal: string | null } | undefined
171      try {
172        while (true) {
173          const r = await stream.next()
174          if (r.done) {
175            exit = r.value
176            break
177          }
178          if (r.value.stream !== 'stdout') continue
179          buf += r.value.text
180          const parts = buf.split('\n')
181          buf = parts.pop() ?? ''
182          for (const l of parts) if (l) push(l)
183        }
184      } catch (err: any) {
185        stopReason ??= 'failed: ' + (err?.message ?? err)
186      }
187      if (buf) push(buf)
188      flushTimer?.cancel?.()
189      flush()
190      timer?.cancel?.()
191      watches.delete(id)
192      const label = `[${o.kind ?? 'monitor'} ${id}: ${description}]`
193      if (stopReason === null && o.endText) {
194        notify(`${label} ${o.endText(exit)}`)
195      } else {
196        const how = stopReason ?? (exit ? `exited with code ${exit.code ?? exit.signal}` : 'ended')
197        notify(`${label} ended: ${how} (${events} event${events === 1 ? '' : 's'}). Re-arm if you still need the watch.`)
198      }
199    })()
200    return id
201  }
202
203  on('tool.call', { tool: 'mcp__persistent-monitor__monitor' }, async ($: any, e: any) => {
204    const io: Io = {
205      spawn: r => $.process.spawn(r),
206      after: (ms, fn) => $.clock.after(ms, fn),
207      submit: text => $.prompt.submit({ text }),
208    }
209    if (typeof e.command !== 'string' || !e.command.trim()) return { result: 'Error: command is required', isError: true }
210    const description = String(e.description ?? e.command).slice(0, 200)
211    const persistent = e.persistent === true
212    const timeoutMs = Number.isFinite(e.timeout_ms) ? Math.max(1000, e.timeout_ms) : DEFAULT_TIMEOUT_MS
213    const id = startWatch(io, { argv: ['bash', '-c', e.command], description, persistent, timeoutMs })
214    return {
215      result: persistent
216        ? `Monitor ${id} started: ${description}. persistent: runs until monitor_stop or session end. ${RESULT_NOTE}`
217        : `Monitor ${id} started: ${description}. Expires in ${Math.round(timeoutMs / 1000)}s. ${RESULT_NOTE}`,
218    }
219  })
220
221  on('tool.call', { tool: 'mcp__persistent-monitor__waitpid' }, async ($: any, e: any) => {
222    const io: Io = {
223      spawn: r => $.process.spawn(r),
224      after: (ms, fn) => $.clock.after(ms, fn),
225      submit: text => $.prompt.submit({ text }),
226    }
227    const pid = Number(e.pid)
228    if (!Number.isInteger(pid) || pid <= 0) return { result: 'Error: pid must be a positive integer', isError: true }
229    const description = String(e.description ?? `process ${pid}`).slice(0, 200)
230    // tail exits when the process does; nothing is written to /dev/null
231    const id = startWatch(io, {
232      argv: ['tail', `--pid=${pid}`, '-f', '/dev/null'],
233      description,
234      kind: 'waitpid',
235      persistent: true,
236      timeoutMs: 0,
237      endText: () => `process ${pid} has exited.`,
238    })
239    return { result: `waitpid ${id} started: ${description}. You will get one message when process ${pid} exits (no deadline; monitor_stop to cancel). ${RESULT_NOTE}` }
240  })
241
242  on('tool.call', { tool: 'mcp__persistent-monitor__waitfile' }, async ($: any, e: any) => {
243    const io: Io = {
244      spawn: r => $.process.spawn(r),
245      after: (ms, fn) => $.clock.after(ms, fn),
246      submit: text => $.prompt.submit({ text }),
247    }
248    if (typeof e.path !== 'string' || !e.path) return { result: 'Error: path is required', isError: true }
249    let re: RegExp | undefined
250    if (e.pattern) {
251      try {
252        re = new RegExp(e.pattern)
253      } catch (err: any) {
254        return { result: 'Error: invalid pattern: ' + err.message, isError: true }
255      }
256    }
257    const once = e.once === true
258    const description = String(e.description ?? `${e.path}${re ? ` /${e.pattern}/` : ''}`).slice(0, 200)
259    // -F follows by name and retries, so the file may not exist yet or may be rotated
260    const id = startWatch(io, {
261      argv: ['tail', '-F', '-n', e.from_start === true ? '+1' : '0', '--', e.path],
262      description,
263      kind: 'waitfile',
264      persistent: e.persistent !== false,
265      timeoutMs: Number.isFinite(e.timeout_ms) ? Math.max(1000, e.timeout_ms) : DEFAULT_TIMEOUT_MS,
266      filter: re ? l => re!.test(l) : undefined,
267      once,
268    })
269    return {
270      result: `waitfile ${id} started: ${description}. ${once ? 'One message on the first' : 'A message for each'} ${re ? 'matching ' : ''}line appended to ${e.path}. ${e.persistent !== false ? 'No deadline' : 'Has a deadline'}; monitor_stop to cancel. ${RESULT_NOTE}`,
271    }
272  })
273
274  on('tool.call', { tool: 'mcp__persistent-monitor__monitor_stop' }, async ($: any, e: any) => {
275    if (!e.id) {
276      const list = [...watches.values()].map(w => `${w.id}: ${w.description}`)
277      return { result: list.length ? list.join('\n') : 'No monitors running.' }
278    }
279    const w = watches.get(e.id)
280    if (!w) return { result: `No running monitor with id ${e.id}`, isError: true }
281    w.stop('stopped by monitor_stop')
282    return { result: `Stopped ${e.id}.` }
283  })
284}
285