wake

1import { atom, read, update } from 'claude-code'
2import type { EngineInterface, Register, Timer } from 'claude-code'
3
4import type { WakeAlert, WakeEvent, WakeEventKind, WakeUp, WakeWatch } from '../types'
5
6const watches = atom({ plugin: 'wake', key: 'watches' } as const, [] as WakeWatch[])
7const after = atom({ plugin: 'wake', key: 'after' } as const, 0)
8const handled = atom({ plugin: 'wake', key: 'handled' } as const, [] as number[])
9const held = atom({ plugin: 'wake', key: 'held' } as const, [] as WakeUp[])
10const alerts = atom({ plugin: 'wake', key: 'alerts' } as const, [] as WakeAlert[])
11const busy = atom({ plugin: 'wake', key: 'busy' } as const, false)
12const isDown = atom({ plugin: 'wake', key: 'isDown' } as const, false)
13const hinted = atom({ plugin: 'wake', key: 'hinted' } as const, false)
14
15/** The daemon's own long-poll slice: it answers within it, events or not. */
16const SLICE = '25s'
17/** A slice the daemon never answered is abandoned after this. */
18const GUARD_MS = 40_000
19/** Wait before reaching for a daemon that did not answer. */
20const RETRY_MS = 30_000
21const MAX_ALERTS = 5
22const ENDING: ReadonlySet<WakeEventKind> = new Set(['met', 'error', 'expired'])
23const KINDS: ReadonlySet<string> = new Set(['change', 'met', 'error', 'expired'])
24const PR = /^pr:(.+)#(\d+)$/
25
26const DESCRIPTION = [
27  'Wake this session when a PR, CI run, Eve review, devbox, vitrinka task or deployik deploy reaches a state,',
28  'instead of a sleep/until loop or a Monitor armed only to wait for it. Subscribe, then END YOUR TURN:',
29  'one shared poller (vybava watch) watches it, and a prompt from the wake plugin starts your next turn when',
30  'the condition holds, or the watch errors or expires.',
31  'Targets: pr:<n> (this repo), pr:<owner>/<name>#<n>, devbox:<box>, devbox-run:<workspace>,',
32  'vitrinka:<workspace>/<id>, deployik:<slug>[/<env>].',
33  'Conditions: pr: merged, closed, checks-settled (green OR red - use it to wait for CI), checks-green,',
34  'checks-red, eve-approved, ready; devbox: up, down; devbox-run: idle, running; vitrinka: done, closed;',
35  'deployik: live, failed, building, settled; any target: changed, <field>=<value>.',
36  'Not a replacement for /prm\'s `vybava gitkit pr-events` Monitor, which also carries review comments; keep it.',
37  'It wakes the main session only - a subagent must not call it.',
38].join(' ')
39
40const SCHEMA = {
41  type: 'object',
42  properties: {
43    target: { type: 'string', description: 'What to watch, e.g. pr:155, devbox-run:my-ws, vitrinka:fixit/4759' },
44    until: { type: 'string', description: 'The condition that wakes you, e.g. checks-settled, merged, idle, done' },
45    note: { type: 'string', description: 'Optional: what you will do when woken; echoed back in the wake prompt' },
46  },
47  required: ['target', 'until'],
48  additionalProperties: false,
49}
50
51// The poll loop's handles live with the module: a reload drops its timers,
52// and session.start arms a fresh loop from the daemon's own list.
53let timer: Timer | null = null
54let isPolling = false
55
56export const register: Register = on => {
57  on('session.start', async ($, e, next) => {
58    timer = null
59    isPolling = false
60    await $.tool.register({ name: 'when', description: DESCRIPTION, inputSchema: SCHEMA })
61    await sync($)
62    return next(e)
63  })
64
65  // /clear and /resume start a new conversation without session.start; the
66  // watches of the old one stay with the daemon until their TTL.
67  on('classic.SessionStart', async ($, e, next) => {
68    const ran = await next(e)
69    if (e.source === 'clear' || e.source === 'resume') {
70      await sync($)
71    }
72    return ran
73  })
74
75  on('tool.call', { tool: 'mcp__wake__when' }, async ($, e) => {
76    const target = e['target']
77    const until = e['until']
78    const note = e['note']
79    if (typeof target !== 'string' || target.trim() === '' || typeof until !== 'string' || until.trim() === '') {
80      return { deny: 'wake: `target` and `until` are required strings (e.g. target "pr:155", until "checks-settled").' }
81    }
82    const want = { target: target.trim(), until: until.trim(), note: typeof note === 'string' && note.trim() !== '' ? note.trim() : null }
83    const session = await $.session.id()
84    const dir = await $.session.cwd()
85    const reply = await call($, 'POST', '/v1/subscriptions', { session, target: want.target, until: want.until, dir })
86    if (reply.kind === 'down') {
87      await markDown($, reply.reason)
88      if (!(await read($, hinted))) {
89        await update($, hinted, () => true)
90        $.ui.status('the watch daemon is not running — `vybava watch agent install`')
91      }
92      return {
93        deny:
94          `wake: the watch daemon is not running (${reply.reason}). Wait with a Monitor instead: ` +
95          `\`vybava watch until ${want.target} ${want.until} --timeout 2h\` (it probes directly). ` +
96          'The human can install the daemon with `vybava watch agent install`.',
97      }
98    }
99    await markUp($)
100    if (reply.status !== 201) {
101      return { deny: `wake: ${errorOf(reply.json) ?? `the daemon answered ${reply.status}`}` }
102    }
103    const added = parseAdded(reply.json)
104    if (added === null) {
105      $.ui.log(`unexpected subscribe answer: ${reply.text.slice(0, 200)}`, { to: 'debug' })
106      $.ui.log('the watch daemon answered a subscribe it could not read; see the debug log')
107      return { deny: 'wake: the watch daemon answered in a shape wake cannot read (mod and daemon versions differ?).' }
108    }
109    const seqs = added.events.map(ev => ev.seq)
110    if (seqs.length > 0) {
111      await update($, handled, list => [...list, ...seqs])
112    }
113    const met = added.events.find(ev => ev.kind === 'met')
114    if (met !== undefined) {
115      $.ui.toast(`◉ ${label(met.target)} · ${met.summary || `already ${met.until}`}`)
116      return { result: `Already ${want.until}: ${label(met.target)} — ${met.summary || 'condition holds now'}. Nothing to wait for; carry on.` }
117    }
118    const watch: WakeWatch = { id: added.id, target: added.target, until: want.until, note: want.note }
119    await update($, watches, list => [...list.filter(one => one.id !== watch.id), watch])
120    await refreshStatus($)
121    arm($, 0)
122    return {
123      result:
124        `Subscribed ${watch.id}: this session will be woken when ${label(watch.target)} is ${watch.until}. ` +
125        'End your turn now; do not poll, sleep or arm a Monitor for it.',
126    }
127  })
128
129  on('turn.start', async ($, e, next) => {
130    await update($, busy, () => true)
131    return next(e)
132  })
133
134  on('turn.complete', async ($, e, next) => {
135    const ran = await next(e)
136    if (e.agentId !== undefined) {
137      return ran
138    }
139    await update($, busy, () => false)
140    const waiting = await read($, held)
141    if (waiting.length > 0) {
142      await update($, held, () => [])
143      submit($, waiting)
144    }
145    return ran
146  })
147
148  on('ui.render', { component: 'AbovePrompt' }, async ($, e, next) => {
149    const list = await read($, alerts)
150    const top = list.at(-1)
151    if (e.props.hasSurvey || top === undefined) {
152      return next(e)
153    }
154    const theirs = await next(e)
155    const { Box, Text, Button } = $.ui.resolve(e)
156    const more = list.length > 1 ? ` (+${list.length - 1} more)` : ''
157    return (
158      <Box flexDirection="column">
159        {theirs}
160        <Box flexDirection="row" gap={1}>
161          <Text wrap="truncate-end">
162            ◉ {top.line}
163            {more}
164          </Text>
165          <Button key="wake-prm" label="/prm" hotkey="4" onPress={() => runPrm($, top)} />
166          <Button key="wake-dismiss" label="dismiss" hotkey="5" onPress={() => update($, alerts, () => [])} />
167        </Box>
168      </Box>
169    )
170  })
171}
172
173/** Re-reads this session's watches from the daemon and (re)arms the poll. */
174async function sync($: EngineInterface): Promise<void> {
175  const session = await $.session.id()
176  const reply = await call($, 'GET', `/v1/subscriptions?session=${encodeURIComponent(session)}`)
177  if (reply.kind === 'down') {
178    await markDown($, reply.reason)
179    if ((await read($, watches)).length > 0) {
180      arm($, RETRY_MS)
181    }
182    return
183  }
184  await markUp($)
185  const listed = reply.status === 200 ? parseListed(reply.json) : null
186  if (listed === null) {
187    $.ui.log(`could not read this session's watches (${reply.status}): ${reply.text.slice(0, 200)}`, { to: 'debug' })
188    return
189  }
190  const known = await read($, watches)
191  await update($, watches, () =>
192    listed.map(one => ({ ...one, note: known.find(old => old.id === one.id)?.note ?? null })),
193  )
194  await refreshStatus($)
195  if (listed.length > 0) {
196    arm($, 0)
197    return
198  }
199  await drainOnce($)
200}
201
202/** With no live watch nothing polls, yet a wake-up queued while the mod was
203 * unloaded is still waiting: a met or expired event ends its subscription
204 * at the daemon the moment it is queued. Takes it once and acknowledges it,
205 * since no later slice will. */
206async function drainOnce($: EngineInterface): Promise<void> {
207  const session = await $.session.id()
208  const path = (seq: number): string => `/v1/events?session=${encodeURIComponent(session)}&after=${seq}&timeout=0s`
209  const reply = await call($, 'GET', path(await read($, after)))
210  if (reply.kind === 'down') {
211    return
212  }
213  const events = reply.status === 200 ? parseEvents(reply.json) : null
214  if (events === null) {
215    $.ui.log(`unreadable events answer (${reply.status}): ${reply.text.slice(0, 200)}`, { to: 'debug' })
216    return
217  }
218  if (events.length === 0) {
219    return
220  }
221  await deliver($, events)
222  await call($, 'GET', path(await read($, after)))
223}
224
225/** Schedules the next long-poll slice unless one is due or running. */
226function arm($: EngineInterface, delayMs: number): void {
227  if (timer !== null || isPolling) {
228    return
229  }
230  timer = $.clock.after(delayMs, () => {
231    timer = null
232    void poll($)
233  })
234}
235
236/** Runs one slice, then re-arms while watches remain, else stops. */
237async function poll($: EngineInterface): Promise<void> {
238  if (isPolling) {
239    return
240  }
241  isPolling = true
242  let retry: number
243  try {
244    retry = await slice($)
245  } catch (err) {
246    $.ui.log(`poll failed: ${messageOf(err)}`, { to: 'debug' })
247    retry = RETRY_MS
248  } finally {
249    isPolling = false
250  }
251  if ((await read($, watches)).length > 0) {
252    arm($, retry)
253    return
254  }
255  // Nothing left to watch: acknowledge the last batch so a later watch
256  // never sees it again, and stop.
257  const session = await $.session.id()
258  const seq = await read($, after)
259  if (seq > 0) {
260    await call($, 'GET', `/v1/events?session=${encodeURIComponent(session)}&after=${seq}&timeout=0s`)
261  }
262}
263
264/** One long-poll: acknowledge what was handled, wait up to SLICE for more.
265 * Answers the delay before the next slice. */
266async function slice($: EngineInterface): Promise<number> {
267  if ((await read($, watches)).length === 0) {
268    return 0
269  }
270  const session = await $.session.id()
271  const seq = await read($, after)
272  const reply = await call($, 'GET', `/v1/events?session=${encodeURIComponent(session)}&after=${seq}&timeout=${SLICE}`)
273  if (reply.kind === 'down') {
274    await markDown($, reply.reason)
275    return RETRY_MS
276  }
277  await markUp($)
278  const events = reply.status === 200 ? parseEvents(reply.json) : null
279  if (events === null) {
280    $.ui.log(`unreadable events answer (${reply.status}): ${reply.text.slice(0, 200)}`, { to: 'debug' })
281    return RETRY_MS
282  }
283  await deliver($, events)
284  return 0
285}
286
287/** Takes one answered batch: advance the watermark, react to what is new. */
288async function deliver($: EngineInterface, events: readonly WakeEvent[]): Promise<void> {
289  if (events.length === 0) {
290    return
291  }
292  const seen = await read($, after)
293  const skip = await read($, handled)
294  const fresh = events.filter(ev => ev.seq > seen && !skip.includes(ev.seq)).sort((a, b) => a.seq - b.seq)
295  const top = Math.max(seen, ...events.map(ev => ev.seq))
296  await update($, after, n => Math.max(n, top))
297  await update($, handled, list => list.filter(seq => seq > top))
298  if (fresh.length > 0) {
299    await react($, fresh)
300  }
301}
302
303/** Toasts every transition, ends finished watches, wakes the model on them. */
304async function react($: EngineInterface, fresh: readonly WakeEvent[]): Promise<void> {
305  const known = await read($, watches)
306  const rows: WakeAlert[] = []
307  for (const ev of fresh) {
308    $.ui.toast(`◉ ${label(ev.target)} · ${ev.kind === 'error' ? `error: ${ev.error}` : ev.summary || ev.kind}`)
309    const pr = PR.exec(ev.target)
310    if (pr !== null && ev.kind !== 'expired') {
311      rows.push({ seq: ev.seq, target: ev.target, prm: `${pr[2]} in ${pr[1]}`, line: `${label(ev.target)} · ${ev.summary || ev.kind}` })
312    }
313  }
314  if (rows.length > 0) {
315    await update($, alerts, list =>
316      [...list.filter(one => !rows.some(row => row.target === one.target)), ...rows].slice(-MAX_ALERTS),
317    )
318  }
319  const ending: WakeUp[] = fresh
320    .filter(ev => ENDING.has(ev.kind))
321    .map(ev => ({ ...ev, note: known.find(one => one.id === ev.subscription)?.note ?? null }))
322  if (ending.length > 0) {
323    for (const ev of ending.filter(one => one.kind === 'error')) {
324      // The daemon keeps an erroring subscription and backs off; the watch
325      // ends here, so it must end there too.
326      const reply = await call($, 'DELETE', `/v1/subscriptions/${encodeURIComponent(ev.subscription)}`)
327      if (reply.kind === 'down' || (reply.status >= 300 && reply.status !== 404)) {
328        $.ui.log(`could not end ${label(ev.target)} at the daemon: ${reply.kind === 'down' ? reply.reason : reply.status}`, { to: 'debug' })
329      }
330    }
331    await update($, watches, list => list.filter(one => !ending.some(ev => ev.subscription === one.id)))
332    if (await read($, busy)) {
333      await update($, held, list => [...list, ...ending])
334    } else {
335      submit($, ending)
336    }
337  }
338  await refreshStatus($)
339}
340
341/** Starts the model's next turn with the facts, framed as this plugin. */
342function submit($: EngineInterface, events: readonly WakeUp[]): void {
343  const lines = events.map(ev => {
344    const parts = [`- ${ev.target} · until ${ev.until} · ${ev.kind}`]
345    if (ev.summary !== '') {
346      parts.push(ev.summary)
347    }
348    if (ev.changed.length > 0) {
349      parts.push(`changed: ${ev.changed.join(', ')}`)
350    }
351    if (ev.error !== '') {
352      parts.push(`error: ${ev.error}`)
353    }
354    if (ev.note !== null) {
355      parts.push(`your note: ${ev.note}`)
356    }
357    return parts.join(' · ')
358  })
359  const text = ['A watch you subscribed to with mcp__wake__when has settled:', ...lines].join('\n')
360  $.prompt.submit({ text }).catch(err => {
361    $.ui.log(`could not start the woken turn: ${messageOf(err)}`)
362  })
363}
364
365async function runPrm($: EngineInterface, alert: WakeAlert): Promise<void> {
366  await update($, alerts, list => list.filter(one => one.seq !== alert.seq))
367  try {
368    await $.command.run({ command: 'prm', args: alert.prm })
369  } catch (err) {
370    $.ui.log(`/prm did not run as a command (${messageOf(err)}); filling the prompt`, { to: 'debug' })
371    const filled = await $.prompt.fill({ text: `/prm ${alert.prm}` })
372    $.ui.toast(filled.isFilled ? '/prm is in the prompt — press Enter' : `run /prm ${alert.prm}`)
373  }
374}
375
376async function refreshStatus($: EngineInterface): Promise<void> {
377  const list = await read($, watches)
378  const down = await read($, isDown)
379  if (list.length > 0) {
380    const names = list.map(one => label(one.target))
381    $.ui.status(`watching ${names.join(', ')}${down ? ' — watch daemon unreachable, retrying' : ''}`)
382    return
383  }
384  // Down with nothing watched: leave whatever is shown (the install hint a
385  // refused subscribe put up), so a session that never subscribes stays quiet.
386  if (!down) {
387    $.ui.status(undefined)
388  }
389}
390
391async function markDown($: EngineInterface, reason: string): Promise<void> {
392  if (!(await read($, isDown))) {
393    $.ui.log(`watch daemon unreachable: ${reason}`, { to: 'debug' })
394  }
395  await update($, isDown, () => true)
396  await refreshStatus($)
397}
398
399async function markUp($: EngineInterface): Promise<void> {
400  if (await read($, isDown)) {
401    await update($, isDown, () => false)
402    await refreshStatus($)
403  }
404}
405
406type Reply = { kind: 'ok'; status: number; text: string; json: unknown } | { kind: 'down'; reason: string }
407
408/** One request to the daemon over its socket, never waiting past GUARD_MS. */
409async function call($: EngineInterface, method: string, path: string, body?: unknown): Promise<Reply> {
410  const home = await $.env.get('HOME')
411  if (home === undefined || home === '') {
412    return { kind: 'down', reason: 'HOME is not set' }
413  }
414  const socketPath = `${home}/.local/state/vybava/watch/watchd.sock`
415  const guard: { timer?: Timer } = {}
416  const expired = new Promise<null>(resolve => {
417    guard.timer = $.clock.after(GUARD_MS, () => resolve(null))
418  })
419  try {
420    const init = body === undefined
421      ? { method, socketPath }
422      : { method, socketPath, headers: { 'content-type': 'application/json' }, body: JSON.stringify(body) }
423    const res = await Promise.race([$.http.fetch(`http://watchd${path}`, init), expired])
424    if (res === null) {
425      return { kind: 'down', reason: `no answer in ${GUARD_MS / 1000}s` }
426    }
427    return { kind: 'ok', status: res.status, text: res.text, json: parseJson(res.text) }
428  } catch (err) {
429    return { kind: 'down', reason: messageOf(err) }
430  } finally {
431    guard.timer?.cancel()
432  }
433}
434
435function parseJson(text: string): unknown {
436  try {
437    return JSON.parse(text)
438  } catch {
439    return undefined
440  }
441}
442
443function isRecord(value: unknown): value is Record<string, unknown> {
444  return typeof value === 'object' && value !== null && !Array.isArray(value)
445}
446
447function stringOf(record: Record<string, unknown>, key: string): string {
448  const value = record[key]
449  return typeof value === 'string' ? value : ''
450}
451
452function parseEvent(value: unknown): WakeEvent | null {
453  if (!isRecord(value) || typeof value['seq'] !== 'number' || !KINDS.has(stringOf(value, 'kind'))) {
454    return null
455  }
456  const changed = value['changed']
457  return {
458    seq: value['seq'],
459    subscription: stringOf(value, 'subscription'),
460    target: stringOf(value, 'target'),
461    until: stringOf(value, 'until'),
462    kind: stringOf(value, 'kind') as WakeEventKind,
463    summary: stringOf(value, 'summary'),
464    changed: Array.isArray(changed) ? changed.filter((one): one is string => typeof one === 'string') : [],
465    error: stringOf(value, 'error'),
466  }
467}
468
469/** `{events: [...]}`; null when the shape is not that (never a silent empty). */
470function parseEvents(json: unknown): WakeEvent[] | null {
471  if (!isRecord(json) || !Array.isArray(json['events'])) {
472    return null
473  }
474  const events = json['events'].map(parseEvent)
475  return events.every((one): one is WakeEvent => one !== null) ? events : null
476}
477
478function parseAdded(json: unknown): { id: string; target: string; events: WakeEvent[] } | null {
479  if (!isRecord(json) || !isRecord(json['subscription'])) {
480    return null
481  }
482  const id = stringOf(json['subscription'], 'id')
483  const target = stringOf(json['subscription'], 'target')
484  const events = parseEvents({ events: json['events'] ?? [] })
485  return id === '' || target === '' || events === null ? null : { id, target, events }
486}
487
488function parseListed(json: unknown): Array<Omit<WakeWatch, 'note'>> | null {
489  if (!isRecord(json) || !Array.isArray(json['subscriptions'])) {
490    return null
491  }
492  const out: Array<Omit<WakeWatch, 'note'>> = []
493  for (const one of json['subscriptions']) {
494    if (!isRecord(one) || stringOf(one, 'id') === '') {
495      return null
496    }
497    out.push({ id: stringOf(one, 'id'), target: stringOf(one, 'target'), until: stringOf(one, 'until') })
498  }
499  return out
500}
501
502function errorOf(json: unknown): string | null {
503  return isRecord(json) && typeof json['error'] === 'string' ? json['error'] : null
504}
505
506/** `pr:owner/name#155` → `PR #155`; other targets as given. */
507function label(target: string): string {
508  const pr = PR.exec(target)
509  return pr === null ? target : `PR #${pr[2]}`
510}
511
512function messageOf(err: unknown): string {
513  return err instanceof Error ? err.message : String(err)
514}
515

