agent-bus

1import { atom, read, update } from 'claude-code'
2import type { Register } from 'claude-code'
3
4import type { BusComposer, BusDetail, BusSnapshot, BusTask } from '../types'
5import { paneView, selectedOf } from './pane-view'
6import type { PaneActions, Ui } from './pane-view'
7import * as Words from './words'
8
9const snapshotAtom = atom(
10  { plugin: 'agent-bus', key: 'snapshot' } as const,
11  null,
12)
13const selectedAtom = atom({ plugin: 'agent-bus', key: 'selected' } as const, null)
14const detailAtom = atom({ plugin: 'agent-bus', key: 'detail' } as const, null)
15const armedAtom = atom({ plugin: 'agent-bus', key: 'armed' } as const, null)
16const composerAtom = atom(
17  { plugin: 'agent-bus', key: 'composer' } as const,
18  null,
19)
20
21/**
22 * How often this machine's queue folders' modification times are compared:
23 * a task queued, claimed or finished changes one, and only then is the bus
24 * asked.
25 */
26const WATCH_MS = 2_000
27
28/**
29 * How often another machine's bus is asked: no folder to watch over ssh.
30 */
31const REMOTE_POLL_MS = 5_000
32
33/**
34 * While a task runs, how often the bus is asked anyway, so its time moves.
35 */
36const ACTIVE_REFRESH_MS = 15_000
37
38const QUEUES = ['inbox', 'running', 'done', 'failed'] as const
39
40/**
41 * The `$.store` key remembering the folder last dispatched to on a machine.
42 */
43const cwdKeyOf = (host: string): string => `cwd:${host}`
44
45type RunResult = { exitCode: number; stdout: string; stderr: string }
46
47/**
48 * The engine as the plugin's background work reaches it: each member one
49 * call on the `$` that `session.start` was handed, bound there.
50 */
51type Host = {
52  run: (
53    argv: string[],
54    timeoutMs: number,
55    env?: Record<string, string>,
56  ) => Promise<RunResult>
57  now: () => Promise<number>
58  mtime: (path: string) => Promise<number>
59  toast: (text: string) => void
60  status: (text: string | undefined) => void
61  note: (text: string) => Promise<void>
62  storeGet: (key: string) => Promise<unknown>
63  storeSet: (key: string, value: unknown) => Promise<void>
64  getSelected: () => Promise<string | null>
65  setSelected: (key: string) => Promise<unknown>
66  getDetail: () => Promise<BusDetail | null>
67  setDetail: (detail: BusDetail) => Promise<unknown>
68  setSnapshot: (snapshot: BusSnapshot) => Promise<unknown>
69  toggleArmed: (key: string) => Promise<unknown>
70  getComposer: () => Promise<BusComposer | null>
71  editComposer: (
72    edit: (composer: BusComposer | null) => BusComposer | null,
73  ) => Promise<unknown>
74}
75
76const firstLine = (text: string): string =>
77  text.trim().split('\n')[0]?.slice(0, 200) ?? ''
78
79export const register: Register = (on, options) => {
80  const hosts = Words.hostsOf(options.hosts)
81  const hasLocal = hosts.includes(Words.LOCAL)
82  const hasRemote = hosts.some(machine => machine !== Words.LOCAL)
83
84  let host: Host | null = null
85  let home: string | undefined
86  let busHome = ''
87  let dispatch = 'agent-dispatch'
88  let sessionId = ''
89  let sessionCwd = ''
90
91  let signature = ''
92  let lastPollAt = 0
93  let isPolling = false
94  let latest: BusSnapshot | null = null
95  let seenFinished: Set<string> | null = null
96  let statusText: string | undefined
97  let carrying: string | null = null
98
99  /**
100   * Runs agent-dispatch with these arguments on a machine: this one's
101   * directly, another's over ssh; the origin, when given, rides in the
102   * dispatcher's environment either way.
103   */
104  function bus(
105    engine: Host,
106    machine: string,
107    args: string[],
108    timeoutMs: number,
109    origin?: string,
110  ): Promise<RunResult> {
111    if (machine === Words.LOCAL) {
112      return engine.run(
113        [dispatch, ...args],
114        timeoutMs,
115        origin ? { AGENT_BUS_ORIGIN: origin } : undefined,
116      )
117    }
118
119    return engine.run(
120      Words.sshArgvOf(machine, Words.remoteCommandOf(args, origin)),
121      timeoutMs + 5_000,
122    )
123  }
124
125  async function answerOf(
126    engine: Host,
127    machine: string,
128  ): Promise<Words.HostAnswer> {
129    const failure =
130      machine === Words.LOCAL
131        ? `Couldn't run ${Words.homeFolded(dispatch, home)}. Is the agent bus installed?`
132        : `Couldn't reach the bus on ${machine}.`
133
134    const run = await bus(
135      engine,
136      machine,
137      ['status', '--json', '--recent', '20'],
138      15_000,
139    ).catch(() => null)
140
141    if (run === null || run.exitCode !== 0) {
142      const why = run === null ? '' : firstLine(run.stderr)
143
144      return {
145        host: machine,
146        status: null,
147        error: why ? `${failure} ${why}` : failure,
148      }
149    }
150
151    try {
152      return { host: machine, status: JSON.parse(run.stdout), error: null }
153    } catch {
154      return {
155        host: machine,
156        status: null,
157        error: `${failure} It answered something other than JSON.`,
158      }
159    }
160  }
161
162  /**
163   * A finished task's result text, asked of the machine that ran it.
164   */
165  async function resultOf(engine: Host, task: BusTask): Promise<string | null> {
166    const run = await bus(engine, task.host, ['result', task.id], 15_000).catch(
167      () => null,
168    )
169
170    return run !== null && run.exitCode === 0 ? run.stdout : null
171  }
172
173  /**
174   * Toasts each task that finished since the last poll, unless another
175   * session dispatched it; a task this session dispatched is also written
176   * into the conversation, so the model knows where its result is.
177   *
178   * The first poll after a load only learns what had already finished.
179   */
180  async function announce(engine: Host, snapshot: BusSnapshot) {
181    const finished = snapshot.tasks.filter(task => !Words.isActive(task))
182
183    if (seenFinished === null) {
184      seenFinished = new Set(finished.map(Words.taskKeyOf))
185
186      return
187    }
188
189    for (const task of finished) {
190      const key = Words.taskKeyOf(task)
191
192      if (seenFinished.has(key)) continue
193
194      seenFinished.add(key)
195
196      if (task.origin && task.origin !== sessionId) continue
197
198      engine.toast(Words.toastOf(task))
199
200      if (task.origin === sessionId) {
201        await engine.note(Words.finishedNoteOf(task)).catch(() => undefined)
202      }
203    }
204  }
205
206  function syncStatus(engine: Host, snapshot: BusSnapshot) {
207    const text = Words.statusTextOf(snapshot)
208
209    if (text !== statusText) {
210      statusText = text
211      engine.status(text)
212    }
213  }
214
215  /**
216   * Reads the selected task's result into the detail once it has finished;
217   * an active task's detail draws its prompt from the snapshot instead.
218   */
219  async function refreshDetail(engine: Host) {
220    const task = selectedOf(latest, await engine.getSelected())
221    const detail = await engine.getDetail()
222
223    if (!task) return
224
225    const key = Words.taskKeyOf(task)
226
227    if (Words.isActive(task)) {
228      if (detail?.id !== key || detail.text !== null) {
229        await engine.setDetail({ id: key, text: null })
230      }
231
232      return
233    }
234
235    if (detail?.id === key && detail.text !== null) return
236
237    const text = await resultOf(engine, task)
238
239    const shown =
240      text === null
241        ? null
242        : Words.sanitize(
243            text.length > Words.DETAIL_MAX_CHARS
244              ? `${text.slice(0, Words.DETAIL_MAX_CHARS)}\n\n…`
245              : text,
246          )
247
248    await engine.setDetail({ id: key, text: shown })
249  }
250
251  async function poll(engine: Host) {
252    if (isPolling) return
253
254    isPolling = true
255
256    try {
257      const answers = await Promise.all(
258        hosts.map(machine => answerOf(engine, machine)),
259      )
260
261      lastPollAt = await engine.now()
262
263      const snapshot = Words.mergedSnapshotOf(answers, lastPollAt / 1000)
264
265      latest = snapshot
266
267      await announce(engine, snapshot)
268      await engine.setSnapshot(snapshot)
269      syncStatus(engine, snapshot)
270      await refreshDetail(engine)
271    } finally {
272      isPolling = false
273    }
274  }
275
276  async function tick(engine: Host) {
277    const now = await engine.now()
278
279    const stamps = hasLocal
280      ? await Promise.all(
281          QUEUES.map(queue => engine.mtime(`${busHome}/${queue}`)),
282        )
283      : []
284
285    const next = stamps.join(':')
286    const hasActive = latest?.tasks.some(Words.isActive) ?? false
287
288    const isDue =
289      next !== signature ||
290      (hasRemote && now - lastPollAt >= REMOTE_POLL_MS) ||
291      (hasActive && now - lastPollAt >= ACTIVE_REFRESH_MS)
292
293    if (isDue) {
294      signature = next
295      await poll(engine)
296    }
297  }
298
299  /**
300   * The folder a new task on this machine starts in: the one last
301   * dispatched to there, else the session's here and the home elsewhere.
302   */
303  async function cwdFor(engine: Host, machine: string): Promise<string> {
304    const kept = await engine.storeGet(cwdKeyOf(machine)).catch(() => null)
305
306    if (typeof kept === 'string' && kept) return kept
307
308    return machine === Words.LOCAL ? sessionCwd : '~'
309  }
310
311  function actionsOf(engine: Host): PaneActions {
312    const taskOf = (key: string) =>
313      latest?.tasks.find(task => Words.taskKeyOf(task) === key)
314
315    return {
316      select: key => {
317        void engine.setSelected(key).then(() => refreshDetail(engine))
318      },
319      cancel: key => {
320        const task = taskOf(key)
321
322        if (!task) return
323
324        void (async () => {
325          const run = await bus(
326            engine,
327            task.host,
328            ['cancel', task.id],
329            30_000,
330          ).catch(() => null)
331
332          engine.toast(
333            run === null
334              ? `Couldn't run agent-dispatch cancel ${task.id}`
335              : firstLine(run.exitCode === 0 ? run.stdout : run.stderr),
336          )
337
338          await poll(engine)
339        })()
340      },
341      toggleAsk: key => {
342        void engine.toggleArmed(key)
343      },
344      compose: () => {
345        void (async () => {
346          const current = await engine.getComposer()
347          const machine = current?.host ?? hosts[0] ?? Words.LOCAL
348          const cwd = current?.cwd ?? (await cwdFor(engine, machine))
349
350          await engine.editComposer(composer => ({
351            isOpen: true,
352            host: machine,
353            lane: composer?.lane ?? 'luna',
354            mode: composer?.mode ?? 'read-only',
355            cwd,
356            prompt: composer?.prompt ?? '',
357            isSending: false,
358          }))
359        })()
360      },
361      discard: () => {
362        void engine.editComposer(
363          composer => composer && { ...composer, isOpen: false, prompt: '' },
364        )
365      },
366      setHost: machine => {
367        void (async () => {
368          const cwd = await cwdFor(engine, machine)
369
370          await engine.editComposer(
371            composer => composer && { ...composer, host: machine, cwd },
372          )
373        })()
374      },
375      setLane: lane => {
376        void engine.editComposer(composer => composer && { ...composer, lane })
377      },
378      setMode: mode => {
379        const next = mode === 'write' ? 'write' : 'read-only'
380
381        void engine.editComposer(
382          composer => composer && { ...composer, mode: next },
383        )
384      },
385      setCwd: cwd => {
386        void engine.editComposer(composer => {
387          if (!composer) return composer
388
389          const isHere = composer.host === Words.LOCAL
390
391          const path =
392            isHere && home && cwd.startsWith('~') ? home + cwd.slice(1) : cwd
393
394          return { ...composer, cwd: path }
395        })
396      },
397      setPrompt: prompt => {
398        void engine.editComposer(composer => composer && { ...composer, prompt })
399      },
400      dispatch: prompt => {
401        void submit(engine, prompt)
402      },
403    }
404  }
405
406  /**
407   * Dispatches the form in the background on its machine, under this
408   * session's origin so its finish reports here; then shows the new task,
409   * remembers the folder, and clears the form.
410   */
411  async function submit(engine: Host, prompt: string) {
412    const composer = await engine.getComposer()
413
414    if (!composer || composer.isSending) return
415
416    const id = Words.taskIdOf(prompt, new Date(await engine.now()))
417    const args = Words.submitArgsOf({ ...composer, prompt }, id)
418
419    if (!args) {
420      engine.toast('Type a prompt and a folder first')
421
422      return
423    }
424
425    await engine.editComposer(
426      current => current && { ...current, prompt, isSending: true },
427    )
428
429    const run = await bus(engine, composer.host, args, 30_000, sessionId).catch(
430      () => null,
431    )
432
433    if (run === null || run.exitCode !== 0) {
434      engine.toast(
435        run === null
436          ? "Couldn't run agent-dispatch submit"
437          : firstLine(run.stderr || run.stdout),
438      )
439
440      await engine.editComposer(
441        current => current && { ...current, isSending: false },
442      )
443
444      return
445    }
446
447    const where = composer.host === Words.LOCAL ? '' : ` on ${composer.host}`
448
449    engine.toast(`Dispatched ${id} to ${composer.lane}${where}`)
450
451    await engine
452      .storeSet(cwdKeyOf(composer.host), composer.cwd)
453      .catch(() => undefined)
454
455    await engine.editComposer(
456      current =>
457        current && { ...current, isOpen: false, prompt: '', isSending: false },
458    )
459
460    await engine.setSelected(Words.taskKeyOf({ host: composer.host, id }))
461    await poll(engine)
462  }
463
464  on('session.start', async ($, e, next) => {
465    home = await $.env.get('HOME')
466    busHome = (await $.env.get('AGENT_BUS_HOME')) ?? `${home}/.agent-bus`
467    sessionId = await $.session.id()
468    sessionCwd = e.cwd
469
470    const installed = `${home}/.local/bin/agent-dispatch`
471
472    dispatch = (await $.fs.exists(installed)) ? installed : 'agent-dispatch'
473
474    const engine: Host = {
475      run: (argv, timeoutMs, env) => $.process.run(argv, { timeoutMs, env }),
476      now: () => $.clock.now(),
477      mtime: path =>
478        $.fs
479          .stat(path)
480          .then(stat => stat.mtimeMs)
481          .catch(() => 0),
482      toast: text => $.ui.toast(text, { timeoutMs: 6_000 }),
483      status: text => $.ui.status(text),
484      note: async text => {
485        await $.session.append({
486          message: { type: 'user', content: [{ type: 'text', text }] },
487        })
488      },
489      storeGet: key => $.store.get(key),
490      storeSet: (key, value) => $.store.set(key, value),
491      getSelected: () => read($, selectedAtom),
492      setSelected: key => update($, selectedAtom, () => key),
493      getDetail: () => read($, detailAtom),
494      setDetail: detail => update($, detailAtom, () => detail),
495      setSnapshot: snapshot => update($, snapshotAtom, () => snapshot),
496      toggleArmed: key =>
497        update($, armedAtom, armed => (armed === key ? null : key)),
498      getComposer: () => read($, composerAtom),
499      editComposer: edit => update($, composerAtom, edit),
500    }
501
502    host = engine
503
504    await $.command.register({
505      name: Words.COMMAND_NAME,
506      description: Words.COMMAND_DESCRIPTION,
507    })
508
509    const started = await next(e)
510
511    void poll(engine).catch(() => undefined)
512
513    $.clock.every(WATCH_MS, () => {
514      void tick(engine).catch(() => undefined)
515    })
516
517    if ((await $.store.get(Words.STORE_OPEN_KEY)) === true) {
518      void $.ui.open({ id: Words.PANE_ID, title: Words.PANE_TITLE })
519    }
520
521    return started
522  })
523
524  on('command.run', { command: 'bus' }, async ($, e) => {
525    const panes = await $.ui.panes().catch(() => [])
526
527    if (panes.some(pane => pane.id === Words.PANE_ID)) {
528      await $.ui.close({ id: Words.PANE_ID })
529      await $.store.set(Words.STORE_OPEN_KEY, false)
530
531      return { text: Words.PANEL_HIDDEN_TEXT }
532    }
533
534    const isInline = e.presentation.isFullscreen === false
535
536    await $.ui.open(
537      isInline
538        ? {
539            id: Words.PANE_ID,
540            title: Words.PANE_TITLE,
541            focus: true,
542            closeOnEscape: true,
543          }
544        : { id: Words.PANE_ID, title: Words.PANE_TITLE },
545    )
546
547    if (host) void poll(host).catch(() => undefined)
548
549    if (isInline) return {}
550
551    await $.store.set(Words.STORE_OPEN_KEY, true)
552
553    return { text: Words.PANEL_SHOWN_TEXT }
554  })
555
556  on('ui.close', { id: 'bus' }, async ($, e, next) => {
557    if (e.origin.kind === 'person') {
558      await $.store.set(Words.STORE_OPEN_KEY, false).catch(() => undefined)
559    }
560
561    return next(e)
562  })
563
564  on(
565    'ui.render',
566    { component: 'Pane', requestId: 'bus' },
567    async ($, e, next) => {
568      if (!host) return next(e)
569
570      const { Box, Text, Button, Markdown, Input, Select } = (await $.ui.resolve(
571        e,
572      )) as Ui
573
574      const isDocked = e.props.placement === 'dock'
575
576      const [snapshot, selected, detail, armed, composer] = await Promise.all([
577        read($, snapshotAtom),
578        read($, selectedAtom),
579        read($, detailAtom),
580        read($, armedAtom),
581        read($, composerAtom),
582      ])
583
584      return paneView(
585        {
586          ui: { Box, Text, Button, Markdown, Input, Select },
587          actions: actionsOf(host),
588          columns: Math.max(1, e.props.bodyColumns - (isDocked ? 1 : 0)),
589          home,
590          canCompose: e.surface !== 'mobile',
591        },
592        { snapshot, selected, detail, armed, composer, isDocked },
593      )
594    },
595  )
596
597  on('prompt.submit', async ($, e, next) => {
598    const armed = await read($, armedAtom)
599    const task = latest?.tasks.find(one => Words.taskKeyOf(one) === armed)
600
601    if (!armed || !task || !host || carrying === armed) return next(e)
602
603    const context = e.context ?? []
604
605    const room =
606      Words.PROMPT_CONTEXT_MAX_CHARS -
607      context.reduce((sum, entry) => sum + entry.length, 0)
608
609    const result = await resultOf(host, task)
610    const text = result === null ? undefined : Words.askTextOf(task, result, room)
611
612    if (text === undefined) {
613      await update($, armedAtom, () => null)
614      $.ui.toast(`${task.id}'s result did not fit in the prompt`)
615
616      return next(e)
617    }
618
619    carrying = armed
620
621    try {
622      const submitted = await next({ ...e, context: [...context, text] })
623
624      if (submitted.drop === undefined) {
625        await update($, armedAtom, current => (current === armed ? null : current))
626      }
627
628      return submitted
629    } finally {
630      carrying = null
631    }
632  })
633}
634

