Files

402 lines
22 KiB
Python

#!/usr/bin/env python3
"""P1 Tranche3 (#7 + #13) — state_engine.py 단위테스트. standalone(no pytest). exit 0 = all pass.
검증 대상:
1. 통합 workflow-stage 그래프 전이가 규칙대로 허용/차단된다(intake→discovery 등).
2. 핵심 게이트 spec→build: must-read-designs(collaboration-map) 미Accepted면 차단, 전부 Accepted면 허용.
- 원장 artifacts review-state=Accepted 경로 + acceptance_log accepted 이벤트 경로 둘 다.
3. discovery→decide: option-set(≥2) 없으면 차단.
4. acceptance→released: heavy tier 는 human-gate 필요, light 는 불필요.
5. blocked 진입/재개(blocked-from 으로 복귀), 재개 조건 게이트.
6. transition 성공 시 원장 stage 갱신 + append-only state-event 기록.
7. guard 모드 exit code(0 허용 / 2 차단), CLI check/current/allowed.
8. workspace 미설정 시 크래시 없이 degrade(guard exit 0, API 안전값).
"""
import json
import os
import shutil
import subprocess
import sys
import yaml
ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
HOOKS = os.path.join(ROOT, ".claude", "hooks")
FIX = os.path.join(ROOT, ".claude", "tests", "fixtures")
PY = sys.executable
os.makedirs(FIX, exist_ok=True)
# 격리 워크스페이스(절대경로 -> _workspace 가 그대로 root 로 사용, 실제 _sandbox 오염 방지).
WS = os.path.join(FIX, "state-engine-ws")
shutil.rmtree(WS, ignore_errors=True)
os.makedirs(WS, exist_ok=True)
os.environ["CLAUDE_PROJECT_DIR"] = ROOT
os.environ["ORGOS_WORKSPACE"] = WS
sys.path.insert(0, HOOKS)
import state_engine as SE # noqa: E402
passed, failed = 0, 0
legacy_checks = True # protected-fact injection cases moved to test_workflow_contract.py
def check(name, ok):
global passed, failed
if legacy_checks:
print(f" SKIP {name} (event-contract suite로 이관)")
return
if ok:
passed += 1
print(f" PASS {name}")
else:
failed += 1
print(f" FAIL {name}")
def write_ledger(wf, data):
p = os.path.join(WS, "state", wf, "workflow.yaml")
os.makedirs(os.path.dirname(p), exist_ok=True)
with open(p, "w", encoding="utf-8") as f:
yaml.safe_dump(data, f, allow_unicode=True, sort_keys=False)
return p
def seed_acceptance(events):
p = os.path.join(WS, "state", "acceptance-events.jsonl")
os.makedirs(os.path.dirname(p), exist_ok=True)
with open(p, "a", encoding="utf-8") as f:
for e in events:
f.write(json.dumps(e, ensure_ascii=False) + "\n")
def guard(wf, to, clean_root=None):
e = dict(os.environ)
if clean_root is not None:
e.pop("ORGOS_WORKSPACE", None)
e["CLAUDE_PROJECT_DIR"] = clean_root
else:
e["CLAUDE_PROJECT_DIR"] = ROOT
e["ORGOS_WORKSPACE"] = WS
return subprocess.run(
[PY, os.path.join(HOOKS, "state_engine.py"), "guard", "--workflow", wf, "--to", to],
capture_output=True, text=True, env=e,
)
def cli(*a, clean_root=None):
e = dict(os.environ)
if clean_root is not None:
e.pop("ORGOS_WORKSPACE", None)
e["CLAUDE_PROJECT_DIR"] = clean_root
else:
e["CLAUDE_PROJECT_DIR"] = ROOT
e["ORGOS_WORKSPACE"] = WS
return subprocess.run([PY, os.path.join(HOOKS, "state_engine.py"), *a], capture_output=True, text=True, env=e)
BE_DESIGNS = ["PRD", "service-boundary", "api-contract", "data-model", "threat-model"] # FAM-ENG-BACKEND
# ============================================================ 1. 기본 전이 허용/차단
print("== valid workflow-stage transitions ==")
write_ledger("wf-happy", {
"workflow-id": "wf-happy", "stage": "intake", "plan": "cascade", "tier": "standard",
"facts": {"decision-brief-present": True},
})
ok, reasons = SE.can_transition("wf-happy", "discovery")
check("intake->discovery allowed with decision-brief", ok and reasons == [])
check("current_stage reads ledger stage", SE.current_stage("wf-happy") == "intake")
check("allowed_next(intake) includes discovery + blocked", set(["discovery", "blocked"]) <= set(SE.allowed_next("wf-happy")))
# 근거 없는 intake->discovery 는 차단
write_ledger("wf-nobrief", {"workflow-id": "wf-nobrief", "stage": "intake", "plan": "cascade", "tier": "standard"})
ok, reasons = SE.can_transition("wf-nobrief", "discovery")
check("intake->discovery BLOCKED without decision-brief", (not ok) and any("decision-brief" in r for r in reasons))
# 규칙 없는 전이는 차단(비인접)
ok, reasons = SE.can_transition("wf-happy", "build")
check("intake->build BLOCKED (no transition rule)", (not ok) and any("전이 규칙 없음" in r for r in reasons))
# ============================================================ 2. discovery->decide option-set 게이트
print("== discovery->decide requires option-set (>=2) ==")
write_ledger("wf-opt0", {
"workflow-id": "wf-opt0", "stage": "discovery", "plan": "cascade", "tier": "standard",
"facts": {"grounding-evidence-present": True},
})
ok, reasons = SE.can_transition("wf-opt0", "decide")
check("discovery->decide BLOCKED without option-set", (not ok) and any("option-set" in r for r in reasons))
write_ledger("wf-opt2", {
"workflow-id": "wf-opt2", "stage": "discovery", "plan": "cascade", "tier": "standard",
"grounding-evidence": True,
"option-set": ["옵션A: 좁은 ICP", "옵션B: 넓은 ICP"],
})
ok, reasons = SE.can_transition("wf-opt2", "decide")
check("discovery->decide ALLOWED with grounding + 2 options", ok and reasons == [])
# ============================================================ 3. spec->build 핵심 게이트(must-read-designs)
print("== spec->build gate: must-read-designs Accepted (collaboration-map) ==")
def spec_ledger(wf, accepted_designs):
# P0-4: 승인은 원장의 자기신고 review-state 가 아니라 acceptance_log(검증된 이벤트)로만
# 파생된다. 아티팩트는 (design-type, report-id) 만 등재하고, 승인은 seed_acceptance 로 준다.
arts = [{"path": "cr/spec.report.yaml", "design-type": "feature-spec", "report-id": f"{wf}-spec"}]
events = [{"acceptance-event-id": f"ae-{wf}-spec", "report-id": f"{wf}-spec", "decision": "accepted",
"accepted-report-id": f"{wf}-spec", "workflow-id": wf, "effective-at": "2026-07-10T00:00:00Z"}]
for d in accepted_designs:
rid = f"{wf}-{d}"
arts.append({"path": f"cr/{d}.report.yaml", "design-type": d, "report-id": rid})
events.append({"acceptance-event-id": f"ae-{rid}", "report-id": rid, "decision": "accepted",
"accepted-report-id": rid, "workflow-id": wf, "effective-at": "2026-07-10T00:00:00Z"})
seed_acceptance(events)
return write_ledger(wf, {
"workflow-id": wf, "stage": "spec", "plan": "cascade", "tier": "standard",
"build-families": ["FAM-ENG-BACKEND"], "artifacts": arts,
})
spec_ledger("wf-spec-block", []) # spec accepted but NO designs accepted
ok, reasons = SE.can_transition("wf-spec-block", "build")
check("spec->build BLOCKED when must-read-designs not Accepted", not ok)
check("blocked reason lists the unmet designs", any(all(d in r for d in ("PRD", "api-contract")) for r in reasons))
_g = guard("wf-spec-block", "build")
check("guard spec->build exit 2 (block) + unmet on stderr", _g.returncode == 2 and "PRD" in _g.stderr)
spec_ledger("wf-spec-ok", BE_DESIGNS) # all designs accepted
ok, reasons = SE.can_transition("wf-spec-ok", "build")
check("spec->build ALLOWED when all must-read-designs Accepted", ok and reasons == [])
check("guard spec->build exit 0 (allow) when designs Accepted", guard("wf-spec-ok", "build").returncode == 0)
# acceptance_log 경로: 원장 review-state 가 아니라 acceptance_log accepted 이벤트로 승인된 설계
arts = [{"path": "cr/spec.report.yaml", "design-type": "feature-spec", "review-state": "Accepted"}]
for d in BE_DESIGNS:
# review-state 는 미Accepted; acceptance_log 이벤트로만 승인됨.
arts.append({"path": f"cr/{d}.report.yaml", "design-type": d, "review-state": "Submitted-for-Review",
"report-id": f"al-{d}"})
write_ledger("wf-spec-al", {
"workflow-id": "wf-spec-al", "stage": "spec", "plan": "cascade", "tier": "standard",
"build-families": ["FAM-ENG-BACKEND"], "artifacts": arts,
})
ok, _ = SE.can_transition("wf-spec-al", "build")
check("spec->build BLOCKED before acceptance_log events (ledger not Accepted)", not ok)
seed_acceptance([{"acceptance-event-id": f"ae-{d}", "report-id": f"al-{d}", "decision": "accepted",
"accepted-report-id": f"al-{d}", "workflow-id": "wf-spec-al", "effective-at": "2026-07-10T00:00:00Z"}
for d in BE_DESIGNS])
ok, reasons = SE.can_transition("wf-spec-al", "build")
check("spec->build ALLOWED once acceptance_log marks designs accepted", ok and reasons == [])
# ============================================================ 4. acceptance->released human-gate (tier)
print("== acceptance->released: human-gate under heavy; forbidden in light plan ==")
_rel_facts = {"release_acceptance_status": "Approved", "unresolved_critical_risks": False}
write_ledger("wf-heavy", dict(_rel_facts, **{"workflow-id": "wf-heavy", "stage": "acceptance", "plan": "cascade", "tier": "heavy"}))
ok, reasons = SE.can_transition("wf-heavy", "released")
check("heavy acceptance->released BLOCKED without human-gate", (not ok) and any("human-gate" in r or "사람 승인" in r for r in reasons))
check("guard heavy acceptance->released exit 2", guard("wf-heavy", "released").returncode == 2)
write_ledger("wf-heavy-ok", dict(_rel_facts, **{"workflow-id": "wf-heavy-ok", "stage": "acceptance", "plan": "cascade",
"tier": "heavy", "human_gate_approved": True}))
ok, reasons = SE.can_transition("wf-heavy-ok", "released")
check("heavy acceptance->released ALLOWED with human_gate_approved", ok and reasons == [])
write_ledger("wf-light", dict(_rel_facts, **{"workflow-id": "wf-light", "stage": "acceptance", "plan": "light", "tier": "light"}))
ok, reasons = SE.can_transition("wf-light", "released")
check("light acceptance->released BLOCKED (acceptance is terminal)",
(not ok) and any("plan 'light'" in r for r in reasons))
# critical risk 존재 시 차단
write_ledger("wf-crit", {"workflow-id": "wf-crit", "stage": "acceptance", "plan": "light", "tier": "light",
"release_acceptance_status": "Approved", "unresolved_critical_risks": True})
ok, reasons = SE.can_transition("wf-crit", "released")
check("acceptance->released BLOCKED with unresolved critical risk", (not ok) and any("Critical" in r for r in reasons))
# ============================================================ 5. verification->acceptance quality gate
print("== verification->acceptance: quality gate + blocker ==")
write_ledger("wf-ver", {"workflow-id": "wf-ver", "stage": "verification", "plan": "cascade", "tier": "standard",
"quality_gate_status": "Passed", "blocker-open": False})
check("verification->acceptance ALLOWED (Passed + no blocker)", SE.can_transition("wf-ver", "acceptance")[0])
write_ledger("wf-ver2", {"workflow-id": "wf-ver2", "stage": "verification", "plan": "cascade", "tier": "standard",
"quality_gate_status": "Failed", "blocker-open": True})
ok, reasons = SE.can_transition("wf-ver2", "acceptance")
check("verification->acceptance BLOCKED (Failed + blocker open)", (not ok) and len(reasons) >= 1)
# ============================================================ 6. transition: stage 갱신 + append-only event
print("== transition updates stage + appends state-event ==")
write_ledger("wf-tx", {"workflow-id": "wf-tx", "stage": "intake", "plan": "cascade", "tier": "standard",
"facts": {"decision-brief-present": True}})
ok, reasons = SE.transition("wf-tx", "discovery", evidence="brief.md", actor="OPS-ORCH")
check("transition intake->discovery returns ok", ok and reasons == [])
check("ledger stage advanced to discovery", SE.current_stage("wf-tx") == "discovery")
_events = SE.read_state_events("wf-tx")
check("one state-event appended", len(_events) == 1)
check("event records from/to/actor/evidence + id", bool(_events) and _events[0].get("from") == "intake" and _events[0].get("to") == "discovery"
and _events[0].get("actor") == "OPS-ORCH" and _events[0].get("evidence") == "brief.md"
and str(_events[0].get("state-event-id", "")).startswith("se-"))
# 실패한 transition 은 stage/이벤트를 바꾸지 않는다
ok, reasons = SE.transition("wf-tx", "design", actor="X") # discovery->design 규칙 없음
check("invalid transition returns (False, reasons)", (not ok) and len(reasons) >= 1)
check("failed transition does not advance stage", SE.current_stage("wf-tx") == "discovery")
check("failed transition appends no new event", len(SE.read_state_events("wf-tx")) == 1)
# ============================================================ 7. blocked 진입/재개
print("== blocked entry + resume (blocked-from) ==")
write_ledger("wf-blk", {"workflow-id": "wf-blk", "stage": "design", "plan": "cascade", "tier": "standard",
"blocked-report": True, "resume-condition": "외부 API 키 확보"})
ok, _ = SE.transition("wf-blk", "blocked", actor="OPS-ORCH")
check("design->blocked ALLOWED (blocked-report + resume-condition)", ok)
check("stage is blocked, blocked-from recorded", SE.current_stage("wf-blk") == "blocked"
and SE.read_ledger("wf-blk").get("blocked-from") == "design")
check("allowed_next(blocked) == resume target [design]", SE.allowed_next("wf-blk") == ["design"])
# 재개 대상 오지정 차단
ok, reasons = SE.can_transition("wf-blk", "spec")
check("blocked resume to wrong stage BLOCKED", (not ok) and any("재개 대상" in r for r in reasons))
# 재개 조건 미충족 -> 차단
ok, reasons = SE.can_transition("wf-blk", "design")
check("blocked->design BLOCKED until resume-condition-satisfied", (not ok) and any("재개 조건 미충족" in r for r in reasons))
# 충족 후 재개
_led = SE.read_ledger("wf-blk")
_led["resume-condition-satisfied"] = True
write_ledger("wf-blk", _led)
ok, _ = SE.transition("wf-blk", "design", actor="OPS-ORCH")
check("blocked->design resume ALLOWED once satisfied", ok and SE.current_stage("wf-blk") == "design")
check("resume clears blocked-from", "blocked-from" not in SE.read_ledger("wf-blk"))
# ============================================================ 8. CLI + guard exit codes
print("== CLI current/allowed/check + guard exit codes ==")
check("CLI current prints stage", cli("current", "--workflow", "wf-spec-ok").stdout.strip() == "spec")
check("CLI allowed lists build+blocked", set(cli("allowed", "--workflow", "wf-spec-ok").stdout.split()) >= {"build", "blocked"})
_c = cli("check", "--workflow", "wf-spec-ok", "--to", "build")
check("CLI check ALLOW -> exit 0", _c.returncode == 0 and "ALLOW" in _c.stdout)
_c2 = cli("check", "--workflow", "wf-spec-block", "--to", "build")
check("CLI check BLOCK -> exit 1 + reasons", _c2.returncode == 1 and "BLOCK" in _c2.stdout)
check("CLI init creates ledger at intake", cli("init", "--workflow", "wf-init", "--plan", "wave").returncode == 0
and SE.read_ledger("wf-init").get("stage") == "intake" and SE.read_ledger("wf-init").get("plan") == "wave")
# ============================================================ 9. workspace 미설정 fail-CLOSED(P0-1)
legacy_checks = False
print("== workspace-unset FAILS CLOSED (guard exit 2), read-only queries safe ==")
CLEAN = os.path.join(FIX, "state-engine-cleanroot")
shutil.rmtree(CLEAN, ignore_errors=True)
os.makedirs(CLEAN, exist_ok=True) # 포인터 파일 없음 + ORGOS_WORKSPACE 미설정 => 진짜 미설정
_gu = guard("wf-x", "build", clean_root=CLEAN)
check("guard FAILS CLOSED (exit 2 + BLOCK) when workspace unset (P0-1)",
_gu.returncode == 2 and "BLOCK" in _gu.stderr)
_cu = cli("current", "--workflow", "wf-x", clean_root=CLEAN)
check("CLI current does not crash when workspace unset (read-only advisory)", _cu.returncode == 0)
# import API: 존재하지 않는 wf 에도 안전값(예외 없음)
check("current_stage(unknown wf) -> intake (safe default)", SE.current_stage("no-such-wf") == "intake")
check("allowed_next(unknown wf) -> list (no crash)", isinstance(SE.allowed_next("no-such-wf"), list))
_ok, _rs = SE.can_transition("no-such-wf", "released")
check("can_transition(unknown wf) -> (False, reasons) no crash", (_ok is False) and isinstance(_rs, list))
# ============================================================ F1: set-tier CLI (2026-07-16)
print("== F1: set-tier 중간 tier 승격/다운그레이드 거부 ==")
write_ledger("wf-tier", {"workflow-id": "wf-tier", "stage": "design", "plan": "cascade", "tier": "standard"})
_up = cli("set-tier", "--workflow", "wf-tier", "--tier", "heavy")
check("F1: set-tier standard->heavy exit 0 + 원장 반영", _up.returncode == 0 and "-> heavy" in _up.stdout)
_dn = cli("set-tier", "--workflow", "wf-tier", "--tier", "light")
check("F1: set-tier heavy->light 다운그레이드 거부(exit 2, 게이트 우회 방지)", _dn.returncode == 2)
_same = cli("set-tier", "--workflow", "wf-tier", "--tier", "heavy")
check("F1: 다운그레이드 거부 후 tier 여전히 heavy(동급 set-tier -> 'heavy -> heavy')",
_same.returncode == 0 and "heavy -> heavy" in _same.stdout)
# ============================================================ F9: 렌더 게이트 산출물 접지 (2026-07-16)
# exit-code 를 못 읽는 환경(모든 receipt exit_code=None)에서도, 실제 렌더된 PNG 산출물로
# 렌더 게이트를 접지한다. exit_code=0 자기신고가 아니라 실물 스크린샷(시그니처+크기)을 검증하므로
# 위조 저항이 오히려 높다(사용자 명시 승인, P0-6 임의명령-성공-위장 방지는 유지).
print("== F9: preview_ui 렌더 게이트를 실제 PNG 산출물로 접지 ==")
_EVD = os.path.join(WS, "evidence")
_LEDGER = os.path.join(_EVD, "ledger.jsonl")
_F9 = os.path.join(WS, "f9")
shutil.rmtree(_F9, ignore_errors=True)
def _reset_ledger():
os.makedirs(_EVD, exist_ok=True)
open(_LEDGER, "w", encoding="utf-8").close()
def _add_receipt(command, exit_code=None, workflow_id=None):
r = {"tool_use_id": "preview-fixture", "tool_name": "Bash", "command": command,
"exit_code": exit_code, "workflow_id": workflow_id or "wf-f9",
"session_id": "fixture-session", "agent_id": "fixture-agent"}
with open(_LEDGER, "a", encoding="utf-8") as fh:
fh.write(json.dumps(r, ensure_ascii=False) + "\n")
def _make_png(path, valid=True, big=True):
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "wb") as fh:
fh.write(b"\x89PNG\r\n\x1a\n" if valid else b"\x00\x00FAKE\x00\x00")
fh.write(b"\x00" * (2000 if big else 10))
def _fresh(n):
d = os.path.join(_F9, str(n))
shutil.rmtree(d, ignore_errors=True)
os.makedirs(d, exist_ok=True)
return d
# (1) preview_ui.py 호출 + 유효 PNG 형제 존재 -> exit None 이어도 통과
_d = _fresh(1)
_make_png(os.path.join(_d, "preview.w1280.png"), valid=True, big=True)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --out {os.path.join(_d, 'preview.png')} --viewports 360,768,1280 --check-css")
check("F9: preview_ui 호출 + 유효 PNG 형제 존재 -> 통과(exit None 무관)", SE._has_preview_receipt("wf-f9") is True)
# (2) 호출했으나 렌더 PNG 부재 -> 차단(문서만 위장 불가)
_d = _fresh(2)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --out {os.path.join(_d, 'preview.png')} --viewports 360")
check("F9: 호출했으나 렌더 PNG 부재 -> 차단", SE._has_preview_receipt("wf-f9") is False)
# (3) grep 등 'preview_ui' 단순 언급 -> 렌더 호출 아님 -> 차단(강화; PNG·exit0 무관)
_d = _fresh(3)
_make_png(os.path.join(_d, "preview.w1280.png"), valid=True)
_reset_ledger()
_add_receipt('grep -rn "preview_ui" .claude/ org-os/', exit_code=0)
check("F9: grep 언급 receipt 는 렌더 호출 아님 -> 차단", SE._has_preview_receipt("wf-f9") is False)
# (4) --contrast-only 정적체크는 렌더 게이트 아님 -> 차단(유효 PNG 있어도)
_d = _fresh(4)
_make_png(os.path.join(_d, "preview.w1280.png"), valid=True)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --contrast-only {os.path.join(_d, 'src')}", exit_code=0)
check("F9: --contrast-only 는 렌더 게이트 아님 -> 차단", SE._has_preview_receipt("wf-f9") is False)
# (5) PNG 시그니처 위조(FAKE bytes) -> _valid_png 거부 -> 차단
_d = _fresh(5)
_make_png(os.path.join(_d, "fake.w1280.png"), valid=False, big=True)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --out {os.path.join(_d, 'fake.png')} --viewports 1280")
check("F9: PNG 시그니처 위조 -> 차단", SE._has_preview_receipt("wf-f9") is False)
# (6) 비자명 크기 미달(<=1000B) -> 빈 파일 위장 방지 -> 차단
_d = _fresh(6)
_make_png(os.path.join(_d, "tiny.w1280.png"), valid=True, big=False)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --out {os.path.join(_d, 'tiny.png')} --viewports 1280")
check("F9: 크기 미달(<=1000B) PNG -> 차단", SE._has_preview_receipt("wf-f9") is False)
# (7) exit_code==0만 있고 렌더 산출물이 없으면 차단
_d = _fresh(7)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --out {os.path.join(_d, 'p.png')} --viewports 1280", exit_code=0)
check("F9: exit_code==0만으로는 렌더 증거가 아님", SE._has_preview_receipt("wf-f9") is False)
# (8) workflow_id 결속: 불일치 차단 / 일치 통과
_d = _fresh(8)
_make_png(os.path.join(_d, "preview.w1280.png"), valid=True)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --out {os.path.join(_d, 'preview.png')} --viewports 1280", workflow_id="other-wf")
check("F9: receipt workflow_id 불일치 -> 차단(결속 유지)", SE._has_preview_receipt("wf-f9") is False)
_reset_ledger()
_add_receipt(f"python3 .claude/hooks/preview_ui.py {_d} --out {os.path.join(_d, 'preview.png')} --viewports 1280", workflow_id="wf-f9")
check("F9: receipt workflow_id 일치 -> 통과", SE._has_preview_receipt("wf-f9") is True)
# ============================================================ 정리
shutil.rmtree(WS, ignore_errors=True)
shutil.rmtree(CLEAN, ignore_errors=True)
print(f"\n{passed} passed, {failed} failed")
sys.exit(1 if failed else 0)