mc-lane

1// mc-lane — reports this Claude Code lane to Mission Control.
2//
3// Spec: docs/specs/claude-mods-integration.md §3.1 (telemetry only; no controls).
4// Server: POST /api/mesh/lane-events (app/server/routes/mesh_lane_events.py).
5//
6// Runs in interactive sessions, `claude -p` and Agent SDK lanes alike: mods'
7// hooks load in all three (docs/reference/claude-mods/overview.md, "Where mods
8// run"). Nothing here draws, so a headless lane loses nothing.
9//
10// Configuration (environment, read at each session.start):
11//   MC_LANE_URL     Pi-CEO backend base URL, e.g. https://pi-dev-ops.up.railway.app
12//   MC_LANE_SECRET  the X-Pi-CEO-Secret the fleet already uses for heartbeats
13//   MC_LANE_HOST    this machine's fleet name (as in mesh_machines.host)
14// With URL or secret unset the mod records nothing and says so once in the
15// transcript: an unconfigured lane must read as "not reporting", never as idle.
16//
17// HOT RELOAD. Every reload re-runs `register` and `session.start`, resets the
18// module variables and cancels the timers (docs/reference/claude-mods/api.md,
19// interface.md "Keep state"). So what must survive one — the unsent queue, the
20// counter, and whether this session's `session_start` was already recorded —
21// lives in `$.state` (declared in types/index.d.ts). What must survive the
22// session itself is parked in `$.store` at session.end. `/clear`, `/resume` and
23// `/branch` reset `$.state` too; events still unsent at that moment (at most one
24// flush interval's worth while the backend answers) are lost, not duplicated.
25//
26// Functions that take `$` are top-level declarations: the engine scans the
27// module before loading it and refuses `$` handed to anything else.
28
29import type { EngineInterface, Register } from 'claude-code'
30import { atom, read, update } from 'claude-code'
31import {
32  dropAcked, enqueue, makeSeq, MAX_BATCH, modelName, repoSlug, toolName, worstRate,
33  type LaneEvent, type LaneEventKind,
34} from './lane'
35
36const FLUSH_MS = 5_000
37const PENDING = 'pending:'
38
39type Config = { url: string; secret: string; host: string }
40
41// Session state: survives a reload of this module.
42const queue = atom({ plugin: 'mc-lane', key: 'queue' } as const, [] as LaneEvent[])
43const counter = atom({ plugin: 'mc-lane', key: 'counter' } as const, 0)
44/** The session whose `session_start` is recorded; '' until then. */
45const started = atom({ plugin: 'mc-lane', key: 'started' } as const, '')
46
47// Module state: rebuilt by every session.start, so a reload loses nothing here.
48const st = {
49  sessionId: 'unknown',
50  cfg: null as Config | null,
51  flushing: false, // the flush timer of this module instance is running
52  inFlight: false,
53  lastFailure: undefined as string | undefined,
54}
55
56function append(events: LaneEvent[]): (q: LaneEvent[]) => LaneEvent[] {
57  return q => {
58    const next = q.slice()
59    for (const ev of events) enqueue(next, ev)
60    return next
61  }
62}
63
64async function record($: EngineInterface, kind: LaneEventKind, extra: Partial<LaneEvent>): Promise<void> {
65  const now = await $.clock.now()
66  const n = await update($, counter, c => c + 1)
67  const ev: LaneEvent = {
68    session_id: st.sessionId,
69    seq: makeSeq(now, n),
70    kind,
71    at: new Date(now).toISOString(),
72    ...extra,
73  }
74  await update($, queue, append([ev]))
75}
76
77// One POST at a time; a tick that lands while one is out is skipped.
78async function flush($: EngineInterface): Promise<void> {
79  const cfg = st.cfg
80  if (!cfg || st.inFlight) return
81  const pending = await read($, queue)
82  if (pending.length === 0) return
83  st.inFlight = true
84  const batch = pending.slice(0, MAX_BATCH)
85  let failure: string | undefined
86  try {
87    const r = await $.http.fetch(`${cfg.url}/api/mesh/lane-events`, {
88      method: 'POST',
89      headers: { 'Content-Type': 'application/json', 'X-Pi-CEO-Secret': cfg.secret },
90      body: JSON.stringify({ host: cfg.host, events: batch }),
91    })
92    if (r.ok) {
93      const acked = (JSON.parse(r.text) as { acked?: Record<string, number> }).acked ?? {}
94      // Drops only what was acked: events recorded during the POST stay.
95      await update($, queue, q => dropAcked(q, acked))
96    } else {
97      failure = `HTTP ${r.status}`
98    }
99  } catch (err) {
100    failure = err instanceof Error ? err.message.slice(0, 80) : 'request failed'
101  } finally {
102    st.inFlight = false
103  }
104  if (failure !== st.lastFailure) {
105    // Only on change, so a long outage is one status line, not one per tick.
106    const left = (await read($, queue)).length
107    $.ui.status(failure ? `Mission Control unreachable (${failure}); ${left} queued` : undefined)
108    st.lastFailure = failure
109  }
110}
111
112export const register: Register = on => {
113  // Before the first prompt, and again after every reload of this module.
114  on('session.start', async ($, e, next) => {
115    const url = (await $.env.get('MC_LANE_URL'))?.replace(/\/+$/, '')
116    const secret = await $.env.get('MC_LANE_SECRET')
117    const host = (await $.env.get('MC_LANE_HOST')) || 'unknown'
118    st.sessionId = await $.session.id()
119
120    if (!url || !secret) {
121      st.cfg = null
122      $.ui.log('not reporting to Mission Control: MC_LANE_URL or MC_LANE_SECRET is unset')
123      return next(e)
124    }
125    st.cfg = { url, secret, host }
126
127    // Events a previous session on this machine could not send before it ended.
128    for (const key of await $.store.keys()) {
129      if (!key.startsWith(PENDING)) continue
130      const saved = await $.store.get(key)
131      if (Array.isArray(saved)) await update($, queue, append(saved as LaneEvent[]))
132      await $.store.delete(key)
133    }
134
135    // A reload re-runs this hook in the same session: record its start once.
136    if ((await read($, started)) !== st.sessionId) {
137      const repo = await $.session.repo()
138      const model = await $.session.model()
139      await record($, 'session_start', { repo: repoSlug(repo?.remote), model: modelName(model) })
140      await update($, started, () => st.sessionId)
141    }
142
143    // A reload cancelled the previous instance's timer; this instance starts its own once.
144    if (!st.flushing) {
145      st.flushing = true
146      $.clock.every(FLUSH_MS, () => flush($))
147    }
148    return next(e)
149  })
150
151  on('tool.call', async ($, e, next) => {
152    if (!st.cfg) return next(e)
153    const t0 = await $.clock.now()
154    const ran = await next(e)
155    const ms = Math.max(0, (await $.clock.now()) - t0)
156    // A denied call never ran: recorded as not ok, with no timing.
157    const denied = ran.deny !== undefined
158    await record($, 'tool', { tool: toolName(e.tool), ok: !denied && ran.isError !== true, ms: denied ? undefined : ms })
159    return ran
160  })
161
162  // Subagents and agent-team teammates (e.isTeammate). Passed through as given;
163  // only the agent type's name and the model it started on are recorded.
164  on('agent.spawn', async ($, e, next) => {
165    const ran = await next(e)
166    if (st.cfg && ran.deny === undefined) {
167      await record($, 'agent_start', {
168        model: modelName(ran.model),
169        tool: e.subagentType ? toolName(e.subagentType) : undefined,
170        ok: true,
171      })
172    }
173    return ran
174  })
175
176  on('session.measure', async ($, e, next) => {
177    if (st.cfg) {
178      await record($, 'usage', {
179        ctx_pct: e.context.percent,
180        rate_pct: worstRate(e.rateLimits),
181        // Absent where the host keeps no cost ledger: sent as absent, shown as unknown.
182        cost_usd: e.cost?.usd,
183      })
184    }
185    return next(e)
186  })
187
188  on('session.end', async ($, e, next) => {
189    if (st.cfg) {
190      await record($, 'session_end', {})
191      // session.end hooks share ~1.5 s; there is no time to POST. Park what is
192      // left for the next session on this machine to send.
193      const left = await read($, queue)
194      if (left.length > 0) await $.store.set(PENDING + st.sessionId, left)
195      await update($, queue, () => [])
196    }
197    return next(e)
198  })
199}
200

