pr-review-watch

1import type { Register } from 'claude-code'
2
3import { registerBand } from './band'
4import { registerPhase5Context } from './phase5Scan'
5import { registerTracker } from './tracker'
6
7export const register: Register = on => {
8  registerTracker(on)
9  registerPhase5Context(on)
10  registerBand(on)
11}
12

1import type { On } from 'claude-code'
2import { atom, read } from 'claude-code'
3
4import { bandText } from './runs'
5
6/** The row above the prompt: the active review run and hot Phase 5 sentinels. Reads state only. */
7
8const runsAtom = atom({ plugin: 'pr-review-watch', key: 'runs' } as const, {})
9const activeAtom = atom({ plugin: 'pr-review-watch', key: 'active' } as const, null)
10const phase5Atom = atom({ plugin: 'pr-review-watch', key: 'phase5' } as const, [])
11const missingAtom = atom({ plugin: 'pr-review-watch', key: 'missing' } as const, [])
12
13export const registerBand = (on: On): void => {
14  on('ui.render', { component: 'AbovePrompt' }, async ($, e, next) => {
15    if (e.props.hasSurvey) return next(e)
16    const id = await read($, activeAtom)
17    const runs = await read($, runsAtom)
18    const missing = await read($, missingAtom)
19    const run = id === null ? null : (runs[id] ?? null)
20    const text = bandText({ run, phase5: await read($, phase5Atom), missing })
21    if (text === undefined) return next(e)
22    const { Box, Text } = $.ui.resolve(e)
23    return (
24      <Box>
25        <Text color={missing.length > 0 ? 'warning' : 'subtle'}>
26          {text}
27        </Text>
28      </Box>
29    )
30  })
31}
32

1import type { EngineInterface, FsEntry, On, SessionCompactInput, SessionCompactResult } from 'claude-code'
2import { atom, update } from 'claude-code'
3
4import { sentinelDirOf, under } from './config'
5import { isHot, isSentinelName, parseSentinel, SENTINEL_PREFIX, type SentinelText } from './sentinels'
6import { resolveWatch } from './watch'
7
8/**
9 * Phase 5 sentinels on disk: the scan of `<state_dir>/pr-review` the band, the Stop guard and the
10 * per-prompt `PHASE5_PENDING=` context line share. Every failure degrades to "no sentinel". Functions
11 * taking `$` are function declarations: the engine follows `$` past an import only into one.
12 */
13
14type Dollar = EngineInterface
15
16/** A hot sentinel's PR id and the session that armed it (null when unowned). */
17export type HotSentinel = { pr: string; owner: string | null }
18
19const phase5Atom = atom({ plugin: 'pr-review-watch', key: 'phase5' } as const, [])
20
21const noop = (): undefined => undefined
22const byName = (a: FsEntry, b: FsEntry): number => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0)
23
24/** Write time of one sentinel in unix seconds: line 2, else the listed mtime, else the stat mtime. */
25async function sentinelBasis(
26  $: Dollar,
27  path: string,
28  entry: FsEntry,
29  text: SentinelText | undefined,
30): Promise<number | undefined> {
31  if (text !== undefined && text.writtenAt !== null) return text.writtenAt
32  const listed = entry.mtimeMs > 0 ? entry.mtimeMs : undefined
33  const mtimeMs = listed ?? (await $.fs.stat(path).catch(noop))?.mtimeMs
34  return mtimeMs !== undefined && mtimeMs > 0 ? Math.floor(mtimeMs / 1000) : undefined
35}
36
37/** One sentinel when hot; undefined when stale or without any usable age. */
38async function hotSentinel($: Dollar, dir: string, entry: FsEntry, now: number): Promise<HotSentinel | undefined> {
39  const path = `${dir}/${entry.name}`
40  const raw = await $.fs.read(path).catch(noop)
41  const text = typeof raw === 'string' ? parseSentinel(raw) : undefined
42  const basis = await sentinelBasis($, path, entry, text)
43  if (basis === undefined || !isHot(basis, now)) return undefined
44  return { pr: text?.pr ?? entry.name.slice(SENTINEL_PREFIX.length), owner: text?.sessionId ?? null }
45}
46
47/** Hot sentinels in `dir` (absolute), sorted by file name; empty when the folder is absent. */
48export async function scanPhase5($: Dollar, dir: string): Promise<HotSentinel[]> {
49  const entries = await $.fs.list(dir).catch(() => [] as FsEntry[])
50  const named = entries.filter(e => e.kind !== 'dir' && isSentinelName(e.name)).sort(byName)
51  if (named.length === 0) return []
52  const now = Math.floor((await $.clock.now()) / 1000)
53  const hot: HotSentinel[] = []
54  for (const entry of named) {
55    const found = await hotSentinel($, dir, entry, now).catch(noop)
56    if (found !== undefined) hot.push(found)
57  }
58  return hot
59}
60
61/** The prompt context line for hot sentinel PR ids. */
62export const phase5Context = (ids: readonly string[]): string => `PHASE5_PENDING=${ids.join(',')}`
63
64// ---------------------------------------------------------------------------
65// Per-prompt context
66
67/** The line last attached in this conversation; '' forces the next non-empty one on. */
68const context = { lastAttached: '' }
69
70/** This prompt's line when hot sentinels exist and it differs from the one last attached. */
71async function promptLine($: Dollar): Promise<string | undefined> {
72  const w = await resolveWatch($)
73  if (w === undefined) return undefined
74  const ids = (await scanPhase5($, under(w.sentinelRoot, sentinelDirOf(w.cfg)))).map(s => s.pr)
75  await update($, phase5Atom, () => ids).catch(noop)
76  if (ids.length === 0) context.lastAttached = ''
77  const line = ids.length === 0 ? undefined : phase5Context(ids)
78  return line === context.lastAttached ? undefined : line
79}
80
81/** Session-end reasons after which the same process goes on with a fresh conversation. */
82const FRESH_CONVERSATION = new Set<string>(['clear', 'resume'])
83
84/** Whether a compaction result replaced the main conversation (its context line may be gone). */
85const isMainCompaction = (e: SessionCompactInput, result: SessionCompactResult): boolean =>
86  result.skip === undefined && e.agentId === undefined && e.trigger !== 'precompute'
87
88/** Attaches `PHASE5_PENDING=<ids>` to a prompt while sentinels are hot, once per change of the set. */
89export const registerPhase5Context = (on: On): void => {
90  on('prompt.submit', async ($, e, next) => {
91    const line = await promptLine($).catch(noop)
92    if (line === undefined || next.signal.aborted) return next(e)
93    const result = await next({ ...e, context: [...(e.context ?? []), line] })
94    if (result.drop === undefined && (result.context ?? []).includes(line)) context.lastAttached = line
95    return result
96  }).catch((_$, e, next) => next(e))
97
98  on('session.end', async ($, e, next) => {
99    if (FRESH_CONVERSATION.has(e.reason)) context.lastAttached = ''
100    return next(e)
101  }).catch((_$, e, next) => next(e))
102
103  on('session.compact', async ($, e, next) => {
104    const result = await next(e)
105    if (isMainCompaction(e, result)) context.lastAttached = ''
106    return result
107  }).catch((_$, e, next) => next(e))
108}
109