1import type { ElementTable, RenderElement } from 'claude-code'
2
3import type { BusComposer, BusDetail, BusSnapshot, BusTask } from '../types'
4import {
5  homeFolded,
6  isActive,
7  LANES,
8  LOCAL,
9  MODES,
10  longSpanOf,
11  markOf,
12  runSecondsOf,
13  sanitize,
14  spanOf,
15  tailOf,
16  taskKeyOf,
17  tokensOf,
18} from './words'
19
20/**
21 * The elements the pane draws with, from `$.ui.resolve(e)`.
22 */
23export type Ui = Pick<
24  ElementTable<'terminal' | 'desktop'>,
25  'Box' | 'Text' | 'Button' | 'Markdown' | 'Input' | 'Select'
26>
27
28/**
29 * What the pane's Buttons do, each closing over the plugin's `$`.
30 */
31export type PaneActions = {
32  select: (id: string) => void
33  cancel: (id: string) => void
34  toggleAsk: (id: string) => void
35  compose: () => void
36  discard: () => void
37  setHost: (host: string) => void
38  setLane: (lane: string) => void
39  setMode: (mode: string) => void
40  setCwd: (cwd: string) => void
41  setPrompt: (prompt: string) => void
42  dispatch: (prompt: string) => void
43}
44
45/**
46 * What every part of one drawing is handed.
47 */
48export type Kit = {
49  ui: Ui
50  actions: PaneActions
51  columns: number
52  home: string | undefined
53
54  /**
55   * Whether the surface draws text fields (the mobile app does not), so the
56   * pane offers a new task there.
57   */
58  canCompose: boolean
59}
60
61export type PaneModel = {
62  snapshot: BusSnapshot | null
63  selected: string | null
64  detail: BusDetail | null
65  armed: string | null
66  composer: BusComposer | null
67  isDocked: boolean
68}
69
70/**
71 * Finished tasks the docked list shows under the active ones; inline, fewer.
72 */
73const RECENT_ROWS_DOCKED = 8
74const RECENT_ROWS_INLINE = 3
75
76/**
77 * The task the detail shows: the one picked (by its key across machines),
78 * else the first active, else the most recent.
79 */
80export function selectedOf(
81  snapshot: BusSnapshot | null,
82  selected: string | null,
83): BusTask | null {
84  const tasks = snapshot?.tasks ?? []
85
86  return (
87    tasks.find(task => taskKeyOf(task) === selected) ??
88    tasks.find(isActive) ??
89    tasks[0] ??
90    null
91  )
92}
93
94const present = (nodes: (RenderElement | null | false)[]): RenderElement[] =>
95  nodes.filter((node): node is RenderElement => Boolean(node))
96
97function dimNote(kit: Kit, text: string): RenderElement {
98  const { Text } = kit.ui
99
100  return (
101    <Text dimColor wrap="truncate-end">
102      {text}
103    </Text>
104  )
105}
106
107function divider(kit: Kit): RenderElement {
108  const { Text } = kit.ui
109
110  return (
111    <Text dimColor wrap="truncate-end">
112      {'─'.repeat(Math.max(1, kit.columns))}
113    </Text>
114  )
115}
116
117/**
118 * The first line: what is running and queued in bold, or `Idle` and when the
119 * bus last finished something.
120 */
121function headerView(
122  kit: Kit,
123  snapshot: BusSnapshot,
124  isComposable: boolean,
125): RenderElement {
126  const { Box, Text, Button } = kit.ui
127  const tasks = snapshot.tasks
128  const running = tasks.filter(task => task.state === 'running').length
129  const queued = tasks.filter(task => task.state === 'queued').length
130  const last = tasks.find(task => !isActive(task))
131
132  const [lead, aside] =
133    running > 0
134      ? [`${running} running`, queued > 0 ? ` · ${queued} queued` : '']
135      : queued > 0
136        ? [`${queued} queued`, '']
137        : [
138            'Idle',
139            last?.finished_at
140              ? ` · last finished ${spanOf(snapshot.now - last.finished_at)} ago`
141              : '',
142          ]
143
144  const unreachable = snapshot.hosts.filter(host => host.error !== null)
145
146  const line = (
147    <Box flexDirection="row" height={1}>
148      <Text wrap="truncate-end">
149        <Text bold>{lead}</Text>
150        <Text dimColor>{aside}</Text>
151      </Text>
152      <Box flexGrow={1} />
153      {isComposable ? (
154        <Button key="compose" plain dimColor hotkey="n" onPress={kit.actions.compose}>
155          New task
156        </Button>
157      ) : null}
158    </Box>
159  )
160
161  return unreachable.length === 0 ? (
162    line
163  ) : (
164    <Box flexDirection="column">
165      {line}
166      {unreachable.map(host => (
167        <Text color="warning" wrap="truncate-end">
168          {`Couldn't reach ${host.name}`}
169        </Text>
170      ))}
171    </Box>
172  )
173}
174
175/**
176 * One task's row: the pointer where selected, its mark, its id as a plain
177 * Button, and the lane and time at the right edge.
178 */
179function taskRow(
180  kit: Kit,
181  task: BusTask,
182  now: number,
183  isSelected: boolean,
184  isTagged: boolean,
185): RenderElement {
186  const { Box, Text, Button } = kit.ui
187  const mark = markOf(task)
188  const key = taskKeyOf(task)
189
190  return (
191    <Box flexDirection="row" key={`row-${key}`}>
192      <Text>{isSelected ? '❯ ' : '  '}</Text>
193      <Text color={mark.color}>{`${mark.glyph} `}</Text>
194      <Button
195        key={`task-${key}`}
196        plain
197        dimColor={!isSelected && !isActive(task)}
198        onPress={() => kit.actions.select(key)}
199      >
200        {sanitize(task.id)}
201      </Button>
202      <Box flexGrow={1} />
203      <Text dimColor wrap="truncate-start">{` ${tailOf(task, now, isTagged)}`}</Text>
204    </Box>
205  )
206}
207
208/**
209 * The selected task under the list: its id and actions, how it ran, then
210 * its result (finished) or its prompt (still active).
211 */
212function detailView(kit: Kit, model: PaneModel, task: BusTask): RenderElement {
213  const { Box, Text, Button, Markdown } = kit.ui
214  const now = model.snapshot?.now ?? 0
215  const key = taskKeyOf(task)
216  const isArmed = model.armed === key
217
218  const action = isActive(task) ? (
219    <Button key="cancel" hotkey="c" onPress={() => kit.actions.cancel(key)}>
220      Cancel
221    </Button>
222  ) : (
223    <Button key="ask" hotkey="a" onPress={() => kit.actions.toggleAsk(key)}>
224      {isArmed ? 'Asked ✓' : 'Ask'}
225    </Button>
226  )
227
228  const ran = runSecondsOf(task, now)
229
230  const how = [
231    task.agent,
232    [task.model, task.effort].filter(part => part).join(' '),
233    task.mode,
234  ]
235    .filter(part => part)
236    .join(' · ')
237
238  const when = [
239    task.state === 'queued' ? 'queued' : null,
240    ran === null ? null : longSpanOf(ran),
241    task.tokens ? tokensOf(task.tokens) : null,
242  ]
243    .filter(part => part)
244    .join(' · ')
245
246  const folder = task.worktree ?? task.cwd
247
248  const where =
249    task.host === LOCAL
250      ? folder && homeFolded(folder, kit.home)
251      : [task.host, folder].filter(part => part).join(' · ')
252
253  const body =
254    model.detail?.id === key && model.detail.text !== null ? (
255      <Markdown key="result" text={model.detail.text} />
256    ) : isActive(task) ? (
257      <Text wrap="wrap">{sanitize(task.prompt)}</Text>
258    ) : (
259      dimNote(kit, 'No result file.')
260    )
261
262  return (
263    <Box flexDirection="column">
264      {present([
265        <Box flexDirection="row" height={1}>
266          <Text bold wrap="truncate-end">
267            {sanitize(task.id)}
268          </Text>
269          <Box flexGrow={1} />
270          {action}
271        </Box>,
272        dimNote(kit, how),
273        where ? dimNote(kit, where) : null,
274        when ? dimNote(kit, when) : null,
275        task.note ? (
276          <Text color={task.cancelled ? 'inactive' : 'error'} wrap="wrap">
277            {sanitize(task.note.split('\n')[0] ?? '')}
278          </Text>
279        ) : null,
280        <Box height={1} />,
281        isActive(task) ? dimNote(kit, 'Prompt') : null,
282        body,
283      ])}
284    </Box>
285  )
286}
287
288/**
289 * The new-task form in the detail's place: lane, mode, folder, then the
290 * prompt, whose Enter dispatches in the background.
291 */
292function composerView(
293  kit: Kit,
294  composer: BusComposer,
295  hosts: string[],
296): RenderElement {
297  const { Box, Text, Button, Input, Select } = kit.ui
298  const lane = LANES.find(one => one.value === composer.lane)
299
300  const hint = composer.isSending
301    ? 'Dispatching…'
302    : `Enter dispatches to ${lane?.value ?? composer.lane}, ${composer.mode}` +
303      (hosts.length > 1 || composer.host !== LOCAL ? `, on ${composer.host}` : '')
304
305  return (
306    <Box flexDirection="column">
307      <Box flexDirection="row" height={1}>
308        <Text bold>New task</Text>
309        <Box flexGrow={1} />
310        <Button key="discard" role="dismiss" onPress={kit.actions.discard}>
311          Discard
312        </Button>
313      </Box>
314      <Box height={1} />
315      {hosts.length > 1 ? (
316        <Select
317          key="host"
318          label="Machine "
319          value={composer.host}
320          options={hosts.map(host => ({
321            value: host,
322            label: host === LOCAL ? 'this machine' : host,
323          }))}
324          onSelect={kit.actions.setHost}
325        />
326      ) : null}
327      <Select
328        key="lane"
329        label="Lane "
330        value={composer.lane}
331        options={LANES.map(({ value, label }) => ({ value, label }))}
332        onSelect={kit.actions.setLane}
333      />
334      <Select
335        key="mode"
336        label="Mode "
337        value={composer.mode}
338        options={MODES.map(({ value, label }) => ({ value, label }))}
339        onSelect={kit.actions.setMode}
340      />
341      <Input
342        key="cwd"
343        label="Folder "
344        value={
345          composer.host === LOCAL
346            ? homeFolded(composer.cwd, kit.home)
347            : composer.cwd
348        }
349        onInput={kit.actions.setCwd}
350        onSubmit={kit.actions.setCwd}
351      />
352      <Box height={1} />
353      <Input
354        key="prompt"
355        label="Prompt "
356        placeholder="What should the worker do?"
357        submitLabel="dispatch"
358        value={composer.prompt}
359        autoFocus
360        onInput={kit.actions.setPrompt}
361        onSubmit={kit.actions.dispatch}
362      />
363      <Box height={1} />
364      {dimNote(kit, hint)}
365    </Box>
366  )
367}
368
369/**
370 * The pane's body for one `ui.render`. Docked: the header, the active
371 * tasks, the recent ones, then the selected task in full. Inline above the
372 * prompt: the header and the rows alone.
373 */
374export function paneView(kit: Kit, model: PaneModel): RenderElement {
375  const { Box } = kit.ui
376  const snapshot = model.snapshot
377
378  const frame = (children: RenderElement[]): RenderElement => (
379    <Box
380      flexDirection="column"
381      paddingTop={model.isDocked ? 1 : 0}
382      paddingRight={model.isDocked ? 1 : 0}
383    >
384      {children}
385    </Box>
386  )
387
388  if (!snapshot) {
389    return frame([dimNote(kit, 'Reading the bus…')])
390  }
391
392  if (snapshot.error) {
393    return frame([dimNote(kit, snapshot.error)])
394  }
395
396  const hosts = snapshot.hosts.map(host => host.name)
397  const isTagged = hosts.length > 1
398  const isComposing = model.isDocked && model.composer?.isOpen === true
399  const isComposable = model.isDocked && kit.canCompose && !isComposing
400
401  if (snapshot.tasks.length === 0) {
402    return frame(
403      present([
404        headerView(kit, snapshot, isComposable),
405        <Box height={1} />,
406        isComposing && model.composer
407          ? composerView(kit, model.composer, hosts)
408          : dimNote(kit, 'No bus tasks yet.'),
409      ]),
410    )
411  }
412
413  const selected = selectedOf(snapshot, model.selected)
414  const active = snapshot.tasks.filter(isActive)
415
416  const recent = snapshot.tasks
417    .filter(task => !isActive(task))
418    .slice(0, model.isDocked ? RECENT_ROWS_DOCKED : RECENT_ROWS_INLINE)
419
420  const rowsOf = (tasks: BusTask[]): RenderElement[] =>
421    tasks.map(task =>
422      taskRow(
423        kit,
424        task,
425        snapshot.now,
426        model.isDocked && task === selected,
427        isTagged,
428      ),
429    )
430
431  return frame(
432    present([
433      headerView(kit, snapshot, isComposable),
434      <Box height={1} />,
435      ...rowsOf(active),
436      active.length > 0 && recent.length > 0 ? <Box height={1} /> : null,
437      recent.length > 0 ? dimNote(kit, 'Recent') : null,
438      ...rowsOf(recent),
439      ...(isComposing && model.composer
440        ? [
441            <Box height={1} />,
442            divider(kit),
443            composerView(kit, model.composer, hosts),
444          ]
445        : model.isDocked && selected
446          ? [
447              <Box height={1} />,
448              divider(kit),
449              detailView(kit, model, selected),
450            ]
451          : []),
452    ]),
453  )
454}
455
456

