feat(controller): Phase 1d-1 — state machine + reaper + pickup guard

The deterministic spine of the master controller. State machine is
pure data with 6 load-bearing invariants enforced via property tests.
Reaper resets stale-heartbeat workflow_attempts to pending. Pickup
guard transitions workflows to STUCK when an attempt has been
re-pended too many times without success.

tools/controller/state_machine.py:

- KNOWN_STATES = 12; TERMINAL_STATES = {MERGED, ABANDONED, STUCK,
  CREATED_PR}. STUCK's only allowed exit is the operator-driven
  operator_unstick event (back to DISCOVERED).
- 32 TRANSITIONS entries covering DISCOVERED → ANALYZING →
  IMPLEMENTING ↔ AWAITING_CI / CONFLICT_RESOLVING / ESCALATING →
  REVIEWING → MERGING → MERGED. Plus pickup_exhausted exits from
  IMPLEMENTING/CONFLICT_RESOLVING/REVIEWING.
- 27 named events with descriptions. apply_event() lookup raises
  IllegalTransitionError (lists legal events from current state)
  or ValueError on unknown state (per v6 unknown-state guard).
- 6 LOAD-BEARING invariants for v1 (per v9 simplification):
  1. no_path_implementing_to_reviewing_skips_ci (Hard Rule #1
     constructional fix for the no-mans-land race)
  2. terminal_states_have_no_exits (only STUCK→operator_unstick OK)
  3. tier_monotonic_non_decreasing
  4. every_pr_workflow_includes_reviewing
  5. conflict_resolving_bounded (1st→IMPLEMENTING, 2nd→ESCALATING,
     3rd→STUCK; structurally encoded)
  6. escalation_deterministic
- reachable_from() honors cycles (DISCOVERED ∈ reachable(DISCOVERED)
  via STUCK→operator_unstick path; AWAITING_CI self-loops via
  ci_flake_retry).

tools/controller/reaper.py:

- reap_stale_attempts(): SELECT in_progress attempts whose
  lock_heartbeat_at + lock_ttl_seconds < NOW (per-row TTL respects
  per-role differences — estimator 180s, reviewer 720s, tier-2
  implementer 2160s). UPDATEs status='pending', clears lock columns,
  preserves pickup_count (the pickup guard handles that). Inserts
  controller_events row with reason='lock-ttl-expired' per reap.
- Dialect-portable: Postgres uses interval arithmetic; SQLite uses
  julianday(). Same logic either way.

tools/controller/pickup_guard.py:

- transition_exhausted_to_stuck(): finds attempts with status='pending'
  AND pickup_count >= MAX_PICKUPS (default 3 per v6 blocker fix)
  AND workflow not already terminal. Transitions workflow → STUCK,
  marks attempt as 'reaped', inserts controller_events with
  reason='attempt-pickup-exhausted' + pickup_count + max_pickups.

45 new tests:
- state_machine: basic shape (states partition, every transition uses
  known states + defined events), apply_event success/error paths,
  events_from + reachable_from helpers (including cycle awareness),
  per-invariant zero-violations against the live table, per-invariant
  monkeypatch-violations to prove the checks catch the bug class they
  claim to, parametrised sanity check "every non-terminal can reach
  some terminal".
- reaper: empty DB / fresh heartbeat / stale heartbeat reaped /
  per-row TTL respected / event row created / only-in-progress
  reaped / multiple stale attempts.
- pickup guard: empty DB / below limit / at limit / in-progress not
  checked / terminal workflow skipped / event payload content /
  default max_pickups matches v6.

Total: 229 controller tests; full auto_agents suite 2591 pass.

tests/auto_agents/controller/test_reaper_and_pickup_guard.py

