#!/usr/bin/env python3 """P1 Tranche3 (#7 + #13) E2 — cascade 순서 교정 + 커맨드↔상태엔진 배선 테스트. standalone(no pytest). exit 0 = all pass. test_enforcement.py 는 건드리지 않는다(별 파일). 검증 대상: 1. cascade 순서가 ground(discovery) → decide 로 교정됐다(commands + collaboration-map 파싱). 2. 9개 cascade/wave 커맨드가 모두 state_engine.py 를 참조한다 (guard 진입 / complete-stage 완료 / enter-stage 진입). 3. /build 절차가 spec→build 설계 게이트를 위해 state_engine guard 를 호출한다(#13 핵심 게이트). 4. collaboration-map 에 DECIDE 앞에 discovery/GROUND phase 가 있다(families 는 실존 FAM-*). 5. wave(plan-wave/run-wave)가 별도 progress.yaml 이 아니라 통합 원장(workflow.yaml progress:)을 쓴다. 6. (functional) 교정된 순서의 엔진 게이트가 실제로 강제된다: discovery→decide 는 option-set(≥2), spec→build 는 must-read-designs Accepted 를 요구. """ import os import subprocess import sys import tempfile import hashlib import yaml ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) HOOKS = os.path.join(ROOT, ".claude", "hooks") CMDS = os.path.join(ROOT, ".claude", "commands") PY = sys.executable passed, failed = 0, 0 def check(name, ok): global passed, failed if ok: passed += 1 print(f" PASS {name}") else: failed += 1 print(f" FAIL {name}") def cmd(name): with open(os.path.join(CMDS, name + ".md"), encoding="utf-8") as fh: return fh.read() NINE = ["ground", "decide", "design", "spec", "build", "plan-wave", "run-wave", "review-output", "release-check"] # ============================================================ 2. 9 커맨드가 엔진을 참조 print("== all 9 cascade/wave commands reference state_engine.py ==") _texts = {c: cmd(c) for c in NINE} for c in NINE: check(f"{c}.md references state_engine.py", "state_engine.py" in _texts[c]) check(f"{c}.md calls guard (진입 게이트)", "state_engine.py guard" in _texts[c]) # 현재 stage 완료와 다음 stage 진입은 별도 API여야 한다(current ≠ last-completed). for c in NINE: check(f"{c}.md calls complete-stage", "complete-stage" in _texts[c]) check(f"{c}.md calls enter-stage", "enter-stage" in _texts[c]) check(f"{c}.md does not use deprecated transition", "state_engine.py transition" not in _texts[c]) # ============================================================ 1. cascade 순서 ground(discovery)→decide print("== cascade order corrected: ground(discovery) -> decide ==") _g, _d = _texts["ground"], _texts["decide"] # ground = discovery, 옵션셋 발산, 입력은 intake 브리프(결정 packet 아님) check("ground is discovery stage", "--to discovery" in _g and "discovery" in _g) check("ground produces option-set (발산, not decision)", "option-set" in _g) check("ground input is intake Decision Brief (not a decision packet)", "Decision Brief" in _g and ("결정 packet이 아니라" in _g.lower() or "결정 Packet이 아니라" in _g or "결정이 아니라" in _g)) check("ground hands off to /decide", "/decide" in _g) # decide = converge, discovery→decide, 입력은 ground option-set, 다음은 /design check("decide is decide stage (converge)", "--to decide" in _d and ("converge" in _d or "수렴" in _d)) check("decide reads discovery option-set as input", "option-set" in _d) check("decide hands off to /design (not back to /ground)", "/design" in _d and "다음**: `/ground`" not in _d) # 핸드오프 사슬: ground -> decide -> design -> spec -> build -> review-output -> release-check check("design hands off to /spec", "/spec" in _texts["design"]) check("spec hands off to /build", "/build" in _texts["spec"]) check("build hands off to /review-output", "/review-output" in _texts["build"]) check("review-output hands off to /release-check", "/release-check" in _texts["review-output"]) # ============================================================ 3. /build spec→build 설계 게이트 print("== /build guards spec->build design gate (#13) ==") _b = _texts["build"] check("build calls engine guard --to build", "state_engine.py guard --workflow --to build" in _b) check("build names the spec->build gate", "spec→build" in _b or "spec->build" in _b) check("build gate is must-read-designs (설계 미승인 차단)", "must-read-designs" in _b) # finding #8 light-path 보존: 단순 변경은 이 guard 를 건너뜀(false Blocked 금지) check("build preserves light-path skip (finding #8)", "light 경로는 이 guard를 호출하지 않는다" in _b) # ============================================================ 4. collaboration-map GROUND phase before DECIDE print("== collaboration-map: discovery/GROUND phase before DECIDE ==") _cm = yaml.safe_load(open(os.path.join(ROOT, "org-os/06-agent-work/collaboration-map.yaml")))["collaboration-map"] _phases = [p["phase"] for p in _cm["cascade-phases"]] check("cascade-phases has a GROUND phase", "GROUND" in _phases) check("GROUND precedes DECIDE", "GROUND" in _phases and "DECIDE" in _phases and _phases.index("GROUND") < _phases.index("DECIDE")) # test_enforcement 가 검증하는 DECIDE/DESIGN/BUILD 는 여전히 존재해야 한다 check("DECIDE/DESIGN/BUILD still present (test_enforcement 유지)", set(_phases) >= {"DECIDE", "DESIGN", "BUILD"}) _ground_phase = next(p for p in _cm["cascade-phases"] if p["phase"] == "GROUND") check("GROUND workflow-stage = discovery", _ground_phase.get("workflow-stage") == "discovery") check("GROUND produces grounding-evidence + option-set", "option-set" in str(_ground_phase.get("synthesis-output", ""))) check("GROUND next-input-to DECIDE", _ground_phase.get("next-input-to") == "DECIDE") # GROUND families 는 실존 FAM-* 여야(test_enforcement 의 family-ids-all-exist 와 일관) _fams = yaml.safe_load(open(os.path.join(ROOT, "org-os/00-role-registry/capability-families.yaml")))["capability-families"]["families"] _famids = {f["family-id"] for f in _fams} check("GROUND families all exist", set(_ground_phase.get("families", [])) <= _famids and len(_ground_phase.get("families", [])) >= 2) # 6 cross-group edges 불변(test_enforcement 유지) check("6 cross-group edges unchanged", len(_cm["cross-group-edges"]["edges"]) == 6) # ============================================================ 5. wave writes the unified ledger print("== wave uses unified ledger (workflow.yaml progress:), not a separate progress.yaml ==") _pw, _rw = _texts["plan-wave"], _texts["run-wave"] check("plan-wave writes progress into unified ledger", "state_engine.py progress" in _pw and "workflow.yaml" in _pw) check("plan-wave abolishes separate progress.yaml", "별도 `progress.yaml`을 만들지 않는다" in _pw) check("run-wave updates progress via engine", "state_engine.py progress" in _rw and "workflow.yaml" in _rw) check("run-wave routes via unified ledger progress.next", "progress.next" in _rw) check("light path formalized (run-wave without plan-wave)", "light" in _rw and "intake→run" in _rw) # ============================================================ 6. functional: gates actually enforce print("== functional: engine enforces corrected-order gates ==") with tempfile.TemporaryDirectory() as WS: 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) def _guard(wf, to): e = dict(os.environ) e["CLAUDE_PROJECT_DIR"] = ROOT e["ORGOS_WORKSPACE"] = WS # 절대경로 -> _workspace 가 root 로 사용 return subprocess.run( [PY, os.path.join(HOOKS, "state_engine.py"), "guard", "--workflow", wf, "--to", to], capture_output=True, text=True, env=e, ) def _engine(*args): e = dict(os.environ) e["CLAUDE_PROJECT_DIR"] = ROOT e["ORGOS_WORKSPACE"] = WS return subprocess.run([PY, os.path.join(HOOKS, "state_engine.py"), *args], capture_output=True, text=True, env=e) def _write_artifact(wf, artifact_id, kind, stage, producer, payload): directory = os.path.join(WS, "completion-records", wf) os.makedirs(directory, exist_ok=True) path = os.path.join(directory, f"{artifact_id}.report.yaml") ledger_path = os.path.join(WS, "state", wf, "workflow.yaml") ledger = yaml.safe_load(open(ledger_path, encoding="utf-8")) if os.path.exists(ledger_path) else {} report = { "report-type": "workflow-artifact", "artifact-kind": kind, "artifact-version": 1, "tier": ledger.get("tier", "light"), "identity": {"artifact-id": artifact_id, "workflow-id": wf, "stage": stage, "producer-role-id": producer}, "payload": payload, "report-header": { "bottom-line": f"{kind} contract fixture", "decision-needed": {"needed": False}, "confidence": {"value": "Med", "derived-from": "evidence"}, "risks": [], "evidence": [{"source-uri": "README.md", "grade": "E3"}], }, } with open(path, "w", encoding="utf-8") as fh: yaml.safe_dump(report, fh, allow_unicode=True, sort_keys=False) return path def _submit(wf, artifact_id, kind, stage, producer, payload, reviewer=None): # This legacy functional fixture skips unrelated upstream gates. Keep # its materialized current stage truthful before using the trusted # stage-scoped submit API. ledger_path = os.path.join(WS, "state", wf, "workflow.yaml") ledger = yaml.safe_load(open(ledger_path, encoding="utf-8")) or {} ledger["stage"] = stage ledger["stage-status"] = "running" with open(ledger_path, "w", encoding="utf-8") as fh: yaml.safe_dump(ledger, fh, allow_unicode=True, sort_keys=False) path = _write_artifact(wf, artifact_id, kind, stage, producer, payload) submitted = _engine("submit-artifact", "--workflow", wf, "--report", path, "--actor", "OPS-ORCH") assert submitted.returncode == 0, submitted.stderr if reviewer: reviewed = _engine("review-artifact", "--workflow", wf, "--report", path, "--decision", "accepted", "--reviewer", reviewer) assert reviewed.returncode == 0, reviewed.stderr return path, hashlib.sha256(open(path, "rb").read()).hexdigest() # discovery -> decide 는 option-set(>=2) 없으면 차단, 있으면 허용 _write_ledger("wf-noopt", {"workflow-id": "wf-noopt", "stage": "discovery", "plan": "cascade", "tier": "standard", "facts": {"grounding-evidence-present": True}}) check("guard discovery->decide BLOCK without option-set (exit 2)", _guard("wf-noopt", "decide").returncode == 2) _write_ledger("wf-opt", {"workflow-id": "wf-opt", "stage": "discovery", "plan": "cascade", "tier": "light", "grounding-evidence": True}) _submit("wf-opt", "wf-opt-ground", "grounding-package", "discovery", "STR-ANALYST", {"problem-structure": {"question": "ICP 범위"}, "analysis-synthesis": {"insight": "범위와 비용의 trade-off"}, "options": [{"id": "A", "problem": "좁은 ICP", "tradeoffs": ["scope"], "evidence-refs": ["README.md"]}, {"id": "B", "problem": "넓은 ICP", "tradeoffs": ["cost"], "evidence-refs": ["README.md"]}], "evidence": ["README.md"], "source-contributions": [ {"report-id": f"untrusted-source-{index}", "report-ref": f"fixture/source-{index}.report.yaml", "report-sha256": str(index) * 64, "producer-role-id": "EXEC-CEO", "context-package-ref": f"fixture/context-{index}.yaml", "context-package-sha256": str(index + 3) * 64, "assigned-lens": lens, "producer-run-id": f"fixture-run-{index}"} for index, lens in enumerate(["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH"], 1)], "lens-coverage": {"required-min": 3, "covered": ["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH"], "contrarian-report-id": None}}) check("guard discovery->decide BLOCK when lens sources are not exact trusted reports", _guard("wf-opt", "decide").returncode == 2) # spec -> build 는 must-read-designs 전부 Accepted 여야 허용(#13 핵심 게이트). # P0-4: 승인은 원장 자기신고 review-state 가 아니라 acceptance_log(검증된 이벤트)로만 파생. def _seed_acceptance(events): import json as _json 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") # 미승인: 아티팩트만 등재(승인 이벤트 없음) -> 차단 _arts = [{"path": "cr/spec.report.yaml", "design-type": "feature-spec", "report-id": "nd-spec"}] _write_ledger("wf-nodesign", {"workflow-id": "wf-nodesign", "stage": "spec", "plan": "cascade", "tier": "standard", "build-families": ["FAM-ENG-BACKEND"], "artifacts": _arts}) _res = _guard("wf-nodesign", "build") check("guard spec->build BLOCK when designs not Accepted (exit 2 + unmet on stderr)", _res.returncode == 2 and ("PRD" in _res.stderr or "must-read" in _res.stderr)) _write_ledger("wf-design-ok", {"workflow-id": "wf-design-ok", "stage": "spec", "plan": "cascade", "tier": "light", "artifacts": []}) _submit("wf-design-ok", "ok-profile", "workload-profile", "intake", "EXEC-CEO", { "surfaces": {"ui": False, "public-api": False, "persistence": False, "infrastructure": False}, "risk": {"security-bearing": False, "data-migration": False, "external-side-effect": False, "risk-level": "Low", "reversibility": "two-way-door", "blast-radius": "single-role", "privacy": False, "regulatory": False, "slo-impact": False}, "required-capabilities": ["product-delivery"], "product-feature": True, }) packet_path, packet_sha = _submit( "wf-design-ok", "ok-packet", "executive-decision-packet", "decide", "EXEC-CEO", {"recommendation": "option A"}, "HUMAN-001") design_basis = {"basis-artifact-id": "ok-packet", "basis-artifact-sha256": packet_sha} design_path, design_sha = _submit( "wf-design-ok", "ok-overall", "overall-design", "design", "ARCH-SOLUTION", design_basis, "ARCH-EA") spec_basis = {"basis-artifact-id": "ok-overall", "basis-artifact-sha256": design_sha} _submit("wf-design-ok", "ok-prd", "prd", "spec", "PROD-PM", spec_basis, "PROD-PO") _submit("wf-design-ok", "ok-ac", "acceptance-criteria", "spec", "PROD-PM", {**spec_basis, "criteria": [{"criterion-id": "AC-1", "preconditions": [], "input": {}, "expected-result": "works", "risk-level": "Low", "verification-method": "automated-test"}]}, "PROD-PO") check("guard spec->build ALLOW when all must-read-designs Accepted (exit 0)", _guard("wf-design-ok", "build").returncode == 0) # wave: 통합 원장 progress 를 엔진 helper 로 기록(별도 파일 아님) -> workflow.yaml progress: 에 반영 e = dict(os.environ); e["CLAUDE_PROJECT_DIR"] = ROOT; e["ORGOS_WORKSPACE"] = WS subprocess.run([PY, os.path.join(HOOKS, "state_engine.py"), "init", "--workflow", "wf-wave", "--plan", "wave"], capture_output=True, text=True, env=e) subprocess.run([PY, os.path.join(HOOKS, "state_engine.py"), "progress", "--workflow", "wf-wave", "--round", "2", "--next", "FAM-ENG-BACKEND", "--progressing", "true"], capture_output=True, text=True, env=e) _lp = os.path.join(WS, "state", "wf-wave", "workflow.yaml") _led = yaml.safe_load(open(_lp)) if os.path.exists(_lp) else {} check("wave progress written INTO unified workflow.yaml (progress:)", _led.get("progress", {}).get("round") == 2 and _led.get("progress", {}).get("next") == "FAM-ENG-BACKEND") check("no separate progress.yaml created for the workflow", not os.path.exists(os.path.join(WS, "state", "wf-wave", "progress.yaml"))) print(f"\n{passed} passed, {failed} failed") sys.exit(1 if failed else 0)