rabe

1import type { Register } from 'claude-code'
2
3import { sources } from './sources'
4import { band } from './ui/band'
5import { builtin } from './ui/builtin'
6import { pane } from './ui/pane'
7import { hoursOf } from './ui/views/timeline'
8
9export const register: Register = (on, options) => {
10  sources(on, String(options.pricesFile ?? ''))
11  band(on)
12  pane(on, hoursOf(options.timelineHours))
13  if (options.hideBuiltinTasks !== false) builtin(on)
14}
15

1import type { On } from 'claude-code'
2
3import { agents } from './agents'
4import { codex } from './codex'
5import { crons } from './crons'
6import { monitors } from './monitors'
7import { shells } from './shells'
8import { workflows } from './workflows'
9import { worktrees } from './worktrees'
10
11export type Source = (on: On) => void
12
13// `file`: the user's price table (option `pricesFile`), '' for none.
14export function sources(on: On, file: string): void {
15  agents(on, file)
16  workflows(on)
17  codex(on, file)
18  shells(on)
19  monitors(on)
20  crons(on)
21  worktrees(on)
22}
23

1import type { On } from 'claude-code'
2
3import { render } from './render'
4import { bounded, NO_SELECTION } from './view'
5import { bandView } from './views/band'
6
7export function band(on: On): void {
8  on('ui.render', { component: 'AbovePrompt' }, async ($, e, next) => {
9    if (e.props.hasSurvey) return next(e)
10    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
11    const { value: removed = [] } = await $.state.get({ plugin: 'rabe', key: 'removed' })
12    if (!items.some(item => item.status === 'running')) return next(e)
13    const usage = await $.session.usage().catch(() => undefined)
14    const model = {
15      items,
16      removed,
17      turns: {},
18      lines: {},
19      now: await $.clock.now(),
20      usd: usage?.cost?.usd,
21    }
22    const size = bounded({
23      columns: e.props.bodyColumns,
24      rows: e.props.maxRows,
25      surface: e.surface,
26      hasInput: false,
27    })
28
29    return render($.ui.resolve(e), e.surface, bandView(model, size, NO_SELECTION), () => {}, 'band')
30  })
31}
32

1import type { On } from 'claude-code'
2
3const TASK =
4  /^\d+ (background |dynamic )?(shell|monitor|agent|subagent|workflow|task|teammate|MCP task|cloud session)s?$/
5
6// Removes the part of Claude Code's prompt hint that counts background work
7// ("2 shells, 1 monitor") and its "↓ to manage"; keeps every other part.
8export function stripTasks(hint: string): string {
9  return hint
10    .split(' · ')
11    .filter(part => part !== '↓ to manage' && !part.split(', ').every(one => TASK.test(one)))
12    .join(' · ')
13}
14
15// The engine's duration format on the turn line: 5s, 1m 4s, 1h 2m 5s.
16export function turnDuration(ms: number): string {
17  const s = Math.floor(ms / 1000)
18  const h = Math.floor(s / 3600)
19  const m = Math.floor(s / 60) % 60
20  if (h > 0) return `${h}h ${m}m ${s % 60}s`
21
22  return m > 0 ? `${m}m ${s % 60}s` : `${s}s`
23}
24
25// Hides Claude Code's own count of background work, since the band shows it.
26export function builtin(on: On): void {
27  on('ui.render', { component: 'PromptHint' }, async (_$, e, next) => {
28    const hint = stripTasks(e.props.hint)
29
30    return hint === e.props.hint ? next(e) : next({ ...e, props: { ...e.props, hint } })
31  })
32
33  // The line's "N shells, M monitors still running" part is not a prop, so the
34  // hook draws the line itself, only while Rabe knows of a running shell or
35  // monitor and no running agent or workflow (then the engine waits for those
36  // and draws no such part). That line leaves out "done 1:16": no prop has it.
37  on('ui.render', { component: 'TurnDuration' }, async ($, e, next) => {
38    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
39    const running = items.filter(item => item.status === 'running')
40    const isWaiting = running.some(item => item.kind === 'agent' || item.kind === 'workflow')
41    const hasTasks = running.some(item => item.kind === 'shell' || item.kind === 'monitor')
42    if (isWaiting || !hasTasks) return next(e)
43    const { Box, Text } = $.ui.resolve(e)
44
45    return (
46      <Box flexDirection="row" marginTop={1} width="100%">
47        <Box minWidth={2}>
48          <Text dimColor>✻</Text>
49        </Box>
50        <Text dimColor>
51          {e.props.word} for {turnDuration(e.props.durationMs)}
52        </Text>
53      </Box>
54    )
55  })
56}
57