1// wake's state contract: the session values its hooks keep in $.state.
2// The event and subscription shapes mirror `vybava watch` (docs/watch.md).
3
4/** One subscription this session holds at the watch daemon. */
5export type WakeWatch = {
6  /** The daemon's subscription id (`w…`). */
7  id: string
8  /** The canonical target, as the daemon answered it (`pr:owner/name#155`). */
9  target: string
10  /** The condition it waits for (`merged`, `checks-settled`, `idle`, …). */
11  until: string
12  /** Why the model subscribed, echoed back when it is woken. */
13  note: string | null
14}
15
16/** What the daemon tells a subscriber. */
17export type WakeEventKind = 'change' | 'met' | 'error' | 'expired'
18
19/** One event of `GET /v1/events`, as wake reads it. */
20export type WakeEvent = {
21  seq: number
22  subscription: string
23  target: string
24  until: string
25  kind: WakeEventKind
26  summary: string
27  changed: string[]
28  error: string
29}
30
31/** An event that ends a watch, with the note it was subscribed with. */
32export type WakeUp = WakeEvent & { note: string | null }
33
34/** The band row: the newest PR transition the person can act on. */
35export type WakeAlert = {
36  seq: number
37  target: string
38  /** `/prm`'s argument for the PR (`155 in owner/name`). */
39  prm: string
40  line: string
41}
42
43declare module 'claude-code' {
44  interface PluginState {
45    wake: {
46      watches: WakeWatch[]
47      /** The highest event seq handled; the next poll acknowledges up to it. */
48      after: number
49      /** Seqs handled outside a poll (an `add` answered met); skipped when they arrive. */
50      handled: number[]
51      /** Wake-ups that arrived while a turn ran, submitted when it ends. */
52      held: WakeUp[]
53      alerts: WakeAlert[]
54      /** A main-loop turn is running. */
55      busy: boolean
56      /** The last call to the daemon failed to reach it. */
57      isDown: boolean
58      /** The install hint was shown this session. */
59      hinted: boolean
60    }
61  }
62}
63