@@ -0,0 +1,277 @@
+"""Tests for the reaper + pickup-exhaustion guard."""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import text
+
+from tools.controller.db import (
+    Workflow,
+    WorkflowAttempt,
+    build_engine,
+    create_all,
+    session_scope,
+)
+from tools.controller.pickup_guard import (
+    DEFAULT_MAX_PICKUPS,
+    PickupGuardReport,
+    transition_exhausted_to_stuck,
+)
+from tools.controller.reaper import ReaperReport, reap_stale_attempts
+
+
+@pytest.fixture
+def engine():
+    eng = build_engine("sqlite:///:memory:")
+    create_all(eng)
+    yield eng
+
+
+def _seed_workflow(
+    engine, *, current_state="IMPLEMENTING", current_tier=0,
+):
+    with session_scope(engine) as s:
+        w = Workflow(
+            kind="pr", owner="o", repo="r", entity_number=30,
+            current_state=current_state, current_tier=current_tier,
+        )
+        s.add(w); s.flush()
+        return w.workflow_id
+
+
+def _seed_in_progress_attempt(
+    engine, workflow_id, *, instance="host/1/uuidA",
+    lock_ttl_seconds=600, heartbeat_age_seconds=0,
+    pickup_count=1,
+):
+    """Insert one in_progress attempt. ``heartbeat_age_seconds`` >
+    ``lock_ttl_seconds`` makes it reapable."""
+    now = datetime.now(timezone.utc)
+    heartbeat_at = now - timedelta(seconds=heartbeat_age_seconds)
+    with session_scope(engine) as s:
+        a = WorkflowAttempt(
+            workflow_id=workflow_id, attempt_number=1, role="implementer",
+            tier=0, input_payload={}, input_version="V1",
+            status="in_progress", locked_by_instance=instance,
+            locked_at=heartbeat_at,
+            lock_heartbeat_at=heartbeat_at,
+            lock_ttl_seconds=lock_ttl_seconds,
+            pickup_count=pickup_count,
+        )
+        s.add(a); s.flush()
+        return a.attempt_id
+
+
+# ─── reaper ───────────────────────────────────────────────────────────
+
+
+class TestReaper:
+    def test_empty_db_no_op(self, engine):
+        report = reap_stale_attempts(engine)
+        assert report.rows_reaped == 0
+        assert report.reaped_attempts == []
+
+    def test_fresh_heartbeat_not_reaped(self, engine):
+        wf_id = _seed_workflow(engine)
+        _seed_in_progress_attempt(
+            engine, wf_id, lock_ttl_seconds=600, heartbeat_age_seconds=10,
+        )
+        report = reap_stale_attempts(engine)
+        assert report.rows_reaped == 0
+
+    def test_stale_heartbeat_reaped(self, engine):
+        wf_id = _seed_workflow(engine)
+        attempt_id = _seed_in_progress_attempt(
+            engine, wf_id, lock_ttl_seconds=60, heartbeat_age_seconds=120,
+        )
+        report = reap_stale_attempts(engine)
+        assert report.rows_reaped == 1
+        assert report.reaped_attempts == [(attempt_id, "host/1/uuidA")]
+        # DB row reset to pending.
+        with session_scope(engine) as s:
+            a = s.query(WorkflowAttempt).filter_by(attempt_id=attempt_id).one()
+            assert a.status == "pending"
+            assert a.locked_by_instance is None
+            assert a.lock_heartbeat_at is None
+            # Pickup count preserved — that's the pickup guard's job.
+            assert a.pickup_count == 1
+
+    def test_per_row_ttl_respected(self, engine):
+        """Different attempts can have different lock_ttl_seconds
+        (per-role tuning per plan v9)."""
+        wf_id = _seed_workflow(engine)
+        # Fresh-heartbeat attempt with short TTL (180s) — NOT reaped.
+        a1 = _seed_in_progress_attempt(
+            engine, wf_id, instance="host/A/uuid1",
+            lock_ttl_seconds=180, heartbeat_age_seconds=60,
+        )
+        # Stale-heartbeat attempt with long TTL (1800s) — NOT reaped
+        # (300s age but 1800s TTL).
+        # Need a second workflow because (workflow_id, attempt_number) is
+        # NOT unique but the per-test fixture inserts attempt_number=1 each
+        # time. Insert a second attempt on the same workflow with number=2.
+        with session_scope(engine) as s:
+            now = datetime.now(timezone.utc)
+            from datetime import timedelta as _td
+            hb = now - _td(seconds=300)
+            a2 = WorkflowAttempt(
+                workflow_id=wf_id, attempt_number=2, role="reviewer",
+                tier=0, input_payload={}, input_version="V1",
+                status="in_progress", locked_by_instance="host/B/uuid2",
+                locked_at=hb, lock_heartbeat_at=hb,
+                lock_ttl_seconds=1800,
+            )
+            s.add(a2); s.flush()
+            a2_id = a2.attempt_id
+        report = reap_stale_attempts(engine)
+        # Neither was reaped.
+        assert report.rows_reaped == 0
+
+    def test_reaper_creates_event_row(self, engine):
+        wf_id = _seed_workflow(engine)
+        _seed_in_progress_attempt(
+            engine, wf_id, lock_ttl_seconds=60, heartbeat_age_seconds=120,
+        )
+        reap_stale_attempts(engine)
+        with session_scope(engine) as s:
+            events = s.execute(
+                text("SELECT event_type, payload FROM controller_events")
+            ).all()
+        assert len(events) == 1
+        assert events[0].event_type == "lock-ttl-expired"
+        import json
+        payload = json.loads(events[0].payload)
+        assert payload["previously_locked_by"] == "host/1/uuidA"
+
+    def test_only_in_progress_attempts_reaped(self, engine):
+        """Pending / complete / failed attempts are never touched."""
+        wf_id = _seed_workflow(engine)
+        with session_scope(engine) as s:
+            for status in ("pending", "complete", "failed", "reaped"):
+                a = WorkflowAttempt(
+                    workflow_id=wf_id, attempt_number=ord(status[0]),
+                    role="implementer", tier=0,
+                    input_payload={}, input_version="V1",
+                    status=status,
+                )
+                s.add(a)
+        report = reap_stale_attempts(engine)
+        assert report.rows_reaped == 0
+
+    def test_multiple_stale_attempts_all_reaped(self, engine):
+        wf1 = _seed_workflow(engine)
+        with session_scope(engine) as s:
+            # Second workflow with a different entity_number so we
+            # don't trip the unique constraint.
+            w2 = Workflow(
+                kind="pr", owner="o", repo="r", entity_number=31,
+                current_state="IMPLEMENTING",
+            )
+            s.add(w2); s.flush()
+            wf2 = w2.workflow_id
+        _seed_in_progress_attempt(
+            engine, wf1, instance="A", lock_ttl_seconds=60,
+            heartbeat_age_seconds=120,
+        )
+        _seed_in_progress_attempt(
+            engine, wf2, instance="B", lock_ttl_seconds=60,
+            heartbeat_age_seconds=120,
+        )
+        report = reap_stale_attempts(engine)
+        assert report.rows_reaped == 2
+
+
+# ─── pickup guard ─────────────────────────────────────────────────────
+
+
+class TestPickupGuard:
+    def test_empty_db_no_op(self, engine):
+        report = transition_exhausted_to_stuck(engine)
+        assert report.workflows_stuck == 0
+
+    def test_attempt_below_limit_not_stuck(self, engine):
+        wf_id = _seed_workflow(engine)
+        with session_scope(engine) as s:
+            s.add(WorkflowAttempt(
+                workflow_id=wf_id, attempt_number=1, role="implementer",
+                tier=0, input_payload={}, input_version="V1",
+                status="pending", pickup_count=2,
+            ))
+        report = transition_exhausted_to_stuck(engine, max_pickups=3)
+        assert report.workflows_stuck == 0
+
+    def test_attempt_at_limit_transitions_workflow_to_stuck(self, engine):
+        wf_id = _seed_workflow(engine)
+        with session_scope(engine) as s:
+            a = WorkflowAttempt(
+                workflow_id=wf_id, attempt_number=1, role="implementer",
+                tier=0, input_payload={}, input_version="V1",
+                status="pending", pickup_count=3,
+            )
+            s.add(a); s.flush()
+            attempt_id = a.attempt_id
+        report = transition_exhausted_to_stuck(engine, max_pickups=3)
+        assert report.workflows_stuck == 1
+        assert report.stuck_workflows == [(wf_id, attempt_id)]
+        # Workflow + attempt updated.
+        with session_scope(engine) as s:
+            w = s.query(Workflow).filter_by(workflow_id=wf_id).one()
+            assert w.current_state == "STUCK"
+            a = s.query(WorkflowAttempt).filter_by(attempt_id=attempt_id).one()
+            assert a.status == "reaped"
+
+    def test_in_progress_attempts_not_checked(self, engine):
+        """Only pending attempts trigger the guard. An in_progress
+        attempt is being worked; let the reaper handle it if stale."""
+        wf_id = _seed_workflow(engine)
+        with session_scope(engine) as s:
+            s.add(WorkflowAttempt(
+                workflow_id=wf_id, attempt_number=1, role="implementer",
+                tier=0, input_payload={}, input_version="V1",
+                status="in_progress", pickup_count=5,
+                locked_by_instance="host/X/u",
+            ))
+        report = transition_exhausted_to_stuck(engine, max_pickups=3)
+        assert report.workflows_stuck == 0
+
+    def test_terminal_workflow_skipped(self, engine):
+        """A workflow that's already MERGED/ABANDONED/STUCK isn't
+        re-transitioned."""
+        wf_id = _seed_workflow(engine, current_state="MERGED")
+        with session_scope(engine) as s:
+            s.add(WorkflowAttempt(
+                workflow_id=wf_id, attempt_number=1, role="implementer",
+                tier=0, input_payload={}, input_version="V1",
+                status="pending", pickup_count=5,
+            ))
+        report = transition_exhausted_to_stuck(engine, max_pickups=3)
+        assert report.workflows_stuck == 0
+
+    def test_stuck_event_row_carries_reason(self, engine):
+        wf_id = _seed_workflow(engine)
+        with session_scope(engine) as s:
+            s.add(WorkflowAttempt(
+                workflow_id=wf_id, attempt_number=1, role="implementer",
+                tier=0, input_payload={}, input_version="V1",
+                status="pending", pickup_count=3,
+            ))
+        transition_exhausted_to_stuck(engine, max_pickups=3)
+        import json
+        with session_scope(engine) as s:
+            events = s.execute(
+                text("SELECT event_type, from_state, to_state, payload "
+                     "FROM controller_events WHERE event_type = 'transition'")
+            ).all()
+        assert len(events) == 1
+        ev = events[0]
+        assert ev.from_state == "IMPLEMENTING"
+        assert ev.to_state == "STUCK"
+        payload = json.loads(ev.payload)
+        assert payload["reason"] == "attempt-pickup-exhausted"
+        assert payload["pickup_count"] == 3
+        assert payload["max_pickups"] == 3
+
+    def test_default_max_pickups_env_default(self):
+        """Default value matches plan v6 (3 pickups)."""
+        assert DEFAULT_MAX_PICKUPS == 3

