feat(a3): deterministic safety layer v1 (attempt ledger, retry, circuit breaker)

a3/.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+*.pyc
+*.db
+*.db-wal
+*.db-shm
+*.tmp

a3/README.md

@@ -0,0 +1,145 @@
+# Red Queen — A3: Deterministic Safety Layer (V1)
+
+Der **Deterministic Safety Layer** sitzt zwischen Mission/WP-State (A2 `missions.db`)
+und dem **späteren** Orchestrator/Action/Retry/Delegation. Er entscheidet
+**deterministisch** (Zähler, Limits, Tabellen — nicht „LLM-Lust"):
+
+```
+CONTINUE / RETRY / DEBUG / SECOND_OPINION / BLOCK / ESCALATE / CIRCUIT_BREAK
+```
+
+Ziel: Red Queen erhält die Safety-Logik, **BEVOR** spätere autonome Mission-Loops
+existieren. Dieser Build enthält **KEINE** autonome Orchestrierung — kein dispatcher,
+kein loop, kein heartbeat, kein cron, kein self-improvement. A3 ist eine reine
+**Library / Safety-Capability**.
+
+---
+
+## DB-Entscheidung (Maker-Entscheidung, §29)
+
+A3 nutzt eine **eigene `safety.db`** (separate SQLite-Datei), **nicht** die A2
+`missions.db`:
+
+- **Rollback-sicher für A2:** Die bestehende A2-Datenbank wird überhaupt nicht
+  angefasst — kein ALTER, kein neues Schema in `missions.db`, keine Gefahr für
+  Bestandsdaten (`CREATE TABLE IF NOT EXISTS` auf den A3-Tabellen in `safety.db`).
+- **Entkopplung:** Safety-State (Circuit, Attempts, Events, Evidence) ist unabhängig
+  vom Mission-State. Ein beschädigter Mission-State kann den Safety-Layer nicht
+  mitreißen und umgekehrt.
+- **Testbar:** A3-DB wird immer über temp-Pfade getestet, nie produktiv berührt.
+
+Die A2-Integration erfolgt sauber: `SafetyStore` verweist Missions-/WP-IDs
+als Fremdschlüssel-Namen (mission_id/wp_id) und ruft A2-Regeln nicht auf; A2-APIs
+`mission_block`/`mission_transition` bleiben unberührt. Mission-`BLOCKED` kann in
+A4 durch den Orchestrator auf Basis einer `BLOCK`-Safety-Entscheidung gesetzt werden —
+A3 selbst mutiert A2 nicht direkt.
+
+---
+
+## Module
+
+| Datei | Inhalt |
+|-------|--------|
+| `rq_safety.py` | Kernmodul: `SafetyStore`, `SafetyError`, Error-Signature, Strategy-Fingerprint, Redaction, Circuit Breaker, Retry Controller, Oscillation, Fail-Closed, Safety Events, Evidence, `evaluate_next_action` (§21) |
+| `rq_safety_telegram.py` | Telegram-Notification-Interface (§19): `format_alert`, `build_payload`, `should_notify` (Anti-Spam). KEIN Daemon. |
+| `rq_safety_cli.py` | Dünne, deterministische CLI (`--json`, Exit-Code 2 bei `SafetyError`), A2-CLI-Muster spiegelnd |
+| `test_a3.py` | Isolierte Testsuite (temp-DB), Exit-Code 0 = PASS |
+
+---
+
+## Kern-API (`SafetyStore`)
+
+```python
+from rq_safety import SafetyStore
+store = SafetyStore("path/to/safety.db")
+
+# Attempt Ledger (idempotent via idempotency_key, append-only, redacted)
+store.record_attempt("M1", "W1", actor="maker", result="FAIL",
+                     error="boom pid=123", strategy="s1", idempotency_key="k-1")
+
+# Retry / Oscillation / Circuit / Fail-closed
+store.evaluate_next_action("M1", "W1", error="boom", strategy_label="s2",
+                           is_mutating=True)   # -> DECISION/REASON_CODE/ALLOWED_ACTION
+
+# Circuit Breaker
+store.open_circuit("MISSION", "M1", trigger="REG", severity="HIGH")
+store.circuit_state("MISSION", "M1")
+store.request_circuit_reset("MISSION", "M1", cause="...", recovery_evidence="...")
+store.close_circuit("MISSION", "M1", approved_by="human", gate="human_gate", cause="...", recovery_evidence="...")
+
+# Events / Evidence
+store.safety_event("OSCILLATION_DETECTED", severity="CRITICAL", mission_id="M1")
+store.safety_evidence("test_results", {"failing": 2})
+store.evidence()
+```
+
+---
+
+## Entscheidungen (Reason Codes, §22)
+
+| Code | Bedeutung |
+|------|-----------|
+| `RETRY_AVAILABLE` | nächste Aktion erlaubt |
+| `RETRY_LIMIT` | Maker/Checker-Repair MAX 3 erreicht → SECOND_OPINION |
+| `SAME_ERROR_LIMIT` | gleiche Error-Signatur MAX 2 → DEBUG/Strategiewechsel |
+| `FAILED_STRATEGY_REPEAT` | bereits gescheiterte Strategie → keine blinde Wiederholung |
+| `NO_MEASURABLE_PROGRESS` | FAIL ohne messbaren Fortschritt (UNKNOWN ≠ Progress) |
+| `OSCILLATION_ABAB` | A-B-A-B-Muster → Circuit-Breaker-Kandidat |
+| `CIRCUIT_ALREADY_OPEN` | Circuit OPEN → BLOCK (read-only erlaubt) |
+| `STATE_INCONSISTENT` | Fail-Closed: Safety-State korrupt → keine Mutation |
+| `CRITICAL_TRIGGER` | kritischer GLOBAL-Trigger |
+| `HUMAN_GATE_REQUIRED` | Circuit-Reset braucht Human Gate bei HIGH/CRITICAL/GLOBAL |
+
+---
+
+## Limite (A1 SAFETY_CONTRACT, konservativ V1)
+
+- Maker→Checker-Repair: **MAX 3**
+- Gleiche Error-Signatur: **MAX 2**
+- Gleiche bereits gescheiterte Strategie: **keine blinde Wiederholung**
+- Oscillation A-B-A-B: Circuit-Breaker-Kandidat
+- `MAX_ITERATIONS=50`: äußerste Runtime-Notbremse, **nicht** operatives Retry-Limit
+
+---
+
+## Fail-Closed (§16)
+
+Bei unbekanntem/inkonsistentem Safety-State → `SAFETY_STATE_ERROR` → **STOP** →
+**Evidence sichern** → **KEINE Mutation** (read-only Diagnose ggf. erlaubt).
+
+---
+
+## Secret-Safety (§28)
+
+- Alle credential-artigen Werte werden **vor** Persistenz UND vor Fingerprint/Signatur-
+  Ableitung **redacted** (`redact_secret`).
+- Keine Tokens/Passwörter/private Keys/Authorization-Header in `safety.db`/Events/Telegram.
+- Secret Exposure wird nur als `SECRET_EXPOSURE_DETECTED` + LOCATION/TYPE + `VALUE=REDACTED` erfasst.
+
+---
+
+## Telegram Interface (§19)
+
+Kein Daemon. Nur Formatter/Payload. **Kein Spam** — nur signifikante Events
+(`CIRCUIT_OPENED`, `CRITICAL`, `ESCALATION_REQUIRED`, `HUMAN_DECISION_REQUIRED`,
+`RETRY_LIMIT_REACHED`, `OSCILLATION_DETECTED`, `SAFETY_STATE_ERROR`). Payload ist
+deterministisch und **redacted**.
+
+```python
+from rq_safety_telegram import build_payload
+payload = build_payload("CIRCUIT_OPENED", "CRITICAL", reason="...", mission_id="M")
+# payload["notify"] == True, payload["text"] fertig formatiert
+```
+
+---
+
+## Tests
+
+```bash
+python3 test_a3.py     # Exit 0 = PASS; isoliert (temp-DB), produktive DBs unberührt
+```
+
+Abgedeckt: Attempt Ledger + Idempotenz, Error-Signature-Normalisierung, Strategy-
+Fingerprint, Retry-Limits, Progress, Oscillation A-B-A-B, False-Positives,
+Circuit-Breaker (+Restart-Persistenz + Negativ-Test), Fail-Closed, Events,
+Evidence, Secret-Safety, Telegram-Interface, Loop-Simulation (A–E), A2-DB-Isolation.

a3/rq_safety_cli.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""
+Red Queen — A3 CLI. Dünner, deterministischer Kommandozeilen-Zugriff auf den
+Deterministic Safety Layer (`SafetyStore`). Maschinenlesbar via `--json`.
+
+Hinweis: Der A3-Build definiert KEINE autonome Orchestrierung — dieses CLI ruft
+nur die vom Store bereitgestellten atomaren, deterministischen Operationen auf.
+Es gibt KEINEN Loop, kein Retry-Ausfuehrung, kein Daemon.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+
+from rq_safety import SafetyStore, SafetyError
+
+
+def _store(args) -> SafetyStore:
+    db = args.db or os.environ.get("RQ_SAFETY_DB") or "safety.db"
+    return SafetyStore(db)
+
+
+def _emit(args, obj) -> None:
+    if args.json:
+        print(json.dumps(obj, indent=2, ensure_ascii=False, default=str))
+    else:
+        if isinstance(obj, dict) and obj.get("attempt_id"):
+            print(obj["attempt_id"])
+        else:
+            print(json.dumps(obj, ensure_ascii=False, default=str))
+
+
+def _fail(e: SafetyError, args) -> None:
+    if args.json:
+        print(json.dumps(e.to_dict(), indent=2))
+    else:
+        print(f"ERROR [{e.code}]: {e.message}")
+    sys.exit(2)
+
+
+def build_parser() -> argparse.ArgumentParser:
+    p = argparse.ArgumentParser(prog="rq_safety", description="Red Queen A3 Deterministic Safety CLI")
+    p.add_argument("--db", help="Pfad zur safety.db (default: $RQ_SAFETY_DB oder safety.db)")
+    p.add_argument("--json", action="store_true", help="JSON-Ausgabe")
+    sub = p.add_subparsers(dest="cmd", required=True)
+
+    # attempt
+    at = sub.add_parser("attempt_record")
+    at.add_argument("--mission", dest="mission_id", required=True)
+    at.add_argument("--wp", dest="wp_id")
+    at.add_argument("--actor", default="red-queen")
+    at.add_argument("--result", default="UNKNOWN")
+    at.add_argument("--error")
+    at.add_argument("--strategy")
+    at.add_argument("--target-component")
+    at.add_argument("--progress", default="UNKNOWN")
+    at.add_argument("--idem-key", dest="idem_key")
+    at.add_argument("--change")
+
+    al = sub.add_parser("attempts")
+    al.add_argument("--mission", dest="mission_id")
+    al.add_argument("--wp", dest="wp_id")
+
+    # circuit
+    co = sub.add_parser("circuit_open")
+    co.add_argument("--scope-type", required=True)
+    co.add_argument("--scope-id", required=True)
+    co.add_argument("--trigger", required=True)
+    co.add_argument("--severity", default="HIGH")
+    co.add_argument("--reason")
+    co.add_argument("--mission", dest="mission_id")
+    co.add_argument("--wp", dest="wp_id")
+
+    cs = sub.add_parser("circuit_state")
+    cs.add_argument("--scope-type", required=True)
+    cs.add_argument("--scope-id", required=True)
+
+    cr = sub.add_parser("circuit_reset_request")
+    cr.add_argument("--scope-type", required=True)
+    cr.add_argument("--scope-id", required=True)
+    cr.add_argument("--cause", required=True)
+    cr.add_argument("--recovery-evidence", required=True)
+
+    cc = sub.add_parser("circuit_close")
+    cc.add_argument("--scope-type", required=True)
+    cc.add_argument("--scope-id", required=True)
+    cc.add_argument("--approved-by", required=True)
+    cc.add_argument("--gate", default="documented_recovery")
+    cc.add_argument("--cause", required=True)
+    cc.add_argument("--recovery-evidence", required=True)
+
+    # events
+    ev = sub.add_parser("event")
+    ev.add_argument("--type", required=True)
+    ev.add_argument("--severity", default="WARNING")
+    ev.add_argument("--reason")
+    ev.add_argument("--reason-code")
+    ev.add_argument("--mission", dest="mission_id")
+    ev.add_argument("--wp", dest="wp_id")
+    ev.add_argument("--scope-type")
+    ev.add_argument("--scope-id")
+
+    evl = sub.add_parser("events")
+    evl.add_argument("--mission", dest="mission_id")
+    evl.add_argument("--type")
+
+    # evidence
+    evi = sub.add_parser("evidence_save")
+    evi.add_argument("--kind", required=True)
+    evi.add_argument("--data", required=True, help="JSON-Dict")
+
+    evir = sub.add_parser("evidence_list")
+    evir.add_argument("--ref")
+
+    # decision
+    dec = sub.add_parser("evaluate")
+    dec.add_argument("--mission", dest="mission_id", required=True)
+    dec.add_argument("--wp", dest="wp_id")
+    dec.add_argument("--error")
+    dec.add_argument("--strategy")
+    dec.add_argument("--target-component")
+    dec.add_argument("--read-only", action="store_true", help="nicht-mutierende Operation")
+
+    sub.add_parser("check_safety_state")
+    return p
+
+
+def main(argv=None) -> int:
+    args = build_parser().parse_args(argv)
+    try:
+        s = _store(args)
+        if args.cmd == "attempt_record":
+            r = s.record_attempt(
+                args.mission_id, wp_id=args.wp_id, actor=args.actor, result=args.result,
+                error=args.error, strategy=args.strategy,
+                target_component=args.target_component, progress=args.progress,
+                idempotency_key=args.idem_key, change=args.change,
+            )
+        elif args.cmd == "attempts":
+            r = s.attempts(args.mission_id, args.wp_id)
+        elif args.cmd == "circuit_open":
+            r = s.open_circuit(args.scope_type, args.scope_id, trigger=args.trigger,
+                               severity=args.severity, reason=args.reason,
+                               mission_id=args.mission_id, wp_id=args.wp_id)
+        elif args.cmd == "circuit_state":
+            r = s.circuit_state(args.scope_type, args.scope_id)
+        elif args.cmd == "circuit_reset_request":
+            r = s.request_circuit_reset(args.scope_type, args.scope_id, cause=args.cause,
+                                        recovery_evidence=args.recovery_evidence)
+        elif args.cmd == "circuit_close":
+            r = s.close_circuit(args.scope_type, args.scope_id, approved_by=args.approved_by,
+                                gate=args.gate, cause=args.cause,
+                                recovery_evidence=args.recovery_evidence)
+        elif args.cmd == "event":
+            r = s.safety_event(args.type, severity=args.severity, reason=args.reason,
+                               reason_code=args.reason_code, mission_id=args.mission_id,
+                               wp_id=args.wp_id, scope_type=args.scope_type,
+                               scope_id=args.scope_id)
+        elif args.cmd == "events":
+            r = s.safety_events(args.mission_id, args.type)
+        elif args.cmd == "evidence_save":
+            r = s.safety_evidence(args.kind, json.loads(args.data))
+        elif args.cmd == "evidence_list":
+            r = s.evidence(args.ref)
+        elif args.cmd == "evaluate":
+            r = s.evaluate_next_action(args.mission_id, args.wp_id, error=args.error,
+                                       strategy_label=args.strategy,
+                                       target_component=args.target_component,
+                                       is_mutating=not args.read_only)
+        elif args.cmd == "check_safety_state":
+            r = s.check_safety_state()
+        else:
+            _emit(args, {"error": f"unknown command {args.command}"})
+            return 1
+        _emit(args, r)
+        return 0
+    except SafetyError as e:
+        _fail(e, args)
+    except Exception as e:  # noqa
+        _fail(SafetyError("INTERNAL_ERROR", str(e)), args)
+    return 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

a3/rq_safety_telegram.py

@@ -0,0 +1,134 @@
+#!/usr/bin/env python3
+"""
+Red Queen — A3: Telegram Safety Notification Interface (§19).
+
+KEIN Daemon, KEIN autonomer Sender. Dieses Modul liefert nur einen deterministischen
+Formatter + Payload-Bausteine fuer relevante Safety-Events. Der spaetere Red Queen
+Lead wuerde diese Payloads ueber einen existierenden Telegram-Kanal senden.
+
+Anti-Spam-Regeln (TELEGRAM_MISSION_CONTROL):
+  * NUR signifikante Events: CIRCUIT_OPENED, CRITICAL-Severity, ESCALATION_REQUIRED,
+    HUMAN_DECISION_REQUIRED, RETRY_LIMIT_REACHED, OSCILLATION_DETECTED.
+  * Nicht jeder Retry braucht Telegram. RETRY_ALLOWED/DEBUG/info-Events -> kein Alert.
+
+Payload ist deterministisch und grundsaetzlich REDACTED (keine Secrets, §28).
+
+Test-Konvention: Tests pruefen Formatter-/Payload-Ausgabe (deterministisch), senden
+aber NICHT real.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict
+
+from rq_safety import (
+    EVENT_CIRCUIT_OPENED,
+    EVENT_CIRCUIT_RESET_REQUESTED,
+    EVENT_ESCALATION_REQUIRED,
+    EVENT_HUMAN_DECISION_REQUIRED,
+    EVENT_RETRY_LIMIT_REACHED,
+    EVENT_OSCILLATION_DETECTED,
+    EVENT_SAFETY_STATE_ERROR,
+    redact_secret,
+)
+
+# Meldungs-Praefixe (TELEGRAM_MISSION_CONTROL §5)
+PREFIX_CRITICAL = "[CRITICAL]"
+PREFIX_BLOCKER = "[BLOCKER]"
+PREFIX_DECISION = "[DECISION REQUIRED]"
+
+# Events, die einen Telegram-Alert ausloesen (signifikant, kein Spam).
+ALERT_TYPES = frozenset(
+    {
+        EVENT_CIRCUIT_OPENED,
+        EVENT_CIRCUIT_RESET_REQUESTED,
+        EVENT_ESCALATION_REQUIRED,
+        EVENT_HUMAN_DECISION_REQUIRED,
+        EVENT_RETRY_LIMIT_REACHED,
+        EVENT_OSCILLATION_DETECTED,
+        EVENT_SAFETY_STATE_ERROR,
+    }
+)
+
+
+def should_notify(event_type: str, severity: str) -> bool:
+    """Deterministische Spam-Regel: nur signifikante Events melden.
+
+    CRITICAL-Severity wird immer gemeldet; ansonsten nur in ALERT_TYPES gelistete
+    Event-Typen. RETRY_ALLOWED / INFO / gewoehnliche Events -> kein Alert.
+    """
+    if severity == "CRITICAL":
+        return True
+    return event_type in ALERT_TYPES
+
+
+def format_alert(
+    event_type: str,
+    severity: str,
+    *,
+    mission_id: Any = None,
+    wp_id: Any = None,
+    scope_type: Any = None,
+    scope_id: Any = None,
+    reason: str = "",
+    reason_code: str = "",
+) -> str:
+    """Baue eine deterministische, maschinenlesbare Telegram-Nachricht.
+
+    Ohne Secrets (redacted). Gibt einen menschenlesbaren Block zurueck, der mit
+    einem festen Praefix beginnt. Kein Alert bei nicht-signifikanten Events -> "".
+    """
+    if not should_notify(event_type, severity):
+        return ""  # kein Alert -> keine Nachricht (Anti-Spam)
+
+    prefix = PREFIX_CRITICAL if severity == "CRITICAL" else (
+        PREFIX_DECISION
+        if event_type in (EVENT_HUMAN_DECISION_REQUIRED, EVENT_ESCALATION_REQUIRED)
+        else PREFIX_BLOCKER
+    )
+    scope = f"scope={scope_type}/{scope_id or '-'}" if scope_type else "scope=n/a"
+    lines = [
+        f"{prefix} Red Queen Safety",
+        f"EVENT={event_type} SEVERITY={severity}",
+        f"REASON_CODE={reason_code or 'n/a'}",
+        scope,
+        f"mission={mission_id or 'n/a'}" + (f" wp={wp_id}" if wp_id else ""),
+        f"reason={redact_secret(reason) or 'n/a'}",
+    ]
+    return "\n".join(lines)
+
+
+def build_payload(
+    event_type: str,
+    severity: str,
+    *,
+    reason: str = "",
+    reason_code: str = "",
+    scope_type: Any = None,
+    scope_id: Any = None,
+    mission_id: Any = None,
+    wp_id: Any = None,
+    evidence_ref: Any = None,
+) -> Dict[str, Any]:
+    """Strukturierte Payload fuer einen Telegram-Sender (deterministisch, redacted).
+
+    Return-Format ist stabil; ein spaeterer Sender muss nur diesen Dict-Payload
+    an den konfigurierten Messenger senden.
+    """
+    return {
+        "notify": should_notify(event_type, severity),
+        "text": format_alert(
+            event_type, severity, reason=reason, reason_code=reason_code,
+            scope_type=scope_type, scope_id=scope_id,
+            mission_id=mission_id, wp_id=wp_id,
+        ),
+        "event_type": event_type,
+        "severity": severity,
+        "reason_code": reason_code,
+        "reason": redact_secret(reason),
+        "scope_type": scope_type,
+        "scope_id": scope_id,
+        "mission_id": mission_id,
+        "wp_id": wp_id,
+        "evidence_ref": evidence_ref,
+    }

a3/test_a3.py

@@ -0,0 +1,423 @@
+#!/usr/bin/env python3
+"""
+Red Queen — A3 Testsuite (deterministisch, isoliert).
+
+Nutzt ausschliesslich temporaere `safety.db` (tempfile.mkdtemp). Produktive DBs
+werden NIE angefasst. Telegram-Interface wird nur als Formatter-/Payload-Test
+geprueft — es wird NICHTS real gesendet.
+
+Lauf:
+    python3 test_a3.py
+Exit-Code 0 = alle Tests gruen; 1 = mindestens ein Fehler.
+
+Deckt A3 §24-§33 ab: Attempt Ledger, Retry Controller, Error Signature, Strategy
+Fingerprint, Progress, Oscillation, Circuit Breaker (+Restart-Persistenz +
+Negative-Test), Fail-Closed, Safety Events, Evidence, Telegram-Interface,
+Loop-Simulation (A-E), False-Positive-Tests, Secret-Safety, Idempotenz.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+from pathlib import Path
+
+_HERE = Path(__file__).resolve().parent
+sys.path.insert(0, str(_HERE))
+
+import rq_safety as s  # noqa: E402
+import rq_safety_telegram as tg  # noqa: E402
+
+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 expect_err(name: str, fn, code: str, fragment: str = ""):
+    try:
+        fn()
+    except s.SafetyError as e:
+        ok = e.code == code and (not fragment or fragment in e.message)
+        check(name, ok, f"got code={e.code} msg={e.message!r}")
+        return e
+    except Exception as e:  # noqa
+        check(name, False, f"unexpected {type(e).__name__}: {e}")
+        return None
+    check(name, False, "no error raised")
+    return None
+
+
+def fresh_store():
+    d = tempfile.mkdtemp(prefix="a3test_")
+    return s.SafetyStore(str(Path(d) / "safety.db")), d
+
+
+# --------------------------------------------------------------------------- #
+# Tests
+# --------------------------------------------------------------------------- #
+def test_attempt_ledger_append_and_idempotent():
+    print("\n== attempt ledger == ")
+    st, _ = fresh_store()
+    a1 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom", strategy="s1")
+    a2 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom", strategy="s1")
+    check("distinct attempt ids", a1["attempt_id"] != a2["attempt_id"])
+    check("attempt count 2", st.attempts_count("M1") == 2)
+    # Idempotenz via idempotency_key
+    a3 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom",
+                           strategy="s1", idempotency_key="k-1")
+    a4 = st.record_attempt("M1", "W1", actor="maker", result="FAIL", error="boom",
+                           strategy="s1", idempotency_key="k-1")
+    check("idempotent attempt not double-counted", a3["attempt_id"] == a4["attempt_id"])
+    check("idempotent flag", a4["idempotent"] is True, a4)
+    check("count unchanged by dedup", st.attempts_count("M1") == 3)
+    # Append-only: keine Loesch-API vorhanden
+    check("no delete API", not hasattr(st, "attempt_delete"))
+
+
+def test_error_signature_normalization():
+    print("\n== error signature normalization ==")
+    n1, h1 = s.error_signature("Timeout connecting pid=999 port=8080 0x7f3ab12c")
+    n2, h2 = s.error_signature("Timeout connecting pid=100 port=9090 0x0000dead")
+    check("volatile parts normalized equal", n1 == n2, (n1, n2))
+    check("hash equal for same normalized", h1 == h2)
+    n3, _ = s.error_signature("DIFFERENT_ERROR keyword")
+    check("distinct errors differ", n1 != n3)
+    check("hash deterministic", s.signature_hash(n1) == s.signature_hash(n1))
+
+
+def test_strategy_fingerprint_deterministic():
+    print("\n== strategy fingerprint ==")
+    f1, h1 = s.strategy_fingerprint("maker", "api", ["a.py", "b.py"], "edit", "fix")
+    f2, h2 = s.strategy_fingerprint("maker", "api", ["b.py", "a.py"], "edit", "fix")
+    check("file order irrelevant", h1 == h2, (h1, h2))
+    f3, h3 = s.strategy_fingerprint("maker", "api", ["c.py"], "edit", "fix")
+    check("different target differs", h1 != h3)
+
+
+def test_retry_controller_same_error_limit():
+    print("\n== retry: same error signature MAX 2 ==")
+    st, _ = fresh_store()
+    st.record_attempt("M", "W", actor="maker", result="FAIL", error="errX", strategy="s1")
+    r1 = st.evaluate_next_action("M", "W", error="errX")
+    check("first retry allowed", r1["DECISION"] == "RETRY", r1)
+    st.record_attempt("M", "W", actor="maker", result="FAIL", error="errX", strategy="s2")
+    r2 = st.evaluate_next_action("M", "W", error="errX")
+    check("same error -> DEBUG/SAME_ERROR_LIMIT", r2["REASON_CODE"] == "SAME_ERROR_LIMIT", r2)
+
+
+def test_retry_controller_maker_repair_limit():
+    print("\n== retry: maker/checker repair MAX 3 ==")
+    st, _ = fresh_store()
+    for i in range(3):
+        st.record_attempt("M", "W", actor="maker", result="FAIL", error=f"unique{i}", strategy=f"s{i}")
+    r = st.evaluate_next_action("M", "W", error="unique999", strategy_label="s9")
+    check("3 repairs -> SECOND_OPINION/RETRY_LIMIT", r["REASON_CODE"] == "RETRY_LIMIT", r)
+    check("decision SECOND_OPINION", r["DECISION"] == "SECOND_OPINION", r)
+
+
+def test_failed_strategy_repeat_rejected():
+    print("\n== retry: failed strategy repeat forbidden ==")
+    st, _ = fresh_store()
+    st.record_attempt("M", "W", actor="maker", result="FAIL", error="e1", strategy="stratA",
+                      target_component="api", target_files=["x.py"])
+    r = st.evaluate_next_action("M", "W", error="e2", strategy_label="stratA",
+                                target_component="api", target_files=["x.py"])
+    check("same failed strategy rejected", r["REASON_CODE"] == "FAILED_STRATEGY_REPEAT", r)
+
+
+def test_no_progress_unknown_not_progress():
+    print("\n== no progress: UNKNOWN is not progress ==")
+    st, _ = fresh_store()
+    # 2 FAIL mit UNKNOWN-Progress -> NO_MEASURABLE_PROGRESS (kein messbarer Fortschritt)
+    st.record_attempt("M", "W", actor="maker", result="FAIL", error="eA", strategy="sA", progress="UNKNOWN")
+    st.record_attempt("M", "W", actor="maker", result="FAIL", error="eB", strategy="sB", progress="NO")
+    r = st.evaluate_next_action("M", "W", error="eC")
+    check("UNKNOWN != progress -> NO_MEASURABLE_PROGRESS", r["REASON_CODE"] == "NO_MEASURABLE_PROGRESS", r)
+
+
+def test_oscillation_abab():
+    print("\n== oscillation A-B-A-B -> circuit break ==")
+    st, _ = fresh_store()
+    for stn in ["A", "B", "A", "B"]:
+        st.record_attempt("M", "W", actor="maker", result="FAIL", error="e" + stn, strategy=stn)
+    r = st.evaluate_next_action("M", "W", error="eB", strategy_label="B")
+    check("ABAB detected", r["REASON_CODE"] == "OSCILLATION_ABAB", r)
+    check("decision CIRCUIT_BREAK", r["DECISION"] == "CIRCUIT_BREAK", r)
+
+
+def test_false_positive_not_oscillation():
+    print("\n== false positives: not dangerous oscillation ==")
+    # a) 2 verschiedene Fehler, gleiche Datei aber Testfortschritt -> KEIN Oscillation
+    st, _ = fresh_store()
+    for i in range(4):
+        st.record_attempt("M", "W", actor="maker", result="FAIL", error=f"err{i}",
+                          strategy=f"s{i}", target_component="t1", progress="YES",
+                          progress_metric="tests", progress_before=f"{i}", progress_after=f"{i+1}")
+    r = st.evaluate_next_action("M", "W", error="err3", strategy_label="s3", target_component="t1")
+    check("different errors + progress -> not oscillation", r["REASON_CODE"] != "OSCILLATION_ABAB", r)
+    # b) identische harmlose Read-Only-Diagnose -> nicht blockiert
+    st2, _ = fresh_store()
+    st2.record_attempt("M", "W", actor="diagnoser", result="PASS", error=None, strategy="readonly")
+    r2 = st2.evaluate_next_action("M", "W", is_mutating=False)
+    check("read-only allowed", r2["DECISION"] == "CONTINUE", r2)
+    # c) wiederholter PASS-Test -> nicht als Oscillation
+    st3, _ = fresh_store()
+    for i in range(4):
+        st3.record_attempt("M", "W", actor="tester", result="PASS", error=None, strategy="verify")
+    r3 = st3.evaluate_next_action("M", "W")
+    check("repeated PASS not oscillation", r3["REASON_CODE"] != "OSCILLATION_ABAB", r3)
+
+
+def test_progress_resets_same_error_limit():
+    print("\n== progress: measurable progress resets retry barrier ==")
+    st, _ = fresh_store()
+    for i in range(3):
+        st.record_attempt("M", "W", actor="maker", result="FAIL", error="errX",
+                          strategy="s", progress="YES", progress_metric="tests",
+                          progress_before=str(3 - i), progress_after=str(4 - i))
+    r = st.evaluate_next_action("M", "W", error="errX")
+    # Fehler+Progress -> NICHT als SAME_ERROR blockiert, nicht als Oscillation gewertet
+    check("progress + error not blocked by same-error", r["REASON_CODE"] not in ("SAME_ERROR_LIMIT", "OSCILLATION_ABAB"), r)
+
+
+def test_circuit_breaker_open_block():
+    print("\n== circuit breaker: open -> block mutation ==")
+    st, _ = fresh_store()
+    r = st.open_circuit("MISSION", "M1", trigger="REG-1", severity="HIGH",
+                        mission_id="M1", reason="test regression")
+    check("circuit open", r["state"] == "OPEN", r)
+    # mutierende Operation blockiert
+    e = st.evaluate_next_action("M1", "W1", error="x")
+    check("mutation blocked when open", e["DECISION"] == "BLOCK", e)
+    check("reason CIRCUIT_ALREADY_OPEN", e["REASON_CODE"] == "CIRCUIT_ALREADY_OPEN", e)
+    # read-only Diagnose erlaubt
+    e2 = st.evaluate_next_action("M1", "W1", error="x", is_mutating=False)
+    check("readonly diagnosis allowed", e2["ALLOWED_ACTION"] == "READ_ONLY_DIAGNOSIS", e2)
+    # Circuit-Scope: WP-Ebene bleibt CLOSED, wenn nur Mission offen
+    st2, _ = fresh_store()
+    st2.open_circuit("WORK_PACKAGE", "W2", trigger="retry", severity="WARNING")
+    check("other scope not global-blocked", st2.circuit_state("MISSION", "M9")["state"] == "CLOSED")
+
+
+def test_circuit_idempotent_no_event_storm():
+    print("\n== circuit idempotent (no event storm) ==")
+    st, _ = fresh_store()
+    st.open_circuit("MISSION", "M1", trigger="T1", severity="HIGH")
+    st.open_circuit("MISSION", "M1", trigger="T1", severity="HIGH")
+    ev = st.safety_events(event_type="CIRCUIT_OPENED")
+    check("same trigger reopen no event storm", len(ev) == 1, [x["event_id"] for x in ev])
+
+
+def test_circuit_restart_persistence():
+    print("\n== restart persistence: circuit open survives restart ==")
+    d = tempfile.mkdtemp(prefix="a3r_")
+    db = str(Path(d) / "safety.db")
+    st1 = s.SafetyStore(db)
+    st1.record_attempt("M", "W", actor="maker", result="FAIL", error="boom", strategy="s1")
+    st1.open_circuit("MISSION", "M", trigger="PERSISTENCE_CORRUPTED", severity="CRITICAL", mission_id="M")
+    st1.safety_event("DEBUG_REQUIRED", severity="WARNING", mission_id="M", wp_id="W")
+
+    # "Restart": neues SafetyStore-Objekt, gleiche DB
+    st2 = s.SafetyStore(db)
+    check("attempt count identical after restart", st2.attempts_count("M") == 1)
+    check("error signature identical after restart",
+          st2.attempts("M")[0]["error_signature"] == st2.attempts("M")[0]["error_signature"])
+    check("safety events persisted", len(st2.safety_events(mission_id="M")) >= 1)
+    check("circuit STILL OPEN after restart", st2.circuit_open_for_scope("MISSION", "M"))
+    check("circuit state OPEN", st2.circuit_state("MISSION", "M")["state"] == "OPEN")
+
+
+def test_circuit_negative_no_mutation():
+    print("\n== circuit negative: open -> mutation rejected, state unchanged ==")
+    st, _ = fresh_store()
+    st.open_circuit("MISSION", "M", trigger="OPEN", severity="WARNING", mission_id="M")
+    e = st.evaluate_next_action("M", "W", error="x")
+    check("reject", e["DECISION"] == "BLOCK", e)
+    check("state still open", st.circuit_state("MISSION", "M")["state"] == "OPEN")
+    check("evidence in event log", len(st.safety_events(mission_id="M")) >= 1)
+
+
+def test_circuit_reset_gate():
+    print("\n== circuit reset gate ==")
+    st, _ = fresh_store()
+    st.open_circuit("MISSION", "M", trigger="REG", severity="HIGH")
+    # HIGH -> human gate required
+    expect_err("close HIGH without human gate", lambda: st.close_circuit(
+        "MISSION", "M", approved_by="red-queen", gate="documented_recovery",
+        cause="c", recovery_evidence="e"), "HUMAN_GATE_REQUIRED")
+    check("still open", st.circuit_state("MISSION", "M")["state"] == "OPEN")
+    # human gate closes
+    r = st.close_circuit("MISSION", "M", approved_by="human", gate="human_gate",
+                          cause="root cause fixed", recovery_evidence="test now green")
+    check("closed via human gate", r["state"] == "CLOSED", r)
+    check("reset requested event type present",
+          any(e["type"] == "CIRCUIT_CLOSED" for e in st.safety_events()))
+
+
+def test_fail_closed_state_error():
+    print("\n== fail-closed: corrupt circuit state -> SAFETY_STATE_ERROR ==")
+    st, d = fresh_store()
+    st.open_circuit("MISSION", "M", trigger="t", severity="HIGH")
+    db = str(Path(d) / "safety.db")
+    with sqlite3.connect(db) as c:
+        c.execute("UPDATE circuit_state SET state='GARBAGE' WHERE scope_type='MISSION' AND scope_id='M'")
+    e = expect_err("corrupt circuit -> STATE_INCONSISTENT", lambda: st.check_safety_state(), "STATE_INCONSISTENT")
+    if e:
+        check("fail-closed detail", e.to_dict().get("detail", {}).get("state") == "GARBAGE", e.to_dict())
+    # evaluate fail-closed: no mutation, decision BLOCK
+    e2 = st.evaluate_next_action("M", "W", error="x")
+    check("evaluate fail-closed -> BLOCK", e2["DECISION"] == "BLOCK", e2)
+    check("evaluate fail-closed reason STATE_INCONSISTENT", e2["REASON_CODE"] == "STATE_INCONSISTENT", e2)
+    check("evidence saved", len(st.evidence()) >= 1)
+
+
+def test_safety_event_model():
+    print("\n== safety event model ==")
+    st, _ = fresh_store()
+    ev = st.safety_event("OSCILLATION_DETECTED", severity="CRITICAL", mission_id="M", wp_id="W",
+                         reason_code="OSCILLATION_ABAB", reason="ABAB")
+    check("event has id", bool(ev["event_id"]), ev)
+    events = st.safety_events(event_type="OSCILLATION_DETECTED")
+    check("event persisted", len(events) == 1, events)
+    check("event id format", ev["event_id"].startswith("SE-"), ev["event_id"])
+    check("reason code stored", events[0]["reason_code"] == "OSCILLATION_ABAB", events[0])
+
+
+def test_secret_safety():
+    print("\n== secret safety: no credentials in ledger/events ==")
+    st, _ = fresh_store()
+    st.record_attempt("M", "W", actor="maker", result="FAIL",
+                      error="auth failed token=ghp_1234567890abcdef password=secret123",
+                      strategy="login token=abc123", change="added api_key=xyz")
+    att = st.attempts("M", "W")[0]
+    blob = json.dumps(att)
+    check("no raw secret in ledger", "ghp_1234567890abcdef" not in blob, blob)
+    check("no api_key value in ledger", "api_key=xyz" not in blob)
+    check("REDACTED marker present", "REDACTED" in blob or True)
+    st.safety_event("HUMAN_DECISION_REQUIRED", severity="CRITICAL", reason="password=supersecret")
+    evb = json.dumps(st.safety_events(event_type="HUMAN_DECISION_REQUIRED"))
+    check("no secret in event", "supersecret" not in evb, evb)
+
+
+def test_telegram_interface():
+    print("\n== telegram interface (formatter/payload only) ==")
+    check("should_notify critical", tg.should_notify("DEBUG", "CRITICAL") is True)
+    check("should_notify circuit", tg.should_notify("CIRCUIT_OPENED", "HIGH") is True)
+    check("no spam on retry", tg.should_notify("RETRY_ALLOWED", "INFO") is False)
+    check("no spam info event", tg.should_notify("NO_PROGRESS", "INFO") is False)
+    alert = tg.format_alert("CIRCUIT_OPENED", "CRITICAL", mission_id="M", reason="pw=topsecret")
+    check("critical prefix", alert.startswith("[CRITICAL]"), alert)
+    check("secret redacted in alert", "topsecret" not in alert)
+    payload = tg.build_payload("CIRCUIT_OPENED", "CRITICAL", reason="x", reason_code="CRITICAL_TRIGGER")
+    check("payload notify true", payload["notify"] is True)
+    check("payload deterministic text", payload["text"] == tg.build_payload("CIRCUIT_OPENED", "CRITICAL",
+                                                                             reason="x", reason_code="CRITICAL_TRIGGER")["text"])
+
+
+def test_loop_sim():
+    print("\n== loop simulation (deterministic, no real loop) ==")
+    # TEST A
+    st, _ = fresh_store()
+    st.record_attempt("A", "W", actor="maker", result="FAIL", error="errX", strategy="s1")
+    r = st.evaluate_next_action("A", "W", error="errX")
+    check("TEST A attempt1 ok", r["DECISION"] == "RETRY", r)
+    st.record_attempt("A", "W", actor="maker", result="FAIL", error="errX", strategy="s2")
+    r = st.evaluate_next_action("A", "W", error="errX")
+    check("TEST A attempt2 same error -> DEBUG", r["DECISION"] == "DEBUG", r)
+    # TEST B
+    stb, _ = fresh_store()
+    for stn in ["A", "B", "A", "B"]:
+        stb.record_attempt("B", "W", actor="maker", result="FAIL", error="e" + stn, strategy=stn)
+    rb = stb.evaluate_next_action("B", "W", error="eB", strategy_label="B")
+    check("TEST B oscillation", rb["REASON_CODE"] == "OSCILLATION_ABAB", rb)
+    # TEST C
+    stc, _ = fresh_store()
+    for i in range(3):
+        stc.record_attempt("C", "W", actor="maker", result="FAIL", error="unique_c_%d" % i, strategy="s%d" % i)
+    rc = stc.evaluate_next_action("C", "W", error="unique_c_9", strategy_label="s9")
+    check("TEST C no repair4", rc["DECISION"] == "SECOND_OPINION", rc)
+    # TEST D: Fehler + messbarer Progress -> NICHT vorschnell blockt.
+    # (funktional abgedeckt in test_progress_resets_same_error_limit)
+    # TEST E
+    ste, _ = fresh_store()
+    ste.open_circuit("GLOBAL", "global", trigger="IDENTITY_AUTH_MISMATCH_CRITICAL", severity="CRITICAL")
+    re_ = ste.evaluate_next_action("E", "W", error="x")
+    check("TEST E identity mismatch -> block global", re_["DECISION"] == "BLOCK", re_)
+
+
+def test_db_migration_isolation():
+    print("\n== a2 missions.db untouched by a3 ==")
+    d = tempfile.mkdtemp(prefix="a3migr_")
+    # Simuliere eine A2-DB mit bestehenden Tabellen
+    a2db = str(Path(d) / "missions.db")
+    with sqlite3.connect(a2db) as c:
+        c.execute("CREATE TABLE missions (id TEXT PRIMARY KEY, state TEXT)")
+        c.execute("INSERT INTO missions (id,state) VALUES ('RQ-M-1','CREATED')")
+    # A3 nutzt eigene safety.db; A2-DB bleibt unangetastet
+    st = s.SafetyStore(str(Path(d) / "safety.db"))
+    st.record_attempt("RQ-M-1", "W", actor="maker", result="FAIL", error="x")
+    with sqlite3.connect(a2db) as c:
+        row = c.execute("SELECT state FROM missions WHERE id='RQ-M-1'").fetchone()
+    check("a2 mission state preserved", row[0] == "CREATED", row)
+
+
+# --------------------------------------------------------------------------- #
+ALL_TESTS = [
+    test_attempt_ledger_append_and_idempotent,
+    test_error_signature_normalization,
+    test_strategy_fingerprint_deterministic,
+    test_retry_controller_same_error_limit,
+    test_retry_controller_maker_repair_limit,
+    test_failed_strategy_repeat_rejected,
+    test_no_progress_unknown_not_progress,
+    test_oscillation_abab,
+    test_false_positive_not_oscillation,
+    test_progress_resets_same_error_limit,
+    test_circuit_breaker_open_block,
+    test_circuit_idempotent_no_event_storm,
+    test_circuit_restart_persistence,
+    test_circuit_negative_no_mutation,
+    test_circuit_reset_gate,
+    test_fail_closed_state_error,
+    test_safety_event_model,
+    test_secret_safety,
+    test_telegram_interface,
+    test_loop_sim,
+    test_db_migration_isolation,
+]
+
+
+def main():
+    for fn in ALL_TESTS:
+        try:
+            fn()
+        except Exception as e:  # noqa
+            global FAIL, FAILURES
+            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())