CASAN/packages/casan-harness/scripts/bash/chat-agent-resolver.py

#!/usr/bin/env python3
"""Plan-18 MVP-2 agent/skill governance resolver.

This is a harness-owned policy primitive. It binds a chat turn to an agent,
skill, delegation level, model role, and tool allowlist before any runtime is
invoked. Unknown policy, RBAC failure, role mismatch, delegation escalation, or
tool outside the agent allowlist all fail closed.
"""
import argparse
import datetime
import hashlib
import json
import os
import subprocess
import sys

PYTHON_BIN = sys.executable or "/usr/bin/python3"


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")
HARNESS_BIN = os.path.join(HARNESS_ROOT, "scripts", "bash")
RBAC = os.path.join(HARNESS_BIN, "rbac-check.py")
TENANT_STORE = os.path.join(HARNESS_BIN, "tenant-store.sh")
GENESIS_HASH = "0" * 64


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


def sha(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def state_root() -> str:
    return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")


def tenant_path(logical: str, fallback: str) -> str:
    if os.environ.get("CASAN_TENANT_ID"):
        r = subprocess.run(["bash", TENANT_STORE, "resolve", logical], cwd=ROOT, capture_output=True, text=True)
        if r.returncode != 0:
            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
        return r.stdout.strip()
    return os.path.join(state_root(), fallback)


def guarded_override(path: str) -> str:
    if path and os.environ.get("CASAN_TENANT_ID"):
        r = subprocess.run(["bash", TENANT_STORE, "guard", path], cwd=ROOT, capture_output=True, text=True)
        if r.returncode != 0:
            raise SystemExit((r.stderr or r.stdout or "TENANT_DENIED").strip())
    return path


def registry_path() -> str:
    return os.environ.get("CASAN_AGENT_REGISTRY_FILE") or os.path.join(HARNESS_ROOT, "config", "agent-registry.yaml")


def audit_path() -> str:
    return guarded_override(os.environ["CASAN_AGENT_BIND_AUDIT_LOG"]) if os.environ.get("CASAN_AGENT_BIND_AUDIT_LOG") else tenant_path("chat/agent-bindings.jsonl", "logs/chat/agent-bindings.jsonl")


def head_path() -> str:
    return guarded_override(os.environ["CASAN_AGENT_BIND_AUDIT_HEAD"]) if os.environ.get("CASAN_AGENT_BIND_AUDIT_HEAD") else tenant_path("chat/agent-bindings-head.txt", "logs/chat/agent-bindings-head.txt")


def load_registry():
    try:
        with open(registry_path(), encoding="utf-8") as fh:
            data = json.load(fh)
        agents = data.get("agents")
        if not isinstance(agents, list):
            raise ValueError("agents_not_list")
        return data
    except Exception as exc:
        return {"_error": f"agent_registry_unreadable:{exc}", "agents": []}


def find_agent(registry, agent_id: str):
    return next((a for a in registry.get("agents", []) if a.get("id") == agent_id), None)


def load_head() -> str:
    try:
        return open(head_path(), encoding="utf-8").read().strip() or GENESIS_HASH
    except OSError:
        return GENESIS_HASH


def append_audit(base):
    path = audit_path()
    os.makedirs(os.path.dirname(path), exist_ok=True)
    seq = 1
    if os.path.isfile(path):
        with open(path, encoding="utf-8") as fh:
            seq = sum(1 for line in fh if line.strip()) + 1
    prev = load_head()
    core = {"seq": seq, **base, "prev_hash": prev}
    record_hash = sha(json.dumps(core, sort_keys=True, ensure_ascii=False))
    rec = {**core, "record_hash": record_hash}
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps(rec, ensure_ascii=False) + "\n")
    os.makedirs(os.path.dirname(head_path()), exist_ok=True)
    with open(head_path(), "w", encoding="utf-8") as fh:
        fh.write(record_hash + "\n")
    return rec


def decision_payload(args, decision: str, reason: str, agent=None, skill="", tools=None):
    tools = tools or []
    success = decision == "BOUND"
    payload = {
        "success": success,
        "decision": decision,
        "reason": reason,
        "agent_selected": (agent or {}).get("id") or args.agent,
        "skill_selected": skill or "",
        "delegation_level": args.delegation_level,
        "model_role": (agent or {}).get("model_role"),
        "mode": (agent or {}).get("mode"),
        "tool_allowlist": (agent or {}).get("tool_allowlist", []),
        "requested_tools": tools,
        "side_effect_allowed": bool((agent or {}).get("tool_allowlist")),
        "requires_approval": bool((agent or {}).get("requires_approval")) or decision == "REQUIRES_APPROVAL",
    }
    rec = append_audit({
        "timestamp": now_iso(),
        "harness": "H5-chat-agent-bind",
        "actor": args.actor,
        "role": args.role,
        "project": args.project,
        "tenant": args.tenant,
        "agent_selected": payload["agent_selected"],
        "skill_selected": payload["skill_selected"],
        "delegation_level": args.delegation_level,
        "decision": decision,
        "reason": reason,
        "tool_allowlist_ref": sha(json.dumps(payload["tool_allowlist"], sort_keys=True)),
        "requested_tools_ref": sha(json.dumps(tools, sort_keys=True)),
    })
    payload["audit"] = {"seq": rec["seq"], "record_hash": rec["record_hash"], "head": rec["record_hash"]}
    return payload