tests/auto_agents/controller/test_state_machine.py

@@ -0,0 +1,220 @@
+"""Tests for the controller state machine + load-bearing invariants.
+
+Per plan v9: 6 load-bearing invariants for v1. Each one has a
+property test that asserts the check returns 0 violations against
+the live TRANSITIONS table.
+"""
+from __future__ import annotations
+
+import pytest
+
+from tools.controller.state_machine import (
+    EVENTS,
+    IllegalTransitionError,
+    KNOWN_STATES,
+    LOAD_BEARING_INVARIANTS,
+    NON_TERMINAL_STATES,
+    TERMINAL_STATES,
+    TRANSITIONS,
+    apply_event,
+    check_all_invariants,
+    check_conflict_resolving_bounded,
+    check_escalation_deterministic,
+    check_every_pr_workflow_includes_reviewing,
+    check_no_path_implementing_to_reviewing_skips_ci,
+    check_terminal_states_have_no_exits,
+    check_tier_monotonic_non_decreasing,
+    events_from,
+    reachable_from,
+)
+
+
+# ─── basic shape ──────────────────────────────────────────────────────
+
+
+class TestBasicShape:
+    def test_known_states_partition(self):
+        """TERMINAL + NON_TERMINAL partition KNOWN_STATES."""
+        assert TERMINAL_STATES | NON_TERMINAL_STATES == KNOWN_STATES
+        assert TERMINAL_STATES & NON_TERMINAL_STATES == set()
+
+    def test_every_transition_uses_known_states(self):
+        for (frm, _evt), to in TRANSITIONS.items():
+            assert frm in KNOWN_STATES, f"unknown from-state {frm!r}"
+            assert to in KNOWN_STATES, f"unknown to-state {to!r}"
+
+    def test_every_transition_event_has_definition(self):
+        for (_frm, evt), _to in TRANSITIONS.items():
+            assert evt in EVENTS, f"event {evt!r} has no definition in EVENTS"
+
+    def test_six_load_bearing_invariants(self):
+        """Plan v9 ships exactly 6 load-bearing invariants for v1."""
+        assert len(LOAD_BEARING_INVARIANTS) == 6
+
+
+# ─── apply_event ──────────────────────────────────────────────────────
+
+
+class TestApplyEvent:
+    def test_known_transition(self):
+        assert apply_event("DISCOVERED", "discovery_picked_up") == "ANALYZING"
+
+    def test_unknown_state_raises_value_error(self):
+        with pytest.raises(ValueError, match="unknown state"):
+            apply_event("GARBAGE", "discovery_picked_up")
+
+    def test_illegal_event_raises(self):
+        with pytest.raises(IllegalTransitionError) as excinfo:
+            apply_event("DISCOVERED", "merge_ok")
+        # Error message lists the legal events from this state for
+        # operator clarity.
+        assert "known events from this state" in str(excinfo.value)
+        assert "discovery_picked_up" in str(excinfo.value)
+
+
+# ─── events_from + reachable_from helpers ─────────────────────────────
+
+
+class TestHelpers:
+    def test_events_from_discovered(self):
+        assert events_from("DISCOVERED") == ["discovery_picked_up"]
+
+    def test_events_from_terminal_state(self):
+        assert events_from("MERGED") == []
+
+    def test_reachable_from_discovered_includes_terminals(self):
+        reach = reachable_from("DISCOVERED")
+        assert "MERGED" in reach
+        assert "STUCK" in reach
+        assert "ABANDONED" in reach
+
+    def test_reachable_from_includes_self_only_if_in_cycle(self):
+        """``reachable_from(s)`` includes ``s`` IFF ``s`` is part of a
+        cycle.
+
+        - AWAITING_CI has a self-loop via ci_flake_retry → included.
+        - DISCOVERED is part of the operator_unstick cycle
+          (DISCOVERED → ... → STUCK → operator_unstick → DISCOVERED)
+          → included.
+        - ABANDONED is terminal (no exits) → not included in its own
+          reachable set.
+        """
+        # AWAITING_CI self-loops.
+        assert "AWAITING_CI" in reachable_from("AWAITING_CI")
+        # DISCOVERED via the STUCK operator-unstick cycle.
+        assert "DISCOVERED" in reachable_from("DISCOVERED")
+        # ABANDONED is a sink with zero outgoing.
+        assert "ABANDONED" not in reachable_from("ABANDONED")
+
+
+# ─── load-bearing invariants ──────────────────────────────────────────
+
+
+class TestLoadBearingInvariants:
+    def test_check_all_invariants_zero_violations(self):
+        """The canonical TRANSITIONS table must satisfy every
+        load-bearing invariant."""
+        violations = check_all_invariants()
+        assert violations == [], (
+            "\n".join(f"  {v.invariant}: {v.message}" for v in violations)
+        )
+
+    def test_no_path_implementing_to_reviewing_skips_ci(self):
+        assert check_no_path_implementing_to_reviewing_skips_ci() == []
+
+    def test_terminal_states_have_no_exits(self):
+        assert check_terminal_states_have_no_exits() == []
+
+    def test_tier_monotonic_non_decreasing(self):
+        assert check_tier_monotonic_non_decreasing() == []
+
+    def test_every_pr_workflow_includes_reviewing(self):
+        assert check_every_pr_workflow_includes_reviewing() == []
+
+    def test_conflict_resolving_bounded(self):
+        assert check_conflict_resolving_bounded() == []
+
+    def test_escalation_deterministic(self):
+        assert check_escalation_deterministic() == []
+
+
+# ─── invariant negative-tests (mutate transitions; expect violations) ─
+
+
+class TestInvariantsCatchViolations:
+    """Verify the invariant checks would FAIL if the TRANSITIONS
+    table were corrupted. Monkeypatch in a violation; check it's caught."""
+
+    def test_no_path_invariant_catches_direct_edge(self, monkeypatch):
+        bad = dict(TRANSITIONS)
+        bad[("IMPLEMENTING", "synthetic_bug")] = "REVIEWING"
+        monkeypatch.setattr(
+            "tools.controller.state_machine.TRANSITIONS", bad
+        )
+        violations = check_no_path_implementing_to_reviewing_skips_ci()
+        assert violations
+        assert "IMPLEMENTING → REVIEWING" in violations[0].message
+
+    def test_terminal_no_exit_invariant_catches_merged_exit(self, monkeypatch):
+        bad = dict(TRANSITIONS)
+        bad[("MERGED", "synthetic_bug")] = "DISCOVERED"
+        monkeypatch.setattr(
+            "tools.controller.state_machine.TRANSITIONS", bad
+        )
+        violations = check_terminal_states_have_no_exits()
+        assert violations
+        assert "MERGED" in violations[0].message
+
+    def test_terminal_no_exit_invariant_catches_stuck_unauthorized_exit(
+        self, monkeypatch
+    ):
+        bad = dict(TRANSITIONS)
+        bad[("STUCK", "auto_recover_bug")] = "DISCOVERED"
+        monkeypatch.setattr(
+            "tools.controller.state_machine.TRANSITIONS", bad
+        )
+        violations = check_terminal_states_have_no_exits()
+        assert violations
+        assert "auto_recover_bug" in violations[0].message
+
+    def test_every_pr_includes_reviewing_catches_direct_merge(
+        self, monkeypatch
+    ):
+        bad = dict(TRANSITIONS)
+        bad[("ANALYZING", "fast_path_bug")] = "MERGED"
+        monkeypatch.setattr(
+            "tools.controller.state_machine.TRANSITIONS", bad
+        )
+        violations = check_every_pr_workflow_includes_reviewing()
+        assert violations
+        assert "MERGED" in violations[0].message
+        assert "MERGING" in violations[0].message  # the "not from MERGING" complaint
+
+    def test_conflict_resolving_bounded_catches_unmapped_event(
+        self, monkeypatch
+    ):
+        bad = dict(TRANSITIONS)
+        bad[("CONFLICT_RESOLVING", "wildcard_resolve")] = "REVIEWING"
+        monkeypatch.setattr(
+            "tools.controller.state_machine.TRANSITIONS", bad
+        )
+        violations = check_conflict_resolving_bounded()
+        assert violations
+        assert "wildcard_resolve" in violations[0].message
+
+
+# ─── coverage check: every non-terminal has a path to a terminal ─────
+
+
+class TestNonTerminalHasTerminalPath:
+    """Every non-terminal state must have SOME path to a terminal
+    state. This isn't one of the 6 load-bearing invariants but it's
+    a sanity check that the graph isn't dead-ended."""
+
+    @pytest.mark.parametrize("state", sorted(NON_TERMINAL_STATES))
+    def test_state_can_reach_a_terminal(self, state):
+        reach = reachable_from(state)
+        assert reach & TERMINAL_STATES, (
+            f"{state!r} cannot reach any terminal state; "
+            f"reachable: {sorted(reach)}"
+        )

