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