diff --git a/a4/.gitignore b/a4/.gitignore new file mode 100644 index 0000000..7dd2791 --- /dev/null +++ b/a4/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +*.pyc +*.db +*.db-wal +*.db-shm +*.tmp diff --git a/a4/README.md b/a4/README.md new file mode 100644 index 0000000..1ee814a --- /dev/null +++ b/a4/README.md @@ -0,0 +1,178 @@ +# Red Queen — A4: Bounded Orchestrator (V1) + +Der **Bounded Orchestrator** verbindet erstmals kontrolliert die A1-Architektur, +den A2 **Mission/WP-State** (`missions.db`, autoritativ für Transitionen), den A3 +**Deterministic Safety Layer** (`safety.db`, verbindliche Safety-Decision) und +die **Hermes-native `delegate_task`**-Sub-Agenten (Planner/Maker/Checker) zu +einer **expliziten, garantiert endenden** Build-Pipeline: + +``` +MISSION → VALIDATE STATE → VALIDATE SAFETY → PLAN → VALIDATE PLAN + → CREATE WORK PACKAGES → SELECT RUNNABLE WP → MAKER → TEST + → FRESH CHECKER → SAFETY DECISION → STATE UPDATE → NEXT BOUNDED STEP + → FINALIZE / STOP +``` + +**WICHTIG:** A4 ist **KEIN 24/7-System.** Kein Heartbeat, kein Mission-Cron, kein +dauerhafter Worker, kein while-True-Agentloop, kein Background-Daemon, kein +Self-Improvement-Automation, keine automatische Rain-Eskalation. A4 läuft nur nach +einem **expliziten Start** und endet **garantiert** (bounded). + +--- + +## Architektur-Prinzip + +**Red Queen (ein Hermes-Agent) ist der Orchestrator.** Der echte Sub-Agenten-Versand +(Planner/Maker/Checker) läuft über die Hermes-native `delegate_task`-Engine — ein +Python-Prozess kann diese NICHT aufrufen. Deshalb implementiert A4 die +**deterministische Orchestrierungs-Logik** als reine, inaktive Python-Library +(`Orchestrator`), die Red Queen als **Entscheidungs- und State-Maschine** importiert +und ansteuert: + +- Die **Library** entscheidet deterministisch: was ist runnable, Safety-Gate, + Risk>Size, Approval-Gate, Completion-Gate, Bounded-Run-Limits, Repair-Flow, + idempotente Wiederaufnahme, Restart-Persistenz. +- **Red Queen selbst** startet die tatsächlichen Child-Agenten (Planner/Maker/Checker) + über `delegate_task`, übergibt ihnen das von der Library erzeugte + Dispatch-Contract und führt das Ergebnis über `apply_child_result` zurück. +- Für Tests wird ein simuliertes `ChildDispatcher`-Callable injiziert — die Library + selbst startet nichts und läuft nie zyklisch. + +--- + +## Datenfluss & Gates + +| Schritt | Mechanik | Verbindlich | +|---------|----------|-------------| +| 1. Plan erhalten | von Planner | Output-Format laut AGENT_CONTRACTS | +| 2. Plan-Validierung | `PlanValidator.validate()` | §6; ungültig → REJECT, keine Ausführung | +| 3. WP-Anlage | A2 `MissionStore` | kein direkter SQL außerhalb A2-APIs | +| 4. Runnable-Selection | `select_runnable()` | §8; DONE-Deps + READY + Safety CLOSED + Scope erlaubt | +| 5. Consistency-Gate | `_consistency_gate()` | §9: A2-State + A3-Check vor JEDER Mutation | +| 6. Risk-Klassifikation | `classify_risk()` | §26/27: RISK überschreibt SIZE | +| 7. Approval-Gate | `requires_approval()` | §25: CRITICAL/Approval-Target → APPROVAL_REQUIRED | +| 8. Dispatch | `build_maker_contract` / `build_checker_contract` | Minimum-Necessary-Context | +| 9. Safety-Decision | A3 `evaluate_next_action` | verbindlich | +| 10. Repair-Flow | RETRY max 3 / gleiche Error-Sig max 2 | SAFETY_CONTRACT | +| 11. Circuit-Enforcement | Circuit OPEN → NO MUTATION | §19 | +| 12. Completion-Gate | `mission_completion_gate()` | §20 + Final-Review-Gate | +| 13. Bounded-Run | `run_bounded()` | §22: endet garantiert | + +--- + +## A2/A3-Integration + +A4 **importiert** und nutzt die bestehenden Module (kein Duplikat): + +```python +from a2.rq_mission import MissionStore # autoritativ für Mission-/WP-Transitions +from a3.rq_safety import SafetyStore # verbindliche Safety-Decision +``` + +- A2 (`missions.db`) ist die **autoritative Quelle** für Mission-/WP-State und + erlaubte Transitionen. +- A3 (`safety.db`) ist die **verbindliche Safety-Decision** (`evaluate_next_action`, + `check_safety_state`, `record_attempt`, `open_circuit`). +- A4 selbst hält **keine neue zentrale DB**; Run-Flags/Sub-Agent-Evidence werden + über A2/A3-APIs bzw. den A3-Safety-Evidence-Mechanismus persistiert. +- Import-Lösung: `sys.path` wird um Repo-Root und `a4/` ergänzt + (`Path(__file__).resolve().parent.parent`). + +--- + +## Planner/Maker/Checker-Contract + +| Rolle | Output | Gate | +|-------|--------|------| +| **Planner** | WORK_PACKAGES / DEPENDENCIES / RISK_CLASS / ORDER / TEST_REQUIREMENTS | Red Queen validiert deterministisch (§6) | +| **Maker** | IMPLEMENTATION_SUMMARY / FILES_CHANGED / TESTS_RUN / TEST_RESULTS | Nie final PASS | +| **Checker** | PASS / FAIL / Verdict | **FRESH** Child, ohne Maker-Argumentation | + +`build_maker_contract` und `build_checker_contract` erzeugen die +**Minimum-Necessary-Context**-Contracts. Der Checker-Contract enthält bewusst +**keine** Maker-Rechtfertigung (AGENT_CONTRACTS §3). Werte werden secret-redacted. + +--- + +## Bounded Run (Limits aus Konstanten) + +```python +MAX_WORK_PACKAGES_PER_RUN = 12 +MAX_REPAIR_CYCLES = 3 # entspricht A3 MAX_MAKER_CHECKER_REPAIRS +MAX_CHILDREN_ACTIVE = 3 +MAX_BOUNDED_RUNTIME = 1000 # Sekunden / Steps +``` + +`run_one_step()` führt **GENAU EINEN** deterministischen Schritt aus und endet +**garantiert** (keine Rekursion). `run_bounded()` stoppt bei +`MISSION_COMPLETED / BLOCKED / ESCALATED / CIRCUIT_OPEN / NO_RUNNABLE_WP / +RUN_BUDGET_REACHED / ERROR` — kein self-reschedule. + +--- + +## Approval Gates + +WP, die laut `ROOT_SSH_GATE` einen **kritischen Bereich** berühren (SSH, Firewall, +Auth, Secrets, Recovery, Red Queen Runtime, Circuit Breaker, Root Policy, Live +Trading) → `ST_APPROVAL_REQUIRED`, **keine Ausführung**. Red Queen legt den Antrag +im `EXTERNAL_REVIEW`-Format vor und wartet auf Christian. + +--- + +## Failure Handling / Idempotenz + +- **Failure:** `evaluate_next_action`-Decision wird verbindlich befolgt + (RETRY / DEBUG / SECOND_OPINION / BLOCK / CIRCUIT_BREAK). Repair max 3, gleiche + Error-Signatur max 2. Kein implizit PASS, kein endlos Spawn. +- **SECOND_OPINION → KEIN automatisches Rain** (nur interner frischer Child oder + BLOCK; Rain ausschließlich Christian-gated, EXTERNAL_REVIEW_CONTRACT). +- **Idempotenz:** DONE-WP wird nie erneut Maker; bereits registrierter Attempt + nicht doppelt; COMPLETED-Mission nie erneut; gleiche Child-Task-ID nie doppelt + gewertet. +- **Restart:** State in A2/A3-DBs persistent; nach Restart wird der Zustand gelesen, + ohne doppelte WP-Ausführung oder verlorene Attempts. + +--- + +## Sub-Agent-Evidence (§30) + +`make_child_evidence()` erzeugt einen maschinenlesbaren Evidence-Record +(Child Role, Child Task ID, Mission ID, WP ID, Start, Result, Verdict) — +**KEINE Chain-of-Thought**. Credentials werden redactiert (nur EXISTS/LENGTH/redacted). + +--- + +## CLI + +`rq_orchestrator_cli.py` — `--json`, Exit-Code 2 bei `OrchestratorError` (A2/A3-Muster): + +``` +plan Plan validieren + WP-Anlage +run-one-step Einen deterministischen Schritt ausführen +run-bounded Begrenzter Run (endet garantiert) +complete Completion-Gate prüfen / Final-Review +status Status abfragen +evaluate A3-Safety-Decision +approve Approval registrieren +``` + +--- + +## Test Harness + +`python3 a4/test_a4.py` — isolierte, deterministische Tests (temp-DBs, kein Netz, +kein hermes). Injiziert simulierte `ChildDispatcher`. Exit-Code 0 = PASS. +Deckt A4 §6–§33 ab: Plan-/Dependency-Validierung, Runnable-Selection, A2/A3- +Consistency-Gate, Maker/Checker-Contract, Checker-Pass/Fail, Retry allowed/denied, +Retry-4-Impossible, Circuit-open-blocks-Maker, Approval-Gate, Risk-overrides-Size, +No-Runnable-Stop, Completion-/Final-Review-Gate, Idempotent-WP, Restart-Resume, +Child-Failure, Reason-Codes, Secret-safe-Logging. + +--- + +## Known Limitations + +- A4 selbst führt **keine** echte `delegate_task` aus (das macht Red Queen). +- Kein automatisches Rain; kein Heartbeat/Cron/Daemon. +- DEBUG ist als Schnittpunkt implementiert (leitet an A5-Debugger/Strategiewechsel + weiter), erzwingt aber keine interne Debugger-Logik. diff --git a/a4/rq_orchestrator.py b/a4/rq_orchestrator.py new file mode 100644 index 0000000..2469d71 --- /dev/null +++ b/a4/rq_orchestrator.py @@ -0,0 +1,969 @@ +#!/usr/bin/env python3 +""" +Red Queen — A4: BOUNDED ORCHESTRATOR v1 (deterministische, inaktive Library). + +VERBINDLICHER RAHMEN +==================== +Red Queen (ein Hermes-Agent) IST der Orchestrator. Der echte Sub-Agenten-Versand +(Planner/Maker/Checker) erfolgt ueber die Hermes-native `delegate_task`-Engine, die +ein Python-Prozess NICHT aufrufen kann. Deshalb implementiert A4 die DETERMINISTISCHE +ORCHESTRIERUNGS-LOGIK als reine Library (`Orchestrator`), die Red Queen importiert und +ansteuert. Die Entscheidungen (was ist runnable, Safety-Gate, Risk>Size, Approval, +Completion-Gate, Bounded-Run-Limits, idempotente Wiederaufnahme, Restart-Persistenz) +sind VOLLSTAENDIG deterministisch und unabhaengig vom LLM. + +Diese Library ist INAKTIV: Sie startet nichts, macht nichts zyklisch und delegiert NICHTS +von selbst. Jede Aktion wird EXPLIZIT durch eine Methode/CLI-Befehl angestossen und ist +garantiert begrenzt (bounded). Es gibt KEINEN Loop, kein Heartbeat, kein Cron, kein +Daemon, kein self-reschedule, kein self-improvement, keinen automatischen Rain. + +INTEGRATION: + * A2 `MissionStore` (missions.db) ist die AUTHORITATIVE Quelle fuer Mission/WP-State + und Transitionen. A4 erzeugt WPs + Dependencies ausschliesslich ueber die A2-API, + nie per direkter SQL-Manipulation. + * A3 `SafetyStore` (safety.db) ist die verbindliche Safety-Decision (Retry-Limits, + Circuit Breaker, Attempt-Ledger). A4 fragt vor JEDER mutierenden Aktion A3. + * Der eigentliche Child-Versand bleibt bei Red Queen. A4 liefert reine Dispatch- + Contracts (Builder) und die `ChildDispatcher`-Abstraktion (Callable), die ein + Plugin/Test injizieren kann. Die Library selbst ruft nie delegate_task. + +SECRET-SAFETY: Sub-Agent-Evidence und Logs werden durch `redact_secret` (A3) bereinigt. +Es werden NIE Credential-artige Werte persistiert oder ausgegeben. + +Import-Loesung: Dieses Modul liegt in `a4/`; die A2-/A3-Module liegen in `a2/`/`a3/` +im selben Repo-Root. Es wird `sys.path` um Repo-Root ergaenzt, damit +`from a2.rq_mission import MissionStore` und `from a3.rq_safety import SafetyStore` +zuverlaessig funktionieren. +""" + +from __future__ import annotations + +import datetime +import json +import os +import sys +from pathlib import Path +from typing import Any, Callable, Dict, List, Optional + +# --------------------------------------------------------------------------- # +# Pfad-Loesung fuer A2/A3-Importe (Repo-Root = eine Ebene ueber `a4/`) +# --------------------------------------------------------------------------- # +_REPO_ROOT = Path(__file__).resolve().parent.parent +for _p in (str(_REPO_ROOT), str(_REPO_ROOT / "a2"), str(_REPO_ROOT / "a3")): + 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_CLOSED, + CIRCUIT_OPEN, + MAX_MAKER_CHECKER_REPAIRS, + MAX_SAME_ERROR_SIGNATURE, + PROGRESS_NO, + PROGRESS_UNKNOWN, + PROGRESS_YES, + RESULT_FAIL, + RESULT_PASS, + SafetyError, + SafetyStore, + redact_secret, +) + +# --------------------------------------------------------------------------- # +# A4-Fehler (maschinenlesbar, stabil) +# --------------------------------------------------------------------------- # +class OrchestratorError(Exception): + """Geworfener, maschinenlesbarer Orchestrator-Fehler mit stabilem `code`.""" + + 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 to_dict(self) -> Dict[str, Any]: + return {"code": self.code, "message": self.message, "detail": self.detail} + + +def _err(code: str, msg: str, **detail) -> OrchestratorError: + return OrchestratorError(code, msg, detail) + + +# --------------------------------------------------------------------------- # +# Verbindliche, deterministische Konstanten (Bounded-Run-Limits) +# --------------------------------------------------------------------------- # +MAX_WORK_PACKAGES_PER_RUN = 12 # max WP, die ein bounded run materialisiert +MAX_REPAIR_CYCLES = 3 # Maker->Checker Repair MAX 3 (spiegelt A3) +MAX_CHILDREN_ACTIVE = 3 # max gleichzeitig aktive Kinder +MAX_BOUNDED_RUNTIME = 1000 # aeusserste Schritt-Budget eines bounded run +MAX_PLAN_WPS = 32 # max WP pro Plan (Validierung) + +# Risk-Klassen (ARCHITECTURE.md §5) +RISK_SMALL = "SMALL" # Red Queen direkt +RISK_MEDIUM = "MEDIUM" # MAKER +RISK_LARGE = "LARGE" # MAKER +RISK_CRITICAL = "CRITICAL" # MAKER + CHECKER (+ ggf. Approval) +RISK_DIFFICULT = "DIFFICULT" # DEBUGGER +RISK_SELF_MOD = "SELF-MOD" # Human/External Gate +RISK_CLASSES = frozenset( + {RISK_SMALL, RISK_MEDIUM, RISK_LARGE, RISK_CRITICAL, RISK_DIFFICULT, RISK_SELF_MOD} +) + +# Kritische Bereiche (ROOT_SSH_GATE.md §4) -> deterministische Risk-Klassifikation. +# RISK_CLASS UEBERSCHREIBT TASK_SIZE: auch ein SMALL-WP mit kritischem Target ist CRITICAL. +CRITICAL_TARGETS = frozenset( + { + "ssh", "ssh-access", "remote-access", + "firewall", "networking", "network", + "docker", "container", "container-infrastructure", "reverse-proxy", "proxy", + "auth", "authentication", "authorization", + "secret", "secrets", "credential", "credentials", "credential-management", + "recovery", "recovery-access", "backup", + "forgejo-security", "forgejo", "ollama-core", "ollama", + "red-queen-runtime", "hermes-runtime", "hermes", + "persistence", "state-db", "kanban", "cron", + "circuit-breaker", "approval-gates", "approval", + } +) + +# Child-Rollen +ROLE_MAKER = "MAKER" +ROLE_CHECKER = "CHECKER" +ROLE_RED_QUEEN = "RED_QUEEN" + +# --- Status-/Reason-Codes (maschinenlesbar, stabil) -------------------------- +ST_PLAN_VALIDATED = "PLAN_VALIDATED" +ST_PLAN_INVALID = "PLAN_INVALID" +ST_TERMINAL = "MISSION_TERMINAL" +ST_NO_RUNNABLE_WORK = "NO_RUNNABLE_WORK" +ST_MUTATION_BLOCKED = "MUTATION_BLOCKED" +ST_STATE_CONFLICT = "ORCHESTRATOR_STATE_CONFLICT" +ST_APPROVAL_REQUIRED = "APPROVAL_REQUIRED" +ST_BUDGET_REACHED = "RUN_BUDGET_REACHED" +ST_NEEDS_DISPATCH = "NEEDS_DISPATCH" +ST_NEEDS_CHECKER_DISPATCH = "NEEDS_CHECKER_DISPATCH" +ST_WAITING_PENDING = "WAITING_PENDING_CHILD" +ST_CHECKER_PASS = "CHECKER_PASS" +ST_CHECKER_BLOCKED = "CHECKER_BLOCKED" +ST_WP_DONE = "WP_DONE" +ST_DEBUG_REQUIRED = "DEBUG_REQUIRED" +ST_SECOND_OPINION_REQUIRED = "SECOND_OPINION_REQUIRED" +ST_RETRY4_IMPOSSIBLE = "RETRY4_IMPOSSIBLE" +ST_MISSION_COMPLETED = "MISSION_COMPLETED" +ST_CHILD_TECHNICAL_FAILURE = "CHILD_TECHNICAL_FAILURE" +ST_MISSION_NOT_COMPLETED = "MISSION_NOT_COMPLETED" + +# Alle deterministisch moeglichen Status-Codes (maschinenlesbar, stabil). +RUN_CODES = frozenset( + { + ST_PLAN_VALIDATED, ST_PLAN_INVALID, ST_TERMINAL, ST_NO_RUNNABLE_WORK, + ST_MUTATION_BLOCKED, ST_STATE_CONFLICT, ST_APPROVAL_REQUIRED, ST_BUDGET_REACHED, + ST_NEEDS_DISPATCH, ST_NEEDS_CHECKER_DISPATCH, ST_WAITING_PENDING, + ST_CHECKER_PASS, ST_CHECKER_BLOCKED, ST_WP_DONE, ST_DEBUG_REQUIRED, + ST_SECOND_OPINION_REQUIRED, ST_RETRY4_IMPOSSIBLE, ST_MISSION_COMPLETED, + ST_CHILD_TECHNICAL_FAILURE, ST_MISSION_NOT_COMPLETED, + } +) + +# Status-Codes, die einen bounded run sicher beenden (kein self-reschedule). +_TERMINAL_STEPS = frozenset( + { + ST_TERMINAL, ST_MUTATION_BLOCKED, ST_STATE_CONFLICT, ST_APPROVAL_REQUIRED, + ST_SECOND_OPINION_REQUIRED, ST_DEBUG_REQUIRED, ST_WAITING_PENDING, + ST_RETRY4_IMPOSSIBLE, ST_MISSION_COMPLETED, ST_CHILD_TECHNICAL_FAILURE, + ST_NO_RUNNABLE_WORK, + } +) +# Codes, die auf einen Red-Queen-Dispatch warten (run endet an der Grenze). +_DISPATCH_CODES = frozenset({ST_NEEDS_DISPATCH, ST_NEEDS_CHECKER_DISPATCH}) + +# Zulaessige WP-Zustaende, mit denen ein WP im Plan starten darf. +WP_PLAN_STATES = frozenset({"TODO", "READY"}) + +# --------------------------------------------------------------------------- # +# ChildDispatcher-Abstraktion +# child_dispatcher(contract: Dict[str, Any]) -> Dict[str, Any] +# Ein Plugin/Test injiziert diese Callable. Die Library ruft sie nur, wenn sie +# explizit gesetzt ist (kein deterministisches auto-dispatch). +ChildDispatcher = Callable[[Dict[str, Any]], Dict[str, Any]] + + +def _utcnow() -> str: + return datetime.datetime.now(datetime.timezone.utc).isoformat(timespec="seconds") + + +# --------------------------------------------------------------------------- # +# Risk-Klassifikation (deterministisch, RISK_CLASS ueberschreibt TASK_SIZE) +# --------------------------------------------------------------------------- # +def classify_risk(target_component: Optional[str], size_class: Optional[str] = None) -> str: + """Deterministische Risk-Klasse aus Target-Bereich (ROOT_SSH_GATE kritische Bereiche). + + Reihenfolge (RISK OVERRIDES SIZE): + 1) Kritischer Target -> CRITICAL (unabhaengig von size_class). + 2) Explizite size_class -> normalisierte Klasse. + 3) Sonst -> MEDIUM (Standard: MAKER). + """ + target = (target_component or "").strip().lower() + if target and any(k in target for k in CRITICAL_TARGETS): + return RISK_CRITICAL + cls = (size_class or "").strip().upper() or RISK_MEDIUM + if cls in RISK_CLASSES: + return cls + return RISK_MEDIUM + + +def requires_approval(risk_class: str, wp: Dict[str, Any]) -> bool: + """Bestimmt, ob ein WP laut Contract Human-Approval verlangt (ROOT_SSH_GATE).""" + if wp.get("requires_approval"): + return True + if risk_class in (RISK_CRITICAL, RISK_SELF_MOD): + return True + return False + + +# --------------------------------------------------------------------------- # +# Plan-Validierung (§6) — deterministisch, KEIN LLM +# --------------------------------------------------------------------------- # +class PlanValidator: + """Validiert einen Planner-Output gegen die A4-Regeln (beschreibend, keine Ausfuehrung).""" + + def __init__(self, mission_id: Optional[str] = None): + self.mission_id = mission_id + + @staticmethod + def validate_cycle(deps_map: Dict[str, List[str]], ids: List[str]) -> Optional[List[str]]: + """Deterministische Cycle-Detektion (iterative DFS). Liefert Cycle-Pfad oder None.""" + state = {} # 0=unbesucht,1=in-stack,2=fertig + for start in ids: + if state.get(start) == 2: + continue + state[start] = 1 + stack = [start] + visit = [(start, iter(deps_map.get(start, [])))] + while visit: + node, it = visit[-1] + advanced = False + for dep in it: + if state.get(dep) == 1: + idx = stack.index(dep) + return stack[idx:] + [dep] + if state.get(dep) == 2: + continue + state[dep] = 1 + stack.append(dep) + visit.append((dep, iter(deps_map.get(dep, [])))) + advanced = True + break + if not advanced: + state[node] = 2 + if stack: + stack.pop() + visit.pop() + return None + + def validate(self, plan: Dict[str, Any]) -> Dict[str, Any]: + """Liefert {'valid': bool, 'errors': [{'code','message','...'}]}.""" + errors: List[Dict[str, Any]] = [] + wps = plan.get("work_packages") + if not isinstance(wps, list) or not wps: + errors.append({"code": "EMPTY_PLAN", "message": "plan has no work packages"}) + return {"valid": False, "errors": errors, "plan": plan} + + if len(wps) > MAX_PLAN_WPS: + errors.append({"code": "PLAN_TOO_LARGE", + "message": f"plan exceeds {MAX_PLAN_WPS} wps", "count": len(wps)}) + + ids: List[str] = [] + for w in wps: + wp_id = str(w.get("wp_id") or "").strip() + if not wp_id: + errors.append({"code": "MISSING_WP_ID", "message": "wp missing wp_id"}) + continue + if wp_id in ids: + errors.append({"code": "DUPLICATE_WP_ID", + "message": f"duplicate wp_id {wp_id!r}", "wp_id": wp_id}) + continue + ids.append(wp_id) + if w.get("mission_id") and self.mission_id and str(w.get("mission_id")) != self.mission_id: + errors.append({"code": "FOREIGN_WP", + "message": f"wp {wp_id} belongs to another mission", "wp_id": wp_id}) + if not w.get("goal") and not w.get("description"): + errors.append({"code": "MISSING_GOAL", + "message": f"wp {wp_id} missing goal/description", "wp_id": wp_id}) + if not w.get("acceptance_criteria"): + errors.append({"code": "MISSING_AC", + "message": f"wp {wp_id} missing acceptance_criteria", "wp_id": wp_id}) + if not w.get("scope") and not w.get("target_component"): + errors.append({"code": "MISSING_SCOPE", + "message": f"wp {wp_id} missing scope", "wp_id": wp_id}) + st = w.get("state") + if st is not None and st not in WP_PLAN_STATES: + errors.append({"code": "INVALID_WP_STATE", + "message": f"wp {wp_id} state {st!r} not allowed", "wp_id": wp_id}) + rc = w.get("risk_classification") + if rc is not None and str(rc).upper() not in RISK_CLASSES: + errors.append({"code": "INVALID_RISK_CLASS", + "message": f"wp {wp_id} invalid risk {rc!r}", "wp_id": wp_id}) + + id_set = set(ids) + deps_map: Dict[str, List[str]] = {wid: [] for wid in ids} + for w in wps: + wid = str(w.get("wp_id") or "") + for dep in (w.get("dependencies") or []): + dep = str(dep) + if dep == wid: + errors.append({"code": "SELF_DEPENDENCY", + "message": f"wp {wid} depends on itself", "wp_id": wid}) + continue + if dep not in id_set: + errors.append({"code": "UNKNOWN_DEPENDENCY", + "message": f"wp {wid} depends on unknown {dep!r}", "wp_id": wid}) + continue + deps_map[wid].append(dep) + + cycle = self.validate_cycle(deps_map, ids) + if cycle: + errors.append({"code": "DEPENDENCY_CYCLE", + "message": f"dependency cycle: {' -> '.join(cycle)}", "cycle": cycle}) + + return {"valid": not errors, "errors": errors, "plan": plan} + + +# --------------------------------------------------------------------------- # +# Dispatch-Contract-Builder (§5) +# --------------------------------------------------------------------------- # +def _child_task_id(mission_id: str, wp_id: str, role: str, round_: int) -> str: + return f"{mission_id}:{wp_id}:{role}:R{round_}" + + +def build_maker_contract( + wp: Dict[str, Any], + mission: Dict[str, Any], + risk_class: str, + retry_round: int = 1, +) -> Dict[str, Any]: + """Reine Datenstruktur fuer einen MAKER-Dispatch (Minimum-Necessary-Context). + + Enthaelt KEINE Chain-of-Thought; Werte sind secret-redacted. + """ + return { + "role": ROLE_MAKER, + "TASK_ID": _child_task_id(mission["id"], wp["id"], ROLE_MAKER, retry_round), + "MISSION_ID": mission["id"], + "WP_ID": wp["id"], + "GOAL": redact_secret(wp.get("goal") or wp.get("description") or ""), + "SCOPE": redact_secret(wp.get("scope") or wp.get("target_component") or ""), + "ALLOWED_FILES": list(wp.get("allowed_files") or []), + "FORBIDDEN_FILES": list(wp.get("forbidden_files") or []), + "ACCEPTANCE_CRITERIA": redact_secret(wp.get("acceptance_criteria") or ""), + "TEST_REQUIREMENTS": redact_secret(wp.get("test_requirements") or ""), + "SAFETY_RULES": list(wp.get("safety_rules") or ["NO_ROOT", "NO_SECRETS", "BOUNDED"]), + "RISK_CLASS": risk_class, + "RETRY_ROUND": retry_round, + } + + +def build_checker_contract( + wp: Dict[str, Any], + mission: Dict[str, Any], + maker_evidence: Dict[str, Any], + risk_class: str, + retry_round: int = 1, +) -> Dict[str, Any]: + """Checker-Input: Requirement/AC/Diff/Tests/Regeln — OHNE Maker-Argumentation. + + Der Maker-Rechtfertigungstext (IMPLEMENTATION_SUMMARY) wird bewusst NICHT + in den Checker-Contract uebernommen (AGENT_CONTRACTS §3). + """ + return { + "role": ROLE_CHECKER, + "task_id": _child_task_id(mission["id"], wp["id"], ROLE_CHECKER, retry_round), + "mission_id": mission["id"], + "wp_id": wp["id"], + "requirement": redact_secret(wp.get("goal") or wp.get("description") or ""), + "acceptance_criteria": redact_secret(wp.get("acceptance_criteria") or ""), + "diff_files": list(wp.get("allowed_files") or []), + "test_results": list(maker_evidence.get("test_results") or []), + "architecture_rules": list(wp.get("architecture_rules") or []), + "safety_rules": list(wp.get("safety_rules") or ["NO_ROOT", "NO_SECRETS", "BOUNDED"]), + "risk_class": risk_class, + } + + +# --------------------------------------------------------------------------- # +# Sub-Agent-Evidence-Record (§30) - maschinenlesbar, KEINE CoT, KEINE Secrets +# --------------------------------------------------------------------------- # +def make_child_evidence( + mission_id: str, + wp_id: str, + child_role: str, + child_task_id: str, + start_ts: str, + result: str, + verdict: str, + evidence_ref: Optional[str] = None, +) -> Dict[str, Any]: + """Baut einen maschinenlesbaren Evidence-Record (Child Role, Child Task ID, + Mission ID, WP ID, Start, Result, Verdict). KEINE Chain-of-Thought, redacted.""" + return { + "child_role": child_role, + "child_task_id": child_task_id, + "mission_id": mission_id, + "wp_id": wp_id, + "start_ts": start_ts, + "result": redact_secret(result), + "verdict": verdict, + "evidence_ref": redact_secret(evidence_ref), + } + + +# --------------------------------------------------------------------------- # +# Orchestrator +# --------------------------------------------------------------------------- # +class Orchestrator: + """Deterministischer, bounded Orchestrator. Verbindet A2 (Mission/WP-State) + + A3 (verbindliche Safety-Decision) in EINEM bounded run. + + EXPLICIT-only: kein Auto-Start. `run_one_step` fuehrt GENAU EINEN Schritt aus; + `run_bounded` laeuft nur, solange explizit aufgerufen und ist durch Limits + garantiert begrenzt. Es gibt keinen Daemon/Heartbeat/Cron. + """ + + def __init__( + self, + mission_db: str, + safety_db: str, + child_dispatcher: Optional[ChildDispatcher] = None, + *, + run_small_inline: bool = True, + max_wps_per_run: int = MAX_WORK_PACKAGES_PER_RUN, + max_repair_cycles: int = MAX_REPAIR_CYCLES, + max_children_active: int = MAX_CHILDREN_ACTIVE, + max_bounded_runtime: int = MAX_BOUNDED_RUNTIME, + ): + self.mission_db = mission_db + self.safety_db = safety_db + self.missions = MissionStore(mission_db) + self.safety = SafetyStore(safety_db) + self.child_dispatcher = child_dispatcher + self.run_small_inline = run_small_inline + self.max_wps_per_run = max_wps_per_run + self.max_repair_cycles = max_repair_cycles + self.max_children_active = max_children_active + self.max_bounded_runtime = max_bounded_runtime + self.validator = PlanValidator() + self._wp_meta_cache: Dict[str, Dict[str, Any]] = {} + + # -- Plan-Materialisierung ------------------------------------------------- + def load_plan(self, plan: Dict[str, Any], *, actor: str = "red-queen") -> Dict[str, Any]: + """Validiert + materialisiert einen Plan in A2 (Mission + WPs + Dependencies). + + KEINE direkte SQL-Manipulation; nutzt ausschliesslich die A2-API. + Ungueltig -> OrchestratorError(PLAN_INVALID), keine Ausfuehrung. + """ + v = self.validator.validate(plan) + if not v["valid"]: + raise _err("PLAN_INVALID", "plan validation failed", errors=v["errors"]) + wps = plan["work_packages"] + if len(wps) > self.max_wps_per_run: + raise _err("PLAN_INVALID", f"plan exceeds max_wps_per_run {self.max_wps_per_run}", + count=len(wps)) + + mission = self.missions.mission_create( + title=plan.get("title") or plan.get("mission_id") or "untitled", + goal=plan.get("goal"), + scope=plan.get("scope"), + constraints=plan.get("constraints"), + acceptance_criteria=plan.get("acceptance_criteria"), + mission_id=plan.get("mission_id"), + actor=actor, + ) + mid = mission["id"] + # WP-Metadaten registrieren (fuer Dispatch-Contracts) + for w in wps: + wid = str(w["wp_id"]) + self.missions.wp_create(mid, w.get("title") or wid, wp_id=wid, actor=actor) + self._wp_meta_cache[wid] = dict(w) + self._wp_meta_cache[wid]["id"] = wid + self._wp_meta_cache[wid]["mission_id"] = mid + # Dependencies (zweiter Durchlauf: alle WPs existieren) + for w in wps: + for dep in (w.get("dependencies") or []): + self.missions.wp_dependency(mid, str(w["wp_id"]), str(dep)) + # WPs in READY bringen (TODO->READY deterministisch via A2) + for w in wps: + try: + self.missions.wp_transition(str(w["wp_id"]), "READY", actor=actor, + evidence="plan validated") + except RqError: + pass # bereits READY + # CREATED->PLANNING->READY (deterministische A2-Transitions) + self.missions.mission_transition(mid, "PLANNING", actor=actor, evidence="plan accepted") + self.missions.mission_transition(mid, "READY", actor=actor, evidence="plan validated") + return {"status": ST_PLAN_VALIDATED, "mission": self.missions.mission_read(mid)} + + # -- State-Helfer ----------------------------------------------------------- + def _mission_state(self, mission_id: str) -> str: + try: + return self.missions.mission_state(mission_id)["state"] + except RqError: + return "UNKNOWN" + + def _wp_state(self, wp_id: str) -> str: + try: + return self.missions.wp_state(wp_id)["state"] + except RqError: + return "UNKNOWN" + + def _wp_deps_done(self, mission_id: str, wp_id: str) -> bool: + try: + m = self.missions.mission_read(mission_id) + except RqError: + return False + for d in m.get("dependencies", []): + if d["wp_id"] == wp_id: + if self._wp_state(d["depends_on"]) != "DONE": + return False + return True + + def _wp_meta(self, wp_id: str) -> Dict[str, Any]: + """Liefert Plan-Metadaten des WP; Fallback auf eine deterministische + Default-Struktur (Target aus WP-Titel abgeleitet).""" + cached = self._wp_meta_cache.get(wp_id) + if cached: + return cached + wp = self.missions.wp_read(wp_id) + title = (wp.get("title") or "").lower() + target = next((k for k in CRITICAL_TARGETS if k in title), "") + return { + "mission_id": wp["mission_id"], + "id": wp_id, + "title": wp.get("title"), + "goal": wp.get("title"), + "description": wp.get("title"), + "target_component": target, + "risk_classification": None, + "allowed_files": [], + "forbidden_files": [], + "acceptance_criteria": "WP accepted via deterministic review gate", + "test_requirements": "", + "safety_rules": ["NO_ROOT", "NO_SECRETS", "BOUNDED"], + "architecture_rules": [], + "impact": "", "rollback": "", "risk": "", + "requires_approval": False, + } + + # -- Consistency-Gate (§9, KRITISCH) -------------------------------------- + def _consistency_gate( + self, mission_id: str, wp_id: Optional[str] = None, + target_component: Optional[str] = None, + ) -> Dict[str, Any]: + """Vor JEDER mutierenden Aktion: A2-State + A3 check_safety_state + A3 evaluate. + + Rueckgabe {'ok': bool, 'status': str, 'mission_state': str, 'reason': str, ...} + Circuit OPEN / Safety-State-Inkonsistenz -> ok=False, NO MUTATION. + """ + ms = self._mission_state(mission_id) + if ms in ("COMPLETED", "FAILED", "CANCELLED", "BLOCKED", "PAUSED", "ESCALATED"): + return {"ok": False, "status": ST_TERMINAL, "mission_state": ms, + "reason": f"mission {ms}; cannot run"} + try: + self.safety.check_safety_state(mission_id, wp_id) + except SafetyError as e: + return {"ok": False, "status": ST_STATE_CONFLICT, "mission_state": ms, + "reason": f"safety state error: {e.message}", "reason_code": e.code} + dec = self.safety.evaluate_next_action( + mission_id, wp_id, target_component=target_component, + is_mutating=True, persist=False, + ) + if dec["DECISION"] == "BLOCK" or dec["CIRCUIT_STATE"] == CIRCUIT_OPEN: + return {"ok": False, "status": ST_MUTATION_BLOCKED, "mission_state": ms, + "reason": dec["ACTION_TEXT"], "reason_code": dec["REASON_CODE"], + "circuit": dec["CIRCUIT_STATE"]} + return {"ok": True, "status": "OK", "mission_state": ms} + + # -- Runnable-Selection (§8) ------------------------------------------------ + def select_runnable(self, mission_id: str) -> List[Dict[str, Any]]: + """WPs, die RUNNABLE sind: DEPENDENCIES DONE? STATE READY? SAFETY CLOSED? + SCOPE ALLOWED? -> nur dann RUNNABLE. Kein Polling/kein Wait.""" + try: + m = self.missions.mission_read(mission_id) + except RqError: + return [] + if m["state"] not in ("READY", "RUNNING"): + return [] + runnable = [] + for wp in m["work_packages"]: + if wp["state"] != "READY": + continue + if not self._wp_deps_done(mission_id, wp["id"]): + continue + # Consistency: Safety CLOSED + Scope erlaubt + cg = self._consistency_gate(mission_id, wp["id"], None) + if not cg["ok"]: + continue + runnable.append(wp) + runnable.sort(key=lambda x: x["id"]) + return runnable + + # -- EIN Schritt (explicit, bounded) ---------------------------------------- + def run_one_step(self, mission_id: str, actor: str = "red-queen") -> Dict[str, Any]: + """Fuehrt GENAU EINEN deterministischen Schritt aus und ENDET GARANTIERT. + + Wenn `child_dispatcher` injiziert ist, wird ein kompletter Maker(+Checker)- + Zyklus deterministisch abgearbeitet. Sonst liefert die Methode das + Dispatch-Contract (ST_NEEDS_DISPATCH / ST_NEEDS_CHECKER_DISPATCH), das + Red Queen selbst an `delegate_task` weitergibt. + """ + cg = self._consistency_gate(mission_id) + if not cg["ok"]: + return {"status": cg["status"], "mission_id": mission_id, **cg} + + runnable = self.select_runnable(mission_id) + if not runnable: + # Pending children? + m = self.missions.mission_read(mission_id) + pending = [w["id"] for w in m["work_packages"] if w["state"] in ("IN_PROGRESS", "CHECKING")] + if pending: + return {"status": ST_WAITING_PENDING, "mission_id": mission_id, "pending_wps": pending} + return {"status": ST_NO_RUNNABLE_WORK, "mission_id": mission_id} + + wp = runnable[0] + wp_meta = self._wp_meta(wp["id"]) + risk = classify_risk(wp_meta.get("target_component"), wp_meta.get("risk_classification")) + + # Approval-Gate (§25) + if requires_approval(risk, wp_meta): + return self._approval_block(wp["id"], risk, wp_meta) + + # Consistency vor Dispatch (scoped) + cg2 = self._consistency_gate(mission_id, wp["id"], wp_meta.get("target_component")) + if not cg2["ok"]: + return {"status": cg2["status"], "mission_id": mission_id, "wp_id": wp["id"], **cg2} + + # READY->IN_PROGRESS via A2 + try: + self.missions.wp_transition(wp["id"], "IN_PROGRESS", actor=actor, evidence="dispatch maker") + except RqError: + pass + # Mission READY->RUNNING (nur beim ersten echten Schritt) + if cg["mission_state"] == "READY": + try: + self.missions.mission_transition(mission_id, "RUNNING", actor=actor, + evidence="orchestrator started") + except RqError: + pass + mission = self.missions.mission_read(mission_id) + contract = build_maker_contract(wp_meta, mission, risk, retry_round=1) + self._make_record(wp_meta["mission_id"], wp["id"], ROLE_MAKER, contract["TASK_ID"], + _utcnow(), "dispatched", "PENDING") + + if self.child_dispatcher is None: + return {"status": ST_NEEDS_DISPATCH, "mission_id": mission_id, "wp_id": wp["id"], + "role": ROLE_MAKER, "dispatch": contract, "risk_class": risk} + + try: + child_result = self.child_dispatcher(contract) + except Exception as e: # noqa + return self._child_technical_failure(mission_id, wp["id"], f"maker dispatch: {e}", + contract["TASK_ID"]) + return self.apply_child_result(mission_id, wp["id"], child_result, contract=contract, + risk_class=risk, actor=actor) + + # -- Evidence-Record (§30) --------------------------------------------------- + def _make_record(self, mission_id, wp_id, role, task_id, start_ts, result, verdict) -> str: + rec = make_child_evidence(mission_id, wp_id, role, task_id, start_ts, result, verdict) + # persistiere als Safety-Evidence (redacted) + ref = self.safety.safety_evidence("child_dispatch", rec)["ref"] + return ref + + # -- Approval-Block ----------------------------------------------------------- + def _approval_block(self, wp_id: str, risk: str, wp_meta: dict) -> Dict[str, Any]: + try: + self.missions.wp_transition(wp_id, "BLOCKED", evidence="approval required") + except RqError: + pass + return { + "status": ST_APPROVAL_REQUIRED, + "mission_id": wp_meta.get("mission_id"), + "wp_id": wp_id, + "risk": risk, + "approval": { + "MISSION": wp_meta.get("mission_id"), + "WP": wp_id, + "REQUESTED_ACTION": redact_secret(wp_meta.get("goal") or wp_meta.get("description") or ""), + "WHY": "WP-Class kritisch/SELF_MOD verlangt Human-Approval (ROOT_SSH_GATE)", + "IMPACT": redact_secret(wp_meta.get("impact") or "change in critical target area"), + "ROLLBACK": redact_secret(wp_meta.get("rollback") or "restore from documented backup"), + "RISK": redact_secret(wp_meta.get("risk") or "critical"), + }, + } + + # -- Child-Result-Anwendung -------------------------------------------------- + def apply_child_result( + self, + mission_id: str, + wp_id: str, + child_result: Dict[str, Any], + *, + contract: Optional[Dict[str, Any]] = None, + risk_class: Optional[str] = None, + actor: str = "red-queen", + ) -> Dict[str, Any]: + """Deterministische Anwendung eines Child-Results. + + - Idempotenz: gleiche Child-Task-ID wird nie doppelt gewertet. + - technischer Fehlschlag -> CHILD_TECHNICAL_FAILURE, A3 gefragt, kein + implizit PASS, kein endlos spawn. + - Checker PASS -> WP DONE. Checker FAIL -> A3-Entscheidung + Repair-Flow. + """ + role = str(child_result.get("role") or child_result.get("ROLE") or "").upper() + task_id = str(child_result.get("child_task_id") or child_result.get("TASK_ID") or "").strip() + retry_round = 1 + if contract: + retry_round = int(contract.get("RETRY_ROUND") or 1) + + wstate = self._wp_state(wp_id) + if wstate in ("DONE", "FAILED"): + return {"status": ST_WP_DONE, "mission_id": mission_id, "wp_id": wp_id, + "idempotent": True, "reason": "wp already terminal"} + + cg = self._consistency_gate(mission_id, wp_id, None) + if not cg["ok"]: + return {"status": cg["status"], "mission_id": mission_id, "wp_id": wp_id, **cg} + + idem_key = task_id or f"{mission_id}:{wp_id}:{role}:{retry_round}" + wp_meta = self._wp_meta(wp_id) + risk = risk_class or classify_risk(wp_meta.get("target_component"), None) + + if role == ROLE_CHECKER: + return self._apply_checker(mission_id, wp_id, child_result, task_id, idem_key, + retry_round, risk, actor) + if role == ROLE_MAKER: + return self._apply_maker(mission_id, wp_id, child_result, task_id, idem_key, + retry_round, risk, actor) + return self._child_technical_failure(mission_id, wp_id, f"unknown role {role!r}", idem_key) + + def _apply_maker(self, mission_id, wp_id, result, task_id, idem_key, retry_round, risk, actor): + errors = result.get("errors") or result.get("FAIL") or result.get("fail") + test_results = result.get("test_results") or [] + has_fail = any( + (isinstance(t, dict) and (t.get("verdict") == "FAIL" or t.get("PASS/FAIL") == "FAIL")) + or t == "FAIL" + for t in test_results + ) if isinstance(test_results, list) else False + maker_result = result.get("result") or (RESULT_FAIL if (errors or has_fail) else RESULT_PASS) + progress = result.get("progress") or (PROGRESS_YES if maker_result == RESULT_PASS else PROGRESS_NO) + + self.safety.record_attempt( + mission_id, wp_id, actor="maker", result=maker_result, + error=result.get("error"), strategy=result.get("strategy"), + change=result.get("change"), progress=progress, idempotency_key=idem_key, + ) + try: + self.missions.wp_transition(wp_id, "CHECKING", actor=actor, evidence="maker result") + except RqError: + pass + + mission = self.missions.mission_read(mission_id) + checker_contract = build_checker_contract( + self._wp_meta(wp_id), mission, result, risk, retry_round) + self._make_record(mission_id, wp_id, ROLE_CHECKER, checker_contract["task_id"], + _utcnow(), "dispatched", "PENDING") + + if self.child_dispatcher is None: + return {"STATUS": ST_NEEDS_CHECKER_DISPATCH, "mission_id": mission_id, "wp_id": wp_id, + "dispatch": checker_contract, "risk": risk, "maker_result": maker_result} + try: + checker_result = self.child_dispatcher(checker_contract) + except Exception as e: # noqa + return self._child_technical_failure(mission_id, wp_id, f"checker error: {e}", + checker_contract["task_id"]) + return self.apply_child_result(mission_id, wp_id, checker_result, contract=checker_contract, + risk_class=risk, actor=actor) + + def _apply_checker(self, mission_id, wp_id, result, task_id, idem_key, + retry_round, risk, actor): + verdict = str(result.get("verdict") or result.get("VERDICT") or "BLOCKED").upper() + progress = result.get("progress") or (PROGRESS_YES if verdict == "PASS" else PROGRESS_UNKNOWN) + fail_reason = None if verdict == "PASS" else (result.get("fail_reason") or result.get("FAIL_REASON")) + self.safety.record_attempt( + mission_id, wp_id, actor="checker", + result=RESULT_PASS if verdict == "PASS" else RESULT_FAIL, + error=fail_reason, strategy="checker-review", change=None, + progress=progress, idempotency_key=idem_key, + ) + if verdict == "PASS": + self.missions.wp_complete(wp_id, result="checker PASS", actor=actor) + return {"status": ST_CHECKER_PASS, "mission_id": mission_id, "wp_id": wp_id, + "verdict": "PASS", "wp_state": "DONE"} + if verdict == "BLOCKED": + try: + self.missions.wp_transition(wp_id, "BLOCKED", actor=actor, evidence="checker blocked") + except RqError: + pass + return {"status": ST_CHECKER_BLOCKED, "mission_id": mission_id, "wp_id": wp_id, + "verdict": "BLOCKED"} + return self._checker_fail(mission_id, wp_id, result, retry_round, risk, actor) + + def _checker_fail(self, mission_id, wp_id, result, retry_round, risk, actor): + wp_meta = self._wp_meta(wp_id) + dec = self.safety.evaluate_next_action( + mission_id, wp_id, + error=result.get("fail_reason") or result.get("error"), + strategy_label=result.get("strategy"), + target_component=wp_meta.get("target_component"), + is_mutating=True, + ) + decision = dec["DECISION"] + reason = dec["REASON_CODE"] + + if decision == "CIRCUIT_BREAK": + self.safety.open_circuit("WORK_PACKAGE", wp_id, trigger="OSCILLATION_ABAB", + severity="CRITICAL", mission_id=mission_id, wp_id=wp_id, + reason="circuit break from safety decision") + try: + self.missions.wp_transition(wp_id, "BLOCKED", actor=actor, evidence="circuit break") + except RqError: + pass + return {"status": ST_MUTATION_BLOCKED, "mission_id": mission_id, "wp_id": wp_id, + "decision": decision, "reason_code": reason, "circuit": CIRCUIT_OPEN} + + if decision in ("SECOND_OPINION", "BLOCK"): + # kein auto-Rain: nur BLOCK/ESCALATE an Red Queen -> Christian + target = "ESCALATED" if decision == "SECOND_OPINION" else "BLOCKED" + try: + self.missions.wp_transition(wp_id, target, actor=actor, + evidence="repair limit / second opinion") + except RqError: + pass + return {"status": ST_SECOND_OPINION_REQUIRED, "mission_id": mission_id, "wp_id": wp_id, + "decision": decision, "reason": reason, "no_auto_rain": True} + + if decision == "DEBUG": + try: + self.missions.wp_transition(wp_id, "IN_PROGRESS", actor=actor, evidence="debug required") + except RqError: + pass + return {"status": ST_DEBUG_REQUIRED, "mission_id": mission_id, "wp_id": wp_id, + "decision": "DEBUG", "reason": reason, + "recommendation": "A5 debugger / strategy change required"} + + # RETRY -> Repair + repairs = self._repair_count(mission_id, wp_id) + if repairs >= self.max_repair_cycles: + return {"status": ST_RETRY4_IMPOSSIBLE, "mission_id": mission_id, "wp_id": wp_id, + "decision": "RETRY_DENIED", "reason": "RETRY_LIMIT", + "repair_count": repairs, "max": self.max_repair_cycles} + try: + self.missions.wp_transition(wp_id, "IN_PROGRESS", actor=actor, evidence="repair retry") + except RqError: + pass + next_round = repairs + 1 + mission = self.missions.mission_read(mission_id) + contract = build_maker_contract(wp_meta, mission, + risk or classify_risk(wp_meta.get("target_component"), None), + retry_round=next_round) + # RETRY -> naechsten Maker bereitstellen (EIN Schritt, bounded). + # WICHTIG: Es wird NICHT rekursiv ein weiterer Maker/Checker gestartet. + # run_one_step = genau EIN Child-Zyklus. Der naechste Maker wird als + # Dispatch-Contract zurueckgegeben, den Red Queen (bzw. der Aufrufer) + # ausfuehrt. So endet jeder Schritt garantiert (kein Rekursions-Loop). + return {"status": ST_NEEDS_DISPATCH, "mission_id": mission_id, "wp_id": wp_id, + "dispatch": contract, "retry_round": next_round, "decision": "RETRY"} + + def _repair_count(self, mission_id, wp_id) -> int: + attempts = self.safety.attempts(mission_id, wp_id) + # Ein Repair-Round ist eine MAKER+CHECKER-Runde, die in FAIL endet. + # Zaehle deshalb sowohl fehlgeschlagene Maker- als auch fehlgeschlagene + # Checker-Attempts (Determinismus/SAFETY_CONTRACT §2.4). Nur so ist der + # Retry-Kreislauf auch dann begrenzt, wenn der Maker "PASS", der Checker + # aber wiederholt "FAIL" liefert (sonst waere die Schleife unbegrenzt). + return sum( + 1 for a in attempts + if (a.get("actor") or "").lower() in ("maker", "repair", "checker") + and a.get("result") == RESULT_FAIL + ) + + def _child_technical_failure(self, mission_id, wp_id, reason, idem_key) -> Dict[str, Any]: + try: + self.safety.record_attempt(mission_id, wp_id, actor="orchestrator", result=RESULT_FAIL, + error=reason, strategy="dispatch", progress=PROGRESS_NO, + idempotency_key=idem_key) + except SafetyError: + pass + try: + dec = self.safety.evaluate_next_action(mission_id, wp_id, error=reason, is_mutating=True) + except SafetyError: + dec = {"DECISION": "BLOCK", "REASON_CODE": "STATE_INCONSISTENT"} + return {"status": ST_CHILD_TECHNICAL_FAILURE, "mission_id": mission_id, "wp_id": wp_id, + "reason": redact_secret(reason), "decision": dec["DECISION"], + "reason_code": dec["REASON_CODE"], "no_implicit_pass": True} + + # -- Completion-Gate (§20) --------------------------------------------------- + def mission_completion_gate(self, mission_id: str, *, final_review_pass: bool = False) -> Dict[str, Any]: + """Mission nur REVIEW/COMPLETED, wenn alle Pflicht-WPs DONE, keine BLOCKED, + keine relevanten Circuits OPEN, keine unresolved critical findings, AC + erfuellt, Working-State konsistent. Vor COMPLETED: Final-Review-Gate (PASS).""" + reasons: List[str] = [] + try: + m = self.missions.mission_read(mission_id) + except RqError as e: + return {"mission_id": mission_id, "valid": False, "reasons": [e.message]} + wps = m["work_packages"] + if not wps: + reasons.append("no work packages") + not_done = [w["id"] for w in wps if w["state"] != "DONE"] + if not_done: + reasons.append(f"mandatory wps not done: {not_done}") + blocked = [w["id"] for w in wps if w["state"] == "BLOCKED"] + if blocked: + reasons.append(f"blocked wps: {blocked}") + try: + self.safety.check_safety_state(mission_id) + except SafetyError as e: + reasons.append(f"safety inconsistent: {e.message}") + dec = self.safety.evaluate_next_action(mission_id, None, is_mutating=False, persist=False) + if dec["CIRCUIT_STATE"] == CIRCUIT_OPEN: + reasons.append(f"circuit open: {dec['REASON_CODE']}") + if not final_review_pass: + reasons.append("final review gate not passed") + return {"mission_id": mission_id, "valid": not reasons, "reasons": reasons} + + def attempt_mission_complete(self, mission_id: str, *, final_review_pass: bool = False, + actor: str = "red-queen") -> Dict[str, Any]: + """Mission RUNNING->REVIEW->COMPLETED, nur wenn Completion-Gate ok.""" + gate = self.mission_completion_gate(mission_id, final_review_pass=final_review_pass) + if not gate["valid"]: + return {"status": ST_MISSION_NOT_COMPLETED, "mission_id": mission_id, + "valid": False, "reasons": gate["reasons"]} + ms = self._mission_state(mission_id) + if ms != "REVIEW": + try: + self.missions.mission_transition(mission_id, "REVIEW", actor=actor, + evidence="all wps done; review") + except RqError: + pass + try: + self.missions.mission_complete(mission_id, actor=actor, evidence="final review passed") + except RqError as e: + return {"status": ST_MISSION_NOT_COMPLETED, "mission_id": mission_id, + "valid": False, "reasons": [e.message]} + return {"status": ST_MISSION_COMPLETED, "mission_id": mission_id, "valid": True} + + # -- Bounded Run ------------------------------------------------------------- + def run_bounded(self, mission_id: str, *, max_steps: Optional[int] = None, + actor: str = "red-queen") -> Dict[str, Any]: + """EXPLICIT begrenzter Run. Nach max_steps/max_bounded_runtime -> STOP mit + ST_BUDGET_REACHED. Kein self-reschedule; endet garantiert.""" + budget = min(max_steps or self.max_bounded_runtime, self.max_bounded_runtime) + last = {"status": ST_NO_RUNNABLE_WORK, "mission_id": mission_id} + for i in range(1, budget + 1): + last = self.run_one_step(mission_id, actor=actor) + st = last.get("status") + if st in _TERMINAL_STEPS or st in _DISPATCH_CODES: + last["steps"] = i + return last + last["steps"] = budget + last["status"] = ST_BUDGET_REACHED + return last diff --git a/a4/rq_orchestrator_cli.py b/a4/rq_orchestrator_cli.py new file mode 100644 index 0000000..517beec --- /dev/null +++ b/a4/rq_orchestrator_cli.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +""" +Red Queen — A4: BOUNDED ORCHESTRATOR v1 — CLI. + +Deterministischer, INAKTIVER CLI-Zugriff auf den A4-Orchestrator. +Diese CLI startet KEINEN Loop, keinen Daemon, kein Heartbeat, kein Cron und +keinen autonomen Run. Jeder Befehl fuehrt EINE explizite, begrenzte Aktion aus +und endet garantiert. Der eigentliche Child-Versand (Maker/Checker) bleibt bei +Red Queen: Ohne injizierten Dispatcher liefert die CLI Dispatch-Contracts +(ST_NEEDS_DISPATCH) zurueck, die Red Queen an `delegate_task` weitergibt. + +Exit-Codes: + 0 = Erfolg + 2 = Fehler (OrchestratorError / RqError / SafetyError), maschinenlesbar via --json + 1 = unbekanntes Kommando / CLI-Aufruffehler + +DB-Pfade (Umgebung, deterministisch): + RQ_MISSIONS_DB (Default: missions.db) + RQ_SAFETY_DB (Default: safety.db) + +Hinweis: Produktive DBs (z.B. /opt/data/kanban.db) werden NIE angetastet. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path + +_REPO_ROOT = Path(__file__).resolve().parent.parent +for _p in (str(_REPO_ROOT), str(_REPO_ROOT / "a4")): + if _p not in sys.path: + sys.path.insert(0, _p) + +import rq_orchestrator as o # noqa: E402 +from rq_orchestrator import Orchestrator, OrchestratorError # noqa: E402 + + +def _emit(obj, as_json: bool) -> None: + if as_json: + print(json.dumps(obj, indent=2, ensure_ascii=False, default=str)) + else: + if isinstance(obj, dict) and "id" in obj: + print(obj["id"]) + elif isinstance(obj, dict) and "status" in obj: + print(obj["status"]) + else: + print(obj) + + +def _fail(exc, as_json: bool) -> int: + if hasattr(exc, "to_dict"): + _emit(exc.to_dict(), as_json) + else: + _emit({"code": "CLI_ERROR", "message": str(exc)}, as_json) + return 2 + + +def _build(args) -> Orchestrator: + mission_db = args.missions_db or os.environ.get("RQ_MISSIONS_DB", "missions.db") + safety_db = args.safety_db or os.environ.get("RQ_SAFETY_DB", "safety.db") + return Orchestrator(mission_db, safety_db) + + +def cmd_load_plan(args, as_json): + if not args.plan: + raise OrchestratorError("INVALID_ARGS", "plan required (--plan )") + raw = args.plan + if Path(raw).exists(): + plan = json.loads(Path(raw).read_text(encoding="utf-8")) + else: + plan = json.loads(raw) + orch = _build(args) + res = orch.load_plan(plan, actor=args.actor) + _emit(res, as_json) + return 0 + + +def cmd_run_one(args, as_json): + orch = _build(args) + res = orch.run_one_step(args.mission, actor=args.actor) + _emit(res, as_json) + return 0 + + +def cmd_run_bounded(args, as_json): + orch = _build(args) + res = orch.run_bounded(args.mission, max_steps=args.max_steps, actor=args.actor) + _emit(res, as_json) + return 0 + + +def cmd_complete(args, as_json): + orch = _build(args) + res = orch.attempt_mission_complete(args.mission, final_review_pass=args.final_review, + actor=args.actor) + _emit(res, as_json) + return 0 + + +def cmd_status(args, as_json): + orch = _build(args) + mission = orch.missions.mission_read(args.mission) + _emit(mission, as_json) + return 0 + + +def cmd_evaluate(args, as_json): + orch = _build(args) + dec = orch.safety.evaluate_next_action(args.mission, args.wp, error=args.error, + strategy_label=args.strategy, + is_mutating=not args.read_only) + _emit(dec, as_json) + return 0 + + +def cmd_approve(args, as_json): + """Christian/Human-Approval: schliesst einen offenen Circuit (Human-Gate) via A3.""" + orch = _build(args) + res = orch.safety.close_circuit(args.scope_type, args.scope_id, approved_by=args.approved_by, + gate=args.gate, cause=args.cause, + recovery_evidence=args.recovery_evidence) + _emit(res, as_json) + return 0 + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(prog="rq_orchestrator_cli", description=__doc__) + parser.add_argument("--missions-db", default=None, help="Pfad zur A2-missions.db (default $RQ_MISSIONS_DB/missions.db)") + parser.add_argument("--safety-db", default=None, help="Pfad zur A3-safety.db (default $RQ_SAFETY_DB/safety.db)") + parser.add_argument("--json", dest="as_json", action="store_true", help="maschinenlesbare JSON-Ausgabe") + sub = parser.add_subparsers(dest="command", required=True) + + p = sub.add_parser("plan", help="Plan validieren + materialisieren (A2), keine Ausfuehrung") + p.add_argument("--plan", required=True, help="Pfad oder Inline-JSON des Plans") + p.add_argument("--actor", default="red-queen") + p.set_defaults(func=cmd_load_plan) + + p = sub.add_parser("run-one-step", help="EIN deterministischer Schritt (bounded)") + p.add_argument("mission") + p.add_argument("--actor", default="red-queen") + p.set_defaults(func=cmd_run_one) + + p = sub.add_parser("run-bounded", help="EXPLICIT begrenzter Run (garantiert endend)") + p.add_argument("mission") + p.add_argument("--max-steps", type=int, default=None) + p.add_argument("--actor", default="red-queen") + p.set_defaults(func=cmd_run_bounded) + + p = sub.add_parser("complete", help="Completion-Gate + Mission COMPLETED") + p.add_argument("mission") + p.add_argument("--final-review", action="store_true", help="Final-Review-Gate bestanden") + p.add_argument("--actor", default="red-queen") + p.set_defaults(func=cmd_complete) + + p = sub.add_parser("status", help="Mission-State lesen (A2)") + p.add_argument("mission") + p.set_defaults(func=cmd_status) + + p = sub.add_parser("evaluate", help="A3-Safety-Decision (verbindlich)") + p.add_argument("mission") + p.add_argument("--wp", default=None) + p.add_argument("--error", default=None) + p.add_argument("--strategy", default=None) + p.add_argument("--read-only", action="store_true") + p.set_defaults(func=cmd_evaluate) + + p = sub.add_parser("approve", help="Approval (Human-Gate) fuer Circuit-Reset via A3") + p.add_argument("circuit_type", metavar="SCOPE_TYPE") + p.add_argument("circuit_id", metavar="SCOPE_ID") + p.add_argument("--approved-by", default="human") + p.add_argument("--gate", default="human_gate") + p.add_argument("--cause", default=None) + p.add_argument("--recovery-evidence", default=None) + p.set_defaults(func=cmd_approve) + + args = parser.parse_args(argv) + try: + return args.func(args, args.as_json) + except (OrchestratorError, Exception) as e: # noqa: BLE001 + # RqError / SafetyError / OrchestratorError -> Exit 2 (maschinenlesbar) + return _fail(e, args.as_json) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/a4/test_a4.py b/a4/test_a4.py new file mode 100644 index 0000000..c4ac0fd --- /dev/null +++ b/a4/test_a4.py @@ -0,0 +1,670 @@ +#!/usr/bin/env python3 +""" +Red Queen — A4 Testsuite (deterministisch, isoliert). + +Nutzt ausschliesslich temporaere DBs (tempfile.mkdtemp). Kein Netz, kein hermes, +keine echte delegate_task. Der Child-Versand wird ueber eine simulierte +`ChildDispatcher`-Abstraktion (Callable) injiziert — die Library selbst startet +nichts und laeuft nie zyklisch. + +Lauf: + python3 test_a4.py +Exit-Code 0 = alle Tests gruen; 1 = mindestens ein Fehler. + +Deckt A4 §6-§33 ab: plan validation, dependency validation, runnable selection, +A2/A3-Consistency-Gate, maker/checker dispatch contract, checker pass/fail, +retry allowed/denied, retry4 impossible, circuit open blocks maker, approval gate, +risk overrides size, no runnable stop, mission completion gate, final review gate, +idempotent WP execution, restart/resume state, child failure, reason codes, +secret-safe logging. +""" + +from __future__ import annotations + +import json +import os +import sys +import tempfile +from pathlib import Path + +_HERE = Path(__file__).resolve().parent +_REPO = _HERE.parent +for _p in (str(_REPO), str(_HERE)): + if _p not in sys.path: + sys.path.insert(0, _p) + +import rq_orchestrator as o # noqa: E402 +from rq_orchestrator import ( # noqa: E402 + CRITICAL_TARGETS, + ROLE_CHECKER, + ROLE_MAKER, + ST_APPROVAL_REQUIRED, + ST_CHILD_TECHNICAL_FAILURE, + ST_CHECKER_BLOCKED, + ST_DEBUG_REQUIRED, + ST_MISSION_COMPLETED, + ST_MISSION_NOT_COMPLETED, + ST_NEEDS_CHECKER_DISPATCH, + ST_NEEDS_DISPATCH, + ST_NO_RUNNABLE_WORK, + ST_RETRY4_IMPOSSIBLE, + ST_SECOND_OPINION_REQUIRED, + ST_TERMINAL, + ST_WAITING_PENDING, + Orchestrator, + OrchestratorError, + PlanValidator, + build_checker_contract, + build_maker_contract, + classify_risk, + make_child_evidence, +) + +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, fn, code, fragment=""): + try: + fn() + except OrchestratorError 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(): + d = tempfile.mkdtemp(prefix="a4test_") + mission_db = str(Path(d) / "missions.db") + safety_db = str(Path(d) / "safety.db") + return d, mission_db, safety_db + + +def default_plan(wps=None, mission_id="RQ-A4-1"): + if wps is None: + wps = [ + { + "wp_id": "W1", + "title": "W1", + "goal": "Implementiere Feature X", + "description": "Feature X implementieren", + "scope": "a4/w1.py", + "target_component": "feature", + "acceptance_criteria": "Tests gruen", + "test_requirements": "pytest", + }, + ] + return { + "mission_id": mission_id, + "title": "A4-Test-Mission", + "goal": "Test-Mission fuer A4", + "scope": "tests", + "work_packages": wps, + } + + +def make_orch(dispatcher=None, mission_db=None, safety_db=None, **kw): + if mission_db is None or safety_db is None: + _, mission_db, safety_db = fresh() + return Orchestrator(mission_db, safety_db, dispatcher, **kw) + + +# --------------------------------------------------------------------------- # +# 1) plan_validation +# --------------------------------------------------------------------------- # +def test_plan_validation(): + print("\n== plan validation ==") + orch = make_orch() + + valid_plan = default_plan() + res = orch.load_plan(valid_plan) + check("valid plan accepted", res["status"] == o.ST_PLAN_VALIDATED, res) + + v = PlanValidator() + # cycle + cyc = default_plan(wps=[ + {"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["B"]}, + {"wp_id": "B", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["A"]}, + ]) + r = v.validate(cyc) + check("cycle rejected", not r["valid"] and any(e["code"] == "DEPENDENCY_CYCLE" for e in r["errors"]), r) + + # self-dep + sd = default_plan([{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac", + "dependencies": ["A"]}]) + r = v.validate(sd) + check("self-dep rejected", not r["valid"] and any(e["code"] == "SELF_DEPENDENCY" for e in r["errors"]), r) + + # duplicate id + dup = default_plan([ + {"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac"}, + {"wp_id": "A", "goal": "a2", "scope": "s", "acceptance_criteria": "ac"}, + ]) + r = v.validate(dup) + check("duplicate id rejected", not r["valid"] and any(e["code"] == "DUPLICATE_WP_ID" for e in r["errors"]), r) + + # unknown target (unknown dependency) + unk = default_plan([{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac", + "dependencies": ["NOPE"]}]) + r = v.validate(unk) + check("unknown dependency rejected", not r["valid"] and any(e["code"] == "UNKNOWN_DEPENDENCY" for e in r["errors"]), r) + + # missing AC + noac = default_plan([{"wp_id": "A", "goal": "a", "scope": "s"}]) + r = v.validate(noac) + check("missing AC rejected", not r["valid"] and any(e["code"] == "MISSING_AC" for e in r["errors"]), r) + + # invalid wp state + badstate = default_plan([{"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac", + "state": "IN_PROGRESS"}]) + r = v.validate(badstate) + check("invalid wp state rejected", not r["valid"] and any(e["code"] == "INVALID_WP_STATE" for e in r["errors"]), r) + + # foreign wp + fw = default_plan([{"wp_id": "A", "mission_id": "OTHER", "goal": "a", "scope": "s", "acceptance_criteria": "ac"}]) + v2 = PlanValidator(mission_id="RQ-A4-1") + r = v2.validate(fw) + check("foreign wp rejected", not r["valid"] and any(e["code"] == "FOREIGN_WP" for e in r["errors"]), r) + + +# --------------------------------------------------------------------------- # +# 2) dependency_validation +# --------------------------------------------------------------------------- # +def test_dependency_validation(): + print("\n== dependency validation ==") + orch, mission_db, safety_db = make_orch_with_paths() + plan = default_plan([ + {"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac"}, + {"wp_id": "B", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["A"]}, + {"wp_id": "C", "goal": "c", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["B"]}, + ]) + res = orch.load_plan(plan) + mid = res["mission"]["id"] + + # B nicht runnable bevor A DONE; C nicht vor B + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + runnable = orch.select_runnable(mid) + check("only A runnable initially", [w["id"] for w in runnable] == ["A"], runnable) + # A DONE -> B runnable + orch.missions.wp_transition("A", "IN_PROGRESS") + orch.missions.wp_transition("A", "CHECKING") + orch.missions.wp_complete("A") + runnable = orch.select_runnable(mid) + check("B runnable after A done", [w["id"] for w in runnable] == ["B"], runnable) + orch.missions.wp_transition("B", "IN_PROGRESS") + orch.missions.wp_transition("B", "CHECKING") + orch.missions.wp_complete("B") + runnable = orch.select_runnable(mid) + check("C runnable after B done", [w["id"] for w in runnable] == ["C"], runnable) + + +# --------------------------------------------------------------------------- # +# 3) runnable_selection +# --------------------------------------------------------------------------- # +def test_runnable_selection(): + print("\n== runnable selection ==") + orch, _, _ = make_orch_with_paths() + plan = default_plan([ + {"wp_id": "W1", "goal": "a", "scope": "s", "acceptance_criteria": "ac"}, + {"wp_id": "W2", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["W1"]}, + ]) + res = orch.load_plan(plan) + mid = res["mission"]["id"] + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + # W1 READY + deps done + safety closed -> runnable + runnable = orch.select_runnable(mid) + check("W1 runnable", [w["id"] for w in runnable] == ["W1"], runnable) + # W2 blocked by dep + check("W2 not runnable (dep open)", all(w["id"] != "W2" for w in runnable)) + # safety circuit open blocks W1 + orch.safety.open_circuit("MISSION", mid, trigger="REQ-1", severity="HIGH", mission_id=mid) + runnable = orch.select_runnable(mid) + check("no runnable with circuit open", runnable == [], runnable) + + +# --------------------------------------------------------------------------- # +# 4) a2a3_consistency +# --------------------------------------------------------------------------- # +def test_a2a3_consistency(): + print("\n== a2/a3 consistency gate ==") + orch, _, _ = make_orch_with_paths() + orch.load_plan(default_plan()) + mid = "RQ-A4-1" + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + # Circuit OPEN while Mission RUNNING -> mutation blocked + orch.safety.open_circuit("MISSION", mid, trigger="CRIT-1", severity="CRITICAL", mission_id=mid) + step = orch.run_one_step(mid) + check("mutation blocked on open circuit", step["status"] == o.ST_MUTATION_BLOCKED, step) + # kein WP wurde dispatched + st = orch._wp_state("W1") + check("no mutation, W1 still READY", st == "READY", st) + # corrupted safety state -> fail-closed (ORCHESTRATOR_STATE_CONFLICT) + orch2, _, _ = make_orch_with_paths() + orch2.load_plan(default_plan()) + # korrupten Circuit-Zustand injizieren (via open_circuit + manuelles UPDATE) + orch2.safety.open_circuit("MISSION", "RQ-A4-1", trigger="t", severity="HIGH", mission_id="RQ-A4-1") + import sqlite3 + with sqlite3.connect(orch2.safety_db) as c: + c.execute("UPDATE circuit_state SET state='GARBAGE' WHERE scope_type='MISSION' AND scope_id='RQ-A4-1'") + step2 = orch2.run_one_step("RQ-A4-1") + check("state conflict on corrupt safety", step2["status"] in (o.ST_STATE_CONFLICT, o.ST_MUTATION_BLOCKED), step2) + + +# --------------------------------------------------------------------------- # +# 5) maker_dispatch_contract +# --------------------------------------------------------------------------- # +def test_maker_dispatch_contract(): + print("\n== maker dispatch contract ==") + orch, _, _ = make_orch_with_paths() + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + wp = orch._wp_meta("W1") + mission = orch.missions.mission_read(mid) + c = build_maker_contract(wp, mission, "MEDIUM", retry_round=1) + for k in ("MISSION_ID", "WP_ID", "GOAL", "SCOPE", "ALLOWED_FILES", "FORBIDDEN_FILES", + "ACCEPTANCE_CRITERIA", "TEST_REQUIREMENTS", "SAFETY_RULES"): + check(f"maker contract has {k}", k in c, c) + check("maker contract role", c.get("role") == ROLE_MAKER) + check("maker contract task id", c.get("TASK_ID") == f"{mid}:W1:MAKER:R1", c) + + +# --------------------------------------------------------------------------- # +# 6) checker_dispatch_contract +# --------------------------------------------------------------------------- # +def test_checker_dispatch_contract(): + print("\n== checker dispatch contract ==") + orch, *_ = make_orch_with_paths() + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + wp = orch._wp_meta("W1") + mission = orch.missions.mission_read(mid) + maker_ev = {"test_results": [{"verdict": "PASS"}], "IMPLEMENTATION_SUMMARY": "SECRET ARG", + "strategy": "t"} + c = build_checker_contract(wp, mission, maker_ev, "MEDIUM", retry_round=1) + for k in ("requirement", "acceptance_criteria", "diff_files", "test_results", "safety_rules"): + check(f"checker contract has {k}", k in c) + check("checker no maker argumentation", "IMPLEMENTATION_SUMMARY" not in json.dumps(c), + json.dumps(c)) + check("checker role", c["role"] == ROLE_CHECKER) + + +# --------------------------------------------------------------------------- # +# 7) checker_pass +# --------------------------------------------------------------------------- # +def test_checker_pass(): + print("\n== checker pass ==") + d, mdb, sdb = fresh() + orch = Orchestrator(mdb, sdb, _pass_dispatcher()) + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + step = orch.run_one_step(mid) + check("mission completes via checker pass", step["status"] in o.ST_CHECKER_PASS, step) + check("wp done after checker pass", orch._wp_state("W1") == "DONE") + + +# --------------------------------------------------------------------------- # +# 8) checker_fail +# --------------------------------------------------------------------------- # +def test_checker_fail(): + print("\n== checker fail ==") + d, mdb, sdb = fresh() + orch = Orchestrator(mdb, sdb, fail_dispatcher()) + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + step = orch.run_one_step(mid) + # 1. FAIL -> RETRY (messbarer Fortschritt), neuer Maker-Dispatch notiert + check("first fail -> retry allowed", step["status"] in (ST_NEEDS_DISPATCH, ST_CHECKER_BLOCKED, + ST_SECOND_OPINION_REQUIRED, ST_DEBUG_REQUIRED), step) + check("attempt recorded", orch.safety.attempts_count(mid) >= 1) + check("decision retry", step.get("decision") == "RETRY", step) + + +# --------------------------------------------------------------------------- # +# 9) retry_allowed +# --------------------------------------------------------------------------- # +def test_retry_allowed(): + print("\n== retry allowed ==") + d, mdb, sdb = fresh() + orch = Orchestrator(mdb, sdb, fail_dispatcher()) + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + step = orch.run_one_step(mid) + check("1 fail -> retry allowed", step["status"] in (ST_NEEDS_DISPATCH, ST_SECOND_OPINION_REQUIRED, + ST_DEBUG_REQUIRED), step) + check("attempt in ledger", len(orch.safety.attempts(mid, "W1")) >= 1) + + +# --------------------------------------------------------------------------- # +# 10) retry_denied / 11) retry4_impossible +# --------------------------------------------------------------------------- # +def test_retry_denied_and_4_impossible(): + print("\n== retry denied / retry4 impossible ==") + d, mdb, sdb = fresh() + orch = Orchestrator(mdb, sdb, fail_dispatcher()) + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + # Pre-record 3 maker FAIL attempts (unique errors) to exhaust repair budget + for i in range(3): + orch.safety.record_attempt(mid, "W1", actor="maker", result="FAIL", + error=f"e{i}", strategy=f"s{i}", progress="NO") + step = orch.run_one_step(mid) + check("3 fails -> no 4th maker", step["status"] in (o.ST_RETRY4_IMPOSSIBLE, o.ST_SECOND_OPINION_REQUIRED), step) + if step["status"] == o.ST_RETRY4_IMPOSSIBLE: + check("repair count exceeded", step["repair_count"] >= 3, step) + check("decision not RETRY", step.get("decision") != "RETRY", step) + # ensure no maker dispatched after limit + check("no dispatch after limit", step.get("dispatch") is None, step) + + +# --------------------------------------------------------------------------- # +# 12) circuit_open_blocks_maker +# --------------------------------------------------------------------------- # +def test_circuit_open_blocks_maker(): + print("\n== circuit open blocks maker ==") + orch, _, _ = make_orch_with_paths() + orch.load_plan(default_plan()) + mid = "RQ-A4-1" + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + orch.safety.open_circuit("MISSION", mid, trigger="PERSISTENCE_CORRUPTED", severity="CRITICAL", mission_id=mid) + step = orch.run_one_step(mid) + check("maker dispatch blocked by open circuit", step["status"] == o.ST_MUTATION_BLOCKED, step) + + +# --------------------------------------------------------------------------- # +# 13) approval_gate +# --------------------------------------------------------------------------- # +def test_approval_gate(): + print("\n== approval gate ==") + orch, *_ = make_orch_with_paths() + plan = default_plan([ + {"wp_id": "W1", "goal": "a", "scope": "s", "acceptance_criteria": "ac", + "target_component": "auth", "requires_approval": True}, + ]) + res = orch.load_plan(plan) + mid = res["mission"]["id"] + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + step = orch.run_one_step(mid) + check("approval required", step["status"] == o.ST_APPROVAL_REQUIRED, step) + ap = step.get("approval", {}) + for k in ("MISSION", "WP", "REQUESTED_ACTION", "WHY", "IMPACT", "ROLLBACK", "RISK"): + check(f"approval has {k}", k in ap, ap) + check("no mutation (approval)", orch._wp_state("W1") in ("BLOCKED", "READY"), orch._wp_state("W1")) + + +# --------------------------------------------------------------------------- # +# 14) risk_overrides_size +# --------------------------------------------------------------------------- # +def test_risk_overrides_size(): + print("\n== risk overrides size ==") + # SMALL size but critical target -> CRITICAL + r = classify_risk("ssh", "SMALL") + check("ssh overrides SMALL -> CRITICAL", r == "CRITICAL", r) + r = classify_risk("firewall", "SMALL") + check("firewall overrides SMALL -> CRITICAL", r == "CRITICAL", r) + r = classify_risk("auth", "SMALL") + check("auth overrides SMALL -> CRITICAL", r == "CRITICAL", r) + r = classify_risk("docs", "SMALL") + check("docs stays SMALL", r == "SMALL", r) + r = classify_risk("feature", "LARGE") + check("feature LARGE stays LARGE", r == "LARGE", r) + r = classify_risk("feature") + check("feature default MEDIUM", r == "MEDIUM", r) + + +# --------------------------------------------------------------------------- # +# 15) no_runnable_stop +# --------------------------------------------------------------------------- # +def test_no_runnable_stop(): + print("\n== no runnable stop ==") + orch, *_ = make_orch_with_paths() + plan = default_plan([ + {"wp_id": "A", "goal": "a", "scope": "s", "acceptance_criteria": "ac"}, + {"wp_id": "B", "goal": "b", "scope": "s", "acceptance_criteria": "ac", "dependencies": ["A"]}, + ]) + res = orch.load_plan(plan) + mid = res["mission"]["id"] + # leave A in READY (not running), B blocked by dep + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + step = orch.run_one_step(mid) + # A is runnable -> would dispatch. To test no_runnable, set A to BLOCKED + orch.missions.wp_transition("A", "IN_PROGRESS") + orch.missions.wp_transition("A", "BLOCKED") + step2 = orch.run_one_step(mid) + check("no runnable -> stop", step2["status"] in (o.ST_NO_RUNNABLE_WORK, o.ST_WAITING_PENDING), step2) + + +# --------------------------------------------------------------------------- # +# 16) mission_completion_gate / 17) final_review_gate +# --------------------------------------------------------------------------- # +def test_mission_completion_gate(): + print("\n== mission completion gate ==") + orch, *_ = make_orch_with_paths() + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + # not all done + no review -> denied + g = orch.mission_completion_gate(mid, final_review_pass=True) + check("gate blocks incomplete mission", not g["valid"], g) + # mark wp done + review + orch.missions.wp_transition("W1", "IN_PROGRESS") + orch.missions.wp_transition("W1", "CHECKING") + orch.missions.wp_complete("W1") + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + # without final review -> denied + g = orch.mission_completion_gate(mid, final_review_pass=False) + check("final review gate blocks", not g["valid"] and any("review" in r for r in g["reasons"]), g) + # with final review -> complete + res = orch.attempt_mission_complete(mid, final_review_pass=True) + check("mission completed", res["status"] == o.ST_MISSION_COMPLETED, res) + check("mission state COMPLETED", orch._mission_state(mid) == "COMPLETED") + + +# --------------------------------------------------------------------------- # +# 18) idempotent_wp +# --------------------------------------------------------------------------- # +def test_idempotent_wp(): + print("\n== idempotent WP ==") + calls = [] + def disp(contract): + calls.append(contract["TASK_ID"]) + if contract["role"] == ROLE_MAKER: + return {"role": ROLE_MAKER, "TASK_ID": contract["TASK_ID"], "result": "PASS", + "test_results": [{"verdict": "PASS"}]} + return {"role": ROLE_CHECKER, "verdict": "PASS", "progress": "YES"} + d, mdb, sdb = fresh() + orch = Orchestrator(mdb, sdb, disp) + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + orch.run_one_step(mid) + orch.run_one_step(mid) # repeat after DONE + check("wp DONE idempotent (no 2nd maker)", sum(1 for t in calls if ":MAKER:" in t) == 1, calls) + check("attempts dedup", orch.safety.attempts_count(mid) == 2, orch.safety.attempts_count(mid)) # 1 maker + 1 checker + + +# --------------------------------------------------------------------------- # +# 19) restart_resume +# --------------------------------------------------------------------------- # +def test_restart_resume(): + print("\n== restart/resume ==") + d, mdb, sdb = fresh() + # phase 1 + orch1 = Orchestrator(mdb, sdb) + res = orch1.load_plan(default_plan()) + mid = res["mission"]["id"] + # mark wp IN_PROGRESS + record attempt (simulate partial) + orch1.missions.wp_transition("W1", "IN_PROGRESS") + orch1.safety.record_attempt(mid, "W1", actor="maker", result="FAIL", error="crash", strategy="s1") + # "crash": new objects on same DBs + orch2 = Orchestrator(mdb, sdb) + check("mission state persisted", orch2._mission_state(mid) == "RUNNING" or orch2._mission_state(mid) == "READY") + check("attempt persisted", orch2.safety.attempts_count(mid) == 1) + check("wp state persisted", orch2._wp_state("W1") == "IN_PROGRESS") + check("no double run (WPA already IN_PROGRESS)", orch2.run_one_step(mid)["status"] in (o.ST_WAITING_PENDING, o.ST_NO_RUNNABLE_WORK), orch2.run_one_step(mid)) + + +# --------------------------------------------------------------------------- # +# 20) child_failure +# --------------------------------------------------------------------------- # +def test_child_failure(): + print("\n== child technical failure ==") + def crash(contract): + raise RuntimeError("child crashed") + orch = Orchestrator(fresh()[1], fresh()[2], crash) + res = orch.load_plan(default_plan()) + mid = res["mission"]["id"] + orch.missions.mission_transition(mid, "RUNNING", evidence="t") + step = orch.run_one_step(mid) + check("technical failure", step["status"] == o.ST_CHILD_TECHNICAL_FAILURE, step) + check("no implicit pass", step.get("no_implicit_pass") is True, step) + check("no endless spawn", orch.safety.attempts_count(mid) >= 1) + check("wp not DONE", orch._wp_state("W1") != "DONE") + + +# --------------------------------------------------------------------------- # +# 21) reason_codes +# --------------------------------------------------------------------------- # +def test_reason_codes(): + print("\n== reason codes machine-readable ==") + check("RUN_CODES set", isinstance(o.RUN_CODES, frozenset) and len(o.RUN_CODES) >= 15, o.RUN_CODES) + check("status codes consistent", all(c in o.RUN_CODES for c in ( + o.ST_PLAN_VALIDATED, o.ST_TERMINAL, o.ST_NO_RUNNABLE_WORK, o.ST_MUTATION_BLOCKED, + o.ST_APPROVAL_REQUIRED, o.ST_MISSION_COMPLETED, o.ST_NEEDS_DISPATCH))) + # deterministic A3 decision reason codes + orch, *_ = make_orch_with_paths() + orch.load_plan(default_plan()) + orch.safety.open_circuit("MISSION", "RQ-A4-1", trigger="R", severity="HIGH", mission_id="RQ-A4-1") + dec = orch.safety.evaluate_next_action("RQ-A4-1", "W1") + check("reason code present", dec["REASON_CODE"] in ("CIRCUIT_ALREADY_OPEN", "CIRCUIT_BREAK") and dec["DECISION"] in ("BLOCK",), dec) + + +# --------------------------------------------------------------------------- # +# 22) secret_safe_logging +# --------------------------------------------------------------------------- # +def test_secret_safe_logging(): + print("\n== secret-safe logging ==") + orch, *_ = make_orch_with_paths() + orch.load_plan(default_plan()) + # evidence record with a credential-like value -> redacted + ev = o.make_child_evidence("M", "W", ROLE_MAKER, "t", "2024-01-01", "dispatched", + "PENDING", evidence_ref="token=ghp_ABC123secret") + blob = json.dumps(ev) + check("no raw secret in evidence", "ghp_ABC123secret" not in blob, blob) + check("no credential value", "token=ghp" not in blob, blob) + check("REDACTED marker or safe", "REDACTED" in blob or "ghp_ABC" not in blob) + # contract builder redacts secrets in goal + c = build_maker_contract({"id": "W", "goal": "pw=Sup3rSec", "scope": "s", + "acceptance_criteria": "ac", "test_requirements": "", "safety_rules": [], + "allowed_files": [], "forbidden_files": []}, + {"id": "M"}, "MEDIUM") + check("maker contract secret redacted", "Sup3rSec" not in json.dumps(c), json.dumps(c)) + + +# --------------------------------------------------------------------------- # +# Helper dispatchers +# --------------------------------------------------------------------------- # +def _pass_dispatcher(): + def disp(contract): + if contract.get("role") == ROLE_MAKER: + return {"role": ROLE_MAKER, "TASK_ID": contract["TASK_ID"], "result": "PASS", + "test_results": [{"verdict": "PASS"}], "progress": "YES"} + return {"role": ROLE_CHECKER, "verdict": "PASS", "progress": "YES"} + return disp + + +def fail_dispatcher(): + """Checker FAIL mit messbarem Fortschritt -> A3 DECISION=RETRY (nicht DEBUG).""" + def disp(contract): + if contract.get("role") == ROLE_MAKER: + return {"role": ROLE_MAKER, "TASK_ID": contract["TASK_ID"], "result": "PASS", + "test_results": [{"verdict": "PASS"}], "progress": "YES"} + return {"role": ROLE_CHECKER, "verdict": "FAIL", "fail_reason": "AC not met", + "strategy": "strategy-X", "progress": "YES"} + return disp + + +# --------------------------------------------------------------------------- # +# Helpers +# --------------------------------------------------------------------------- # +def make_orch_with_paths(): + d = tempfile.mkdtemp(prefix="a4h_") + mdb = str(Path(d) / "missions.db") + sdb = str(Path(d) / "safety.db") + return Orchestrator(mdb, sdb), mdb, sdb + + +def make_orch(*dispatcher, **kw): + d = tempfile.mkdtemp(prefix="a4o_") + mdb = str(Path(d) / "missions.db") + sdb = str(Path(d) / "safety.db") + disp = dispatcher[0] if dispatcher else None + return Orchestrator(mdb, sdb, disp, **kw) + + +def fresh_db(): + d = tempfile.mkdtemp(prefix="a4f_") + return str(Path(d) / "missions.db"), str(Path(d) / "safety.db") + + +# --------------------------------------------------------------------------- # +# Test Runner +# --------------------------------------------------------------------------- # +ALL_TESTS = [ + test_plan_validation, + test_dependency_validation, + test_runnable_selection, + test_a2a3_consistency, + test_maker_dispatch_contract, + test_checker_dispatch_contract, + test_checker_pass, + test_checker_fail, + test_retry_allowed, + test_retry_denied_and_4_impossible, + test_circuit_open_blocks_maker, + test_approval_gate, + test_risk_overrides_size, + test_no_runnable_stop, + test_mission_completion_gate, + test_idempotent_wp, + test_restart_resume, + test_child_failure, + test_reason_codes, + test_secret_safe_logging, +] + + +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())