tower-crane

1// Tower Crane for Claude Code: pushes orchestrator wake-ups into the session.
2// One child, tower-crane wait --follow, prints a line of ids per event; this
3// module turns those lines into prompts, so the session never polls. A push
4// names events only: the session reads current findings with tower-crane inbox.
5
6const TOOL = 'mcp__tower-crane__watch'
7const COMMAND = 'tower-crane-watch'
8
9// The same text lib/events.js notice() gives spawned harness homes.
10function notice(rows, agent) {
11  const lines = rows.map(w => `- ${w.id}: ${w.type}${w.decision ? ` ${w.decision}` : ''}${w.task ? ` ${w.task}` : ''} from ${w.agent}`)
12  const last = rows.at(-1)
13  return [
14    `Tower Crane: ${rows.length} new event${rows.length === 1 ? '' : 's'} for ${agent}.`,
15    ...lines,
16    `Run \`tower-crane inbox\` for findings and resolving commands.${last?.offset ? ` Cursor: ${last.offset}.` : ''}`,
17  ].join('\n')
18}
19
20// One module instance per session load: what it watches and what waits.
21const live = { watch: null, pending: [], steered: [], submitting: false, turn: null }
22
23// A prompt queues until the session is idle and resolves as its turn
24// starts, so what arrives meanwhile goes out together in the next one.
25async function flush($) {
26  if (live.submitting) return
27  live.submitting = true
28  try {
29    while (live.pending.length && live.watch) {
30      const batch = live.pending.splice(0)
31      await $.prompt.submit({ text: notice(batch, live.watch.agent) })
32    }
33  } finally {
34    live.submitting = false
35  }
36}
37
38// A steer joins the running turn at its next model request. A turn that
39// makes no further request never read it, so turn.complete queues it again.
40async function deliver($, w) {
41  if (w.steer && live.turn) {
42    const appended = await $.session.append({ message: { type: 'user', content: [{ type: 'text', text: notice([w], live.watch.agent) }] } })
43    if (!appended?.deny) {
44      live.steered.push({ w, step: live.turn.steps })
45      return
46    }
47  }
48  live.pending.push(w)
49  void flush($)
50}
51
52async function follow($, current, child) {
53  let buffer = ''
54  let stderr = ''
55  try {
56    for await (const { stream, text } of child) {
57      if (live.watch !== current) break
58      if (stream === 'stderr') {
59        stderr = (stderr + text).slice(-2000)
60        continue
61      }
62      buffer += text
63      let end
64      while ((end = buffer.indexOf('\n')) !== -1) {
65        const line = buffer.slice(0, end)
66        buffer = buffer.slice(end + 1)
67        let w
68        try {
69          w = JSON.parse(line)
70        } catch {
71          continue
72        }
73        if (typeof w.offset === 'number') current.cursor = String(w.offset)
74        if (w.type !== 'ready' && w.id) await deliver($, w)
75      }
76    }
77  } catch (e) {
78    stderr ||= String(e?.message || e)
79  }
80  if (live.watch !== current) return
81  live.watch = null
82  const why = stderr.trim().split('\n').at(-1)
83  $.ui.toast(`Tower Crane stopped watching${why ? `: ${why}` : ''}. Run /${COMMAND} to resume.`)
84}
85
86async function arm($, input = {}) {
87  const agent = input.agent || (await $.env.get('TOWER_CRANE_AGENT')) || 'orchestrator'
88  const state = input.state || (await $.env.get('TOWER_CRANE_STATE')) || ''
89  const after = input.after !== undefined && input.after !== '' ? String(input.after) : live.watch?.cursor || 'now'
90  const argv = ['node', `${$.plugin.root}/bin/tower-crane.js`, 'wait', '--follow', '--inbox', '--after', after, '--agent', agent, ...(state ? ['--state', state] : [])]
91  const current = { agent, cursor: after }
92  live.watch = current
93  void follow($, current, $.process.spawn({ argv }))
94  $.ui.status(`tower-crane: watching for ${agent}`)
95  return `Tower Crane pushes events for ${agent} after ${after} into this session as they arrive. Run \`tower-crane inbox\` on each wake for findings and resolving commands.`
96}
97
98export function register(on) {
99  on('session.start', async ($, e, next) => {
100    const started = await next(e)
101    await $.tool.register({
102      name: 'watch',
103      description: 'Tower Crane orchestrator: push every orchestrator event (submits, reviews, worker exits and stalls, worker messages, decisions, owner comments, CI, conflicts) into this session as it arrives. Call once after the startup cursor; it replaces background tower-crane wait.',
104      inputSchema: {
105        type: 'object',
106        properties: {
107          after: { type: 'string', description: 'cursor from tower-crane wait --timeout 0, or an event id (default: now, or where the last watch stopped)' },
108          agent: { type: 'string', description: 'identity to wake (default TOWER_CRANE_AGENT, else orchestrator)' },
109          state: { type: 'string', description: 'state directory (default TOWER_CRANE_STATE or the one found from the working directory)' },
110        },
111      },
112    })
113    await $.command.register({ name: COMMAND, description: 'Push Tower Crane orchestrator events into this session', argumentHint: '[cursor]' })
114    if ((await $.env.get('TOWER_CRANE_AGENT')) === 'orchestrator') await arm($)
115    return started
116  })
117
118  on('tool.call', { tool: TOOL }, async ($, e) => ({ result: await arm($, e) }))
119    .catch(() => ({ deny: 'Tower Crane could not start its event watch; run tower-crane wait in the background instead.' }))
120
121  on('command.run', { command: COMMAND }, async ($, e) => ({ text: await arm($, { after: e.args.trim() }) }))
122    .catch(() => ({ text: 'Tower Crane could not start its event watch; run tower-crane wait in the background instead.' }))
123
124  on('turn.start', ($, e, next) => {
125    live.turn = { id: e.turnId, steps: 0 }
126    return next(e)
127  })
128
129  on('turn.step', async function* ($, e, next) {
130    if (live.turn && e.turnId === live.turn.id) live.turn.steps += 1
131    return yield* next(e)
132  })
133
134  on('turn.complete', async ($, e, next) => {
135    const done = await next(e)
136    if (live.turn && e.turnId === live.turn.id) {
137      const unread = live.steered.filter(s => s.step >= live.turn.steps).map(s => s.w)
138      live.steered = []
139      live.turn = null
140      if (unread.length && live.watch) {
141        live.pending.unshift(...unread)
142        void flush($)
143      }
144    }
145    return done
146  })
147
148  on('session.end', ($, e, next) => {
149    live.watch = null
150    return next(e)
151  })
152}
153