1import type { EngineInterface, On, RenderSurface } from 'claude-code'
2
3import type { RabeItem, RabePrevious } from '../../types'
4import { KIND_LABEL, kept, matches, orderOf, previousOf, rowOf } from './lists'
5import { type At, type Held, hold, isRowKey, render, shifts } from './render'
6import { reserve, sequence } from './sequence'
7import {
8  type Action,
9  type ArmEvent,
10  type Arming,
11  arm,
12  bounded,
13  DISARMED,
14  landing,
15  landingOf,
16  layout,
17  type Model,
18  messageLanding,
19  rowKeys,
20  type Selection,
21  stepRow,
22  targetsOf,
23  taskIdOf,
24} from './view'
25import { fallbackOf, isLiveRow, paneView, seatsRows, selectsOnPress } from './views/pane'
26import { widen, windowOf } from './views/timeline'
27
28const PANE = 'rabe'
29const MAX_ACTIONS = 20
30
31// The option `timelineHours`, set by `pane(on, hours)`: the Timeline's window
32// when the pane opens.
33let baseHours = 4
34
35function toastText(text: string): string {
36  return text.replace(/\s+/g, ' ').replace(/[\p{Cc}\p{Cf}]/gu, '?')
37}
38
39// Titles and reasons come from files and tools.
40function toast($: EngineInterface, text: string): void {
41  $.ui.toast(toastText(text))
42}
43
44// Tabs and line breaks stay; an escape could end a bracketed paste.
45function clipText(text: string): string {
46  return text.replace(/(?![\t\n\r])[\p{Cc}\p{Cf}]/gu, '?')
47}
48
49// A cron job's title is its prompt.
50function nameOf(item: RabeItem): string {
51  return item.kind === 'cron' ? item.detail.jobId : item.title
52}
53
54// Esc may have closed the pane before the delayed focus call.
55async function refocus($: EngineInterface): Promise<void> {
56  const pane = (await $.ui.panes()).find(one => one.id === PANE)
57  if (!pane || pane.isFocused) return
58  await $.ui.open({ id: PANE, title: 'Rabe', focus: true, closeOnEscape: true })
59}
60
61async function tick($: EngineInterface): Promise<void> {
62  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
63  if (items.some(item => item.status === 'running')) $.ui.invalidate('ui.render')
64}
65
66// The store is the plugin's own JSON; an older Rabe may have written another
67// shape, and a summary before 0.4 has no session id.
68function asPrevious(value: unknown): RabePrevious | undefined {
69  const prev = value as RabePrevious | undefined
70  if (typeof prev?.endedAt !== 'number' || !Array.isArray(prev.failed) || !prev.counts) {
71    return undefined
72  }
73  const { sessionId, ...rest } = prev
74
75  return typeof sessionId === 'string' && sessionId ? { ...rest, sessionId } : rest
76}
77
78async function previous($: EngineInterface): Promise<RabePrevious | undefined> {
79  return asPrevious(await $.store.get(`previous:${await $.session.cwd()}`))
80}
81
82// Keeps this session's summary for the next one in this project; a session
83// with no background work leaves the last summary in place.
84async function remember($: EngineInterface, sessionId?: string): Promise<void> {
85  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
86  if (items.length === 0) return
87  const usage = await $.session.usage()
88  const summary = previousOf(
89    items,
90    await $.clock.now(),
91    { startedAt: usage.startedAt, usd: usage.cost?.usd },
92    sessionId,
93  )
94  await $.store.set(`previous:${await $.session.cwd()}`, summary)
95}
96
97async function stop($: EngineInterface, ids: string[]): Promise<void> {
98  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
99  const targets = items.filter(item => ids.includes(item.id))
100  const one = targets.length === 1 ? targets[0] : undefined
101  if (one) toast($, `Stopping ${KIND_LABEL[one.kind]} ${nameOf(one)}…`)
102  let stopped = 0
103  const stoppedItems: RabeItem[] = []
104  let ended = 0
105  let refused: string | undefined
106  for (const item of targets) {
107    if (item.kind === 'codex' && item.status === 'running') {
108      const { text = 'n/a' } = await $.command
109        .run({ command: 'rabe-stop', args: item.id })
110        .catch((error: unknown) => ({ text: `Stop refused: ${String(error)}` }))
111      if (text.startsWith('Stopped')) {
112        stopped += 1
113        stoppedItems.push(item)
114      } else refused = text.replace(/^Stop refused: /, '')
115      continue
116    }
117    const task = taskIdOf(item)
118    if (item.status !== 'running' || !task) {
119      ended += 1
120      continue
121    }
122    try {
123      const result = await $.tool.call({ tool: 'TaskStop', task_id: task })
124      if (result.deny) refused = result.deny
125      else if (result.isError) refused = result.text ?? 'n/a'
126      else {
127        stopped += 1
128        stoppedItems.push(item)
129      }
130    } catch (error) {
131      refused = error instanceof Error ? error.message : String(error)
132    }
133  }
134  for (const item of stoppedItems) {
135    await queueAction($, `Rabe: user stopped ${item.kind} ${item.id}`)
136  }
137  if (one && refused) toast($, `Stop refused: ${refused}`)
138  else if (one) toast($, ended ? `${nameOf(one)} had already finished` : `Stopped ${nameOf(one)}`)
139  else {
140    const done = ended ? `; ${ended} had already finished` : ''
141    const no = refused ? `; refused: ${refused}` : ''
142    toast($, `Stopped ${stopped} of ${targets.length}${done}${no}`)
143  }
144}
145
146async function queueAction($: EngineInterface, note: string): Promise<void> {
147  for (;;) {
148    const { value = [], version } = await $.state.get({ plugin: 'rabe', key: 'actions' })
149    const { isSet } = await $.state.set(
150      { plugin: 'rabe', key: 'actions' },
151      [...value, note].slice(-MAX_ACTIONS),
152      { ifVersion: version },
153    )
154    if (isSet) return
155  }
156}
157
158async function clearActions($: EngineInterface, delivered: string[]): Promise<void> {
159  for (;;) {
160    const { value = [], version } = await $.state.get({ plugin: 'rabe', key: 'actions' })
161    if (!delivered.every((note, index) => value[index] === note)) return
162    const { isSet } = await $.state.set(
163      { plugin: 'rabe', key: 'actions' },
164      value.slice(delivered.length),
165      { ifVersion: version },
166    )
167    if (isSet) return
168  }
169}
170
171// Whether stop and delete act: the ring is known to sit on a safe element.
172// A module value, so a reload, which also drops the hold, starts disarmed.
173let arming: Arming = DISARMED
174
175// Where the last `ui.focus` on the pane landed, as the pane's focus hook read it.
176let landedOn: string | undefined
177
178// The element that holds the focus ring, as far as the pane's focus hook saw
179// it land; unknown after a reset, whose view puts another element at its index.
180let ringOn: string | undefined
181
182// The element the ring belongs on after rows came above it in the Items list
183// (the ring kept its index), until the pane's focus hook sees the ring land.
184// `isAsked`: a move back is under way; `tries`: moves asked since the pane
185// last took the keys.
186type Shift = { key: string; isAsked: boolean; tries: number }
187let shifted: Shift | undefined
188
189function feed($: EngineInterface, event: ArmEvent): void {
190  const was = arming
191  arming = arm(arming, event)
192  if (event.type === 'reset') {
193    ringOn = undefined
194    shifted = undefined
195  }
196  if (arming.isArmed !== was.isArmed || arming.isListArmed !== was.isListArmed) {
197    $.ui.invalidate('ui.render')
198  }
199}
200
201// Moves the ring onto one of the pane's safe elements. Refused while the pane
202// does not hold the keys (Enter then goes to the prompt) or when another hook
203// says no: either way the ring may sit anywhere, so the pane disarms.
204async function focusOn($: EngineInterface, key: string): Promise<boolean> {
205  landedOn = undefined
206  const result = await $.ui.focus({ requestId: PANE, key }).catch(() => ({ deny: 'threw' }))
207  const isMoved = !('deny' in result && result.deny) && landedOn === key
208  feed($, { type: 'landed', isMoved })
209
210  return isMoved
211}
212
213// Puts the ring back on the element rows came above, a moment after the
214// drawing: a move asked while the hook draws resolves against the tree
215// before the one it returns, where the element still stands at its old
216// index. From a timer the move reaches no hook (see docs/feasibility/ui.md), so the
217// engine's answer is the evidence: a move it refuses or abandons counts as a
218// refused landing, and the next drawing asks again, at most three times. A
219// landing the focus hook saw since (the person's arrow) makes the move moot.
220const REGAIN_MS = 100
221
222async function regain($: EngineInterface, want: Shift): Promise<void> {
223  if (shifted !== want) return
224  const result = await $.ui
225    .focus({ requestId: PANE, key: want.key })
226    .catch(() => ({ deny: 'threw' }))
227  if (!('deny' in result && result.deny)) {
228    if (shifted === want) shifted = undefined
229    return
230  }
231  want.isAsked = false
232  feed($, { type: 'landed', isMoved: false })
233}
234
235// A row press while the ring is off its element: Enter would press the row
236// that took its index. A click while the pane does not hold the keys is the
237// person's own.
238const isHeldBack = (at: At): boolean =>
239  shifted !== undefined && seats.get(at.surface)?.isFocused === true && isRowKey(at.element)
240
241// A new view starts its hold anew, but the ring keeps its index, where the new
242// view may draw a stop; so the pane disarms and the ring moves (`landing`).
243async function land($: EngineInterface, keys: string[]): Promise<void> {
244  for (const key of keys) if (!(await focusOn($, key))) return
245}
246
247async function landView($: EngineInterface, action: Action, surface: RenderSurface): Promise<void> {
248  const { model, selection } = await look($, true)
249  const drawn = paneView(model, { columns: 80, rows: 24, surface, hasInput: true }, selection)
250  return land($, landing(action, selection.open, drawn))
251}
252
253// A press on a live row of the Items list away from the ring selects it: a
254// click moves no ring, so the press is the person's choice of that row. The
255// ring's row and the selected row open.
256async function choose($: EngineInterface, id: string): Promise<boolean> {
257  const { model, selection } = await look($, true)
258  if (!selectsOnPress(model, selection, id, ringOn)) return false
259  await $.state.set({ plugin: 'rabe', key: 'selected' }, id)
260  feed($, { type: 'press' })
261
262  return true
263}
264
265// Hides ended items for the session: they stay in `rabe.items`, which polls
266// fill again, and only the list leaves them out (`kept`). An id stays even when
267// the cap drops its item, which a poll may find again.
268async function remove($: EngineInterface, action: Action & { type: 'remove' | 'clear' }) {
269  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
270  const { value: query = '' } = await $.state.get({ plugin: 'rabe', key: 'query' })
271  let gone: RabeItem[] = []
272  for (;;) {
273    const { value: removed = [], version } = await $.state.get({ plugin: 'rabe', key: 'removed' })
274    gone = kept(items, removed).filter(
275      item =>
276        item.status !== 'running' &&
277        (action.type === 'remove' ? action.ids.includes(item.id) : matches(item, query)),
278    )
279    if (gone.length === 0) return
280    const next = [...removed, ...gone.map(item => item.id)]
281    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'removed' }, next, {
282      ifVersion: version,
283    })
284    if (isSet) break
285  }
286  const [one] = gone
287  toast(
288    $,
289    gone.length === 1 && one
290      ? `Removed ${KIND_LABEL[one.kind]} ${nameOf(one)}`
291      : `Removed ${gone.length} ended items`,
292  )
293}
294
295async function act($: EngineInterface, action: Action, surface: RenderSurface): Promise<void> {
296  if (action.type === 'open' && (await choose($, action.id))) return
297  if (action.type !== 'message' && landing(action, '').length > 0) feed($, { type: 'reset' })
298  switch (action.type) {
299    case 'tab':
300      await $.state.set({ plugin: 'rabe', key: 'tab' }, action.tab)
301      return landView($, action, surface)
302    case 'fold': {
303      const { value: folded = [] } = await $.state.get({ plugin: 'rabe', key: 'folded' })
304      const next = folded.includes(action.group)
305        ? folded.filter(group => group !== action.group)
306        : [...folded, action.group]
307      await $.state.set({ plugin: 'rabe', key: 'folded' }, next)
308      return land($, landing(action, ''))
309    }
310    case 'open': {
311      // Back selects the item that was open: the focus may have moved onto a
312      // row inside its detail (a workflow's agents). A folded forwarder has no
313      // row, so back selects its job.
314      const { value: open = '' } = await $.state.get({ plugin: 'rabe', key: 'open' })
315      const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
316      const was = rowOf(open, items)
317      const id = action.id || was
318      if (id) {
319        await $.state.set({ plugin: 'rabe', key: 'selected' }, id)
320        await $.state.set({ plugin: 'rabe', key: 'tab' }, 'items')
321      }
322      await $.state.set({ plugin: 'rabe', key: 'open' }, action.id)
323      return land($, landing(action, was))
324    }
325    case 'query':
326      await $.state.set({ plugin: 'rabe', key: 'query' }, action.text)
327      return action.isSubmit ? landView($, action, surface) : land($, landing(action, ''))
328    case 'window': {
329      const now = await $.clock.now()
330      const { value = windowOf(baseHours, now) } = await $.state.get({
331        plugin: 'rabe',
332        key: 'window',
333      })
334      await $.state.set({ plugin: 'rabe', key: 'window' }, widen(value, now))
335      return land($, landing(action, ''))
336    }
337    case 'focus':
338      await focusOn($, action.key)
339      return
340    case 'stop':
341      await stop($, action.ids)
342      return
343    case 'remove':
344    case 'clear':
345      await remove($, action)
346      return
347    case 'delete': {
348      const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
349      const item = items.find(one => one.id === action.id)
350      if (
351        item?.kind !== 'cron' ||
352        item.status !== 'running' ||
353        item.detail.scheduledFor !== undefined
354      )
355        return
356      const result = await $.tool.call({ tool: 'CronDelete', id: item.detail.jobId })
357      const reason = result.deny ?? (result.isError ? (result.text ?? 'n/a') : undefined)
358      toast(
359        $,
360        reason
361          ? `Delete refused: ${reason.replace(/<\/?tool_use_error>/g, '').trim()}`
362          : `Deleted cron ${nameOf(item)}`,
363      )
364      if (!reason) await queueAction($, `Rabe: user deleted cron ${item.id}`)
365      return
366    }
367    case 'copy': {
368      // Over SSH the clipboard needs the terminal's OSC 52.
369      const text = clipText(action.text)
370      const result = await $.ui
371        .copy({ text, surface })
372        .catch((error: unknown) => ({ isCopied: false as const, reason: String(error) }))
373      const preview = toastText(text)
374      // only text the filters left alone is offered for selection
375      const selectable = !action.what && preview === action.text
376      const copied = action.what ? `Copied ${action.what}` : `Copied: ${preview}`
377      toast(
378        $,
379        result.isCopied
380          ? copied
381          : `Copy failed: ${result.reason}.${selectable ? ` Select it: ${text}` : ''}`,
382      )
383      return
384    }
385    case 'none':
386      return
387    case 'message': {
388      const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
389      const item = items.find(one => one.id === action.id)
390      if (item?.kind !== 'agent' || !action.text.trim()) return
391      const sent = await $.session.send({ to: { agentId: item.detail.agentId }, text: action.text })
392      toast(
393        $,
394        sent.isDelivered
395          ? `Message sent to claude ${item.detail.name || item.title}`
396          : `Message not sent: ${sent.reason}`,
397      )
398      if (sent.isDelivered) await queueAction($, `Rabe: user sent a message to agent ${item.id}`)
399      return sequence(async () => {
400        const { selection } = await look($, true)
401        const target = messageLanding(action, selection, ringOn)
402        if (target) await land($, target)
403      })
404    }
405  }
406}
407
408// What the pane draws from: the sources' values and the person's place.
409async function look(
410  $: EngineInterface,
411  isFocused: boolean,
412): Promise<{ model: Model; selection: Selection }> {
413  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
414  const { value: removed = [] } = await $.state.get({ plugin: 'rabe', key: 'removed' })
415  const { value: turns = {} } = await $.state.get({ plugin: 'rabe', key: 'turns' })
416  const { value: lines = {} } = await $.state.get({ plugin: 'rabe', key: 'lines' })
417  const { value: edits = [] } = await $.state.get({ plugin: 'rabe', key: 'edits' })
418  const { value: tab = 'items' } = await $.state.get({ plugin: 'rabe', key: 'tab' })
419  const { value: query = '' } = await $.state.get({ plugin: 'rabe', key: 'query' })
420  const { value: folded = [] } = await $.state.get({ plugin: 'rabe', key: 'folded' })
421  const { value: selected = '' } = await $.state.get({ plugin: 'rabe', key: 'selected' })
422  const { value: open = '' } = await $.state.get({ plugin: 'rabe', key: 'open' })
423  const { value: order } = await $.state.get({ plugin: 'rabe', key: 'order' })
424  const { value: worktrees } = await $.state.get({ plugin: 'rabe', key: 'worktrees' })
425  const { value: window } = await $.state.get({ plugin: 'rabe', key: 'window' })
426  const now = await $.clock.now()
427  const usage = await $.session.usage().catch(() => undefined)
428  const model = {
429    items,
430    removed,
431    turns,
432    lines,
433    now,
434    usd: usage?.cost?.usd,
435    previous: await previous($).catch(() => undefined),
436    edits,
437    cwd: await $.session.cwd().catch(() => undefined),
438    sessionId: await $.session.id().catch(() => undefined),
439    ...(worktrees && { worktrees }),
440  }
441
442  const selection = {
443    tab,
444    query,
445    folded,
446    selected,
447    open,
448    isFocused,
449    isArmed: arming.isArmed,
450    isListArmed: arming.isListArmed,
451    window: window ?? windowOf(baseHours, now),
452    ...(order && { order }),
453  }
454
455  return { model, selection }
456}
457
458// The view the person chose. A change of it is theirs, so its hold starts anew.
459const scopeOf = (sel: Selection) =>
460  JSON.stringify([sel.tab, sel.open, sel.query, sel.folded, sel.window?.hours])
461
462// Per surface, the focusable keys the pane drew since it opened, for one view
463// (`scope`). A render hook may not write `$.state` (drawing is pure), so this
464// is a module value: a reload starts the hold anew, as a new drawing would.
465const holds = new Map<string, { scope: string; keys: Held[] }>()
466
467// Per surface, where the pane sat and whether it held the keys at the last
468// drawing. A move between dock and inline takes the keys from the pane.
469const seats = new Map<string, { placement: string; isFocused: boolean }>()
470
471// The presses and Input events under way, per surface and element, oldest
472// first. Each event's hook adds its own slot before `next(e)`; the closure
473// beneath, which the engine runs with the event, fills the oldest empty slot of
474// its element, so an event carries out only an action of its own element.
475// ponytail: two events on one element in flight at once may swap their
476// actions; both are that element's, and each runs exactly once.
477type Slot = { press?: { action: Action; surface: RenderSurface } }
478const waiting = new Map<string, Slot[]>()
479
480const slotOf = (at: At) => `${at.surface} ${at.element}`
481
482function leave(action: Action, at: At): void {
483  const slot = waiting.get(slotOf(at))?.find(one => !one.press)
484  if (slot) slot.press = { action, surface: at.surface }
485}
486
487// Runs a press or an Input event beneath the hook, then carries out the action
488// its closure left in this event's own slot with the hook's `$`.
489async function carry<R>($: EngineInterface, at: At, run: () => Promise<R>): Promise<R> {
490  const key = slotOf(at)
491  const mine: Slot = {}
492  const focusTurn = reserve()
493  waiting.set(key, [...(waiting.get(key) ?? []), mine])
494  let result: R
495  try {
496    result = await run()
497  } catch (error) {
498    await focusTurn(async () => undefined)
499    throw error
500  } finally {
501    const rest = (waiting.get(key) ?? []).filter(one => one !== mine)
502    if (rest.length > 0) waiting.set(key, rest)
503    else waiting.delete(key)
504  }
505  const press = mine.press
506  if (press && !isHeldBack(at)) {
507    const action = press.action
508    if (
509      action.type === 'tab' ||
510      action.type === 'fold' ||
511      action.type === 'open' ||
512      action.type === 'query' ||
513      action.type === 'window' ||
514      action.type === 'focus'
515    ) {
516      await focusTurn(() => act($, action, press.surface))
517    } else {
518      await focusTurn(async () => undefined)
519      await act($, action, press.surface)
520    }
521  } else {
522    await focusTurn(async () => undefined)
523  }
524
525  return result
526}
527
528const heldOf = (surface: RenderSurface, sel: Selection): Held[] | undefined => {
529  const mine = holds.get(surface)
530  return mine?.scope === scopeOf(sel) ? mine.keys : undefined
531}
532
533// An arrow key in a pane taller than its body scrolls it a row; Rabe moves
534// the focus to the next or previous row instead, and the pane follows the
535// focus. Past either end the scroll goes on, to show what is above or below.
536async function arrow($: EngineInterface, by: number, bodyRows: number): Promise<number | false> {
537  const { model, selection } = await look($, true)
538  const size = { columns: 80, rows: bodyRows, surface: 'terminal', hasInput: true } as const
539  const drawn = layout(paneView(model, size, selection))
540  const keys = rowKeys(hold(drawn, heldOf('terminal', selection), seatsRows(selection)).list)
541  const key = stepRow(keys, selection.selected, by)
542  if (!key) return false
543  const from = keys.indexOf(`row:${selection.selected}`)
544  const to = keys.indexOf(key)
545  const moved = from < 0 ? to + 1 : to - from
546  feed($, { type: 'step', key })
547  await focusOn($, key)
548
549  return by - moved
550}
551
552export function pane(on: On, hours = baseHours): void {
553  baseHours = hours
554  on('session.start', async ($, e, next) => {
555    await $.command.register({
556      name: 'rabe',
557      description:
558        'Show background work: agents, Codex jobs, shells, monitors, cron jobs and workflows',
559    })
560    $.clock.every(1000, () => void tick($))
561
562    return next(e)
563  })
564
565  // Each open sorts the lists once; then they hold their order (see `stable`).
566  on('command.run', { command: 'rabe' }, async $ => {
567    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
568    const { value: edits = [] } = await $.state.get({ plugin: 'rabe', key: 'edits' })
569    await $.state.set({ plugin: 'rabe', key: 'order' }, orderOf(items, edits))
570    await $.state.set({ plugin: 'rabe', key: 'window' }, windowOf(baseHours, await $.clock.now()))
571    holds.clear()
572    feed($, { type: 'reset' })
573    await $.ui.open({ id: PANE, title: 'Rabe', closeOnEscape: true })
574    $.clock.after(1500, () => void refocus($).catch(() => undefined))
575
576    return { text: 'Rabe opened.' }
577  })
578
579  on('session.end', { reason: /^/ }, async ($, e, next) => {
580    await remember($, e.resume?.id ?? e.sessionId)
581
582    return next(e)
583  }).catch((_$, e, next) => next(e))
584
585  on('prompt.submit', { origin: { kind: /^/ } }, async ($, e, next) => {
586    const { value: actions = [] } = await $.state.get({ plugin: 'rabe', key: 'actions' })
587    if (actions.length === 0) return next(e)
588    const result = await next({ ...e, context: [...(e.context ?? []), ...actions] })
589    if (!result.drop) await clearActions($, actions)
590    return result
591  }).catch((_$, e, next) => next(e))
592
593  // The ring lands where the engine's link received the move (`landingOf`),
594  // maybe not `e.element`; the selection follows it. A move that never
595  // reached the engine left the ring where it was, so the pane disarms.
596  on('ui.focus', { requestId: PANE }, ($, e, next) => {
597    const focus = async () => {
598      const { value: before = '' } = await $.state.get({ plugin: 'rabe', key: 'selected' })
599      const asked = e.element?.startsWith('row:') ? e.element.slice(4) : before
600      if (asked !== before) await $.state.set({ plugin: 'rabe', key: 'selected' }, asked)
601      const result = await next(e)
602      const landing = landingOf(next.trace, 'deny' in result && !!result.deny)
603      const landed = landing?.element
604      const selected = landed?.startsWith('row:') ? landed.slice(4) : before
605      if (selected !== asked) await $.state.set({ plugin: 'rabe', key: 'selected' }, selected)
606      landedOn = landed
607      if (landing) {
608        ringOn = landed
609        shifted = undefined
610      }
611      if (!landing) feed($, { type: 'landed', isMoved: false })
612      else if (landed) {
613        const { model, selection } = await look($, true)
614        feed($, {
615          type: 'focus',
616          byPerson: e.origin.kind === 'person',
617          key: landed,
618          requested: e.element,
619          isLiveRow: isLiveRow(model, selection, landed),
620          selected: selection.selected,
621        })
622      }
623
624      return result
625    }
626    return e.origin.kind === 'person' && ringOn && e.element !== ringOn ? sequence(focus) : focus()
627  }).catch((_$, e, next) => next(e))
628
629  // Rabe's own `$.ui.focus` from a Button's or Input's closure reaches no
630  // hook of Rabe's, so its landings would never count; calls from these do.
631  on('ui.press', { requestId: PANE }, ($, e, next) => carry($, e, () => next(e))).catch(
632    (_$, e, next) =>
633      next.called && next.error.message?.includes('no handler is held under handle')
634        ? { element: e.element }
635        : next(e),
636  )
637  on('ui.input', { requestId: PANE }, ($, e, next) => carry($, e, () => next(e))).catch(
638    (_$, e, next) =>
639      next.called && next.error.message?.includes('no handler is held under handle')
640        ? { element: e.element, value: e.value }
641        : next(e),
642  )
643
644  on('ui.scroll', { requestId: PANE }, async ($, e, next) => {
645    if (e.pointer || e.by === 0 || Math.abs(e.by) >= e.bodyRows) return next(e)
646    const remaining = await sequence(() => arrow($, e.by, e.bodyRows)).catch(() => false)
647    if (remaining === 0) return {}
648    if (typeof remaining === 'number')
649      return next(
650        e.offset > 0 && e.offset < e.contentRows - e.bodyRows
651          ? { ...e, offset: e.offset - e.by + remaining }
652          : e,
653      )
654
655    return next(e)
656  })
657
658  on('ui.render', { component: 'Pane', requestId: PANE }, async ($, e) => {
659    const ui = $.ui.resolve(e)
660    const seat = seats.get(e.surface)
661    seats.set(e.surface, { placement: e.props.placement, isFocused: e.props.isFocused })
662    if (seat?.isFocused && seat.placement !== e.props.placement) {
663      $.clock.after(500, () => void refocus($).catch(() => undefined))
664    }
665    const { model, selection: seen } = await look($, e.props.isFocused)
666    const size = bounded({
667      columns: e.props.bodyColumns,
668      rows: e.props.scroll?.bodyRows || 24,
669      surface: e.surface,
670      hasInput: 'Input' in ui,
671      ...(e.props.scroll && {
672        window: { top: e.props.scroll.offset, rows: e.props.scroll.bodyRows },
673      }),
674    })
675
676    // What x and g would act on decides whether they may.
677    const armed = paneView(model, size, { ...seen, isArmed: true, isListArmed: true })
678    const { selected } = seen
679    feed($, {
680      type: 'drawn',
681      fallback: fallbackOf(model, seen),
682      targets: targetsOf(armed),
683      selected,
684    })
685    const selection = { ...seen, isArmed: arming.isArmed, isListArmed: arming.isListArmed }
686    const drawn = arming.isArmed && arming.isListArmed ? armed : paneView(model, size, selection)
687    const before = heldOf(e.surface, selection)
688    const out = hold(layout(drawn), before, seatsRows(selection))
689    holds.set(e.surface, { scope: scopeOf(selection), keys: out.held })
690    if (before && ringOn && shifts(before, out.held, ringOn)) {
691      shifted = { key: ringOn, isAsked: false, tries: 0 }
692    }
693    if (shifted && seat && !seat.isFocused && e.props.isFocused) shifted.tries = 0
694    if (shifted && !shifted.isAsked && shifted.tries < 3 && e.props.isFocused) {
695      shifted.isAsked = true
696      shifted.tries += 1
697      const want = shifted
698      $.clock.after(REGAIN_MS, () => void regain($, want).catch(() => undefined))
699    }
700
701    return render(ui, e.surface, out.list, leave)
702  })
703}
704