1import type { EngineInterface, FsEntry, On } from 'claude-code'
2import { atom, read, update } from 'claude-code'
3
4import type { AgentRec, LogFailure, Phase, Run } from '../types'
5import {
6  keywordBlock, mentionsBubble, missingFromReview, readReview, reviewFileFor, sidecarIds, stopBlock, verdictPrefix,
7  type ReviewDoc,
8} from './bubbles'
9import { runtimeBaseOf, sentinelDirOf, under } from './config'
10import { identify, parseReturnBlock, prOfRunId, startedAtOfRunId, type Dispatch } from './contract'
11import { appendRows, buildRow, validateRow, type MatrixRow } from './matrixRow'
12import { scanPhase5, type HotSentinel } from './phase5Scan'
13import {
14  applyCompletion, applyDispatch, currentRunId, fixerBubbledIds, markLogged, newAgent, planPhase, recoverReviewers,
15  reviewersSettled, reviewersToLog, settle, withDiskPhase, type Completion,
16} from './runs'
17import { blocksSession } from './sentinels'
18import { resolveWatch, type Watch } from './watch'
19
20/**
21 * Engine glue for pr-review-watch: tracks pr-team / pr-review `Agent` dispatches, folds subagent usage,
22 * appends invocation-matrix rows and guards Stop on undispositioned bubbled ids. Active only in a project
23 * holding `.claude/harness.json` (./watch); its paths and base branch come from that file. Every host
24 * failure degrades to doing nothing; no hook throws into a tool call, a turn or a stop.
25 */
26
27type Dollar = EngineInterface
28type Key = { runId: string; agentId: string }
29/** The fields of an Agent tool result the watch reads; the engine's union is wider. */
30type AgentResult = {
31  status?: string
32  agentId?: string
33  totalTokens?: number
34  totalToolUseCount?: number
35  totalDurationMs?: number
36  resolvedModel?: string
37  content?: readonly { type: string; text?: string }[]
38}
39/** The fields of a tool result the watch reads. */
40type ToolAnswer = { result?: unknown; text?: string; isError?: true; deny?: string }
41type Usage = {
42  input_tokens: number
43  output_tokens: number
44  cache_read_input_tokens: number
45  cache_creation_input_tokens: number
46  model: string
47}
48type TurnEnd = { answer: string; durationMs: number; usage?: Usage }
49
50const runsAtom = atom({ plugin: 'pr-review-watch', key: 'runs' } as const, {})
51const activeAtom = atom({ plugin: 'pr-review-watch', key: 'active' } as const, null)
52const loopsAtom = atom({ plugin: 'pr-review-watch', key: 'loops' } as const, {})
53const failuresAtom = atom({ plugin: 'pr-review-watch', key: 'failures' } as const, [])
54const phase5Atom = atom({ plugin: 'pr-review-watch', key: 'phase5' } as const, [])
55const missingAtom = atom({ plugin: 'pr-review-watch', key: 'missing' } as const, [])
56
57/** A shell command that drives pr-review's script; its result can move a run's phase or bubbles. */
58const PR_REVIEW_SCRIPT = /pr_review\.py/
59const MARKER = '.matrix_owner'
60const GIT_TIMEOUT_MS = 10_000
61const MAX_FAILURES = 20
62export const OWNER_NOTE =
63  '[pr-review-watch] Matrix rows for this run are recorded by the pr-review-watch mod; ' +
64  'skip the manual JSONL append (pr-review Phase 3 Step 4).'
65
66const noop = (): undefined => undefined
67/** Tool calls per subagent loop id, counted for background launches. */
68const toolUses = new Map<string, number>()
69/** Serializes matrix appends: parallel returns never interleave a read-modify-write. */
70let flushTail: Promise<unknown> = Promise.resolve()
71
72// ---------------------------------------------------------------------------
73// Host access
74
75const readText = async ($: Dollar, path: string): Promise<string | undefined> => {
76  const raw = await $.fs.read(path).catch(noop)
77  return typeof raw === 'string' ? raw : undefined
78}
79
80const listDir = ($: Dollar, path: string): Promise<FsEntry[]> => $.fs.list(path).catch(() => [] as FsEntry[])
81
82const gitLine = async ($: Dollar, root: string, args: readonly string[]): Promise<string | undefined> => {
83  const ran = await $.process.run(['git', '-C', root, ...args], { timeoutMs: GIT_TIMEOUT_MS }).catch(noop)
84  return ran !== undefined && ran.exitCode === 0 ? ran.stdout.trim() : undefined
85}
86
87/** The configured base branch, else the branch `origin/HEAD` names, else `main` (resolve_base_branch). */
88const baseBranchFor = async ($: Dollar, w: Watch): Promise<string> => {
89  if (w.cfg.baseBranch !== null) return w.cfg.baseBranch
90  const head = await gitLine($, w.root, ['symbolic-ref', '--quiet', '--short', 'refs/remotes/origin/HEAD'])
91  const slash = head?.indexOf('/') ?? -1
92  return head !== undefined && slash > 0 ? head.slice(slash + 1) : 'main'
93}
94
95/** `<merge-base> -- <branch> (pr-review-watch auto)`; empty (which fails validation) when git fails. */
96const devCommitFor = async ($: Dollar, w: Watch): Promise<string> => {
97  const sha = await gitLine($, w.root, ['merge-base', 'HEAD', await baseBranchFor($, w)])
98  const branch = (await gitLine($, w.root, ['rev-parse', '--abbrev-ref', 'HEAD'])) ?? '?'
99  return sha === undefined ? '' : `${sha} -- ${branch} (pr-review-watch auto)`
100}
101
102// ---------------------------------------------------------------------------
103// Sentinels and review JSON
104
105const scanSentinels = ($: Dollar, w: Watch): Promise<HotSentinel[]> =>
106  scanPhase5($, under(w.sentinelRoot, sentinelDirOf(w.cfg)))
107
108/** The newest review JSON for a PR, unless it is older than `sinceMs`. */
109const loadReviewFor = async ($: Dollar, w: Watch, pr: number, sinceMs: number): Promise<ReviewDoc | undefined> => {
110  const dir = under(w.root, w.cfg.reviewsDir)
111  const hit = reviewFileFor(pr, await listDir($, dir))
112  if (hit === undefined || hit.mtimeMs < sinceMs) return undefined
113  const text = await readText($, `${dir}/${hit.name}`)
114  return text === undefined ? undefined : readReview(text)
115}
116
117/** The run's review JSON: newest for its PR and not older than the run. */
118const loadReview = ($: Dollar, w: Watch, run: Run): Promise<ReviewDoc | undefined> =>
119  run.pr === null ? Promise.resolve(undefined) : loadReviewFor($, w, run.pr, run.startedAt)
120
121const loadSidecar = async ($: Dollar, w: Watch, run: Run): Promise<string[] | null> => {
122  if (run.runtimeDir === null) return null
123  const text = await readText($, `${under(w.root, run.runtimeDir)}/bubble_ups.json`)
124  return text === undefined ? null : sidecarIds(text)
125}
126
127/** What a pr-review run directory says: the `plan.json` phase and the fixer-bubbled ids. */
128type RunDisk = { phase: Phase | undefined; fixerBubbled: string[] }
129const NO_RUN_DISK: RunDisk = { phase: undefined, fixerBubbled: [] }
130
131/**
132 * The pr-review run directory under `artifacts.scratch` (`<NNNN>/current` names the active run,
133 * `<NNNN>/<RUN_ID>/` holds it), only while the PR folder's `current` names this run.
134 */
135const prReviewRunDir = async ($: Dollar, w: Watch, run: Run): Promise<string | undefined> => {
136  if (run.skill !== 'pr-review' || run.pr === null) return undefined
137  const prDir = `${under(w.root, w.cfg.scratchDir)}/${String(run.pr).padStart(4, '0')}`
138  return currentRunId(await readText($, `${prDir}/current`)) === run.runId ? `${prDir}/${run.runId}` : undefined
139}
140
141const fixerReports = async ($: Dollar, dir: string): Promise<string[]> => {
142  const names = (await listDir($, dir)).filter(e => e.kind === 'file' && e.name.endsWith('.json')).map(e => e.name)
143  const texts = await Promise.all(names.sort().map(name => readText($, `${dir}/${name}`)))
144  return texts.filter((t): t is string => t !== undefined)
145}
146
147const loadRunDisk = async ($: Dollar, w: Watch, run: Run): Promise<RunDisk> => {
148  const dir = await prReviewRunDir($, w, run)
149  if (dir === undefined) return NO_RUN_DISK
150  const plan = await readText($, `${dir}/plan.json`)
151  const phase = plan === undefined ? undefined : planPhase(plan)
152  return { phase, fixerBubbled: fixerBubbledIds(await fixerReports($, `${dir}/fixes`)) }
153}
154
155// ---------------------------------------------------------------------------
156// State updates
157
158const runOf = async ($: Dollar, runId: string): Promise<Run | undefined> => (await read($, runsAtom))[runId]
159
160const updateRun = ($: Dollar, runId: string, fn: (run: Run) => Run): Promise<unknown> =>
161  update($, runsAtom, runs => {
162    const cur = runs[runId]
163    return cur === undefined ? runs : { ...runs, [runId]: fn(cur) }
164  })
165
166const recordDispatch = async ($: Dollar, d: Dispatch, meta: { model?: string; effort?: string }): Promise<void> => {
167  const now = await $.clock.now()
168  await update($, runsAtom, runs => applyDispatch(runs, d, newAgent(d, meta), now))
169  await update($, activeAtom, () => d.runId)
170}
171
172/** Writes the pending ownership marker once the orchestrator's plan is in. */
173const claimMatrix = async ($: Dollar, key: Key, signal: string | undefined): Promise<void> => {
174  if (key.agentId !== 'orchestrator_plan' || signal !== 'DONE') return
175  const run = await runOf($, key.runId)
176  const w = await resolveWatch($)
177  if (run === undefined || run.runtimeDir === null || w === undefined) return
178  await $.fs.write(`${under(w.root, run.runtimeDir)}/${MARKER}`, 'pending\n').catch(noop)
179}
180
181const complete = async ($: Dollar, key: Key, c: Completion): Promise<void> => {
182  await updateRun($, key.runId, run => applyCompletion(run, key.agentId, c))
183  await claimMatrix($, key, c.block?.signal)
184}
185
186const textOf = (r: AgentResult): string => (r.content ?? []).map(b => b.text ?? '').join('\n')
187
188const registerLoop = async ($: Dollar, loopId: string, key: Key): Promise<void> => {
189  await update($, loopsAtom, loops => ({ ...loops, [loopId]: [key.runId, key.agentId] as [string, string] }))
190}
191
192/** Folds an Agent tool result: usage now when synchronous, a loop registration when backgrounded. */
193const recordResult = async ($: Dollar, key: Key, r: ToolAnswer): Promise<void> => {
194  const res = (r.deny === undefined && r.isError !== true ? r.result : undefined) as AgentResult | undefined
195  if (res?.status === 'async_launched' && res.agentId !== undefined) return registerLoop($, res.agentId, key)
196  const isDone = res?.status === 'completed'
197  await complete($, key, {
198    tokens: isDone ? (res.totalTokens ?? 0) : 0,
199    toolUses: isDone ? (res.totalToolUseCount ?? 0) : 0,
200    durationMs: isDone ? (res.totalDurationMs ?? 0) : 0,
201    model: res?.resolvedModel ?? null,
202    block: isDone ? parseReturnBlock(r.text ?? textOf(res)) : undefined,
203    isMeasured: isDone,
204  })
205}
206
207const usageTokens = (u: Usage | undefined): number =>
208  u === undefined ? 0 : u.input_tokens + u.output_tokens + u.cache_read_input_tokens + u.cache_creation_input_tokens
209
210/** A background subagent's turn ended: its usage, tool count and report complete the agent. */
211const recordLoopEnd = async ($: Dollar, loopId: string, e: TurnEnd): Promise<void> => {
212  const used = toolUses.get(loopId) ?? 0
213  toolUses.delete(loopId)
214  const hit = (await read($, loopsAtom))[loopId]
215  if (hit === undefined) return
216  await update($, loopsAtom, loops => Object.fromEntries(Object.entries(loops).filter(([k]) => k !== loopId)))
217  await complete($, { runId: hit[0], agentId: hit[1] }, {
218    tokens: usageTokens(e.usage), toolUses: used, durationMs: e.durationMs, model: e.usage?.model ?? null,
219    block: parseReturnBlock(e.answer), isMeasured: e.usage !== undefined,
220  })
221}
222
223// ---------------------------------------------------------------------------
224// Matrix append
225
226const recordFailures = async ($: Dollar, failures: readonly LogFailure[]): Promise<void> => {
227  if (failures.length === 0) return
228  await update($, failuresAtom, prev => [...prev, ...failures].slice(-MAX_FAILURES)).catch(noop)
229  const first = failures[0]
230  const text = `pr-review-watch: ${failures.length} matrix row(s) not appended (${first?.cell}: ${first?.reason})`
231  await Promise.resolve($.ui.toast(text)).catch(noop)
232}
233
234/**
235 * The matrix text to append to: '' when the file is missing, undefined when it exists (or its existence
236 * is unknown) but cannot be read, so a whole-file write would drop its rows.
237 */
238const readMatrix = async ($: Dollar, path: string): Promise<string | undefined> => {
239  const isThere = await $.fs.exists(path).catch(() => true)
240  return isThere ? readText($, path) : ''
241}
242
243/**
244 * Appends valid rows; returns the failures (invalid rows, or every row when the matrix cannot be read or
245 * the write is refused). An unreadable matrix is never overwritten.
246 */
247const writeRows = async ($: Dollar, w: Watch, rows: readonly MatrixRow[], now: number): Promise<LogFailure[]> => {
248  const checked = rows.map(r => ({ r, why: validateRow(r) }))
249  const good = checked.filter(x => x.why.length === 0).map(x => x.r)
250  const failures = checked.filter(x => x.why.length > 0).map(x => ({ cell: x.r.cell, reason: x.why.join('; '), at: now }))
251  if (good.length === 0) return failures
252  const path = under(w.root, w.cfg.matrixPath)
253  const existing = await readMatrix($, path)
254  const isWritten =
255    existing !== undefined && (await $.fs.write(path, appendRows(existing, good)).then(() => true, () => false))
256  if (isWritten) return failures
257  const reason = existing === undefined ? 'matrix unreadable; not overwritten' : 'matrix write refused'
258  return [...failures, ...good.map(r => ({ cell: r.cell, reason, at: now }))]
259}
260
261/**
262 * Claims the run's unlogged reviewers: marks them logged inside one atom update and returns them, so an
263 * overlapping refresh finds nothing left to claim and never appends the same rows again.
264 */
265const claimReviewers = async ($: Dollar, runId: string, now: number): Promise<AgentRec[]> => {
266  let claimed: AgentRec[] = []
267  await updateRun($, runId, r => {
268    claimed = reviewersToLog(r)
269    return claimed.length === 0 ? r : markLogged(r, claimed.map(a => a.id), now)
270  })
271  return claimed
272}
273
274const flushNow = async ($: Dollar, w: Watch, runId: string): Promise<void> => {
275  const run = await runOf($, runId)
276  const prefix = verdictPrefix(run?.verdict ?? null)
277  if (run === undefined || run.pr === null || prefix === undefined || !reviewersSettled(run)) return
278  const now = await $.clock.now()
279  const agents = await claimReviewers($, runId, now)
280  if (agents.length === 0) return
281  const devCommit = await devCommitFor($, w)
282  const pr = run.pr
283  const rows = agents.map(agent => buildRow({ pr, agent, verdictPrefix: prefix, devCommit }))
284  await recordFailures($, await writeRows($, w, rows, now))
285  if (run.runtimeDir === null) return
286  const stamp = `logged ${new Date(now).toISOString()}\n`
287  await $.fs.write(`${under(w.root, run.runtimeDir)}/${MARKER}`, stamp).catch(noop)
288}
289
290const flush = ($: Dollar, w: Watch, runId: string): Promise<unknown> => {
291  flushTail = flushTail.then(() => flushNow($, w, runId)).catch(noop)
292  return flushTail
293}
294
295// ---------------------------------------------------------------------------
296// Refresh, recovery, stop
297
298/** Re-reads disk for the active run: sentinels, verdict, bubbled set, missing ids; flushes rows. */
299const refreshWith = async ($: Dollar, w: Watch): Promise<void> => {
300  const phase5 = (await scanSentinels($, w).catch(() => [] as HotSentinel[])).map(s => s.pr)
301  await update($, phase5Atom, () => phase5)
302  const id = await read($, activeAtom)
303  const run = id === null ? undefined : await runOf($, id)
304  if (run === undefined) {
305    await update($, missingAtom, () => [])
306    return
307  }
308  const review = await loadReview($, w, run)
309  const sidecar = await loadSidecar($, w, run)
310  const disk = await loadRunDisk($, w, run).catch(() => NO_RUN_DISK)
311  await updateRun($, run.runId, r => withDiskPhase(settle(r, review, sidecar, disk.fixerBubbled), disk.phase))
312  const settled = await runOf($, run.runId)
313  const tracked = settled === undefined || settled.bubbledFrom === 'review_json' ? [] : settled.bubbled
314  await update($, missingAtom, () => missingFromReview(tracked, review))
315  if (review !== undefined) await flush($, w, run.runId)
316}
317
318/** `refreshWith` for the session's project; nothing outside a harness project. */
319const refresh = async ($: Dollar): Promise<void> => {
320  const w = await resolveWatch($)
321  if (w !== undefined) await refreshWith($, w)
322}
323
324/**
325 * A run rebuilt from its runtime dir: plan reviewers joined with terminals rows. Its start is the run id's
326 * stamp, so an older review JSON for the same PR is never taken for its verdict; no stamp, no recovery.
327 */
328const recoveredRun = async ($: Dollar, w: Watch, runId: string): Promise<Run | undefined> => {
329  const runtimeDir = `${runtimeBaseOf(w.cfg)}/${runId}`
330  const dir = under(w.root, runtimeDir)
331  const plan = await readText($, `${dir}/dispatch_plan_review.json`)
332  const pr = prOfRunId(runId)
333  const startedAt = startedAtOfRunId(runId)
334  if (plan === undefined || pr === null || startedAt === null) return undefined
335  const agents = recoverReviewers(plan, (await readText($, `${dir}/terminals.jsonl`)) ?? '')
336  return {
337    runId, skill: 'pr-team', pr, phase: 'consolidate', runtimeDir,
338    agents: Object.fromEntries(agents.map(a => [a.id, a])), bubbled: [], bubbledFrom: 'none', verdict: null,
339    startedAt,
340  }
341}
342
343/** Rows for a pr-team run a previous session claimed but never logged. */
344const recoverRun = async ($: Dollar, w: Watch, runId: string): Promise<void> => {
345  const run = await recoveredRun($, w, runId)
346  if (run === undefined) return
347  await update($, runsAtom, runs => (runs[runId] === undefined ? { ...runs, [runId]: run } : runs))
348  const review = await loadReview($, w, run)
349  if (review === undefined) return
350  await updateRun($, runId, r => settle(r, review, null))
351  await flush($, w, runId)
352}
353
354const recoverPending = async ($: Dollar): Promise<void> => {
355  const w = await resolveWatch($)
356  if (w === undefined) return
357  const known = await read($, runsAtom)
358  const base = under(w.root, runtimeBaseOf(w.cfg))
359  for (const entry of await listDir($, base)) {
360    if (entry.kind !== 'dir' || known[entry.name] !== undefined) continue
361    const marker = await readText($, `${base}/${entry.name}/${MARKER}`)
362    if (marker?.startsWith('pending')) await recoverRun($, w, entry.name).catch(noop)
363  }
364}
365
366const activeRun = async ($: Dollar): Promise<Run | undefined> => {
367  const id = await read($, activeAtom)
368  return id === null ? undefined : runOf($, id)
369}
370
371type PrReview = [number, ReviewDoc | undefined]
372
373/**
374 * The active review for the keyword check: the tracked run's PR, else that of the first hot sentinel
375 * blocking the stopping session (`blocksSession`; another session's sentinel is not this one's review).
376 */
377const keywordReview = async (
378  $: Dollar,
379  w: Watch,
380  run: Run | undefined,
381  sessionId: string | undefined,
382): Promise<PrReview | undefined> => {
383  if (run !== undefined && run.pr !== null) return [run.pr, await loadReview($, w, run)]
384  const hot = await scanSentinels($, w).catch(() => [] as HotSentinel[])
385  const pr = Number(hot.find(s => blocksSession(s.owner, sessionId))?.pr)
386  return Number.isInteger(pr) && pr >= 1 ? [pr, await loadReviewFor($, w, pr, 0)] : undefined
387}
388
389type StopInput = { stop_hook_active: boolean; last_assistant_message?: string; session_id?: string }
390
391/** Tracked bubbled ids out of the review JSON (pr-team), then the "bubble" keyword rule (any review). */
392const stopReason = async ($: Dollar, w: Watch, e: StopInput): Promise<string | undefined> => {
393  const run = await activeRun($)
394  const missing = await read($, missingAtom)
395  const tracked =
396    run?.skill === 'pr-team' ? stopBlock(run.pr, missing, run.bubbledFrom, e.stop_hook_active) : undefined
397  if (tracked !== undefined || !mentionsBubble(e.last_assistant_message)) return tracked
398  const hit = await keywordReview($, w, run, e.session_id)
399  return hit === undefined ? undefined : keywordBlock(hit[0], e.last_assistant_message, hit[1], e.stop_hook_active)
400}
401
402/** The Stop block the watch owes, or undefined; nothing outside a harness project. */
403const stopBlockFor = async ($: Dollar, e: StopInput): Promise<string | undefined> => {
404  const w = await resolveWatch($)
405  if (w === undefined) return undefined
406  await refreshWith($, w).catch(noop)
407  return stopReason($, w, e)
408}
409
410// ---------------------------------------------------------------------------
411// Hooks
412
413/** A shell call that ran pr-review's script refreshes, so the band moves within a turn. */
414const afterShell = async ($: Dollar, command: string | undefined): Promise<void> => {
415  if (PR_REVIEW_SCRIPT.test(command ?? '')) await refresh($).catch(noop)
416}
417
418/** Whether the mod holds a run it can log rows for: recorded in state, with a PR. */
419const ownsRows = async ($: Dollar, runId: string): Promise<boolean> =>
420  ((await runOf($, runId).catch(noop))?.pr ?? null) !== null
421
422/** The tracked identity of a dispatch in a harness project, with that project; else undefined. */
423const trackedDispatch = async (
424  $: Dollar,
425  subagentType: string | undefined,
426  prompt: string,
427): Promise<[Dispatch, Watch] | undefined> => {
428  if (identify(subagentType, prompt, '') === undefined) return undefined
429  const w = await resolveWatch($).catch(noop)
430  const d = w === undefined ? undefined : identify(subagentType, prompt, runtimeBaseOf(w.cfg))
431  return w === undefined || d === undefined ? undefined : [d, w]
432}
433
434/** Tool-call hooks: tracked Agent dispatches, per-loop tool counts and shell refreshes. */
435const registerToolHooks = (on: On): void => {
436  on('tool.call', { tool: 'Agent' }, async ($, e, next) => {
437    const hit = await trackedDispatch($, e.subagent_type, e.prompt)
438    if (hit === undefined) return next(e)
439    const [d, w] = hit
440    await recordDispatch($, d, { model: e.model, effort: e.effort }).catch(noop)
441    const r = await next(e)
442    await recordResult($, { runId: d.runId, agentId: d.agentId }, r).catch(noop)
443    await refreshWith($, w).catch(noop)
444    // Refusals inline (not a helper) so the checker narrows `r` to a result that takes context.
445    if (d.skill !== 'pr-review' || d.role !== 'reviewer' || r.deny !== undefined || r.isError === true) return r
446    if (!(await ownsRows($, d.runId))) return r
447    return { ...r, context: [...(r.context ?? []), OWNER_NOTE] }
448  })
449
450  on('tool.call', ($, e, next) => {
451    if (e.agentId !== undefined) toolUses.set(e.agentId, (toolUses.get(e.agentId) ?? 0) + 1)
452    return next(e)
453  })
454
455  on('tool.call', { tool: 'Bash' }, async ($, e, next) => {
456    const r = await next(e)
457    await afterShell($, e.command)
458    return r
459  })
460
461  on('tool.call', { tool: 'PowerShell' }, async ($, e, next) => {
462    const r = await next(e)
463    await afterShell($, e.command)
464    return r
465  })
466}
467
468/** Turn, session and Stop hooks: loop usage, recovery, refreshes and the Stop guard. */
469const registerLifecycleHooks = (on: On): void => {
470  on('turn.complete', async ($, e, next) => {
471    const r = await next(e)
472    if (e.agentId !== undefined) await recordLoopEnd($, e.agentId, e).catch(noop)
473    await refresh($).catch(noop)
474    return r
475  })
476
477  on('session.start', async ($, e, next) => {
478    const r = await next(e)
479    await recoverPending($).catch(noop)
480    await refresh($).catch(noop)
481    return r
482  })
483
484  on('classic.Stop', async ($, e, next) => {
485    const r = await next(e)
486    if (r.block !== undefined) return r
487    const block = await stopBlockFor($, e).catch(noop)
488    return block === undefined ? r : { ...r, block }
489  })
490}
491
492export const registerTracker = (on: On): void => {
493  registerToolHooks(on)
494  registerLifecycleHooks(on)
495}
496