1// Pure helpers for mc-lane: no `$`, so they unit-test without the kit.
2//
3// WHAT IS SENT. Only names, numbers and booleans. A tool call's input and output
4// never leave the machine — not even redacted — so there is nothing to redact on
5// the server and nothing a leaked row can expose. The server re-validates every
6// field (app/server/routes/mesh_lane_events.py); this file is the first gate,
7// not the only one.
8
9// `agent_start`: a subagent or agent-team teammate started (agent.spawn). Its
10// `tool` is the agent type's name and `model` what it runs on; nothing else.
11export type LaneEventKind = 'session_start' | 'tool' | 'agent_start' | 'usage' | 'session_end'
12
13export type LaneEvent = {
14  session_id: string
15  seq: number
16  kind: LaneEventKind
17  at: string
18  repo?: string
19  model?: string
20  tool?: string
21  ok?: boolean
22  ms?: number
23  ctx_pct?: number
24  rate_pct?: number
25  cost_usd?: number
26}
27
28/** Hard cap on one POST; the server refuses more (MAX_BATCH there). */
29export const MAX_BATCH = 200
30
31/** Hard cap on what a lane holds while the backend is unreachable. Oldest go first. */
32export const MAX_QUEUE = 2000
33
34const TOOL_NAME = /^[A-Za-z0-9_.:-]{1,128}$/
35
36/**
37 * A tool's name as Mission Control may show it. MCP tool names carry the server
38 * name, which is configuration, not content, so they pass; anything that does
39 * not look like a name is replaced rather than sent.
40 */
41export function toolName(raw: unknown): string {
42  return typeof raw === 'string' && TOOL_NAME.test(raw) ? raw : 'unknown'
43}
44
45const MODEL_NAME = /^[A-Za-z0-9_.:[\]-]{1,80}$/
46
47/** A model id or alias as Mission Control may show it (the server's _MODEL), or undefined. */
48export function modelName(raw: unknown): string | undefined {
49  return typeof raw === 'string' && MODEL_NAME.test(raw) ? raw : undefined
50}
51
52/** `owner/name` from a git remote URL, or undefined. Never sends the URL itself (it can hold a token). */
53export function repoSlug(remote: string | null | undefined): string | undefined {
54  if (!remote) return undefined
55  const m = /[:/]([A-Za-z0-9_.-]+)\/([A-Za-z0-9_.-]+?)(?:\.git)?\/?$/.exec(remote.trim())
56  return m ? `${m[1]}/${m[2]}` : undefined
57}
58
59/** Highest rate-limit window used, 0–100, or undefined when none reported. */
60export function worstRate(limits: readonly { percentUsed: number }[] | undefined): number | undefined {
61  if (!limits || limits.length === 0) return undefined
62  return Math.max(...limits.map(l => l.percentUsed))
63}
64
65/**
66 * Sequence numbers are milliseconds × 1000 plus a counter, so they stay
67 * increasing across a module reload (which resets module variables) and the
68 * server's unique (session_id, seq) never drops a new event as a duplicate.
69 */
70export function makeSeq(nowMs: number, counter: number): number {
71  return Math.floor(nowMs) * 1000 + (counter % 1000)
72}
73
74/** Appends and trims to MAX_QUEUE, dropping the oldest. Returns how many were dropped. */
75export function enqueue(queue: LaneEvent[], ev: LaneEvent): number {
76  queue.push(ev)
77  const over = queue.length - MAX_QUEUE
78  if (over > 0) queue.splice(0, over)
79  return Math.max(over, 0)
80}
81
82/** Drops what the server acknowledged: every event of a session at or below its acked seq. */
83export function dropAcked(queue: LaneEvent[], acked: Record<string, number>): LaneEvent[] {
84  return queue.filter(ev => !(ev.session_id in acked && ev.seq <= acked[ev.session_id]))
85}
86

1// mc-lane's $.state contract (docs/reference/claude-mods/interface.md, "Declare the values").
2// Values here survive a reload of the hooks module; see hooks/register.ts "HOT RELOAD".
3// McLaneEvent mirrors LaneEvent in hooks/lane.ts — keep them equal. This file stands
4// alone: with an `import` from ../hooks, `claude plugin validate` (2.1.289) found no
5// declared state.
6
7export type McLaneEvent = {
8  session_id: string
9  seq: number
10  kind: 'session_start' | 'tool' | 'agent_start' | 'usage' | 'session_end'
11  at: string
12  repo?: string
13  model?: string
14  tool?: string
15  ok?: boolean
16  ms?: number
17  ctx_pct?: number
18  rate_pct?: number
19  cost_usd?: number
20}
21
22declare module 'claude-code' {
23  interface PluginState {
24    'mc-lane': {
25      /** Events not yet acked by Mission Control, oldest first (capped at MAX_QUEUE). */
26      queue: McLaneEvent[]
27      /** Per-session event counter, the low digits of each seq. */
28      counter: number
29      /** The session whose session_start is recorded; '' until then. */
30      started: string
31    }
32  }
33}
34