gitea-admin/CASAN
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: certify codegen chat drafts through loop

Browse Source
This commit is contained in:
thanhnv
2026-07-08 23:20:54 +09:00
parent 780fe86273
commit b3b0544ba8
15 changed files with 359 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/casan-harness/scripts/bash/chat-turn.py
+169
View File
@@ -12,6 +12,7 @@ import subprocess
import sys
import tempfile
import uuid
from datetime import datetime, timezone
def project_root() -> str:
@@ -36,12 +37,17 @@ 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")
ARTIFACT_SCAN = os.path.join(BIN, "artifact-scan.sh")
def state_root() -> str:
return os.environ.get("CASAN_STATE_ROOT") or os.path.join(ROOT, ".specify")
def now_iso() -> str:
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def audit_path() -> str:
return os.environ.get("CASAN_CHAT_AUDIT_LOG") or os.path.join(state_root(), "logs", "chat", "chat-turns.jsonl")
@@ -54,6 +60,14 @@ def sha(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def sha_file(path: str) -> str:
h = hashlib.sha256()
with open(path, "rb") as fh:
for chunk in iter(lambda: fh.read(65536), b""):
h.update(chunk)
return h.hexdigest()
def classify(message: str):
r = subprocess.run(["python3", ROUTER, "classify", "--message", message], cwd=ROOT, capture_output=True, text=True)
try:
@@ -136,6 +150,10 @@ def loop_dir(run_id: str) -> str:
return os.path.join(state_root(), "logs", "chat", "loop-runs", run_id)
def codegen_dir(run_id: str) -> str:
return os.path.join(state_root(), "logs", "chat", "codegen-artifacts", run_id)
def load_chat_head() -> str:
try:
return open(head_path(), encoding="utf-8").read().strip() or ("0" * 64)
@@ -173,6 +191,11 @@ def run_preflight_and_context(run_id: str, draft_path: str):
return True, "preflight_pass", (r.stdout + r.stderr).strip()
def run_artifact_scan(path: str, label: str):
r = subprocess.run(["bash", ARTIFACT_SCAN, path, label], cwd=ROOT, capture_output=True, text=True)
return r.returncode == 0, (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)
@@ -250,6 +273,147 @@ def certify_operator_draft(args, router, binding):
}, (0 if certified else 3)
def render_codegen_draft(args, binding) -> str:
fn = "generated_chat_draft"
return "\n".join([
"# CODEGEN_DRAFT",
"# GENERATED_BY_CASAN_CHAT",
f"# agent={binding.get('agent_selected')}",
f"# skill={binding.get('skill_selected')}",
f"# user_message_preview={args.message[:200]}",
"",
f"def {fn}():",
" \"\"\"Deterministic draft generated by governed chat; review before use.\"\"\"",
" return {",
f" \"request_hash\": \"{sha(args.message)}\",",
" \"status\": \"draft_only\",",
" }",
"",
])
def certify_codegen_draft(args, router, binding):
run_id = f"chat-codegen-{sha('|'.join([args.chat_id or 'chat-default', args.turn_id or '', args.message]))[:16]}"
d = codegen_dir(run_id)
artifact_path = os.path.join(d, "draft.py")
criteria_path = os.path.join(d, "success-criteria.json")
draft = render_codegen_draft(args, binding)
write_text(artifact_path, draft)
write_json(criteria_path, {
"must_contain": ["CODEGEN_DRAFT", "GENERATED_BY_CASAN_CHAT", "def generated_chat_draft"],
"must_not_contain": ["BYPASS_LOOP_GATE"],
})
scan_ok, scan_out = run_artifact_scan(artifact_path, "chat-codegen-draft")
if not scan_ok:
return {
"success": False,
"decision": "DENIED",
"reason": "artifact_scan_failed",
"run_id": run_id,
"draft_certified": False,
"side_effect_released": False,
"artifact": artifact_path,
"success_criteria": criteria_path,
"artifact_scan": {"ok": False, "output": scan_out},
}, 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", artifact_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", artifact_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)
output_scan_rc, output_scan_msg = scan_tool_output(draft, "chat-codegen-output")
certified = (
r.returncode == 0
and "final=DONE" in r.stdout
and trace_rc.returncode == 0
and replay_rc.returncode == 0
and output_scan_rc == 0
)
return {
"success": certified,
"decision": "CERTIFIED" if certified else "HALTED",
"reason": "codegen_loop_pass" if certified else "codegen_loop_failed",
"run_id": run_id,
"draft_ref": sha(draft),
"draft_certified": certified,
"side_effect_released": False,
"artifact": artifact_path,
"success_criteria": criteria_path,
"artifact_scan": {"ok": True, "output": scan_out},
"tool_output_scan": {"ok": output_scan_rc == 0, "output": output_scan_msg},
"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 finish_codegen(args, router, binding, loop_run, rc: int):
artifact = loop_run.get("artifact", "")
source = []
if artifact and os.path.isfile(artifact):
source.append({
"path": os.path.relpath(artifact, ROOT) if artifact.startswith(ROOT) else artifact,
"line": 1,
"excerpt": "CODEGEN_DRAFT generated as review-only artifact.",
"score": 10 if rc == 0 else 1,
"hash": sha_file(artifact),
})
decision = "ANSWERED" if rc == 0 else ("DENIED" if loop_run.get("decision") == "DENIED" else "HALTED")
rec = append_chat_turn({
"timestamp": now_iso(),
"trace_id": loop_run.get("run_id"),
"chat_id": args.chat_id or "chat-default",
"turn_id": args.turn_id or str(uuid.uuid4()),
"tenant_id": args.tenant,
"actor": args.actor,
"mode": "CODEGEN",
"risk": router.get("risk", "high"),
"decision": decision,
"answer": f"Codegen draft certified: {os.path.relpath(artifact, ROOT) if artifact else 'n/a'}" if rc == 0 else f"Codegen draft held: {loop_run.get('reason')}",
"sources": source,
"router": router,
"agent_binding": binding,
"loop_run": loop_run,
})
print(json.dumps({
"success": rc == 0,
"mode": "CODEGEN",
"risk": router.get("risk", "high"),
"decision": decision,
"answer": f"Codegen draft certified: {os.path.relpath(artifact, ROOT) if artifact else 'n/a'}" if rc == 0 else f"Codegen draft held: {loop_run.get('reason')}",
"sources": source,
"certified": rc == 0,
"audit": {"seq": rec["seq"], "record_hash": rec["record_hash"], "head": rec["record_hash"], "path": audit_path()},
"router": router,
"agent_binding": binding,
"loop_run": loop_run,
"codegen": {
"artifact": os.path.relpath(artifact, ROOT) if artifact and artifact.startswith(ROOT) else artifact,
"artifact_scan": loop_run.get("artifact_scan"),
"tool_output_scan": loop_run.get("tool_output_scan"),
},
}, ensure_ascii=False))
return rc
def denied_from_loop(router, binding, loop_run):
print(json.dumps({
"success": False,
@@ -278,6 +442,8 @@ def bind_agent(args, router):
# 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")
if router.get("mode") == "CODEGEN":
tools.extend(["artifact-scan", "tool-output-scan"])
cmd = [
"python3", AGENT_RESOLVER, "bind",
"--agent", agent,
@@ -385,6 +551,9 @@ def ask(args) -> int:
denied_from_loop(router, binding, loop_run)
return loop_rc
return run_mode(["python3", OPERATOR, "run", *common], binding, loop_run, "chat-operator")
if router.get("mode") == "CODEGEN":
loop_run, loop_rc = certify_codegen_draft(args, router, binding)
return finish_codegen(args, router, binding, loop_run, loop_rc)
return run_mode(["python3", READONLY, "ask", *common], binding)
packages/casan-harness/scripts/bash/ci-harness-gate.sh
+1
View File
@@ -154,6 +154,7 @@ run "phase-chat-pipeline" bash "$TESTS/phase-chat-pipeline-tests.sh"
run "phase-chat-stream-hold" bash "$TESTS/phase-chat-stream-hold-tests.sh"
run "phase-chat-replay" bash "$TESTS/phase-chat-replay-tests.sh"
run "phase-chat-approval" bash "$TESTS/phase-chat-approval-tests.sh"
run "phase-chat-codegen" bash "$TESTS/phase-chat-codegen-tests.sh"
# ARCH-02: coverage cannot silently drop; ARCH-01: harness/policy cannot silently
# drift. Both SKIP cleanly when no manifest is provisioned (non-strict dev/CI).
packages/casan-harness/scripts/bash/prompt-mode-router.py
+21 -4
View File
@@ -2,7 +2,8 @@
"""Plan-18 deterministic prompt mode router.
MVP-0 emits READ_ONLY/BLOCK/NOT_SUPPORTED. MVP-1 adds OPERATOR only for
registered action phrases; free commands remain NOT_SUPPORTED/BLOCK.
registered action phrases; MVP-2 adds CODEGEN draft mode. Free commands remain
NOT_SUPPORTED/BLOCK.
"""
import argparse
import json
@@ -120,23 +121,39 @@ def classify(message: str, model_verdict: str = ""):
"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(),
}
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"}:
if mv in {"BLOCK", "NOT_SUPPORTED", "OPERATOR", "CODEGEN"}:
mode = mv
cfg = policy["block" if mode == "BLOCK" else ("operator" if mode == "OPERATOR" else "not_supported")]
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_operator_without_registered_action",
"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(),