swarmery-link

1// swarmery-link 0.2: a sensor, plus an opt-in command channel. It reports this
2// session's live state to a local swarmery daemon over POST /api/link/events
3// (tools/swarmery/docs/mods-link-protocol.md) and changes nothing: every hook
4// passes its event on, a failure is logged and swallowed, and with no token
5// file the whole mod stays silent for the session and makes no request.
6//
7// With `control` on, one background loop long-polls GET /api/link/commands
8// and runs what the dashboard queued for this session: `stop` aborts the
9// running turn, `prompt` submits a prompt once the session is idle. Commands
10// run only from that loop, never from inside a hook. With `control` off none
11// of it is created and no request goes to the command endpoints.
12//
13// With `forecast` on, each main-loop turn that starts with the person's text
14// asks a small model, without waiting on it, what the turn will touch; the
15// turn's own tool calls make the actual. Both go to the daemon as
16// `turn.forecast` and `turn.actual` (paths and numbers only). With `forecast`
17// off no model call is made and neither event is sent.
18//
19// Every `$` call is spelled in this module, as the engine requires: `$` is
20// handed only to the functions declared at the top of this file, and ./lib/*
21// is pure (data in, data out).
22
23import type { EngineInterface, PromptSubmitArgs, Register, Timer } from 'claude-code'
24import {
25  AUTH_RETRY_MS,
26  DEFAULT_DAEMON_URL,
27  END_FLUSH_TIMEOUT_MS,
28  FLUSH_EVERY_MS,
29  REQUEST_TIMEOUT_MS,
30  daemonEndpoint,
31  describeFailure,
32  linkRequest,
33  outcome,
34  parseToken,
35} from './lib/client'
36import {
37  type Ack,
38  type Command,
39  type Executed,
40  POLL_TIMEOUT_MS,
41  PROMPT_START_TIMEOUT_MS,
42  ackBody,
43  ackUrl,
44  commandsUrl,
45  decide,
46  done,
47  dropAck,
48  failed,
49  nextPollDelay,
50  parseCommand,
51  pollOutcome,
52  pollRequest,
53  remember,
54  shortError,
55  submitFailure,
56} from './lib/control'
57import {
58  FORECAST_MAX_TOKENS,
59  FORECAST_MODEL,
60  FORECAST_TIMEOUT_MS,
61  type ForecastState,
62  INITIAL_FORECAST,
63  NO_CLAIMS,
64  RUN_ID_RANDOM,
65  type ToolOutcome,
66  closeForecast,
67  endTurn,
68  forecastPayload,
69  forecastRequest,
70  noteToolCall,
71  parseForecast,
72  resetForecast,
73  runIdOf,
74  startTurn,
75  usageTokens,
76} from './lib/forecast'
77import { inputHash, toolInput } from './lib/hash'
78import { type LinkEvent, type LinkKind, type Stamps, drop, enqueue, stamp, takeBatch } from './lib/queue'
79import { measurePayload } from './lib/usage'
80
81const NAME = 'swarmery-link'
82
83/** Options reserved for later releases: each logs once when switched on, and hooks nothing else. */
84const RESERVED = ['planBand', 'swarm'] as const
85
86type Reply = { readonly status?: number; readonly error?: unknown }
87
88/** The link's state for one load of the module (a reload starts it over). */
89type Link = {
90  readonly url: string
91  token: string | undefined
92  sessionId: string
93  cwd: string
94  /** The real session.start's `{ surface, isInteractive }`, sent again for the session a `/clear` starts. */
95  start: Readonly<Record<string, unknown>> | undefined
96  /**
97   * A `/clear` ended the session and the process goes on under a new id with
98   * no session.start: nothing is queued until the next turn.start re-reads it.
99   */
100  isClearing: boolean
101  /** Off for the rest of this load: 503, or a 401 that survived a re-read. */
102  isOff: boolean
103  /** The last request got no answer or a 5xx: turn boundaries leave sending to the timer until one succeeds. */
104  isFailing: boolean
105  /** A 401 was answered with a token re-read; the one retry is pending. */
106  isAuthWait: boolean
107  hasRetriedAuth: boolean
108  queue: readonly LinkEvent[]
109  stamps: Stamps
110  timer: Timer | undefined
111  flushing: Promise<void> | undefined
112  readonly logged: Set<string>
113}
114
115const isActive = (link: Link): boolean => link.token !== undefined && !link.isOff && link.sessionId !== ''
116
117function log($: EngineInterface, text: string, toDebug = false): void {
118  try {
119    $.ui.log(`${NAME}: ${text}`, toDebug ? { to: 'debug' } : undefined)
120  } catch {
121    // a log line is a courtesy
122  }
123}
124
125/** Logs `text` to the transcript the first time `key` is seen, to the debug log after. */
126function logOnce($: EngineInterface, link: Link, key: string, text: string): void {
127  log($, text, link.logged.has(key))
128  link.logged.add(key)
129}
130
131/** The daemon's token (M7: `~` is not expanded, so HOME is read and spelled out). */
132async function readToken($: EngineInterface): Promise<string | undefined> {
133  try {
134    const home = await $.env.get('HOME')
135    return parseToken(await $.fs.read(`${home}/.swarmery/link-token`))
136  } catch {
137    return undefined
138  }
139}
140
141function turnOff($: EngineInterface, link: Link, why: string): void {
142  link.isOff = true
143  link.queue = []
144  link.timer?.cancel()
145  link.timer = undefined
146  log($, `link off for this session: ${why}`)
147}
148
149/** `pending`, or `'timeout'` once `ms` pass on `$.clock`, whichever comes first. */
150async function within<T>($: EngineInterface, pending: Promise<T>, ms: number): Promise<T | 'timeout'> {
151  let guard: Timer | undefined
152  const timeout = new Promise<'timeout'>(resolve => {
153    try {
154      guard = $.clock.after(ms, () => resolve('timeout'))
155    } catch {
156      // no timer: the pending work alone decides
157    }
158  })
159  try {
160    return await Promise.race([pending, timeout])
161  } finally {
162    guard?.cancel()
163  }
164}
165
166/** One POST of `body`, raced against a `$.clock` timer: HttpInit has no timeout or signal. */
167async function post($: EngineInterface, link: Link, body: string, timeoutMs: number): Promise<Reply> {
168  const request = $.http.fetch(link.url, linkRequest(link.token ?? '', body)).then(
169    (r): Reply => ({ status: r.status }),
170    (error: unknown): Reply => ({ error }),
171  )
172  const reply = await within($, request, timeoutMs)
173  return reply === 'timeout' ? { error: 'timeout' } : reply
174}
175
176async function send($: EngineInterface, link: Link, timeoutMs: number): Promise<void> {
177  while (isActive(link) && !link.isAuthWait && link.queue.length > 0) {
178    const { batch, body, oversized } = takeBatch(link.queue)
179    if (oversized.length > 0) link.queue = drop(link.queue, oversized)
180    if (batch.length === 0) return
181    const reply = await post($, link, body, timeoutMs)
182    switch (outcome(reply.status)) {
183      case 'sent':
184        link.queue = drop(link.queue, batch)
185        link.isFailing = false
186        link.hasRetriedAuth = false
187        continue
188      case 'discard':
189        link.queue = drop(link.queue, batch)
190        logOnce($, link, `discard-${reply.status}`, `the daemon refused a batch (${describeFailure(reply.status, undefined)}); dropped`)
191        continue
192      case 'off':
193        turnOff($, link, 'the daemon has the link disabled (503)')
194        return
195      case 'auth':
196        if (link.hasRetriedAuth) {
197          turnOff($, link, 'the daemon refused the token after a re-read (401)')
198          return
199        }
200        link.hasRetriedAuth = true
201        link.isAuthWait = true
202        link.token = (await readToken($)) ?? link.token
203        log($, 'token refused (401): re-read it, retrying once', true)
204        try {
205          $.clock.after(AUTH_RETRY_MS, () => {
206            link.isAuthWait = false
207            void flush($, link, REQUEST_TIMEOUT_MS)
208          })
209        } catch {
210          link.isAuthWait = false
211        }
212        return
213      case 'retry':
214        link.isFailing = true
215        logOnce($, link, 'unreachable', `daemon unreachable (${describeFailure(reply.status, reply.error)}); keeping up to 200 events`)
216        return
217    }
218  }
219}
220
221/** Sends what is queued, one flush at a time. Never rejects. */
222function flush($: EngineInterface, link: Link, timeoutMs: number): Promise<void> {
223  if (link.flushing !== undefined) return link.flushing
224  link.flushing = send($, link, timeoutMs)
225    .catch(() => undefined)
226    .finally(() => {
227      link.flushing = undefined
228    })
229  return link.flushing
230}
231
232/** Arms the periodic flush once; it re-arms while events remain. */
233function arm($: EngineInterface, link: Link): void {
234  if (link.timer !== undefined || !isActive(link)) return
235  try {
236    link.timer = $.clock.after(FLUSH_EVERY_MS, () => {
237      link.timer = undefined
238      void flush($, link, REQUEST_TIMEOUT_MS).then(() => {
239        if (link.queue.length > 0) arm($, link)
240      })
241    })
242  } catch {
243    link.timer = undefined
244  }
245}
246
247async function push($: EngineInterface, link: Link, kind: LinkKind, payload?: Readonly<Record<string, unknown>>): Promise<void> {
248  if (!isActive(link) || link.isClearing) return
249  const stamped = stamp(link.stamps, kind, await $.clock.now())
250  link.stamps = stamped.stamps
251  const event: LinkEvent = { sessionId: link.sessionId, cwd: link.cwd, kind, at: stamped.at, ...(payload !== undefined ? { payload } : {}) }
252  link.queue = enqueue(link.queue, event)
253  arm($, link)
254}
255
256/**
257 * At a turn boundary: start a send now while the daemon answers, else leave it
258 * to the timer. Never waited on: a hook never blocks on the daemon.
259 */
260function flushAtTurn($: EngineInterface, link: Link): void {
261  if (isActive(link) && !link.isFailing) void flush($, link, REQUEST_TIMEOUT_MS)
262}
263
264/**
265 * session.end's last send: joins a flush already in flight (which may have
266 * started under the 5 s request timeout), then sends what that one left. The
267 * caller bounds the whole wait.
268 */
269async function flushForEnd($: EngineInterface, link: Link): Promise<void> {
270  await flush($, link, END_FLUSH_TIMEOUT_MS)
271  if (link.queue.length > 0 && !link.isFailing) await flush($, link, END_FLUSH_TIMEOUT_MS)
272}
273
274/**
275 * After a `/clear`, at the first main-loop turn.start: the process goes on
276 * under a new session id and no session.start fires for it (SessionEndInput),
277 * so the id is read again here and, when it moved, the new session is
278 * announced with the real session.start's payload.
279 */
280async function resync($: EngineInterface, link: Link): Promise<void> {
281  const sessionId = await $.session.id()
282  const cwd = (await $.session.cwd()) || link.cwd
283  link.isClearing = false
284  if (sessionId === link.sessionId) return
285  link.sessionId = sessionId
286  link.cwd = cwd
287  await push($, link, 'session.start', link.start)
288}
289
290/** The command channel's state for one session's loop. Exists only with `control` on. */
291type Control = {
292  readonly origin: string
293  readonly sessionId: string
294  /** Set for good by session.end, a 503, or a 401 that survived a re-read. */
295  isStopped: boolean
296  /** The main loop's running turn: `turn.start`'s id until its `turn.complete`. */
297  turnId: string | undefined
298  executed: Executed
299  hasRetriedAuth: boolean
300  /** The pending wait between polls, cancelled on stop. */
301  pause: Timer | undefined
302  /** Settles every race of the loop the moment it stops. */
303  readonly stopped: Promise<'stopped'>
304  readonly settleStopped: () => void
305}
306
307type PollReply = { readonly status?: number; readonly text?: string; readonly error?: unknown }
308
309const newControl = (origin: string, sessionId: string, previous: Control | undefined): Control => {
310  let settleStopped: () => void = () => undefined
311  const stopped = new Promise<'stopped'>(resolve => {
312    settleStopped = () => resolve('stopped')
313  })
314  return {
315    origin,
316    sessionId,
317    isStopped: false,
318    turnId: previous?.turnId,
319    // ids are the daemon's, unique across sessions: a new loop keeps what ran
320    executed: previous?.executed ?? new Map(),
321    hasRetriedAuth: false,
322    pause: undefined,
323    stopped,
324    settleStopped,
325  }
326}
327
328function stopControl($: EngineInterface, control: Control, why: string | undefined): void {
329  if (control.isStopped) return
330  control.isStopped = true
331  control.pause?.cancel()
332  control.pause = undefined
333  control.settleStopped()
334  if (why !== undefined) log($, `control off for this session: ${why}`)
335}
336
337/** Waits `ms` on `$.clock` between polls, or less when the loop stops. A clock that refuses stops the loop. */
338async function pause($: EngineInterface, control: Control, ms: number): Promise<void> {
339  if (ms <= 0 || control.isStopped) return
340  const waited = new Promise<void>(resolve => {
341    try {
342      control.pause = $.clock.after(ms, () => {
343        control.pause = undefined
344        resolve()
345      })
346    } catch {
347      stopControl($, control, 'no timer for the poll loop')
348      resolve()
349    }
350  })
351  await Promise.race([waited, control.stopped])
352}
353
354/** One long poll, raced against its 28 s timer and the loop's stop. */
355async function fetchCommand($: EngineInterface, link: Link, control: Control): Promise<PollReply | 'stopped'> {
356  const request = $.http.fetch(commandsUrl(control.origin, control.sessionId), pollRequest(link.token ?? '')).then(
357    (r): PollReply => ({ status: r.status, text: r.text }),
358    (error: unknown): PollReply => ({ error }),
359  )
360  const reply = await Promise.race([within($, request, POLL_TIMEOUT_MS), control.stopped])
361  return reply === 'timeout' ? { error: 'timeout' } : reply
362}
363
364/** Acks a command. A lost ack is not retried: the daemon delivers the command again and the executed set answers it. */
365async function sendAck($: EngineInterface, link: Link, control: Control, id: string, ack: Ack): Promise<void> {
366  try {
367    const request = $.http.fetch(ackUrl(control.origin, id), linkRequest(link.token ?? '', ackBody(ack))).then(
368      (r): Reply => ({ status: r.status }),
369      (error: unknown): Reply => ({ error }),
370    )
371    const reply = await within($, request, REQUEST_TIMEOUT_MS)
372    const status = reply === 'timeout' ? undefined : reply.status
373    if (status === undefined || status < 200 || status >= 300) {
374      const error = reply === 'timeout' ? 'timeout' : reply.error
375      logOnce($, link, 'ack-failed', `command ${id}: ack not taken (${describeFailure(status, error)}); the daemon delivers it again`)
376    }
377  } catch {
378    // the daemon re-delivers an unacked command after 30 s
379  }
380}
381
382async function abortTurn($: EngineInterface, turnId: string): Promise<Ack> {
383  try {
384    await $.turn.abort({ turnId })
385    return done()
386  } catch (error) {
387    return failed(shortError(error))
388  }
389}
390
391/** Submits the prompt; it resolves as its turn starts, so a session that stays busy past the bound leaves it queued. */
392async function submitPrompt($: EngineInterface, text: string, asUser: boolean): Promise<Ack> {
393  const args: PromptSubmitArgs = asUser ? { text, asUser: true } : { text }
394  try {
395    const result = await within($, $.prompt.submit(args), PROMPT_START_TIMEOUT_MS)
396    if (result === 'timeout') return done('queued')
397    if (result.drop !== undefined) return dropAck(result.drop)
398    return done()
399  } catch (error) {
400    return submitFailure(error)
401  }
402}
403
404/** Runs one delivered command (or answers a re-delivered one from the executed set) and acks it. */
405async function execute($: EngineInterface, link: Link, control: Control, command: Command): Promise<void> {
406  const action = decide(command, control.executed, control.turnId)
407  let ack: Ack
408  switch (action.do) {
409    case 'ack':
410      ack = action.ack
411      break
412    case 'abort':
413      ack = await abortTurn($, action.turnId)
414      break
415    case 'submit':
416      ack = await submitPrompt($, action.text, action.asUser)
417      break
418  }
419  // recorded before the ack goes out: a lost ack must not run it twice
420  control.executed = remember(control.executed, command.id, ack)
421  if (!control.isStopped) await sendAck($, link, control, command.id, ack)
422}
423
424/** The command loop: poll, execute, ack, poll again, until it stops. Never rejects. */
425async function poll($: EngineInterface, link: Link, control: Control): Promise<void> {
426  try {
427    while (!control.isStopped && !link.isOff) {
428      const startedAt = await $.clock.now()
429      const reply = await fetchCommand($, link, control)
430      if (reply === 'stopped' || control.isStopped || link.isOff) return
431      switch (pollOutcome(reply.status)) {
432        case 'command': {
433          const command = parseCommand(reply.text)
434          control.hasRetriedAuth = false
435          if (command === undefined) {
436            logOnce($, link, 'command-unreadable', 'control: a command reply named no id; skipped')
437            await pause($, control, nextPollDelay('retry', 0))
438            continue
439          }
440          await execute($, link, control, command)
441          continue
442        }
443        case 'idle':
444          control.hasRetriedAuth = false
445          await pause($, control, nextPollDelay('idle', (await $.clock.now()) - startedAt))
446          continue
447        case 'auth':
448          if (control.hasRetriedAuth) {
449            stopControl($, control, 'the daemon refused the token after a re-read (401)')
450            return
451          }
452          control.hasRetriedAuth = true
453          link.token = (await readToken($)) ?? link.token
454          log($, 'control: token refused (401): re-read it, retrying once', true)
455          await pause($, control, AUTH_RETRY_MS)
456          continue
457        case 'off':
458          stopControl($, control, undefined)
459          turnOff($, link, 'the daemon has the link disabled (503)')
460          return
461        case 'retry':
462          logOnce($, link, 'control-unreachable', `control: command poll failed (${describeFailure(reply.status, reply.error)}); retrying every 3s`)
463          await pause($, control, nextPollDelay('retry', 0))
464          continue
465      }
466    }
467  } catch {
468    stopControl($, control, 'the poll loop failed')
469  }
470}
471
472/** Starts the session's one command loop, detached from the hook that starts it. */
473function startControl($: EngineInterface, link: Link, control: Control): void {
474  try {
475    $.clock.after(0, () => {
476      void poll($, link, control)
477    })
478    log($, 'control: polling the daemon for commands', true)
479  } catch {
480    stopControl($, control, 'no timer for the poll loop')
481  }
482}
483
484/** Replaces `previous` (stopped first) with a loop for the link's current session id, and starts it. */
485function openControl($: EngineInterface, link: Link, origin: string, previous: Control | undefined): Control {
486  if (previous !== undefined) stopControl($, previous, undefined)
487  const control = newControl(origin, link.sessionId, previous)
488  startControl($, link, control)
489  return control
490}
491
492/** The forecast loop's state for one load of the module. Exists only with `forecast` on. */
493type Forecaster = {
494  state: ForecastState
495  /** An in-process resume ended the session: the next turn.start starts a new run (resync). */
496  isResumed: boolean
497}
498
499/** RUN_ID_RANDOM random bytes from the runtime's `crypto`, or undefined where it has none. */
500const randomBytes = (): Uint8Array | undefined => {
501  try {
502    if (typeof crypto === 'undefined' || typeof crypto.getRandomValues !== 'function') return undefined
503    return crypto.getRandomValues(new Uint8Array(RUN_ID_RANDOM))
504  } catch {
505    return undefined
506  }
507}
508
509/** Starts the turn counter over under a fresh run id: `<ms of $.clock.now()>-<random base36>`. */
510async function newRun($: EngineInterface, forecaster: Forecaster): Promise<void> {
511  let now = 0
512  try {
513    now = await $.clock.now()
514  } catch {
515    // the random part alone tells the runs apart
516  }
517  forecaster.state = resetForecast(forecaster.state, runIdOf(now, randomBytes()))
518}
519
520/**
521 * Asks the model for turn `turnSeq`'s forecast and queues it as `turn.forecast`,
522 * however the call ended (no readable claim: all-null claims, tokens still
523 * counted). A reset while the call was in flight (`generation` moved) drops
524 * it: the session id and the run it was asked under are gone. Never rejects;
525 * nothing waits on it.
526 */
527async function runForecast(
528  $: EngineInterface,
529  link: Link,
530  forecaster: Forecaster,
531  runId: string,
532  turnSeq: number,
533  generation: number,
534  ask: { readonly system: string; readonly prompt: string },
535): Promise<void> {
536  let claims = NO_CLAIMS
537  let tokens: number | null = null
538  try {
539    const reply = await $.model.complete({
540      model: FORECAST_MODEL,
541      system: ask.system,
542      prompt: ask.prompt,
543      maxTokens: FORECAST_MAX_TOKENS,
544      timeoutMs: FORECAST_TIMEOUT_MS,
545    })
546    tokens = usageTokens(reply.usage)
547    if (reply.isAnswered) claims = parseForecast(reply.text, link.cwd)
548    else logOnce($, link, `forecast-${reply.reason}`, `forecast: no reply (${reply.reason}); sent without claims`)
549  } catch {
550    logOnce($, link, 'forecast-refused', 'forecast: the model call was refused; sent without claims')
551  } finally {
552    forecaster.state = closeForecast(forecaster.state, generation)
553  }
554  if (forecaster.state.generation !== generation) return
555  try {
556    await push($, link, 'turn.forecast', forecastPayload(runId, turnSeq, claims, FORECAST_MODEL, tokens))
557    flushAtTurn($, link)
558  } catch {
559    // unreported
560  }
561}
562
563/**
564 * A main-loop turn starts: count it, and when it carries the person's text and
565 * no forecast is still open, start one detached from the hook, so the turn
566 * never waits on the model.
567 */
568function beginForecast($: EngineInterface, link: Link, forecaster: Forecaster, turnId: string, text: string): void {
569  const { state, shouldAsk } = startTurn(forecaster.state, turnId, text, isActive(link))
570  forecaster.state = state
571  if (!shouldAsk) return
572  const run = (): void => {
573    void runForecast($, link, forecaster, state.runId, state.turnSeq, state.generation, forecastRequest(text, state.history))
574  }
575  try {
576    $.clock.after(0, run)
577  } catch {
578    run()
579  }
580}
581
582/** A main-loop turn ends: queue its actual when the turn was forecast. */
583async function finishForecast($: EngineInterface, link: Link, forecaster: Forecaster, turnId: string): Promise<void> {
584  const { state, actual } = endTurn(forecaster.state, turnId)
585  forecaster.state = state
586  if (actual !== undefined) await push($, link, 'turn.actual', actual)
587}
588
589/** How a tool call ended, from what `next(e)` resolved to. */
590const toolOutcome = (result: unknown): ToolOutcome => {
591  const r = result as { readonly deny?: unknown; readonly isError?: unknown } | undefined
592  if (r?.deny !== undefined) return 'denied'
593  return r?.isError === true ? 'error' : 'ok'
594}
595
596export const register: Register = (on, options) => {
597  const reserved = RESERVED.filter(feature => options[feature] === true)
598  const isControlOn = options.control === true
599  const endpoint = daemonEndpoint(options.daemonUrl)
600  /** The command loop; only ever set with `control` on. */
601  let control: Control | undefined
602  /** The forecast loop; undefined with `forecast` off, so none of it runs. */
603  const forecaster: Forecaster | undefined = options.forecast === true ? { state: INITIAL_FORECAST, isResumed: false } : undefined
604  const noted = new Set<string>()
605  const link: Link = {
606    url: endpoint.url,
607    token: undefined,
608    sessionId: '',
609    cwd: '',
610    start: undefined,
611    isClearing: false,
612    isOff: false,
613    isFailing: false,
614    isAuthWait: false,
615    hasRetriedAuth: false,
616    queue: [],
617    stamps: {},
618    timer: undefined,
619    flushing: undefined,
620    logged: new Set<string>(),
621  }
622
623  on('session.start', async ($, e, next) => {
624    const result = await next(e)
625    // also on `--resume`, which keeps the session id: a new run, so turn 1 is a new row
626    if (forecaster !== undefined) {
627      forecaster.isResumed = false
628      await newRun($, forecaster)
629    }
630    for (const feature of reserved) {
631      if (noted.has(feature)) continue
632      noted.add(feature)
633      log($, `${feature}: not implemented in 0.3`)
634    }
635    if (endpoint.isRefused && !noted.has('daemonUrl')) {
636      // the configured value is not echoed: it may carry credentials
637      noted.add('daemonUrl')
638      log($, `daemonUrl refused: not a loopback http origin (127.0.0.1, [::1], localhost); using ${DEFAULT_DAEMON_URL}`)
639    }
640    try {
641      link.token = await readToken($)
642      if (link.token === undefined) {
643        log($, 'no token, sensor off')
644        return result
645      }
646      link.sessionId = await $.session.id()
647      link.cwd = (await $.session.cwd()) || e.cwd
648      link.start = { surface: e.surface, isInteractive: e.isInteractive }
649      link.isClearing = false
650      await push($, link, 'session.start', link.start)
651      // one loop per session: a second session.start of the same session keeps it
652      if (isControlOn && !link.isOff && (control === undefined || control.isStopped || control.sessionId !== link.sessionId)) {
653        control = openControl($, link, endpoint.origin, control)
654      }
655    } catch {
656      // the sensor stays off; the session goes on
657    }
658    return result
659  })
660
661  // turn.start is the main loop's alone (TurnCompleteFields.turnId: "a subagent's run raises no turn.start").
662  on('turn.start', async ($, e, next) => {
663    if (link.isClearing && isActive(link)) {
664      try {
665        await resync($, link)
666        if (isControlOn && !link.isOff) control = openControl($, link, endpoint.origin, control)
667      } catch {
668        // the id is read again at the next turn.start
669      }
670    }
671    if (control !== undefined) control.turnId = e.turnId
672    // the resync after an in-process resume: the counter starts over under a new run
673    if (forecaster !== undefined && forecaster.isResumed) {
674      forecaster.isResumed = false
675      await newRun($, forecaster)
676    }
677    try {
678      await push($, link, 'turn.start')
679      flushAtTurn($, link)
680    } catch {
681      // never fail a turn over the sensor
682    }
683    // turn.start is the main loop's alone: a subagent's run raises none
684    if (forecaster !== undefined) beginForecast($, link, forecaster, e.turnId, e.text)
685    return next(e)
686  })
687
688  on('tool.call', async ($, e, next) => {
689    if (isActive(link)) {
690      try {
691        const fields = e as unknown as Readonly<Record<string, unknown>>
692        await push($, link, 'tool.call', {
693          tool: String(e.tool),
694          inputHash: await inputHash(toolInput(fields)),
695          ...(e.agentId !== undefined ? { agentId: e.agentId } : {}),
696        })
697      } catch {
698        // the call goes on unreported
699      }
700    }
701    if (forecaster === undefined || e.agentId !== undefined || forecaster.state.turnId === undefined) return next(e)
702    const result = await next(e)
703    try {
704      const facts = { tool: String(e.tool), agentId: e.agentId, fields: e as unknown as Readonly<Record<string, unknown>> }
705      forecaster.state = noteToolCall(forecaster.state, facts, toolOutcome(result), link.cwd)
706    } catch {
707      // the journal misses one call; the call's result stands
708    }
709    return result
710  }).catch(($, e, next) => next(e))
711
712  on('session.measure', async ($, e, next) => {
713    try {
714      const payload = measurePayload(e)
715      if (payload !== undefined) await push($, link, 'session.measure', payload)
716    } catch {
717      // unreported
718    }
719    return next(e)
720  })
721
722  on('turn.complete', async ($, e, next) => {
723    // A subagent's run ends in a turn.complete too; only the main loop's means the session waits.
724    if (control !== undefined && e.agentId === undefined && control.turnId === e.turnId) control.turnId = undefined
725    if (e.agentId !== undefined || !isActive(link)) return next(e)
726    try {
727      await push($, link, 'turn.complete', { durationMs: e.durationMs, stopReason: e.reason })
728      if (forecaster !== undefined) await finishForecast($, link, forecaster, e.turnId)
729      const snapshot = measurePayload(await $.session.usage())
730      if (snapshot !== undefined) await push($, link, 'session.measure', snapshot)
731      flushAtTurn($, link)
732    } catch {
733      // unreported
734    }
735    return next(e)
736  })
737
738  on('session.end', async ($, e, next) => {
739    if (control !== undefined) stopControl($, control, undefined)
740    // a /clear goes on in this process with no session.start: the turn count starts over here;
741    // an in-process resume starts it over at the next turn.start's resync
742    if (forecaster !== undefined && e.reason === 'clear') await newRun($, forecaster)
743    if (forecaster !== undefined && e.reason === 'resume') forecaster.isResumed = true
744    if (!isActive(link)) return next(e)
745    try {
746      await push($, link, 'session.end', { reason: e.reason })
747      link.timer?.cancel()
748      link.timer = undefined
749      // bounded by its own timer: a flush in flight may have a 5 s request timeout
750      await within($, flushForEnd($, link), END_FLUSH_TIMEOUT_MS)
751    } catch {
752      // best effort inside session.end's 1.5 s bound
753    }
754    // `/clear` or an in-process resume: the process goes on under another id with no
755    // session.start (SessionEndInput.sessionId); the next turn.start picks it up
756    if (e.reason === 'clear' || e.reason === 'resume') link.isClearing = true
757    return next(e)
758  })
759}
760

