#!/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())