report-agent-state

1import type { EngineInterface, PluginOptions, Register } from 'claude-code'
2import { type Run, agentStates, anyEnvSet, reportStates, reportWithCommand } from './watch'
3
4// strings reads a `multiple` string option. The engine validates options
5// against the manifest before the module loads, so anything else is only an
6// unset option.
7function strings(v: PluginOptions[string] | undefined): readonly string[] {
8  return Array.isArray(v) ? v.filter(s => s !== '') : []
9}
10
11// Each read of the listing spawns a Claude Code process, which costs a good
12// part of a second, so the interval stays well above that.
13const POLL_MS = 2000
14
15// Whatever records the state may have restarted, or refused a report, and has
16// to hear the state again even when it did not change.
17const RESEND_MS = 10_000
18
19// A reporter that stays down would otherwise put a line in the debug log every
20// tick, so the first failure of a run is logged, then one in every WARN_EVERY:
21// at the 2s interval, one line every five minutes.
22const WARN_EVERY = 150
23
24function throttledLog($: EngineInterface) {
25  let failures = 0
26  return {
27    fail(message: string) {
28      failures++
29      if (failures === 1 || failures % WARN_EVERY === 0) {
30        $.ui.log(`report-agent-state: ${message} (failures: ${failures})`, { to: 'debug' })
31      }
32    },
33    ok() {
34      failures = 0
35    },
36  }
37}
38
39export const register: Register = (on, options) => {
40  const reportArgv = strings(options.reportCommand)
41  const requireEnv = strings(options.requireEnv)
42
43  on('session.start', async ($, e, next) => {
44    const result = await next(e)
45    if (reportArgv.length === 0) return result
46    const run: Run = (argv, init) => $.process.run(argv, init)
47    // A session the reporter has no identity for stays quiet instead of
48    // failing on every tick.
49    if (!(await anyEnvSet(run, requireEnv))) return result
50    const log = throttledLog($)
51    const states = agentStates({
52      every: (ms, fn) => $.clock.every(ms, fn),
53      sessionId: () => $.session.id(),
54      run,
55      intervalMs: POLL_MS,
56      onError: log.fail,
57    })
58    void reportStates(states, {
59      report: reportWithCommand(run, reportArgv),
60      now: () => $.clock.now(),
61      resendMs: RESEND_MS,
62      onError: log.fail,
63      onReported: log.ok,
64    })
65    return result
66  })
67}
68