1import type { RabeWindow } from '../../../types'
2import type { RabeItem } from '../../model'
3import { nextRun } from '../../schedule'
4import { fit } from '../cells/grid'
5import { C, CHIP, type Style, tone } from '../cells/palette'
6import { clockTime, day, duration, tokens, usd } from '../format'
7import {
8  bar,
9  byStart,
10  forwarderOf,
11  glyph,
12  KIND_LABEL,
13  kept,
14  nameSpans,
15  shown,
16  stable,
17  tree,
18} from '../lists'
19import type { Drawn, Line, Model, View, ViewButton } from '../view'
20import { resumeButton } from './detail'
21import { beside, fitLine, focusOn, text } from './lines'
22
23const dim = { fg: C.dim }
24const BOX = 40
25const SIDE_COLUMNS = 90
26
27const LEGEND: [string, Style][] = [
28  ['shell', CHIP.shell],
29  ['claude', CHIP.agent],
30  ['codex', CHIP.codex],
31  ['monitor', CHIP.monitor],
32  ['scheduled run', CHIP.cron],
33  ['done', CHIP.done],
34  ['failed', CHIP.failed],
35]
36
37const COUNTS: [RabeItem['kind'], string, string][] = [
38  ['agent', 'agent', 'agents'],
39  ['codex', 'codex', 'codex'],
40  ['workflow', 'workflow', 'workflows'],
41  ['shell', 'shell', 'shells'],
42  ['monitor', 'monitor', 'monitors'],
43]
44
45function color(item: RabeItem): number {
46  if (item.status === 'failed') return C.red
47  if (item.status === 'done') return C.green
48  if (item.status === 'stopped') return C.dim
49
50  return CHIP[item.kind].fg
51}
52
53// One cell per slice of the window: a bar over the run, or for a cron job a
54// tick at each scheduled run since Rabe saw it (jitter not counted).
55// What began before the window starts with `◂` at the left edge.
56function track(item: RabeItem, start: number, now: number, width: number): string {
57  const out = cells(item, start, now, width)
58
59  return (item.startedAt ?? item.seenAt) < start ? `◂${out.slice(1)}` : out
60}
61
62function cells(item: RabeItem, start: number, now: number, width: number): string {
63  if (item.kind !== 'cron') {
64    return bar(item.startedAt ?? item.seenAt, item.endedAt ?? now, start, now, width)
65  }
66  const cells = Array<string>(width).fill(' ')
67  const schedule = item.detail.schedule
68  const span = Math.max(1, now - start)
69  const end = Math.min(now, item.endedAt ?? now)
70  let at = schedule ? nextRun(schedule, Math.max(start, item.startedAt ?? item.seenAt)) : undefined
71  for (let n = 0; at !== undefined && at <= end && n < width; n++) {
72    cells[Math.min(width - 1, Math.floor(((at - start) / span) * width))] = '█'
73    at = nextRun(schedule as string, at)
74  }
75
76  return cells.join('')
77}
78
79function treeLines(items: RabeItem[]): Line[] {
80  return [
81    {
82      spans: [
83        ['WHO STARTED WHAT?', { fg: C.bright }],
84        ['  agents and their children', dim],
85      ],
86    },
87    { spans: [['main session']] },
88    ...tree(items).map(
89      (line): Line => ({
90        spans: [
91          [line.prefix, dim],
92          [`${glyph(line.item)} `, { fg: tone(line.item) }],
93          [line.item.title],
94          [line.item.kind === 'agent' ? '' : ` (${KIND_LABEL[line.item.kind]})`, dim],
95        ],
96      }),
97    ),
98  ]
99}
100
101function previousLines(model: Model, width: number): Line[] {
102  const prev = model.previous
103  const head: Line = { spans: [[' PREVIOUS SESSION', { fg: C.bright }]] }
104  if (!prev) {
105    return [
106      head,
107      ...text('No earlier session with background work in this project.', width, dim, ' '),
108    ]
109  }
110  const counts = COUNTS.flatMap(([kind, one, many]) => {
111    const n = prev.counts[kind] ?? 0
112    return n ? [`${n} ${n === 1 ? one : many}`] : []
113  })
114  const crons = prev.counts.cron ?? 0
115  const took = prev.startedAt === undefined ? '' : ` · ${duration(prev.endedAt - prev.startedAt)}`
116  const failed = prev.failed
117    .slice(0, 2)
118    .map((title): Line => ({ spans: [[' ✗ failed', { fg: C.red }], [` ${title}`]] }))
119  const more = prev.failed.length - failed.length
120
121  const out: Line[] = [
122    head,
123    { spans: [[` this project · ended ${day(prev.endedAt, model.now)}`, dim]] },
124    { spans: [[' ✓', { fg: C.green }], [` ${counts.join(' · ') || 'no items'}`]] },
125    {
126      spans: [
127        [' session ', dim],
128        [prev.usd === undefined ? 'cost n/a' : `≈ ${usd(prev.usd)}`, { fg: C.bright }],
129        [took, dim],
130      ],
131    },
132    { spans: [[` background ${tokens(prev.tokens)} tok`, dim]] },
133    ...failed,
134    ...(more ? [{ spans: [[` +${more} more failed`, dim]] } as Line] : []),
135    ...(crons
136      ? [
137          {
138            spans: [
139              [' ⟳', { fg: C.purple }],
140              [` ${crons} cron job${crons === 1 ? '' : 's'} ended with the session`, dim],
141            ],
142          } as Line,
143        ]
144      : []),
145    { spans: [[` id ${prev.sessionId ?? 'n/a'}`, dim]] },
146  ]
147
148  return out.map(line => ({ ...line, bg: C.raised }))
149}
150
151const MIN = 60_000
152const STEPS = [1, 2, 5, 10, 15, 30, 60, 120, 180, 360, 720, 1440].map(one => one * MIN)
153
154// The time axis: round clock times, at most four, spaced for the window, then
155// `now`. A label that would repeat the one before it, run into it or into
156// `now` is left out.
157function axis(start: number, span: number, labelWidth: number, columns: number): string {
158  const barWidth = columns - labelWidth
159  const cells = Array<string>(columns).fill(' ')
160  const put = (x: number, value: string) => {
161    ;[...value].forEach((ch, i) => {
162      if (x + i >= 0 && x + i < columns) cells[x + i] = ch
163    })
164  }
165  const step = STEPS.find(one => span / one <= 4) ?? (STEPS.at(-1) as number)
166  // Round in local time: the zone's offset from UTC at the start.
167  const offset = new Date(start).getTimezoneOffset() * MIN
168  let last = ''
169  let end = labelWidth
170  for (let at = Math.ceil((start - offset) / step) * step + offset; at < start + span; at += step) {
171    const label = clockTime(at).slice(0, 5)
172    const x = labelWidth + Math.round((barWidth * (at - start)) / span) - 2
173    if (label === last || x < end || x + label.length > columns - 4) continue
174    put(x, label)
175    last = label
176    end = x + label.length + 1
177  }
178  if (!last && labelWidth + 5 <= columns - 4) put(labelWidth, clockTime(start).slice(0, 5))
179  put(columns - 3, 'now')
180
181  return cells.join('')
182}
183
184const HOUR = 3_600_000
185const DEFAULT_HOURS = 4
186const WIDE_HOURS = 12
187
188// The option `timelineHours`: hours, 0 for no limit; anything else the default.
189export function hoursOf(value: unknown): number {
190  const hours = typeof value === 'string' && value.trim() ? Number(value) : value
191  return typeof hours === 'number' && Number.isFinite(hours) && hours >= 0 ? hours : DEFAULT_HOURS
192}
193
194// The window `/rabe` opens with: the last `base` hours, or the whole session.
195export const windowOf = (base: number, now: number): RabeWindow => ({
196  base,
197  hours: base,
198  since: base > 0 ? now - base * HOUR : 0,
199})
200
201const steps = (base: number) => [base, ...(base < WIDE_HOURS ? [WIDE_HOURS] : []), 0]
202
203const next = (w: RabeWindow) => {
204  const list = steps(w.base)
205  return list[(list.indexOf(w.hours) + 1) % list.length] ?? 0
206}
207
208// The next window `w` steps to: base hours, 12 hours, the whole session, then
209// base again. A wider one never starts later than the one before.
210export function widen(w: RabeWindow, now: number): RabeWindow {
211  const hours = next(w)
212  if (hours === 0) return { ...w, hours, since: 0 }
213  const since = now - hours * HOUR
214
215  return { ...w, hours, since: w.hours === 0 || hours < w.hours ? since : Math.min(w.since, since) }
216}
217
218const spanText = (ms: number) => (ms < 2 * HOUR ? `${Math.round(ms / 60_000)} min` : duration(ms))
219
220// The Timeline tab: a bar per item over the window with a legend and a time
221// axis, then the tree of who started what beside (or above) the previous
222// session in this project. The rows keep the order the pane opened with.
223// Items that ended before the window (`sel.window`) fold into one line of
224// plain text, so the rows' focus order stays; `w` widens the window.
225export const timelineView: View = (model, size, sel): Drawn => {
226  const since = sel.window?.since ?? 0
227  const all = stable(kept(shown(model.items), model.removed), sel.order?.timeline, byStart)
228  const list = all.filter(item => item.status === 'running' || (item.endedAt ?? since) >= since)
229  const folded = all.length - list.length
230  const selected = list.find(item => item.id === sel.selected) ?? list[0]
231  const start = Math.max(
232    since,
233    Math.min(model.now - 60_000, ...list.map(item => item.startedAt ?? item.seenAt)),
234  )
235  const span = model.now - start
236  const labelWidth = Math.min(24, Math.floor(size.columns / 3))
237  const barWidth = size.columns - labelWidth
238  const isSide = size.columns >= SIDE_COLUMNS
239  const bars: Line[] = list.length
240    ? list.map(item => {
241        const name = fit(
242          nameSpans(item)
243            .map(([one]) => one)
244            .join(''),
245          labelWidth - 2,
246        ).trimEnd()
247        return {
248          spans: [
249            [item === selected ? '▌' : ' ', { fg: C.orange }],
250            {
251              key: `row:${item.id}`,
252              label: name,
253              action: { type: 'open', id: item.id },
254              ...(item !== selected && { dim: true }),
255            },
256            [' '.repeat(labelWidth - 2 - [...name].length + 1)],
257            [track(item, start, model.now, barWidth), { fg: color(item) }],
258          ],
259          ...(item === selected && { bg: C.selected }),
260        }
261      })
262    : folded
263      ? []
264      : [{ spans: [[' Nothing ran yet.', dim]] }]
265  const fold: Line[] = folded
266    ? [
267        {
268          spans: [
269            [
270              ` +${folded} older item${folded === 1 ? '' : 's'}, ended before ${day(since, model.now).replace(/^today /, '')}`,
271              dim,
272            ],
273          ],
274        },
275      ]
276    : []
277  // a forwarder has no row, but stays in the tree above its job
278  const who = treeLines(
279    model.items.filter(
280      item => list.includes(item) || list.some(one => forwarderOf(one, model.items) === item),
281    ),
282  )
283  const prev = previousLines(model, isSide ? BOX : size.columns)
284  const below = isSide
285    ? beside(who, size.columns - BOX - 2, [['  ']], prev, BOX)
286    : [...who, { spans: [] }, ...prev].map(line => fitLine(line, size.columns))
287  const lines: Line[] = [
288    {
289      spans: [
290        ['WHEN DID THINGS RUN?', { fg: C.bright }],
291        [`  this session, last ${spanText(span)}`, dim],
292      ],
293    },
294    { spans: LEGEND.flatMap(([label, chip]) => [[` ${label} `, chip], [' ']]) },
295    { spans: [] },
296    { spans: [[axis(start, span, labelWidth, size.columns), dim]] },
297    ...fold,
298    ...focusOn(bars, selected?.id ?? ''),
299    { spans: [] },
300  ]
301  const w = sel.window
302  const buttons: ViewButton[] =
303    w && w.base > 0
304      ? [
305          {
306            key: 'window',
307            label: `w: show ${next(w) ? `${next(w)} h` : 'all'}`,
308            hotkey: 'w',
309            action: { type: 'window' },
310          },
311        ]
312      : []
313
314  const id = model.previous?.sessionId
315  if (id) buttons.push(resumeButton(`claude --resume ${id}`, id))
316
317  return { nodes: [...lines.map(line => fitLine(line, size.columns)), ...below], buttons }
318}
319

