feat: enforce pre-commit hooks in worker worktrees

.forge/quality-manifesto.md

@@ -8,6 +8,18 @@ exists, not just what it says.
 
 ## Rules
 
+### Q6. No `--no-verify` without a justified reason in the PR body. Pre-commit gates are not optional.
+
+Pre-commit exists to enforce the same quality gates locally that CI will
+enforce later. Bypassing it hides defects from the worker loop and turns
+review into the first real gate. If a bypass is genuinely required, the
+PR body MUST include a `## Pre-commit bypass justification` section that
+explains why the hook could not run.
+
+**Rationale:** see #158. Forge-loop had a pre-commit config but the hook
+was never installed in worker worktrees, so commits silently bypassed
+ruff, mypy, pyright, and formatting until CI or production found them.
+
 ### Q1. No shared mutable module-level state. Use a Container or per-instance State.
 
 Module-level globals (caches, counters, "current run" dicts, singleton

src/forge_loop/_testing/precommit.py

@@ -0,0 +1,26 @@
+"""Test fakes for the pre-commit adapter."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from forge_loop.precommit import git_hook_path
+
+
+class FakePreCommitRunner:
+    def __init__(self, *, available: bool = True, returncode: int = 0) -> None:
+        self.available = available
+        self.returncode = returncode
+        self.installs: list[Path] = []
+
+    def is_available(self) -> bool:
+        return self.available
+
+    def install(self, repo: Path) -> int:
+        self.installs.append(repo)
+        if self.returncode == 0:
+            hook = git_hook_path(repo)
+            hook.parent.mkdir(parents=True, exist_ok=True)
+            hook.write_text("#!/bin/sh\necho fake pre-commit\nexit 0\n")
+            hook.chmod(0o755)
+        return self.returncode

src/forge_loop/briefs/critic.md.tmpl

@@ -13,7 +13,11 @@ DO:
    acceptance criteria. Note any "Acceptance" or "Out of scope" sections.
 2. Read the PR diff: `gh pr diff {pr_url}`.
 3. Read the PR description: `gh pr view {pr_url} --json title,body,additions,deletions`.
-4. Decide overall + per-finding. Use the rubric:
+4. Check worker commit commands / PR text for `git commit --no-verify`.
+   If `--no-verify` appears and the PR body lacks a non-empty
+   `## Pre-commit bypass justification` section, emit a sev1 finding
+   tagged `precommit_bypass`. Pre-commit gates are not optional.
+5. Decide overall + per-finding. Use the rubric:
    - sev1 = correctness/security bug, missing acceptance criterion,
      or test that doesn't actually exercise the change. Blocks merge.
    - sev2 = meaningful concern (untested error path, weak assertion,
@@ -75,4 +79,4 @@ Hard rules:
   sev3 noting what you reviewed.
 - If genuinely uncertain about a finding, lean toward emitting it as sev3
   rather than swallowing it.
-- The JSON object MUST be on the LAST line of your output and parse cleanly.
+- The JSON object MUST be on the LAST line of your output and parse cleanly.

src/forge_loop/briefs/worker.md.tmpl

@@ -54,6 +54,9 @@ CONTRACT (content-grade — NOT the smallest possible diff):
    the discovered ones. Avoid full-suite runs; target the specific tests
    you touched.
 7. Run your project's pre-commit gates (lint, format, type-check).
+   `git commit` MUST run without `--no-verify`. Any `--no-verify` use
+   REQUIRES a justified reason in the PR body under a
+   `## Pre-commit bypass justification` heading.
 8. git add -A; git commit referencing #{n}. {coauthor_line}
    The commit message MUST have a body (the "why"), not just a one-line title.
 9. git push -u origin <branch>.

src/forge_loop/cli_operator_commands.py

@@ -70,7 +70,6 @@ def _cmd_cluster_status(self, args: SimpleNamespace) -> int:
         )
         return 2
 
-
     def _cmd_doctor(self, _args: SimpleNamespace) -> int:
         """One-shot health check with a Rich table."""
         import glob
@@ -124,7 +123,9 @@ def line(status: str, label: str, detail: str = "") -> None:
 
         tmux_bin = shutil.which("tmux")
         if tmux_bin is None:
-            line("yellow", "tmux not installed", "operator usually runs forge-loop in a tmux session")
+            line(
+                "yellow", "tmux not installed", "operator usually runs forge-loop in a tmux session"
+            )
         else:
             try:
                 r = _sp.run([tmux_bin, "ls"], capture_output=True, text=True, timeout=5)
@@ -276,6 +277,10 @@ def _cmd_init(self, args: SimpleNamespace) -> int:
             typer.echo(f"  + {path}")
         for path in result["skipped"]:
             typer.echo(f"  · skipped (exists; pass --force to overwrite): {path}")
+        for outcome in result.get("precommit", []):
+            typer.echo(f"  · {outcome}")
+        for hint in result.get("precommit_hint", []):
+            typer.echo(f"    hint: {hint}")
 
         if args.create_labels:
             created = _init_mod.ensure_labels_via_gh(repo, _init_mod.DEFAULT_LABELS)

src/forge_loop/critic.py

@@ -30,6 +30,7 @@
 import re
 import time
 from collections.abc import Callable
+from contextlib import suppress
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -39,6 +40,22 @@
 VALID_OVERALL = {"approve", "request_changes", "block"}
 VALID_SEVERITY = {"sev1", "sev2", "sev3"}
 VALID_CATEGORY = {"correctness", "security", "style", "tests", "docs", "product"}
+PRECOMMIT_BYPASS_TAG = "precommit_bypass"
+_NO_VERIFY_RE = re.compile(r"\bgit(?:\s+-[cC]\s+\S+)*\s+commit\b[^\n]*\s(?:--no-verify|-n)\b")
+_BODY_NO_VERIFY_ACTION_RE = re.compile(
+    r"\b(?:i\s+)?(?:ran|run|used|use|called|call|executed|execute)\s+"
+    r"git(?:\s+-[cC]\s+\S+)*\s+commit\b[^\n]*\s(?:--no-verify|-n)\b",
+    re.IGNORECASE,
+)
+_NO_VERIFY_STATIC_CONTEXT_RE = re.compile(
+    r"\b(?:flag|detect|test|tests|rule|brief|manifesto|requires|mention|mentions|"
+    r"statement|statements)\b",
+    re.IGNORECASE,
+)
+_BYPASS_HEADING_RE = re.compile(
+    r"^##\s+Pre-commit bypass justification\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
 
 
 @dataclass
@@ -58,8 +75,10 @@ class ManifestoViolation:
 
     def is_valid(self) -> bool:
         return (
-            isinstance(self.rule_id, str) and bool(self.rule_id.strip())
-            and isinstance(self.manifesto, str) and bool(self.manifesto.strip())
+            isinstance(self.rule_id, str)
+            and bool(self.rule_id.strip())
+            and isinstance(self.manifesto, str)
+            and bool(self.manifesto.strip())
             and isinstance(self.quote, str)
             and isinstance(self.suggested_fix, str)
             and self.severity in VALID_SEVERITY
@@ -103,15 +122,13 @@ def severities(self) -> set[str]:
         return {f.severity for f in self.findings}
 
     def has_sev1(self) -> bool:
-        return (
-            any(f.severity == "sev1" for f in self.findings)
-            or any(v.severity == "sev1" for v in self.manifesto_violations)
+        return any(f.severity == "sev1" for f in self.findings) or any(
+            v.severity == "sev1" for v in self.manifesto_violations
         )
 
     def has_sev2(self) -> bool:
-        return (
-            any(f.severity == "sev2" for f in self.findings)
-            or any(v.severity == "sev2" for v in self.manifesto_violations)
+        return any(f.severity == "sev2" for f in self.findings) or any(
+            v.severity == "sev2" for v in self.manifesto_violations
         )
 
     def has_sev1_manifesto_violation(self) -> bool:
@@ -129,6 +146,134 @@ class CriticOutcome:
     parse_retries: int = 0
 
 
+def detect_precommit_bypass(commit_text: str, *, pr_body: str) -> CriticReport:
+    """Flag `git commit --no-verify` unless the PR body justifies it."""
+
+    if not _has_no_verify_command(commit_text) and not _has_no_verify_body_action(pr_body):
+        return CriticReport(overall="approve", findings=[])
+    if _has_precommit_bypass_justification(pr_body):
+        return CriticReport(overall="approve", findings=[])
+    return CriticReport(
+        overall="request_changes",
+        findings=[
+            Finding(
+                severity="sev1",
+                category="correctness",
+                file=None,
+                line=None,
+                message=(
+                    f"{PRECOMMIT_BYPASS_TAG}: `git commit --no-verify` requires a "
+                    "`## Pre-commit bypass justification` section in the PR body."
+                ),
+            )
+        ],
+    )
+
+
+def _has_no_verify_command(text: str) -> bool:
+    for line in _command_segments(text):
+        match = _NO_VERIFY_RE.search(line)
+        if match is not None and not _is_static_no_verify_context(line, match):
+            return True
+    return False
+
+
+def _worker_command_context(issue_number: int, logs_dir: Path) -> str:
+    chunks: list[str] = []
+    for pattern in (f"worker-{issue_number}-*.log", f"repair-{issue_number}-*.log"):
+        for path in sorted(logs_dir.glob(pattern)):
+            try:
+                lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
+            except OSError:
+                continue
+            for line in lines:
+                try:
+                    event = json.loads(line)
+                except json.JSONDecodeError:
+                    continue
+                item = event.get("item")
+                if not isinstance(item, dict):
+                    continue
+                if item.get("type") != "command_execution":
+                    continue
+                command = item.get("command")
+                if isinstance(command, str):
+                    chunks.append(command)
+    return "\n".join(chunks)
+
+
+def _command_segments(text: str) -> list[str]:
+    lines = text.splitlines()
+    segments: list[str] = []
+    i = 0
+    while i < len(lines):
+        line = lines[i]
+        segment = line.rstrip()
+        while segment.endswith("\\") and i + 1 < len(lines):
+            segment = f"{segment[:-1]} {lines[i + 1].strip()}"
+            i += 1
+        segments.append(segment)
+        if i + 1 < len(lines) and "git" in line and "commit" in line:
+            segments.append(f"{line.rstrip()} {lines[i + 1].strip()}")
+        i += 1
+    return segments
+
+
+def _has_no_verify_body_action(text: str) -> bool:
+    for line in text.splitlines():
+        match = _BODY_NO_VERIFY_ACTION_RE.search(line)
+        if match is not None and not _is_static_no_verify_context(line, match):
+            return True
+    return False
+
+
+def _is_static_no_verify_context(line: str, match: re.Match[str]) -> bool:
+    return _NO_VERIFY_STATIC_CONTEXT_RE.search(line[: match.start()]) is not None
+
+
+def _has_precommit_bypass_justification(pr_body: str) -> bool:
+    match = _BYPASS_HEADING_RE.search(pr_body)
+    if match is None:
+        return False
+    tail = pr_body[match.end() :]
+    body = tail.split("\n## ", 1)[0].strip()
+    return bool(body)
+
+
+def _fetch_pr_precommit_context(pr_url: str, repo: Path) -> tuple[str, str]:
+    """Return PR body plus commit metadata for deterministic local checks."""
+
+    from forge_loop import gh
+
+    return gh.pr_precommit_context(pr_url, repo)
+
+
+def _with_deterministic_precommit_findings(
+    report: CriticReport,
+    *,
+    pr_url: str,
+    repo: Path,
+    issue_number: int,
+    logs_dir: Path,
+) -> CriticReport:
+    pr_body, commit_text = _fetch_pr_precommit_context(pr_url, repo)
+    worker_text = _worker_command_context(issue_number, logs_dir)
+    if worker_text:
+        commit_text = f"{commit_text}\n{worker_text}"
+    deterministic = detect_precommit_bypass(commit_text, pr_body=pr_body)
+    if not deterministic.findings:
+        return report
+    overall = report.overall
+    if overall == "approve":
+        overall = deterministic.overall
+    return CriticReport(
+        overall=overall,
+        findings=[*report.findings, *deterministic.findings],
+        manifesto_violations=report.manifesto_violations,
+        raw=report.raw,
+    )
+
+
 def review_pr(
     pr_url: str,
     issue_number: int,
@@ -206,6 +351,13 @@ def review_pr(
                 stdout_tail=tail,
                 error=parse_error or result.error or "critic_parse_failed",
             )
+        report = _with_deterministic_precommit_findings(
+            report,
+            pr_url=pr_url,
+            repo=repo,
+            issue_number=issue_number,
+            logs_dir=logs_dir,
+        )
         verdict = _verdict_from_overall(report.overall)
         reasons = [f"[{f.severity}/{f.category}] {f.message}" for f in report.findings]
         return CriticOutcome(
@@ -237,10 +389,8 @@ def review_pr(
         # Mirror the final assistant text to disk so the existing
         # _tail(log_path, 500) read for stdout_tail keeps working and
         # operators can grep critic-*.log as before.
-        try:
+        with suppress(OSError):
             log_path.write_text(sdk_result.last_message or "")
-        except OSError:
-            pass
         if sdk_result.timed_out:
             return CriticOutcome(
                 verdict="error",
@@ -290,6 +440,13 @@ def review_pr(
             parse_retries=retries,
         )
 
+    report = _with_deterministic_precommit_findings(
+        report,
+        pr_url=pr_url,
+        repo=repo,
+        issue_number=issue_number,
+        logs_dir=logs_dir,
+    )
     verdict = _verdict_from_overall(report.overall)
     reasons = [f"[{f.severity}/{f.category}] {f.message}" for f in report.findings]
     return CriticOutcome(

src/forge_loop/events.py

@@ -41,6 +41,8 @@
 
 from pydantic import BaseModel, ConfigDict, Field
 
+from forge_loop.precommit import PreCommitInstallMethod
+
 _DEFAULT_DURABLE_MIRROR = object()
 
 
@@ -287,6 +289,16 @@ class StuckSweepDemotedEvent(EventBase):
     reason: str = ""
 
 
+@register_event
+class WorkerPreCommitInstalledEvent(EventBase):
+    """Pre-commit hook propagation result for one worker worktree."""
+
+    KIND: ClassVar[str] = "worker_precommit_installed"
+    worktree_path: str = ""
+    method: PreCommitInstallMethod
+    reason: str | None = None
+
+
 # ---------------------------------------------------------------------------
 # Emit + back-compat shim. ``emit`` is the typed path; ``append_event_with_
 # registry_check`` is the back-compat wrapper called by state.append_event.
@@ -430,6 +442,7 @@ def _default_durable_mirror(events_path: Path) -> tuple[object | None, str | Non
     "TickStartEvent",
     "WorkerSessionRecoveredEvent",
     "WorkerSessionTransitionEvent",
+    "WorkerPreCommitInstalledEvent",
     "WorktreeReapedEvent",
     "append_event_with_registry_check",
     "emit",