Files
CASAN/packages/casan-harness/scripts/bash/chat-turn.py
T

317 lines
12 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 hashlib
import json
import os
import subprocess
import sys
import tempfile
import uuid
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")
AGENT_RESOLVER = os.path.join(BIN, "chat-agent-resolver.py")
READONLY = os.path.join(BIN, "chat-readonly.py")
OPERATOR = os.path.join(BIN, "chat-operator.py")
LOOP_RUN = os.path.join(BIN, "loop-run.sh")
LOOP_TRACE = os.path.join(BIN, "loop-trace.py")
PREFLIGHT = os.path.join(BIN, "harness-preflight.sh")
CONTEXT_SCAN = os.path.join(BIN, "context-assemble-scan.sh")
TOOL_OUTPUT_SCAN = os.path.join(BIN, "tool-output-scan.sh")
def state_root() -> str:
return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
def sha(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
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 scan_tool_output(text: str, label: str):
with tempfile.TemporaryDirectory() as td:
path = os.path.join(td, "tool-output.txt")
with open(path, "w", encoding="utf-8") as fh:
fh.write(text)
r = subprocess.run(["bash", TOOL_OUTPUT_SCAN, path, label], cwd=ROOT, capture_output=True, text=True)
return r.returncode, (r.stdout + r.stderr).strip()
def run_mode(args, binding, loop_run=None, scan_output_label=""):
env = os.environ.copy()
if loop_run is not None:
env["CASAN_CHAT_LOOP_RUN_JSON"] = json.dumps(loop_run, ensure_ascii=False, sort_keys=True)
r = subprocess.run(args, cwd=ROOT, capture_output=True, text=True, env=env)
if scan_output_label:
scan_rc, scan_msg = scan_tool_output(r.stdout + "\n" + r.stderr, scan_output_label)
if scan_rc != 0:
print(json.dumps({
"success": False,
"mode": binding.get("mode") if binding else "OPERATOR",
"risk": "high",
"decision": "DENIED",
"answer": "Denied by H4 tool-output scan.",
"sources": [],
"certified": False,
"audit": {},
"router": {"reason": "tool_output_scan_denied", "matched_rules": [scan_msg]},
"agent_binding": binding,
"loop_run": loop_run,
}, ensure_ascii=False))
return 2
try:
payload = json.loads(r.stdout)
payload["agent_binding"] = binding
if loop_run is not None:
payload["loop_run"] = loop_run
print(json.dumps(payload, ensure_ascii=False))
except Exception:
if r.stdout:
sys.stdout.write(r.stdout)
if r.stderr:
sys.stderr.write(r.stderr)
return r.returncode
def run_and_passthrough(args):
r = subprocess.run(args, cwd=ROOT, text=True)
return r.returncode
def default_agent_for_mode(mode: str) -> str:
if mode == "OPERATOR":
return "ops-operator"
if mode == "CODEGEN":
return "codegen-draft"
return "evidence-reader"
def write_json(path: str, payload):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
json.dump(payload, fh, ensure_ascii=False, indent=2, sort_keys=True)
def write_text(path: str, text: str):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
fh.write(text)
def loop_dir(run_id: str) -> str:
return os.path.join(state_root(), "logs", "chat", "loop-runs", run_id)
def run_preflight_and_context(run_id: str, draft_path: str):
preflight_out = os.path.join(loop_dir(run_id), "preflight.json")
r = subprocess.run(["bash", PREFLIGHT, draft_path, preflight_out, "--model", "local:chat-turn"], cwd=ROOT, capture_output=True, text=True)
if r.returncode != 0:
return False, "harness_preflight_failed", (r.stdout + r.stderr).strip()
r = subprocess.run(["bash", CONTEXT_SCAN, draft_path], cwd=ROOT, capture_output=True, text=True)
if r.returncode != 0:
return False, "context_assemble_scan_failed", (r.stdout + r.stderr).strip()
return True, "preflight_pass", (r.stdout + r.stderr).strip()
def certify_operator_draft(args, router, binding):
run_id = f"chat-{sha('|'.join([args.chat_id or 'chat-default', args.turn_id or '', args.message]))[:16]}"
d = loop_dir(run_id)
draft_path = os.path.join(d, "draft.txt")
criteria_path = os.path.join(d, "success-criteria.json")
draft = "\n".join([
"CHAT_TURN",
"ACTION_HELD",
f"chat_id={args.chat_id or 'chat-default'}",
f"turn_id={args.turn_id or 'auto'}",
f"mode={router.get('mode')}",
f"agent={binding.get('agent_selected')}",
f"skill={binding.get('skill_selected')}",
f"delegation_level={binding.get('delegation_level')}",
f"user_msg_ref={sha(args.message)}",
f"user_message_preview={args.message[:160]}",
"",
])
write_text(draft_path, draft)
write_json(criteria_path, {
"must_contain": ["CHAT_TURN", "ACTION_HELD", f"agent={binding.get('agent_selected')}"],
"must_not_contain": ["BYPASS_LOOP_GATE"],
})
ok, preflight_reason, preflight_detail = run_preflight_and_context(run_id, draft_path)
if not ok:
return {
"success": False,
"decision": "DENIED",
"reason": preflight_reason,
"detail": preflight_detail,
"run_id": run_id,
"draft_ref": sha(draft),
"draft_certified": False,
"side_effect_released": False,
"artifact": draft_path,
"success_criteria": criteria_path,
}, 2
loop_env = {
**os.environ,
"CASAN_LOOP_STATE_ROOT": os.environ.get("CASAN_LOOP_STATE_ROOT") or os.path.join(state_root(), "logs", "chat", "loop-state"),
"CASAN_TENANT_ID": args.tenant,
}
dlevel = f"L{binding.get('delegation_level', 0)}"
r = subprocess.run([
"bash", LOOP_RUN,
"--run-id", run_id,
"--artifact", draft_path,
"--success-criteria", criteria_path,
"--profile", os.environ.get("CASAN_PROFILE", "dev"),
"--delegation-level", dlevel,
"--project", args.project,
"--max-steps", "2",
"--tokens-per-step", "128",
"--cost-per-step", "0",
"--context", draft_path,
], cwd=ROOT, capture_output=True, text=True, env=loop_env)
trace_rc = subprocess.run(["python3", LOOP_TRACE, "verify-chain", "--run-id", run_id], cwd=ROOT, capture_output=True, text=True, env=loop_env)
replay_rc = subprocess.run(["python3", LOOP_TRACE, "replay", "--run-id", run_id, "--profile", os.environ.get("CASAN_PROFILE", "dev")], cwd=ROOT, capture_output=True, text=True, env=loop_env)
certified = r.returncode == 0 and "final=DONE" in r.stdout and trace_rc.returncode == 0 and replay_rc.returncode == 0
return {
"success": certified,
"decision": "CERTIFIED" if certified else "HALTED",
"reason": "loop_run_pass" if certified else "loop_run_failed",
"run_id": run_id,
"draft_ref": sha(draft),
"draft_certified": certified,
"side_effect_released": certified,
"artifact": draft_path,
"success_criteria": criteria_path,
"output": (r.stdout + r.stderr).strip(),
"trace_verify": {"ok": trace_rc.returncode == 0, "output": (trace_rc.stdout + trace_rc.stderr).strip()},
"replay": {"ok": replay_rc.returncode == 0, "output": (replay_rc.stdout + replay_rc.stderr).strip()},
}, (0 if certified else 3)
def denied_from_loop(router, binding, loop_run):
print(json.dumps({
"success": False,
"mode": router.get("mode", "OPERATOR"),
"risk": router.get("risk", "medium"),
"decision": "DENIED" if loop_run.get("decision") == "DENIED" else "HALTED",
"answer": f"Operator side-effect held: {loop_run.get('reason')}",
"sources": [{
"path": os.path.relpath(loop_run.get("artifact", ""), ROOT) if loop_run.get("artifact") else "",
"line": 1,
"excerpt": "UNCERTIFIED draft held before side-effect.",
"score": 1,
}],
"certified": False,
"audit": {},
"router": router,
"agent_binding": binding,
"loop_run": loop_run,
}, ensure_ascii=False))
def bind_agent(args, router):
agent = args.agent or default_agent_for_mode(router.get("mode", "READ_ONLY"))
tools = []
if router.get("mode") == "OPERATOR":
# The operator primitive resolves the exact registered action later; the
# agent bind still proves this turn is allowed to use the operator class.
tools.append("run-chat-tests")
cmd = [
"python3", AGENT_RESOLVER, "bind",
"--agent", agent,
"--actor", args.actor,
"--role", args.role,
"--project", args.project,
"--tenant", args.tenant,
"--delegation-level", str(args.delegation_level),
]
if args.skill:
cmd += ["--skill", args.skill]
for tool in tools:
cmd += ["--tool", tool]
r = subprocess.run(cmd, cwd=ROOT, capture_output=True, text=True)
if r.returncode != 0:
sys.stdout.write((r.stdout or r.stderr).strip() + "\n")
return r.returncode, None
try:
return 0, json.loads(r.stdout)
except Exception:
print(json.dumps({"success": False, "decision": "DENIED", "reason": "agent_resolver_invalid_json"}, ensure_ascii=False))
return 2, None
def ask(args) -> int:
router = classify(args.message)
bind_rc, binding = bind_agent(args, router)
if bind_rc != 0:
return bind_rc
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":
loop_run, loop_rc = certify_operator_draft(args, router, binding)
if loop_rc != 0:
denied_from_loop(router, binding, loop_run)
return loop_rc
return run_mode(["python3", OPERATOR, "run", *common], binding, loop_run, "chat-operator")
return run_mode(["python3", READONLY, "ask", *common], binding)
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("--role", default="viewer")
askp.add_argument("--project", default="default")
askp.add_argument("--chat-id", default="")
askp.add_argument("--turn-id", default="")
askp.add_argument("--tenant", default="default")
askp.add_argument("--agent", default="")
askp.add_argument("--skill", default="")
askp.add_argument("--delegation-level", type=int, default=0)
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())