1import { type HarnessState, stateOf } from './state'
2
3// The engine calls the helpers make. The engine refuses a $ passed across an
4// import, so register.tsx hands in closures over its own $ instead.
5export type Every = (ms: number, fn: () => void) => { cancel: () => void }
6export type Now = () => Promise<number>
7export type Run = (
8  argv: readonly string[],
9  init?: { timeoutMs?: number },
10) => Promise<{ exitCode: number; stdout: string; stderr: string }>
11
12// ticks yields once at once, then once per period. A period that ends while
13// the consumer is still busy with the last tick is folded into the next one,
14// so a slow listing never stacks reads up.
15//
16// The loop rides on $.clock.every rather than $.clock.sleep: a sleep is charged
17// to the budget of the hook that started it, while a timer runs until it is
18// cancelled or the module reloads.
19export async function* ticks(every: Every, ms: number): AsyncGenerator<void> {
20  let wake: (() => void) | undefined
21  let pending = false
22  const timer = every(ms, () => {
23    pending = true
24    wake?.()
25  })
26  try {
27    yield
28    for (;;) {
29      if (!pending) await new Promise<void>(resolve => (wake = resolve))
30      wake = undefined
31      pending = false
32      yield
33    }
34  } finally {
35    timer.cancel()
36  }
37}
38
39export type WatchOptions = {
40  every: Every
41  sessionId: () => Promise<string>
42  run: Run
43  intervalMs: number
44  onError?: (message: string) => void
45}
46
47// agentStates yields the session's state as `claude agents --json` shows it,
48// once per tick. The state is never inferred from this session's own events:
49// an interrupted turn fires nothing, and a background subagent, shell or
50// monitor keeps the session working after its main turn ended. The listing is
51// Claude Code's own account of both.
52//
53// The session id is read on every tick, since /clear, /resume and their kin
54// switch it inside the same process without loading the mod again. A tick
55// whose listing does not carry the session yields nothing.
56export async function* agentStates(opts: WatchOptions): AsyncGenerator<HarnessState> {
57  for await (const _ of ticks(opts.every, opts.intervalMs)) {
58    let state: HarnessState | undefined
59    try {
60      const sessionId = await opts.sessionId()
61      // Each read spawns a Claude Code process; one that hangs past the
62      // interval would otherwise hold every later tick.
63      const listed = await opts.run(['claude', 'agents', '--json'], { timeoutMs: opts.intervalMs })
64      if (listed.exitCode !== 0) {
65        opts.onError?.(`claude agents --json exited ${listed.exitCode}: ${listed.stderr.trim()}`)
66        continue
67      }
68      state = stateOf(listed.stdout, sessionId)
69    } catch (err) {
70      opts.onError?.(String(err))
71      continue
72    }
73    if (state !== undefined) yield state
74  }
75}
76
77// Report hands one state to whatever records it for the room, and rejects when
78// it did not get there.
79export type Report = (state: HarnessState) => Promise<void>
80
81const REPORT_TIMEOUT_MS = 5000
82
83// reportWithCommand runs argv with the state word appended, in the session's
84// environment, and counts a non-zero exit as a failed report.
85export function reportWithCommand(run: Run, argv: readonly string[]): Report {
86  return async (state) => {
87    const r = await run([...argv, state], { timeoutMs: REPORT_TIMEOUT_MS })
88    if (r.exitCode !== 0) {
89      throw new Error(`${[...argv, state].join(' ')} exited ${r.exitCode}: ${r.stderr.trim()}`)
90    }
91  }
92}
93
94// anyEnvSet reports whether the session's environment sets at least one of
95// names to a non-empty value; no names at all is a yes.
96//
97// It asks printenv rather than $.env.get: the engine takes only a string
98// literal as a variable name there, and these names come from the options.
99export async function anyEnvSet(run: Run, names: readonly string[]): Promise<boolean> {
100  if (names.length === 0) return true
101  for (const name of names) {
102    const r = await run(['printenv', name], { timeoutMs: REPORT_TIMEOUT_MS })
103    if (r.exitCode === 0 && r.stdout.trim() !== '') return true
104  }
105  return false
106}
107
108export type ReportOptions = {
109  report: Report
110  now: Now
111  resendMs: number
112  onError?: (message: string) => void
113  onReported?: () => void
114}
115
116// reportStates reports each state that differs from the last one reported, and
117// an unchanged one again once resendMs passed, so a daemon that restarted or
118// refused a report hears it again. A failed report is retried on the next
119// state, whatever it is.
120export async function reportStates(states: AsyncIterable<HarnessState>, opts: ReportOptions): Promise<void> {
121  let last: { state: HarnessState; at: number } | undefined
122  for await (const state of states) {
123    const now = await opts.now()
124    if (last?.state === state && now - last.at < opts.resendMs) continue
125    try {
126      await opts.report(state)
127      last = { state, at: now }
128      opts.onReported?.()
129    } catch (err) {
130      last = undefined
131      opts.onError?.(err instanceof Error ? err.message : String(err))
132    }
133  }
134}
135

1// The state words the report command is handed.
2export type HarnessState = 'working' | 'waiting' | 'done'
3
4// One entry of `claude agents --json`, the fields a state is read from.
5type ClaudeAgent = {
6  pid?: number
7  sessionId?: string
8  status?: string
9  state?: string
10}
11
12// stateOf picks the entry of sessionId out of a listing and maps it onto a
13// state word. It answers undefined when the listing does not carry
14// the session or says nothing it can map.
15//
16// An entry without a pid is never picked. Those come from the background job
17// records under <config home>/jobs/, which describe sessions running elsewhere
18// or not at all; a session resumed from a job leaves the job's record behind
19// under the same id, often with a stale blocked state.
20export function stateOf(listing: string, sessionId: string): HarnessState | undefined {
21  let entries: unknown
22  try {
23    entries = JSON.parse(listing)
24  } catch {
25    return undefined
26  }
27  if (!Array.isArray(entries)) return undefined
28  const entry = (entries as ClaudeAgent[]).find(
29    a => typeof a === 'object' && a !== null && !!a.pid && a.sessionId === sessionId,
30  )
31  return entry === undefined ? undefined : stateOfEntry(entry)
32}
33
34// status is what the session is doing at this moment and is read first. An
35// entry with no status is a session whose process the listing could not see,
36// and its state, the session's own account of its progress, is the fallback.
37function stateOfEntry(a: ClaudeAgent): HarnessState | undefined {
38  switch (a.status) {
39    case 'busy':
40      return 'working'
41    case 'waiting':
42      return 'waiting'
43    case 'idle':
44      return 'done'
45    case undefined:
46    case '':
47      break
48    default:
49      return undefined
50  }
51  switch (a.state) {
52    case 'working':
53      return 'working'
54    case 'blocked':
55      return 'waiting'
56    case 'done':
57    case 'failed':
58    case 'stopped':
59      return 'done'
60    default:
61      return undefined
62  }
63}
64