pipeline

1import type { Register } from 'claude-code'
2import { LOOP_PLUGIN, TAG } from '../pipeline-agents/src/agents.ts'
3import { pass } from '../src/loop.ts'
4import { type Prompts, promptsChanged } from '../src/prompts.ts'
5
6const FIRST_MS = 10_000
7// STALE in src/loop.ts is 3 of these intervals: change it with this.
8const EVERY_MS = 5 * 60_000
9
10// The loop only. The hooks its agents need are in pipeline-agents: a plugin never hears the agents it spawned from
11// its own timer (docs/adr/0001).
12export const register: Register = on => {
13  on('session.start', async ($, e, next) => {
14    const started = await next(e)
15    if ($.plugin.name !== LOOP_PLUGIN) throw new Error(`the loop plugin must be named ${LOOP_PLUGIN}: pipeline-agents recognises agents by it`)
16    const root = await $.session.root()
17    // read once, so every claim and heartbeat of this loop carries the same owner
18    const session = await $.session.id()
19    const home = await $.env.get('HOME')
20    if (home === undefined) throw new Error('HOME is unset: the agents\' worktrees live under it')
21    try {
22      const entries = await $.fs.list(`${$.plugin.root}/agents`)
23      const now: Prompts = Object.fromEntries(entries.filter(f => f.name.endsWith('.md')).map(f => [f.name, f.mtimeMs]))
24      const { value: atStart } = await $.state.get({ plugin: 'pipeline', key: 'agentPrompts' })
25      if (atStart === undefined) await $.state.set({ plugin: 'pipeline', key: 'agentPrompts' }, now)
26      else {
27        const line = promptsChanged(atStart, now)
28        if (line) $.ui.log(line)
29      }
30    } catch (err) {
31      $.ui.log(`agent prompt check failed: ${(err as Error).message}`)
32    }
33    let isRunning = false
34    const tick = async () => {
35      if (isRunning) return
36      isRunning = true
37      try {
38        const lines = await pass({
39          run: (argv, init) => $.process.run(argv, { cwd: root, ...init }),
40          complete: request => $.model.complete(request),
41          spawn: args => $.agent.spawn(args),
42          agents: () => $.agent.list(),
43          plugin: $.plugin.name,
44          session,
45          root,
46          home,
47        })
48        for (const line of lines) $.ui.log(line)
49      } catch (err) {
50        $.ui.log(`pass failed: ${(err as Error).message}`)
51      } finally {
52        isRunning = false
53      }
54    }
55    // first pass shortly after start rather than a full interval in
56    $.clock.after(FIRST_MS, tick)
57    $.clock.every(EVERY_MS, tick)
58    return started
59  })
60
61  // the loop's agents finishing must not start a turn on the main thread
62  const finished = new RegExp(`<summary>(Agent "${TAG} [^"]*" [^<]*)</summary>`)
63  on('prompt.submit', ($, e, next) => {
64    const done = e.origin.kind === 'task-notification' && e.text.match(finished)
65    return done ? { drop: done[1] ?? done[0] } : next(e)
66  }).catch(($, e, next) => next(e))
67}
68

1import type { AgentInfo } from 'claude-code'
2
3/** Every pipeline agent's description starts with this; the notification filter keys on it too. */
4export const TAG = 'pipeline'
5
6/** The loop plugin's name: every pipeline agent is spawned by it, which is how pipeline-agents' hooks recognise them. */
7export const LOOP_PLUGIN = 'pipeline'
8
9/** The review agents, `agents/<role>.md`, each described `pipeline <ticket_id> <role>`; the implementer is `pipeline <ticket_id>`. */
10export const ROLES = ['reviewer-spec', 'reviewer-quality', 'verifier', 'fixer'] as const
11export type Role = 'implementer' | (typeof ROLES)[number]
12
13/** The ticket and role of an agent this plugin spawned for a ticket, or null for any other agent. */
14export const agentOf = (agent: AgentInfo | undefined, plugin: string): { ticket: string; role: Role } | null => {
15  if (agent?.spawnedBy !== plugin || !agent.description.startsWith(`${TAG} `)) return null
16  const rest = agent.description.slice(TAG.length + 1)
17  const role = ROLES.find(r => rest.endsWith(` ${r}`))
18  return role === undefined ? { ticket: rest, role: 'implementer' } : { ticket: rest.slice(0, -role.length - 1), role }
19}
20
21

1import type { AgentInfo, AgentSpawnArgs, AgentSpawnResult } from 'claude-code'
2import { ROLES, type Role, TAG, agentOf } from '../pipeline-agents/src/agents.ts'
3import { worktreeOf } from '../pipeline-agents/src/guards.ts'
4import { reviewFrom } from '../pipeline-agents/src/routing.ts'
5import { DB, type Run, type Stage, type Ticket, lit, migrate, sql } from '../pipeline-agents/src/store.ts'
6import { closeMerged, sync } from './sync.ts'
7import { type Complete, MODELS, REVIEW_MODEL, triage } from './triage.ts'
8
9export type Deps = {
10  /** `$.process.run` with the project root as cwd. */
11  run: Run
12  complete: Complete
13  spawn: (args: AgentSpawnArgs) => Promise<AgentSpawnResult>
14  agents: () => Promise<AgentInfo[]>
15  /** `$.plugin.name`, which namespaces the implementer agents. */
16  plugin: string
17  /** `$.session.id()`: the owner this session's claims carry. */
18  session: string
19  /** `$.session.root()`: the main checkout, which the agents never change. */
20  root: string
21  /** `$HOME`, under which the agents' worktrees live. */
22  home: string
23}
24
25/** The loop chooses each agent's worktree and states it in the spawn prompt; the guard in pipeline-agents holds it to that path. */
26const worktreeLine = (d: Pick<Deps, 'root' | 'home'>, ticketId: string, role: Role) => `Worktree: ${worktreeOf(d.home, d.root, ticketId, role)}`
27
28/** Removes every worktree left under the ticket's paths, one per role, then prunes; touches nothing else. Resolves what to log. */
29export const removeWorktrees = async (d: Pick<Deps, 'run' | 'root' | 'home'>, ticketId: string): Promise<string[]> => {
30  const log: string[] = []
31  for (const role of ['implementer', ...ROLES] as const) {
32    const path = worktreeOf(d.home, d.root, ticketId, role)
33    if ((await d.run(['test', '-e', path])).exitCode !== 0) continue
34    const r = await d.run(['git', 'worktree', 'remove', '--force', path])
35    log.push(r.exitCode === 0 ? `${ticketId} ${role} left a worktree; removed ${path}` : `${ticketId} ${role} left ${path}, which could not be removed: ${r.stderr.trim()}`)
36  }
37  await d.run(['git', 'worktree', 'prune'])
38  return log
39}
40
41const LIVE: readonly string[] = ['pending', 'running', 'waiting']
42
43/** An owner unseen for 3 pass intervals (5 minutes each) is taken as stopped. */
44const STALE = `(owner_seen is null or owner_seen < datetime('now', '-15 minutes'))`
45
46/** The agents a stage starts. */
47const STARTS: Record<Exclude<Stage, 'implement'>, readonly Role[]> = { review: ['reviewer-spec', 'reviewer-quality'], verify: ['verifier'], fix: ['fixer'] }
48
49/** Spawns the agents of a stage the ticket has just entered: reviewers and verifier on opus, the fixer on the routed model. Resolves what to log. */
50export const start = async (d: Pick<Deps, 'run' | 'spawn' | 'plugin' | 'root' | 'home'>, ticketId: string, stage: Exclude<Stage, 'implement'>): Promise<string> => {
51  const [t] = (await sql(d.run, DB, `select ticket_id, title, url, model, pr_url from tickets where ticket_id = ${lit(ticketId)}`)) as Ticket[]
52  if (t === undefined) throw new Error(`no ticket ${ticketId} to start ${stage} for`)
53  const routed = t.model as (typeof MODELS)[number]
54  if (!MODELS.includes(routed) || t.pr_url === null) throw new Error(`${ticketId} is at ${stage} with no routed model or PR: ${t.model}, ${t.pr_url}`)
55  const model = stage === 'fix' ? routed : REVIEW_MODEL
56  for (const role of STARTS[stage]) {
57    const s = await d.spawn({
58      description: `${TAG} ${ticketId} ${role}`,
59      subagentType: `${d.plugin}:${role}`,
60      model,
61      prompt: `Ticket ${ticketId}: ${t.title}\n${t.url ?? ''}\nPR: ${t.pr_url}\n${worktreeLine(d, ticketId, role)}`,
62    })
63    if (s.deny === undefined) continue
64    const question = `The ${role} could not start: ${s.deny}. The PR is still a draft: finish its review, then set the ticket pr_open.`
65    await sql(d.run, DB, `update tickets set status = 'needs_human', restarts = 0, question = ${lit(question)} where ticket_id = ${lit(ticketId)} and status = 'in_progress'`)
66    return `${ticketId} needs you: its ${role} could not start: ${s.deny}`
67  }
68  await sql(d.run, DB, `update tickets set started = 1 where ticket_id = ${lit(ticketId)}`)
69  return `${model} is at ${stage} for ${ticketId}`
70}
71
72const read = async (run: Run, name: string) => {
73  const r = await run(['cat', `.claude/pipeline/${name}.md`])
74  return r.exitCode === 0 ? r.stdout : null
75}
76
77/** One pass: mark merged PRs done and sync (a tracker failure is logged and the pass goes on), triage every new ticket, move each in_progress ticket this session owns on, send those of a stopped session to a human, claim and spawn the implementer for the next ready one if none is in progress. Resolves what to log. */
78export const pass = async (d: Deps): Promise<string[]> => {
79  const [tracker, readiness, rubric, routing] = await Promise.all([
80    read(d.run, 'tracker'),
81    read(d.run, 'readiness'),
82    read(d.run, 'rubric'),
83    read(d.run, 'routing'),
84  ])
85  if (tracker === null || readiness === null || rubric === null || routing === null) return [`no .claude/pipeline setup here; run /tracker-setup`]
86  await migrate(d.run, DB)
87  // heartbeat first, so a pass that fails later still shows this session alive; a ticket from before owners is
88  // adopted, so the stage rules below apply to it as they did then
89  await sql(d.run, DB, `update tickets set owner = coalesce(owner, ${lit(d.session)}), owner_seen = datetime('now') where status = 'in_progress' and (owner is null or owner = ${lit(d.session)})`)
90  const log: string[] = []
91  const live = (await d.agents()).filter(a => LIVE.includes(a.status))
92
93  // 1. a merged PR marks its ticket done before sync, so dependents it was holding are released in this pass
94  // 2. sync, so what it adds is triaged in this pass
95  // a tracker that fails or times out (a laptop waking from sleep) must not hold up the tickets already in the table
96  const done: string[] = []
97  try {
98    const closed = await closeMerged(d.run, DB, tracker)
99    for (const id of closed) log.push(`${id} is done: its PR merged`)
100    done.push(...closed, ...(await sync(d.run, DB, tracker)))
101  } catch (err) {
102    log.push(`tracker unreachable: ${(err as Error).message}`)
103  }
104  // a done ticket's agents are gone, however it got there: whatever worktree one left behind goes
105  for (const id of done) log.push(...(await removeWorktrees(d, id)))
106
107  // 3. triage; a failed ticket stays new and is retried next pass
108  const fresh = (await sql(d.run, DB, `select ticket_id, title, url, body from tickets where status = 'new' order by rowid`)) as Ticket[]
109  for (const t of fresh) {
110    try {
111      const v = await triage(d.run, d.complete, DB, { readiness, rubric, routing }, t)
112      log.push(v.ready ? `${t.ticket_id} is ready for ${v.model}` : `${t.ticket_id} needs you: ${v.question}`)
113    } catch (err) {
114      log.push(`${(err as Error).message}`)
115    }
116  }
117
118  // 4. only the owner moves a ticket on: $.agent.list() sees this session's agents alone. Another session's ticket
119  // counts toward the one-at-a-time limit while its owner is seen, and goes to a human once it is not.
120  // A stage with no agent of its ticket still live moves on: a stage report_result entered gets its agents started
121  // here, since only this plugin's spawns are heard by pipeline-agents (docs/adr/0001); finished reviewers hand over
122  // to the verifier; a verifier or fixer gone with no result reported (crash, sleep) is started again once; any other
123  // agent gone, or one gone a second time, leaves its ticket to a human: an implementer may have pushed a branch
124  const busy = (await sql(d.run, DB, `select ticket_id, stage, started, restarts, owner, ${STALE} as stale from tickets where status = 'in_progress' order by rowid`)) as (Ticket & { stale: 0 | 1 })[]
125  const working = new Set(live.map(a => agentOf(a, d.plugin)?.ticket))
126  let waiting = 0
127  for (const t of busy) {
128    // the heartbeat adopted every ownerless one, and a claim always sets the owner
129    if (t.owner === null) throw new Error(`${t.ticket_id} is in_progress with no owner`)
130    if (t.owner !== d.session && t.stale === 0) waiting++
131    else if (t.owner !== d.session) {
132      const question = `The session that owned this ticket (${t.owner}) stopped, so whatever agent it was running is gone. Check the branch and any PR, then send the ticket back to new or set it pr_open.`
133      const moved = await sql(d.run, DB, `update tickets set status = 'needs_human', restarts = 0, question = ${lit(question)} where ticket_id = ${lit(t.ticket_id)} and status = 'in_progress' and owner = ${lit(t.owner)} and ${STALE} returning ticket_id`)
134      if (moved.length > 0) log.push(`${t.ticket_id} needs you: its owning session stopped`)
135    } else if (working.has(t.ticket_id)) waiting++
136    else if (t.started === 0 && t.stage !== null && t.stage !== 'implement') {
137      waiting++
138      log.push(await start(d, t.ticket_id, t.stage))
139    } else if (t.stage === 'review') {
140      waiting++
141      const moved = await sql(d.run, DB, `update tickets set stage = 'verify', started = 0 where ticket_id = ${lit(t.ticket_id)} and status = 'in_progress' and stage = 'review' returning ticket_id`)
142      if (moved.length > 0) log.push(await start(d, t.ticket_id, 'verify'))
143    } else if ((t.stage === 'verify' || t.stage === 'fix') && t.restarts === 0) {
144      waiting++
145      const moved = await sql(d.run, DB, `update tickets set started = 0, restarts = 1 where ticket_id = ${lit(t.ticket_id)} and status = 'in_progress' and stage = ${lit(t.stage)} and restarts = 0 returning ticket_id`)
146      if (moved.length > 0) log.push(`${t.ticket_id} ${STARTS[t.stage].join(', ')} vanished; restarting it once`, await start(d, t.ticket_id, t.stage))
147    } else {
148      const who = t.stage === 'verify' ? 'verifier' : t.stage === 'fix' ? 'fixer' : 'implementer'
149      const question =
150        who === 'implementer'
151          ? 'The implementer stopped without recording a PR or a question. Check its branch, then send the ticket back to new.'
152          : `The ${who} stopped without recording a result. The PR is still a draft: check it, then set the ticket pr_open or send it back to new.`
153      await sql(d.run, DB, `update tickets set status = 'needs_human', restarts = 0, question = ${lit(question)} where ticket_id = ${lit(t.ticket_id)} and status = 'in_progress'`)
154      log.push(`${t.ticket_id} needs you: its ${who} stopped without a result`)
155    }
156  }
157  // one ticket at a time: no new implementer while any ticket is at any stage
158  if (waiting > 0) return log
159
160  const [next] = (await sql(d.run, DB, `select ticket_id, title, url, model, difficulty from tickets where status = 'ready' order by rowid limit 1`)) as Ticket[]
161  if (next === undefined) return log
162  if (!MODELS.includes(next.model as (typeof MODELS)[number])) throw new Error(`${next.ticket_id} is ready with no routed model: ${next.model}`)
163  if (next.difficulty === null) throw new Error(`${next.ticket_id} is ready with no difficulty`)
164  const draft = next.difficulty >= reviewFrom(routing)
165  // the claim: one statement, so of passes racing (in this session or another) only one gets the row back, and only
166  // while nothing is in_progress
167  const claimed = await sql(
168    d.run,
169    DB,
170    `update tickets set status = 'in_progress', stage = 'implement', started = 1, owner = ${lit(d.session)}, owner_seen = datetime('now')
171     where ticket_id = ${lit(next.ticket_id)} and status = 'ready' and not exists (select 1 from tickets where status = 'in_progress') returning ticket_id`,
172  )
173  if (claimed.length === 0) return log
174  const s = await d.spawn({
175    description: `${TAG} ${next.ticket_id}`,
176    subagentType: `${d.plugin}:implementer-${next.model}`,
177    prompt: `Ticket ${next.ticket_id}: ${next.title}\n${next.url ?? ''}\n${worktreeLine(d, next.ticket_id, 'implementer')}\n\n${
178      draft ? 'Open the PR as a draft on the host, with the title prefix "Draft: ": agents review it before the user does.' : 'Open the PR without a "Draft: " title prefix.'
179    }`,
180  })
181  if (s.deny !== undefined) {
182    await sql(d.run, DB, `update tickets set status = 'ready', owner = null, owner_seen = null, restarts = 0 where ticket_id = ${lit(next.ticket_id)} and status = 'in_progress'`)
183    log.push(`implementer for ${next.ticket_id} not started: ${s.deny}`)
184  } else log.push(`${next.model} is implementing ${next.ticket_id}`)
185  return log
186}
187

