usage-feed

1// usage-feed: keeps each Claude Code session's state from its events and sends it, with the plan's usage, to
2// upMonitor over a Unix socket inside upMonitor's own folder. Every hook only observes: it records what it saw and
3// returns what the hooks beneath returned. The classic hooks carry a `.catch` that only passes the event on, so one
4// that fails (or overruns its time) leaves the event as the hooks beneath made it; the others are not gating.
5import type { Register, EngineInterface } from 'claude-code'
6import type { UsageFeedSnap } from '../types'
7import {
8  HEARTBEAT_MS,
9  SEND_TIMEOUT_MS,
10  afterClear,
11  applyWord,
12  buildEnd,
13  buildRecord,
14  buildUpdate,
15  candidateIds,
16  emptySnap,
17  socketPath,
18  type Word,
19} from './model'
20
21const SNAP = { plugin: 'usage-feed', key: 'snap' } as const
22
23type Step = (snap: UsageFeedSnap, nowS: number) => UsageFeedSnap | null
24type Result = { status?: number; error?: string }
25
26const agentOf = (e: unknown): string | undefined => {
27  const a = (e as { agent_id?: unknown }).agent_id
28  return typeof a === 'string' && a !== '' ? a : undefined
29}
30
31// The main thread's effort level from an event that carries it
32const effortStep = (e: unknown): Step => snap => {
33  if (agentOf(e)) return null
34  const level = (e as { effort?: { level?: unknown } }).effort?.level
35  if (typeof level !== 'string' || level === snap.effort) return null
36  return { ...snap, effort: level }
37}
38
39const wordStep = (agentId: string | undefined, word: Word | null): Step => (snap, nowS) => applyWord(snap, agentId, word, nowS)
40
41const titleStep = (title: unknown): Step => snap => (typeof title === 'string' && title !== '' && title !== snap.title ? { ...snap, title } : null)
42
43// The first step that changes the snapshot carries on from its result; null when none changes it
44const steps = (...all: Step[]): Step => (snap, nowS) => {
45  let cur = snap
46  let changed = false
47  for (const s of all) {
48    const n = s(cur, nowS)
49    if (n !== null) {
50      cur = n
51      changed = true
52    }
53  }
54  return changed ? cur : null
55}
56
57// One sender per Claude Code process. Its state is made by `register`, so a test that loads the mod twice has two.
58type Ctx = {
59  candidates: string[]
60  inFlight: boolean
61  pending: boolean
62  closed: boolean
63  timer: { cancel: () => void } | null
64  tail: Promise<unknown>
65}
66
67// Read-modify-write of the snapshot one at a time: hooks run side by side
68const serial = <T>(ctx: Ctx, fn: () => Promise<T>): Promise<T> => {
69  const run = ctx.tail.then(fn)
70  ctx.tail = run.then(
71    () => undefined,
72    () => undefined,
73  )
74  return run
75}
76
77const getSnap = async ($: EngineInterface): Promise<UsageFeedSnap> => (await $.state.get(SNAP)).value ?? emptySnap()
78
79const update = ($: EngineInterface, ctx: Ctx, step: Step): Promise<boolean> =>
80  serial(ctx, async () => {
81    const next = step(await getSnap($), Math.floor((await $.clock.now()) / 1000))
82    if (next === null) return false
83    await $.state.set(SNAP, next)
84    return true
85  })
86
87// One POST to one app, never rejecting. The host gives up on a listener that accepts and never answers only after
88// 30 s and HttpInit has no timeout, so the fetch is raced against 2 s and the late answer dropped.
89const post = async ($: EngineInterface, home: string, bundleId: string, text: string): Promise<Result> => {
90  try {
91    const fetched: Promise<Result> = $.http
92      .fetch('http://localhost/v1/session', {
93        method: 'POST',
94        headers: { 'content-type': 'application/json' },
95        body: text,
96        socketPath: socketPath(home, bundleId),
97      })
98      .then(
99        r => ({ status: r.status }),
100        (err: unknown) => ({ error: String(err).slice(0, 200) }),
101      )
102    return await new Promise<Result>(resolve => {
103      const t = $.clock.after(SEND_TIMEOUT_MS, () => resolve({ error: 'timeout' }))
104      void fetched.then(r => {
105        t.cancel()
106        resolve(r)
107      })
108    })
109  } catch (err) {
110    return { error: String(err).slice(0, 200) }
111  }
112}
113
114// The next sequence number, kept in the snapshot
115const bump = ($: EngineInterface, ctx: Ctx): Promise<UsageFeedSnap> =>
116  serial(ctx, async () => {
117    const snap = await getSnap($)
118    const next = { ...snap, seq: snap.seq + 1 }
119    await $.state.set(SNAP, next)
120    return next
121  })
122
123const sendAll = async ($: EngineInterface, ctx: Ctx, text: string): Promise<void> => {
124  const home = await $.env.get('HOME')
125  if (typeof home !== 'string' || home === '') return
126  await Promise.all(ctx.candidates.map(id => post($, home, id, text)))
127}
128
129const sendUpdate = async ($: EngineInterface, ctx: Ctx): Promise<void> => {
130  const snap = await bump($, ctx)
131  const nowMs = await $.clock.now()
132  const interactive = snap.interactive ?? (await $.session.surfaces()).length > 0
133  const record = buildRecord({
134    nowMs,
135    sessionId: await $.session.id(),
136    root: await $.session.root(),
137    model: await $.session.model(),
138    version: (await $.session.version()).version,
139    usage: await $.session.usage(),
140    title: snap.title,
141    effort: snap.effort,
142  })
143  await sendAll($, ctx, JSON.stringify(buildUpdate(snap, interactive, record, Math.floor(nowMs / 1000))))
144}
145
146// One request in flight at most; a change during it is sent once more after it, as the latest snapshot
147const kick = ($: EngineInterface, ctx: Ctx): void => {
148  if (ctx.closed) return
149  ctx.pending = true
150  if (ctx.inFlight) return
151  ctx.inFlight = true
152  void (async () => {
153    try {
154      while (ctx.pending && !ctx.closed) {
155        ctx.pending = false
156        await sendUpdate($, ctx)
157      }
158    } catch {
159      // the next change or heartbeat tries again
160    } finally {
161      ctx.inFlight = false
162    }
163  })()
164}
165
166const ensureTimer = ($: EngineInterface, ctx: Ctx): void => {
167  if (ctx.timer !== null || ctx.closed) return
168  ctx.timer = $.clock.every(HEARTBEAT_MS, () => kick($, ctx))
169}
170
171// Run an observer's work: it never throws into the hook, and a change asks for a send
172const observe = async ($: EngineInterface, ctx: Ctx, step: Step | null): Promise<void> => {
173  try {
174    ensureTimer($, ctx)
175    const changed = step === null ? false : await update($, ctx, step)
176    if (changed) kick($, ctx)
177  } catch {
178    // an observer must not fail the event
179  }
180}
181
182export const register: Register = (on, options) => {
183  const ctx: Ctx = { candidates: candidateIds(options.extraBundleIds), inFlight: false, pending: false, closed: false, timer: null, tail: Promise.resolve() }
184
185  on('session.start', async ($, e, next) => {
186    const r = await next(e)
187    try {
188      ctx.closed = false
189      await update($, ctx, snap => (snap.interactive === e.isInteractive ? null : { ...snap, interactive: e.isInteractive }))
190      ensureTimer($, ctx)
191      kick($, ctx)
192    } catch {
193      // as observe
194    }
195    return r
196  })
197  on('classic.SessionStart', async ($, e, next) => {
198    const r = await next(e)
199    await observe($, ctx, titleStep(e.session_title))
200    kick($, ctx)
201    return r
202
203  }).catch((_, e, next) => next(e))
204  on('classic.UserPromptSubmit', async ($, e, next) => {
205    const r = await next(e)
206    await observe($, ctx, steps(titleStep(r.sessionTitle ?? e.session_title), wordStep(agentOf(e), 'busy')))
207    return r
208
209  }).catch((_, e, next) => next(e))
210  on('classic.PostToolUse', async ($, e, next) => {
211    await observe($, ctx, steps(effortStep(e), wordStep(agentOf(e), 'busy')))
212    return next(e)
213
214  }).catch((_, e, next) => next(e))
215  on('classic.PostToolUseFailure', async ($, e, next) => {
216    await observe($, ctx, wordStep(agentOf(e), 'busy'))
217    return next(e)
218
219  }).catch((_, e, next) => next(e))
220  // Before next: the hooks beneath may wait for the person's answer
221  on('classic.PermissionRequest', async ($, e, next) => {
222    await observe($, ctx, steps(effortStep(e), wordStep(agentOf(e), 'waiting')))
223    return next(e)
224
225  }).catch((_, e, next) => next(e))
226  on('classic.Stop', async ($, e, next) => {
227    await observe($, ctx, steps(effortStep(e), agentOf(e) ? () => null : wordStep(undefined, 'idle')))
228    return next(e)
229
230  }).catch((_, e, next) => next(e))
231  on('classic.StopFailure', async ($, e, next) => {
232    await observe($, ctx, agentOf(e) ? null : wordStep(undefined, 'idle'))
233    return next(e)
234
235  }).catch((_, e, next) => next(e))
236  on('classic.SubagentStart', async ($, e, next) => {
237    const a = agentOf(e)
238    await observe($, ctx, a ? wordStep(a, 'busy') : null)
239    return next(e)
240
241  }).catch((_, e, next) => next(e))
242  on('classic.SubagentStop', async ($, e, next) => {
243    const a = agentOf(e)
244    await observe($, ctx, a ? wordStep(a, null) : null)
245    return next(e)
246
247  }).catch((_, e, next) => next(e))
248  // Every turn's end, whatever its reason: main idle, or that agent removed (an Escape on a permission prompt gives
249  // only this event, and a stopped agent gets no SubagentStop)
250  on('turn.complete', async ($, e, next) => {
251    await observe($, ctx, e.agentId ? wordStep(e.agentId, null) : wordStep(undefined, 'idle'))
252    return next(e)
253  })
254  on('session.measure', async ($, e, next) => {
255    await observe($, ctx, null)
256    kick($, ctx)
257    return next(e)
258  })
259  on('session.end', async ($, e, next) => {
260    try {
261      const clearing = e.reason === 'clear' || e.reason === 'resume'
262      ctx.pending = false
263      if (!clearing) {
264        ctx.closed = true
265        ctx.timer?.cancel()
266        ctx.timer = null
267      }
268      const snap = await bump($, ctx)
269      await sendAll($, ctx, JSON.stringify(buildEnd(snap.seq, e.sessionId)))
270      // The process goes on under a new id: idle, no agents, no name, until its events arrive
271      if (clearing) await update($, ctx, (s, nowS) => afterClear(s, nowS))
272    } catch {
273      // the app drops a session whose process has gone
274    }
275    return next(e)
276  })
277}
278

