From db4a72f95ef5416b667ce1703014b9da59be5750 Mon Sep 17 00:00:00 2001 From: Eric Cao Date: Mon, 6 Jul 2026 17:15:51 +0000 Subject: [PATCH] feat: Phase 4 parallel file-disjoint workers (#37) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Completes the final phase of the loop-engineering roadmap: run multiple specs concurrently, safely, by scheduling on file-disjointness rather than lock coordination — and only now that Phase 1 stall detection exists to bound the cost of missing progress. File-disjoint scheduling (spec_queue.py) — specs declare a `## Files` scope; get_parallel_batch() greedily selects a set of ready specs whose scopes don't overlap (glob + directory-prefix aware), up to N workers. A spec with no declared scope runs alone — correctness over parallelism. Ready specs are already dependency-independent, so scheduling only reasons about file scope. Shared verification gate (verification.py) — the deterministic gate (acceptance criteria + backpressure via subprocess, plus the tamper-guard hash) is extracted into one module that both the sequential engine and the parallel orchestrator call. There is exactly one definition of "verified"; the engine now delegates to it (no behavior change). Parallel orchestrator (parallel.py) + owloop run --workers N — each worker runs one target spec in its own git worktree on its own branch: implement -> shared gate -> commit. The orchestrator merges each passed branch into the base branch; because the batch is file-disjoint those merges never conflict. A round with zero progress counts toward a stall (StopReason.STALLED) so a parallel run can't spin forever. Fresh adapter per worker (adapters hold per-run streaming state). Default --workers 1 keeps the existing sequential engine untouched. Spec template + generator now teach the `## Files` scope and disjoint decomposition; CLAUDE.md documents the file-disjoint principle and the shared gate. Tests: test_parallel.py (real worktrees + merges, stall, preflight) and file-disjoint scheduling tests in test_spec_queue.py. Full suite 307 passed; ruff + mypy clean. --- CLAUDE.md | 4 + src/owloop/cli.py | 61 ++++++- src/owloop/engine.py | 75 ++------- src/owloop/parallel.py | 304 +++++++++++++++++++++++++++++++++++ src/owloop/spec_generator.py | 6 + src/owloop/spec_queue.py | 110 +++++++++++++ src/owloop/verification.py | 92 +++++++++++ templates/spec-template.md | 8 + tests/test_parallel.py | 147 +++++++++++++++++ tests/test_spec_queue.py | 90 +++++++++++ 10 files changed, 835 insertions(+), 62 deletions(-) create mode 100644 src/owloop/parallel.py create mode 100644 src/owloop/verification.py create mode 100644 tests/test_parallel.py diff --git a/CLAUDE.md b/CLAUDE.md index a85ca41..b4e0953 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -16,6 +16,9 @@ owloop is a spec-driven autonomous coding loop for Claude Code — "Your code ev |---|---| | `src/owloop/cli.py` | Python CLI (`init` / `run` / `plan` / `status` / `version` subcommands), rich console output | | `src/owloop/engine.py` | Python loop engine — spawns agent per iteration, manages worktree, drives spec queue | +| `src/owloop/verification.py` | The shared deterministic gate (acceptance criteria + backpressure via `subprocess`, tamper hash) used by both engine and parallel orchestrator | +| `src/owloop/parallel.py` | File-disjoint parallel workers (`owloop run --workers N`): per-worker worktrees, shared gate, merge-back | +| `src/owloop/notifications.py` | Best-effort completion notifications (webhook/desktop) when a run stops | | `src/owloop/adapters.py` | Agent adapter abstraction (`ClaudeCodeAdapter`, `MockAdapter`) | | `src/owloop/tui.py` | Full-screen Rich TUI with owl animation | | `src/owloop/reporter.py` | Plain-text event reporter for non-interactive terminals and for `owloop run --no-tui` / `--plain` | @@ -60,3 +63,4 @@ owloop is a spec-driven autonomous coding loop for Claude Code — "Your code ev - **Named terminal states, never infinite retry** — runs stop with a `TerminalState` (`success` / `blocked` / `decide` / `stalled` / `exhausted` / `tampered`). A stall (N consecutive failures or the same error N times) hard-stops with `stalled`; `--keep-retrying` restores the old warn-and-back-off behavior. `exhausted` (hit an iteration/duration/token budget) must never be rendered or exit-coded as success. - **Don't stop silently** — an unattended run that halts at 2 a.m. must be able to reach the operator. `owloop run --notify-webhook URL` / `--notify-desktop` (or `OWLOOP_NOTIFY_WEBHOOK`) fire a best-effort completion notification (`notifications.py`, zero deps); a failed notification never changes the run outcome. - **Empty queue ≠ goal met** — `owloop run --converge N` runs up to N post-queue audit sweeps that compare the codebase against the specs' collective intent and append *linted* gap specs until it converges (#36). The spec-generation clarify gate records unanswered questions as a `## Assumptions` section when non-interactive. +- **Parallelism is file-disjoint, not lock-coordinated** — `owloop run --workers N` schedules specs whose declared `## Files` scopes don't overlap (`spec_queue.get_parallel_batch`), runs each in its own worktree via `parallel.py`, verifies with the *same* shared gate (`verification.run_gate`), and merges the passed branches back — disjoint scopes mean those merges never conflict. A spec with no `## Files` scope runs alone (correctness over parallelism). Never add lock coordination; invest in decomposition instead. The single verification gate lives in `verification.py` so the engine and the orchestrator grade work identically. diff --git a/src/owloop/cli.py b/src/owloop/cli.py index 57119e0..7814939 100644 --- a/src/owloop/cli.py +++ b/src/owloop/cli.py @@ -730,7 +730,16 @@ def _run_engine( notify_webhook: str | None = None, notify_desktop: bool = False, converge_sweeps: int = 0, + workers: int = 1, ) -> None: + resolved_webhook = notify_webhook or os.environ.get("OWLOOP_NOTIFY_WEBHOOK") or None + if workers > 1: + _run_parallel( + workers=workers, model=model, agent=agent, idle_timeout=idle_timeout, + ascii=ascii, no_color=no_color, + notify_webhook=resolved_webhook, notify_desktop=notify_desktop, + ) + return config = EngineConfig( project_dir=Path.cwd(), max_iterations=max_iterations, @@ -745,7 +754,7 @@ def _run_engine( dry_run=dry_run, keep_retrying=keep_retrying, rollback=rollback, - notify_webhook=notify_webhook or os.environ.get("OWLOOP_NOTIFY_WEBHOOK") or None, + notify_webhook=resolved_webhook, notify_desktop=notify_desktop, converge_sweeps=converge_sweeps, ) @@ -837,6 +846,47 @@ def _confirm_worktree_plain() -> bool: raise SystemExit(1) +def _run_parallel( + *, workers: int, model: str, agent: str, idle_timeout: float, + ascii: bool, no_color: bool, + notify_webhook: str | None, notify_desktop: bool, +) -> None: + """Run the file-disjoint parallel worker mode (`owloop run --workers N`).""" + from owloop.parallel import ParallelConfig, ParallelOrchestrator + + console = Console(no_color=no_color) + console.print() + console.print(_banner_text(ascii=ascii, no_color=no_color)) + console.print(f"[{_brand.AMBER}]Starting {workers} parallel workers...[/]") + + def _adapter_factory() -> Any: + # Fresh adapter per worker: adapters hold per-run streaming state, so + # concurrent workers must not share one instance. + return get_adapter( + agent, model=model, + claude_cmd=os.environ.get("CLAUDE_CMD", "claude"), + kimi_cmd=os.environ.get("KIMI_CMD", "kimi"), + idle_timeout=idle_timeout, + ) + + config = ParallelConfig( + project_dir=Path.cwd(), + workers=workers, + notify_webhook=notify_webhook, + notify_desktop=notify_desktop, + ) + reporter = ConsoleReporter(console, ascii=ascii) + orchestrator = ParallelOrchestrator(config, _adapter_factory, on_event=reporter.on_event) + try: + summary = orchestrator.run() + except KeyboardInterrupt: + console.print("\n[dim]owloop stopped.[/]") + raise SystemExit(0) from None + reporter.print_summary(summary) + if summary.stopped_reason in STOPPED_REASON_EXIT_1: + raise SystemExit(1) + + def _print_dry_run_report(console: Console, summary: RunSummary) -> None: """Print the concise pass/fail report produced by ``--dry-run`` / ``--one-shot``.""" report = summary.dry_run_report @@ -934,10 +984,16 @@ def _print_dry_run_report(console: Console, summary: RunSummary) -> None: "specs until the codebase converges on the goal (0 = disabled).", show_default=True, ) +@click.option( + "--workers", type=int, default=1, metavar="N", + help="Run up to N file-disjoint specs concurrently, each in its own worktree " + "(1 = sequential). Specs need a `## Files` scope to be scheduled in parallel.", + show_default=True, +) @_common_run_options def run(max_iterations: int, resume: bool, dry_run: bool, no_tui: bool, max_tokens_per_iteration: int, max_turns_per_iteration: int, max_budget_usd: float, keep_retrying: bool, rollback: bool, - notify_webhook: str | None, notify_desktop: bool, converge_sweeps: int, + notify_webhook: str | None, notify_desktop: bool, converge_sweeps: int, workers: int, worktree: bool, model: str, agent: str, verifier_model: str | None, subagents: bool, idle_timeout: float, max_duration: int, max_tokens: int) -> None: """Start the autonomous coding loop.""" @@ -966,6 +1022,7 @@ def run(max_iterations: int, resume: bool, dry_run: bool, no_tui: bool, max_toke notify_webhook=notify_webhook, notify_desktop=notify_desktop, converge_sweeps=converge_sweeps, + workers=workers, ) diff --git a/src/owloop/engine.py b/src/owloop/engine.py index 548368f..bf2ae47 100644 --- a/src/owloop/engine.py +++ b/src/owloop/engine.py @@ -18,7 +18,6 @@ from __future__ import annotations -import hashlib import json import re import shutil @@ -33,9 +32,8 @@ from pathlib import Path from typing import Any -from owloop import notifications, spec_queue +from owloop import notifications, spec_queue, verification from owloop.adapters import AgentAdapter, AgentResult -from owloop.backpressure import load_backpressure from owloop.learnings import ( append_learning, extract_learnings, @@ -922,82 +920,39 @@ def _push(self, branch: str) -> None: self._emit("push_retry", branch=branch) self._run_git("push", "-u", "origin", branch) - def _run_commands(self, commands: list[str]) -> tuple[int, int]: - """Run shell commands from the engine (not the agent); count pass/fail.""" - passed = failed = 0 - for command in commands: - result = subprocess.run( # noqa: S602 - spec-authored commands, same trust model as the agent's own execution - command, shell=True, cwd=self.cwd, capture_output=True, text=True, - ) - if result.returncode == 0: - passed += 1 - else: - failed += 1 - return passed, failed - def _run_acceptance_criteria(self, spec_name: str | None) -> tuple[int, int]: """Run a spec's Acceptance Criteria shell commands; count passes vs failures.""" - if not spec_name: - return 0, 0 - commands = spec_queue.get_acceptance_criteria_commands(self.specs_dir / spec_name) - return self._run_commands(commands) + return verification.run_acceptance_criteria(self.cwd, self.specs_dir, spec_name) def _guarded_hash(self, spec_name: str | None) -> str: - """Hash the spec sections + backpressure file the agent must not edit. - - The engine snapshots this before an iteration and re-checks it after, - so an agent that rewrites its own success conditions (acceptance - criteria, verification section, or the project's backpressure - commands) is caught regardless of whether the commands then "pass". - """ - h = hashlib.sha256() - if spec_name: - section = spec_queue.get_acceptance_criteria_section(self.specs_dir / spec_name) - h.update(section.encode("utf-8")) - backpressure = resolve_owloop_dir(self.cwd) / "backpressure.json" - if backpressure.is_file(): - h.update(backpressure.read_bytes()) - return h.hexdigest() + """Hash the spec sections + backpressure file the agent must not edit.""" + return verification.guarded_hash(self.cwd, self.specs_dir, spec_name) def _run_verification_gate( self, iteration: int, spec_name: str | None, guard_before: str ) -> tuple[bool, bool, int, int]: """Deterministically verify an iteration outside the agent's control. - Returns ``(passed, tampered, passed_count, failed_count)``. The engine - — never the agent — runs the spec's acceptance-criteria commands and - the project's backpressure commands via ``subprocess`` and lets exit - codes decide. A tampered guard region fails the gate immediately with - a distinct ``spec_tampered`` event, regardless of command results. + Delegates to the shared gate in ``verification.py`` (the single + definition of "verified" used by both the sequential engine and the + parallel orchestrator) and emits the gate events. A tampered guard + region fails the iteration with a distinct ``spec_tampered`` event. """ self._emit("verification_gate_start", iteration=iteration, spec=spec_name) + result = verification.run_gate(self.cwd, self.specs_dir, spec_name, guard_before) - if self._guarded_hash(spec_name) != guard_before: + if result.tampered: self._emit("spec_tampered", iteration=iteration, spec=spec_name) - return False, True, 0, 0 - - acc_passed, acc_failed = self._run_acceptance_criteria(spec_name) - bp_commands = [cmd.command for cmd in load_backpressure(self.cwd)] - bp_passed, bp_failed = self._run_commands(bp_commands) - - passed_count = acc_passed + bp_passed - failed_count = acc_failed + bp_failed - gate_ok = failed_count == 0 - - if gate_ok: - self._emit( - "verification_gate_passed", - iteration=iteration, - passed=passed_count, - ) + elif result.passed: + self._emit("verification_gate_passed", iteration=iteration, passed=result.passed_count) else: self._emit( "verification_gate_failed", iteration=iteration, - passed=passed_count, - failed=failed_count, + passed=result.passed_count, + failed=result.failed_count, ) - return gate_ok, False, passed_count, failed_count + return result.passed, result.tampered, result.passed_count, result.failed_count def _head(self) -> str: return str(self._run_git("rev-parse", "HEAD").stdout).strip() diff --git a/src/owloop/parallel.py b/src/owloop/parallel.py new file mode 100644 index 0000000..ae71cc3 --- /dev/null +++ b/src/owloop/parallel.py @@ -0,0 +1,304 @@ +"""Parallel workers over file-disjoint specs (Phase 4 of the #37 roadmap). + +The community lesson for parallel Ralph-style loops is: invest in **file-disjoint +spec decomposition**, not lock coordination — schedule specs whose declared +``## Files`` scopes don't overlap so their worktrees can never conflict, then +merge the successful branches back with a plain ``git merge``. And never +parallelize before stall detection exists (Phase 1), because parallelism +multiplies the cost of missing progress. + +Each worker runs one target spec in its own ``git worktree`` on its own branch: +implement → the *shared* deterministic gate (``verification.run_gate``) → +commit. The orchestrator then merges each passed branch into the base branch. +Because a batch is file-disjoint, those merges apply cleanly. + +The gate is the same one the sequential engine uses, so the "harness verifies, +never the agent" invariant holds identically here. +""" + +from __future__ import annotations + +import shutil +import subprocess +import uuid +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +from pathlib import Path + +from owloop import notifications, spec_queue, verification +from owloop.adapters import AgentAdapter +from owloop.engine import RunSummary, StopReason, classify_terminal_state +from owloop.paths import resolve_owloop_dir, resolve_specs_dir +from owloop.promise import parse_promise_signal + +WORKER_PROMPT = """\ +# Owloop — Parallel Build Worker + +You are one of several workers running concurrently, each on a disjoint set of +files. Implement EXACTLY ONE spec and nothing else: + + {spec_name} + +Read `AGENTS.md` / `CLAUDE.md` if present. Implement the spec completely and +run its `## Acceptance Criteria` commands yourself to check your work. Then +output `DONE`. + +Stay strictly inside the files listed in that spec's `## Files` section — other +workers own the other files right now, and touching them will lose their work +at merge time. Do NOT work on any other spec. + +The loop — not you — owns git and completion: +- Do NOT commit or push. The loop commits only after re-running the acceptance + criteria itself and they pass. +- Do NOT add a `Status: COMPLETE` line, and do NOT edit the spec's + `## Acceptance Criteria` / `## Verification` sections or + `.owloop/backpressure.json`. Rewriting your own success conditions fails the + iteration. + +If you hit an external blocker, output `BLOCKED:reason`. +""" + +EventCallback = Callable[[str, dict], None] +AdapterFactory = Callable[[], AgentAdapter] + + +@dataclass +class WorkerResult: + spec_name: str + passed: bool + branch: str + worktree: Path + tampered: bool = False + tokens_used: int = 0 + reason: str = "" + + +@dataclass +class ParallelConfig: + project_dir: Path + workers: int = 2 + max_rounds: int = 0 # 0 = until queue empty / stall + max_consecutive_failed_rounds: int = 3 + notify_webhook: str | None = None + notify_desktop: bool = False + + +class ParallelOrchestrator: + """Run file-disjoint specs concurrently across per-worker git worktrees.""" + + def __init__( + self, + config: ParallelConfig, + adapter_factory: AdapterFactory, + on_event: EventCallback | None = None, + ) -> None: + self.config = config + self.adapter_factory = adapter_factory + self.on_event = on_event or (lambda *_a: None) + self.base_dir = config.project_dir + self.session_id = uuid.uuid4().hex[:8] + self.tokens_used = 0 + + # ── git helpers ── + + def _git(self, *args: str, cwd: Path | None = None) -> subprocess.CompletedProcess: + return subprocess.run( + ["git", *args], + cwd=cwd or self.base_dir, + capture_output=True, + text=True, + ) + + def _is_git_repo(self) -> bool: + return self._git("rev-parse", "--is-inside-work-tree").returncode == 0 + + def _current_branch(self) -> str: + return self._git("branch", "--show-current").stdout.strip() or "main" + + def _head(self) -> str: + return str(self._git("rev-parse", "HEAD").stdout).strip() + + def _emit(self, kind: str, **data: object) -> None: + self.on_event(kind, data) + + @property + def specs_dir(self) -> Path: + return resolve_specs_dir(self.base_dir) + + def preflight(self) -> list[str]: + issues = self.adapter_factory().preflight() + if not self._is_git_repo(): + issues.append("current directory is not a git repository") + if not spec_queue.get_root_specs(self.specs_dir): + issues.append("no .md files in specs/") + return issues + + # ── worker ── + + def _run_worker(self, spec_name: str, base_head: str, index: int) -> WorkerResult: + wt_base = self.base_dir.parent / f"{self.base_dir.name}-owloop-wt" + wt_base.mkdir(parents=True, exist_ok=True) + branch = f"owloop/{self.session_id}-w{index}" + wt_path = wt_base / f"owloop-{self.session_id}-w{index}" + + created = self._git("worktree", "add", str(wt_path), "-b", branch, base_head) + if created.returncode != 0: + return WorkerResult(spec_name, False, branch, wt_path, reason="worktree_failed") + + # Materialize loop metadata (specs/backpressure) that may be untracked. + source = resolve_owloop_dir(self.base_dir) + if source.is_dir(): + shutil.copytree(source, wt_path / ".owloop", dirs_exist_ok=True) + + self._emit("worker_start", spec=spec_name, worker=index, branch=branch) + + wt_specs = resolve_specs_dir(wt_path) + guard_before = verification.guarded_hash(wt_path, wt_specs, spec_name) + + adapter = self.adapter_factory() + result = adapter.run( + WORKER_PROMPT.format(spec_name=spec_name), + cwd=wt_path, + on_line=lambda line: self._emit("output_line", worker=index, line=line), + ) + tokens = result.tokens_used + parsed = parse_promise_signal(result.stdout) + state = parsed[0] if parsed else "" + + if state != "DONE": + self._emit("worker_no_done", spec=spec_name, worker=index) + return WorkerResult(spec_name, False, branch, wt_path, tokens_used=tokens, + reason="no_done_signal") + + gate = verification.run_gate(wt_path, wt_specs, spec_name, guard_before) + if gate.tampered: + self._emit("spec_tampered", spec=spec_name, worker=index) + return WorkerResult(spec_name, False, branch, wt_path, tampered=True, + tokens_used=tokens, reason="tampered") + if not gate.passed: + self._emit("verification_gate_failed", spec=spec_name, worker=index, + failed=gate.failed_count) + return WorkerResult(spec_name, False, branch, wt_path, tokens_used=tokens, + reason="verification_failed") + + # Verified: mark complete + commit inside the worker's worktree. + spec_queue.mark_spec_complete(wt_specs / spec_name) + self._git("add", "-A", cwd=wt_path) + self._git("reset", "--quiet", "--", ".owloop/run-notes.md", ".owloop/logs", + ".owloop/PROMPT_build.md", cwd=wt_path) + self._git("commit", "-m", f"owloop: complete {spec_name}", cwd=wt_path) + self._emit("verification_gate_passed", spec=spec_name, worker=index, + passed=gate.passed_count) + return WorkerResult(spec_name, True, branch, wt_path, tokens_used=tokens) + + def _cleanup_worker(self, result: WorkerResult, *, keep_branch: bool) -> None: + self._git("worktree", "remove", "--force", str(result.worktree)) + if not keep_branch: + self._git("branch", "-D", result.branch) + + # ── run ── + + def run(self) -> RunSummary: + issues = self.preflight() + if issues: + self._emit("preflight_failed", issues=issues) + return self._summary(StopReason.PREFLIGHT_FAILED, rounds=0, branch="", issues=issues) + + base_branch = self._current_branch() + stopped_reason: StopReason = StopReason.SUCCESS + rounds = 0 + consecutive_failed_rounds = 0 + + self._emit("parallel_session_info", workers=self.config.workers, + branch=base_branch, session_id=self.session_id) + + while True: + if self.config.max_rounds and rounds >= self.config.max_rounds: + stopped_reason = StopReason.MAX_ITERATIONS + break + + batch = spec_queue.get_parallel_batch(self.specs_dir, self.config.workers) + if not batch: + stopped_reason = StopReason.SUCCESS + self._emit("all_specs_complete", + spec_count=spec_queue.count_root_specs(self.specs_dir)) + break + + rounds += 1 + base_head = self._head() + names = [p.name for p in batch] + self._emit("round_start", round=rounds, specs=names, size=len(names)) + + results = self._run_batch(names, base_head) + for r in results: + self.tokens_used += r.tokens_used + + passed = [r for r in results if r.passed] + self._merge_passed(passed, base_branch) + for r in results: + self._cleanup_worker(r, keep_branch=False) + + self._emit("round_end", round=rounds, passed=[r.spec_name for r in passed], + failed=[r.spec_name for r in results if not r.passed]) + + if passed: + consecutive_failed_rounds = 0 + else: + consecutive_failed_rounds += 1 + if consecutive_failed_rounds >= self.config.max_consecutive_failed_rounds: + stopped_reason = StopReason.STALLED + self._emit("stalled", reason="no_progress_rounds", + rounds=consecutive_failed_rounds) + break + + summary = self._summary(stopped_reason, rounds=rounds, branch=base_branch) + notifications.notify_run_complete( + summary, + webhook_url=self.config.notify_webhook, + desktop=self.config.notify_desktop, + emit=lambda kind, **data: self._emit(kind, **data), + ) + return summary + + def _run_batch(self, names: list[str], base_head: str) -> list[WorkerResult]: + if len(names) == 1: + return [self._run_worker(names[0], base_head, 0)] + with ThreadPoolExecutor(max_workers=len(names)) as pool: + futures = [ + pool.submit(self._run_worker, name, base_head, i) + for i, name in enumerate(names) + ] + return [f.result() for f in futures] + + def _merge_passed(self, passed: list[WorkerResult], base_branch: str) -> None: + """Merge each verified worker branch into the base branch. + + The batch is file-disjoint, so these merges never conflict. Merges run + sequentially in the base worktree in deterministic (spec-name) order. + """ + for r in sorted(passed, key=lambda r: r.spec_name): + merged = self._git("merge", "--no-ff", "-m", + f"owloop: merge {r.spec_name}", r.branch) + if merged.returncode == 0: + self._emit("worker_merged", spec=r.spec_name, branch=r.branch) + else: + # Should not happen for disjoint scopes; abort and surface it. + self._git("merge", "--abort") + self._emit("worker_merge_conflict", spec=r.spec_name, branch=r.branch) + + def _summary( + self, stopped_reason: StopReason, *, rounds: int, branch: str, + issues: list[str] | None = None, + ) -> RunSummary: + return RunSummary( + iterations=rounds, + branch=branch, + cwd=self.base_dir, + main_repo_dir=self.base_dir, + stopped_reason=str(stopped_reason), + terminal_state=classify_terminal_state(stopped_reason), + issues=issues, + tokens_used=self.tokens_used, + session_id=self.session_id, + ) diff --git a/src/owloop/spec_generator.py b/src/owloop/spec_generator.py index 5134724..d1bc681 100644 --- a/src/owloop/spec_generator.py +++ b/src/owloop/spec_generator.py @@ -63,6 +63,8 @@ - Touch 1-5 files / < 300 lines - Be independently verifiable (has its own acceptance criteria) - Have a clear dependency order (which specs must complete first) +- Declare its file scope in a `## Files` section, and prefer DISJOINT scopes + across independent specs so they can be run concurrently by parallel workers Plan the decomposition as an ordered list before writing any spec. Common decomposition patterns: @@ -112,6 +114,10 @@ ## Depends On - [list spec names this depends on, or "none"] +## Files +- [each file/dir this spec may touch, one per line — paths or globs. Keep scopes + DISJOINT across sibling specs so they can run in parallel (`--workers N`).] + ## Requirements - [ ] Concrete, scoped task description. Prefer EARS-style phrasing where it fits — "WHEN , THE SYSTEM SHALL ", "WHILE , THE diff --git a/src/owloop/spec_queue.py b/src/owloop/spec_queue.py index 1ecfbd5..161d534 100644 --- a/src/owloop/spec_queue.py +++ b/src/owloop/spec_queue.py @@ -24,6 +24,7 @@ number, then the lexicographically earliest filename. """ +import fnmatch import re from pathlib import Path @@ -43,6 +44,10 @@ r"^##\s+Verification\s*$\n(.*?)(?=^#{1,2}\s|\Z)", re.IGNORECASE | re.MULTILINE | re.DOTALL, ) +_FILES_SECTION_RE = re.compile( + r"^##\s+Files\s*$\n(.*?)(?=^#{1,2}\s|\Z)", + re.IGNORECASE | re.MULTILINE | re.DOTALL, +) DEFAULT_PRIORITY = 999 @@ -277,3 +282,108 @@ def get_next_ready_spec(specs_dir: Path) -> Path | None: ready.sort(key=lambda spec: (get_spec_priority(spec), spec.name)) return ready[0] + + +def get_ready_specs(specs_dir: Path) -> list[Path]: + """Return all incomplete specs whose dependencies are complete, in run order. + + Same readiness rule as ``get_next_ready_spec`` but returns the whole ready + frontier, sorted by (Priority, filename). Because two *ready* specs can only + depend on already-complete specs, they are guaranteed dependency-independent + of each other — so parallel scheduling only has to reason about file scope. + """ + all_specs = get_root_specs(specs_dir) + incomplete = get_incomplete_root_specs(specs_dir) + incomplete_set = set(incomplete) + ready = [ + spec + for spec in incomplete + if all(dep not in incomplete_set for dep in get_spec_dependencies(spec, all_specs)) + ] + ready.sort(key=lambda spec: (get_spec_priority(spec), spec.name)) + return ready + + +def get_spec_file_scope(spec_file: Path) -> list[str]: + """Return the path/glob tokens a spec declares under its ``## Files`` section. + + This is the contract that makes safe parallelism possible: a spec that + declares which files it touches can be scheduled alongside other specs whose + scopes don't overlap. A missing/empty section yields ``[]`` — an unknown + scope, which the scheduler treats conservatively (never parallel-safe). + """ + if not spec_file.is_file(): + return [] + content = spec_file.read_text(encoding="utf-8", errors="replace") + match = _FILES_SECTION_RE.search(content) + if match is None: + return [] + scope: list[str] = [] + for item in _LIST_ITEM_RE.findall(match.group(1)): + cleaned = item.strip().strip("`").strip() + if cleaned and cleaned.lower() != "none": + scope.append(cleaned) + return scope + + +def _paths_conflict(a: str, b: str) -> bool: + """True if two path/glob tokens could touch a common file.""" + if a == b: + return True + if fnmatch.fnmatch(a, b) or fnmatch.fnmatch(b, a): + return True + an, bn = a.rstrip("/"), b.rstrip("/") + # Directory-prefix containment (only for non-glob tokens). + if "*" not in a and "?" not in a and (bn == an or bn.startswith(an + "/")): + return True + if "*" not in b and "?" not in b and (an == bn or an.startswith(bn + "/")): # noqa: SIM103 + return True + return False + + +def scopes_overlap(scope_a: list[str], scope_b: list[str]) -> bool: + """True if any token in one scope could touch a file in the other.""" + return any(_paths_conflict(a, b) for a in scope_a for b in scope_b) + + +def specs_are_disjoint(spec_a: Path, spec_b: Path) -> bool: + """True if two specs declare non-empty, non-overlapping file scopes. + + An empty (undeclared) scope is never disjoint — the safe default is that an + unscoped spec might touch anything, so it must not run in parallel. + """ + scope_a = get_spec_file_scope(spec_a) + scope_b = get_spec_file_scope(spec_b) + if not scope_a or not scope_b: + return False + return not scopes_overlap(scope_a, scope_b) + + +def get_parallel_batch(specs_dir: Path, max_workers: int) -> list[Path]: + """Select a batch of ready specs safe to run concurrently. + + Greedy: start from the highest-priority ready spec and add further ready + specs whose declared ``## Files`` scope is disjoint from every spec already + in the batch, up to ``max_workers``. If the lead spec has no declared scope + the batch is just ``[lead]`` (run it alone) — correctness over parallelism. + Returns ``[]`` only when nothing is ready. + """ + ready = get_ready_specs(specs_dir) + if not ready: + return [] + if max_workers <= 1: + return [ready[0]] + + lead = ready[0] + batch = [lead] + if not get_spec_file_scope(lead): + return batch # unknown scope → run alone + + for candidate in ready[1:]: + if len(batch) >= max_workers: + break + if get_spec_file_scope(candidate) and all( + specs_are_disjoint(candidate, chosen) for chosen in batch + ): + batch.append(candidate) + return batch diff --git a/src/owloop/verification.py b/src/owloop/verification.py new file mode 100644 index 0000000..b9a1fa9 --- /dev/null +++ b/src/owloop/verification.py @@ -0,0 +1,92 @@ +"""The deterministic verification gate — the single place work is graded. + +owloop's central invariant (#32): completion is decided by the *harness* running +the spec's acceptance criteria + the project's backpressure commands via +``subprocess``, outside the agent's control, never by the agent's own say-so. A +snapshot hash of the sections the agent must not touch (acceptance criteria, +verification, ``backpressure.json``) catches an agent that rewrites its own +success conditions. + +Both the sequential engine and the parallel orchestrator call these functions, +so the gate is shared, not duplicated — there is exactly one definition of +"verified". +""" + +from __future__ import annotations + +import hashlib +import subprocess +from dataclasses import dataclass +from pathlib import Path + +from owloop import spec_queue +from owloop.backpressure import load_backpressure +from owloop.paths import resolve_owloop_dir + + +@dataclass +class GateResult: + passed: bool + tampered: bool + passed_count: int + failed_count: int + + +def run_commands(cwd: Path, commands: list[str]) -> tuple[int, int]: + """Run shell commands from the harness (not the agent); count pass/fail.""" + passed = failed = 0 + for command in commands: + result = subprocess.run( # noqa: S602 - spec-authored commands, same trust model as the agent's own execution + command, shell=True, cwd=cwd, capture_output=True, text=True, + ) + if result.returncode == 0: + passed += 1 + else: + failed += 1 + return passed, failed + + +def run_acceptance_criteria(cwd: Path, specs_dir: Path, spec_name: str | None) -> tuple[int, int]: + """Run a spec's Acceptance Criteria shell commands; count passes vs failures.""" + if not spec_name: + return 0, 0 + commands = spec_queue.get_acceptance_criteria_commands(specs_dir / spec_name) + return run_commands(cwd, commands) + + +def guarded_hash(cwd: Path, specs_dir: Path, spec_name: str | None) -> str: + """Hash the spec sections + backpressure file the agent must not edit.""" + h = hashlib.sha256() + if spec_name: + section = spec_queue.get_acceptance_criteria_section(specs_dir / spec_name) + h.update(section.encode("utf-8")) + backpressure = resolve_owloop_dir(cwd) / "backpressure.json" + if backpressure.is_file(): + h.update(backpressure.read_bytes()) + return h.hexdigest() + + +def run_gate( + cwd: Path, specs_dir: Path, spec_name: str | None, guard_before: str +) -> GateResult: + """Deterministically verify one iteration's work outside the agent's control. + + A guard region that changed since ``guard_before`` fails immediately as + ``tampered``; otherwise the acceptance-criteria and backpressure commands + decide pass/fail by exit code. + """ + if guarded_hash(cwd, specs_dir, spec_name) != guard_before: + return GateResult(passed=False, tampered=True, passed_count=0, failed_count=0) + + acc_passed, acc_failed = run_acceptance_criteria(cwd, specs_dir, spec_name) + bp_commands = [cmd.command for cmd in load_backpressure(cwd)] + bp_passed, bp_failed = run_commands(cwd, bp_commands) + + passed_count = acc_passed + bp_passed + failed_count = acc_failed + bp_failed + return GateResult( + passed=failed_count == 0, + tampered=False, + passed_count=passed_count, + failed_count=failed_count, + ) diff --git a/templates/spec-template.md b/templates/spec-template.md index c20579a..f514a27 100644 --- a/templates/spec-template.md +++ b/templates/spec-template.md @@ -2,6 +2,14 @@ ## Priority: [1-5] +## Files +[The files/directories this spec is allowed to touch, one per line — paths or +globs. This scope is what lets `owloop run --workers N` schedule this spec +alongside others whose scopes don't overlap. Omit it and the spec runs alone.] + +- src/[module]/ +- tests/test_[module].py + ## Requirements [What to build — the functional description of the feature. EARS-style phrasing maps cleanly onto shell-verifiable criteria: "WHEN , THE SYSTEM SHALL diff --git a/tests/test_parallel.py b/tests/test_parallel.py new file mode 100644 index 0000000..4df038b --- /dev/null +++ b/tests/test_parallel.py @@ -0,0 +1,147 @@ +"""Tests for the file-disjoint parallel worker orchestrator (Phase 4).""" + +from __future__ import annotations + +import subprocess +from pathlib import Path + +from owloop.adapters import AgentResult, MockAdapter +from owloop.parallel import ParallelConfig, ParallelOrchestrator + + +def _git(cwd: Path, *args: str) -> None: + subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True) + + +def _repo_with_specs(tmp_path: Path, specs: dict[str, list[str]]) -> Path: + """Create a git repo with committed, file-scoped specs (acceptance: `true`).""" + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init") + _git(repo, "config", "user.email", "t@example.com") + _git(repo, "config", "user.name", "T") + (repo / "README.md").write_text("# t", encoding="utf-8") + specs_dir = repo / ".owloop" / "specs" + specs_dir.mkdir(parents=True) + for name, files in specs.items(): + files_block = "\n".join(f"- {f}" for f in files) + (specs_dir / name).write_text( + f"# Spec: {name}\n\n## Priority: 1\n\n## Files\n{files_block}\n\n" + f"## Acceptance Criteria\n- `true`\n\n## Exclusions\n- Do NOT touch other files\n", + encoding="utf-8", + ) + _git(repo, "add", "-A") + _git(repo, "commit", "-m", "init with specs") + return repo + + +def _done() -> AgentResult: + return AgentResult( + stdout="DONE", + returncode=0, + success=True, + has_completion_signal=True, + done_signal="DONE", + ) + + +def _done_factory(): + return MockAdapter(responses=[_done()]) + + +def _is_complete(repo: Path, spec: str) -> bool: + from owloop.spec_queue import is_root_spec_complete + + return is_root_spec_complete(repo / ".owloop" / "specs" / spec) + + +def test_parallel_completes_disjoint_specs(tmp_path: Path) -> None: + repo = _repo_with_specs(tmp_path, { + "01-a.md": ["src/a/"], + "02-b.md": ["src/b/"], + }) + config = ParallelConfig(project_dir=repo, workers=2) + events: list[tuple[str, dict]] = [] + orch = ParallelOrchestrator(config, _done_factory, on_event=lambda k, d: events.append((k, d))) + + summary = orch.run() + + assert summary.stopped_reason == "success" + assert _is_complete(repo, "01-a.md") + assert _is_complete(repo, "02-b.md") + # First round scheduled both specs together (disjoint scopes). + round_starts = [d for k, d in events if k == "round_start"] + assert round_starts and round_starts[0]["size"] == 2 + assert sum(1 for k, _ in events if k == "worker_merged") == 2 + + +def test_parallel_merges_land_on_base_branch(tmp_path: Path) -> None: + repo = _repo_with_specs(tmp_path, {"01-a.md": ["src/a/"], "02-b.md": ["src/b/"]}) + orch = ParallelOrchestrator(ParallelConfig(project_dir=repo, workers=2), _done_factory) + + orch.run() + + log = subprocess.run( + ["git", "log", "--oneline"], cwd=repo, capture_output=True, text=True + ).stdout + assert "owloop: complete 01-a.md" in log + assert "owloop: complete 02-b.md" in log + # Worktrees are cleaned up. + assert subprocess.run( + ["git", "worktree", "list"], cwd=repo, capture_output=True, text=True + ).stdout.count("owloop-") == 0 + + +def test_parallel_overlapping_specs_run_in_separate_rounds(tmp_path: Path) -> None: + # Both specs touch src/shared/ → never batched together. + repo = _repo_with_specs(tmp_path, { + "01-a.md": ["src/shared/a.py"], + "02-b.md": ["src/shared/b.py", "src/shared/"], + }) + # Force conflict: 02 declares the whole dir, 01 a file inside it. + events: list[tuple[str, dict]] = [] + orch = ParallelOrchestrator( + ParallelConfig(project_dir=repo, workers=2), _done_factory, + on_event=lambda k, d: events.append((k, d)), + ) + summary = orch.run() + + assert summary.stopped_reason == "success" + sizes = [d["size"] for k, d in events if k == "round_start"] + assert all(s == 1 for s in sizes) # never parallelized + assert len(sizes) == 2 # two sequential rounds + + +def test_parallel_stalls_when_no_worker_makes_progress(tmp_path: Path) -> None: + repo = _repo_with_specs(tmp_path, {"01-a.md": ["src/a/"], "02-b.md": ["src/b/"]}) + + # Workers never emit DONE → no spec ever completes. + def _fail_factory(): + return MockAdapter(responses=[AgentResult( + stdout="nope", returncode=0, success=True, has_completion_signal=False, + )]) + + events: list[tuple[str, dict]] = [] + orch = ParallelOrchestrator( + ParallelConfig(project_dir=repo, workers=2, max_consecutive_failed_rounds=2), + _fail_factory, on_event=lambda k, d: events.append((k, d)), + ) + summary = orch.run() + + assert summary.stopped_reason == "stalled" + assert summary.state == "stalled" + assert any(k == "stalled" for k, _ in events) + assert not _is_complete(repo, "01-a.md") + + +def test_parallel_preflight_fails_without_specs(tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + _git(repo, "init") + _git(repo, "config", "user.email", "t@example.com") + _git(repo, "config", "user.name", "T") + (repo / ".owloop" / "specs").mkdir(parents=True) + orch = ParallelOrchestrator(ParallelConfig(project_dir=repo, workers=2), _done_factory) + + summary = orch.run() + assert summary.stopped_reason == "preflight_failed" diff --git a/tests/test_spec_queue.py b/tests/test_spec_queue.py index 58106ce..fb15f04 100644 --- a/tests/test_spec_queue.py +++ b/tests/test_spec_queue.py @@ -209,3 +209,93 @@ def test_get_acceptance_criteria_section_missing_returns_empty(tmp_path: Path) - spec = tmp_path / "01-t.md" spec.write_text("# Spec\n\n## Requirements\n- x\n", encoding="utf-8") assert spec_queue.get_acceptance_criteria_section(spec) == "" + + +# ── file-disjoint parallel scheduling (Phase 4) ── + + +def _scoped_spec(priority: int, files: list[str], depends_on: list[str] | None = None, + complete: bool = False) -> str: + lines = ["# Spec: test", ""] + if complete: + lines += ["**Status**: COMPLETE", ""] + lines += [f"## Priority: {priority}", ""] + if depends_on is not None: + lines += ["## Depends On"] + ([f"- {d}" for d in depends_on] or ["- none"]) + [""] + lines += ["## Files"] + [f"- {f}" for f in files] + [""] + lines += ["## Requirements", "Do a thing.", ""] + return "\n".join(lines) + + +def test_get_spec_file_scope_parses_files_section(tmp_path: Path) -> None: + spec = _write(tmp_path, "001-a.md", _scoped_spec(1, ["src/a/", "`tests/test_a.py`"])) + assert spec_queue.get_spec_file_scope(spec) == ["src/a/", "tests/test_a.py"] + + +def test_get_spec_file_scope_missing_section_is_empty(tmp_path: Path) -> None: + spec = _write(tmp_path, "001-a.md", _spec(priority=1)) + assert spec_queue.get_spec_file_scope(spec) == [] + + +def test_specs_are_disjoint_true_for_separate_dirs(tmp_path: Path) -> None: + a = _write(tmp_path, "001-a.md", _scoped_spec(1, ["src/a/"])) + b = _write(tmp_path, "002-b.md", _scoped_spec(1, ["src/b/"])) + assert spec_queue.specs_are_disjoint(a, b) is True + + +def test_specs_are_disjoint_false_on_overlap(tmp_path: Path) -> None: + a = _write(tmp_path, "001-a.md", _scoped_spec(1, ["src/a/"])) + b = _write(tmp_path, "002-b.md", _scoped_spec(1, ["src/a/util.py"])) + assert spec_queue.specs_are_disjoint(a, b) is False + + +def test_specs_are_disjoint_false_on_glob_match(tmp_path: Path) -> None: + a = _write(tmp_path, "001-a.md", _scoped_spec(1, ["src/*.py"])) + b = _write(tmp_path, "002-b.md", _scoped_spec(1, ["src/main.py"])) + assert spec_queue.specs_are_disjoint(a, b) is False + + +def test_specs_are_disjoint_false_when_scope_unknown(tmp_path: Path) -> None: + a = _write(tmp_path, "001-a.md", _scoped_spec(1, ["src/a/"])) + b = _write(tmp_path, "002-b.md", _spec(priority=1)) # no ## Files + assert spec_queue.specs_are_disjoint(a, b) is False + + +def test_get_parallel_batch_groups_disjoint_specs(tmp_path: Path) -> None: + specs = tmp_path / "specs" + _write(specs, "001-a.md", _scoped_spec(1, ["src/a/"])) + _write(specs, "002-b.md", _scoped_spec(1, ["src/b/"])) + _write(specs, "003-c.md", _scoped_spec(1, ["src/a/deep.py"])) # conflicts with a + batch = spec_queue.get_parallel_batch(specs, max_workers=4) + names = {p.name for p in batch} + assert names == {"001-a.md", "002-b.md"} # c excluded (overlaps a) + + +def test_get_parallel_batch_respects_max_workers(tmp_path: Path) -> None: + specs = tmp_path / "specs" + _write(specs, "001-a.md", _scoped_spec(1, ["src/a/"])) + _write(specs, "002-b.md", _scoped_spec(1, ["src/b/"])) + _write(specs, "003-c.md", _scoped_spec(1, ["src/c/"])) + batch = spec_queue.get_parallel_batch(specs, max_workers=2) + assert len(batch) == 2 + + +def test_get_parallel_batch_lead_without_scope_runs_alone(tmp_path: Path) -> None: + specs = tmp_path / "specs" + _write(specs, "001-a.md", _spec(priority=1)) # highest priority, no ## Files + _write(specs, "002-b.md", _scoped_spec(2, ["src/b/"])) + batch = spec_queue.get_parallel_batch(specs, max_workers=4) + assert [p.name for p in batch] == ["001-a.md"] + + +def test_get_parallel_batch_max_workers_one_is_sequential(tmp_path: Path) -> None: + specs = tmp_path / "specs" + _write(specs, "001-a.md", _scoped_spec(1, ["src/a/"])) + _write(specs, "002-b.md", _scoped_spec(1, ["src/b/"])) + assert len(spec_queue.get_parallel_batch(specs, max_workers=1)) == 1 + + +def test_get_parallel_batch_empty_when_nothing_ready(tmp_path: Path) -> None: + specs = tmp_path / "specs" + _write(specs, "001-a.md", _scoped_spec(1, ["src/a/"], complete=True)) + assert spec_queue.get_parallel_batch(specs, max_workers=4) == []