1import type { AgentRec, Counts, Phase, Run } from '../types'
2import { allDispositioned, type ReviewDoc } from './bubbles'
3import type { Dispatch, ReturnBlock } from './contract'
4
5/** Pure state transitions over tracked runs, plus the band text. No `$`. */
6
7/** What one subagent turn (or a synchronous Agent result) contributes. */
8export type Completion = {
9  tokens: number
10  toolUses: number
11  durationMs: number
12  model: string | null
13  block: ReturnBlock | undefined
14  isMeasured: boolean
15}
16
17const PHASE_LABEL: Readonly<Record<Phase, string>> = {
18  plan: 'planning', review: 'reviewers', consolidate: 'consolidating', 'fix-plan': 'fix planning',
19  fix: 'fixers', closeout: 'closeout', done: 'done',
20}
21
22export const newAgent = (d: Dispatch, meta: { model?: string; effort?: string }): AgentRec => ({
23  id: d.agentId, role: d.role, model: meta.model ?? null, effort: meta.effort ?? null,
24  tokens: 0, toolUses: 0, durationMs: 0, isMeasured: false, signal: null, counts: null, loggedAt: null,
25})
26
27const newRun = (d: Dispatch, now: number): Run => ({
28  runId: d.runId, skill: d.skill, pr: d.pr, phase: d.phase, runtimeDir: d.runtimeDir,
29  agents: {}, bubbled: [], bubbledFrom: 'none', verdict: null, startedAt: now,
30})
31
32/** Records a dispatch: the run is created on first sight, the phase follows the dispatch. */
33export const applyDispatch = (
34  runs: Readonly<Record<string, Run>>,
35  d: Dispatch,
36  rec: AgentRec,
37  now: number,
38): Record<string, Run> => {
39  const run = runs[d.runId] ?? newRun(d, now)
40  const agents = { ...run.agents, [d.agentId]: rec }
41  const next: Run = { ...run, pr: run.pr ?? d.pr, runtimeDir: run.runtimeDir ?? d.runtimeDir, phase: d.phase, agents }
42  return { ...runs, [d.runId]: next }
43}
44
45const foldAgent = (a: AgentRec, c: Completion): AgentRec => ({
46  ...a,
47  tokens: a.tokens + c.tokens,
48  toolUses: a.toolUses + c.toolUses,
49  durationMs: a.durationMs + c.durationMs,
50  model: c.model ?? a.model,
51  isMeasured: a.isMeasured || c.isMeasured,
52  signal: c.block?.signal ?? 'NO_BLOCK',
53  counts: c.block?.counts ?? a.counts,
54})
55
56/** The closer's DONE ends the run; its BUBBLED list becomes the run's bubbled set. */
57const applyCloser = (run: Run, block: ReturnBlock): Run => {
58  const closed: Run = block.signal === 'DONE' ? { ...run, phase: 'done' } : run
59  return block.bubbled === null ? closed : { ...closed, bubbled: block.bubbled, bubbledFrom: 'closer' }
60}
61
62export const applyCompletion = (run: Run, agentId: string, c: Completion): Run => {
63  const prev = run.agents[agentId]
64  if (prev === undefined) return run
65  const next: Run = { ...run, agents: { ...run.agents, [agentId]: foldAgent(prev, c) } }
66  return agentId === 'closer' && c.block !== undefined ? applyCloser(next, c.block) : next
67}
68
69const withBubbled = (run: Run, ids: string[], from: 'bubble_ups' | 'review_json'): Run =>
70  run.bubbledFrom === 'closer' ? run : { ...run, bubbled: ids, bubbledFrom: from }
71
72/** pr-review's bubbled set: the review's `bubbles`, plus fixer-bubbled ids no resolution has closed. */
73export const prReviewBubbled = (review: ReviewDoc | undefined, fixerBubbled: readonly string[]): string[] => {
74  const closed = review?.closed ?? []
75  return [...new Set([...(review?.bubbles ?? []), ...fixerBubbled.filter(id => !closed.includes(id))])]
76}
77
78/**
79 * Folds what disk says: verdict, bubbled set (closer > sidecar > pr-review's review JSON and fixer
80 * reports), completion. `fixerBubbled`: ids bubbled in a pr-review run's `fixes/*.json`.
81 */
82export const settle = (
83  run: Run,
84  review: ReviewDoc | undefined,
85  sidecar: string[] | null,
86  fixerBubbled: readonly string[] = [],
87): Run => {
88  const isPrReview = run.skill === 'pr-review'
89  if (review === undefined && sidecar === null && !(isPrReview && fixerBubbled.length > 0)) return run
90  const base: Run = review === undefined ? run : { ...run, verdict: review.verdict }
91  const bubbled =
92    sidecar !== null ? withBubbled(base, sidecar, 'bubble_ups')
93      : isPrReview ? withBubbled(base, prReviewBubbled(review, fixerBubbled), 'review_json')
94        : base
95  return review !== undefined && allDispositioned(review) ? { ...bubbled, phase: 'done' } : bubbled
96}
97
98/** The phase a pr-review run's `plan.json` records wins over the dispatch-derived one; none keeps it. */
99export const withDiskPhase = (run: Run, phase: Phase | undefined): Run =>
100  phase === undefined || phase === run.phase ? run : { ...run, phase }
101
102const reviewers = (run: Run): AgentRec[] => Object.values(run.agents).filter(a => a.role === 'reviewer')
103
104export const reviewersSettled = (run: Run): boolean => reviewers(run).every(a => a.signal !== null)
105
106export const reviewersToLog = (run: Run): AgentRec[] =>
107  reviewers(run).filter(a => a.signal !== null && a.loggedAt === null)
108
109export const markLogged = (run: Run, ids: readonly string[], now: number): Run => ({
110  ...run,
111  agents: Object.fromEntries(
112    Object.entries(run.agents).map(([k, a]) => [k, ids.includes(k) ? { ...a, loggedAt: now } : a]),
113  ),
114})
115
116const isRecord = (v: unknown): v is Record<string, unknown> =>
117  typeof v === 'object' && v !== null && !Array.isArray(v)
118const str = (v: unknown): string | null => (typeof v === 'string' ? v : null)
119
120const parseJson = (text: string): unknown => {
121  try {
122    return JSON.parse(text)
123  } catch {
124    return undefined
125  }
126}
127
128/** Reviewer terminals rows by agent id. */
129const terminalsByAgent = (text: string): Map<string, Record<string, unknown>> => {
130  const out = new Map<string, Record<string, unknown>>()
131  for (const line of text.split(/\r?\n/)) {
132    const row = parseJson(line)
133    if (isRecord(row) && row.agent_role === 'reviewer' && typeof row.agent_id === 'string') out.set(row.agent_id, row)
134  }
135  return out
136}
137
138const countsOf = (v: unknown): Counts | null => {
139  if (!isRecord(v)) return null
140  const n = (k: string): number => (typeof v[k] === 'number' ? (v[k] as number) : 0)
141  return { findings: n('findings'), critical: n('critical'), important: n('important'), minor: n('minor') }
142}
143
144const recovered = (agent: Record<string, unknown>, row: Record<string, unknown>): AgentRec => ({
145  id: String(agent.id), role: 'reviewer', model: str(agent.model), effort: str(agent.effort),
146  tokens: 0, toolUses: 0, durationMs: 0, isMeasured: false,
147  signal: str(row.signal) ?? 'NO_BLOCK', counts: countsOf(row.counts), loggedAt: null,
148})
149
150/** Reviewers rebuilt after a restart: plan `agents[]` that have a reviewer terminals row. */
151export const recoverReviewers = (planText: string, terminalsText: string): AgentRec[] => {
152  const plan = parseJson(planText)
153  const agents = isRecord(plan) && Array.isArray(plan.agents) ? plan.agents.filter(isRecord) : []
154  const rows = terminalsByAgent(terminalsText)
155  return agents.flatMap(a => {
156    const row = typeof a.id === 'string' ? rows.get(a.id) : undefined
157    return row === undefined ? [] : [recovered(a, row)]
158  })
159}
160
161const isPhase = (v: unknown): v is Phase => typeof v === 'string' && Object.hasOwn(PHASE_LABEL, v)
162
163/** A pr-review `plan.json`'s `phase`; undefined when absent, unknown or the text is not JSON. */
164export const planPhase = (planText: string): Phase | undefined => {
165  const plan = parseJson(planText)
166  return isRecord(plan) && isPhase(plan.phase) ? plan.phase : undefined
167}
168
169/** The `current` run id of a pr-review PR folder: its first line, trimmed; undefined when blank. */
170export const currentRunId = (text: string | undefined): string | undefined =>
171  text?.split(/\r?\n/)[0]?.trim() || undefined
172
173/** Ids bubbled across pr-review fixer reports (`fixes/*.json` `bubbled[].id`), first-seen order. */
174export const fixerBubbledIds = (reportTexts: readonly string[]): string[] => {
175  const ids = reportTexts.flatMap(text => {
176    const doc = parseJson(text)
177    const bubbled = isRecord(doc) && Array.isArray(doc.bubbled) ? doc.bubbled : []
178    return bubbled.flatMap(b => (isRecord(b) && typeof b.id === 'string' ? [b.id] : []))
179  })
180  return [...new Set(ids)]
181}
182
183const progress = (run: Run, role: string): string => {
184  const agents = Object.values(run.agents).filter(a => a.role === role)
185  return `${agents.filter(a => a.signal !== null).length}/${agents.length}`
186}
187
188const phaseLabel = (run: Run): string => {
189  if (run.phase === 'review') return `reviewers ${progress(run, 'reviewer')}`
190  if (run.phase === 'fix') return `fixers ${progress(run, 'fixer')}`
191  return PHASE_LABEL[run.phase]
192}
193
194/** The run's phase, then the Phase 5 marker when its PR's sentinel is hot. */
195const phaseText = (run: Run, isPhase5: boolean): string =>
196  isPhase5 ? `${phaseLabel(run)} · Phase 5 pending` : phaseLabel(run)
197
198const runText = (run: Run, phase5: readonly string[], missing: readonly string[]): string => {
199  const parts = [`PR #${run.pr ?? '?'}`, phaseText(run, phase5.includes(String(run.pr)))]
200  if (run.bubbled.length > 0) parts.push(`${run.bubbled.length} bubbled`)
201  if (missing.length > 0) parts.push(`bubbles[] missing ${missing.join(',')}`)
202  return parts.join(' · ')
203}
204
205export type BandInput = { run: Run | null; phase5: readonly string[]; missing: readonly string[] }
206
207/** One band line: the active run (unless done), then any other hot sentinel; undefined when empty. */
208export const bandText = (b: BandInput): string | undefined => {
209  const run = b.run !== null && b.run.phase !== 'done' ? b.run : null
210  const others = b.phase5
211    .filter(pr => run === null || pr !== String(run.pr))
212    .map(pr => `PR #${pr} · Phase 5 pending`)
213  const lines = run === null ? others : [runText(run, b.phase5, b.missing), ...others]
214  return lines.length === 0 ? undefined : lines.join(' | ')
215}
216