1import type { BusSnapshot, BusTask } from '../types'
2
3/**
4 * The pane's one id: its `requestId` at `ui.render`, and what `/bus` toggles.
5 */
6export const PANE_ID = 'bus'
7
8/**
9 * The pane's tab label while another pane is open too.
10 */
11export const PANE_TITLE = 'Bus'
12
13export const COMMAND_NAME = 'bus'
14
15export const COMMAND_DESCRIPTION =
16  'Toggle the agent bus panel showing delegated tasks'
17
18export const PANEL_SHOWN_TEXT = 'Bus panel shown'
19export const PANEL_HIDDEN_TEXT = 'Bus panel hidden'
20
21/**
22 * The `$.store` key remembering whether the person left the pane open.
23 */
24export const STORE_OPEN_KEY = 'open'
25
26/**
27 * The context a prompt may carry in all; an ask is cut to what is left.
28 */
29export const PROMPT_CONTEXT_MAX_CHARS = 32_000
30
31/**
32 * What a result drawn in the detail is cut to (Markdown takes 10,000).
33 */
34export const DETAIL_MAX_CHARS = 9_000
35
36/**
37 * The lanes the form dispatches to, as agent-ops ROUTING.md names them:
38 * each one's `agent-dispatch submit` arguments.
39 */
40export const LANES: readonly { value: string; label: string; argv: string[] }[] =
41  [
42    {
43      value: 'luna',
44      label: 'luna · throughput',
45      argv: ['--to', 'codex', '--model', 'gpt-6-luna', '--effort', 'high'],
46    },
47    {
48      value: 'sol',
49      label: 'sol · implementation',
50      argv: ['--to', 'codex', '--model', 'gpt-6.1-sol', '--effort', 'medium'],
51    },
52    {
53      value: 'sonnet',
54      label: 'sonnet · fallback',
55      argv: ['--to', 'claude', '--effort', 'medium'],
56    },
57  ]
58
59export const MODES = [
60  { value: 'read-only', label: 'read-only' },
61  { value: 'write', label: 'may edit files' },
62] as const
63
64/**
65 * The id a task dispatched from the pane is queued under: the time and the
66 * prompt's first words, as agent-dispatch names one itself.
67 */
68export function taskIdOf(prompt: string, at: Date): string {
69  const pad = (n: number) => String(n).padStart(2, '0')
70
71  const stamp =
72    `${at.getFullYear()}${pad(at.getMonth() + 1)}${pad(at.getDate())}-` +
73    `${pad(at.getHours())}${pad(at.getMinutes())}${pad(at.getSeconds())}`
74
75  const slug = prompt
76    .toLowerCase()
77    .replace(/[^a-z0-9]+/g, '-')
78    .replace(/^-+|-+$/g, '')
79    .slice(0, 40)
80    .replace(/-+$/g, '')
81
82  return `${stamp}-${slug || 'task'}`
83}
84
85/**
86 * The agent-dispatch arguments that submit the form, run in the background
87 * so the call returns at once; the folder bounds a writing task's scope.
88 */
89export function submitArgsOf(
90  composer: { lane: string; mode: string; cwd: string; prompt: string },
91  id: string,
92): string[] | null {
93  const lane = LANES.find(one => one.value === composer.lane)
94  const prompt = composer.prompt.trim()
95
96  if (!lane || !prompt || !composer.cwd.trim()) return null
97
98  return [
99    'submit',
100    ...lane.argv,
101    '--mode',
102    composer.mode,
103    '--cwd',
104    composer.cwd.trim(),
105    ...(composer.mode === 'write'
106      ? ['--scope', `files under ${composer.cwd.trim()} only`]
107      : []),
108    '--id',
109    id,
110    '--bg',
111    prompt,
112  ]
113}
114
115/**
116 * This machine's bus, as the `hosts` option names it.
117 */
118export const LOCAL = 'local'
119
120/**
121 * Where a remote machine's dispatcher is, left unquoted so its shell
122 * expands the home directory.
123 */
124export const REMOTE_DISPATCH = '~/.local/bin/agent-dispatch'
125
126/**
127 * The machines the `hosts` option names, in order and once each: `local` or
128 * ssh aliases (letters, digits, `.`, `_`, `-`); anything else is dropped.
129 */
130export function hostsOf(option: unknown): string[] {
131  const names = String(option ?? '')
132    .split(/[\s,]+/)
133    .map(name => name.trim())
134    .filter(name => /^[A-Za-z0-9._-]+$/.test(name) && !name.startsWith('-'))
135
136  const unique = [...new Set(names)]
137
138  return unique.length > 0 ? unique : [LOCAL]
139}
140
141/**
142 * One word for a POSIX shell: single-quoted, with each `'` closed, escaped
143 * and reopened.
144 */
145export const shellQuote = (word: string): string =>
146  `'${word.replace(/'/g, `'\\''`)}'`
147
148/**
149 * The command a remote shell runs for these dispatcher arguments, the
150 * origin set in its environment when given.
151 */
152export function remoteCommandOf(args: string[], origin?: string): string {
153  const env = origin ? `env AGENT_BUS_ORIGIN=${shellQuote(origin)} ` : ''
154
155  return `${env}${REMOTE_DISPATCH} ${args.map(shellQuote).join(' ')}`
156}
157
158/**
159 * The ssh argument vector that runs a command on `alias`: never prompting,
160 * giving up on an unreachable machine in seconds, and sharing one
161 * connection between polls.
162 */
163export function sshArgvOf(alias: string, command: string): string[] {
164  return [
165    'ssh',
166    '-o',
167    'BatchMode=yes',
168    '-o',
169    'ConnectTimeout=5',
170    '-o',
171    'ControlMaster=auto',
172    '-o',
173    'ControlPath=/tmp/agent-bus-ssh-%C',
174    '-o',
175    'ControlPersist=120',
176    alias,
177    command,
178  ]
179}
180
181/**
182 * A task's address across machines.
183 */
184export const taskKeyOf = (task: { host: string; id: string }): string =>
185  `${task.host}/${task.id}`
186
187/**
188 * What one machine's bus answered: its parsed `status --json`, or why not.
189 */
190export type HostAnswer = {
191  host: string
192  status: { tasks?: Omit<BusTask, 'host'>[] } | null
193  error: string | null
194}
195
196/**
197 * The machines' answers as one snapshot: every active task in the order the
198 * machines are listed, then the most recent finished ones across them all.
199 * It fails as a whole only when no machine answered.
200 */
201export function mergedSnapshotOf(
202  answers: HostAnswer[],
203  now: number,
204  recent = 20,
205): BusSnapshot {
206  const tasks = answers.flatMap(answer =>
207    (answer.status?.tasks ?? []).map(task => ({ ...task, host: answer.host })),
208  )
209
210  const active = tasks.filter(task => ACTIVE_STATES.includes(task.state))
211
212  const finished = tasks
213    .filter(task => !ACTIVE_STATES.includes(task.state))
214    .sort((a, b) => (b.finished_at ?? 0) - (a.finished_at ?? 0))
215    .slice(0, recent)
216
217  const count = (state: BusTask['state']) =>
218    tasks.filter(task => task.state === state).length
219
220  const failed = answers.filter(answer => answer.error !== null)
221
222  return {
223    hosts: answers.map(({ host, error }) => ({ name: host, error })),
224    now,
225    counts: {
226      queued: count('queued'),
227      running: count('running'),
228      done: count('done'),
229      failed: count('failed'),
230    },
231    tasks: [...active, ...finished],
232    error:
233      answers.length > 0 && failed.length === answers.length
234        ? (failed[0]?.error ?? null)
235        : null,
236  }
237}
238
239export const ACTIVE_STATES: readonly BusTask['state'][] = [
240  'running',
241  'queued',
242  'orphaned',
243]
244
245export const isActive = (task: BusTask): boolean =>
246  ACTIVE_STATES.includes(task.state)
247
248/**
249 * The lane a task ran in, as the routing doc names them: the model's family
250 * where it has one (`luna`, `sol`, `opus`), else the worker CLI.
251 */
252export function laneOf(task: BusTask): string {
253  const model = (task.model ?? '').toLowerCase()
254  const lanes = ['luna', 'sol', 'opus', 'sonnet', 'haiku', 'fable', 'argon']
255  const lane = lanes.find(name => model.includes(name))
256
257  return lane ?? (task.agent === 'antigravity' ? 'agy' : task.agent)
258}
259
260/**
261 * A span in its largest whole unit: `45s`, `12m`, `3h`, `2d`.
262 */
263export function spanOf(seconds: number): string {
264  const s = Math.max(0, Math.round(seconds))
265
266  if (s < 60) return `${s}s`
267  if (s < 3600) return `${Math.floor(s / 60)}m`
268  if (s < 86_400) return `${Math.floor(s / 3600)}h`
269
270  return `${Math.floor(s / 86_400)}d`
271}
272
273/**
274 * A span to the minute: `12m 4s`, `1h 12m`.
275 */
276export function longSpanOf(seconds: number): string {
277  const s = Math.max(0, Math.round(seconds))
278
279  if (s < 60) return `${s}s`
280  if (s < 3600) return `${Math.floor(s / 60)}m ${s % 60}s`
281
282  return `${Math.floor(s / 3600)}h ${Math.floor((s % 3600) / 60)}m`
283}
284
285export function tokensOf(count: number): string {
286  if (count < 1000) return `${count} tokens`
287  if (count < 1_000_000) return `${(count / 1000).toFixed(1)}k tokens`
288
289  return `${(count / 1_000_000).toFixed(2)}M tokens`
290}
291
292/**
293 * How long a task has run, or ran, as of the snapshot.
294 */
295export function runSecondsOf(task: BusTask, now: number): number | null {
296  if (task.elapsed_s !== null) return task.elapsed_s
297  if (task.started_at !== null) return now - task.started_at
298  if (task.queued_at !== null) return now - task.queued_at
299
300  return null
301}
302
303/**
304 * A list row's right edge: the lane and, while active, how long it has run;
305 * once finished, how long ago.
306 */
307export function tailOf(task: BusTask, now: number, isTagged = false): string {
308  const lane = isTagged ? `${task.host} · ${laneOf(task)}` : laneOf(task)
309
310  if (task.state === 'queued') return `${lane} · queued`
311  if (task.state === 'orphaned') return `${lane} · stranded`
312
313  if (task.state === 'running') {
314    const ran = runSecondsOf(task, now)
315
316    return ran === null ? lane : `${lane} · ${spanOf(ran)}`
317  }
318
319  return task.finished_at === null
320    ? lane
321    : `${lane} · ${spanOf(now - task.finished_at)} ago`
322}
323
324/**
325 * The glyph before a row, and the theme colour it draws in.
326 */
327export function markOf(task: BusTask): { glyph: string; color?: string } {
328  if (task.cancelled) return { glyph: '⊘', color: 'inactive' }
329
330  switch (task.state) {
331    case 'running':
332      return { glyph: '●', color: 'warning' }
333    case 'queued':
334      return { glyph: '○', color: 'inactive' }
335    case 'orphaned':
336      return { glyph: '!', color: 'error' }
337    case 'done':
338      return { glyph: '✓', color: 'success' }
339    case 'failed':
340      return { glyph: '✗', color: 'error' }
341  }
342}
343
344/**
345 * The status line's entry: what is active, or nothing while the bus is idle.
346 */
347export function statusTextOf(snapshot: BusSnapshot | null): string | undefined {
348  if (!snapshot || snapshot.error) return undefined
349
350  const running = snapshot.tasks.filter(task => task.state === 'running')
351  const queued = snapshot.tasks.filter(task => task.state === 'queued').length
352
353  if (running.length === 0 && queued === 0) return undefined
354
355  const lanes = [...new Set(running.map(laneOf))].join(', ')
356
357  const parts = [
358    running.length > 0 ? `${running.length} running (${lanes})` : null,
359    queued > 0 ? `${queued} queued` : null,
360  ].filter(part => part !== null)
361
362  return `bus · ${parts.join(' · ')}`
363}
364
365/**
366 * The toast for a task that finished while this session watched.
367 */
368export function toastOf(task: BusTask): string {
369  const verb = task.cancelled
370    ? 'cancelled'
371    : task.state === 'done'
372      ? 'done'
373      : 'failed'
374
375  const ran = task.elapsed_s === null ? '' : ` · ${spanOf(task.elapsed_s)}`
376
377  return `${markOf(task).glyph} ${task.id} ${verb} · ${laneOf(task)}${ran}`
378}
379
380/**
381 * The note the model reads when a task this session dispatched finishes:
382 * enough to act on without asking, and where the whole result is.
383 */
384export function finishedNoteOf(task: BusTask): string {
385  const verb = task.cancelled
386    ? 'was cancelled'
387    : task.state === 'done'
388      ? 'finished'
389      : 'failed'
390
391  const how = [
392    task.agent,
393    task.model,
394    task.elapsed_s === null ? null : longSpanOf(task.elapsed_s),
395  ]
396    .filter(part => part)
397    .join(', ')
398
399  const isRemote = task.host !== LOCAL
400
401  const lines = [
402    `[agent-bus] Task \`${task.id}\` that this session dispatched ${verb} (${how}${isRemote ? `, on ${task.host}` : ''}).`,
403    isRemote
404      ? `Result on ${task.host}: read it with \`ssh ${task.host} '${REMOTE_DISPATCH} result ${task.id}'\``
405      : `Result: ${task.result}`,
406  ]
407
408  if (task.note) lines.push(`Dispatcher note: ${task.note.split('\n')[0]}`)
409
410  return lines.join('\n')
411}
412
413/**
414 * What an ask adds to the next prompt: the task, then its result, cut by
415 * whole lines to the room left.
416 */
417export function askTextOf(
418  task: BusTask,
419  text: string,
420  room: number,
421): string | undefined {
422  const head =
423    `Result of agent-bus task \`${task.id}\` (${laneOf(task)}, ` +
424    `${task.state}) from ${task.result}:\n\n`
425
426  const cut = '\n[… result cut to fit; the whole file is at the path above]'
427
428  if (head.length + text.length <= room) return head + text
429
430  const kept: string[] = []
431
432  let used = head.length + cut.length
433
434  for (const line of text.split('\n')) {
435    if (used + line.length + 1 > room) break
436
437    kept.push(line)
438    used += line.length + 1
439  }
440
441  return kept.length > 0 ? head + kept.join('\n') + cut : undefined
442}
443
444/**
445 * A path with the home directory folded to `~`.
446 */
447export const homeFolded = (path: string, home: string | undefined): string =>
448  home && path.startsWith(home) ? `~${path.slice(home.length)}` : path
449
450/**
451 * Text safe to draw: no control characters but newline and tab.
452 */
453export const sanitize = (text: string): string =>
454  text.replace(/[\u0000-\u0008\u000b-\u001f\u007f-\u009f]/g, '')
455

