79 lines
2.4 KiB
Python
Executable File
79 lines
2.4 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Plan-18 Chat turn entrypoint.
|
|
|
|
Thin harness-owned router for Control Panel: classify once, then delegate to the
|
|
mode primitive. NestJS calls this file only; it does not own governance verdicts.
|
|
"""
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
|
|
def project_root() -> str:
|
|
d = os.path.abspath(os.path.dirname(__file__))
|
|
p = d
|
|
while p != os.path.dirname(p):
|
|
if os.path.isdir(os.path.join(p, ".specify")) or os.path.isdir(os.path.join(p, "packages/casan-harness")):
|
|
return p
|
|
p = os.path.dirname(p)
|
|
return os.path.abspath(os.path.join(d, "..", "..", ".."))
|
|
|
|
|
|
ROOT = project_root()
|
|
BIN = os.path.join(ROOT, "packages", "casan-harness", "scripts", "bash")
|
|
ROUTER = os.path.join(BIN, "prompt-mode-router.py")
|
|
READONLY = os.path.join(BIN, "chat-readonly.py")
|
|
OPERATOR = os.path.join(BIN, "chat-operator.py")
|
|
|
|
|
|
def classify(message: str):
|
|
r = subprocess.run(["python3", ROUTER, "classify", "--message", message], cwd=ROOT, capture_output=True, text=True)
|
|
try:
|
|
return json.loads(r.stdout)
|
|
except Exception:
|
|
return {"mode": "BLOCK", "risk": "high", "reason": "router_invalid_json", "matched_rules": [r.stderr.strip()]}
|
|
|
|
|
|
def run_and_passthrough(args):
|
|
r = subprocess.run(args, cwd=ROOT, text=True)
|
|
return r.returncode
|
|
|
|
|
|
def ask(args) -> int:
|
|
router = classify(args.message)
|
|
common = [
|
|
"--message", args.message,
|
|
"--actor", args.actor,
|
|
"--chat-id", args.chat_id,
|
|
"--turn-id", args.turn_id,
|
|
"--tenant", args.tenant,
|
|
]
|
|
if router.get("mode") == "OPERATOR":
|
|
return run_and_passthrough(["python3", OPERATOR, "run", *common])
|
|
return run_and_passthrough(["python3", READONLY, "ask", *common])
|
|
|
|
|
|
def verify_audit() -> int:
|
|
return run_and_passthrough(["python3", READONLY, "verify-audit"])
|
|
|
|
|
|
def main() -> int:
|
|
ap = argparse.ArgumentParser()
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
askp = sub.add_parser("ask")
|
|
askp.add_argument("--message", required=True)
|
|
askp.add_argument("--actor", default="anonymous")
|
|
askp.add_argument("--chat-id", default="")
|
|
askp.add_argument("--turn-id", default="")
|
|
askp.add_argument("--tenant", default="default")
|
|
askp.set_defaults(func=ask)
|
|
sub.add_parser("verify-audit").set_defaults(func=lambda _args: verify_audit())
|
|
args = ap.parse_args()
|
|
return args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|