tools/controller/__init__.py

@@ -21,4 +21,8 @@ The package layout mirrors the plan's responsibilities:
 - (later) ``mcp/`` — per-role response-builder MCP servers.
 """
 
-__all__ = ["contracts"]
+from . import state_machine  # re-exported for tests
+from . import reaper           # re-exported for tests
+from . import pickup_guard     # re-exported for tests
+
+__all__ = ["contracts", "state_machine", "reaper", "pickup_guard"]

tools/controller/pickup_guard.py

@@ -0,0 +1,130 @@
+"""Master-side pickup-exhaustion guard.
+
+Per plan v6 blocker fix: an attempt that's been dequeued
+``MAX_PICKUPS`` times without success means something is structurally
+wrong (worker bug, persistent OpenCode failure, infrastructure
+issue). The master detects these and transitions the workflow to
+STUCK with ``reason='attempt-pickup-exhausted'``.
+
+The dequeue helper (``db/dequeue.py``) already enforces the
+``pickup_count < MAX_PICKUPS`` filter — so exhausted attempts are
+NEVER picked up. This guard handles the secondary case: an attempt
+that JUST hit the limit after a failed run; the workflow needs to
+transition to STUCK so it stops being re-enqueued.
+
+Runs on the master's tick loop. Cheap query; safe to run every tick.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import os
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+from sqlalchemy import text
+from sqlalchemy.engine import Engine
+
+from .db.session import session_scope
+
+logger = logging.getLogger(__name__)
+
+
+DEFAULT_MAX_PICKUPS = int(os.environ.get("CONTROLLER_MAX_ATTEMPT_PICKUPS", "3"))
+
+
+@dataclass
+class PickupGuardReport:
+    """Per-tick summary: which workflows got transitioned to STUCK."""
+
+    workflows_stuck: int = 0
+    stuck_workflows: list[tuple[int, int]] = field(default_factory=list)
+    # Each entry: (workflow_id, attempt_id_that_exhausted)
+
+
+def transition_exhausted_to_stuck(
+    engine: Engine, *, max_pickups: int = DEFAULT_MAX_PICKUPS,
+) -> PickupGuardReport:
+    """Find attempts at or past MAX_PICKUPS that are also currently
+    pending (i.e., the last run failed and re-pended), and transition
+    their workflow to STUCK.
+
+    "At or past MAX_PICKUPS" means ``pickup_count >= max_pickups``
+    AND ``status = 'pending'``. The dequeue helper won't pick these
+    up (its filter is ``pickup_count < max_pickups``), so they'd loop
+    in the pending pool forever without this guard.
+    """
+    report = PickupGuardReport()
+    now = datetime.now(timezone.utc)
+    with session_scope(engine) as session:
+        # Find candidate attempts + their workflows. A workflow in
+        # an already-terminal state doesn't need re-transitioning.
+        rows = session.execute(
+            text(
+                "SELECT a.attempt_id, a.workflow_id, a.pickup_count, "
+                "       w.current_state "
+                "FROM workflow_attempts a "
+                "JOIN workflows w ON w.workflow_id = a.workflow_id "
+                "WHERE a.status = 'pending' "
+                "  AND a.pickup_count >= :limit "
+                "  AND w.current_state NOT IN ('MERGED', 'ABANDONED', 'STUCK', 'CREATED_PR')"
+            ),
+            {"limit": max_pickups},
+        ).all()
+
+        if not rows:
+            return report
+
+        for r in rows:
+            # Transition workflow → STUCK; mark the attempt as reaped
+            # (so the pending pool is clean).
+            session.execute(
+                text(
+                    "UPDATE workflows SET "
+                    "  current_state = 'STUCK', "
+                    "  last_transition_at = :now, "
+                    "  entered_state_at = :now "
+                    "WHERE workflow_id = :wf_id"
+                ),
+                {"now": now, "wf_id": r.workflow_id},
+            )
+            session.execute(
+                text(
+                    "UPDATE workflow_attempts SET status = 'reaped' "
+                    "WHERE attempt_id = :aid"
+                ),
+                {"aid": r.attempt_id},
+            )
+            session.execute(
+                text(
+                    "INSERT INTO controller_events "
+                    "(workflow_id, ts, event_type, from_state, to_state, "
+                    " attempt_id, payload, forgejo_write_pending, replay_attempts) "
+                    "VALUES (:wf_id, :ts, 'transition', :from_state, 'STUCK', "
+                    "        :aid, :payload, 0, 0)"
+                ),
+                {
+                    "wf_id": r.workflow_id,
+                    "ts": now,
+                    "from_state": r.current_state,
+                    "aid": r.attempt_id,
+                    "payload": json.dumps({
+                        "reason": "attempt-pickup-exhausted",
+                        "pickup_count": r.pickup_count,
+                        "max_pickups": max_pickups,
+                    }),
+                },
+            )
+            report.workflows_stuck += 1
+            report.stuck_workflows.append((r.workflow_id, r.attempt_id))
+
+    if report.workflows_stuck:
+        logger.warning(
+            "pickup guard: %d workflow(s) transitioned to STUCK "
+            "(pickup-exhausted): %s",
+            report.workflows_stuck, report.stuck_workflows,
+        )
+    return report
+
+
+__all__ = ["DEFAULT_MAX_PICKUPS", "PickupGuardReport", "transition_exhausted_to_stuck"]

