pipeline-agents

1import type { ModelUsage, Register } from 'claude-code'
2import { LOOP_PLUGIN, agentOf } from '../src/agents.ts'
3import { GUARDED, worktreeRefusal } from '../src/guards.ts'
4import { REPORT_TOOL, dbRefusal, onReport, stop } from '../src/result.ts'
5import { type Verdict, decide, markTold, nudgeText, openRun, readRun, recordRequest, softStopText } from '../src/limits.ts'
6import { limitsFrom, readRouting } from '../src/routing.ts'
7import { DB, type Run, migrate } from '../src/store.ts'
8import { recordTurn } from '../src/usage.ts'
9
10// Every hook that must hear the loop's agents lives here, not in pipeline: a plugin never hears the agents it
11// spawned from its own timer (docs/adr/0001). Agents are recognised as spawned by LOOP_PLUGIN.
12export const register: Register = on => {
13  on('session.start', async ($, e, next) => {
14    // awaited before next, so the tool is listed by the implementers' first turn
15    await $.tool.register(REPORT_TOOL)
16    // the limits write agent_runs from an agent's first request, which may come before the loop's first pass migrates
17    try {
18      const root = await $.session.root()
19      const run: Run = (argv, init) => $.process.run(argv, { cwd: root, ...init })
20      if ((await run(['test', '-d', '.claude/pipeline'])).exitCode === 0) await migrate(run, DB)
21    } catch (err) {
22      $.ui.log(`migrating the pipeline database failed: ${(err as Error).message}`)
23    }
24    return next(e)
25  })
26
27  // the limits, per model request of a pipeline agent: its start and context are recorded, and once the soft stop's
28  // grace is spent the request is answered without being sent, so the agent ends. The soft stop itself is the
29  // tool-call hook's. They fail open: an error is logged and the request is sent
30  on('turn.step', async function* ($, e, next) {
31    let hard: string | null = null
32    let count: ((usage: ModelUsage | null) => Promise<unknown>) | null = null
33    try {
34      const agentId = e.agentId
35      const who = agentId === undefined ? null : agentOf((await $.agent.list()).find(a => a.id === agentId), LOOP_PLUGIN)
36      if (agentId !== undefined && who !== null) {
37        const root = await $.session.root()
38        const run: Run = (argv, init) => $.process.run(argv, { cwd: root, ...init })
39        const now = await $.clock.now()
40        const r = await openRun(run, DB, agentId, who, now)
41        count = usage => recordRequest(run, DB, agentId, usage)
42        const v = decide(r, now, limitsFrom(await readRouting(run)))
43        if (v?.kind === 'hard') {
44          $.ui.log(`${who.ticket} ${who.role} hard-stopped at its ${v.limit} limit`)
45          hard = `Stopped: the ${v.limit} limit was reached.`
46        }
47      }
48    } catch (err) {
49      $.ui.log(`applying the agent limits to a model request failed: ${(err as Error).message}`)
50    }
51    if (hard !== null) {
52      yield { kind: 'text', index: 0, text: hard }
53      yield { kind: 'stop', stopReason: 'end_turn', usage: null }
54      return { turnId: e.turnId, index: e.index, answer: hard, toolUses: [], stopReason: 'end_turn', usage: null }
55    }
56    const response = yield* next(e)
57    try {
58      if (count !== null) await count(response.usage)
59    } catch (err) {
60      $.ui.log(`recording an agent's model request failed: ${(err as Error).message}`)
61    }
62    return response
63  })
64
65  // one tool.call hook without a matcher, as the engine allows: a report_result call, then the limits on every other
66  // call of a pipeline agent
67  on('tool.call', async ($, e, next) => {
68    // a pipeline agent's result; refused for any other caller, written only while its ticket is at that agent's stage
69    if (e.tool === `mcp__${$.plugin.name}__${REPORT_TOOL.name}`) {
70      const agent = (await $.agent.list()).find(a => a.id === e.agentId)
71      try {
72        const root = await $.session.root()
73        const run: Run = (argv, init) => $.process.run(argv, { cwd: root, ...init })
74        const r = await onReport(run, LOOP_PLUGIN, agent, e as Record<string, unknown>)
75        $.ui.log(r.log)
76        if (r.deny !== undefined) return { deny: r.deny }
77        return { result: r.changed ? 'Recorded.' : 'Not recorded: the ticket is no longer at your stage.' }
78      } catch (err) {
79        $.ui.log(`report_result failed: ${(err as Error).message}`)
80        return { deny: (err as Error).message }
81      }
82    }
83
84    // the limits: from 80% of a limit a nudge rides on the result, once per limit; at the limit the call is refused.
85    // They fail open: an error is logged and the call runs
86    const agentId = e.agentId
87    if (agentId === undefined) return next(e)
88    let v: Verdict = null
89    try {
90      const who = agentOf((await $.agent.list()).find(a => a.id === agentId), LOOP_PLUGIN)
91      const root = await $.session.root()
92      const run: Run = (argv, init) => $.process.run(argv, { cwd: root, ...init })
93      // the run's first request opens it: none means the turn.step hook failed, and the call is not held on that
94      const r = who === null ? undefined : await readRun(run, DB, agentId)
95      if (who !== null && r !== undefined) {
96        v = decide(r, await $.clock.now(), limitsFrom(await readRouting(run)))
97        // claimed before the call runs: parallel calls of one step run their hooks concurrently, and only one carries it
98        if (v?.kind === 'nudge') {
99          if ((await markTold(run, DB, agentId, { kind: 'nudge', limit: v.limit })).length === 0) v = null
100          else $.ui.log(`${who.ticket} ${who.role} nudged at 80% of its ${v.limit} limit`)
101        } else if (v !== null && r.soft_stop_at === null) {
102          await markTold(run, DB, agentId, { kind: 'soft', limit: v.limit })
103          $.ui.log(`${who.ticket} ${who.role} soft-stopped at its ${v.limit} limit`)
104        }
105      }
106    } catch (err) {
107      $.ui.log(`applying the agent limits to a tool call failed: ${(err as Error).message}`)
108      v = null
109    }
110    if (v === null) return next(e)
111    if (v.kind !== 'nudge') return { deny: softStopText(v.limit) }
112    const done = await next(e)
113    if (done.deny !== undefined) return done
114    return { ...done, context: [...(done.context ?? []), nudgeText(v.limit)] }
115  })
116
117  // a pipeline agent's Bash command that mentions pipeline.db is refused: its result goes through report_result
118  on('tool.call', { tool: 'Bash' }, async ($, e, next) => {
119    if (e.agentId === undefined) return next(e)
120    const deny = dbRefusal((await $.agent.list()).find(a => a.id === e.agentId), LOOP_PLUGIN, e.tool, e as Record<string, unknown>)
121    return deny === undefined ? next(e) : { deny }
122  })
123
124  // the worktree guard: a pipeline agent's Edit, Write or NotebookEdit inside the main checkout, or its state-changing
125  // git command not placed in its own worktree, is refused naming that worktree. It fails closed: a check that throws
126  // refuses the call
127  on('tool.call', { tool: GUARDED }, async ($, e, next) => {
128    if (e.agentId === undefined) return next(e)
129    const agent = (await $.agent.list()).find(a => a.id === e.agentId)
130    if (agentOf(agent, LOOP_PLUGIN) === null) return next(e)
131    let deny: string | undefined
132    try {
133      const home = await $.env.get('HOME')
134      if (home === undefined) throw new Error('HOME is unset')
135      deny = worktreeRefusal(agent, LOOP_PLUGIN, await $.session.root(), home, e.tool, e as Record<string, unknown>)
136    } catch (err) {
137      deny = `The worktree guard could not check this call, so it is refused: ${(err as Error).message}`
138    }
139    return deny === undefined ? next(e) : { deny }
140  })
141
142  // an implementer, verifier or fixer must report before it stops: blocked once, then its ticket goes to a human
143  on('classic.SubagentStop', async ($, e, next) => {
144    const done = await next(e)
145    const agent = (await $.agent.list()).find(a => a.id === e.agent_id)
146    const root = await $.session.root()
147    const s = await stop((argv, init) => $.process.run(argv, { cwd: root, ...init }), DB, LOOP_PLUGIN, agent, e.stop_hook_active)
148    if (s.log !== undefined) $.ui.log(s.log)
149    return s.block === undefined ? done : { ...done, block: s.block }
150  })
151
152  // each finished turn of a pipeline agent or an agent it spawned: its tokens and minutes, per ticket and role, for /pipeline-status
153  on('turn.complete', async ($, e, next) => {
154    if (e.agentId === undefined) return next(e)
155    try {
156      const root = await $.session.root()
157      await recordTurn((argv, init) => $.process.run(argv, { cwd: root, ...init }), DB, LOOP_PLUGIN, await $.agent.list(), e.agentId, e)
158    } catch (err) {
159      $.ui.log(`recording a turn's usage failed: ${(err as Error).message}`)
160    }
161    return next(e)
162  })
163
164  // the loop's agents run unattended: a call that would prompt is allowed; deny rules and an org ceiling still stand
165  on('tool.check', async ($, e, next) => {
166    const verdict = await next(e)
167    if (verdict.decision !== 'ask' || e.agentId === undefined || (e.ceiling !== undefined && e.ceiling !== 'allow')) return verdict
168    const agent = (await $.agent.list()).find(a => a.id === e.agentId)
169    return agentOf(agent, LOOP_PLUGIN) !== null ? { decision: 'allow', reason: `${$.plugin.name} allows the pipeline agents' calls` } : verdict
170  })
171}
172

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 } 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 { AgentInfo } from 'claude-code'
2import { type Role, agentOf } from './agents.ts'
3import { isHardStopped, readRun } from './limits.ts'
4import { DB, type Run, type Stage, type Ticket, lit, sql } from './store.ts'
5import { readRouting, reviewFrom } from './routing.ts'
6
7export const REPORT_TOOL = {
8  name: 'report_result',
9  description:
10    'Reports your result to the pipeline. Call it once, as your last action: status "pr_open" with pr_url once the PR is open, status "reviewed" with findings once the checked findings are posted (verifier only), or status "blocked" with the one question a human must answer.',
11  inputSchema: {
12    type: 'object',
13    properties: {
14      status: { type: 'string', enum: ['pr_open', 'reviewed', 'blocked'] },
15      pr_url: { type: 'string', description: 'The PR url; required for pr_open.' },
16      findings: { type: 'integer', minimum: 0, description: 'How many checked findings the verifier posted; required for reviewed.' },
17      question: { type: 'string', description: 'The one question a human must answer; required for blocked.' },
18    },
19    required: ['status'],
20    additionalProperties: false,
21  },
22  isDeferred: false,
23} as const
24
25/** The stage at which each reporting role's result is written; reviewers do not report. */
26const REPORTS_AT: Partial<Record<Role, Stage>> = { implementer: 'implement', verifier: 'verify', fixer: 'fix' }
27
28/** The statuses each reporting role may send. */
29const SENDS: Partial<Record<Role, readonly string[]>> = { implementer: ['pr_open', 'blocked'], verifier: ['reviewed', 'blocked'], fixer: ['pr_open', 'blocked'] }
30
31/** Updates the ticket only while `in_progress` at `stage` (null reads as implement); resolves whether a row changed. */
32const settle = async (run: Run, db: string, ticketId: string, stage: Stage, set: string) =>
33  (await sql(run, db, `update tickets set ${set} where ticket_id = ${lit(ticketId)} and status = 'in_progress' and coalesce(stage, 'implement') = ${lit(stage)} returning ticket_id`)).length > 0
34
35const text = (v: unknown) => (typeof v === 'string' && v.trim() !== '' ? v.trim() : null)
36
37/** What a report_result call logs: `changed` is whether the row moved, `deny` the refusal the agent reads, `next` the stage the loop's pass must now start. */
38export type Reported = { log: string; changed: boolean; next?: 'review' | 'fix'; deny?: string }
39
40/** Settles the ticket, resolving the changed log, or one saying why nothing changed (`what` is what the agent reported). */
41const apply = async (run: Run, db: string, id: string, stage: Stage, set: string, what: string, log: string, next?: Reported['next']): Promise<Reported> => {
42  if (await settle(run, db, id, stage, set)) return { log, changed: true, next }
43  const [t] = (await sql(run, db, `select status, stage from tickets where ticket_id = ${lit(id)}`)) as Ticket[]
44  const now = t === undefined ? 'missing' : t.status === 'in_progress' ? `in_progress at ${t.stage ?? 'implement'}` : t.status
45  return { log: `${id} reported ${what} but was already ${now}; nothing changed`, changed: false }
46}
47
48/** Records a report_result call; throws the refusal the agent reads. */
49const record = async (run: Run, db: string, plugin: string, agent: AgentInfo | undefined, input: Record<string, unknown>, reviewFrom: number): Promise<Reported> => {
50  const who = agentOf(agent, plugin)
51  if (who === null) throw new Error('report_result is only for the pipeline agents')
52  const stage = REPORTS_AT[who.role]
53  if (stage === undefined) throw new Error('reviewers post their findings as one ticket comment and do not call report_result')
54  const id = who.ticket
55  if (!SENDS[who.role]?.includes(input.status as string)) throw new Error(`the ${who.role} reports status ${SENDS[who.role]?.join(' or ')}, not ${JSON.stringify(input.status)}`)
56
57  if (input.status === 'blocked') {
58    const question = text(input.question)
59    if (question === null) throw new Error('status blocked needs question')
60    return apply(run, db, id, stage, `status = 'needs_human', restarts = 0, question = ${lit(question)}`, 'a question', `${id} needs you: ${question}`)
61  }
62
63  if (input.status === 'reviewed') {
64    const n = input.findings
65    if (!Number.isInteger(n) || (n as number) < 0) throw new Error('status reviewed needs findings, a whole number from 0')
66    if (n === 0) return apply(run, db, id, stage, `status = 'pr_open', restarts = 0`, 'no findings', `${id} passed review: its PR is yours`)
67    return apply(run, db, id, stage, `stage = 'fix', started = 0`, `${n} findings`, `${id} has ${n} review findings to fix`, 'fix')
68  }
69
70  const url = text(input.pr_url)
71  if (url === null) throw new Error('status pr_open needs pr_url')
72  if (who.role === 'implementer') {
73    const [t] = (await sql(run, db, `select difficulty, stage from tickets where ticket_id = ${lit(id)}`)) as Ticket[]
74    // a null stage is a ticket started before review existed: its implementer was not told to open a draft
75    if (t?.stage === 'implement') {
76      if (t.difficulty === null) throw new Error(`${id} is in progress with no difficulty`)
77      if (t.difficulty >= reviewFrom)
78        return apply(run, db, id, stage, `stage = 'review', started = 0, pr_url = ${lit(url)}`, 'a draft PR', `${id} has a draft PR for review: ${url}`, 'review')
79    }
80  }
81  return apply(run, db, id, stage, `status = 'pr_open', restarts = 0, pr_url = ${lit(url)}`, 'a PR', `${id} has a PR: ${url}`)
82}
83
84/**
85 * Handles a report_result call; `reviewFrom` is routing.md's review threshold. Every call resolves one log line:
86 * a refusal carries the reason the agent reads in `deny`.
87 */
88export const report = async (run: Run, db: string, plugin: string, agent: AgentInfo | undefined, input: Record<string, unknown>, reviewFrom: number): Promise<Reported> => {
89  try {
90    return await record(run, db, plugin, agent, input, reviewFrom)
91  } catch (err) {
92    const deny = (err as Error).message
93    return { log: `${agentOf(agent, plugin)?.ticket ?? agent?.description ?? 'unknown agent'} report_result refused: ${deny}`, changed: false, deny }
94  }
95}
96
97/**
98 * A report_result call end to end, as the pipeline-agents plugin handles it. It only records: a stage it moves the
99 * ticket into is left `started = 0` for the loop's pass, since an agent spawned from these hooks would be nested in
100 * this plugin and its own calls would skip them (docs/adr/0001). Resolves one log line and, for a refusal, `deny`.
101 */
102export const onReport = async (run: Run, plugin: string, agent: AgentInfo | undefined, input: Record<string, unknown>): Promise<{ log: string; changed: boolean; deny?: string }> => {
103  const r = await report(run, DB, plugin, agent, input, reviewFrom(await readRouting(run)))
104  return { log: r.log, changed: r.changed, deny: r.deny }
105}
106
107/**
108 * The refusal for a pipeline agent's Bash command that mentions pipeline.db, or undefined to let the call run, so
109 * report_result is the only way an agent records its result. A text match: it stops the habit, not a workaround.
110 */
111export const dbRefusal = (agent: AgentInfo | undefined, plugin: string, tool: string, input: Record<string, unknown>): string | undefined =>
112  tool === 'Bash' && agentOf(agent, plugin) !== null && String(input.command ?? '').includes('pipeline.db')
113    ? `The pipeline agents do not touch pipeline.db: call the ${REPORT_TOOL.name} tool to record your result.`
114    : undefined
115
116export type Stop = { block?: string; log?: string }
117
118/** The stage each role works at, reviewers included. */
119const WORKS_AT: Record<Role, Stage> = { implementer: 'implement', 'reviewer-spec': 'review', 'reviewer-quality': 'review', verifier: 'verify', fixer: 'fix' }
120
121/**
122 * Handles a pipeline agent stopping: one hard-stopped at a limit, or a reviewer soft-stopped at one, sends its
123 * ticket, if still at its stage, to needs_human naming the limit; an agent that must report is blocked once while its ticket is still at its stage,
124 * then the ticket goes to needs_human.
125 */
126export const stop = async (run: Run, db: string, plugin: string, agent: AgentInfo | undefined, isContinued: boolean): Promise<Stop> => {
127  const who = agentOf(agent, plugin)
128  if (who === null || agent === undefined) return {}
129  const limited = await readRun(run, db, agent.id)
130  // a soft-stopped reviewer cannot post its comment, so it ends without its review even inside the grace
131  if (limited !== undefined && (isHardStopped(limited) || (REPORTS_AT[who.role] === undefined && limited.soft_stop_at !== null))) {
132    const question = `The ${who.role} reached its ${limited.soft_limit} limit (Agent ${limited.soft_limit} limit in routing.md) and was stopped without reporting. Check its branch and PR, then send the ticket back to new or set it pr_open.`
133    const moved = await settle(run, db, who.ticket, WORKS_AT[who.role], `status = 'needs_human', restarts = 0, question = ${lit(question)}`)
134    return moved ? { log: `${who.ticket} needs you: its ${who.role} was stopped at its ${limited.soft_limit} limit` } : {}
135  }
136  const stage = REPORTS_AT[who.role]
137  if (stage === undefined) return {}
138  const [open] = await sql(run, db, `select 1 from tickets where ticket_id = ${lit(who.ticket)} and status = 'in_progress' and coalesce(stage, 'implement') = ${lit(stage)}`)
139  if (open === undefined) return {}
140  if (!isContinued) return { block: `Call the ${REPORT_TOOL.name} tool before you stop: status ${SENDS[who.role]?.join(' or ')}, with the field it needs.` }
141  const question = `The ${who.role} stopped without calling report_result. Check its branch and PR, then send the ticket back to new.`
142  return (await settle(run, db, who.ticket, stage, `status = 'needs_human', restarts = 0, question = ${lit(question)}`)) ? { log: `${who.ticket} needs you: its ${who.role} stopped without reporting` } : {}
143}
144