1import type {
2  AgentInfo,
3  AgentStatus,
4  EngineInterface,
5  On,
6  TurnCompleteReason,
7  TurnStepResult,
8  TurnUsage,
9} from 'claude-code'
10
11import type { RabeCodexCall, RabeEdit, RabeToolUse, RabeTurn } from '../../types'
12import { companionCall, companionOutput } from '../forwarders'
13import {
14  clip,
15  type EndStatus,
16  itemId,
17  type RabeItem,
18  type RabeItemOf,
19  type RabeItemStatus,
20  type RabeTokens,
21} from '../model'
22import { claudeUsd, type Prices, parsePrices, withOverride } from '../prices'
23import { addItem, type Change, commit, endItem, pastEnd, prune, updateItem } from '../registry'
24import { changed, type Seen, shellWrites } from '../writes'
25
26type Turns = Record<string, RabeTurn[]>
27type AgentItem = RabeItemOf<'agent'>
28
29const POLL_MS = 3000
30const MAX_TURNS = 30
31const MAX_ANSWER = 16000
32const MAX_EDITS = 100
33const MAX_CALLS = 10
34const MAX_CHECKS = 20
35const LOOK_MS = 1000
36const MAX_PROMPT = 600
37const MAX_SUMMARY = 80
38const SUMMARY_FIELDS = ['file_path', 'command', 'pattern', 'path', 'url', 'query', 'description']
39const META_FIELDS = ['cwd', 'worktreePath', 'worktreeBranch', 'workflowPhase'] as const
40
41const LISTED: Record<AgentStatus, RabeItemStatus> = {
42  pending: 'running',
43  running: 'running',
44  waiting: 'running',
45  idle: 'running',
46  completed: 'done',
47  failed: 'failed',
48  killed: 'stopped',
49}
50
51const ENDED: Record<TurnCompleteReason, EndStatus> = {
52  answer: 'done',
53  aborted: 'stopped',
54  refusal: 'failed',
55  error: 'failed',
56}
57
58export function toolSummary(input: unknown): string | undefined {
59  if (!input || typeof input !== 'object') return undefined
60  const fields = input as Record<string, unknown>
61  const name = SUMMARY_FIELDS.find(one => typeof fields[one] === 'string')
62  if (!name) return undefined
63  const line = String(fields[name]).split('\n')[0] ?? ''
64
65  // Effects tells files apart by the whole path.
66  return name === 'file_path' ? line : clip(line, MAX_SUMMARY)
67}
68
69export function metaPatch(text: string): Partial<AgentItem['detail']> | undefined {
70  try {
71    const meta = JSON.parse(text) as Record<string, unknown>
72
73    return Object.fromEntries(
74      META_FIELDS.filter(name => typeof meta[name] === 'string').map(name => [name, meta[name]]),
75    )
76  } catch {
77    return undefined
78  }
79}
80
81export function metaPath(item: AgentItem, items: RabeItem[]): string | undefined {
82  const parent = items.find(one => one.id === item.parentId)
83  if (parent?.kind === 'workflow') {
84    const folder =
85      parent.detail.transcriptDir ??
86      parent.detail.scriptPath?.replace(
87        /\/workflows\/scripts\/[^/]+$/,
88        `/subagents/workflows/${parent.detail.runId}`,
89      )
90    if (folder && folder !== parent.detail.scriptPath)
91      return `${folder}/agent-${item.detail.agentId}.meta.json`
92  }
93
94  return item.detail.transcriptPath?.replace(/\.jsonl$/, '.meta.json')
95}
96
97export function agentTranscript(path: string, agentId: string): string | undefined {
98  if (path.endsWith(`/agent-${agentId}.jsonl`)) return path
99  if (!path.endsWith('.jsonl')) return undefined
100
101  return `${path.slice(0, -'.jsonl'.length)}/subagents/agent-${agentId}.jsonl`
102}
103
104function workflowTranscript(
105  path: string | undefined,
106  agentId: string,
107  runId: string,
108): string | undefined {
109  const generic = `/subagents/agent-${agentId}.jsonl`
110  return path?.endsWith(generic)
111    ? `${path.slice(0, -generic.length)}/subagents/workflows/${runId}/agent-${agentId}.jsonl`
112    : path
113}
114
115// The file an Edit or Write changed: the engine ran it (a result, no error) and
116// did not only stage it for review. The path comes from the result when it has one.
117export function changedFile(
118  answer: { result?: unknown; isError?: boolean },
119  input: string,
120): string | undefined {
121  if (answer.isError || !answer.result || typeof answer.result !== 'object') return undefined
122  const result = answer.result as { filePath?: unknown; staged?: unknown }
123  if (result.staged === true) return undefined
124
125  return typeof result.filePath === 'string' ? result.filePath : input
126}
127
128// A Bash call that ran to its end: a result, no error, not interrupted and not
129// moved to the background.
130export function ranBash(answer: { result?: unknown; isError?: boolean }): boolean {
131  if (answer.isError || !answer.result || typeof answer.result !== 'object') return false
132  const result = answer.result as { interrupted?: unknown; backgroundTaskId?: unknown }
133
134  return result.interrupted !== true && result.backgroundTaskId === undefined
135}
136
137// Changes to files of an agent; its newest MAX_EDITS are kept.
138function edited(items: RabeItem[], id: string, changes: RabeEdit[]): RabeItem[] {
139  const agent = asAgent(items, id)
140  if (!agent) return items
141  const edits = [...(agent.detail.edits ?? []), ...changes].slice(-MAX_EDITS)
142
143  return updateItem(items, id, { detail: { edits } })
144}
145
146// The companion calls of an agent: a new one, or what its answer told.
147function called(
148  items: RabeItem[],
149  id: string,
150  at: number,
151  patch: Partial<RabeCodexCall>,
152): RabeItem[] {
153  const agent = asAgent(items, id)
154  if (!agent) return items
155  const calls = agent.detail.codexCalls ?? []
156  const codexCalls = calls.some(one => one.at === at)
157    ? calls.map(one => (one.at === at ? { ...one, ...patch } : one))
158    : [...calls, { at, ...patch } as RabeCodexCall].slice(-MAX_CALLS)
159
160  return updateItem(items, id, { detail: { codexCalls } })
161}
162
163// A call that went to the background still runs: it keeps no end. A failed
164// run still printed its thread.
165function answered(answer: { result?: unknown }, now: number): Partial<RabeCodexCall> {
166  const result = (answer.result ?? {}) as Record<string, unknown>
167  if (result.backgroundTaskId) return {}
168
169  return { endedAt: now, ...companionOutput(`${result.stdout ?? ''}\n${result.stderr ?? ''}`) }
170}
171
172export function addTurn(turns: Turns, id: string, turn: RabeTurn): Turns {
173  return { ...turns, [id]: [...(turns[id] ?? []), turn].slice(-MAX_TURNS) }
174}
175
176function addUsage(tokens: RabeTokens | undefined, usage: TurnUsage): RabeTokens {
177  return {
178    input:
179      (tokens?.input ?? 0) +
180      usage.input_tokens +
181      usage.cache_creation_input_tokens +
182      usage.cache_read_input_tokens,
183    output: (tokens?.output ?? 0) + usage.output_tokens,
184    cached: (tokens?.cached ?? 0) + usage.cache_read_input_tokens,
185  }
186}
187
188// Each step adds its own price, so a long-context step bills at its own tier. The
189// sum is known only while every step since the start was: an agent Rabe did not
190// see start, or one step without a price, stays n/a.
191function addCost(item: AgentItem, usd: number | undefined): number | undefined {
192  if (usd === undefined || item.startedAt === undefined) return undefined
193  if (!item.tokens) return usd
194
195  return item.costUsd === undefined ? undefined : item.costUsd + usd
196}
197
198function asAgent(items: RabeItem[], id: string): AgentItem | undefined {
199  const item = items.find(one => one.id === id)
200
201  return item?.kind === 'agent' ? item : undefined
202}
203
204// Statuses of a turn in flight: an ended agent in one was resumed. `idle` is
205// not one: a teammate is idle after each turn, which already ended its item.
206const RESUMED: AgentStatus[] = ['pending', 'running', 'waiting']
207
208// An ended agent Rabe holds no item for takes `end` (see `pastEnd`), and none
209// is added when `end` is undefined, so a poll never gives one a new recency.
210function listed(
211  items: RabeItem[],
212  info: AgentInfo,
213  now: number,
214  end: number | undefined,
215): RabeItem[] {
216  const id = itemId('agent', info.id)
217  const status = LISTED[info.status] ?? 'running'
218  const held = asAgent(items, id)
219  if (!held) {
220    if (status !== 'running' && end === undefined) return items
221    return addItem(
222      items,
223      {
224        id,
225        kind: 'agent',
226        title: info.name || info.description || info.type,
227        status,
228        ...(status !== 'running' && { endedAt: end }),
229        parentId: info.parentId ? itemId('agent', info.parentId) : undefined,
230        detail: {
231          agentId: info.id,
232          ...(info.name && { name: info.name }),
233          type: info.type,
234          description: info.description,
235        },
236      },
237      now,
238    )
239  }
240  const named = info.name
241    ? updateItem(items, id, { title: info.name, detail: { name: info.name } })
242    : items
243  if (held.status !== 'running' && RESUMED.includes(info.status)) {
244    return updateItem(named, id, { status: 'running', endedAt: undefined })
245  }
246
247  return status === 'running' ? named : endItem(named, id, status, now)
248}
249
250function stepped(
251  items: RabeItem[],
252  id: string,
253  result: TurnStepResult,
254  tools: RabeToolUse[],
255  now: number,
256  usd: number | undefined,
257): RabeItem[] {
258  const item = asAgent(items, id)
259  if (!item) return items
260  const last = tools.at(-1)
261
262  return updateItem(items, id, {
263    status: 'running',
264    endedAt: undefined,
265    tokens: result.usage ? addUsage(item.tokens, result.usage) : item.tokens,
266    costUsd: result.usage ? addCost(item, usd) : item.costUsd,
267    detail: {
268      model: result.usage?.model ?? item.detail.model,
269      toolCount: (item.detail.toolCount ?? 0) + tools.length,
270      lastTool: last?.name ?? item.detail.lastTool,
271      lastToolAt: last ? now : item.detail.lastToolAt,
272    },
273  })
274}
275
276async function write($: Pick<EngineInterface, 'state'>, change: Change): Promise<void> {
277  for (;;) {
278    const { value, version } = await $.state.get({ plugin: 'rabe', key: 'items' })
279    const next = commit(value, change)
280    if (next === undefined) return
281    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'items' }, next.items, {
282      ifVersion: version,
283    })
284    if (isSet) return forget($, next.dropped)
285  }
286}
287
288// Drops the lines and turns of the items a write dropped.
289async function forget($: Pick<EngineInterface, 'state'>, dropped: string[]): Promise<void> {
290  if (dropped.length === 0) return
291  for (;;) {
292    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
293    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
294    const next = prune(value, dropped, items)
295    if (next === value) break
296    const lines = await $.state.set({ plugin: 'rabe', key: 'lines' }, next, { ifVersion: version })
297    if (lines.isSet) break
298  }
299  for (;;) {
300    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'turns' })
301    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
302    const next = prune(value, dropped, items)
303    if (next === value) return
304    const turns = await $.state.set({ plugin: 'rabe', key: 'turns' }, next, { ifVersion: version })
305    if (turns.isSet) return
306  }
307}
308
309async function writeTurns(
310  $: Pick<EngineInterface, 'state'>,
311  change: (turns: Turns) => Turns,
312): Promise<void> {
313  for (;;) {
314    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'turns' })
315    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'turns' }, change(value), {
316      ifVersion: version,
317    })
318    if (isSet) return
319  }
320}
321
322// The main session's changes go to `rabe.edits`, an agent's to its item.
323async function record(
324  $: EngineInterface,
325  agentId: string | undefined,
326  changes: RabeEdit[],
327): Promise<void> {
328  if (changes.length === 0) return
329  if (agentId) {
330    const id = itemId('agent', agentId)
331    return write($, items => edited(items, id, changes))
332  }
333  for (;;) {
334    const { value = [], version } = await $.state.get({ plugin: 'rabe', key: 'edits' })
335    const next = [...value, ...changes].slice(-MAX_EDITS)
336    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'edits' }, next, {
337      ifVersion: version,
338    })
339    if (isSet) return
340  }
341}
342
343type Listings = Map<string, Promise<Set<string> | undefined>>
344
345// A path whose stat failed is missing only when its folder lists without it,
346// or that folder is missing itself; error text is never read.
347function hasName(names: Set<string>, name: string): boolean {
348  if (names.has(name)) return true
349  const fold = (one: string) => one.normalize('NFC').toLowerCase()
350  const wanted = fold(name)
351  for (const one of names) if (fold(one) === wanted) return true
352
353  return false
354}
355
356async function isMissing($: EngineInterface, path: string, lists: Listings): Promise<boolean> {
357  const cut = path.lastIndexOf('/')
358  if (cut < 0 || path === '/') return false
359  const folder = path.slice(0, cut) || '/'
360  let names = lists.get(folder)
361  if (!names) {
362    names = $.fs.list(folder).then(
363      entries => new Set(entries.map(entry => entry.name)),
364      () => undefined,
365    )
366    lists.set(folder, names)
367  }
368  const listed = await names
369  // A name the folder lists in another case or Unicode form may be this file (macOS, Windows).
370  if (listed) return !hasName(listed, path.slice(cut + 1))
371
372  return folder !== '/' && isMissing($, folder, lists)
373}
374
375// What is on disk at each path, or undefined when the looks outlast LOOK_MS.
376async function look($: EngineInterface, paths: string[]): Promise<Seen[] | undefined> {
377  const stop = new AbortController()
378  const late = $.clock.sleep(LOOK_MS, { signal: stop.signal }).then(
379    () => undefined,
380    () => undefined,
381  )
382  const lists: Listings = new Map()
383  const seen = Promise.all(
384    paths.map(path =>
385      $.fs.stat(path).then(
386        (stat): Seen => stat,
387        async (): Promise<Seen> => ((await isMissing($, path, lists)) ? 'none' : undefined),
388      ),
389    ),
390  )
391  try {
392    return await Promise.race([seen, late])
393  } finally {
394    stop.abort()
395  }
396}
397
398type Looked = { paths: string[]; seen: Seen[] }
399
400async function beforeBash(
401  $: EngineInterface,
402  command: string,
403  agentId: string | undefined,
404): Promise<Looked | undefined> {
405  const home = await $.env.get('HOME')
406  let candidates = shellWrites(command, undefined, home)
407  if (candidates.some(one => !one.path.startsWith('/'))) {
408    let cwd: string | undefined
409    if (agentId) {
410      const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
411      cwd = asAgent(items, itemId('agent', agentId))?.detail.cwd
412    } else {
413      cwd = await $.session.cwd().catch(() => undefined)
414    }
415    candidates = shellWrites(command, cwd, home)
416  }
417  const paths = candidates.map(one => one.path).filter(path => path.startsWith('/'))
418  if (paths.length === 0 || paths.length > MAX_CHECKS) return undefined
419  const seen = await look($, paths)
420
421  return seen && { paths, seen }
422}
423
424async function afterBash($: EngineInterface, agentId: string | undefined, before: Looked) {
425  const after = await look($, before.paths)
426  if (!after) return
427  const at = await $.clock.now()
428  const changes = before.paths.flatMap((path, n): RabeEdit[] => {
429    const change = changed(before.seen[n], after[n])
430    if (!change) return []
431    return [{ path, at, via: 'shell', ...(change === 'delete' && { change }) }]
432  })
433  await record($, agentId, changes)
434}
435
436async function readMeta($: EngineInterface, id: string): Promise<void> {
437  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
438  const item = asAgent(items, id)
439  const path = item && metaPath(item, items)
440  if (!path) return
441  const text = await $.fs.read(path).catch(() => undefined)
442  const patch = typeof text === 'string' ? metaPatch(text) : undefined
443  if (patch) await write($, list => updateItem(list, id, { detail: patch }))
444}
445
446async function refresh($: EngineInterface): Promise<void> {
447  const agents = await $.agent.list().catch(() => [])
448  const now = await $.clock.now()
449  await write($, items => {
450    const end = pastEnd(items, now)
451
452    return agents.reduce((list, info) => listed(list, info, now, end), items)
453  })
454  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
455  for (const item of items) {
456    if (item.kind !== 'agent') continue
457    const workflow = items.find(parent => parent.id === item.parentId && parent.kind === 'workflow')
458    if (
459      item.status === 'running' ||
460      (workflow?.kind === 'workflow' &&
461        workflow.detail.phases?.length &&
462        item.detail.workflowPhase === undefined)
463    )
464      await readMeta($, item.id)
465  }
466}
467
468// The built-in table, or with the user's file (`pricesFile`) its rows first. A
469// file that is set but cannot be read gives no prices: costs show n/a.
470async function prices($: EngineInterface, file: string): Promise<Prices> {
471  const read = (path: string) => $.fs.read(path).then(text => parsePrices(String(text)))
472  const built = await read(`${$.plugin.root}/data/prices.csv`).catch(() => [])
473  if (!file) return built
474  const home = file.startsWith('~/') && (await $.env.get('HOME').catch(() => undefined))
475  const own = await read(home ? `${home}${file.slice(1)}` : file).catch(() => undefined)
476
477  return own ? withOverride(built, own) : []
478}
479
480async function recordStep(
481  $: EngineInterface,
482  agentId: string,
483  result: TurnStepResult,
484  file: string,
485) {
486  const id = itemId('agent', agentId)
487  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
488  if (!asAgent(items, id)) return
489  const now = await $.clock.now()
490  const tools = result.toolUses.map(use => ({ name: use.name, summary: toolSummary(use.input) }))
491  const usage = result.usage
492  const usd = usage ? claudeUsd(await prices($, file), usage, usage.model) : undefined
493  await write($, list => stepped(list, id, result, tools, now, usd))
494  if (!result.answer && tools.length === 0) return
495  await writeTurns($, turns => {
496    const index = (turns[id]?.at(-1)?.index ?? 0) + 1
497    const turn = { index, at: now, text: clip(result.answer, MAX_ANSWER), tools }
498
499    return addTurn(turns, id, turn)
500  })
501}
502
503export function agents(on: On, file: string): void {
504  on('session.start', { cwd: /^/ }, async ($, e, next) => {
505    $.clock.every(POLL_MS, () => refresh($))
506    await refresh($)
507
508    return next(e)
509  })
510
511  on('agent.spawn', { subagentType: /^/ }, async ($, e, next) => {
512    const result = await next(e)
513    const agentId = result.agentId
514    if (!agentId) return result
515    const now = await $.clock.now()
516    const runId = e.workflow?.runId
517    const parent = e.parentAgentId ? itemId('agent', e.parentAgentId) : undefined
518    const id = itemId('agent', agentId)
519    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
520    const transcriptPath = runId
521      ? workflowTranscript(asAgent(items, id)?.detail.transcriptPath, agentId, runId)
522      : undefined
523    await write($, items =>
524      addItem(
525        items,
526        {
527          id: itemId('agent', agentId),
528          kind: 'agent',
529          title: e.description || e.subagentType,
530          status: 'running',
531          startedAt: now,
532          parentId: runId ? itemId('workflow', runId) : parent,
533          detail: {
534            agentId,
535            ...(transcriptPath && { transcriptPath }),
536            type: e.subagentType,
537            model: result.model,
538            description: e.description,
539            prompt: clip(e.prompt, MAX_PROMPT),
540            workflowIndex: e.workflow?.agentIndex,
541          },
542        },
543        now,
544      ),
545    )
546
547    return result
548  }).catch((_$, e, next) => next(e))
549
550  on('classic.SubagentStart', { agent_id: /^/ }, async ($, e, next) => {
551    const transcriptPath = agentTranscript(e.transcript_path, e.agent_id)
552    const detail = {
553      agentId: e.agent_id,
554      ...(transcriptPath && { transcriptPath }),
555      ...(e.cwd && { cwd: e.cwd }),
556    }
557    const id = itemId('agent', e.agent_id)
558    const now = await $.clock.now()
559    await write($, items =>
560      items.some(item => item.id === id)
561        ? updateItem(items, id, { detail })
562        : addItem(
563            items,
564            { id, kind: 'agent', title: e.agent_type, status: 'running', startedAt: now, detail },
565            now,
566          ),
567    )
568
569    return next(e)
570  }).catch((_$, e, next) => next(e))
571
572  on('turn.step', { agentId: /^/ }, async function* ($, e, next) {
573    const result = yield* next(e)
574    if (e.agentId) await recordStep($, e.agentId, result, file)
575
576    return result
577  })
578
579  on('tool.call', { tool: ['Edit', 'Write'] }, async ($, e, next) => {
580    const answer = await next(e)
581    try {
582      const path = (e.tool === 'Edit' || e.tool === 'Write') && changedFile(answer, e.file_path)
583      if (path) {
584        const via = e.tool === 'Edit' ? 'edit' : 'write'
585        await record($, e.agentId, [{ path, at: await $.clock.now(), via }])
586      }
587    } catch {}
588
589    return answer
590  })
591
592  // The command line proposes files; a look on disk before and after decides.
593  // A call of the Codex companion is kept on its agent, so the Codex source can
594  // link the job it starts (see forwarders.ts).
595  on('tool.call', { tool: 'Bash' }, async ($, e, next) => {
596    const before =
597      e.tool === 'Bash'
598        ? await beforeBash($, e.command, e.agentId).catch(() => undefined)
599        : undefined
600    const call = e.tool === 'Bash' && e.agentId ? companionCall(e.command) : undefined
601    const id = itemId('agent', e.agentId ?? '')
602    const at = call && (await $.clock.now().catch(() => undefined))
603    if (call && at !== undefined) {
604      await write($, items => called(items, id, at, call)).catch(() => undefined)
605    }
606    const answer = await next(e)
607    try {
608      if (at !== undefined) {
609        const patch = answered(answer, await $.clock.now())
610        await write($, items => called(items, id, at, patch))
611      }
612    } catch {}
613    try {
614      if (before && ranBash(answer)) await afterBash($, e.agentId, before)
615    } catch {}
616
617    return answer
618  })
619
620  on('turn.complete', { agentId: /^/ }, async ($, e, next) => {
621    const result = await next(e)
622    const agentId = e.agentId
623    if (!agentId) return result
624    const now = await $.clock.now()
625    await write($, items => endItem(items, itemId('agent', agentId), ENDED[e.reason], now))
626    await readMeta($, itemId('agent', agentId))
627
628    return result
629  })
630}
631