tools/controller/reaper.py

@@ -0,0 +1,155 @@
+"""Master-side reaper: reset stale-heartbeat workflow_attempts to
+pending so another worker can re-acquire.
+
+Per plan v9 simplified: TTL-only reaper. No JOIN-with-workflows
+superseded check (deferred — orphan re-attempts no-op cheaply against
+terminal workflows when the next worker tries to advance them).
+
+Runs on the master's tick loop every ``CONTROLLER_REAPER_INTERVAL_S``
+(default 60s). The reaper:
+- Selects ``workflow_attempts WHERE status='in_progress' AND
+  lock_heartbeat_at < NOW() - lock_ttl_seconds``
+- For each: status → 'pending'; clear locked_by_instance / locked_at /
+  lock_heartbeat_at; insert a ``controller_events`` row with
+  reason='lock-ttl-expired'.
+
+The reset preserves ``pickup_count`` so the master's separate guard
+catches "this attempt has been re-pended too many times → STUCK."
+"""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+
+from sqlalchemy import text
+from sqlalchemy.engine import Engine
+
+from .db.session import session_scope
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ReaperReport:
+    """Per-sweep summary; emitted to structured logs + Prometheus."""
+
+    rows_reaped: int = 0
+    reaped_attempts: list[tuple[int, str | None]] = field(default_factory=list)
+    # Each entry: (attempt_id, locked_by_instance) — useful for
+    # operator forensics ("which machine's worker died?")
+
+
+def reap_stale_attempts(engine: Engine) -> ReaperReport:
+    """Reset stale-heartbeat in_progress attempts to pending.
+
+    The TTL is per-row (``lock_ttl_seconds``) — different roles can
+    have different TTLs (estimator ~180s; reviewer ~720s; tier-2
+    implementer ~2160s). The WHERE clause uses the per-row column.
+    """
+    report = ReaperReport()
+    now = datetime.now(timezone.utc)
+    # Two-step approach for portability:
+    # 1. SELECT the candidates (so we can log them).
+    # 2. UPDATE them.
+    # We could do this in one statement on Postgres (RETURNING) and
+    # SQLite (RETURNING is supported in 3.35+). Splitting for
+    # observability — the reaper isn't a hot path.
+    with session_scope(engine) as session:
+        dialect = session.bind.dialect.name if session.bind else "sqlite"
+        if dialect == "postgresql":
+            select_sql = text(
+                "SELECT attempt_id, locked_by_instance, lock_heartbeat_at "
+                "FROM workflow_attempts "
+                "WHERE status = 'in_progress' "
+                "  AND lock_heartbeat_at IS NOT NULL "
+                "  AND lock_heartbeat_at + (lock_ttl_seconds || ' seconds')::interval < :now"
+            )
+        else:
+            # SQLite: TIMESTAMP arithmetic via julianday().
+            select_sql = text(
+                "SELECT attempt_id, locked_by_instance, lock_heartbeat_at "
+                "FROM workflow_attempts "
+                "WHERE status = 'in_progress' "
+                "  AND lock_heartbeat_at IS NOT NULL "
+                "  AND (julianday(:now) - julianday(lock_heartbeat_at)) "
+                "      * 86400 > lock_ttl_seconds"
+            )
+        rows = session.execute(select_sql, {"now": now}).all()
+        if not rows:
+            return report
+
+        # 2. UPDATE — use IN-clause with the candidate ids.
+        ids = [r.attempt_id for r in rows]
+        # SQLAlchemy's expanding bindparam handles IN over a list.
+        from sqlalchemy import bindparam
+        update_sql = text(
+            "UPDATE workflow_attempts SET "
+            "  status = 'pending', "
+            "  locked_by_instance = NULL, "
+            "  locked_at = NULL, "
+            "  lock_heartbeat_at = NULL "
+            "WHERE attempt_id IN :ids"
+        ).bindparams(bindparam("ids", expanding=True))
+        result = session.execute(update_sql, {"ids": ids})
+        report.rows_reaped = result.rowcount
+
+        # 3. Log + record controller_events.
+        events_sql = text(
+            "INSERT INTO controller_events "
+            "(workflow_id, ts, event_type, payload, forgejo_write_pending, replay_attempts) "
+            "VALUES ("
+            "  (SELECT workflow_id FROM workflow_attempts WHERE attempt_id = :attempt_id), "
+            "  :ts, 'lock-ttl-expired', :payload, 0, 0"
+            ")"
+        )
+        for r in rows:
+            report.reaped_attempts.append(
+                (r.attempt_id, r.locked_by_instance)
+            )
+            session.execute(events_sql, {
+                "attempt_id": r.attempt_id,
+                "ts": now,
+                "payload": _payload_for_event(
+                    r.attempt_id, r.locked_by_instance,
+                    r.lock_heartbeat_at, now,
+                ),
+            })
+
+    if report.rows_reaped:
+        logger.warning(
+            "reaper reset %d stale in_progress attempt(s): %s",
+            report.rows_reaped,
+            [(aid, inst) for (aid, inst) in report.reaped_attempts],
+        )
+    return report
+
+
+def _payload_for_event(
+    attempt_id: int,
+    locked_by_instance: str | None,
+    heartbeat_at: datetime | None,
+    now: datetime,
+) -> str:
+    """Build a JSON string payload for the controller_events row."""
+    import json
+    age_s: float | None = None
+    if heartbeat_at is not None:
+        try:
+            # heartbeat_at may be naive depending on the DB; normalise.
+            if heartbeat_at.tzinfo is None:
+                from datetime import timezone as _tz
+                hb = heartbeat_at.replace(tzinfo=_tz.utc)
+            else:
+                hb = heartbeat_at
+            age_s = (now - hb).total_seconds()
+        except Exception:
+            age_s = None
+    return json.dumps({
+        "reaped_attempt_id": attempt_id,
+        "previously_locked_by": locked_by_instance,
+        "heartbeat_age_s": age_s,
+    })
+
+
+__all__ = ["ReaperReport", "reap_stale_attempts"]

tools/controller/state_machine.py