1/**
2 * The harness keys the watch reads from `<project>/.claude/harness.json`, merged over copies of the harness
3 * plugin's `config/defaults.json` values: an installed mod sits in its own plugin cache folder and cannot
4 * read that file. `tests/mods/test_pr_review_watch_config_mirror.py` pins the copies. Pure: no `$`.
5 */
6
7export const DEFAULT_STATE_DIR = '.claude/state/harness'
8export const DEFAULT_MATRIX_PATH = 'docs/_harness/pr_review_invocation_test_matrix.jsonl'
9export const DEFAULT_REVIEWS = 'docs/PR_REVIEWS'
10export const DEFAULT_SCRATCH = 'docs/_scratch/pr_review'
11export const DEFAULT_BASE_BRANCH: string | null = null
12
13/** Project-relative POSIX paths and the configured base branch (null: resolve it from git). */
14export type WatchConfig = {
15  stateDir: string
16  matrixPath: string
17  reviewsDir: string
18  scratchDir: string
19  baseBranch: string | null
20}
21
22type Json = Readonly<Record<string, unknown>>
23
24const isObject = (v: unknown): v is Json => typeof v === 'object' && v !== null && !Array.isArray(v)
25
26const clean = (p: string): string => p.trim().replace(/\\/g, '/').replace(/\/+$/, '')
27
28/** A non-blank string value, cleaned; else the fallback. */
29const pathOr = (v: unknown, fallback: string): string =>
30  typeof v === 'string' && clean(v) !== '' ? clean(v) : fallback
31
32const parse = (text: string): Json => {
33  try {
34    const data: unknown = JSON.parse(text)
35    return isObject(data) ? data : {}
36  } catch {
37    return {}
38  }
39}
40
41/**
42 * The watch's view of a `harness.json` text: each key it reads replaces its default, `artifacts` merges
43 * key by key. Malformed JSON or a wrong-typed value keeps the default (the harness hooks report it).
44 */
45export const readConfig = (text: string): WatchConfig => {
46  const data = parse(text)
47  const artifacts = isObject(data.artifacts) ? data.artifacts : {}
48  const base = data.base_branch
49  return {
50    stateDir: pathOr(data.state_dir, DEFAULT_STATE_DIR),
51    matrixPath: pathOr(data.matrix_path, DEFAULT_MATRIX_PATH),
52    reviewsDir: pathOr(artifacts.reviews, DEFAULT_REVIEWS),
53    scratchDir: pathOr(artifacts.scratch, DEFAULT_SCRATCH),
54    baseBranch: typeof base === 'string' && base.trim() !== '' ? base.trim() : DEFAULT_BASE_BRANCH,
55  }
56}
57
58/**
59 * A configured path against the project root: a relative one is joined to it, an absolute one (POSIX,
60 * drive or UNC) is kept, as harness_lib's `cfg.path()` does.
61 */
62export const under = (root: string, path: string): string =>
63  /^([A-Za-z]:)?[\\/]/.test(path) ? path : `${root}/${path}`
64
65/** `<state_dir>/pr-review`: where Phase 5 sentinels live. */
66export const sentinelDirOf = (c: WatchConfig): string => `${c.stateDir}/pr-review`
67
68/** `<state_dir>/pr-team/_runtime`: the pr-team run folders. */
69export const runtimeBaseOf = (c: WatchConfig): string => `${c.stateDir}/pr-team/_runtime`
70