1/**
2 * One task as `agent-dispatch status --json` reports it (contract version 1).
3 */
4export type BusTask = {
5  /**
6   * The machine whose bus holds the task: `local`, or the ssh alias the
7   * pane reached it by. Set by the pane, not the bus.
8   */
9  host: string
10  id: string
11  state: 'queued' | 'running' | 'orphaned' | 'done' | 'failed'
12  to: string
13  agent: string
14  model: string | null
15  effort: string | null
16  mode: string | null
17  cwd: string | null
18  prompt: string
19  origin: string | null
20  queued_at: number | null
21  started_at: number | null
22  finished_at: number | null
23  elapsed_s: number | null
24  exit_code: number | null
25  cancelled: boolean
26  tokens: number | null
27  result: string
28  log: string | null
29  note: string | null
30  worktree: string | null
31}
32
33/**
34 * The last poll: its tasks, active first then the most recent finished, and
35 * when it was taken (epoch seconds); or why the poll failed.
36 */
37export type BusSnapshot = {
38  /**
39   * Every machine the pane watches, in order, and why it could not be read.
40   */
41  hosts: { name: string; error: string | null }[]
42  now: number
43  counts: { queued: number; running: number; done: number; failed: number }
44  tasks: BusTask[]
45  error: string | null
46}
47
48/**
49 * What the detail under the list shows for the selected task: its result's
50 * text once finished, or nothing yet.
51 */
52export type BusDetail = {
53  id: string
54  text: string | null
55}
56
57/**
58 * The new-task form under the list: which lane, whether the worker may
59 * write, where it runs, and the prompt typed so far.
60 */
61export type BusComposer = {
62  isOpen: boolean
63  host: string
64  lane: string
65  mode: 'read-only' | 'write'
66  cwd: string
67  prompt: string
68  isSending: boolean
69}
70
71declare module 'claude-code' {
72  interface PluginState {
73    'agent-bus': {
74      snapshot: BusSnapshot | null
75      selected: string | null
76      detail: BusDetail | null
77      armed: string | null
78      composer: BusComposer | null
79    }
80  }
81}
82