@@ -0,0 +1,565 @@
+"""Controller state machine — TRANSITIONS table + invariants.
+
+Plan v9 simplified scope: ship the 6 LOAD-BEARING property invariants
+for v1; add the other 15 as bugs surface.
+
+States (per plan v6+v9, with v8 metadata-only-no-bypass + v6
+CONFLICT_RESOLVING + v6 MERGING + v6 CREATED_PR):
+
+    DISCOVERED
+        → ANALYZING (estimator picks tier OR routes to metadata_only)
+        → STUCK (estimator failed twice)
+
+    ANALYZING
+        → IMPLEMENTING(tier) (normal path)
+        → REVIEWING (metadata-only path; per v8 Hard Rule #1 still
+                     goes through reviewer with lightweight profile)
+        → STUCK (estimator output schema violation)
+
+    IMPLEMENTING
+        → AWAITING_CI (worker pushed; head_sha advanced)
+        → ESCALATING (worker emitted competence-failure; no push)
+        → CONFLICT_RESOLVING (worker emitted rebase-failed)
+        → STUCK (worker emitted blocked)
+
+    AWAITING_CI
+        → REVIEWING (CI green)
+        → IMPLEMENTING(tier) (CI red, attempts-per-tier remaining)
+        → ESCALATING (CI red, attempts-per-tier exhausted)
+        → AWAITING_CI (flake retry; one retry per flake-classifier
+                       per gate; same state)
+
+    CONFLICT_RESOLVING
+        → IMPLEMENTING(same tier) (1st conflict resolved)
+        → ESCALATING (2nd conflict at same tier)
+        → STUCK (3rd conflict; or resolver outcome=irreconcilable)
+
+    ESCALATING
+        → IMPLEMENTING(tier+1) (next tier available)
+        → ABANDONED (max tier exhausted)
+
+    REVIEWING
+        → MERGING (verdict=approve)
+        → IMPLEMENTING(tier) (verdict=request-changes,
+                              attempts-per-tier remaining)
+        → ESCALATING (verdict=request-changes, attempts-per-tier
+                      exhausted)
+        → STUCK (verdict=abstain; retry-once-then-stuck handled
+                 in v9 retry policy)
+
+    MERGING
+        → MERGED (Forgejo 200)
+        → IMPLEMENTING(tier_last_succeeded) (Forgejo 409; post-
+                                              approval base conflict)
+        → AWAITING_CI (Forgejo 422; CI status expired race)
+        → STUCK (Forgejo 403 branch protection; or retry_count>=5)
+        → MERGED OR ABANDONED (Forgejo 404; check external state)
+
+Terminal states:
+    MERGED, ABANDONED, STUCK, CREATED_PR
+
+CREATED_PR is reachable only from issue-kind workflows (IMPLEMENTING
+on an issue successfully creates a PR + spawns a new pr-kind workflow
+with parent_workflow_id; the issue workflow terminates at CREATED_PR).
+For v1 simplicity, this module models only the PR-kind state machine;
+issue-kind handling lands in Phase 4.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Iterable
+
+
+# ─── states ──────────────────────────────────────────────────────────
+
+
+KNOWN_STATES: frozenset[str] = frozenset({
+    "DISCOVERED",
+    "ANALYZING",
+    "IMPLEMENTING",
+    "AWAITING_CI",
+    "CONFLICT_RESOLVING",
+    "ESCALATING",
+    "REVIEWING",
+    "MERGING",
+    "MERGED",
+    "ABANDONED",
+    "STUCK",
+    "CREATED_PR",
+})
+
+TERMINAL_STATES: frozenset[str] = frozenset({
+    "MERGED", "ABANDONED", "STUCK", "CREATED_PR",
+})
+
+NON_TERMINAL_STATES: frozenset[str] = KNOWN_STATES - TERMINAL_STATES
+
+
+# Events that drive transitions. The event name captures "what the
+# controller observed" so the transition table reads as data, not
+# code.
+@dataclass(frozen=True)
+class Event:
+    name: str
+    description: str = ""
+
+    def __repr__(self) -> str:
+        return f"Event({self.name!r})"
+
+
+# ─── canonical events ────────────────────────────────────────────────
+
+
+EVENTS = {
+    # ANALYZING
+    "estimator_done": Event("estimator_done", "Estimator returned tier"),
+    "estimator_metadata_only": Event(
+        "estimator_metadata_only",
+        "Estimator flagged is_metadata_only=true",
+    ),
+    "estimator_failed_twice": Event(
+        "estimator_failed_twice",
+        "Estimator strict-parse failed twice (v6 retry policy)",
+    ),
+    # IMPLEMENTING
+    "implementer_pushed": Event(
+        "implementer_pushed",
+        "Worker emitted outcome=resolved; head_sha advanced",
+    ),
+    "implementer_competence_failure": Event(
+        "implementer_competence_failure",
+        "Worker emitted outcome=competence-failure",
+    ),
+    "implementer_rebase_failed": Event(
+        "implementer_rebase_failed",
+        "Worker emitted outcome=rebase-failed",
+    ),
+    "implementer_blocked": Event(
+        "implementer_blocked",
+        "Worker emitted outcome=blocked",
+    ),
+    # AWAITING_CI
+    "ci_green": Event("ci_green", "CI overall_state=success"),
+    "ci_red_retry_same_tier": Event(
+        "ci_red_retry_same_tier",
+        "CI failed; attempts-per-tier remaining",
+    ),
+    "ci_red_escalate": Event(
+        "ci_red_escalate",
+        "CI failed; attempts-per-tier exhausted",
+    ),
+    "ci_flake_retry": Event(
+        "ci_flake_retry",
+        "CI failed but classified as flake; retry once",
+    ),
+    # CONFLICT_RESOLVING
+    "conflict_resolved_first": Event(
+        "conflict_resolved_first",
+        "Resolver returned outcome=resolved (1st conflict at tier)",
+    ),
+    "conflict_resolved_second_same_tier": Event(
+        "conflict_resolved_second_same_tier",
+        "Resolver returned outcome=resolved (2nd conflict at tier)",
+    ),
+    "conflict_repeated_three_plus": Event(
+        "conflict_repeated_three_plus",
+        "3rd or later conflict — likely structural; STUCK",
+    ),
+    "conflict_irreconcilable": Event(
+        "conflict_irreconcilable",
+        "Resolver returned outcome=irreconcilable",
+    ),
+    # ESCALATING
+    "escalate_next_tier_available": Event(
+        "escalate_next_tier_available",
+        "Tier+1 ≤ MAX_TIER; transition to IMPLEMENTING(tier+1)",
+    ),
+    "escalate_max_tier_exhausted": Event(
+        "escalate_max_tier_exhausted",
+        "Tier was MAX_TIER; ABANDONED",
+    ),
+    # REVIEWING
+    "reviewer_approve": Event(
+        "reviewer_approve",
+        "Reviewer verdict=approve",
+    ),
+    "reviewer_request_changes_retry": Event(
+        "reviewer_request_changes_retry",
+        "Reviewer verdict=request-changes; attempts-per-tier remaining",
+    ),
+    "reviewer_request_changes_escalate": Event(
+        "reviewer_request_changes_escalate",
+        "Reviewer verdict=request-changes; attempts-per-tier exhausted",
+    ),
+    "reviewer_abstain": Event(
+        "reviewer_abstain",
+        "Reviewer verdict=abstain (rare); per v6 retry-once-then-STUCK",
+    ),
+    # MERGING
+    "merge_ok": Event("merge_ok", "Forgejo merge returned 200"),
+    "merge_base_conflict": Event(
+        "merge_base_conflict",
+        "Forgejo merge returned 409; base advanced post-approval",
+    ),
+    "merge_ci_required_missing": Event(
+        "merge_ci_required_missing",
+        "Forgejo merge returned 422; race with CI status",
+    ),
+    "merge_branch_protection_blocked": Event(
+        "merge_branch_protection_blocked",
+        "Forgejo merge returned 403; operator must intervene",
+    ),
+    "merge_retry_exhausted": Event(
+        "merge_retry_exhausted",
+        "merging_retry_count >= 5; STUCK",
+    ),
+    "merge_external_action": Event(
+        "merge_external_action",
+        "Forgejo merge returned 404; reconcile via PR state",
+    ),
+    # DISCOVERED → ANALYZING
+    "discovery_picked_up": Event(
+        "discovery_picked_up",
+        "Master saw DISCOVERED workflow; queue estimator attempt",
+    ),
+    # Operator escape hatch (reachable from every non-terminal state)
+    "operator_unstick": Event(
+        "operator_unstick",
+        "Operator manually unstucking; back to DISCOVERED",
+    ),
+    "pickup_exhausted": Event(
+        "pickup_exhausted",
+        "Attempt picked up MAX_PICKUPS times w/o success; STUCK",
+    ),
+}
+
+
+# ─── transitions ─────────────────────────────────────────────────────
+
+
+# Map of (from_state, event_name) → to_state. Pure data; tests
+# enumerate every key to verify invariants.
+TRANSITIONS: dict[tuple[str, str], str] = {
+    ("DISCOVERED", "discovery_picked_up"): "ANALYZING",
+
+    ("ANALYZING", "estimator_done"): "IMPLEMENTING",
+    ("ANALYZING", "estimator_metadata_only"): "REVIEWING",
+    ("ANALYZING", "estimator_failed_twice"): "STUCK",
+
+    ("IMPLEMENTING", "implementer_pushed"): "AWAITING_CI",
+    ("IMPLEMENTING", "implementer_competence_failure"): "ESCALATING",
+    ("IMPLEMENTING", "implementer_rebase_failed"): "CONFLICT_RESOLVING",
+    ("IMPLEMENTING", "implementer_blocked"): "STUCK",
+    ("IMPLEMENTING", "pickup_exhausted"): "STUCK",
+
+    ("AWAITING_CI", "ci_green"): "REVIEWING",
+    ("AWAITING_CI", "ci_red_retry_same_tier"): "IMPLEMENTING",
+    ("AWAITING_CI", "ci_red_escalate"): "ESCALATING",
+    ("AWAITING_CI", "ci_flake_retry"): "AWAITING_CI",
+
+    ("CONFLICT_RESOLVING", "conflict_resolved_first"): "IMPLEMENTING",
+    ("CONFLICT_RESOLVING", "conflict_resolved_second_same_tier"): "ESCALATING",
+    ("CONFLICT_RESOLVING", "conflict_repeated_three_plus"): "STUCK",
+    ("CONFLICT_RESOLVING", "conflict_irreconcilable"): "STUCK",
+    ("CONFLICT_RESOLVING", "pickup_exhausted"): "STUCK",
+
+    ("ESCALATING", "escalate_next_tier_available"): "IMPLEMENTING",
+    ("ESCALATING", "escalate_max_tier_exhausted"): "ABANDONED",
+
+    ("REVIEWING", "reviewer_approve"): "MERGING",
+    ("REVIEWING", "reviewer_request_changes_retry"): "IMPLEMENTING",
+    ("REVIEWING", "reviewer_request_changes_escalate"): "ESCALATING",
+    ("REVIEWING", "reviewer_abstain"): "STUCK",
+    ("REVIEWING", "pickup_exhausted"): "STUCK",
+
+    ("MERGING", "merge_ok"): "MERGED",
+    ("MERGING", "merge_base_conflict"): "IMPLEMENTING",
+    ("MERGING", "merge_ci_required_missing"): "AWAITING_CI",
+    ("MERGING", "merge_branch_protection_blocked"): "STUCK",
+    ("MERGING", "merge_retry_exhausted"): "STUCK",
+    ("MERGING", "merge_external_action"): "MERGED",  # reconciliation refines
+
+    # Operator escape hatch from STUCK.
+    ("STUCK", "operator_unstick"): "DISCOVERED",
+}
+
+
+# ─── transition lookup ────────────────────────────────────────────────
+
+
+class IllegalTransitionError(Exception):
+    """Raised when ``apply_event`` is called with an event that has no
+    mapped transition from the current state. The master catches this
+    and transitions the workflow to STUCK with reason='illegal-event'."""
+
+
+def apply_event(current_state: str, event_name: str) -> str:
+    """Return the next state for ``event_name`` from ``current_state``.
+
+    Raises ``ValueError`` if ``current_state`` isn't in ``KNOWN_STATES``
+    (callers should validate state at load time per v6 unknown-state
+    guard). Raises ``IllegalTransitionError`` if no transition is
+    mapped.
+    """
+    if current_state not in KNOWN_STATES:
+        raise ValueError(
+            f"unknown state {current_state!r}; not in KNOWN_STATES"
+        )
+    try:
+        return TRANSITIONS[(current_state, event_name)]
+    except KeyError:
+        raise IllegalTransitionError(
+            f"no transition from {current_state!r} on event "
+            f"{event_name!r}; known events from this state: "
+            f"{sorted(e for (s, e) in TRANSITIONS if s == current_state)}"
+        ) from None
+
+
+def events_from(state: str) -> list[str]:
+    """Return the list of event names that have a transition mapped
+    from ``state``. Useful for operator CLI 'what can I do from here?'
+    queries + property tests."""
+    return sorted(e for (s, e) in TRANSITIONS if s == state)
+
+
+def reachable_from(state: str) -> set[str]:
+    """Set of states reachable from ``state`` in one or more transitions.
+
+    Includes the starting state IFF it's part of a cycle (e.g.,
+    AWAITING_CI has a self-loop via ``ci_flake_retry``).
+    """
+    seen: set[str] = set()
+    stack = [state]
+    cycles_back = False
+    while stack:
+        s = stack.pop()
+        for (frm, _evt), to in TRANSITIONS.items():
+            if frm != s:
+                continue
+            if to == state and s != state:
+                # A transition leads back to the start — that's a cycle.
+                cycles_back = True
+            if to == state and s == state:
+                # Self-loop on the start state.
+                cycles_back = True
+            if to in seen:
+                continue
+            seen.add(to)
+            stack.append(to)
+    if not cycles_back:
+        seen.discard(state)
+    return seen
+
+
+# ─── property invariants (v9: ship 6 for v1) ──────────────────────────
+
+
+@dataclass(frozen=True)
+class InvariantViolation:
+    """Returned by ``check_*`` helpers when an invariant fails. The
+    property tests assert ``check_*() == []``."""
+
+    invariant: str
+    message: str
+
+
+def _violation(invariant: str, message: str) -> InvariantViolation:
+    return InvariantViolation(invariant=invariant, message=message)
+
+
+def check_no_path_implementing_to_reviewing_skips_ci(
+) -> list[InvariantViolation]:
+    """Invariant #1 (v9 load-bearing): REVIEWING is reachable from
+    IMPLEMENTING only through AWAITING_CI.
+
+    Closes the no-mans-land race by construction.
+    """
+    name = "no_path_implementing_to_reviewing_skips_ci"
+    # Direct edge IMPLEMENTING → REVIEWING would violate this.
+    for (frm, evt), to in TRANSITIONS.items():
+        if frm == "IMPLEMENTING" and to == "REVIEWING":
+            return [_violation(
+                name,
+                f"direct IMPLEMENTING → REVIEWING via {evt!r}; must "
+                f"route through AWAITING_CI"
+            )]
+    # AWAITING_CI is the only path; check it's reachable from
+    # IMPLEMENTING.
+    if "AWAITING_CI" not in reachable_from("IMPLEMENTING"):
+        return [_violation(
+            name, "IMPLEMENTING cannot reach AWAITING_CI"
+        )]
+    return []
+
+
+def check_terminal_states_have_no_exits() -> list[InvariantViolation]:
+    """Invariant #2: terminal states are sinks — no outgoing transitions.
+
+    Note: ``STUCK`` has the ``operator_unstick`` exit, which is the
+    OPERATOR-driven escape hatch. Treated as "outgoing for
+    operator-driven events, but not for controller-driven events" —
+    we permit it.
+    """
+    name = "terminal_states_have_no_exits"
+    out: list[InvariantViolation] = []
+    for (frm, evt), to in TRANSITIONS.items():
+        if frm in {"MERGED", "ABANDONED", "CREATED_PR"}:
+            out.append(_violation(
+                name,
+                f"terminal {frm!r} has transition via {evt!r} → {to!r}"
+            ))
+        # STUCK's only allowed exit is operator_unstick.
+        if frm == "STUCK" and evt != "operator_unstick":
+            out.append(_violation(
+                name,
+                f"STUCK has unauthorized exit via {evt!r} → {to!r}; "
+                f"only 'operator_unstick' is allowed"
+            ))
+    return out
+
+
+def check_tier_monotonic_non_decreasing() -> list[InvariantViolation]:
+    """Invariant #3: when ESCALATING fires, the next tier is strictly
+    greater than the previous. Encoded structurally: ESCALATING has
+    only two events — escalate_next_tier_available (→ IMPLEMENTING,
+    interpreted by master as tier+1) and escalate_max_tier_exhausted
+    (→ ABANDONED).
+    """
+    name = "tier_monotonic_non_decreasing"
+    out: list[InvariantViolation] = []
+    escalating_events = events_from("ESCALATING")
+    if set(escalating_events) != {
+        "escalate_next_tier_available", "escalate_max_tier_exhausted",
+    }:
+        out.append(_violation(
+            name,
+            f"ESCALATING has unexpected events {escalating_events}; "
+            f"expected {{'escalate_next_tier_available', "
+            f"'escalate_max_tier_exhausted'}}"
+        ))
+    return out
+
+
+def check_every_pr_workflow_includes_reviewing() -> list[InvariantViolation]:
+    """Invariant #4: every path from DISCOVERED to MERGED must pass
+    through REVIEWING.
+
+    Per Hard Rule #1: every PR receives a fresh LLM review.
+    """
+    name = "every_pr_workflow_includes_reviewing"
+    out: list[InvariantViolation] = []
+    # Check: no transition lands in MERGED from anything but MERGING.
+    for (frm, evt), to in TRANSITIONS.items():
+        if to == "MERGED" and frm != "MERGING":
+            out.append(_violation(
+                name,
+                f"MERGED reachable from {frm!r} via {evt!r} (not MERGING)"
+            ))
+    # Check: MERGING is reachable only via REVIEWING.
+    for (frm, evt), to in TRANSITIONS.items():
+        if to == "MERGING" and frm != "REVIEWING":
+            out.append(_violation(
+                name,
+                f"MERGING reachable from {frm!r} via {evt!r} (not REVIEWING)"
+            ))
+    return out
+
+
+def check_conflict_resolving_bounded() -> list[InvariantViolation]:
+    """Invariant #5: CONFLICT_RESOLVING transitions for 1st-time
+    resolution go to IMPLEMENTING; 2nd same-tier resolution goes to
+    ESCALATING; 3rd-or-later goes to STUCK. Structurally enforced
+    via exactly these named events.
+    """
+    name = "conflict_resolving_bounded"
+    out: list[InvariantViolation] = []
+    expected_events = {
+        "conflict_resolved_first": "IMPLEMENTING",
+        "conflict_resolved_second_same_tier": "ESCALATING",
+        "conflict_repeated_three_plus": "STUCK",
+        "conflict_irreconcilable": "STUCK",
+        "pickup_exhausted": "STUCK",
+    }
+    actual = {
+        evt: to for (frm, evt), to in TRANSITIONS.items()
+        if frm == "CONFLICT_RESOLVING"
+    }
+    for evt, expected_to in expected_events.items():
+        got_to = actual.get(evt)
+        if got_to != expected_to:
+            out.append(_violation(
+                name,
+                f"CONFLICT_RESOLVING + {evt!r}: expected → {expected_to!r}; "
+                f"got → {got_to!r}"
+            ))
+    unexpected = set(actual) - set(expected_events)
+    if unexpected:
+        out.append(_violation(
+            name,
+            f"CONFLICT_RESOLVING has unexpected event(s) {sorted(unexpected)}"
+        ))
+    return out
+
+
+def check_escalation_deterministic() -> list[InvariantViolation]:
+    """Invariant #6: ESCALATING has exactly two transitions, both
+    deterministic. Combined with tier-monotonic-non-decreasing, this
+    means the escalation policy is static (min(tier+1, MAX_TIER)).
+    """
+    name = "escalation_deterministic"
+    out: list[InvariantViolation] = []
+    actual = {
+        evt: to for (frm, evt), to in TRANSITIONS.items()
+        if frm == "ESCALATING"
+    }
+    if actual != {
+        "escalate_next_tier_available": "IMPLEMENTING",
+        "escalate_max_tier_exhausted": "ABANDONED",
+    }:
+        out.append(_violation(
+            name,
+            f"ESCALATING transitions are non-deterministic: {actual}"
+        ))
+    return out
+
+
+# Registry of all v1 load-bearing invariants. Tests iterate this.
+LOAD_BEARING_INVARIANTS: dict[str, callable] = {  # type: ignore[type-arg]
+    "no_path_implementing_to_reviewing_skips_ci":
+        check_no_path_implementing_to_reviewing_skips_ci,
+    "terminal_states_have_no_exits":
+        check_terminal_states_have_no_exits,
+    "tier_monotonic_non_decreasing":
+        check_tier_monotonic_non_decreasing,
+    "every_pr_workflow_includes_reviewing":
+        check_every_pr_workflow_includes_reviewing,
+    "conflict_resolving_bounded":
+        check_conflict_resolving_bounded,
+    "escalation_deterministic":
+        check_escalation_deterministic,
+}
+
+
+def check_all_invariants() -> list[InvariantViolation]:
+    """Run every load-bearing invariant; return concatenated violations."""
+    out: list[InvariantViolation] = []
+    for check in LOAD_BEARING_INVARIANTS.values():
+        out.extend(check())
+    return out
+
+
+__all__ = [
+    "EVENTS",
+    "Event",
+    "IllegalTransitionError",
+    "InvariantViolation",
+    "KNOWN_STATES",
+    "LOAD_BEARING_INVARIANTS",
+    "NON_TERMINAL_STATES",
+    "TERMINAL_STATES",
+    "TRANSITIONS",
+    "apply_event",
+    "check_all_invariants",
+    "events_from",
+    "reachable_from",
+]