thread-pulse

1import type { EngineInterface, Register } from 'claude-code'
2
3import {
4  composeStatus,
5  DEFAULT_HEADER,
6  describeCall,
7  formatElapsed,
8  longCalls,
9  pulseLine,
10  type Running,
11} from './pulse'
12import { type CallDoc, docIdOf, exitCodeOf, tailOutput } from './watch'
13
14// In a Claude project thread the checklist the person sees is written by
15// the hearthbot server's update_status tool. Elsewhere it is absent and the
16// mod keeps to the status line.
17const HEARTH_SERVER = 'hearthbot'
18const HEARTH_STATUS = 'mcp__hearthbot__update_status'
19const TICK_MS = 15_000
20// Command Watch: how often running output is read and sent, and how often a
21// quiet call still says the device is alive.
22const WATCH_TICK_MS = 2_000
23const HEARTBEAT_MS = 15_000
24const MAX_READ = 4 * 1024 * 1024
25
26// Module state; a reload starts it over, which only drops the calls in flight.
27const running = new Map<string, Running>()
28let thresholdMs = 60_000
29let warnMs = 600_000
30// The last checklist the model wrote, without our lines.
31let base: string | undefined
32// What was last written to the thread, so a tick only writes on change.
33let pushed: string | undefined
34let statusLine: string | undefined
35let hasHearth: boolean | undefined
36let hearthCheckedAt = 0
37
38// One Bash call fed to the Command Watch page.
39type Watch = {
40  docId: string
41  // The command's letters and digits, as found in its shell's command line.
42  key: string
43  doc: CallDoc
44  file?: string
45  size: number
46  lastWriteAt: number
47  taskId?: string
48  isEnded: boolean
49  chain: Promise<void>
50}
51// Command Watch: one JSON file per call, read by command-watch-server.py.
52let isLocalWatch = true
53let localDir = ''
54let sessionId = ''
55let sessionLabel = ''
56// undefined: not looked for yet; null: looked for and not found.
57let tasksDir: string | null | undefined
58const claimed = new Set<string>()
59const watches = new Map<string, Watch>()
60// The page server: every session tries to start it, the first to bind the
61// port serves, and the others try again in a minute in case it went away.
62const SERVE_RETRY_MS = 60_000
63let isServeWatch = true
64let watchPort = 8765
65// 'off': not started; 'serving': this session's child holds the port;
66// 'elsewhere': another process holds it; 'no-python': python3 did not start.
67let serveState: 'off' | 'serving' | 'elsewhere' | 'no-python' = 'off'
68
69const alnum = (text: string): string => text.replace(/[^A-Za-z0-9]/g, '')
70
71const isWatching = (): boolean => localDir !== ''
72
73const num = (value: unknown, fallback: number): number =>
74  typeof value === 'number' && Number.isFinite(value) && value > 0 ? value : fallback
75
76async function findHearth($: EngineInterface, now: number): Promise<boolean> {
77  if (hasHearth === true) return true
78  if (hasHearth === false && now - hearthCheckedAt < 60_000) return false
79  hearthCheckedAt = now
80  const tools = await $.tool.list()
81  hasHearth = tools.some(t => t.name === HEARTH_STATUS)
82  return hasHearth
83}
84
85async function pushStatus($: EngineInterface, text: string): Promise<void> {
86  if (text === pushed) return
87  const res = await $.mcp.call(HEARTH_SERVER, 'update_status', { text })
88  if (!res.isError) pushed = text
89}
90
91async function tick($: EngineInterface): Promise<void> {
92  const now = await $.clock.now()
93  const long = longCalls(running.values(), now, thresholdMs)
94
95  const line =
96    long.length === 0
97      ? undefined
98      : `⏳ ${long.map(c => `${formatElapsed(now - c.startedAt)} ${c.label}`).join(' · ')}`
99  if (line !== statusLine) {
100    statusLine = line
101    $.ui.status(line)
102  }
103
104  if (!(await findHearth($, now))) return
105  if (long.length > 0) {
106    await pushStatus($, composeStatus(base, long.map(c => pulseLine(c, now, warnMs))))
107  } else if (pushed !== undefined && pushed !== (base ?? DEFAULT_HEADER)) {
108    // The long calls ended: put the model's own checklist back.
109    await pushStatus($, base ?? DEFAULT_HEADER)
110  }
111}
112
113// The engine writes each shell command's output to
114// <tmp>/claude-<uid>/<project>/<session>/tasks/<task>.output as it runs.
115async function findTasksDir($: EngineInterface): Promise<string | null> {
116  if (tasksDir !== undefined) return tasksDir
117  tasksDir = null
118  const tmp = (await $.env.get('CLAUDE_CODE_TMPDIR')) ?? (await $.env.get('TMPDIR')) ?? '/tmp'
119  const roots = (await $.fs.list(tmp).catch(() => [])).filter(
120    d => d.kind === 'dir' && d.name.startsWith('claude-'),
121  )
122  // The project folder is named after the session's directory.
123  const slug = (await $.session.root()).replace(/[^A-Za-z0-9]/g, '-')
124  for (const root of roots) {
125    const dir = `${tmp}/${root.name}/${slug}/${sessionId}/tasks`
126    if (await $.fs.exists(dir)) {
127      tasksDir = dir
128      return dir
129    }
130  }
131  for (const root of roots) {
132    for (const project of await $.fs.list(`${tmp}/${root.name}`).catch(() => [])) {
133      const dir = `${tmp}/${root.name}/${project.name}/${sessionId}/tasks`
134      if (project.kind === 'dir' && (await $.fs.exists(dir))) {
135        tasksDir = dir
136        return dir
137      }
138    }
139  }
140  return null
141}
142
143// Writes go one at a time per card, so a later state never lands first.
144function writeDoc($: EngineInterface, w: Watch, patch: Partial<CallDoc>): void {
145  Object.assign(w.doc, patch)
146  w.chain = w.chain
147    .then(async () => {
148      const now = await $.clock.now()
149      w.lastWriteAt = now
150      await $.fs.write(`${localDir}/${w.docId}.json`, JSON.stringify({ id: w.docId, ...w.doc, updatedAt: now }))
151    })
152    .catch(() => undefined)
153}
154
155async function readOutput($: EngineInterface, w: Watch): Promise<boolean> {
156  if (w.file === undefined) return false
157  const stat = await $.fs.stat(w.file).catch(() => undefined)
158  if (stat === undefined || stat.size === w.size) return false
159  w.size = stat.size
160  if (stat.size > MAX_READ) {
161    writeDoc($, w, { output: `(output is ${Math.round(stat.size / 1048576)} MiB, too large to show live)`, outputTruncated: true })
162    return true
163  }
164  const text = await $.fs.read(w.file)
165  writeDoc($, w, { ...tailOutput(text), lastOutputAt: await $.clock.now() })
166  return true
167}
168
169// Each running shell has its output file open as stdout; its command line
170// holds the command, so the two are matched through /proc (Linux).
171const SCAN = 'for p in /proc/[0-9]*; do f=$(readlink "$p/fd/1" 2>/dev/null) || continue; ' +
172  'case "$f" in */tasks/*.output) printf "%s\\t" "$f"; tr "\\0\\n" "  " < "$p/cmdline"; echo;; esac; done'
173
174async function claimFiles($: EngineInterface): Promise<void> {
175  const waiting = [...watches.values()].filter(w => w.file === undefined && !w.isEnded)
176  if (waiting.length === 0) return
177  // Only this session's own output files: another session's shell may
178  // mention the same command (claude -p "run X").
179  const dir = await findTasksDir($)
180  if (dir === null) return
181  const scan = await $.process.run(['sh', '-c', SCAN]).catch(() => undefined)
182  if (scan === undefined) return
183  const shells = scan.stdout
184    .split('\n')
185    .map(line => line.split('\t'))
186    .filter((parts): parts is [string, string] => parts.length === 2 && (parts[0] ?? '').startsWith(`${dir}/`))
187  for (const w of waiting) {
188    const hit = shells.find(([file, cmdline]) => !claimed.has(file) && alnum(cmdline).includes(w.key))
189    if (hit === undefined) continue
190    claimed.add(hit[0])
191    w.file = hit[0]
192  }
193}
194
195async function watchTick($: EngineInterface): Promise<void> {
196  if (!isWatching() || watches.size === 0) return
197  await claimFiles($)
198  const now = await $.clock.now()
199  for (const w of watches.values()) {
200    if (w.isEnded) continue
201    const changed = await readOutput($, w)
202    if (!changed && now - w.lastWriteAt > HEARTBEAT_MS) writeDoc($, w, {})
203  }
204}
205
206async function endWatch($: EngineInterface, key: string, w: Watch, patch: Partial<CallDoc>): Promise<void> {
207  w.isEnded = true
208  w.size = -1
209  await readOutput($, w)
210  writeDoc($, w, { ...patch, endedAt: await $.clock.now() })
211  watches.delete(key)
212}
213
214async function serveWatch($: EngineInterface): Promise<void> {
215  const script = `${$.plugin.root}/command-watch/command-watch-server.py`
216  for (;;) {
217    try {
218      const child = $.process.spawn({ argv: ['python3', script, '--port', String(watchPort)] })
219      // The server prints its address once it holds the port, then nothing.
220      for await (const { stream, text } of child) {
221        if (stream === 'stdout' && text.includes('Command Watch on')) serveState = 'serving'
222      }
223    } catch {
224      serveState = 'no-python'
225      return
226    }
227    // The port was taken, or our server stopped: someone else may serve now.
228    serveState = 'elsewhere'
229    await $.clock.sleep(SERVE_RETRY_MS)
230  }
231}
232
233export const register: Register = (on, options) => {
234  thresholdMs = num(options.thresholdSeconds, 60) * 1000
235  warnMs = num(options.warnMinutes, 10) * 60_000
236  isLocalWatch = options.localWatch !== false
237  isServeWatch = options.serveWatch !== false
238  watchPort = num(options.watchPort, 8765)
239
240  on('session.start', async ($, e, next) => {
241    sessionId = await $.session.id()
242    const home = await $.env.get('HOME')
243    localDir = isLocalWatch && home !== undefined ? `${home}/.claude/command-watch/calls` : ''
244    const host = await $.process.run(['hostname']).then(r => r.stdout.trim()).catch(() => '')
245    const dir = e.cwd.split('/').filter(Boolean).at(-1) ?? e.cwd
246    sessionLabel = `${host === '' ? '' : `${host} · `}${dir} · ${sessionId.slice(0, 8)}`
247    $.clock.every(TICK_MS, () => {
248      tick($).catch(() => undefined)
249    })
250    $.clock.every(WATCH_TICK_MS, () => {
251      watchTick($).catch(() => undefined)
252    })
253    if (isWatching() && isServeWatch) void serveWatch($)
254    // A second copy of the mod may hold the names; the mod runs without them.
255    await $.command
256      .register({
257        name: 'pulse',
258        description: 'List the tool calls running now and how long each has run',
259      })
260      .catch(() => undefined)
261    await $.command
262      .register({
263        name: 'command-watch',
264        description: 'Show where the Command Watch page is served',
265      })
266      .catch(() => undefined)
267    return next(e)
268  })
269
270  on('command.run', { command: 'pulse' }, async $ => {
271    const now = await $.clock.now()
272    const all = longCalls(running.values(), now, 0, 50)
273    if (all.length === 0) return { text: 'No tool call is running.' }
274    return {
275      text: all
276        .map(c => `${formatElapsed(now - c.startedAt)}  ${c.isSubagent ? 'subagent ' : ''}${c.label}`)
277        .join('\n'),
278    }
279  })
280
281  on('command.run', { command: 'command-watch' }, async $ => {
282    if (!isWatching()) return { text: 'Command Watch is off (option localWatch).' }
283    const url = `http://127.0.0.1:${watchPort}/`
284    const start = `python3 ${$.plugin.root}/command-watch/command-watch-server.py --port ${watchPort}`
285    switch (serveState) {
286      case 'serving':
287        return { text: `Command Watch: ${url} (served by this session)` }
288      case 'elsewhere':
289        return { text: `Command Watch: ${url} (served by another session or process)` }
290      case 'no-python':
291        return { text: `Command Watch could not start: python3 was not found. Start it yourself:\n${start}` }
292      default:
293        return { text: `Command Watch is not served by the mod (option serveWatch). Start it yourself:\n${start}\nthen open ${url}` }
294    }
295  })
296
297  // The model's own checklist writes: keep the latest as the base.
298  on('tool.call', { tool: HEARTH_STATUS }, async ($, e, next) => {
299    const text = (e as { text?: unknown }).text
300    if (typeof text === 'string') base = text
301    const ran = await next(e)
302    if (typeof text === 'string' && ran.deny === undefined && ran.isError !== true) pushed = text
303    return ran
304  })
305
306  // Every shell command becomes a card on the Command Watch page.
307  on('tool.call', { tool: 'Bash' }, async ($, e, next) => {
308    if (!isWatching()) return next(e)
309    const now = await $.clock.now()
310    const w: Watch = {
311      docId: docIdOf(sessionId, e.tool_use_id),
312      key: alnum(e.command).slice(0, 80),
313      size: -1,
314      lastWriteAt: now,
315      isEnded: false,
316      chain: Promise.resolve(),
317      doc: {
318        session: sessionId,
319        sessionLabel,
320        tool: 'Bash',
321        command: tailOutput(e.command).output,
322        description: e.description ?? '',
323        timeoutMs: e.timeout ?? (e.run_in_background === true ? 1_800_000 : 120_000),
324        background: e.run_in_background === true,
325        startedAt: now,
326        updatedAt: now,
327        lastOutputAt: now,
328        endedAt: null,
329        status: 'running',
330        exitCode: null,
331        output: '',
332        outputTruncated: false,
333      },
334    }
335    watches.set(e.tool_use_id, w)
336    writeDoc($, w, {})
337
338    const ran = await next(e).catch((err: unknown) => {
339      void endWatch($, e.tool_use_id, w, { status: 'interrupted' })
340      throw err
341    })
342    if (ran.deny !== undefined) {
343      await endWatch($, e.tool_use_id, w, { status: 'failed', output: ran.deny })
344      await w.chain
345      return ran
346    }
347    const r = ran.result as Partial<{ stdout: string; stderr: string; interrupted: boolean; backgroundTaskId: string }>
348    const taskId = ran.isError === true ? undefined : r?.backgroundTaskId
349    if (taskId !== undefined) {
350      // Moved to the background (asked for, Ctrl+B or a timeout): the task's
351      // own file keeps growing until its notification arrives.
352      const dir = await findTasksDir($)
353      w.taskId = taskId
354      if (dir !== null) w.file = `${dir}/${taskId}.output`
355      if (w.file !== undefined) claimed.add(w.file)
356      writeDoc($, w, { background: true })
357      await w.chain
358      return ran
359    }
360    if (w.file === undefined) {
361      writeDoc($, w, tailOutput([r?.stdout, r?.stderr].filter(Boolean).join('\n') || (ran.text ?? '')))
362    }
363    await endWatch($, e.tool_use_id, w, {
364      status: r?.interrupted === true ? 'interrupted' : ran.isError === true ? 'failed' : 'done',
365      exitCode: ran.isError === true ? exitCodeOf(ran.text) : r?.interrupted === true ? null : 0,
366    })
367    await w.chain
368    return ran
369  })
370
371  // A background command ends with a task notification naming its id.
372  on('session.append', async ($, e, next) => {
373    if (isWatching()) {
374      const text = JSON.stringify(e.message.content ?? '')
375      for (const [key, w] of watches) {
376        if (w.taskId === undefined || !text.includes(`<task-id>${w.taskId}</task-id>`)) continue
377        const status = /<status>(\w+)<\/status>/.exec(text)?.[1] ?? ''
378        if (status === '' || status === 'running') continue
379        const code = /exit code (\d+)/i.exec(text)
380        await endWatch($, key, w, {
381          status: status === 'completed' ? 'done' : status === 'killed' ? 'interrupted' : 'failed',
382          exitCode: code === null ? null : Number(code[1]),
383        })
384        await w.chain
385      }
386    }
387    return next(e)
388  })
389
390  on('tool.call', async ($, e, next) => {
391    if (e.tool.startsWith('mcp__hearthbot__')) return next(e)
392    const { tool, tool_use_id, agentId, ...args } = e as {
393      tool: string
394      tool_use_id: string
395      agentId?: string
396    } & Record<string, unknown>
397    running.set(tool_use_id, {
398      tool,
399      label: describeCall(tool, args),
400      startedAt: await $.clock.now(),
401      isSubagent: agentId !== undefined,
402    })
403    try {
404      return await next(e)
405    } finally {
406      running.delete(tool_use_id)
407      // Clear the line now rather than at the next tick.
408      if (statusLine !== undefined || (pushed !== undefined && pushed !== (base ?? DEFAULT_HEADER))) {
409        await tick($).catch(() => undefined)
410      }
411    }
412  })
413}
414

