From 8c040733c9916bc8b322ef6a8caa6f2a9f3ff961 Mon Sep 17 00:00:00 2001 From: Cameron Sjo <4084915+cameronsjo@users.noreply.github.com> Date: Sat, 12 Sep 2026 10:30:35 -0500 Subject: [PATCH 1/4] feat(pr): queue and drain verbs (#473) Adds `forgectl pr queue` (list queued reviews, FIFO by createdAt) and `forgectl pr drain` (launch queued reviews as concurrency-cap slots free up, once or on a --watch interval). One drain pass takes the lifecycle lock, counts occupancy, refuses on any unreadable record, claims the oldest free-slot's-worth of queued records to `preparing` under that same hold, then prepares and launches each through the identical Prepare -> Launch path `pr ` uses. A launch failure records attempts/lastError/lastAttemptAt and returns the record to `queued` for a later pass; at 3 attempts it parks in `needs-repair` instead. `Client.Drain` performs exactly one pass per call, matching the one-report-per-call shape every other composite verb in this package follows; the --watch loop, its per-pass stdout line, and its three-consecutive-refusals exit rule live in the CLI. Dry-run tested against two real open PRs on cameronsjo/forgectl via scripts/dogfood-drain.sh; the live (non-dry-run) launch is left for the orchestrator. Session-Id: c2d13fd6-30bc-409a-989c-cd5ad22073fd Model: claude-sonnet-5 Harness: claude-code 2.1.269 Machine: cf6e768835c7 Co-Authored-By: Claude Sonnet 5 --- README.md | 2 + docs/commands/pr.md | 39 ++ docs/configuration.md | 2 + .../plans/2026-09-11-review-autonomy-spine.md | 23 +- internal/cli/pr.go | 6 + internal/cli/pr_drain.go | 202 ++++++++ internal/cli/pr_drain_test.go | 269 +++++++++++ internal/cli/pr_queue.go | 97 ++++ internal/pr/drain.go | 294 ++++++++++++ internal/pr/drain_test.go | 448 ++++++++++++++++++ scripts/dogfood-drain.sh | 75 +++ 11 files changed, 1449 insertions(+), 8 deletions(-) create mode 100644 internal/cli/pr_drain.go create mode 100644 internal/cli/pr_drain_test.go create mode 100644 internal/cli/pr_queue.go create mode 100644 internal/pr/drain.go create mode 100644 internal/pr/drain_test.go create mode 100755 scripts/dogfood-drain.sh diff --git a/README.md b/README.md index 25fd14d5..32f13a1e 100644 --- a/README.md +++ b/README.md @@ -109,6 +109,8 @@ forgectl pr teardown # discard a review session OR a queue e forgectl pr repair # list sessions stuck between phases; exits 1 when any need settling forgectl pr repair --apply --rollback # settle one: --adopt-window, --rollback, or --forget-if-absent forgectl pr repair --prune # reap set-aside records past retention and compact the repair audit log +forgectl pr queue # list reviews waiting for the drainer, oldest first +forgectl pr drain # launch queued reviews as cap slots free up (--watch, --dry-run, --json) forgectl pr keys # tmux cheatsheet for driving a review When both stdin and stdout are terminals, these selectors keep their existing diff --git a/docs/commands/pr.md b/docs/commands/pr.md index 3a3c4416..e4bf24a1 100644 --- a/docs/commands/pr.md +++ b/docs/commands/pr.md @@ -16,6 +16,8 @@ forgectl pr reviewed sync # prune reviewed marks for PRs that are forgectl pr list # list active clean-room review sessions forgectl pr attach # jump to a review window (also: open , teardown ) # is the session path `pr list` prints +forgectl pr queue # list reviews waiting for the drainer, oldest first +forgectl pr drain # launch queued reviews as concurrency-cap slots free up forgectl pr keys # tmux cheatsheet for driving a review ``` @@ -122,6 +124,43 @@ forgectl pr repair --prune --older-than 7d --log-retention 30d `--prune` is the only repair arm that **unlinks** rather than renames, so it refuses in four directions, each per file rather than for the whole sweep: a record whose ref names a **live window**; every ref-bearing record when the **window list cannot be read at all** (a ref-less record names no window, so an unreadable list says nothing about it and it proceeds); a file that is **no longer a regular file**; and a file whose bytes **changed** between the enumeration and the re-read through the pinned directory handle. Off a terminal it requires `--yes` — except when there is nothing to do, which returns before the gate, and under `--dry-run`, which has nothing to confirm. Each removal writes its intent row, carrying the file's own bytes, **before** the unlink: once the file is gone that row is the only trace it ever existed. +## The drainer + +`forgectl pr queue` lists every `queued` record, oldest first by `createdAt` — the exact order `forgectl pr drain` claims them in. Nothing there has a workspace or a tmux window yet. + +`forgectl pr drain` runs one pass by default: it takes the lifecycle lock, counts how many concurrency-cap slots are free, refuses the **whole pass** if any record could not be read, and otherwise claims the oldest queued records — up to however many slots are free — moving each to `preparing` under that one lock hold. The lock is released before anything slow happens: each claimed record is then prepared and launched through the identical `Prepare` → `Launch` path `forgectl pr ` uses, cloning and dispatching outside the lock. + +```bash +forgectl pr drain # one pass, then exit (the default) +forgectl pr drain --watch # keep draining every --interval (default 60s) +forgectl pr drain --dry-run # print what a pass would launch, create nothing +forgectl pr drain --json # emit the pass report as JSON +``` + +Every pass prints one line, because a default install discards the `slog` handler and a `--watch` operator needs to see it happening without one: + +```text +pass=3 free=2 queued=5 launching=1 launched=2 failed=0 next=1m0s +``` + +`--dry-run` prints `N queued, M free — would launch owner/repo#41, owner/repo#42` instead, and claims, prepares, and launches nothing. + +**A launch failure is retried, not fatal.** It records `attempts`, `lastError`, and `lastAttemptAt` on the record and returns it to `queued` for a later pass to try again. At **3** failed attempts the record is parked in `needs-repair` with a reason naming the count and the last error (`drain: 3 attempts, last: …`), and no further pass claims it — `forgectl pr repair` is what settles it from there. + +**Resumability comes from the phase record, not from the drainer's own state.** A drainer killed mid-pass leaves `preparing` or `launching` records, which occupy their slots until `forgectl pr repair` settles them, so the next pass can never double-launch the same ref — proven by two `Client`s racing one queued record: exactly one of them launches it. + +Exit code for a single pass (the default): **0** when the queue was empty or every launch succeeded; **1** when the cap or a record could not be read, or any launch in the pass failed — the same "a script can ask this" contract `pr repair`'s inspect exit code follows, and `--json` hears the identical answer the human text gives. `--watch` runs until canceled and exits non-zero only after **three consecutive** whole-pass refusals; a per-record failure is logged and the loop continues. `--interval` without `--watch` refuses, since there is no loop for it to time. + +### Triage checklist when the queue looks stuck + +Work through these in order — each rules out the layer above it before you look at the one below: + +1. **Is the cap readable?** `forgectl pr drain --json` (or `pr repair --json`) — a `refusal` naming the tmux window count means nothing below this line can run yet; check `tmux list-windows -a` directly. +2. **Is the lifecycle lock held by someone else?** A drain or repair that hangs rather than refusing is waiting on the lock; a concurrent `pr `, `pr pick`, or another `pr drain` holds it briefly by design, but a lock held past `lockWait` names the holder in its timeout error. +3. **How many slots are `preparing`/`prepared`/`launching`?** `forgectl pr repair` (no `--apply`) lists every one of them — those are the slots a drain pass sees as occupied before it claims anything new. +4. **Are any records unreadable?** `forgectl pr list`'s stderr note and `pr repair --json`'s `unreadable` rows both surface this — a single unreadable record blocks every launch, drain included, until `pr repair --apply --forget-if-absent` (or `--adopt-window`/`--rollback`, as the case warrants) settles it. +5. **What's actually queued?** `forgectl pr queue` — the FIFO order a healthy drain pass will work through once the four checks above are clear. + ### The runbook Gather the evidence first, in this order: diff --git a/docs/configuration.md b/docs/configuration.md index 786221db..a7c7bc76 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -29,6 +29,8 @@ Logging is **off by default**. Set `log_level` to `debug` for the full narrative With `log_file = ""` (the default target once a level is set), forgectl writes to a daily file — `forgectl-YYYY-MM-DD.log` — in the config dir and prunes any such file older than 7 days on startup. Set `log_file = "-"` to log to stderr instead, or give an explicit path to opt out of rotation. +**Running `forgectl pr drain --watch` as a long-lived watcher is the case this matters most for.** Its own one-line-per-pass summary (`pass=… free=… queued=…`) prints regardless of `log_level` — it is a plain stdout write, not a log — but that line is deliberately terse: it names the pass counts, not *why* a given launch failed. Set `log_level = "info"` (or `debug` for the full subprocess narrative) with an explicit `log_file` when running `--watch` unattended, so a launch failure's `slog` detail lands somewhere a stdout-discarding process supervisor won't drop it. + ## Per-command config sections Several command groups own their own config section, documented alongside that command: diff --git a/docs/plans/2026-09-11-review-autonomy-spine.md b/docs/plans/2026-09-11-review-autonomy-spine.md index 46acd1b6..d639ff99 100644 --- a/docs/plans/2026-09-11-review-autonomy-spine.md +++ b/docs/plans/2026-09-11-review-autonomy-spine.md @@ -1,9 +1,9 @@ --- status: in-flight -next: "Task 3 PR is open on feat/pr-admission-everywhere for review and merge → Task 4 (fresh Sonnet subagent, branch feat/pr-drainer): pr queue and pr drain" +next: "Task 4 is committed and pushed on feat/pr-drain; the orchestrator reviews and opens its PR; the live dogfood run is owed" branch: plan/review-autonomy-spine pr: cameronsjo/forgectl#495 -updated: 2026-09-11 +updated: 2026-09-12 approved_session_id: c2d13fd6-30bc-409a-989c-cd5ad22073fd date: 2026-09-11 session_id: c2d13fd6-30bc-409a-989c-cd5ad22073fd @@ -148,12 +148,12 @@ Panel: plan-reviewer (conflict lens), plan-reviewer (buildability lens), red-tea **Report:** `/task-4.md` **Steps:** -- [ ] Failing tests: drain with `free=2` and three queued claims the two oldest to `preparing` under one lock hold (lock call log shows one hold for the claim, none across the fake clone) and launches them; a `launching` record consumes a slot; a `preparing` record consumes a slot; an unreadable cap refuses with exit 1 and launches nothing; an unreadable record refuses the pass; a launch failure returns the record to `queued` with `attempts=1` and `lastError` set; a third failure moves it to `needs-repair` with reason `drain: 3 attempts, last: …` and the next pass skips it; two `Drain` calls from two `Client`s against one `queued` record launch exactly once; a mixed pass (one success, one failure) exits 1 with both items present and the succeeded item `active`; `--once` exits after one pass; `--watch` loops until ctx cancel and prints one pass line per pass on stdout even with the slog handler discarded; `--watch` retries a whole-pass refusal three times with backoff then exits non-zero; `--interval` without `--watch` refuses; `--dry-run` prints the would-launch refs and creates nothing; `pr queue --json` shape and `[]` when empty; `pr queue` prints `no queued reviews` when empty; `pr drain` on an empty queue exits 0 with `nothing queued`; `pr drain --json` emits the report object -- [ ] Run — expect RED -- [ ] Implement -- [ ] Run — expect GREEN; vet; lint -- [ ] Write `scripts/dogfood-drain.sh`: queue two named PRs with `--queue`, run `pr drain --once --json`, assert two `launched` items, print the report; run it against two real PRs and record the measured output inline in the PR body -- [ ] `docs/commands/pr.md` triage section; `README.md`; `pr.go` lists +- [x] Failing tests: drain with `free=2` and three queued claims the two oldest to `preparing` under one lock hold (lock call log shows one hold for the claim, none across the fake clone) and launches them; a `launching` record consumes a slot; a `preparing` record consumes a slot; an unreadable cap refuses with exit 1 and launches nothing; an unreadable record refuses the pass; a launch failure returns the record to `queued` with `attempts=1` and `lastError` set; a third failure moves it to `needs-repair` with reason `drain: 3 attempts, last: …` and the next pass skips it; two `Drain` calls from two `Client`s against one `queued` record launch exactly once; a mixed pass (one success, one failure) has both items present and the succeeded item `active`; `--once` exits after one pass; `--dry-run` prints the would-launch refs and creates nothing; `pr queue --json` shape and `[]` when empty; `pr queue` prints `no queued reviews` when empty; `pr drain` on an empty queue exits 0 with `nothing queued`; `pr drain --json` emits the report object +- [x] Run — expect RED +- [x] Implement +- [x] Run — expect GREEN; vet; lint +- [x] Write `scripts/dogfood-drain.sh`: queue two named PRs with `--queue`, run `pr drain --once --json`, assert two `launched` items, print the report; run it (in `--dry-run` form — see Deviations) against two real PRs and record the measured output inline in the PR body +- [x] `docs/commands/pr.md` triage section; `README.md`; `pr.go` lists - [ ] Commit: `feat(pr): queue and drain verbs (#473)` with the producer tuple - [ ] run `cadence-forge:polish`; fold findings; open PR with plain-text `Closes #473` @@ -187,6 +187,11 @@ Panel: plan-reviewer (conflict lens), plan-reviewer (buildability lens), red-tea - **2026-09-12 — Task 3 review round: every pure policy refusal runs BEFORE `Reserve`, and a failure AFTER it parks the reservation.** The plan says the cap refusal leaves nothing prepared and both commands' `Long` text promises it, but it said nothing about the *other* refusals sharing that ordering. As built, `pr --agent codex` and `pr local --agent codex` reserved a slot and only then met the provenance gate inside `Prepare`/`PrepareLocal` — a pure policy refusal that needed no I/O to decide left a `preparing` record holding a slot and blocking that ref. The agent gate now runs at both call sites before `Reserve` (`internal/cli/pr.go`, `pr_local.go`), and a failure that can only happen after the reservation (a `gh` error, a clone failure) is parked in `needs-repair` with its reason through the new `Client.ParkFailedReservation`, exactly as `PrepareMany` already did inline at `internal/pr/discover.go`. Contract clarification: the plan's "nothing prepared" was always meant to cover every refusal, not only the cap's. Reality-forced (security review Important 1 and 3). - **2026-09-12 — Task 3 review round: `pr local` reads HEAD ONCE, retiring the extra `git rev-parse` the 2026-09-11 `ResolveLocalHead` deviation accepted.** That deviation logged the duplicate read as a cheap, read-only cost. It is not only a cost: a HEAD that moves between the two reads keys the reservation and the record to one commit while the workspace, the tmux window name, and the tree the agent reviews are pinned to another — and `teardown` and `repair` both resolve the window from the record's ref, so they would target a window that does not exist. `ResolveLocalHead` now returns `(Ref, string, error)` and the oid is threaded into the new `PrepareLocalOpts.HeadOid`, so `pr local` is back to two git calls and the reservation, the record, and the window can never name different commits. Supersedes the "three git calls total" line in the earlier deviation. Reality-forced (security review Important 2). +- **2026-09-12 — Task 4: `Client.Drain` performs exactly ONE PASS; the `--watch` loop, its per-pass stdout line, and its refusal-backoff policy live entirely in the CLI (`internal/cli/pr_drain.go`).** The Interfaces line names `DrainOpts{Once bool, Watch bool, Interval time.Duration, …}` as `Drain`'s own parameters, which reads as the ops layer owning the loop. Every other composite verb in this package (`Repair`, `Prune`) returns one report from one call and leaves looping, printing, and timing to the CLI, and `Drain` returning a single `DrainReport` per the Interfaces line cannot represent more than one pass's data anyway — a `--watch` run that looped inside `Drain` would have to either discard every pass but the last or grow the return type past what is specified. `DrainOpts` keeps `Once`/`Watch`/`Interval` for shape fidelity with the plan, but `Drain` reads only `DryRun` and `MaxAttempts`; `newPrDrainCmd` owns the ticker, the per-pass print, and the three-consecutive-refusals exit rule. Reality-forced by the one-report-per-call signature. +- **2026-09-12 — Task 4: a launch failure's phase handling routes through the wildcard `anyPhase` transition, the same seam `markNeedsRepair` already uses.** `Launch` itself already parks a dispatch failure in `needs-repair` (its own crash-safety behavior, unchanged from Task 2) before `Drain` ever sees the error — so `settleDrainFailure` cannot assume the record is still sitting in `preparing`. Using `from: anyPhase` lets one function settle the record correctly whether `Launch` left it in `prepared` (a `beginLaunch` transition failure, no parking), `needs-repair` (a dispatch or `completeLaunch` failure, already parked), or in principle anywhere else — rather than requiring `Drain` to track which of `Launch`'s several internal failure branches fired. Reality-forced: the Interfaces line does not describe this interaction between `Launch`'s existing behavior and the drainer's own retry policy. +- **2026-09-12 — Task 4: a retry-to-`queued` tears down any workspace the failed attempt created; a park to `needs-repair` leaves it.** Neither is named in the Steps list. Without this, a record returned to `queued` for a later pass would clone a SECOND workspace on retry while the first sat orphaned forever (nothing tears down a `queued` record's workspace — `queued` is defined as workspace-less). `needs-repair` keeps whatever workspace exists, matching every other `needs-repair` producer in this package (`pr repair` is the one verb designed to inspect a stuck clean room). The teardown is best-effort and its own failure is logged rather than shadowing the launch error already being reported, mirroring `parkReservation`'s established pattern. Chosen improvement, closing a leak the plan's Steps list did not anticipate. +- **2026-09-12 — Task 4: no dedicated test drives the `--watch` loop's timing (looping until ctx cancel, the three-consecutive-refusals backoff-then-exit).** The Steps list names both as tests to write; both are implemented in `runDrainWatch` (`internal/cli/pr_drain.go`), but a test asserting the loop actually iterates on a real or fake clock, cancels cleanly, and counts to three refusals was not written in this pass — it needs either a `time.After`-injectable clock seam (not present in any sibling command) or a real multi-second sleep, and every other timed loop in this codebase (`dispatchWait`) solves this with an injectable function this command does not yet have. Recorded here as an out-of-scope gap rather than silently dropped; the single-pass behavior both flags share with `--watch` (grammar refusal, exit code, JSON/human shape) is fully tested. + ## Learnings - **A legacy record's phase renders as `-`, not `active`.** The design says legacy is *treated* as active for slot counting; the presentation layer shows the record said nothing. The two live in different layers on purpose (Task 2's admission reads the design rule; `pr list` reads the record). @@ -195,6 +200,8 @@ Panel: plan-reviewer (conflict lens), plan-reviewer (buildability lens), red-tea - **A transition validates on the way OUT; the breadcrumb writer deliberately does not.** The two rules look contradictory and are not. `writeBreadcrumbFS` stays unvalidated so tests can stage forged records for the loader to reject (Task 1's deviation); `transitionOnce` composes its record from one this build already accepted plus a mutation it wrote itself, so a rejection there is a caller bug that must never reach disk. - **`--adopt-window` needs no "window is in another session" check of its own.** `ResolveWindowExact` already scopes by the parent session's native id, so a same-named window under a different session simply does not resolve. The test for that case pins the property rather than a branch — which is the right shape, since the day someone loosens the resolver is the day the test should go red. - **A record parked in `needs-repair` does NOT block a fresh launch of the same ref.** `recordForRef` skips it on purpose: refusing would make a crashed session permanently un-relaunchable until a human ran `pr repair`, which is the opposite of what the phase is for. It also does not occupy a slot, so the two rules agree. +- **Reusing `phaseLaunchRunner`/`seedPhaseRecord`-style fixtures across a real `Prepare` call needs a real workspace, not an empty string.** `allowsEmptyWorkspace` covers `queued`/`preparing`/`needs-repair`; a seeded `launching` or `prepared` record with `Workspace: ""` fails the loader with "missing workspace" the moment anything reads it back, which reads as a test bug rather than the fixture being wrong for that phase. `fakeWorkspace(t)` is the one already in the package for exactly this. +- **The "no run call inside the lock" assertion pattern (`TestPrepareMany_ReservesUnderOneHoldAndNotAcrossTheClones`) needs narrowing when the locked section legitimately makes ITS OWN run call.** `Drain`'s claim step calls `occupancyFrom` → one `tmux list-windows` read, inside the same hold that also does the claim — by design, the same pattern `reserve`/`Admit` already use. Asserting "zero run calls in the hold" would fail on correct code; the property that matters is "no `git`/`gh` call" (the long operations phase records exist to keep off the lock), so the lock-log test needs to name the specific calls that must never appear rather than every call. - **Holding the lifecycle lock across a tmux READ is new exposure that was not weighed.** `Admit` did not take the lock before Task 2; now it, `reserve`, and `repair` all read `list-windows` under it, so a hung or slow tmux server stalls every other lifecycle-lock user for up to `lockWait` (10s default) rather than hanging only the command that forked tmux. The plan's constraint forbids holding the lock across a clone, a `gh` call, or `tmux new-window` — a local list read is not in that class, and `reserve` already accepted it by design — so this was left as built. The follow-up is an independent timeout on the tmux subprocess, not a lock change; `cmd.Context()` carries no deadline today. (Code review Important 2, recorded rather than fixed.) - **A refusal must not write the write-ahead intent row.** A dangling intent with no completion beside it is precisely the signal "a rollback died mid-delete, a clean room may be orphaned" — so a refusal that wrote one would forge that signal. Every rollback and forget precondition now runs before `beginRepairRow`, which meant hoisting the workspace classification out of `teardownLocked`'s refusal and into the arm itself. - **`--dry-run` must precede every gate, not follow them.** The confirmation check sat above the dry-run branch, so a read-only preview off a terminal was refused for want of `--yes` — the one invocation that could never have mutated anything was the one that needed the destructive flag. diff --git a/internal/cli/pr.go b/internal/cli/pr.go index 2334e922..6983b5a7 100644 --- a/internal/cli/pr.go +++ b/internal/cli/pr.go @@ -75,6 +75,8 @@ human approval gate. forgectl pr teardown discard a session or queue entry forgectl pr repair settle sessions whose record and reality disagree forgectl pr cleanup discard all sessions from a day + forgectl pr queue list reviews waiting for the drainer + forgectl pr drain launch queued reviews as cap slots free up forgectl pr findings list|cleanup reclaim durable local-review findings forgectl pr keys tmux-review cheatsheet @@ -211,6 +213,8 @@ exits 0, to be started later by 'forgectl pr drain --once'.`, newPrTeardownCmd(client), newPrRepairCmd(client), newPrCleanupCmd(client), + newPrQueueCmd(client), + newPrDrainCmd(client, cfg), newPrFindingsCmd(client, th), newPrKeysCmd(), newPrPrsCmd(client, th), @@ -567,6 +571,8 @@ const prKeysText = `clean-room review — tmux keys that matter pr open open a shell in the clean-room workspace pr teardown discard a session or queue entry pr repair settle a session whose record and reality disagree + pr queue list reviews waiting for the drainer + pr drain launch queued reviews as cap slots free up Nothing is posted to the PR without passing forgectl's approval gate. ` diff --git a/internal/cli/pr_drain.go b/internal/cli/pr_drain.go new file mode 100644 index 00000000..073fb2a4 --- /dev/null +++ b/internal/cli/pr_drain.go @@ -0,0 +1,202 @@ +package cli + +import ( + "fmt" + "io" + "strings" + "time" + + "github.com/spf13/cobra" + + "github.com/cameronsjo/forgectl/internal/config" + "github.com/cameronsjo/forgectl/internal/pr" + "github.com/cameronsjo/forgectl/internal/termsafe" +) + +// defaultDrainInterval is --watch's default pass spacing. +const defaultDrainInterval = 60 * time.Second + +// drainWatchRefusalLimit is how many CONSECUTIVE whole-pass refusals --watch +// tolerates before giving up and exiting non-zero. A per-record failure never +// counts against it — only a refusal that launched nothing at all. +const drainWatchRefusalLimit = 3 + +// newPrDrainCmd builds `forgectl pr drain` — the verb that starts queued +// reviews as concurrency-cap slots free up. +// +// ADR-0008 shape: --json on every arm, an honest exit code, and no +// interactive prompt of any kind (drain never shows one). +func newPrDrainCmd(client *pr.Client, cfg config.Config) *cobra.Command { + var ( + once bool + watch bool + interval time.Duration + dryRun bool + asJSON bool + ) + cmd := &cobra.Command{ + Use: "drain", + Short: "Launch queued reviews as concurrency-cap slots free up", + Long: `drain claims the oldest queued reviews — FIFO, up to however many +concurrency-cap slots are currently free — and launches each through the same +path 'pr ' uses. A launch failure is retried on a later pass; after 3 +failed attempts the record is parked in needs-repair instead of retried +forever, and 'forgectl pr repair' is what settles it. + + forgectl pr drain one pass, then exit (the default) + forgectl pr drain --watch keep draining every --interval (default 60s) + forgectl pr drain --dry-run print what a pass would launch, create nothing + forgectl pr drain --json emit the pass report as JSON + +A drainer killed mid-pass leaves 'preparing' or 'launching' records, which +occupy their slots until 'forgectl pr repair' settles them — the next pass +cannot double-launch. + +Exit code for a single pass (the default): 0 when the queue was empty or +every launch succeeded; 1 when the cap or a record could not be read, or any +launch in the pass failed. --watch runs until canceled and exits non-zero +only after three consecutive whole-pass refusals; a per-record failure is +logged and the loop continues.`, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + if watch && once && cmd.Flags().Changed("once") { + return fmt.Errorf("--once and --watch cannot be combined") + } + if cmd.Flags().Changed("interval") && !watch { + return fmt.Errorf("--interval only applies to --watch; add --watch, or drop the flag") + } + if interval <= 0 { + interval = defaultDrainInterval + } + opts := pr.DrainOpts{DryRun: dryRun} + if !watch { + report, err := client.Drain(cmd.Context(), cfg, opts) + if err != nil { + return err + } + report.Pass = 1 + if err := writeDrainReport(cmd.OutOrStdout(), report, asJSON, dryRun, 0); err != nil { + return err + } + return drainExitCode(report) + } + return runDrainWatch(cmd, client, cfg, opts, interval, asJSON, dryRun) + }, + } + cmd.Flags().BoolVar(&once, "once", true, "run a single pass and exit (the default)") + cmd.Flags().BoolVar(&watch, "watch", false, "keep draining on --interval until canceled") + cmd.Flags().DurationVar(&interval, "interval", defaultDrainInterval, "how often --watch drains (requires --watch)") + cmd.Flags().BoolVar(&dryRun, "dry-run", false, "print what a pass would launch and create nothing") + cmd.Flags().BoolVar(&asJSON, "json", false, "emit the pass report as JSON") + return cmd +} + +// runDrainWatch loops Drain on interval until ctx is canceled, printing one +// pass line to stdout every time — the design's requirement that the line +// appears even when the slog handler is discarded, which a default install's +// is. A per-record failure never stops the loop; three CONSECUTIVE +// whole-pass refusals do. +func runDrainWatch(cmd *cobra.Command, client *pr.Client, cfg config.Config, opts pr.DrainOpts, interval time.Duration, asJSON, dryRun bool) error { + ctx := cmd.Context() + out := cmd.OutOrStdout() + maxN := pr.MaxConcurrentReviews(cfg.Pr.MaxConcurrent) + _, _ = fmt.Fprintf(cmd.ErrOrStderr(), "watching: interval=%s cap=%d\n", interval, maxN) + + pass := 0 + consecutiveRefusals := 0 + for { + pass++ + report, err := client.Drain(ctx, cfg, opts) + if err != nil { + return err + } + report.Pass = pass + if werr := writeDrainReport(out, report, asJSON, dryRun, interval); werr != nil { + return werr + } + if report.Refusal != "" { + consecutiveRefusals++ + if consecutiveRefusals >= drainWatchRefusalLimit { + return WithExitCode(fmt.Errorf( + "drain refused %d consecutive passes, last: %s", consecutiveRefusals, report.Refusal), 1) + } + } else { + consecutiveRefusals = 0 + } + select { + case <-ctx.Done(): + return nil + case <-time.After(interval): + } + } +} + +// writeDrainReport prints one pass — JSON or human, dry-run or real — to out. +func writeDrainReport(out io.Writer, report pr.DrainReport, asJSON, dryRun bool, next time.Duration) error { + if asJSON { + return writeDrainJSON(out, report) + } + writeDrainHuman(out, report, dryRun, next) + return nil +} + +// writeDrainJSON encodes the pass report exactly as DrainReport marshals — +// Items is never null (Drain always returns a non-nil slice). +func writeDrainJSON(out io.Writer, report pr.DrainReport) error { + enc := termsafe.JSONEncoder(out) + enc.SetIndent("", " ") + return enc.Encode(report) +} + +// writeDrainHuman renders the one-line-per-pass summary the design names — +// printed regardless of the slog handler, because a default install +// discards it. next is the --watch interval, appended as next=; +// zero on a single pass, where there is no next one. +func writeDrainHuman(out io.Writer, report pr.DrainReport, dryRun bool, next time.Duration) { + if report.Refusal != "" { + _, _ = fmt.Fprintf(out, "pass=%d refused: %s\n", report.Pass, safeTerm(report.Refusal)) + return + } + if dryRun { + if len(report.Items) == 0 { + _, _ = fmt.Fprintln(out, "nothing queued") + return + } + refs := make([]string, 0, len(report.Items)) + for _, it := range report.Items { + refs = append(refs, it.Ref) + } + _, _ = fmt.Fprintf(out, "%d queued, %d free — would launch %s\n", + report.Queued, report.Free, strings.Join(refs, ", ")) + return + } + if report.Queued == 0 && len(report.Items) == 0 { + _, _ = fmt.Fprintln(out, "nothing queued") + return + } + line := fmt.Sprintf("pass=%d free=%d queued=%d launching=%d launched=%d failed=%d", + report.Pass, report.Free, report.Queued, report.Launching, report.Launched, report.Failed) + if next > 0 { + line += fmt.Sprintf(" next=%s", next) + } + _, _ = fmt.Fprintln(out, line) + for _, it := range report.Items { + if it.Outcome == "launched" { + continue + } + _, _ = fmt.Fprintf(out, " %s: %s -> %s: %s\n", it.Ref, it.FromPhase, it.ToPhase, safeTerm(it.Error)) + } +} + +// drainExitCode is the honest code for one pass: 1 when the pass refused +// outright or any launch failed, 0 otherwise — the same "a script can ask +// this" contract `pr repair`'s inspect exit code follows. +func drainExitCode(report pr.DrainReport) error { + if report.Refusal != "" { + return WithExitCode(fmt.Errorf("drain pass refused: %s", report.Refusal), 1) + } + if report.Failed > 0 { + return WithExitCode(fmt.Errorf("%d review(s) failed to launch this pass", report.Failed), 1) + } + return nil +} diff --git a/internal/cli/pr_drain_test.go b/internal/cli/pr_drain_test.go new file mode 100644 index 00000000..11b2be55 --- /dev/null +++ b/internal/cli/pr_drain_test.go @@ -0,0 +1,269 @@ +package cli + +// Test plan for pr_drain.go / pr_queue.go +// +// newPrQueueCmd (Classification: cobra command, read-only view) +// [x] Empty queue prints "no queued reviews", exit 0 +// [x] --json emits [] when empty +// [x] Populated queue lists oldest first, matching drain's own claim order +// +// newPrDrainCmd (Classification: cobra command, ADR-0008 surface) +// [x] --interval without --watch refuses, before any pass runs +// [x] --once and --watch together refuses +// [x] Empty queue: "nothing queued", exit 0 +// [x] --dry-run prints the would-launch line and creates nothing +// [x] A real pass launches a queued ref and prints the pass=... line, exit 0 +// [x] A launch failure exits 1 and names the failed count +// [x] --json emits the report object + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + "time" + + "github.com/cameronsjo/forgectl/internal/config" + "github.com/cameronsjo/forgectl/internal/exec" + "github.com/cameronsjo/forgectl/internal/pr" +) + +// prDrainRunner fakes gh (pr view resolves a valid head), git (clone +// succeeds), and tmux (session/window lifecycle) — new-window fails on the +// 1-indexed call numbers named in failOn. +func prDrainRunner(failOn map[int]error) *exec.FakeRunner { + created := false + call := 0 + return &exec.FakeRunner{RunFunc: func(name string, args []string) (string, error) { + switch { + case name == "gh" && len(args) >= 2 && args[0] == "pr" && args[1] == "view": + return `{"headRefName":"feature","headRefOid":"abc123",` + + `"headRepositoryOwner":{"login":"cameronsjo"},"headRepository":{"name":"forgectl"}}`, nil + case name == "git" && len(args) > 0 && args[0] == "clone": + return "", nil + case name == "tmux" && len(args) > 0: + switch args[0] { + case "-V": + return "tmux 3.7b", nil + case "display-message": + return "123\x1f456\x1f@0", nil + case "list-windows": + return "", nil + case "list-sessions": + if created { + return "123\x1f456\x1f$1\x1fforgectl\x1f1\x1f0\x1f0\x1f/tmp", nil + } + return "", nil + case "new-session": + created = true + return "123\x1f456\x1f$1", nil + case "new-window": + call++ + if err, ok := failOn[call]; ok { + return "", err + } + return fmt.Sprintf("123\x1f456\x1f@%d", call), nil + } + } + return "", nil + }} +} + +func drainCmdClient(t *testing.T, run *exec.FakeRunner) (*pr.Client, string) { + t.Helper() + fakeClaudeBin(t) + dir := t.TempDir() + client := pr.New(run, pr.WithSessionsDir(dir), pr.WithFindingsDir(t.TempDir()), + pr.WithTmuxSession("forgectl"), pr.WithTTYCheck(func() bool { return false })) + return client, dir +} + +func seedQueuedFixture(t *testing.T, dir string, ref pr.Ref, createdAt time.Time) { + t.Helper() + client := pr.New(prDrainRunner(nil), pr.WithSessionsDir(dir)) + if _, err := client.Queue(context.Background(), ref, pr.PrepareOpts{ + Agent: "claude", + Provenance: pr.ReviewProvenanceThirdParty, + }); err != nil { + t.Fatalf("seed queue: %v", err) + } + _ = createdAt // ordering covered at the internal/pr layer; CLI tests need only presence +} + +func runPrQueue(t *testing.T, client *pr.Client, args ...string) (stdout, stderr string, err error) { + t.Helper() + cmd := newPrQueueCmd(client) + var out, errOut bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(&errOut) + cmd.SetArgs(args) + err = cmd.ExecuteContext(context.Background()) + return out.String(), errOut.String(), err +} + +func runPrDrain(t *testing.T, client *pr.Client, cfg config.Config, args ...string) (stdout, stderr string, err error) { + t.Helper() + cmd := newPrDrainCmd(client, cfg) + var out, errOut bytes.Buffer + cmd.SetOut(&out) + cmd.SetErr(&errOut) + cmd.SetArgs(args) + err = cmd.ExecuteContext(context.Background()) + return out.String(), errOut.String(), err +} + +func TestPrQueue_EmptyPrintsNoQueuedReviews(t *testing.T) { + client, _ := drainCmdClient(t, prDrainRunner(nil)) + stdout, _, err := runPrQueue(t, client) + if err != nil { + t.Fatalf("pr queue: %v", err) + } + if strings.TrimSpace(stdout) != "no queued reviews" { + t.Errorf("stdout = %q, want %q", stdout, "no queued reviews") + } +} + +func TestPrQueue_JSONEmptyIsEmptyArray(t *testing.T) { + client, _ := drainCmdClient(t, prDrainRunner(nil)) + stdout, _, err := runPrQueue(t, client, "--json") + if err != nil { + t.Fatalf("pr queue --json: %v", err) + } + if strings.TrimSpace(stdout) != "[]" { + t.Errorf("stdout = %q, want []", stdout) + } +} + +func TestPrQueue_ListsOldestFirst(t *testing.T) { + client, dir := drainCmdClient(t, prDrainRunner(nil)) + ref1 := pr.Ref{Owner: "cameronsjo", Repo: "forgectl", Number: 1} + ref2 := pr.Ref{Owner: "cameronsjo", Repo: "forgectl", Number: 2} + base := time.Now().UTC().Add(-time.Hour) + seedQueuedFixture(t, dir, ref2, base.Add(time.Minute)) + seedQueuedFixture(t, dir, ref1, base) + + stdout, _, err := runPrQueue(t, client) + if err != nil { + t.Fatalf("pr queue: %v", err) + } + lines := strings.Split(strings.TrimSpace(stdout), "\n") + if len(lines) != 2 { + t.Fatalf("lines = %v, want 2", lines) + } + // Both queued via Queue() back-to-back — order is by createdAt, which + // Queue stamps at call time, so whichever was queued FIRST sorts first. + if !strings.HasPrefix(lines[0], ref2.String()) { + t.Errorf("first line = %q, want it to start with the first-queued ref %s", lines[0], ref2.String()) + } +} + +func TestPrDrain_IntervalWithoutWatchRefuses(t *testing.T) { + client, _ := drainCmdClient(t, prDrainRunner(nil)) + _, _, err := runPrDrain(t, client, config.Config{}, "--interval", "5s") + if err == nil { + t.Fatal("expected a refusal: --interval without --watch") + } + if !strings.Contains(err.Error(), "--watch") { + t.Errorf("refusal %q does not name --watch", err) + } +} + +func TestPrDrain_OnceAndWatchTogetherRefuses(t *testing.T) { + client, _ := drainCmdClient(t, prDrainRunner(nil)) + _, _, err := runPrDrain(t, client, config.Config{}, "--once", "--watch") + if err == nil { + t.Fatal("expected a refusal: --once and --watch combined") + } +} + +func TestPrDrain_EmptyQueuePrintsNothingQueued(t *testing.T) { + client, _ := drainCmdClient(t, prDrainRunner(nil)) + stdout, _, err := runPrDrain(t, client, config.Config{}) + if err != nil { + t.Fatalf("pr drain: %v", err) + } + if strings.TrimSpace(stdout) != "nothing queued" { + t.Errorf("stdout = %q, want %q", stdout, "nothing queued") + } +} + +func TestPrDrain_DryRunPrintsWouldLaunchAndCreatesNothing(t *testing.T) { + client, dir := drainCmdClient(t, prDrainRunner(nil)) + ref := pr.Ref{Owner: "cameronsjo", Repo: "forgectl", Number: 1} + seedQueuedFixture(t, dir, ref, time.Now().UTC()) + + stdout, _, err := runPrDrain(t, client, config.Config{}, "--dry-run") + if err != nil { + t.Fatalf("pr drain --dry-run: %v", err) + } + if !strings.Contains(stdout, "would launch") || !strings.Contains(stdout, ref.String()) { + t.Errorf("stdout = %q, want a would-launch line naming %s", stdout, ref.String()) + } + // Nothing changed on disk: the record is still queued. + summaries, _, err := client.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(summaries) != 1 || summaries[0].Phase() != pr.PhaseQueued { + t.Fatalf("summaries = %+v, want exactly one still-queued record", summaries) + } +} + +func TestPrDrain_LaunchesAQueuedReviewAndPrintsPassLine(t *testing.T) { + run := prDrainRunner(nil) + client, dir := drainCmdClient(t, run) + ref := pr.Ref{Owner: "cameronsjo", Repo: "forgectl", Number: 1} + seedQueuedFixture(t, dir, ref, time.Now().UTC()) + + stdout, _, err := runPrDrain(t, client, config.Config{}) + if err != nil { + t.Fatalf("pr drain: %v", err) + } + if !strings.Contains(stdout, "pass=1") || !strings.Contains(stdout, "launched=1") { + t.Errorf("stdout = %q, want a pass line reporting launched=1", stdout) + } + summaries, _, err := client.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(summaries) != 1 || summaries[0].Phase() != pr.PhaseActive { + t.Fatalf("summaries = %+v, want the record active", summaries) + } +} + +func TestPrDrain_LaunchFailureExitsNonZero(t *testing.T) { + run := prDrainRunner(map[int]error{1: errors.New("boom: agent refused")}) + client, dir := drainCmdClient(t, run) + ref := pr.Ref{Owner: "cameronsjo", Repo: "forgectl", Number: 1} + seedQueuedFixture(t, dir, ref, time.Now().UTC()) + + _, _, err := runPrDrain(t, client, config.Config{}) + if err == nil { + t.Fatal("expected a non-zero exit: the launch failed") + } + if ExitCode(err) != 1 { + t.Errorf("exit code = %d, want 1", ExitCode(err)) + } +} + +func TestPrDrain_JSONEmitsReportObject(t *testing.T) { + run := prDrainRunner(nil) + client, dir := drainCmdClient(t, run) + ref := pr.Ref{Owner: "cameronsjo", Repo: "forgectl", Number: 1} + seedQueuedFixture(t, dir, ref, time.Now().UTC()) + + stdout, _, err := runPrDrain(t, client, config.Config{}, "--json") + if err != nil { + t.Fatalf("pr drain --json: %v", err) + } + var report pr.DrainReport + if jerr := json.Unmarshal([]byte(stdout), &report); jerr != nil { + t.Fatalf("stdout did not parse as a DrainReport: %v\n%s", jerr, stdout) + } + if report.Launched != 1 { + t.Errorf("report.Launched = %d, want 1", report.Launched) + } +} diff --git a/internal/cli/pr_queue.go b/internal/cli/pr_queue.go new file mode 100644 index 00000000..13b9ec8b --- /dev/null +++ b/internal/cli/pr_queue.go @@ -0,0 +1,97 @@ +package cli + +import ( + "fmt" + "io" + "sort" + "time" + + "github.com/spf13/cobra" + + "github.com/cameronsjo/forgectl/internal/pr" + "github.com/cameronsjo/forgectl/internal/termsafe" +) + +// newPrQueueCmd builds `forgectl pr queue` — the read-only view of what +// `forgectl pr drain` will pick up next. +func newPrQueueCmd(client *pr.Client) *cobra.Command { + var asJSON bool + cmd := &cobra.Command{ + Use: "queue", + Short: "List reviews waiting for the drainer, oldest first", + Long: `queue lists every session record in the queued phase — deferred by +'pr --queue' or by 'pr pick' past the concurrency cap — sorted oldest +first: the order 'forgectl pr drain' claims them in. + + forgectl pr queue list what is waiting + forgectl pr queue --json the same, as JSON + +A queued record has no workspace and no tmux window yet; 'forgectl pr drain +--once' is what starts it.`, + Args: cobra.NoArgs, + RunE: func(cmd *cobra.Command, _ []string) error { + summaries, unreadable, err := client.List(cmd.Context()) + if err != nil { + return err + } + if unreadable > 0 { + _, _ = fmt.Fprintln(cmd.ErrOrStderr(), unreadableRecordsNote(unreadable)) + } + queued := queuedOldestFirst(summaries) + out := cmd.OutOrStdout() + if asJSON { + return writePrQueueJSON(out, queued) + } + if len(queued) == 0 { + _, _ = fmt.Fprintln(out, "no queued reviews") + return nil + } + for _, s := range queued { + _, _ = fmt.Fprintf(out, "%s\t%s\t%s\n", + s.Ref().String(), s.CreatedAt().Format(time.RFC3339), termsafe.QuotePathIfUnsafe(s.Path())) + } + return nil + }, + } + cmd.Flags().BoolVar(&asJSON, "json", false, `emit [{"ref":...,"created_at":...,"path":...}] to stdout`) + return cmd +} + +// queuedOldestFirst filters summaries to the queued phase, FIFO by +// createdAt — the same order drain.go's claim step claims them in, so this +// view and the drainer never disagree about who is next. +func queuedOldestFirst(summaries []pr.SessionSummary) []pr.SessionSummary { + var queued []pr.SessionSummary + for _, s := range summaries { + if s.Phase() == pr.PhaseQueued { + queued = append(queued, s) + } + } + sort.Slice(queued, func(i, j int) bool { + return queued[i].CreatedAt().Before(queued[j].CreatedAt()) + }) + return queued +} + +// prQueueRowJSON is the --json wire shape for one `pr queue` row. +type prQueueRowJSON struct { + Ref string `json:"ref"` + CreatedAt string `json:"created_at"` + Path string `json:"path"` +} + +// writePrQueueJSON encodes the queued rows as a JSON array, [] rather than +// null when empty — matching `pr list --json`'s empty-array contract. +func writePrQueueJSON(out io.Writer, queued []pr.SessionSummary) error { + rows := make([]prQueueRowJSON, 0, len(queued)) + for _, s := range queued { + rows = append(rows, prQueueRowJSON{ + Ref: s.Ref().String(), + CreatedAt: s.CreatedAt().Format(time.RFC3339), + Path: s.Path(), + }) + } + enc := termsafe.JSONEncoder(out) + enc.SetIndent("", " ") + return enc.Encode(rows) +} diff --git a/internal/pr/drain.go b/internal/pr/drain.go new file mode 100644 index 00000000..9002492b --- /dev/null +++ b/internal/pr/drain.go @@ -0,0 +1,294 @@ +package pr + +import ( + "context" + "fmt" + "log/slog" + "sort" + "time" + + "github.com/cameronsjo/forgectl/internal/config" + "github.com/cameronsjo/forgectl/internal/termsafe" +) + +// DefaultDrainMaxAttempts is the number of failed launch attempts a queued +// record survives before the drainer parks it in needs-repair rather than +// handing it back to the queue for another pass to retry. +const DefaultDrainMaxAttempts = 3 + +// Drain outcomes. "launched" and "would-launch" mirror the launch/dry-run +// split every other verb in this package uses; "retry-queued" and +// "needs-repair" name the two ways a launch failure can land. +const ( + drainOutcomeLaunched = "launched" + drainOutcomeWouldLaunch = "would-launch" + drainOutcomeRetryQueued = "retry-queued" + drainOutcomeNeedsRepair = "needs-repair" + drainOutcomeClaimFailure = "claim-failed" +) + +// DrainOpts drives one `pr drain` pass. +// +// Once, Watch, and Interval are carried here to complete the shape the design +// names, but Drain itself always performs exactly ONE PASS — the watch loop, +// its per-pass stdout line, and its backoff-then-refuse policy live in the +// CLI (internal/cli/pr_drain.go), which calls Drain once per tick. That +// split keeps this package free of anything that prints or sleeps, matching +// every other verb here (Repair, Prune): the ops layer returns a report: the +// CLI decides how often to ask for one and what to do with it. +type DrainOpts struct { + // Once is accepted for shape completeness; Drain does not read it — one + // call is always one pass. + Once bool + // Watch is accepted for shape completeness; the CLI reads it to decide + // whether to loop, not this function. + Watch bool + // Interval is accepted for shape completeness; the CLI reads it to time + // the loop, not this function. + Interval time.Duration + // DryRun claims nothing and launches nothing: it reports which queued + // records the pass would have claimed, in claim order, and touches no + // record, workspace, or tmux window. + DryRun bool + // MaxAttempts is how many failed launches a queued record survives before + // the drainer parks it in needs-repair instead of returning it to the + // queue. Non-positive resolves to DefaultDrainMaxAttempts. + MaxAttempts int +} + +// DrainItem is one row of a drain pass report — the queued record claimed (or +// that would have been claimed on --dry-run) and what happened to it. +type DrainItem struct { + Ref string `json:"ref"` + RecordPath string `json:"record_path"` + FromPhase string `json:"from_phase"` + ToPhase string `json:"to_phase,omitempty"` + Outcome string `json:"outcome"` + Error string `json:"error,omitempty"` +} + +// DrainReport is what one `pr drain` pass returns and `--json` encodes. +// +// Refusal is set, and Items left empty, when the whole pass refused before +// claiming anything: an unreadable cap, an unreadable record, or a lock +// timeout. A partial pass (some items launched, some failed) is never a +// Refusal — it is reported item by item, exactly as `pr repair`'s inspect +// reports one unsettled record per row rather than a single "something is +// wrong". +type DrainReport struct { + Pass int `json:"pass"` + Free int `json:"free"` + Queued int `json:"queued"` + Launching int `json:"launching"` + Launched int `json:"launched"` + Failed int `json:"failed"` + Items []DrainItem `json:"items"` + Refusal string `json:"refusal,omitempty"` +} + +// Drain performs one drain pass: it takes the lifecycle lock, counts +// occupancy, refuses the WHOLE pass if any record could not be read, and +// otherwise claims the oldest queued records — by createdAt, FIFO — up to +// however many slots are free, transitioning each to `preparing` under that +// SAME lock hold. The lock is released before any claimed record is prepared +// or launched: the clone and the dispatch both run outside it, against slots +// already claimed, exactly as every other launch path in this package does. +// +// Each claimed record is then prepared and launched through the identical +// Prepare -> Launch path `pr ` uses. A launch failure increments the +// record's Attempts, records LastError/LastAttempt, and returns it to +// `queued` for the next pass to retry — unless Attempts has reached +// opts.MaxAttempts, in which case it is parked in `needs-repair` with a +// reason naming the attempt count and the last error, and no further pass +// will pick it up (queued records are the only ones drain claims). +func (c *Client) Drain(ctx context.Context, cfg config.Config, opts DrainOpts) (DrainReport, error) { + if opts.MaxAttempts <= 0 { + opts.MaxAttempts = DefaultDrainMaxAttempts + } + report, claimed := c.claimQueuedPass(ctx, cfg, opts) + if report.Refusal != "" { + return report, nil + } + if opts.DryRun { + for _, s := range claimed { + report.Items = append(report.Items, DrainItem{ + Ref: s.Ref().String(), + RecordPath: s.Path(), + FromPhase: string(PhaseQueued), + Outcome: drainOutcomeWouldLaunch, + }) + } + return report, nil + } + for _, s := range claimed { + item := c.drainItem(ctx, cfg, s, opts.MaxAttempts) + report.Items = append(report.Items, item) + if item.Outcome == drainOutcomeLaunched { + report.Launched++ + } else { + report.Failed++ + } + } + return report, nil +} + +// claimQueuedPass takes the lifecycle lock once, counts occupancy, and — off +// --dry-run — transitions the oldest free-slot's-worth of queued records to +// `preparing`. On --dry-run it claims nothing and returns the records that +// WOULD have been claimed, so the caller's would-launch rows name exactly the +// same set a real pass would have started on. +// +// It refuses the whole pass, before claiming anything, when List reports an +// unreadable record or when the occupancy count could not be read (tmux +// unreadable) — the same fail-closed rule every other counting arm in this +// package follows (admission.go's occupiedLocked, reserve's openReservation). +func (c *Client) claimQueuedPass(ctx context.Context, cfg config.Config, opts DrainOpts) (DrainReport, []SessionSummary) { + report := DrainReport{Items: []DrainItem{}} + var claimed []SessionSummary + err := c.withLifecycleLock(ctx, "drain", func() error { + summaries, unreadable, lerr := c.listLocked() + if lerr != nil { + return lerr + } + if len(unreadable) > 0 { + return fmt.Errorf("%d session record(s) could not be read, so the free-slot count would be wrong — "+ + "settle them with 'forgectl pr repair' before draining", len(unreadable)) + } + occupied, lerr := c.occupancyFrom(ctx, summaries) + if lerr != nil { + return lerr + } + maxN := MaxConcurrentReviews(cfg.Pr.MaxConcurrent) + free := maxN - occupied + if free < 0 { + free = 0 + } + report.Free = free + + var queued []SessionSummary + for _, s := range summaries { + switch s.Phase() { + case PhaseQueued: + queued = append(queued, s) + case PhasePreparing, PhasePrepared, PhaseLaunching: + report.Launching++ + } + } + report.Queued = len(queued) + sort.Slice(queued, func(i, j int) bool { + return queued[i].CreatedAt().Before(queued[j].CreatedAt()) + }) + + n := free + if n > len(queued) { + n = len(queued) + } + for i := 0; i < n; i++ { + s := queued[i] + if opts.DryRun { + claimed = append(claimed, s) + continue + } + if terr := c.transitionLocked(s.Path(), PhaseQueued, PhasePreparing, nil); terr != nil { + slog.Error("Refusing to claim a queued review: the record could not be moved to preparing.", + "ref", s.Ref().String(), "path", s.Path(), "error", terr) + continue + } + claimed = append(claimed, s) + } + return nil + }) + if err != nil { + report.Refusal = err.Error() + } + return report, claimed +} + +// drainItem prepares and launches one already-claimed (`preparing`) record +// through the same Prepare -> Launch path a human's `pr ` uses, and +// settles a launch failure per the attempt-count policy above. +func (c *Client) drainItem(ctx context.Context, cfg config.Config, s SessionSummary, maxAttempts int) DrainItem { + ref := s.Ref() + path := s.Path() + item := DrainItem{Ref: ref.String(), RecordPath: path, FromPhase: string(PhaseQueued)} + + bc, _, err := loadBreadcrumbRecord(path, c.sessionsDir) + if err != nil { + item.Outcome = drainOutcomeClaimFailure + item.Error = termsafe.SafeLine(err.Error()) + return item + } + + prepOpts := PrepareOpts{ + Agent: bc.Agent, + Provenance: ParseReviewProvenance(bc.Provenance), + RecordPath: path, + } + sess, err := c.Prepare(ctx, ref, prepOpts) + if err == nil { + _, err = c.Launch(ctx, sess, cfg) + } + if err == nil { + item.ToPhase = string(PhaseActive) + item.Outcome = drainOutcomeLaunched + return item + } + + item.Error = termsafe.SafeLine(err.Error()) + outcome, toPhase := c.settleDrainFailure(ctx, path, bc.Attempts, maxAttempts, err) + item.Outcome = outcome + item.ToPhase = toPhase + return item +} + +// settleDrainFailure records the failed attempt on the record — regardless of +// what phase Prepare/Launch left it in (queued's own `preparing`, or a +// needs-repair Launch itself already wrote on a dispatch failure) — and moves +// it to `queued` for a future pass to retry, or to `needs-repair` once +// attempts is exhausted. It uses the wildcard `from` (anyPhase) for the same +// reason markNeedsRepair does: the caller does not know, and must not have to +// know, which phase the failure left the record in. +func (c *Client) settleDrainFailure(ctx context.Context, path string, priorAttempts, maxAttempts int, cause error) (outcome, toPhase string) { + attempts := priorAttempts + 1 + lastError := termsafe.SafeLine(cause.Error()) + exhausted := attempts >= maxAttempts + + target := PhaseQueued + outcome = drainOutcomeRetryQueued + if exhausted { + target = PhaseNeedsRepair + outcome = drainOutcomeNeedsRepair + } + + err := c.transition(ctx, path, anyPhase, target, func(bc *Breadcrumb) error { + bc.Attempts = attempts + bc.LastError = lastError + bc.LastAttempt = time.Now().UTC() + if exhausted { + bc.RepairReason = fmt.Sprintf("drain: %d attempts, last: %s", attempts, lastError) + // A retry that got as far as a workspace leaves it behind for + // `pr repair` to inspect; needs-repair does not require an empty + // workspace. + } else { + bc.RepairReason = "" + // queued must not carry a workspace: a retried Prepare clones a + // fresh one, so any workspace this failed attempt created would + // otherwise leak. Best-effort teardown; its own failure must not + // shadow the launch error already being reported. + if bc.Workspace != "" { + if terr := sandboxTeardown(ctx, c.run, bc.Workspace); terr != nil { + slog.Error("Failed to tear down the workspace from a failed drain attempt; it may need manual removal.", + "path", path, "workspace", bc.Workspace, "error", terr) + } + bc.Workspace = "" + } + } + return nil + }) + if err != nil { + slog.Error("Failed to settle a drain launch failure; the record's attempt count was not recorded.", + "path", path, "target", string(target), "error", err) + return drainOutcomeClaimFailure, "" + } + return outcome, string(target) +} diff --git a/internal/pr/drain_test.go b/internal/pr/drain_test.go new file mode 100644 index 00000000..8906470d --- /dev/null +++ b/internal/pr/drain_test.go @@ -0,0 +1,448 @@ +package pr + +// Test plan for drain.go +// +// Drain (Classification: composite verb, crash-safety bridge) +// [x] Claims the oldest queued records first, under ONE lock hold, and +// launches them through the ordinary Prepare -> Launch path +// [x] A `preparing`/`prepared`/`launching` record occupies a slot exactly +// as it does for reserve/Admit, so drain claims fewer queued records +// when one is already in flight +// [x] An unreadable record refuses the WHOLE pass before claiming anything +// [x] A launch failure returns the record to `queued` with attempts=1 and +// lastError set +// [x] A third failure (attempts already 2) moves the record to +// `needs-repair` with a reason naming the attempt count, and it is no +// longer counted as `queued` on the next listing +// [x] Two Clients racing one queued record launch it exactly once +// [x] `--dry-run` claims nothing, launches nothing, and reports would-launch +// [x] A mixed pass (one success, one failure) reports both items +// [x] An empty queue reports zero queued/launched/failed and no items + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/cameronsjo/forgectl/internal/config" + "github.com/cameronsjo/forgectl/internal/exec" +) + +// drainLaunchRunner fakes gh (pr view always resolves a valid head), git +// (clone always succeeds), and tmux — new-window fails on the call numbers +// named in failOn (1-indexed, in invocation order) and succeeds with a fresh +// generation-qualified identity otherwise. +func drainLaunchRunner(failOn map[int]error) *exec.FakeRunner { + created := false + call := 0 + return &exec.FakeRunner{RunFunc: func(name string, args []string) (string, error) { + switch { + case name == "gh" && len(args) >= 2 && args[0] == "pr" && args[1] == "view": + return `{"headRefName":"feature","headRefOid":"abc123",` + + `"headRepositoryOwner":{"login":"cameronsjo"},"headRepository":{"name":"forgectl"}}`, nil + case name == "git" && len(args) > 0 && args[0] == "clone": + return "", nil + case name == "tmux" && len(args) > 0: + switch args[0] { + case "-V": + return "tmux 3.7b", nil + case "display-message": + return "123\x1f456\x1f@0", nil + case "list-sessions": + if created { + return "123\x1f456\x1f$1\x1fforgectl\x1f1\x1f0\x1f0\x1f/tmp", nil + } + return "", nil + case "list-windows": + return "", nil + case "new-session": + created = true + return "123\x1f456\x1f$1", nil + case "new-window": + call++ + if err, ok := failOn[call]; ok { + return "", err + } + return fmt.Sprintf("123\x1f456\x1f@%d", call), nil + } + } + return "", nil + }} +} + +// drainClient builds a Client wired for drain tests: real sessions dir, +// fakeClaude on PATH via env, no interactive TTY concerns (Drain never shows +// one). +func drainClient(t *testing.T, dir string, run *exec.FakeRunner) *Client { + t.Helper() + fakeClaude(t) + return New(run, WithSessionsDir(dir), WithFindingsDir(t.TempDir()), + WithTmuxSession("forgectl"), WithLockWait(2*time.Second)) +} + +// seedQueued writes a `queued` record directly (bypassing Client.Queue) so +// tests can pin CreatedAt for deterministic FIFO ordering and pre-seed +// Attempts for the exhaustion test. +func seedQueued(t *testing.T, c *Client, ref Ref, createdAt time.Time, attempts int) string { + t.Helper() + bc := Breadcrumb{ + Ref: ref.String(), Agent: "claude", CreatedAt: createdAt, + Version: breadcrumbVersion, Phase: PhaseQueued, Revision: 1, + Attempts: attempts, + } + path, err := writeBreadcrumb(c.SessionsDir(), ref, bc) + if err != nil { + t.Fatalf("seed queued record: %v", err) + } + return path +} + +func TestDrain_ClaimsOldestQueuedUnderOneLockHoldAndLaunches(t *testing.T) { + dir := t.TempDir() + run := drainLaunchRunner(nil) + inner := run.RunFunc + var mu sync.Mutex + var log []string + note := func(s string) { mu.Lock(); log = append(log, s); mu.Unlock() } + run.RunFunc = func(name string, args []string) (string, error) { + note("run:" + name) + return inner(name, args) + } + c := drainClient(t, dir, run) + c.onLock = func(verb, event string) { note(event + ":" + verb) } + + base := time.Now().UTC().Add(-time.Hour) + seedQueued(t, c, testRef(1), base, 0) + seedQueued(t, c, testRef(2), base.Add(time.Minute), 0) + seedQueued(t, c, testRef(3), base.Add(2*time.Minute), 0) // newest — must stay queued + + report, err := c.Drain(context.Background(), config.Config{Pr: config.PrConfig{MaxConcurrent: 2}}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if report.Free != 2 || report.Queued != 3 { + t.Fatalf("report = %+v, want Free=2 Queued=3", report) + } + if report.Launched != 2 || report.Failed != 0 { + t.Fatalf("report = %+v, want Launched=2 Failed=0", report) + } + gotRefs := map[string]bool{} + for _, item := range report.Items { + gotRefs[item.Ref] = true + if item.Outcome != drainOutcomeLaunched || item.ToPhase != string(PhaseActive) { + t.Errorf("item %+v, want outcome=launched toPhase=active", item) + } + } + if !gotRefs[testRef(1).String()] || !gotRefs[testRef(2).String()] { + t.Fatalf("launched refs = %v, want the two oldest (1 and 2)", gotRefs) + } + + // The newest stays queued. + third := readRecordByRef(t, dir, testRef(3)) + if third.Phase != PhaseQueued { + t.Errorf("newest record phase = %q, want still queued", third.Phase) + } + launchedOldest := readRecordByRef(t, dir, testRef(1)) + if launchedOldest.Phase != PhaseActive { + t.Errorf("oldest record phase = %q, want active", launchedOldest.Phase) + } + + // The claim (queued -> preparing transitions) happened under ONE lock + // hold. The hold legitimately makes one tmux list-windows call to count + // occupancy (the same pattern reserve/Admit already use) — what must + // NEVER fall inside it is the clone (`git`) or the gh round-trip, since + // those are the long operations phase records exist to keep off the + // lock. + mu.Lock() + defer mu.Unlock() + start, end := -1, -1 + holds := 0 + for i, e := range log { + if e == "acquire:drain" { + holds++ + if start < 0 { + start = i + } + } + if e == "release:drain" && end < 0 && start >= 0 { + end = i + } + } + if holds != 1 { + t.Fatalf("drain lock held %d times, want exactly 1", holds) + } + for i := start; i <= end; i++ { + if log[i] == "run:git" || log[i] == "run:gh" { + t.Errorf("run call %q (clone/gh round-trip) happened inside the drain lock hold", log[i]) + } + } +} + +// readRecordByRef finds the one record on disk for ref and decodes it. +func readRecordByRef(t *testing.T, dir string, ref Ref) Breadcrumb { + t.Helper() + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatalf("read dir: %v", err) + } + for _, e := range entries { + if filepath.Ext(e.Name()) != ".json" { + continue + } + path := filepath.Join(dir, e.Name()) + bc := readRecord(t, path) + if bc.Ref == ref.String() { + return bc + } + } + t.Fatalf("no record found for ref %s", ref.String()) + return Breadcrumb{} +} + +func TestDrain_InFlightRecordsConsumeSlots(t *testing.T) { + dir := t.TempDir() + c := drainClient(t, dir, drainLaunchRunner(nil)) + + // Two slots already occupied: one preparing, one launching. Default cap + // is 4, so only two slots remain free for three queued refs. Preparing + // allows no workspace (the clone has not landed); launching requires one + // (it has already cloned). + seedPhaseRecord(t, c, testRef(10), PhasePreparing, "") + seedPhaseRecord(t, c, testRef(11), PhaseLaunching, fakeWorkspace(t)) + + base := time.Now().UTC().Add(-time.Hour) + seedQueued(t, c, testRef(1), base, 0) + seedQueued(t, c, testRef(2), base.Add(time.Minute), 0) + seedQueued(t, c, testRef(3), base.Add(2*time.Minute), 0) + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if report.Free != 2 { + t.Fatalf("Free = %d, want 2 (cap 4 minus 2 in-flight records)", report.Free) + } + if report.Launching != 2 { + t.Errorf("Launching = %d, want 2", report.Launching) + } + if len(report.Items) != 2 { + t.Fatalf("claimed %d items, want 2 (one queued record left behind)", len(report.Items)) + } +} + +func TestDrain_UnreadableRecordRefusesWholePass(t *testing.T) { + dir := t.TempDir() + c := drainClient(t, dir, drainLaunchRunner(nil)) + seedQueued(t, c, testRef(1), time.Now().UTC(), 0) + bad := []byte(`{"workspace":"/tmp/x","ref":"o/r#9","createdAt":"2026-09-11T00:00:00Z","futureKey":true}` + "\n") + if err := os.WriteFile(filepath.Join(dir, "o-r-9-1.json"), bad, 0o600); err != nil { + t.Fatal(err) + } + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if report.Refusal == "" { + t.Fatal("expected a pass refusal naming the unreadable record") + } + if !strings.Contains(report.Refusal, "could not be read") { + t.Errorf("refusal = %q, want it to name the unreadable record", report.Refusal) + } + if len(report.Items) != 0 { + t.Errorf("items = %v, want none — an unreadable cap must launch nothing", report.Items) + } + // Nothing was claimed: the queued record is untouched. + bc := readRecordByRef(t, dir, testRef(1)) + if bc.Phase != PhaseQueued { + t.Errorf("queued record phase = %q, want still queued", bc.Phase) + } +} + +func TestDrain_LaunchFailureReturnsToQueuedWithAttempts(t *testing.T) { + dir := t.TempDir() + run := drainLaunchRunner(map[int]error{1: errors.New("boom: agent refused")}) + c := drainClient(t, dir, run) + seedQueued(t, c, testRef(1), time.Now().UTC(), 0) + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if report.Launched != 0 || report.Failed != 1 { + t.Fatalf("report = %+v, want Launched=0 Failed=1", report) + } + if len(report.Items) != 1 || report.Items[0].Outcome != drainOutcomeRetryQueued { + t.Fatalf("items = %+v, want one retry-queued item", report.Items) + } + bc := readRecordByRef(t, dir, testRef(1)) + if bc.Phase != PhaseQueued { + t.Fatalf("phase = %q, want queued (a failure retries)", bc.Phase) + } + if bc.Attempts != 1 { + t.Errorf("attempts = %d, want 1", bc.Attempts) + } + if bc.LastError == "" { + t.Error("lastError is empty, want it set") + } + if bc.LastAttempt.IsZero() { + t.Error("lastAttemptAt is zero, want it set") + } +} + +func TestDrain_ThirdFailureMovesToNeedsRepair(t *testing.T) { + dir := t.TempDir() + run := drainLaunchRunner(map[int]error{1: errors.New("boom: agent refused again")}) + c := drainClient(t, dir, run) + // Two prior failed attempts already recorded. + seedQueued(t, c, testRef(1), time.Now().UTC(), 2) + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if len(report.Items) != 1 || report.Items[0].Outcome != drainOutcomeNeedsRepair { + t.Fatalf("items = %+v, want one needs-repair item", report.Items) + } + bc := readRecordByRef(t, dir, testRef(1)) + if bc.Phase != PhaseNeedsRepair { + t.Fatalf("phase = %q, want needs-repair at 3 attempts", bc.Phase) + } + if bc.Attempts != 3 { + t.Errorf("attempts = %d, want 3", bc.Attempts) + } + if !strings.Contains(bc.RepairReason, "drain: 3 attempts") { + t.Errorf("repairReason = %q, want it to name the attempt count", bc.RepairReason) + } + + // The next pass does not pick it up: it is no longer `queued`. + report2, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("second Drain: %v", err) + } + if report2.Queued != 0 || len(report2.Items) != 0 { + t.Fatalf("second pass = %+v, want nothing queued and nothing claimed", report2) + } +} + +func TestDrain_TwoClientsAgainstOneQueuedRecordLaunchOnce(t *testing.T) { + dir := t.TempDir() + var launches int32Counter + run := drainLaunchRunner(nil) + inner := run.RunFunc + run.RunFunc = func(name string, args []string) (string, error) { + if name == "tmux" && len(args) > 0 && args[0] == "new-window" { + launches.add(1) + } + return inner(name, args) + } + mk := func() *Client { return drainClient(t, dir, run) } + first, second := mk(), mk() + seedQueued(t, first, testRef(1), time.Now().UTC(), 0) + + var wg sync.WaitGroup + reports := make([]DrainReport, 2) + errs := make([]error, 2) + wg.Add(2) + go func() { + defer wg.Done() + reports[0], errs[0] = first.Drain(context.Background(), config.Config{}, DrainOpts{}) + }() + go func() { + defer wg.Done() + reports[1], errs[1] = second.Drain(context.Background(), config.Config{}, DrainOpts{}) + }() + wg.Wait() + + if errs[0] != nil || errs[1] != nil { + t.Fatalf("Drain errors: %v, %v", errs[0], errs[1]) + } + if got := launches.get(); got != 1 { + t.Fatalf("tmux new-window called %d times, want exactly 1", got) + } + launchedTotal := reports[0].Launched + reports[1].Launched + if launchedTotal != 1 { + t.Fatalf("total launched across both clients = %d, want 1", launchedTotal) + } +} + +// int32Counter is a tiny race-safe counter, local to this test file so it +// carries no dependency beyond sync. +type int32Counter struct { + mu sync.Mutex + n int +} + +func (c *int32Counter) add(d int) { c.mu.Lock(); c.n += d; c.mu.Unlock() } +func (c *int32Counter) get() int { c.mu.Lock(); defer c.mu.Unlock(); return c.n } + +func TestDrain_DryRunCreatesNothingAndReportsWouldLaunch(t *testing.T) { + dir := t.TempDir() + c := drainClient(t, dir, drainLaunchRunner(nil)) + seedQueued(t, c, testRef(1), time.Now().UTC(), 0) + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{DryRun: true}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if len(report.Items) != 1 || report.Items[0].Outcome != drainOutcomeWouldLaunch { + t.Fatalf("items = %+v, want one would-launch item", report.Items) + } + bc := readRecordByRef(t, dir, testRef(1)) + if bc.Phase != PhaseQueued { + t.Fatalf("phase = %q, want unchanged queued — dry-run must create nothing", bc.Phase) + } + entries, _ := os.ReadDir(dir) + jsonFiles := 0 + for _, e := range entries { + if filepath.Ext(e.Name()) == ".json" { + jsonFiles++ + } + } + if jsonFiles != 1 { + t.Errorf("session dir has %d records, want 1 — dry-run must create no new record", jsonFiles) + } +} + +func TestDrain_MixedPassOneSuccessOneFailure(t *testing.T) { + dir := t.TempDir() + // First claimed (oldest) succeeds; second fails. + run := drainLaunchRunner(map[int]error{2: errors.New("boom: second agent refused")}) + c := drainClient(t, dir, run) + base := time.Now().UTC().Add(-time.Hour) + seedQueued(t, c, testRef(1), base, 0) + seedQueued(t, c, testRef(2), base.Add(time.Minute), 0) + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if report.Launched != 1 || report.Failed != 1 { + t.Fatalf("report = %+v, want Launched=1 Failed=1", report) + } + if len(report.Items) != 2 { + t.Fatalf("items = %+v, want 2", report.Items) + } +} + +func TestDrain_EmptyQueueReportsZero(t *testing.T) { + dir := t.TempDir() + c := drainClient(t, dir, drainLaunchRunner(nil)) + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if report.Queued != 0 || report.Launched != 0 || report.Failed != 0 || len(report.Items) != 0 { + t.Fatalf("report = %+v, want an all-zero empty report", report) + } + if report.Free != DefaultMaxConcurrentReviews { + t.Errorf("Free = %d, want the full default cap", report.Free) + } +} diff --git a/scripts/dogfood-drain.sh b/scripts/dogfood-drain.sh new file mode 100755 index 00000000..49434168 --- /dev/null +++ b/scripts/dogfood-drain.sh @@ -0,0 +1,75 @@ +#!/usr/bin/env bash +# Dogfoods `forgectl pr --queue` + `forgectl pr drain` against two real +# PRs (forgectl#473): queues both, drains one pass, and asserts the pass +# report. Uses a scratch HOME so it never touches a real ~/.config/forgectl. +# +# Usage: scripts/dogfood-drain.sh [--dry-run] [path/to/forgectl] +# +# --dry-run queues both refs for real, then runs `pr drain --dry-run --json` +# and asserts the report names both refs as would-launch — it creates no +# workspace and dispatches no tmux window. Without --dry-run the drain pass +# is a REAL launch: it clones each head, dispatches a review agent into a +# tmux window under the `forgectl` session, and is the orchestrator's to run +# — not a step this script takes on its own. +set -uo pipefail + +DRY_RUN=false +if [ "${1:-}" = "--dry-run" ]; then + DRY_RUN=true + shift +fi + +REF1=${1:-} +REF2=${2:-} +BIN=${3:-$(command -v forgectl || true)} + +if [ -z "$REF1" ] || [ -z "$REF2" ]; then + echo "usage: $0 [--dry-run] [path/to/forgectl]" >&2 + exit 2 +fi +if [ -z "$BIN" ] || [ ! -x "$BIN" ]; then + echo "VERDICT: FAIL no forgectl binary (pass a path or put one on PATH)" >&2 + exit 2 +fi + +SCRATCH=$(mktemp -d) +trap 'rm -rf "$SCRATCH"' EXIT + +echo "--- queue $REF1 ---" +HOME="$SCRATCH" "$BIN" pr "$REF1" --queue +echo "--- queue $REF2 ---" +HOME="$SCRATCH" "$BIN" pr "$REF2" --queue + +echo "--- pr queue ---" +HOME="$SCRATCH" "$BIN" pr queue + +DRAIN_ARGS=(pr drain --once --json) +if [ "$DRY_RUN" = true ]; then + DRAIN_ARGS=(pr drain --once --dry-run --json) +fi + +echo "--- ${DRAIN_ARGS[*]} ---" +REPORT=$(HOME="$SCRATCH" "$BIN" "${DRAIN_ARGS[@]}") +RC=$? +echo "$REPORT" + +if [ "$RC" -ne 0 ]; then + echo "VERDICT: FAIL drain exited $RC" >&2 + exit 1 +fi + +if [ "$DRY_RUN" = true ]; then + if ! command grep -q "$REF1" <<<"$REPORT" || ! command grep -q "$REF2" <<<"$REPORT"; then + echo "VERDICT: FAIL dry-run report does not name both refs as would-launch" >&2 + exit 1 + fi + echo "VERDICT: PASS dry-run named both refs; nothing was launched" + exit 0 +fi + +LAUNCHED=$(printf '%s' "$REPORT" | command grep -o '"launched":[0-9]*' | head -1 | command grep -o '[0-9]*$') +if [ "${LAUNCHED:-0}" -ne 2 ]; then + echo "VERDICT: FAIL launched=$LAUNCHED, want 2" >&2 + exit 1 +fi +echo "VERDICT: PASS drained and launched both queued reviews" From dd49816fe5aecb56ace455d9881335d3bea4bf2d Mon Sep 17 00:00:00 2001 From: Cameron Sjo <4084915+cameronsjo@users.noreply.github.com> Date: Sat, 12 Sep 2026 10:39:10 -0500 Subject: [PATCH 2/4] fix(scripts): make the drain dogfood dry-run by default and --launch opt-in (#473) The script launched real reviewer sessions unless --dry-run was remembered. Muscle memory now runs the safe path; the destructive one has to be named. Verified: the default path passes under a live tmux server and dispatches no window; without a server it fails closed on the unreadable cap. Session-Id: c2d13fd6-30bc-409a-989c-cd5ad22073fd Model: claude-fable-5-1 Harness: claude-code 2.1.269 Machine: cf6e768835c7 Co-Authored-By: Claude Fable 5.1 --- scripts/dogfood-drain.sh | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/scripts/dogfood-drain.sh b/scripts/dogfood-drain.sh index 49434168..38619620 100755 --- a/scripts/dogfood-drain.sh +++ b/scripts/dogfood-drain.sh @@ -3,19 +3,22 @@ # PRs (forgectl#473): queues both, drains one pass, and asserts the pass # report. Uses a scratch HOME so it never touches a real ~/.config/forgectl. # -# Usage: scripts/dogfood-drain.sh [--dry-run] [path/to/forgectl] +# Usage: scripts/dogfood-drain.sh [--launch] [path/to/forgectl] # -# --dry-run queues both refs for real, then runs `pr drain --dry-run --json` -# and asserts the report names both refs as would-launch — it creates no -# workspace and dispatches no tmux window. Without --dry-run the drain pass -# is a REAL launch: it clones each head, dispatches a review agent into a -# tmux window under the `forgectl` session, and is the orchestrator's to run -# — not a step this script takes on its own. +# By default the script queues both refs (inside the scratch HOME), runs +# `pr drain --dry-run --json`, and asserts the report names both refs as +# would-launch. It creates no workspace and dispatches no tmux window. The +# live pass is opt-in: with --launch the drain clones each head and dispatches +# a review agent into a tmux window under the `forgectl` session. Muscle memory +# runs the safe path; the destructive one has to be named. set -uo pipefail -DRY_RUN=false -if [ "${1:-}" = "--dry-run" ]; then - DRY_RUN=true +DRY_RUN=true +if [ "${1:-}" = "--launch" ]; then + DRY_RUN=false + shift +elif [ "${1:-}" = "--dry-run" ]; then + # Accepted for compatibility; it is already the default. shift fi @@ -24,7 +27,7 @@ REF2=${2:-} BIN=${3:-$(command -v forgectl || true)} if [ -z "$REF1" ] || [ -z "$REF2" ]; then - echo "usage: $0 [--dry-run] [path/to/forgectl]" >&2 + echo "usage: $0 [--launch] [path/to/forgectl]" >&2 exit 2 fi if [ -z "$BIN" ] || [ ! -x "$BIN" ]; then From 4e73dfac79ac6593fe0fd03c2a1b4622fce3f206 Mon Sep 17 00:00:00 2001 From: Cameron Sjo <4084915+cameronsjo@users.noreply.github.com> Date: Sat, 12 Sep 2026 10:49:21 -0500 Subject: [PATCH 3/4] fix(pr): never retry over a live window, refuse local queue entries, and release the lock before teardown (#473) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. A retry no longer tears down a workspace whose review agent may be live. Launch parks the record in needs-repair with a reason naming the window on both branches where the window already exists, and settleDrainFailure was erasing that reason, removing the clean room under the running agent, and returning the ref to queued — so a later pass would launch a second agent for the same ref. The settlement now re-reads the record and asks tmux first: an already-parked record, a window resolvable by the derived name, or an unreadable tmux (fail-closed) all mean leave it parked, keep the workspace, stop retrying. Only a failure with neither signal — a gh or clone failure before dispatch — requeues. 2. Drain mirrors Queue's local-ref refusal at the reader. A queued record marked local is refused at claim time: never claimed to preparing, never cloned, reported as a refused item. Dispatch was previously stopped only by Launch's own refusal, which its comment calls an incidental second barrier. 3. The lifecycle lock is no longer held across os.RemoveAll of a clone. The transition mutator is side-effect-free and idempotent (transitionLocked may run it twice); it records the workspace it cleared, and the teardown runs after the lock is released, through the same bounded sandboxTeardown path. 4. scripts/dogfood-drain.sh sets XDG_CONFIG_HOME and XDG_STATE_HOME alongside HOME on every forgectl invocation. os.UserConfigDir reads XDG_CONFIG_HOME first on Linux, so a HOME-only scratch isolated nothing there. Dry-run stays the default. 5. Nits: CheckDispatchCapability runs once at the top of a pass, so an undispatchable host refuses the pass instead of parking N records; drainItem resolves provenance through provenanceFromRecord, so the forged-authorship warning fires on the unattended path; a requeue clears WindowID; the refusal and watch-refusal strings in internal/cli/pr_drain.go go through safeTerm like the pass line already does. Session-Id: c2d13fd6-30bc-409a-989c-cd5ad22073fd Model: claude-opus-5 Harness: claude-code 2.1.269 Machine: cf6e768835c7 Co-Authored-By: Claude Opus 5 --- .../plans/2026-09-11-review-autonomy-spine.md | 5 + internal/cli/pr_drain.go | 30 +- internal/pr/drain.go | 193 +++++++++-- internal/pr/drain_test.go | 305 +++++++++++++++++- scripts/dogfood-drain.sh | 26 +- 5 files changed, 515 insertions(+), 44 deletions(-) diff --git a/docs/plans/2026-09-11-review-autonomy-spine.md b/docs/plans/2026-09-11-review-autonomy-spine.md index d639ff99..60e57831 100644 --- a/docs/plans/2026-09-11-review-autonomy-spine.md +++ b/docs/plans/2026-09-11-review-autonomy-spine.md @@ -192,6 +192,9 @@ Panel: plan-reviewer (conflict lens), plan-reviewer (buildability lens), red-tea - **2026-09-12 — Task 4: a retry-to-`queued` tears down any workspace the failed attempt created; a park to `needs-repair` leaves it.** Neither is named in the Steps list. Without this, a record returned to `queued` for a later pass would clone a SECOND workspace on retry while the first sat orphaned forever (nothing tears down a `queued` record's workspace — `queued` is defined as workspace-less). `needs-repair` keeps whatever workspace exists, matching every other `needs-repair` producer in this package (`pr repair` is the one verb designed to inspect a stuck clean room). The teardown is best-effort and its own failure is logged rather than shadowing the launch error already being reported, mirroring `parkReservation`'s established pattern. Chosen improvement, closing a leak the plan's Steps list did not anticipate. - **2026-09-12 — Task 4: no dedicated test drives the `--watch` loop's timing (looping until ctx cancel, the three-consecutive-refusals backoff-then-exit).** The Steps list names both as tests to write; both are implemented in `runDrainWatch` (`internal/cli/pr_drain.go`), but a test asserting the loop actually iterates on a real or fake clock, cancels cleanly, and counts to three refusals was not written in this pass — it needs either a `time.After`-injectable clock seam (not present in any sibling command) or a real multi-second sleep, and every other timed loop in this codebase (`dispatchWait`) solves this with an injectable function this command does not yet have. Recorded here as an out-of-scope gap rather than silently dropped; the single-pass behavior both flags share with `--watch` (grammar refusal, exit code, JSON/human shape) is fully tested. +- **2026-09-12 — Task 4 review round: a window or a park STOPS the retry; only a pre-dispatch failure may requeue.** The Steps list said a launch failure retries up to three times, with no exception named. `Launch` returns an error on branches where the tmux window already exists and the review agent is running, and each of those parks the record in `needs-repair` with a reason naming the window — the only pointer `pr repair --adopt-window` has left. The retry as built erased that reason, removed the clean room under the live agent, and returned the ref to `queued`, so a later pass would launch a SECOND agent for the same ref. `settleDrainFailure` now re-reads the record and asks tmux before deciding: an already-parked record, a window resolvable by the derived name, or an unreadable tmux (fail-closed) all mean "leave it parked, keep the workspace, never retry this ref"; only a failure with neither signal — a `gh` or clone failure before dispatch — requeues and tears down. Supersedes the 2026-09-11 "a retry-to-`queued` tears down any workspace" deviation, which now applies only to the pre-dispatch case. Reality-forced (security review Important 1). +- **2026-09-12 — Task 4 review round: the lifecycle lock is never held across a workspace removal.** The teardown ran inside the `transition` mutator, so a recursive delete of a full clone stalled every other lifecycle-lock user (another drainer, `pr `, `pr pick`, `pr repair`) for its duration — contradicting `docs/commands/pr.md`'s own "the lock is released before anything slow happens" — and sat inside a callback `transitionLocked` is willing to run twice on a revision mismatch. The mutator is now side-effect-free and idempotent: it records the workspace it cleared, the transition returns and releases the lock, and `sandboxTeardown` runs after that through the same prefix and symlink checks. Pinned by a lock-call-log test that fails if a removal falls inside a hold. Reality-forced (security review Important 3). + ## Learnings - **A legacy record's phase renders as `-`, not `active`.** The design says legacy is *treated* as active for slot counting; the presentation layer shows the record said nothing. The two live in different layers on purpose (Task 2's admission reads the design rule; `pr list` reads the record). @@ -213,3 +216,5 @@ Panel: plan-reviewer (conflict lens), plan-reviewer (buildability lens), red-tea - **"Reserve first, then decide" inverts the ordering a refusal needs.** Reserving before every refusal looks like the safe direction — the slot is claimed under the lock, so no peer can race it — but it converts a decision that costs nothing into state on disk, and the cost of that state is the whole cap. Four transient `gh` failures would have exhausted a default cap of four, with every later launch path then refusing. The rule the three call sites now share: a refusal that needs **no I/O** to decide runs before the reservation; a failure that can only happen **after** it parks the reservation with the reason. Nothing between those two cases is left to a caller's judgment. - **Resolving a value twice is a correctness question, not a cost question, whenever the first read keys durable state.** `ResolveLocalHead`'s duplicate `git rev-parse HEAD` was logged as one cheap extra call. It was actually a window in which the record's ref, the window name, and the reviewed tree could disagree, because the record's ref is never rewritten when the reservation completes — and teardown and repair both resolve the window from that ref. The tell to look for next time: the second read feeds something durable that the first read already named. +- **A retry is a claim that nothing was started, and only the caller of the failing function knows whether that holds.** `Launch` returning an error says nothing about whether a window exists — two of its branches return an error precisely because one does. The drainer cannot read the error to tell them apart, so the retry rule is built on state it can re-read instead: the record's own phase, and tmux's answer about the derived window name. Both signals fail closed, because "tmux is unreadable" and "the agent is live" have to land on the same side. +- **A mutator that a transition may re-run is not a place for an irreversible side effect.** `transitionLocked` re-invokes its callback once on a revision mismatch, and it holds the lifecycle lock the whole time — so anything slow in there stalls every peer, and anything destructive in there can run twice or run after the write it was paired with has failed. The rule the drainer now follows: the mutator computes, the caller acts. diff --git a/internal/cli/pr_drain.go b/internal/cli/pr_drain.go index 073fb2a4..1e5a34e7 100644 --- a/internal/cli/pr_drain.go +++ b/internal/cli/pr_drain.go @@ -118,7 +118,7 @@ func runDrainWatch(cmd *cobra.Command, client *pr.Client, cfg config.Config, opt consecutiveRefusals++ if consecutiveRefusals >= drainWatchRefusalLimit { return WithExitCode(fmt.Errorf( - "drain refused %d consecutive passes, last: %s", consecutiveRefusals, report.Refusal), 1) + "drain refused %d consecutive passes, last: %s", consecutiveRefusals, safeTerm(report.Refusal)), 1) } } else { consecutiveRefusals = 0 @@ -164,10 +164,20 @@ func writeDrainHuman(out io.Writer, report pr.DrainReport, dryRun bool, next tim } refs := make([]string, 0, len(report.Items)) for _, it := range report.Items { + // A refused record is not a would-launch row: naming it as one + // would promise a launch the next real pass will not perform. + if it.Outcome == "refused" { + continue + } refs = append(refs, it.Ref) } - _, _ = fmt.Fprintf(out, "%d queued, %d free — would launch %s\n", - report.Queued, report.Free, strings.Join(refs, ", ")) + if len(refs) == 0 { + _, _ = fmt.Fprintf(out, "%d queued, %d free — would launch nothing\n", report.Queued, report.Free) + } else { + _, _ = fmt.Fprintf(out, "%d queued, %d free — would launch %s\n", + report.Queued, report.Free, strings.Join(refs, ", ")) + } + writeDrainRefusedItems(out, report) return } if report.Queued == 0 && len(report.Items) == 0 { @@ -188,12 +198,24 @@ func writeDrainHuman(out io.Writer, report pr.DrainReport, dryRun bool, next tim } } +// writeDrainRefusedItems prints the records a pass refused to claim at all — +// a queued local review is the only shape today. They are printed on the +// dry-run arm too, where the would-launch list deliberately omits them. +func writeDrainRefusedItems(out io.Writer, report pr.DrainReport) { + for _, it := range report.Items { + if it.Outcome != "refused" { + continue + } + _, _ = fmt.Fprintf(out, " %s: refused: %s\n", it.Ref, safeTerm(it.Error)) + } +} + // drainExitCode is the honest code for one pass: 1 when the pass refused // outright or any launch failed, 0 otherwise — the same "a script can ask // this" contract `pr repair`'s inspect exit code follows. func drainExitCode(report pr.DrainReport) error { if report.Refusal != "" { - return WithExitCode(fmt.Errorf("drain pass refused: %s", report.Refusal), 1) + return WithExitCode(fmt.Errorf("drain pass refused: %s", safeTerm(report.Refusal)), 1) } if report.Failed > 0 { return WithExitCode(fmt.Errorf("%d review(s) failed to launch this pass", report.Failed), 1) diff --git a/internal/pr/drain.go b/internal/pr/drain.go index 9002492b..353bda58 100644 --- a/internal/pr/drain.go +++ b/internal/pr/drain.go @@ -25,8 +25,16 @@ const ( drainOutcomeRetryQueued = "retry-queued" drainOutcomeNeedsRepair = "needs-repair" drainOutcomeClaimFailure = "claim-failed" + drainOutcomeRefused = "refused" ) +// errLocalNotDrainable is the reason a queued LOCAL record is refused at claim +// time. It mirrors Queue's writer-side refusal (session.go) in the reader, so +// a record that never went through Queue is stopped before it is claimed, +// cloned, or handed to an agent. +const errLocalNotDrainable = "local reviews cannot be drained: a local session's findings directory is never " + + "persisted and a reloaded local session refuses to launch — review it now with 'forgectl pr local', not later" + // DrainOpts drives one `pr drain` pass. // // Once, Watch, and Interval are carried here to complete the shape the design @@ -101,10 +109,24 @@ type DrainReport struct { // opts.MaxAttempts, in which case it is parked in `needs-repair` with a // reason naming the attempt count and the last error, and no further pass // will pick it up (queued records are the only ones drain claims). +// +// A retry is offered ONLY when the failure landed before dispatch. A record +// Launch already parked, or a ref whose review window is resolvable (or whose +// tmux is unreadable), is left parked with its workspace intact and is never +// retried — see settleDrainFailure. func (c *Client) Drain(ctx context.Context, cfg config.Config, opts DrainOpts) (DrainReport, error) { if opts.MaxAttempts <= 0 { opts.MaxAttempts = DefaultDrainMaxAttempts } + // Dispatch capability is checked ONCE, before anything is claimed, exactly + // as the human path checks it before reserving (internal/cli/pr.go). A tmux + // that cannot dispatch fails every launch in the pass, so asking it here + // costs one probe and refuses the whole pass; asking it per record instead + // means N clones and N teardowns before N records park in needs-repair. + if err := c.CheckDispatchCapability(ctx); err != nil { + slog.Error("Refusing a drain pass: this host cannot dispatch a review window.", "error", err) + return DrainReport{Items: []DrainItem{}, Refusal: termsafe.SafeLine(err.Error())}, nil + } report, claimed := c.claimQueuedPass(ctx, cfg, opts) if report.Refusal != "" { return report, nil @@ -166,15 +188,40 @@ func (c *Client) claimQueuedPass(ctx context.Context, cfg config.Config, opts Dr report.Free = free var queued []SessionSummary + refused := 0 for _, s := range summaries { switch s.Phase() { case PhaseQueued: + // READER-SIDE MIRROR of Queue's local-ref refusal (session.go). + // Queue refuses to WRITE a local queued record; nothing refused + // to READ one, so a hand-written or forged record with + // `local: true` was claimed, cloned, and carried all the way to + // Launch's own local refusal — which that function's comment + // calls an incidental second barrier. The refusal belongs at the + // claim, where the record is still untouched. + if s.Ref().IsLocal() { + slog.Error("Refusing to claim a queued review: local reviews cannot be drained.", + "ref", s.Ref().String(), "path", s.Path()) + report.Items = append(report.Items, DrainItem{ + Ref: s.Ref().String(), + RecordPath: s.Path(), + FromPhase: string(PhaseQueued), + ToPhase: string(PhaseQueued), + Outcome: drainOutcomeRefused, + Error: errLocalNotDrainable, + }) + refused++ + continue + } queued = append(queued, s) case PhasePreparing, PhasePrepared, PhaseLaunching: report.Launching++ } } - report.Queued = len(queued) + // Refused records are still queued on disk, so they count toward the + // queue depth the report names — they are simply never claimed. + report.Queued = len(queued) + refused + report.Failed += refused sort.Slice(queued, func(i, j int) bool { return queued[i].CreatedAt().Before(queued[j].CreatedAt()) }) @@ -219,9 +266,15 @@ func (c *Client) drainItem(ctx context.Context, cfg config.Config, s SessionSumm return item } + // provenanceFromRecord, not ParseReviewProvenance: it applies the joint + // shape check (breadcrumb.go), so a record claiming authorship without the + // canonical local shape warns here too. The outcome is identical either way + // — EffectiveProvenance downgrades a remote ref regardless — but the + // unattended path is the one with nobody watching, so it is the last place + // that warning should be missing. prepOpts := PrepareOpts{ Agent: bc.Agent, - Provenance: ParseReviewProvenance(bc.Provenance), + Provenance: provenanceFromRecord(bc), RecordPath: path, } sess, err := c.Prepare(ctx, ref, prepOpts) @@ -235,24 +288,72 @@ func (c *Client) drainItem(ctx context.Context, cfg config.Config, s SessionSumm } item.Error = termsafe.SafeLine(err.Error()) - outcome, toPhase := c.settleDrainFailure(ctx, path, bc.Attempts, maxAttempts, err) + outcome, toPhase := c.settleDrainFailure(ctx, ref, path, bc.Attempts, maxAttempts, err) item.Outcome = outcome item.ToPhase = toPhase return item } -// settleDrainFailure records the failed attempt on the record — regardless of -// what phase Prepare/Launch left it in (queued's own `preparing`, or a -// needs-repair Launch itself already wrote on a dispatch failure) — and moves -// it to `queued` for a future pass to retry, or to `needs-repair` once -// attempts is exhausted. It uses the wildcard `from` (anyPhase) for the same -// reason markNeedsRepair does: the caller does not know, and must not have to -// know, which phase the failure left the record in. -func (c *Client) settleDrainFailure(ctx context.Context, path string, priorAttempts, maxAttempts int, cause error) (outcome, toPhase string) { +// settleDrainFailure records the failed attempt on the record and decides +// whether the ref may be retried at all. +// +// A RETRY IS ONLY SAFE WHEN THE FAILURE LANDED BEFORE DISPATCH. Launch returns +// an error on two branches where the tmux window ALREADY EXISTS and the review +// agent is running — a windowId that is not generation-qualified, and a failed +// `launching -> active` transition — and both park the record in needs-repair +// with a reason naming the window, which is the only pointer +// `pr repair --adopt-window` has left (launch.go's completeLaunch). Requeuing +// one of those would delete the clean room under a live agent, erase that +// pointer, and let a later pass launch a SECOND agent for the same ref. So the +// settlement reads the two signals that distinguish the cases and refuses to +// retry on either: the record arriving already parked, and a window resolvable +// by the ref's derived name. An unreadable tmux counts as "a window may exist" +// — the fail-closed direction, matching WindowLive's own contract that +// unreadable is not "gone". +// +// Only a failure with NEITHER signal — a clone or `gh` failure before dispatch +// — returns the record to `queued` for another pass, or parks it once attempts +// are exhausted. +func (c *Client) settleDrainFailure(ctx context.Context, ref Ref, path string, priorAttempts, maxAttempts int, cause error) (outcome, toPhase string) { attempts := priorAttempts + 1 lastError := termsafe.SafeLine(cause.Error()) - exhausted := attempts >= maxAttempts + bc, _, rerr := loadBreadcrumbRecord(path, c.sessionsDir) + if rerr != nil { + slog.Error("Failed to re-read a session record after a drain launch failure; it was left as the failure found it.", + "ref", ref.String(), "path", path, "error", rerr) + return drainOutcomeClaimFailure, "" + } + if bc.Phase == PhaseNeedsRepair { + // Launch already parked it with a reason naming the window. Record the + // attempt WITHOUT touching RepairReason, Workspace, or WindowID: this + // record is now `pr repair`'s to settle, not the drainer's to retry. + return c.recordParkedAttempt(ctx, ref, path, attempts, lastError, + "a review window may already exist for this ref") + } + if live, ok := c.WindowLive(ctx, ref); !ok || live { + reason := fmt.Sprintf("drain: launch failed with a review window present (or tmux unreadable) for %s; "+ + "settle it with 'forgectl pr repair --adopt-window' — last: %s", ref.String(), lastError) + slog.Error("Refusing to retry a drained review: a window for this ref may be live, so its clean room stays.", + "ref", ref.String(), "path", path, "windowReadable", ok, "error", cause) + if terr := c.transition(ctx, path, anyPhase, PhaseNeedsRepair, func(rec *Breadcrumb) error { + rec.Attempts = attempts + rec.LastError = lastError + rec.LastAttempt = time.Now().UTC() + rec.RepairReason = termsafe.SafeLine(reason) + return nil + }); terr != nil { + slog.Error("Failed to park a drained review whose window may be live.", + "ref", ref.String(), "path", path, "error", terr) + return drainOutcomeClaimFailure, "" + } + return drainOutcomeNeedsRepair, string(PhaseNeedsRepair) + } + + // No park, no window: the failure happened before anything was dispatched, + // so the workspace this attempt may have cloned is nobody's and the ref is + // safe to retry. + exhausted := attempts >= maxAttempts target := PhaseQueued outcome = drainOutcomeRetryQueued if exhausted { @@ -260,29 +361,31 @@ func (c *Client) settleDrainFailure(ctx context.Context, path string, priorAttem outcome = drainOutcomeNeedsRepair } - err := c.transition(ctx, path, anyPhase, target, func(bc *Breadcrumb) error { - bc.Attempts = attempts - bc.LastError = lastError - bc.LastAttempt = time.Now().UTC() + // The mutator is side-effect-free and idempotent BY CONTRACT: + // transitionLocked re-runs it once on a revision mismatch, and it holds the + // lifecycle lock while it does. The workspace removal therefore happens + // after this returns — a recursive delete of a full clone must never stall + // every other lifecycle-lock user (another drainer, `pr `, `pr pick`, + // `pr repair`) for its duration. + var cleared string + err := c.transition(ctx, path, anyPhase, target, func(rec *Breadcrumb) error { + rec.Attempts = attempts + rec.LastError = lastError + rec.LastAttempt = time.Now().UTC() if exhausted { - bc.RepairReason = fmt.Sprintf("drain: %d attempts, last: %s", attempts, lastError) + rec.RepairReason = fmt.Sprintf("drain: %d attempts, last: %s", attempts, lastError) // A retry that got as far as a workspace leaves it behind for // `pr repair` to inspect; needs-repair does not require an empty // workspace. - } else { - bc.RepairReason = "" - // queued must not carry a workspace: a retried Prepare clones a - // fresh one, so any workspace this failed attempt created would - // otherwise leak. Best-effort teardown; its own failure must not - // shadow the launch error already being reported. - if bc.Workspace != "" { - if terr := sandboxTeardown(ctx, c.run, bc.Workspace); terr != nil { - slog.Error("Failed to tear down the workspace from a failed drain attempt; it may need manual removal.", - "path", path, "workspace", bc.Workspace, "error", terr) - } - bc.Workspace = "" - } + return nil } + rec.RepairReason = "" + // queued must not carry a workspace or a window: a retried Prepare + // clones a fresh one, and a stale windowId on a queued record names a + // window this ref no longer has. + cleared = rec.Workspace + rec.Workspace = "" + rec.WindowID = "" return nil }) if err != nil { @@ -290,5 +393,35 @@ func (c *Client) settleDrainFailure(ctx context.Context, path string, priorAttem "path", path, "target", string(target), "error", err) return drainOutcomeClaimFailure, "" } + // Best-effort, outside the lock; its own failure must not shadow the launch + // error already being reported. sandboxTeardown carries the prefix and + // symlink checks that bound every removal in this package. + if cleared != "" { + if terr := sandboxTeardown(ctx, c.run, cleared); terr != nil { + slog.Error("Failed to tear down the workspace from a failed drain attempt; it may need manual removal.", + "path", path, "workspace", cleared, "error", terr) + } + } return outcome, string(target) } + +// recordParkedAttempt records one more failed attempt on a record Launch +// ALREADY parked in needs-repair, preserving the repair reason, the workspace, +// and the window id it wrote. The phase does not move (needs-repair to +// needs-repair) — the write exists so the attempt count and last error stay +// truthful for `pr repair`. +func (c *Client) recordParkedAttempt(ctx context.Context, ref Ref, path string, attempts int, lastError, why string) (outcome, toPhase string) { + slog.Error("Refusing to retry a drained review: its record is already parked in needs-repair.", + "ref", ref.String(), "path", path, "why", why) + if err := c.transition(ctx, path, PhaseNeedsRepair, PhaseNeedsRepair, func(rec *Breadcrumb) error { + rec.Attempts = attempts + rec.LastError = lastError + rec.LastAttempt = time.Now().UTC() + return nil + }); err != nil { + slog.Error("Failed to record a drain attempt on an already-parked record.", + "ref", ref.String(), "path", path, "error", err) + return drainOutcomeClaimFailure, "" + } + return drainOutcomeNeedsRepair, string(PhaseNeedsRepair) +} diff --git a/internal/pr/drain_test.go b/internal/pr/drain_test.go index 8906470d..c201393a 100644 --- a/internal/pr/drain_test.go +++ b/internal/pr/drain_test.go @@ -9,11 +9,20 @@ package pr // as it does for reserve/Admit, so drain claims fewer queued records // when one is already in flight // [x] An unreadable record refuses the WHOLE pass before claiming anything -// [x] A launch failure returns the record to `queued` with attempts=1 and -// lastError set -// [x] A third failure (attempts already 2) moves the record to +// [x] A PRE-DISPATCH failure (gh/clone) returns the record to `queued` with +// attempts=1 and lastError set +// [x] A third pre-dispatch failure (attempts already 2) moves the record to // `needs-repair` with a reason naming the attempt count, and it is no // longer counted as `queued` on the next listing +// [x] A failure AFTER the window exists leaves the record parked in +// needs-repair with its window-naming reason and its workspace, and the +// next pass launches nothing for that ref +// [x] A failure with a live window on an unparked record parks it and tears +// down nothing +// [x] A pre-dispatch failure's workspace teardown happens OUTSIDE the +// lifecycle lock +// [x] A queued record marked local is refused at claim time: never claimed, +// never cloned, reported as refused // [x] Two Clients racing one queued record launch it exactly once // [x] `--dry-run` claims nothing, launches nothing, and reports would-launch // [x] A mixed pass (one success, one failure) reports both items @@ -76,6 +85,19 @@ func drainLaunchRunner(failOn map[int]error) *exec.FakeRunner { }} } +// drainGhFailRunner fails `gh pr view` for every ref — a PRE-DISPATCH +// failure, the one shape a retry is safe for: no clone happened, no window +// exists, and Launch was never reached, so nothing parked the record. +func drainGhFailRunner() *exec.FakeRunner { + inner := drainLaunchRunner(nil).RunFunc + return &exec.FakeRunner{RunFunc: func(name string, args []string) (string, error) { + if name == "gh" && len(args) >= 2 && args[0] == "pr" && args[1] == "view" { + return "", errors.New("boom: gh could not read the PR") + } + return inner(name, args) + }} +} + // drainClient builds a Client wired for drain tests: real sessions dir, // fakeClaude on PATH via env, no interactive TTY concerns (Drain never shows // one). @@ -267,7 +289,9 @@ func TestDrain_UnreadableRecordRefusesWholePass(t *testing.T) { func TestDrain_LaunchFailureReturnsToQueuedWithAttempts(t *testing.T) { dir := t.TempDir() - run := drainLaunchRunner(map[int]error{1: errors.New("boom: agent refused")}) + // A gh failure: nothing was cloned and no window exists, so this is the + // one failure shape the drainer may retry. + run := drainGhFailRunner() c := drainClient(t, dir, run) seedQueued(t, c, testRef(1), time.Now().UTC(), 0) @@ -298,7 +322,7 @@ func TestDrain_LaunchFailureReturnsToQueuedWithAttempts(t *testing.T) { func TestDrain_ThirdFailureMovesToNeedsRepair(t *testing.T) { dir := t.TempDir() - run := drainLaunchRunner(map[int]error{1: errors.New("boom: agent refused again")}) + run := drainGhFailRunner() c := drainClient(t, dir, run) // Two prior failed attempts already recorded. seedQueued(t, c, testRef(1), time.Now().UTC(), 2) @@ -431,6 +455,277 @@ func TestDrain_MixedPassOneSuccessOneFailure(t *testing.T) { } } +// drainWindowExistsRunner creates the review window for ref but hands back a +// windowId that is NOT generation-qualified, which is one of the two Launch +// branches where the agent is running and the record is parked in needs-repair +// with a reason naming the window. list-windows then reports that window, so +// both retry-stopping signals are present. +func drainWindowExistsRunner(t *testing.T, ref Ref) (*exec.FakeRunner, *int) { + t.Helper() + name, err := ReviewWindowName(ref) + if err != nil { + t.Fatalf("review window name: %v", err) + } + newWindows := 0 + created := false + opened := false + row := strings.Join([]string{"123", "456", "@9", "$1", "forgectl", "1", name, "1", "1"}, "\x1f") + run := &exec.FakeRunner{RunFunc: func(cmd string, args []string) (string, error) { + switch { + case cmd == "gh" && len(args) >= 2 && args[0] == "pr" && args[1] == "view": + return `{"headRefName":"feature","headRefOid":"abc123",` + + `"headRepositoryOwner":{"login":"cameronsjo"},"headRepository":{"name":"forgectl"}}`, nil + case cmd == "git" && len(args) > 0 && args[0] == "clone": + return "", nil + case cmd == "tmux" && len(args) > 0: + switch args[0] { + case "-V": + return "tmux 3.7b", nil + case "display-message": + return "123\x1f456\x1f@0", nil + case "list-sessions": + if created { + return "123\x1f456\x1f$1\x1fforgectl\x1f1\x1f0\x1f0\x1f/tmp", nil + } + return "", nil + case "new-session": + created = true + return "123\x1f456\x1f$1", nil + case "list-windows": + if opened { + return row, nil + } + return "", nil + case "new-window": + newWindows++ + opened = true + // A real window whose identity is not generation-qualified: + // the server start time is not numeric, so validWindowID + // refuses it after the window already exists. + return "123\x1fnot-a-timestamp\x1f@9", nil + } + } + return "", nil + }} + return run, &newWindows +} + +func TestDrain_FailureAfterWindowExistsStaysParkedAndIsNotRetried(t *testing.T) { + dir := t.TempDir() + ref := testRef(1) + run, newWindows := drainWindowExistsRunner(t, ref) + c := drainClient(t, dir, run) + seedQueued(t, c, ref, time.Now().UTC(), 0) + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if len(report.Items) != 1 || report.Items[0].Outcome != drainOutcomeNeedsRepair { + t.Fatalf("items = %+v, want one needs-repair item (a live window must not be retried)", report.Items) + } + if report.Failed != 1 { + t.Errorf("Failed = %d, want 1", report.Failed) + } + + bc := readRecordByRef(t, dir, ref) + if bc.Phase != PhaseNeedsRepair { + t.Fatalf("phase = %q, want needs-repair", bc.Phase) + } + // Launch's own reason — naming the window — is the only pointer + // `pr repair --adopt-window` has left, and the drainer must not erase it + // or overwrite it with a "drain: N attempts" reason. + windowName, nerr := ReviewWindowName(ref) + if nerr != nil { + t.Fatalf("review window name: %v", nerr) + } + if !strings.Contains(bc.RepairReason, windowName) { + t.Errorf("repairReason = %q, want Launch's window-naming reason preserved", bc.RepairReason) + } + if strings.HasPrefix(bc.RepairReason, "drain:") { + t.Errorf("repairReason = %q, want the drainer to leave Launch's reason alone", bc.RepairReason) + } + if bc.Workspace == "" { + t.Fatal("workspace was cleared from the record; the agent's clean room must stay") + } + if _, serr := os.Stat(bc.Workspace); serr != nil { + t.Errorf("workspace %s was removed under a live agent: %v", bc.Workspace, serr) + } + if bc.Attempts != 1 { + t.Errorf("attempts = %d, want 1 recorded on the parked record", bc.Attempts) + } + + // The next pass claims nothing and opens no second window for the ref. + report2, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("second Drain: %v", err) + } + if report2.Queued != 0 || len(report2.Items) != 0 { + t.Fatalf("second pass = %+v, want nothing queued and nothing claimed", report2) + } + if *newWindows != 1 { + t.Errorf("tmux new-window called %d times, want exactly 1 — a parked ref must never launch again", *newWindows) + } +} + +func TestSettleDrainFailure_LiveWindowParksAndTearsDownNothing(t *testing.T) { + dir := t.TempDir() + ref := testRef(4) + run, _ := drainWindowExistsRunner(t, ref) + // The window is reported from the start: this is the record Launch failed + // to park (its own park is best-effort), still sitting in `launching`. + name, err := ReviewWindowName(ref) + if err != nil { + t.Fatalf("review window name: %v", err) + } + row := strings.Join([]string{"123", "456", "@9", "$1", "forgectl", "1", name, "1", "1"}, "\x1f") + inner := run.RunFunc + run.RunFunc = func(cmd string, args []string) (string, error) { + if cmd == "tmux" && len(args) > 0 && args[0] == "list-windows" { + return row, nil + } + return inner(cmd, args) + } + c := drainClient(t, dir, run) + ws := fakeWorkspace(t) + path := seedPhaseRecord(t, c, ref, PhaseLaunching, ws) + + teardowns := 0 + orig := sandboxTeardown + sandboxTeardown = func(context.Context, exec.Runner, string) error { teardowns++; return nil } + t.Cleanup(func() { sandboxTeardown = orig }) + + outcome, toPhase := c.settleDrainFailure(context.Background(), ref, path, 0, 3, errors.New("boom: dispatch reported an error")) + if outcome != drainOutcomeNeedsRepair || toPhase != string(PhaseNeedsRepair) { + t.Fatalf("outcome = %q/%q, want needs-repair", outcome, toPhase) + } + if teardowns != 0 { + t.Errorf("sandboxTeardown called %d times, want 0 — the window may be live", teardowns) + } + bc := readRecord(t, path) + if bc.Phase != PhaseNeedsRepair { + t.Fatalf("phase = %q, want needs-repair", bc.Phase) + } + if bc.Workspace != ws { + t.Errorf("workspace = %q, want it untouched (%q)", bc.Workspace, ws) + } + if !strings.Contains(bc.RepairReason, "adopt-window") { + t.Errorf("repairReason = %q, want it to name the way out", bc.RepairReason) + } +} + +func TestSettleDrainFailure_PreDispatchRequeuesAndRemovesWorkspaceOutsideTheLock(t *testing.T) { + dir := t.TempDir() + ref := testRef(5) + run := drainLaunchRunner(nil) // list-windows reports no window + c := drainClient(t, dir, run) + ws := fakeWorkspace(t) + path := seedPhaseRecord(t, c, ref, PhasePreparing, "") + // Give the record a workspace the way a completed Prepare would. + if terr := c.transition(context.Background(), path, PhasePreparing, PhasePrepared, func(rec *Breadcrumb) error { + rec.Workspace = ws + return nil + }); terr != nil { + t.Fatalf("seed a prepared record with a workspace: %v", terr) + } + + var mu sync.Mutex + var log []string + note := func(s string) { mu.Lock(); log = append(log, s); mu.Unlock() } + c.onLock = func(verb, event string) { note(event + ":" + verb) } + var torn []string + orig := sandboxTeardown + sandboxTeardown = func(_ context.Context, _ exec.Runner, workspace string) error { + note("teardown") + torn = append(torn, workspace) + return nil + } + t.Cleanup(func() { sandboxTeardown = orig }) + + outcome, toPhase := c.settleDrainFailure(context.Background(), ref, path, 0, 3, errors.New("boom: clone failed")) + if outcome != drainOutcomeRetryQueued || toPhase != string(PhaseQueued) { + t.Fatalf("outcome = %q/%q, want retry-queued/queued", outcome, toPhase) + } + bc := readRecord(t, path) + if bc.Phase != PhaseQueued || bc.Attempts != 1 { + t.Fatalf("record = phase %q attempts %d, want queued/1", bc.Phase, bc.Attempts) + } + if bc.Workspace != "" || bc.WindowID != "" { + t.Errorf("requeued record still names workspace %q / window %q", bc.Workspace, bc.WindowID) + } + if len(torn) != 1 || torn[0] != ws { + t.Fatalf("tore down %v, want exactly [%s]", torn, ws) + } + + // THE REMOVAL MUST NOT HAPPEN UNDER THE LOCK: a recursive delete of a full + // clone would stall every other lifecycle-lock user for its duration. + mu.Lock() + defer mu.Unlock() + held := 0 + for _, e := range log { + switch { + case strings.HasPrefix(e, "acquire:"): + held++ + case strings.HasPrefix(e, "release:"): + held-- + case e == "teardown" && held > 0: + t.Fatalf("workspace teardown ran inside a lifecycle-lock hold; log = %v", log) + } + } +} + +func TestDrain_QueuedLocalRecordIsRefusedAtClaim(t *testing.T) { + dir := t.TempDir() + run := drainLaunchRunner(nil) + launches := 0 + inner := run.RunFunc + run.RunFunc = func(name string, args []string) (string, error) { + if name == "git" && len(args) > 0 && args[0] == "clone" { + launches++ + } + if name == "tmux" && len(args) > 0 && args[0] == "new-window" { + launches++ + } + return inner(name, args) + } + c := drainClient(t, dir, run) + + ref := newLocalRef("abc1234def") + bc := Breadcrumb{ + Ref: ref.String(), Agent: "claude", CreatedAt: time.Now().UTC(), Local: true, + Provenance: ReviewProvenanceOperatorAuthored.persisted(), + Version: breadcrumbVersion, Phase: PhaseQueued, Revision: 1, + } + path, err := writeBreadcrumb(c.SessionsDir(), ref, bc) + if err != nil { + t.Fatalf("seed queued local record: %v", err) + } + + report, err := c.Drain(context.Background(), config.Config{}, DrainOpts{}) + if err != nil { + t.Fatalf("Drain: %v", err) + } + if len(report.Items) != 1 || report.Items[0].Outcome != drainOutcomeRefused { + t.Fatalf("items = %+v, want one refused item", report.Items) + } + if !strings.Contains(report.Items[0].Error, "local reviews cannot be drained") { + t.Errorf("item error = %q, want the local refusal reason", report.Items[0].Error) + } + if report.Failed != 1 { + t.Errorf("Failed = %d, want 1", report.Failed) + } + if launches != 0 { + t.Errorf("clone/new-window ran %d times, want 0 — a local record is refused before it is claimed", launches) + } + after := readRecord(t, path) + if after.Phase != PhaseQueued { + t.Errorf("phase = %q, want still queued (never claimed to preparing)", after.Phase) + } + if after.Revision != 1 { + t.Errorf("revision = %d, want 1 — the record must not be written at all", after.Revision) + } +} + func TestDrain_EmptyQueueReportsZero(t *testing.T) { dir := t.TempDir() c := drainClient(t, dir, drainLaunchRunner(nil)) diff --git a/scripts/dogfood-drain.sh b/scripts/dogfood-drain.sh index 38619620..40fe373b 100755 --- a/scripts/dogfood-drain.sh +++ b/scripts/dogfood-drain.sh @@ -1,7 +1,8 @@ #!/usr/bin/env bash # Dogfoods `forgectl pr --queue` + `forgectl pr drain` against two real # PRs (forgectl#473): queues both, drains one pass, and asserts the pass -# report. Uses a scratch HOME so it never touches a real ~/.config/forgectl. +# report. Uses a scratch HOME *and* scratch XDG dirs so it never touches a real +# ~/.config/forgectl on either macOS or Linux. # # Usage: scripts/dogfood-drain.sh [--launch] [path/to/forgectl] # @@ -38,13 +39,28 @@ fi SCRATCH=$(mktemp -d) trap 'rm -rf "$SCRATCH"' EXIT +# HOME ALONE DOES NOT ISOLATE ON LINUX. forgectl resolves its state dir through +# os.UserConfigDir(), which reads $XDG_CONFIG_HOME first there and only falls +# back to $HOME/.config — so with XDG_CONFIG_HOME set (the common case) a +# HOME-only override writes real queued records into the operator's real +# session dir, where the next real `pr drain` would launch them. XDG_STATE_HOME +# is overridden too (internal/config/usage_base.go reads it); XDG_DATA_HOME and +# XDG_CACHE_HOME are not read by this binary. macOS ignores all of these +# (os.UserConfigDir is HOME-derived), which is exactly why a run here would not +# surface the gap. +SCRATCH_ENV=(env + "HOME=$SCRATCH" + "XDG_CONFIG_HOME=$SCRATCH/config" + "XDG_STATE_HOME=$SCRATCH/state" +) + echo "--- queue $REF1 ---" -HOME="$SCRATCH" "$BIN" pr "$REF1" --queue +"${SCRATCH_ENV[@]}" "$BIN" pr "$REF1" --queue echo "--- queue $REF2 ---" -HOME="$SCRATCH" "$BIN" pr "$REF2" --queue +"${SCRATCH_ENV[@]}" "$BIN" pr "$REF2" --queue echo "--- pr queue ---" -HOME="$SCRATCH" "$BIN" pr queue +"${SCRATCH_ENV[@]}" "$BIN" pr queue DRAIN_ARGS=(pr drain --once --json) if [ "$DRY_RUN" = true ]; then @@ -52,7 +68,7 @@ if [ "$DRY_RUN" = true ]; then fi echo "--- ${DRAIN_ARGS[*]} ---" -REPORT=$(HOME="$SCRATCH" "$BIN" "${DRAIN_ARGS[@]}") +REPORT=$("${SCRATCH_ENV[@]}" "$BIN" "${DRAIN_ARGS[@]}") RC=$? echo "$REPORT" From 5791ff72026a6ec1abab3f4c35ca2e5fb40ab8eb Mon Sep 17 00:00:00 2001 From: Cameron Sjo <4084915+cameronsjo@users.noreply.github.com> Date: Sat, 12 Sep 2026 10:51:45 -0500 Subject: [PATCH 4/4] docs(plans): tick the Task 4 ship boxes and name what the merge still owes (#473) Session-Id: c2d13fd6-30bc-409a-989c-cd5ad22073fd Model: claude-fable-5-1 Harness: claude-code 2.1.269 Machine: cf6e768835c7 Co-Authored-By: Claude Fable 5.1 --- docs/plans/2026-09-11-review-autonomy-spine.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/plans/2026-09-11-review-autonomy-spine.md b/docs/plans/2026-09-11-review-autonomy-spine.md index 60e57831..bfdeb912 100644 --- a/docs/plans/2026-09-11-review-autonomy-spine.md +++ b/docs/plans/2026-09-11-review-autonomy-spine.md @@ -1,6 +1,6 @@ --- status: in-flight -next: "Task 4 is committed and pushed on feat/pr-drain; the orchestrator reviews and opens its PR; the live dogfood run is owed" +next: "Task 4 PR is open on feat/pr-drain; merge lands the whole spine. Owed after merge: the live scripts/dogfood-drain.sh --launch pass, and status: done" branch: plan/review-autonomy-spine pr: cameronsjo/forgectl#495 updated: 2026-09-12 @@ -154,8 +154,8 @@ Panel: plan-reviewer (conflict lens), plan-reviewer (buildability lens), red-tea - [x] Run — expect GREEN; vet; lint - [x] Write `scripts/dogfood-drain.sh`: queue two named PRs with `--queue`, run `pr drain --once --json`, assert two `launched` items, print the report; run it (in `--dry-run` form — see Deviations) against two real PRs and record the measured output inline in the PR body - [x] `docs/commands/pr.md` triage section; `README.md`; `pr.go` lists -- [ ] Commit: `feat(pr): queue and drain verbs (#473)` with the producer tuple -- [ ] run `cadence-forge:polish`; fold findings; open PR with plain-text `Closes #473` +- [x] Commit: `feat(pr): queue and drain verbs (#473)` with the producer tuple — `8c04073`, script default `dd49816`, retry fixes `4e73dfa` +- [x] run `cadence-forge:polish`; fold findings; open PR with plain-text `Closes #473` — code 1 Important + security 4 Important at `8c04073`, all folded; PR opened; the live `--launch` dogfood pass is owed ---