1import type { EngineInterface, FsStat, On } from 'claude-code'
2
3import type { RabeCodexStep, RabeEdit, RabeTokens } from '../../types'
4import { linkForwarders } from '../forwarders'
5import {
6  clip,
7  type EndStatus,
8  itemId,
9  type NewItem,
10  type RabeItem,
11  type RabeItemOf,
12} from '../model'
13import { type CodexRequest, codexUsd, type Prices, parsePrices, withOverride } from '../prices'
14import { addItem, type Change, commit, endItem, prune, updateItem } from '../registry'
15import { shellWrites } from '../writes'
16
17const POLL_MS = 2000
18const CODEX_DATA = 'plugins/data/codex-openai-codex'
19const CODEX_PLUGIN = 'codex@openai-codex'
20const MAX_READ = 4 * 1024 * 1024
21const TAIL_LINES = 200
22const RUN_MS = 5000
23const MAX_STEPS = 50
24const MAX_TEXT = 300
25const MAX_EDITS = 100
26const CHANGES = ['add', 'update', 'delete'] as const
27const DAY = 24 * 60 * 60 * 1000
28const MAX_CHECKS = 20
29const LATE_MS = 2000
30const LOOK_MS = 1000
31
32type Rec = Record<string, unknown>
33
34export type CodexJob = Rec & { id: string }
35
36export type Rollout = {
37  model?: string
38  effort?: string
39  sandbox?: string
40  prompt?: string
41  tokens?: RabeTokens
42  // One per model request, from `last_token_usage`, with the model then in use.
43  requests: CodexRequest[]
44  commandCount: number
45  steps: RabeCodexStep[]
46  edits?: RabeEdit[]
47  checks?: Check[]
48}
49
50// A file a shell command may have written while it ran, from `from` to `to`.
51export type Check = { path: string; from: number; to: number }
52
53export type CodexSession = {
54  path?: string
55  updatedAt?: number
56  isMissing?: boolean
57  isPartial?: boolean
58  rollout?: Rollout
59  // Set with `rollout`: the job's dollars, undefined when not known.
60  usd?: number
61}
62
63function rec(value: unknown): Rec {
64  return value && typeof value === 'object' && !Array.isArray(value) ? (value as Rec) : {}
65}
66
67function list(value: unknown): unknown[] {
68  return Array.isArray(value) ? value : []
69}
70
71function str(value: unknown): string | undefined {
72  return typeof value === 'string' && value !== '' ? value : undefined
73}
74
75function num(value: unknown): number | undefined {
76  return typeof value === 'number' ? value : undefined
77}
78
79function parseJson(text: string): Rec | undefined {
80  try {
81    const value = JSON.parse(text)
82    return value && typeof value === 'object' ? (value as Rec) : undefined
83  } catch {
84    return undefined
85  }
86}
87
88function defined<T extends object>(value: T): T {
89  return Object.fromEntries(Object.entries(value).filter(([, one]) => one !== undefined)) as T
90}
91
92function texts(content: unknown, type: string): string {
93  return list(content)
94    .map(rec)
95    .filter(part => part.type === type)
96    .map(part => str(part.text) ?? '')
97    .join('')
98}
99
100function shellLine(command: unknown): string {
101  const argv = list(command).map(String)
102  return argv.length === 3 && argv[1] === '-lc' ? (argv[2] as string) : argv.join(' ')
103}
104
105function pendingCommand(input: unknown): string {
106  const text = String(input ?? '')
107  const quoted = /cmd:\s*("(?:[^"\\]|\\.)*")/.exec(text)?.[1]
108  if (quoted) {
109    try {
110      return JSON.parse(quoted)
111    } catch {}
112  }
113
114  return text.split('\n')[0] ?? ''
115}
116
117// Codex writes the cwd as a `file://` URL.
118function folder(cwd: unknown): string | undefined {
119  const text = str(cwd)
120  if (!text?.startsWith('file://')) return text
121  try {
122    return text.slice('file://'.length).split('/').map(decodeURIComponent).join('/')
123  } catch {
124    return undefined
125  }
126}
127
128// The absolute files a command line may write; a command with more is skipped.
129function commandChecks(item: Rec, from: number, to: number, home?: string): Check[] {
130  const paths = shellWrites(shellLine(item.command), folder(item.cwd), home)
131    .filter(one => !one.isDeleted && one.path.startsWith('/'))
132    .map(one => one.path)
133
134  return paths.length > MAX_CHECKS ? [] : paths.map(path => ({ path, from, to }))
135}
136
137function lineCount(output: unknown): number | undefined {
138  const text = str(output)
139
140  return text === undefined ? undefined : text.replace(/\n$/, '').split('\n').length
141}
142
143// The files a completed FileChange changed; a moved file leaves its old path.
144function fileChanges(changes: unknown, at: number): RabeEdit[] {
145  return Object.entries(rec(changes)).flatMap(([path, value]) => {
146    const change = CHANGES.find(one => one === rec(value).type)
147    const moved = str(rec(value).move_path)
148    if (!change) return []
149    const edit = (one: string, kind: RabeEdit['change']) =>
150      ({ path: one, at, via: 'codex', change: kind }) as RabeEdit
151
152    return moved ? [edit(path, 'delete'), edit(moved, 'add')] : [edit(path, change)]
153  })
154}
155
156// A count that is missing makes the request one without a model: no price.
157function request(value: unknown, model: string | undefined): CodexRequest {
158  const usage = rec(value)
159  const counts = [
160    usage.input_tokens,
161    usage.cached_input_tokens,
162    usage.cache_write_input_tokens,
163    usage.output_tokens,
164  ].map(num)
165  const [input = 0, cached = 0, write = 0, output = 0] = counts
166
167  return { ...(!counts.includes(undefined) && { model }), input, cached, write, output }
168}
169
170// `since` is the job's start: a job that resumed a thread finds the turns of
171// the jobs before it in the same file, and counts none of them. `until` is a
172// finished job's end: the turns of jobs that resumed the thread later count
173// nothing either.
174export function parseRollout(text: string, home?: string, since?: number, until?: number): Rollout {
175  const out: Rollout = { requests: [], commandCount: 0, steps: [] }
176  let total = ''
177  // the thread's total when the job started; undefined while none was seen
178  let base: Rec | undefined
179  let isResumed = false
180  const edits: RabeEdit[] = []
181  const checks: Check[] = []
182  const pending = new Map<string, RabeCodexStep>()
183  for (const raw of text.split('\n')) {
184    const record = parseJson(raw)
185    if (!record) continue
186    const payload = rec(record.payload)
187    if (until !== undefined && Date.parse(str(record.timestamp) ?? '') > until) continue
188    if (record.type === 'turn_context') {
189      const settings = rec(rec(payload.collaboration_mode).settings)
190      out.model = str(payload.model) ?? str(settings.model) ?? out.model
191      out.effort = str(payload.effort) ?? str(settings.reasoning_effort) ?? out.effort
192      out.sandbox = str(rec(payload.sandbox_policy).type) ?? out.sandbox
193    } else if (since !== undefined && Date.parse(str(record.timestamp) ?? '') < since) {
194      isResumed = true
195      const info = rec(payload.info)
196      if (payload.type === 'token_count' && Object.keys(info).length) {
197        base = rec(info.total_token_usage)
198        total = JSON.stringify(base)
199      }
200    } else if (record.type === 'event_msg' && payload.type === 'token_count') {
201      const info = rec(payload.info)
202      const usage = rec(info.total_token_usage)
203      // the total counts the whole thread: a resumed job's share needs the total it started at
204      const own = (key: string) => {
205        const now = num(usage[key])
206        if (!isResumed || now === undefined) return now
207        const before = num(base?.[key])
208        return before === undefined ? undefined : now - before
209      }
210      const input = own('input_tokens')
211      const output = own('output_tokens')
212      if (input !== undefined && output !== undefined) {
213        out.tokens = defined({ input, output, cached: own('cached_input_tokens') })
214      }
215      // a record without counts (rate limits only) or that repeats the total adds no request
216      const key = JSON.stringify(usage)
217      if (Object.keys(info).length) {
218        if (key !== total) out.requests.push(request(info.last_token_usage, out.model))
219        total = key
220      }
221    } else if (record.type === 'event_msg' && payload.type === 'item_completed') {
222      const item = rec(payload.item)
223      const at = num(payload.completed_at_ms) ?? (Date.parse(str(record.timestamp) ?? '') || 0)
224      if (item.type === 'UserMessage') out.prompt ??= str(texts(item.content, 'text'))
225      if (item.type === 'FileChange' && item.status === 'completed') {
226        edits.push(...fileChanges(item.changes, at))
227      }
228      const from = num(payload.started_at_ms)
229      if (item.type === 'CommandExecution' && item.exit_code === 0 && from !== undefined) {
230        checks.push(...commandChecks(item, from, at, home))
231      }
232      if (item.type === 'CommandExecution') {
233        out.commandCount += 1
234        out.steps.push(
235          defined({
236            kind: 'command' as const,
237            text: clip(shellLine(item.command), MAX_TEXT),
238            exitCode: num(item.exit_code),
239            lines: lineCount(item.aggregated_output),
240          }),
241        )
242      }
243    } else if (record.type === 'response_item') {
244      if (payload.type === 'message' && payload.role === 'assistant') {
245        const message = texts(payload.content, 'output_text')
246        if (message) out.steps.push({ kind: 'message', text: clip(message, MAX_TEXT) })
247      } else if (payload.type === 'reasoning') {
248        const summary = list(payload.summary)
249          .map(part => str(rec(part).text))
250          .filter(Boolean)
251          .join('\n')
252        if (summary) out.steps.push({ kind: 'reasoning', text: clip(summary, MAX_TEXT) })
253      } else if (payload.type === 'custom_tool_call') {
254        const step: RabeCodexStep = {
255          kind: 'command',
256          text: clip(pendingCommand(payload.input), MAX_TEXT),
257          isRunning: true,
258        }
259        pending.set(String(payload.call_id), step)
260        out.steps.push(step)
261      } else if (payload.type === 'custom_tool_call_output') {
262        const step = pending.get(String(payload.call_id))
263        out.steps = out.steps.filter(one => one !== step)
264      }
265    }
266  }
267  out.steps = out.steps.slice(-MAX_STEPS)
268  if (edits.length) out.edits = edits.slice(-MAX_EDITS)
269  if (checks.length) out.checks = checks.slice(-MAX_EDITS)
270
271  return defined(out)
272}
273
274export function jobEnd(status: unknown): EndStatus | undefined {
275  if (status === 'completed') return 'done'
276  if (status === 'failed') return 'failed'
277  if (status === 'cancelled') return 'stopped'
278
279  return undefined
280}
281
282function jobStart(job: CodexJob): number | undefined {
283  return Date.parse(str(job.startedAt) ?? str(job.createdAt) ?? '') || undefined
284}
285
286// a running job, or a finished one without its time, has no end to cut at
287function jobStop(job: CodexJob): number | undefined {
288  return (jobEnd(job.status) && Date.parse(str(job.completedAt) ?? '')) || undefined
289}
290
291export function codexItem(job: CodexJob, session: CodexSession | undefined): NewItem {
292  const rollout = session?.rollout
293  // an unchanged session file is not parsed again; the held item keeps what it gave
294  const request = session?.path && !rollout ? {} : rec(job.request)
295
296  const item = defined({
297    id: itemId('codex', job.id),
298    kind: 'codex' as const,
299    title: str(String(job.summary ?? '').split('\n')[0]) ?? str(job.title) ?? 'Codex job',
300    status: 'running' as const,
301    startedAt: jobStart(job),
302    tokens: rollout?.tokens,
303    detail: defined({
304      jobId: job.id,
305      jobKind: str(job.kindLabel),
306      threadId: str(job.threadId),
307      model: rollout?.model ?? str(request.model),
308      effort: rollout?.effort ?? str(request.effort),
309      sandbox: rollout?.sandbox,
310      prompt: str(request.prompt) ?? rollout?.prompt,
311      workspaceRoot: str(job.workspaceRoot),
312      logPath: str(job.logFile),
313      sessionPath: session?.path,
314      sessionUpdatedAt: session?.updatedAt,
315      isSessionMissing: session?.isMissing,
316      isSessionPartial: session?.isPartial,
317      commandCount: rollout?.commandCount,
318      steps: rollout?.steps,
319      edits: rollout?.edits,
320    }),
321  })
322
323  // only an unchanged file keeps the cost; a parse, a gone or unreadable file sets it, also to unknown
324  const isUnchanged = session?.updatedAt !== undefined && !rollout
325
326  return session && !isUnchanged ? { ...item, costUsd: session.usd } : item
327}
328
329async function write($: Pick<EngineInterface, 'state'>, change: Change): Promise<void> {
330  for (;;) {
331    const { value, version } = await $.state.get({ plugin: 'rabe', key: 'items' })
332    const next = commit(value, change)
333    if (next === undefined) return
334    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'items' }, next.items, {
335      ifVersion: version,
336    })
337    if (isSet) return forget($, next.dropped)
338  }
339}
340
341// Drops the lines and turns of the items a write dropped.
342async function forget($: Pick<EngineInterface, 'state'>, dropped: string[]): Promise<void> {
343  if (dropped.length === 0) return
344  for (;;) {
345    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
346    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
347    const next = prune(value, dropped, items)
348    if (next === value) break
349    const lines = await $.state.set({ plugin: 'rabe', key: 'lines' }, next, { ifVersion: version })
350    if (lines.isSet) break
351  }
352  for (;;) {
353    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'turns' })
354    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
355    const next = prune(value, dropped, items)
356    if (next === value) return
357    const turns = await $.state.set({ plugin: 'rabe', key: 'turns' }, next, { ifVersion: version })
358    if (turns.isSet) return
359  }
360}
361
362async function homes($: EngineInterface): Promise<{ home: string; claude: string; codex: string }> {
363  const home = (await $.env.get('HOME')) ?? ''
364
365  return {
366    home,
367    claude: (await $.env.get('CLAUDE_CONFIG_DIR')) ?? `${home}/.claude`,
368    codex: (await $.env.get('CODEX_HOME')) ?? `${home}/.codex`,
369  }
370}
371
372async function findRollout(
373  $: EngineInterface,
374  codexHome: string,
375  threadId: string,
376  startedAt: number,
377): Promise<string | undefined> {
378  for (const offset of [0, -DAY, DAY]) {
379    const day = new Date(startedAt + offset).toISOString().slice(0, 10).replaceAll('-', '/')
380    const dir = `${codexHome}/sessions/${day}`
381    const entries = await $.fs.list(dir).catch(() => [])
382    const found = entries.find(entry => entry.name.endsWith(`-${threadId}.jsonl`))
383    if (found) return `${dir}/${found.name}`
384  }
385
386  return undefined
387}
388
389// One poll's shell checks: each path is statted once, and all share one deadline.
390type Looks = {
391  asked: Map<string, Promise<void>>
392  known: Map<string, FsStat>
393  late?: Promise<void>
394  stop: AbortController
395}
396
397function statOnce($: EngineInterface, looks: Looks, path: string): Promise<void> {
398  let asked = looks.asked.get(path)
399  if (!asked) {
400    asked = $.fs.stat(path).then(
401      stat => {
402        looks.known.set(path, stat)
403      },
404      () => undefined,
405    )
406    looks.asked.set(path, asked)
407  }
408
409  return asked
410}
411
412// A shell write counts when its file was changed while the command ran; a
413// delete leaves nothing to look at, so only FileChange records give those.
414async function confirmed($: EngineInterface, rollout: Rollout, looks: Looks): Promise<Rollout> {
415  const { checks, ...rest } = rollout
416  if (!checks) return rollout
417  looks.late ??= $.clock.sleep(LOOK_MS, { signal: looks.stop.signal }).catch(() => undefined)
418  await Promise.race([Promise.all(checks.map(one => statOnce($, looks, one.path))), looks.late])
419  const written = checks.flatMap(({ path, from, to }): RabeEdit[] => {
420    const stat = looks.known.get(path)
421    const isWritten = stat?.kind === 'file' && stat.mtimeMs >= from && stat.mtimeMs <= to + LATE_MS
422
423    return isWritten ? [{ path, at: to, via: 'shell' }] : []
424  })
425  const edits = [...(rest.edits ?? []), ...written].sort((a, b) => a.at - b.at).slice(-MAX_EDITS)
426
427  return edits.length ? { ...rest, edits } : rest
428}
429
430// The built-in table, or with the user's file (`pricesFile`) its rows first. A
431// file that is set but cannot be read gives no prices: costs show n/a.
432async function prices($: EngineInterface, file: string): Promise<Prices> {
433  const read = (path: string) => $.fs.read(path).then(text => parsePrices(String(text)))
434  const built = await read(`${$.plugin.root}/data/prices.csv`).catch(() => [])
435  if (!file) return built
436  const home = file.startsWith('~/') && (await $.env.get('HOME').catch(() => undefined))
437  const own = await read(home ? `${home}${file.slice(1)}` : file).catch(() => undefined)
438
439  return own ? withOverride(built, own) : []
440}
441
442async function readSession(
443  $: EngineInterface,
444  codexHome: string,
445  job: CodexJob,
446  held: RabeItemOf<'codex'> | undefined,
447  looks: Looks,
448  file: string,
449): Promise<CodexSession | undefined> {
450  const threadId = str(job.threadId)
451  if (!threadId) return undefined
452  const path =
453    held?.detail.sessionPath ??
454    (await findRollout($, codexHome, threadId, jobStart(job) ?? (await $.clock.now())))
455  const stat = path ? await $.fs.stat(path).catch(() => undefined) : undefined
456  if (!path || !stat) {
457    if (jobEnd(job.status)) return { isMissing: true }
458
459    // a file seen before and gone now: no new fields, but the cost is no longer known
460    return held?.detail.sessionPath ? { path } : undefined
461  }
462  const updatedAt = stat.mtimeMs
463  if (updatedAt === held?.detail.sessionUpdatedAt) return { path, updatedAt }
464  // no updatedAt on failure, so the next poll tries again
465  const home = await $.env.get('HOME')
466  if (stat.size <= MAX_READ) {
467    const text = await $.fs.read(path).catch(() => undefined)
468    if (text === undefined) return { path }
469    const rollout = await confirmed(
470      $,
471      parseRollout(String(text), home, jobStart(job), jobStop(job)),
472      looks,
473    )
474
475    return { path, updatedAt, rollout, usd: codexUsd(await prices($, file), rollout.requests) }
476  }
477  const head = await $.process
478    .run(['grep', '-m', '2', '-E', '"type":"(turn_context|UserMessage)"', '--', path], {
479      timeoutMs: RUN_MS,
480    })
481    .catch(() => undefined)
482  const tail = await $.process
483    .run(['tail', '-n', String(TAIL_LINES), '--', path], { timeoutMs: RUN_MS })
484    .catch(() => undefined)
485  if (!head || !tail || tail.exitCode !== 0) return { path }
486
487  // the head and tail miss requests between them, so no usd
488  return {
489    path,
490    updatedAt,
491    isPartial: true,
492    rollout: await confirmed(
493      $,
494      parseRollout(`${head.stdout}\n${tail.stdout}`, home, jobStart(job), jobStop(job)),
495      looks,
496    ),
497  }
498}
499
500async function refresh(
501  $: EngineInterface,
502  sessionId: string,
503  codexHome: string,
504  jobPath: string,
505  entry: CodexJob,
506  items: RabeItem[],
507  looks: Looks,
508  file: string,
509): Promise<void> {
510  const saved = parseJson(String(await $.fs.read(jobPath).catch(() => '')))
511  const job: CodexJob = { ...entry, ...saved, id: entry.id }
512  if (job.sessionId !== sessionId) return
513  const end = jobEnd(job.status)
514  const now = await $.clock.now()
515  const endedAt = Date.parse(str(job.completedAt) ?? '') || now
516  const change = (item: NewItem) => (list: RabeItem[]) => {
517    const added = addItem(list, item, now)
518
519    return end ? endItem(added, item.id, end, endedAt) : added
520  }
521  const held = items.find((one): one is RabeItemOf<'codex'> => one.id === itemId('codex', job.id))
522  // a finished job the cap would drop again: its session file is not read
523  if (!held && commit(items, change(codexItem(job, undefined))) === undefined) return
524  const session = await readSession($, codexHome, job, held, looks, file)
525  await write($, change(codexItem(job, session)))
526}
527
528async function poll($: EngineInterface, file: string): Promise<void> {
529  const { claude, codex } = await homes($)
530  const stateRoot = `${claude}/${CODEX_DATA}/state`
531  const sessionId = await $.session.id()
532  const { startedAt: since } = await $.session.usage()
533  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
534  const looks: Looks = { asked: new Map(), known: new Map(), stop: new AbortController() }
535  // ponytail: one stat per workspace folder per poll; keep a folder list in $.state if there are hundreds
536  for (const dir of await $.fs.list(stateRoot).catch(() => [])) {
537    if (dir.kind !== 'dir') continue
538    const statePath = `${stateRoot}/${dir.name}/state.json`
539    const stat = await $.fs.stat(statePath).catch(() => undefined)
540    if (!stat || stat.mtimeMs < since) continue
541    const state = parseJson(String(await $.fs.read(statePath).catch(() => '')))
542    for (const entry of list(state?.jobs).map(rec)) {
543      const id = str(entry.id)
544      // the plugin's state.json can hold a bare status patch; its job file has the session
545      // the id names a file in jobs/, so it must stay a plain file name
546      if (!id || !/^\w[\w.-]*$/.test(id)) continue
547      if (entry.sessionId !== undefined && entry.sessionId !== sessionId) continue
548      const held = items.find((one): one is RabeItemOf<'codex'> => one.id === itemId('codex', id))
549      if (held && held.status !== 'running') continue
550      await refresh(
551        $,
552        sessionId,
553        codex,
554        `${stateRoot}/${dir.name}/jobs/${id}.json`,
555        { ...entry, id },
556        items,
557        looks,
558        file,
559      ).catch(() => undefined)
560    }
561  }
562  looks.stop.abort()
563  await write($, linkForwarders)
564}
565
566function firstLine(text: string): string {
567  return text.trim().split('\n')[0] ?? ''
568}
569
570async function cancel($: EngineInterface, id: string): Promise<{ text: string }> {
571  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
572  const item = items.find((one: RabeItem): one is RabeItemOf<'codex'> => one.id === id)
573  if (item?.status !== 'running') return { text: `No running Codex job ${id}.` }
574  const { claude } = await homes($)
575  const installed = parseJson(
576    String(await $.fs.read(`${claude}/plugins/installed_plugins.json`).catch(() => '')),
577  )
578  const root = str(rec(list(rec(installed?.plugins)[CODEX_PLUGIN])[0]).installPath)
579  if (!root) return { text: 'Stop refused: the Codex plugin is not installed.' }
580  const { workspaceRoot } = item.detail
581  const argv = [
582    'node',
583    `${root}/scripts/codex-companion.mjs`,
584    'cancel',
585    item.detail.jobId,
586    '--json',
587    ...(workspaceRoot ? ['--cwd', workspaceRoot] : []),
588  ]
589  const result = await $.process
590    .run(argv, { env: { CLAUDE_PLUGIN_DATA: `${claude}/${CODEX_DATA}` } })
591    .catch((error: unknown) => ({ exitCode: 1, stdout: '', stderr: String(error) }))
592  if (result.exitCode !== 0) {
593    return { text: `Stop refused: ${firstLine(result.stderr) || `exit ${result.exitCode}`}` }
594  }
595  const now = await $.clock.now()
596  // later polls skip the stopped job: its cost stays only while the session
597  // file is the one it was priced from
598  const { sessionPath, sessionUpdatedAt } = item.detail
599  const stat = sessionPath ? await $.fs.stat(sessionPath).catch(() => undefined) : undefined
600  const isPriced = stat !== undefined && stat.mtimeMs === sessionUpdatedAt
601  await write($, held => {
602    const ended = endItem(held, id, 'stopped', now)
603    return isPriced ? ended : updateItem(ended, id, { costUsd: undefined })
604  })
605
606  return { text: `Stopped codex ${item.detail.jobId}` }
607}
608
609export function codex(on: On, file: string): void {
610  on('session.start', { isInteractive: true }, async ($, e, next) => {
611    const started = await next(e)
612    await $.command.register({
613      name: 'rabe-stop',
614      description: 'Stop a background item that Rabe shows, by its id',
615      argumentHint: '<item id>',
616      immediate: true,
617    })
618    let isPolling = false
619    $.clock.every(POLL_MS, () => {
620      if (isPolling) return
621      isPolling = true
622      poll($, file)
623        .catch(() => undefined)
624        .finally(() => {
625          isPolling = false
626        })
627    })
628
629    return started
630  })
631
632  on('command.run', { command: 'rabe-stop', args: /^\s*codex:/ }, async ($, e) =>
633    cancel($, e.args.trim()),
634  )
635}
636