1/** The Phase 5 sentinel rule of the harness plugin's `hooks/check_phase5_sentinel.py`. Pure: no `$`. */
2
3/** Mirrors `MAX_AGE_SECONDS` in check_phase5_sentinel.py. */
4export const MAX_AGE_SECONDS = 86400
5export const SENTINEL_PREFIX = '.phase5_pending_'
6
7export type SentinelText = { pr: string | null; writtenAt: number | null; sessionId: string | null }
8
9const orNull = (s: string): string | null => (s === '' ? null : s)
10
11/**
12 * Line 1 the PR id, line 2 an optional unix-int write time, line 3 the optional id of the session
13 * that armed it (CRLF tolerated).
14 */
15export const parseSentinel = (text: string): SentinelText => {
16  const lines = text.split(/\r?\n/)
17  const stamp = (lines[1] ?? '').trim()
18  return {
19    pr: orNull((lines[0] ?? '').trim()),
20    writtenAt: /^\d+$/.test(stamp) ? Number(stamp) : null,
21    sessionId: orNull((lines[2] ?? '').trim()),
22  }
23}
24
25export const isHot = (basisSeconds: number, nowSeconds: number): boolean =>
26  nowSeconds - basisSeconds < MAX_AGE_SECONDS
27
28/**
29 * Mirrors `blocks_session`: a hot sentinel spares a session only when both ids are known and differ.
30 * A blank session id (after trimming, as the Python hook strips it) counts as unknown.
31 */
32export const blocksSession = (owner: string | null, sessionId: string | null | undefined): boolean => {
33  const id = sessionId?.trim() ?? ''
34  return owner === null || id === '' || owner === id
35}
36
37export const isSentinelName = (name: string): boolean =>
38  name.startsWith(SENTINEL_PREFIX) && name.length > SENTINEL_PREFIX.length
39

