183 lines
8.8 KiB
Python
183 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""test_p2_orchestrator.py — 항목4: 얇은 end-to-end 오케스트레이터(state_engine next + /run-cascade).
|
|
|
|
리뷰 항목4: 상위 오케스트레이터가 없어 사람이 각 stage 커맨드를 수동 호출해야 했다. `state_engine.py
|
|
next` 는 현재/다음 stage·guard·사람게이트·커맨드를 결정론적으로 반환하는 얇은 조회층이고,
|
|
`/run-cascade` 는 그걸 따라 걷되 **사람 결정 지점에서 멈춘다**(자동 승인·완주 금지). 검증:
|
|
1. next_info: current/next stage·command, terminal, advance{ok,reasons,human-gate} 정확.
|
|
2. _human_gate_for: DECIDE go/no-go · RELEASE 는 human-gate, 그 외 전이는 아님.
|
|
3. next CLI 가 유효 JSON.
|
|
4. guard block 이 advance.reasons 로 노출(선행조건 미충족).
|
|
5. resumable: 사람 signoff 후 human-gate 해제.
|
|
6. terminal 정확.
|
|
7. /run-cascade 배선: state_engine next 재사용 + 사람게이트 정지 + 자동승인 금지.
|
|
|
|
실행: CLAUDE_PROJECT_DIR="$PWD" python3 .claude/tests/test_p2_orchestrator.py
|
|
"""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
|
|
ROOT = os.environ.get("CLAUDE_PROJECT_DIR") or 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)
|
|
|
|
_WS = tempfile.mkdtemp(prefix="p2orch_")
|
|
os.environ["CLAUDE_PROJECT_DIR"] = ROOT
|
|
os.environ["ORGOS_WORKSPACE"] = _WS
|
|
|
|
import state_engine as SE # noqa: E402
|
|
import acceptance_log as AL # noqa: E402
|
|
|
|
PY = sys.executable
|
|
SEP = os.path.join(HOOKS, "state_engine.py")
|
|
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 _cli(args):
|
|
e = dict(os.environ)
|
|
return subprocess.run([PY, SEP] + args, capture_output=True, text=True, env=e)
|
|
|
|
|
|
# ── 1) next_info: 신선한 cascade 워크플로(intake) ───────────────────────
|
|
print("== next_info: intake 상태 ==")
|
|
SE.init_ledger("wf-a", plan="cascade", tier="standard", overwrite=True)
|
|
info = SE.next_info("wf-a")
|
|
check("current-stage=intake", info.get("current-stage") == "intake")
|
|
check("current-command=/ceo-intake", info.get("current-command") == "/ceo-intake")
|
|
check("next-stage=discovery", info.get("next-stage") == "discovery")
|
|
check("next-command=/ground", info.get("next-command") == "/ground")
|
|
check("not terminal", info.get("terminal") is False)
|
|
adv = info.get("advance", {})
|
|
check("advance present", isinstance(adv, dict) and "ok" in adv)
|
|
check("intake->discovery blocked (no decision-brief)", adv.get("ok") is False)
|
|
check("guard reason surfaces decision-brief", any("decision-brief" in r for r in adv.get("reasons", [])))
|
|
check("intake->discovery not human-gate", adv.get("human-gate", {}).get("required") is False)
|
|
|
|
|
|
# ── 2) _human_gate_for ─────────────────────────────────────────────────
|
|
print("== _human_gate_for: DECIDE/RELEASE 만 사람게이트 ==")
|
|
hg_dec = SE._human_gate_for("decide", "design")
|
|
check("decide->design human-gate (go/no-go)", hg_dec[0] is True and hg_dec[1] is not None)
|
|
check("decide->design approver is HUMAN-001", "HUMAN-001" in (hg_dec[1] or ""))
|
|
check("acceptance->released human-gate (release)", SE._human_gate_for("acceptance", "released")[0] is True)
|
|
check("design->spec NOT human-gate", SE._human_gate_for("design", "spec")[0] is False)
|
|
check("spec->build NOT human-gate", SE._human_gate_for("spec", "build")[0] is False)
|
|
check("build->verification NOT human-gate", SE._human_gate_for("build", "verification")[0] is False)
|
|
check("discovery->decide NOT human-gate", SE._human_gate_for("discovery", "decide")[0] is False)
|
|
|
|
|
|
# ── 3) next CLI -> 유효 JSON ────────────────────────────────────────────
|
|
print("== next CLI -> JSON ==")
|
|
r = _cli(["next", "--workflow", "wf-a"])
|
|
check("next CLI exit 0", r.returncode == 0)
|
|
try:
|
|
cli_info = json.loads(r.stdout)
|
|
ok_json = cli_info.get("current-stage") == "intake"
|
|
except Exception:
|
|
ok_json = False
|
|
check("next CLI emits parseable JSON", ok_json)
|
|
check("next CLI needs --workflow (usage)", _cli(["next"]).returncode != 0)
|
|
|
|
|
|
# ── 4) DECIDE 사람게이트 + resumable(signoff) ───────────────────────────
|
|
print("== DECIDE human-gate + resumable ==")
|
|
led = SE._default_ledger("wf-d")
|
|
led["stage"] = "decide"
|
|
led["plan"] = "cascade"
|
|
led["tier"] = "standard"
|
|
SE._write_ledger("wf-d", led)
|
|
info_d = SE.next_info("wf-d")
|
|
check("decide current-command=/decide", info_d.get("current-command") == "/decide")
|
|
check("decide next-stage=design", info_d.get("next-stage") == "design")
|
|
check("decide->design human-gate REQUIRED (before signoff)",
|
|
info_d.get("advance", {}).get("human-gate", {}).get("required") is True)
|
|
check("human-gate names approver", info_d.get("advance", {}).get("human-gate", {}).get("approver"))
|
|
|
|
# 사람이 세션 밖에서 승인(signoff) — 이후 게이트 해제(재개 가능).
|
|
ok_sign, _ = SE.record_signoff("wf-d", "decide", "HUMAN-001")
|
|
check("record_signoff(HUMAN-001) ok", ok_sign is True)
|
|
info_d2 = SE.next_info("wf-d")
|
|
check("decide->design human-gate CLEARED after signoff",
|
|
info_d2.get("advance", {}).get("human-gate", {}).get("required") is False)
|
|
|
|
# 동일한 결정의 exact HUMAN artifact acceptance 자체가 게이트를 충족한다. 별도 signoff를
|
|
# 다시 요구하지 않는다(id+sha가 다른 이벤트나 비-human reviewer는 충족하지 못함).
|
|
led_exact = SE._default_ledger("wf-exact")
|
|
led_exact["stage"] = "decide"
|
|
led_exact["plan"] = "cascade"
|
|
SE._write_ledger("wf-exact", led_exact)
|
|
_orig_trusted, _orig_read_events = SE._trusted_artifacts, AL.read_events
|
|
try:
|
|
SE._trusted_artifacts = lambda wf: [{
|
|
"artifact-id": "packet-1", "artifact-sha256": "a" * 64,
|
|
"artifact-kind": "executive-decision-packet",
|
|
}] if wf == "wf-exact" else _orig_trusted(wf)
|
|
AL.read_events = lambda: [{
|
|
"workflow-id": "wf-exact", "report-id": "packet-1", "artifact-sha256": "a" * 64,
|
|
"decision": "accepted", "role-id": "HUMAN-001",
|
|
"reviewer": {"actor-id": "HUMAN-001"},
|
|
}]
|
|
info_exact = SE.next_info("wf-exact")
|
|
check("exact HUMAN decision-packet acceptance clears gate without duplicate signoff",
|
|
info_exact.get("advance", {}).get("human-gate", {}).get("required") is False)
|
|
finally:
|
|
SE._trusted_artifacts, AL.read_events = _orig_trusted, _orig_read_events
|
|
|
|
# 비-HUMAN signoff 는 거부(사람만 유효)
|
|
ok_bad, _ = SE.record_signoff("wf-d2", "decide", "OPS-ORCH")
|
|
check("non-HUMAN signoff rejected", ok_bad is False)
|
|
|
|
|
|
# ── 5) terminal: terminal stage 진입과 완료를 구분 ─────────────────────
|
|
print("== terminal 감지(current vs last-completed) ==")
|
|
ledr = SE._default_ledger("wf-r")
|
|
ledr["stage"] = "released"
|
|
SE._write_ledger("wf-r", ledr)
|
|
info_r = SE.next_info("wf-r")
|
|
check("released.running is not terminal yet", info_r.get("terminal") is False)
|
|
check("released -> no next-command", info_r.get("next-command") is None)
|
|
ok_complete, _ = SE.complete_stage("wf-r", "OPS-ORCH")
|
|
info_r_done = SE.next_info("wf-r")
|
|
check("released completed successfully", ok_complete is True)
|
|
check("released.completed -> terminal true", info_r_done.get("terminal") is True)
|
|
|
|
led_light = SE._default_ledger("wf-light-terminal")
|
|
led_light["plan"] = "light"
|
|
led_light["tier"] = "light"
|
|
led_light["stage"] = "acceptance"
|
|
led_light["stage-status"] = "completed"
|
|
SE._write_ledger("wf-light-terminal", led_light)
|
|
info_light = SE.next_info("wf-light-terminal")
|
|
check("light acceptance.completed is terminal", info_light.get("terminal") is True)
|
|
check("light terminal has no release command", info_light.get("next-command") is None)
|
|
|
|
|
|
# ── 6) 배선: /run-cascade.md ────────────────────────────────────────────
|
|
print("== 배선: run-cascade.md ==")
|
|
RC = os.path.join(ROOT, ".claude/commands/run-cascade.md")
|
|
check("/run-cascade command exists", os.path.exists(RC))
|
|
_rc = open(RC, encoding="utf-8").read() if os.path.exists(RC) else ""
|
|
check("run-cascade reuses state_engine next", "state_engine.py next" in _rc)
|
|
check("run-cascade stops at human-gate", "human-gate" in _rc and ("정지" in _rc or "멈춘" in _rc or "멈춘다" in _rc))
|
|
check("run-cascade forbids auto-approve/auto-complete", "자동 승인" in _rc and ("자동 완주" in _rc or "완주" in _rc))
|
|
check("run-cascade forbids parallel engine", "평행 엔진" in _rc)
|
|
check("run-cascade names approver path (HUMAN)", "HUMAN-001" in _rc)
|
|
|
|
|
|
print(f"\n{passed} passed, {failed} failed")
|
|
sys.exit(1 if failed else 0)
|