1import type { EngineInterface, On } from 'claude-code'
2
3import { type EndStatus, itemId, type RabeItem, type RabeItemOf } from '../model'
4import { addItem, type Change, commit, endItem, prune, updateItem } from '../registry'
5
6const LOOP_SENTINEL = '<<autonomous-loop-dynamic>>'
7const SENTINEL = /^<<[\w.-]+>>$/
8const WAKEUP = 'wakeup-'
9// A wakeup fires when the session is idle, up to 90 s early for :00 and :30.
10const FIRE_SLACK_MS = 90_000
11
12// Each event takes one hook without a matcher per plugin; these match every event.
13const ANY_START = { isInteractive: [true, false] }
14const ANY_STOP = { stop_hook_active: [true, false] }
15
16type Job = { id: string; schedule: string; prompt: string; humanSchedule?: string }
17
18async function write($: Pick<EngineInterface, 'state'>, change: Change): Promise<void> {
19  for (;;) {
20    const { value, version } = await $.state.get({ plugin: 'rabe', key: 'items' })
21    const next = commit(value, change)
22    if (next === undefined) return
23    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'items' }, next.items, {
24      ifVersion: version,
25    })
26    if (isSet) return forget($, next.dropped)
27  }
28}
29
30// Drops the lines and turns of the items a write dropped.
31async function forget($: Pick<EngineInterface, 'state'>, dropped: string[]): Promise<void> {
32  if (dropped.length === 0) return
33  for (;;) {
34    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
35    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
36    const next = prune(value, dropped, items)
37    if (next === value) break
38    const lines = await $.state.set({ plugin: 'rabe', key: 'lines' }, next, { ifVersion: version })
39    if (lines.isSet) break
40  }
41  for (;;) {
42    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'turns' })
43    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
44    const next = prune(value, dropped, items)
45    if (next === value) return
46    const turns = await $.state.set({ plugin: 'rabe', key: 'turns' }, next, { ifVersion: version })
47    if (turns.isSet) return
48  }
49}
50
51function runningCrons(items: RabeItem[]): RabeItemOf<'cron'>[] {
52  return items.flatMap(item => (item.kind === 'cron' && item.status === 'running' ? [item] : []))
53}
54
55function isWakeup(item: RabeItemOf<'cron'>): boolean {
56  return item.detail.jobId.startsWith(WAKEUP)
57}
58
59function endWakeups(items: RabeItem[], status: EndStatus, now: number, until = Infinity) {
60  return dueWakeups(items, until).reduce((next, item) => endItem(next, item.id, status, now), items)
61}
62
63function dueWakeups(items: RabeItem[], until: number) {
64  return runningCrons(items).filter(
65    item => isWakeup(item) && (item.detail.scheduledFor ?? 0) <= until,
66  )
67}
68
69// A due wakeup ends only on evidence that it fired: its own prompt, or the
70// sentinel it was given (`<<autonomous-loop-dynamic>>`), which its expanded
71// tick names. Other text, such as a recurring loop's tick, ends none.
72function fired(items: RabeItem[], text: string, now: number): RabeItem[] {
73  const prompt = text.trim()
74  const ends = dueWakeups(items, now + FIRE_SLACK_MS).filter(item => {
75    const own = item.detail.prompt.trim()
76
77    return own === prompt || (SENTINEL.test(own) && text.includes(own))
78  })
79
80  return ends.reduce((next, item) => endItem(next, item.id, 'done', now), items)
81}
82
83function sync(items: RabeItem[], jobs: Job[], now: number): RabeItem[] {
84  const listed = new Set(jobs.map(job => itemId('cron', job.id)))
85  const running = runningCrons(items)
86  const wakeupPrompts = new Set(running.filter(isWakeup).map(item => item.detail.prompt))
87  const ended = running
88    .filter(item => !isWakeup(item) && !listed.has(item.id))
89    .reduce((next, item) => endItem(next, item.id, 'done', now), items)
90
91  return jobs.reduce((next, job) => {
92    const id = itemId('cron', job.id)
93    if (wakeupPrompts.has(job.prompt) || next.some(item => item.id === id)) return next
94
95    return addItem(
96      next,
97      {
98        id,
99        kind: 'cron',
100        title: job.prompt,
101        status: 'running',
102        detail: {
103          jobId: job.id,
104          prompt: job.prompt,
105          schedule: job.schedule,
106          ...(job.humanSchedule && { humanSchedule: job.humanSchedule }),
107        },
108      },
109      now,
110    )
111  }, ended)
112}
113
114async function list($: EngineInterface): Promise<void> {
115  try {
116    const answer = await $.tool.call({ tool: 'CronList' })
117    if (!answer.result || answer.isError) return
118    const jobs = answer.result.jobs.map(job => ({
119      id: job.id,
120      schedule: job.cron,
121      prompt: job.prompt,
122      humanSchedule: job.humanSchedule,
123    }))
124    const now = await $.clock.now()
125    await write($, held => sync(held, jobs, now))
126  } catch {}
127}
128
129export function crons(on: On): void {
130  on('tool.call', { tool: 'CronCreate' }, async ($, e, next) => {
131    const answer = await next(e)
132    try {
133      if (e.tool === 'CronCreate' && answer.result && !answer.isError) {
134        const { id, humanSchedule } = answer.result as { id: string; humanSchedule?: string }
135        const now = await $.clock.now()
136        await write($, held =>
137          addItem(
138            held,
139            {
140              id: itemId('cron', id),
141              kind: 'cron',
142              title: e.prompt,
143              status: 'running',
144              startedAt: now,
145              ...(e.agentId && { parentId: itemId('agent', e.agentId) }),
146              detail: {
147                jobId: id,
148                prompt: e.prompt,
149                schedule: e.cron,
150                ...(humanSchedule && { humanSchedule }),
151              },
152            },
153            now,
154          ),
155        )
156      }
157    } catch {}
158
159    return answer
160  })
161
162  on('tool.call', { tool: 'CronDelete' }, async ($, e, next) => {
163    const answer = await next(e)
164    try {
165      if (e.tool === 'CronDelete' && answer.result && !answer.isError) {
166        const now = await $.clock.now()
167        const id = itemId('cron', e.id)
168        await write($, held =>
169          updateItem(endItem(held, id, 'stopped', now), id, { detail: { isDeleted: true } }),
170        )
171      }
172    } catch {}
173
174    return answer
175  })
176
177  on('tool.call', { tool: 'ScheduleWakeup' }, async ($, e, next) => {
178    const answer = await next(e)
179    try {
180      if (e.tool === 'ScheduleWakeup' && answer.result && !answer.isError) {
181        const { scheduledFor, stopped } = answer.result as {
182          scheduledFor: number
183          stopped?: boolean
184        }
185        const now = await $.clock.now()
186        const prompt = e.prompt ?? ''
187        await write($, held =>
188          stopped || e.stop
189            ? endWakeups(held, 'stopped', now)
190            : addItem(
191                endWakeups(held, 'done', now),
192                {
193                  id: itemId('cron', `${WAKEUP}${scheduledFor}`),
194                  kind: 'cron',
195                  title: prompt === LOOP_SENTINEL ? 'autonomous loop' : prompt,
196                  status: 'running',
197                  startedAt: now,
198                  ...(e.agentId && { parentId: itemId('agent', e.agentId) }),
199                  detail: { jobId: `${WAKEUP}${scheduledFor}`, prompt, scheduledFor },
200                },
201                now,
202              ),
203        )
204      }
205    } catch {}
206
207    return answer
208  })
209
210  on('prompt.submit', { origin: { kind: 'scheduled-trigger' } }, async ($, e, next) => {
211    try {
212      const now = await $.clock.now()
213      await write($, held => fired(held, e.text, now))
214    } catch {}
215
216    return next(e)
217  })
218
219  on('classic.Stop', ANY_STOP, async ($, e, next) => {
220    try {
221      if (e.session_crons) {
222        const jobs = e.session_crons.map(job => ({
223          id: job.id,
224          schedule: job.schedule,
225          prompt: job.prompt,
226        }))
227        const now = await $.clock.now()
228        await write($, held => sync(held, jobs, now))
229      }
230    } catch {}
231
232    return next(e)
233  })
234
235  on('session.start', ANY_START, async ($, e, next) => {
236    await list($)
237
238    return next(e)
239  })
240}
241

