feat(a5): Controlled Heartbeat & Resume v1

- a5/rq_heartbeat.py: HeartbeatStore + Heartbeat.run_tick (thin scheduler/resume layer)
- a5/rq_heartbeat_cli.py: status/tick/enable/disable/resume/approve/deny/priority
- a5/test_a5.py: 31 deterministic tests (tick-lock incl. ownership, eligibility, kill-switch, approval, circuit, git-conflict, bounded single A4)
- a5/DESIGN.md, a5/README.md, a5/scripts/a5_heartbeat_tick.sh
- a2/rq_mission.py: add read-only mission_list() for A5 enumeration (additive)
- Fresh checker: PASS after tick-lock ownership repair (Regressionschutz lock_owned)

a2/rq_mission.py

@@ -332,6 +332,20 @@ class MissionStore:
         finally:
             conn.close()
 
+    def mission_list(self) -> List[Dict[str, Any]]:
+        """Read-only: alle Missionen (id, title, state, created_at, updated_at),
+        deterministisch sortiert (created_at ASC, id ASC). Reine Enumeration —
+        keine Mutation, keine Transition. Dient A5 der Heartbeat-Eligibility."""
+        conn = _connect(self.db_path)
+        try:
+            rows = conn.execute(
+                "SELECT id,title,state,created_at,updated_at "
+                "FROM missions ORDER BY created_at ASC, id ASC"
+            ).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
     def mission_state(self, mission_id: str) -> Dict[str, Any]:
         conn = _connect(self.db_path)
         try:

a5/DESIGN.md

@@ -0,0 +1,34 @@
+# A5 DESIGN — CONTROLLED HEARTBEAT & RESUME v1
+
+## Integrations (verified from source)
+- **A2 `MissionStore`** (`a2/rq_mission.py`, `missions.db`):
+  - `mission_list()` [NEW, read-only, added for A5] — enumerate missions.
+  - `mission_read(id)` -> dict with `work_packages`, `dependencies`, `state`.
+  - `mission_state(id)`, `wp_state(id)`.
+  - Transitions: `mission_transition(id,target,actor,evidence)`, `mission_pause`, `mission_resume`, `mission_complete`, `wp_transition`, `wp_complete`.
+  - Transition tables in `a2/rq_state_machine.py`: eligible mission states {READY,RUNNING,REVIEW,PAUSED-resume-capable}; PAUSED->RUNNING allowed; BLOCKED->IN_PROGRESS (wp) allowed; REVIEW->RUNNING allowed.
+- **A3 `SafetyStore`** (`a3/rq_safety.py`, `safety_db`):
+  - `check_safety_state(mid,wp)` — fail-closed STATE_INCONSISTENT.
+  - `evaluate_next_action(mid,wp,is_mutating,persist)` — decision + circuit.
+  - `circuit_state(scope_type,scope_id)`, `open_circuit`, `attempts`, `attempts_count`, `safety_evidence`, `redact_secret`.
+- **A4 `Orchestrator`** (`a4/rq_orchestrator.py`):
+  - `__init__(mission_db, safety_db, child_dispatcher=None, max_bounded_runtime=...)`.
+  - `run_bounded(mission_id, max_steps=1, actor)` — EXACTLY-N bounded steps, guarantees return. Without child_dispatcher returns dispatch contracts; with dispatcher runs full Maker+Checker.
+  - `attempt_mission_complete(mission_id, final_review_pass=...)`.
+  - `run_one_step` = 1 child cycle. A4 has own limits (no loop).
+- **Hermes native cron**: `.tick.lock`, `executions.db`, no_agent, workdir, delivery. Baseline regression A2=61, A3=69, A4=83 PASS.
+
+## A5 result codes (§10)
+HB_NO_ACTIVE_MISSION, HB_MISSION_NOT_ELIGIBLE, HB_SAFETY_BLOCKED,
+HB_STATE_CONFLICT, HB_NO_RUNNABLE_WORK, HB_BOUNDED_RUN_COMPLETE,
+HB_MISSION_COMPLETED, HB_APPROVAL_REQUIRED, HB_ESCALATED, HB_CIRCUIT_OPEN,
+HB_TIMEOUT, HB_INTERNAL_ERROR, plus HB_LOCKED, HB_KILL_SWITCH_OFF,
+HB_PAUSED, HB_CANCELLED, HB_GIT_CONFLICT, HB_RECOVERY_REQUIRED.
+
+## Key invariants (spec §5,§9,§18,§25,§29,§30)
+- A5 is a thin layer: never fabricates states/retries/circuits. Delegates to A2/A3/A4.
+- Exactly ONE A4 `run_bounded(max_steps=1)` per tick per mission; MAX 1 mission per tick.
+- No while-true, no self-reschedule, no background child dependency.
+- Kill switch default OFF; blocks all mutation, read-only health allowed.
+- Fail-closed: any gate failure -> NO mutation, stable result code.
+- Dedup notifications by (mission, reason_code, evidence_hash).

a5/README.md

