diff --git a/README.md b/README.md index d0ad170..f5d02cc 100644 --- a/README.md +++ b/README.md @@ -49,7 +49,7 @@ stays resumable). See [Usage](#usage) for every flag. - [How it works](#how-it-works) — the loop, the two gates, the verify ladder - [Install](#install) · [Usage](#usage) — every flag and mode -- [Phased goals](#phased-goals---phased) · [Worktrees](#worktrees---worktree) · [Sandboxing](#sandboxing) · [Per-run spend report](#per-run-spend-report) — going further +- [Phased goals](#phased-goals---phased) · [Parallel waves](#cooperative-parallel-waves---parallel-phases-experimental) · [Worktrees](#worktrees---worktree) · [Sandboxing](#sandboxing) · [Per-run spend report](#per-run-spend-report) — going further - [Reliability](#reliability-preflight-retries-interrupts-crash-safety) — preflight, retries, Ctrl-C, crash-safety - [Operator control](#watching-steering--extending-a-run-operator-control) — watch live, steer with `--note`, extend caps on `--resume` - [Web UI](#web-ui-goaly-ui) — `goaly ui`: runs, live feeds, and worktrees in the browser @@ -190,7 +190,10 @@ PHASE 2 · the loop (🔁 ≤ --max-iterations, default 10; bails early on STUCK 0) and can't win. Each candidate completes write-ahead, so `--resume` re-runs only the not-yet-logged ones and re-selects deterministically — or, with `--resume-best-of-incomplete collapse`, collapses to the best already-logged candidate and re-runs nothing. Needs a committed HEAD (it refuses to start fail-closed - on an unborn branch). See [Best-of-N parallel worker](#best-of-n-parallel-worker---candidates). + on an unborn branch). You can also request it in **natural language** — *"fix the flaky test, work with 4 + subagents"* in the goal, or *"try 4 parallel attempts"* in a `--resume` note — a deterministic directive + grammar (never an LLM parse) maps it onto `--candidates`, strips the clause from the frozen goal, and + logs the interpretation loudly. See [Best-of-N parallel worker](#best-of-n-parallel-worker---candidates). - **Compile is resilient, not one-shot.** A `COMPILE_FAILED` (a correctable authoring mistake — bad path, transient parse miss) re-authors the verification with the error fed back as guidance, up to `--max-compile-retries` (default 2; `0` disables), before the run fails — so one bad compile output @@ -304,6 +307,53 @@ ACCEPT (a cumulative contract on the ORIGINAL goal) ──both keys──► DO every phase). `--resume` re-enters mid-plan without repeating completed phases. `goaly runs show` prints the frozen plan and stamps each iteration with its phase. +### Cooperative parallel waves (`--parallel-phases`, EXPERIMENTAL) + +Sequential phases leave wall-clock on the table when sub-goals are **independent**. `--parallel-phases` +(opt-in, experimental) runs them as **cooperating agents**: consecutive plan phases sharing a `group` +value form a **wave** that executes concurrently, then **merges** — without weakening a single +guarantee. + +```jsonc +// plan.json — phases 1+2 are one wave; phase 3 runs after the merged result +{ "phases": [ + { "goal": "implement the parser", "group": 1 }, + { "goal": "implement the formatter", "group": 1 }, + { "goal": "wire parser + formatter into the CLI" } +] } +``` +```bash +goaly "build the tool" --verify-cmd "npm test" --phased --parallel-phases --autonomous \ + --plan-file plan.json +``` + +How a wave works, and why it can't cheat: + +- **Fork: every wave member is a full goaly run.** Each phase gets an isolated git worktree off the + wave-start checkpoint and runs as its own **child run** — its own compiled + frozen contract, its own + iterations, verifier ladder, and veto-only Sign-off (both keys per child), its own write-ahead log + inside the worktree — all children metered by the **one shared `--budget-tokens`** cap. +- **Merge: plumbing, not prayer.** DONE children merge in phase order with a real 3-way + `git merge-tree` against the fork point — objects only, no commits, no HEAD/branch/index movement. A + **textual conflict applies nothing** of that child. +- **Re-verify: a merge is never trusted.** After promotion, each merged child's **frozen deterministic + rungs re-run on the combined tree** — two individually-green changes can still break each other. A + red re-verify un-lands nothing silently: that phase **downgrades to the classic sequential run** on + the merged tree, under a fresh frozen contract for the same sub-goal (the bar never moves, only the + starting tree). Merge conflicts and children that never reach DONE downgrade the same way. +- **Acceptance still gates the whole.** The final cumulative acceptance contract (both keys, LLM + included) runs on the original goal exactly as in a sequential phased run — so no decomposition, + parallel or not, can green a goal whose parts pass but whole doesn't. +- **Fail-closed to sequential.** Every failure shape — a conflicted merge, a red re-verify, a crashed + child, even a missing/broken wave executor — degrades to phases running one at a time, which is + byte-for-byte today's `--phased`. The reducer stays pure: it emits one `RUN_WAVE` command and folds + one `WAVE_RAN` event; children never enter the parent's state machine. +- **Experimental limits (v1):** requires `--autonomous` (children seal their frozen contracts + concurrently — still frozen + loudly logged) and a `--plan-file` with `group` fields (the LLM planner + does not author groups yet); a crash mid-wave re-runs the whole wave on `--resume` (children live in + ephemeral worktrees); wave-child spend is reported under the parent's `harness` layer. Grouped plans + run **strictly sequentially** without the flag, and the plan's grouping is frozen into `planHash`. + ## Best-of-N parallel worker (`--candidates`) Some iterations are a coin-flip — one attempt half-finishes, another nails it. `--candidates N` (alias @@ -368,6 +418,33 @@ each iteration, with --candidates N: HEAD** — `git worktree` can't check out an unborn tree, so a `--candidates > 1` run on a HEAD-less repo **refuses to start** (fail-closed) with a clear message; make an initial commit or use `--candidates 1`. +### Natural-language delegation — just say it + +You don't have to remember the flag: a **delegation directive in the goal** maps onto the same +tournament, and mid-run the same grammar reads your **resume note**. + +```bash +goaly "fix the flaky auth test, work with 4 subagents" # ⇒ --candidates 4 +goaly "make the linter pass using 3 parallel attempts" # ⇒ --candidates 3 +goaly "port the parser to TS, use subagents" # ⇒ --candidates 3 (documented default) +goaly --resume run-… --note "focus on the parser, try 4 parallel attempts" # raises the fan-out mid-run +``` + +- **Deterministic, never an LLM parse.** Detection is a small directive grammar (`src/cli/delegation.ts`) + — an LLM interpreting your config would be exactly the "LLM in control flow" goaly exists to avoid. The + grammar is deliberately narrow: only **`subagents`** (with a delegation verb — `use / spawn / work + with / delegate to …`) and **`N parallel attempts|candidates|tries`** trigger it, so goals about + app-domain parallelism (*"implement a job queue with 4 parallel workers"*, *"handle 5 parallel login + attempts"*) never match. No match ⇒ the classic single attempt — fail-closed, never a guess. +- **The directive is stripped from the goal** before the contract is compiled — the goal is frozen and + read by the judge/approver, and a leftover *"use 4 subagents"* would become an unverifiable success + criterion. The interpretation is **loudly logged** (`phrase → N`), and the explicit `--candidates` / + `--best-of` flag (or config file) **always wins**. +- **Mid-run steering rides ADR 0012.** A directive in `--resume … --note "…"` becomes a `candidates` + overlay on the `RUN_EXTENDED` marker (an operational knob, like a raised `--max-iterations` — the + frozen contract stays unreachable); the directive clause leaves the note and any remaining text still + steers the worker. Same 16 cap, same fail-closed error above it. + ## Worktrees (`--worktree`) Sometimes the run shouldn't touch your working tree at all — you keep editing while goaly builds, or @@ -1179,14 +1256,18 @@ goaly --resume run- --max-iterations 25 # revive FAILED at the i goaly --resume run- --budget-tokens 900000 # revive a budget abort (prior spend still counts) goaly --resume run- --stuck-no-diff false --note "try editing src/parser.ts directly" # revive a stuck abort, with direction +goaly --resume run- --candidates 4 # widen the best-of-N fan-out for what's left +goaly --resume run- --note "try 4 parallel attempts" # same, said in natural language ``` Only the operational knobs are extendable (`--max-iterations`, `--budget-tokens`, -`--budget-wall-ms`, the `--stuck-*` thresholds) — the goal, verifier, and rubric are structurally -not part of an extension, so autonomy never becomes "renegotiate the bar." A DONE run refuses to -extend and points you at `--from-run`. Rule of thumb: **same goal, more room → `--resume` with -caps/note; new or refined goal → `--from-run`** (a fresh contract, compiled aware of what just -happened — see [Following up](#following-up-after-a-run-ends---from-run)). +`--budget-wall-ms`, the `--stuck-*` thresholds, `--candidates`) — the goal, verifier, and rubric are +structurally not part of an extension, so autonomy never becomes "renegotiate the bar." A DONE run +refuses to extend and points you at `--from-run`. A resume also **continues the run's own harness** +(recorded in the log) rather than silently switching to the default — session ids are +harness-specific; pass `--harness` explicitly to override. Rule of thumb: **same goal, more room → +`--resume` with caps/note; new or refined goal → `--from-run`** (a fresh contract, compiled aware of +what just happened — see [Following up](#following-up-after-a-run-ends---from-run)). ### Inspecting past runs diff --git a/docs/adr/0017-cooperative-parallel-waves.md b/docs/adr/0017-cooperative-parallel-waves.md new file mode 100644 index 0000000..e121866 --- /dev/null +++ b/docs/adr/0017-cooperative-parallel-waves.md @@ -0,0 +1,70 @@ +# ADR 0017 — cooperative parallel waves (`--parallel-phases`, experimental) + +## Status +Accepted (experimental, opt-in). + +## Context + +goaly had exactly one parallelism: best-of-N (`--candidates`, issue #85) — K **competing** attempts +at the SAME sub-goal, keep one, discard the rest. What it lacked was **cooperation**: running +*different, independent* sub-goals concurrently and **combining** their work. Phased decomposition +(issue #48) already produces exactly the right unit — a frozen plan of sub-goals, each executed as +its own frozen, two-key contract, finished by a cumulative acceptance contract on the original goal +— but executes strictly sequentially, leaving wall-clock on the table whenever phases touch +disjoint parts of the tree. The plan schema explicitly deferred this ("no DAG/parallelism in v1"). + +The tension: combining trees is where fail-closed guarantees usually die. A merge can conflict +(textually) or lie (two clean merges that break each other semantically), and any "merge agent" +that resolves conflicts with an LLM would put unverified writes on the DONE path. + +## Decision + +**Waves of child runs + merge-and-reverify, fail-closed to sequential.** + +1. **The plan carries the grouping, frozen.** `SubGoal` gains an optional `group`; CONSECUTIVE + phases sharing a group form a wave. The grouping is canonicalized into `planHash` (groupless + plans keep their legacy hash byte-for-byte), so no transition can re-shuffle it. v1 sources + groups from `--plan-file` only. +2. **The reducer sees one command, one event.** When the current phase heads a not-yet-attempted + group and `config.parallelPhases` is on, `startPhaseCompile` emits ONE `RUN_WAVE` (per-phase + configs derived exactly as for sequential phases) and folds ONE `WAVE_RAN` — still exactly one + command per state, still pure. `PhaseCtx` gains `skip` (indices completed by a wave) and `waved` + (indices already attempted — a group can never re-fan-out), both optional so classic runs are + untouched. +3. **Every wave member is a full goaly run.** The Driver-side `WaveRunner` gives each phase an + ephemeral worktree off the wave-start checkpoint and an embedded `drive()` — its own compiled + + frozen contract, iterations, ladder, veto-only Sign-off, and write-ahead log inside the + worktree — all children on the PARENT's budget meter and interrupt probe. Worktree creation is + sequential (git lock contention); only the runs are concurrent. +4. **Merge is plumbing; the merged tree is re-verified.** DONE children merge in phase order via + `git merge-tree --write-tree --merge-base=` (objects only; a conflicted merge + applies nothing). After promotion, each merged child's frozen DETERMINISTIC rungs re-run on the + combined tree. Judge rungs are not re-run here: each child already turned both keys in + isolation, and the final acceptance contract still gates the whole run — the merged-tree guard + is the ungameable deterministic bar in between. +5. **Every failure downgrades, nothing greens.** A merge conflict, a red re-verify, a child that + never reaches DONE, a thrown wave runner, or a missing wave seam all resolve to `unmerged`, + which the reducer turns into the CLASSIC sequential phase on the merged-so-far tree — a fresh + frozen contract for the same sub-goal. The worst case of the feature is exactly today's + `--phased`. + +## Consequences + +- **Invariants hold.** #1: the fan-out is Driver-side data flow (`RUN_WAVE`/`WAVE_RAN`); children + are separate pure folds over separate logs. #2: grouping frozen in `planHash`; child contracts + frozen at their own Seals. #3: two keys per child AND at acceptance. #4: a merge is never + trusted; every failure shape is a typed downgrade. #7: `WAVE_RAN` carries the post-merge + checkpoint tree (replay re-points the baseline like `PHASE_ADVANCED`). +- **Cost profile.** ~1× the sequential token cost (+ deterministic re-verification + any conflict + re-runs) for wall-clock ≈ slowest child + merge — the complement of best-of-N, which buys + quality at ~N× cost for one phase's work. +- **Experimental limits (v1), by design:** requires `--autonomous` (children seal concurrently; + an interactive gate cannot pause K children at once); a crash mid-wave re-runs the whole wave on + `--resume` (children live in ephemeral worktrees; their logs die with them); wave-child spend is + bucketed under the parent's `harness` usage layer (totals and the budget cap stay exact); + compiler-authored (git-excluded) verification files are copied from each merged child's worktree + so their frozen commands keep their inputs — colliding authored paths across children surface as + a red re-verify, never a silent overwrite that greens. +- **Deferred:** planner-authored groups (and the natural-language "split this across N subagents" + directive mapping onto them), durable child logs for fine-grained wave resume, LLM-judge re-runs + on the merged tree. diff --git a/docs/index.html b/docs/index.html index 6b33c3b..151cde1 100644 --- a/docs/index.html +++ b/docs/index.html @@ -150,6 +150,30 @@

Best-of-N parallel worker

above 16 is a fail-closed error). Composes with --phased / --delta-verify / --sandbox; needs a committed HEAD.

+

Or just say it: a natural-language directive in the goal — "fix the flaky test, + work with 4 subagents", "… using 3 parallel attempts" — maps onto + --candidates via a small deterministic grammar (never an LLM parse), and the + clause is stripped from the goal so it can't enter the frozen contract; the + interpretation is loudly logged and the explicit flag always wins. Mid-run, the same grammar + reads a resume note: --resume <id> --note "try 4 parallel attempts" raises + the fan-out as a RUN_EXTENDED candidates overlay (operator control, ADR 0012). + Domain goals like "a queue with 4 parallel workers" never match — no directive means the + classic single attempt.

+ +
+

Cooperative parallel waves (experimental)

+

--parallel-phases (opt-in, with --phased --autonomous) turns + consecutive plan phases sharing a "group" into a concurrent wave of + cooperating child runs — each phase a FULL goaly run (own frozen contract, iterations, + two-key gate, write-ahead log) in an isolated worktree, on the one shared + --budget-tokens. DONE children then merge in phase order via real 3-way + git merge-tree plumbing (no commits), and each merged phase's frozen + deterministic rungs re-run on the combined tree — a merge is never trusted. A + conflict, a red re-verify, a failed child, or even a missing wave executor all + downgrade fail-closed to the classic sequential phase; the cumulative ACCEPTANCE + contract still gates the whole run, so parallel decomposition can't green a broken whole. + The reducer stays pure: one RUN_WAVE command out, one WAVE_RAN + event back — children never enter the parent's state machine.

Named worktrees

@@ -234,8 +258,10 @@

Watch it & steer it

never disturbs the run). Ctrl-C stops cleanly between steps, then --resume <id> --note "…" hands the worker your guidance on its next turn. A run that hit an operational limit isn't a dead end — --resume with - --max-iterations / --budget-tokens / --stuck-* - continues it in place, auditable in the log. The extension schema simply has no field for + --max-iterations / --budget-tokens / --stuck-* / + --candidates continues it in place, auditable in the log — and it continues the + run's own recorded harness (session ids are harness-specific; --harness + overrides explicitly). The extension schema simply has no field for the goal or verifier: you can add room, never lower the frozen bar. ADR 0012 →