1import type { EngineInterface, On } from 'claude-code'
2
3import { type EndStatus, itemId, type RabeItem } from '../model'
4import { addItem, type Change, commit, endItem, prune, updateItem } from '../registry'
5import { appendLines, endStatus, parseNotifications, parseOutput, taskOutput } from '../tasks'
6
7const POLL_MS = 2000
8const MAX_READ = 4 * 1024 * 1024
9
10// Each event takes one hook without a matcher per plugin; these match every event.
11const ANY_START = { isInteractive: [true, false] }
12const ANY_STOP = { stop_hook_active: [true, false] }
13
14type BackgroundTask = { id: string; type: string; description: string; command?: string }
15
16async function write($: Pick<EngineInterface, 'state'>, change: Change): Promise<void> {
17  for (;;) {
18    const { value, version } = await $.state.get({ plugin: 'rabe', key: 'items' })
19    const next = commit(value, change)
20    if (next === undefined) return
21    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'items' }, next.items, {
22      ifVersion: version,
23    })
24    if (isSet) return forget($, next.dropped)
25  }
26}
27
28// Drops the lines and turns of the items a write dropped.
29async function forget($: Pick<EngineInterface, 'state'>, dropped: string[]): Promise<void> {
30  if (dropped.length === 0) return
31  for (;;) {
32    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
33    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
34    const next = prune(value, dropped, items)
35    if (next === value) break
36    const lines = await $.state.set({ plugin: 'rabe', key: 'lines' }, next, { ifVersion: version })
37    if (lines.isSet) break
38  }
39  for (;;) {
40    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'turns' })
41    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
42    const next = prune(value, dropped, items)
43    if (next === value) return
44    const turns = await $.state.set({ plugin: 'rabe', key: 'turns' }, next, { ifVersion: version })
45    if (turns.isSet) return
46  }
47}
48
49// ponytail: reads files up to 4 MiB whole; a larger monitor output is not followed
50async function follow(
51  $: EngineInterface,
52  id: string,
53  path: string,
54): Promise<EndStatus | undefined> {
55  try {
56    const { size } = await $.fs.stat(path)
57    if (size > MAX_READ) return undefined
58    const { lines, ended } = parseOutput(String(await $.fs.read(path)))
59    const now = await $.clock.now()
60    for (;;) {
61      const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
62      const next = appendLines(value[id], lines, now)
63      if (!next) break
64      const { isSet } = await $.state.set(
65        { plugin: 'rabe', key: 'lines' },
66        { ...value, [id]: next },
67        { ifVersion: version },
68      )
69      if (isSet) break
70    }
71
72    return ended
73  } catch {
74    return undefined
75  }
76}
77
78function runningMonitors(items: RabeItem[]) {
79  return items.flatMap(item => (item.kind === 'monitor' && item.status === 'running' ? [item] : []))
80}
81
82async function poll($: EngineInterface): Promise<void> {
83  try {
84    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
85    for (const item of runningMonitors(items)) {
86      const path =
87        item.detail.outputPath ?? (item.detail.taskId && taskOutput(items, item.detail.taskId))
88      if (!path) continue
89      const ended = await follow($, item.id, path)
90      const now = await $.clock.now()
91      await write($, held => {
92        const next = updateItem(held, item.id, { detail: { outputPath: path } })
93
94        return ended ? endItem(next, item.id, ended, now) : next
95      })
96    }
97  } catch {}
98}
99
100async function onNotifications($: EngineInterface, text: string): Promise<void> {
101  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
102  for (const one of parseNotifications(text)) {
103    const id = itemId('monitor', one.taskId)
104    if (!one.status || !items.some(item => item.id === id)) continue
105    if (one.outputFile) await follow($, id, one.outputFile)
106    const now = await $.clock.now()
107    await write($, held =>
108      endItem(
109        updateItem(held, id, { detail: one.outputFile ? { outputPath: one.outputFile } : {} }),
110        id,
111        endStatus(one.status),
112        now,
113      ),
114    )
115  }
116}
117
118async function correct($: EngineInterface, tasks: readonly BackgroundTask[]): Promise<void> {
119  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
120  const inFlight = new Set(tasks.map(task => task.id))
121  const now = await $.clock.now()
122  for (const item of runningMonitors(items)) {
123    if (!item.detail.taskId || inFlight.has(item.detail.taskId)) continue
124    // A subagent's task may be missing from the main session's list; the poll
125    // ends it from the exit line of its output file.
126    if (item.parentId?.startsWith('agent:')) continue
127    const ended = item.detail.outputPath && (await follow($, item.id, item.detail.outputPath))
128    await write($, held => endItem(held, item.id, ended || 'stopped', now))
129  }
130  await write($, held =>
131    tasks.reduce((next, task) => {
132      const id = itemId('monitor', task.id)
133      if (task.type !== 'monitor' || next.some(item => item.id === id)) return next
134      const outputPath = taskOutput(next, task.id)
135
136      return addItem(
137        next,
138        {
139          id,
140          kind: 'monitor',
141          title: task.description,
142          status: 'running',
143          detail: {
144            command: task.command ?? task.description,
145            description: task.description,
146            taskId: task.id,
147            ...(outputPath && { outputPath }),
148          },
149        },
150        now,
151      )
152    }, held),
153  )
154}
155
156export function monitors(on: On): void {
157  on('tool.call', { tool: 'Monitor' }, async ($, e, next) => {
158    const answer = await next(e)
159    try {
160      const result = answer.result as
161        | { taskId?: string; timeoutMs?: number; persistent?: boolean }
162        | undefined
163      const taskId = result?.taskId
164      if (e.tool === 'Monitor' && taskId) {
165        const now = await $.clock.now()
166        await write($, held => {
167          const outputPath = taskOutput(held, taskId)
168
169          return addItem(
170            held,
171            {
172              id: itemId('monitor', taskId),
173              kind: 'monitor',
174              title: e.description,
175              status: 'running',
176              startedAt: now,
177              ...(e.agentId && { parentId: itemId('agent', e.agentId) }),
178              detail: {
179                command: e.command ?? e.ws?.url ?? '',
180                description: e.description,
181                taskId,
182                timeoutMs: result.timeoutMs ?? e.timeout_ms,
183                isPersistent: result.persistent === true,
184                ...(outputPath && { outputPath }),
185              },
186            },
187            now,
188          )
189        })
190      }
191    } catch {}
192
193    return answer
194  })
195
196  on('prompt.submit', { origin: { kind: 'task-notification' } }, async ($, e, next) => {
197    try {
198      await onNotifications($, e.text)
199    } catch {}
200
201    return next(e)
202  })
203
204  on('classic.Stop', ANY_STOP, async ($, e, next) => {
205    try {
206      if (e.background_tasks) await correct($, e.background_tasks)
207    } catch {}
208
209    return next(e)
210  })
211
212  on('session.start', ANY_START, async ($, e, next) => {
213    $.clock.every(POLL_MS, () => void poll($))
214
215    return next(e)
216  })
217}
218

