diff --git a/a2/rq_mission.py b/a2/rq_mission.py index ae3a1af..ca02b21 100644 --- a/a2/rq_mission.py +++ b/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: diff --git a/a5/DESIGN.md b/a5/DESIGN.md new file mode 100644 index 0000000..524bde9 --- /dev/null +++ b/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). diff --git a/a5/README.md b/a5/README.md new file mode 100644 index 0000000..62b162d --- /dev/null +++ b/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. diff --git a/a5/rq_heartbeat.py b/a5/rq_heartbeat.py new file mode 100644 index 0000000..7b119d2 --- /dev/null +++ b/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" diff --git a/a5/rq_heartbeat_cli.py b/a5/rq_heartbeat_cli.py new file mode 100644 index 0000000..aab5c2a --- /dev/null +++ b/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()) diff --git a/a5/scripts/a5_heartbeat_tick.sh b/a5/scripts/a5_heartbeat_tick.sh new file mode 100755 index 0000000..c82567b --- /dev/null +++ b/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 diff --git a/a5/test_a5.py b/a5/test_a5.py new file mode 100644 index 0000000..b4a6ac3 --- /dev/null +++ b/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())