@@ -0,0 +1,145 @@
+# Red Queen — A5: Controlled Heartbeat & Resume (V1)
+
+Der **Controlled Heartbeat & Resume** verbindet den A4 **Bounded Orchestrator** mit
+einem **dünnen Scheduling-/Resume-Layer** (Heartbeat), sodass eine längere Mission
+über **mehrere getrennte, sichere, restart-feste Arbeitszyklen** fortgesetzt werden
+kann — **ohne unkontrollierten Dauer-LLM-Loop**.
+
+Der Heartbeat ist **NICHT der Orchestrator** und **keine neue Mission-Engine**:
+
+- **A2** (`missions.db`) bleibt autoritativ für Mission-/WP-State.
+- **A3** (`safety.db`) bleibt verbindliche Safety-Decision (Circuit, Evaluate, Redact).
+- **A4** bleibt die Orchestrierung (`run_bounded` = garantiert endender Zyklus).
+- **A5** ist eine **reine, inaktive Scheduling-/Resume-Schicht**, die pro Tick
+  **GENAU EINEN** bounded A4-Zyklus anstößt, das Ergebnis persistiert und sofort
+  endet. **Kein while-True, kein Daemon, kein self-reschedule.**
+
+```
+SCHEDULER (Hermes Cron) → Heartbeat.run_tick()
+  → Run-ID (Idempotenz) → Kill-Switch-Gate → Tick-Lock
+  → Mission-Eligibility (inkl. Circuit + Resume-Pending)
+    → A3 check_safety_state → optional Git-Preflight → Approval-Gate
+      → GENAU EIN A4 run_bounded(max_steps=1)
+        → Ergebnis persistieren → Lock freigeben → EXIT
+```
+
+---
+
+## Integrations (verified from source)
+
+- **A2 `MissionStore`** (`a2/rq_mission.py`, `missions.db`):
+  - `mission_list()` [NEW, read-only, additiv — A5-Enumeration]
+  - `mission_read(id)`, `mission_state(id)`, `wp_state(id)`
+  - Transitionen: `mission_transition`, `mission_pause`, `mission_resume`,
+    `mission_complete`, `wp_transition`, `wp_complete`.
+- **A3 `SafetyStore`** (`a3/rq_safety.py`, `safety_db`):
+  - `check_safety_state(mid, wp)` — fail-closed `STATE_INCONSISTENT`
+  - `evaluate_next_action(mid, wp, is_mutating, persist)`, `circuit_state`,
+    `open_circuit`, `attempts`, `safety_evidence`, `redact_secret`.
+- **A4 `Orchestrator`** (`a4/rq_orchestrator.py`):
+  - `run_bounded(mission_id, max_steps=1, actor)` — EXACTLY-N bounded steps,
+    garantiertes Ende. **Heartbeat ruft DAS GENAU EINMAL pro Tick.**
+- **Hermes native Cron**: `no_agent`, `executions.db`, `.tick.lock`, `context_from`,
+  `workdir`, `delivery` — die bevorzugte Scheduling-Plattform (kein zweiter Scheduler).
+
+---
+
+## Eigene Registry (A5-spezifisch, `heartbeat.db`)
+
+A5 führt **keine neue Missions-/Safety-DB**, sondern nur eine schlanke Registry für
+Heartbeat-eigene Steuerung:
+
+- `kill_switch` — `AUTONOMOUS_EXECUTION_ENABLED`, **Default OFF**. OFF → `run_tick`
+  liefert `HB_KILL_SWITCH_OFF` vor jeder A4-Mutation. `status`/`health` bleiben read-only.
+- `lock` — **EIN-Tick-Lock** (idempotent). `acquire_lock` → `ok|locked|recover`.
+  Stale nach `TICK_LOCK_STALE_SECONDS` (600s).
+- `run_id`-Ledger + `tick_log` — jede Tick-Ausführung wird persistiert (Idempotenz).
+- `approval_gates`, `resume_pending`, `notification_dedup`, `priority`.
+
+**Restart-Festigkeit:** Alle DBs (missions, safety, heartbeat) liegen auf **persistenter
+Volume**, nicht process-local. Ein frischer Prozess liest den State neu; neue Run-ID,
+keine Doppel-/Neu-Ausführung, kein verlorener Pending-State.
+
+---
+
+## A5-Result-Codes (§10)
+
+```
+HB_NO_ACTIVE_MISSION, HB_MISSION_NOT_ELIGIBLE, HB_SAFETY_BLOCKED,
+HB_STATE_CONFLICT, HB_NO_RUNNABLE_WORK, HB_BOUNDED_RUN_COMPLETE,
+HB_MISSION_COMPLETED, HB_APPROVAL_REQUIRED, HB_ESCALATED, HB_CIRCUIT_OPEN,
+HB_TIMEOUT, HB_INTERNAL_ERROR,
+HB_LOCKED, HB_KILL_SWITCH_OFF, HB_PAUSED, HB_CANCELLED,
+HB_GIT_CONFLICT, HB_RECOVERY_REQUIRED
+```
+
+`health()`/`status` sind read-only und immer erlaubt. Jeder gescheiterte Gate liefert
+einen **stabilen HB_*-Code** (fail-closed, keine partielle Mutation).
+
+---
+
+## CLI
+
+`python3 a5/rq_heartbeat_cli.py` — deterministisch, kein Loop, kein Root:
+
+```
+status                 Kill-Switch, letzter Tick, Registry-Infos
+tick [--mission ID]    EIN bounded Heartbeat-Zyklus (ohne --execute read-only)
+enable                 Kill-Switch ON (bewusst; erlaubt A4-Mutation)
+disable                Kill-Switch OFF (Fail-closed)
+resume --mission ID    PAUSED-Mission explizit als resume-fähig markieren
+approve --gate ID      Approval-Gate bewilligen
+deny    --gate ID      Approval-Gate ablehnen
+priority --mission --prio N
+```
+`--execute` ist ein **globales** Flag (vor dem Subcommand): ohne ist `tick` read-only,
+mit injiziert es einen echten `Orchestrator` und erlaubt einen A4-Mutation.
+
+---
+
+## Cron-Integration (Hermes native)
+
+```yaml
+cron:
+  # One-Shot oder wiederkehrend, no_agent=True, Script aus $HERMES_HOME/scripts
+  script: a5_heartbeat_tick.sh   # ruft EINEN bounded Tick; kein Loop
+```
+
+One-Shot-Jobs werden nach Ausführung automatisch entfernt → es bleibt **kein
+unkontrollierter Dauer-Job** zurück (`hermes cron list` = 0). Der Tick läuft mit
+`no_agent` über die Heartbeat-Library, **ohne** eigenständigen LLM-Loop.
+
+---
+
+## Test Harness
+
+`python3 a5/test_a5.py` — isolierte, deterministische Tests (temp-DBs, kein Netz,
+kein echter A4-Dispatch außer einem bounded real-Orchestrator-Test). Exit-Code 0 = PASS.
+Deckt: Kill-Switch off/on, **Tick-Lock (overlap → `HB_LOCKED` und Lock-Erhalt)**,
+Lock-Release nach Owner-Tick, Eligibility (cancelled/paused/resumable), Priorität,
+**GENAU EIN** bounded A4-Call, No-Runnable, Approval-Block/Grant/Resume,
+Circuit-Block, Git-Konflikt, unbekannte Mission, read-only health,
+bounded-real-Orchestrator-Contract.
+
+Regressions: `a2/test_a2.py` (61), `a3/test_a3.py` (69), `a4/test_a4.py` (83).
+
+---
+
+## Fail-closed & Approval
+
+- `mission_completion_gate(..., final_review_pass=False)` — A5 übergibt standardmäßig
+  `False` → **kein autonomes Final-Review/Auto-Complete in V1**.
+- Approval-Gates persistieren in der Registry; erst nach `approve` wird die Mission
+  wieder runnable.
+- `open_circuit` (A3) ist **keyword-only** nach scope_type/scope_id:
+  `open_circuit("MISSION", mid, trigger=..., reason=...)`.
+
+---
+
+## Known Limitations
+
+- A5 selbst startet **keine** echte A4-Mutation außerhalb eines ausdrücklichen
+  `--execute`-Ticks. Produktiver Heartbeat/Cron ist erst nach expliziter Freigabe
+  (Kill-Switch) aktiv; Default OFF.
+- Kein automatisches Rain; kein Self-Improvement-Automation; kein unkontrollierter
+  Dauerlauf.

a5/rq_heartbeat.py