1import type { EngineInterface, On } from 'claude-code'
2
3import { itemId, type RabeItem } from '../model'
4import { addItem, type Change, commit, endItem, prune, updateItem } from '../registry'
5import {
6  appendLines,
7  endStatus,
8  guessPort,
9  outputPathOf,
10  parseNotifications,
11  parseOutput,
12  taskOutput,
13} from '../tasks'
14
15const POLL_MS = 2000
16const MAX_READ = 4 * 1024 * 1024
17const TAIL_BYTES = '1048576'
18const RUN_MS = 5000
19
20// Each event takes one hook without a matcher per plugin; these match every event.
21const ANY_START = { isInteractive: [true, false] }
22const ANY_STOP = { stop_hook_active: [true, false] }
23
24type BackgroundTask = { id: string; type: string; description?: string; command?: string }
25
26const shellTitle = (description: string | undefined, command: string, id: string) =>
27  `${(description || command).replace(/\s+/g, ' ').trim()} · ${id}`
28
29async function write($: Pick<EngineInterface, 'state'>, change: Change): Promise<void> {
30  for (;;) {
31    const { value, version } = await $.state.get({ plugin: 'rabe', key: 'items' })
32    const next = commit(value, change)
33    if (next === undefined) return
34    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'items' }, next.items, {
35      ifVersion: version,
36    })
37    if (isSet) return forget($, next.dropped)
38  }
39}
40
41// Drops the lines and turns of the items a write dropped.
42async function forget($: Pick<EngineInterface, 'state'>, dropped: string[]): Promise<void> {
43  if (dropped.length === 0) return
44  for (;;) {
45    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
46    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
47    const next = prune(value, dropped, items)
48    if (next === value) break
49    const lines = await $.state.set({ plugin: 'rabe', key: 'lines' }, next, { ifVersion: version })
50    if (lines.isSet) break
51  }
52  for (;;) {
53    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'turns' })
54    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
55    const next = prune(value, dropped, items)
56    if (next === value) return
57    const turns = await $.state.set({ plugin: 'rabe', key: 'turns' }, next, { ifVersion: version })
58    if (turns.isSet) return
59  }
60}
61
62async function readOutput($: EngineInterface, path: string): Promise<string | undefined> {
63  return (await readWhole($, path)).text
64}
65
66// `isWhole` is false when only the file's tail was read.
67async function readWhole(
68  $: EngineInterface,
69  path: string,
70): Promise<{ text?: string; isWhole: boolean }> {
71  try {
72    const { size } = await $.fs.stat(path)
73    if (size <= MAX_READ) return { text: String(await $.fs.read(path)), isWhole: true }
74    const { stdout } = await $.process.run(['tail', '-c', TAIL_BYTES, '--', path], {
75      timeoutMs: RUN_MS,
76    })
77
78    return { text: stdout, isWhole: false }
79  } catch {
80    return { isWhole: false }
81  }
82}
83
84// ponytail: lines are counted from the file's start, so a shell whose output
85// passed 4 MiB keeps the lines it had; the pane shows those.
86async function keepLines($: EngineInterface, id: string, lines: string[]): Promise<void> {
87  const now = await $.clock.now()
88  for (;;) {
89    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
90    const next = appendLines(value[id], lines, now)
91    if (!next) return
92    const { isSet } = await $.state.set(
93      { plugin: 'rabe', key: 'lines' },
94      { ...value, [id]: next },
95      { ifVersion: version },
96    )
97    if (isSet) return
98  }
99}
100
101function runningShells(items: RabeItem[]) {
102  return items.flatMap(item => (item.kind === 'shell' && item.status === 'running' ? [item] : []))
103}
104
105async function poll($: EngineInterface): Promise<void> {
106  try {
107    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
108    for (const item of runningShells(items)) {
109      const { text, isWhole } = item.detail.outputPath
110        ? await readWhole($, item.detail.outputPath)
111        : { isWhole: false }
112      if (!text) continue
113      const { exitCode, ended, lines } = parseOutput(text)
114      if (isWhole) await keepLines($, item.id, lines)
115      const port = item.detail.port ?? guessPort(text)
116      const now = await $.clock.now()
117      const portAt = item.detail.portAt ?? (port ? now : undefined)
118      await write($, held => {
119        const next = updateItem(held, item.id, {
120          detail: {
121            ...(port && { port, portAt }),
122            ...(exitCode !== undefined && { exitCode }),
123          },
124        })
125
126        return ended ? endItem(next, item.id, ended, now) : next
127      })
128    }
129  } catch {}
130}
131
132async function onNotifications($: EngineInterface, text: string): Promise<void> {
133  const notes = parseNotifications(text).filter(one => one.status)
134  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
135  for (const one of notes) {
136    const id = itemId('shell', one.taskId)
137    const item = items.find(held => held.id === id)
138    const path = one.outputFile ?? (item?.kind === 'shell' ? item.detail.outputPath : undefined)
139    const read = item && path ? await readWhole($, path) : undefined
140    if (read?.text && read.isWhole) await keepLines($, id, parseOutput(read.text).lines)
141  }
142  const now = await $.clock.now()
143  await write($, held =>
144    notes.reduce((items, one) => {
145      const id = itemId('shell', one.taskId)
146      const next = updateItem(items, id, {
147        detail: {
148          ...(one.exitCode !== undefined && { exitCode: one.exitCode }),
149          ...(one.outputFile && { outputPath: one.outputFile }),
150        },
151      })
152
153      return endItem(next, id, endStatus(one.status, one.exitCode), now)
154    }, held),
155  )
156}
157
158async function correct($: EngineInterface, tasks: readonly BackgroundTask[]): Promise<void> {
159  const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
160  const inFlight = new Set(tasks.map(task => task.id))
161  const now = await $.clock.now()
162  for (const item of runningShells(items)) {
163    if (!item.detail.taskId || inFlight.has(item.detail.taskId)) continue
164    // A subagent's task may be missing from the main session's list; the poll
165    // ends it from the exit line of its output file.
166    if (item.parentId?.startsWith('agent:')) continue
167    const text = item.detail.outputPath && (await readOutput($, item.detail.outputPath))
168    const { exitCode, ended = 'stopped' } = parseOutput(text || '')
169    await write($, held =>
170      endItem(
171        updateItem(held, item.id, { detail: exitCode === undefined ? {} : { exitCode } }),
172        item.id,
173        ended,
174        now,
175      ),
176    )
177  }
178  await write($, held =>
179    tasks.reduce((next, task) => {
180      const id = itemId('shell', task.id)
181      // Claude Code lists a Monitor tool task as type 'shell' too.
182      const known = [id, itemId('monitor', task.id)]
183      if (task.type !== 'shell' || !task.command || next.some(item => known.includes(item.id))) {
184        return next
185      }
186      const outputPath = taskOutput(items, task.id)
187
188      return addItem(
189        next,
190        {
191          id,
192          kind: 'shell',
193          title: shellTitle(task.description, task.command, task.id),
194          status: 'running',
195          detail: { command: task.command, taskId: task.id, ...(outputPath && { outputPath }) },
196        },
197        now,
198      )
199    }, held),
200  )
201}
202
203export function shells(on: On): void {
204  on('tool.call', { tool: 'Bash' }, async ($, e, next) => {
205    const answer = await next(e)
206    try {
207      const result = answer.result as { backgroundTaskId?: string } | undefined
208      const taskId = result?.backgroundTaskId
209      if (e.tool === 'Bash' && taskId) {
210        const now = await $.clock.now()
211        const outputPath = outputPathOf(answer.text)
212        await write($, held =>
213          addItem(
214            held,
215            {
216              id: itemId('shell', taskId),
217              kind: 'shell',
218              title: shellTitle(e.description, e.command, taskId),
219              status: 'running',
220              startedAt: now,
221              ...(e.agentId && { parentId: itemId('agent', e.agentId) }),
222              detail: { command: e.command, taskId, ...(outputPath && { outputPath }) },
223            },
224            now,
225          ),
226        )
227      }
228    } catch {}
229
230    return answer
231  })
232
233  on('prompt.submit', { origin: { kind: 'task-notification' } }, async ($, e, next) => {
234    try {
235      await onNotifications($, e.text)
236    } catch {}
237
238    return next(e)
239  })
240
241  on('classic.Stop', ANY_STOP, async ($, e, next) => {
242    try {
243      if (e.background_tasks) await correct($, e.background_tasks)
244    } catch {}
245
246    return next(e)
247  })
248
249  on('session.start', ANY_START, async ($, e, next) => {
250    $.clock.every(POLL_MS, () => void poll($))
251
252    return next(e)
253  })
254}
255

1import type { BuiltinToolResults, EngineInterface, On } from 'claude-code'
2
3import { itemId } from '../model'
4import { addItem, type Change, commit, endItem, prune, updateItem } from '../registry'
5import { endStatus, parseNotifications } from '../tasks'
6
7type Launched = BuiltinToolResults['Workflow']
8
9export function phaseNames(script: string): string[] | undefined {
10  // ponytail: a `}` inside an object entry's string still ends that entry.
11  const list =
12    /\bphases\s*:\s*\[\s*((?:(?:'[^']*'|"[^"]*"|`[^`]*`|\{[^}]*\})\s*(?:,\s*)?)*)\]/.exec(
13      script,
14    )?.[1]
15  if (list === undefined) return undefined
16
17  return [...list.matchAll(/\{[^}]*\}|(['"`])(.*?)\1/g)].flatMap(([whole, quote, text]) => {
18    if (quote) return [text ?? '']
19    const title = /\btitle\s*:\s*(['"`])(.*?)\1/.exec(whole)?.[2]
20    return title === undefined ? [] : [title]
21  })
22}
23
24async function write($: Pick<EngineInterface, 'state'>, change: Change): Promise<void> {
25  for (;;) {
26    const { value, version } = await $.state.get({ plugin: 'rabe', key: 'items' })
27    const next = commit(value, change)
28    if (next === undefined) return
29    const { isSet } = await $.state.set({ plugin: 'rabe', key: 'items' }, next.items, {
30      ifVersion: version,
31    })
32    if (isSet) return forget($, next.dropped)
33  }
34}
35
36// Drops the lines and turns of the items a write dropped.
37async function forget($: Pick<EngineInterface, 'state'>, dropped: string[]): Promise<void> {
38  if (dropped.length === 0) return
39  for (;;) {
40    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'lines' })
41    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
42    const next = prune(value, dropped, items)
43    if (next === value) break
44    const lines = await $.state.set({ plugin: 'rabe', key: 'lines' }, next, { ifVersion: version })
45    if (lines.isSet) break
46  }
47  for (;;) {
48    const { value = {}, version } = await $.state.get({ plugin: 'rabe', key: 'turns' })
49    const { value: items = [] } = await $.state.get({ plugin: 'rabe', key: 'items' })
50    const next = prune(value, dropped, items)
51    if (next === value) return
52    const turns = await $.state.set({ plugin: 'rabe', key: 'turns' }, next, { ifVersion: version })
53    if (turns.isSet) return
54  }
55}
56
57async function launched($: EngineInterface, result: Launched, name?: string): Promise<void> {
58  const runId = result.runId
59  if (!runId) return
60  const id = itemId('workflow', runId)
61  const now = await $.clock.now()
62  await write($, items =>
63    addItem(
64      items,
65      {
66        id,
67        kind: 'workflow',
68        title: result.workflowName ?? name ?? 'workflow',
69        status: 'running',
70        startedAt: now,
71        endedAt: undefined,
72        detail: {
73          runId,
74          taskId: result.taskId,
75          scriptPath: result.scriptPath,
76          transcriptDir: result.transcriptDir,
77        },
78      },
79      now,
80    ),
81  )
82  if (!result.scriptPath) return
83  const script = await $.fs.read(result.scriptPath).catch(() => undefined)
84  const phases = typeof script === 'string' ? phaseNames(script) : undefined
85  if (phases) await write($, items => updateItem(items, id, { detail: { phases } }))
86}
87
88async function notified($: EngineInterface, text: string): Promise<void> {
89  const notes = parseNotifications(text).filter(note => note.status)
90  if (notes.length === 0) return
91  const now = await $.clock.now()
92  await write($, items =>
93    notes.reduce((list, note) => {
94      const run = list.find(item => item.kind === 'workflow' && item.detail.taskId === note.taskId)
95      return run ? endItem(list, run.id, endStatus(note.status), now) : list
96    }, items),
97  )
98}
99
100export function workflows(on: On): void {
101  on('tool.call', { tool: 'Workflow' }, async ($, e, next) => {
102    const answer = await next(e)
103    if (answer.result && !answer.isError) await launched($, answer.result as Launched, e.name)
104
105    return answer
106  }).catch((_$, e, next) => next(e))
107
108  on('prompt.submit', { origin: { kind: 'task-notification' } }, async ($, e, next) => {
109    await notified($, e.text)
110
111    return next(e)
112  }).catch((_$, e, next) => next(e))
113}
114