CASAN/packages/casan-harness/scripts/bash/prompt-mode-router.py

#!/usr/bin/env python3
"""Plan-18 deterministic prompt mode router.

MVP-0 emits READ_ONLY/BLOCK/NOT_SUPPORTED. MVP-1 adds OPERATOR only for
registered action phrases; MVP-2 adds CODEGEN draft mode. Free commands remain
NOT_SUPPORTED/BLOCK.
"""
import argparse
import json
import os
import re
import sys
from datetime import datetime, timezone


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()
HARNESS_ROOT = os.path.join(ROOT, "packages", "casan-harness")


def policy_path() -> str:
    return os.environ.get("CASAN_PROMPT_MODES_FILE") or os.path.join(HARNESS_ROOT, "config", "prompt-modes.yaml")


def now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def load_policy():
    try:
        with open(policy_path(), encoding="utf-8") as fh:
            return json.load(fh)
    except Exception as exc:
        return {"_error": f"policy_unreadable:{exc}"}


def lower(s: str) -> str:
    return re.sub(r"\s+", " ", s.lower()).strip()


def contains_any(text: str, patterns):
    hits = []
    for pattern in patterns:
        if pattern and re.search(re.escape(pattern.lower()), text):
            hits.append(pattern)
    return hits


def classify(message: str, model_verdict: str = ""):
    policy = load_policy()
    if policy.get("_error"):
        return {
            "mode": "BLOCK",
            "risk": "high",
            "gates": ["H5_CHAT_AUDIT"],
            "needs_approval": False,
            "reason": policy["_error"],
            "matched_rules": ["policy_fail_closed"],
            "side_effect_allowed": False,
            "classified_at": now_iso(),
        }

    text = lower(message)
    if not text:
        return {
            "mode": "BLOCK",
            "risk": "high",
            "gates": policy["block"]["gates"],
            "needs_approval": False,
            "reason": "empty_message",
            "matched_rules": ["empty_message"],
            "side_effect_allowed": False,
            "classified_at": now_iso(),
        }

    block_hits = contains_any(text, policy.get("block_patterns", []))
    if block_hits:
        return {
            "mode": "BLOCK",
            "risk": "high",
            "gates": policy["block"]["gates"],
            "needs_approval": False,
            "reason": "blocked_by_rule",
            "matched_rules": block_hits,
            "side_effect_allowed": False,
            "classified_at": now_iso(),
        }

    unsupported_hits = contains_any(text, policy.get("not_supported_patterns", []))
    if unsupported_hits:
        return {
            "mode": "NOT_SUPPORTED",
            "risk": "medium",
            "gates": policy["not_supported"]["gates"],
            "needs_approval": True,
            "reason": "side_effect_not_supported_in_mvp0",
            "matched_rules": unsupported_hits,
            "side_effect_allowed": False,
            "classified_at": now_iso(),
        }

    operator_hits = contains_any(text, policy.get("operator_terms", []))
    if operator_hits:
        return {
            "mode": "OPERATOR",
            "risk": policy["operator"]["risk"],
            "gates": policy["operator"]["gates"],
            "needs_approval": bool(policy["operator"]["needs_approval"]),
            "reason": "registered_operator_action",
            "matched_rules": operator_hits,
            "side_effect_allowed": True,
            "classified_at": now_iso(),
        }

    codegen_hits = contains_any(text, policy.get("codegen_terms", []))
    if codegen_hits:
        return {
            "mode": "CODEGEN",
            "risk": policy["codegen"]["risk"],
            "gates": policy["codegen"]["gates"],
            "needs_approval": bool(policy["codegen"]["needs_approval"]),
            "reason": "codegen_draft_requested",
            "matched_rules": codegen_hits,
            "side_effect_allowed": False,
            "classified_at": now_iso(),
        }

    analysis_hits = contains_any(text, policy.get("analysis_terms", []))
    if analysis_hits:
        acfg = policy.get("analysis", policy["read_only"])
        return {
            "mode": "ANALYSIS",
            "risk": acfg["risk"],
            "gates": acfg["gates"],
            "needs_approval": bool(acfg.get("needs_approval", False)),
            "reason": "analysis_reasoning_requested",
            "matched_rules": analysis_hits,
            "side_effect_allowed": False,
            "classified_at": now_iso(),
        }

    read_terms = policy.get("read_only_terms", [])
    read_hits = [t for t in read_terms if re.search(rf"\b{re.escape(t.lower())}\b", text)]
    # Model-assisted verdict can only increase caution. In MVP-0 an unsafe model
    # verdict is refused, while READ_ONLY from the model cannot override rules.
    mv = (model_verdict or "").strip().upper()
    if mv in {"BLOCK", "NOT_SUPPORTED", "OPERATOR", "CODEGEN"}:
        mode = mv
        cfg = policy["block" if mode == "BLOCK" else ("operator" if mode == "OPERATOR" else ("codegen" if mode == "CODEGEN" else "not_supported"))]
        if mode == "OPERATOR" and not operator_hits:
            mode = "NOT_SUPPORTED"
            cfg = policy["not_supported"]
        if mode == "CODEGEN" and not codegen_hits:
            mode = "NOT_SUPPORTED"
            cfg = policy["not_supported"]
        return {
            "mode": mode,
            "risk": cfg["risk"],
            "gates": cfg["gates"],
            "needs_approval": bool(cfg["needs_approval"]),
            "reason": "model_escalated" if mode != "NOT_SUPPORTED" else "model_requested_unsupported_action",
            "matched_rules": [f"model:{mode}"],
            "side_effect_allowed": mode == "OPERATOR",
            "classified_at": now_iso(),
        }

    return {
        "mode": "READ_ONLY",
        "risk": "low",
        "gates": policy["read_only"]["gates"],
        "needs_approval": False,
        "reason": "read_only_terms" if read_hits else "default_read_only_no_side_effect",
        "matched_rules": read_hits,
        "side_effect_allowed": False,
        "classified_at": now_iso(),
    }


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("classify", nargs="?")
    ap.add_argument("--message", default="")
    ap.add_argument("--input", default="")
    ap.add_argument("--model-verdict", default="")
    args = ap.parse_args()
    message = args.message
    if args.input:
        with open(args.input, encoding="utf-8") as fh:
            message = fh.read()
    print(json.dumps(classify(message, args.model_verdict), ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())