From 134b97d6eee011a766f3e034eee95ae2918d655a Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 00:00:29 -0400 Subject: [PATCH 1/9] docs(issue-32): plan + decisions for human-in-loop & review resume --- planning/issues/32/decisions.md | 55 ++++++++++++++++++++++++++++ planning/issues/32/plan.md | 63 +++++++++++++++++++++++++++++++++ 2 files changed, 118 insertions(+) create mode 100644 planning/issues/32/decisions.md create mode 100644 planning/issues/32/plan.md diff --git a/planning/issues/32/decisions.md b/planning/issues/32/decisions.md new file mode 100644 index 00000000..64058693 --- /dev/null +++ b/planning/issues/32/decisions.md @@ -0,0 +1,55 @@ +# Decisions — Issue #32 (Human-in-the-loop + review-driven resume) + +## bunqueue cannot express the spec's nested waitFor graph; use a top-level waitFor spine +**File(s):** `packages/dispatcher/src/workflows/implementation.ts` +**Date:** 2026-05-23 + +**Decision:** Model park/resume as a **top-level `waitFor` node** reached after a branch, with +additional review rounds achieved by re-enqueue — not as `.path((w) => w.step().waitFor().step())` +as the build spec's idealized example shows. + +**Why:** The installed `bunqueue@2.7.12` `Workflow` builder filters branch `.path()` bodies and +loop (`doUntil`/`doWhile`/`forEach`) bodies to `type === 'step'` only (`workflow.js:46-48, 83-85`). +A `waitFor` nested inside a path or loop is **silently dropped** — it never executes. The executor +also has no goto/loop-back: `advance()` only moves forward or completes. So a `waitFor` only works +as a top-level node in the workflow's `nodes` array. The spec's annotation `// and loop back via +re-enqueue` confirms re-enqueue was always the intended looping mechanism. + +**Evidence:** `node_modules/.bun/bunqueue@2.7.12*/dist/client/workflow/workflow.js:39-51` (path +filters to steps), `executor.js:129-138` (runBranch runs path steps inline then advances), +`executor.js:149-181` (runWaitFor), `types.d.ts:51-54` (BranchDefinition.paths is `StepDefinition[]`). + +## Conditional parking via pre-seeding `ctx.signals` (by-reference) +**File(s):** `packages/dispatcher/src/workflows/implementation.ts` +**Date:** 2026-05-23 + +**Decision:** A single top-level `waitFor` follows the outcome branch. Park-worthy outcomes +(asked-question, done) leave the signal unset so the `waitFor` genuinely parks; terminal outcomes +(bare-stop, failed, rate-limited) **pre-seed `ctx.signals[RESUME_EVENT]`** in their branch step so +the same `waitFor` falls through immediately and the workflow finalizes without waiting. + +**Why:** A top-level `waitFor` always executes (no skip primitive). `buildContext` returns +`signals: exec.signals` by reference (`runner.js:169-181`), and `runWaitFor` advances when +`exec.signals[node.event] !== undefined` (`executor.js:150`). Mutating `ctx.signals` in a step +therefore satisfies the wait for terminal paths. Validated by a spike against the real embedded +engine before building the production workflow (build-to-learn). + +**Evidence:** spike test (see commit); `runner.js:178`, `executor.js:150`. + +## One generic engine event name; epic-specific naming lives in `waitfor_signals` +**File(s):** `packages/dispatcher/src/workflows/implementation.ts`, `workflow-record.ts` +**Date:** 2026-05-23 + +**Decision:** The bunqueue `waitFor` uses a single constant event string (`"resume"`). The +durable, poller-facing name (`epic--answered` / `epic--review-resolved`) is the +`waitfor_signals.signal_name`. The poller looks up the workflow by its armed row and calls +`engine.signal(workflowId, "resume", payload)` regardless of reason; the reason + data ride in the +payload and the DB row. + +**Why:** `waitFor(event)` takes a **static string** in this bunqueue version (not the spec's +`(ctx) => ...`), and `engine.signal` already targets a specific execution by id, so the event name +need not be parameterized to avoid cross-execution signal collisions. This keeps the workflow +definition static while preserving the epic-scoped, reason-scoped naming the poller and dashboard need. + +**Evidence:** `workflow.d.ts:24` (`waitFor(event: string, ...)`), `executor.js:83-97` (signal +targets one execution), spec §"implementation workflow". diff --git a/planning/issues/32/plan.md b/planning/issues/32/plan.md new file mode 100644 index 00000000..c0f6afc7 --- /dev/null +++ b/planning/issues/32/plan.md @@ -0,0 +1,63 @@ +# Issue #32: Human-in-the-loop and review-driven resume flow + +**Link:** https://github.com/thejustinwalsh/middle/issues/32 +**Branch:** middle-issue-32 + +## Goal +Give the `implementation` workflow a **park → external-signal → resume** spine so an agent +can hand control back to a human (asked a question) or to a reviewer (PR-ready), and later +resume a fresh session in the same worktree with the answer / review threads in context. +`APPROVED` ends the loop; a never-satisfied review loop is bounded to 5 rounds. + +## Approach +- The Epic's 4 open sub-issues are the phases. Build down them on one branch / one PR. +- **bunqueue reality check (load-bearing):** the installed `bunqueue@2.7.12` `Workflow` DSL + filters `.path()` / loop bodies to **steps only** — a `waitFor` nested in a branch path is + silently dropped (`workflow.js:46`). `waitFor` must be a **top-level node**. `engine.signal(execId, event, payload)` + targets a specific execution and sets `exec.signals[event]`; the matching top-level `waitFor` + then advances. `buildContext` passes `signals` **by reference** (`runner.js:178`), so a step can + pre-seed `ctx.signals[event]` to make a downstream top-level `waitFor` fall through without parking. + There is no goto/loop-back. The spec's idealized nested graph (§"implementation workflow") is + therefore expressed as: **a top-level `waitFor` spine + re-enqueue for additional rounds**, which + matches the spec's own `// loop back via re-enqueue` annotation. +- Reuse the existing `waitfor_signals` table + `armWaitForSignal`/`isWaitForArmed` (built in + Phase 2 for the watchdog sentinel re-arm). Add a `consumeWaitForSignal` (delete on resume) and a + per-workflow round counter (`meta_json` or a column). +- Poller talks to GitHub via the `gh` CLI subprocess pattern already used in `state-issue.ts`. +- Tests follow the existing `implementation-workflow.test.ts` / adapter test style: stub tmux + + SessionGate + adapter, drive the real embedded engine, assert DB state + signal flow. + +## Phases +1. **#33 waitFor signal spine** — branch on `classifyStop` outcome; asked-question + done paths arm + a `waitfor_signals` row, end the session (keep the worktree), set state `waiting-human`, park on a + top-level `waitFor`; resume re-enters carrying the resume reason; row consumed on resume. +2. **#34 classifyStop sentinel** — `.middle/blocked.json` → `{kind:'asked-question', sentinelPath}` + with the question/context surfaced to the workflow; no sentinel → `done`/`bare-stop`. +3. **#35 GitHub poller** — for Epics with an armed wait, fire `epic--answered` on a new human + reply, and `epic--review-resolved` on a review transition (CHANGES_REQUESTED/label → resume; + APPROVED **or** 0-actionable re-review → resolved). Idempotent + rate-limit resilient. +4. **#36 resume logic** — fresh session re-primed per reason; review-changes follows the skill's + "Addressing review feedback" per-round procedure (batch → internal review loop → single push → + reply in-thread → re-request → re-park); round counter per pass; cap (default 5) → `waiting-human`; + APPROVED ends the loop. + +## Files likely to change +- `packages/dispatcher/src/workflows/implementation.ts` — the park/resume spine (#33, #36) +- `packages/dispatcher/src/workflow-record.ts` — `consumeWaitForSignal`, round-counter helpers (#33, #36) +- `packages/dispatcher/src/db/migrations/00X_*.sql` — round counter / signal metadata if a column is needed +- `packages/adapters/claude/src/classify.ts` + `prompt.ts` — sentinel contents, resume prompt framing (#34, #36) +- `packages/core/src/adapter.ts` — `StopClassification` enrichment if contents are surfaced via the type (#34) +- `packages/dispatcher/src/poller.ts` (new) + wiring in `main.ts` — the GitHub poller (#35) +- `packages/dispatcher/test/*` + `packages/adapters/claude/test/*` — tests per phase + +## Out of scope +- Mechanical verification gates (Phase 6) — the poller fires on review state; it does not run gates. +- Auto-dispatch / slot enforcement (Phase 8) — parking frees the slot conceptually; the auto-dispatch + loop that consumes freed slots is Phase 8. +- Dashboard surfaces (Phase 9) — "asked question" / "waiting review" rendering. +- middle never merges — APPROVED is terminal; the human merges. + +## Open questions +- None blocking. The round-counter storage (new column vs `meta_json`) and the exact re-enqueue shape + for multi-round loops will be resolved during #36 by building (the spike in #33 validates the core + park/signal mechanic against the real engine first). From 9a6958e94e24e5a578dfe1dbc7a898cd9bc02064 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 00:08:40 -0400 Subject: [PATCH 2/9] feat(dispatcher): park/resume waitFor spine in implementation workflow Restructure createImplementationWorkflow into a park -> external-signal -> resume spine: launch-and-drive -> branch(park|terminal) -> top-level waitFor -> resume-or-finalize. asked-question and done stops arm a durable waitfor_signals row (epic--answered / epic--review-resolved), end the session, set waiting-human, and park; terminal stops pre-seed the signal so the single top-level waitFor falls through. Resume consumes the row and re-drives a fresh session re-primed per reason. bunqueue 2.7.12 filters branch/loop bodies to steps only and signal() targets one execution, so waitFor is a single top-level node (the spec's nested graph isn't expressible); additional review rounds loop via re-enqueue (#36). Adds getWaitForSignal/consumeWaitForSignal to workflow-record. Closes #33 --- packages/dispatcher/src/workflow-record.ts | 30 +++ .../src/workflows/implementation.ts | 250 ++++++++++++++---- .../test/implementation-workflow.test.ts | 209 ++++++++++----- 3 files changed, 380 insertions(+), 109 deletions(-) diff --git a/packages/dispatcher/src/workflow-record.ts b/packages/dispatcher/src/workflow-record.ts index ad4b5aae..67650735 100644 --- a/packages/dispatcher/src/workflow-record.ts +++ b/packages/dispatcher/src/workflow-record.ts @@ -188,6 +188,36 @@ export function armWaitForSignal( ); } +export type ArmedSignal = { signalName: string; payloadJson: string | null }; + +/** + * The signal armed for this workflow, or null. The poller reads this to learn + * what an Epic is waiting on (the epic-scoped, reason-scoped `signal_name`) + * without consuming it; only a successful resume consumes the row. + */ +export function getWaitForSignal(db: Database, workflowId: string): ArmedSignal | null { + const row = db + .query( + "SELECT signal_name, payload_json FROM waitfor_signals WHERE workflow_id = ? LIMIT 1", + ) + .get(workflowId) as { signal_name: string; payload_json: string | null } | null; + if (!row) return null; + return { signalName: row.signal_name, payloadJson: row.payload_json }; +} + +/** + * Consume (delete) the armed signal for a workflow on resume, returning what it + * was. The durable `waitfor_signals` row is middle's own record that the + * workflow is parked — distinct from bunqueue's in-memory `exec.signals`. It is + * armed when the workflow parks and consumed exactly once when it resumes, so a + * resumed workflow no longer reads as waiting and the poller stops watching it. + */ +export function consumeWaitForSignal(db: Database, workflowId: string): ArmedSignal | null { + const armed = getWaitForSignal(db, workflowId); + if (armed) db.run("DELETE FROM waitfor_signals WHERE workflow_id = ?", [workflowId]); + return armed; +} + type WorkflowRow = { id: string; kind: string; diff --git a/packages/dispatcher/src/workflows/implementation.ts b/packages/dispatcher/src/workflows/implementation.ts index 0162575d..cba9d5b1 100644 --- a/packages/dispatcher/src/workflows/implementation.ts +++ b/packages/dispatcher/src/workflows/implementation.ts @@ -8,6 +8,8 @@ import type { SessionGate } from "../hook-server.ts"; import { markAvailableOnSuccess, parseResetAt, setRateLimited } from "../rate-limits.ts"; import type { CreateWorktreeOpts, WorktreeHandle } from "../worktree.ts"; import { + armWaitForSignal, + consumeWaitForSignal, createWorkflowRecord, updateWorkflow, type WorkflowState, @@ -20,6 +22,32 @@ export type ImplementationInput = { adapter: string; }; +/** + * Which pause kind the workflow parked on. The two pause kinds share one + * park → external-signal → resume spine; the reason is what the resume step + * uses to pick its re-priming framing (`answer` vs `resume`/review-changes). + */ +export type ResumeReason = "answered-question" | "review-changes"; + +/** + * The single bunqueue signal event the workflow's top-level `waitFor` listens + * on. bunqueue's `waitFor(event)` takes a *static* string and `engine.signal` + * targets a specific execution by id, so one constant event name suffices — + * the epic-scoped, reason-scoped name lives in the durable `waitfor_signals` + * row (see `signalNameFor`), which is what the poller and dashboard read. + */ +export const RESUME_EVENT = "resume"; + +/** The durable, poller-facing signal name for a workflow's armed wait. */ +export function signalNameFor(epicNumber: number, reason: ResumeReason): string { + return reason === "review-changes" + ? `epic-${epicNumber}-review-resolved` + : `epic-${epicNumber}-answered`; +} + +/** The `waitFor` timeout — a parked workflow waits up to a week for its signal. */ +const WAITFOR_TIMEOUT_MS = 7 * 24 * 3600 * 1000; + /** The tmux surface the workflow drives — structural so tests can stub it. */ export type TmuxOps = { newSession(opts: { @@ -51,6 +79,17 @@ export type ImplementationDeps = { dispatcherUrl: string; launchTimeoutMs?: number; stopTimeoutMs?: number; + /** + * Post the agent's open question on the Epic for human visibility when it + * parks on `asked-question`. Optional + injectable so tests need no `gh`; + * the default (wired by the dispatcher) reads `.middle/blocked.json` and + * comments on the issue. + */ + postQuestion?: (opts: { + repo: string; + epicNumber: number; + worktreePath: string; + }) => Promise; }; const DEFAULT_LAUNCH_TIMEOUT_MS = 90_000; @@ -101,6 +140,17 @@ time. Operating rules for this dispatch: ); } +/** A park-worthy stop ends the session and waits for a human/reviewer signal. */ +function isParkKind(kind: StopClassification["kind"]): boolean { + return kind === "asked-question" || kind === "done"; +} + +/** The resume reason a park-worthy classification maps to. */ +function reasonFor(kind: StopClassification["kind"]): ResumeReason { + return kind === "done" ? "review-changes" : "answered-question"; +} + +/** The terminal `workflows.state` a settled classification resolves to. */ function finalStateFor(classification: StopClassification): WorkflowState { switch (classification.kind) { case "done": @@ -110,9 +160,12 @@ function finalStateFor(classification: StopClassification): WorkflowState { case "rate-limited": return "rate-limited"; case "asked-question": + // A resumed asked-question that did not settle stays parked for a human; + // a single-cycle resume cannot re-park in this execution (re-park is the + // re-enqueue path, sub-issue #36). return "waiting-human"; case "bare-stop": - // the minimal 3-step workflow has no nudge loop — a clean stop is terminal here + // the minimal spine has no nudge loop — a clean stop is terminal here return "completed"; } } @@ -121,11 +174,25 @@ type PrepareResult = { handle: WorktreeHandle }; type DriveResult = { classification: StopClassification; sessionName: string }; /** - * The Phase 1 `implementation` workflow — deliberately just three steps: - * prepare-worktree → launch-and-drive → cleanup. No skill enforcement, no - * sub-issue plan resolution, no hook-driven heartbeats; those land in Phases - * 2 and 4. `launch-and-drive` runs the launch → drive → observe loop and reacts - * to the `Stop` boundary via the adapter's `classifyStop`. + * The `implementation` workflow with the Phase 5 park → external-signal → + * resume spine: + * + * prepare-worktree → launch-and-drive → branch(park | terminal) + * → waitFor(RESUME_EVENT) → resume-or-finalize + * + * `launch-and-drive` runs the launch → drive → observe loop and ends the + * session at the `Stop` boundary (every classify outcome frees the slot). The + * branch arms a durable `waitfor_signals` row and parks the workflow in + * `waiting-human` for park-worthy stops (`asked-question`, `done`), or — for + * terminal stops — pre-seeds the signal so the single top-level `waitFor` falls + * through without parking. `resume-or-finalize` consumes the signal and + * re-drives a fresh session on resume, then finalizes (worktree teardown + + * terminal state). + * + * bunqueue's branch `.path()` bodies and loop bodies are *steps only* — a + * `waitFor` nested inside is silently dropped — and `engine.signal(id, event)` + * targets one execution, so the `waitFor` is a single top-level node and the + * loop-back for additional review rounds is re-enqueue (sub-issue #36). * * Built as a factory so the dispatcher injects real collaborators and tests * inject stubs. The workflow's `executionId` doubles as the `workflows.id`. @@ -164,8 +231,17 @@ export function createImplementationWorkflow( updateWorkflow(deps.db, ctx.executionId, { state: "compensated" }); } - async function launchAndDrive(ctx: StepContext): Promise { - const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; + /** + * Launch (or resume) one interactive session in the worktree, drive one turn, + * and classify the `Stop`. Ends the session before returning — at `Stop` the + * turn is over and the slot frees regardless of outcome ("END SESSION" in the + * dispatch lifecycle). Shared by the initial drive and the resume drive. + */ + async function driveOnce( + ctx: StepContext, + handle: WorktreeHandle, + promptKind: "initial" | "resume" | "answer", + ): Promise { const adapter = deps.getAdapter(ctx.input.adapter); const sessionName = sessionNameFor(ctx.input); const sessionToken = crypto.randomUUID(); @@ -197,9 +273,9 @@ export function createImplementationWorkflow( }, }); // Clear any orphaned session of the same name left by a prior dispatch - // that was interrupted (Ctrl-C / crash) before its cleanup ran — - // otherwise newSession fails with "duplicate session". killSession is a - // no-op when nothing's there. + // (or this workflow's own prior drive) before its cleanup ran — otherwise + // newSession fails with "duplicate session". killSession is a no-op when + // nothing's there. await deps.tmux.killSession(sessionName); console.error(`${tag} launching tmux session: ${argv.join(" ")} (cwd=${handle.path})`); await deps.tmux.newSession({ sessionName, command: argv, cwd: handle.path, env }); @@ -220,8 +296,6 @@ export function createImplementationWorkflow( console.error( `${tag} SessionStart received — session_id=${startPayload.session_id ?? ""}`, ); - // dismissPromise will resolve on its own (answered the prompt, or never - // saw it within the polling window). No further enterAutoMode call. void dismissPromise; const transcriptPath = adapter.resolveTranscriptPath(startPayload); @@ -234,10 +308,10 @@ export function createImplementationWorkflow( const promptText = adapter.buildPromptText({ promptFile: ".middle/prompt.md", - kind: "initial", + kind: promptKind, epicNumber: ctx.input.epicNumber, }); - console.error(`${tag} sending prompt: "${promptText}"`); + console.error(`${tag} sending prompt (${promptKind}): "${promptText}"`); await deps.tmux.sendText(sessionName, promptText); await deps.tmux.sendEnter(sessionName); @@ -252,59 +326,143 @@ export function createImplementationWorkflow( worktree: handle.path, }); console.error(`${tag} Stop received — classification=${classification.kind}`); + // END SESSION — the turn is over; free the slot before parking/finalizing. + await deps.tmux.killSession(sessionName); return { classification, sessionName }; } catch (error) { // never leak a tmux session on the failure path; the compensation rolls // back the worktree - console.error(`${tag} step failed: ${(error as Error).message}`); + console.error(`${tag} drive failed: ${(error as Error).message}`); await deps.tmux.killSession(sessionName); throw error; } } - async function cleanup(ctx: StepContext): Promise { + async function launchAndDrive(ctx: StepContext): Promise { + const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; + return driveOnce(ctx, handle, "initial"); + } + + /** + * Park-worthy stop: arm the durable `waitfor_signals` row under the + * epic-scoped, reason-scoped name the poller watches, set `waiting-human`, + * and (for `asked-question`) post the question for human visibility. The + * session already ended in `driveOnce`. The top-level `waitFor` that follows + * then parks the execution because RESUME_EVENT is unset. + */ + async function parkForResume(ctx: StepContext): Promise { + const { classification } = ctx.steps["launch-and-drive"] as DriveResult; const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; - const { classification, sessionName } = ctx.steps["launch-and-drive"] as DriveResult; - await deps.tmux.killSession(sessionName); - await deps.worktree.destroyWorktree(handle); + const reason = reasonFor(classification.kind); + armWaitForSignal( + deps.db, + signalNameFor(ctx.input.epicNumber, reason), + ctx.executionId, + JSON.stringify({ reason }), + ); + updateWorkflow(deps.db, ctx.executionId, { state: "waiting-human" }); + if (classification.kind === "asked-question" && deps.postQuestion) { + try { + await deps.postQuestion({ + repo: ctx.input.repo, + epicNumber: ctx.input.epicNumber, + worktreePath: handle.path, + }); + } catch (error) { + // Visibility is best-effort — the wait is already armed and durable, so + // a failed comment must not abort the park. + console.error(`[workflow] postQuestion failed: ${(error as Error).message}`); + } + } + } - const finalState = finalStateFor(classification); + /** + * Terminal stop: record rate-limit bookkeeping and pre-seed RESUME_EVENT so + * the single top-level `waitFor` falls through without parking. The final + * `workflows.state` is set in `resume-or-finalize` (alongside worktree + * teardown), so all terminal handling lives in one place. `ctx.signals` is + * the live `exec.signals` (passed by reference), which is exactly what the + * downstream `waitFor` reads. + */ + async function recordTerminal(ctx: StepContext): Promise { + const { classification } = ctx.steps["launch-and-drive"] as DriveResult; if (classification.kind === "rate-limited") { - // Reactive rate-limit: record the durable signal the auto-dispatch loop - // (Phase 8) reads to delay re-enqueue until reset_at. resetAt is the raw - // text the transcript carried after "Resets at "; parse it to unix ms, - // null when unrecognized (RATE_LIMITED with an unknown reset). setRateLimited(deps.db, { adapter: ctx.input.adapter, resetAt: parseResetAt(classification.resetAt), source: "transcript", detail: classification.resetAt, }); - } else if (finalState === "completed") { + } + (ctx.signals as Record)[RESUME_EVENT] = { terminal: true }; + } + + /** + * Reached after the `waitFor` resolves. Terminal stops fall straight through + * (the signal was pre-seeded). Park-worthy stops only reach here once the + * poller has fired `engine.signal(id, RESUME_EVENT, …)` — so consume the + * durable row and re-drive a fresh session re-primed per reason, then + * finalize on the resumed outcome. Worktree teardown + terminal state happen + * here, once, for every path. + */ + async function resumeOrFinalize(ctx: StepContext): Promise { + const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; + const initial = ctx.steps["launch-and-drive"] as DriveResult; + + let settled = initial.classification; + if (isParkKind(initial.classification.kind)) { + // We were resumed: consume the durable wait record and re-drive. + consumeWaitForSignal(deps.db, ctx.executionId); + const reason = reasonFor(initial.classification.kind); + const promptKind = reason === "answered-question" ? "answer" : "resume"; + const resumed = await driveOnce(ctx, handle, promptKind); + settled = resumed.classification; + } + + // Finalize: tear the worktree down and resolve the terminal state. + await deps.worktree.destroyWorktree(handle); + if (settled.kind !== "rate-limited" && finalStateFor(settled) === "completed") { // Probe-via-real-work: a completed dispatch proves the adapter is serving // again, so a previously RATE_LIMITED adapter reverts to AVAILABLE. markAvailableOnSuccess(deps.db, ctx.input.adapter); + } else if (settled.kind === "rate-limited") { + setRateLimited(deps.db, { + adapter: ctx.input.adapter, + resetAt: parseResetAt(settled.resetAt), + source: "transcript", + detail: settled.resetAt, + }); } - - updateWorkflow(deps.db, ctx.executionId, { state: finalState }); + updateWorkflow(deps.db, ctx.executionId, { state: finalStateFor(settled) }); } - return new Workflow("implementation") - .step("prepare-worktree", prepareWorktree, { compensate: cleanupWorktree }) - // timeout: must exceed the step's OWN internal waits (launchTimeout for - // SessionStart + stopTimeout for Stop), or bunqueue's default 30s step - // timeout fires mid-work and kills the live session. The internal - // awaitSessionStart/awaitStop timeouts stay the controlling ones (they give - // specific errors); this is a backstop just above them. - // retry: 1 — bunqueue's `retry` is `maxAttempts` (loop runs `attempt = 1 - // … <= retry`), not "retries after the first attempt". `1` means exactly - // one attempt, no retries. Phase 1 fails fast and compensates: retrying a - // launch piles up tmux/branch state and aggravates bunqueue's - // job-lifecycle race on the failure path. The full workflow's retry - // budgets (spec) live on `plan` / `implement-loop`. - .step("launch-and-drive", launchAndDrive, { - retry: 1, - timeout: launchTimeout + stopTimeout + 60_000, - }) - .step("cleanup", cleanup); + return ( + new Workflow("implementation") + .step("prepare-worktree", prepareWorktree, { compensate: cleanupWorktree }) + // timeout: must exceed the step's OWN internal waits (launchTimeout for + // SessionStart + stopTimeout for Stop), or bunqueue's default 30s step + // timeout fires mid-work and kills the live session. The internal + // awaitSessionStart/awaitStop timeouts stay the controlling ones (they + // give specific errors); this is a backstop just above them. retry: 1 — + // bunqueue's `retry` is `maxAttempts`; `1` means one attempt, no retries. + .step("launch-and-drive", launchAndDrive, { + retry: 1, + timeout: launchTimeout + stopTimeout + 60_000, + }) + .branch((ctx) => + isParkKind((ctx.steps["launch-and-drive"] as DriveResult).classification.kind) + ? "park" + : "terminal", + ) + .path("park", (w) => w.step("park-for-resume", parkForResume)) + .path("terminal", (w) => w.step("record-terminal", recordTerminal)) + // Single top-level `waitFor`: parks park-worthy stops until the poller + // fires RESUME_EVENT; terminal stops pre-seeded the signal and fall + // through. Same timeout budget as the drive step. + .waitFor(RESUME_EVENT, { timeout: WAITFOR_TIMEOUT_MS }) + .step("resume-or-finalize", resumeOrFinalize, { + retry: 1, + timeout: launchTimeout + stopTimeout + 60_000, + }) + ); } diff --git a/packages/dispatcher/test/implementation-workflow.test.ts b/packages/dispatcher/test/implementation-workflow.test.ts index 5c97489c..7c7abadf 100644 --- a/packages/dispatcher/test/implementation-workflow.test.ts +++ b/packages/dispatcher/test/implementation-workflow.test.ts @@ -8,9 +8,11 @@ import { Engine } from "bunqueue/workflow"; import { openAndMigrate } from "../src/db.ts"; import type { SessionGate } from "../src/hook-server.ts"; import { getRateLimitState, setRateLimited } from "../src/rate-limits.ts"; -import { getWorkflow } from "../src/workflow-record.ts"; +import { getWaitForSignal, getWorkflow } from "../src/workflow-record.ts"; import { createImplementationWorkflow, + RESUME_EVENT, + signalNameFor, type ImplementationDeps, } from "../src/workflows/implementation.ts"; import { createWorktree, destroyWorktree, listWorktrees } from "../src/worktree.ts"; @@ -87,14 +89,28 @@ const readyGate: SessionGate = { awaitStop: async () => ({ reason: "turn-end" }) as HookPayload, }; -/** A minimal AgentAdapter stub with a configurable classifyStop outcome. */ -function makeAdapterStub(classification: StopClassification): AgentAdapter { +/** + * A minimal AgentAdapter stub. `classifyStop` returns each supplied + * classification in turn (one per drive); the last value repeats — so a single + * value behaves as a constant, and a `[asked-question, done]` pair models a + * park that resumes to completion. `prompts` records every `buildPromptText` + * kind so tests can assert resume framing. + */ +function makeAdapterStub( + classifications: StopClassification | StopClassification[], + prompts: string[] = [], +): AgentAdapter { + const seq = Array.isArray(classifications) ? [...classifications] : [classifications]; + let i = 0; return { name: "stub", readyEvent: "session.started", async installHooks() {}, buildLaunchCommand: () => ({ argv: ["true"], env: {} }), - buildPromptText: () => "@.middle/prompt.md", + buildPromptText: (opts) => { + prompts.push(opts.kind); + return `@.middle/prompt.md (${opts.kind})`; + }, async enterAutoMode() {}, resolveTranscriptPath: (payload) => payload.transcript_path as string, readTranscriptState: () => ({ @@ -103,7 +119,7 @@ function makeAdapterStub(classification: StopClassification): AgentAdapter { turnCount: 0, lastToolUse: null, }), - classifyStop: () => classification, + classifyStop: () => seq[Math.min(i++, seq.length - 1)]!, }; } @@ -131,86 +147,165 @@ function expectNoSessionLeak(tmux: { created: string[]; killed: string[] }): voi } } -async function runToEnd(deps: ImplementationDeps): Promise { +const EPIC = 6; +const INPUT = { repo: "thejustinwalsh/middle", epicNumber: EPIC, adapter: "stub" }; + +async function start(deps: ImplementationDeps): Promise { engine.register(createImplementationWorkflow(deps)); - const handle = await engine.start("implementation", { - repo: "thejustinwalsh/middle", - epicNumber: 6, - adapter: "stub", - }); - const deadline = Date.now() + 5000; + const handle = await engine.start("implementation", INPUT); + return handle.id; +} + +/** + * Wait until the execution is genuinely parked on the `waitFor` node — bunqueue + * `exec.state === 'waiting'`. Signalling before the branch has advanced to the + * `waitFor` would race the park; in production the poller only fires after a + * real reply, long after parking. Asserts the `workflows` row reads + * `waiting-human` once parked. + */ +async function awaitParked(id: string, timeoutMs = 5000): Promise { + const deadline = Date.now() + timeoutMs; while (Date.now() < deadline) { - const execution = engine.getExecution(handle.id); - if (execution && execution.state !== "running" && execution.state !== "compensating") { - return handle.id; + if (engine.getExecution(id)?.state === "waiting") { + expect(getWorkflow(db, id)?.state).toBe("waiting-human"); + return; } await Bun.sleep(15); } - throw new Error("workflow did not settle within 5s"); + throw new Error( + `workflow ${id} did not park within ${timeoutMs}ms (exec '${engine.getExecution(id)?.state}', row '${getWorkflow(db, id)?.state}')`, + ); } -describe("implementation workflow — happy path", () => { - test("runs prepare → drive → cleanup, ends 'completed', leaks nothing", async () => { +/** Run the engine until the workflow row reaches a terminal-ish state. */ +async function awaitSettled(id: string, timeoutMs = 5000): Promise { + const terminal = new Set(["completed", "failed", "rate-limited", "compensated", "cancelled"]); + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const s = getWorkflow(db, id)?.state; + if (s && terminal.has(s)) return s; + await Bun.sleep(15); + } + throw new Error(`workflow ${id} did not settle within ${timeoutMs}ms (was '${getWorkflow(db, id)?.state}')`); +} + +describe("implementation workflow — terminal stops fall through the waitFor", () => { + test("a 'failed' classifyStop ends 'failed', destroys the worktree, leaks no session", async () => { const tmux = makeTmuxStub(); const deps = makeDeps({ tmux: tmux.ops, - getAdapter: () => makeAdapterStub({ kind: "done" }), + getAdapter: () => makeAdapterStub({ kind: "failed", reason: "stub failure" }), }); - const id = await runToEnd(deps); + const id = await start(deps); - const record = getWorkflow(db, id)!; - expect(record.state).toBe("completed"); - expect(record.epicNumber).toBe(6); - expect(record.sessionName).toBe("middle-thejustinwalsh-middle-6"); - expect(record.sessionId).toBe("stub-session"); - expect(record.transcriptPath).toBe("/tmp/stub.jsonl"); - - // no worktree leak + expect(await awaitSettled(id)).toBe("failed"); + expect(getWaitForSignal(db, id)).toBeNull(); // never armed a wait expect(await listWorktrees({ repoPath, worktreeRoot })).toEqual([]); - // no session leak — every created session was killed expectNoSessionLeak(tmux); }); - test("a 'failed' classifyStop ends the workflow 'failed' but still cleans up", async () => { + test("a 'bare-stop' ends 'completed' without parking", async () => { + const deps = makeDeps({ getAdapter: () => makeAdapterStub({ kind: "bare-stop" }) }); + const id = await start(deps); + expect(await awaitSettled(id)).toBe("completed"); + expect(getWaitForSignal(db, id)).toBeNull(); + }); + + test("a rate-limited classifyStop ends 'rate-limited' and records rate_limit_state", async () => { const tmux = makeTmuxStub(); + const resetAt = "2026-05-23T18:00:00Z"; const deps = makeDeps({ tmux: tmux.ops, - getAdapter: () => makeAdapterStub({ kind: "failed", reason: "stub failure" }), + getAdapter: () => makeAdapterStub({ kind: "rate-limited", resetAt }), }); - const id = await runToEnd(deps); + const id = await start(deps); - expect(getWorkflow(db, id)!.state).toBe("failed"); + expect(await awaitSettled(id)).toBe("rate-limited"); + const state = getRateLimitState(db, "stub")!; + expect(state.status).toBe("RATE_LIMITED"); + expect(state.resetAt).toBe(Date.parse(resetAt)); expect(await listWorktrees({ repoPath, worktreeRoot })).toEqual([]); expectNoSessionLeak(tmux); }); }); -describe("implementation workflow — rate-limit state", () => { - test("a rate-limited classifyStop ends 'rate-limited' and records rate_limit_state", async () => { +describe("implementation workflow — asked-question park → answer → resume", () => { + test("parks on asked-question (waiting-human, answered signal armed, worktree kept), then a signal resumes to completion", async () => { const tmux = makeTmuxStub(); - const resetAt = "2026-05-23T18:00:00Z"; + const prompts: string[] = []; + const postQuestionCalls: number[] = []; + // First drive asks a question; the resumed drive finishes (done). + // One stub instance shared across drives so its classification sequence + // advances (initial → asked-question, resume → done). + const adapter = makeAdapterStub( + [{ kind: "asked-question", sentinelPath: "/x/.middle/blocked.json" }, { kind: "done" }], + prompts, + ); const deps = makeDeps({ tmux: tmux.ops, - getAdapter: () => makeAdapterStub({ kind: "rate-limited", resetAt }), + getAdapter: () => adapter, + postQuestion: async ({ epicNumber }) => { + postQuestionCalls.push(epicNumber); + }, }); - const id = await runToEnd(deps); + const id = await start(deps); - expect(getWorkflow(db, id)!.state).toBe("rate-limited"); - const state = getRateLimitState(db, "stub")!; - expect(state.status).toBe("RATE_LIMITED"); - expect(state.resetAt).toBe(Date.parse(resetAt)); - expect(state.source).toBe("transcript"); - // worktree + session still cleaned up + // Parked: waiting-human, the epic-scoped 'answered' signal armed, worktree preserved. + await awaitParked(id); + expect(getWaitForSignal(db, id)).toEqual({ + signalName: signalNameFor(EPIC, "answered-question"), + payloadJson: JSON.stringify({ reason: "answered-question" }), + }); + expect(postQuestionCalls).toEqual([EPIC]); + expect((await listWorktrees({ repoPath, worktreeRoot })).length).toBe(1); + expect(prompts).toEqual(["initial"]); // resume drive not yet run + + // Human reply fires the signal → resume re-drives with the 'answer' prompt. + await engine.signal(id, RESUME_EVENT, { answer: "use option B" }); + expect(await awaitSettled(id)).toBe("completed"); + expect(prompts).toEqual(["initial", "answer"]); + expect(getWaitForSignal(db, id)).toBeNull(); // consumed on resume expect(await listWorktrees({ repoPath, worktreeRoot })).toEqual([]); expectNoSessionLeak(tmux); }); +}); + +describe("implementation workflow — done park → review-resolved → resume", () => { + test("parks on done (waiting-human, review-resolved signal armed), then a signal resumes", async () => { + const tmux = makeTmuxStub(); + const prompts: string[] = []; + const adapter = makeAdapterStub([{ kind: "done" }, { kind: "done" }], prompts); + const deps = makeDeps({ tmux: tmux.ops, getAdapter: () => adapter }); + const id = await start(deps); - test("a completed dispatch reverts a previously RATE_LIMITED adapter to AVAILABLE", async () => { - setRateLimited(db, { adapter: "stub", resetAt: Date.parse("2026-05-23T18:00:00Z"), source: "transcript" }); - const deps = makeDeps({ getAdapter: () => makeAdapterStub({ kind: "done" }) }); - const id = await runToEnd(deps); + await awaitParked(id); + expect(getWaitForSignal(db, id)).toEqual({ + signalName: signalNameFor(EPIC, "review-changes"), + payloadJson: JSON.stringify({ reason: "review-changes" }), + }); + // No postQuestion for the done/review path. + expect((await listWorktrees({ repoPath, worktreeRoot })).length).toBe(1); + + await engine.signal(id, RESUME_EVENT, { decision: "CHANGES_REQUESTED" }); + expect(await awaitSettled(id)).toBe("completed"); + expect(prompts).toEqual(["initial", "resume"]); // review-changes resumes with the 'resume' framing + expect(getWaitForSignal(db, id)).toBeNull(); + expect(await listWorktrees({ repoPath, worktreeRoot })).toEqual([]); + expectNoSessionLeak(tmux); + }); - expect(getWorkflow(db, id)!.state).toBe("completed"); + test("a completed resume reverts a previously RATE_LIMITED adapter to AVAILABLE", async () => { + setRateLimited(db, { + adapter: "stub", + resetAt: Date.parse("2026-05-23T18:00:00Z"), + source: "transcript", + }); + const adapter = makeAdapterStub([{ kind: "done" }, { kind: "done" }]); + const deps = makeDeps({ getAdapter: () => adapter }); + const id = await start(deps); + await awaitParked(id); + await engine.signal(id, RESUME_EVENT, {}); + expect(await awaitSettled(id)).toBe("completed"); expect(getRateLimitState(db, "stub")!.status).toBe("AVAILABLE"); }); }); @@ -226,20 +321,8 @@ describe("implementation workflow — compensation", () => { }; const deps = makeDeps({ tmux: tmux.ops, sessionGate: failingGate }); - engine.register(createImplementationWorkflow(deps)); - const handle = await engine.start("implementation", { - repo: "thejustinwalsh/middle", - epicNumber: 6, - adapter: "stub", - }); - const deadline = Date.now() + 5000; - while (Date.now() < deadline) { - const execution = engine.getExecution(handle.id); - if (execution && execution.state !== "running" && execution.state !== "compensating") break; - await Bun.sleep(15); - } - - expect(getWorkflow(db, handle.id)!.state).toBe("compensated"); + const id = await start(deps); + expect(await awaitSettled(id)).toBe("compensated"); expect(await listWorktrees({ repoPath, worktreeRoot })).toEqual([]); expectNoSessionLeak(tmux); }); From 7b3cf3d26694f658602a500d754bd67c73ddea68 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 00:12:02 -0400 Subject: [PATCH 3/9] feat(adapter-claude): surface blocked.json question/context in classifyStop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit classifyStop now reads the .middle/blocked.json question sentinel and carries its parsed contents on the asked-question classification (new BlockedSentinel type in core). Parsing is tolerant — a missing/malformed/contentless sentinel yields sentinel: null while still classifying asked-question (presence is the signal). The workflow's postQuestion seam receives the question + context so the parked Epic can show the human what's being asked. Closes #34 --- packages/adapters/claude/src/classify.ts | 31 +++++++++++++++++-- packages/adapters/claude/test/adapter.test.ts | 27 ++++++++++++++-- packages/core/src/adapter.ts | 14 ++++++++- packages/core/src/index.ts | 1 + .../src/workflows/implementation.ts | 13 ++++---- .../test/implementation-workflow.test.ts | 24 +++++++++++--- 6 files changed, 94 insertions(+), 16 deletions(-) diff --git a/packages/adapters/claude/src/classify.ts b/packages/adapters/claude/src/classify.ts index 96c49dd9..abd0abc6 100644 --- a/packages/adapters/claude/src/classify.ts +++ b/packages/adapters/claude/src/classify.ts @@ -1,6 +1,11 @@ import { existsSync, readFileSync } from "node:fs"; import { join } from "node:path"; -import type { HookPayload, RateLimitDetection, StopClassification } from "@middle/core"; +import type { + BlockedSentinel, + HookPayload, + RateLimitDetection, + StopClassification, +} from "@middle/core"; const USAGE_LIMIT_RE = /You've hit your usage limit\. Resets at (.+?)\./; @@ -27,7 +32,8 @@ export function classifyStop(opts: { const middleDir = join(opts.worktree, ".middle"); if (opts.sentinelPresent) { - return { kind: "asked-question", sentinelPath: join(middleDir, "blocked.json") }; + const sentinelPath = join(middleDir, "blocked.json"); + return { kind: "asked-question", sentinelPath, sentinel: readBlockedSentinel(sentinelPath) }; } const match = USAGE_LIMIT_RE.exec(readTail(opts.transcriptPath)); @@ -68,6 +74,27 @@ function readTail(path: string): string { } } +/** + * Read and tolerantly parse the `.middle/blocked.json` question sentinel so the + * workflow can surface the agent's question (and any context) to the human — + * e.g. posted on the Epic when it parks on `asked-question`. Returns `null` when + * the file is missing, unreadable, not JSON, or carries no string `question`: + * the Stop is still classified `asked-question` (the sentinel's *presence* is + * the signal), the contents are just best-effort. + */ +function readBlockedSentinel(path: string): BlockedSentinel | null { + try { + const parsed = JSON.parse(readFileSync(path, "utf8")) as Record; + if (typeof parsed.question !== "string" || parsed.question.length === 0) return null; + const context = typeof parsed.context === "string" ? parsed.context : undefined; + return context === undefined + ? { question: parsed.question } + : { question: parsed.question, context }; + } catch { + return null; + } +} + function readFailedReason(path: string): string { try { const parsed = JSON.parse(readFileSync(path, "utf8")) as { reason?: unknown }; diff --git a/packages/adapters/claude/test/adapter.test.ts b/packages/adapters/claude/test/adapter.test.ts index 09be4fbf..600a8992 100644 --- a/packages/adapters/claude/test/adapter.test.ts +++ b/packages/adapters/claude/test/adapter.test.ts @@ -171,8 +171,12 @@ function writeMiddleDir(): { cwd: string; middle: string; transcript: string } { } describe("classifyStop", () => { - test("sentinelPresent → asked-question, with the worktree-anchored blocked.json path", () => { - const { cwd, transcript } = writeMiddleDir(); + test("sentinelPresent → asked-question, surfacing the blocked.json path + question/context", () => { + const { cwd, middle, transcript } = writeMiddleDir(); + writeFileSync( + join(middle, "blocked.json"), + JSON.stringify({ question: "Use option A or B?", context: "Both pass typecheck." }), + ); const result = claudeAdapter.classifyStop({ payload: { cwd }, transcriptPath: transcript, @@ -182,6 +186,25 @@ describe("classifyStop", () => { expect(result.kind).toBe("asked-question"); if (result.kind === "asked-question") { expect(result.sentinelPath).toBe(join(cwd, ".middle", "blocked.json")); + expect(result.sentinel).toEqual({ + question: "Use option A or B?", + context: "Both pass typecheck.", + }); + } + }); + + test("asked-question tolerates a malformed/contentless blocked.json (sentinel → null)", () => { + const { cwd, middle, transcript } = writeMiddleDir(); + writeFileSync(join(middle, "blocked.json"), "{ not valid json"); + const result = claudeAdapter.classifyStop({ + payload: { cwd }, + transcriptPath: transcript, + sentinelPresent: true, + worktree: cwd, + }); + expect(result.kind).toBe("asked-question"); + if (result.kind === "asked-question") { + expect(result.sentinel).toBeNull(); } }); diff --git a/packages/core/src/adapter.ts b/packages/core/src/adapter.ts index d37bc3b2..06b9e825 100644 --- a/packages/core/src/adapter.ts +++ b/packages/core/src/adapter.ts @@ -86,9 +86,21 @@ export type TranscriptState = { lastToolUse: string | null; }; +/** + * The contents of a `.middle/blocked.json` question sentinel: the question the + * agent needs answered to proceed, plus optional supporting context the human + * needs to answer it. The skill writes this when it parks on `asked-question`. + * Parsed tolerantly — a sentinel that is missing or malformed yields `null` on + * the classification rather than failing the Stop. + */ +export type BlockedSentinel = { + question: string; + context?: string; +}; + export type StopClassification = | { kind: "done" } // agent marked the PR ready - | { kind: "asked-question"; sentinelPath: string } + | { kind: "asked-question"; sentinelPath: string; sentinel: BlockedSentinel | null } | { kind: "rate-limited"; resetAt: string /* ISO */ } | { kind: "bare-stop" } // stopped, no sentinel, not done | { kind: "failed"; reason: string }; diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index bb4c3f52..74927eb9 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -25,6 +25,7 @@ export type { LaunchOpts, TranscriptState, StopClassification, + BlockedSentinel, RateLimitDetection, } from "./adapter.ts"; diff --git a/packages/dispatcher/src/workflows/implementation.ts b/packages/dispatcher/src/workflows/implementation.ts index cba9d5b1..97b9c367 100644 --- a/packages/dispatcher/src/workflows/implementation.ts +++ b/packages/dispatcher/src/workflows/implementation.ts @@ -81,14 +81,15 @@ export type ImplementationDeps = { stopTimeoutMs?: number; /** * Post the agent's open question on the Epic for human visibility when it - * parks on `asked-question`. Optional + injectable so tests need no `gh`; - * the default (wired by the dispatcher) reads `.middle/blocked.json` and - * comments on the issue. + * parks on `asked-question`. Receives the sentinel contents `classifyStop` + * surfaced (`question` + optional `context`). Optional + injectable so tests + * need no `gh`; the default (wired by the dispatcher) comments on the issue. */ postQuestion?: (opts: { repo: string; epicNumber: number; - worktreePath: string; + question: string; + context?: string; }) => Promise; }; @@ -352,7 +353,6 @@ export function createImplementationWorkflow( */ async function parkForResume(ctx: StepContext): Promise { const { classification } = ctx.steps["launch-and-drive"] as DriveResult; - const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; const reason = reasonFor(classification.kind); armWaitForSignal( deps.db, @@ -366,7 +366,8 @@ export function createImplementationWorkflow( await deps.postQuestion({ repo: ctx.input.repo, epicNumber: ctx.input.epicNumber, - worktreePath: handle.path, + question: classification.sentinel?.question ?? "(question text unavailable)", + context: classification.sentinel?.context, }); } catch (error) { // Visibility is best-effort — the wait is already armed and durable, so diff --git a/packages/dispatcher/test/implementation-workflow.test.ts b/packages/dispatcher/test/implementation-workflow.test.ts index 7c7abadf..a86de149 100644 --- a/packages/dispatcher/test/implementation-workflow.test.ts +++ b/packages/dispatcher/test/implementation-workflow.test.ts @@ -233,19 +233,30 @@ describe("implementation workflow — asked-question park → answer → resume" test("parks on asked-question (waiting-human, answered signal armed, worktree kept), then a signal resumes to completion", async () => { const tmux = makeTmuxStub(); const prompts: string[] = []; - const postQuestionCalls: number[] = []; + const postQuestionCalls: Array<{ epicNumber: number; question: string; context?: string }> = []; // First drive asks a question; the resumed drive finishes (done). // One stub instance shared across drives so its classification sequence // advances (initial → asked-question, resume → done). const adapter = makeAdapterStub( - [{ kind: "asked-question", sentinelPath: "/x/.middle/blocked.json" }, { kind: "done" }], + [ + { + kind: "asked-question", + sentinelPath: "/x/.middle/blocked.json", + sentinel: { question: "Option A or B?", context: "Both compile." }, + }, + { kind: "done" }, + ], prompts, ); const deps = makeDeps({ tmux: tmux.ops, getAdapter: () => adapter, - postQuestion: async ({ epicNumber }) => { - postQuestionCalls.push(epicNumber); + postQuestion: async (opts) => { + postQuestionCalls.push({ + epicNumber: opts.epicNumber, + question: opts.question, + context: opts.context, + }); }, }); const id = await start(deps); @@ -256,7 +267,10 @@ describe("implementation workflow — asked-question park → answer → resume" signalName: signalNameFor(EPIC, "answered-question"), payloadJson: JSON.stringify({ reason: "answered-question" }), }); - expect(postQuestionCalls).toEqual([EPIC]); + // The sentinel's question + context are surfaced to the workflow's poster. + expect(postQuestionCalls).toEqual([ + { epicNumber: EPIC, question: "Option A or B?", context: "Both compile." }, + ]); expect((await listWorktrees({ repoPath, worktreeRoot })).length).toBe(1); expect(prompts).toEqual(["initial"]); // resume drive not yet run From 94f27ae13e4ad7b344ffed443acd9a5bc048e0aa Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 00:18:32 -0400 Subject: [PATCH 4/9] feat(dispatcher): GitHub poller for human replies and PR review state MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds runPoller: a pure pass over parked workflows (waiting-human + armed waitfor_signals) behind an injected GitHubPollGateway. For epic--answered it fires on a new non-bot reply; for epic--review-resolved it classifies the PR verdict — CHANGES_REQUESTED (review or changes-requested label) -> changes-requested, APPROVED or a fresh 0-actionable re-review -> resolved (so a bot reviewer that never flips its verdict doesn't hang the loop). Detect-and-fire only; #36 interprets the payload. Idempotent via a new waitfor_signals.fired_at column (migration 002): a fired wait is skipped until the workflow resumes and re-parks. Per-workflow failures (rate limits, transient errors) are isolated and retried next pass. Production gh gateway in poller-gateway.ts; 60s cron wired into main.ts alongside the watchdog (full cross-process resume hosting is Phase 8). Closes #35 --- .../src/db/migrations/002_waitfor_fired.sql | 9 + packages/dispatcher/src/main.ts | 20 ++ packages/dispatcher/src/poller-cron.ts | 34 ++ packages/dispatcher/src/poller-gateway.ts | 103 ++++++ packages/dispatcher/src/poller.ts | 185 +++++++++++ packages/dispatcher/src/workflow-record.ts | 50 +++ packages/dispatcher/test/db.test.ts | 23 +- packages/dispatcher/test/poller.test.ts | 301 ++++++++++++++++++ planning/issues/32/decisions.md | 39 +++ 9 files changed, 757 insertions(+), 7 deletions(-) create mode 100644 packages/dispatcher/src/db/migrations/002_waitfor_fired.sql create mode 100644 packages/dispatcher/src/poller-cron.ts create mode 100644 packages/dispatcher/src/poller-gateway.ts create mode 100644 packages/dispatcher/src/poller.ts create mode 100644 packages/dispatcher/test/poller.test.ts diff --git a/packages/dispatcher/src/db/migrations/002_waitfor_fired.sql b/packages/dispatcher/src/db/migrations/002_waitfor_fired.sql new file mode 100644 index 00000000..a2b57059 --- /dev/null +++ b/packages/dispatcher/src/db/migrations/002_waitfor_fired.sql @@ -0,0 +1,9 @@ +-- 002_waitfor_fired.sql +-- The GitHub poller (Phase 5) fires a workflow's resume signal once per park. +-- `fired_at` records when a signal was fired so a subsequent poll pass does not +-- re-fire the same wait before the workflow has resumed and consumed the row. +-- A fresh park (next review round) deletes-and-reinserts the row, clearing it. + +ALTER TABLE waitfor_signals ADD COLUMN fired_at INTEGER; + +INSERT OR IGNORE INTO schema_version VALUES (2); diff --git a/packages/dispatcher/src/main.ts b/packages/dispatcher/src/main.ts index b83d2214..3a89a02c 100644 --- a/packages/dispatcher/src/main.ts +++ b/packages/dispatcher/src/main.ts @@ -13,8 +13,11 @@ import { Engine } from "bunqueue/workflow"; import { openAndMigrate } from "./db.ts"; import { HookServer } from "./hook-server.ts"; import { DbHookStore } from "./hook-store.ts"; +import { ghPollGateway } from "./poller-gateway.ts"; +import { startPoller } from "./poller-cron.ts"; import { killSession, status } from "./tmux.ts"; import { startWatchdog } from "./watchdog-cron.ts"; +import { RESUME_EVENT } from "./workflows/implementation.ts"; /** Phase 2 adapter registry — only `claude` is implemented. */ function getAdapter(name: string): AgentAdapter { @@ -44,6 +47,18 @@ async function main(): Promise { getAdapter, }); + // GitHub poller: every 60s, for each parked workflow with an armed wait, fire + // its resume signal when the unblocking event appears (a human reply, or a PR + // review verdict). `fireSignal` delivers it to the engine that hosts the + // parked execution. NOTE: routing dispatches through this long-lived engine + // (so parked executions live here to be resumed) is the Phase 8 auto-dispatch + // integration; the poller + signal seam are in place ahead of it. + const stopPoller = await startPoller({ + db, + github: ghPollGateway, + fireSignal: (workflowId, payload) => engine.signal(workflowId, RESUME_EVENT, payload), + }); + console.log( `middle dispatcher up — hooks on :${hookServer.port}, db ${config.global.dbPath}`, ); @@ -59,6 +74,11 @@ async function main(): Promise { } catch (error) { console.error(`shutdown: stopWatchdog failed — ${(error as Error).message}`); } + try { + await stopPoller(); + } catch (error) { + console.error(`shutdown: stopPoller failed — ${(error as Error).message}`); + } try { hookServer.stop(); } catch (error) { diff --git a/packages/dispatcher/src/poller-cron.ts b/packages/dispatcher/src/poller-cron.ts new file mode 100644 index 00000000..4c3b7dd7 --- /dev/null +++ b/packages/dispatcher/src/poller-cron.ts @@ -0,0 +1,34 @@ +import { Bunqueue } from "bunqueue/client"; +import { runPoller, type PollerDeps } from "./poller.ts"; + +/** + * How often the poller checks GitHub for resume triggers. Slower than the + * watchdog (30s) — a human reply or a review verdict is not latency-sensitive, + * and a gentler cadence is kinder to GitHub rate limits. + */ +export const POLLER_INTERVAL_MS = 60_000; + +/** + * Stand up the GitHub poller as a bunqueue cron: every {@link POLLER_INTERVAL_MS} + * it runs one {@link runPoller} pass over parked workflows with an armed wait, + * firing the resume signal when the unblocking event appears. Returns a stop + * function that tears the cron down. The pass is resilient on its own (per- + * workflow failures are isolated); this wrapper guards the whole pass too so a + * thrown pass never crashes the cron worker. + */ +export async function startPoller(deps: PollerDeps): Promise<() => Promise> { + const queue = new Bunqueue("middle-poller", { + embedded: true, + processor: async () => { + try { + await runPoller(deps); + } catch (error) { + console.error(`[poller] pass failed: ${(error as Error).message}`); + } + }, + }); + await queue.every("poller-tick", POLLER_INTERVAL_MS); + return async () => { + await queue.close(true); + }; +} diff --git a/packages/dispatcher/src/poller-gateway.ts b/packages/dispatcher/src/poller-gateway.ts new file mode 100644 index 00000000..58d2f205 --- /dev/null +++ b/packages/dispatcher/src/poller-gateway.ts @@ -0,0 +1,103 @@ +import type { GitHubPollGateway, IssueComment, PrReview, PrSnapshot } from "./poller.ts"; + +/** + * The production {@link GitHubPollGateway} — reads issue comments and PR review + * state through the `gh` CLI. The poller's logic is unit-tested against an + * injected stub gateway; this is the thin subprocess glue that backs it in the + * dispatcher. Read-only: the poller never writes to GitHub. + */ + +async function gh(argv: string[]): Promise { + const proc = Bun.spawn(["gh", ...argv], { stdout: "pipe", stderr: "pipe", stdin: "ignore" }); + const [stdout, stderr] = await Promise.all([ + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + ]); + if ((await proc.exited) !== 0) { + throw new Error(`gh ${argv.join(" ")} failed: ${stderr.trim()}`); + } + return stdout; +} + +function isBotLogin(login: string, type: string | undefined): boolean { + return type === "Bot" || login.endsWith("[bot]"); +} + +export const ghPollGateway: GitHubPollGateway = { + async listIssueComments(repo: string, issueNumber: number): Promise { + const out = await gh([ + "api", + "--paginate", + `repos/${repo}/issues/${issueNumber}/comments`, + ]); + const rows = JSON.parse(out) as Array<{ + id: number; + body: string; + created_at: string; + user: { login: string; type?: string } | null; + }>; + return rows.map((r) => ({ + id: r.id, + body: r.body ?? "", + createdAt: Date.parse(r.created_at), + authorLogin: r.user?.login ?? "", + authorIsBot: isBotLogin(r.user?.login ?? "", r.user?.type), + })); + }, + + async findPrForEpic(repo: string, epicNumber: number): Promise { + // The Epic's one PR closes the Epic — find the open PR referencing it. + const listOut = await gh([ + "pr", + "list", + "--repo", + repo, + "--state", + "open", + "--search", + `in:body Closes #${epicNumber}`, + "--json", + "number", + ]); + const prs = JSON.parse(listOut) as Array<{ number: number }>; + const prNumber = prs[0]?.number; + if (prNumber === undefined) return null; + + const viewOut = await gh([ + "pr", + "view", + String(prNumber), + "--repo", + repo, + "--json", + "reviewDecision,labels", + ]); + const view = JSON.parse(viewOut) as { + reviewDecision: string | null; + labels: Array<{ name: string }>; + }; + + const reviewsOut = await gh(["api", "--paginate", `repos/${repo}/pulls/${prNumber}/reviews`]); + const reviewRows = JSON.parse(reviewsOut) as Array<{ + id: number; + state: string; + body: string; + submitted_at: string | null; + user: { login: string } | null; + }>; + const reviews: PrReview[] = reviewRows.map((r) => ({ + id: r.id, + state: r.state, + body: r.body ?? "", + submittedAt: r.submitted_at ? Date.parse(r.submitted_at) : 0, + authorLogin: r.user?.login ?? "", + })); + + return { + number: prNumber, + reviewDecision: view.reviewDecision ?? null, + reviews, + labels: view.labels.map((l) => l.name), + }; + }, +}; diff --git a/packages/dispatcher/src/poller.ts b/packages/dispatcher/src/poller.ts new file mode 100644 index 00000000..5c2567c9 --- /dev/null +++ b/packages/dispatcher/src/poller.ts @@ -0,0 +1,185 @@ +import type { Database } from "bun:sqlite"; +import type { ResumeReason } from "./workflows/implementation.ts"; +import { loadPollableWaits, markSignalFired } from "./workflow-record.ts"; + +/** + * The GitHub poller fires a parked workflow's resume signal when its unblocking + * event appears on GitHub — for both pause kinds, which share the + * park → external-signal → resume spine: + * + * - `answered-question` — a new human (non-bot) reply on the Epic resumes it. + * - `review-changes` — a PR review verdict. `CHANGES_REQUESTED` (a review or + * the `changes-requested` label) resumes the agent to address feedback; + * **resolved** — `APPROVED`, or a fresh re-review reporting **0 actionable + * comments** — ends the loop. The 0-actionable case matters because a bot + * reviewer (CodeRabbit) often won't flip `CHANGES_REQUESTED → APPROVED` on + * its own, so a clean re-review must count as resolved or the loop hangs on + * an approval that never comes. + * + * The poller only *detects and fires*; the resume step (sub-issue #36) + * interprets the payload (re-prime with the answer / review threads, the round + * cap, terminating on resolved). Firing is idempotent: a fired wait is marked + * (`fired_at`) and skipped until the workflow resumes and a fresh park rearms. + * + * Source of truth: build spec → "Build sequence" → "Phase 5". + */ + +/** One issue comment, normalized for bot detection + recency. */ +export type IssueComment = { + id: number; + authorLogin: string; + authorIsBot: boolean; + createdAt: number; // unix ms + body: string; +}; + +/** One PR review, normalized. `state` is GitHub's review state verb. */ +export type PrReview = { + id: number; + state: string; // 'APPROVED' | 'CHANGES_REQUESTED' | 'COMMENTED' | 'DISMISSED' | ... + authorLogin: string; + submittedAt: number; // unix ms + body: string; +}; + +/** A PR's review-relevant snapshot. */ +export type PrSnapshot = { + number: number; + reviewDecision: string | null; // 'APPROVED' | 'CHANGES_REQUESTED' | 'REVIEW_REQUIRED' | null + reviews: PrReview[]; + labels: string[]; +}; + +/** The read-only GitHub surface the poller needs — injectable so tests need no `gh`. */ +export type GitHubPollGateway = { + listIssueComments(repo: string, issueNumber: number): Promise; + /** The Epic's one open PR, or null if it hasn't been opened yet. */ + findPrForEpic(repo: string, epicNumber: number): Promise; +}; + +/** What the poller fires into the workflow's resume signal for #36 to interpret. */ +export type ResumeSignalPayload = + | { + reason: "answered-question"; + reply: { commentId: number; authorLogin: string; body: string }; + } + | { + reason: "review-changes"; + outcome: ReviewOutcome; + reviewId: number | null; + decision: string | null; + }; + +export type ReviewOutcome = "changes-requested" | "resolved"; + +export type PollerDeps = { + db: Database; + github: GitHubPollGateway; + /** Deliver the resume signal to the parked workflow (engine.signal in prod). */ + fireSignal: (workflowId: string, payload: ResumeSignalPayload) => Promise; + now?: () => number; +}; + +const ACTIONABLE_RE = /actionable comments posted:\s*(\d+)/i; + +/** The resume reason a durable signal name encodes, or null if not poller-driven. */ +export function reasonFromSignalName(name: string): ResumeReason | null { + if (name.endsWith("-review-resolved")) return "review-changes"; + // `epic--answered` (workflow) and `blocked:` (watchdog re-arm fallback) + // are both the question-sentinel pause. + if (name.endsWith("-answered") || name.startsWith("blocked:")) return "answered-question"; + return null; +} + +/** The newest non-bot reply posted after the wait armed, or null. */ +export function classifyNewHumanReply(comments: IssueComment[], sinceMs: number): IssueComment | null { + const fresh = comments + .filter((c) => !c.authorIsBot && c.createdAt > sinceMs) + .sort((a, b) => b.createdAt - a.createdAt); + return fresh[0] ?? null; +} + +/** + * Classify the PR's review state into a resume verdict, or null when nothing + * actionable has changed since the wait armed. The newest review submitted this + * round is authoritative; a 0-actionable re-review counts as **resolved** even + * while the PR's `reviewDecision` still reads `CHANGES_REQUESTED`. Falls back to + * the standing decision / `changes-requested` label when no fresh review exists. + */ +export function classifyReviewOutcome( + snapshot: PrSnapshot, + sinceMs: number, +): { outcome: ReviewOutcome; reviewId: number | null; decision: string | null } | null { + const fresh = snapshot.reviews + .filter((r) => r.submittedAt > sinceMs) + .sort((a, b) => b.submittedAt - a.submittedAt); + const latest = fresh[0]; + if (latest) { + if (latest.state === "APPROVED") { + return { outcome: "resolved", reviewId: latest.id, decision: "APPROVED" }; + } + const m = ACTIONABLE_RE.exec(latest.body); + if (m && Number(m[1]) === 0) { + // Clean re-review — resolved even if the decision hasn't flipped. + return { outcome: "resolved", reviewId: latest.id, decision: snapshot.reviewDecision }; + } + if (latest.state === "CHANGES_REQUESTED" || (m && Number(m[1]) > 0)) { + return { outcome: "changes-requested", reviewId: latest.id, decision: "CHANGES_REQUESTED" }; + } + } + // No fresh verdict from a review this round — fall back to standing state. + if (snapshot.reviewDecision === "APPROVED") { + return { outcome: "resolved", reviewId: null, decision: "APPROVED" }; + } + if (snapshot.reviewDecision === "CHANGES_REQUESTED" || snapshot.labels.includes("changes-requested")) { + return { outcome: "changes-requested", reviewId: null, decision: "CHANGES_REQUESTED" }; + } + return null; +} + +/** + * One poll pass over every parked workflow with an armed, not-yet-fired wait. + * Fires the resume signal when the unblocking event appears, then marks the + * wait fired (idempotent). Per-workflow failures (GitHub rate limits, transient + * errors) are isolated and logged — they skip that workflow this pass and are + * retried next pass; they never abort the pass for the others. Returns the + * number of signals fired (for logging/tests). + */ +export async function runPoller(deps: PollerDeps): Promise { + const now = (deps.now ?? Date.now)(); + let fired = 0; + for (const wait of loadPollableWaits(deps.db)) { + if (wait.firedAt !== null || wait.epicNumber === null) continue; + const reason = reasonFromSignalName(wait.signalName); + if (!reason) continue; + try { + if (reason === "answered-question") { + const comments = await deps.github.listIssueComments(wait.repo, wait.epicNumber); + const reply = classifyNewHumanReply(comments, wait.createdAt); + if (!reply) continue; + await deps.fireSignal(wait.workflowId, { + reason, + reply: { commentId: reply.id, authorLogin: reply.authorLogin, body: reply.body }, + }); + } else { + const pr = await deps.github.findPrForEpic(wait.repo, wait.epicNumber); + if (!pr) continue; + const verdict = classifyReviewOutcome(pr, wait.createdAt); + if (!verdict) continue; + await deps.fireSignal(wait.workflowId, { + reason, + outcome: verdict.outcome, + reviewId: verdict.reviewId, + decision: verdict.decision, + }); + } + markSignalFired(deps.db, wait.workflowId, now); + fired++; + } catch (error) { + console.error( + `[poller] poll failed for workflow ${wait.workflowId} (${wait.signalName}): ${(error as Error).message}`, + ); + } + } + return fired; +} diff --git a/packages/dispatcher/src/workflow-record.ts b/packages/dispatcher/src/workflow-record.ts index 67650735..761286da 100644 --- a/packages/dispatcher/src/workflow-record.ts +++ b/packages/dispatcher/src/workflow-record.ts @@ -190,6 +190,56 @@ export function armWaitForSignal( export type ArmedSignal = { signalName: string; payloadJson: string | null }; +/** A parked workflow the poller is watching: its armed wait joined to repo/epic. */ +export type PollableWait = { + workflowId: string; + repo: string; + epicNumber: number | null; + signalName: string; + createdAt: number; + firedAt: number | null; +}; + +/** + * Every armed wait on a parked (`waiting-human`) workflow, joined to its + * repo/epic — the poller's working set. Already-fired waits are included so the + * poller can decide idempotently; the poller filters on `firedAt`. + */ +export function loadPollableWaits(db: Database): PollableWait[] { + return db + .query( + `SELECT s.workflow_id, s.signal_name, s.created_at, s.fired_at, + w.repo, w.epic_number + FROM waitfor_signals s + JOIN workflows w ON w.id = s.workflow_id + WHERE w.state = 'waiting-human'`, + ) + .all() + .map((r) => { + const row = r as { + workflow_id: string; + signal_name: string; + created_at: number; + fired_at: number | null; + repo: string; + epic_number: number | null; + }; + return { + workflowId: row.workflow_id, + repo: row.repo, + epicNumber: row.epic_number, + signalName: row.signal_name, + createdAt: row.created_at, + firedAt: row.fired_at, + }; + }); +} + +/** Mark a workflow's armed wait as fired so the poller won't re-fire it. */ +export function markSignalFired(db: Database, workflowId: string, ts: number = Date.now()): void { + db.run("UPDATE waitfor_signals SET fired_at = ? WHERE workflow_id = ?", [ts, workflowId]); +} + /** * The signal armed for this workflow, or null. The poller reads this to learn * what an Epic is waiting on (the epic-scoped, reason-scoped `signal_name`) diff --git a/packages/dispatcher/test/db.test.ts b/packages/dispatcher/test/db.test.ts index bfe0a470..dc756860 100644 --- a/packages/dispatcher/test/db.test.ts +++ b/packages/dispatcher/test/db.test.ts @@ -58,10 +58,10 @@ describe("runMigrations", () => { db.close(); }); - test("applies 001_initial and reports version 1", () => { + test("applies every migration and reports the latest version", () => { const db = openDb(dbPath); - expect(runMigrations(db)).toBe(1); - expect(currentSchemaVersion(db)).toBe(1); + expect(runMigrations(db)).toBe(2); + expect(currentSchemaVersion(db)).toBe(2); db.close(); }); @@ -81,11 +81,20 @@ describe("runMigrations", () => { db.close(); }); - test("is idempotent — running twice leaves version at 1 and does not throw", () => { + test("is idempotent — running twice leaves version at the latest and does not throw", () => { const db = openDb(dbPath); runMigrations(db); - expect(runMigrations(db)).toBe(1); - expect(currentSchemaVersion(db)).toBe(1); + expect(runMigrations(db)).toBe(2); + expect(currentSchemaVersion(db)).toBe(2); + db.close(); + }); + + test("002 adds the waitfor_signals.fired_at column", () => { + const db = openAndMigrate(dbPath); + const cols = (db.query("PRAGMA table_info(waitfor_signals)").all() as { name: string }[]).map( + (c) => c.name, + ); + expect(cols).toContain("fired_at"); db.close(); }); @@ -117,7 +126,7 @@ describe("runMigrations", () => { describe("openAndMigrate", () => { test("opens, migrates, and returns a ready database", () => { const db = openAndMigrate(dbPath); - expect(currentSchemaVersion(db)).toBe(1); + expect(currentSchemaVersion(db)).toBe(2); db.close(); }); }); diff --git a/packages/dispatcher/test/poller.test.ts b/packages/dispatcher/test/poller.test.ts new file mode 100644 index 00000000..f77e69b7 --- /dev/null +++ b/packages/dispatcher/test/poller.test.ts @@ -0,0 +1,301 @@ +import { afterEach, beforeEach, describe, expect, test } from "bun:test"; +import type { Database } from "bun:sqlite"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { openAndMigrate } from "../src/db.ts"; +import { + classifyNewHumanReply, + classifyReviewOutcome, + reasonFromSignalName, + runPoller, + type GitHubPollGateway, + type IssueComment, + type PrSnapshot, + type ResumeSignalPayload, +} from "../src/poller.ts"; +import { + armWaitForSignal, + createWorkflowRecord, + getWaitForSignal, + updateWorkflow, +} from "../src/workflow-record.ts"; +import { signalNameFor, type ResumeReason } from "../src/workflows/implementation.ts"; + +let scratch: string; +let db: Database; + +beforeEach(() => { + scratch = mkdtempSync(join(tmpdir(), "middle-poll-")); + db = openAndMigrate(join(scratch, "db.sqlite3")); +}); + +afterEach(() => { + db.close(); + rmSync(scratch, { recursive: true, force: true }); +}); + +const REPO = "thejustinwalsh/middle"; +const EPIC = 32; +const ARMED_AT = 1_000_000; + +/** Seed a parked workflow with an armed wait for `reason`, armed at ARMED_AT. */ +function seedParked(reason: ResumeReason, epic = EPIC): string { + const id = crypto.randomUUID(); + createWorkflowRecord(db, { id, kind: "implementation", repo: REPO, epicNumber: epic, adapter: "claude" }); + updateWorkflow(db, id, { state: "waiting-human" }); + // armWaitForSignal stamps created_at = Date.now(); normalize it to ARMED_AT so + // recency comparisons in the poller are deterministic. + armWaitForSignal(db, signalNameFor(epic, reason), id, JSON.stringify({ reason })); + db.run("UPDATE waitfor_signals SET created_at = ? WHERE workflow_id = ?", [ARMED_AT, id]); + return id; +} + +function comment(over: Partial): IssueComment { + return { + id: 1, + authorLogin: "octocat", + authorIsBot: false, + createdAt: ARMED_AT + 1000, + body: "hi", + ...over, + }; +} + +function prSnapshot(over: Partial): PrSnapshot { + return { number: 90, reviewDecision: null, reviews: [], labels: [], ...over }; +} + +/** A gateway stub returning fixed comments / PR snapshot, recording calls. */ +function makeGateway(opts: { + comments?: IssueComment[]; + pr?: PrSnapshot | null; +}): GitHubPollGateway & { commentCalls: number; prCalls: number } { + const g = { + commentCalls: 0, + prCalls: 0, + async listIssueComments() { + g.commentCalls++; + return opts.comments ?? []; + }, + async findPrForEpic() { + g.prCalls++; + return opts.pr ?? null; + }, + }; + return g; +} + +function captureFires(): { + fired: Array<{ workflowId: string; payload: ResumeSignalPayload }>; + fireSignal: (id: string, p: ResumeSignalPayload) => Promise; +} { + const fired: Array<{ workflowId: string; payload: ResumeSignalPayload }> = []; + return { + fired, + fireSignal: async (workflowId, payload) => { + fired.push({ workflowId, payload }); + }, + }; +} + +describe("reasonFromSignalName", () => { + test("maps the durable signal names to resume reasons", () => { + expect(reasonFromSignalName("epic-32-answered")).toBe("answered-question"); + expect(reasonFromSignalName("epic-32-review-resolved")).toBe("review-changes"); + expect(reasonFromSignalName("blocked:wf_123")).toBe("answered-question"); + expect(reasonFromSignalName("something-else")).toBeNull(); + }); +}); + +describe("classifyNewHumanReply", () => { + test("returns the newest non-bot reply posted after the wait armed", () => { + const reply = classifyNewHumanReply( + [ + comment({ id: 1, createdAt: ARMED_AT + 100, body: "first" }), + comment({ id: 2, createdAt: ARMED_AT + 500, body: "newest" }), + comment({ id: 3, authorIsBot: true, createdAt: ARMED_AT + 900, body: "bot noise" }), + comment({ id: 4, createdAt: ARMED_AT - 100, body: "stale (pre-armed)" }), + ], + ARMED_AT, + ); + expect(reply?.id).toBe(2); + expect(reply?.body).toBe("newest"); + }); + + test("returns null when only bot/stale comments exist", () => { + expect( + classifyNewHumanReply( + [comment({ authorIsBot: true, createdAt: ARMED_AT + 100 }), comment({ createdAt: ARMED_AT - 1 })], + ARMED_AT, + ), + ).toBeNull(); + }); +}); + +describe("classifyReviewOutcome", () => { + test("a fresh CHANGES_REQUESTED review → changes-requested", () => { + const v = classifyReviewOutcome( + prSnapshot({ + reviewDecision: "CHANGES_REQUESTED", + reviews: [{ id: 7, state: "CHANGES_REQUESTED", authorLogin: "coderabbitai[bot]", submittedAt: ARMED_AT + 10, body: "Actionable comments posted: 3" }], + }), + ARMED_AT, + ); + expect(v).toEqual({ outcome: "changes-requested", reviewId: 7, decision: "CHANGES_REQUESTED" }); + }); + + test("a fresh APPROVED review → resolved", () => { + const v = classifyReviewOutcome( + prSnapshot({ + reviewDecision: "APPROVED", + reviews: [{ id: 8, state: "APPROVED", authorLogin: "human", submittedAt: ARMED_AT + 10, body: "lgtm" }], + }), + ARMED_AT, + ); + expect(v).toEqual({ outcome: "resolved", reviewId: 8, decision: "APPROVED" }); + }); + + test("a fresh 0-actionable re-review → resolved even while decision stays CHANGES_REQUESTED", () => { + const v = classifyReviewOutcome( + prSnapshot({ + reviewDecision: "CHANGES_REQUESTED", // bot didn't flip its standing verdict + reviews: [{ id: 9, state: "COMMENTED", authorLogin: "coderabbitai[bot]", submittedAt: ARMED_AT + 10, body: "**Actionable comments posted: 0**\n\nLooks good." }], + }), + ARMED_AT, + ); + expect(v).toEqual({ outcome: "resolved", reviewId: 9, decision: "CHANGES_REQUESTED" }); + }); + + test("the `changes-requested` label alone (no fresh review) → changes-requested", () => { + const v = classifyReviewOutcome(prSnapshot({ labels: ["changes-requested"] }), ARMED_AT); + expect(v).toEqual({ outcome: "changes-requested", reviewId: null, decision: "CHANGES_REQUESTED" }); + }); + + test("only stale reviews and no actionable label → null (nothing changed)", () => { + const v = classifyReviewOutcome( + prSnapshot({ + reviews: [{ id: 1, state: "CHANGES_REQUESTED", authorLogin: "x", submittedAt: ARMED_AT - 5, body: "old" }], + }), + ARMED_AT, + ); + expect(v).toBeNull(); + }); +}); + +describe("runPoller — answered-question", () => { + test("a new human reply fires epic--answered exactly once (idempotent across passes)", async () => { + const id = seedParked("answered-question"); + const github = makeGateway({ + comments: [comment({ id: 42, authorLogin: "maintainer", body: "Go with option B." })], + }); + const { fired, fireSignal } = captureFires(); + + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 5000 })).toBe(1); + expect(fired).toEqual([ + { + workflowId: id, + payload: { + reason: "answered-question", + reply: { commentId: 42, authorLogin: "maintainer", body: "Go with option B." }, + }, + }, + ]); + + // Second pass must NOT re-fire (fired_at guards it). + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 9000 })).toBe(0); + expect(fired.length).toBe(1); + }); + + test("a bot-only reply does not fire", async () => { + seedParked("answered-question"); + const github = makeGateway({ + comments: [comment({ id: 1, authorLogin: "coderabbitai[bot]", authorIsBot: true })], + }); + const { fired, fireSignal } = captureFires(); + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 5000 })).toBe(0); + expect(fired).toEqual([]); + }); +}); + +describe("runPoller — review-changes", () => { + test("CHANGES_REQUESTED fires review-resolved with outcome 'changes-requested'", async () => { + const id = seedParked("review-changes"); + const github = makeGateway({ + pr: prSnapshot({ + reviewDecision: "CHANGES_REQUESTED", + reviews: [{ id: 7, state: "CHANGES_REQUESTED", authorLogin: "coderabbitai[bot]", submittedAt: ARMED_AT + 10, body: "Actionable comments posted: 2" }], + }), + }); + const { fired, fireSignal } = captureFires(); + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 5000 })).toBe(1); + expect(fired[0]).toEqual({ + workflowId: id, + payload: { reason: "review-changes", outcome: "changes-requested", reviewId: 7, decision: "CHANGES_REQUESTED" }, + }); + }); + + test("APPROVED fires review-resolved as resolved", async () => { + seedParked("review-changes"); + const github = makeGateway({ + pr: prSnapshot({ + reviewDecision: "APPROVED", + reviews: [{ id: 8, state: "APPROVED", authorLogin: "human", submittedAt: ARMED_AT + 10, body: "ship it" }], + }), + }); + const { fired, fireSignal } = captureFires(); + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 5000 })).toBe(1); + expect(fired[0]!.payload).toEqual({ + reason: "review-changes", + outcome: "resolved", + reviewId: 8, + decision: "APPROVED", + }); + }); + + test("a 0-actionable re-review fires review-resolved as resolved", async () => { + seedParked("review-changes"); + const github = makeGateway({ + pr: prSnapshot({ + reviewDecision: "CHANGES_REQUESTED", + reviews: [{ id: 9, state: "COMMENTED", authorLogin: "coderabbitai[bot]", submittedAt: ARMED_AT + 10, body: "**Actionable comments posted: 0**" }], + }), + }); + const { fired, fireSignal } = captureFires(); + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 5000 })).toBe(1); + expect(fired[0]!.payload).toMatchObject({ reason: "review-changes", outcome: "resolved" }); + }); + + test("no PR yet → no fire", async () => { + seedParked("review-changes"); + const github = makeGateway({ pr: null }); + const { fired, fireSignal } = captureFires(); + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 5000 })).toBe(0); + expect(fired).toEqual([]); + }); +}); + +describe("runPoller — resilience", () => { + test("a gateway error for one workflow is isolated; others still fire", async () => { + const good = seedParked("answered-question", 100); + seedParked("answered-question", 200); // this one's gateway throws + + let n = 0; + const github: GitHubPollGateway = { + async listIssueComments(_repo, epicNumber) { + n++; + if (epicNumber === 200) throw new Error("API rate limit exceeded"); + return [comment({ id: 1, authorLogin: "human", body: "answer" })]; + }, + async findPrForEpic() { + return null; + }, + }; + const { fired, fireSignal } = captureFires(); + // One fires, one throws-and-is-skipped — the pass still completes. + expect(await runPoller({ db, github, fireSignal, now: () => ARMED_AT + 5000 })).toBe(1); + expect(n).toBe(2); + expect(fired.map((f) => f.workflowId)).toEqual([good]); + expect(getWaitForSignal(db, good)).not.toBeNull(); // row still present until resume consumes it + }); +}); diff --git a/planning/issues/32/decisions.md b/planning/issues/32/decisions.md index 64058693..265121d8 100644 --- a/planning/issues/32/decisions.md +++ b/planning/issues/32/decisions.md @@ -53,3 +53,42 @@ definition static while preserving the epic-scoped, reason-scoped naming the pol **Evidence:** `workflow.d.ts:24` (`waitFor(event: string, ...)`), `executor.js:83-97` (signal targets one execution), spec §"implementation workflow". + +## Poller idempotency via a `fired_at` column; detect-only, interpret in #36 +**File(s):** `packages/dispatcher/src/poller.ts`, `db/migrations/002_waitfor_fired.sql` +**Date:** 2026-05-24 + +**Decision:** The poller is a pure pass over parked workflows (`waiting-human` + an +armed `waitfor_signals` row) behind an injected `GitHubPollGateway`, mirroring the +`watchdog.ts` / `state-issue.ts` gateway pattern. It *detects and fires* only — it +classifies the trigger (new non-bot reply; review verdict) and calls `fireSignal`; +the resume step (#36) interprets the payload. Idempotency is a `fired_at` column on +`waitfor_signals`: a fired wait is skipped until the workflow resumes and a fresh +park (next round) deletes-and-reinserts the row. + +**Why:** Keeps the poller unit-testable without `gh` and keeps "what to do on resume" +(round cap, threads into the prompt, terminate-on-resolved) in one place (#36). The +0-actionable-re-review-counts-as-resolved rule lives in the classifier because the +poller must decide *whether* to fire and *what outcome* to report — a bot reviewer +often won't flip `CHANGES_REQUESTED → APPROVED`, so without it the loop would hang. + +**Evidence:** `poller.test.ts` (15 tests); acceptance §#35. + +## Poller wired into `main.ts`; cross-process resume hosting is Phase 8 +**File(s):** `packages/dispatcher/src/main.ts`, `poller-cron.ts` +**Date:** 2026-05-24 + +**Decision:** `startPoller` runs as a 60s bunqueue cron in the long-running dispatcher +alongside the watchdog, with `fireSignal = (id, p) => engine.signal(id, RESUME_EVENT, p)`. + +**Why:** The poller and the signal-delivery seam belong in the persistent process. But +today dispatches run through `dispatchEpic`'s throwaway engine (which drains when the +workflow parks — `waitForSettle` returns on `waiting`), so a parked execution does not +yet live on `main.ts`'s engine to be resumed. Routing dispatches through the persistent +engine + durable bunqueue + `recover()` is the **Phase 8 auto-dispatch** integration +(explicitly out of scope for Phase 5). Wiring the seam now keeps it ready; until Phase 8, +`fireSignal` for a not-yet-hosted execution is caught by the poller's per-workflow guard +and retried — it never crashes the pass. + +**Evidence:** `dispatch.ts:42-55` (`waitForSettle` returns on non-running/non-compensating, +i.e. `waiting`); spec §"Phase 8 — Auto-dispatch + limits". From 3804fbf5bf9a87df6d6e7f6da8a1da281bf32ce1 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 00:56:06 -0400 Subject: [PATCH 5/9] feat(dispatcher): re-enqueue continuation resume for answered-question and review-changes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Phase 5 spine (#33) parked and did a single inline resume drive. #36 replaces that with the re-enqueue continuation model the bunqueue reality forces: a park happens once per execution (no loop-back; loop bodies can't hold a waitFor), so every resume is a fresh execution re-primed from a ResumeInput that reuses the prior round's worktree — same branch, same PR. - resume-or-finalize interprets the fired signal: terminal stop / review resolved (APPROVED or clean re-review) -> finalize; answered-question or CHANGES_REQUESTED under the cap -> enqueueContinuation; a CHANGES_REQUESTED that would exceed the round cap (default 5) -> park waiting-human, no re-arm, no continuation (bounded loop). - prepare-worktree reuses input.resume.worktree and writes the reason-specific resume brief to .middle/prompt.md (answer injected, or the address-review brief pointing at the skill procedure). launch-and-drive picks answer/resume framing from the reason. - enqueueContinuation injected; wired in dispatch.ts to engine.start (the long-lived-engine hosting of parked execs is the Phase 8 integration). - Tests: asked-question e2e (reply injected into continuation brief), review e2e (address-review brief + APPROVED ends the loop), round-cap boundary (waiting-human at the cap, no further round). --- packages/dispatcher/src/dispatch.ts | 8 + .../src/workflows/implementation.ts | 258 ++++++++++++++++-- .../test/implementation-workflow.test.ts | 205 +++++++++++--- planning/issues/32/decisions.md | 48 ++++ 4 files changed, 456 insertions(+), 63 deletions(-) diff --git a/packages/dispatcher/src/dispatch.ts b/packages/dispatcher/src/dispatch.ts index 3e761260..8061c927 100644 --- a/packages/dispatcher/src/dispatch.ts +++ b/packages/dispatcher/src/dispatch.ts @@ -148,6 +148,14 @@ export async function dispatchEpic(opts: DispatchEpicOptions): Promise opts.repoPath, worktreeRoot: opts.worktreeRoot, dispatcherUrl: `http://127.0.0.1:${hookServer.port}`, + // Resume hand-off: a continuation round re-enters the same workflow on + // this engine. NOTE: `dispatchEpic`'s engine drains once the workflow + // parks (`waitForSettle` returns on `waiting`), so the continuation only + // actually runs once dispatches are hosted on the long-lived engine — + // the Phase 8 auto-dispatch integration. The seam is wired here ahead of it. + enqueueContinuation: async (input) => { + await engine.start("implementation", input); + }, }), ); diff --git a/packages/dispatcher/src/workflows/implementation.ts b/packages/dispatcher/src/workflows/implementation.ts index 97b9c367..3f2c539b 100644 --- a/packages/dispatcher/src/workflows/implementation.ts +++ b/packages/dispatcher/src/workflows/implementation.ts @@ -5,6 +5,7 @@ import type { AgentAdapter, StopClassification } from "@middle/core"; import { Workflow } from "bunqueue/workflow"; import type { StepContext } from "bunqueue/workflow"; import type { SessionGate } from "../hook-server.ts"; +import type { ResumeSignalPayload } from "../poller.ts"; import { markAvailableOnSuccess, parseResetAt, setRateLimited } from "../rate-limits.ts"; import type { CreateWorktreeOpts, WorktreeHandle } from "../worktree.ts"; import { @@ -15,11 +16,34 @@ import { type WorkflowState, } from "../workflow-record.ts"; +/** + * The handoff carried by a continuation execution. A park can only happen once + * per bunqueue execution (no loop-back; loop bodies can't hold a `waitFor`), so + * every resume is a *fresh* execution re-primed from this — reusing the prior + * round's worktree (no new branch / PR) and re-driving from the resume brief. + */ +export type ResumeInput = { + reason: ResumeReason; + /** Review-pass counter; one round = one whole `CHANGES_REQUESTED` pass. */ + round: number; + /** The worktree handle from the prior round — reused verbatim. */ + worktree: WorktreeHandle; + /** What the poller fired: the human's reply, or the review verdict. */ + payload: ResumeSignalPayload; +}; + /** A dispatch unit: an Epic (or standalone issue) pointed at one adapter. */ export type ImplementationInput = { repo: string; epicNumber: number; adapter: string; + /** + * Present only on a continuation execution (a resume). Absent on the initial + * dispatch. When set, `prepare-worktree` reuses `resume.worktree` instead of + * creating one, and writes the reason-specific resume brief to + * `.middle/prompt.md` before the drive. + */ + resume?: ResumeInput; }; /** @@ -91,10 +115,25 @@ export type ImplementationDeps = { question: string; context?: string; }) => Promise; + /** + * Enqueue a continuation execution for the next round (a resume). Injected so + * the workflow stays free of the engine: in prod the dispatcher wires this to + * `engine.start("implementation", input)` on the long-lived engine that hosts + * parked executions; tests wire it to their embedded engine. The continuation + * reuses the prior round's worktree via `input.resume.worktree`. + */ + enqueueContinuation: (input: ImplementationInput) => Promise; + /** + * The review-round ceiling: after this many `CHANGES_REQUESTED` passes without + * an `APPROVED`, the workflow parks in `waiting-human` and stops auto-resuming + * (a never-satisfied loop must not run forever). Defaults to 5. + */ + reviewRoundCap?: number; }; const DEFAULT_LAUNCH_TIMEOUT_MS = 90_000; const DEFAULT_STOP_TIMEOUT_MS = 4 * 60 * 60 * 1000; +const DEFAULT_REVIEW_ROUND_CAP = 5; /** * Session names are deterministic so compensations can recompute them, and @@ -141,6 +180,89 @@ time. Operating rules for this dispatch: ); } +/** + * Overwrite `.middle/prompt.md` with the reason-specific resume brief for a + * continuation execution. The agent re-reads this on its `@`-referenced resume + * drive (`buildPromptText` kind `answer` / `resume`): + * + * - `answered-question` — inlines the human's reply so the agent reads the + * answer and continues the workstream. + * - `review-changes` — an "address review" brief. The agent pulls the PR's + * review threads itself (`gh`) and follows the `implementing-github-issues` + * skill's "Addressing review feedback" procedure (batch → internal review + * loop → push once → reply in-thread → re-request review → re-park). Carries + * the round and cap so a bounded loop is visible to the agent. + * + * This unconditionally overwrites (unlike `ensurePromptFile`, which preserves an + * operator brief on the *initial* dispatch) — a resume's brief is the live one. + */ +function writeResumeBrief( + worktreePath: string, + epicNumber: number, + resume: ResumeInput, + reviewRoundCap: number, +): void { + const middleDir = join(worktreePath, ".middle"); + mkdirSync(middleDir, { recursive: true }); + const promptPath = join(middleDir, "prompt.md"); + const operatingRules = `## Operating rules for this dispatch + +- You are running autonomously under middle. There is no human watching in real + time. Continue the workstream — do not restart it. The branch, draft PR, + \`plan.md\`, and \`decisions.md\` are all intact. +- Work continuously; pause only if you are genuinely blocked (write + \`.middle/blocked.json\` and exit). The terminal state is the PR marked ready. +`; + + if (resume.reason === "answered-question") { + const reply = resume.payload.reason === "answered-question" ? resume.payload.reply : undefined; + const answer = reply + ? `> ${reply.body.replace(/\n/g, "\n> ")}\n\n— @${reply.authorLogin}` + : "(the human's reply text was unavailable — check the Epic thread on GitHub)"; + writeFileSync( + promptPath, + `# middle dispatch brief — Epic #${epicNumber} (resumed: a human answered) + +A human answered the open question you parked on. Their reply: + +${answer} + +Read this answer, fold it into your plan / decisions log, and continue the +workstream from where you left off. + +${operatingRules}`, + ); + return; + } + + // review-changes + const decision = resume.payload.reason === "review-changes" ? resume.payload.decision : null; + writeFileSync( + promptPath, + `# middle dispatch brief — Epic #${epicNumber} (resumed: address review — round ${resume.round} of ${reviewRoundCap}) + +A reviewer requested changes on the PR${decision ? ` (decision: ${decision})` : ""}. Address this +review pass now, following the \`implementing-github-issues\` skill's +**"Addressing review feedback"** procedure: + +1. Pull **every** open review thread on the PR yourself via \`gh\` (the review + comments and the review bodies). Read the whole pass before changing anything. +2. **Batch** the findings and resolve each **class-wide** — a fix plus a test per + fix, not one comment at a time. +3. Run the **internal clean-eyes review loop** over the batched diff (a review + subagent), looping until it surfaces nothing new, to catch adjacent edges + before re-review. +4. **Push once** — one push for the whole pass, not per fix. +5. Reply in-thread to each addressed comment, **re-request review**, then stop. + The workflow re-parks for the next verdict. + +This is review round ${resume.round} of ${reviewRoundCap}. After ${reviewRoundCap} rounds without an +\`APPROVED\` the workflow parks for a human and stops auto-resuming. + +${operatingRules}`, + ); +} + /** A park-worthy stop ends the session and waits for a human/reviewer signal. */ function isParkKind(kind: StopClassification["kind"]): boolean { return kind === "asked-question" || kind === "done"; @@ -161,9 +283,10 @@ function finalStateFor(classification: StopClassification): WorkflowState { case "rate-limited": return "rate-limited"; case "asked-question": - // A resumed asked-question that did not settle stays parked for a human; - // a single-cycle resume cannot re-park in this execution (re-park is the - // re-enqueue path, sub-issue #36). + // Defensive only: `finalize` is reached for terminal stops and the + // synthesized review-resolved `done`. Park kinds (`asked-question`, + // `done`) route to `parkForResume`, not here — a resume re-enqueues a + // continuation rather than finalizing in place. return "waiting-human"; case "bare-stop": // the minimal spine has no nudge loop — a clean stop is terminal here @@ -203,6 +326,7 @@ export function createImplementationWorkflow( ): Workflow { const launchTimeout = deps.launchTimeoutMs ?? DEFAULT_LAUNCH_TIMEOUT_MS; const stopTimeout = deps.stopTimeoutMs ?? DEFAULT_STOP_TIMEOUT_MS; + const reviewRoundCap = deps.reviewRoundCap ?? DEFAULT_REVIEW_ROUND_CAP; async function prepareWorktree(ctx: StepContext): Promise { createWorkflowRecord(deps.db, { @@ -212,6 +336,15 @@ export function createImplementationWorkflow( epicNumber: ctx.input.epicNumber, adapter: ctx.input.adapter, }); + const resume = ctx.input.resume; + if (resume) { + // Continuation: reuse the prior round's worktree (same branch, same PR — + // no new branch, no new PR) and re-prime the brief for this resume reason. + const handle = resume.worktree; + updateWorkflow(deps.db, ctx.executionId, { worktreePath: handle.path }); + writeResumeBrief(handle.path, ctx.input.epicNumber, resume, reviewRoundCap); + return { handle }; + } const handle = await deps.worktree.createWorktree({ repoPath: deps.resolveRepoPath(ctx.input.repo), repo: ctx.input.repo, @@ -341,7 +474,13 @@ export function createImplementationWorkflow( async function launchAndDrive(ctx: StepContext): Promise { const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; - return driveOnce(ctx, handle, "initial"); + const resume = ctx.input.resume; + const promptKind = !resume + ? "initial" + : resume.reason === "answered-question" + ? "answer" + : "resume"; + return driveOnce(ctx, handle, promptKind); } /** @@ -399,44 +538,103 @@ export function createImplementationWorkflow( } /** - * Reached after the `waitFor` resolves. Terminal stops fall straight through - * (the signal was pre-seeded). Park-worthy stops only reach here once the - * poller has fired `engine.signal(id, RESUME_EVENT, …)` — so consume the - * durable row and re-drive a fresh session re-primed per reason, then - * finalize on the resumed outcome. Worktree teardown + terminal state happen - * here, once, for every path. + * Tear the worktree down and resolve the terminal `workflows.state` for a + * settled classification. Called for genuinely-terminal stops and for a + * review-resolved (`APPROVED` / clean re-review) `done`. middle never merges — + * the human merges; this just records the terminal state and frees the worktree. */ - async function resumeOrFinalize(ctx: StepContext): Promise { - const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; - const initial = ctx.steps["launch-and-drive"] as DriveResult; - - let settled = initial.classification; - if (isParkKind(initial.classification.kind)) { - // We were resumed: consume the durable wait record and re-drive. - consumeWaitForSignal(deps.db, ctx.executionId); - const reason = reasonFor(initial.classification.kind); - const promptKind = reason === "answered-question" ? "answer" : "resume"; - const resumed = await driveOnce(ctx, handle, promptKind); - settled = resumed.classification; - } - - // Finalize: tear the worktree down and resolve the terminal state. + async function finalize( + ctx: StepContext, + handle: WorktreeHandle, + settled: StopClassification, + ): Promise { await deps.worktree.destroyWorktree(handle); - if (settled.kind !== "rate-limited" && finalStateFor(settled) === "completed") { - // Probe-via-real-work: a completed dispatch proves the adapter is serving - // again, so a previously RATE_LIMITED adapter reverts to AVAILABLE. - markAvailableOnSuccess(deps.db, ctx.input.adapter); - } else if (settled.kind === "rate-limited") { + if (settled.kind === "rate-limited") { setRateLimited(deps.db, { adapter: ctx.input.adapter, resetAt: parseResetAt(settled.resetAt), source: "transcript", detail: settled.resetAt, }); + } else if (finalStateFor(settled) === "completed") { + // Probe-via-real-work: a completed dispatch proves the adapter is serving + // again, so a previously RATE_LIMITED adapter reverts to AVAILABLE. + markAvailableOnSuccess(deps.db, ctx.input.adapter); } updateWorkflow(deps.db, ctx.executionId, { state: finalStateFor(settled) }); } + /** + * Reached after the `waitFor` resolves. Three outcomes: + * + * - **Terminal stop** — `record-terminal` pre-seeded `{ terminal: true }`, so + * this drive's own classification is final; `finalize` ends it. + * - **Review resolved** — the poller fired `outcome: "resolved"` (`APPROVED` + * or a clean re-review). The loop ends (terminal); the human merges. + * - **A continuing resume** — an answered question, or a `CHANGES_REQUESTED` + * pass under the round cap. Hand off to a fresh continuation execution that + * reuses this worktree (re-primed per reason); this round ends `completed` + * and the continuation becomes the Epic's live (latest non-terminal) row. + * + * The review-round counter increments **per pass**. Once a `CHANGES_REQUESTED` + * verdict would exceed `reviewRoundCap`, the workflow parks in `waiting-human` + * with no re-arm and no continuation — a never-satisfied loop is bounded. + */ + async function resumeOrFinalize(ctx: StepContext): Promise { + const { handle } = ctx.steps["prepare-worktree"] as PrepareResult; + const initial = ctx.steps["launch-and-drive"] as DriveResult; + const signal = (ctx.signals as Record)[RESUME_EVENT] as + | { terminal?: boolean } + | ResumeSignalPayload + | undefined; + + // Terminal stop: the branch pre-seeded the signal; this drive is final. + if (signal && (signal as { terminal?: boolean }).terminal) { + await finalize(ctx, handle, initial.classification); + return; + } + + // We genuinely parked, and the poller fired a resume verdict. Consume the + // durable wait record so the workflow no longer reads as parked. + const payload = signal as ResumeSignalPayload; + consumeWaitForSignal(deps.db, ctx.executionId); + + // A resolved review (APPROVED, or a 0-actionable re-review) ends the loop. + if (payload.reason === "review-changes" && payload.outcome === "resolved") { + await finalize(ctx, handle, { kind: "done" }); + return; + } + + // A continuing resume. Only a `CHANGES_REQUESTED` pass advances the review + // counter; an answered question carries the round through unchanged. + const currentRound = ctx.input.resume?.round ?? 0; + let nextRound = currentRound; + if (payload.reason === "review-changes") { + nextRound = currentRound + 1; + if (nextRound > reviewRoundCap) { + // Bounded: stop auto-resuming and park for a human. Keep the worktree; + // do not re-arm a wait (the poller stops watching) and do not re-enqueue. + // Everything the agent has pushed stays on the branch / PR. + updateWorkflow(deps.db, ctx.executionId, { state: "waiting-human" }); + return; + } + } + + // The drive that just parked ran a working adapter; revert any stale + // RATE_LIMITED before handing off. + markAvailableOnSuccess(deps.db, ctx.input.adapter); + // Hand control to a fresh continuation that reuses this worktree. + await deps.enqueueContinuation({ + repo: ctx.input.repo, + epicNumber: ctx.input.epicNumber, + adapter: ctx.input.adapter, + resume: { reason: payload.reason, round: nextRound, worktree: handle, payload }, + }); + // This round handed off — terminal in the bunqueue sense. The worktree is + // NOT torn down; the continuation reuses it. + updateWorkflow(deps.db, ctx.executionId, { state: "completed" }); + } + return ( new Workflow("implementation") .step("prepare-worktree", prepareWorktree, { compensate: cleanupWorktree }) diff --git a/packages/dispatcher/test/implementation-workflow.test.ts b/packages/dispatcher/test/implementation-workflow.test.ts index a86de149..352d69bd 100644 --- a/packages/dispatcher/test/implementation-workflow.test.ts +++ b/packages/dispatcher/test/implementation-workflow.test.ts @@ -1,6 +1,6 @@ import { afterEach, beforeEach, describe, expect, test } from "bun:test"; import type { Database } from "bun:sqlite"; -import { mkdtempSync, realpathSync, rmSync } from "node:fs"; +import { mkdtempSync, readFileSync, realpathSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; import type { AgentAdapter, HookPayload, StopClassification } from "@middle/core"; @@ -135,10 +135,59 @@ function makeDeps(overrides: Partial): ImplementationDeps { dispatcherUrl: "http://127.0.0.1:8822", launchTimeoutMs: 2000, stopTimeoutMs: 2000, + // Default: no continuation expected. Tests exercising the re-enqueue loop + // override this with the engine-backed harness below. + enqueueContinuation: async () => { + throw new Error("unexpected continuation enqueue"); + }, ...overrides, }; } +/** + * Wire `enqueueContinuation` to the test engine so a resume actually starts the + * next round as a fresh execution, recording each continuation's id. This is + * the production seam (`engine.start("implementation", input)`) under test — + * the re-enqueue loop the spec annotates `// loop back via re-enqueue`. + */ +function withContinuations(overrides: Partial): { + deps: ImplementationDeps; + continuationIds: string[]; +} { + const continuationIds: string[] = []; + const deps = makeDeps({ + ...overrides, + enqueueContinuation: async (input) => { + const handle = await engine.start("implementation", input); + continuationIds.push(handle.id); + }, + }); + return { deps, continuationIds }; +} + +/** Wait until the indexed continuation has been enqueued, returning its id. */ +async function awaitContinuation(ids: string[], index: number, timeoutMs = 5000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (ids[index]) return ids[index]!; + await Bun.sleep(15); + } + throw new Error(`continuation #${index} was not enqueued within ${timeoutMs}ms`); +} + +const CHANGES_REQUESTED = { + reason: "review-changes" as const, + outcome: "changes-requested" as const, + reviewId: 1, + decision: "CHANGES_REQUESTED", +}; +const APPROVED = { + reason: "review-changes" as const, + outcome: "resolved" as const, + reviewId: 2, + decision: "APPROVED", +}; + /** No session leak: every tmux session that was created was also killed. */ function expectNoSessionLeak(tmux: { created: string[]; killed: string[] }): void { expect(tmux.created.length).toBeGreaterThanOrEqual(1); @@ -229,14 +278,20 @@ describe("implementation workflow — terminal stops fall through the waitFor", }); }); -describe("implementation workflow — asked-question park → answer → resume", () => { - test("parks on asked-question (waiting-human, answered signal armed, worktree kept), then a signal resumes to completion", async () => { +/** The `.middle/prompt.md` written into the (shared) worktree, by workflow id. */ +function readPromptBrief(workflowId: string): string { + const path = getWorkflow(db, workflowId)?.worktreePath; + if (!path) throw new Error(`workflow ${workflowId} has no worktree path`); + return readFileSync(join(path, ".middle", "prompt.md"), "utf8"); +} + +describe("implementation workflow — asked-question park → answer → resume (e2e)", () => { + test("parks on asked-question, a human reply resumes a fresh continuation with the answer injected", async () => { const tmux = makeTmuxStub(); const prompts: string[] = []; const postQuestionCalls: Array<{ epicNumber: number; question: string; context?: string }> = []; - // First drive asks a question; the resumed drive finishes (done). - // One stub instance shared across drives so its classification sequence - // advances (initial → asked-question, resume → done). + // One shared stub instance so its classification sequence advances across + // both executions: initial → asked-question, the continuation → done. const adapter = makeAdapterStub( [ { @@ -248,7 +303,7 @@ describe("implementation workflow — asked-question park → answer → resume" ], prompts, ); - const deps = makeDeps({ + const { deps, continuationIds } = withContinuations({ tmux: tmux.ops, getAdapter: () => adapter, postQuestion: async (opts) => { @@ -259,71 +314,155 @@ describe("implementation workflow — asked-question park → answer → resume" }); }, }); - const id = await start(deps); + const id0 = await start(deps); - // Parked: waiting-human, the epic-scoped 'answered' signal armed, worktree preserved. - await awaitParked(id); - expect(getWaitForSignal(db, id)).toEqual({ + // Parked: waiting-human, the epic-scoped 'answered' signal armed, worktree kept. + await awaitParked(id0); + expect(getWaitForSignal(db, id0)).toEqual({ signalName: signalNameFor(EPIC, "answered-question"), payloadJson: JSON.stringify({ reason: "answered-question" }), }); - // The sentinel's question + context are surfaced to the workflow's poster. expect(postQuestionCalls).toEqual([ { epicNumber: EPIC, question: "Option A or B?", context: "Both compile." }, ]); expect((await listWorktrees({ repoPath, worktreeRoot })).length).toBe(1); - expect(prompts).toEqual(["initial"]); // resume drive not yet run + expect(prompts).toEqual(["initial"]); // continuation not yet driven - // Human reply fires the signal → resume re-drives with the 'answer' prompt. - await engine.signal(id, RESUME_EVENT, { answer: "use option B" }); - expect(await awaitSettled(id)).toBe("completed"); + // The poller fires the human's reply → a fresh continuation execution. + await engine.signal(id0, RESUME_EVENT, { + reason: "answered-question", + reply: { commentId: 7, authorLogin: "alice", body: "Use option B." }, + }); + // The original execution hands off and ends; its wait is consumed. + expect(await awaitSettled(id0)).toBe("completed"); + expect(getWaitForSignal(db, id0)).toBeNull(); + + // The continuation re-drives with the 'answer' prompt, reusing the worktree, + // and the human's reply is injected into its brief. + const id1 = await awaitContinuation(continuationIds, 0); + await awaitParked(id1); // the answered continuation reaches done → parks on review expect(prompts).toEqual(["initial", "answer"]); - expect(getWaitForSignal(db, id)).toBeNull(); // consumed on resume + expect(getWorkflow(db, id1)?.worktreePath).toBe(getWorkflow(db, id0)?.worktreePath); + const brief = readPromptBrief(id1); + expect(brief).toContain("a human answered"); + expect(brief).toContain("Use option B."); + expect(brief).toContain("@alice"); + // An answered question does not advance the review counter; it parks on review. + expect(getWaitForSignal(db, id1)).toEqual({ + signalName: signalNameFor(EPIC, "review-changes"), + payloadJson: JSON.stringify({ reason: "review-changes" }), + }); + + // Approve to end the loop cleanly and prove the worktree is torn down once. + await engine.signal(id1, RESUME_EVENT, APPROVED); + expect(await awaitSettled(id1)).toBe("completed"); expect(await listWorktrees({ repoPath, worktreeRoot })).toEqual([]); expectNoSessionLeak(tmux); }); }); -describe("implementation workflow — done park → review-resolved → resume", () => { - test("parks on done (waiting-human, review-resolved signal armed), then a signal resumes", async () => { +describe("implementation workflow — done park → review-changes → resume (e2e)", () => { + test("a CHANGES_REQUESTED pass resumes a continuation with the address-review brief; APPROVED ends the loop", async () => { const tmux = makeTmuxStub(); const prompts: string[] = []; - const adapter = makeAdapterStub([{ kind: "done" }, { kind: "done" }], prompts); - const deps = makeDeps({ tmux: tmux.ops, getAdapter: () => adapter }); - const id = await start(deps); + const adapter = makeAdapterStub({ kind: "done" }, prompts); + const { deps, continuationIds } = withContinuations({ tmux: tmux.ops, getAdapter: () => adapter }); + const id0 = await start(deps); - await awaitParked(id); - expect(getWaitForSignal(db, id)).toEqual({ + await awaitParked(id0); + expect(getWaitForSignal(db, id0)).toEqual({ signalName: signalNameFor(EPIC, "review-changes"), payloadJson: JSON.stringify({ reason: "review-changes" }), }); - // No postQuestion for the done/review path. expect((await listWorktrees({ repoPath, worktreeRoot })).length).toBe(1); - await engine.signal(id, RESUME_EVENT, { decision: "CHANGES_REQUESTED" }); - expect(await awaitSettled(id)).toBe("completed"); - expect(prompts).toEqual(["initial", "resume"]); // review-changes resumes with the 'resume' framing - expect(getWaitForSignal(db, id)).toBeNull(); + // A reviewer requests changes → resume a continuation to address them. + await engine.signal(id0, RESUME_EVENT, CHANGES_REQUESTED); + expect(await awaitSettled(id0)).toBe("completed"); + + const id1 = await awaitContinuation(continuationIds, 0); + await awaitParked(id1); + // Resumes with the 'resume' framing; the brief is the address-review brief + // (round 1 of the default cap 5) that points at the skill's procedure. + expect(prompts).toEqual(["initial", "resume"]); + const brief = readPromptBrief(id1); + expect(brief).toContain("address review — round 1 of 5"); + expect(brief).toContain("Addressing review feedback"); + expect(brief).toContain("Push once"); + expect(brief).toContain("CHANGES_REQUESTED"); + + // The agent re-requested review; an APPROVED verdict ends the loop (terminal). + await engine.signal(id1, RESUME_EVENT, APPROVED); + expect(await awaitSettled(id1)).toBe("completed"); + expect(continuationIds).toHaveLength(1); // no further round after APPROVED + expect(getWaitForSignal(db, id1)).toBeNull(); expect(await listWorktrees({ repoPath, worktreeRoot })).toEqual([]); expectNoSessionLeak(tmux); }); - test("a completed resume reverts a previously RATE_LIMITED adapter to AVAILABLE", async () => { + test("a resolved review reverts a previously RATE_LIMITED adapter to AVAILABLE", async () => { setRateLimited(db, { adapter: "stub", resetAt: Date.parse("2026-05-23T18:00:00Z"), source: "transcript", }); - const adapter = makeAdapterStub([{ kind: "done" }, { kind: "done" }]); - const deps = makeDeps({ getAdapter: () => adapter }); + const { deps } = withContinuations({ getAdapter: () => makeAdapterStub({ kind: "done" }) }); const id = await start(deps); await awaitParked(id); - await engine.signal(id, RESUME_EVENT, {}); + await engine.signal(id, RESUME_EVENT, APPROVED); expect(await awaitSettled(id)).toBe("completed"); expect(getRateLimitState(db, "stub")!.status).toBe("AVAILABLE"); }); }); +describe("implementation workflow — review-round cap", () => { + test("after the configured cap of CHANGES_REQUESTED passes without APPROVED, it parks in waiting-human and stops auto-resuming", async () => { + const tmux = makeTmuxStub(); + const adapter = makeAdapterStub({ kind: "done" }); + // Cap of 2: rounds 1 and 2 re-enqueue; the 3rd CHANGES_REQUESTED caps. + const { deps, continuationIds } = withContinuations({ + tmux: tmux.ops, + getAdapter: () => adapter, + reviewRoundCap: 2, + }); + const id0 = await start(deps); + + // Round 0 (initial) parks; request changes → round 1. + await awaitParked(id0); + await engine.signal(id0, RESUME_EVENT, CHANGES_REQUESTED); + expect(await awaitSettled(id0)).toBe("completed"); + + // Round 1 parks; request changes → round 2. + const id1 = await awaitContinuation(continuationIds, 0); + await awaitParked(id1); + expect(readPromptBrief(id1)).toContain("round 1 of 2"); + await engine.signal(id1, RESUME_EVENT, CHANGES_REQUESTED); + expect(await awaitSettled(id1)).toBe("completed"); + + // Round 2 parks; request changes again → would be round 3 > cap → capped. + const id2 = await awaitContinuation(continuationIds, 1); + await awaitParked(id2); + expect(readPromptBrief(id2)).toContain("round 2 of 2"); + await engine.signal(id2, RESUME_EVENT, CHANGES_REQUESTED); + + // Both "parked" and "capped" read as `waiting-human`, so wait on the + // definitive barrier: the bunqueue execution fully settling (the cap path + // runs `resume-or-finalize` to completion, which consumes id2's armed wait). + const deadline = Date.now() + 5000; + while (Date.now() < deadline) { + const s = engine.getExecution(id2)?.state; + if (s === "completed" || s === "failed") break; + await Bun.sleep(15); + } + // Capped: parks in waiting-human, no continuation enqueued, no armed wait + // (poller stops watching), worktree preserved for the human. + expect(getWorkflow(db, id2)?.state).toBe("waiting-human"); + expect(continuationIds).toHaveLength(2); // id1, id2 — no third round + expect(getWaitForSignal(db, id2)).toBeNull(); // consumed, not re-armed + expect((await listWorktrees({ repoPath, worktreeRoot })).length).toBe(1); + }); +}); + describe("implementation workflow — compensation", () => { test("a launch failure compensates: worktree rolled back, session freed, state 'compensated'", async () => { const tmux = makeTmuxStub(); diff --git a/planning/issues/32/decisions.md b/planning/issues/32/decisions.md index 265121d8..62776a84 100644 --- a/planning/issues/32/decisions.md +++ b/planning/issues/32/decisions.md @@ -92,3 +92,51 @@ and retried — it never crashes the pass. **Evidence:** `dispatch.ts:42-55` (`waitForSettle` returns on non-running/non-compensating, i.e. `waiting`); spec §"Phase 8 — Auto-dispatch + limits". + +## Multi-round resume = re-enqueue a continuation execution (one round = one execution) +**File(s):** `packages/dispatcher/src/workflows/implementation.ts` +**Date:** 2026-05-24 + +**Decision:** Each park/resume cycle is one bunqueue execution. `resume-or-finalize` +**interprets** the fired verdict and either finalizes (terminal / review *resolved*) +or **re-enqueues a continuation execution** (via an injected `enqueueContinuation` +dep) that carries `resume = { reason, round, worktree, payload }` in its input. The +continuation reuses the same worktree (its `prepare-worktree` skips `createWorktree` +and reuses the handle from `input.resume.worktree`) and drives the resume prompt in +its own `launch-and-drive`. The addressing drive therefore happens in the continuation, +not inline in `resume-or-finalize`. + +**Why:** A single execution can park only once (bunqueue has one top-level `waitFor` +per linear graph and no loop-back; loop bodies can't hold a `waitFor`). The review +loop needs up to `cap` real parks (each frees the session for a reviewer who may take +days), so the only expressible loop is re-enqueue — which the spec annotates twice +(`// loop back via re-enqueue`). The `waitfor_signals.workflow_id` must equal the +bunqueue execution id for `engine.signal` to target the parked execution, so each +round is necessarily a fresh execution (and a fresh `workflows` row, keyed by the same +`epic_number`); the live one is the latest non-terminal row. The round counter rides in +`input.resume.round`; `resume-or-finalize` increments per pass and parks in +`waiting-human` (no re-arm, no re-enqueue) once it would exceed the cap (default 5). + +**Evidence:** `#36` tests (asked-question e2e, review-changes single-round, cap boundary); +`executor.js` (no loop-back); spec §"implementation workflow". + +## The agent fetches review threads; the dispatcher writes the "address review" brief +**File(s):** `packages/dispatcher/src/workflows/implementation.ts`, +`packages/skills/implementing-github-issues/SKILL.md` +**Date:** 2026-05-24 + +**Decision:** On a `review-changes` continuation, the dispatcher overwrites +`.middle/prompt.md` with an "address review" brief (round, decision, the skill's +per-round procedure) and the agent pulls the PR's review threads itself via `gh`, +following the new **"Addressing review feedback"** section of the +`implementing-github-issues` skill (batch → internal clean-eyes review loop → push +once → reply in-thread → re-request review → re-park). + +**Why:** The agent is a full Claude session with `gh`; having it fetch live threads is +more robust than the dispatcher embedding a stale snapshot, and it keeps the dispatcher +GitHub-read-light. Codifying the procedure in the skill is what makes the autonomous +daemon loop and a hand-driven agent behave identically (the #36 acceptance's explicit +requirement). The brief in `.middle/prompt.md` is the "address-review brief" the threads +are pulled behind. + +**Evidence:** skill "Addressing review feedback" section; `prompt.ts` resume framing. From c6184b66cea3224d11b8cc6e0261db68eb050bed Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 00:57:03 -0400 Subject: [PATCH 6/9] docs(skill): codify 'Addressing review feedback' procedure in implementing-github-issues Adds Phase 11 (the review-changes loop the #36 resume brief points at) so the autonomous daemon loop and a hand-driven agent follow the same per-round procedure: batch the whole pass -> resolve class-wide with a test per fix -> internal clean-eyes review loop over the batched diff -> push once -> reply in-thread -> re-request review -> stop and wait for the next verdict. APPROVED ends the loop; the agent never merges. Also expands 'You may be resumed mid-workstream' under the headless-dispatch section: fresh session in the same worktree, the two resume reasons (answer injected / address-review brief), and the bounded review loop (round cap 5). Synced to the bootstrap-assets mirror and the dogfood .claude/.codex copies. --- .../implementing-github-issues/SKILL.md | 35 ++++++++++++++++++- .../implementing-github-issues/SKILL.md | 35 ++++++++++++++++++- .../implementing-github-issues/SKILL.md | 35 ++++++++++++++++++- .../implementing-github-issues/SKILL.md | 35 ++++++++++++++++++- 4 files changed, 136 insertions(+), 4 deletions(-) diff --git a/.claude/skills/implementing-github-issues/SKILL.md b/.claude/skills/implementing-github-issues/SKILL.md index d76a71c0..98953152 100644 --- a/.claude/skills/implementing-github-issues/SKILL.md +++ b/.claude/skills/implementing-github-issues/SKILL.md @@ -519,6 +519,34 @@ Closes # **Stop here.** The skill does not merge the PR. The human reviews and merges; that's the final gate. +## Phase 11 — Addressing review feedback (when changes are requested) + +Marking the PR ready (Phase 10) is not always the end. A reviewer — a human, or an automated reviewer like CodeRabbit — may request changes. **This is a loop, not a one-shot:** you address the pass, re-request review, and wait for the next verdict. An `APPROVED` ends it (the human merges; the skill never does). Each `CHANGES_REQUESTED` pass is **one round**; a never-satisfied loop is bounded (see "Running under middle" for the round cap). + +When you pick up a review pass, run this procedure **in order** — it's the same whether you're a human-driven agent or resumed by middle, so the two behave identically: + +1. **Pull the whole pass first — batch, don't drip.** Fetch *every* open review thread before changing a line: the inline comments and the review bodies. + ```bash + gh api repos/{owner}/{repo}/pulls//comments # inline thread comments + gh pr view --json reviews --jq '.reviews[] | {author: .author.login, state, body}' + ``` + Reading the whole pass first is what lets you fix by *class* instead of comment-by-comment. + +2. **Resolve each finding class-wide, with a test per fix.** If a comment flags one instance of a bug, grep for the rest and fix them all in one change. A fix without a regression test is not a fix — the test is the evidence the finding is closed and stays closed. + +3. **Run the internal clean-eyes review loop before re-requesting.** Dispatch a review subagent (`superpowers:requesting-code-review`) over the *batched* diff — the whole pass's changes, read fresh — and loop: address what it surfaces, re-run it, until a pass surfaces nothing new. This catches the adjacent edges the reviewer's spot-checks implied but didn't enumerate, so you don't burn a whole review round on something a clean-eyes pass would have caught. + +4. **Push once.** One push for the entire pass — not one per fix. The reviewer (and any re-review bot) reacts to a single new head commit, so batch-then-push keeps the review history legible and avoids re-triggering the bot mid-edit. + +5. **Reply in-thread, then re-request review.** Reply to each addressed comment on its own thread (what changed, where), then formally re-request review so the verdict re-fires: + ```bash + gh api repos/{owner}/{repo}/pulls//comments//replies -f body="Fixed in — " + gh pr edit --add-reviewer # or the API re-request; a bot re-reviews on the new push + ``` + Then **stop and wait for the next verdict.** Don't pre-emptively mark anything resolved you haven't addressed. + +**Do not** open a new PR, rebase away the review history, or merge. The branch and PR are the same long-lasting context they've always been; you're adding commits to them. + ## Quick reference | Step | Command | @@ -628,7 +656,12 @@ Completed sub-issues stay done on the branch; only the current one is paused. On ### You may be resumed mid-workstream -When a human answers, middle re-spawns you with the answer injected into your prompt. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. +When you parked — either because you asked a question (`blocked.json`) or because you marked the PR ready and a reviewer's verdict is pending — middle ends your session to free the slot and parks the workflow on a wait signal. When the unblocking event lands on GitHub, it **re-spawns you as a fresh session in the same worktree** (same branch, same PR — never a new one), re-primed via a rewritten `.middle/prompt.md`. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. Two resume reasons: + +- **A human answered your question.** The reply is injected into your brief. Fold it into your plan / decisions log and continue. +- **A reviewer requested changes.** Your brief is an "address review" brief naming the round. Pull the PR's review threads yourself (`gh`) and follow **Phase 11 — Addressing review feedback** exactly (batch → resolve class-wide with a test per fix → internal clean-eyes review loop → push once → reply in-thread → re-request review → stop). Then the workflow re-parks for the next verdict. + +The review loop is bounded: after the repo's configured round cap (default **5**) of `CHANGES_REQUESTED` passes without an `APPROVED`, the workflow stops auto-resuming and parks for a human. An `APPROVED` (or a clean re-review with zero actionable comments) ends the loop — and, as ever, **you never merge**; the human does. ### The plan comment is mechanically gated (reinforces Phase 4) diff --git a/.codex/skills/implementing-github-issues/SKILL.md b/.codex/skills/implementing-github-issues/SKILL.md index d76a71c0..98953152 100644 --- a/.codex/skills/implementing-github-issues/SKILL.md +++ b/.codex/skills/implementing-github-issues/SKILL.md @@ -519,6 +519,34 @@ Closes # **Stop here.** The skill does not merge the PR. The human reviews and merges; that's the final gate. +## Phase 11 — Addressing review feedback (when changes are requested) + +Marking the PR ready (Phase 10) is not always the end. A reviewer — a human, or an automated reviewer like CodeRabbit — may request changes. **This is a loop, not a one-shot:** you address the pass, re-request review, and wait for the next verdict. An `APPROVED` ends it (the human merges; the skill never does). Each `CHANGES_REQUESTED` pass is **one round**; a never-satisfied loop is bounded (see "Running under middle" for the round cap). + +When you pick up a review pass, run this procedure **in order** — it's the same whether you're a human-driven agent or resumed by middle, so the two behave identically: + +1. **Pull the whole pass first — batch, don't drip.** Fetch *every* open review thread before changing a line: the inline comments and the review bodies. + ```bash + gh api repos/{owner}/{repo}/pulls//comments # inline thread comments + gh pr view --json reviews --jq '.reviews[] | {author: .author.login, state, body}' + ``` + Reading the whole pass first is what lets you fix by *class* instead of comment-by-comment. + +2. **Resolve each finding class-wide, with a test per fix.** If a comment flags one instance of a bug, grep for the rest and fix them all in one change. A fix without a regression test is not a fix — the test is the evidence the finding is closed and stays closed. + +3. **Run the internal clean-eyes review loop before re-requesting.** Dispatch a review subagent (`superpowers:requesting-code-review`) over the *batched* diff — the whole pass's changes, read fresh — and loop: address what it surfaces, re-run it, until a pass surfaces nothing new. This catches the adjacent edges the reviewer's spot-checks implied but didn't enumerate, so you don't burn a whole review round on something a clean-eyes pass would have caught. + +4. **Push once.** One push for the entire pass — not one per fix. The reviewer (and any re-review bot) reacts to a single new head commit, so batch-then-push keeps the review history legible and avoids re-triggering the bot mid-edit. + +5. **Reply in-thread, then re-request review.** Reply to each addressed comment on its own thread (what changed, where), then formally re-request review so the verdict re-fires: + ```bash + gh api repos/{owner}/{repo}/pulls//comments//replies -f body="Fixed in — " + gh pr edit --add-reviewer # or the API re-request; a bot re-reviews on the new push + ``` + Then **stop and wait for the next verdict.** Don't pre-emptively mark anything resolved you haven't addressed. + +**Do not** open a new PR, rebase away the review history, or merge. The branch and PR are the same long-lasting context they've always been; you're adding commits to them. + ## Quick reference | Step | Command | @@ -628,7 +656,12 @@ Completed sub-issues stay done on the branch; only the current one is paused. On ### You may be resumed mid-workstream -When a human answers, middle re-spawns you with the answer injected into your prompt. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. +When you parked — either because you asked a question (`blocked.json`) or because you marked the PR ready and a reviewer's verdict is pending — middle ends your session to free the slot and parks the workflow on a wait signal. When the unblocking event lands on GitHub, it **re-spawns you as a fresh session in the same worktree** (same branch, same PR — never a new one), re-primed via a rewritten `.middle/prompt.md`. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. Two resume reasons: + +- **A human answered your question.** The reply is injected into your brief. Fold it into your plan / decisions log and continue. +- **A reviewer requested changes.** Your brief is an "address review" brief naming the round. Pull the PR's review threads yourself (`gh`) and follow **Phase 11 — Addressing review feedback** exactly (batch → resolve class-wide with a test per fix → internal clean-eyes review loop → push once → reply in-thread → re-request review → stop). Then the workflow re-parks for the next verdict. + +The review loop is bounded: after the repo's configured round cap (default **5**) of `CHANGES_REQUESTED` passes without an `APPROVED`, the workflow stops auto-resuming and parks for a human. An `APPROVED` (or a clean re-review with zero actionable comments) ends the loop — and, as ever, **you never merge**; the human does. ### The plan comment is mechanically gated (reinforces Phase 4) diff --git a/packages/cli/src/bootstrap-assets/skills/implementing-github-issues/SKILL.md b/packages/cli/src/bootstrap-assets/skills/implementing-github-issues/SKILL.md index d76a71c0..98953152 100644 --- a/packages/cli/src/bootstrap-assets/skills/implementing-github-issues/SKILL.md +++ b/packages/cli/src/bootstrap-assets/skills/implementing-github-issues/SKILL.md @@ -519,6 +519,34 @@ Closes # **Stop here.** The skill does not merge the PR. The human reviews and merges; that's the final gate. +## Phase 11 — Addressing review feedback (when changes are requested) + +Marking the PR ready (Phase 10) is not always the end. A reviewer — a human, or an automated reviewer like CodeRabbit — may request changes. **This is a loop, not a one-shot:** you address the pass, re-request review, and wait for the next verdict. An `APPROVED` ends it (the human merges; the skill never does). Each `CHANGES_REQUESTED` pass is **one round**; a never-satisfied loop is bounded (see "Running under middle" for the round cap). + +When you pick up a review pass, run this procedure **in order** — it's the same whether you're a human-driven agent or resumed by middle, so the two behave identically: + +1. **Pull the whole pass first — batch, don't drip.** Fetch *every* open review thread before changing a line: the inline comments and the review bodies. + ```bash + gh api repos/{owner}/{repo}/pulls//comments # inline thread comments + gh pr view --json reviews --jq '.reviews[] | {author: .author.login, state, body}' + ``` + Reading the whole pass first is what lets you fix by *class* instead of comment-by-comment. + +2. **Resolve each finding class-wide, with a test per fix.** If a comment flags one instance of a bug, grep for the rest and fix them all in one change. A fix without a regression test is not a fix — the test is the evidence the finding is closed and stays closed. + +3. **Run the internal clean-eyes review loop before re-requesting.** Dispatch a review subagent (`superpowers:requesting-code-review`) over the *batched* diff — the whole pass's changes, read fresh — and loop: address what it surfaces, re-run it, until a pass surfaces nothing new. This catches the adjacent edges the reviewer's spot-checks implied but didn't enumerate, so you don't burn a whole review round on something a clean-eyes pass would have caught. + +4. **Push once.** One push for the entire pass — not one per fix. The reviewer (and any re-review bot) reacts to a single new head commit, so batch-then-push keeps the review history legible and avoids re-triggering the bot mid-edit. + +5. **Reply in-thread, then re-request review.** Reply to each addressed comment on its own thread (what changed, where), then formally re-request review so the verdict re-fires: + ```bash + gh api repos/{owner}/{repo}/pulls//comments//replies -f body="Fixed in — " + gh pr edit --add-reviewer # or the API re-request; a bot re-reviews on the new push + ``` + Then **stop and wait for the next verdict.** Don't pre-emptively mark anything resolved you haven't addressed. + +**Do not** open a new PR, rebase away the review history, or merge. The branch and PR are the same long-lasting context they've always been; you're adding commits to them. + ## Quick reference | Step | Command | @@ -628,7 +656,12 @@ Completed sub-issues stay done on the branch; only the current one is paused. On ### You may be resumed mid-workstream -When a human answers, middle re-spawns you with the answer injected into your prompt. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. +When you parked — either because you asked a question (`blocked.json`) or because you marked the PR ready and a reviewer's verdict is pending — middle ends your session to free the slot and parks the workflow on a wait signal. When the unblocking event lands on GitHub, it **re-spawns you as a fresh session in the same worktree** (same branch, same PR — never a new one), re-primed via a rewritten `.middle/prompt.md`. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. Two resume reasons: + +- **A human answered your question.** The reply is injected into your brief. Fold it into your plan / decisions log and continue. +- **A reviewer requested changes.** Your brief is an "address review" brief naming the round. Pull the PR's review threads yourself (`gh`) and follow **Phase 11 — Addressing review feedback** exactly (batch → resolve class-wide with a test per fix → internal clean-eyes review loop → push once → reply in-thread → re-request review → stop). Then the workflow re-parks for the next verdict. + +The review loop is bounded: after the repo's configured round cap (default **5**) of `CHANGES_REQUESTED` passes without an `APPROVED`, the workflow stops auto-resuming and parks for a human. An `APPROVED` (or a clean re-review with zero actionable comments) ends the loop — and, as ever, **you never merge**; the human does. ### The plan comment is mechanically gated (reinforces Phase 4) diff --git a/packages/skills/implementing-github-issues/SKILL.md b/packages/skills/implementing-github-issues/SKILL.md index d76a71c0..98953152 100644 --- a/packages/skills/implementing-github-issues/SKILL.md +++ b/packages/skills/implementing-github-issues/SKILL.md @@ -519,6 +519,34 @@ Closes # **Stop here.** The skill does not merge the PR. The human reviews and merges; that's the final gate. +## Phase 11 — Addressing review feedback (when changes are requested) + +Marking the PR ready (Phase 10) is not always the end. A reviewer — a human, or an automated reviewer like CodeRabbit — may request changes. **This is a loop, not a one-shot:** you address the pass, re-request review, and wait for the next verdict. An `APPROVED` ends it (the human merges; the skill never does). Each `CHANGES_REQUESTED` pass is **one round**; a never-satisfied loop is bounded (see "Running under middle" for the round cap). + +When you pick up a review pass, run this procedure **in order** — it's the same whether you're a human-driven agent or resumed by middle, so the two behave identically: + +1. **Pull the whole pass first — batch, don't drip.** Fetch *every* open review thread before changing a line: the inline comments and the review bodies. + ```bash + gh api repos/{owner}/{repo}/pulls//comments # inline thread comments + gh pr view --json reviews --jq '.reviews[] | {author: .author.login, state, body}' + ``` + Reading the whole pass first is what lets you fix by *class* instead of comment-by-comment. + +2. **Resolve each finding class-wide, with a test per fix.** If a comment flags one instance of a bug, grep for the rest and fix them all in one change. A fix without a regression test is not a fix — the test is the evidence the finding is closed and stays closed. + +3. **Run the internal clean-eyes review loop before re-requesting.** Dispatch a review subagent (`superpowers:requesting-code-review`) over the *batched* diff — the whole pass's changes, read fresh — and loop: address what it surfaces, re-run it, until a pass surfaces nothing new. This catches the adjacent edges the reviewer's spot-checks implied but didn't enumerate, so you don't burn a whole review round on something a clean-eyes pass would have caught. + +4. **Push once.** One push for the entire pass — not one per fix. The reviewer (and any re-review bot) reacts to a single new head commit, so batch-then-push keeps the review history legible and avoids re-triggering the bot mid-edit. + +5. **Reply in-thread, then re-request review.** Reply to each addressed comment on its own thread (what changed, where), then formally re-request review so the verdict re-fires: + ```bash + gh api repos/{owner}/{repo}/pulls//comments//replies -f body="Fixed in — " + gh pr edit --add-reviewer # or the API re-request; a bot re-reviews on the new push + ``` + Then **stop and wait for the next verdict.** Don't pre-emptively mark anything resolved you haven't addressed. + +**Do not** open a new PR, rebase away the review history, or merge. The branch and PR are the same long-lasting context they've always been; you're adding commits to them. + ## Quick reference | Step | Command | @@ -628,7 +656,12 @@ Completed sub-issues stay done on the branch; only the current one is paused. On ### You may be resumed mid-workstream -When a human answers, middle re-spawns you with the answer injected into your prompt. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. +When you parked — either because you asked a question (`blocked.json`) or because you marked the PR ready and a reviewer's verdict is pending — middle ends your session to free the slot and parks the workflow on a wait signal. When the unblocking event lands on GitHub, it **re-spawns you as a fresh session in the same worktree** (same branch, same PR — never a new one), re-primed via a rewritten `.middle/prompt.md`. Your branch, draft PR, `plan.md`, and `decisions.md` are all intact — **continue the workstream from where it is**, don't restart. Re-read the PR's Status section and `plan.md` to orient. Two resume reasons: + +- **A human answered your question.** The reply is injected into your brief. Fold it into your plan / decisions log and continue. +- **A reviewer requested changes.** Your brief is an "address review" brief naming the round. Pull the PR's review threads yourself (`gh`) and follow **Phase 11 — Addressing review feedback** exactly (batch → resolve class-wide with a test per fix → internal clean-eyes review loop → push once → reply in-thread → re-request review → stop). Then the workflow re-parks for the next verdict. + +The review loop is bounded: after the repo's configured round cap (default **5**) of `CHANGES_REQUESTED` passes without an `APPROVED`, the workflow stops auto-resuming and parks for a human. An `APPROVED` (or a clean re-review with zero actionable comments) ends the loop — and, as ever, **you never merge**; the human does. ### The plan comment is mechanically gated (reinforces Phase 4) From c3c86a3b43ca090dcb66d60ad96c951a9a193b42 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 00:58:06 -0400 Subject: [PATCH 7/9] docs(issue-32): record the completed-vs-superseded continuation-row decision --- planning/issues/32/decisions.md | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/planning/issues/32/decisions.md b/planning/issues/32/decisions.md index 62776a84..8bf68556 100644 --- a/planning/issues/32/decisions.md +++ b/planning/issues/32/decisions.md @@ -140,3 +140,29 @@ requirement). The brief in `.middle/prompt.md` is the "address-review brief" the are pulled behind. **Evidence:** skill "Addressing review feedback" section; `prompt.ts` resume framing. + +## A handed-off continuation round terminates as `completed`, not a new `superseded` state +**File(s):** `packages/dispatcher/src/workflows/implementation.ts` +**Date:** 2026-05-24 + +**Decision:** When `resume-or-finalize` re-enqueues a continuation, the round that +handed off is marked `state = 'completed'` (the continuation becomes the Epic's +latest non-terminal row). I did **not** add a dedicated `superseded` state. + +**Why:** The handed-off row must be terminal so `findActiveWorkflowBySession` +(hook correlation) and `loadPollableWaits` ignore it — otherwise a stale round +would compete with the live continuation for the deterministic session name. +Adding a `superseded` state means modifying the `workflows.state` CHECK +constraint, which SQLite can't `ALTER` — it needs a full table rebuild (create ++ copy + drop + rename, with the `events` FK in tow). That's disproportionate +for what is, today, a cosmetic distinction: the only consumer that would tell +`completed` from `superseded` apart is the Phase 9 dashboard (out of scope), and +`markAvailableOnSuccess` firing on a handoff is *correct* (the round's drive ran +a working adapter). The honest accounting (one Epic can have several `completed` +rows, one per round) is a Phase 8/9 concern — Phase 8 routes dispatches through +the persistent engine and revisits the `workflows` row lifecycle, which is the +natural place to introduce `superseded` if the dashboard needs it. + +**Evidence:** `001_initial.sql` (state CHECK; no in-place ALTER for CHECK in +SQLite); `workflow-record.ts` `TERMINAL_STATES` / `findActiveWorkflowBySession`; +`rate-limits.ts:88` (`markAvailableOnSuccess` no-ops unless RATE_LIMITED). From 0e486303dd34e67edf7e2296dc39568cf8cc3d9b Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 01:57:36 -0400 Subject: [PATCH 8/9] fix(dispatcher): harden poller pagination, review classification, and handoff teardown Address CodeRabbit review on PR #90: - poller-gateway: slurp + flatten paginated `gh api --paginate` output so multi-page issue-comment / PR-review reads don't break JSON.parse past page 1. - poller: drop the stale standing `reviewDecision === CHANGES_REQUESTED` resume fallback. A bot reviewer never flips its standing verdict, so it would re-dispatch the agent every pass with no new feedback; only a fresh review or an explicit `changes-requested` label is a trustworthy resume signal. The 0-actionable body still wins over an un-flipped state for a fresh review. - implementation: preserve the worktree on a `waiting-human` finalize so the human can inspect / resume the in-progress state. - test: fail fast if the cap-path execution never settles; add a regression test for the stale-standing-decision fix. - plan: note the review-round cap is configurable (default 5), not fixed. --- packages/dispatcher/src/poller-gateway.ts | 44 +++++++++++++------ packages/dispatcher/src/poller.ts | 7 ++- .../src/workflows/implementation.ts | 7 ++- .../test/implementation-workflow.test.ts | 9 +++- packages/dispatcher/test/poller.test.ts | 8 ++++ planning/issues/32/plan.md | 2 +- 6 files changed, 58 insertions(+), 19 deletions(-) diff --git a/packages/dispatcher/src/poller-gateway.ts b/packages/dispatcher/src/poller-gateway.ts index 58d2f205..9f115990 100644 --- a/packages/dispatcher/src/poller-gateway.ts +++ b/packages/dispatcher/src/poller-gateway.ts @@ -25,17 +25,24 @@ function isBotLogin(login: string, type: string | undefined): boolean { export const ghPollGateway: GitHubPollGateway = { async listIssueComments(repo: string, issueNumber: number): Promise { + // `--slurp` wraps the per-page arrays into one outer array; `gh` without it + // emits one JSON array *per page*, which `JSON.parse` chokes on past page 1. const out = await gh([ "api", "--paginate", + "--slurp", `repos/${repo}/issues/${issueNumber}/comments`, ]); - const rows = JSON.parse(out) as Array<{ - id: number; - body: string; - created_at: string; - user: { login: string; type?: string } | null; - }>; + const rows = ( + JSON.parse(out) as Array< + Array<{ + id: number; + body: string; + created_at: string; + user: { login: string; type?: string } | null; + }> + > + ).flat(); return rows.map((r) => ({ id: r.id, body: r.body ?? "", @@ -77,14 +84,23 @@ export const ghPollGateway: GitHubPollGateway = { labels: Array<{ name: string }>; }; - const reviewsOut = await gh(["api", "--paginate", `repos/${repo}/pulls/${prNumber}/reviews`]); - const reviewRows = JSON.parse(reviewsOut) as Array<{ - id: number; - state: string; - body: string; - submitted_at: string | null; - user: { login: string } | null; - }>; + const reviewsOut = await gh([ + "api", + "--paginate", + "--slurp", + `repos/${repo}/pulls/${prNumber}/reviews`, + ]); + const reviewRows = ( + JSON.parse(reviewsOut) as Array< + Array<{ + id: number; + state: string; + body: string; + submitted_at: string | null; + user: { login: string } | null; + }> + > + ).flat(); const reviews: PrReview[] = reviewRows.map((r) => ({ id: r.id, state: r.state, diff --git a/packages/dispatcher/src/poller.ts b/packages/dispatcher/src/poller.ts index 5c2567c9..edd5ee4a 100644 --- a/packages/dispatcher/src/poller.ts +++ b/packages/dispatcher/src/poller.ts @@ -131,7 +131,12 @@ export function classifyReviewOutcome( if (snapshot.reviewDecision === "APPROVED") { return { outcome: "resolved", reviewId: null, decision: "APPROVED" }; } - if (snapshot.reviewDecision === "CHANGES_REQUESTED" || snapshot.labels.includes("changes-requested")) { + // Deliberately NOT a `reviewDecision === "CHANGES_REQUESTED"` fallback: a bot + // reviewer leaves the PR's standing decision at CHANGES_REQUESTED even after a + // clean re-review, so re-firing off it would re-dispatch the agent every pass + // with no new feedback (and burn a round). A fresh review (handled above) or an + // explicit human `changes-requested` label is the only trustworthy resume signal. + if (snapshot.labels.includes("changes-requested")) { return { outcome: "changes-requested", reviewId: null, decision: "CHANGES_REQUESTED" }; } return null; diff --git a/packages/dispatcher/src/workflows/implementation.ts b/packages/dispatcher/src/workflows/implementation.ts index 577cc5f8..cd6ccb77 100644 --- a/packages/dispatcher/src/workflows/implementation.ts +++ b/packages/dispatcher/src/workflows/implementation.ts @@ -656,7 +656,12 @@ export function createImplementationWorkflow( settled: DriveOutcome, ): Promise { const finalState = finalStateFor(settled); - await deps.worktree.destroyWorktree(handle); + // A `waiting-human` handoff (round cap exhausted, or nudge-exhausted mid-work) + // keeps the worktree so the human can inspect / resume the in-progress state. + // Every other terminal state frees it — the work is in the PR or abandoned. + if (finalState !== "waiting-human") { + await deps.worktree.destroyWorktree(handle); + } if (settled.kind === "rate-limited") { setRateLimited(deps.db, { adapter: ctx.input.adapter, diff --git a/packages/dispatcher/test/implementation-workflow.test.ts b/packages/dispatcher/test/implementation-workflow.test.ts index d9b9709e..b227aad4 100644 --- a/packages/dispatcher/test/implementation-workflow.test.ts +++ b/packages/dispatcher/test/implementation-workflow.test.ts @@ -476,11 +476,16 @@ describe("implementation workflow — review-round cap", () => { // definitive barrier: the bunqueue execution fully settling (the cap path // runs `resume-or-finalize` to completion, which consumes id2's armed wait). const deadline = Date.now() + 5000; + let settledState: string | undefined; while (Date.now() < deadline) { - const s = engine.getExecution(id2)?.state; - if (s === "completed" || s === "failed") break; + settledState = engine.getExecution(id2)?.state; + if (settledState === "completed" || settledState === "failed") break; await Bun.sleep(15); } + // Fail fast: if the cap path never ran resume-or-finalize to settle, the + // assertions below could still pass off the park-time `waiting-human` state + // and mask the regression. Require the execution to have actually settled. + expect(settledState === "completed" || settledState === "failed").toBe(true); // Capped: parks in waiting-human, no continuation enqueued, no armed wait // (poller stops watching), worktree preserved for the human. expect(getWorkflow(db, id2)?.state).toBe("waiting-human"); diff --git a/packages/dispatcher/test/poller.test.ts b/packages/dispatcher/test/poller.test.ts index f77e69b7..70363841 100644 --- a/packages/dispatcher/test/poller.test.ts +++ b/packages/dispatcher/test/poller.test.ts @@ -181,6 +181,14 @@ describe("classifyReviewOutcome", () => { ); expect(v).toBeNull(); }); + + test("a stale standing CHANGES_REQUESTED decision (no fresh review, no label) → null", () => { + // A bot reviewer leaves the PR's standing decision at CHANGES_REQUESTED even + // after the agent addressed it, so the standing decision alone must NOT + // re-fire a resume every pass — only a fresh review or an explicit label does. + const v = classifyReviewOutcome(prSnapshot({ reviewDecision: "CHANGES_REQUESTED" }), ARMED_AT); + expect(v).toBeNull(); + }); }); describe("runPoller — answered-question", () => { diff --git a/planning/issues/32/plan.md b/planning/issues/32/plan.md index c0f6afc7..1834f9d8 100644 --- a/planning/issues/32/plan.md +++ b/planning/issues/32/plan.md @@ -7,7 +7,7 @@ Give the `implementation` workflow a **park → external-signal → resume** spine so an agent can hand control back to a human (asked a question) or to a reviewer (PR-ready), and later resume a fresh session in the same worktree with the answer / review threads in context. -`APPROVED` ends the loop; a never-satisfied review loop is bounded to 5 rounds. +`APPROVED` ends the loop; a never-satisfied review loop is bounded by a configurable round cap (default 5). ## Approach - The Epic's 4 open sub-issues are the phases. Build down them on one branch / one PR. From fb2e47eae119891608cb889c3cea9933df3eaa61 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Sun, 24 May 2026 02:13:41 -0400 Subject: [PATCH 9/9] fix(dispatcher): tighten epic-PR match and consolidate terminal bookkeeping MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address CodeRabbit's post-approval batch on PR #90: - poller-gateway: the `in:body Closes #N` search is a prefix match, so `#3` surfaced `#30`/`#300`. Re-confirm the exact closing reference client-side with a non-digit-boundary regex over the returned PR bodies, making epic→PR resolution deterministic as the issue space grows. - implementation: drop the duplicate `setRateLimited` in `recordTerminal`; `finalize` is the single authoritative terminal handler and always runs after the pre-seed falls the `waitFor` through. - implementation: in `resumeOrFinalize`, enqueue the continuation before clearing RATE_LIMITED so a throwing enqueue leaves both the rate-limit state and the row untouched for a clean poller retry. --- packages/dispatcher/src/poller-gateway.ts | 10 ++++--- .../src/workflows/implementation.ts | 27 ++++++++----------- 2 files changed, 18 insertions(+), 19 deletions(-) diff --git a/packages/dispatcher/src/poller-gateway.ts b/packages/dispatcher/src/poller-gateway.ts index 9f115990..4cfe4840 100644 --- a/packages/dispatcher/src/poller-gateway.ts +++ b/packages/dispatcher/src/poller-gateway.ts @@ -54,6 +54,9 @@ export const ghPollGateway: GitHubPollGateway = { async findPrForEpic(repo: string, epicNumber: number): Promise { // The Epic's one PR closes the Epic — find the open PR referencing it. + // The server-side search is a prefix match, so `Closes #3` also surfaces + // `Closes #30`/`#300`; re-confirm the exact closing reference client-side on + // the returned bodies, anchoring the number with a non-digit boundary. const listOut = await gh([ "pr", "list", @@ -64,10 +67,11 @@ export const ghPollGateway: GitHubPollGateway = { "--search", `in:body Closes #${epicNumber}`, "--json", - "number", + "number,body", ]); - const prs = JSON.parse(listOut) as Array<{ number: number }>; - const prNumber = prs[0]?.number; + const closesRe = new RegExp(`\\bcloses\\s+#${epicNumber}(?!\\d)`, "i"); + const prs = JSON.parse(listOut) as Array<{ number: number; body: string | null }>; + const prNumber = prs.find((pr) => closesRe.test(pr.body ?? ""))?.number; if (prNumber === undefined) return null; const viewOut = await gh([ diff --git a/packages/dispatcher/src/workflows/implementation.ts b/packages/dispatcher/src/workflows/implementation.ts index cd6ccb77..90f78ebe 100644 --- a/packages/dispatcher/src/workflows/implementation.ts +++ b/packages/dispatcher/src/workflows/implementation.ts @@ -624,23 +624,16 @@ export function createImplementationWorkflow( } /** - * Terminal stop: record rate-limit bookkeeping and pre-seed RESUME_EVENT so - * the single top-level `waitFor` falls through without parking. The final - * `workflows.state` is set in `resume-or-finalize` (alongside worktree + * Terminal stop: pre-seed RESUME_EVENT so the single top-level `waitFor` + * falls through without parking. Rate-limit bookkeeping and the final + * `workflows.state` are set in `resume-or-finalize` (alongside worktree * teardown), so all terminal handling lives in one place. `ctx.signals` is * the live `exec.signals` (passed by reference), which is exactly what the * downstream `waitFor` reads. */ async function recordTerminal(ctx: StepContext): Promise { - const { outcome } = ctx.steps["launch-and-drive"] as DriveResult; - if (outcome.kind === "rate-limited") { - setRateLimited(deps.db, { - adapter: ctx.input.adapter, - resetAt: parseResetAt(outcome.resetAt), - source: "transcript", - detail: outcome.resetAt, - }); - } + // Rate-limit bookkeeping lives solely in `finalize` (the authoritative terminal + // handler), which always runs after this pre-seed falls the `waitFor` through. (ctx.signals as Record)[RESUME_EVENT] = { terminal: true }; } @@ -733,16 +726,18 @@ export function createImplementationWorkflow( } } - // The drive that just parked ran a working adapter; revert any stale - // RATE_LIMITED before handing off. - markAvailableOnSuccess(deps.db, ctx.input.adapter); - // Hand control to a fresh continuation that reuses this worktree. + // Hand control to a fresh continuation that reuses this worktree. Enqueue + // FIRST: if it throws, neither the rate-limit state nor the row state has + // changed, so the poller retries cleanly on its next pass. await deps.enqueueContinuation({ repo: ctx.input.repo, epicNumber: ctx.input.epicNumber, adapter: ctx.input.adapter, resume: { reason: payload.reason, round: nextRound, worktree: handle, payload }, }); + // The drive that just parked ran a working adapter; revert any stale + // RATE_LIMITED now that the hand-off is committed. + markAvailableOnSuccess(deps.db, ctx.input.adapter); // This round handed off — terminal in the bunqueue sense. The worktree is // NOT torn down; the continuation reuses it. updateWorkflow(deps.db, ctx.executionId, { state: "completed" });