1import type { EngineInterface } from 'claude-code'
2
3import { readConfig, type WatchConfig } from './config'
4import { HARNESS_CONFIG_REL, rootCandidates } from './root'
5
6/**
7 * The project the watch serves: its root (forward slashes), its harness configuration, and the main
8 * checkout per-repository state (Phase 5 sentinels) lives under; the root itself outside a linked worktree.
9 */
10export type Watch = { root: string; cfg: WatchConfig; sentinelRoot: string }
11
12const GIT_TIMEOUT_MS = 10_000
13
14const noop = (): undefined => undefined
15
16/** Top levels already answered, by start folder (a reload starts afresh). */
17const toplevels = new Map<string, string>()
18
19/** `git rev-parse --show-toplevel` from `start`; undefined on any failure. */
20async function toplevelOf($: EngineInterface, start: string | undefined): Promise<string | undefined> {
21  if (start === undefined || start.trim() === '') return undefined
22  const known = toplevels.get(start)
23  if (known !== undefined) return known
24  const argv = ['git', '-C', start, 'rev-parse', '--show-toplevel']
25  const ran = await $.process.run(argv, { timeoutMs: GIT_TIMEOUT_MS }).catch(noop)
26  const top = ran === undefined || ran.exitCode !== 0 ? '' : ran.stdout.trim()
27  if (top === '') return undefined
28  toplevels.set(start, top)
29  return top
30}
31
32/** Main checkout of `root`'s repository (`git rev-parse --git-common-dir` minus `/.git`); else `root`. */
33async function mainCheckoutOf($: EngineInterface, root: string): Promise<string> {
34  const argv = ['git', '-C', root, 'rev-parse', '--path-format=absolute', '--git-common-dir']
35  const ran = await $.process.run(argv, { timeoutMs: GIT_TIMEOUT_MS }).catch(noop)
36  const common = ran === undefined || ran.exitCode !== 0 ? '' : ran.stdout.trim().replace(/\\/g, '/')
37  return common.endsWith('/.git') ? common.slice(0, -'/.git'.length) : root
38}
39
40/**
41 * The first candidate root (./root) whose `.claude/harness.json` reads, with that file's configuration;
42 * undefined when none does, which leaves the watch inactive. Never throws. Function declarations: the
43 * engine follows `$` past an import only into one.
44 */
45export async function resolveWatch($: EngineInterface): Promise<Watch | undefined> {
46  const env = await $.env.get('CLAUDE_PROJECT_DIR').catch(noop)
47  const cwd = await $.session.cwd().catch(noop)
48  const top = await toplevelOf($, cwd).catch(noop)
49  for (const root of rootCandidates(env, cwd, top)) {
50    const text = await $.fs.read(`${root}/${HARNESS_CONFIG_REL}`).catch(noop)
51    if (typeof text === 'string') return { root, cfg: readConfig(text), sentinelRoot: await mainCheckoutOf($, root) }
52  }
53  return undefined
54}
55

1/**
2 * Review-JSON reading for the bubble guard: a port of the harness library's `validation/id_handling.py`
3 * (`issue_ids`, `disposition_ids`, `all_issues_dispositioned`) plus file selection. Pure: no `$`.
4 */
5
6/**
7 * The parts of a consolidated review JSON (`<artifacts.reviews>/*.json`) the watch reads. `closed`: ids
8 * whose latest `post_review_resolutions[]` status is `RESOLVED` or `CLOSED_NO_ACTION`.
9 */
10export type ReviewDoc = {
11  verdict: string | null
12  issueIds: string[]
13  dispositioned: string[]
14  bubbles: string[]
15  closed: string[]
16}
17
18/** A directory entry, as `$.fs.list` reports one. */
19export type Listed = { name: string; mtimeMs: number; kind: string }
20
21const VERDICT_PREFIX = /^(NOT READY TO MERGE|READY TO MERGE)/
22
23const isRecord = (v: unknown): v is Record<string, unknown> =>
24  typeof v === 'object' && v !== null && !Array.isArray(v)
25const list = (v: unknown): unknown[] => (Array.isArray(v) ? v : [])
26const idOf = (v: unknown): string | undefined => (isRecord(v) && typeof v.id === 'string' ? v.id : undefined)
27const isDefined = <T>(v: T | undefined): v is T => v !== undefined
28const uniq = (ids: readonly string[]): string[] => [...new Set(ids)]
29
30/** Issue ids in C, I, M order, then by number. */
31const byId = (a: string, b: string): number =>
32  a[0] === b[0] ? Number(a.slice(1)) - Number(b.slice(1)) : a < b ? -1 : 1
33
34const parseObject = (text: string): Record<string, unknown> | undefined => {
35  try {
36    const doc: unknown = JSON.parse(text)
37    return isRecord(doc) ? doc : undefined
38  } catch {
39    return undefined
40  }
41}
42
43const issueIdsOf = (doc: Record<string, unknown>): string[] => {
44  const issues = isRecord(doc.issues) ? doc.issues : {}
45  return Object.values(issues).flatMap(bucket => list(bucket).map(idOf).filter(isDefined))
46}
47
48const bubblesOf = (doc: Record<string, unknown>): string[] =>
49  list(doc.bubbles).filter((b): b is string => typeof b === 'string')
50
51const dispositionedOf = (doc: Record<string, unknown>): string[] => [
52  ...list(doc.post_review_resolutions).map(idOf).filter(isDefined),
53  ...bubblesOf(doc),
54  ...list(doc.bubble_up).map(idOf).filter(isDefined),
55]
56
57const CLOSING_STATUSES: readonly unknown[] = ['RESOLVED', 'CLOSED_NO_ACTION']
58
59/** Ids whose last resolution entry carries a closing status; a later non-closing entry reopens. */
60const closedOf = (doc: Record<string, unknown>): string[] => {
61  const latest = new Map<string, unknown>()
62  for (const entry of list(doc.post_review_resolutions)) {
63    const id = idOf(entry)
64    if (id !== undefined && isRecord(entry)) latest.set(id, entry.status)
65  }
66  return [...latest].filter(([, status]) => CLOSING_STATUSES.includes(status)).map(([id]) => id)
67}
68
69/** The review parts, or undefined when the text is not a JSON object. */
70export const readReview = (text: string): ReviewDoc | undefined => {
71  const doc = parseObject(text)
72  if (doc === undefined) return undefined
73  return {
74    verdict: typeof doc.verdict === 'string' ? doc.verdict : null,
75    issueIds: uniq(issueIdsOf(doc)),
76    dispositioned: uniq(dispositionedOf(doc)),
77    bubbles: uniq(bubblesOf(doc)),
78    closed: closedOf(doc),
79  }
80}
81
82/** `bubble_ups.json` `items[].id`, or null when the text is not a JSON object. */
83export const sidecarIds = (text: string): string[] | null => {
84  const doc = parseObject(text)
85  return doc === undefined ? null : uniq(list(doc.items).map(idOf).filter(isDefined))
86}
87
88/** At least one issue, and every issue dispositioned. */
89export const allDispositioned = (r: ReviewDoc): boolean =>
90  r.issueIds.length > 0 && r.issueIds.every(id => r.dispositioned.includes(id))
91
92/** Tracked bubbled ids the review does not disposition, sorted; none without a review. */
93export const missingFromReview = (tracked: readonly string[], r: ReviewDoc | undefined): string[] =>
94  r === undefined ? [] : uniq(tracked.filter(id => !r.dispositioned.includes(id))).sort(byId)
95
96export const verdictPrefix = (verdict: string | null): string | undefined => verdict?.match(VERDICT_PREFIX)?.[1]
97
98/** The newest `<NNNN>_*.json` file entry for a PR, or undefined. */
99export const reviewFileFor = (pr: number, entries: readonly Listed[]): Listed | undefined => {
100  const prefix = `${String(pr).padStart(4, '0')}_`
101  const hits = entries.filter(e => e.kind === 'file' && e.name.startsWith(prefix) && e.name.endsWith('.json'))
102  return [...hits].sort((a, b) => b.mtimeMs - a.mtimeMs)[0]
103}
104
105const BUBBLE_WORD = /bubble/i
106
107/** The message talks about bubbles: the trigger `BUBBLE-watch.sh` used, any case and inflection. */
108export const mentionsBubble = (text: string | undefined): boolean => BUBBLE_WORD.test(text ?? '')
109
110/**
111 * The keyword-triggered Stop block (the `has_bubbles.py` rule): a message that mentions bubbles needs a
112 * populated `bubbles[]`, unless every issue is already dispositioned. Undefined to let the stop through.
113 */
114export const keywordBlock = (
115  pr: number,
116  text: string | undefined,
117  r: ReviewDoc | undefined,
118  isLoopBreak: boolean,
119): string | undefined => {
120  if (isLoopBreak || r === undefined || !mentionsBubble(text) || r.bubbles.length > 0) return undefined
121  const open = r.issueIds.filter(id => !r.dispositioned.includes(id)).sort(byId)
122  if (open.length === 0) return undefined
123  return (
124    `PR #${pr}: the last message discusses bubbles, but the review JSON's bubbles[] is empty while ` +
125    `${open.join(', ')} remain undispositioned. Populate bubbles[] before discussing bubbles, unless all ` +
126    'issues are already resolved.'
127  )
128}
129
130/** The state-triggered Stop block text, or undefined to let the stop through. */
131export const stopBlock = (
132  pr: number | null,
133  missing: readonly string[],
134  from: string,
135  isLoopBreak: boolean,
136): string | undefined => {
137  if (isLoopBreak || missing.length === 0) return undefined
138  return (
139    `PR #${pr ?? '?'}: the review JSON does not disposition bubbled ${missing.join(', ')} ` +
140    `(tracked from ${from}). Add them to bubbles[] (or bubble_up[] / post_review_resolutions[]) before stopping.`
141  )
142}
143

