feat: Phase 4 parallel file-disjoint workers (#37)

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`, `KimiCodeAdapter`, `MockAdapter`) |
 | `src/owloop/presets.py` | Agent preset registry — per-tool launch commands/env as data (user presets via `.owloop/agents.toml`) |
 | `src/owloop/acp.py` | `AcpAdapter` — one Agent Client Protocol client covering all non-native agents |
@@ -63,3 +66,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.

src/owloop/cli.py

@@ -797,7 +797,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,
@@ -812,7 +821,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,
     )
@@ -906,6 +915,47 @@ def _confirm_worktree_plain() -> bool:
         raise SystemExit(1)
 
 
+def _run_parallel(
+    *, workers: int, model: str | None, 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
@@ -1003,10 +1053,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."""
@@ -1035,6 +1091,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,
     )
 
 

src/owloop/engine.py

@@ -19,7 +19,6 @@
 from __future__ import annotations
 
 import contextlib
-import hashlib
 import json
 import re
 import shutil
@@ -34,9 +33,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 DEFAULT_IDLE_TIMEOUT, AgentAdapter, AgentResult
-from owloop.backpressure import load_backpressure
 from owloop.learnings import (
     append_learning,
     extract_learnings,
@@ -1021,97 +1019,46 @@ 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, list[dict[str, Any]]]:
-        """Run shell commands from the engine (not the agent).
 
-        Returns ``(passed, failed, failures)`` where each failure records the
-        command, its exit code, and an output tail — the raw material for the
-        failure-feedback file that primes the next iteration's retry.
-        """
-        passed = 0
-        failures: list[dict[str, Any]] = []
-        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:
-                output = f"{result.stdout or ''}\n{result.stderr or ''}".strip()
-                tail = "\n".join(output.splitlines()[-30:])[-2000:]
-                failures.append(
-                    {"command": command, "returncode": result.returncode, "output": tail}
-                )
-        return passed, len(failures), failures
-
-    def _run_acceptance_criteria(
-        self, spec_name: str | None
-    ) -> tuple[int, int, list[dict[str, Any]]]:
-        """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)
 
     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, list[dict[str, Any]]]:
         """Deterministically verify an iteration outside the agent's control.
 
-        Returns ``(passed, tampered, failures)``. 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. ``failures``
-        carries per-command details for the next iteration's failure feedback.
+        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. ``failures`` carries
+        per-command details for the next iteration's failure feedback. 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, []
 
-        acc_passed, acc_failed, acc_failures = self._run_acceptance_criteria(spec_name)
-        bp_commands = [cmd.command for cmd in load_backpressure(self.cwd)]
-        bp_passed, bp_failed, bp_failures = self._run_commands(bp_commands)
-
-        passed_count = acc_passed + bp_passed
-        failed_count = acc_failed + bp_failed
-        failures = acc_failures + bp_failures
-        gate_ok = failed_count == 0
-
-        if gate_ok:
+        if result.passed:
             self._emit(
                 "verification_gate_passed",
                 iteration=iteration,
-                passed=passed_count,
+                passed=result.passed_count,
             )
         else:
             self._emit(
                 "verification_gate_failed",
                 iteration=iteration,
-                passed=passed_count,
-                failed=failed_count,
-                commands=[f["command"] for f in failures],
+                passed=result.passed_count,
+                failed=result.failed_count,
+                commands=[f["command"] for f in result.failures],
             )
-        return gate_ok, False, failures
+        return result.passed, False, result.failures
 
     def _head(self) -> str:
         return str(self._run_git("rev-parse", "HEAD").stdout).strip()
@@ -1541,8 +1488,8 @@ def run(self) -> RunSummary:
 
                 if self.config.dry_run:
                     self._append_run_note(iteration, result.success, result.summary)
-                    acceptance_passed, acceptance_failed, _ = self._run_acceptance_criteria(
-                        active_spec
+                    acceptance_passed, acceptance_failed, _ = verification.run_acceptance_criteria(
+                        self.cwd, self.specs_dir, active_spec
                     )
                     current_head = self._head()
                     if current_head and current_head != dry_run_original_head: