#!/usr/bin/env python3
"""test_p0_trust_boundary.py — the re-review's P0 bypasses must now BLOCK.

이 스위트는 '정상 경로가 동작한다'가 아니라 '나쁜 경로가 막힌다'를 증명한다(재리뷰 지적의 핵심).
각 케이스는 리뷰가 재현한 우회를 그대로 재현하고, 이제 차단됨을 확인한다.

  P0-1  workspace 미설정 → 운영 훅 fail-closed(exit 2)
  P0-2  context-package 없이/위장 패키지로 Org OS 워커 spawn → 차단
  P0-4a 원장(state/evidence/acceptance/registry) 직접 쓰기·evidence_ledger 수동호출·signoff → 차단
  P0-4b plan 불일치 전이·actor 미지정 전이 → 차단; 승인은 acceptance_log 파생만
  P0-4c 존재하지 않는/미검증 report 를 accepted 로 등록 → 차단
  P0-5  report-type 누락/미지·위장 primary-artifacts → 차단
  P0-6  receipt substring/basename 매칭 제거(정확 일치만)
"""
import os
import subprocess
import sys
import tempfile

ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
HOOKS = os.path.join(ROOT, ".claude", "hooks")
sys.path.insert(0, HOOKS)
os.environ["CLAUDE_PROJECT_DIR"] = ROOT

passed = failed = 0


def check(name, ok):
    global passed, failed
    if ok:
        passed += 1
        print(f"  PASS  {name}")
    else:
        failed += 1
        print(f"  FAIL  {name}")


def _run(script, args, env_extra=None, stdin=None):
    e = dict(os.environ)
    e["CLAUDE_PROJECT_DIR"] = ROOT
    e.pop("ORGOS_WORKSPACE", None)
    if env_extra:
        e.update(env_extra)
    return subprocess.run([sys.executable, os.path.join(HOOKS, script)] + args,
                          capture_output=True, text=True, input=stdin, env=e)


# ============================================================ P0-1 workspace fail-closed
print("== P0-1 workspace-unset FAILS CLOSED across operational hooks ==")
nows = {}  # ORGOS_WORKSPACE removed by _run
# Isolate the "unset" cases from the repository's optional .orgos-workspace
# pointer.  A real workflow may legitimately populate that pointer while this
# suite is running; using ROOT here would then test "pointer configured", not
# "workspace unset".  Seed only the worker card needed by stop_validate's
# unregistered-agent classifier.
clean_project = tempfile.mkdtemp(prefix="company-haness-p0-clean-", dir="/tmp")
clean_agents = os.path.join(clean_project, ".claude", "agents")
os.makedirs(clean_agents, exist_ok=True)
with open(os.path.join(clean_agents, "fam-eng-backend.md"), "w", encoding="utf-8") as fh:
    fh.write("---\nname: fam-eng-backend\n---\n")
check("state_engine guard (unset ws) -> exit 2",
      _run("state_engine.py", ["guard", "--workflow", "wf", "--to", "build"], nows).returncode == 2)
check("acceptance_log append (unset ws) -> exit 2",
      _run("acceptance_log.py", ["append", "--report-id", "x", "--decision", "accepted"], nows).returncode == 2)
check("subagent_register report-producing (unset ws) -> exit 2",
      _run("subagent_register.py", [], {"CLAUDE_PROJECT_DIR": clean_project},
           stdin='{"agent_id":"a","agent_type":"eng-be","workflow_id":"wf"}').returncode == 2)
check("subagent_register helper (unset ws) -> exit 0 (not over-blocked)",
      _run("subagent_register.py", [], nows,
           stdin='{"agent_id":"h","agent_type":"general-purpose"}').returncode == 0)
check("stop_validate Org OS agent (unset ws) -> exit 2",
      _run("stop_validate.py", [], {"CLAUDE_PROJECT_DIR": clean_project},
           stdin='{"agent_id":"a","agent_type":"fam-eng-backend"}').returncode == 2)
check("stop_validate --main (unset ws) -> exit 0 (advisory)",
      _run("stop_validate.py", ["--main"], nows, stdin='{"agent_id":"m"}').returncode == 0)

# ============================================================ direct hook API cases
import guard_tools as G  # noqa: E402
import context_package as CP  # noqa: E402
import validate_report as VR  # noqa: E402

print("== P0-4a ledgers are a trust boundary (guard_tools) ==")
LED = [
    ("Write", {"file_path": "_sandbox/state/wf/workflow.yaml", "content": "x"}, "Write workflow.yaml"),
    ("Write", {"file_path": "_sandbox/evidence/wf/ledger.jsonl", "content": "x"}, "Write evidence ledger"),
    ("Edit", {"file_path": "_sandbox/state/acceptance-events.jsonl", "old_string": "a", "new_string": "b"}, "Edit acceptance-events"),
    ("Write", {"file_path": "_sandbox/state/wf/human-signoff.jsonl", "content": "x"}, "Write human-signoff"),
    ("Bash", {"command": "echo x >" + "> _sandbox/state/wf/state-events.jsonl"}, "append state-events"),
    ("Bash", {"command": "python3 -c \"open('_sandbox/evidence/x/ledger.jsonl','a').write('{}')\""}, "python -c open ledger"),
    ("Bash", {"command": "echo {} | python3 .claude/hooks/" + "evidence_ledger.py"}, "manual evidence_ledger.py"),
    ("Bash", {"command": "python3 .claude/hooks/state_engine.py signoff --workflow wf --stage acceptance --by HUMAN-x"}, "agent signoff"),
    ("Bash", {"command": "python3 .claude/hooks/state_engine.py record-human-signoff --workflow wf --stage acceptance --by HUMAN-001"}, "agent signoff alias"),
    ("Bash", {"command": "python3 .claude/hooks/state_engine.py review-artifact --workflow wf --report r.report.yaml --decision accepted --reviewer HUMAN-001"}, "agent impersonates human reviewer"),
    ("Bash", {"command": "python3 -c \"import acceptance_log as AL; AL.append_event(AL.build_event('x','accepted'))\""}, "low-level acceptance append"),
]
for t, ti, label in LED:
    cat, _ = G.check(t, ti)
    check(f"BLOCK {label}", cat is not None)
# legitimate CLIs still pass
for t, ti, label in [
    ("Bash", {"command": "python3 .claude/hooks/state_engine.py transition --workflow wf --to build --actor OPS-ORCH"}, "state transition CLI"),
    ("Bash", {"command": "python3 .claude/hooks/acceptance_log.py append --report-id r --decision accepted"}, "acceptance append CLI"),
    ("Bash", {"command": "python3 .claude/hooks/verify_run.py --workflow wf --agent QA --session s --category test --subject unit -- pytest -q"}, "typed verification runner"),
    ("Bash", {"command": "python3 -m py_compile .claude/hooks/" + "evidence_ledger.py"}, "py_compile mention"),
]:
    cat, _ = G.check(t, ti)
    check(f"ALLOW {label}", cat is None)

print("== P0-2 context-package spawn gate ==")
check("Org OS worker spawn WITHOUT package ref -> BLOCK",
      G.check("Task", {"subagent_type": "eng-be", "prompt": "do it"})[0] is not None)
check("helper spawn (general-purpose) -> ALLOW",
      G.check("Task", {"subagent_type": "general-purpose", "prompt": "search"})[0] is None)
# placeholder package rejected by strengthened validator
ph = {"workflow-id": "w", "task-id": "t", "mode": "converge", "tier": "standard",
      "target-role-agent": "fam-eng-backend", "workspace": "_sandbox", "target-repo": "repo",
      "objective": "x", "output-format": "y", "allowed-tools": ["ALL"], "task-boundaries": "b",
      "non-goals": ["n"], "must-read": ["none"], "inherited-decisions": [],
      "acceptance-tests": ["none"], "evidence-plan": ["self-assertion"],
      "expected-output": {}, "token-budget": "unlimited"}
viol = CP.validate(ph)
check("placeholder package (none/ALL/unlimited/self-assertion) -> violations",
      len(viol) >= 4)
check("fake target-role-agent card -> violation",
      any("에이전트 카드" in v or "card" in v.lower() for v in CP.validate(dict(ph, **{"target-role-agent": "not-a-real-agent"}))))

print("== P0-5 report validator bypasses ==")
_hdr = {"bottom-line": "x", "decision-needed": {"needed": False},
        "confidence": {"value": "Med"}, "risks": [], "evidence": [{"grade": "E1", "source-uri": "README.md"}]}
_idn = {"report-id": "r", "workflow-id": "wf", "role-id": "EXEC-CTO"}
check("no report-type -> BLOCK", len(VR.validate({"report-header": _hdr})) > 0)
check("unknown report-type -> BLOCK", len(VR.validate({"report-type": "made-up", **_idn, "report-header": _hdr})) > 0)
check("build w/ empty kind + null verification + null verification-performed -> BLOCK",
      len(VR.validate({"report-type": "build", **_idn, "report-header": _hdr,
                       "primary-artifacts": [{"path": "README.md", "kind": "", "verification": None}],
                       "verification-performed": None})) > 0)
check("E3 self-report cmd (no receipt) + High confidence -> BLOCK (overconfidence)",
      len(VR.validate({"report-type": "work", **_idn, "work-summary": "s",
                       "report-header": dict(_hdr, **{"confidence": {"value": "High"},
                       "evidence": [{"grade": "E3", "command": "pytest -q", "exit-code": 0}]})})) > 0)
_abs_company = os.path.join(ROOT, "org-os", "01-company", "company-context.yaml")
check("E3 abs-path company ref @ High -> capped (BLOCK)",
      len(VR.validate({"report-type": "work", **_idn, "work-summary": "s",
                       "report-header": dict(_hdr, **{"confidence": {"value": "High"},
                       "evidence": [{"grade": "E3", "source-uri": _abs_company}]})})) > 0)
check("E3 CLAUDE.md @ High -> capped (BLOCK)",
      len(VR.validate({"report-type": "work", **_idn, "work-summary": "s",
                       "report-header": dict(_hdr, **{"confidence": {"value": "High"},
                       "evidence": [{"grade": "E3", "source-uri": "CLAUDE.md"}]})})) > 0)

print("== P0-6 receipt matching is exact (no substring/basename) ==")
receipts = [{"command": "python3 -m pytest tests && echo done", "exit_code": 0},
            {"artifact_path": "/tmp/a/result.json", "artifact_sha256": "abc"}]
check("substring command claim ('echo') no longer matches full receipt",
      VR._cmd_receipt(receipts, "echo") is None)
check("exact command claim matches",
      VR._cmd_receipt([{"command": "pytest -q", "exit_code": 0}], "pytest -q") is not None)
check("basename artifact claim (different dir) no longer matches",
      VR._artifact_receipt(receipts, "/different/project/result.json", None) is None)

print(f"\n{passed} passed, {failed} failed")
sys.exit(0 if failed == 0 else 1)