1import type { Counts, Phase, Skill } from '../types'
2
3/**
4 * Parsers for pr-team / pr-review `Agent` prompts and return-contract blocks (the harness pr-team skill's
5 * `return_contract.md` and `canonical_prompt_template.md`). Pure: no `$`.
6 */
7
8/** Header fields of an `Agent` prompt: `KEY=value` and `KEY: value` lines at column 1. */
9export type Header = Readonly<Record<string, string>>
10
11/** Who a tracked `Agent` dispatch is. */
12export type Dispatch = {
13  runId: string
14  skill: Skill
15  agentId: string
16  role: string
17  pr: number | null
18  runtimeDir: string | null
19  phase: Phase
20}
21
22/** A parsed return-contract block; `bubbled` is null when the block has no `BUBBLED:` line. */
23export type ReturnBlock = {
24  signal: string
25  role: string | null
26  counts: Counts | null
27  fixed: string[]
28  bubbled: string[] | null
29  output: string | null
30  notes: string | null
31}
32
33export const MODEL_FAMILIES = ['haiku', 'sonnet', 'opus'] as const
34export type ModelFamily = (typeof MODEL_FAMILIES)[number]
35
36const HEADER_LINE = /^([A-Z][A-Z_]*)(?:=|: )(.*)$/
37const BLOCK_LINE = /^([A-Z][A-Z_]*):(?: (.*))?$/
38const COUNTS_VALUE = /^findings=(\d+) critical=(\d+) important=(\d+) minor=(\d+)$/
39const ISSUE_ID = /^[CIM]\d+$/
40const RUN_ID_PR = /^pr(\d{1,7})_\d{8}T\d{6}Z$/
41const RUN_ID_STAMP = /^pr\d{1,7}_(\d{4})(\d{2})(\d{2})T(\d{2})(\d{2})(\d{2})Z$/
42
43/** Phase, agent id and role per pr-team agent family and MODE. */
44const MODES: Readonly<Record<string, Readonly<Record<string, readonly [Phase, string, string]>>>> = {
45  orchestrator: {
46    PLAN: ['plan', 'orchestrator_plan', 'orchestrator'],
47    CONSOLIDATE: ['consolidate', 'orchestrator_consolidate', 'orchestrator'],
48  },
49  fixer: { PLAN: ['fix-plan', 'fixer_plan', 'planner'], CLOSEOUT: ['closeout', 'closer', 'closer'] },
50}
51
52/** Key/value lines matching `pattern`; the first occurrence of a key wins. */
53const fieldsOf = (lines: readonly string[], pattern: RegExp): Record<string, string> => {
54  const out: Record<string, string> = {}
55  for (const line of lines) {
56    const m = line.trimEnd().match(pattern)
57    const key = m?.[1]
58    if (key !== undefined && !(key in out)) out[key] = (m?.[2] ?? '').trim()
59  }
60  return out
61}
62
63export const parseHeader = (prompt: string): Header => fieldsOf(prompt.split(/\r?\n/), HEADER_LINE)
64
65/** A positive integer PR number, or null. */
66export const toPr = (raw: string | undefined): number | null =>
67  raw !== undefined && /^\d+$/.test(raw) && Number(raw) >= 1 ? Number(raw) : null
68
69/** The PR a run id names (`pr1122_20260518T060416Z`), or null. */
70export const prOfRunId = (runId: string): number | null => toPr(runId.match(RUN_ID_PR)?.[1])
71
72/** Epoch ms of the UTC stamp a run id carries (`pr1122_20260518T060416Z`), or null when absent or invalid. */
73export const startedAtOfRunId = (runId: string): number | null => {
74  const m = runId.match(RUN_ID_STAMP)
75  if (m === null) return null
76  const [y, mo, d, h, mi, s] = m.slice(1).map(Number) as [number, number, number, number, number, number]
77  const ms = Date.UTC(y, mo - 1, d, h, mi, s)
78  const back = new Date(ms)
79  return back.getUTCMonth() === mo - 1 && back.getUTCDate() === d && h < 24 && mi < 60 && s < 60 ? ms : null
80}
81
82/** The agent family a pr-team subagent type names, or undefined. */
83const familyOf = (subagentType: string): string | undefined =>
84  /orchestrator/.test(subagentType) ? 'orchestrator' : /fixer/.test(subagentType) ? 'fixer' : undefined
85
86/** A pr-team `MODE=` dispatch to the orchestrator or fixer agent. */
87const modeDispatch = (subagentType: string, h: Header, runId: string): Dispatch | undefined => {
88  const family = familyOf(subagentType)
89  const hit = family === undefined ? undefined : MODES[family]?.[h.MODE ?? '']
90  if (hit === undefined) return undefined
91  const [phase, agentId, role] = hit
92  const pr = toPr(h.PR_NUMBER) ?? prOfRunId(runId)
93  return { runId, skill: 'pr-team', agentId, role, pr, runtimeDir: h.RUNTIME_DIR ?? null, phase }
94}
95
96/** The runtime dir an OUTPUT_FILE slot sits in, else `<runtimeBase>/<runId>`. */
97const runtimeDirOf = (h: Header, runId: string, runtimeBase: string): string => {
98  const m = (h.OUTPUT_FILE ?? '').match(/^(.*)[\\/]outputs[\\/][^\\/]+$/)
99  return m?.[1] ?? `${runtimeBase}/${runId}`
100}
101
102/** A slot dispatch (`AGENT_ID:` / `AGENT_ROLE:`): pr-review when a `PR:` line is present. */
103const slotDispatch = (h: Header, runId: string, runtimeBase: string): Dispatch | undefined => {
104  const { AGENT_ID: agentId, AGENT_ROLE: role } = h
105  if (!agentId || (role !== 'reviewer' && role !== 'fixer')) return undefined
106  const isReview = h.PR !== undefined
107  const pr = (isReview ? toPr(h.PR) : null) ?? prOfRunId(runId)
108  const skill: Skill = isReview ? 'pr-review' : 'pr-team'
109  const runtimeDir = isReview ? null : runtimeDirOf(h, runId, runtimeBase)
110  return { runId, skill, agentId, role, pr, runtimeDir, phase: role === 'reviewer' ? 'review' : 'fix' }
111}
112
113/**
114 * The tracked identity of an `Agent` call, or undefined for any other dispatch. `runtimeBase` is the
115 * project's pr-team runtime folder, used when a pr-team slot names no OUTPUT_FILE.
116 */
117export const identify = (
118  subagentType: string | undefined,
119  prompt: string,
120  runtimeBase: string,
121): Dispatch | undefined => {
122  const h = parseHeader(prompt)
123  const runId = h.RUN_ID
124  if (runId === undefined || runId === '') return undefined
125  return h.MODE !== undefined ? modeDispatch(subagentType ?? '', h, runId) : slotDispatch(h, runId, runtimeBase)
126}
127
128/** Comma-separated issue ids; malformed entries dropped. */
129export const idList = (raw: string): string[] =>
130  raw.split(',').map(s => s.trim()).filter(s => ISSUE_ID.test(s))
131
132/** A COUNTS value, or null when absent or malformed. */
133export const parseCounts = (raw: string | undefined): Counts | null => {
134  const m = raw?.match(COUNTS_VALUE)
135  if (!m) return null
136  return { findings: Number(m[1]), critical: Number(m[2]), important: Number(m[3]), minor: Number(m[4]) }
137}
138
139/** Indexes of column-1 `---` fence lines. */
140const fences = (lines: readonly string[]): number[] => lines.flatMap((l, i) => (l.trimEnd() === '---' ? [i] : []))
141
142const toBlock = (f: Readonly<Record<string, string>>): ReturnBlock => ({
143  signal: f.SIGNAL ?? '',
144  role: f.AGENT_ROLE ?? null,
145  counts: parseCounts(f.COUNTS),
146  fixed: idList(f.FIXED ?? ''),
147  bubbled: f.BUBBLED === undefined ? null : idList(f.BUBBLED),
148  output: f.OUTPUT ?? null,
149  notes: f.NOTES ?? null,
150})
151
152/** The last fenced block that carries a SIGNAL line; undefined when there is none. */
153export const parseReturnBlock = (text: string): ReturnBlock | undefined => {
154  const lines = text.split(/\r?\n/)
155  const at = fences(lines)
156  for (let i = at.length - 1; i >= 1; i--) {
157    const f = fieldsOf(lines.slice((at[i - 1] ?? 0) + 1, at[i]), BLOCK_LINE)
158    if (f.SIGNAL) return toBlock(f)
159  }
160  return undefined
161}
162
163/** The family token in a model id, case-insensitive; undefined when none appears. */
164export const modelFamily = (raw: string | null | undefined): ModelFamily | undefined => {
165  const id = (raw ?? '').toLowerCase()
166  return MODEL_FAMILIES.find(t => id.includes(t))
167}
168

