dispatch-watch

1import { atom, read, update } from 'claude-code'
2import type { Register } from 'claude-code'
3
4import type { Watch } from '../types'
5import { decide, parseDispatch, parseReply, parseReqId, POLL_MS, replySql, REPLY_STALE_MS, statusSql } from './logic'
6
7// Why (LCH-528): a dispatched workflow is watched by the mod, not by a check-in the session
8// schedules and forgets. Decisions live in ./logic.ts; this file gathers facts and acts.
9
10const watches = atom({ plugin: 'dispatch-watch', key: 'watches' } as const, [] as Watch[])
11
12function textOf(r: { content?: readonly unknown[] }): string {
13  return (r.content ?? []).map(c => String((c as { text?: unknown }).text ?? '')).join('')
14}
15
16// Wakes the host refused (submitted from inside a tool call) wait for the turn's end.
17const pendingWakes: string[] = []
18let ticking = false
19let busy = false
20
21function wake($: any, line: string) {
22  pendingWakes.push(line)
23  flush($)
24}
25
26function flush($: any) {
27  if (pendingWakes.length === 0) return
28  const lines = pendingWakes.splice(0)
29  try { $.ui.toast(lines[0]!.slice(0, 200)) } catch { /* a toast never changes a wake */ }
30  Promise.resolve()
31    .then(() => $.prompt.submit({ text: lines.join('\n') }))
32    .catch(() => { pendingWakes.unshift(...lines) })
33}
34
35async function sql($: any, w: Watch, query: string): Promise<string | null> {
36  try {
37    const r = await $.mcp.call(w.server, 'execute_sql', { project_id: w.projectId, query })
38    return r.isError ? null : textOf(r)
39  } catch { return null }
40}
41
42async function tick($: any) {
43  if (busy) return
44  busy = true
45  try {
46    const now = await $.clock.now()
47    for (const w of await read($, watches)) {
48      if (w.reqId === undefined || now - (w.reqAt ?? 0) > REPLY_STALE_MS) {
49        const id = parseReqId((await sql($, w, statusSql(w))) ?? '')
50        await update($, watches, l => l.map(x => (x === w || x.startedAt === w.startedAt
51          ? { ...x, reqId: id ?? undefined, reqAt: now } : x)))
52        continue
53      }
54      const d = decide(w, parseReply((await sql($, w, replySql(w.reqId))) ?? ''), now)
55      if (d.wake) wake($, d.wake)
56      await update($, watches, l => (d.done
57        ? l.filter(x => x.startedAt !== w.startedAt)
58        // The answer is used: the next tick sends a fresh status request.
59        : l.map(x => (x.startedAt === w.startedAt ? { ...x, reqId: undefined, warnedLong: x.warnedLong || d.warnedLong } : x))))
60    }
61  } catch { /* a failed tick is retried by the next one */ } finally {
62    busy = false
63  }
64}
65
66export const register: Register = on => {
67  on('tool.call', { tool: /^mcp__.*__execute_sql$/ }, async ($, e, next) => {
68    const ran = await next(e)
69    if (ran.deny !== undefined || ran.isError) return ran
70    const workflow = parseDispatch(String((e as any).query ?? ''))
71    if (!workflow) return ran
72    const now = await $.clock.now()
73    const w: Watch = {
74      workflow,
75      server: String((e as any).tool).split('__')[1] ?? 'Supabase',
76      projectId: String((e as any).project_id ?? ''),
77      since: new Date(now - 60_000).toISOString(),
78      startedAt: now,
79    }
80    await update($, watches, l => [...l, w])
81    if (!ticking) {
82      ticking = true
83      $.clock.every(POLL_MS, () => { void tick($) })
84    }
85    return { ...ran, context: [...(ran.context ?? []),
86      `[mod:dispatch-watch] Watching ${workflow}: its run is checked every minute and this session is woken when it ` +
87      'finishes (or fails to start). Do not schedule a long check-in for it.'] }
88  // Fail open: a bug here must never change the dispatch's outcome.
89  }).catch(($, e, next) => next(e))
90
91  on('turn.complete', async ($, e, next) => {
92    const res = await next(e)
93    flush($)
94    return res
95  })
96}
97