def run_rbac(args):
    r = subprocess.run([
        PYTHON_BIN, RBAC, "check",
        "--role", args.role,
        "--resource", "chat",
        "--action", "select_agent",
        "--role-project", args.project,
        "--target-project", args.target_project or args.project,
        "--role-tenant", args.tenant,
        "--target-tenant", args.target_tenant or args.tenant,
    ], cwd=ROOT, capture_output=True, text=True)
    return r.returncode, (r.stdout + r.stderr).strip()


def bind(args) -> int:
    if args.tenant and args.tenant != "default":
        os.environ["CASAN_TENANT_ID"] = args.tenant
    registry = load_registry()
    if registry.get("_error"):
        out = decision_payload(args, "DENIED", registry["_error"])
        print(json.dumps(out, ensure_ascii=False))
        return 2

    agent = find_agent(registry, args.agent)
    if not agent:
        out = decision_payload(args, "DENIED", "unknown_agent")
        print(json.dumps(out, ensure_ascii=False))
        return 2

    rc, msg = run_rbac(args)
    if rc != 0:
        out = decision_payload(args, "DENIED", f"rbac_denied:{msg}", agent)
        print(json.dumps(out, ensure_ascii=False))
        return 2

    if args.role not in agent.get("roles_allowed", []):
        out = decision_payload(args, "DENIED", "role_not_allowed_for_agent", agent)
        print(json.dumps(out, ensure_ascii=False))
        return 2

    skill = args.skill or (agent.get("skills_allowed", [""])[0] if agent.get("skills_allowed") else "")
    if skill and skill not in agent.get("skills_allowed", []):
        out = decision_payload(args, "DENIED", "skill_not_allowed_for_agent", agent, skill)
        print(json.dumps(out, ensure_ascii=False))
        return 2

    max_level = int(agent.get("max_delegation_level", 0))
    if args.delegation_level > max_level:
        out = decision_payload(args, "REQUIRES_APPROVAL", "delegation_exceeds_agent_max", agent, skill)
        print(json.dumps(out, ensure_ascii=False))
        return 3

    requested_tools = list(args.tool or [])
    allow = set(agent.get("tool_allowlist", []))
    outside = [t for t in requested_tools if t not in allow]
    if outside:
        out = decision_payload(args, "DENIED", "tool_not_allowlisted", agent, skill, requested_tools)
        out["action_gate"] = {"outcome": "BLOCK", "reason": f"tool_not_allowlisted:{','.join(outside)}"}
        print(json.dumps(out, ensure_ascii=False))
        return 2

    out = decision_payload(args, "BOUND", "agent_bound", agent, skill, requested_tools)
    print(json.dumps(out, ensure_ascii=False))
    return 0


def list_agents(args) -> int:
    registry = load_registry()
    if registry.get("_error"):
        print(json.dumps({"success": False, "error": registry["_error"], "agents": []}, ensure_ascii=False))
        return 2
    agents = []
    for agent in registry.get("agents", []):
        item = {k: agent.get(k) for k in (
            "id", "label", "mode", "model_role", "roles_allowed", "skills_allowed",
            "tool_allowlist", "max_delegation_level", "requires_approval",
        )}
        item["allowed_for_role"] = args.role in agent.get("roles_allowed", []) if args.role else None
        agents.append(item)
    print(json.dumps({"success": True, "version": registry.get("version"), "agents": agents}, ensure_ascii=False))
    return 0


def verify_audit() -> int:
    prev = GENESIS_HASH
    count = 0
    try:
        fh = open(audit_path(), encoding="utf-8")
    except OSError:
        print("CHAT_AGENT_AUDIT ok=true records=0 head=" + prev)
        return 0
    with fh:
        for line in fh:
            if not line.strip():
                continue
            count += 1
            rec = json.loads(line)
            got = rec.get("record_hash")
            rest = {k: v for k, v in rec.items() if k != "record_hash"}
            if rest.get("prev_hash") != prev or sha(json.dumps(rest, sort_keys=True, ensure_ascii=False)) != got:
                print(f"CHAT_AGENT_AUDIT ok=false brokenAt={count}")
                return 1
            prev = got
    print(f"CHAT_AGENT_AUDIT ok=true records={count} head={prev}")
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    sub = ap.add_subparsers(dest="cmd", required=True)
    b = sub.add_parser("bind")
    b.add_argument("--agent", required=True)
    b.add_argument("--skill", default="")
    b.add_argument("--actor", default="anonymous")
    b.add_argument("--role", default="viewer")
    b.add_argument("--project", default="default")
    b.add_argument("--tenant", default="default")
    b.add_argument("--target-project", default="")
    b.add_argument("--target-tenant", default="")
    b.add_argument("--delegation-level", type=int, default=0)
    b.add_argument("--tool", action="append", default=[])
    b.set_defaults(func=bind)

    l = sub.add_parser("list-agents")
    l.add_argument("--role", default="")
    l.set_defaults(func=list_agents)

    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())