1import type { AgentRec } from '../types'
2import { MODEL_FAMILIES, modelFamily } from './contract'
3
4/**
5 * Rows for the project's invocation matrix (harness `matrix_path`), validated against the rules of the
6 * harness plugin's `schemas/pr_review_invocation_test_matrix.schema.json`. The constants below are pinned
7 * to that schema by `tests/mods/test_pr_review_watch_matrix_contract.py`. Pure: no `$`.
8 */
9
10export const MATRIX_REQUIRED = [
11  'cell', 'pr', 'model_family', 'effort', 'dev_commit', 'dispatch_shape', 'tokens', 'tool_uses', 'duration_s',
12  'findings_count', 'C_count', 'I_count', 'M_count', 'other_count', 'verdict_prefix', 'notes',
13] as const
14export const MATRIX_OPTIONAL = ['phase4_script', 'legacy'] as const
15export const DISPATCH_SHAPES = ['parallel', 'inline'] as const
16export const DEV_COMMIT_PATTERN = '^[0-9a-f]{40}( -- .*)?$'
17
18export type MatrixRow = {
19  cell: string
20  pr: number
21  model_family: string
22  effort: string
23  dev_commit: string
24  dispatch_shape: 'parallel'
25  tokens: number
26  tool_uses: number
27  duration_s: number
28  phase4_script: null
29  findings_count: number
30  C_count: number
31  I_count: number
32  M_count: number
33  other_count: number
34  verdict_prefix: string
35  notes: string
36}
37
38export type RowInput = { pr: number; agent: AgentRec; verdictPrefix: string; devCommit: string }
39
40const STRING_FIELDS = [
41  'cell', 'model_family', 'effort', 'dev_commit', 'dispatch_shape', 'verdict_prefix', 'notes',
42] as const
43const COUNT_FIELDS = [
44  'tokens', 'tool_uses', 'findings_count', 'C_count', 'I_count', 'M_count', 'other_count',
45] as const
46const ALLOWED = new Set<string>([...MATRIX_REQUIRED, ...MATRIX_OPTIONAL])
47const DEV_COMMIT = new RegExp(DEV_COMMIT_PATTERN)
48
49const notesFor = (a: AgentRec): string => {
50  const base = `pr-review-watch auto row; ${a.role} ${a.signal ?? 'pending'}`
51  return a.isMeasured ? base : `${base}; usage not measured (session restarted mid-run)`
52}
53
54const whole = (n: number): number => Math.max(0, Math.round(n))
55
56export const buildRow = (i: RowInput): MatrixRow => {
57  const { agent: a } = i
58  const c = a.counts
59  return {
60    cell: `pr${i.pr}_${a.id}`, pr: i.pr, model_family: modelFamily(a.model) ?? a.model ?? 'unknown',
61    effort: a.effort ?? 'default', dev_commit: i.devCommit, dispatch_shape: 'parallel',
62    tokens: whole(a.tokens), tool_uses: whole(a.toolUses), duration_s: whole(a.durationMs) / 1000,
63    phase4_script: null, findings_count: c?.findings ?? 0, C_count: c?.critical ?? 0, I_count: c?.important ?? 0,
64    M_count: c?.minor ?? 0, other_count: 0, verdict_prefix: i.verdictPrefix, notes: notesFor(a),
65  }
66}
67
68const isCount = (v: unknown): boolean => Number.isInteger(v) && (v as number) >= 0
69const isOneOf = (v: string, options: readonly string[]): boolean => options.includes(v)
70
71/** Range checks on numeric fields that are present. */
72const numberProblems = (row: Readonly<Record<string, unknown>>): string[] => {
73  const out: string[] = []
74  if ('pr' in row && !(Number.isInteger(row.pr) && (row.pr as number) >= 1)) out.push('pr: not a positive integer')
75  const d = row.duration_s
76  const isDuration = typeof d === 'number' && Number.isFinite(d) && d >= 0
77  if ('duration_s' in row && !isDuration) out.push('duration_s: not a non-negative number')
78  return out
79}
80
81/** Enum, pattern and optional-field checks on fields that are present with the right primitive type. */
82const valueProblems = (row: Readonly<Record<string, unknown>>): string[] => {
83  const out = numberProblems(row)
84  const m = row.model_family
85  if (typeof m === 'string' && !isOneOf(m, MODEL_FAMILIES)) {
86    out.push(`model_family: '${m}' is not one of ${MODEL_FAMILIES.join('/')}`)
87  }
88  const s = row.dispatch_shape
89  if (typeof s === 'string' && !isOneOf(s, DISPATCH_SHAPES)) {
90    out.push(`dispatch_shape: '${s}' is not one of ${DISPATCH_SHAPES.join('/')}`)
91  }
92  if (row.legacy !== true && typeof row.dev_commit === 'string' && !DEV_COMMIT.test(row.dev_commit)) {
93    out.push('dev_commit: does not start with a 40-char merge-base SHA')
94  }
95  const p4 = row.phase4_script
96  if ('phase4_script' in row && p4 !== null && typeof p4 !== 'string') out.push('phase4_script: not a string or null')
97  if ('legacy' in row && typeof row.legacy !== 'boolean') out.push('legacy: not a boolean')
98  return out
99}
100
101/** Every way the row breaks the schema; empty when it validates. */
102export const validateRow = (row: Readonly<Record<string, unknown>>): string[] => [
103  ...Object.keys(row).filter(k => !ALLOWED.has(k)).map(k => `${k}: not in the schema`),
104  ...MATRIX_REQUIRED.filter(k => !(k in row)).map(k => `${k}: missing`),
105  ...STRING_FIELDS.filter(k => k in row && typeof row[k] !== 'string').map(k => `${k}: not a string`),
106  ...COUNT_FIELDS.filter(k => k in row && !isCount(row[k])).map(k => `${k}: not a non-negative integer`),
107  ...valueProblems(row),
108]
109
110/** The JSONL text with rows appended, one compact object per line. */
111export const appendRows = (existing: string, rows: readonly MatrixRow[]): string => {
112  if (rows.length === 0) return existing
113  const head = existing === '' || existing.endsWith('\n') ? existing : `${existing}\n`
114  return head + rows.map(r => `${JSON.stringify(r)}\n`).join('')
115}
116

1/**
2 * The project-root rule: the watch is active only in a project holding `.claude/harness.json`. Candidates
3 * follow harness_lib.project.project_root: the git top level of the session cwd (a linked worktree, or a
4 * subfolder of one), then CLAUDE_PROJECT_DIR, then the cwd itself; the first that holds the file is the
5 * root. Pure: the `$` probe lives in watch.ts.
6 */
7
8export const HARNESS_CONFIG_REL = '.claude/harness.json'
9
10const clean = (p: string): string => p.trim().replace(/\\/g, '/').replace(/\/+$/, '')
11const keyOf = (p: string): string => clean(p).toLowerCase()
12const nonBlank = (v: string | undefined): string | undefined =>
13  v !== undefined && v.trim() !== '' ? clean(v) : undefined
14
15/**
16 * Candidate roots in probe order: `rawTop` (git's top level of the cwd), CLAUDE_PROJECT_DIR, the cwd;
17 * blanks and case-insensitive repeats dropped.
18 */
19export const rootCandidates = (
20  rawEnv: string | undefined,
21  rawCwd: string | undefined,
22  rawTop?: string,
23): string[] => {
24  const out: string[] = []
25  for (const c of [nonBlank(rawTop), nonBlank(rawEnv), nonBlank(rawCwd)]) {
26    if (c !== undefined && !out.some(o => keyOf(o) === keyOf(c))) out.push(c)
27  }
28  return out
29}
30