1import type { Watch } from '../types'
2
3// Pure decisions for dispatch-watch (LCH-528). register.ts only gathers facts and acts.
4//
5// Why this exists: a session that starts a workflow through the github-dispatch bridge has no
6// GitHub access to the target repo (LC-1879), so it could not see the run. On 2026-10-20 a
7// refresh-run.yml run finished in 85 s having done nothing, and the session waited hours on a
8// check-in it had scheduled for later. Memory 1440431b had already said "check within 1-2
9// minutes; a fast finish is the signal". This mod does that check, every minute, by itself.
10
11export const POLL_MS = 60_000
12export const NO_RUN_MS = 6 * 60_000 // no run listed this long after dispatch: it never registered
13export const LONG_MS = 90 * 60_000 // still running this long: say so once
14export const FAST_MS = 2 * 60_000 // finished faster than this: likely did nothing (1440431b)
15export const REPLY_STALE_MS = 5 * 60_000 // pg_net never answered the status request: send another
16
17const FN_URL = 'https://tlcbwwgfqnasmiipgmmd.supabase.co/functions/v1/github-dispatch'
18
19/** The workflow a bridge DISPATCH names, or null (a status request is not a dispatch). */
20export function parseDispatch(query: string): string | null {
21  // The real bridge only: its function URL, not any http_post that mentions the name.
22  if (!/\/functions\/v1\/github-dispatch\b/.test(query) || !/net\.http_post/i.test(query)) return null
23  if (/"action"\s*:\s*"status"/.test(query)) return null
24  const m = query.match(/"workflow"\s*:\s*"([\w.-]+)"/)
25  return m ? m[1]! : null
26}
27
28/** SQL that sends the bridge's read-only status request; the secret is read inside the statement. */
29export function statusSql(w: Pick<Watch, 'workflow' | 'since'>): string {
30  const body = JSON.stringify({ action: 'status', workflow: w.workflow, since: w.since }).replace(/'/g, "''")
31  return `select net.http_post(url := '${FN_URL}?secret=' || (select decrypted_secret from vault.decrypted_secrets ` +
32    `where name = 'edge_function_shared_secret'), headers := '{"Content-Type":"application/json"}'::jsonb, ` +
33    `body := '${body}'::jsonb) as req_id`
34}
35
36export function replySql(reqId: number): string {
37  return `select status_code, content::text as content from net._http_response where id = ${Math.trunc(reqId)}`
38}
39
40/** The JSON rows inside an execute_sql result text (the MCP wraps them in an untrusted-data block). */
41export function sqlRows(text: string): Record<string, unknown>[] | null {
42  let body = text
43  try {
44    const outer = JSON.parse(text)
45    if (outer && typeof outer.result === 'string') body = outer.result
46  } catch { /* already the inner text */ }
47  const m = body.match(/<untrusted-data-[^>]*>\s*([\s\S]*?)\s*<\/untrusted-data/)
48  try {
49    const rows = JSON.parse(m ? m[1]! : body)
50    return Array.isArray(rows) ? rows.filter(r => r && typeof r === 'object') : null
51  } catch { return null }
52}
53
54export function parseReqId(text: string): number | null {
55  const id = Number(sqlRows(text)?.[0]?.req_id)
56  return Number.isFinite(id) && id > 0 ? id : null
57}
58
59export type Run = { id: number; status: string; conclusion: string | null; created_at: string;
60  updated_at: string; run_started_at: string | null; html_url: string }
61export type Reply = { kind: 'pending' } | { kind: 'error'; detail: string } | { kind: 'runs'; runs: Run[] }
62
63/** The bridge's answer from a net._http_response read. */
64export function parseReply(text: string): Reply {
65  const row = sqlRows(text)?.[0]
66  if (!row) return { kind: 'pending' }
67  const code = Number(row.status_code)
68  let content: any
69  try { content = JSON.parse(String(row.content ?? '')) } catch { content = null }
70  if (code !== 200 || !content?.ok || !Array.isArray(content.runs)) {
71    return { kind: 'error', detail: `HTTP ${row.status_code}: ${String(content?.message ?? row.content ?? '').slice(0, 160)}` }
72  }
73  return { kind: 'runs', runs: content.runs as Run[] }
74}
75
76export type Decision = { done: boolean; wake?: string; warnedLong?: boolean }
77
78const mins = (ms: number) => (ms < 120_000 ? `${Math.round(ms / 1000)} s` : `${Math.round(ms / 60_000)} min`)
79
80/** What one status answer means for a watch. */
81export function decide(w: Watch, reply: Reply, now: number): Decision {
82  const tag = `[mod:dispatch-watch] ${w.workflow}`
83  if (reply.kind === 'pending') return { done: false }
84  if (reply.kind === 'error') {
85    return { done: true, wake: `${tag}: the status check failed (${reply.detail}). The run's state is unknown; ` +
86      'check it another way before waiting on it.' }
87  }
88  const floor = Date.parse(w.since)
89  const run = reply.runs.find(r => Date.parse(r.created_at) >= floor)
90  if (!run) {
91    return now - w.startedAt > NO_RUN_MS
92      ? { done: true, wake: `${tag}: no run appeared ${mins(now - w.startedAt)} after the dispatch. It likely never ` +
93          'registered: read the dispatch reply in net._http_response before waiting further.' }
94      : { done: false }
95  }
96  if (run.status !== 'completed') {
97    if (now - w.startedAt > LONG_MS && !w.warnedLong) {
98      return { done: false, warnedLong: true, wake: `${tag} run ${run.id} is still ${run.status} after ` +
99        `${mins(now - w.startedAt)}. ${run.html_url}` }
100    }
101    return { done: false }
102  }
103  const took = Date.parse(run.updated_at) - Date.parse(run.run_started_at ?? run.created_at)
104  const fast = Number.isFinite(took) && took < FAST_MS
105  return { done: true, wake: `${tag} run ${run.id} finished: ${run.conclusion ?? 'no conclusion'} after ${mins(took)}. ` +
106    `${run.html_url}` + (fast ? ' A finish under 2 minutes usually means it did nothing (memory 1440431b: a section ' +
107    'parked at `partial`; read the newest receipts and redispatch it by name).' : ' Read its outcome and continue.') }
108}
109

1/** One workflow run this session dispatched through the github-dispatch bridge. */
2export type Watch = {
3  workflow: string
4  /** Supabase MCP server and project the dispatch went through; the status check uses the same. */
5  server: string
6  projectId: string
7  /** ISO time a minute before the dispatch: runs created from here on are candidates. */
8  since: string
9  startedAt: number
10  /** The pending status request (net.http_post id) and when it was sent. */
11  reqId?: number
12  reqAt?: number
13  warnedLong?: boolean
14}
15
16declare module 'claude-code' {
17  interface PluginState {
18    'dispatch-watch': { watches: Watch[] }
19  }
20}
21