@@ -0,0 +1,707 @@
+#!/usr/bin/env python3
+"""
+Red Queen — A5: CONTROLLED HEARTBEAT & RESUME v1 (dünne Scheduling-/Resume-Schicht).
+
+VERBINDLICHER RAHMEN
+====================
+A5 ist KEINE neue Mission-Engine und KEIN Orchestrator. A5 ist eine DUENNE,
+deterministische Scheduler-/Resume-Schicht, die Red Queen (Hermes) aufweckt,
+einen GENAU BEGRENZTEN A4-Arbeitszyklus anstösst, den Zustand persistiert und
+dann stoppt. Es gibt KEINEN while-true, keinen unbounded Agent-Loop, kein
+Selbst-Reschedule, kein Auto-Rain, kein Self-Improvement.
+
+Authority:
+  * A2 `MissionStore` = Mission-/WP-State Authority (Transitions, Dependencies).
+  * A3 `SafetyStore`  = Safety Authority (Circuit, Retry, Attempt-Ledger).
+  * A4 `Orchestrator` = Orchestrierungs-Logik (bounded run, Approval, Completion).
+A5 erfindet NICHTS davon — es delegiert ausschliesslich und bewertet den
+Rueckgabecode. A5 implementiert NUR: Kill-Switch, Tick-Lock, Run-ID, Mission-
+Eligibility & -Priorität, exakt-ein-bounded-A4-Aufruf pro Tick, Notification-
+Dedup, Approval-Gate-Warten, Recovery-Reconciliation (fail-closed) und das
+Heartbeat-Registry.
+
+Kill-Switch (§25/§26): persistierter globaler Gate AUTONOMOUS_EXECUTION_ENABLED.
+Default OFF. Wenn OFF -> KEINE A4-Mutation; read-only Health erlaubt. Das Gate
+wird nie von einem LLM-Satz überschrieben (Gated Config, >= SI-3).
+
+V1: MAX 1 Mission pro Heartbeat. Pro Tick genau EIN A4 bounded run (max_steps=1).
+Keine parallelen Missionen.
+
+RESTART-FEST: Alle Zustände (Registry, Tick-Log, Dedup, Approval-Gate,
+Kill-Switch) liegen in einer eigenen, isolierten SQLite-DB (heartbeat.db).
+Nach Container-/Hermes-Restart werden die DBs neu geladen und deterministisch
+fortgesetzt.
+"""
+
+from __future__ import annotations
+
+import datetime
+import hashlib
+import json
+import sqlite3
+import sys
+import uuid
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional
+
+_REPO_ROOT = Path(__file__).resolve().parent.parent
+for _p in (str(_REPO_ROOT), str(_REPO_ROOT / "a2"), str(_REPO_ROOT / "a3"), str(_REPO_ROOT / "a4")):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from a2.rq_mission import MissionStore, RqError  # noqa: E402
+from a3.rq_safety import (  # noqa: E402
+    CIRCUIT_OPEN,
+    SafetyError,
+    SafetyStore,
+    redact_secret,
+)
+from a4.rq_orchestrator import (  # noqa: E402
+    CIRCUIT_CLOSED,
+    ST_APPROVAL_REQUIRED,
+    ST_BUDGET_REACHED,
+    ST_CHECKER_PASS,
+    ST_DEBUG_REQUIRED,
+    ST_MISSION_COMPLETED,
+    ST_MISSION_NOT_COMPLETED,
+    ST_MUTATION_BLOCKED,
+    ST_NEEDS_CHECKER_DISPATCH,
+    ST_NEEDS_DISPATCH,
+    ST_NO_RUNNABLE_WORK,
+    ST_SECOND_OPINION_REQUIRED,
+    ST_STATE_CONFLICT,
+    ST_TERMINAL,
+    ST_WAITING_PENDING,
+    ST_WP_DONE,
+    Orchestrator,
+    OrchestratorError,
+)
+
+# --------------------------------------------------------------------------- #
+# A5-Fehler
+# --------------------------------------------------------------------------- #
+class HeartbeatError(Exception):
+    def __init__(self, code: str, message: str, detail: Optional[Dict[str, Any]] = None):
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.detail = detail or {}
+
+
+def _err(code: str, msg: str, **detail) -> HeartbeatError:
+    return HeartbeatError(code, msg, detail)
+
+
+# --------------------------------------------------------------------------- #
+# A5-Konstanten (deterministisch)
+# --------------------------------------------------------------------------- #
+HB_NO_ACTIVE_MISSION = "HB_NO_ACTIVE_MISSION"
+HB_MISSION_NOT_ELIGIBLE = "HB_MISSION_NOT_ELIGIBLE"
+HB_SAFETY_BLOCKED = "HB_SAFETY_BLOCKED"
+HB_STATE_CONFLICT = "HB_STATE_CONFLICT"
+HB_NO_RUNNABLE_WORK = "HB_NO_RUNNABLE_WORK"
+HB_BOUNDED_RUN_COMPLETE = "HB_BOUNDED_RUN_COMPLETE"
+HB_MISSION_COMPLETED = "HB_MISSION_COMPLETED"
+HB_APPROVAL_REQUIRED = "HB_APPROVAL_REQUIRED"
+HB_ESCALATED = "HB_ESCALATED"
+HB_CIRCUIT_OPEN = "HB_CIRCUIT_OPEN"
+HB_TIMEOUT = "HB_TIMEOUT"
+HB_INTERNAL_ERROR = "HB_INTERNAL_ERROR"
+HB_LOCKED = "HB_LOCKED"
+HB_KILL_SWITCH_OFF = "HB_KILL_SWITCH_OFF"
+HB_PAUSED = "HB_PAUSED"
+HB_CANCELLED = "HB_CANCELLED"
+HB_TERMINAL = "HB_TERMINAL"
+HB_GIT_CONFLICT = "HB_GIT_CONFLICT"
+HB_RECOVERY_REQUIRED = "HB_RECOVERY_REQUIRED"
+HB_APPROVAL_PENDING = "HB_APPROVAL_PENDING"
+HB_MISSION_NOT_COMPLETED = "HB_MISSION_NOT_COMPLETED"
+
+ELIGIBLE_STATES = frozenset({"READY", "RUNNING", "REVIEW"})
+RESUME_CAPABLE = frozenset({"PAUSED"})
+
+TICK_LOCK_STALE_SECONDS = 600  # 10 min ohne Heartbeat -> stale (recovery)
+
+# --------------------------------------------------------------------------- #
+# DB-Hilfsfunktionen
+# --------------------------------------------------------------------------- #
+def _connect(path: str) -> sqlite3.Connection:
+    conn = sqlite3.connect(path)
+    conn.row_factory = sqlite3.Row
+    conn.execute("PRAGMA journal_mode=WAL")
+    conn.execute("PRAGMA foreign_keys=ON")
+    return conn
+
+
+def _utcnow() -> str:
+    return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds")
+
+
+def _now_epoch() -> float:
+    return datetime.datetime.now(datetime.timezone.utc).timestamp()
+
+
+def _new_run_id() -> str:
+    return f"HB-{datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%dT%H%M%S')}-{uuid.uuid4().hex[:8]}"
+
+
+# --------------------------------------------------------------------------- #
+# HeartbeatStore: persistente A5-Registry
+# --------------------------------------------------------------------------- #
+class HeartbeatStore:
+    """Isolierte, durable A5-Registry (heartbeat.db)."""
+
+    def __init__(self, db_path: str):
+        self.db_path = db_path
+        self._ensure_schema()
+
+    def _ensure_schema(self) -> None:
+        conn = _connect(self.db_path)
+        try:
+            conn.executescript(
+                """
+                CREATE TABLE IF NOT EXISTS config (
+                    key TEXT PRIMARY KEY,
+                    value TEXT NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS tick_log (
+                    run_id TEXT PRIMARY KEY,
+                    started_at TEXT NOT NULL,
+                    finished_at TEXT,
+                    result_code TEXT,
+                    mission_id TEXT,
+                    exit_reason TEXT,
+                    detail TEXT
+                );
+                CREATE TABLE IF NOT EXISTS notification_dedup (
+                    dedup_key TEXT PRIMARY KEY,
+                    sent_at TEXT NOT NULL,
+                    mission_id TEXT,
+                    reason_code TEXT,
+                    message_id TEXT
+                );
+                CREATE TABLE IF NOT EXISTS lock (
+                    id INTEGER PRIMARY KEY CHECK (id = 1),
+                    owner TEXT,
+                    acquired_at TEXT,
+                    heartbeat_at TEXT
+                );
+                CREATE TABLE IF NOT EXISTS approval_gates (
+                    gate_id TEXT PRIMARY KEY,
+                    mission_id TEXT NOT NULL,
+                    wp_id TEXT NOT NULL,
+                    action TEXT NOT NULL,
+                    requested_action TEXT,
+                    risk TEXT,
+                    state TEXT NOT NULL,
+                    requested_at TEXT NOT NULL,
+                    approved_at TEXT,
+                    approved_by TEXT,
+                    evidence TEXT
+                );
+                """
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    # -- config ---------------------------------------------------------------
+    def get_config(self, key: str, default: Optional[str] = None) -> Optional[str]:
+        conn = _connect(self.db_path)
+        try:
+            row = conn.execute("SELECT value FROM config WHERE key = ?", (key,)).fetchone()
+            return row["value"] if row else default
+        finally:
+            conn.close()
+
+    def set_config(self, key: str, value: str) -> None:
+        conn = _connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO config (key,value) VALUES (?,?) "
+                "ON CONFLICT(key) DO UPDATE SET value=excluded.value",
+                (key, value),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    # -- Kill Switch ----------------------------------------------------------
+    def kill_switch_on(self) -> bool:
+        v = self.get_config("AUTONOMOUS_EXECUTION_ENABLED", "OFF")
+        return str(v).strip().upper() == "ON"
+
+    def set_kill_switch(self, on: bool) -> None:
+        self.set_config("AUTONOMOUS_EXECUTION_ENABLED", "ON" if on else "OFF")
+
+    # -- Resume-Flag (PAUSED -> resume-faehig) --------------------------------
+    def set_resume_pending(self, mission_id: str, pending: bool) -> None:
+        self.set_config(f"resume_pending:{mission_id}", "1" if pending else "0")
+
+    def resume_pending(self, mission_id: str) -> bool:
+        return self.get_config(f"resume_pending:{mission_id}", "0") == "1"
+
+    def set_priority(self, mission_id: str, priority: int) -> None:
+        self.set_config(f"priority:{mission_id}", str(int(priority)))
+
+    # -- Tick Lock ------------------------------------------------------------
+    def acquire_lock(self, run_id: str) -> str:
+        """EIN-Tick-Lock. Rueckgabe 'ok'|'locked'|'recover'."""
+        now = _now_epoch()
+        conn = _connect(self.db_path)
+        try:
+            row = conn.execute("SELECT * FROM lock WHERE id=1").fetchone()
+            if row is None:
+                conn.execute(
+                    "INSERT INTO lock(id,owner,acquired_at,heartbeat_at) VALUES (1,?,?,?)",
+                    (run_id, _utcnow(), now),
+                )
+                conn.commit()
+                return "ok"
+            heartbeat_at = float(row["heartbeat_at"]) if row["heartbeat_at"] else 0
+            if now - heartbeat_at <= TICK_LOCK_STALE_SECONDS:
+                return "locked"
+            return "recover"
+        finally:
+            conn.close()
+
+    def heartbeat_mark(self, run_id: str) -> None:
+        conn = _connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO lock(id,owner,acquired_at,heartbeat_at) VALUES (1,?,?,?) "
+                "ON CONFLICT(id) DO UPDATE SET owner=excluded.owner, heartbeat_at=excluded.heartbeat_at",
+                (run_id, _utcnow(), _now_epoch()),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    def release_lock(self) -> None:
+        conn = _connect(self.db_path)
+        try:
+            conn.execute("DELETE FROM lock WHERE id=1")
+            conn.commit()
+        finally:
+            conn.close()
+
+    # -- tick_log -------------------------------------------------------------
+    def log_tick(self, run_id, started_at, finished_at, result_code, mission_id,
+                 exit_reason, detail=None) -> None:
+        conn = _connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT OR REPLACE INTO tick_log "
+                "(run_id,started_at,finished_at,result_code,mission_id,exit_reason,detail) "
+                "VALUES (?,?,?,?,?,?,?)",
+                (run_id, started_at, finished_at, result_code, mission_id, exit_reason,
+                 json.dumps(detail or {}, default=str)),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    def last_tick(self) -> Optional[Dict[str, Any]]:
+        conn = _connect(self.db_path)
+        try:
+            row = conn.execute("SELECT * FROM tick_log ORDER BY started_at DESC LIMIT 1").fetchone()
+            if row is None:
+                return None
+            d = dict(row)
+            try:
+                d["detail"] = json.loads(d["detail"]) if d["detail"] else {}
+            except (ValueError, TypeError):
+                d["detail"] = {}
+            return d
+        finally:
+            conn.close()
+
+    def run_id_known(self, run_id: str) -> bool:
+        conn = _connect(self.db_path)
+        try:
+            return conn.execute("SELECT 1 FROM tick_log WHERE run_id=?", (run_id,)).fetchone() is not None
+        finally:
+            conn.close()
+
+    # -- Notification-Dedup ---------------------------------------------------
+    def notification_key(self, mission_id: str, reason_code: str, evidence: str) -> str:
+        raw = f"{mission_id}|{reason_code}|{evidence}"
+        return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+    def notification_seen(self, dedup_key: str) -> bool:
+        conn = _connect(self.db_path)
+        try:
+            return conn.execute("SELECT 1 FROM notification_dedup WHERE dedup_key=?", (dedup_key,)).fetchone() is not None
+        finally:
+            conn.close()
+
+    def notification_record(self, dedup_key: str, mission_id: str, reason_code: str, message_id: str) -> None:
+        conn = _connect(self.db_path)
+        try:
+            conn.execute(
+                "INSERT INTO notification_dedup (dedup_key,sent_at,mission_id,reason_code,message_id) "
+                "VALUES (?,?,?,?,?)",
+                (dedup_key, _utcnow(), mission_id, reason_code, message_id),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+
+    # -- Approval Gate --------------------------------------------------------
+    def approval_pending(self, mission_id: str) -> Optional[Dict[str, Any]]:
+        conn = _connect(self.db_path)
+        try:
+            row = conn.execute(
+                "SELECT * FROM approval_gates WHERE mission_id=? AND state='PENDING' "
+                "ORDER BY requested_at LIMIT 1",
+                (mission_id,),
+            ).fetchone()
+            return dict(row) if row else None
+        finally:
+            conn.close()
+
+    def approval_issue(self, mission_id: str, wp_id: str, action: str, risk: str) -> str:
+        """Persistiert einen wartenden Approval-Gate. Idempotent (kein Duplikat
+        fuer denselben WP im PENDING-Zustand)."""
+        conn = _connect(self.db_path)
+        try:
+            existing = conn.execute(
+                "SELECT * FROM approval_gates WHERE mission_id=? AND wp_id=? AND state='PENDING'",
+                (mission_id, wp_id),
+            ).fetchone()
+            if existing:
+                return existing["gate_id"]
+            gate_id = str(uuid.uuid4())
+            conn.execute(
+                "INSERT INTO approval_gates "
+                "(gate_id,mission_id,wp_id,action,requested_action,risk,state,requested_at) "
+                "VALUES (?,?,?,?,?,?,?,?)",
+                (gate_id, mission_id, wp_id, "APPROVE", action, risk, "PENDING", _utcnow()),
+            )
+            conn.commit()
+            return gate_id
+        finally:
+            conn.close()
+
+    def approval_grant(self, gate_id: str, approved_by: str, evidence: str) -> Dict[str, Any]:
+        conn = _connect(self.db_path)
+        try:
+            row = conn.execute("SELECT * FROM approval_gates WHERE gate_id=?", (gate_id,)).fetchone()
+            if row is None:
+                raise _err("UNKNOWN_GATE", f"unknown approval gate {gate_id!r}", gate_id=gate_id)
+            if row["state"] == "APPROVED":
+                return {"gate_id": gate_id, "state": "APPROVED", "idempotent": True}
+            conn.execute(
+                "UPDATE approval_gates SET state='APPROVED', approved_at=?, approved_by=?, evidence=? "
+                "WHERE gate_id=?",
+                (_utcnow(), approved_by, evidence, gate_id),
+            )
+            conn.commit()
+            return {"gate_id": gate_id, "state": "APPROVED", "idempotent": False}
+        finally:
+            conn.close()
+
+    def approval_deny(self, gate_id: str, denied_by: str, evidence: str) -> Dict[str, Any]:
+        conn = _connect(self.db_path)
+        try:
+            row = conn.execute("SELECT * FROM approval_gates WHERE gate_id=?", (gate_id,)).fetchone()
+            if row is None:
+                raise _err("UNKNOWN_GATE", f"unknown approval gate {gate_id!r}", gate_id=gate_id)
+            conn.execute(
+                "UPDATE approval_gates SET state='DENIED', approved_at=?, approved_by=?, evidence=? "
+                "WHERE gate_id=?",
+                (_utcnow(), denied_by, evidence, gate_id),
+            )
+            conn.commit()
+            return {"gate_id": gate_id, "state": "DENIED"}
+        finally:
+            conn.close()
+
+
+# --------------------------------------------------------------------------- #
+# Heartbeat
+# --------------------------------------------------------------------------- #
+class Heartbeat:
+    """A5: Kontrollierter Heartbeat & Resume. `run_tick()` = genau ein bounded cycle."""
+
+    def __init__(
+        self,
+        registry_db: str,
+        mission_db: str,
+        safety_db: str,
+        orchestrator: Optional[Orchestrator] = None,
+        *,
+        max_bounded_steps: int = 1,
+        git_preflight: Optional[Callable[[str], Dict[str, Any]]] = None,
+    ):
+        self.registry = HeartbeatStore(registry_db)
+        self.missions = MissionStore(mission_db)
+        self.safety = SafetyStore(safety_db)
+        self.orchestrator = orchestrator
+        self.max_bounded_steps = max_bounded_steps
+        self.git_preflight = git_preflight
+
+    # -- Helper: Mission-Enumeration ------------------------------------------
+    def _mission(self) -> List[Dict[str, Any]]:
+        try:
+            return self.missions.mission_list()
+        except RqError:
+            return []
+
+    # -- Eligibility & Priority (deterministisch) ------------------------------
+    def _eligible_states(self, state: str, resume_pending: bool) -> bool:
+        if state in ELIGIBLE_STATES:
+            return True
+        if state == "PAUSED" and resume_pending:
+            return True
+        return False
+
+    def _mission_circuit(self, mission_id: str) -> Optional[str]:
+        try:
+            g = self.safety.circuit_state("GLOBAL", "global")
+            if g["state"] == CIRCUIT_OPEN:
+                return "global"
+        except SafetyError:
+            pass
+        try:
+            m = self.safety.circuit_state("MISSION", mission_id)
+            if m["state"] == CIRCUIT_OPEN:
+                return f"mission:{mission_id}"
+        except SafetyError:
+            pass
+        return None
+
+    def _not_eligible_code(self, state: str) -> str:
+        if state in ("COMPLETED", "FAILED", "ESCALATED"):
+            return HB_TERMINAL
+        if state == "CANCELLED":
+            return HB_CANCELLED
+        if state == "PAUSED":
+            return HB_PAUSED
+        return HB_MISSION_NOT_ELIGIBLE
+
+    def select_eligible(self, mission_id: Optional[str] = None) -> Dict[str, Any]:
+        """Deterministische Auswahl EINER heartbeat-eligible Mission.
+
+        Rueckgabe {'eligible', 'mission', 'reason', 'code'}."""
+        if mission_id is not None:
+            try:
+                m = self.missions.mission_read(mission_id)
+            except RqError as e:
+                return {"eligible": False, "mission": None, "reason": e.message, "code": HB_NO_ACTIVE_MISSION}
+            st = m["state"]
+            resume_pending = self.registry.resume_pending(mission_id)
+            if not self._eligible_states(m["state"], resume_pending):
+                return {"eligible": False, "mission": {"id": m["id"], "state": st},
+                        "reason": f"mission {st}; not eligible", "code": self._not_eligible_code(st)}
+            circ = self._mission_circuit(mission_id)
+            if circ is not None:
+                return {"eligible": False, "mission": {"id": m["id"], "state": st},
+                        "reason": f"circuit open: {circ}", "code": HB_CIRCUIT_OPEN}
+            return {"eligible": True, "mission": {"id": m["id"], "state": st, "created_at": m["created_at"]},
+                    "reason": "", "code": "OK"}
+
+        cands = []
+        for m in self._mission():
+            resume_pending = self.registry.resume_pending(m["id"])
+            if not self._eligible_states(m["state"], resume_pending):
+                continue
+            if self._mission_circuit(m["id"]) is not None:
+                continue
+            cands.append({"id": m["id"], "state": m["state"], "created_at": m["created_at"],
+                          "priority": int(self.registry.get_config(f"priority:{m['id']}", "0") or 0)})
+        if not cands:
+            return {"eligible": False, "mission": None, "reason": "no active eligible mission",
+                    "code": HB_NO_ACTIVE_MISSION}
+        cands.sort(key=lambda x: (-x["priority"], x["created_at"], x["id"]))
+        return {"eligible": True, "mission": cands[0], "reason": "", "code": "OK"}
+
+    # -- run_tick --------------------------------------------------------------
+    def run_tick(self, mission_id: Optional[str] = None, *, execute: bool = True,
+                 actor: str = "red-queen") -> Dict[str, Any]:
+        """EIN Heartbeat-Tick. Endet garantiert."""
+        started = _utcnow()
+        run_id = _new_run_id()
+        result_code = HB_INTERNAL_ERROR
+        exit_reason = ""
+        mission_processed: Optional[str] = None
+        detail: Dict[str, Any] = {}
+        lock_owned = False  # nur TRUE, wenn DIESER Tick das Tick-Lock erworben hat
+
+        try:
+            # Idempotenz: gleiche Run-ID nie doppelt
+            if self.registry.run_id_known(run_id):
+                return self._finalize(run_id, started, HB_NO_RUNNABLE_WORK, None,
+                                      "run id already processed", {"idempotent": True})
+
+            # Kill-Switch-Gate
+            if not self.registry.kill_switch_on():
+                return self._finalize(run_id, started, HB_KILL_SWITCH_OFF, None,
+                                     "autonomous execution disabled (kill switch OFF)",
+                                     {"mutating": False})
+
+            # Tick-Lock
+            lock = self.registry.acquire_lock(run_id)
+            if lock == "locked":
+                return self._finalize(run_id, started, HB_LOCKED, None,
+                                     "tick lock held: overlapping heartbeat prevented",
+                                     {"lock": "locked"})
+            if lock == "recover":
+                return self._finalize(run_id, started, HB_RECOVERY_REQUIRED, None,
+                                 "stale tick lock: recovery required, no mutation",
+                                 {"lock": "recover"})
+            lock_owned = True
+            self.registry.heartbeat_mark(run_id)
+
+            try:
+                # Eligibility
+                sel = self.select_eligible(mission_id)
+                if not sel["eligible"]:
+                    mission_processed = sel.get("mission", {}).get("id") if sel.get("mission") else None
+                    return self._finalize(run_id, started, sel["code"], mission_processed,
+                                         sel["reason"], {"mission": sel.get("mission")})
+
+                mid = sel["mission"]["id"]
+                mission_processed = mid
+
+                # A2-Load
+                try:
+                    m = self.missions.mission_read(mid)
+                except RqError as e:
+                    return self._finalize(run_id, started, HB_STATE_CONFLICT, mid,
+                                         f"mission read failed: {e.message}", {})
+                if m["state"] in ("COMPLETED", "FAILED", "CANCELLED", "ESCALATED"):
+                    return self._finalize(run_id, started, HB_TERMINAL, mid,
+                                         f"mission {m['state']}; terminal", {})
+
+                # A3 Safety
+                try:
+                    self.safety.check_safety_state(mid)
+                except SafetyError as e:
+                    return self._finalize(run_id, started, HB_SAFETY_BLOCKED, mid,
+                                         f"safety state: {e.message}", {"reason_code": e.code})
+
+                # Git-Preflight
+                if self.git_preflight is not None:
+                    git = self.git_preflight(mid)
+                    if not git.get("ok", False):
+                        return self._finalize(run_id, started, HB_GIT_CONFLICT, mid,
+                                             f"git state conflict: {git.get('reason')}", {"git": git})
+
+                # Approval-Pending (persistierter Gate)
+                pending = self.registry.approval_pending(mid)
+                if pending is not None:
+                    return self._finalize(run_id, started, HB_APPROVAL_PENDING, mid,
+                                         "approval gate pending; cannot dispatch",
+                                         {"gate_id": pending["gate_id"], "wp_id": pending["wp_id"]})
+
+                # Execute-Gate (read-only)
+                if not execute:
+                    return self._finalize(run_id, started, HB_NO_RUNNABLE_WORK, mid,
+                                         "execute=False (read-only); no A4 mutation", {})
+
+                # EXACTLY ONE bounded A4 run
+                a4 = self._orchestrator()
+                a4_result = a4.run_bounded(mid, max_steps=self.max_bounded_steps, actor=actor)
+                st = a4_result.get("status")
+                detail["a4"] = a4_result
+
+                # Map A4 -> A5
+                if st in (ST_NEEDS_DISPATCH, ST_NEEDS_CHECKER_DISPATCH, ST_WAITING_PENDING,
+                          ST_CHECKER_PASS, ST_WP_DONE, ST_BUDGET_REACHED):
+                    result_code = HB_BOUNDED_RUN_COMPLETE
+                    exit_reason = f"bounded a4 cycle complete: {st}"
+                elif st == ST_NO_RUNNABLE_WORK:
+                    # Completion-Gate pruefen
+                    cc = a4.mission_completion_gate(mid, final_review_pass=False)
+                    if cc["valid"]:
+                        cmp = a4.attempt_mission_complete(mid, final_review_pass=False)
+                        if cmp["status"] == ST_MISSION_COMPLETED:
+                            result_code = HB_MISSION_COMPLETED
+                            exit_reason = "mission completed (all wps done)"
+                            detail["completed"] = True
+                        else:
+                            result_code = HB_MISSION_NOT_COMPLETED
+                            exit_reason = f"completion gate passed but complete failed: {cmp.get('reasons')}"
+                    else:
+                        result_code = HB_NO_RUNNABLE_WORK
+                        exit_reason = "no runnable work"
+                elif st == ST_APPROVAL_REQUIRED:
+                    result_code = HB_APPROVAL_REQUIRED
+                    wp_id = a4_result.get("wp_id") or a4_result.get("approval", {}).get("WP")
+                    act = a4_result.get("approval", {}).get("REQUESTED_ACTION") or ""
+                    risk = a4_result.get("risk") or "CRITICAL"
+                    if wp_id:
+                        self.registry.approval_issue(mid, wp_id, act, risk)
+                    exit_reason = "approval required; gate persisted"
+                elif st == ST_MUTATION_BLOCKED:
+                    result_code = HB_CIRCUIT_OPEN
+                    exit_reason = f"mutation blocked: {a4_result.get('reason')}"
+                elif st == ST_STATE_CONFLICT:
+                    result_code = HB_STATE_CONFLICT
+                    exit_reason = a4_result.get("reason", "orchestrator state conflict")
+                elif st in (ST_DEBUG_REQUIRED, ST_SECOND_OPINION_REQUIRED):
+                    result_code = HB_ESCALATED
+                    exit_reason = f"escalated: {st}"
+                elif st == ST_MISSION_COMPLETED:
+                    result_code = HB_MISSION_COMPLETED
+                    exit_reason = "mission completed"
+                elif st == ST_TERMINAL:
+                    result_code = HB_TERMINAL
+                    exit_reason = "mission terminal"
+                else:
+                    result_code = HB_INTERNAL_ERROR
+                    exit_reason = f"unhandled orchestrator status {st!r}"
+
+                return self._finalize(run_id, started, result_code, mid, exit_reason, detail)
+
+            except HeartbeatError as e:
+                return self._finalize(run_id, started, HB_INTERNAL_ERROR, mission_processed,
+                                     e.message, {"code": e.code})
+            finally:
+                if lock_owned:
+                    self.registry.release_lock()
+
+        except HeartbeatError as e:
+            return self._finalize(run_id, started, HB_INTERNAL_ERROR, mission_processed,
+                                 e.message, {"code": e.code})
+        except Exception as e:  # noqa
+            return self._finalize(run_id, started, HB_INTERNAL_ERROR, mission_processed,
+                                 f"internal error: {redact_secret(str(e))}", {})
+        finally:
+            if lock_owned:
+                try:
+                    self.registry.release_lock()
+                except Exception:
+                    pass
+
+    def _orchestrator(self) -> Orchestrator:
+        if self.orchestrator is not None:
+            return self.orchestrator
+        return Orchestrator(mission_db=self.missions.db_path, safety_db=self.safety.db_path)
+
+    def _finalize(self, run_id, started, result_code, mission_id, exit_reason, detail) -> Dict[str, Any]:
+        finished = _utcnow()
+        self.registry.log_tick(run_id, started, finished, result_code, mission_id, exit_reason, detail)
+        return {
+            "run_id": run_id,
+            "code": result_code,
+            "result": result_code,
+            "mission_id": mission_id,
+            "exit_reason": exit_reason,
+            "detail": detail,
+            "started_at": started,
+            "finished_at": finished,
+        }
+
+    # -- Health (read-only) ----------------------------------------------------
+    def health(self) -> Dict[str, Any]:
+        return {
+            "kill_switch": "ON" if self.registry.kill_switch_on() else "OFF",
+            "last_tick": self.registry.last_tick(),
+            "missions": [{"id": m["id"], "state": m["state"]} for m in self._mission()],
+        }
+
+
+# Referenz: Wir brauchen einen Code fuer "Completion-Gate ok aber complete failed".
+HB_MISSION_NOT_FOUND_CODE = "HB_MISSION_NOT_COMPLETED"

a5/rq_heartbeat_cli.py

@@ -0,0 +1,150 @@
+#!/usr/bin/env python3
+"""Red Queen — A5: Heartbeat CLI (CONTROLLED HEARTBEAT & RESUME v1).
+
+Operational Commands (§52):
+  status                 Read-only: Kill-Switch, letzter Tick, Missionen, offene Gates.
+  tick [MISSION_ID]      EIN bounded Heartbeat-Zyklus (Kill-Switch, Lock, Eligibility, A4).
+  enable                 Kill-Switch ON  (bewusst; autonome A4-Mutation erlaubt).
+  disable                Kill-Switch OFF (Fail-closed: keine A4-Mutation).
+  resume --mission ID    Explizite Resume einer PAUSED-Mission markieren.
+  approve --gate GATE_ID [--by NAME --evidence TEXT]
+  deny    --gate GATE_ID [--by NAME --evidence TEXT]
+  priority --mission ID --prio N
+  status                  Registry-/Tick-Log-Anzeige.
+
+Alle Kommandos sind deterministisch und laufen nie zyklisch. Kein Root.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from pathlib import Path
+
+_HERE = Path(__file__).resolve().parent
+_REPO = _HERE.parent
+for _p in (str(_REPO), str(_REPO / "a2"), str(_REPO / "a3"), str(_REPO / "a4"), str(_HERE)):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from a4.rq_orchestrator import Orchestrator  # noqa: E402
+from a5.rq_heartbeat import Heartbeat, HeartbeatStore  # noqa: E402
+
+
+def _default_paths():
+    base = os.environ.get("RQ_DATA", str(_HERE))
+    return {
+        "registry": os.path.join(base, "heartbeat.db"),
+        "mission": os.path.join(base, "missions.db"),
+        "safety": os.path.join(base, "safety.db"),
+    }
+
+
+def _make_hb(args) -> Heartbeat:
+    p = _default_paths()
+    orch = None
+    if args.execute:
+        orch = Orchestrator(mission_db=p["mission"], safety_db=p["safety"])
+    return Heartbeat(p["registry"], p["mission"], p["safety"], orchestrator=orch)
+
+
+def _cmd_tick(args):
+    hb = _make_hb(args)
+    res = hb.run_tick(mission_id=args.mission, execute=args.execute)
+    print(json.dumps(res, indent=2, default=str))
+    return 0
+
+
+def _cmd_status(args):
+    reg = HeartbeatStore(_default_paths()["registry"])
+    print(f"kill_switch: {'ON' if reg.kill_switch_on() else 'OFF'}")
+    lt = reg.last_tick()
+    if lt:
+        print("last_tick:", json.dumps(lt, default=str))
+    else:
+        print("last_tick: none")
+    return 0
+
+
+def _cmd_switch(args, value: bool):
+    reg = HeartbeatStore(_default_paths()["registry"])
+    reg.set_kill_switch(value)
+    print(f"kill_switch -> {'ON' if value else 'OFF'}")
+    return 0
+
+
+def _cmd_resume(args):
+    reg = HeartbeatStore(_default_paths()["registry"])
+    reg.set_resume_pending(args.mission, True)
+    print(f"resume_pending -> {args.mission}")
+    return 0
+
+
+def _cmd_approve(args):
+    reg = HeartbeatStore(_default_paths()["registry"])
+    out = reg.approval_grant(args.gate_id, args.by, args.evidence)
+    print(json.dumps(out, default=str))
+    return 0
+
+
+def _cmd_deny(args):
+    reg = HeartbeatStore(_default_paths()["registry"])
+    out = reg.approval_deny(args.gate_id, args.by, args.evidence)
+    print(json.dumps(out, default=str))
+    return 0
+
+
+def _cmd_priority(args):
+    reg = HeartbeatStore(_default_paths()["registry"])
+    reg.set_config(f"priority:{args.mission}", str(args.prio))
+    print(f"priority[{args.mission}] -> {args.prio}")
+    return 0
+
+
+def build_parser():
+    ap = argparse.ArgumentParser(prog="rq-heartbeat", description="A5 Heartbeat CLI")
+    ap.add_argument("--execute", action="store_true",
+                    help="(tick) erlaubt A4-Mutation; ohne ist tick read-only")
+    sub = ap.add_subparsers(dest="cmd", required=True)
+
+    p = sub.add_parser("tick")
+    p.add_argument("--mission", default=None)
+    p.set_defaults(func=_cmd_tick)
+
+    sub.add_parser("status").set_defaults(func=_cmd_status)
+    sub.add_parser("enable").set_defaults(func=lambda a: _cmd_switch(a, True))
+    sub.add_parser("disable").set_defaults(func=lambda a: _cmd_switch(a, False))
+
+    p = sub.add_parser("resume")
+    p.add_argument("--mission", required=True)
+    p.set_defaults(func=_cmd_resume)
+
+    p = sub.add_parser("approve")
+    p.add_argument("--gate", dest="gate_id", required=True)
+    p.add_argument("--by", default="christian")
+    p.add_argument("--evidence", default="manual approval")
+    p.set_defaults(func=_cmd_approve)
+
+    p = sub.add_parser("deny")
+    p.add_argument("--gate", dest="gate_id", required=True)
+    p.add_argument("--by", default="christian")
+    p.add_argument("--evidence", default="manual deny")
+    p.set_defaults(func=_cmd_deny)
+
+    p = sub.add_parser("priority")
+    p.add_argument("--mission", required=True)
+    p.add_argument("--prio", type=int, required=True)
+    p.set_defaults(func=_cmd_priority)
+
+    return ap
+
+
+def main():
+    args = build_parser().parse_args()
+    return args.func(args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

a5/scripts/a5_heartbeat_tick.sh

@@ -0,0 +1,14 @@
+#!/usr/bin/env bash
+# A5 E2E: EIN bounded Heartbeat-Tick (synthetische Mission, isolierte DBs).
+# Wird vom nativen Hermes Cron im no_agent-Modus aufgerufen. Fuehrt KEINEN
+# Dauer-Loop aus — ein Tick, Ergebnis auf stdout, EXIT.
+set -uo pipefail
+
+REPO="/opt/data/forgejo/trading-system-docs"
+export RQ_DATA="/opt/data/a5_e2e"
+cd "$REPO"
+
+out=$(python3 a5/rq_heartbeat_cli.py --execute tick --mission E2E-1 2>&1)
+code=$(echo "$out" | python3 -c 'import sys,json; d=json.load(sys.stdin); print(d.get("code","?")); print("exit_reason:", d.get("exit_reason","")); print("mission:", d.get("mission_id",""))')
+echo "A5-HEARTBEAT-TICK-RESULT $code"
+echo "$code" | tail -3

a5/test_a5.py

@@ -0,0 +1,426 @@
+#!/usr/bin/env python3
+"""Red Queen — A5: deterministische Tests (CONTROLLED HEARTBEAT & RESUME v1).
+
+Lauf:
+    python3 test_a5.py
+Exit-Code 0 = alle Tests gruen; 1 = mindestens ein Fehler.
+Nutzt ausschliesslich temporaere DBs. Kein Netz, kein echter A4-Dispatch (Stub)
+ausser einem echten bounded Orchestrator-Test (ohne dispatcher).
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+_HERE = Path(__file__).resolve().parent
+_REPO = _HERE.parent
+for _p in (str(_REPO), str(_REPO / "a2"), str(_REPO / "a3"), str(_REPO / "a4"), str(_HERE)):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+from a2.rq_mission import MissionStore
+from a3.rq_safety import SafetyStore
+from a4.rq_orchestrator import (  # noqa: E402
+    ST_NEEDS_DISPATCH,
+    ST_APPROVAL_REQUIRED,
+    ST_NO_RUNNABLE_WORK,
+    Orchestrator,
+)
+from a5.rq_heartbeat import (  # noqa: E402
+    HB_APPROVAL_PENDING,
+    HB_APPROVAL_REQUIRED,
+    HB_BOUNDED_RUN_COMPLETE,
+    HB_CANCELLED,
+    HB_CIRCUIT_OPEN,
+    HB_GIT_CONFLICT,
+    HB_KILL_SWITCH_OFF,
+    HB_LOCKED,
+    HB_MISSION_COMPLETED,
+    HB_NO_ACTIVE_MISSION,
+    HB_NO_RUNNABLE_WORK,
+    HB_PAUSED,
+    HB_SAFETY_BLOCKED,
+    HB_TERMINAL,
+    Heartbeat,
+)
+
+PASS = 0
+FAIL = 0
+FAILURES = []
+
+
+def check(name: str, cond: bool, extra: str = ""):
+    global PASS, FAIL
+    if cond:
+        PASS += 1
+        print(f"  [PASS] {name}")
+    else:
+        FAIL += 1
+        FAILURES.append(name)
+        print(f"  [FAIL] {name} {extra}")
+
+
+def _mk_env():
+    d = tempfile.mkdtemp(prefix="a5_test_")
+    return {
+        "dir": d,
+        "registry": os.path.join(d, "heartbeat.db"),
+        "mission": os.path.join(d, "missions.db"),
+        "safety": os.path.join(d, "safety.db"),
+    }
+
+
+class StubOrchestrator:
+    """Deterministischer A4-Stub: liefert genau den injizierten Status."""
+
+    def __init__(self, status=ST_NEEDS_DISPATCH, **extra):
+        self._status = status
+        self._extra = extra
+        self.run_bounded_calls = 0
+
+    def run_bounded(self, mission_id, *, max_steps=None, actor="red-queen"):
+        self.run_bounded_calls += 1
+        return {"status": self._status, "mission_id": mission_id, **self._extra}
+
+    def mission_completion_gate(self, mission_id, *, final_review_pass=False):
+        return {"mission_id": mission_id, "valid": False, "reasons": ["no wps done"]}
+
+    def attempt_mission_complete(self, mission_id, *, final_review_pass=False, actor="red-queen"):
+        return {"status": "MISSION_NOT_COMPLETED", "mission_id": mission_id, "valid": False}
+
+
+def _mission_running(ms: MissionStore, mid: str, wp_ids=None):
+    ms.mission_create(title=f"M {mid}", goal="g", scope="s", mission_id=mid)
+    for c, t in (("CREATED", "PLANNING"), ("PLANNING", "READY"), ("READY", "RUNNING")):
+        ms.mission_transition(mid, t, evidence=f"to {t}")
+    if wp_ids:
+        for w in wp_ids:
+            ms.wp_create(mid, f"WP {w}", wp_id=f"{mid}-{w}")
+            ms.wp_transition(f"{mid}-{w}", "READY")
+
+
+# --------------------------------------------------------------------------- #
+# 1. Kill Switch
+# --------------------------------------------------------------------------- #
+def test_kill_switch_off_blocks_mutation():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-KS")
+    stub = StubOrchestrator(ST_NEEDS_DISPATCH)
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub)
+    check("kill switch default OFF", hb.registry.kill_switch_on() is False)
+    res = hb.run_tick(mission_id="M-KS")
+    check("kill switch OFF -> HB_KILL_SWITCH_OFF", res["code"] == HB_KILL_SWITCH_OFF, res["code"])
+    check("kill switch OFF -> keine A4-Mutation", stub.run_bounded_calls == 0, str(stub.run_bounded_calls))
+
+
+def test_kill_switch_on_allows_bounded():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-KSON")
+    stub = StubOrchestrator(ST_NEEDS_DISPATCH)
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub)
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-KSON")
+    check("kill switch ON -> bounded run", res["code"] == HB_BOUNDED_RUN_COMPLETE, res["code"])
+    check("exactly one A4 call", stub.run_bounded_calls == 1, str(stub.run_bounded_calls))
+
+
+# --------------------------------------------------------------------------- #
+# 2. Tick Lock
+# --------------------------------------------------------------------------- #
+def test_tick_lock_prevents_overlap():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-TL")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    assert hb.registry.acquire_lock("RUN-A") == "ok"
+    res = hb.run_tick(mission_id="M-TL")
+    check("overlapping tick -> HB_LOCKED", res["code"] == HB_LOCKED, res["code"])
+
+
+def test_tick_lock_preserved_on_overlap():
+    """Regressionsschutz: Ein HB_LOCKED-Tick darf das FREMD-aktive Lock NICHT
+    freigeben. Defekt, den der Fresh Checker fand: blindes release_lock im
+    finally loeschte das Lock von Tick A, sodass Tick C sofort wieder mutieren
+    konnte. Jetzt: nur der Owner-Tick gibt sein eigenes Lock frei."""
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-TLP")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    # Tick A haelt das Lock aktiv
+    assert hb.registry.acquire_lock("RUN-A") == "ok"
+    # Tick B (ueberlappend) -> HB_LOCKED; darf A's Lock NICHT loeschen
+    res = hb.run_tick(mission_id="M-TLP")
+    check("overlapping tick -> HB_LOCKED", res["code"] == HB_LOCKED, res["code"])
+    # Das Lock von A MUSS weiterhin bestehen bleiben (Owner A)
+    lock = hb.registry.acquire_lock("RUN-C")
+    check("fremdes Lock nach HB_LOCKED NICHT freigegeben", lock == "locked", lock)
+
+
+def test_tick_lock_release_after_run():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-TL2")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-TL2")
+    check("run completes", res["code"] == HB_BOUNDED_RUN_COMPLETE, res["code"])
+    lock = hb.registry.acquire_lock("RUN-NEXT")
+    check("lock released after run", lock == "ok", lock)
+
+
+# --------------------------------------------------------------------------- #
+# 3. Mission Eligibility
+# --------------------------------------------------------------------------- #
+def test_eligibility_cancelled():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-TERM")
+    ms.mission_transition("M-TERM", "CANCELLED")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-TERM")
+    check("cancelled -> HB_CANCELLED", res["code"] == HB_CANCELLED, res["code"])
+
+
+def test_eligibility_paused_not_resumable():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-PAUSE")
+    ms.mission_transition("M-PAUSE", "PAUSED")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-PAUSE")
+    check("paused (no resume) -> HB_PAUSED", res["code"] == HB_PAUSED, res["code"])
+
+
+def test_eligibility_paused_resumable():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-PR")
+    ms.mission_transition("M-PR", "PAUSED")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    hb.registry.set_resume_pending("M-PR", True)
+    res = hb.run_tick(mission_id="M-PR")
+    check("paused (resumable) -> bounded", res["code"] == HB_BOUNDED_RUN_COMPLETE, res["code"])
+
+
+# --------------------------------------------------------------------------- #
+# 4. Priority (deterministisch)
+# --------------------------------------------------------------------------- #
+def test_priority_selection():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-B")
+    _mission_running(ms, "M-A")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    hb.registry.set_priority("M-A", 10)
+    res = hb.run_tick()
+    check("priority selects M-A", res["mission_id"] == "M-A", str(res["mission_id"]))
+
+
+# --------------------------------------------------------------------------- #
+# 5. Bounded: EXACTLY ONE A4 run
+# --------------------------------------------------------------------------- #
+def test_bounded_single_a4_call():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-BD")
+    stub = StubOrchestrator(ST_NEEDS_DISPATCH)
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub)
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-BD")
+    check("exactly one A4 run", stub.run_bounded_calls == 1, str(stub.run_bounded_calls))
+    check("bounded complete code", res["code"] == HB_BOUNDED_RUN_COMPLETE, res["code"])
+
+
+# --------------------------------------------------------------------------- #
+# 6. NO RUNNABLE WORK
+# --------------------------------------------------------------------------- #
+def test_no_runnable_work():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-NR")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NO_RUNNABLE_WORK))
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-NR")
+    check("no runnable -> HB_NO_RUNNABLE_WORK", res["code"] == HB_NO_RUNNABLE_WORK, res["code"])
+
+
+# --------------------------------------------------------------------------- #
+# 7. Approval
+# --------------------------------------------------------------------------- #
+def test_approval_required_blocks_retry():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-APP")
+    stub = StubOrchestrator(ST_APPROVAL_REQUIRED, wp_id="M-APP-WP1", risk="CRITICAL",
+                            approval={"WP": "M-APP-WP1", "REQUESTED_ACTION": "a"})
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub)
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-APP")
+    check("approval required", res["code"] == HB_APPROVAL_REQUIRED, res["code"])
+    check("gate persisted", hb.registry.approval_pending("M-APP") is not None)
+    # naechster Tick darf NICHT erneut dispatchen
+    stub2 = StubOrchestrator(ST_NEEDS_DISPATCH)
+    hb2 = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub2)
+    hb2.registry.set_kill_switch(True)
+    res2 = hb2.run_tick(mission_id="M-APP")
+    check("approval pending blocks re-dispatch", res2["code"] == HB_APPROVAL_PENDING, res2["code"])
+    check("no re-dispatch A4 call", stub2.run_bounded_calls == 0, str(stub2.run_bounded_calls))
+
+
+def test_approval_grant_resumes():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-AGR")
+    stub = StubOrchestrator(ST_APPROVAL_REQUIRED, wp_id="M-AGR-WP1", risk="CRITICAL",
+                            approval={"WP": "M-AGR-WP1", "REQUESTED_ACTION": "a"})
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub)
+    hb.registry.set_kill_switch(True)
+    hb.run_tick(mission_id="M-AGR")
+    gate = hb.registry.approval_pending("M-AGR")
+    check("gate exists", gate is not None)
+    hb.registry.approval_grant(gate["gate_id"], "christian", "approved")
+    check("gate cleared after grant", hb.registry.approval_pending("M-AGR") is None)
+    stub2 = StubOrchestrator(ST_NEEDS_DISPATCH)
+    hb2 = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub2)
+    hb2.registry.set_kill_switch(True)
+    res = hb2.run_tick(mission_id="M-AGR")
+    check("resume after approval", res["code"] == HB_BOUNDED_RUN_COMPLETE, res["code"])
+
+
+# --------------------------------------------------------------------------- #
+# 8. Circuit
+# --------------------------------------------------------------------------- #
+def test_circuit_open_blocks():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-CIR")
+    ss = SafetyStore(env["safety"])
+    ss.open_circuit("MISSION", "M-CIR", trigger="test circuit", reason="test")
+    stub = StubOrchestrator(ST_NEEDS_DISPATCH)
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=stub)
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-CIR")
+    check("circuit open -> HB_CIRCUIT_OPEN", res["code"] == HB_CIRCUIT_OPEN, res["code"])
+    check("circuit open -> kein A4", stub.run_bounded_calls == 0, str(stub.run_bounded_calls))
+
+
+# --------------------------------------------------------------------------- #
+# 9. Git Conflict
+# --------------------------------------------------------------------------- #
+def test_git_conflict_blocks_resume():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-GIT")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH),
+                   git_preflight=lambda mid: {"ok": False, "reason": "uncommitted"})
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-GIT")
+    check("git conflict -> HB_GIT_CONFLICT", res["code"] == HB_GIT_CONFLICT, res["code"])
+
+
+# --------------------------------------------------------------------------- #
+# 10. Unknown mission / state conflict
+# --------------------------------------------------------------------------- #
+def test_unknown_mission():
+    env = _mk_env()
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-UNKNOWN")
+    check("unknown mission -> HB_NO_ACTIVE_MISSION", res["code"] == HB_NO_ACTIVE_MISSION, res["code"])
+
+
+# --------------------------------------------------------------------------- #
+# 11. Real bounded orchestrator (kein dispatcher)
+# --------------------------------------------------------------------------- #
+def test_real_orchestrator_bounded_dispatch_contract():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-REAL")
+    a4 = Orchestrator(mission_db=env["mission"], safety_db=env["safety"])
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"], orchestrator=a4)
+    hb.registry.set_kill_switch(True)
+    res = hb.run_tick(mission_id="M-REAL", execute=True)
+    check("real orchestrator bounded -> complete|no_runnable",
+          res["code"] in (HB_BOUNDED_RUN_COMPLETE, HB_NO_RUNNABLE_WORK), res["code"])
+
+
+# --------------------------------------------------------------------------- #
+# 12. Health (read-only, Kill-Switch OFF erlaubt)
+# --------------------------------------------------------------------------- #
+def test_health_readonly():
+    env = _mk_env()
+    ms = MissionStore(env["mission"])
+    _mission_running(ms, "M-HEALTH")
+    hb = Heartbeat(env["registry"], env["mission"], env["safety"],
+                   orchestrator=StubOrchestrator(ST_NEEDS_DISPATCH))
+    h = hb.health()
+    check("health kill switch OFF", h["kill_switch"] == "OFF")
+    check("health lists mission", any(m["id"] == "M-HEALTH" for m in h["missions"]))
+
+
+# --------------------------------------------------------------------------- #
+# Test Runner
+# --------------------------------------------------------------------------- #
+ALL_TESTS = [
+    test_kill_switch_off_blocks_mutation,
+    test_kill_switch_on_allows_bounded,
+    test_tick_lock_prevents_overlap,
+    test_tick_lock_preserved_on_overlap,
+    test_tick_lock_release_after_run,
+    test_eligibility_cancelled,
+    test_eligibility_paused_not_resumable,
+    test_eligibility_paused_resumable,
+    test_priority_selection,
+    test_bounded_single_a4_call,
+    test_no_runnable_work,
+    test_approval_required_blocks_retry,
+    test_approval_grant_resumes,
+    test_circuit_open_blocks,
+    test_git_conflict_blocks_resume,
+    test_unknown_mission,
+    test_real_orchestrator_bounded_dispatch_contract,
+    test_health_readonly,
+]
+
+
+def main():
+    for fn in ALL_TESTS:
+        try:
+            fn()
+        except Exception as e:  # noqa
+            global FAIL
+            FAIL += 1
+            FAILURES.append(fn.__name__)
+            print(f"  [ERROR] {fn.__name__}: {type(e).__name__}: {e}")
+    print("\n" + "=" * 60)
+    print(f"PASS={PASS} FAIL={FAIL}")
+    if FAIL:
+        print("FAILURES:", FAILURES)
+        return 1
+    print("ALL TESTS PASSED")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())