From 976817fa221bcd0aaaada51d674d8d74ea2ced38 Mon Sep 17 00:00:00 2001 From: drew Date: Mon, 18 May 2026 13:34:28 -0400 Subject: [PATCH] =?UTF-8?q?feat(controller):=20Phase=201d-1=20=E2=80=94=20?= =?UTF-8?q?state=20machine=20+=20reaper=20+=20pickup=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../test_reaper_and_pickup_guard.py | 277 +++++++++ .../controller/test_state_machine.py | 220 +++++++ tools/controller/__init__.py | 6 +- tools/controller/pickup_guard.py | 130 ++++ tools/controller/reaper.py | 155 +++++ tools/controller/state_machine.py | 565 ++++++++++++++++++ 6 files changed, 1352 insertions(+), 1 deletion(-) create mode 100644 tests/auto_agents/controller/test_reaper_and_pickup_guard.py create mode 100644 tests/auto_agents/controller/test_state_machine.py create mode 100644 tools/controller/pickup_guard.py create mode 100644 tools/controller/reaper.py create mode 100644 tools/controller/state_machine.py diff --git a/tests/auto_agents/controller/test_reaper_and_pickup_guard.py b/tests/auto_agents/controller/test_reaper_and_pickup_guard.py new file mode 100644 index 000000000..853f2a8b0 --- /dev/null +++ b/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 diff --git a/tests/auto_agents/controller/test_state_machine.py b/tests/auto_agents/controller/test_state_machine.py new file mode 100644 index 000000000..a67a97dcd --- /dev/null +++ b/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)}" + ) diff --git a/tools/controller/__init__.py b/tools/controller/__init__.py index fd245e1c1..3ea372346 100644 --- a/tools/controller/__init__.py +++ b/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"] diff --git a/tools/controller/pickup_guard.py b/tools/controller/pickup_guard.py new file mode 100644 index 000000000..d642a9e60 --- /dev/null +++ b/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"] diff --git a/tools/controller/reaper.py b/tools/controller/reaper.py new file mode 100644 index 000000000..05f5a4266 --- /dev/null +++ b/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"] diff --git a/tools/controller/state_machine.py b/tools/controller/state_machine.py new file mode 100644 index 000000000..829023303 --- /dev/null +++ b/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", +]