1// The pure part of usage-feed: events to a session's state, and state plus usage to the message the sender posts.
2// Nothing here calls the engine, so claude plugin test checks it by plain calls.
3import type { UsageFeedSnap } from '../types'
4
5export const RELEASE_BUNDLE_ID = 'com.halfmarblellc.upmonitor'
6// Keep equal to "version" in .claude-plugin/plugin.json
7export const MOD_VERSION = '1.0.0'
8export const WIRE_VERSION = 1
9export const HEARTBEAT_MS = 10000
10export const SEND_TIMEOUT_MS = 2000
11// A busy subagent whose state is older than this is no longer sent
12export const AGENT_BUSY_MAX_AGE_S = 7200
13export const AGENTS_MAX = 64
14export const EXTRA_IDS_MAX = 4
15
16export type Word = 'busy' | 'waiting' | 'idle'
17
18export const emptySnap = (): UsageFeedSnap => ({ seq: 0, main: null, agents: {}, effort: null, title: null, interactive: null })
19
20// "<word> <epoch seconds>"
21export const wordOf = (line: string | null | undefined): string | undefined => line?.split(' ')[0]
22const epochOf = (line: string): number => {
23  const n = Number(line.split(' ')[1])
24  return Number.isFinite(n) ? n : 0
25}
26
27// The UTF-8 bytes of a string, without TextEncoder
28const utf8 = (s: string): number[] => {
29  const out: number[] = []
30  for (const ch of s) {
31    const c = ch.codePointAt(0) ?? 0
32    if (c < 0x80) out.push(c)
33    else if (c < 0x800) out.push(0xc0 | (c >> 6), 0x80 | (c & 0x3f))
34    else if (c < 0x10000) out.push(0xe0 | (c >> 12), 0x80 | ((c >> 6) & 0x3f), 0x80 | (c & 0x3f))
35    else out.push(0xf0 | (c >> 18), 0x80 | ((c >> 12) & 0x3f), 0x80 | ((c >> 6) & 0x3f), 0x80 | (c & 0x3f))
36  }
37  return out
38}
39
40// FNV-1a, 64 bits, as 16 hex digits
41export const fnv1a64 = (s: string): string => {
42  let h = 0xcbf29ce484222325n
43  for (const b of utf8(s)) h = BigInt.asUintN(64, (h ^ BigInt(b)) * 0x100000001b3n)
44  return h.toString(16).padStart(16, '0')
45}
46
47// An agent id the app takes as it is stays; any other (a teammate's "name@team") is sent as "h" and its hash
48export const agentKey = (id: string): string => (/^[A-Za-z0-9]{1,64}$/.test(id) ? id : `h${fnv1a64(id)}`)
49
50// The snapshot after one state change, or null when it changes nothing. word null removes an agent; for the main
51// thread a null word changes nothing.
52export const applyWord = (snap: UsageFeedSnap, agentId: string | undefined, word: Word | null, nowS: number): UsageFeedSnap | null => {
53  const line = word === null ? null : `${word} ${nowS}`
54  if (agentId === undefined || agentId === '') {
55    if (line === null || wordOf(snap.main) === word) return null
56    return { ...snap, main: line }
57  }
58  const key = agentKey(agentId)
59  const agents = { ...snap.agents }
60  if (line === null) {
61    if (!(key in agents)) return null
62    delete agents[key]
63  } else {
64    if (wordOf(agents[key]) === word) return null
65    agents[key] = line
66  }
67  return { ...snap, agents }
68}
69
70// The agents a message carries: no busy one older than two hours, and the 64 with the latest epochs
71export const agentsToSend = (agents: Record<string, string>, nowS: number): Record<string, string> => {
72  const live = Object.entries(agents).filter(([, line]) => !(wordOf(line) === 'busy' && nowS - epochOf(line) > AGENT_BUSY_MAX_AGE_S))
73  live.sort((a, b) => epochOf(b[1]) - epochOf(a[1]) || (a[0] < b[0] ? -1 : 1))
74  return Object.fromEntries(live.slice(0, AGENTS_MAX))
75}
76
77// The bundle IDs to feed: the release ID, then the valid ones of the option (comma-separated, four at most)
78export const candidateIds = (extra: unknown): string[] => {
79  const ids = [RELEASE_BUNDLE_ID]
80  if (typeof extra !== 'string') return ids
81  let n = 0
82  for (const part of extra.split(',')) {
83    const id = part.trim()
84    if (!/^[A-Za-z0-9.-]{1,64}$/.test(id) || ids.includes(id)) continue
85    if (n >= EXTRA_IDS_MAX) break
86    ids.push(id)
87    n += 1
88  }
89  return ids
90}
91
92export const socketPath = (home: string, bundleId: string): string => `${home}/Library/Containers/${bundleId}/Data/m.sock`
93
94export const lastPart = (path: string): string | null => path.split('/').filter(p => p !== '').pop() ?? null
95
96export type UsageIn = {
97  startedAt: number
98  context: { tokens?: number; window: number; percent?: number }
99  rateLimits: { kind: string; percentUsed: number; resetsAt?: string }[]
100  cost?: { usd: number }
101}
102
103export type RecordIn = {
104  nowMs: number
105  sessionId: string
106  root: string
107  model: string
108  version: string
109  usage: UsageIn
110  title: string | null
111  effort: string | null
112}
113
114// the v1 record, in the fields the app reads
115export const buildRecord = (r: RecordIn) => {
116  const limits: Record<string, { used_percentage: number; resets_at: number | null }> = {}
117  for (const l of r.usage.rateLimits) {
118    if (l.kind !== 'five_hour' && l.kind !== 'seven_day') continue
119    const at = l.resetsAt === undefined ? Number.NaN : Date.parse(l.resetsAt)
120    limits[l.kind] = { used_percentage: l.percentUsed, resets_at: Number.isFinite(at) ? Math.floor(at / 1000) : null }
121  }
122  return {
123    v: WIRE_VERSION,
124    written_at: Math.floor(r.nowMs / 1000),
125    session_id: r.sessionId,
126    session_name: r.title,
127    project: lastPart(r.root),
128    model: { id: r.model },
129    effort: { level: r.effort },
130    context_window: {
131      used_percentage: r.usage.context.percent ?? null,
132      context_window_size: r.usage.context.window,
133      used_tokens: r.usage.context.tokens ?? null,
134    },
135    cost: { total_cost_usd: r.usage.cost?.usd ?? null, total_duration_ms: r.nowMs - r.usage.startedAt },
136    ...(Object.keys(limits).length > 0 ? { rate_limits: limits } : {}),
137    version: r.version,
138  }
139}
140
141export const buildUpdate = (snap: UsageFeedSnap, interactive: boolean, record: ReturnType<typeof buildRecord>, nowS: number) => ({
142  v: WIRE_VERSION,
143  kind: 'update' as const,
144  seq: snap.seq,
145  mod: MOD_VERSION,
146  interactive,
147  record,
148  state: { main: snap.main, agents: agentsToSend(snap.agents, nowS) },
149})
150
151export const buildEnd = (seq: number, sessionId: string) => ({ v: WIRE_VERSION, kind: 'end' as const, seq, mod: MOD_VERSION, session_id: sessionId })
152
153// The session's snapshot after /clear or a resume: the same process under a new id, main idle, no agents, no name
154export const afterClear = (snap: UsageFeedSnap, nowS: number): UsageFeedSnap => ({ ...snap, main: `idle ${nowS}`, agents: {}, title: null })
155

1// The state this mod keeps in $.state: one snapshot per session, which a reload of the mod keeps.
2export type UsageFeedSnap = {
3  // Counts the messages sent for this session, so a receiver can tell a late one from a newer one
4  seq: number
5  // The main thread's state as "<word> <epoch seconds>", or null until the mod has seen a main-thread event
6  main: string | null
7  // Each subagent's state in the same form, by its id (an id outside [A-Za-z0-9]{1,64} is kept as "h" and 16 hex digits)
8  agents: Record<string, string>
9  effort: string | null
10  title: string | null
11  // The session's isInteractive from session.start, null until the mod has seen it
12  interactive: boolean | null
13}
14
15declare module 'claude-code' {
16  interface PluginState {
17    'usage-feed': { snap: UsageFeedSnap }
18  }
19}
20