trading-system-docs/a2/rq_mission_cli.py

#!/usr/bin/env python3
"""
Red Queen — A2 CLI. Dünner, deterministischer Kommandozeilen-Zugriff auf den
MissionStore. Maschinenlesbar via `--json`.

Hinweis: Der A2-Build definiert KEINE autonome Orchestrierung — dieses CLI ruft
nur die vom Store bereitgestellten atomaren Operationen auf.
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from typing import Optional

from rq_mission import MissionStore, RqError


def _store(args) -> MissionStore:
    db = args.db or os.environ.get("RQ_MISSIONS_DB") or "missions.db"
    mirror = args.mirror
    return MissionStore(db, mirror_db=mirror)


def _emit(args, obj) -> None:
    if args.json:
        print(json.dumps(obj, indent=2, ensure_ascii=False))
    else:
        if isinstance(obj, dict) and "id" in obj:
            print(obj["id"])
        else:
            print(json.dumps(obj, ensure_ascii=False))


def _fail(e: RqError, 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_mission", description="Red Queen A2 Mission/WP CLI")
    p.add_argument("--db", help="Pfad zur missions.db (default: $RQMISSIONS_DB oder missions.db)")
    p.add_argument("--mirror", help="Kanban-DB-Pfad fuer nativen Mirror ('off' deaktiviert)")
    p.add_argument("--json", action="store_true", help="JSON-Ausgabe")
    sub = p.add_subparsers(dest="cmd", required=True)

    # mission_create
    mc = sub.add_parser("mission_create")
    mc.add_argument("--title", required=True)
    mc.add_argument("--goal")
    mc.add_argument("--scope")
    mc.add_argument("--constraints")
    mc.add_argument("--acceptance")
    mc.add_argument("--id", dest="mission_id")
    mc.add_argument("--actor", default="red-queen")

    # mission_read / state
    mr = sub.add_parser("mission_read"); mr.add_argument("mission_id")
    ms = sub.add_parser("mission_state"); ms.add_argument("mission_id")

    # mission transitions
    mt = sub.add_parser("mission_transition")
    mt.add_argument("--mission", dest="mission_id", required=True)
    mt.add_argument("--target", required=True)
    mt.add_argument("--evidence")
    mt.add_argument("--actor", default="red-queen")
    sub.add_parser("mission_pause").add_argument("mission_id")
    sub.add_parser("mission_block").add_argument("mission_id")
    sub.add_parser("mission_resume").add_argument("mission_id")
    sub.add_parser("mission_complete").add_argument("mission_id")

    # wp_create
    wc = sub.add_parser("wp_create")
    wc.add_argument("--mission", dest="mission_id", required=True)
    wc.add_argument("--title", required=True)
    wc.add_argument("--wp-id", dest="wp_id")
    wc.add_argument("--actor", default="red-queen")

    wr = sub.add_parser("wp_read"); wr.add_argument("wp_id")
    ws = sub.add_parser("wp_state"); ws.add_argument("wp_id")
    wt = sub.add_parser("wp_transition")
    wt.add_argument("--wp", dest="wp_id", required=True)
    wt.add_argument("--target", required=True)
    wt.add_argument("--evidence")
    wt.add_argument("--actor", default="red-queen")
    wd = sub.add_parser("wp_dependency")
    wd.add_argument("--mission", dest="mission_id", required=True)
    wd.add_argument("--wp", dest="wp_id", required=True)
    wd.add_argument("--depends-on", required=True)
    wb = sub.add_parser("wp_block")
    wb.add_argument("wp_id")
    wb.add_argument("--reason", default="blocked")
    wc2 = sub.add_parser("wp_complete")
    wc2.add_argument("wp_id")
    wc2.add_argument("--result")
    return p


def main(argv=None) -> int:
    args = build_parser().parse_args(argv)
    try:
        s = _store(args)
        if args.cmd == "mission_create":
            r = s.mission_create(args.title, goal=args.goal, scope=args.scope,
                                 constraints=args.constraints,
                                 acceptance_criteria=args.acceptance,
                                 mission_id=args.mission_id, actor=args.actor)
        elif args.cmd == "mission_read":
            r = s.mission_read(args.mission_id)
        elif args.cmd == "mission_state":
            r = s.mission_state(args.mission_id)
        elif args.cmd == "mission_transition":
            r = s.mission_transition(args.mission_id, args.target, actor=args.actor, evidence=args.evidence)
        elif args.cmd == "mission_pause":
            r = s.mission_pause(args.mission_id)
        elif args.cmd == "mission_block":
            r = s.mission_block(args.mission_id)
        elif args.cmd == "mission_resume":
            r = s.mission_resume(args.mission_id)
        elif args.cmd == "mission_complete":
            r = s.mission_complete(args.mission_id)
        elif args.cmd == "wp_create":
            r = s.wp_create(args.mission_id, args.title, wp_id=args.wp_id, actor=args.actor)
        elif args.cmd == "wp_read":
            r = s.wp_read(args.wp_id)
        elif args.cmd == "wp_state":
            r = s.wp_state(args.wp_id)
        elif args.cmd == "wp_transition":
            r = s.wp_transition(args.wp_id, args.target, actor=args.actor, evidence=args.evidence)
        elif args.cmd == "wp_dependency":
            r = s.wp_dependency(args.mission_id, args.wp_id, args.depends_on)
        elif args.cmd == "wp_block":
            r = s.wp_block(args.wp_id, reason=args.reason)
        elif args.cmd == "wp_complete":
            r = s.wp_complete(args.wp_id, result=args.result)
        else:
            _emit(args, {"error": f"unknown command {args.cmd}"})
            return 1
        _emit(args, r)
        return 0
    except RqError as e:
        _fail(e, args)
    except Exception as e:  # noqa
        _fail(RqError("INTERNAL_ERROR", str(e)), args)
    return 2  # unreachable normally; _fail exits


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