274 lines
16 KiB
Python
274 lines
16 KiB
Python
#!/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 <wf> --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)
|