diff --git a/src/agent-cli/codec.test.ts b/src/agent-cli/codec.test.ts index afb9d63..772d570 100644 --- a/src/agent-cli/codec.test.ts +++ b/src/agent-cli/codec.test.ts @@ -176,6 +176,55 @@ describe('runCodecHarness', () => { expect(result.sessionId).toBe('s-2'); }); + it('does NOT resume a FOREIGN harness sentinel (a resumed run that switched harness)', async () => { + // Regression: the guard only knew THIS codec's sentinel, so a run resumed under a different + // --harness threaded the prior harness's sentinel into `claude --resume noop-session`, crashing + // every candidate/turn. Every entry in the shared sentinel skip-list must start fresh instead. + const captured = { args: [] as string[][] }; + const exec: AgentExecFn = async () => ({ + stdout: JSON.stringify({ result: 'ok', session_id: 's-3', usage: { total_tokens: 3 } }), + stderr: '', + code: 0, + }); + const result = await runCodecHarness(spyCodec(captured), exec, undefined, 'go', sid('noop-session')); + expect(captured.args[0]).not.toContain('--resume'); + expect(captured.args[0]).not.toContain('noop-session'); + expect(result.status).toBe('completed'); + expect(result.sessionId).toBe('s-3'); + }); + + it('never trusts the AMBIENT session id (goaly nested under Claude Code)', async () => { + // Regression: a nested `claude -p` adopts and reports the ambient CLAUDE_CODE_SESSION_ID instead + // of minting its own — the LLM provider already refuses it, but the HARNESS recorded it as the + // worker's session, so iteration 2 (or --resume / --inherit-session) would resume the OUTER + // conversation into the worker. The harness must scrub it on both sides of the seam. + const prev = process.env['CLAUDE_CODE_SESSION_ID']; + process.env['CLAUDE_CODE_SESSION_ID'] = 'ambient-1'; + try { + const captured = { args: [] as string[][] }; + const exec: AgentExecFn = async () => ({ + // The wrapped CLI reports the ambient id instead of a fresh session. + stdout: JSON.stringify({ result: 'ok', session_id: 'ambient-1', usage: { total_tokens: 3 } }), + stderr: '', + code: 0, + }); + + // The ambient id is never SURFACED: it becomes the codec's unknown-session sentinel, which + // every downstream consumer (resume hint, --inherit-session, next-turn resume) already skips. + const first = await runCodecHarness(spyCodec(captured), exec, undefined, 'go'); + expect(first.status).toBe('completed'); + expect(first.sessionId).toBe(claudeCodec.unknownSession); + + // …and never THREADED into --resume when a prior turn recorded it anyway. + await runCodecHarness(spyCodec(captured), exec, undefined, 'go', sid('ambient-1')); + expect(captured.args[1]).not.toContain('--resume'); + expect(captured.args[1]).not.toContain('ambient-1'); + } finally { + if (prev === undefined) delete process.env['CLAUDE_CODE_SESSION_ID']; + else process.env['CLAUDE_CODE_SESSION_ID'] = prev; + } + }); + it('never throws when the exec seam itself rejects — fails closed to crashed', async () => { const exec: AgentExecFn = async () => { throw new Error('spawn ENOENT'); diff --git a/src/agent-cli/codec.ts b/src/agent-cli/codec.ts index 990159f..5237a5c 100644 --- a/src/agent-cli/codec.ts +++ b/src/agent-cli/codec.ts @@ -14,7 +14,7 @@ * provider and the composition root import the codec from this neutral `agent-cli/` layer. */ -import { SessionId, coerceSessionId } from '../domain/ids'; +import { SessionId, coerceSessionId, isSentinelSession } from '../domain/ids'; import { HarnessRunResult } from '../domain/events'; import { runProcess } from '../util/spawn'; import { parseAgentOutput, type AgentOutput, type FieldExtractor } from './output'; @@ -176,6 +176,22 @@ export function defaultAgentExec( }; } +/** + * The AMBIENT session id when goaly itself runs nested under Claude Code (e.g. inside a Claude Code + * remote environment). A spawned `claude -p` there adopts and REPORTS this id instead of minting a + * fresh per-call session, and every call in a cwd appends to that ONE shared session file — so + * resuming it would replay the OUTER conversation's turns (and every sibling goaly LLM step) into + * the worker's context, not the worker's own working memory. Observed empirically; scrubbing the + * variable from the child env does NOT stop the pinning (the wrapped CLI keeps it), so the only + * safe policy is to never TRUST the ambient id: treat it exactly like the codec's unknown-session + * sentinel — never surface it as a resumable session and never thread it into `--resume`. Shared by + * the harness core (below) and the read-only {@link ../llm/agent-cli-provider!AgentCliLlmProvider}. + */ +export function ambientSessionId(): string | undefined { + const v = process.env['CLAUDE_CODE_SESSION_ID']; + return v !== undefined && v.length > 0 ? v : undefined; +} + /** * The one harness `run()` body, parameterised by a codec. Builds the optional stream tap (and the * issue-#24 token estimator), asks the codec for the write-mode argv, runs the injected `exec` @@ -209,7 +225,17 @@ export async function runCodecHarness( // continuation turn, turning one slow/timed-out turn into a dead run — a false STUCK_HARNESS_CRASH. // Drop it so the next turn starts a FRESH session instead; the worker loses that turn's chat memory // but keeps making real progress against the frozen contract (which alone governs DONE). - const resumeId = sessionId === codec.unknownSession ? undefined : sessionId; + // EVERY sentinel is refused, not just this codec's own: a run resumed under a different --harness + // carries the PRIOR harness's sentinel (e.g. the fake harness's `noop-session`), and + // `claude --resume noop-session` crashes every turn the same way. The AMBIENT id (goaly nested + // under Claude Code) is refused too: resuming it would pull the OUTER conversation — and every + // sibling LLM step sharing that session file — into the worker (see {@link ambientSessionId}). + const resumeId = + sessionId === codec.unknownSession || + (sessionId !== undefined && isSentinelSession(sessionId)) || + sessionId === ambientSessionId() + ? undefined + : sessionId; const args = codec.harnessArgs({ prompt, model, @@ -231,7 +257,7 @@ export async function runCodecHarness( } tap?.end(); // flush a final unterminated JSONL line before classification - return codec.classify({ + const classified = codec.classify({ stdout: result.stdout, stderr: result.stderr, code: result.code, @@ -239,6 +265,14 @@ export async function runCodecHarness( ...(resumeId !== undefined ? { sessionId: resumeId } : {}), ...(estimator !== undefined ? { estimator } : {}), }); + // Never SURFACE the ambient id either: a nested CLI reports it as its session_id, and anything + // downstream that stores it (the run log, the resume hint, --inherit-session) would later resume + // the outer conversation. Coerce it to the codec's unknown-session sentinel — "no resumable + // session" — which every consumer already skips. + if (classified.sessionId === ambientSessionId()) { + return { ...classified, sessionId: coerceSessionId(undefined, codec.unknownSession) }; + } + return classified; } /** diff --git a/src/cli/args.test.ts b/src/cli/args.test.ts index a5d1c7f..4a068a8 100644 --- a/src/cli/args.test.ts +++ b/src/cli/args.test.ts @@ -140,6 +140,101 @@ describe('parseArgs', () => { }); }); + describe('natural-language parallel delegation (goal directive → best-of-N)', () => { + it('maps "work with N subagents" in the goal to candidates and STRIPS the clause', async () => { + const a = await parseArgs([ + 'run', '--goal', 'fix the flaky test, work with 4 subagents', '--verify-cmd', 'true', + ]); + expect(a.config.candidates).toBe(4); + expect(a.config.goal).toBe('fix the flaky test'); // the directive never enters the contract + expect(a.delegation).toEqual({ + candidates: 4, + phrase: expect.stringContaining('work with 4 subagents'), + overriddenByFlag: false, + }); + }); + + it('the explicit --candidates flag wins over the directive (still stripped + reported)', async () => { + const a = await parseArgs([ + 'run', '--goal', 'fix it using 4 subagents', '--verify-cmd', 'true', '--candidates', '2', + ]); + expect(a.config.candidates).toBe(2); + expect(a.config.goal).toBe('fix it'); + expect(a.delegation?.overriddenByFlag).toBe(true); + }); + + it('a directive above the candidate cap fails closed like the flag', async () => { + await expect( + parseArgs(['run', '--goal', 'fix it with 20 subagents', '--verify-cmd', 'true']), + ).rejects.toThrow(/at most 16/); + }); + + it('a goal that is ONLY a directive fails closed (no goal left to freeze)', async () => { + await expect( + parseArgs(['run', '--goal', 'use 4 subagents', '--verify-cmd', 'true']), + ).rejects.toThrow(/only a delegation directive/); + }); + + it('an application-domain goal about parallelism does NOT trigger delegation', async () => { + const a = await parseArgs([ + 'run', '--goal', 'implement a job queue with 4 parallel workers', '--verify-cmd', 'true', + ]); + expect(a.config.candidates).toBe(1); + expect(a.config.goal).toBe('implement a job queue with 4 parallel workers'); + expect(a.delegation).toBeUndefined(); + }); + + it('a note directive at resume becomes a candidates extension and the clause leaves the note', async () => { + const a = await parseArgs([ + 'run', '--resume', 'run-abc', '--note', 'focus on the parser, try 4 parallel attempts', + ]); + expect(a.resumeExtend?.candidates).toBe(4); + expect(a.resumeExtend?.note).toBe('focus on the parser'); + expect(a.delegation?.candidates).toBe(4); + }); + + it('a note that is ONLY a directive still extends candidates, with no note left', async () => { + const a = await parseArgs(['run', '--resume', 'run-abc', '--note', 'use 3 subagents']); + expect(a.resumeExtend?.candidates).toBe(3); + expect(a.resumeExtend?.note).toBeUndefined(); + }); + + it('an explicit --candidates at resume rides the extension (flag wins over the note)', async () => { + const a = await parseArgs([ + 'run', '--resume', 'run-abc', '--candidates', '2', '--note', 'try 5 parallel attempts', + ]); + expect(a.resumeExtend?.candidates).toBe(2); + expect(a.delegation?.overriddenByFlag).toBe(true); + }); + }); + + describe('--parallel-phases (EXPERIMENTAL cooperative waves)', () => { + it('parses with --phased --autonomous', async () => { + const a = await parseArgs([ + 'run', '--goal', 'g', '--verify-cmd', 'true', '--phased', '--autonomous', '--parallel-phases', + ]); + expect(a.config.parallelPhases).toBe(true); + expect(a.config.phased).toBe(true); + }); + + it('defaults OFF (grouped plans run sequentially without the flag)', async () => { + const a = await parseArgs(['run', '--goal', 'g', '--verify-cmd', 'true', '--phased', '--autonomous']); + expect(a.config.parallelPhases).toBe(false); + }); + + it('rejects --parallel-phases without --phased (fail-closed)', async () => { + await expect( + parseArgs(['run', '--goal', 'g', '--verify-cmd', 'true', '--autonomous', '--parallel-phases']), + ).rejects.toThrow(/--phased/); + }); + + it('rejects --parallel-phases without --autonomous (children seal concurrently)', async () => { + await expect( + parseArgs(['run', '--goal', 'g', '--verify-cmd', 'true', '--phased', '--parallel-phases']), + ).rejects.toThrow(/--autonomous/); + }); + }); + describe('--resume-best-of-incomplete (issue #85 follow-up)', () => { it('defaults to rerun when absent (byte-for-byte the historical behavior)', async () => { const a = await parseArgs(['run', '--goal', 'g', '--verify-cmd', 'true']); diff --git a/src/cli/args.ts b/src/cli/args.ts index 90cc2f0..5c75e8e 100644 --- a/src/cli/args.ts +++ b/src/cli/args.ts @@ -6,6 +6,7 @@ import { LogLevel } from '../log/logger'; import { ModelSelection, type ModelSelectionInput } from './models'; import { resolveInputSources, defaultReaders, type InputReaders } from './input-sources'; import { loadConfig, type LoadedConfig } from './config-file'; +import { parseDelegationDirective } from './delegation'; import type { AgentCli } from '../agent-cli/registry'; import type { WorktreeCommand } from './worktree-cmd'; import { WorktreeName } from '../workspace/worktree-manager'; @@ -16,6 +17,21 @@ import { WorktreeName } from '../workspace/worktree-manager'; */ export type HarnessChoice = AgentCli | 'fake' | 'goaly-code'; +/** All valid harness choices — one source of truth for the flag parser and the resume adoption. */ +export const HARNESS_CHOICES: readonly HarnessChoice[] = [ + 'claude', + 'codex', + 'droid', + 'pi', + 'fake', + 'goaly-code', +]; + +/** Whether a stored/untrusted string names a known harness. */ +export function isHarnessChoice(value: string): value is HarnessChoice { + return (HARNESS_CHOICES as readonly string[]).includes(value); +} + /** * Which provider runs the LLM workflow steps (judge / approver / compiler). Any bundled CLI, plus * `openai` — a direct OpenAI-compatible chat-completions endpoint (no coding CLI installed). @@ -79,6 +95,14 @@ export type ParsedArgs = { worktreeRun: string | true | undefined; config: RunConfig; harness: HarnessChoice; + /** + * Whether `harness` came from an EXPLICIT `--harness` CLI flag this invocation (never the + * config-file overlay — same explicitness rule as the resume extension). A `--resume` without it + * ADOPTS the resumed run's recorded harness instead of silently switching to the default: session + * ids are harness-specific, and continuing a fake/codex run under `claude` mid-run is never what + * the user meant. + */ + harnessExplicit: boolean; models: ModelSelection; llmProvider: LlmProviderChoice; workspace: string; @@ -163,6 +187,14 @@ export type ParsedArgs = { * run or a plain resume. Operational knobs only — never the goal / verifier / rubric. */ resumeExtend: RunExtension | undefined; + /** + * Natural-language parallel delegation, when a directive in the GOAL text was mapped onto the + * best-of-N tournament ("work with 4 subagents" ⇒ `candidates: 4` — see `delegation.ts`). Carried + * so the CLI can log the interpretation loudly (the matched phrase and the count). The directive + * clause is already stripped from `config.goal`; an explicit `--candidates` always wins (this is + * then still set, with `overriddenByFlag: true`, so the log can say so). Undefined ⇒ no directive. + */ + delegation: { candidates: number; phrase: string; overriddenByFlag: boolean } | undefined; }; export const USAGE = `goaly — run a coding agent until a frozen success contract is met. @@ -180,7 +212,7 @@ Usage: [--install-missing-tools true|false] [--rubric ""] [--autonomous] [--max-iterations N] [--candidates N] [--phased [--max-phases N] [--max-plan-revisions N] [--plan-file

] - [--planner-model ]] + [--planner-model ] [--parallel-phases]] [--max-seal-revisions N] [--max-compile-retries N] [--verify-dir

] [--budget-tokens N] [--budget-wall-ms N] [--diff-ignore ""] [--stuck-no-diff true|false] [--stuck-repeat-threshold N] @@ -319,6 +351,19 @@ Phased decomposition (issue #48 — split one big goal into a frozen plan of sma --max-plan-revisions N cap the free-text plan-Seal revise rounds (default 10; 0 disables revision). --planner-model model for the planner step only (cascades like the other LLM-step models). --autonomous also auto-accepts the plan AND each phase contract — still frozen + logged loudly. + --parallel-phases EXPERIMENTAL, opt-in — cooperative parallel WAVES: consecutive plan phases + sharing a "group" value (plan-file: {"goal": …, "group": 1}) execute + CONCURRENTLY, each as its own frozen, two-key CHILD goaly run in an isolated + git worktree on the SHARED --budget-tokens meter. The children are then merged + in phase order (3-way git merge-tree — plumbing only, no commits) and each + merged phase's frozen DETERMINISTIC rungs are RE-VERIFIED on the combined tree + — a merge is never trusted. Fail-closed everywhere: a merge conflict, a red + re-verify, or a child that can't reach DONE simply DOWNGRADES that phase to + the classic sequential run on the merged tree (the bar never moves, only the + starting tree); the cumulative ACCEPTANCE contract still gates the whole run. + Requires --phased and --autonomous (children seal concurrently). Without this + flag, grouped plans run strictly sequentially. Resume note: a crash mid-wave + re-runs the WHOLE wave on --resume (children live in ephemeral worktrees). Best-of-N parallel worker (issue #85 — tournament-select candidates against the frozen ladder): --candidates N (alias --best-of N) run N independent worker attempts EACH loop iteration in @@ -338,6 +383,18 @@ Best-of-N parallel worker (issue #85 — tournament-select candidates against th the N execs goes through the same jail). Needs a committed HEAD: on a repo with no resolvable HEAD (unborn branch) a --candidates > 1 run refuses to start (fail-closed) — make an initial commit or run with --candidates 1. + Natural-language delegation: you can also just SAY it — a delegation directive in the goal + ("fix the flaky test, work with 4 subagents", "… using 3 parallel attempts", + "use subagents" ⇒ 3) maps onto --candidates and the directive clause is + STRIPPED from the goal (it must never enter the frozen contract), loudly + logged as phrase → N. Detection is a small DETERMINISTIC grammar, never an + LLM parse, and deliberately narrow: only "subagents"/"parallel attempts| + candidates|tries" introduced by a delegation verb trigger it — goals about + app-domain parallelism ("a queue with 4 parallel workers") never match. The + explicit --candidates/--best-of always wins. Mid-run, the same grammar reads + your resume note: goaly --resume --note "try 4 parallel attempts" raises + the fan-out as a RUN_EXTENDED candidates overlay (ADR 0012) and keeps any + remaining note text as worker guidance. Compile resilience (issue #51): --max-compile-retries N on a COMPILE_FAILED, re-author the verification with the error as @@ -540,8 +597,11 @@ Plain-language run narration (opt-in observability — issue #8): Resume, steer & extend (operator control over ONE run — the frozen contract never changes): --resume re-enter an INCOMPLETE run's loop exactly where the write-ahead log left it - (crash, Ctrl-C, kill — nothing completed is repeated). Pass any of the flags - below WITH --resume to extend/steer the resumed run; each is recorded in the + (crash, Ctrl-C, kill — nothing completed is repeated). The resumed run + CONTINUES ITS OWN harness (recorded in the run log) — session ids are + harness-specific, so it is never silently switched to the default; pass + --harness explicitly to override. Pass any of the flags below WITH --resume + to extend/steer the resumed run; each is recorded in the log (a RUN_EXTENDED marker) so the extension is auditable and later resumes keep it. Only these OPERATIONAL knobs are extendable — never the goal, the verifier, or the rubric (the contract stays frozen; both keys still gate DONE): @@ -549,10 +609,14 @@ Resume, steer & extend (operator control over ONE run — the frozen contract ne --budget-tokens N also revives a budget-ABORTED run (spend re-judged --budget-wall-ms N against the new cap; prior spend still counts) --stuck-* flags raise/toggle a tripped stuck detector to continue + --candidates N raise/lower the best-of-N fan-out for the remaining + iterations (also via a --note directive, see above) Live in another terminal: goaly runs watch . --note "" (with --resume) operator guidance appended to the NEXT agent prompt — steer the worker without touching the bar. Combine with Ctrl-C for mid-run steering: - interrupt, then 'goaly --resume --note "try the other approach"'. + interrupt, then 'goaly --resume --note "try the other approach"'. A + delegation directive in the note ("try 4 parallel attempts") is lifted out + into a --candidates extension (see natural-language delegation above). Follow-up after a run ends (build on a finished run — keeps every invariant by construction): --from-run start a NEW run whose contract is authored AWARE of a finished run: a concise, @@ -738,9 +802,12 @@ function boolFlag(flags: RawFlags, key: string): boolean | undefined { * effective config says. `--note` is resume-only: on a fresh run there is no next-turn boundary to * attach it to, so it fails closed with the fix. */ -function collectResumeExtension(flags: RawFlags, config: RunConfig): RunExtension | undefined { +function collectResumeExtension( + flags: RawFlags, + config: RunConfig, +): { extension: RunExtension | undefined; delegation: ParsedArgs['delegation'] } { const resuming = str(flags, 'resume') !== undefined; - const note = str(flags, 'note'); + let note = str(flags, 'note'); if (!resuming) { if (note !== undefined) { throw new UsageError( @@ -748,9 +815,32 @@ function collectResumeExtension(flags: RawFlags, config: RunConfig): RunExtensio '--resume . To guide a fresh run, put the guidance in the goal or --intent.', ); } - return undefined; + return { extension: undefined, delegation: undefined }; } const has = (key: string): boolean => flags[key] !== undefined; + // Natural-language delegation in a resume note ("try 4 parallel attempts"): the steering intent + // is goaly's to ACT on (a `candidates` overlay on the extension), not the worker's to read — so + // the directive clause is stripped and any remaining guidance stays the note. The explicit + // `--candidates` / `--best-of` flag wins, exactly as on a fresh run. + const explicit = has('candidates') || has('best-of'); + let delegation: ParsedArgs['delegation']; + if (note !== undefined) { + const directive = parseDelegationDirective(note); + if (directive !== null) { + if (directive.candidates > MAX_CANDIDATES) { + throw new UsageError( + `"${directive.phrase}": at most ${MAX_CANDIDATES} parallel candidates are supported ` + + `(each is a full concurrent worker + worktree) — ask for ${MAX_CANDIDATES} or fewer`, + ); + } + delegation = { + candidates: directive.candidates, + phrase: directive.phrase, + overriddenByFlag: explicit, + }; + note = directive.cleaned.length > 0 ? directive.cleaned : undefined; + } + } const stuck = { ...(has('stuck-no-diff') ? { noDiff: config.stuckPolicy.noDiff } : {}), ...(has('stuck-repeat-threshold') @@ -773,9 +863,17 @@ function collectResumeExtension(flags: RawFlags, config: RunConfig): RunExtensio ? { budgetWallMs: config.budget.wallClockMs } : {}), ...(Object.keys(stuck).length > 0 ? { stuck } : {}), + ...(explicit + ? { candidates: config.candidates } + : delegation !== undefined + ? { candidates: delegation.candidates } + : {}), ...(note !== undefined ? { note } : {}), }; - return Object.keys(extension).length > 0 ? extension : undefined; + return { + extension: Object.keys(extension).length > 0 ? extension : undefined, + delegation, + }; } /** Fields that may be sourced inline / from a file / from stdin; a CLI source overrides config. */ @@ -871,7 +969,38 @@ export async function parseArgs( '--goal - (stdin)', ); } - const goalForParse = resolved.goal ?? RESUMED_GOAL_PLACEHOLDER; + // Natural-language parallel delegation: an explicit directive in the goal ("work with 4 + // subagents") maps onto the best-of-N tournament (issue #85) and its clause is STRIPPED — the + // goal is frozen into the contract and read by the judge/approver, so a leftover directive would + // become an unverifiable success criterion. Deterministic grammar (see `delegation.ts`), loudly + // logged by the CLI; the explicit `--candidates` / `--best-of` flag (or config) always wins. + const explicitCandidates = candidatesFlag(flags); + let goalText = resolved.goal; + let delegation: ParsedArgs['delegation']; + if (goalText !== undefined) { + const directive = parseDelegationDirective(goalText); + if (directive !== null) { + if (directive.candidates > MAX_CANDIDATES) { + throw new UsageError( + `"${directive.phrase}": at most ${MAX_CANDIDATES} parallel candidates are supported ` + + `(each is a full concurrent worker + worktree) — ask for ${MAX_CANDIDATES} or fewer`, + ); + } + if (directive.cleaned.length === 0) { + throw new UsageError( + `the goal '${goalText}' is only a delegation directive — say WHAT to achieve too, ` + + `e.g. goaly "fix the flaky test, ${directive.phrase}"`, + ); + } + goalText = directive.cleaned; + delegation = { + candidates: directive.candidates, + phrase: directive.phrase, + overriddenByFlag: explicitCandidates !== undefined, + }; + } + } + const goalForParse = goalText ?? RESUMED_GOAL_PLACEHOLDER; const cliInput = CliInput.parse({ goal: goalForParse, @@ -889,11 +1018,16 @@ export async function parseArgs( ...(str(flags, 'max-iterations') !== undefined ? { maxIterations: str(flags, 'max-iterations') } : {}), - ...(candidatesFlag(flags) !== undefined ? { candidates: candidatesFlag(flags) } : {}), + ...(explicitCandidates !== undefined + ? { candidates: explicitCandidates } + : delegation !== undefined + ? { candidates: String(delegation.candidates) } + : {}), ...(parseResumeBestOfIncomplete(flags) !== undefined ? { resumeBestOfIncomplete: parseResumeBestOfIncomplete(flags) } : {}), ...(flags['phased'] !== undefined ? { phased: true } : {}), + ...(flags['parallel-phases'] !== undefined ? { parallelPhases: true } : {}), ...(str(flags, 'max-phases') !== undefined ? { maxPhases: str(flags, 'max-phases') } : {}), ...(str(flags, 'max-plan-revisions') !== undefined ? { maxPlanRevisions: str(flags, 'max-plan-revisions') } @@ -950,10 +1084,29 @@ export async function parseArgs( const harness = parseHarness(str(flags, 'harness')); const config = cliInputToRunConfig(cliInput); + + // EXPERIMENTAL parallel waves: the fan-out only exists inside a phased plan (grouped sub-goals), + // and wave children compile + Seal their contracts CONCURRENTLY — an interactive gate cannot pause + // K children at once, so autonomy is required (the contracts are still frozen + logged loudly). + if (config.parallelPhases && !resuming) { + if (!config.phased) { + throw new UsageError( + "--parallel-phases parallelizes a phased plan's grouped sub-goals — pair it with --phased " + + '(and mark consecutive phases with a shared "group" in the plan)', + ); + } + if (!config.autonomous) { + throw new UsageError( + '--parallel-phases requires --autonomous: wave children seal their frozen contracts ' + + 'concurrently and cannot pause at interactive gates (each contract is still frozen + logged)', + ); + } + } // Explicitness for the resume extension is judged on CLI flags ONLY (never the config-file // overlay): a `.goalyrc` default like "budget-tokens" must not append a RUN_EXTENDED marker to // the log on every resume — an extension is an explicit per-invocation operator act. - const resumeExtend = collectResumeExtension(cliFlags, config); + const resumed = collectResumeExtension(cliFlags, config); + const resumeExtend = resumed.extension; // Piping a field via stdin (`--goal -`) drains the ONLY stdin stream, so the interactive Seal // prompt that a non-autonomous run needs would read EOF / hang. That used to be a doc-note @@ -974,6 +1127,7 @@ export async function parseArgs( worktreeRun: parseWorktreeRun(flags), config, harness, + harnessExplicit: cliFlags['harness'] !== undefined, models: parseModels(flags), llmProvider: parseLlmProvider(str(flags, 'llm-provider')), workspace: str(flags, 'workspace') ?? process.cwd(), @@ -982,6 +1136,9 @@ export async function parseArgs( planFile: str(flags, 'plan-file'), resumeRunId: str(flags, 'resume'), resumeExtend, + // A directive can come from the goal (fresh run) or the resume note — never both in one + // invocation (a resumed run's goal is the placeholder; a fresh run rejects --note). + delegation: delegation ?? resumed.delegation, fromRunId: str(flags, 'from-run'), inheritSession: flags['inherit-session'] !== undefined, logLevel: parseLogLevel(str(flags, 'log-level')), @@ -1223,14 +1380,7 @@ function parseSandbox(flags: RawFlags): SandboxPolicy { function parseHarness(value: string | undefined): HarnessChoice { if (value === undefined) return 'claude'; - if ( - value === 'claude' || - value === 'codex' || - value === 'droid' || - value === 'pi' || - value === 'fake' || - value === 'goaly-code' - ) { + if (isHarnessChoice(value)) { return value; } // The `claude-code` value was renamed to `claude` (one name per CLI across the harness and the @@ -1445,6 +1595,7 @@ function baseArgs( // a placeholder config; never used for the help / runs commands. config: cliInputToRunConfig(CliInput.parse({ goal: 'help', verifyCmd: 'true' })), harness: 'claude', + harnessExplicit: false, models: ModelSelection.parse({}), llmProvider: 'claude', workspace, @@ -1469,5 +1620,6 @@ function baseArgs( baseUrl: undefined, llmApiKeyEnv: 'OPENAI_API_KEY', resumeExtend: undefined, + delegation: undefined, }; } diff --git a/src/cli/compose.ts b/src/cli/compose.ts index 4a1ecfd..940de8c 100644 --- a/src/cli/compose.ts +++ b/src/cli/compose.ts @@ -38,6 +38,7 @@ import { AgentCliHarness } from '../harness/agent-cli-harness'; import { SystemClock } from '../driver/clock'; import { SystemBudgetMeter } from '../driver/budget'; import { LlmTokenMeter, meterLlm } from '../driver/llm-meter'; +import { DefaultWaveRunner } from '../driver/wave-runner'; import { buildLogger, type FileLogOptions } from '../log/build'; import type { Logger, LogLevel } from '../log/logger'; import type { LogFs } from '../log/sinks'; @@ -191,6 +192,13 @@ export type ComposeOptions = { sealGate?: SealGate; /** Inject the plan-Seal gate (phased runs), same rules as {@link sealGate}. */ planGate?: PlanGate; + /** + * Inject the harness adapter per workspace root (tests/embedders) — bypasses {@link harness} + * selection. The FACTORY shape (not a single adapter) exists for EXPERIMENTAL parallel waves, + * where each wave child composes its own deps rooted at its worktree: the factory receives that + * root so a scripted test harness can write into the right tree. + */ + harnessFactory?: (workspaceRoot: string) => HarnessAdapter; }; /** @@ -361,11 +369,12 @@ export function composeDeps(config: RunConfig, options: ComposeOptions): DriverD options.egressProxy, ); const workspace = new GitWorkspace(options.workspaceRoot, undefined, excludes, true, runLauncher); - // Best-of-N worktree host (issue #85): only wired when `--candidates > 1` (a `--candidates 1` run - // never touches it). It shares the canonical root / exec / excludes / verify-jail so each candidate's - // isolated worktree hashes + scores identically to the canonical workspace. + // Worktree host: wired for best-of-N (issue #85, `--candidates > 1`) and for EXPERIMENTAL + // cooperative parallel waves (`--parallel-phases`) — a run using neither never touches it. It + // shares the canonical root / exec / excludes / verify-jail so each isolated worktree hashes + + // scores identically to the canonical workspace. const worktrees = - config.candidates > 1 + config.candidates > 1 || (config.phased && config.parallelPhases) ? new GitWorktreeHost({ root: options.workspaceRoot, exec: realExec, @@ -470,6 +479,39 @@ export function composeDeps(config: RunConfig, options: ComposeOptions): DriverD // detected, never assumed — a non-code workspace yields `undefined` and nothing is injected. const workspaceFacts = detectWorkspaceFacts(options.workspaceRoot); + // ONE budget meter for the whole run — hoisted so EXPERIMENTAL parallel-wave children share it + // (the `--budget-tokens` cap governs the fan-out, not each child separately). + const budget = new SystemBudgetMeter(config.budget, clock); + + // EXPERIMENTAL cooperative parallel waves (`--parallel-phases`): each wave CHILD is a FULL goaly + // run composed by this very function, rooted at its ephemeral worktree — its own frozen contract, + // two-key gate, and write-ahead log (under `/.goaly`), on the parent's budget meter and + // interrupt probe. Parent-anchored artifact paths (log/stream/state overrides, the diff baseline) + // are stripped so children never write into the parent's files. + const wave = + config.phased && config.parallelPhases && worktrees !== undefined + ? new DefaultWaveRunner({ + host: worktrees, + workspace, + workspaceRoot: options.workspaceRoot, + ...(timeouts.verifyMs !== undefined ? { verifyTimeoutMs: timeouts.verifyMs } : {}), + logger, + composeChild: async (spec, worktree, childRunId, interrupted) => { + const { logFile: _lf, streamFile: _sf, stateDir: _sd, baseline: _b, ...rest } = options; + const childDeps = composeDeps(spec.config, { + ...rest, + workspaceRoot: worktree.root, + runId: childRunId, + }); + return { + ...childDeps, + budget, + ...(interrupted !== undefined ? { interrupted } : {}), + }; + }, + }) + : undefined; + return { compiler: seedCompiler( critiqueCompiler( @@ -500,14 +542,16 @@ export function composeDeps(config: RunConfig, options: ComposeOptions): DriverD : new HumanSealGate({ allowRevise: config.maxSealRevisions > 0 })), ...(phasedSeams !== undefined ? phasedSeams : {}), harness: - options.harness === 'goaly-code' - ? makeGoalyCodeHarness(options, models, stateDir, logger, launcher) - : makeHarness(options.harness, models.harness, timeouts.harnessMs, timeouts.harnessIdleMs, { - launcher, - workspace: options.workspaceRoot, - policy: options.sandbox ?? defaultPolicy(), - ...(options.egressProxy !== undefined ? { proxy: options.egressProxy } : {}), - }), + options.harnessFactory !== undefined + ? options.harnessFactory(options.workspaceRoot) + : options.harness === 'goaly-code' + ? makeGoalyCodeHarness(options, models, stateDir, logger, launcher) + : makeHarness(options.harness, models.harness, timeouts.harnessMs, timeouts.harnessIdleMs, { + launcher, + workspace: options.workspaceRoot, + policy: options.sandbox ?? defaultPolicy(), + ...(options.egressProxy !== undefined ? { proxy: options.egressProxy } : {}), + }), makeLadder: (contract) => { // Surface the frozen authored bar (`generatedFiles`) in the diff the two LLM keys review, even // though it's git-excluded (issue #52) from the user's `git status`. Without this the judge sees @@ -540,8 +584,9 @@ export function composeDeps(config: RunConfig, options: ComposeOptions): DriverD prepareLlm: llmFor(models.judge, 'preflight'), workspace, ...(worktrees !== undefined ? { worktrees } : {}), + ...(wave !== undefined ? { wave } : {}), clock, - budget: new SystemBudgetMeter(config.budget, clock), + budget, llmMeter, runlog: new FileRunLog(path.join(stateDir, options.runId)), logger, diff --git a/src/cli/compose.wave.test.ts b/src/cli/compose.wave.test.ts new file mode 100644 index 0000000..43a16f8 --- /dev/null +++ b/src/cli/compose.wave.test.ts @@ -0,0 +1,196 @@ +import { describe, it, expect, afterEach } from 'vitest'; +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { composeDeps } from './compose'; +import { drive } from '../driver/driver'; +import { makeConfig } from '../testing/fakes'; +import { asRunId, coerceSessionId, type SessionId } from '../domain/ids'; +import type { HarnessAdapter } from '../harness/adapter'; +import type { LlmProvider } from '../llm/provider'; +import { runProcess } from '../util/spawn'; + +/** + * The parallel-wave pipeline END TO END on REAL git — real worktrees, real 3-way merges, real + * promotion — with zero LLM tokens and zero agent CLIs: the LLM is routed by prompt content (each + * step's schema marker) so the CONCURRENT children can't race a scripted queue, and the harness is + * a scripted writer that creates whatever file its sub-goal names, inside its own worktree. + */ + +const routedLlm: LlmProvider = { + name: 'routed-fake-llm', + async complete(req) { + // The schema/marker may live in `system` (the compiler's session-style calls) or the prompt. + const p = `${req.system ?? ''}\n${req.prompt}`; + // The usage-gate shape classification (compile phase). + if (p.includes('"buildAndUse"')) { + return { text: '{"buildAndUse":false,"targetArtifact":null,"reason":"n/a"}' }; + } + // The Sign-off approver (child sign-offs + the acceptance sign-off). + if (p.includes('{"veto"')) return { text: '{"veto": false}' }; + // The per-child authoring compiler: a deterministic bar per sub-goal, no rubric (no judge rung). + if (p.includes('"command": string')) { + if (p.includes('a.txt')) return { text: '{"command":"test -f a.txt","rubric":""}' }; + if (p.includes('b.txt')) return { text: '{"command":"test -f b.txt","rubric":""}' }; + return { text: '{"command":"true","rubric":""}' }; + } + throw new Error(`unrouted LLM prompt: ${p.slice(0, 160)}`); + }, +}; + +/** A worker that "achieves" its sub-goal by writing the file the prompt names — in ITS OWN root. */ +function scriptedWriter(root: string): HarnessAdapter { + return { + name: 'scripted-wave-writer', + async run(prompt: string, sessionId?: SessionId) { + const id = sessionId ?? coerceSessionId('scripted', 'scripted'); + if (prompt.includes('a.txt')) await writeFile(path.join(root, 'a.txt'), 'alpha\n'); + if (prompt.includes('b.txt')) await writeFile(path.join(root, 'b.txt'), 'beta\n'); + return { output: 'did the work', sessionId: id, status: 'completed' as const }; + }, + }; +} + +async function initRepo(): Promise { + const dir = await mkdtemp(path.join(tmpdir(), 'goaly-wave-e2e-')); + await runProcess('git', ['-C', dir, 'init', '-q']); + await runProcess('git', ['-C', dir, 'config', 'user.email', 't@example.com']); + await runProcess('git', ['-C', dir, 'config', 'user.name', 'tester']); + await writeFile(path.join(dir, 'README.md'), '# fixture\n'); + await runProcess('git', ['-C', dir, 'add', '-A']); + await runProcess('git', ['-C', dir, 'commit', '-qm', 'init']); + return dir; +} + +describe('parallel waves END TO END (compose + drive, real git, routed fake LLM)', () => { + let dir: string | null = null; + afterEach(async () => { + if (dir !== null) await rm(dir, { recursive: true, force: true }); + dir = null; + }); + + it('runs a grouped plan as one wave: children fork, merge cleanly, re-verify, acceptance gates DONE', async () => { + dir = await initRepo(); + // Two INDEPENDENT sub-goals sharing wave group 1 — the whole plan is one wave + acceptance. + await writeFile( + path.join(dir, 'plan.json'), + JSON.stringify({ + phases: [ + { goal: 'create a file a.txt containing alpha', group: 1 }, + { goal: 'create a file b.txt containing beta', group: 1 }, + ], + }), + ); + const config = makeConfig({ + goal: 'produce both fixture files', + // The ORIGINAL verifier becomes the cumulative acceptance bar on the whole merged tree. + verifier: { kind: 'existing', ref: 'test -f a.txt && test -f b.txt' }, + autonomous: true, + phased: true, + parallelPhases: true, + }); + const runId = asRunId('run-wave-e2e'); + const deps = composeDeps(config, { + harness: 'fake', + harnessFactory: scriptedWriter, + workspaceRoot: dir, + runId, + noLogConsole: true, + llm: routedLlm, + planFile: path.join(dir, 'plan.json'), + }); + + const outcome = await drive(deps, config, runId); + + // The whole run reaches DONE through the acceptance contract's two keys. + expect(outcome.status).toBe('DONE'); + // BOTH children's work landed in the canonical tree via the 3-way merge (disjoint files). + expect(await readFile(path.join(dir, 'a.txt'), 'utf8')).toBe('alpha\n'); + expect(await readFile(path.join(dir, 'b.txt'), 'utf8')).toBe('beta\n'); + + // The parent log carries ONE WAVE_RAN with both phases merged — and the reducer never saw the + // children's iterations (each child kept its own write-ahead log in its worktree). + const stored = await deps.runlog.read(); + const wave = stored?.entries.find((e) => e.event.tag === 'WAVE_RAN'); + expect(wave?.event.tag).toBe('WAVE_RAN'); + if (wave?.event.tag === 'WAVE_RAN') { + expect(wave.event.outcomes.map((o) => o.kind)).toEqual(['merged', 'merged']); + } + // No stray worktrees left behind. + const wt = await runProcess('git', ['-C', dir, 'worktree', 'list']); + expect(wt.stdout.trim().split('\n')).toHaveLength(1); + }); + + it('a conflicting wave member downgrades to the classic sequential phase and the run still finishes', async () => { + dir = await initRepo(); + // BOTH sub-goals write the SAME file with different content — the second merge must conflict, + // downgrade to a sequential re-run on the merged tree, and the run must still reach DONE. + await writeFile( + path.join(dir, 'plan.json'), + JSON.stringify({ + phases: [ + { goal: 'create clash.txt saying alpha', group: 1 }, + { goal: 'make clash.txt say beta instead', group: 1 }, + ], + }), + ); + const conflictLlm: LlmProvider = { + name: 'routed-fake-llm', + async complete(req) { + const p = `${req.system ?? ''}\n${req.prompt}`; + if (p.includes('"buildAndUse"')) { + return { text: '{"buildAndUse":false,"targetArtifact":null,"reason":"n/a"}' }; + } + if (p.includes('{"veto"')) return { text: '{"veto": false}' }; + if (p.includes('"command": string')) { + if (p.includes('beta')) return { text: '{"command":"grep -q beta clash.txt","rubric":""}' }; + return { text: '{"command":"grep -q alpha clash.txt","rubric":""}' }; + } + throw new Error(`unrouted LLM prompt: ${p.slice(0, 160)}`); + }, + }; + const conflictWriter = (root: string): HarnessAdapter => ({ + name: 'scripted-conflict-writer', + async run(prompt: string, sessionId?: SessionId) { + const id = sessionId ?? coerceSessionId('scripted', 'scripted'); + // Each child rewrites the SAME file; the sequential fallback then runs on the merged tree. + if (prompt.includes('beta')) await writeFile(path.join(root, 'clash.txt'), 'beta\n'); + else await writeFile(path.join(root, 'clash.txt'), 'alpha\n'); + return { output: 'did the work', sessionId: id, status: 'completed' as const }; + }, + }); + const config = makeConfig({ + goal: 'end with clash.txt saying beta', + verifier: { kind: 'existing', ref: 'grep -q beta clash.txt' }, + autonomous: true, + phased: true, + parallelPhases: true, + }); + const runId = asRunId('run-wave-conflict'); + const deps = composeDeps(config, { + harness: 'fake', + harnessFactory: conflictWriter, + workspaceRoot: dir, + runId, + noLogConsole: true, + llm: conflictLlm, + planFile: path.join(dir, 'plan.json'), + }); + + const outcome = await drive(deps, config, runId); + + expect(outcome.status).toBe('DONE'); + expect(await readFile(path.join(dir, 'clash.txt'), 'utf8')).toBe('beta\n'); + + const stored = await deps.runlog.read(); + const wave = stored?.entries.find((e) => e.event.tag === 'WAVE_RAN'); + expect(wave?.event.tag).toBe('WAVE_RAN'); + if (wave?.event.tag === 'WAVE_RAN') { + const kinds = wave.event.outcomes.map((o) => o.kind).sort(); + expect(kinds).toEqual(['merged', 'unmerged']); // one landed, one downgraded fail-closed + } + // The downgraded phase re-ran through the CLASSIC sequential path: its own frozen contract. + const contracts = stored?.entries.filter((e) => e.event.tag === 'CONTRACT_COMPILED') ?? []; + expect(contracts.length).toBeGreaterThanOrEqual(2); // the fallback phase + acceptance + }); +}); diff --git a/src/cli/delegation.test.ts b/src/cli/delegation.test.ts new file mode 100644 index 0000000..016c9e6 --- /dev/null +++ b/src/cli/delegation.test.ts @@ -0,0 +1,96 @@ +import { describe, it, expect } from 'vitest'; +import { DEFAULT_DELEGATION_CANDIDATES, parseDelegationDirective } from './delegation'; + +describe('parseDelegationDirective — natural-language parallel delegation', () => { + describe('counted subagent directives (verb required)', () => { + it('"work with N subagents" parses and strips cleanly', () => { + const d = parseDelegationDirective('fix the flaky auth test, work with 4 subagents'); + expect(d).not.toBeNull(); + expect(d!.candidates).toBe(4); + expect(d!.cleaned).toBe('fix the flaky auth test'); + }); + + it('a leading directive strips its dangling connector ("use N subagents to …")', () => { + const d = parseDelegationDirective('use 4 subagents to fix the flaky auth test'); + expect(d!.candidates).toBe(4); + expect(d!.cleaned).toBe('fix the flaky auth test'); + }); + + it('hyphenated "sub-agents" and other verbs parse too', () => { + expect(parseDelegationDirective('fix it, delegate to 2 sub-agents')!.candidates).toBe(2); + expect(parseDelegationDirective('fix it, spawn 5 subagents')!.candidates).toBe(5); + expect(parseDelegationDirective('fix it using 3 concurrent subagents')!.candidates).toBe(3); + }); + + it('a mid-sentence directive keeps the surrounding goal intact', () => { + const d = parseDelegationDirective('use 3 subagents and make the linter pass'); + expect(d!.candidates).toBe(3); + expect(d!.cleaned).toBe('make the linter pass'); + }); + + it('a sentence-final directive keeps the terminator', () => { + const d = parseDelegationDirective('Make the linter pass, use 3 subagents.'); + expect(d!.candidates).toBe(3); + expect(d!.cleaned).toBe('Make the linter pass.'); + }); + }); + + describe('parallel-attempt directives', () => { + it('"N parallel attempts" parses with or without a verb', () => { + expect(parseDelegationDirective('fix the parser with 3 parallel attempts')!.candidates).toBe(3); + expect(parseDelegationDirective('fix the parser, 2 parallel attempts')!.candidates).toBe(2); + expect(parseDelegationDirective('make 4 parallel attempts at fixing the parser')!.candidates).toBe(4); + }); + + it('"N parallel candidates/tries" parse too', () => { + expect(parseDelegationDirective('fix it, run 4 parallel candidates')!.candidates).toBe(4); + expect(parseDelegationDirective('fix it with 2 parallel tries')!.candidates).toBe(2); + }); + }); + + describe('uncounted subagent directives (documented default)', () => { + it('"use subagents" defaults the count', () => { + const d = parseDelegationDirective('fix the flaky test, use subagents'); + expect(d!.candidates).toBe(DEFAULT_DELEGATION_CANDIDATES); + expect(d!.cleaned).toBe('fix the flaky test'); + }); + + it('"spawn several subagents" defaults the count', () => { + expect(parseDelegationDirective('spawn several subagents to fix the test')!.candidates).toBe( + DEFAULT_DELEGATION_CANDIDATES, + ); + }); + }); + + describe('false-positive guard — application-domain goals never trigger', () => { + it.each([ + 'make the tests run in parallel', + 'implement a job queue with 4 parallel workers', + 'add retry logic with 3 attempts', + 'handle 5 parallel login attempts without racing', + 'document the 3 subagents in the README', // subagents as a domain noun, no delegation verb + 'implement a worker pool with 8 threads', + 'parallelize the build across CI shards', + ])('%s', (goal) => { + expect(parseDelegationDirective(goal)).toBeNull(); + }); + + it('a zero count is not a directive', () => { + expect(parseDelegationDirective('use 0 subagents to fix it')).toBeNull(); + }); + }); + + it('the matched phrase is surfaced for the interpretation log', () => { + const d = parseDelegationDirective('fix the test, work with 4 subagents'); + expect(d!.phrase).toContain('work with 4 subagents'); + }); + + it('only the FIRST directive is consumed', () => { + const d = parseDelegationDirective('use 4 subagents, then use 2 subagents'); + expect(d!.candidates).toBe(4); + }); + + it('a goal that is ONLY a directive cleans to the empty string (caller fails closed)', () => { + expect(parseDelegationDirective('use 4 subagents')!.cleaned).toBe(''); + }); +}); diff --git a/src/cli/delegation.ts b/src/cli/delegation.ts new file mode 100644 index 0000000..2c15d7b --- /dev/null +++ b/src/cli/delegation.ts @@ -0,0 +1,113 @@ +/** + * Natural-language parallel delegation: let the user tell goaly to parallelize in plain language — + * in the goal ("fix the flaky test, work with 4 subagents") or in a `--resume` note ("try 4 + * parallel attempts") — and map it onto the EXISTING best-of-N tournament (`--candidates`, issue + * #85). Nothing new runs: N independent worker attempts per iteration in isolated worktrees, scored + * against the same frozen ladder, best tree wins; the reducer never learns N existed. + * + * Detection is DETERMINISTIC — a small directive grammar, never an LLM parse (an LLM interpreting + * config would be exactly the "LLM in control flow" this codebase exists to avoid). The grammar is + * deliberately narrow to keep false positives out of real goals: + * + * - ` N subagents` — "use 4 subagents", "work with 3 sub-agents", "delegate to 2 subagents". + * A delegation VERB is required ("document the 3 subagents" is a goal about subagents, not a + * directive) and only unambiguous agent nouns participate — never "workers"/"threads"/"jobs", + * which routinely describe the application domain ("a queue with 4 parallel workers"). + * - ` subagents` (uncounted) — "use subagents", "spawn subagents" ⇒ a documented default + * of {@link DEFAULT_DELEGATION_CANDIDATES}. + * - `N parallel attempts|candidates|tries` — "make 3 parallel attempts". The word "parallel" must + * be ADJACENT to the noun ("5 parallel login attempts" does not match). + * + * The matched clause is STRIPPED from the text: the goal is frozen into the contract and read by + * the judge/approver, and a leftover "use 4 subagents" would become an unverifiable success + * criterion. The caller logs the interpretation loudly (phrase → candidates) so the rewrite is + * always auditable. Anything the grammar does not match is left untouched — fail-closed to the + * classic single attempt, never a guess. The explicit `--candidates` flag always wins. + */ + +/** Candidate count when the directive names no number ("use subagents"). */ +export const DEFAULT_DELEGATION_CANDIDATES = 3; + +export type DelegationDirective = { + /** The parsed candidate count (uncapped here — the CLI seam enforces MAX_CANDIDATES). */ + readonly candidates: number; + /** The exact matched directive text (for the loud interpretation log). */ + readonly phrase: string; + /** The input with the directive clause removed and punctuation/whitespace tidied. */ + readonly cleaned: string; +}; + +/** Delegation verbs that must introduce a subagent directive (counted or bare). */ +const VERB = String.raw`(?:us(?:e|ing)|spawn(?:ing)?|launch(?:ing)?|run(?:ning)?|try(?:ing)?|delegate\s+to|work(?:ing)?\s+with|with|across)`; + +/** Optional lead-in words a directive clause often carries ("please and then …"). */ +const LEAD_IN = String.raw`(?:please\s+)?(?:and\s+)?(?:then\s+)?(?:in\s+parallel\s+)?`; + +/** The unambiguous agent noun. Deliberately NOT workers/threads/jobs (application-domain words). */ +const SUBAGENTS = String.raw`sub-?agents?`; + +/** + * The three directive shapes, tried in order; the FIRST match wins and only it is stripped. + * Each pattern consumes its leading separator run (comma/semicolon/dash) so the strip is clean. + */ +const PATTERNS: readonly { re: RegExp; count: (m: RegExpMatchArray) => number }[] = [ + // " N subagents" — counted, verb required. + { + re: new RegExp( + String.raw`[\s,;:–—-]*\b${LEAD_IN}${VERB}\s+(\d+)\s+(?:parallel\s+|concurrent\s+)?${SUBAGENTS}\b`, + 'i', + ), + count: (m) => Number(m[1]), + }, + // "N parallel attempts|candidates|tries" — "parallel" adjacent to the noun disambiguates. + { + re: new RegExp( + String.raw`[\s,;:–—-]*\b${LEAD_IN}(?:${VERB}\s+|make\s+|making\s+)?(\d+)\s+parallel\s+(?:attempts?|candidates?|tries)\b`, + 'i', + ), + count: (m) => Number(m[1]), + }, + // " subagents" — uncounted, verb required, documented default count. + { + re: new RegExp( + String.raw`[\s,;:–—-]*\b${LEAD_IN}${VERB}\s+(?:multiple\s+|several\s+|some\s+|a\s+few\s+|parallel\s+|concurrent\s+)?${SUBAGENTS}\b`, + 'i', + ), + count: () => DEFAULT_DELEGATION_CANDIDATES, + }, +]; + +/** + * Parse (and strip) a natural-language delegation directive. Returns `null` when the text carries + * none — the classic single-attempt run. Pure and deterministic; the caller owns validation + * (candidate cap) and the loud interpretation log. + */ +export function parseDelegationDirective(text: string): DelegationDirective | null { + for (const { re, count } of PATTERNS) { + const m = text.match(re); + if (m === null || m.index === undefined) continue; + const candidates = count(m); + if (!Number.isInteger(candidates) || candidates < 1) continue; + const cleaned = tidy(text.slice(0, m.index), text.slice(m.index + m[0].length)); + // The match consumes its leading separator run for a clean strip — drop it from the reported + // phrase so the log reads "work with 3 subagents", not ", work with 3 subagents". + return { candidates, phrase: m[0].replace(/^[\s,;:–—-]+/, ''), cleaned }; + } + return null; +} + +/** + * Re-join the text around a stripped clause and tidy the seam: collapse doubled whitespace, drop a + * connector or punctuation run left dangling at the join ("use 4 subagents to fix X" → "fix X"; + * "fix X, use 4 subagents." → "fix X."), and never leave a space before closing punctuation. + */ +function tidy(before: string, after: string): string { + let rest = after.replace(/^\s*(?:(?:to|and|then)\s+|[,;:\s]+)*/i, ''); + // A directive that ended the sentence leaves its terminator on `after` — keep exactly one. + if (/^[.!?]/.test(after.trimStart()) && rest.length === 0) rest = after.trimStart().charAt(0); + const joined = before.trimEnd().length > 0 ? `${before.trimEnd()} ${rest}` : rest; + return joined + .replace(/\s+([.,;:!?])/g, '$1') + .replace(/\s{2,}/g, ' ') + .trim(); +} diff --git a/src/cli/main.test.ts b/src/cli/main.test.ts index aba6296..1fec887 100644 --- a/src/cli/main.test.ts +++ b/src/cli/main.test.ts @@ -288,6 +288,28 @@ describe('main() — resume extension end-to-end (operator control, ADR 0012)', expect(second.out).toContain('reached maxIterations'); }); + it("a --resume without --harness ADOPTS the run's recorded harness (never silently the default)", async () => { + // Regression: a resume that omitted --harness silently switched to the default CLI (claude), + // threading the prior harness's session/sentinel into a different tool — observed live as + // `claude --resume noop-session` crashing every candidate of a resumed fake-harness run. + const first = await captureAll(() => + main(['run', 'g', '--verify-cmd', 'false', '--harness', 'fake', '--autonomous', + '--max-iterations', '2', '--workspace', root]), + ); + expect(first.code).toBe(1); + const runId = /── goaly run (run-[0-9a-f-]+) ──/.exec(first.out)?.[1]; + expect(runId).toBeDefined(); + + // No --harness here: the run log's 'fake' must be adopted (a silent claude default would spawn + // the real CLI on iteration 2). + const second = await captureAll(() => + main(['run', '--workspace', root, '--resume', runId!, '--stuck-no-diff', 'false']), + ); + expect(second.out).toContain("continuing with this run's harness 'fake'"); + expect(second.code).toBe(1); + expect(second.out).toContain('reached maxIterations'); + }); + it('refuses to extend a DONE run, pointing at --from-run', async () => { // Fabricate a DONE run log directly (no LLM/harness involved): both keys turned. const contract = makeFakeContract({ goal: 'g' }); diff --git a/src/cli/run-cmd.ts b/src/cli/run-cmd.ts index 9d5bf32..d40f67b 100644 --- a/src/cli/run-cmd.ts +++ b/src/cli/run-cmd.ts @@ -1,7 +1,7 @@ import { randomUUID } from 'node:crypto'; import path from 'node:path'; import { readFile } from 'node:fs/promises'; -import { USAGE, type ParsedArgs } from './args'; +import { USAGE, isHarnessChoice, type ParsedArgs } from './args'; import { composeDeps, STATE_DIR, EndpointConfigError } from './compose'; import { SandboxUnavailableError, isAllowlist, startEgressProxy, type EgressProxy } from '../sandbox'; import { drive } from '../driver/driver'; @@ -210,35 +210,21 @@ export async function executeRun(parsed: ParsedArgs, io: RunIo): Promise o.kind === 'merged').length; + const fallback = e.outcomes.length - merged; + const tail = fallback > 0 ? `, ${fallback} downgraded to sequential` : ''; + return `${at} wave: ${merged}/${e.outcomes.length} phase(s) merged + re-verified${tail}`; + } case 'CHECKPOINTED': return null; // internal diff-baseline plumbing — noise for a human watcher } diff --git a/src/domain/config.ts b/src/domain/config.ts index d58af16..bfbc31b 100644 --- a/src/domain/config.ts +++ b/src/domain/config.ts @@ -139,6 +139,17 @@ export const RunConfig = z.object({ * reads this in `initial()` to seed PLANNING instead of COMPILING. */ phased: z.boolean().default(false), + /** + * EXPERIMENTAL — cooperative parallel waves (`--parallel-phases`, opt-in). When true, CONSECUTIVE + * plan phases sharing a `group` value execute as one concurrent WAVE: each phase runs as its own + * frozen, two-key CHILD goaly run in an isolated worktree (sharing this run's budget), then the + * children are merged in phase order and each merged phase's frozen ladder is RE-VERIFIED on the + * combined tree — a merge is never trusted. Any conflict / red re-verify downgrades that phase to + * the classic sequential run (fail-closed; the bar never moves, only the starting tree). Requires + * `phased` + `autonomous` (child contracts cannot pause at interactive Seals mid-fan-out). Default + * false ⇒ grouped plans still run strictly sequentially — byte-for-byte the classic phased run. + */ + parallelPhases: z.boolean().default(false), /** Max sub-goals a phased plan may contain; a planner that exceeds it is a fail-closed PLAN_FAILED. */ maxPhases: z.number().int().positive().default(10), /** @@ -266,6 +277,7 @@ export type LoopPolicy = Pick< | 'stuckPolicy' | 'budget' | 'phased' + | 'parallelPhases' | 'maxPhases' | 'installMissingTools' >; @@ -287,6 +299,7 @@ export const pickLoopPolicy = (c: LoopPolicy): LoopPolicy => ({ stuckPolicy: c.stuckPolicy, budget: c.budget, phased: c.phased, + parallelPhases: c.parallelPhases, maxPhases: c.maxPhases, installMissingTools: c.installMissingTools, }); @@ -330,6 +343,8 @@ export const CliInput = z.object({ resumeBestOfIncomplete: z.enum(['rerun', 'collapse']).optional(), /** Phased decomposition (issue #48). */ phased: z.coerce.boolean().optional(), + /** EXPERIMENTAL cooperative parallel waves (`--parallel-phases`; requires phased + autonomous). */ + parallelPhases: z.coerce.boolean().optional(), maxPhases: z.coerce.number().int().positive().optional(), maxPlanRevisions: z.coerce.number().int().nonnegative().optional(), budgetTokens: z.coerce.number().int().positive().optional(), @@ -445,6 +460,7 @@ export function cliInputToRunConfig(input: CliInput): RunConfig { ? { resumeBestOfIncomplete: input.resumeBestOfIncomplete } : {}), ...(input.phased !== undefined ? { phased: input.phased } : {}), + ...(input.parallelPhases !== undefined ? { parallelPhases: input.parallelPhases } : {}), ...(input.maxPhases !== undefined ? { maxPhases: input.maxPhases } : {}), ...(input.maxPlanRevisions !== undefined ? { maxPlanRevisions: input.maxPlanRevisions } diff --git a/src/domain/events.ts b/src/domain/events.ts index ccbb9c3..966af05 100644 --- a/src/domain/events.ts +++ b/src/domain/events.ts @@ -107,6 +107,45 @@ export const OrchestratorEvent = z.discriminatedUnion('tag', [ * (like CHECKPOINTED) AND drives the reducer's advance to the next phase's contract compile. */ z.object({ tag: z.literal('PHASE_ADVANCED'), tree: DiffHash }), + /** + * EXPERIMENTAL — a cooperative parallel WAVE completed (`--parallel-phases`): consecutive grouped + * phases ran concurrently as isolated, frozen, two-key CHILD runs; the Driver merged the DONE + * children in phase order and RE-VERIFIED each merged phase's frozen ladder on the combined tree. + * One outcome per wave member: + * - `merged` — child DONE, merged clean, frozen ladder green on the combined tree ⇒ the phase + * is complete (the reducer SKIPS it when advancing). + * - `unmerged` — the child failed to land (merge conflict, red re-verify, or a non-DONE child + * outcome) ⇒ FAIL-CLOSED downgrade: the phase re-runs as a classic sequential + * phase on the merged-so-far tree (a fresh frozen contract on the same sub-goal — + * the bar never moves, only the starting tree). + * Carries the post-merge checkpoint tree (the baseline for whatever follows, like PHASE_ADVANCED). + * Fed to `step()` (it drives the advance) AND read by replay for baseline reconstruction. + */ + z.object({ + tag: z.literal('WAVE_RAN'), + outcomes: z + .array( + z.discriminatedUnion('kind', [ + z.object({ + kind: z.literal('merged'), + /** The plan phase index this outcome belongs to. */ + index: z.number().int().nonnegative(), + /** The child run's total spend (all layers), for the parent's usage fold. */ + usage: TokenUsage.optional(), + }), + z.object({ + kind: z.literal('unmerged'), + index: z.number().int().nonnegative(), + /** Why the child did not land (conflict / red re-verify / child FAILED-ABORTED / error). */ + reason: z.string(), + usage: TokenUsage.optional(), + }), + ]), + ) + .min(1), + /** The post-merge checkpoint tree — the diff baseline for the phases that follow. */ + tree: DiffHash, + }), z.object({ tag: z.literal('CONTRACT_COMPILED'), contract: CompiledContract, @@ -230,6 +269,14 @@ export const OrchestratorEvent = z.discriminatedUnion('tag', [ budgetWallMs: z.number().int().positive().optional(), /** Stuck-policy overrides (each field replaces its counterpart; absent fields keep the prior). */ stuck: StuckPolicy.partial().optional(), + /** + * Best-of-N candidates override (issue #85): raise/lower the per-iteration parallel fan-out + * mid-run — an OPERATIONAL loop knob like `maxIterations`, never the frozen contract. Capped at + * 16 like the config seam (each candidate is a full concurrent worker + worktree). Typically + * set from an explicit `--candidates` at resume or a natural-language `--note` directive + * ("try 4 parallel attempts" — see `src/cli/delegation.ts`). + */ + candidates: z.number().int().positive().max(16).optional(), /** Operator guidance appended to the NEXT agent prompt (worker steering, never the contract). */ note: z.string().min(1).optional(), }), @@ -296,7 +343,17 @@ export type Command = */ | { tag: 'RUN_AGENT_BEST_OF'; prompt: string; sessionId: SessionId | undefined; candidates: number } | { tag: 'RUN_VERIFIER'; contract: CompiledContract } - | { tag: 'REQUEST_SIGNOFF'; goal: string; rubric: string; verdicts: Verdict[] }; + | { tag: 'REQUEST_SIGNOFF'; goal: string; rubric: string; verdicts: Verdict[] } + /** + * EXPERIMENTAL — run a cooperative parallel WAVE (`--parallel-phases`): the consecutive grouped + * phases at `phases[i].index`, each as its own frozen, two-key CHILD goaly run in an isolated + * worktree (per-phase config derived by the reducer exactly as for a sequential phase), then merge + * the DONE children in phase order and re-verify each merged ladder on the combined tree. The + * Driver performs the whole wave through the injected {@link WaveRunner} seam and feeds back ONE + * `WAVE_RAN` event. Emitted INSTEAD of the first phase's `COMPILE_VERIFIER` when the plan groups + * consecutive phases and `config.parallelPhases` is on — still exactly one command per state. + */ + | { tag: 'RUN_WAVE'; phases: { index: number; config: RunConfig }[] }; /** Terminal result of a whole run. */ export const RunOutcome = z.object({ diff --git a/src/domain/ids.ts b/src/domain/ids.ts index 7cafe30..73fa574 100644 --- a/src/domain/ids.ts +++ b/src/domain/ids.ts @@ -55,6 +55,34 @@ export type PlanHash = z.infer; /** Helpers for constructing branded ids from trusted internal sources. */ export const asSessionId = (s: string): SessionId => SessionId.parse(s); +/** + * The synthesized SENTINEL session ids the adapters/driver mint when no REAL id could be recovered + * from the agent CLI. They are valid {@link SessionId}s on the wire (so the event still parses) but + * mean "no resumable session" — threading one into `claude --resume ` / a goaly-code session + * reload would point at nothing (or, worse, at a DIFFERENT harness's sentinel: a run resumed under + * a new `--harness` carries the OLD harness's sentinel, e.g. `claude --resume noop-session`, which + * crashes every turn). Kept in ONE place — the id domain — so the codec sentinels + * (`-unknown`), the NoopHarness sentinel, the driver's error sentinels, and the generic + * coerce fallback can never drift from this skip-list. The harness core, the follow-up resume-hint + * (Capability A), and session inheritance (Capability C) must all skip them. + */ +export const SENTINEL_SESSION_IDS: ReadonlySet = new Set([ + 'unknown-session', // coerceSessionId default fallback + 'noop-session', // NoopHarness (the fake harness) + 'workspace-error', // driver: a workspace (diffHash) failure synthesizes a crashed run + 'best-of-error', // best-of-N driver/tournament: a fan-out error synthesizes a crashed run + 'claude-unknown', + 'codex-unknown', + 'droid-unknown', + 'pi-unknown', + 'goaly-code-unknown', +]); + +/** Whether `id` is a synthesized sentinel rather than a real, resumable harness session id. */ +export function isSentinelSession(id: string): boolean { + return SENTINEL_SESSION_IDS.has(id); +} + /** * Coerce an untrusted candidate (parsed from harness stdout) into a valid SessionId, falling * back to a safe sentinel when it is absent or fails the allowlist — so an adapter never throws diff --git a/src/domain/plan.ts b/src/domain/plan.ts index f8375d0..894c83f 100644 --- a/src/domain/plan.ts +++ b/src/domain/plan.ts @@ -13,6 +13,14 @@ export const SubGoal = z.object({ intent: z.string().optional(), /** Optional rubric guidance for this phase's LLM-judge portion (frozen with the phase contract). */ rubric: z.string().optional(), + /** + * EXPERIMENTAL — cooperative parallel waves (opt-in via `--parallel-phases`): CONSECUTIVE phases + * sharing a `group` value form a WAVE that executes concurrently (each phase as its own frozen, + * two-key child run in an isolated worktree) and is then merged and RE-VERIFIED fail-closed. + * Absent (the default) ⇒ the phase is strictly sequential, byte-for-byte the classic plan. The + * grouping is part of the frozen plan (hashed), so no transition can re-shuffle it. + */ + group: z.number().int().nonnegative().optional(), }); export type SubGoal = z.infer; @@ -52,6 +60,28 @@ export function canonicalPlanString(p: UnhashedPlan): string { goal: s.goal, intent: s.intent ?? null, rubric: s.rubric ?? null, + // `group` (parallel waves) is included ONLY when set, so every pre-existing plan keeps the + // planHash it always had (back-compat), while a grouped plan's grouping is frozen into the hash. + ...(s.group !== undefined ? { group: s.group } : {}), })); return JSON.stringify({ phases }); } + +/** + * The CONSECUTIVE indices sharing `plan.phases[index]`'s wave group, starting at `index` (which must + * be the group's first member for a fan-out to trigger). Returns `[index]` alone when the phase has + * no group, the group has one member, or `index` is mid-group (a resumed sequential fallback walks + * the remaining members one at a time — never re-fans-out from the middle). Pure and total. + */ +export function waveIndicesAt(plan: PhasePlan, index: number): readonly number[] { + const phase = plan.phases[index]; + if (phase === undefined || phase.group === undefined) return [index]; + // Mid-group entry (a sequential fallback / resume) never re-fans-out. + if (index > 0 && plan.phases[index - 1]?.group === phase.group) return [index]; + const wave: number[] = [index]; + for (let i = index + 1; i < plan.phases.length; i += 1) { + if (plan.phases[i]?.group !== phase.group) break; + wave.push(i); + } + return wave; +} diff --git a/src/driver/driver.ts b/src/driver/driver.ts index 249971e..488c11d 100644 --- a/src/driver/driver.ts +++ b/src/driver/driver.ts @@ -18,6 +18,7 @@ import type { PlanGate } from '../plan/plan-gate'; import type { HarnessAdapter } from '../harness/adapter'; import type { Verifier } from '../verify/verifier'; import type { Approver } from '../verify/approver'; +import type { WaveRunner } from './wave'; import type { LlmProvider } from '../llm/provider'; import type { Workspace, WorktreeHost } from '../workspace/workspace'; import type { Clock } from './clock'; @@ -86,6 +87,12 @@ export type DriverDeps = { * but this is absent, the run refuses to start (fail-closed). */ worktrees?: WorktreeHost; + /** + * EXPERIMENTAL — the cooperative parallel-wave seam (`--parallel-phases`). Used ONLY when a + * grouped, phased run fans a wave out; absent ⇒ a `RUN_WAVE` fails closed by DOWNGRADING every + * wave member to the classic sequential phase (never a crash, never a skipped phase). + */ + wave?: WaveRunner; clock: Clock; budget: BudgetMeter; /** @@ -632,6 +639,40 @@ async function perform( return { event: { tag: 'PLAN_SEAL_DECIDED', decision } }; } + case 'RUN_WAVE': { + // EXPERIMENTAL parallel waves: the whole fan-out + merge + re-verify happens behind the + // injected seam; the reducer sees ONE WAVE_RAN. Fail-closed on every failure shape: a missing + // runner or a thrown runner DOWNGRADES every wave member to the classic sequential phase + // (`unmerged`) — never a crash, never a skipped phase, never an unverified merge. + try { + if (deps.wave === undefined) { + throw new Error('parallel waves require a wave runner, but none was configured'); + } + const result = await deps.wave.run(command.phases, deps.interrupted); + log.info('wave completed', { + phases: command.phases.length, + merged: result.outcomes.filter((o) => o.kind === 'merged').length, + }); + return { event: { tag: 'WAVE_RAN', outcomes: result.outcomes, tree: result.tree } }; + } catch (e) { + log.warn('wave runner failed — downgrading every wave member to sequential', { + reason: errorMessage(e), + }); + const tree = await deps.workspace.checkpoint(); + return { + event: { + tag: 'WAVE_RAN', + outcomes: command.phases.map((p) => ({ + kind: 'unmerged' as const, + index: p.index, + reason: `wave fan-out unavailable: ${errorMessage(e)}`, + })), + tree, + }, + }; + } + } + case 'CHECKPOINT_AND_ADVANCE': { // Between-phase checkpoint (issue #47): snapshot the tree (advancing the diff baseline so the // next phase diffs only its own delta) and return the tree on PHASE_ADVANCED — which both drives diff --git a/src/driver/driver.wave.test.ts b/src/driver/driver.wave.test.ts new file mode 100644 index 0000000..9ca3876 --- /dev/null +++ b/src/driver/driver.wave.test.ts @@ -0,0 +1,72 @@ +import { describe, it, expect } from 'vitest'; +import { drive, type DriverDeps } from './driver'; +import { asRunId } from '../domain/ids'; +import { + FakeApprover, + FakeCompiler, + FakeHarness, + FakePlanGate, + FakePlanner, + FakeSealGate, + FakeVerifier, + FakeWorkspace, + InMemoryRunLog, + ManualBudgetMeter, + ManualClock, + approve, + makeConfig, + makeFakeContract, + makeFakePlan, + passVerdict, +} from '../testing/fakes'; + +describe('driver — RUN_WAVE fail-closed (EXPERIMENTAL parallel waves)', () => { + it('a wave with NO runner configured downgrades EVERY member to sequential and the run still finishes', async () => { + // A grouped, parallel-enabled plan… but the deps carry no `wave` seam (an embedder that never + // wired one). The wave must degrade to the classic sequential phased run — never crash, never + // skip a phase, never green anything unverified. + const plan = makeFakePlan({ + phases: [ + { goal: 'member A', group: 1 }, + { goal: 'member B', group: 1 }, + ], + }); + const config = makeConfig({ phased: true, parallelPhases: true, autonomous: true }); + const workspace = new FakeWorkspace('0000000'); + const runlog = new InMemoryRunLog(); + const deps: DriverDeps = { + planner: new FakePlanner(plan), + planGate: new FakePlanGate(), + compiler: new FakeCompiler(makeFakeContract()), + seal: new FakeSealGate(), + // Three sequential worker turns: fallback phase A, fallback phase B, then acceptance. + harness: new FakeHarness( + [{ postHash: '0000aaa' }, { postHash: '0000bbb' }, { postHash: '0000ccc' }], + workspace, + ), + makeLadder: () => new FakeVerifier([passVerdict()]), + approver: new FakeApprover([approve(), approve(), approve()]), + workspace, + clock: new ManualClock(), + budget: new ManualBudgetMeter(false), + runlog, + // no `wave` — the fail-closed path under test + }; + + const outcome = await drive(deps, config, asRunId('run-wave-noseam')); + + expect(outcome.status).toBe('DONE'); + const stored = await runlog.read(); + const wave = stored?.entries.find((e) => e.event.tag === 'WAVE_RAN'); + expect(wave?.event.tag).toBe('WAVE_RAN'); + if (wave?.event.tag === 'WAVE_RAN') { + expect(wave.event.outcomes.map((o) => o.kind)).toEqual(['unmerged', 'unmerged']); + if (wave.event.outcomes[0]!.kind === 'unmerged') { + expect(wave.event.outcomes[0]!.reason).toContain('wave fan-out unavailable'); + } + } + // Both members + acceptance ran the classic sequential path: three agent turns in the log. + const turns = stored?.entries.filter((e) => e.event.tag === 'AGENT_RAN') ?? []; + expect(turns).toHaveLength(3); + }); +}); diff --git a/src/driver/wave-runner.test.ts b/src/driver/wave-runner.test.ts new file mode 100644 index 0000000..7ad4bee --- /dev/null +++ b/src/driver/wave-runner.test.ts @@ -0,0 +1,179 @@ +import { describe, it, expect } from 'vitest'; +import { DefaultWaveRunner, type ComposeChild } from './wave-runner'; +import type { DriverDeps } from './driver'; +import type { WavePhaseSpec } from './wave'; +import type { Worktree } from '../workspace/workspace'; +import type { CompiledContract } from '../domain/contract'; +import { + FakeApprover, + FakeCompiler, + FakeHarness, + FakeSealGate, + FakeVerifier, + FakeWorkspace, + FakeWorktreeHost, + InMemoryRunLog, + ManualBudgetMeter, + ManualClock, + approve, + failVerdict, + makeConfig, + makeFakeContract, + passVerdict, +} from '../testing/fakes'; + +/** + * The whole wave with ZERO LLM calls and ZERO subprocesses: children are full `drive()` runs on + * fakes, the host merges with the scripted fake `mergeTrees`, and the post-merge re-verify runs the + * child contracts' deterministic rungs against the scripted canonical FakeWorkspace. + */ + +const specA: WavePhaseSpec = { index: 0, config: makeConfig({ goal: 'member A', autonomous: true }) }; +const specB: WavePhaseSpec = { index: 1, config: makeConfig({ goal: 'member B', autonomous: true }) }; +const contractA = makeFakeContract({ goal: 'member A', rungs: [{ kind: 'deterministic', command: 'check-a' }] }); +const contractB = makeFakeContract({ goal: 'member B', rungs: [{ kind: 'deterministic', command: 'check-b' }] }); + +/** Compose one DONE-in-one-iteration child on fakes; `fail` scripts a red ladder (child FAILS). */ +function childDeps(opts: { + worktree: Worktree; + contract: CompiledContract; + tree: string; + budget: ManualBudgetMeter; + fail?: boolean; +}): DriverDeps { + const scope = opts.worktree.scope as FakeWorkspace; + return { + compiler: new FakeCompiler(opts.contract), + seal: new FakeSealGate(), + harness: new FakeHarness([{ postHash: opts.tree, tokensUsed: 111 }], scope), + makeLadder: () => + new FakeVerifier(opts.fail === true ? [failVerdict('child red')] : [passVerdict()]), + approver: new FakeApprover([approve()]), + workspace: scope, + clock: new ManualClock(), + budget: opts.budget, + runlog: new InMemoryRunLog(), + }; +} + +function runner(opts: { + host: FakeWorktreeHost; + canonical: FakeWorkspace; + composeChild: ComposeChild; +}): DefaultWaveRunner { + return new DefaultWaveRunner({ + host: opts.host, + workspace: opts.canonical, + workspaceRoot: '/fake/canonical', + composeChild: opts.composeChild, + }); +} + +/** Standard two-child fixture: canonical at eeeeeee; A edits to aaaa111, B to bbbb222. */ +function fixture(opts: { failB?: boolean } = {}): { + host: FakeWorktreeHost; + canonical: FakeWorkspace; + wave: DefaultWaveRunner; +} { + const canonical = new FakeWorkspace('eeeeeee'); + const host = new FakeWorktreeHost([], canonical); + const budget = new ManualBudgetMeter(false); + const composeChild: ComposeChild = async (spec, worktree) => + spec.index === 0 + ? childDeps({ worktree, contract: contractA, tree: 'aaaa111', budget }) + : childDeps({ worktree, contract: contractB, tree: 'bbbb222', budget, ...(opts.failB === true ? { fail: true } : {}) }); + return { host, canonical, wave: runner({ host, canonical, composeChild }) }; +} + +describe('DefaultWaveRunner — cooperative parallel waves (EXPERIMENTAL)', () => { + it('runs both children to DONE, merges in phase order, re-verifies, and checkpoints', async () => { + const { host, canonical, wave } = fixture(); + const result = await wave.run([specA, specB]); + + expect(result.outcomes.map((o) => o.kind)).toEqual(['merged', 'merged']); + // Merges are 3-way against the WAVE-START base, accumulating in phase order. + expect(host.mergedCalls).toEqual([ + { base: 'eeeeeee', ours: 'eeeeeee', theirs: 'aaaa111' }, + { base: 'eeeeeee', ours: 'eeeaaaa', theirs: 'bbbb222' }, + ]); + // The combined tree was promoted into the canonical workspace and checkpointed as the baseline. + expect(host.promoted).toEqual(['eeebbbb']); + expect(result.tree).toBe('eeebbbb'); + expect(await canonical.diffHash()).toBe('eeebbbb'); + // Every worktree torn down on the happy path. + expect(host.live.size).toBe(0); + // Child spend is surfaced for the parent's usage fold (shared budget already metered it). + expect(result.outcomes[0]!.usage?.tokens).toBe(111); + }); + + it('a merge CONFLICT downgrades that child to unmerged; the rest still land', async () => { + const { host, wave } = fixture(); + host.conflicts.add('eeeaaaa+bbbb222'); // B conflicts when merged onto A's result + const result = await wave.run([specA, specB]); + + expect(result.outcomes[0]).toMatchObject({ kind: 'merged', index: 0 }); + expect(result.outcomes[1]).toMatchObject({ kind: 'unmerged', index: 1 }); + if (result.outcomes[1]!.kind === 'unmerged') { + expect(result.outcomes[1]!.reason).toContain('merge conflict'); + } + // Only A's tree was promoted — nothing of B was applied (fail-closed). + expect(host.promoted).toEqual(['eeeaaaa']); + expect(host.live.size).toBe(0); + }); + + it('a child that cannot reach DONE is unmerged with its terminal status as the reason', async () => { + const { host, wave } = fixture({ failB: true }); + const result = await wave.run([specA, specB]); + + expect(result.outcomes[0]!.kind).toBe('merged'); + expect(result.outcomes[1]).toMatchObject({ kind: 'unmerged', index: 1 }); + if (result.outcomes[1]!.kind === 'unmerged') { + // The fake red ladder makes the child run terminate without both keys. + expect(result.outcomes[1]!.reason).toContain('child run'); + } + expect(host.live.size).toBe(0); + }); + + it('a RED post-merge re-verify downgrades that child — a merge is never trusted', async () => { + // The canonical workspace scripts the two re-verify rungs: A's `check-a` green, B's `check-b` red + // (the semantic-conflict case: two CLEAN merges that break each other). + const canonical = new FakeWorkspace('eeeeeee', '', [ + { exitCode: 0, stdout: '', stderr: '' }, + { exitCode: 1, stdout: '', stderr: 'check-b broke after the merge' }, + ]); + const host = new FakeWorktreeHost([], canonical); + const budget = new ManualBudgetMeter(false); + const composeChild: ComposeChild = async (spec, worktree) => + spec.index === 0 + ? childDeps({ worktree, contract: contractA, tree: 'aaaa111', budget }) + : childDeps({ worktree, contract: contractB, tree: 'bbbb222', budget }); + const wave = runner({ host, canonical, composeChild }); + + const result = await wave.run([specA, specB]); + expect(result.outcomes[0]!.kind).toBe('merged'); + expect(result.outcomes[1]).toMatchObject({ kind: 'unmerged', index: 1 }); + if (result.outcomes[1]!.kind === 'unmerged') { + expect(result.outcomes[1]!.reason).toContain('post-merge re-verify failed'); + expect(result.outcomes[1]!.reason).toContain('check-b broke after the merge'); + } + }); + + it('a composeChild failure is a fail-closed unmerged outcome, never a thrown wave', async () => { + const canonical = new FakeWorkspace('eeeeeee'); + const host = new FakeWorktreeHost([], canonical); + const budget = new ManualBudgetMeter(false); + const composeChild: ComposeChild = async (spec, worktree) => { + if (spec.index === 1) throw new Error('no deps for you'); + return childDeps({ worktree, contract: contractA, tree: 'aaaa111', budget }); + }; + const wave = runner({ host, canonical, composeChild }); + + const result = await wave.run([specA, specB]); + expect(result.outcomes[0]!.kind).toBe('merged'); + expect(result.outcomes[1]).toMatchObject({ kind: 'unmerged', index: 1 }); + if (result.outcomes[1]!.kind === 'unmerged') { + expect(result.outcomes[1]!.reason).toContain('no deps for you'); + } + expect(host.live.size).toBe(0); + }); +}); diff --git a/src/driver/wave-runner.ts b/src/driver/wave-runner.ts new file mode 100644 index 0000000..2eba5a7 --- /dev/null +++ b/src/driver/wave-runner.ts @@ -0,0 +1,311 @@ +import { randomUUID } from 'node:crypto'; +import { copyFile, mkdir } from 'node:fs/promises'; +import { dirname, resolve, sep } from 'node:path'; +import type { CompiledContract } from '../domain/contract'; +import type { TokenUsage } from '../domain/usage'; +import { asRunId, type DiffHash, type RunId } from '../domain/ids'; +import type { Workspace, Worktree, WorktreeHost } from '../workspace/workspace'; +import { DeterministicVerifier } from '../verify/deterministic'; +import type { Logger } from '../log/logger'; +import { noopLogger } from '../log/logger'; +import { drive, type DriverDeps } from './driver'; +import type { WavePhaseSpec, WaveOutcome, WaveResult, WaveRunner } from './wave'; + +/** + * Compose the FULL driver dependencies for one wave CHILD, rooted at its worktree. The composition + * root provides the real thing (harness/ladder/approver/runlog scoped to the worktree — see + * `makeWaveRunner` in compose.ts); tests inject fakes so the whole wave runs with zero LLM and zero + * subprocesses. `runId` names the child's OWN write-ahead log dir; `interrupted` is the parent's + * cooperative stop probe. Contract for implementers: the child's `budget` MUST be the PARENT's + * meter (the wave shares the run's one budget) and `interrupted` should be threaded through so + * Ctrl-C stops children cleanly between steps. + */ +export type ComposeChild = ( + spec: WavePhaseSpec, + worktree: Worktree, + runId: RunId, + interrupted?: () => boolean, +) => Promise; + +/** A child after the sequential preparation stage (worktree + deps), before its concurrent run. */ +type Prepared = { + readonly spec: WavePhaseSpec; + readonly worktree: Worktree | null; + readonly deps: DriverDeps | null; + readonly runId: RunId | null; + readonly reason: string | null; +}; + +/** What one finished child contributes to the merge stage. */ +type ChildResult = { + readonly spec: WavePhaseSpec; + readonly worktree: Worktree | null; + /** Set only when the child reached DONE (both keys) — the merge candidates. */ + readonly done: { + readonly tree: DiffHash; + readonly contract: CompiledContract | null; + } | null; + readonly reason: string | null; + readonly usage: TokenUsage | undefined; +}; + +/** + * EXPERIMENTAL — the real cooperative-wave executor (`--parallel-phases`). One `run()`: + * + * 1. **Fork.** Checkpoint the canonical tree (the merge BASE) and give each phase an isolated + * worktree + a full CHILD goaly run (`drive()` — its own frozen contract, iterations, two-key + * gate, and write-ahead log inside the worktree), all children concurrent on the SHARED budget. + * 2. **Merge.** In phase order, 3-way merge each DONE child's tree onto the accumulated result + * (`mergeTrees(base, acc, child)`), copying the child's compiler-authored verification files + * across (they are git-excluded, so no tree snapshot carries them). A textual conflict marks + * that child `unmerged` — nothing of it is applied. + * 3. **Promote + re-verify.** Promote the merged tree into the canonical workspace, then re-run + * each merged child's frozen DETERMINISTIC rungs against the combined tree — clean merges can + * still break each other semantically, and a merge is NEVER trusted. A red re-verify marks that + * child `unmerged` (its sub-goal re-runs sequentially on this very tree, so nothing is lost and + * nothing is greened). Judge rungs are not re-run here: each child already turned both keys in + * isolation, and the run's final ACCEPTANCE contract still gates the whole (two keys, LLM + * included) — the merged-tree guard is the ungameable deterministic bar in between. + * 4. **Checkpoint.** Snapshot the final canonical tree — the `WAVE_RAN.tree` baseline. + * + * Every failure shape degrades to `unmerged` (the classic sequential phase), never a throw out of + * `run()` for a per-child problem; the Driver additionally catches a wholesale throw and downgrades + * the entire wave. + */ +export class DefaultWaveRunner implements WaveRunner { + readonly #host: WorktreeHost; + readonly #workspace: Workspace; + readonly #workspaceRoot: string; + readonly #composeChild: ComposeChild; + readonly #verifyTimeoutMs: number | undefined; + readonly #log: Logger; + + constructor(opts: { + host: WorktreeHost; + /** The CANONICAL workspace (fork point, promotion target, and re-verify scope). */ + workspace: Workspace; + /** The canonical workspace's filesystem root (authored-file copy target). */ + workspaceRoot: string; + composeChild: ComposeChild; + /** Per-rung kill timeout for the post-merge deterministic re-verify (the run's verify cap). */ + verifyTimeoutMs?: number; + logger?: Logger; + }) { + this.#host = opts.host; + this.#workspace = opts.workspace; + this.#workspaceRoot = opts.workspaceRoot; + this.#composeChild = opts.composeChild; + this.#verifyTimeoutMs = opts.verifyTimeoutMs; + this.#log = opts.logger ?? noopLogger; + } + + async run(phases: readonly WavePhaseSpec[], interrupted?: () => boolean): Promise { + const base = await this.#workspace.checkpoint(); + this.#log.info('wave: forking children', { + phases: phases.map((p) => p.index).join(','), + base, + }); + + // Worktree creation is SEQUENTIAL (concurrent `git worktree add` calls contend on repo locks); + // only the child RUNS are concurrent. A preparation failure is already a fail-closed result. + const prepared: Prepared[] = []; + for (const spec of phases) prepared.push(await this.#prepareChild(spec, base, interrupted)); + const children = await Promise.all(prepared.map((p) => this.#driveChild(p))); + try { + const { merged, outcomes, tree } = await this.#mergeAndReverify(base, children); + this.#log.info('wave: merged + re-verified', { + merged: merged.length, + total: children.length, + tree, + }); + return { outcomes, tree }; + } finally { + for (const child of children) { + if (child.worktree !== null) await this.#host.removeWorktree(child.worktree); + } + } + } + + /** Create ONE child's worktree + deps (sequential stage). Never throws — a failure is a reason. */ + async #prepareChild( + spec: WavePhaseSpec, + base: DiffHash, + interrupted?: () => boolean, + ): Promise { + // The worktree handle survives a later failure so the teardown sweep still removes it. + let worktree: Worktree | null = null; + try { + worktree = await this.#host.addWorktree(base); + const runId = asRunId(`run-wave-p${spec.index}-${randomUUID()}`); + const deps = await this.#composeChild(spec, worktree, runId, interrupted); + return { spec, worktree, deps, runId, reason: null }; + } catch (e) { + return { + spec, + worktree, + deps: null, + runId: null, + reason: `child failed to start: ${e instanceof Error ? e.message : String(e)}`, + }; + } + } + + /** Drive ONE prepared child to a terminal outcome (concurrent stage). Never throws. */ + async #driveChild(prepared: Prepared): Promise { + const { spec, worktree, deps, runId } = prepared; + if (worktree === null || deps === null || runId === null) { + return { spec, worktree, done: null, reason: prepared.reason ?? 'child not prepared', usage: undefined }; + } + try { + this.#log.info('wave child starting', { phase: spec.index, runId, root: worktree.root }); + const outcome = await drive(deps, spec.config, runId); + const usage = outcome.usage?.total; + if (outcome.status !== 'DONE') { + return { + spec, + worktree, + done: null, + reason: `child run ${outcome.status}${outcome.reason !== undefined ? `: ${outcome.reason}` : ''}`, + usage, + }; + } + // The child's frozen contract (for the post-merge re-verify + authored-file copy) comes from + // ITS OWN write-ahead log. Fail-closed: no recoverable contract ⇒ unmerged — never an + // unverified merge. + const contract = await lastContract(deps); + if (contract === null) { + return { spec, worktree, done: null, reason: 'child log carried no frozen contract', usage }; + } + const tree = await worktree.scope.diffHash(); + return { spec, worktree, done: { tree, contract }, reason: null, usage }; + } catch (e) { + return { + spec, + worktree, + done: null, + reason: `child failed to run: ${e instanceof Error ? e.message : String(e)}`, + usage: undefined, + }; + } + } + + /** Stages 2–4: sequential merge (+ authored-file copy), promote, deterministic re-verify, checkpoint. */ + async #mergeAndReverify( + base: DiffHash, + children: readonly ChildResult[], + ): Promise<{ merged: ChildResult[]; outcomes: WaveOutcome[]; tree: DiffHash }> { + const ordered = [...children].sort((a, b) => a.spec.index - b.spec.index); + const outcomes: WaveOutcome[] = []; + const merged: ChildResult[] = []; + let acc: string = base; + + for (const child of ordered) { + const { index } = child.spec; + const usage = child.usage !== undefined ? { usage: child.usage } : {}; + if (child.done === null) { + outcomes.push({ kind: 'unmerged', index, reason: child.reason ?? 'child did not finish', ...usage }); + continue; + } + try { + const m = await this.#host.mergeTrees(base, acc, child.done.tree); + if (m.kind === 'conflict') { + this.#log.warn('wave: merge conflict — phase downgrades to sequential', { + phase: index, + detail: m.detail, + }); + outcomes.push({ kind: 'unmerged', index, reason: `merge conflict: ${m.detail}`, ...usage }); + continue; + } + acc = m.tree; + merged.push(child); + } catch (e) { + outcomes.push({ + kind: 'unmerged', + index, + reason: `merge failed: ${e instanceof Error ? e.message : String(e)}`, + ...usage, + }); + } + } + + if (merged.length > 0) { + await this.#host.promoteTree(acc); + // Authored verification files are git-excluded (never in a tree snapshot) — carry them over + // from each merged child's worktree so its frozen commands still have their inputs. + for (const child of merged) await this.#copyGeneratedFiles(child); + } + + // Re-verify each merged child's frozen deterministic rungs against the COMBINED tree. + for (const child of merged) { + const verdict = await this.#reverify(child); + const usage = child.usage !== undefined ? { usage: child.usage } : {}; + if (verdict === null) { + outcomes.push({ kind: 'merged', index: child.spec.index, ...usage }); + } else { + this.#log.warn('wave: post-merge re-verify red — phase downgrades to sequential', { + phase: child.spec.index, + detail: verdict, + }); + outcomes.push({ + kind: 'unmerged', + index: child.spec.index, + reason: `post-merge re-verify failed: ${verdict}`, + ...usage, + }); + } + } + + const tree = await this.#workspace.checkpoint(); + outcomes.sort((a, b) => a.index - b.index); + return { merged, outcomes, tree }; + } + + /** Run the child's frozen DETERMINISTIC rungs on the canonical tree; null = green, else the red detail. */ + async #reverify(child: ChildResult): Promise { + const contract = child.done?.contract; + if (contract === undefined || contract === null) return 'no frozen contract to re-verify'; + for (const rung of contract.rungs) { + if (rung.kind !== 'deterministic') continue; + const verifier = new DeterministicVerifier(rung.command, rung.label, this.#verifyTimeoutMs); + const verdict = await verifier.verify(this.#workspace, contract.goal, contract.rubric); + if (!verdict.pass) return verdict.detail; + } + return null; + } + + /** Copy a merged child's compiler-authored (git-excluded) verification files into the canonical root. */ + async #copyGeneratedFiles(child: ChildResult): Promise { + const contract = child.done?.contract; + const worktree = child.worktree; + if (contract === undefined || contract === null || worktree === null) return; + const canonicalRoot = resolve(this.#workspaceRoot); + for (const file of contract.generatedFiles) { + // Containment: the paths were validated at compile, but re-check before writing (fail-closed). + const src = resolve(worktree.root, file.path); + const dst = resolve(canonicalRoot, file.path); + if (!src.startsWith(resolve(worktree.root) + sep) || !dst.startsWith(canonicalRoot + sep)) { + throw new Error(`generated file escapes the workspace: ${file.path}`); + } + await mkdir(dirname(dst), { recursive: true }); + await copyFile(src, dst); + } + } +} + +/** The LAST frozen contract in a child's write-ahead log (revisions re-freeze; the last one ran). */ +async function lastContract(deps: DriverDeps): Promise { + try { + const stored = await deps.runlog.read(); + if (stored === null) return null; + let contract: CompiledContract | null = null; + for (const entry of stored.entries) { + if (entry.event.tag === 'CONTRACT_COMPILED') contract = entry.event.contract; + } + return contract; + } catch { + return null; + } +} + +/** Re-export the seam types so the composition root imports one module. */ +export type { WavePhaseSpec, WaveOutcome, WaveResult, WaveRunner }; diff --git a/src/driver/wave.ts b/src/driver/wave.ts new file mode 100644 index 0000000..4f4b0fc --- /dev/null +++ b/src/driver/wave.ts @@ -0,0 +1,35 @@ +import type { RunConfig } from '../domain/config'; +import type { OrchestratorEvent } from '../domain/events'; +import type { DiffHash } from '../domain/ids'; + +/** + * EXPERIMENTAL — the cooperative parallel-wave seam (`--parallel-phases`). The Driver performs a + * `RUN_WAVE` command through this interface and feeds the reducer ONE `WAVE_RAN` event; everything + * concurrent, git-shaped, or LLM-adjacent lives behind it (invariant #1). The real implementation + * (`src/cli/wave-runner.ts`) runs each phase as its own frozen, two-key CHILD goaly run in an + * isolated worktree, merges the DONE children in phase order, and RE-VERIFIES each merged phase's + * frozen ladder on the combined tree; tests inject fakes. Like every seam it must not reject in + * normal operation — per-child failures become `unmerged` outcomes (the fail-closed sequential + * downgrade); the Driver additionally catches a thrown runner and downgrades the WHOLE wave. + */ + +/** One wave member: the plan phase index + the phase config the reducer derived for it. */ +export type WavePhaseSpec = { readonly index: number; readonly config: RunConfig }; + +/** Per-phase wave outcome — exactly the shape persisted in the `WAVE_RAN` event. */ +export type WaveOutcome = Extract['outcomes'][number]; + +/** The whole wave's result: one outcome per member + the post-merge checkpoint tree. */ +export type WaveResult = { + readonly outcomes: WaveOutcome[]; + /** The post-merge checkpoint tree (the diff baseline for the phases that follow). */ + readonly tree: DiffHash; +}; + +export interface WaveRunner { + /** + * Run the wave. `interrupted` is the parent run's cooperative stop probe (Ctrl-C/SIGTERM) — + * threaded into every child's deps so children stop cleanly between steps like the parent does. + */ + run(phases: readonly WavePhaseSpec[], interrupted?: () => boolean): Promise; +} diff --git a/src/index.ts b/src/index.ts index 8403734..ac80f7a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -15,6 +15,9 @@ export { type CheckpointDeps, } from './driver/driver'; export { noopTelemetry, type Telemetry, type TelemetryEvent } from './telemetry/telemetry'; +// EXPERIMENTAL cooperative parallel waves (--parallel-phases): the seam + the composable executor. +export type { WaveRunner, WavePhaseSpec, WaveOutcome, WaveResult } from './driver/wave'; +export { DefaultWaveRunner, type ComposeChild } from './driver/wave-runner'; export { composeDeps, buildLadder, diff --git a/src/llm/agent-cli-provider.ts b/src/llm/agent-cli-provider.ts index f4d169e..a7e0b3f 100644 --- a/src/llm/agent-cli-provider.ts +++ b/src/llm/agent-cli-provider.ts @@ -4,7 +4,7 @@ import { runProcess } from '../util/spawn'; import { parseAgentOutput } from '../agent-cli/output'; import { StreamTap, type AgentEventSink } from '../agent-cli/stream'; import { accountTokens, streamingEstimator } from '../agent-cli/estimate'; -import type { AgentCliCodec } from '../agent-cli/codec'; +import { ambientSessionId, type AgentCliCodec } from '../agent-cli/codec'; /** * Injectable subprocess seam: takes the full argv plus the prompt (delivered on stdin only when the @@ -25,21 +25,6 @@ const BACKOFF_MS = 1000; const realSleep = (ms: number): Promise => new Promise((r) => setTimeout(r, ms)); -/** - * The AMBIENT session id when goaly itself runs nested under Claude Code (e.g. inside a Claude Code - * remote environment). A spawned `claude -p` there adopts and REPORTS this id instead of minting a - * fresh per-call session, and every call in a cwd appends to that ONE shared session file — so - * resuming it would replay sibling steps' turns (the shape classifier, the red-team critics, …) - * into the authoring context, not the author's own conversation. Observed empirically; scrubbing - * the variable from the child env does NOT stop the pinning (the wrapped CLI keeps it), so the - * only safe policy is to never TRUST the ambient id: drop it from completions and refuse it as a - * resume target — authoring then degrades to fresh full-prompt calls, the pre-feature behavior. - */ -function ambientSessionId(): string | undefined { - const v = process.env['CLAUDE_CODE_SESSION_ID']; - return v !== undefined && v.length > 0 ? v : undefined; -} - /** * The ONE {@link LlmProvider} backed by a coding-agent CLI, driven entirely by that CLI's * {@link AgentCliCodec}. The judge / approver / compiler use a CLI's model in a READ-ONLY dialect diff --git a/src/orchestrator/decide.test.ts b/src/orchestrator/decide.test.ts index af41b0c..8013cd8 100644 --- a/src/orchestrator/decide.test.ts +++ b/src/orchestrator/decide.test.ts @@ -18,12 +18,12 @@ describe('DECIDE truth table', () => { it('ladder fail → CONTINUE with verifier detail as feedback', () => { const d = decide(makeCtx(), failVerdict('tests are red'), null); - expect(d).toEqual({ kind: 'CONTINUE', feedback: 'tests are red' }); + expect(d).toEqual({ kind: 'CONTINUE', feedback: 'tests are red', source: 'verifier' }); }); it('ladder pass + veto → CONTINUE with veto reason as feedback', () => { const d = decide(makeCtx(), passVerdict(), veto('the test is empty')); - expect(d).toEqual({ kind: 'CONTINUE', feedback: 'the test is empty' }); + expect(d).toEqual({ kind: 'CONTINUE', feedback: 'the test is empty', source: 'veto' }); }); it('DONE wins over maxIterations (goal met on the last allowed iteration)', () => { @@ -64,21 +64,53 @@ describe('DECIDE truth table', () => { // has not yet seen — it must get one real turn to act on the (correct) critique first. const ctx = makeCtx({ iteration: 1, lastNoDiff: true, feedback: undefined }); const d = decide(ctx, passVerdict(), veto('power-ups are inert')); - expect(d).toEqual({ kind: 'CONTINUE', feedback: 'power-ups are inert' }); + expect(d).toEqual({ kind: 'CONTINUE', feedback: 'power-ups are inert', source: 'veto' }); }); it('a SECOND unproductive no-diff on the same veto → ABORTED (issue #54)', () => { - // The agent was already told this veto last turn (feedback === reason) and still made no edits. - const ctx = makeCtx({ iteration: 2, lastNoDiff: true, feedback: 'power-ups are inert' }); + // The agent was already told this veto last turn (feedbackSource 'veto') and still made no edits. + const ctx = makeCtx({ + iteration: 2, + lastNoDiff: true, + feedback: 'power-ups are inert', + feedbackSource: 'veto', + }); const d = decide(ctx, passVerdict(), veto('power-ups are inert')); expect(d.kind).toBe('ABORTED'); if (d.kind === 'ABORTED') expect(d.reason).toContain('no-diff'); }); + it('a REWORDED veto after an unproductive no-diff still ABORTS — the excuse is per source, not per wording', () => { + // Regression: an LLM approver rewords its veto every round. If freshness were keyed on the + // reason TEXT, every veto would look fresh and a worker that never edits would burn the whole + // iteration budget in approver spend. The excuse must not renew just because the words changed. + const ctx = makeCtx({ + iteration: 2, + lastNoDiff: true, + feedback: 'the diff is empty and the goal is vague', + feedbackSource: 'veto', + }); + const d = decide(ctx, passVerdict(), veto('no evidence of any actual change was provided')); + expect(d.kind).toBe('ABORTED'); + if (d.kind === 'ABORTED') expect(d.reason).toContain('no-diff'); + }); + + it('a no-diff on a veto AFTER verifier-red feedback is excused — the just-run turn was not answering a veto', () => { + // The turn that produced no diff was chewing on a red-ladder detail; the veto is genuinely new. + const ctx = makeCtx({ + iteration: 2, + lastNoDiff: true, + feedback: 'tests are red', + feedbackSource: 'verifier', + }); + const d = decide(ctx, passVerdict(), veto('power-ups are inert')); + expect(d).toEqual({ kind: 'CONTINUE', feedback: 'power-ups are inert', source: 'veto' }); + }); + it('a turn killed by timeout does not immediately trip no-diff (issue #54)', () => { const ctx = makeCtx({ iteration: 1, lastNoDiff: true, lastRunStatus: 'timeout' }); const d = decide(ctx, failVerdict('still red'), null); - expect(d).toEqual({ kind: 'CONTINUE', feedback: 'still red' }); + expect(d).toEqual({ kind: 'CONTINUE', feedback: 'still red', source: 'verifier' }); }); it('stuck (oscillation) → ABORTED', () => { diff --git a/src/orchestrator/decide.ts b/src/orchestrator/decide.ts index e7161cd..0c9bb29 100644 --- a/src/orchestrator/decide.ts +++ b/src/orchestrator/decide.ts @@ -7,7 +7,7 @@ import { detectStuck } from './stuck'; * commands. No LLM, no IO, no clock. */ export type Decision = - | { kind: 'CONTINUE'; feedback: string } + | { kind: 'CONTINUE'; feedback: string; source: 'verifier' | 'veto' } | { kind: 'DONE' } | { kind: 'FAILED'; reason: string } | { kind: 'ABORTED'; reason: string }; @@ -55,23 +55,27 @@ export function decide( // Continue: feed back the verifier detail (failed ladder) or the veto reason. if (!ladder.pass) { - return { kind: 'CONTINUE', feedback: ladder.detail }; + return { kind: 'CONTINUE', feedback: ladder.detail, source: 'verifier' }; } // ladder.pass && veto return { kind: 'CONTINUE', feedback: approval?.reason ?? 'rejected by the approval gate', + source: 'veto', }; } /** * The in-flight half of the no-diff excuse (issue #54): a green ladder blocked only by a FRESH - * Sign-off veto — one whose reason differs from the feedback the just-run turn was already given - * (`ctx.feedback`) — so the worker has not yet had a real turn to act on it. One-shot by construction: - * once that veto reason becomes the prior feedback, it no longer differs and the no-diff abort trips. + * Sign-off veto — one the just-run turn was NOT already answering (`ctx.feedbackSource !== 'veto'`) + * — so the worker has not yet had a real turn to act on a veto-class critique. One-shot by + * construction: the excused turn's feedback is recorded as `source: 'veto'`, so a second + * consecutive no-diff-on-veto aborts. Keyed on the feedback's SOURCE, not its text: an LLM + * approver rewords its veto every round, so a reason-string comparison would classify every veto + * as fresh and let a worker that never edits burn the whole iteration budget in approver spend. * Pure; lives in DECIDE because it needs the in-flight `ladder`/`approval` the reducer is deciding on. */ function freshVeto(ctx: LoopCtx, ladder: Verdict, approval: ApprovalVerdict | null): boolean { if (ladder.pass !== true || approval?.veto !== true) return false; - return (approval.reason ?? '') !== (ctx.feedback ?? ''); + return ctx.feedbackSource !== 'veto'; } diff --git a/src/orchestrator/state.ts b/src/orchestrator/state.ts index 780ceff..0d291aa 100644 --- a/src/orchestrator/state.ts +++ b/src/orchestrator/state.ts @@ -20,6 +20,18 @@ export type PhaseCtx = { readonly plan: PhasePlan; /** 0-based phase index; `plan.phases.length` denotes the final cumulative acceptance phase. */ readonly index: number; + /** + * EXPERIMENTAL parallel waves: phase indices already COMPLETED by a merged-and-reverified wave + * child — the sequential advance skips them. Absent on a classic/sequential run (no field, so + * every existing PhaseCtx construction and equality stays byte-for-byte). + */ + readonly skip?: readonly number[]; + /** + * EXPERIMENTAL parallel waves: phase indices whose wave fan-out was already ATTEMPTED. A phase in + * this list never re-fans-out — an unmerged member re-runs as a classic sequential phase (the + * fail-closed downgrade), so a fully-conflicted wave can never fan out forever. + */ + readonly waved?: readonly number[]; }; /** @@ -61,6 +73,14 @@ export type LoopCtx = { readonly lastVerdict: Verdict | undefined; /** Feedback text threaded into the next agent prompt. */ readonly feedback: string | undefined; + /** + * Where the current `feedback` came from: `'verifier'` (a red ladder's detail) or `'veto'` (a + * green-ladder Sign-off veto reason). Drives the one-shot no-diff excuse (issue #54): a no-diff + * iteration is excused for a veto only when the just-run turn was NOT already answering a veto — + * an LLM approver rewords its veto every round, so comparing reason strings would renew the + * excuse forever and burn maxIterations of approver spend on a worker that never edits. + */ + readonly feedbackSource: 'verifier' | 'veto' | undefined; /** * The phase position within a frozen plan (issue #48), or undefined on a classic single-contract * run. When set, a phase reaching both keys ADVANCES (checkpoint + next phase's compile) instead of @@ -125,6 +145,17 @@ export type OrchestratorState = /** The phase position when this prepare belongs to a phased run (issue #48); else undefined. */ readonly phase?: PhaseCtx; } + | { + /** + * EXPERIMENTAL — a cooperative parallel WAVE is in flight (`--parallel-phases`): the Driver is + * running the grouped phases at `indices` as concurrent, isolated, frozen two-key CHILD runs, + * then merging + re-verifying. Resolved by ONE `WAVE_RAN` event. Carries the wave's FIRST + * member's PhaseCtx (`phase.index === indices[0]`). + */ + readonly tag: 'RUNNING_WAVE'; + readonly phase: PhaseCtx; + readonly indices: readonly number[]; + } | { readonly tag: 'RUNNING_AGENT'; readonly ctx: LoopCtx } | { readonly tag: 'VERIFYING'; readonly ctx: LoopCtx } | { readonly tag: 'AWAIT_SIGNOFF'; readonly ctx: LoopCtx } @@ -172,6 +203,8 @@ export function iterationCount(state: OrchestratorState): number { case 'PREPARING': case 'PLANNING': case 'AWAIT_PLAN_SEAL': + // A wave's iterations belong to its CHILD runs (each has its own log); the parent counts none. + case 'RUNNING_WAVE': return 0; } } @@ -204,6 +237,7 @@ export function initialCtx( lastBudget: undefined, lastVerdict: undefined, feedback: undefined, + feedbackSource: undefined, phase, }; } diff --git a/src/orchestrator/step.ts b/src/orchestrator/step.ts index bc49d50..44a0e77 100644 --- a/src/orchestrator/step.ts +++ b/src/orchestrator/step.ts @@ -3,6 +3,7 @@ import type { RunConfig, VerifierIntent } from '../domain/config'; import { pickGatePolicy, pickLoopPolicy, pickDriverWiring } from '../domain/config'; import type { CompiledContract, Rung } from '../domain/contract'; import type { PhasePlan } from '../domain/plan'; +import { waveIndicesAt } from '../domain/plan'; import type { OrchestratorState, LoopCtx, PhaseCtx } from './state'; import { initialCtx } from './state'; import { decide, type Decision } from './decide'; @@ -42,6 +43,8 @@ export function step(state: OrchestratorState, event: OrchestratorEvent): StepRe return stepAwaitPlanSeal(state.config, state.plan, state.reviseRound, event); case 'ADVANCING_PHASE': return stepAdvancingPhase(state.phase, event); + case 'RUNNING_WAVE': + return stepRunningWave(state.phase, state.indices, event); case 'COMPILING': return stepCompiling(state.config, state.reviseRound, state.compileRound, state.phase, event); case 'AWAIT_SEAL': @@ -146,12 +149,39 @@ function stepAwaitPlanSeal( */ function stepAdvancingPhase(phase: PhaseCtx, event: OrchestratorEvent): StepResult { if (event.tag !== 'PHASE_ADVANCED') throw invalidTransition('ADVANCING_PHASE', event); - const next: PhaseCtx = { ...phase, index: phase.index + 1 }; + const next: PhaseCtx = { ...phase, index: nextPhaseIndex(phase, phase.index + 1) }; return startPhaseCompile(next); } -/** Begin a phase: COMPILING its derived config, carrying the phase position for the eventual advance. */ +/** The next phase index at or after `from`, skipping indices a wave already completed (merged). */ +function nextPhaseIndex(phase: PhaseCtx, from: number): number { + const skip = phase.skip ?? []; + let next = from; + while (skip.includes(next)) next += 1; + return next; +} + +/** + * Begin a phase: COMPILING its derived config, carrying the phase position for the eventual advance. + * EXPERIMENTAL parallel waves: when the phase heads a not-yet-attempted group of consecutive + * same-`group` sub-goals AND `--parallel-phases` is on, the whole group is emitted as ONE `RUN_WAVE` + * command instead (still exactly one command per state — the Driver invariant). Everything else — + * ungrouped plans, the acceptance phase, a re-entered (already-attempted) member, the feature off — + * takes the classic sequential compile, byte-for-byte. + */ function startPhaseCompile(phase: PhaseCtx): StepResult { + const wave = pendingWaveAt(phase); + if (wave.length > 1) { + return [ + { tag: 'RUNNING_WAVE', phase, indices: wave }, + [ + { + tag: 'RUN_WAVE', + phases: wave.map((index) => ({ index, config: phaseConfigFor({ ...phase, index }) })), + }, + ], + ]; + } const config = phaseConfigFor(phase); return [ { tag: 'COMPILING', config, reviseRound: 0, compileRound: 0, phase }, @@ -159,6 +189,45 @@ function startPhaseCompile(phase: PhaseCtx): StepResult { ]; } +/** + * The wave the current phase would fan out, or a singleton when it must run sequentially: the + * feature is off, the index is the acceptance phase, the group was ALREADY attempted (`waved` — an + * unmerged member re-runs sequentially, never re-fans-out), or the group has one live member. + */ +function pendingWaveAt(phase: PhaseCtx): readonly number[] { + if (!phase.baseConfig.parallelPhases) return [phase.index]; + if (phase.index >= phase.plan.phases.length) return [phase.index]; + if ((phase.waved ?? []).includes(phase.index)) return [phase.index]; + const skip = phase.skip ?? []; + return waveIndicesAt(phase.plan, phase.index).filter((i) => !skip.includes(i)); +} + +/** + * EXPERIMENTAL parallel waves: fold the ONE `WAVE_RAN` event. `merged` members are recorded in + * `skip` (complete — the advance walks past them); every attempted index is recorded in `waved` + * (never re-fans-out); the machine advances to the FIRST not-merged wave member — a classic + * sequential re-run on the merged tree (the fail-closed downgrade) — or past the group when all + * merged. The plan, the contracts, and the two-key gate are untouched: an unmerged phase re-enters + * the same compile → Seal → loop path any sequential phase takes. + */ +function stepRunningWave( + phase: PhaseCtx, + indices: readonly number[], + event: OrchestratorEvent, +): StepResult { + if (event.tag !== 'WAVE_RAN') throw invalidTransition('RUNNING_WAVE', event); + // Only indices that were actually part of this wave count (defense in depth on a replayed log). + const merged = event.outcomes + .filter((o) => o.kind === 'merged' && indices.includes(o.index)) + .map((o) => o.index); + const next: PhaseCtx = { + ...phase, + skip: [...(phase.skip ?? []), ...merged], + waved: [...(phase.waved ?? []), ...indices], + }; + return startPhaseCompile({ ...next, index: nextPhaseIndex(next, indices[0] ?? phase.index) }); +} + /** * Derive the RunConfig for a phase from the frozen plan + the original config. A sub-goal phase * (`index < phases.length`) inherits the operational knobs (iterations, budget, stuck policy, @@ -171,7 +240,7 @@ function startPhaseCompile(phase: PhaseCtx): StepResult { function phaseConfigFor(phase: PhaseCtx): RunConfig { const base = phase.baseConfig; if (phase.index >= phase.plan.phases.length) { - return { ...base, phased: false }; + return { ...base, phased: false, parallelPhases: false }; } // A sub-goal phase: FRESH contract inputs authored per sub-goal (goal/verifier/rubric), but the // SAME operational policy as the run — inherited wholesale by lifetime VIEW (gate / loop / wiring) @@ -196,6 +265,9 @@ function phaseConfigFor(phase: PhaseCtx): RunConfig { // frozen into the contract, so each phase's Sign-off uses the same panel.) ...(sub.rubric !== undefined ? { rubric: sub.rubric } : {}), phased: false, + // A phase (whether run inline or as a wave CHILD) is a single-contract run — it must never + // decompose or fan out again (no nested waves). + parallelPhases: false, }; } @@ -527,7 +599,7 @@ function stepAwaitSignoff(ctx: LoopCtx, event: OrchestratorEvent): StepResult { function applyDecision(ctx: LoopCtx, decision: Decision): StepResult { switch (decision.kind) { case 'CONTINUE': { - const next: LoopCtx = { ...ctx, feedback: decision.feedback }; + const next: LoopCtx = { ...ctx, feedback: decision.feedback, feedbackSource: decision.source }; const prompt = buildLoopPrompt(ctx.contract, decision.feedback, ctx.lastRunStatus); return startIteration(next, prompt, ctx.sessionId); } diff --git a/src/orchestrator/step.wave.test.ts b/src/orchestrator/step.wave.test.ts new file mode 100644 index 0000000..582f317 --- /dev/null +++ b/src/orchestrator/step.wave.test.ts @@ -0,0 +1,171 @@ +import { describe, it, expect } from 'vitest'; +import { initial, step } from './step'; +import type { OrchestratorState } from './state'; +import type { OrchestratorEvent, BudgetSnapshot } from '../domain/events'; +import { makeConfig, makeFakeContract, makeFakePlan, passVerdict, dh } from '../testing/fakes'; + +const budget: BudgetSnapshot = { exceeded: false }; +/** Phases 0+1 share wave group 1; phase 2 is sequential. */ +const plan = makeFakePlan({ + phases: [ + { goal: 'wave member A', group: 1 }, + { goal: 'wave member B', group: 1 }, + { goal: 'sequential tail' }, + ], +}); +const contract = makeFakeContract(); + +function agentRan(prev: string, post: string): OrchestratorEvent { + const [p, q] = dh(prev, post); + return { + tag: 'AGENT_RAN', + run: { output: '', sessionId: 'sess-1' as never, status: 'completed' }, + prevDiffHash: p!, + diffHash: q!, + budget, + }; +} + +/** A phased+parallel run folded up to the plan-Seal approve (the wave decision point). */ +function approvedPlan(parallel: boolean): readonly [OrchestratorState, readonly unknown[]] { + const config = makeConfig({ phased: true, parallelPhases: parallel, autonomous: true }); + const [s0] = initial(config); + const [s1] = step(s0, { tag: 'PLAN_COMPILED', plan }); + return step(s1, { tag: 'PLAN_SEAL_DECIDED', decision: { kind: 'approve' } }); +} + +/** Drive a compiling phase through Seal → run → verify(pass) → sign-off(approve). */ +function runPhaseToBothKeys(compiling: OrchestratorState): OrchestratorState { + const [sealed] = step(compiling, { tag: 'CONTRACT_COMPILED', contract }); + const [running] = step(sealed, { tag: 'SEAL_DECIDED', decision: { kind: 'approve' } }); + const [verifying] = step(running, agentRan('0000000', '0000aaa')); + const [awaitSignoff] = step(verifying, { tag: 'VERIFIED', verdict: passVerdict() }); + return step(awaitSignoff, { tag: 'SIGNOFF_DECIDED', approval: { veto: false } })[0]; +} + +const waveTree = dh('00cafe0')[0]!; + +describe('parallel waves reducer (EXPERIMENTAL --parallel-phases)', () => { + it('plan approve fans a grouped prefix out as ONE RUN_WAVE with per-phase derived configs', () => { + const [state, cmds] = approvedPlan(true); + expect(state.tag).toBe('RUNNING_WAVE'); + if (state.tag === 'RUNNING_WAVE') expect(state.indices).toEqual([0, 1]); + expect(cmds).toHaveLength(1); // driver invariant: exactly one command per state + const cmd = cmds[0] as Extract; + expect(cmd.tag).toBe('RUN_WAVE'); + expect(cmd.phases.map((p) => p.index)).toEqual([0, 1]); + expect(cmd.phases[0]!.config.goal).toBe('wave member A'); + expect(cmd.phases[1]!.config.goal).toBe('wave member B'); + // Each wave child is a normal single-contract, non-fanning run authored per sub-goal. + for (const p of cmd.phases) { + expect(p.config.verifier.kind).toBe('generate'); + expect(p.config.phased).toBe(false); + expect(p.config.parallelPhases).toBe(false); + } + }); + + it('the feature is OPT-IN: a grouped plan without --parallel-phases runs strictly sequentially', () => { + const [state, cmds] = approvedPlan(false); + expect(state.tag).toBe('COMPILING'); + if (state.tag === 'COMPILING') expect(state.config.goal).toBe('wave member A'); + expect(cmds[0]).toMatchObject({ tag: 'COMPILE_VERIFIER' }); + }); + + it('all members merged → the machine advances PAST the group to the next phase', () => { + const [wave] = approvedPlan(true); + const [next, cmds] = step(wave, { + tag: 'WAVE_RAN', + outcomes: [ + { kind: 'merged', index: 0 }, + { kind: 'merged', index: 1 }, + ], + tree: waveTree, + }); + expect(next.tag).toBe('COMPILING'); + if (next.tag === 'COMPILING') { + expect(next.config.goal).toBe('sequential tail'); + expect(next.phase).toMatchObject({ index: 2, skip: [0, 1] }); + } + expect(cmds[0]).toMatchObject({ tag: 'COMPILE_VERIFIER' }); + }); + + it('a partially-merged wave re-runs ONLY the unmerged member sequentially, then skips the merged one', () => { + const [wave] = approvedPlan(true); + const [fallback] = step(wave, { + tag: 'WAVE_RAN', + outcomes: [ + { kind: 'merged', index: 0 }, + { kind: 'unmerged', index: 1, reason: 'merge conflict: file.txt' }, + ], + tree: waveTree, + }); + // The unmerged member re-enters the CLASSIC sequential path (fresh compile, same sub-goal). + expect(fallback.tag).toBe('COMPILING'); + if (fallback.tag === 'COMPILING') { + expect(fallback.config.goal).toBe('wave member B'); + expect(fallback.phase).toMatchObject({ index: 1, skip: [0], waved: [0, 1] }); + } + // When it completes both keys, the advance walks past the group to the tail — never back to 0. + const advancing = runPhaseToBothKeys(fallback); + expect(advancing.tag).toBe('ADVANCING_PHASE'); + const [tail] = step(advancing, { tag: 'PHASE_ADVANCED', tree: waveTree }); + expect(tail.tag).toBe('COMPILING'); + if (tail.tag === 'COMPILING') expect(tail.config.goal).toBe('sequential tail'); + }); + + it('a fully-unmerged wave NEVER re-fans-out — every member downgrades to sequential', () => { + const [wave] = approvedPlan(true); + const [first] = step(wave, { + tag: 'WAVE_RAN', + outcomes: [ + { kind: 'unmerged', index: 0, reason: 'child run FAILED' }, + { kind: 'unmerged', index: 1, reason: 'child run FAILED' }, + ], + tree: waveTree, + }); + expect(first.tag).toBe('COMPILING'); // sequential, NOT another RUNNING_WAVE + if (first.tag === 'COMPILING') expect(first.config.goal).toBe('wave member A'); + + const advancing = runPhaseToBothKeys(first); + const [second] = step(advancing, { tag: 'PHASE_ADVANCED', tree: waveTree }); + expect(second.tag).toBe('COMPILING'); // member B also sequential — the `waved` guard holds + if (second.tag === 'COMPILING') expect(second.config.goal).toBe('wave member B'); + }); + + it('a wave covering the LAST sub-goals advances into the cumulative ACCEPTANCE phase', () => { + const twoPhase = makeFakePlan({ + phases: [ + { goal: 'wave member A', group: 7 }, + { goal: 'wave member B', group: 7 }, + ], + }); + const config = makeConfig({ phased: true, parallelPhases: true, autonomous: true }); + const [s0] = initial(config); + const [s1] = step(s0, { tag: 'PLAN_COMPILED', plan: twoPhase }); + const [wave] = step(s1, { tag: 'PLAN_SEAL_DECIDED', decision: { kind: 'approve' } }); + expect(wave.tag).toBe('RUNNING_WAVE'); + const [accept] = step(wave, { + tag: 'WAVE_RAN', + outcomes: [ + { kind: 'merged', index: 0 }, + { kind: 'merged', index: 1 }, + ], + tree: waveTree, + }); + expect(accept.tag).toBe('COMPILING'); + if (accept.tag === 'COMPILING') { + // The acceptance phase is the ORIGINAL goal (decomposition can't green a broken whole). + expect(accept.config.goal).toBe(config.goal); + expect(accept.phase).toMatchObject({ index: 2 }); + } + }); + + it('ungrouped plans and the acceptance phase never fan out', () => { + const linear = makeFakePlan({ phases: [{ goal: 'only phase' }] }); + const config = makeConfig({ phased: true, parallelPhases: true, autonomous: true }); + const [s0] = initial(config); + const [s1] = step(s0, { tag: 'PLAN_COMPILED', plan: linear }); + const [state] = step(s1, { tag: 'PLAN_SEAL_DECIDED', decision: { kind: 'approve' } }); + expect(state.tag).toBe('COMPILING'); + }); +}); diff --git a/src/plan/plan.test.ts b/src/plan/plan.test.ts index ceed953..81ac99c 100644 --- a/src/plan/plan.test.ts +++ b/src/plan/plan.test.ts @@ -4,6 +4,7 @@ import { StaticPlanner } from './static-planner'; import { AutoPlanGate, HumanPlanGate } from './plan-gates'; import { FakeLlm } from '../llm/provider'; import { freezePlan, hashPlan } from '../util/hash'; +import { canonicalPlanString, waveIndicesAt } from '../domain/plan'; import { makeConfig } from '../testing/fakes'; const config = makeConfig({ phased: true, goal: 'build a CLI', maxPhases: 5 }); @@ -22,6 +23,48 @@ describe('freezePlan / hashPlan (issue #48)', () => { expect(frozen.planHash).toBe(hashPlan({ phases: [{ goal: 'only' }] })); expect(frozen.phases).toHaveLength(1); }); + + it('a wave `group` is FROZEN into the hash, and groupless plans keep their legacy hash (back-compat)', () => { + // Grouping is part of the frozen plan — re-shuffling it would be a different plan. + const grouped = hashPlan({ phases: [{ goal: 'x', group: 1 }, { goal: 'y', group: 1 }] }); + const ungrouped = hashPlan({ phases: [{ goal: 'x' }, { goal: 'y' }] }); + const regrouped = hashPlan({ phases: [{ goal: 'x', group: 1 }, { goal: 'y', group: 2 }] }); + expect(grouped).not.toBe(ungrouped); + expect(grouped).not.toBe(regrouped); + // Back-compat: a plan WITHOUT groups canonicalizes exactly as before the field existed, so every + // pre-existing run log's planHash still matches on replay. + expect(canonicalPlanString({ phases: [{ goal: 'x' }] })).toBe( + JSON.stringify({ phases: [{ goal: 'x', intent: null, rubric: null }] }), + ); + }); +}); + +describe('waveIndicesAt — consecutive same-group members (EXPERIMENTAL parallel waves)', () => { + const plan = freezePlan({ + phases: [ + { goal: 'a', group: 1 }, + { goal: 'b', group: 1 }, + { goal: 'c' }, + { goal: 'd', group: 2 }, + ], + }); + + it('the group head fans out over its consecutive members', () => { + expect(waveIndicesAt(plan, 0)).toEqual([0, 1]); + }); + + it('a MID-group index never fans out (a sequential fallback walks members one at a time)', () => { + expect(waveIndicesAt(plan, 1)).toEqual([1]); + }); + + it('an ungrouped phase and a singleton group are singletons', () => { + expect(waveIndicesAt(plan, 2)).toEqual([2]); + expect(waveIndicesAt(plan, 3)).toEqual([3]); + }); + + it('an out-of-range index is a singleton (the acceptance phase)', () => { + expect(waveIndicesAt(plan, 4)).toEqual([4]); + }); }); describe('AgentPlanner — LLM-authored plan (issue #48)', () => { diff --git a/src/runlog/replay.test.ts b/src/runlog/replay.test.ts index cb5b523..c310c08 100644 --- a/src/runlog/replay.test.ts +++ b/src/runlog/replay.test.ts @@ -257,6 +257,14 @@ describe('replay() — RUN_EXTENDED (operator extension, ADR 0012)', () => { expect(cfg.stuckPolicy.oscillation).toBe(true); // untouched fields keep their values }); + it('a candidates extension overlays the best-of-N fan-out (NL delegation at resume)', () => { + const cfg = extendedRunConfig(makeConfig({ maxIterations: 5 }), [ + extensionEntry(1, { candidates: 4, note: 'try 4 parallel attempts' }), + ]); + expect(cfg.candidates).toBe(4); + expect(cfg.maxIterations).toBe(5); // untouched fields keep their values + }); + it('a raised maxIterations UN-TERMINATES a FAILED-at-cap fold (the run continues)', async () => { const runlog = await driveToIterationCap(1); const stored = await runlog.read(); diff --git a/src/runlog/replay.ts b/src/runlog/replay.ts index 5215e81..319e1e6 100644 --- a/src/runlog/replay.ts +++ b/src/runlog/replay.ts @@ -30,6 +30,7 @@ export function applyRunExtension(cfg: RunConfig, x: RunExtension): RunConfig { return { ...cfg, ...(x.maxIterations !== undefined ? { maxIterations: x.maxIterations } : {}), + ...(x.candidates !== undefined ? { candidates: x.candidates } : {}), budget: { ...cfg.budget, ...(x.budgetTokens !== undefined ? { tokens: x.budgetTokens } : {}), @@ -167,6 +168,12 @@ export function replay(config: RunConfig, entries: readonly RunLogEntry[]): Repl baseline = entry.event.tree; phaseBaseline = entry.event.tree; } + // EXPERIMENTAL parallel waves: like PHASE_ADVANCED, a wave both DRIVES the reducer (skip/advance + // bookkeeping) and records the post-merge checkpoint tree for baseline reconstruction on resume. + if (entry.event.tag === 'WAVE_RAN') { + baseline = entry.event.tree; + phaseBaseline = entry.event.tree; + } // With extended budget caps, the persisted `exceeded` flags are re-judged against the new caps // (raw spent numbers stay the persisted facts) — else the fold would re-abort at the old cap. [state, commands] = step(state, budgetExtended ? rejudgeBudget(entry.event, effective) : entry.event); diff --git a/src/runlog/session-id.ts b/src/runlog/session-id.ts index 5db2fe6..30f2a23 100644 --- a/src/runlog/session-id.ts +++ b/src/runlog/session-id.ts @@ -1,32 +1,10 @@ import type { SessionId } from '../domain/ids'; import type { RunLogEntry } from './runlog'; -/** - * The synthesized SENTINEL session ids the adapters/driver mint when no REAL id could be recovered - * from the agent CLI. They are valid `SessionId`s on the wire (so the event still parses) but mean - * "no resumable session" — threading one into `claude --resume ` / a goaly-code session reload - * would point at nothing. The follow-up resume-hint (Capability A) and session inheritance - * (Capability C) must skip them and recover the last id that actually came back from the CLI. - * - * Kept in ONE place so the codec sentinels (`-unknown`), the NoopHarness sentinel - * (`noop-session`), the workspace-error sentinel (`workspace-error`), and the generic coerce - * fallback (`unknown-session`) can never drift from this skip-list. - */ -export const SENTINEL_SESSION_IDS: ReadonlySet = new Set([ - 'unknown-session', // coerceSessionId default fallback - 'noop-session', // NoopHarness - 'workspace-error', // driver: a workspace (diffHash) failure synthesizes a crashed run - 'claude-unknown', - 'codex-unknown', - 'droid-unknown', - 'pi-unknown', - 'goaly-code-unknown', -]); - -/** Whether `id` is a synthesized sentinel rather than a real, resumable harness session id. */ -export function isSentinelSession(id: string): boolean { - return SENTINEL_SESSION_IDS.has(id); -} +// The sentinel skip-list lives in the id DOMAIN (src/domain/ids.ts) so the harness core can refuse +// sentinels at the resume seam without importing persistence; re-exported here for its consumers. +export { SENTINEL_SESSION_IDS, isSentinelSession } from '../domain/ids'; +import { isSentinelSession } from '../domain/ids'; /** * Walk the `AGENT_RAN` entries BACKWARDS to the last REAL session id — the most recent agent turn diff --git a/src/runlog/usage.ts b/src/runlog/usage.ts index 5ce2965..2bbd6e7 100644 --- a/src/runlog/usage.ts +++ b/src/runlog/usage.ts @@ -43,6 +43,13 @@ export function summarizeUsage(events: OrchestratorEvent[], budget: BudgetConfig case 'SIGNOFF_DECIDED': addLlmStep(approver, event.llm); break; + case 'WAVE_RAN': + // EXPERIMENTAL parallel waves: each outcome carries its CHILD run's total spend (the child + // spends across all layers internally, metered by the SHARED budget). The parent report has + // no per-child columns, so the whole child total is bucketed under `harness` — the run's + // `total`/`budget` stay exact, which is what the cap and the summary line need. + for (const outcome of event.outcomes) addLlmStep(harness, outcome.usage); + break; } } diff --git a/src/testing/fakes.ts b/src/testing/fakes.ts index def8324..e422988 100644 --- a/src/testing/fakes.ts +++ b/src/testing/fakes.ts @@ -365,6 +365,28 @@ export class FakeWorktreeHost implements WorktreeHost { this.promoted.push(treeish); this.canonical?.setHash(treeish); } + + /** Scripted conflict paths (ours+theirs keys) — a pair listed here merges as a typed conflict. */ + readonly conflicts = new Set(); + /** Record of every mergeTrees call, for assertions. */ + readonly mergedCalls: { base: string; ours: string; theirs: string }[] = []; + + /** + * Fake 3-way merge (parallel waves): a pair scripted via {@link conflicts} (`"ours+theirs"`) + * conflicts; anything else merges "clean" to a deterministic synthetic tree id derived from the + * inputs, so tests can assert exactly which trees were combined without real git. + */ + async mergeTrees( + base: string, + ours: string, + theirs: string, + ): Promise<{ kind: 'clean'; tree: string } | { kind: 'conflict'; detail: string }> { + this.mergedCalls.push({ base, ours, theirs }); + if (this.conflicts.has(`${ours}+${theirs}`)) { + return { kind: 'conflict', detail: `scripted conflict merging ${theirs} onto ${ours}` }; + } + return { kind: 'clean', tree: `${ours.slice(0, 3)}${theirs.slice(0, 4)}` }; + } } export class ManualClock implements Clock { diff --git a/src/ui/web/format.ts b/src/ui/web/format.ts index 91b593d..66e98bc 100644 --- a/src/ui/web/format.ts +++ b/src/ui/web/format.ts @@ -74,6 +74,12 @@ export function feedLine(entry: RunLogEntry, iteration: number): FeedLine | null ]; return plain(`operator extension: ${parts.join(', ')}`); } + case 'WAVE_RAN': { + const merged = e.outcomes.filter((o) => o.kind === 'merged').length; + const fallback = e.outcomes.length - merged; + const text = `wave: ${merged}/${e.outcomes.length} phase(s) merged + re-verified${fallback > 0 ? `, ${fallback} downgraded to sequential` : ''}`; + return { at, text, tone: fallback > 0 ? 'plain' : 'pass' }; + } case 'CHECKPOINTED': return null; // internal diff-baseline plumbing — noise for a human } diff --git a/src/workspace/git-worktree-host.test.ts b/src/workspace/git-worktree-host.test.ts index 4f746e4..51dab93 100644 --- a/src/workspace/git-worktree-host.test.ts +++ b/src/workspace/git-worktree-host.test.ts @@ -97,6 +97,54 @@ describe('GitWorktreeHost (integration, real git) — best-of-N (issue #85)', () expect(git(root, 'rev-parse', 'HEAD')).toBe(headBefore); }); + it('mergeTrees merges DISJOINT edits cleanly into a promotable tree (parallel waves)', async () => { + const h = host(root); + const base = await new GitWorkspace(root).diffHash(); // the fork point + + // Two children fork from base and edit DIFFERENT files. + const a = await h.addWorktree('HEAD'); + await writeFile(join(a.root, 'a.txt'), 'from child A\n'); + const treeA = await a.scope.diffHash(); + await h.removeWorktree(a); + + const b = await h.addWorktree('HEAD'); + await writeFile(join(b.root, 'b.txt'), 'from child B\n'); + const treeB = await b.scope.diffHash(); + await h.removeWorktree(b); + + const merged = await h.mergeTrees(base, treeA, treeB); + expect(merged.kind).toBe('clean'); + if (merged.kind !== 'clean') return; + + // The merged tree promotes into the canonical workspace with BOTH children's work. + await h.promoteTree(merged.tree); + expect(await readFile(join(root, 'a.txt'), 'utf8')).toBe('from child A\n'); + expect(await readFile(join(root, 'b.txt'), 'utf8')).toBe('from child B\n'); + expect(await readFile(join(root, 'file.txt'), 'utf8')).toBe('base\n'); + }); + + it('mergeTrees reports OVERLAPPING edits as a typed conflict and applies nothing', async () => { + const h = host(root); + const base = await new GitWorkspace(root).diffHash(); + + const a = await h.addWorktree('HEAD'); + await writeFile(join(a.root, 'file.txt'), 'child A version\n'); + const treeA = await a.scope.diffHash(); + await h.removeWorktree(a); + + const b = await h.addWorktree('HEAD'); + await writeFile(join(b.root, 'file.txt'), 'child B version\n'); + const treeB = await b.scope.diffHash(); + await h.removeWorktree(b); + + const merged = await h.mergeTrees(base, treeA, treeB); + expect(merged.kind).toBe('conflict'); + if (merged.kind !== 'conflict') return; + expect(merged.detail).toContain('file.txt'); + // Nothing was applied anywhere — the canonical tree is untouched. + expect(await readFile(join(root, 'file.txt'), 'utf8')).toBe('base\n'); + }); + it('promoteTree deletes a tracked file the winning tree dropped', async () => { // Add a second tracked file in the canonical tree. await writeFile(join(root, 'drop-me.txt'), 'temp\n'); diff --git a/src/workspace/git-worktree-host.ts b/src/workspace/git-worktree-host.ts index 06f6715..2bcf5ad 100644 --- a/src/workspace/git-worktree-host.ts +++ b/src/workspace/git-worktree-host.ts @@ -101,6 +101,39 @@ export class GitWorktreeHost implements WorktreeHost { } } + /** + * EXPERIMENTAL (parallel waves) — 3-way merge `ours` and `theirs` against `base` using the modern + * plumbing `git merge-tree --write-tree --merge-base=` (git ≥ 2.40): a REAL recursive merge + * that writes only objects, never touching HEAD / index / working tree. Tree SHAs are wrapped in + * dangling commits first (the plumbing takes commit-ish). Exit 0 ⇒ clean (first stdout line is the + * merged tree OID); exit 1 ⇒ textual conflict (typed, with the conflicted paths — nothing is + * applied anywhere); anything else throws fail-closed. + */ + async mergeTrees( + base: string, + ours: string, + theirs: string, + ): Promise<{ kind: 'clean'; tree: string } | { kind: 'conflict'; detail: string }> { + const b = await this.#toCommitish(base); + const o = await this.#toCommitish(ours); + const t = await this.#toCommitish(theirs); + const r = await this.#git(['merge-tree', '--write-tree', `--merge-base=${b}`, o, t]); + if (r.code === 0) { + const tree = r.stdout.trim().split('\n')[0] ?? ''; + if (tree.length === 0) throw new Error('git merge-tree returned no tree OID'); + return { kind: 'clean', tree }; + } + if (r.code === 1) { + // Conflicted: stdout is \n. Surface the file names for the log. + const lines = splitLines(r.stdout).slice(1); + return { + kind: 'conflict', + detail: lines.length > 0 ? lines.slice(0, 10).join(', ') : 'textual merge conflict', + }; + } + throw new Error(`git merge-tree failed (code ${r.code}): ${r.stderr.trim()}`); + } + /** Resolve `treeish` to a commit-ish: a ref/commit as-is, else wrap a bare tree SHA in a commit. */ async #toCommitish(treeish: string): Promise { const commit = await this.#git(['rev-parse', '--verify', '--quiet', `${treeish}^{commit}`]); diff --git a/src/workspace/workspace.ts b/src/workspace/workspace.ts index 56f2716..087943a 100644 --- a/src/workspace/workspace.ts +++ b/src/workspace/workspace.ts @@ -64,6 +64,18 @@ export interface WorktreeHost { * surfaces it to the outer loop, never a silent half-applied tree). */ promoteTree(treeish: string): Promise; + /** + * EXPERIMENTAL (parallel waves) — 3-way merge two trees against an explicit base, entirely with + * plumbing (`git merge-tree --write-tree`): no commit, no HEAD/branch/index movement, no working-tree + * touch. Returns the merged tree SHA on a clean merge, or a typed `conflict` (with the conflicted + * paths) — NEVER a half-merged tree: a conflicted merge writes nothing anywhere. Throws fail-closed + * only on a real git error (e.g. an unknown SHA), which the caller downgrades to `unmerged`. + */ + mergeTrees( + base: string, + ours: string, + theirs: string, + ): Promise<{ kind: 'clean'; tree: string } | { kind: 'conflict'; detail: string }>; } /**