1// Pure helpers: what a running call is called, how long it ran, and the
2// checklist text with the "still running" lines folded in.
3
4export type Running = {
5  tool: string
6  label: string
7  startedAt: number
8  isSubagent: boolean
9}
10
11export const DEFAULT_HEADER = 'Thread activity'
12
13const MAX_LABEL = 70
14
15const clip = (text: string): string => {
16  const line = text.replace(/\s+/g, ' ').trim().replace(/`/g, "'")
17  return line.length > MAX_LABEL ? `${line.slice(0, MAX_LABEL - 1)}…` : line
18}
19
20const firstString = (args: Record<string, unknown>): string | undefined => {
21  for (const key of ['description', 'command', 'url', 'file_path', 'pattern', 'prompt', 'query']) {
22    const value = args[key]
23    if (typeof value === 'string' && value.trim() !== '') return value
24  }
25  return undefined
26}
27
28// "Bash: k3d cluster start lab-infra", "subagent: Review the diff",
29// "remote-devices device_bash: helm upgrade ...".
30export const describeCall = (tool: string, args: Record<string, unknown>): string => {
31  if (tool === 'Agent' || tool === 'Task') {
32    const what = firstString(args)
33    return what === undefined ? 'subagent' : `subagent: ${clip(what)}`
34  }
35  const mcp = /^mcp__(.+?)__(.+)$/.exec(tool)
36  const name = mcp === null ? tool : `${mcp[1]} ${mcp[2]}`
37  // For a shell call the command says more than the model's description.
38  const what =
39    typeof args.command === 'string' && args.command.trim() !== ''
40      ? args.command
41      : firstString(args)
42  return what === undefined ? name : `${name}: ${clip(what)}`
43}
44
45export const formatElapsed = (ms: number): string => {
46  const s = Math.max(0, Math.floor(ms / 1000))
47  if (s < 60) return `${s}s`
48  const m = Math.floor(s / 60)
49  if (m < 60) return `${m}m${String(s % 60).padStart(2, '0')}s`
50  return `${Math.floor(m / 60)}h${String(m % 60).padStart(2, '0')}m`
51}
52
53// Minutes only, so the thread checklist changes at most once a minute.
54export const formatMinutes = (ms: number): string => {
55  const m = Math.max(1, Math.floor(ms / 60000))
56  if (m < 60) return `${m} min`
57  return `${Math.floor(m / 60)} h ${m % 60} min`
58}
59
60// The calls past the threshold, oldest first, at most `limit`.
61export const longCalls = (
62  running: Iterable<Running>,
63  now: number,
64  thresholdMs: number,
65  limit = 3,
66): Running[] =>
67  [...running]
68    .filter(call => now - call.startedAt >= thresholdMs)
69    .sort((a, b) => a.startedAt - b.startedAt)
70    .slice(0, limit)
71
72export const pulseLine = (call: Running, now: number, warnMs: number): string => {
73  const age = now - call.startedAt
74  const who = call.isSubagent ? 'subagent ' : ''
75  return age >= warnMs
76    ? `✱ ⚠ No result after ${formatMinutes(age)}, may be stuck: ${who}${call.label}`
77    : `✱ Still running (${formatMinutes(age)}): ${who}${call.label}`
78}
79
80// The model's checklist with the pulse lines placed right after its current
81// step (the last ✱ line), or at the end when it has none.
82export const composeStatus = (base: string | undefined, lines: string[]): string => {
83  const rows = (base ?? DEFAULT_HEADER).split('\n')
84  if (lines.length === 0) return rows.join('\n')
85  let at = -1
86  rows.forEach((row, i) => {
87    if (row.replace(/^\*\*/, '').trimStart().startsWith('✱')) at = i
88  })
89  if (at === -1) {
90    const tail = rows.length === 1 ? ['', ...lines] : lines
91    return [...rows, ...tail].join('\n')
92  }
93  return [...rows.slice(0, at + 1), ...lines, ...rows.slice(at + 1)].join('\n')
94}
95

1// Pure helpers for the Command Watch page: what a card holds, masking
2// secrets, cutting long output.
3
4export const OUTPUT_KEEP = 16_000
5
6// One Bash call as the page reads it (one JSON file per call).
7export type CallDoc = {
8  session: string
9  sessionLabel: string
10  tool: string
11  command: string
12  description: string
13  timeoutMs: number
14  background: boolean
15  startedAt: number
16  updatedAt: number
17  lastOutputAt: number
18  endedAt: number | null
19  status: 'running' | 'done' | 'failed' | 'interrupted'
20  exitCode: number | null
21  output: string
22  outputTruncated: boolean
23}
24
25const hasMixedBase64 = (s: string): boolean => /[A-Z]/.test(s) && /[a-z]/.test(s) && /\d/.test(s)
26
27// Best effort: keys, tokens and `password: x` style values become ***.
28export const redact = (text: string): string =>
29  text
30    .replace(/-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?(-----END [A-Z ]*PRIVATE KEY-----|$)/g, '[private key masked]')
31    .replace(/AGE-SECRET-KEY-1[0-9A-Z]+/g, 'AGE-SECRET-KEY-***')
32    .replace(/\b(gh[pousr]_[A-Za-z0-9]{20,}|github_pat_[A-Za-z0-9_]{20,}|AKIA[0-9A-Z]{16}|xox[abposr]-[A-Za-z0-9-]{10,})\b/g, '***')
33    .replace(/(Bearer\s+)[A-Za-z0-9._~+/-]{8,}=*/gi, '$1***')
34    .replace(
35      /((?:password|passwd|pass|secret|token|api[_-]?key|access[_-]?key|secret[_-]?key|client[_-]?secret|private[_-]?key)["']?\s*[:=]\s*["']?)([^\s"',}]{3,})/gi,
36      '$1***',
37    )
38    .replace(/[A-Za-z0-9+/]{40,}={0,2}/g, m => (hasMixedBase64(m) ? '[masked]' : m))
39
40// Terminal output as a person would see it: no colour codes, and a line
41// redrawn with \r (a progress bar) shows its last state.
42export const cleanTerminal = (text: string): string =>
43  text
44    // eslint-disable-next-line no-control-regex
45    .replace(/\u001b\[[0-9;?]*[ -/]*[@-~]|\u001b\][^\u0007]*\u0007/g, '')
46    .split('\n')
47    .map(line => {
48      const parts = line.split('\r')
49      for (let i = parts.length - 1; i >= 0; i--) if (parts[i] !== '') return parts[i]
50      return ''
51    })
52    .join('\n')
53
54export const tailOutput = (text: string): { output: string; outputTruncated: boolean } => {
55  const clean = redact(cleanTerminal(text))
56  if (clean.length <= OUTPUT_KEEP) return { output: clean, outputTruncated: false }
57  const cut = clean.slice(-OUTPUT_KEEP)
58  const nl = cut.indexOf('\n')
59  return { output: nl >= 0 && nl < 400 ? cut.slice(nl + 1) : cut, outputTruncated: true }
60}
61
62// "Exit code 2" at the head of a failed Bash result.
63export const exitCodeOf = (text: string | undefined): number | null => {
64  const m = /Exit code (\d+)/.exec(text ?? '')
65  return m === null ? null : Number(m[1])
66}
67
68export const docIdOf = (session: string, toolUseId: string): string =>
69  `${session.slice(0, 8)}-${toolUseId}`.replace(/[^A-Za-z0-9_\-.~:@+]/g, '_').slice(0, 200)
70