diff --git a/src/owloop/cli.py b/src/owloop/cli.py
index 48566da..f52c803 100644
--- a/src/owloop/cli.py
+++ b/src/owloop/cli.py
@@ -250,6 +250,13 @@ def agents() -> None:
"(no committed changes are left behind). Use to validate specs without "
"burning a full overnight run.",
)
+@click.option(
+ "--no-push",
+ is_flag=True,
+ default=False,
+ help="Commit completed specs locally but do not push. Useful for "
+ "review-before-push workflows or CI jobs that should leave commits on disk.",
+)
@click.option(
"--no-tui", "--plain", "no_tui",
is_flag=True,
@@ -304,16 +311,18 @@ def agents() -> None:
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, workers: int,
- worktree: bool, model: str, agent: str, verifier_model: str | None, subagents: bool,
- idle_timeout: float, max_duration: int, max_tokens: int) -> None:
+def run(max_iterations: int, resume: bool, dry_run: bool, no_push: 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, 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."""
run_cmd(
max_iterations=max_iterations,
resume=resume,
dry_run=dry_run,
+ no_push=no_push,
no_tui=no_tui,
max_tokens_per_iteration=max_tokens_per_iteration,
max_turns_per_iteration=max_turns_per_iteration,
diff --git a/src/owloop/cli_options.py b/src/owloop/cli_options.py
index cafa9bf..2de9374 100644
--- a/src/owloop/cli_options.py
+++ b/src/owloop/cli_options.py
@@ -99,7 +99,9 @@ def _agent_run_options(f: Callable[..., Any]) -> Callable[..., Any]:
"--subagents",
is_flag=True,
default=False,
- help="Split large iterations into Orient/Implement/Verify subagent phases.",
+ help="Split large iterations into Orient/Implement/Verify subagent phases. "
+ "Small/scoped specs (≤3 files by default) still run as a single agent "
+ "to save tokens.",
)(f)
f = click.option(
"--idle-timeout", type=float, default=DEFAULT_IDLE_TIMEOUT,
diff --git a/src/owloop/commands/run.py b/src/owloop/commands/run.py
index 6613030..bc5b26b 100644
--- a/src/owloop/commands/run.py
+++ b/src/owloop/commands/run.py
@@ -44,7 +44,7 @@ def _run_engine(
ascii: bool = False, no_color: bool = False, compact: bool = False,
verifier_model: str | None = None, subagents: bool = False,
session_id: str | None = None, resume: bool = False,
- no_tui: bool = False, dry_run: bool = False,
+ no_tui: bool = False, dry_run: bool = False, no_push: bool = False,
max_tokens_per_iteration: int = 0,
max_turns_per_iteration: int = 0,
max_budget_usd: float = 0.0,
@@ -73,6 +73,7 @@ def _run_engine(
resume=resume,
no_tui=no_tui,
dry_run=dry_run,
+ no_push=no_push,
max_tokens_per_iteration=max_tokens_per_iteration,
max_turns_per_iteration=max_turns_per_iteration,
max_budget_usd=max_budget_usd,
@@ -99,6 +100,7 @@ def _run_engine(
resume = kwargs["resume"]
no_tui = kwargs["no_tui"]
dry_run = kwargs["dry_run"]
+ no_push = kwargs["no_push"]
max_tokens_per_iteration = kwargs["max_tokens_per_iteration"]
max_turns_per_iteration = kwargs["max_turns_per_iteration"]
max_budget_usd = kwargs["max_budget_usd"]
@@ -129,6 +131,7 @@ def _run_engine(
session_id=session_id,
resume=resume,
dry_run=dry_run,
+ no_push=no_push,
keep_retrying=keep_retrying,
rollback=rollback,
notify_webhook=resolved_webhook,
@@ -277,11 +280,12 @@ def _print_dry_run_report(console: Console, summary: RunSummary) -> None:
def run_cmd(
- 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, workers: int,
- worktree: bool, model: str, agent: str, verifier_model: str | None, subagents: bool,
- idle_timeout: float, max_duration: int, max_tokens: int,
+ max_iterations: int, resume: bool, dry_run: bool, no_push: 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, 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."""
ascii, no_color, compact, verbose = _cli_options()
@@ -301,6 +305,7 @@ def run_cmd(
resume=resume,
no_tui=no_tui,
dry_run=dry_run,
+ no_push=no_push,
max_tokens_per_iteration=max_tokens_per_iteration,
max_turns_per_iteration=max_turns_per_iteration,
max_budget_usd=max_budget_usd,
diff --git a/src/owloop/config.py b/src/owloop/config.py
index 35e8fb1..ea6821c 100644
--- a/src/owloop/config.py
+++ b/src/owloop/config.py
@@ -32,11 +32,12 @@
"notify_webhook": str,
"no_tui": bool,
"dry_run": bool,
+ "no_push": bool,
}
# Boolean CLI flags default to False; config turns them on.
_BOOL_FLAG_KEYS: frozenset[str] = frozenset(
- {"notify_desktop", "rollback", "keep_retrying", "no_tui", "dry_run"}
+ {"notify_desktop", "rollback", "keep_retrying", "no_tui", "dry_run", "no_push"}
)
# Numeric CLI options use 0 as the "not set" sentinel.
diff --git a/src/owloop/engine.py b/src/owloop/engine.py
index 8039e34..49c2219 100644
--- a/src/owloop/engine.py
+++ b/src/owloop/engine.py
@@ -115,6 +115,7 @@ class TerminalState(str, Enum):
TAMPERED = "tampered"
INTERRUPTED = "interrupted"
FAILED = "failed"
+ SOFT_FAILURE = "soft_failure"
def __str__(self) -> str: # so f-strings / json render the value, not "TerminalState.SUCCESS"
return self.value
@@ -142,6 +143,7 @@ class StopReason(str, Enum):
INTERRUPTED = "interrupted"
PREFLIGHT_FAILED = "preflight_failed"
DIRTY_WORKSPACE_DECLINED = "dirty_workspace_declined"
+ SOFT_FAILURE = "soft_failure"
def __str__(self) -> str:
return self.value
@@ -181,6 +183,7 @@ def __str__(self) -> str:
StopReason.INTERRUPTED: TerminalState.INTERRUPTED,
StopReason.PREFLIGHT_FAILED: TerminalState.FAILED,
StopReason.DIRTY_WORKSPACE_DECLINED: TerminalState.FAILED,
+ StopReason.SOFT_FAILURE: TerminalState.SOFT_FAILURE,
}
@@ -349,6 +352,9 @@ class EngineConfig:
# When True, run exactly one iteration, skip push, revert any commit the
# iteration made, and produce a DryRunReport instead of looping.
dry_run: bool = False
+ # When True, commit and mark specs complete locally but never push. Useful
+ # for review-before-push workflows or CI dry runs that want real commits.
+ no_push: bool = False
# Completion notifications: fire a webhook and/or desktop notification when
# the run stops on an attention-worthy terminal state. None/False = off.
notify_webhook: str | None = None
@@ -532,8 +538,29 @@ def _log_line(self, text: str) -> None:
def _is_git_repo(self) -> bool:
return self._run_git("rev-parse", "--is-inside-work-tree").returncode == 0
+ _OWLOOP_OWNED_PREFIXES = (".owloop",)
+
+ def _is_path_owned(self, path: str) -> bool:
+ """Return True if a path belongs to owloop's own state directories."""
+ for prefix in self._OWLOOP_OWNED_PREFIXES:
+ if path == prefix or path.startswith(prefix + "/"):
+ return True
+ return False
+
def _is_dirty(self) -> bool:
- return bool(self._run_git("status", "--porcelain").stdout.strip())
+ result = self._run_git("status", "--porcelain")
+ for line in result.stdout.splitlines():
+ # Porcelain format: "XY path" or "XY orig -> dest".
+ path_part = line[3:].strip()
+ # If this is a rename, check both sides.
+ if " -> " in path_part:
+ source, _, dest = path_part.partition(" -> ")
+ if self._is_path_owned(source) and self._is_path_owned(dest):
+ continue
+ elif self._is_path_owned(path_part):
+ continue
+ return True
+ return False
def _resolve_main_repo_dir(self) -> Path:
if not self._is_git_repo():
@@ -1045,7 +1072,7 @@ def _guarded_hash(self, spec_name: str | None) -> str:
def _run_verification_gate(
self, iteration: int, spec_name: str | None, guard_before: str
- ) -> tuple[bool, bool, list[dict[str, Any]]]:
+ ) -> verification.GateResult:
"""Deterministically verify an iteration outside the agent's control.
Delegates to the shared gate in ``verification.py`` (the single
@@ -1060,7 +1087,7 @@ def _run_verification_gate(
if result.tampered:
self._emit("spec_tampered", iteration=iteration, spec=spec_name)
- return False, True, []
+ return verification.GateResult(passed=False, tampered=True, passed_count=0, failed_count=0)
if result.passed:
self._emit(
@@ -1068,6 +1095,14 @@ def _run_verification_gate(
iteration=iteration,
passed=result.passed_count,
)
+ elif result.soft_failure:
+ self._emit(
+ "verification_gate_soft_failure",
+ iteration=iteration,
+ passed=result.passed_count,
+ failed=result.failed_count,
+ commands=[f["command"] for f in result.failures],
+ )
else:
self._emit(
"verification_gate_failed",
@@ -1076,7 +1111,7 @@ def _run_verification_gate(
failed=result.failed_count,
commands=[f["command"] for f in result.failures],
)
- return result.passed, False, result.failures
+ return result
def _head(self) -> str:
return str(self._run_git("rev-parse", "HEAD").stdout).strip()
@@ -1144,6 +1179,23 @@ def _rollback_iteration(self, iteration: int, last_good: str) -> None:
patch=str(patch_path) if diff.stdout.strip() else None,
)
+ def _should_use_subagents(self, target_spec: str | None) -> bool:
+ """Skip expensive subagent orchestration for small/scoped specs.
+
+ Subagents are great for cross-file refactors, but for a single-file or
+ tightly-scoped change they burn tokens on coordination overhead. When
+ the spec declares a `## Files` scope that is at or below the threshold,
+ run a single agent iteration instead.
+ """
+ if not self.config.use_subagents:
+ return False
+ if not target_spec:
+ return True
+ scope = spec_queue.get_spec_file_scope(self.specs_dir / target_spec)
+ if not scope:
+ return True
+ return len(scope) > self.config.subagent_file_threshold
+
def run_iteration(self, iteration: int, target_spec: str | None = None) -> IterationResult:
owloop_dir = resolve_owloop_dir(self.cwd)
prompt_file = owloop_dir / "PROMPT_build.md"
@@ -1173,7 +1225,7 @@ def _on_line(line: str) -> None:
raise IterationTokenLimitExceededError(iteration_tokens)
try:
- if self.config.use_subagents:
+ if self._should_use_subagents(target_spec):
orchestrator = SubagentOrchestrator(
self.adapter, self.verifier_adapter, self.cwd, on_line=_on_line
)
@@ -1494,7 +1546,7 @@ def run(self) -> RunSummary:
if self.config.dry_run:
self._append_run_note(iteration, result.success, result.summary)
- acceptance_passed, acceptance_failed, _ = verification.run_acceptance_criteria(
+ acceptance_passed, acceptance_failed, _, _, _ = verification.run_acceptance_criteria(
self.cwd, self.specs_dir, active_spec
)
current_head = self._head()
@@ -1531,23 +1583,26 @@ def run(self) -> RunSummary:
break
# ── Deterministic verification gate (engine-owned) ──
- gate_passed = False
- tampered = False
- gate_failures: list[dict[str, Any]] = []
+ gate_result = verification.GateResult(passed=False, tampered=False, passed_count=0, failed_count=0)
if result.promise_state == "DONE":
- gate_passed, tampered, gate_failures = self._run_verification_gate(
+ gate_result = self._run_verification_gate(
iteration, active_spec, guard_before
)
# Shell-first ordering: the expensive LLM verifier runs only
# on work that already survived the mechanical gate.
if (
- gate_passed
+ gate_result.passed
and self.verifier_adapter is not None
and not self.config.use_subagents
):
- gate_passed = self._apply_llm_verifier(iteration, result)
+ gate_result = verification.GateResult(
+ passed=self._apply_llm_verifier(iteration, result),
+ tampered=False,
+ passed_count=gate_result.passed_count,
+ failed_count=gate_result.failed_count,
+ )
- if gate_passed:
+ if gate_result.passed:
# Verified success: only now does the engine commit, mark the
# spec complete, and push — never the agent.
consecutive_failures = 0
@@ -1563,7 +1618,10 @@ def run(self) -> RunSummary:
self._append_run_note(iteration, True, note_summary)
self._mark_spec_complete(active_spec)
self._commit_iteration(iteration, active_spec)
- self._push(branch)
+ if self.config.no_push:
+ self._emit("push_skipped", branch=branch)
+ else:
+ self._push(branch)
spec_status = self._spec_status()
self._emit(
"iteration_end",
@@ -1602,8 +1660,22 @@ def run(self) -> RunSummary:
# iteration is minutes of work; waiting here buys nothing.
continue
+ # ── Soft failure: functional checks pass, meta-check fails ──
+ if gate_result.soft_failure:
+ self._append_run_note(iteration, False, result.summary)
+ diff = self._run_git("diff", last_good, "--", ".", ":!.owloop")
+ self._emit(
+ "soft_failure",
+ iteration=iteration,
+ spec=active_spec,
+ diff=diff.stdout or "",
+ commands=[f["command"] for f in gate_result.failures],
+ )
+ stopped_reason = StopReason.SOFT_FAILURE
+ break
+
# ── Failure: classify, record feedback, roll back, stop on a stall ──
- if tampered:
+ if gate_result.tampered:
failure_reason = FailureReason.TAMPERED
elif result.timed_out:
failure_reason = FailureReason.TIMEOUT
@@ -1622,7 +1694,7 @@ def run(self) -> RunSummary:
self._append_run_note(iteration, False, result.summary)
# Written before rollback; .owloop/ is excluded from the reset,
# so the next iteration's prompt starts from this diagnosis.
- self._write_failure_feedback(iteration, failure_reason, result, gate_failures)
+ self._write_failure_feedback(iteration, failure_reason, result, gate_result.failures)
self._rollback_iteration(iteration, last_good)
consecutive_failures += 1
@@ -1700,10 +1772,29 @@ def run(self) -> RunSummary:
dry_run_report=dry_run_report,
)
self._write_summary(summary)
+ self._generate_report()
self._notify(summary)
self._close_append_handles()
return summary
+ def _generate_report(self) -> None:
+ """Generate a static HTML report inside the worktree (best-effort)."""
+ try:
+ from owloop.report import ReportGenerator
+ from owloop.report_ai import ReportInsights
+
+ report_path = self.cwd / ".owloop" / "reports" / "owloop_report_latest.html"
+ report_path.parent.mkdir(parents=True, exist_ok=True)
+ generator = ReportGenerator(self.cwd)
+ generator.generate(
+ output_path=report_path,
+ insights=ReportInsights(),
+ use_tailwind=False,
+ )
+ except Exception:
+ # Report generation must never change the run outcome.
+ pass
+
def _notify(self, summary: RunSummary) -> None:
"""Fire completion notifications for the finished run (best-effort)."""
if not self.config.notify_webhook and not self.config.notify_desktop:
diff --git a/src/owloop/report.py b/src/owloop/report.py
index f5df43f..e896068 100644
--- a/src/owloop/report.py
+++ b/src/owloop/report.py
@@ -52,6 +52,24 @@ def _load_summary(self) -> dict[str, Any]:
if self.summary_path.exists():
with self.summary_path.open(encoding="utf-8") as f:
return json.load(f) # type: ignore[no-any-return]
+ # The engine writes the detailed summary inside the worktree; the main
+ # repo only gets a lightweight session file. Fall back to it so reports
+ # generated from the main repo still show branch/iterations/status.
+ session_path = self.summary_path.with_name("session_latest.json")
+ if session_path.exists():
+ with session_path.open(encoding="utf-8") as f:
+ session = json.load(f) # type: ignore[no-any-return]
+ status = session.get("status", "unknown")
+ # The session file uses "completed" as a generic completion flag;
+ # the classifier expects the granular stopped_reason "success".
+ stopped_reason = "success" if status == "completed" else status
+ return {
+ "iterations": session.get("iterations", 0),
+ "branch": session.get("branch", "unknown"),
+ "tokens_used": session.get("tokens_used", 0),
+ "estimated_cost_usd": 0,
+ "stopped_reason": stopped_reason,
+ }
return {}
def _load_events(self) -> list[dict[str, Any]]:
diff --git a/src/owloop/reporter.py b/src/owloop/reporter.py
index f46f0f7..e4b1e52 100644
--- a/src/owloop/reporter.py
+++ b/src/owloop/reporter.py
@@ -131,6 +131,11 @@ def on_event(self, kind: str, data: dict) -> None:
f"[{_brand.RED}]{self._mark('fail')} verification gate failed "
f"({data.get('failed', 0)} of {data.get('passed', 0) + data.get('failed', 0)} checks)[/]"
)
+ elif kind == "soft_failure":
+ c.print(
+ f"[{_brand.AMBER}]{self._mark('warn')} soft failure: functional checks passed but a "
+ f"meta-check failed — work is preserved for review ({data.get('spec')})[/]"
+ )
elif kind == "iteration_rolled_back":
c.print(f"[{_brand.CYAN}]{self._mark('info')} rolled back to {data.get('to_commit')} (failed iteration discarded)[/]")
elif kind == "iteration_exhausted":
@@ -146,6 +151,8 @@ def on_event(self, kind: str, data: dict) -> None:
c.print(f"[{_brand.AMBER}]{self._mark('clock')} token budget reached ({data['tokens']:,} / {data['limit']:,}), stopping loop[/]")
elif kind == "push_retry":
c.print(f"[{_brand.AMBER}]{self._mark('warn')} push failed, creating remote branch {data['branch']}...[/]")
+ elif kind == "push_skipped":
+ c.print(f"[{_brand.CYAN}]{self._mark('info')} push skipped (--no-push); commits remain on {data['branch']}[/]")
elif kind == "interrupted":
c.print("\n[dim]owloop stopped[/]")
elif kind == "parallel_session_info":
diff --git a/src/owloop/spec_queue.py b/src/owloop/spec_queue.py
index 161d534..4a10e14 100644
--- a/src/owloop/spec_queue.py
+++ b/src/owloop/spec_queue.py
@@ -26,6 +26,7 @@
import fnmatch
import re
+from dataclasses import dataclass
from pathlib import Path
_COMPLETE_RE = re.compile(r"^(#{1,3} )?(\*\*)?Status(\*\*)?:\s+COMPLETE", re.MULTILINE)
@@ -48,10 +49,31 @@
r"^##\s+Files\s*$\n(.*?)(?=^#{1,2}\s|\Z)",
re.IGNORECASE | re.MULTILINE | re.DOTALL,
)
+_EXCLUSIONS_SECTION_RE = re.compile(
+ r"^##\s+Exclusions\s*$\n(.*?)(?=^#{1,2}\s|\Z)",
+ re.IGNORECASE | re.MULTILINE | re.DOTALL,
+)
DEFAULT_PRIORITY = 999
+@dataclass(frozen=True)
+class AcceptanceCriterion:
+ """A single shell command parsed from a spec's Acceptance Criteria bullet.
+
+ The optional ``expect_no_output`` flag is set when the bullet carries an
+ expectation such as ``→ no output``. In that case the verifier treats a
+ command that exits 0/1 with empty stdout as passing, instead of requiring a
+ zero exit code (which grep naturally returns when it finds no matches).
+ """
+
+ command: str
+ expect_no_output: bool = False
+
+
+_NO_OUTPUT_RE = re.compile(r"(?:→|->)\s*no output", re.IGNORECASE)
+
+
def get_root_specs(specs_dir: Path) -> list[Path]:
if not specs_dir.is_dir():
return []
@@ -158,10 +180,12 @@ def get_spec_dependencies(spec_file: Path, specs: list[Path]) -> list[Path]:
return resolved
-def get_acceptance_criteria_commands(spec_file: Path) -> list[str]:
+def get_acceptance_criteria_commands(spec_file: Path) -> list[AcceptanceCriterion]:
"""Extract the first backtick-quoted shell command from each Acceptance Criteria bullet.
Bullets without a backtick-quoted command (free-form descriptions) are skipped.
+ A trailing expectation such as ``→ no output`` is parsed so the verifier can
+ treat grep-style "no match" exit codes as passing.
"""
if not spec_file.is_file():
return []
@@ -170,7 +194,7 @@ def get_acceptance_criteria_commands(spec_file: Path) -> list[str]:
if match is None:
return []
- commands: list[str] = []
+ commands: list[AcceptanceCriterion] = []
for line in match.group(1).splitlines():
stripped = line.strip()
if "`" not in stripped:
@@ -179,8 +203,11 @@ def get_acceptance_criteria_commands(spec_file: Path) -> list[str]:
if len(parts) < 3:
continue
command = parts[1].strip()
- if command:
- commands.append(command)
+ if not command:
+ continue
+ tail = "".join(parts[2:]).strip()
+ expect_no_output = bool(_NO_OUTPUT_RE.search(tail))
+ commands.append(AcceptanceCriterion(command, expect_no_output))
return commands
@@ -326,6 +353,25 @@ def get_spec_file_scope(spec_file: Path) -> list[str]:
return scope
+def get_spec_exclusions(spec_file: Path) -> list[str]:
+ """Return the path/glob tokens listed under a spec's ``## Exclusions`` section.
+
+ Missing/empty sections or ``none`` entries yield an empty list.
+ """
+ if not spec_file.is_file():
+ return []
+ content = spec_file.read_text(encoding="utf-8", errors="replace")
+ match = _EXCLUSIONS_SECTION_RE.search(content)
+ if match is None:
+ return []
+ exclusions: list[str] = []
+ for item in _LIST_ITEM_RE.findall(match.group(1)):
+ cleaned = item.strip().strip("`").strip()
+ if cleaned and cleaned.lower() != "none":
+ exclusions.append(cleaned)
+ return exclusions
+
+
def _paths_conflict(a: str, b: str) -> bool:
"""True if two path/glob tokens could touch a common file."""
if a == b:
diff --git a/src/owloop/tui.py b/src/owloop/tui.py
index e9424b8..af996ef 100644
--- a/src/owloop/tui.py
+++ b/src/owloop/tui.py
@@ -344,8 +344,16 @@ def _handle(self, kind: str, data: dict) -> None:
s.current_action = "Maximum run time reached"
self._log(f"⏱ reached maximum run time ({data['minutes']} min), stopping loop")
self._flash("⏱ time up", f"bold {AMBER}")
+ elif kind == "soft_failure":
+ s.done = True
+ s.phase = "stuck"
+ s.current_action = f"Soft failure on {data.get('spec')} — review preserved diff"
+ self._log(f"⚠ soft failure on {data.get('spec')}: meta-check failed; work preserved for review")
+ self._flash("⚠ soft failure — review diff", f"bold {AMBER}")
elif kind == "push_retry":
self._log(f"push failed, creating remote branch {data['branch']}...")
+ elif kind == "push_skipped":
+ self._log(f"push skipped (--no-push); commits remain on {data['branch']}")
elif kind == "iteration_end":
if "specs" in data:
s.specs = data["specs"]
diff --git a/src/owloop/verification.py b/src/owloop/verification.py
index 67a2a4c..318e1e8 100644
--- a/src/owloop/verification.py
+++ b/src/owloop/verification.py
@@ -32,6 +32,10 @@ class GateResult:
passed_count: int
failed_count: int
failures: list[dict[str, Any]] = field(default_factory=list)
+ # True when functional checks pass but a structural/meta check (e.g. grep)
+ # failed. The engine should surface the diff for human review rather than
+ # blindly rolling back work that may be correct.
+ soft_failure: bool = False
def run_commands(
@@ -60,14 +64,84 @@ def run_commands(
return passed, failed, failures
+def _tail_output(result: subprocess.CompletedProcess[str]) -> str:
+ output = f"{result.stdout or ''}\n{result.stderr or ''}".strip()
+ return "\n".join(output.splitlines()[-30:])[-2000:]
+
+
def run_acceptance_criteria(
cwd: Path, specs_dir: Path, spec_name: str | None
-) -> tuple[int, int, list[dict[str, Any]]]:
- """Run a spec's Acceptance Criteria shell commands; count passes vs failures."""
+) -> tuple[int, int, list[dict[str, Any]], int, int]:
+ """Run a spec's Acceptance Criteria shell commands; count passes vs failures.
+
+ Returns ``(passed, failed, failures, code_failed, meta_failed)`` where
+ ``code_failed`` counts functional checks (tests, lints) and ``meta_failed``
+ counts structural checks such as ``grep ... → no output``.
+ """
+ if not spec_name:
+ return 0, 0, [], 0, 0
+
+ criteria = spec_queue.get_acceptance_criteria_commands(specs_dir / spec_name)
+ passed = failed = code_failed = meta_failed = 0
+ failures: list[dict[str, Any]] = []
+ for criterion in criteria:
+ result = subprocess.run( # noqa: S602
+ criterion.command, shell=True, cwd=cwd, capture_output=True, text=True,
+ )
+ if criterion.expect_no_output:
+ # grep-style tools exit 1 when they find no matches; the expectation
+ # is about empty stdout, not the exit code.
+ ok = result.stdout.strip() == "" and result.returncode in (0, 1)
+ is_meta = True
+ else:
+ ok = result.returncode == 0
+ is_meta = False
+
+ if ok:
+ passed += 1
+ else:
+ failed += 1
+ if is_meta:
+ meta_failed += 1
+ else:
+ code_failed += 1
+ failures.append(
+ {
+ "command": criterion.command,
+ "returncode": result.returncode,
+ "output": _tail_output(result),
+ }
+ )
+ return passed, failed, failures, code_failed, meta_failed
+
+
+def _restore_exclusions(cwd: Path, specs_dir: Path, spec_name: str | None) -> None:
+ """Revert any tracked files listed in the spec's Exclusions section.
+
+ Acceptance-criteria commands (e.g. ``uv run --with pytest``) can mutate
+ files the spec promised not to touch, such as ``uv.lock``. This best-effort
+ cleanup restores those tracked files to HEAD so they never leak into the
+ iteration's commit.
+ """
if not spec_name:
- return 0, 0, []
- commands = spec_queue.get_acceptance_criteria_commands(specs_dir / spec_name)
- return run_commands(cwd, commands)
+ return
+ exclusions = spec_queue.get_spec_exclusions(specs_dir / spec_name)
+ if not exclusions:
+ return
+ for pattern in exclusions:
+ result = subprocess.run(
+ ["git", "ls-files", "--", pattern],
+ cwd=cwd,
+ capture_output=True,
+ text=True,
+ )
+ files = [f.strip() for f in result.stdout.splitlines() if f.strip()]
+ if files:
+ subprocess.run(
+ ["git", "checkout", "--", *files],
+ cwd=cwd,
+ capture_output=True,
+ )
def guarded_hash(cwd: Path, specs_dir: Path, spec_name: str | None) -> str:
@@ -94,16 +168,26 @@ def run_gate(
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, acc_failures = run_acceptance_criteria(cwd, specs_dir, spec_name)
+ acc_passed, acc_failed, acc_failures, acc_code_failed, acc_meta_failed = run_acceptance_criteria(
+ cwd, specs_dir, spec_name
+ )
+ _restore_exclusions(cwd, specs_dir, spec_name)
+
bp_commands = [cmd.command for cmd in load_backpressure(cwd)]
bp_passed, bp_failed, bp_failures = run_commands(cwd, bp_commands)
+ _restore_exclusions(cwd, specs_dir, spec_name)
passed_count = acc_passed + bp_passed
failed_count = acc_failed + bp_failed
+ # Soft failure: functional checks (code + backpressure) pass, but a
+ # structural/meta check failed. Surface for human review instead of rolling
+ # back potentially-correct work.
+ soft_failure = failed_count > 0 and acc_code_failed == 0 and bp_failed == 0
return GateResult(
passed=failed_count == 0,
tampered=False,
passed_count=passed_count,
failed_count=failed_count,
failures=acc_failures + bp_failures,
+ soft_failure=soft_failure,
)
diff --git a/tests/test_engine.py b/tests/test_engine.py
index bf156c5..9d6429d 100644
--- a/tests/test_engine.py
+++ b/tests/test_engine.py
@@ -531,6 +531,43 @@ def test_resolve_worktree_session_resume_falls_back_to_latest_branch(tmp_path: P
assert branch == "owloop/20260706-xyz789"
+def test_resolve_worktree_session_includes_spec_slug(tmp_path: Path) -> None:
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ adapter = MockAdapter()
+ engine = _make_engine(repo, adapter)
+
+ specs = repo / ".owloop" / "specs"
+ specs.mkdir(parents=True)
+ (specs / "04-extract-issue-service.md").write_text(
+ "# Spec: extract issue service\n", encoding="utf-8"
+ )
+
+ session_id, branch, path = engine._resolve_worktree_session()
+
+ assert branch.startswith("owloop/")
+ assert "extract-issue-service" in branch
+ assert session_id in branch
+ assert "extract-issue-service" in str(path)
+
+
+def test_resolve_worktree_session_slugifies_title(tmp_path: Path) -> None:
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ adapter = MockAdapter()
+ engine = _make_engine(repo, adapter)
+
+ specs = repo / ".owloop" / "specs"
+ specs.mkdir(parents=True)
+ (specs / "02-Hello World_Refactor.md").write_text(
+ "# Spec: Hello World Refactor\n", encoding="utf-8"
+ )
+
+ session_id, branch, path = engine._resolve_worktree_session()
+
+ assert "hello-world-refactor" in branch
+
+
def _read_jsonl(path: Path) -> list[dict]:
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
@@ -753,6 +790,119 @@ def test_dry_run_reverts_commit_and_skips_push(tmp_path: Path, monkeypatch) -> N
assert summary.dry_run_report.acceptance_passed == 1
+class _FileWritingAdapter(MockAdapter):
+ """Adapter that writes a file but leaves committing to the engine."""
+
+ def __init__(self, repo: Path, **kwargs) -> None:
+ super().__init__(**kwargs)
+ self._repo = repo
+
+ def run(self, prompt: str, cwd: Path, *, on_line=None) -> AgentResult:
+ (self._repo / "agent_change.txt").write_text("changed", encoding="utf-8")
+ return super().run(prompt, cwd, on_line=on_line)
+
+
+def test_no_push_commits_but_skips_remote_push(tmp_path: Path, monkeypatch) -> None:
+ """--no-push should still commit and complete the spec, but never call _push."""
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ _git_init(repo)
+ (repo / ".owloop" / "specs").mkdir(parents=True)
+ (repo / ".owloop" / "specs" / "01-test.md").write_text(
+ "# spec\n\n## Acceptance Criteria\n- `true`\n", encoding="utf-8"
+ )
+ monkeypatch.setattr("owloop.engine.time.sleep", lambda _: None)
+
+ original_head = subprocess.run(
+ ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True
+ ).stdout.strip()
+
+ adapter = _FileWritingAdapter(
+ repo,
+ responses=[
+ AgentResult(
+ stdout="ok\nDONE",
+ returncode=0,
+ success=True,
+ has_completion_signal=True,
+ done_signal="DONE",
+ )
+ ],
+ )
+ engine = _make_engine(repo, adapter, no_push=True)
+
+ push_calls: list[str] = []
+ monkeypatch.setattr(engine, "_push", lambda branch: push_calls.append(branch))
+ skipped: list[str] = []
+ engine.on_event = lambda k, d: skipped.append(k) if k == "push_skipped" else None
+
+ summary = engine.run()
+
+ assert summary.stopped_reason == "success"
+ assert push_calls == []
+ assert "push_skipped" in skipped
+ assert "**Status**: COMPLETE" in (repo / ".owloop" / "specs" / "01-test.md").read_text()
+ head_after = subprocess.run(
+ ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True
+ ).stdout.strip()
+ assert head_after != original_head
+ assert (repo / "agent_change.txt").is_file()
+
+
+def test_soft_failure_preserves_work_when_meta_check_fails(tmp_path: Path, monkeypatch) -> None:
+ """A functional pass + meta-check fail should stop for review without rollback."""
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ _git_init(repo)
+ (repo / "marker.txt").write_text("removed\n", encoding="utf-8")
+ subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
+ subprocess.run(["git", "commit", "-m", "add marker"], cwd=repo, check=True, capture_output=True)
+
+ (repo / ".owloop" / "specs").mkdir(parents=True)
+ (repo / ".owloop" / "specs" / "01-test.md").write_text(
+ "# spec\n\n"
+ "## Acceptance Criteria\n"
+ "- `true`\n"
+ "- `grep removed marker.txt` → no output\n",
+ encoding="utf-8",
+ )
+ monkeypatch.setattr("owloop.engine.time.sleep", lambda _: None)
+
+ original_head = subprocess.run(
+ ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True
+ ).stdout.strip()
+
+ adapter = _FileWritingAdapter(
+ repo,
+ responses=[
+ AgentResult(
+ stdout="ok\nDONE",
+ returncode=0,
+ success=True,
+ has_completion_signal=True,
+ done_signal="DONE",
+ )
+ ],
+ )
+ engine = _make_engine(repo, adapter)
+
+ soft_events: list[dict] = []
+ engine.on_event = lambda k, d: soft_events.append(d) if k == "soft_failure" else None
+
+ summary = engine.run()
+
+ assert summary.stopped_reason == "soft_failure"
+ assert summary.terminal_state == "soft_failure"
+ assert "**Status**: COMPLETE" not in (repo / ".owloop" / "specs" / "01-test.md").read_text()
+ head_after = subprocess.run(
+ ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True
+ ).stdout.strip()
+ assert head_after == original_head # no rollback
+ assert (repo / "agent_change.txt").is_file()
+ assert len(soft_events) == 1
+ assert "grep removed marker.txt" in soft_events[0]["commands"]
+
+
class _InterruptingAdapter(MockAdapter):
"""Adapter that completes one iteration, then simulates Ctrl+C on the next."""
@@ -762,6 +912,113 @@ def run(self, prompt: str, cwd: Path, *, on_line=None) -> AgentResult:
return super().run(prompt, cwd, on_line=on_line)
+def _make_repo_with_spec(tmp_path: Path, files_scope: list[str]) -> Path:
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
+ subprocess.run(
+ ["git", "config", "user.email", "test@example.com"],
+ cwd=repo, check=True, capture_output=True,
+ )
+ subprocess.run(
+ ["git", "config", "user.name", "Test"],
+ cwd=repo, check=True, capture_output=True,
+ )
+ (repo / "README.md").write_text("# test", encoding="utf-8")
+ specs = repo / ".owloop" / "specs"
+ specs.mkdir(parents=True)
+ scope_lines = "\n".join(f"- `{f}`" for f in files_scope)
+ (specs / "01-test.md").write_text(
+ f"# spec\n\n## Files\n{scope_lines}\n\n## Acceptance Criteria\n- `true`\n",
+ encoding="utf-8",
+ )
+ subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
+ subprocess.run(["git", "commit", "-m", "initial"], cwd=repo, check=True, capture_output=True)
+ return repo
+
+
+def test_subagents_skipped_for_small_scoped_spec(tmp_path: Path, monkeypatch) -> None:
+ """Small/scoped specs should not pay the subagent orchestration token cost."""
+ repo = _make_repo_with_spec(tmp_path, ["src/small.py"])
+ monkeypatch.setattr("owloop.engine.time.sleep", lambda _: None)
+
+ subagent_calls: list[tuple[str, ...]] = []
+
+ class _FakeOrchestrator:
+ def __init__(self, adapter, verifier, cwd, on_line=None) -> None:
+ subagent_calls.append((str(adapter), str(verifier), str(cwd)))
+
+ def run(self) -> AgentResult:
+ return AgentResult(
+ stdout="ok\nDONE",
+ returncode=0,
+ success=True,
+ has_completion_signal=True,
+ done_signal="DONE",
+ )
+
+ monkeypatch.setattr("owloop.engine.SubagentOrchestrator", _FakeOrchestrator)
+
+ engine = _make_engine(
+ repo,
+ MockAdapter(
+ responses=[
+ AgentResult(
+ stdout="ok\nDONE",
+ returncode=0,
+ success=True,
+ has_completion_signal=True,
+ done_signal="DONE",
+ )
+ ]
+ ),
+ use_subagents=True,
+ )
+ monkeypatch.setattr(engine, "_push", lambda b: None)
+
+ summary = engine.run()
+
+ assert summary.stopped_reason == "success"
+ assert subagent_calls == []
+
+
+def test_subagents_used_for_large_scoped_spec(tmp_path: Path, monkeypatch) -> None:
+ """Specs touching many files still get the full subagent orchestration."""
+ repo = _make_repo_with_spec(
+ tmp_path, [f"src/file{i}.py" for i in range(5)]
+ )
+ monkeypatch.setattr("owloop.engine.time.sleep", lambda _: None)
+
+ subagent_calls: list[tuple[str, ...]] = []
+
+ class _FakeOrchestrator:
+ def __init__(self, adapter, verifier, cwd, on_line=None) -> None:
+ subagent_calls.append((str(adapter), str(verifier), str(cwd)))
+
+ def run(self) -> AgentResult:
+ return AgentResult(
+ stdout="ok\nDONE",
+ returncode=0,
+ success=True,
+ has_completion_signal=True,
+ done_signal="DONE",
+ )
+
+ monkeypatch.setattr("owloop.engine.SubagentOrchestrator", _FakeOrchestrator)
+
+ engine = _make_engine(
+ repo,
+ MockAdapter(responses=[]),
+ use_subagents=True,
+ )
+ monkeypatch.setattr(engine, "_push", lambda b: None)
+
+ summary = engine.run()
+
+ assert summary.stopped_reason == "success"
+ assert len(subagent_calls) == 1
+
+
def test_session_state_persisted_on_interrupt(tmp_path: Path) -> None:
repo = tmp_path / "repo"
repo.mkdir()
@@ -1030,3 +1287,40 @@ def test_trim_run_notes_keeps_only_newest_entries() -> None:
# Under the cap (or unstructured content), nothing is touched.
assert trim_run_notes("free-form note", max_entries=3) == "free-form note"
assert trim_run_notes(notes, max_entries=20) == notes
+
+
+def test_is_dirty_ignores_owloop_directory(tmp_path: Path) -> None:
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ _git_init(repo)
+ adapter = MockAdapter()
+ engine = _make_engine(repo, adapter)
+
+ (repo / ".owloop" / "specs").mkdir(parents=True)
+ (repo / ".owloop" / "specs" / "01-test.md").write_text("# test", encoding="utf-8")
+
+ assert not engine._is_dirty()
+
+
+def test_is_dirty_detects_other_untracked_files(tmp_path: Path) -> None:
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ _git_init(repo)
+ adapter = MockAdapter()
+ engine = _make_engine(repo, adapter)
+
+ (repo / "untracked.txt").write_text("hi", encoding="utf-8")
+
+ assert engine._is_dirty()
+
+
+def test_is_dirty_detects_modified_tracked_file(tmp_path: Path) -> None:
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ _git_init(repo)
+ adapter = MockAdapter()
+ engine = _make_engine(repo, adapter)
+
+ (repo / "README.md").write_text("modified", encoding="utf-8")
+
+ assert engine._is_dirty()
diff --git a/tests/test_spec_queue.py b/tests/test_spec_queue.py
index fb15f04..6d16bd8 100644
--- a/tests/test_spec_queue.py
+++ b/tests/test_spec_queue.py
@@ -211,6 +211,50 @@ def test_get_acceptance_criteria_section_missing_returns_empty(tmp_path: Path) -
assert spec_queue.get_acceptance_criteria_section(spec) == ""
+# ── acceptance-criteria command extraction ──
+
+
+def test_get_acceptance_criteria_commands_extracts_plain_command(tmp_path: Path) -> None:
+ spec = tmp_path / "01-t.md"
+ spec.write_text(
+ "# Spec\n\n## Acceptance Criteria\n- `pytest -q`\n- `ruff check .`\n",
+ encoding="utf-8",
+ )
+ criteria = spec_queue.get_acceptance_criteria_commands(spec)
+ assert [c.command for c in criteria] == ["pytest -q", "ruff check ."]
+ assert all(not c.expect_no_output for c in criteria)
+
+
+def test_get_acceptance_criteria_commands_detects_no_output(tmp_path: Path) -> None:
+ spec = tmp_path / "01-t.md"
+ spec.write_text(
+ "# Spec\n\n## Acceptance Criteria\n"
+ "- `grep foo bar.txt` → no output\n"
+ "- `grep baz qux.txt` -> no output\n"
+ "- `echo hi` → at least 1 match\n",
+ encoding="utf-8",
+ )
+ criteria = spec_queue.get_acceptance_criteria_commands(spec)
+ assert [c.command for c in criteria] == [
+ "grep foo bar.txt",
+ "grep baz qux.txt",
+ "echo hi",
+ ]
+ assert criteria[0].expect_no_output is True
+ assert criteria[1].expect_no_output is True
+ assert criteria[2].expect_no_output is False
+
+
+def test_get_acceptance_criteria_commands_ignores_free_form_bullets(tmp_path: Path) -> None:
+ spec = tmp_path / "01-t.md"
+ spec.write_text(
+ "# Spec\n\n## Acceptance Criteria\n- just a free-form note\n- `true`\n",
+ encoding="utf-8",
+ )
+ criteria = spec_queue.get_acceptance_criteria_commands(spec)
+ assert [c.command for c in criteria] == ["true"]
+
+
# ── file-disjoint parallel scheduling (Phase 4) ──
diff --git a/tests/test_verifier.py b/tests/test_verifier.py
index 3e7f6b1..83d3d47 100644
--- a/tests/test_verifier.py
+++ b/tests/test_verifier.py
@@ -8,6 +8,7 @@
import subprocess
from pathlib import Path
+from owloop import verification
from owloop.adapters import AgentResult, MockAdapter
from owloop.engine import EngineConfig, OwloopEngine
@@ -138,3 +139,92 @@ def test_no_verifier_keeps_existing_behavior(tmp_path: Path, monkeypatch) -> Non
summary = engine.run()
assert summary.stopped_reason == "success"
+
+
+# ── deterministic gate: no-output expectations ──
+
+
+def test_run_acceptance_criteria_treats_grep_no_match_as_pass(tmp_path: Path) -> None:
+ """A `→ no output` criterion must pass when grep finds nothing (exit 1)."""
+ specs = tmp_path / ".owloop" / "specs"
+ specs.mkdir(parents=True)
+ spec = specs / "01-test.md"
+ target = tmp_path / "api.py"
+ target.write_text("def keep(): pass\n", encoding="utf-8")
+ spec.write_text(
+ "# Spec\n\n## Acceptance Criteria\n"
+ f"- `grep removed {target.name}` → no output\n",
+ encoding="utf-8",
+ )
+
+ passed, failed, failures, code_failed, meta_failed = verification.run_acceptance_criteria(
+ tmp_path, specs, "01-test.md"
+ )
+
+ assert passed == 1
+ assert failed == 0
+ assert failures == []
+ assert code_failed == 0
+ assert meta_failed == 0
+
+
+def test_run_acceptance_criteria_no_output_fails_on_nonempty_stdout(tmp_path: Path) -> None:
+ specs = tmp_path / ".owloop" / "specs"
+ specs.mkdir(parents=True)
+ spec = specs / "01-test.md"
+ target = tmp_path / "api.py"
+ target.write_text("def removed(): pass\n", encoding="utf-8")
+ spec.write_text(
+ "# Spec\n\n## Acceptance Criteria\n"
+ f"- `grep removed {target.name}` → no output\n",
+ encoding="utf-8",
+ )
+
+ passed, failed, failures, code_failed, meta_failed = verification.run_acceptance_criteria(
+ tmp_path, specs, "01-test.md"
+ )
+
+ assert passed == 0
+ assert failed == 1
+ assert len(failures) == 1
+ assert code_failed == 0
+ assert meta_failed == 1
+
+
+def test_run_gate_restores_excluded_tracked_files(tmp_path: Path) -> None:
+ """Acceptance-criteria commands must not leave mutations in spec-excluded files."""
+ repo = tmp_path / "repo"
+ repo.mkdir()
+ subprocess.run(["git", "init"], cwd=repo, check=True, capture_output=True)
+ subprocess.run(
+ ["git", "config", "user.email", "test@example.com"],
+ cwd=repo, check=True, capture_output=True,
+ )
+ subprocess.run(
+ ["git", "config", "user.name", "Test"],
+ cwd=repo, check=True, capture_output=True,
+ )
+
+ (repo / "uv.lock").write_text("original\n", encoding="utf-8")
+ (repo / "README.md").write_text("# test", encoding="utf-8")
+ subprocess.run(["git", "add", "."], cwd=repo, check=True, capture_output=True)
+ subprocess.run(["git", "commit", "-m", "init"], cwd=repo, check=True, capture_output=True)
+
+ specs = repo / ".owloop" / "specs"
+ specs.mkdir(parents=True)
+ spec = specs / "01-test.md"
+ spec.write_text(
+ "# Spec\n\n"
+ "## Acceptance Criteria\n"
+ "- `echo changed > uv.lock`\n\n"
+ "## Exclusions\n"
+ "- `uv.lock`\n",
+ encoding="utf-8",
+ )
+
+ guard_before = verification.guarded_hash(repo, specs, "01-test.md")
+ result = verification.run_gate(repo, specs, "01-test.md", guard_before)
+
+ assert result.tampered is False
+ assert result.passed is True
+ assert (repo / "uv.lock").read_text(encoding="utf-8") == "original\n"