1/** Each agent prompt file's name in `agents/` and its mtime. */
2export type Prompts = Record<string, number>
3
4/**
5 * The line to log when an agent prompt file was changed, added or removed since session start; null when none was.
6 * A running session keeps the agent definitions it loaded at start, while a reload only swaps in the code.
7 */
8export const promptsChanged = (atStart: Prompts, now: Prompts): string | null => {
9  const changed = [...new Set([...Object.keys(atStart), ...Object.keys(now)])].filter(name => atStart[name] !== now[name]).sort()
10  return changed.length ? `agent prompts changed (${changed.map(name => `agents/${name}`).join(', ')}); restart the session to apply them` : null
11}
12

1import type { AgentInfo } from 'claude-code'
2import { type Role, agentOf } from './agents.ts'
3
4/** Where every pipeline agent's worktree lives, under the user's home: `<home>/.claude/pipeline-worktrees/<project>/<ticket>-<role>`. */
5export const WORKTREES = '.claude/pipeline-worktrees'
6
7/** The tools the worktree guard checks. */
8export const GUARDED = ['Edit', 'Write', 'NotebookEdit', 'Bash'] as const
9
10/** A ticket id names one path segment; a project folder name is only kept out of `.`, `..`, empty and a `/`. */
11const TICKET = /^[\w.@+-]+$/
12const bad = (part: string, strict: boolean) => part === '' || part === '.' || part === '..' || part.includes('/') || (strict && !TICKET.test(part))
13
14/** The worktree of `role` for `ticket` of the project at `root`; throws on a ticket id or project name that cannot be one path segment. */
15export const worktreeOf = (home: string, root: string, ticket: string, role: Role): string => {
16  const project = root.replace(/\/+$/, '').split('/').pop() ?? ''
17  for (const [part, strict] of [[project, false], [ticket, true]] as const) if (bad(part, strict)) throw new Error(`cannot name a worktree from ${JSON.stringify(part)}`)
18  return `${home}/${WORKTREES}/${project}/${ticket}-${role}`
19}
20
21/**
22 * `path` made absolute against `cwd` with `.`, `..` and `~` resolved as text, or null when it cannot be placed:
23 * empty, `-`, a shell expansion (`$VAR`, a glob) or an unknown cwd. Symlinks are not resolved.
24 */
25export const resolve = (path: string, cwd: string | null, home: string): string | null => {
26  const p = path.replace(/^(["'])(.*)\1$/, '$2')
27  if (p === '' || p === '-' || /[$`*?]/.test(p)) return null
28  const abs = p === '~' || p.startsWith('~/') ? home + p.slice(1) : p.startsWith('/') ? p : cwd === null ? null : `${cwd}/${p}`
29  if (abs === null) return null
30  const out: string[] = []
31  for (const part of abs.split('/')) {
32    if (part === '..') out.pop()
33    else if (part !== '' && part !== '.') out.push(part)
34  }
35  return `/${out.join('/')}`
36}
37
38const within = (path: string, dir: string) => path === dir || path.startsWith(`${dir}/`)
39
40/** Git subcommands that change nothing, allowed from anywhere. */
41const READ_ONLY = new Set(['status', 'log', 'diff', 'show', 'fetch', 'rev-parse', 'worktree', 'ls-files', 'ls-tree', 'ls-remote', 'blame', 'grep', 'describe', 'cat-file', 'rev-list', 'merge-base', 'reflog', 'version', 'shortlog', 'for-each-ref', 'show-ref', 'name-rev', 'check-ignore', 'diff-tree', 'var', 'count-objects'])
42/** Subcommands that change nothing with these arguments: a listing `branch`, `tag`, `stash`, `remote` and a `config` read. */
43const READS: Record<string, (args: string[]) => boolean> = {
44  branch: args => !args.some(a => /^-[dDmMcCuf]|^--(delete|move|copy|force|set-upstream-to|unset-upstream)/.test(a)),
45  tag: args => args.length === 0 || args.includes('-l') || args.includes('--list'),
46  stash: args => args[0] === 'list' || args[0] === 'show',
47  remote: args => args.length === 0 || args[0] === 'show' || args[0] === 'get-url' || args[0]!.startsWith('-'),
48  config: args => args.some(a => a === '-l' || a.startsWith('--get') || a.startsWith('--list')),
49}
50const readOnly = (sub: string, args: string[]) => READ_ONLY.has(sub) || (READS[sub]?.(args) ?? false)
51
52/** Each simple command of a shell command line, roughly: split on the shell's separators, subshells and substitutions; a word's outer quotes are dropped so `sh -c 'git …'` shows its git. */
53const commands = (command: string) =>
54  command
55    .split(/\|\||&&|[;|&()\n`]/)
56    .map(s => s.trim().split(/\s+/).map(w => w.replace(/^["']|["']$/g, '')))
57    .filter(w => w[0] !== '')
58
59/**
60 * The state-changing git command of `command` that is not placed in `worktree`, as `[subcommand, where]` (`where`
61 * null when it cannot be placed), or undefined when every git command is read-only or targets the worktree. The
62 * command starts in `root`; `cd` moves it, `git -C` places one command.
63 */
64const misplacedGit = (command: string, root: string, home: string, worktree: string): [string, string | null] | undefined => {
65  let cwd: string | null = root
66  for (const words of commands(command)) {
67    if (words[0] === 'cd') {
68      cwd = words[1] === undefined ? home : resolve(words[1], cwd, home)
69      continue
70    }
71    // git anywhere in the command, so a wrapper the guard does not know (rtk proxy, timeout, sudo, sh -c) fails closed
72    let i = words.indexOf('git')
73    if (i === -1) continue
74    let where = cwd
75    for (i++; i < words.length && words[i]!.startsWith('-'); i++) {
76      if (words[i] === '-C') where = resolve(words[++i] ?? '', where, home)
77      else if (words[i] === '-c') i++
78    }
79    const sub = words[i]
80    if (sub === undefined || readOnly(sub, words.slice(i + 1))) continue
81    if (where === null || !within(where, worktree)) return [sub, where]
82  }
83  return undefined
84}
85
86const target = (input: Record<string, unknown>) => {
87  const p = input.file_path ?? input.notebook_path
88  return typeof p === 'string' ? p : ''
89}
90
91const how = (worktree: string) => `Work in your worktree, ${worktree} (git worktree add ${worktree} …), and target it in git commands with cd ${worktree} && … or git -C ${worktree} ….`
92
93/**
94 * The refusal for a pipeline agent's edit inside the main checkout at `root`, or its state-changing git command
95 * not explicitly placed in its own worktree, or undefined to let the call run. Fails closed: a path or a command it
96 * cannot place is refused. Any other agent is not checked.
97 */
98export const worktreeRefusal = (agent: AgentInfo | undefined, plugin: string, root: string, home: string, tool: string, input: Record<string, unknown>): string | undefined => {
99  const who = agentOf(agent, plugin)
100  if (who === null || !(GUARDED as readonly string[]).includes(tool)) return undefined
101  const main = resolve(root, '/', home)
102  if (main === null) throw new Error(`cannot place the project root ${root}`)
103  const worktree = worktreeOf(home, main, who.ticket, who.role)
104  if (tool === 'Bash') {
105    const bad = misplacedGit(String(input.command ?? ''), main, home, worktree)
106    if (bad === undefined) return undefined
107    const place = bad[1] === null ? 'cannot be placed' : within(bad[1], main) ? `in ${bad[1]}: pipeline agents never change the main checkout` : `in ${bad[1]}, which is not your worktree`
108    return `git ${bad[0]} ${place}. ${how(worktree)}`
109  }
110  const path = resolve(target(input), main, home)
111  if (path === null) return `${tool} to a path that cannot be placed (${JSON.stringify(target(input))}): pipeline agents never change the main checkout. ${how(worktree)}`
112  return within(path, main) ? `${tool} in the main checkout (${path}) is refused: pipeline agents never change it. ${how(worktree)}` : undefined
113}
114

1import type { Run } from './store.ts'
2
3/** `.claude/pipeline/routing.md`, or '' when there is none, so every setting reads as its default. */
4export const readRouting = async (run: Run): Promise<string> => {
5  const r = await run(['cat', '.claude/pipeline/routing.md'])
6  return r.exitCode === 0 ? r.stdout : ''
7}
8
9/** routing.md's `<label>: <n>` line as a whole number from 1; `fallback` when the line is missing, so a routing.md written before it needs no re-setup. */
10const whole = (routing: string, label: string, fallback: number): number => {
11  const line = routing.match(new RegExp(`^${label}:(.*)$`, 'im'))
12  if (line === null) return fallback
13  const n = Number((line[1] ?? '').trim())
14  if (!Number.isInteger(n) || n < 1) throw new Error(`routing.md "${label}:" must be a whole number from 1, not ${JSON.stringify(line[1]?.trim())}`)
15  return n
16}
17
18/** The difficulty at and above which a ticket is reviewed, from routing.md's `Review threshold: <n>` line. */
19export const reviewFrom = (routing: string): number => whole(routing, 'Review threshold', 3)
20
21/** Each pipeline agent run's limits: context tokens and minutes of wall time, from routing.md's `Agent token limit:` and `Agent time limit:` lines. */
22export type Limits = { tokens: number; minutes: number }
23
24export const limitsFrom = (routing: string): Limits => ({ tokens: whole(routing, 'Agent token limit', 500_000), minutes: whole(routing, 'Agent time limit', 120) })
25

1import type { ProcessRunInit, ProcessRunResult } from 'claude-code'
2
3/** The shape of `$.process.run`; the mod passes `(argv, init) => $.process.run(argv, init)`, tests a child_process shim. */
4export type Run = (argv: readonly string[], init?: ProcessRunInit) => Promise<ProcessRunResult>
5
6export type Status = 'new' | 'needs_human' | 'ready' | 'in_progress' | 'pr_open' | 'done'
7
8/** Where an `in_progress` ticket stands; null (a row from before stages) reads as `implement`. */
9export type Stage = 'implement' | 'review' | 'verify' | 'fix'
10
11export type Ticket = {
12  ticket_id: string
13  title: string
14  url: string | null
15  /** The ticket's description plus acceptance criteria, filled by sync. */
16  body: string | null
17  status: Status
18  stage: Stage | null
19  /** 0 while the agents of `stage` are still to be started by the loop's pass; 1 once started. */
20  started: 0 | 1
21  /** The `$.session.id()` of the session whose loop claimed the ticket; it alone moves the ticket's stages on. */
22  owner: string | null
23  /** When the owner's pass last ran, as SQLite `datetime('now')`; stale for 15 minutes, the owner is taken as stopped. */
24  owner_seen: string | null
25  /** 1 once the loop restarted this ticket's vanished verifier or fixer; a second vanishing goes to needs_human. Back to 0 when the ticket leaves in_progress. */
26  restarts: 0 | 1
27  question: string | null
28  difficulty: number | null
29  confidence: number | null
30  model: string | null
31  reason: string | null
32  pr_url: string | null
33  learned: 0 | 1
34}
35
36export const DB = '.claude/pipeline/pipeline.db'
37
38export const SCHEMA = `create table if not exists tickets (
39  ticket_id text primary key,
40  title text not null,
41  url text,
42  body text,
43  status text not null default 'new'
44    check (status in ('new', 'needs_human', 'ready', 'in_progress', 'pr_open', 'done')),
45  stage text check (stage in ('implement', 'review', 'verify', 'fix')),
46  started integer not null default 1 check (started in (0, 1)),
47  owner text,
48  owner_seen text,
49  restarts integer not null default 0 check (restarts in (0, 1)),
50  question text,
51  difficulty integer check (difficulty between 1 and 5),
52  confidence real check (confidence between 0 and 1),
53  model text check (model in ('haiku', 'sonnet', 'opus', 'fable')),
54  reason text,
55  pr_url text,
56  learned integer not null default 0 check (learned in (0, 1))
57);
58create table if not exists turns (
59  ticket_id text not null,
60  role text not null check (role in ('implementer', 'reviewer-spec', 'reviewer-quality', 'verifier', 'fixer')),
61  model text,
62  input_tokens integer not null,
63  output_tokens integer not null,
64  cache_read_tokens integer not null,
65  cache_write_tokens integer not null,
66  duration_ms integer not null
67);
68create table if not exists agent_runs (
69  agent_id text primary key,
70  ticket_id text not null,
71  role text not null check (role in ('implementer', 'reviewer-spec', 'reviewer-quality', 'verifier', 'fixer')),
72  started_at integer not null,
73  requests integer not null default 0,
74  context_tokens integer not null default 0,
75  nudged_token integer not null default 0 check (nudged_token in (0, 1)),
76  nudged_time integer not null default 0 check (nudged_time in (0, 1)),
77  soft_stop_at integer,
78  soft_limit text check (soft_limit in ('token', 'time'))
79);
80drop view if exists role_usage;
81create view role_usage as
82  select ticket_id, role, group_concat(distinct model) as model, count(*) as turns,
83    sum(input_tokens + output_tokens + cache_read_tokens + cache_write_tokens) as tokens,
84    sum(input_tokens) as input_tokens, sum(output_tokens) as output_tokens,
85    sum(cache_read_tokens) as cache_read_tokens, sum(cache_write_tokens) as cache_write_tokens,
86    round(sum(duration_ms) / 60000.0, 1) as minutes
87  from turns group by ticket_id, role;
88drop view if exists ticket_usage;
89create view ticket_usage as
90  select ticket_id, sum(input_tokens + output_tokens + cache_read_tokens + cache_write_tokens) as tokens,
91    round(sum(duration_ms) / 60000.0, 1) as minutes
92  from turns group by ticket_id;`
93
94/** A SQL literal: strings quoted with `'` doubled, so ticket text cannot break out of the statement. */
95export const lit = (v: string | number | null): string => {
96  if (v === null) return 'null'
97  if (typeof v === 'number') {
98    if (!Number.isFinite(v)) throw new Error(`not a finite number: ${v}`)
99    return String(v)
100  }
101  return `'${v.replaceAll("'", "''")}'`
102}
103
104/** Runs one or more statements against `db`; resolves the rows of the last select, empty for writes. */
105export const sql = async (run: Run, db: string, statement: string): Promise<Record<string, unknown>[]> => {
106  // .timeout: the loop and the hooks write the same file from overlapping calls
107  const r = await run(['sqlite3', '-bail', '-json', '-cmd', '.timeout 5000', db, statement])
108  if (r.exitCode !== 0) throw new Error(`sqlite3 exited ${r.exitCode}: ${r.stderr.trim()}`)
109  return r.stdout.trim() === '' ? [] : JSON.parse(r.stdout)
110}
111
112/** Columns added after the first schema, with their definitions as in SCHEMA. */
113const ADDED: Record<string, string> = {
114  body: 'text',
115  stage: `text check (stage in ('implement', 'review', 'verify', 'fix'))`,
116  started: 'integer not null default 1 check (started in (0, 1))',
117  owner: 'text',
118  owner_seen: 'text',
119  restarts: 'integer not null default 0 check (restarts in (0, 1))',
120}
121
122/** Applies SCHEMA, and adds the columns a database made before them is missing. */
123export const migrate = async (run: Run, db: string) => {
124  await sql(run, db, SCHEMA)
125  const have = (await sql(run, db, `select name from pragma_table_info('tickets')`)).map(c => c.name)
126  for (const [name, def] of Object.entries(ADDED)) if (!have.includes(name)) await sql(run, db, `alter table tickets add column ${name} ${def}`)
127}
128

1import { type Run, lit, sql } from '../pipeline-agents/src/store.ts'
2
3/** The headings in tracker.md whose first fenced block is a shell command printing a JSON array. */
4export const WAITING = 'List the tickets waiting to be worked'
5export const DONE = 'List the tickets that are done'
6export const BLOCKERS = 'List the blockers'
7/** The headings whose command takes one value as `$1`: a PR URL, then a tracker ticket id. */
8export const MERGED = 'Check whether a PR is merged'
9export const MARK_DONE = 'Mark a ticket done'
10
11type Row = Record<string, unknown>
12
13const command = (tracker: string, heading: string) => {
14  const section = tracker.split(/^## /m).find(s => s.startsWith(`${heading}\n`))
15  const block = section?.match(/^```[^\n]*\n([\s\S]*?)^```/m)?.[1]
16  if (block === undefined || block.trim() === '') throw new Error(`tracker.md has no command under "## ${heading}"; run /tracker-setup`)
17  return block
18}
19
20/** Runs one tracker.md command; empty output is no rows (`sqlite3 -json` prints nothing for none). */
21const list = async (run: Run, tracker: string, heading: string): Promise<Row[]> => {
22  const r = await run(['sh', '-c', command(tracker, heading)])
23  if (r.exitCode !== 0) throw new Error(`"${heading}" exited ${r.exitCode}: ${r.stderr.trim()}`)
24  const rows: unknown = r.stdout.trim() === '' ? [] : JSON.parse(r.stdout)
25  if (!Array.isArray(rows) || !rows.every(row => typeof row === 'object' && row !== null)) throw new Error(`"${heading}" did not print a JSON array of objects`)
26  return rows
27}
28
29/** Runs one tracker.md command with `value` as its `$1`, so the value is never read as shell text. */
30const call = async (run: Run, cmd: string, heading: string, value: string) => {
31  const r = await run(['sh', '-c', cmd, 'sh', value])
32  if (r.exitCode !== 0) throw new Error(`"${heading}" exited ${r.exitCode} for ${value}: ${r.stderr.trim()}`)
33  return r.stdout.trim()
34}
35
36/** Marks done, in the tracker and the table, every pr_open ticket whose PR has merged, so sync can release its dependents. Resolves the ids marked. */
37export const closeMerged = async (run: Run, db: string, tracker: string): Promise<string[]> => {
38  // both are looked up first, so a tracker.md without them fails even when no PR is open
39  const [merged, markDone] = [command(tracker, MERGED), command(tracker, MARK_DONE)]
40  const open = await sql(run, db, `select ticket_id, pr_url from tickets where status = 'pr_open' order by rowid`)
41  const closed: string[] = []
42  for (const t of open) {
43    const ticket = String(t.ticket_id)
44    if (typeof t.pr_url !== 'string') throw new Error(`${ticket} is pr_open with no pr_url`)
45    const out = await call(run, merged, MERGED, t.pr_url)
46    let answer = ''
47    try {
48      answer = JSON.stringify(JSON.parse(out))
49    } catch {}
50    if (answer !== '{"merged":true}' && answer !== '{"merged":false}') throw new Error(`"${MERGED}" printed ${JSON.stringify(out.slice(0, 200))} for ${t.pr_url}, not {"merged": true|false}`)
51    if (answer === '{"merged":false}') continue
52    await call(run, markDone, MARK_DONE, ticket)
53    await sql(run, db, `update tickets set status = 'done' where ticket_id = ${lit(ticket)} and status = 'pr_open'`)
54    closed.push(ticket)
55  }
56  return closed
57}
58
59const bad = (heading: string, row: Row, field: string): never => {
60  throw new Error(`"${heading}" printed a row with a bad ${field}: ${JSON.stringify(row)}`)
61}
62
63/** Tracker ids are strings or integers; the pipeline stores them as text. */
64const id = (heading: string, row: Row, field = 'id') => {
65  const v = row[field]
66  if ((typeof v === 'string' && v !== '') || Number.isInteger(v)) return String(v)
67  return bad(heading, row, field)
68}
69
70/** Pulls the tracker into the pipeline table, following the three commands in tracker.md, in one transaction. Resolves the ids it marked done. */
71export const sync = async (run: Run, db: string, tracker: string): Promise<string[]> => {
72  const [waiting, done, blockers] = await Promise.all([list(run, tracker, WAITING), list(run, tracker, DONE), list(run, tracker, BLOCKERS)])
73
74  const blocked = new Set<string>()
75  for (const b of blockers) {
76    id(BLOCKERS, b, 'blocked_by')
77    if (![true, false, 0, 1].includes(b.blocker_done as never)) bad(BLOCKERS, b, 'blocker_done')
78    if (!b.blocker_done) blocked.add(id(BLOCKERS, b))
79  }
80
81  const statements = ['begin;']
82  for (const t of waiting) {
83    const ticket = id(WAITING, t)
84    if (typeof t.title !== 'string') bad(WAITING, t, 'title')
85    if (typeof t.url !== 'string' && t.url !== null) bad(WAITING, t, 'url')
86    if (typeof t.body !== 'string' && t.body !== null) bad(WAITING, t, 'body')
87    if (blocked.has(ticket)) continue
88    statements.push(`insert into tickets (ticket_id, title, url, body) values (${lit(ticket)}, ${lit(t.title as string)}, ${lit(t.url as string | null)}, ${lit(t.body as string | null)})
89      on conflict (ticket_id) do update set title = excluded.title, url = excluded.url, body = excluded.body;`)
90  }
91  // a ticket that became blocked leaves until its blockers are done, unless work on it has started
92  for (const ticket of blocked) statements.push(`delete from tickets where ticket_id = ${lit(ticket)} and status in ('new', 'ready');`)
93  const ids = done.map(t => id(DONE, t))
94  for (const ticket of ids) statements.push(`update tickets set status = 'done', restarts = 0 where ticket_id = ${lit(ticket)};`)
95  statements.push('commit;')
96  // read before the write: sqlite3 -json prints one array per statement, so a returning inside the transaction would not parse
97  const marked = ids.length === 0 ? [] : await sql(run, db, `select ticket_id from tickets where status != 'done' and ticket_id in (${ids.map(lit).join(', ')})`)
98  await sql(run, db, statements.join('\n'))
99  return marked.map(r => String(r.ticket_id))
100}
101

1import type { ModelCompleteRequest, ModelCompleteResult } from 'claude-code'
2import { type Run, type Ticket, lit, sql } from '../pipeline-agents/src/store.ts'
3
4/** The shape of `$.model.complete`. */
5export type Complete = (request: ModelCompleteRequest) => Promise<ModelCompleteResult>
6
7/** The project's own guides, `.claude/pipeline/<name>.md`. */
8export type Guides = { readiness: string; rubric: string; routing: string }
9
10/** The models with an implementer agent, `agents/implementer-<model>.md`. */
11export const MODELS = ['sonnet', 'opus', 'fable'] as const
12
13/** The model every reviewer and the verifier run on, whatever the routed model: fable eats tokens, so review never escalates to it. */
14export const REVIEW_MODEL = 'opus'
15
16
17export type Verdict = {
18  ready: boolean
19  question: string | null
20  difficulty: number
21  confidence: number
22  model: (typeof MODELS)[number]
23  reason: string
24}
25
26const system = (g: Guides) => `You triage one ticket for an AI implementer that works without a human.
27Judge readiness against the readiness guide, difficulty against the rubric, and pick the model from the routing table.
28
29<readiness>
30${g.readiness}
31</readiness>
32
33<rubric>
34${g.rubric}
35</rubric>
36
37<routing>
38${g.routing}
39</routing>
40
41Reply with only this JSON object:
42{"ready": true|false, "question": "what a human must answer, when not ready, else null", "difficulty": 1-5, "confidence": 0-1, "model": "sonnet"|"opus"|"fable", "reason": "one or two sentences"}`
43
44/** Reads Haiku's reply; throws when it is not the verdict the prompt asked for. */
45export const parseVerdict = (text: string): Verdict => {
46  const json = text.slice(text.indexOf('{'), text.lastIndexOf('}') + 1)
47  const v = JSON.parse(json) as Verdict
48  const ok =
49    typeof v.ready === 'boolean' &&
50    Number.isInteger(v.difficulty) && v.difficulty >= 1 && v.difficulty <= 5 &&
51    typeof v.confidence === 'number' && v.confidence >= 0 && v.confidence <= 1 &&
52    MODELS.includes(v.model) &&
53    typeof v.reason === 'string' &&
54    (v.ready || (typeof v.question === 'string' && v.question.trim() !== ''))
55  if (!ok) throw new Error(`not a triage verdict: ${text.slice(0, 200)}`)
56  return { ...v, question: v.ready ? null : v.question }
57}
58
59/** Triages one ticket with Haiku and writes the verdict, unless the ticket left `new` meanwhile. */
60export const triage = async (run: Run, complete: Complete, db: string, guides: Guides, t: Pick<Ticket, 'ticket_id' | 'title' | 'url' | 'body'>) => {
61  const r = await complete({
62    model: 'haiku',
63    system: system(guides),
64    prompt: `Ticket ${t.ticket_id}: ${t.title}\n${t.url ?? ''}\n\n${t.body ?? '(no description)'}`,
65    effort: 'low',
66    timeoutMs: 60_000,
67  })
68  if (!r.isAnswered) throw new Error(`triage of ${t.ticket_id}: no answer (${r.reason})`)
69  const v = parseVerdict(r.text)
70  await sql(run, db, `update tickets set
71      status = ${lit(v.ready ? 'ready' : 'needs_human')}, question = ${lit(v.question)},
72      difficulty = ${lit(v.difficulty)}, confidence = ${lit(v.confidence)}, model = ${lit(v.model)}, reason = ${lit(v.reason)}
73    where ticket_id = ${lit(t.ticket_id)} and status = 'new'`)
74  return v
75}
76

1declare module 'claude-code' {
2  interface PluginState {
3    pipeline: {
4      /** Each agent prompt file in agents/ and its mtime, as this session loaded them; kept here to outlive a reload. */
5      agentPrompts: Record<string, number>
6    }
7  }
8}
9