trading-system-docs/a5/rq_heartbeat_cli.py

#!/usr/bin/env python3
"""Red Queen — A5: Heartbeat CLI (CONTROLLED HEARTBEAT & RESUME v1).

Operational Commands (§52):
  status                 Read-only: Kill-Switch, letzter Tick, Missionen, offene Gates.
  tick [MISSION_ID]      EIN bounded Heartbeat-Zyklus (Kill-Switch, Lock, Eligibility, A4).
  enable                 Kill-Switch ON  (bewusst; autonome A4-Mutation erlaubt).
  disable                Kill-Switch OFF (Fail-closed: keine A4-Mutation).
  resume --mission ID    Explizite Resume einer PAUSED-Mission markieren.
  approve --gate GATE_ID [--by NAME --evidence TEXT]
  deny    --gate GATE_ID [--by NAME --evidence TEXT]
  priority --mission ID --prio N
  status                  Registry-/Tick-Log-Anzeige.

Alle Kommandos sind deterministisch und laufen nie zyklisch. Kein Root.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

_HERE = Path(__file__).resolve().parent
_REPO = _HERE.parent
for _p in (str(_REPO), str(_REPO / "a2"), str(_REPO / "a3"), str(_REPO / "a4"), str(_HERE)):
    if _p not in sys.path:
        sys.path.insert(0, _p)

from a4.rq_orchestrator import Orchestrator  # noqa: E402
from a5.rq_heartbeat import Heartbeat, HeartbeatStore  # noqa: E402


def _default_paths():
    base = os.environ.get("RQ_DATA", str(_HERE))
    return {
        "registry": os.path.join(base, "heartbeat.db"),
        "mission": os.path.join(base, "missions.db"),
        "safety": os.path.join(base, "safety.db"),
    }


def _make_hb(args) -> Heartbeat:
    p = _default_paths()
    orch = None
    if args.execute:
        orch = Orchestrator(mission_db=p["mission"], safety_db=p["safety"])
    return Heartbeat(p["registry"], p["mission"], p["safety"], orchestrator=orch)


def _cmd_tick(args):
    hb = _make_hb(args)
    res = hb.run_tick(mission_id=args.mission, execute=args.execute)
    print(json.dumps(res, indent=2, default=str))
    return 0


def _cmd_status(args):
    reg = HeartbeatStore(_default_paths()["registry"])
    print(f"kill_switch: {'ON' if reg.kill_switch_on() else 'OFF'}")
    lt = reg.last_tick()
    if lt:
        print("last_tick:", json.dumps(lt, default=str))
    else:
        print("last_tick: none")
    return 0


def _cmd_switch(args, value: bool):
    reg = HeartbeatStore(_default_paths()["registry"])
    reg.set_kill_switch(value)
    print(f"kill_switch -> {'ON' if value else 'OFF'}")
    return 0


def _cmd_resume(args):
    reg = HeartbeatStore(_default_paths()["registry"])
    reg.set_resume_pending(args.mission, True)
    print(f"resume_pending -> {args.mission}")
    return 0


def _cmd_approve(args):
    reg = HeartbeatStore(_default_paths()["registry"])
    out = reg.approval_grant(args.gate_id, args.by, args.evidence)
    print(json.dumps(out, default=str))
    return 0


def _cmd_deny(args):
    reg = HeartbeatStore(_default_paths()["registry"])
    out = reg.approval_deny(args.gate_id, args.by, args.evidence)
    print(json.dumps(out, default=str))
    return 0


def _cmd_priority(args):
    reg = HeartbeatStore(_default_paths()["registry"])
    reg.set_config(f"priority:{args.mission}", str(args.prio))
    print(f"priority[{args.mission}] -> {args.prio}")
    return 0


def build_parser():
    ap = argparse.ArgumentParser(prog="rq-heartbeat", description="A5 Heartbeat CLI")
    ap.add_argument("--execute", action="store_true",
                    help="(tick) erlaubt A4-Mutation; ohne ist tick read-only")
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("tick")
    p.add_argument("--mission", default=None)
    p.set_defaults(func=_cmd_tick)

    sub.add_parser("status").set_defaults(func=_cmd_status)
    sub.add_parser("enable").set_defaults(func=lambda a: _cmd_switch(a, True))
    sub.add_parser("disable").set_defaults(func=lambda a: _cmd_switch(a, False))

    p = sub.add_parser("resume")
    p.add_argument("--mission", required=True)
    p.set_defaults(func=_cmd_resume)

    p = sub.add_parser("approve")
    p.add_argument("--gate", dest="gate_id", required=True)
    p.add_argument("--by", default="christian")
    p.add_argument("--evidence", default="manual approval")
    p.set_defaults(func=_cmd_approve)

    p = sub.add_parser("deny")
    p.add_argument("--gate", dest="gate_id", required=True)
    p.add_argument("--by", default="christian")
    p.add_argument("--evidence", default="manual deny")
    p.set_defaults(func=_cmd_deny)

    p = sub.add_parser("priority")
    p.add_argument("--mission", required=True)
    p.add_argument("--prio", type=int, required=True)
    p.set_defaults(func=_cmd_priority)

    return ap


def main():
    args = build_parser().parse_args()
    return args.func(args)


if __name__ == "__main__":
    sys.exit(main())