1import type { ModelUsage } from 'claude-code'
2import type { Role } from './agents.ts'
3import type { Limits } from './routing.ts'
4import { type Run, lit, sql } from './store.ts'
5
6export type Limit = 'token' | 'time'
7
8/** One pipeline agent run's row in `agent_runs`; a restarted agent is a new agent, so a new row. */
9export type AgentRun = {
10  agent_id: string
11  ticket_id: string
12  role: Role
13  /** Milliseconds since the epoch at the run's first model request. */
14  started_at: number
15  /** Model requests sent. */
16  requests: number
17  /** The input side (uncached, cache-written, cache-read) of the latest request that reported usage, as Claude Code counts context. */
18  context_tokens: number
19  nudged_token: 0 | 1
20  nudged_time: 0 | 1
21  /** `requests` when the soft stop was reached; null before. */
22  soft_stop_at: number | null
23  soft_limit: Limit | null
24}
25
26/** Model requests a soft-stopped agent still gets to call report_result before its requests stop being sent. */
27const GRACE = 2
28
29export type Verdict = { kind: 'nudge' | 'soft' | 'hard'; limit: Limit } | null
30
31/** Past the soft stop's grace: the run's further requests are answered without being sent. */
32export const isHardStopped = (r: AgentRun) => r.soft_stop_at !== null && r.requests >= r.soft_stop_at + GRACE
33
34/** What the run is told now: nothing, a nudge at 80% of a limit (once per limit), a soft stop at 100%, or a hard stop GRACE requests after it. */
35export const decide = (r: AgentRun, now: number, limits: Limits): Verdict => {
36  if (r.soft_stop_at !== null) {
37    if (r.soft_limit === null) throw new Error(`agent run ${r.agent_id} soft-stopped with no limit`)
38    return { kind: isHardStopped(r) ? 'hard' : 'soft', limit: r.soft_limit }
39  }
40  const used: Record<Limit, number> = { token: r.context_tokens / limits.tokens, time: (now - r.started_at) / (limits.minutes * 60_000) }
41  const over = (['token', 'time'] as const).find(l => used[l] >= 1)
42  if (over !== undefined) return { kind: 'soft', limit: over }
43  const near = (['token', 'time'] as const).find(l => used[l] >= 0.8 && r[`nudged_${l}`] === 0)
44  return near === undefined ? null : { kind: 'nudge', limit: near }
45}
46
47export const nudgeText = (l: Limit) => `You've used 80% of your ${l} limit. Finish the step you're on, push what you have, and report. Don't start new work.`
48
49export const softStopText = (l: Limit) => `Limit reached (${l}): call report_result now with what you have.`
50
51const where = (agentId: string) => `where agent_id = ${lit(agentId)}`
52
53/** The run's row, or undefined before its first model request. */
54export const readRun = async (run: Run, db: string, agentId: string): Promise<AgentRun | undefined> =>
55  ((await sql(run, db, `select * from agent_runs ${where(agentId)}`)) as AgentRun[])[0]
56
57/** The run's row, created at its first model request with `now` as its start; a reload or restarted session reads the one it kept. */
58export const openRun = async (run: Run, db: string, agentId: string, who: { ticket: string; role: Role }, now: number): Promise<AgentRun> => {
59  const [r] = (await sql(
60    run,
61    db,
62    `insert into agent_runs (agent_id, ticket_id, role, started_at) values (${[agentId, who.ticket, who.role, now].map(lit).join(', ')}) on conflict (agent_id) do nothing;
63select * from agent_runs ${where(agentId)}`,
64  )) as AgentRun[]
65  if (r === undefined) throw new Error(`agent run ${agentId} was not written`)
66  return r
67}
68
69/** Counts a sent request; its usage, when it has one, is the run's context now, else the last that had one stands. */
70export const recordRequest = (run: Run, db: string, agentId: string, usage: ModelUsage | null) =>
71  sql(
72    run,
73    db,
74    `update agent_runs set requests = requests + 1, context_tokens = coalesce(${lit(usage === null ? null : usage.input_tokens + usage.cache_creation_input_tokens + usage.cache_read_input_tokens)}, context_tokens) ${where(agentId)}`,
75  )
76
77/**
78 * Records that the run was told a nudge or soft stop, so a nudge is sent once per limit and the soft stop keeps the
79 * request count it came at. A nudge resolves one row only to the call that marked it first: parallel tool calls race.
80 */
81export const markTold = (run: Run, db: string, agentId: string, v: { kind: 'nudge' | 'soft'; limit: Limit }) =>
82  sql(
83    run,
84    db,
85    v.kind === 'nudge'
86      ? `update agent_runs set nudged_${v.limit} = 1 ${where(agentId)} and nudged_${v.limit} = 0 returning agent_id`
87      : `update agent_runs set soft_stop_at = coalesce(soft_stop_at, requests), soft_limit = coalesce(soft_limit, ${lit(v.limit)}) ${where(agentId)}`,
88  )
89

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 { AgentInfo, TurnCompleteInput } from 'claude-code'
2import { agentOf } from './agents.ts'
3import { type Run, lit, sql } from './store.ts'
4
5/** The ticket and role an agent's turns count against: its own, or its nearest ancestor's the loop spawned. */
6const ownerOf = (agents: AgentInfo[], agentId: string, plugin: string) => {
7  const seen = new Set<string>() // a parentId cycle ends the walk instead of looping
8  for (let id: string | undefined = agentId; id !== undefined && !seen.has(id); ) {
9    seen.add(id)
10    const agent = agents.find(a => a.id === id)
11    const who = agentOf(agent, plugin)
12    if (who !== null) return who
13    id = agent?.parentId
14  }
15  return null
16}
17
18/**
19 * Records one finished turn in `turns`, against the ticket and role of the pipeline agent it ran for (the agent
20 * itself, or the pipeline agent beneath which an Agent call or fork spawned it): the tokens and model
21 * `turn.complete` reports (zero and null when it reports none, as on an interrupt) and its duration. Resolves
22 * whether a row was written.
23 */
24export const recordTurn = async (run: Run, db: string, plugin: string, agents: AgentInfo[], agentId: string, e: Pick<TurnCompleteInput, 'usage' | 'durationMs'>): Promise<boolean> => {
25  const who = ownerOf(agents, agentId, plugin)
26  if (who === null) return false
27  const u = e.usage
28  const values = [who.ticket, who.role, u?.model ?? null, u?.input_tokens ?? 0, u?.output_tokens ?? 0, u?.cache_read_input_tokens ?? 0, u?.cache_creation_input_tokens ?? 0, Math.round(e.durationMs)]
29  await sql(run, db, `insert into turns (ticket_id, role, model, input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, duration_ms) values (${values.map(lit).join(', ')})`)
30  return true
31}
32