1// The link client's decisions. Pure: builds the request and turns a response
2// into what the queue does next. The `$.http.fetch` call itself is made in
3// register.ts, where every `$` call is spelled.
4// Contract: tools/swarmery/docs/mods-link-protocol.md.
5
6export const DEFAULT_DAEMON_URL = 'http://127.0.0.1:7777'
7export const EVENTS_PATH = '/api/link/events'
8
9/** How long a request may run before the queue treats it as failed. */
10export const REQUEST_TIMEOUT_MS = 5_000
11/** session.end's whole wall-clock bound is 1.5 s; the final flush keeps inside it. */
12export const END_FLUSH_TIMEOUT_MS = 1_200
13/** The periodic flush. */
14export const FLUSH_EVERY_MS = 2_000
15/** After a 401, the one retry waits past the daemon's 1 s token re-read throttle. */
16export const AUTH_RETRY_MS = 1_100
17
18/**
19 * A loopback http origin: scheme, one of the three loopback hosts, an optional
20 * port, then nothing or a path/query/fragment (dropped). No userinfo, no other
21 * host, no https: the bearer token goes nowhere else.
22 */
23const LOOPBACK_ORIGIN = /^http:\/\/(127\.0\.0\.1|\[::1\]|localhost)(?::(\d{1,5}))?(?:[/?#].*)?$/i
24
25export type DaemonEndpoint = {
26  /** The daemon's loopback origin, no path: the command channel's URLs hang off it. */
27  readonly origin: string
28  /** Where events are POSTed. */
29  readonly url: string
30  /** A daemonUrl was configured but is not a loopback http origin, so the default is used. */
31  readonly isRefused: boolean
32}
33
34/** The events endpoint under the configured loopback origin; the default origin for anything else. */
35export const daemonEndpoint = (daemonUrl: unknown): DaemonEndpoint => {
36  const fallback: DaemonEndpoint = { origin: DEFAULT_DAEMON_URL, url: `${DEFAULT_DAEMON_URL}${EVENTS_PATH}`, isRefused: false }
37  if (daemonUrl === undefined || daemonUrl === null) return fallback
38  const raw = typeof daemonUrl === 'string' ? daemonUrl.trim() : undefined
39  if (raw === '') return fallback
40  const match = raw === undefined ? null : LOOPBACK_ORIGIN.exec(raw)
41  const host = match?.[1]?.toLowerCase()
42  const port = match?.[2] === undefined ? undefined : Number(match[2])
43  if (host === undefined || (port !== undefined && (port < 1 || port > 65_535))) return { ...fallback, isRefused: true }
44  const origin = `http://${host}${port === undefined ? '' : `:${port}`}`
45  return { origin, url: `${origin}${EVENTS_PATH}`, isRefused: false }
46}
47
48export type LinkRequest = {
49  readonly method: 'POST'
50  readonly headers: Record<string, string>
51  readonly body: string
52}
53
54export const linkRequest = (token: string, body: string): LinkRequest => ({
55  method: 'POST',
56  headers: { authorization: `Bearer ${token}`, 'content-type': 'application/json' },
57  body,
58})
59
60/**
61 * What a response means for the queue:
62 * - `sent`: the batch was processed (202); drop it, rejected events included.
63 * - `auth`: 401; re-read the token once and retry once, past the throttle.
64 * - `off`: 503 link disabled; the link is off for this session.
65 * - `discard`: a request the daemon will never take (400, 403, 413 and other
66 *   4xx); drop the batch, log once.
67 * - `retry`: 5xx, a timeout or no answer; keep the batch for the next flush.
68 */
69export type Outcome = 'sent' | 'auth' | 'off' | 'discard' | 'retry'
70
71export const outcome = (status: number | undefined): Outcome => {
72  if (status === undefined) return 'retry'
73  if (status >= 200 && status < 300) return 'sent'
74  if (status === 401) return 'auth'
75  if (status === 503) return 'off'
76  if (status >= 400 && status < 500) return 'discard'
77  return 'retry'
78}
79
80/** The token file's text as a bearer credential, or undefined when it holds none. */
81export const parseToken = (text: unknown): string | undefined => {
82  if (typeof text !== 'string') return undefined
83  const token = text.trim()
84  return token === '' || /\s/.test(token) ? undefined : token
85}
86
87/** A one-line reason for the log, never the token or a request body. */
88export const describeFailure = (status: number | undefined, error: unknown): string => {
89  if (status !== undefined) return `HTTP ${status}`
90  if (error === 'timeout') return `no answer in ${REQUEST_TIMEOUT_MS / 1000}s`
91  const text = error instanceof Error ? error.message : String(error)
92  return text.split('\n')[0]?.slice(0, 160) ?? 'unknown error'
93}
94

1// The command channel's decisions. Pure: builds the poll and ack requests,
2// reads a command out of a reply, decides what to do with it, and keeps the
3// set of commands already executed. Every `$` call (fetch, clock, turn.abort,
4// prompt.submit) is made in register.ts.
5// Contract: tools/swarmery/docs/mods-link-protocol.md, "Commands".
6
7export const COMMANDS_PATH = '/api/link/commands'
8/** How long the daemon may hold a poll open (it clamps anything above 25). */
9export const POLL_WAIT_S = 25
10/** A poll's own bound, past the daemon's 25 s hold: HttpInit has no timeout. */
11export const POLL_TIMEOUT_MS = 28_000
12/** After a failed poll (no answer, a 5xx, an unreadable reply). */
13export const FAILURE_BACKOFF_MS = 3_000
14/** Two polls that came back empty start at least this far apart. */
15export const MIN_POLL_GAP_MS = 1_000
16/** `$.prompt.submit` resolves as its turn starts; past this, the prompt is left queued. */
17export const PROMPT_START_TIMEOUT_MS = 10_000
18/** The exact ack message the daemon reads as "deliver again after the next turn.complete". */
19export const TURN_IN_PROGRESS = 'turn in progress'
20/** The daemon keeps an ack message up to 500 characters. */
21export const MAX_ACK_MESSAGE = 500
22/** Executed ids remembered for re-delivery, newest kept. */
23export const MAX_REMEMBERED = 200
24
25export const commandsUrl = (origin: string, sessionId: string): string =>
26  `${origin}${COMMANDS_PATH}?session=${encodeURIComponent(sessionId)}&wait=${POLL_WAIT_S}`
27
28export const ackUrl = (origin: string, id: string): string => `${origin}${COMMANDS_PATH}/${encodeURIComponent(id)}/ack`
29
30export type PollRequest = {
31  readonly method: 'GET'
32  readonly headers: Record<string, string>
33}
34
35export const pollRequest = (token: string): PollRequest => ({
36  method: 'GET',
37  headers: { authorization: `Bearer ${token}` },
38})
39
40/**
41 * What a poll's reply means for the loop:
42 * - `command`: 200, a command to execute and ack, then poll again at once.
43 * - `idle`: 204, nothing deliverable; poll again.
44 * - `auth`: 401; re-read the token once, else stop.
45 * - `off`: 503 link disabled; stop.
46 * - `retry`: anything else, a timeout or no answer; back off, then poll again.
47 */
48export type PollOutcome = 'command' | 'idle' | 'auth' | 'off' | 'retry'
49
50export const pollOutcome = (status: number | undefined): PollOutcome => {
51  if (status === 200) return 'command'
52  if (status === 204) return 'idle'
53  if (status === 401) return 'auth'
54  if (status === 503) return 'off'
55  return 'retry'
56}
57
58/** How long to wait before the next poll, given how the last one ended and how long it took. */
59export const nextPollDelay = (outcome: 'command' | 'idle' | 'retry', elapsedMs: number): number => {
60  if (outcome === 'command') return 0
61  if (outcome === 'retry') return FAILURE_BACKOFF_MS
62  return Math.max(0, MIN_POLL_GAP_MS - Math.max(0, elapsedMs))
63}
64
65export type Command =
66  | { readonly id: string; readonly kind: 'stop' }
67  | { readonly id: string; readonly kind: 'prompt'; readonly text: string; readonly asUser: boolean }
68  | { readonly id: string; readonly kind: 'invalid'; readonly reason: string }
69
70const isRecord = (value: unknown): value is Readonly<Record<string, unknown>> =>
71  typeof value === 'object' && value !== null && !Array.isArray(value)
72
73const commandId = (value: unknown): string | undefined => {
74  if (typeof value === 'number' && Number.isSafeInteger(value) && value > 0) return String(value)
75  if (typeof value === 'string' && /^[1-9]\d{0,18}$/.test(value)) return value
76  return undefined
77}
78
79/**
80 * The command a 200 carries, or undefined when the reply names no id to ack
81 * (the loop then backs off as on a failure). A command it cannot run comes
82 * back `invalid`, to be acked failed.
83 */
84export const parseCommand = (text: unknown): Command | undefined => {
85  if (typeof text !== 'string') return undefined
86  let body: unknown
87  try {
88    body = JSON.parse(text)
89  } catch {
90    return undefined
91  }
92  if (!isRecord(body)) return undefined
93  const id = commandId(body.id)
94  if (id === undefined) return undefined
95  const payload = isRecord(body.payload) ? body.payload : {}
96  if (body.kind === 'stop') return { id, kind: 'stop' }
97  if (body.kind === 'prompt') {
98    const text = payload.text
99    if (typeof text !== 'string' || text.trim() === '') return { id, kind: 'invalid', reason: 'prompt without text' }
100    return { id, kind: 'prompt', text, asUser: payload.asUser === true }
101  }
102  return { id, kind: 'invalid', reason: `unknown command kind: ${String(body.kind).slice(0, 40)}` }
103}
104
105export type Ack = { readonly result: 'done' | 'failed'; readonly message?: string }
106
107export const done = (message?: string): Ack => (message === undefined ? { result: 'done' } : { result: 'done', message })
108export const failed = (message: string): Ack => ({ result: 'failed', message })
109
110export const ackBody = (ack: Ack): string =>
111  JSON.stringify(ack.message === undefined ? { result: ack.result } : { result: ack.result, message: ack.message.slice(0, MAX_ACK_MESSAGE) })
112
113/** The executed commands by id, with the ack each got: a re-delivery is acked the same, never run again. */
114export type Executed = ReadonlyMap<string, Ack>
115
116/**
117 * Records `ack` under `id`, keeping the newest MAX_REMEMBERED. A prompt
118 * refused for a running turn is not recorded: the daemon delivers it again
119 * after the turn, and then it runs.
120 */
121export const remember = (executed: Executed, id: string, ack: Ack): Executed => {
122  if (ack.result === 'failed' && ack.message === TURN_IN_PROGRESS) return executed
123  const next = new Map(executed)
124  next.delete(id)
125  next.set(id, ack)
126  const extra = next.size - MAX_REMEMBERED
127  if (extra > 0) for (const old of [...next.keys()].slice(0, extra)) next.delete(old)
128  return next
129}
130
131export type Action =
132  | { readonly do: 'ack'; readonly ack: Ack }
133  | { readonly do: 'abort'; readonly turnId: string }
134  | { readonly do: 'submit'; readonly text: string; readonly asUser: boolean }
135
136/**
137 * What to do with a delivered command, given the commands already executed
138 * and the main loop's running turn (`turn.start`'s id, until its
139 * `turn.complete`).
140 */
141export const decide = (command: Command, executed: Executed, runningTurnId: string | undefined): Action => {
142  const previous = executed.get(command.id)
143  if (previous !== undefined) return { do: 'ack', ack: previous }
144  switch (command.kind) {
145    case 'invalid':
146      return { do: 'ack', ack: failed(command.reason) }
147    case 'stop':
148      return runningTurnId === undefined ? { do: 'ack', ack: done('no turn running') } : { do: 'abort', turnId: runningTurnId }
149    case 'prompt':
150      return runningTurnId === undefined
151        ? { do: 'submit', text: command.text, asUser: command.asUser }
152        : { do: 'ack', ack: failed(TURN_IN_PROGRESS) }
153  }
154}
155
156/** A one-line, bounded reason for an ack or a log line. */
157export const shortError = (error: unknown): string => {
158  const text = error instanceof Error ? error.message : String(error)
159  return (text.split('\n')[0] ?? '').slice(0, 160) || 'unknown error'
160}
161
162/** Whether a rejected `$.prompt.submit` says a turn is running. */
163export const isTurnRunning = (error: unknown): boolean =>
164  /\bturn\b.*\b(running|in progress|active|busy)\b|\bnot idle\b|\bbusy\b/i.test(shortError(error))
165
166/** The ack for a `$.prompt.submit` that rejected. */
167export const submitFailure = (error: unknown): Ack => (isTurnRunning(error) ? failed(TURN_IN_PROGRESS) : failed(shortError(error)))
168
169/** The ack for a `$.prompt.submit` a hook answered with `{ drop }`: the same rule, the reason marked as a drop. */
170export const dropAck = (reason: string): Ack => (isTurnRunning(reason) ? failed(TURN_IN_PROGRESS) : failed(`dropped: ${shortError(reason)}`))
171

1// The turn forecast loop (option `forecast`). Pure: every function takes the
2// loop's state or a raw value and returns the next state or a payload. The
3// `$` calls (the model call, the push into the sensor queue) are spelled in
4// register.ts. The wire shapes are the link protocol's `turn.forecast` and
5// `turn.actual` (tools/swarmery/docs/mods-link-protocol.md): paths and numbers
6// only, never prompt text, tool input or model reply text.
7
8/** The model asked: an alias the engine resolves like `--model`. */
9export const FORECAST_MODEL = 'haiku'
10/** The reply's token cap: a forecast is one short JSON object. */
11export const FORECAST_MAX_TOKENS = 300
12/** The model call's own time limit; past it the call resolves `aborted`. */
13export const FORECAST_TIMEOUT_MS = 20_000
14/** The most of the user's text the model is shown. */
15export const PROMPT_TEXT_LIMIT = 2_000
16/** Completed turns whose touched files the model is shown. */
17export const HISTORY_TURNS = 3
18/** The daemon's limits on `files`. */
19export const MAX_FILES = 50
20export const MAX_PATH = 300
21/** Random base36 characters after a run id's time: enough that two loads in the same millisecond differ. */
22export const RUN_ID_RANDOM = 6
23
24/** A Bash command that runs tests (the protocol's pattern). */
25export const TEST_COMMAND = /\b(test|vitest|jest|go test|pytest|cargo test|npm (run )?test)\b/
26
27/** Tools whose input names the file they write, and the field that names it. */
28const FILE_TOOLS: Readonly<Record<string, string>> = {
29  Edit: 'file_path',
30  Write: 'file_path',
31  MultiEdit: 'file_path',
32  NotebookEdit: 'notebook_path',
33}
34
35/** What the model claimed about the coming turn; null where it made no claim. */
36export type Claims = {
37  readonly files: readonly string[]
38  readonly testsPass: boolean | null
39  readonly steps: number | null
40  readonly confidence: number | null
41}
42
43export const NO_CLAIMS: Claims = { files: [], testsPass: null, steps: null, confidence: null }
44
45/** What the main loop did this turn, from its tool calls. */
46export type Journal = {
47  readonly files: readonly string[]
48  readonly testsRan: boolean
49  readonly testsPass: boolean | null
50  readonly steps: number
51}
52
53export const EMPTY_JOURNAL: Journal = { files: [], testsRan: false, testsPass: null, steps: 0 }
54
55/** The loop's state for one load of the module. */
56export type ForecastState = {
57  /**
58   * Names this run of the turn counter; renewed with every reset (session.start,
59   * a /clear, an in-process resume). A `--resume` keeps the session id and counts
60   * from 1 again, so the daemon keys a turn on (session, runId, turnSeq).
61   * '' before the first reset: no turn is forecast without one.
62   */
63  readonly runId: string
64  /** The main loop's turn count since the run started. */
65  readonly turnSeq: number
66  /** Bumped on every reset, so a forecast in flight from before it does not reopen the gate. */
67  readonly generation: number
68  /** The running main-loop turn's id; undefined between turns. */
69  readonly turnId: string | undefined
70  /** This turn asked the model, so its actual is sent. */
71  readonly isForecasted: boolean
72  /** A forecast call is still in flight: the next turn is not forecast. */
73  readonly isOpen: boolean
74  readonly journal: Journal
75  /** Files touched per completed turn, newest last, at most HISTORY_TURNS. */
76  readonly history: ReadonlyArray<readonly string[]>
77}
78
79export const INITIAL_FORECAST: ForecastState = {
80  runId: '',
81  turnSeq: 0,
82  generation: 0,
83  turnId: undefined,
84  isForecasted: false,
85  isOpen: false,
86  journal: EMPTY_JOURNAL,
87  history: [],
88}
89
90/**
91 * The state after session.start, a session.end with reason `clear`, or the
92 * resync after an in-process resume: numbering starts over under `runId`.
93 */
94export const resetForecast = (state: ForecastState, runId: string): ForecastState => ({
95  ...INITIAL_FORECAST,
96  runId,
97  generation: state.generation + 1,
98})
99
100/**
101 * A run id: the reset's time in ms since the epoch, then `-` and RUN_ID_RANDOM
102 * base36 characters drawn from `random` when there is a random source, the
103 * time alone when there is none. Always within the daemon's
104 * `[A-Za-z0-9._:-]{1,64}`.
105 */
106export const runIdOf = (nowMs: number, random: Uint8Array | undefined): string => {
107  const ms = Number.isFinite(nowMs) && nowMs >= 0 ? Math.floor(nowMs) : 0
108  const time = String(ms)
109  if (random === undefined || random.length < RUN_ID_RANDOM) return time
110  const tail = Array.from(random.subarray(0, RUN_ID_RANDOM), b => (b % 36).toString(36)).join('')
111  return `${time}-${tail}`
112}
113
114/**
115 * Whether a turn's text was typed by the person: empty text (a continuation)
116 * and the forms a command or a notification arrives in are not.
117 */
118export const isUserText = (text: string): boolean => {
119  const t = text.trim()
120  return t !== '' && !/^(\/|<(command-|local-command|task-notification|system-reminder))/.test(t)
121}
122
123/** The state at a main-loop turn.start, and whether to ask the model for this turn. */
124export const startTurn = (
125  state: ForecastState,
126  turnId: string,
127  text: string,
128  canSend: boolean,
129): { readonly state: ForecastState; readonly shouldAsk: boolean } => {
130  const shouldAsk = canSend && state.runId !== '' && !state.isOpen && isUserText(text)
131  return {
132    shouldAsk,
133    state: {
134      ...state,
135      turnSeq: state.turnSeq + 1,
136      turnId,
137      isForecasted: shouldAsk,
138      isOpen: state.isOpen || shouldAsk,
139      journal: EMPTY_JOURNAL,
140    },
141  }
142}
143
144/** The state once the forecast call of `generation` settled. */
145export const closeForecast = (state: ForecastState, generation: number): ForecastState =>
146  state.generation === generation ? { ...state, isOpen: false } : state
147
148/**
149 * The state at the main loop's turn.complete, and the turn.actual payload when
150 * the turn was forecast. The actual carries the run the turn started in: a
151 * reset between start and complete clears `turnId`, so no actual is built
152 * across one.
153 */
154export const endTurn = (
155  state: ForecastState,
156  turnId: string,
157): { readonly state: ForecastState; readonly actual: Readonly<Record<string, unknown>> | undefined } => {
158  if (state.turnId === undefined || state.turnId !== turnId) return { state, actual: undefined }
159  const { journal } = state
160  const actual = state.isForecasted
161    ? { runId: state.runId, turnSeq: state.turnSeq, files: journal.files, testsRan: journal.testsRan, testsPass: journal.testsPass, steps: journal.steps }
162    : undefined
163  return {
164    actual,
165    state: {
166      ...state,
167      turnId: undefined,
168      isForecasted: false,
169      journal: EMPTY_JOURNAL,
170      history: [...state.history, journal.files].slice(-HISTORY_TURNS),
171    },
172  }
173}
174
175/** `path` with `./`, duplicate and trailing separators gone, relative to `cwd` when under it. */
176export const normalizePath = (path: string, cwd: string): string => {
177  const clean = (p: string): string =>
178    p
179      .trim()
180      .replace(/\/{2,}/g, '/')
181      .replace(/^(\.\/)+/, '')
182      .replace(/(.)\/$/, '$1')
183  const p = clean(path)
184  const root = clean(cwd)
185  return root !== '' && root !== '/' && p.startsWith(`${root}/`) ? p.slice(root.length + 1) : p
186}
187
188/**
189 * Whether `p` can stand on the wire as a path: one token of at most MAX_PATH
190 * characters naming a directory or an extension. A model that echoes prose
191 * into `files` has it dropped here, so prompt text never rides along.
192 */
193const isWirePath = (p: string): boolean =>
194  p.length > 0 && p.length <= MAX_PATH && !/[\s\u0000-\u001f<>"'`|;]/.test(p) && /[./]/.test(p)
195
196/** The paths added to `files` in order, each once, valid ones only, at most MAX_FILES. */
197const addFiles = (files: readonly string[], more: readonly string[]): readonly string[] => {
198  const out = [...files]
199  for (const p of more) {
200    if (out.length >= MAX_FILES) break
201    if (isWirePath(p) && !out.includes(p)) out.push(p)
202  }
203  return out
204}
205
206/** A tool call as the journal reads it: the tool, its loop, and the input fields it may name. */
207export type ToolCallFacts = {
208  readonly tool: string
209  readonly agentId?: string | undefined
210  readonly fields: Readonly<Record<string, unknown>>
211}
212
213/** How the call ended, from `next(e)`'s result. */
214export type ToolOutcome = 'ok' | 'error' | 'denied'
215
216/**
217 * The journal after one tool call of the running turn ended with `outcome`.
218 * A subagent's call (agentId present), or a call outside a turn, leaves it as
219 * it was. Every main-loop call is a step; a file counts once its write
220 * succeeded; a test command run in the foreground sets testsRan and testsPass
221 * (the last one wins: a red run fixed later in the turn reads as passing).
222 */
223export const noteToolCall = (state: ForecastState, call: ToolCallFacts, outcome: ToolOutcome, cwd: string): ForecastState => {
224  if (state.turnId === undefined || call.agentId !== undefined) return state
225  let journal: Journal = { ...state.journal, steps: state.journal.steps + 1 }
226  const field = FILE_TOOLS[call.tool]
227  const target = field === undefined ? undefined : call.fields[field]
228  if (typeof target === 'string' && outcome === 'ok') {
229    journal = { ...journal, files: addFiles(journal.files, [normalizePath(target, cwd)]) }
230  }
231  const command = call.tool === 'Bash' ? call.fields.command : undefined
232  if (typeof command === 'string' && TEST_COMMAND.test(command) && outcome !== 'denied') {
233    const isBackground = call.fields.run_in_background === true
234    journal = { ...journal, testsRan: true, testsPass: isBackground ? journal.testsPass : outcome === 'ok' }
235  }
236  return { ...state, journal }
237}
238
239/** The model's request: a short fixed instruction, the user's text and the files of the last turns. */
240export const forecastRequest = (text: string, history: ReadonlyArray<readonly string[]>): { readonly system: string; readonly prompt: string } => {
241  const recent = [...new Set(history.flat())]
242  const shown = text.length > PROMPT_TEXT_LIMIT ? `${text.slice(0, PROMPT_TEXT_LIMIT)}…` : text
243  return {
244    system: [
245      'You forecast what a coding agent will do while it answers the request below.',
246      'Reply with one JSON object and nothing else:',
247      '{"files": [paths it will create or edit, relative to the repository root], "testsPass": true | false | null (null when it will not run tests),',
248      '"steps": number of tool calls it will make, "confidence": 0..1 that the forecast is right}.',
249    ].join(' '),
250    prompt: [
251      'Request:',
252      shown,
253      '',
254      'Files edited in the previous turns:',
255      recent.length > 0 ? recent.map(p => `- ${p}`).join('\n') : '(none)',
256    ].join('\n'),
257  }
258}
259
260/** The first balanced `{...}` in `text`, strings respected; undefined when there is none. */
261const firstObject = (text: string): string | undefined => {
262  const start = text.indexOf('{')
263  if (start < 0) return undefined
264  let depth = 0
265  let inString = false
266  let escaped = false
267  for (let i = start; i < text.length; i += 1) {
268    const c = text[i]
269    if (inString) {
270      if (escaped) escaped = false
271      else if (c === '\\') escaped = true
272      else if (c === '"') inString = false
273      continue
274    }
275    if (c === '"') inString = true
276    else if (c === '{') depth += 1
277    else if (c === '}') {
278      depth -= 1
279      if (depth === 0) return text.slice(start, i + 1)
280    }
281  }
282  return undefined
283}
284
285const isRecord = (v: unknown): v is Readonly<Record<string, unknown>> => typeof v === 'object' && v !== null && !Array.isArray(v)
286
287/**
288 * The claims in a model reply: the first JSON object in it, each field held to
289 * the protocol's limits; a field out of shape is null, a reply with no
290 * readable object is NO_CLAIMS.
291 */
292export const parseForecast = (reply: string | undefined, cwd: string): Claims => {
293  const raw = reply === undefined ? undefined : firstObject(reply)
294  if (raw === undefined) return NO_CLAIMS
295  let value: unknown
296  try {
297    value = JSON.parse(raw)
298  } catch {
299    return NO_CLAIMS
300  }
301  if (!isRecord(value)) return NO_CLAIMS
302  const files = Array.isArray(value.files)
303    ? addFiles(
304        [],
305        value.files.filter((f): f is string => typeof f === 'string').map(f => normalizePath(f, cwd)),
306      )
307    : []
308  const steps = typeof value.steps === 'number' && Number.isFinite(value.steps) ? Math.max(0, Math.round(value.steps)) : null
309  const confidence =
310    typeof value.confidence === 'number' && Number.isFinite(value.confidence) ? Math.min(1, Math.max(0, value.confidence)) : null
311  return {
312    files,
313    testsPass: typeof value.testsPass === 'boolean' ? value.testsPass : null,
314    steps,
315    confidence,
316  }
317}
318
319/** The four token counts of a model call, as ModelUsage spells them. */
320export type TokenUsage = {
321  readonly input_tokens: number
322  readonly output_tokens: number
323  readonly cache_read_input_tokens: number
324  readonly cache_creation_input_tokens: number
325}
326
327/** Every token the call was billed for, input (cached or not) and output; null when the reply carried no usage. */
328export const usageTokens = (usage: TokenUsage | undefined): number | null => {
329  if (usage === undefined) return null
330  const sum = usage.input_tokens + usage.output_tokens + usage.cache_read_input_tokens + usage.cache_creation_input_tokens
331  return Number.isFinite(sum) && sum >= 0 ? Math.round(sum) : null
332}
333
334/** The turn.forecast payload; `runId` is the run the forecast turn started in. */
335export const forecastPayload = (
336  runId: string,
337  turnSeq: number,
338  claims: Claims,
339  model: string,
340  forecastTokens: number | null,
341): Readonly<Record<string, unknown>> => ({
342  runId,
343  turnSeq,
344  files: claims.files,
345  testsPass: claims.testsPass,
346  steps: claims.steps,
347  confidence: claims.confidence,
348  model: model.slice(0, 80),
349  forecastTokens,
350})
351

1// The tool-call input hash. Pure: takes a value, returns its digest. The link
2// sends this in place of the input, which never leaves the session
3// (mods-link-protocol.md, "tool.call carries no input").
4
5/** Hex digits kept of a digest: 64 bits, enough to match two calls, too few to search. */
6export const HASH_HEX = 16
7
8const isRecord = (v: unknown): v is Readonly<Record<string, unknown>> =>
9  typeof v === 'object' && v !== null && !Array.isArray(v)
10
11/** JSON with object keys sorted at every depth, so equal inputs hash equal. */
12export const canonicalJson = (value: unknown): string => {
13  if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]`
14  if (isRecord(value)) {
15    const fields = Object.keys(value)
16      .filter(k => value[k] !== undefined)
17      .sort()
18      .map(k => `${JSON.stringify(k)}:${canonicalJson(value[k])}`)
19    return `{${fields.join(',')}}`
20  }
21  const text = JSON.stringify(value)
22  return text === undefined ? 'null' : text
23}
24
25const toHex = (bytes: Uint8Array): string => Array.from(bytes, b => b.toString(16).padStart(2, '0')).join('')
26
27/** FNV-1a, 64 bits, as 16 hex digits: the fallback where `crypto.subtle` is missing. */
28export const fnv1a64 = (text: string): string => {
29  const bytes = new TextEncoder().encode(text)
30  let hash = 0xcbf29ce484222325n
31  for (const b of bytes) {
32    hash ^= BigInt(b)
33    hash = (hash * 0x100000001b3n) & 0xffffffffffffffffn
34  }
35  return hash.toString(16).padStart(16, '0')
36}
37
38/** `HASH_HEX` hex digits of SHA-256 over the canonical JSON, FNV-1a where SHA-256 is unavailable. */
39export const inputHash = async (input: unknown): Promise<string> => {
40  const text = canonicalJson(input)
41  try {
42    const subtle = typeof crypto === 'undefined' ? undefined : crypto.subtle
43    if (subtle !== undefined) {
44      const digest = await subtle.digest('SHA-256', new TextEncoder().encode(text))
45      return toHex(new Uint8Array(digest)).slice(0, HASH_HEX)
46    }
47  } catch {
48    // fall through to the pure hash
49  }
50  return fnv1a64(text).slice(0, HASH_HEX)
51}
52
53/** The tool's arguments alone: the envelope fields the engine spreads beside them removed. */
54export const toolInput = (e: Readonly<Record<string, unknown>>): Record<string, unknown> => {
55  const { tool: _tool, tool_use_id: _id, agentId: _agent, requestMeta: _meta, ...args } = e
56  return args
57}
58

1// The outgoing event queue. Pure: every function takes the queue's state and
2// returns the next one. The wire shape and limits are the link protocol's
3// (tools/swarmery/docs/mods-link-protocol.md, POST /api/link/events).
4
5export type LinkKind =
6  | 'session.start'
7  | 'turn.start'
8  | 'tool.call'
9  | 'session.measure'
10  | 'turn.complete'
11  | 'session.end'
12  | 'turn.forecast'
13  | 'turn.actual'
14
15export type LinkEvent = {
16  readonly sessionId: string
17  readonly cwd: string
18  readonly kind: LinkKind
19  /** RFC 3339, ms, UTC. Never rewritten once stamped: a resend carries the same `at`. */
20  readonly at: string
21  readonly payload?: Readonly<Record<string, unknown>>
22}
23
24/** Events a batch may carry, and the most the queue holds while the daemon is away. */
25export const MAX_EVENTS = 200
26/** The daemon's body limit is 1 MiB; a batch stays under it with room for the envelope. */
27export const MAX_BODY_BYTES = 1_000_000
28
29/** The last `at` (epoch ms) stamped per kind. */
30export type Stamps = Readonly<Partial<Record<LinkKind, number>>>
31
32/**
33 * The event's time: `now`, or one millisecond past the last event of the same
34 * kind when `now` does not move past it. The daemon's dedup key is
35 * (session, kind, at), so two distinct events of a kind must never share an `at`.
36 */
37export const stamp = (stamps: Stamps, kind: LinkKind, now: number): { readonly at: string; readonly stamps: Stamps } => {
38  const last = stamps[kind]
39  const ms = last !== undefined && now <= last ? last + 1 : now
40  return { at: new Date(ms).toISOString(), stamps: { ...stamps, [kind]: ms } }
41}
42
43const key = (e: LinkEvent): string => `${e.kind}\u0000${e.at}`
44
45/**
46 * The queue with `event` appended: an event already queued under the same
47 * (kind, at) is not queued twice, and past `cap` the oldest events are dropped.
48 */
49export const enqueue = (queue: readonly LinkEvent[], event: LinkEvent, cap: number = MAX_EVENTS): readonly LinkEvent[] => {
50  if (queue.some(q => key(q) === key(event))) return queue
51  const next = [...queue, event]
52  return next.length > cap ? next.slice(next.length - cap) : next
53}
54
55/** The queue without the events of `sent` (a batch the daemon answered), matched by (kind, at). */
56export const drop = (queue: readonly LinkEvent[], sent: readonly LinkEvent[]): readonly LinkEvent[] => {
57  const gone = new Set(sent.map(key))
58  return queue.filter(e => !gone.has(key(e)))
59}
60
61const bytes = (text: string): number => new TextEncoder().encode(text).length
62
63/**
64 * The next batch, oldest first: at most `MAX_EVENTS` events and a body under
65 * `MAX_BODY_BYTES`. An event too large to send alone is skipped (it can never
66 * be accepted) and returned in `oversized` so the caller can drop it.
67 */
68export const takeBatch = (
69  queue: readonly LinkEvent[],
70): { readonly batch: readonly LinkEvent[]; readonly body: string; readonly oversized: readonly LinkEvent[] } => {
71  const batch: LinkEvent[] = []
72  const oversized: LinkEvent[] = []
73  const envelope = bytes('{"events":[]}')
74  let size = envelope
75  for (const event of queue) {
76    if (batch.length >= MAX_EVENTS) break
77    const one = bytes(JSON.stringify(event))
78    if (envelope + one > MAX_BODY_BYTES) {
79      oversized.push(event)
80      continue
81    }
82    const added = one + (batch.length > 0 ? 1 : 0)
83    if (size + added > MAX_BODY_BYTES) break
84    batch.push(event)
85    size += added
86  }
87  return { batch, body: JSON.stringify({ events: batch }), oversized }
88}
89

1// The session.measure payload. Pure: normalizes what `session.measure`'s `e`
2// or `$.session.usage()` answered into the M2 shape the daemon decodes
3// (mods-format.md M2; testdata/fixtures/mods/session-usage.json).
4//
5// A unit the engine did not report is left out, never zeroed: the daemon keeps
6// a column's last value for an absent unit ("not reported" is not "zero").
7
8export type UsageUnit = 'context' | 'rateLimits' | 'cost'
9
10export type RateLimit = {
11  readonly kind: string
12  readonly percentUsed: number
13  readonly resetsAt?: string
14}
15
16export type MeasurePayload = {
17  readonly context?: { readonly tokens?: number; readonly window: number; readonly percent?: number }
18  readonly rateLimits?: readonly RateLimit[]
19  readonly cost?: { readonly usd: number }
20  readonly changed: readonly UsageUnit[]
21}
22
23const UNITS: readonly UsageUnit[] = ['context', 'rateLimits', 'cost']
24
25const isRecord = (v: unknown): v is Readonly<Record<string, unknown>> =>
26  typeof v === 'object' && v !== null && !Array.isArray(v)
27
28const num = (v: unknown): number | undefined => (typeof v === 'number' && Number.isFinite(v) ? v : undefined)
29
30const context = (v: unknown): MeasurePayload['context'] => {
31  if (!isRecord(v)) return undefined
32  const window = num(v.window)
33  if (window === undefined) return undefined
34  const tokens = num(v.tokens)
35  const percent = num(v.percent)
36  return {
37    ...(tokens !== undefined ? { tokens } : {}),
38    window,
39    ...(percent !== undefined ? { percent } : {}),
40  }
41}
42
43const rateLimit = (v: unknown): RateLimit | undefined => {
44  if (!isRecord(v) || typeof v.kind !== 'string') return undefined
45  const percentUsed = num(v.percentUsed)
46  if (percentUsed === undefined) return undefined
47  return typeof v.resetsAt === 'string'
48    ? { kind: v.kind, percentUsed, resetsAt: v.resetsAt }
49    : { kind: v.kind, percentUsed }
50}
51
52const rateLimits = (v: unknown): readonly RateLimit[] | undefined => {
53  if (!Array.isArray(v)) return undefined
54  const out = v.map(rateLimit).filter((r): r is RateLimit => r !== undefined)
55  return out.length > 0 ? out : undefined
56}
57
58const cost = (v: unknown): MeasurePayload['cost'] => {
59  if (!isRecord(v)) return undefined
60  const usd = num(v.usd)
61  return usd === undefined ? undefined : { usd }
62}
63
64const changed = (v: unknown, present: readonly UsageUnit[]): readonly UsageUnit[] => {
65  if (!Array.isArray(v)) return present
66  const named = UNITS.filter(u => v.includes(u))
67  return named.length > 0 ? named : present
68}
69
70/**
71 * The payload for one measurement. `raw` is `session.measure`'s `e` (which
72 * names what `changed`) or `$.session.usage()` (which does not: a snapshot
73 * names every unit it carries). Undefined when nothing usable was reported.
74 */
75export const measurePayload = (raw: unknown): MeasurePayload | undefined => {
76  if (!isRecord(raw)) return undefined
77  const ctx = context(raw.context)
78  const limits = rateLimits(raw.rateLimits)
79  const usd = cost(raw.cost)
80  const present = UNITS.filter(
81    u => (u === 'context' && ctx !== undefined) || (u === 'rateLimits' && limits !== undefined) || (u === 'cost' && usd !== undefined),
82  )
83  if (present.length === 0) return undefined
84  return {
85    ...(ctx !== undefined ? { context: ctx } : {}),
86    ...(limits !== undefined ? { rateLimits: limits } : {}),
87    ...(usd !== undefined ? { cost: usd } : {}),
88    changed: changed(raw.changed, present),
89  }
90}
91