1400 lines
90 KiB
Python
1400 lines
90 KiB
Python
#!/usr/bin/env python3
|
|
"""P2 design-direction 강제기 — 격리 워크스페이스 + uuid, standalone check(pytest 아님)."""
|
|
import os, sys, tempfile, importlib.util, yaml, uuid, hashlib, subprocess, contextlib, shutil, json
|
|
|
|
ROOT = os.environ.get("CLAUDE_PROJECT_DIR", os.getcwd())
|
|
HOOKS = os.path.join(ROOT, ".claude", "hooks")
|
|
def _load(mod, path):
|
|
spec = importlib.util.spec_from_file_location(mod, os.path.join(HOOKS, path))
|
|
m = importlib.util.module_from_spec(spec); sys.path.insert(0, HOOKS); spec.loader.exec_module(m); return m
|
|
L = _load("lint_design_direction", "lint_design_direction.py")
|
|
SE = _load("state_engine", "state_engine.py")
|
|
AL = _load("acceptance_log", "acceptance_log.py")
|
|
|
|
passed = failed = 0
|
|
def check(name, ok):
|
|
global passed, failed
|
|
if ok: passed += 1; print(f" ✅ {name}")
|
|
else: failed += 1; print(f" ❌ {name}")
|
|
def _tmp(doc):
|
|
fd, p = tempfile.mkstemp(suffix=".yaml"); os.close(fd); yaml.safe_dump(doc, open(p, "w")); return p
|
|
def _sha(p): return hashlib.sha256(open(p, "rb").read()).hexdigest()
|
|
def _review(lens, reviewer_run_id, **over):
|
|
"""Fix B: 실제 review report 파일을 만들고 report-ref/report-sha256 을 채운 review dict.
|
|
`_critique_panel_ok`가 이제 각 review 의 report-ref 파일 실존+라이브 sha256 대조를 요구하므로,
|
|
합성 원장 테스트도 진짜 파일을 갖춰야 panel-pass 케이스가 통과한다."""
|
|
roles = {"product-fit": "DES-PROD", "usability": "UX-RESEARCHER",
|
|
"distinctiveness": "DES-VISUAL", "visual-craft": "DES-VISUAL",
|
|
"systematizability": "DES-PLATFORM", "market-memorability": "GTM-PMM",
|
|
"implementability": "ENG-FE"}
|
|
source = {"lens": lens, "reviewer-role-id": roles[lens],
|
|
"reviewer-run-id": reviewer_run_id, "verdict": "pass",
|
|
"findings": [], "notes": "synthetic review"}
|
|
source.update({k: v for k, v in over.items() if k in source})
|
|
p = _tmp(source)
|
|
d = {"report-id": f"REVIEW-{reviewer_run_id}", "lens": lens, "reviewer-role-id": source["reviewer-role-id"],
|
|
"reviewer-run-id": reviewer_run_id, "verdict": source["verdict"],
|
|
"report-ref": p, "report-sha256": _sha(p)}
|
|
d.update(over)
|
|
return d
|
|
|
|
@contextlib.contextmanager
|
|
def isolated_ws():
|
|
d = tempfile.mkdtemp(prefix="dd-ws-")
|
|
old = os.environ.get("ORGOS_WORKSPACE")
|
|
os.environ["ORGOS_WORKSPACE"] = d
|
|
try: yield d
|
|
finally:
|
|
os.environ.pop("ORGOS_WORKSPACE", None)
|
|
if old: os.environ["ORGOS_WORKSPACE"] = old
|
|
shutil.rmtree(d, ignore_errors=True)
|
|
|
|
_IB_REQ = {"product-goal": "x", "core-users": "y", "core-tasks": "z", "information-density": "high",
|
|
"required-accessibility": "AA", "brand-constraints": "n/a", "avoid-cliches": "none",
|
|
"representative-screen-requirement": "core-task", "tech-platform-constraints": "web"}
|
|
check("input-brief missing required -> Hard Fail",
|
|
any("필수" in h or "required" in h for h in L.lint_file(_tmp({"product-goal": "x"}), "direction-input-brief")[0]))
|
|
check("input-brief with reference-cluster -> Hard Fail",
|
|
any("reference-cluster" in h for h in L.lint_file(_tmp({**_IB_REQ, "reference-cluster": [1]}), "direction-input-brief")[0]))
|
|
check("clean full input-brief -> ok", not L.lint_file(_tmp(_IB_REQ), "direction-input-brief")[0])
|
|
check("input-brief competitor UI anchor -> Hard Fail",
|
|
any("anchor" in h for h in L.lint_file(
|
|
_tmp({**_IB_REQ, "brand-constraints": "Duolingo like entry UI"}),
|
|
"direction-input-brief")[0]))
|
|
check("input-brief named direction example -> Hard Fail",
|
|
any("메타포" in h for h in L.lint_file(_tmp({**_IB_REQ,
|
|
"representative-screen-requirement": {"id": "s", "kind": "core-task",
|
|
"description": "예: 가이드 트레일"}}),
|
|
"direction-input-brief")[0]))
|
|
|
|
_same_charter_direction = {
|
|
"design-question": "same", "layout-topology": "centered card",
|
|
"navigation-model": "next button", "typography-voice": "system sans",
|
|
"imagery-strategy": "css circles", "motion-model": "fade",
|
|
"dominant-primitives": ["card", "pill"],
|
|
"exclusive-primitives": ["shared-card", "shared-progress"],
|
|
"forbidden-primitives": ["table", "sidebar"],
|
|
}
|
|
_bad_charter = {
|
|
"direction-cycle-id": "C", "representative-screen": {"id": "s", "kind": "core-task", "description": "x"},
|
|
"directions": [{**_same_charter_direction, "id": f"D{i}"} for i in (1, 2, 3)],
|
|
"pairwise-separation": [
|
|
{"directions": [a, b], "differing-axes": ["layout-topology"], "allowed-overlap": "none"}
|
|
for a, b in (("D1", "D2"), ("D1", "D3"), ("D2", "D3"))
|
|
],
|
|
}
|
|
check("same-shell divergence-charter -> Hard Fail",
|
|
any("조형축 차이" in h or "exclusive primitive 충돌" in h
|
|
for h in L.lint_file(_tmp(_bad_charter), "divergence-charter")[0]))
|
|
check("comparative audit with blocker cannot pass -> Hard Fail",
|
|
any("blocking finding" in h for h in L.lint_file(_tmp({
|
|
"direction-cycle-id": "C", "divergence-charter-ref": "c", "divergence-charter-sha256": "s",
|
|
"direction-set-ref": "d", "direction-set-sha256": "s", "reviewer-role-id": "DES-VISUAL",
|
|
"reviewer-run-id": "r", "verdict": "pass", "pairwise-comparisons": [{}],
|
|
"full-size-previews": [{}], "blocking-findings": [{"id": "same-shell"}],
|
|
}), "comparative-divergence-audit")[0]))
|
|
|
|
# --- 리뷰 반영 회귀(C1/C2/I1/I2) ---
|
|
_missing_proto = os.path.join(tempfile.gettempdir(), f"no-such-prototype-{uuid.uuid4().hex}.png")
|
|
_winner_doc = {
|
|
"selected-direction-ref": "x", "selected-direction-sha256": "x", "prototype-path": _missing_proto,
|
|
"prototype-sha256": "deadbeef", "preview-receipt-ref": "x", "revision": 1,
|
|
}
|
|
check("winner-prototype with nonexistent prototype-path -> Hard Fail(파일 없음)",
|
|
any("파일 없음" in h for h in L.lint_file(_tmp(_winner_doc), "winner-prototype")[0]))
|
|
|
|
_ds_list_path = _tmp([1, 2, 3])
|
|
_h_list, _w_list = L.lint_selected_direction(_tmp({"selected-direction-id": "d1"}), _ds_list_path)
|
|
check("lint_selected_direction with list-typed direction-set -> Hard Fail(no crash)", bool(_h_list))
|
|
|
|
_dup_ds = {"directions": [{"id": "d1"}, {"id": "d2"}, {"id": "d3"}]}
|
|
_dup_ds_path = _tmp(_dup_ds)
|
|
_dup_sd = {
|
|
"direction-set-ref": "x", "direction-set-sha256": _sha(_dup_ds_path), "selected-direction-id": "d1",
|
|
"parent-workflow-id": "x", "product-decision-id": "x", "direction-input-brief-sha256": "x",
|
|
"selection-acceptance-receipt": "x",
|
|
"rejected-directions": [{"id": "d1", "reason": "dup"}, {"id": "d2", "reason": "n/a"}, {"id": "d3", "reason": "n/a"}],
|
|
"locked-invariants": ["a", "b", "c"],
|
|
}
|
|
_h_dup, _w_dup = L.lint_selected_direction(_tmp(_dup_sd), _dup_ds_path)
|
|
check("selected-direction-id also in rejected-directions -> Hard Fail(중복)",
|
|
any("중복" in h for h in _h_dup))
|
|
|
|
def _ds_file(ids=("DIR-1","DIR-2","DIR-3")):
|
|
return _tmp({"direction-cycle-id": "C1", "representative-screen": {"id": "S1", "kind": "core-task"},
|
|
"directions": [{"id": i} for i in ids]})
|
|
def _sel_file(dsp, **over):
|
|
d = {"direction-set-ref": dsp, "direction-set-sha256": _sha(dsp), "selected-direction-id": "DIR-2",
|
|
"rejected-directions": [{"id": "DIR-1", "reason": "저밀도"}, {"id": "DIR-3", "reason": "클리셰"}],
|
|
"locked-invariants": ["a","b","c"], "parent-workflow-id": "p", "product-decision-id": "PD",
|
|
"direction-input-brief-sha256": "SHA", "selection-acceptance-receipt": "r"}
|
|
d.update(over); return _tmp(d)
|
|
dsp = _ds_file()
|
|
check("valid selected bundle -> ok", not L.lint_selected_direction(_sel_file(dsp), dsp)[0])
|
|
check("ghost selected id -> Hard Fail", any("유령" in h for h in L.lint_selected_direction(_sel_file(dsp, **{"selected-direction-id": "DIR-9"}), dsp)[0]))
|
|
check("rejected not covering all -> Hard Fail", any("불일치" in h for h in L.lint_selected_direction(_sel_file(dsp, **{"rejected-directions": [{"id": "DIR-1", "reason": "x"}]}), dsp)[0]))
|
|
check("secondary-influence-id -> Hard Fail", any("secondary" in h for h in L.lint_selected_direction(_sel_file(dsp, **{"secondary-influence-id": "DIR-1"}), dsp)[0]))
|
|
check("locked<3 -> Hard Fail", any("locked" in h for h in L.lint_selected_direction(_sel_file(dsp, **{"locked-invariants": ["a"]}), dsp)[0]))
|
|
check("adopted from ghost -> Hard Fail", any("adopted" in h for h in L.lint_selected_direction(_sel_file(dsp, **{"adopted-elements": [{"from-direction-id": "DIR-9", "element-id": "t", "rationale": "r"}]}), dsp)[0]))
|
|
check("adopted vague -> Hard Fail", any("adopted" in h for h in L.lint_selected_direction(_sel_file(dsp, **{"adopted-elements": [{"from-direction-id": "DIR-1", "element-id": "분위기", "rationale": "감성"}]}), dsp)[0]))
|
|
check("direction-set hash mismatch -> Hard Fail", any("불일치" in h for h in L.lint_selected_direction(_sel_file(dsp, **{"direction-set-sha256": "WRONG"}), dsp)[0]))
|
|
|
|
# --- Task 8: design-direction plan(8 stage) + 전이 10종(역전이2+NONE+finalize) ---
|
|
plans = yaml.safe_load(open(os.path.join(ROOT, "org-os/06-agent-work/execution-plans.yaml")))["execution-plans"]["plans"]
|
|
dd = plans.get("design-direction", {})
|
|
check("plan has 8 namespaced stages incl finalize",
|
|
"design-direction-finalize" in dd.get("stages", []) and len(dd.get("stages", [])) == 8)
|
|
trans = yaml.safe_load(open(os.path.join(ROOT, "org-os/00-role-registry/state-transition-rules.yaml")))["state-transition-rules"]["workflow-stage-transitions"]["transitions"]
|
|
ddt = [t for t in trans if str(t.get("from","")).startswith("design-direction")]
|
|
check("design-direction transitions == 10", len(ddt) == 10)
|
|
check("finalize->approved exists", any(t.get("from")=="design-direction-finalize" and t.get("to")=="design-direction-approved" for t in trans))
|
|
check("critique->finalize (not approved)", any(t.get("from")=="design-direction-critique" and t.get("to")=="design-direction-finalize" for t in trans))
|
|
check("critique->prototype reverse transition exists", any(t.get("from")=="design-direction-critique" and t.get("to")=="design-direction-prototype" for t in trans))
|
|
check("critique->divergence reverse transition exists", any(t.get("from")=="design-direction-critique" and t.get("to")=="design-direction-divergence" for t in trans))
|
|
check("HUMAN none-of-the-above -> discovery exists", any(t.get("from")=="design-direction-decision" and t.get("to")=="design-direction-discovery" for t in trans))
|
|
|
|
# --- Task 7: _initial_stage(plan) — namespaced plan 첫 stage에서 원장 시작 ---
|
|
check("_initial_stage design-direction", SE._initial_stage("design-direction") == "design-direction-intake")
|
|
check("_initial_stage cascade = intake", SE._initial_stage("cascade") == "intake")
|
|
|
|
def _se(*a, ws=None):
|
|
a = list(a)
|
|
if a and a[0] == "init" and "--tier" not in a:
|
|
a += ["--tier", "light"]
|
|
env = {**os.environ, "CLAUDE_PROJECT_DIR": ROOT}
|
|
if ws: env["ORGOS_WORKSPACE"] = ws
|
|
return subprocess.run([sys.executable, os.path.join(HOOKS, "state_engine.py"), *a], capture_output=True, text=True, env=env)
|
|
|
|
|
|
def _write_contract_artifact(ws, wf, artifact_id, kind, producer, payload, stage):
|
|
directory = os.path.join(ws, "completion-records", wf)
|
|
os.makedirs(directory, exist_ok=True)
|
|
path = os.path.join(directory, f"{artifact_id}.report.yaml")
|
|
report = {
|
|
"report-type": "workflow-artifact", "artifact-kind": kind, "artifact-version": 1,
|
|
"tier": "light",
|
|
"identity": {"artifact-id": artifact_id, "workflow-id": wf, "stage": stage,
|
|
"producer-role-id": producer},
|
|
"payload": payload,
|
|
"report-header": {"bottom-line": f"{kind} test 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 _accept_product_decision(ws, workflow, artifact_id, supersedes=None):
|
|
if SE.current_stage(workflow) != "decide":
|
|
SE._atomic_event_transaction(workflow, workflow_event={
|
|
"state-event-id": f"fixture-{workflow}-decide", "event-type": "state-transition",
|
|
"workflow-id": workflow, "from": SE.current_stage(workflow), "to": "decide",
|
|
"actor": "OPS-ORCH", "effective-at": SE._now(),
|
|
})
|
|
path = _write_contract_artifact(ws, workflow, artifact_id, "executive-decision-packet", "EXEC-CEO",
|
|
{"recommendation": artifact_id}, "decide")
|
|
submitted = _se("submit-artifact", "--workflow", workflow, "--report", path,
|
|
"--actor", "OPS-ORCH", ws=ws)
|
|
args = ["review-artifact", "--workflow", workflow, "--report", path,
|
|
"--decision", "accepted", "--reviewer", "HUMAN-001"]
|
|
if supersedes:
|
|
args += ["--supersedes", supersedes]
|
|
reviewed = _se(*args, ws=ws)
|
|
return submitted.returncode == 0 and reviewed.returncode == 0
|
|
|
|
|
|
_raw_append_event = AL.append_event
|
|
|
|
|
|
def _contract_aware_append(event):
|
|
"""Migrate old product-decision setup calls to the trusted artifact/review API."""
|
|
rid = str(event.get("report-id") or "")
|
|
workflow = event.get("workflow-id")
|
|
if (event.get("decision") == "accepted" and workflow and rid.startswith("PD")
|
|
and str(event.get("role-id") or "").upper() == "HUMAN-001"):
|
|
return _accept_product_decision(os.environ["ORGOS_WORKSPACE"], workflow, rid,
|
|
event.get("supersedes-report-id"))
|
|
return _raw_append_event(event)
|
|
|
|
|
|
AL.append_event = _contract_aware_append
|
|
|
|
_submitted_paths = {}
|
|
_DIRECTION_KIND_META = {
|
|
"direction-discovery": ("DES-DIRECTOR", "design-direction-discovery"),
|
|
"divergence-charter": ("DES-DIRECTOR", "design-direction-discovery"),
|
|
"direction-set": ("DES-VISUAL", "design-direction-divergence"),
|
|
"comparative-divergence-audit": ("DES-VISUAL", "design-direction-divergence"),
|
|
"selected-direction": ("DES-DIRECTOR", "design-direction-decision"),
|
|
"winner-prototype": ("ENG-FEUX", "design-direction-prototype"),
|
|
"design-review-panel": ("DES-DIRECTOR", "design-direction-critique"),
|
|
"approved-direction": ("DES-DIRECTOR", "design-direction-finalize"),
|
|
}
|
|
|
|
|
|
def _submit_raw_as_artifact(ws, workflow, artifact_id, kind, raw_path):
|
|
producer, stage = _DIRECTION_KIND_META[kind]
|
|
raw = yaml.safe_load(open(raw_path, encoding="utf-8")) or {}
|
|
if isinstance(raw.get(kind), dict):
|
|
raw = raw[kind]
|
|
path = _write_contract_artifact(ws, workflow, artifact_id, kind, producer, raw, stage)
|
|
result = _se("submit-artifact", "--workflow", workflow, "--report", path,
|
|
"--actor", "OPS-ORCH", ws=ws)
|
|
if result.returncode == 0:
|
|
_submitted_paths[(workflow, artifact_id)] = path
|
|
return result
|
|
|
|
_IB_PATH = "org-os/06-agent-work/design-direction-spec.yaml"
|
|
with isolated_ws() as ws:
|
|
# 부모 workflow 를 실제로 만들고(원장 실존) product-decision "PD" 에 대한 진짜 accepted 이벤트를
|
|
# acceptance_log 에 남긴다(fix: "부모가 아무거나 accepted" 나 "literal 부분일치" 우회가 막혔으므로,
|
|
# 이제 이 id 자체의 accepted 이벤트가 있어야만 바인딩이 통과한다).
|
|
_se("init", "--workflow", "pc", ws=ws)
|
|
check("setup: acceptance event appended (pc/PD)",
|
|
AL.append_event(AL.build_event("PD", "accepted", workflow="pc", role="HUMAN-001")))
|
|
wf = f"dd-{uuid.uuid4().hex[:8]}"
|
|
r_init = _se("init", "--workflow", wf, "--plan", "design-direction", "--parent-workflow", "pc",
|
|
"--product-decision", "PD", "--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("design-direction init with real bound parent -> accept(exit 0)", r_init.returncode == 0)
|
|
cur = _se("current", "--workflow", wf, ws=ws).stdout
|
|
check("design-direction init stage = design-direction-intake", "design-direction-intake" in cur)
|
|
|
|
# --- Task 10: init 바인딩 필수화 + 부모 실존 검증 + dedup 조회 ---
|
|
with isolated_ws() as ws:
|
|
r = _se("init", "--workflow", f"dd-{uuid.uuid4().hex[:8]}", "--plan", "design-direction", ws=ws)
|
|
check("design-direction init without binding -> reject", r.returncode != 0)
|
|
with isolated_ws() as ws:
|
|
# 부모 원장 없음 -> 거부(_load_ledger_safe 는 부재 시에도 기본원장을 반환하므로,
|
|
# 그 반환값만으로 판단하면 안 되고 파일 실존을 직접 확인해야 한다)
|
|
r = _se("init", "--workflow", f"dd-{uuid.uuid4().hex[:8]}", "--plan", "design-direction",
|
|
"--parent-workflow", "ghost-parent", "--product-decision", "PD",
|
|
"--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("nonexistent parent -> reject", r.returncode != 0)
|
|
with isolated_ws() as ws:
|
|
# 부모는 실존하지만 product-decision 이 없음(accepted 이벤트가 전혀 없음) -> 거부
|
|
_se("init", "--workflow", "pc2", ws=ws)
|
|
r = _se("init", "--workflow", f"dd-{uuid.uuid4().hex[:8]}", "--plan", "design-direction",
|
|
"--parent-workflow", "pc2", "--product-decision", "PD-NOPE",
|
|
"--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("parent exists but no accepted/matching product-decision -> reject", r.returncode != 0)
|
|
|
|
# --- Critical fix 회귀: _al_accepted_ids 로 "정확히 이 product-decision 이 accepted 되었는가" 만 인정
|
|
# (부모가 아무거나 accepted 했다고 통과시키던 _al_has_accepted 우회, 그리고 str(parent_led) 부분일치
|
|
# 우회를 둘 다 닫는다) ---
|
|
with isolated_ws() as ws:
|
|
# 1) genuine match -> accept: 부모에 정확히 PD-001 의 accepted 이벤트가 있다.
|
|
_se("init", "--workflow", "pc-genuine", ws=ws)
|
|
check("setup: acceptance event appended (pc-genuine/PD-001)",
|
|
AL.append_event(AL.build_event("PD-001", "accepted", workflow="pc-genuine", role="HUMAN-001")))
|
|
r = _se("init", "--workflow", f"dd-{uuid.uuid4().hex[:8]}", "--plan", "design-direction",
|
|
"--parent-workflow", "pc-genuine", "--product-decision", "PD-001",
|
|
"--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("genuine accepted product-decision match -> accept(exit 0)", r.returncode == 0)
|
|
with isolated_ws() as ws:
|
|
# 2) acceptance exists but for a DIFFERENT id -> reject(위조 id 로 편승 불가)
|
|
_se("init", "--workflow", "pc-other", ws=ws)
|
|
check("setup: acceptance event appended (pc-other/PD-001)",
|
|
AL.append_event(AL.build_event("PD-001", "accepted", workflow="pc-other", role="HUMAN-001")))
|
|
r = _se("init", "--workflow", f"dd-{uuid.uuid4().hex[:8]}", "--plan", "design-direction",
|
|
"--parent-workflow", "pc-other", "--product-decision", "PD-OTHER",
|
|
"--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("accepted product-decision exists but id differs -> reject", r.returncode != 0)
|
|
with isolated_ws() as ws:
|
|
# 3) universal-substring bypass -> reject: 부모는 갓 init(accepted 이벤트 전무), product-decision
|
|
# 은 원장 고정 스키마 키("workflow-id" 등)의 부분문자열인 "workflow" -> 예전엔 str(parent_led)
|
|
# 부분일치로 어떤 부모에도 통과했다. 이제는 거부되어야 한다.
|
|
_se("init", "--workflow", "pc-substr", ws=ws)
|
|
r = _se("init", "--workflow", f"dd-{uuid.uuid4().hex[:8]}", "--plan", "design-direction",
|
|
"--parent-workflow", "pc-substr", "--product-decision", "workflow",
|
|
"--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("universal-substring 'workflow' bypass -> reject(substring hole closed)", r.returncode != 0)
|
|
with isolated_ws() as ws:
|
|
# 4) missing brief file -> reject: 부모+accepted product-decision 은 유효하지만 brief 경로가 실존하지 않음
|
|
_se("init", "--workflow", "pc-nobrief", ws=ws)
|
|
check("setup: acceptance event appended (pc-nobrief/PD-001)",
|
|
AL.append_event(AL.build_event("PD-001", "accepted", workflow="pc-nobrief", role="HUMAN-001")))
|
|
r = _se("init", "--workflow", f"dd-{uuid.uuid4().hex[:8]}", "--plan", "design-direction",
|
|
"--parent-workflow", "pc-nobrief", "--product-decision", "PD-001",
|
|
"--direction-input-brief", "/does/not/exist.yaml", ws=ws)
|
|
check("missing direction-input-brief file -> reject", r.returncode != 0)
|
|
|
|
with isolated_ws() as ws:
|
|
# dedup: 같은 (parent, product-decision) 으로 다시 찾으면 이미 만든 자식을 반환해야 한다
|
|
_se("init", "--workflow", "pc3", ws=ws)
|
|
check("setup: acceptance event appended (pc3/PD3)",
|
|
AL.append_event(AL.build_event("PD3", "accepted", workflow="pc3", role="HUMAN-001")))
|
|
wf3 = f"dd-{uuid.uuid4().hex[:8]}"
|
|
_se("init", "--workflow", wf3, "--plan", "design-direction", "--parent-workflow", "pc3",
|
|
"--product-decision", "PD3", "--direction-input-brief", _IB_PATH, ws=ws)
|
|
# isolated_ws() 가 이미 이 with-블록 동안 ORGOS_WORKSPACE=ws 를 설정해 두므로 in-process 직접호출도 같은 워크스페이스를 본다.
|
|
_abs_ib = _IB_PATH if os.path.isabs(_IB_PATH) else os.path.join(ROOT, _IB_PATH)
|
|
_brief_sha = _sha(_abs_ib) if os.path.exists(_abs_ib) else None
|
|
found = SE.find_child_direction_workflow("pc3", "PD3", _brief_sha)
|
|
check("find_child_direction_workflow finds the already-init'd child", bool(found) and found.get("workflow-id") == wf3)
|
|
check("find_child_direction_workflow reports not-stale when sha matches", found is not None and found.get("stale") is False)
|
|
with isolated_ws() as ws:
|
|
found_none = SE.find_child_direction_workflow("no-such-parent", "PD-X", "deadbeef")
|
|
check("find_child_direction_workflow -> None when nothing matches", found_none is None)
|
|
|
|
# --- Task 15 item 5: stale child(brief hash differs since binding) is flagged, not silently reused ---
|
|
with isolated_ws() as ws:
|
|
_se("init", "--workflow", "pc5", ws=ws)
|
|
check("setup(item5): acceptance event appended (pc5/PD5)",
|
|
AL.append_event(AL.build_event("PD5", "accepted", workflow="pc5", role="HUMAN-001")))
|
|
wf5 = f"dd-{uuid.uuid4().hex[:8]}"
|
|
r5 = _se("init", "--workflow", wf5, "--plan", "design-direction", "--parent-workflow", "pc5",
|
|
"--product-decision", "PD5", "--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("setup(item5): child design-direction init -> accept(exit 0)", r5.returncode == 0)
|
|
found_stale = SE.find_child_direction_workflow("pc5", "PD5", "totally-different-sha-than-binding")
|
|
check("item5: find_child_direction_workflow reports stale=True when queried brief-sha differs "
|
|
"from what the child bound at init(brief changed since)",
|
|
found_stale is not None and found_stale.get("stale") is True)
|
|
check("item5: stale lookup still identifies the correct(existing) child workflow-id "
|
|
"(stale != not-found — the caller must decide to open a NEW cycle, not silently resume)",
|
|
found_stale is not None and found_stale.get("workflow-id") == wf5)
|
|
r_find5 = _se("find-child-direction", "--parent-workflow", "pc5", "--product-decision", "PD5",
|
|
"--direction-input-brief-sha256", "totally-different-sha-than-binding", ws=ws)
|
|
check("item5: find-child-direction CLI -> exit 0 when found(even if stale — still reported, not hidden)",
|
|
r5.returncode == 0 and r_find5.returncode == 0)
|
|
check("item5: find-child-direction CLI -> stdout reports stale=true(brief changed since binding)",
|
|
"true" in r_find5.stdout.lower())
|
|
|
|
# --- Task 9: cycle model(active pointer) + facts/helpers + predicate 9 + PROTECTED ---
|
|
for k in ["parent-binding-present", "direction-discovery-present", "divergence-charter-present",
|
|
"directions-diverged", "divergence-audit-passed", "selected-direction-accepted",
|
|
"winner-prototype-present", "critique-revision-requested", "concept-rejection-recorded", "direction-critique-passed"]:
|
|
check(f"predicate {k}", k in SE._PREDICATES)
|
|
for fk in ["divergence_charter_present", "directions_diverged", "divergence_audit_passed",
|
|
"selected_direction_accepted", "design_direction_approved", "parent_binding_present"]:
|
|
check(f"protected {fk}", fk in SE._PROTECTED_FACTS)
|
|
# active cycle: 오래된 pass 패널이 남아도 현재 cycle 아니면 무시
|
|
led = {"design-direction-active": {"cycle-id": "C2", "review-panel-report-id": "RP-C2"},
|
|
"artifacts": [
|
|
{"design-type": "design-review-panel", "report-id": "RP-C1", "direction-cycle-id": "C1",
|
|
"design-review-panel": {"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR"}, "reviews": []}},
|
|
{"design-type": "design-review-panel", "report-id": "RP-C2", "direction-cycle-id": "C2",
|
|
"design-review-panel": {"synthesis": {"verdict": "minor-revision", "role-id": "DES-DIRECTOR"}, "reviews": []}}]}
|
|
f = SE._facts("wf-cycle", led, None)
|
|
check("stale pass panel ignored, active=minor-revision", f.get("critique_revision_requested") and not f.get("direction_critique_passed"))
|
|
|
|
# --- 리뷰 반영 회귀(C1/C2/I3) — _active_artifact 최신선택/fail-closed, panel 직물검증, coded-slice hash ---
|
|
|
|
_good_cs_tmpdir = tempfile.mkdtemp(prefix="dd-good-coded-slice-")
|
|
_good_cs_path = os.path.join(_good_cs_tmpdir, "slice.tsx")
|
|
open(_good_cs_path, "w").write("export const Slice = () => null;\n")
|
|
_good_cs_sha = _sha(_good_cs_path)
|
|
|
|
def _valid_ds(run_ids=("R1", "R2", "R3")):
|
|
"""directions_diverged 를 통과시킬 최소 유효 direction-set(coded-slice 실존+hash 일치 포함)."""
|
|
return {
|
|
"direction-set": {
|
|
"directions": [
|
|
{"id": f"D{i+1}", "producer-run-id": rid, "context-package-id": f"CP{i+1}",
|
|
"coded-slice": _good_cs_path, "coded-slice-sha256": _good_cs_sha}
|
|
for i, rid in enumerate(run_ids)
|
|
],
|
|
"representative-screen": {"id": "S1"},
|
|
"comparison-preview": {"receipt-ref": "r", "gallery-path": "g"},
|
|
}
|
|
}
|
|
|
|
def _broken_ds(run_ids=("R1", "R1", "R1")):
|
|
"""producer-run-id 중복 -> _directions_diverged 가 False 여야 하는 broken direction-set."""
|
|
return {
|
|
"direction-set": {
|
|
"directions": [
|
|
{"id": f"D{i+1}", "producer-run-id": rid, "context-package-id": f"CP{i+1}",
|
|
"coded-slice": _good_cs_path, "coded-slice-sha256": _good_cs_sha}
|
|
for i, rid in enumerate(run_ids)
|
|
],
|
|
"representative-screen": {"id": "S1"},
|
|
"comparison-preview": {"receipt-ref": "r", "gallery-path": "g"},
|
|
}
|
|
}
|
|
|
|
# 1) C1 stale direction-set: 오래된 valid(먼저 append) + 최신 broken(나중 append), 포인터는 report-id 미명시
|
|
led_c1 = {
|
|
"design-direction-active": {"cycle-id": "C1"}, # direction-set-report-id 미명시 -> 최신(마지막) 사용해야 함
|
|
"artifacts": [
|
|
{"design-type": "direction-set", "report-id": "DS-OLD-VALID", **_valid_ds(("R1", "R2", "R3"))},
|
|
{"design-type": "direction-set", "report-id": "DS-NEW-BROKEN", **_broken_ds(("R9", "R9", "R9"))},
|
|
],
|
|
}
|
|
check("C1: no pointer report-id -> uses NEWEST(broken) artifact, not stale valid one",
|
|
SE._directions_diverged(led_c1, led_c1["artifacts"]) is False)
|
|
|
|
# 2) C1 exact pointer: 같은 두 아티팩트, 포인터가 valid(옛) 것의 report-id 를 정확히 지정 -> 그걸 사용해야 True
|
|
led_c1_exact = {
|
|
"design-direction-active": {"cycle-id": "C1", "direction-set-report-id": "DS-OLD-VALID"},
|
|
"artifacts": led_c1["artifacts"],
|
|
}
|
|
check("C1: exact pointer report-id -> exact match wins (valid one) -> True",
|
|
SE._directions_diverged(led_c1_exact, led_c1_exact["artifacts"]) is True)
|
|
|
|
# 3) C2 panel fail-closed: valid-looking 6-lens 패널 + pass/DES-DIRECTOR synthesis, but NO resolvable direction-set
|
|
_lenses = ["product-fit", "usability", "distinctiveness", "visual-craft",
|
|
"systematizability", "market-memorability", "implementability"]
|
|
led_c2 = {
|
|
"design-direction-active": {"cycle-id": "C3", "review-panel-report-id": "RP-C3"},
|
|
"artifacts": [
|
|
{"design-type": "design-review-panel", "report-id": "RP-C3",
|
|
"design-review-panel": {
|
|
"reviews": [_review(l, f"REV-{l}") for l in _lenses],
|
|
"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR", "unresolved-dissent": []},
|
|
}},
|
|
# 의도적으로 direction-set 아티팩트 없음 -> producer-run-id 대조 불가
|
|
],
|
|
}
|
|
check("C2: valid-looking panel but no resolvable direction-set -> fail closed (False)",
|
|
SE._critique_panel_ok(led_c2, led_c2["artifacts"]) is False)
|
|
|
|
# 3b) 같은 패널 + 이번엔 resolvable direction-set(교집合 없는 producer-run-id) -> True 로 통과해야 함
|
|
led_c2_ok = {
|
|
"design-direction-active": {"cycle-id": "C3", "review-panel-report-id": "RP-C3", "direction-set-report-id": "DS-C3"},
|
|
"artifacts": led_c2["artifacts"] + [
|
|
{"design-type": "direction-set", "report-id": "DS-C3", **_valid_ds(("PR-1", "PR-2", "PR-3"))},
|
|
],
|
|
}
|
|
check("C2: panel + resolvable direction-set with disjoint producer/reviewer runs -> ok (True)",
|
|
SE._critique_panel_ok(led_c2_ok, led_c2_ok["artifacts"]) is True)
|
|
|
|
# A concern/revision verdict is work, not prose that synthesis may override.
|
|
_concern_reviews = [_review(l, f"REV-CONCERN-{l}", **({"verdict": "concerns"} if l == "visual-craft" else {}))
|
|
for l in _lenses]
|
|
led_c2_concern = {
|
|
"design-direction-active": {"cycle-id": "C3", "review-panel-report-id": "RP-CONCERN", "direction-set-report-id": "DS-C3"},
|
|
"artifacts": [
|
|
{"design-type": "design-review-panel", "report-id": "RP-CONCERN",
|
|
"design-review-panel": {"reviews": _concern_reviews,
|
|
"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR", "unresolved-dissent": []}}},
|
|
{"design-type": "direction-set", "report-id": "DS-C3", **_valid_ds(("PR-1", "PR-2", "PR-3"))},
|
|
],
|
|
}
|
|
check("visual-craft concerns + synthesis pass -> veto(False)",
|
|
SE._critique_panel_ok(led_c2_concern, led_c2_concern["artifacts"]) is False)
|
|
|
|
_blocking_reviews = [_review(l, f"REV-BLOCK-{l}", **({"findings": [{"severity": "critical"}]} if l == "distinctiveness" else {}))
|
|
for l in _lenses]
|
|
led_c2_block = {
|
|
"design-direction-active": {"cycle-id": "C3", "review-panel-report-id": "RP-BLOCK", "direction-set-report-id": "DS-C3"},
|
|
"artifacts": [
|
|
{"design-type": "design-review-panel", "report-id": "RP-BLOCK",
|
|
"design-review-panel": {"reviews": _blocking_reviews,
|
|
"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR", "unresolved-dissent": []}}},
|
|
{"design-type": "direction-set", "report-id": "DS-C3", **_valid_ds(("PR-1", "PR-2", "PR-3"))},
|
|
],
|
|
}
|
|
check("critical lens finding + synthesis pass -> veto(False)",
|
|
SE._critique_panel_ok(led_c2_block, led_c2_block["artifacts"]) is False)
|
|
|
|
# 3c) Fix B(최종리뷰): review 하나의 report-sha256 이 그 report-ref 파일의 실제 해시와 불일치 -> False
|
|
_bad_hash_reviews = [_review(l, f"REV-{l}") for l in _lenses]
|
|
_bad_hash_reviews[0]["report-sha256"] = "0" * 64 # 위조: 파일은 실존하지만 다른 report 를 가리키는 척
|
|
led_c2_badhash = {
|
|
"design-direction-active": {"cycle-id": "C3B", "review-panel-report-id": "RP-C3B", "direction-set-report-id": "DS-C3"},
|
|
"artifacts": [
|
|
{"design-type": "design-review-panel", "report-id": "RP-C3B",
|
|
"design-review-panel": {
|
|
"reviews": _bad_hash_reviews,
|
|
"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR", "unresolved-dissent": []},
|
|
}},
|
|
{"design-type": "direction-set", "report-id": "DS-C3", **_valid_ds(("PR-1", "PR-2", "PR-3"))},
|
|
],
|
|
}
|
|
check("FixB: one review's report-sha256 mismatches its real report-ref file -> _critique_panel_ok False (fail closed)",
|
|
SE._critique_panel_ok(led_c2_badhash, led_c2_badhash["artifacts"]) is False)
|
|
|
|
# 3d) Fix B: review 가 report-ref 자체를 누락(자기신고만) -> False(우회 방지)
|
|
_missing_ref_reviews = [_review(l, f"REV3-{l}") for l in _lenses]
|
|
del _missing_ref_reviews[0]["report-ref"]
|
|
led_c2_noref = {
|
|
"design-direction-active": {"cycle-id": "C3D", "review-panel-report-id": "RP-C3D", "direction-set-report-id": "DS-C3"},
|
|
"artifacts": [
|
|
{"design-type": "design-review-panel", "report-id": "RP-C3D",
|
|
"design-review-panel": {
|
|
"reviews": _missing_ref_reviews,
|
|
"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR", "unresolved-dissent": []},
|
|
}},
|
|
{"design-type": "direction-set", "report-id": "DS-C3", **_valid_ds(("PR-1", "PR-2", "PR-3"))},
|
|
],
|
|
}
|
|
check("FixB: review missing report-ref entirely -> _critique_panel_ok False (fail closed, no bare-string self-report)",
|
|
SE._critique_panel_ok(led_c2_noref, led_c2_noref["artifacts"]) is False)
|
|
|
|
# 4) I3 coded-slice hash mismatch: 파일은 실존하지만 sha256 이 다르면 False 여야 함
|
|
_cs_tmpdir = tempfile.mkdtemp(prefix="dd-coded-slice-")
|
|
_cs_path = os.path.join(_cs_tmpdir, "slice.tsx")
|
|
open(_cs_path, "w").write("export const Slice = () => null;\n")
|
|
_wrong_sha = "0" * 64
|
|
led_i3 = {
|
|
"design-direction-active": {"cycle-id": "C4"},
|
|
"artifacts": [
|
|
{"design-type": "direction-set", "report-id": "DS-C4", "direction-set": {
|
|
"directions": [
|
|
{"id": f"D{i+1}", "producer-run-id": f"R{i+1}", "context-package-id": f"CP{i+1}",
|
|
"coded-slice": _cs_path, "coded-slice-sha256": _wrong_sha}
|
|
for i in range(3)
|
|
],
|
|
"representative-screen": {"id": "S1"},
|
|
"comparison-preview": {"receipt-ref": "r", "gallery-path": "g"},
|
|
}},
|
|
],
|
|
}
|
|
check("I3: coded-slice exists but sha256 mismatch -> _directions_diverged False",
|
|
SE._directions_diverged(led_i3, led_i3["artifacts"]) is False)
|
|
shutil.rmtree(_cs_tmpdir, ignore_errors=True)
|
|
shutil.rmtree(_good_cs_tmpdir, ignore_errors=True)
|
|
|
|
# 5) Fix1 회귀: coded-slice-sha256 필드 자체를 생략(실존 파일은 있음) -> hash 게이트 우회 차단 -> False
|
|
_cs5_tmpdir = tempfile.mkdtemp(prefix="dd-cs-nosha-")
|
|
_cs5_path = os.path.join(_cs5_tmpdir, "slice.tsx")
|
|
open(_cs5_path, "w").write("export const Slice = () => null;\n")
|
|
led_nosha = {
|
|
"design-direction-active": {"cycle-id": "C5"},
|
|
"artifacts": [
|
|
{"design-type": "direction-set", "report-id": "DS-C5", "direction-set": {
|
|
"directions": [
|
|
{"id": f"D{i+1}", "producer-run-id": f"R{i+1}", "context-package-id": f"CP{i+1}",
|
|
"coded-slice": _cs5_path} # coded-slice-sha256 필드 없음(우회 시도)
|
|
for i in range(3)
|
|
],
|
|
"representative-screen": {"id": "S1"},
|
|
"comparison-preview": {"receipt-ref": "r", "gallery-path": "g"},
|
|
}},
|
|
],
|
|
}
|
|
check("Fix1: coded-slice-sha256 omitted (real coded-slice file exists) -> _directions_diverged False",
|
|
SE._directions_diverged(led_nosha, led_nosha["artifacts"]) is False)
|
|
shutil.rmtree(_cs5_tmpdir, ignore_errors=True)
|
|
|
|
# 6) F6 회귀(2026-07-16 실측): coded-slice 가 파일이 아니라 디렉터리면 hash 대조의 open() 이
|
|
# IsADirectoryError 로 크래시했다. 디렉터리는 '실제 픽셀' 아님 -> 크래시 없이 False 여야 한다.
|
|
_cs6_tmpdir = tempfile.mkdtemp(prefix="dd-cs-dir-") # 디렉터리 경로 자체를 coded-slice 로 지정(오용)
|
|
led_dir = {
|
|
"design-direction-active": {"cycle-id": "C6"},
|
|
"artifacts": [
|
|
{"design-type": "direction-set", "report-id": "DS-C6", "direction-set": {
|
|
"directions": [
|
|
{"id": f"D{i+1}", "producer-run-id": f"R{i+1}", "context-package-id": f"CP{i+1}",
|
|
"coded-slice": _cs6_tmpdir, "coded-slice-sha256": "deadbeef"}
|
|
for i in range(3)
|
|
],
|
|
"representative-screen": {"id": "S1"},
|
|
"comparison-preview": {"receipt-ref": "r", "gallery-path": "g"},
|
|
}},
|
|
],
|
|
}
|
|
check("F6: coded-slice is a directory (not a file) -> _directions_diverged False (no crash)",
|
|
SE._directions_diverged(led_dir, led_dir["artifacts"]) is False)
|
|
check("F6: lint _file_sha(<directory>) -> None (no crash)",
|
|
L._file_sha(_cs6_tmpdir) is None)
|
|
shutil.rmtree(_cs6_tmpdir, ignore_errors=True)
|
|
|
|
# 6) Fix2 회귀: direction-set 의 모든 direction 이 producer-run-id 없음({None} 만 남는 경우) -> panel fail closed
|
|
led_panel_nopid = {
|
|
"design-direction-active": {"cycle-id": "C6", "review-panel-report-id": "RP-C6", "direction-set-report-id": "DS-C6"},
|
|
"artifacts": [
|
|
{"design-type": "design-review-panel", "report-id": "RP-C6",
|
|
"design-review-panel": {
|
|
"reviews": [_review(l, f"REV-{l}") for l in _lenses],
|
|
"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR", "unresolved-dissent": []},
|
|
}},
|
|
{"design-type": "direction-set", "report-id": "DS-C6", "direction-set": {
|
|
"directions": [
|
|
{"id": f"D{i+1}", "context-package-id": f"CP{i+1}"} # producer-run-id 없음(전부 None)
|
|
for i in range(3)
|
|
],
|
|
"representative-screen": {"id": "S1"},
|
|
"comparison-preview": {"receipt-ref": "r", "gallery-path": "g"},
|
|
}},
|
|
],
|
|
}
|
|
check("Fix2: direction-set producers all missing(None) -> _critique_panel_ok False (fail closed)",
|
|
SE._critique_panel_ok(led_panel_nopid, led_panel_nopid["artifacts"]) is False)
|
|
|
|
# --- Task 11: register-direction-approval + _has_direction_approval exact 8점(fail-open 없음) ---
|
|
|
|
# Step 1(brief) — 최소 RED
|
|
check("no approval link -> not approved", not SE._has_direction_approval("wfx", {"plan": "cascade"}))
|
|
_approved_stub = {"parent-workflow-id": "p", "product-decision-id": "PD", "direction-input-brief-sha256": "OLD",
|
|
"child-workflow-id": "c", "selected-direction-sha256": "S", "winner-prototype-sha256": "W"}
|
|
check("brief mismatch -> stale(not matches)", not SE._approval_brief_matches(_approved_stub, "NEW"))
|
|
check("brief match -> ok", SE._approval_brief_matches(_approved_stub, "OLD"))
|
|
with isolated_ws() as ws:
|
|
r = _se("register-direction-approval", "--parent-workflow", "p", "--child-workflow", "ghost",
|
|
"--report", "/nope.yaml", "--report-sha256", "x", ws=ws)
|
|
check("register with ghost child -> reject(non-zero exit)", r.returncode != 0)
|
|
|
|
for k in ["approved-direction-valid", "approval-receipt-bound", "parent-approval-link-recorded", "design-direction-approved"]:
|
|
check(f"predicate {k} wired", k in SE._PREDICATES)
|
|
|
|
|
|
def _setup_dd_scenario(ws, tag, final_stage="design-direction-approved"):
|
|
"""부모+자식 design-direction 워크플로를 approved-direction 8점 검증에 필요한 전 재료(부모
|
|
accepted product-decision, child(기본은 design-direction-approved stage로 force-write —
|
|
8점 나머지 조건을 이미 종료된 상태로 단위검증하기 위함), 승인문서가 참조하는
|
|
selected-direction/winner-prototype 실 파일, approved-direction report 파일)와 함께 만들어
|
|
돌려준다. register-direction-approval 은 각 테스트가 필요에 따라 직접 호출한다.
|
|
|
|
final_stage="design-direction-finalize" 로 호출하면 child 를 finalize 단계에 남겨둔다 —
|
|
실제 finalize->approved 전이를 REAL 하게 구동하는 e2e 테스트(Important 1)가 이 옵션을 쓴다."""
|
|
parent = f"p11-{tag}"
|
|
pd_id = f"PD11-{tag}"
|
|
_se("init", "--workflow", parent, ws=ws)
|
|
check(f"setup({tag}): parent accepted product-decision appended",
|
|
AL.append_event(AL.build_event(pd_id, "accepted", workflow=parent, role="HUMAN-001")))
|
|
child = f"dd11-{tag}-{uuid.uuid4().hex[:6]}"
|
|
r = _se("init", "--workflow", child, "--plan", "design-direction", "--parent-workflow", parent,
|
|
"--product-decision", pd_id, "--direction-input-brief", _IB_PATH, ws=ws)
|
|
check(f"setup({tag}): child design-direction init -> accept(exit 0)", r.returncode == 0)
|
|
child_led = SE._load_ledger_safe(child)
|
|
forced = {
|
|
"workflow-event-id": f"test-force-{uuid.uuid4().hex}", "event-type": "state-transition",
|
|
"workflow-id": child, "from": child_led.get("stage"), "to": "design-direction-finalize",
|
|
"actor": "OPS-ORCH", "effective-at": "2026-07-16T00:00:00Z",
|
|
}
|
|
SE._atomic_event_transaction(child, workflow_event=forced)
|
|
|
|
ib_abs = _IB_PATH if os.path.isabs(_IB_PATH) else os.path.join(ROOT, _IB_PATH)
|
|
ib_sha = _sha(ib_abs)
|
|
sel_path = _tmp({"selected": tag})
|
|
win_path = _tmp({"winner": tag})
|
|
sel_sha, win_sha = _sha(sel_path), _sha(win_path)
|
|
approved_doc = {
|
|
"parent-workflow-id": parent, "child-workflow-id": child, "product-decision-id": pd_id,
|
|
"direction-input-brief-sha256": ib_sha,
|
|
"selected-direction-ref": sel_path, "selected-direction-sha256": sel_sha,
|
|
"winner-prototype-ref": win_path, "winner-prototype-sha256": win_sha,
|
|
}
|
|
report_path = _write_contract_artifact(
|
|
ws, child, f"AD-{child}", "approved-direction", "DES-DIRECTOR", approved_doc,
|
|
"design-direction-finalize")
|
|
assert _se("submit-artifact", "--workflow", child, "--report", report_path,
|
|
"--actor", "OPS-ORCH", ws=ws).returncode == 0
|
|
if final_stage == "design-direction-approved":
|
|
SE._atomic_event_transaction(child, workflow_event={
|
|
"workflow-event-id": f"test-force-approved-{uuid.uuid4().hex}",
|
|
"event-type": "state-transition", "workflow-id": child,
|
|
"from": "design-direction-finalize", "to": "design-direction-approved",
|
|
"actor": "OPS-ORCH", "effective-at": "2026-07-16T00:00:01Z",
|
|
})
|
|
report_sha = _sha(report_path)
|
|
return dict(parent=parent, child=child, pd_id=pd_id, report_path=report_path, report_sha=report_sha,
|
|
sel_path=sel_path, sel_sha=sel_sha, win_path=win_path, win_sha=win_sha, ib_sha=ib_sha)
|
|
|
|
|
|
def _register(ws, s, child=None):
|
|
return _se("register-direction-approval", "--parent-workflow", s["parent"],
|
|
"--child-workflow", child or s["child"], "--report", s["report_path"],
|
|
"--report-sha256", s["report_sha"], ws=ws)
|
|
|
|
|
|
_UNSET = object()
|
|
|
|
|
|
def _accept_report(ws, s, report_id=None, workflow=_UNSET, report_sha256=None):
|
|
"""approved-direction report 를 accepted 로 기록한다. Important 2 fix(리뷰): receipt 는 이제
|
|
report-sha256 필드값만이 아니라, 이벤트가 가리키는 report-id 를 completion-records 경로로
|
|
재해석해 그 파일의 라이브 sha256 까지 재대조하므로, 실제로 그 경로에 report 사본을 심어둬야
|
|
genuine 케이스가 통과한다(_venture_decision_receipt_ok 와 동일 계약). workflow 를 override 하면
|
|
(reject 테스트용, 미지정 sentinel _UNSET 과 명시적 None 을 구분해 '다른 워크플로' 와 '워크플로
|
|
필드 자체 없음' 두 경우를 모두 표현할 수 있다) 파일은 s["child"] 밑에 심되 이벤트의
|
|
workflow-id 필드만 어긋나게 만들어 exact-match 가드를 검증한다."""
|
|
child = s["child"]
|
|
rid = report_id or f"AD-{child}"
|
|
if workflow is _UNSET and report_sha256 is None:
|
|
return _se("review-artifact", "--workflow", child, "--report", s["report_path"],
|
|
"--decision", "accepted", "--reviewer", "EXEC-CPO", ws=ws).returncode == 0
|
|
ev_workflow = child if workflow is _UNSET else workflow
|
|
sha = s["report_sha"] if report_sha256 is None else report_sha256
|
|
return _raw_append_event(AL.build_event(rid, "accepted", workflow=ev_workflow,
|
|
report_sha256=sha))
|
|
|
|
|
|
# --- 8점 전부 통과하는 genuine 왕복(round-trip) — fail-closed 만이 아니라 실제로 동작함을 증명 ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "ok")
|
|
check("register genuine approval -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
check("setup(ok): acceptance receipt for approved-direction report appended",
|
|
_accept_report(ws, s))
|
|
child_led = SE._load_ledger_safe(s["child"])
|
|
check("full valid chain -> _has_direction_approval True (child-shape call)",
|
|
SE._has_direction_approval(s["child"], child_led) is True)
|
|
parent_led = SE._load_ledger_safe(s["parent"])
|
|
check("full valid chain -> _has_direction_approval True (parent-shape call)",
|
|
SE._has_direction_approval(s["parent"], parent_led) is True)
|
|
r_chk = _se("check-direction-approved", "--workflow", s["parent"], ws=ws)
|
|
check("check-direction-approved(parent) -> YES(exit 0)", r_chk.returncode == 0 and "YES" in r_chk.stdout)
|
|
r_chk2 = _se("check-direction-approved", "--workflow", s["child"], ws=ws)
|
|
check("check-direction-approved(child) -> YES(exit 0)", r_chk2.returncode == 0 and "YES" in r_chk2.stdout)
|
|
|
|
# --- #3 fail-closed(parent-shape): register 는 finalize 단계에서도 허용되지만, 상위(부모/외부
|
|
# cascade) 관점에서 본 승인 상태는 child 가 실제로 design-direction-approved stage 로 전이되기
|
|
# 전엔 유효하지 않아야 한다(register != 전이). 이 관점은 parent-shape 호출(led 에
|
|
# design-direction-approval 링크를 직접 보유)로 검사한다.
|
|
# 대조적으로 child-shape(자식 자신의 finalize->approved 전이 평가) 는 바로 이 finalize 단계를
|
|
# 허용해야 한다 — Critical fix(교착 해소): 그렇지 않으면 그 전이 자체가 만드는 stage 를 전이 조건
|
|
# 으로 요구하는 셈이라 전이가 영원히 발동할 수 없다(리뷰가 지적한 deadlock). 아래 e2e 테스트가
|
|
# 이 child-shape True 상태에서 실제 전이가 성공적으로 발동함을 증명한다. ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "notyetstage")
|
|
SE._atomic_event_transaction(s["child"], workflow_event={
|
|
"workflow-event-id": f"test-force-{uuid.uuid4().hex}", "event-type": "state-transition",
|
|
"workflow-id": s["child"], "from": "design-direction-approved",
|
|
"to": "design-direction-finalize", "actor": "OPS-ORCH",
|
|
"effective-at": "2026-07-16T00:00:00Z"})
|
|
check("register while child still at finalize stage -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
_accept_report(ws, s)
|
|
check("parent-shape: child not yet at design-direction-approved stage -> _has_direction_approval False",
|
|
SE._has_direction_approval(s["parent"], SE._load_ledger_safe(s["parent"])) is False)
|
|
check("child-shape(Critical fix): same finalize stage -> _has_direction_approval True "
|
|
"(그래야 finalize->approved 전이가 발동할 수 있다 — 교착 아님)",
|
|
SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is True)
|
|
|
|
# --- #6 fail-closed: 승인 이후 부모가 그 product-decision 을 새 결정으로 supersede -> False ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "superseded")
|
|
check("register(superseded-setup) -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
_accept_report(ws, s)
|
|
check("before supersede -> True", SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is True)
|
|
AL.append_event(AL.build_event(f"{s['pd_id']}-v2", "accepted", workflow=s["parent"], role="HUMAN-001",
|
|
supersedes=s["pd_id"]))
|
|
check("product-decision superseded by newer decision -> _has_direction_approval False",
|
|
SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is False)
|
|
|
|
# --- #7 fail-closed(live staleness): 승인 이후 child 의 direction-input-brief-ref 가 바뀌면(그 사이
|
|
# input-brief 가 갱신됐다는 뜻) 박제된 hash 와 어긋나 stale -> False ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "stale")
|
|
check("register(stale-setup) -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
_accept_report(ws, s)
|
|
_new_ib = _tmp({"changed": True})
|
|
child_led = SE._load_ledger_safe(s["child"])
|
|
child_led["direction-input-brief-ref"] = _new_ib
|
|
SE._write_ledger(s["child"], child_led)
|
|
check("direct workflow.yaml binding edit cannot override initialized event projection",
|
|
SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is True)
|
|
|
|
# --- #8a fail-closed: 승인문서가 가리키는 selected-direction-ref 파일이 사후 변조되면 hash 불일치 -> False ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "tamper")
|
|
check("register(tamper-setup) -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
_accept_report(ws, s)
|
|
open(s["sel_path"], "w").write("tampered-after-approval")
|
|
check("selected-direction-ref tampered after approval -> _has_direction_approval False",
|
|
SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is False)
|
|
|
|
# --- #8b fail-closed: acceptance receipt 자체가 없으면(자기신고만으로는) 승인 불성립 -> False ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "noreceipt")
|
|
check("register(noreceipt-setup) -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
check("no acceptance receipt for approved-direction report -> _has_direction_approval False",
|
|
SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is False)
|
|
|
|
# --- register 자체의 충돌 가드: 같은 parent 에 이미 active 한 approval(child A) 이 있는데 다른
|
|
# child(B) 로 재등록하면 거부해야 한다(부모당 동시 active approval 1건) ---
|
|
with isolated_ws() as ws:
|
|
s1 = _setup_dd_scenario(ws, "conflictA")
|
|
check("register(conflictA) -> accept(exit 0)", _register(ws, s1).returncode == 0)
|
|
pd2 = f"PD11-conflictB-{uuid.uuid4().hex[:6]}"
|
|
AL.append_event(AL.build_event(pd2, "accepted", workflow=s1["parent"], role="HUMAN-001"))
|
|
child2 = f"dd11-conflictB-{uuid.uuid4().hex[:6]}"
|
|
_se("init", "--workflow", child2, "--plan", "design-direction", "--parent-workflow", s1["parent"],
|
|
"--product-decision", pd2, "--direction-input-brief", _IB_PATH, ws=ws)
|
|
SE._atomic_event_transaction(child2, workflow_event={
|
|
"workflow-event-id": f"test-force-{uuid.uuid4().hex}", "event-type": "state-transition",
|
|
"workflow-id": child2, "from": "design-direction-intake",
|
|
"to": "design-direction-approved", "actor": "OPS-ORCH",
|
|
"effective-at": "2026-07-16T00:00:00Z"})
|
|
r2 = _register(ws, s1, child=child2)
|
|
check("register conflicting second child for same parent(existing active link) -> reject",
|
|
r2.returncode != 0)
|
|
|
|
# --- Important 2 reject test(리뷰가 지적한 missing test): receipt 이벤트의 workflow-id 가 이
|
|
# child 와 다르거나(엉뚱한 workflow 로 편승) 아예 없으면(None) — report-sha256 필드가 정확히
|
|
# 같아도 — 승인이 성립하지 않아야 한다. (기존 exact-match 가드가 이미 이를 막고 있었지만 회귀
|
|
# 테스트가 없었다 — 이제 명시적으로 고정한다.) ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "wrongwf")
|
|
check("register(wrongwf-setup) -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
_accept_report(ws, s, workflow=f"not-{s['child']}")
|
|
check("receipt workflow-id != child(다른 workflow, 같은 report-sha256) -> _has_direction_approval False",
|
|
SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is False)
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "nonewf")
|
|
check("register(nonewf-setup) -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
_accept_report(ws, s, workflow=None)
|
|
check("receipt workflow-id missing(None, 같은 report-sha256) -> _has_direction_approval False",
|
|
SE._has_direction_approval(s["child"], SE._load_ledger_safe(s["child"])) is False)
|
|
|
|
# --- Important 1: REAL end-to-end transition test. 위의 happy-path 테스트들은 전부
|
|
# child_led["stage"] = "design-direction-approved" 를 force-write 해서 8점 검증만 단위로 확인할 뿐,
|
|
# 실제 finalize->approved 전이가 발동하는지는 결코 증명하지 않는다 — 그래서 check 3 가 그 전이가
|
|
# 만드는 바로 그 stage 를 조건으로 요구하는 교착(Critical)을 숨겼다. 이 테스트는 child 를
|
|
# design-direction-finalize 에 그대로 남겨둔 채(force-write 로 approved 로 건너뛰지 않음) 부모
|
|
# 링크를 등록하고 receipt 를 남긴 뒤, 실제 `state_engine.py transition` CLI 로 REAL 전이를
|
|
# 구동한다 — check-3 mode fix 이전에는 이 테스트가 반드시 실패(교착으로 exit != 0)하고, 이후에는
|
|
# 반드시 통과(exit 0 + stage 실제 갱신)해야 한다. ---
|
|
with isolated_ws() as ws:
|
|
s = _setup_dd_scenario(ws, "e2e", final_stage="design-direction-finalize")
|
|
check("e2e: register genuine approval -> accept(exit 0)", _register(ws, s).returncode == 0)
|
|
check("e2e: acceptance receipt for approved-direction report appended", _accept_report(ws, s))
|
|
pre_stage = SE._load_ledger_safe(s["child"]).get("stage")
|
|
check("e2e: child still at design-direction-finalize before REAL transition (not force-written to approved)",
|
|
pre_stage == "design-direction-finalize")
|
|
r_t = _se("transition", "--workflow", s["child"], "--to", "design-direction-approved",
|
|
"--actor", "OPS-ORCH", ws=ws)
|
|
check(f"e2e: REAL finalize->approved transition fires -> exit 0 (not deadlocked; stderr: {r_t.stderr.strip()[:300]!r})",
|
|
r_t.returncode == 0)
|
|
post_stage = SE._load_ledger_safe(s["child"]).get("stage")
|
|
check("e2e: child stage after REAL transition == design-direction-approved",
|
|
post_stage == "design-direction-approved")
|
|
|
|
# --- Task 13: 부모 cascade gate — _is_ui_bearing + design-direction-gate-satisfied + find-child-direction CLI ---
|
|
|
|
check("ui-bearing comes from workload-profile.surfaces.ui",
|
|
SE._is_ui_bearing({"artifacts": [{"artifact-kind": "workload-profile",
|
|
"workload-profile": {"surfaces": {"ui": True}}}]}))
|
|
check("ui-bearing explicit false", not SE._is_ui_bearing({"deliverable-profile": {"ui-bearing": False}, "build-families": ["FAM-ENG-FRONTEND"]}))
|
|
check("ui-bearing does not fall back to FE family", not SE._is_ui_bearing({"build-families": ["FAM-ENG-FRONTEND"]}))
|
|
check("ui-bearing fallback non-FE -> False", not SE._is_ui_bearing({"build-families": ["FAM-ENG-BACKEND"]}))
|
|
check("ui-bearing does not fall back to deliverable-kind", not SE._is_ui_bearing({"deliverable-kind": "frontend-screen"}))
|
|
check("FixA: deliverable-kind 'ui-backend-mapping' -> NOT ui-bearing (tightened token match, not raw substring)",
|
|
not SE._is_ui_bearing({"deliverable-kind": "ui-backend-mapping"}))
|
|
check("FixA: deliverable-kind 'ui' alone is not a second SSOT", not SE._is_ui_bearing({"deliverable-kind": "ui"}))
|
|
|
|
# --- Fix A(최종리뷰): 게이트가 죽어있던 production 경로 — accepted decision-brief report 의
|
|
# deliverable-profile.ui-bearing 을 원장에 아무 CLI 도 쓰지 않아도 신호로 쓴다 ---
|
|
with isolated_ws() as ws:
|
|
wf_dbtrue = "wf-fixA-db-true"
|
|
db_true_path = _tmp({"report-header": {"bottom-line": "x"}, "deliverable-profile": {"ui-bearing": True, "ui-kind": "web-app"}})
|
|
led_dbtrue = {"artifacts": [{"design-type": "decision-brief", "report-id": "DB-A-TRUE", "path": db_true_path}]}
|
|
check("FixA setup: accept decision-brief(ui-bearing=true) report-id",
|
|
AL.append_event(AL.build_event("DB-A-TRUE", "accepted", workflow=wf_dbtrue)))
|
|
check("legacy decision-brief deliverable-profile no longer controls UI-bearing",
|
|
SE._is_ui_bearing(led_dbtrue, wf_dbtrue) is False)
|
|
|
|
wf_dbfalse = "wf-fixA-db-false"
|
|
db_false_path = _tmp({"report-header": {"bottom-line": "x"}, "deliverable-profile": {"ui-bearing": False, "ui-kind": "api-only"}})
|
|
led_dbfalse = {"artifacts": [{"design-type": "decision-brief", "report-id": "DB-A-FALSE", "path": db_false_path}],
|
|
"build-families": ["FAM-ENG-FRONTEND"]}
|
|
check("FixA setup: accept decision-brief(ui-bearing=false) report-id",
|
|
AL.append_event(AL.build_event("DB-A-FALSE", "accepted", workflow=wf_dbfalse)))
|
|
check("FixA: accepted decision-brief deliverable-profile.ui-bearing=false -> _is_ui_bearing False "
|
|
"even though build-families has FAM-ENG-FRONTEND (explicit decision-brief signal beats the fallback)",
|
|
SE._is_ui_bearing(led_dbfalse, wf_dbfalse) is False)
|
|
|
|
wf_dbunacc = "wf-fixA-db-unaccepted"
|
|
db_unacc_path = _tmp({"deliverable-profile": {"ui-bearing": True}})
|
|
led_dbunacc = {"artifacts": [{"design-type": "decision-brief", "report-id": "DB-A-UNACC", "path": db_unacc_path}]}
|
|
check("FixA: UNaccepted decision-brief ui-bearing=true is ignored (fail closed, no acceptance_log event) "
|
|
"-> falls through to False (no deliverable-kind/build-families signal either)",
|
|
SE._is_ui_bearing(led_dbunacc, wf_dbunacc) is False)
|
|
|
|
pred = SE._PREDICATES.get("design-direction-gate-satisfied")
|
|
check("predicate design-direction-gate-satisfied wired", pred is not None)
|
|
f_block = {"tier": "standard", "_ui_bearing": True, "design_direction_approved": False}
|
|
f_pass_light = {"tier": "light", "_ui_bearing": True, "design_direction_approved": False}
|
|
f_pass_nonui = {"tier": "standard", "_ui_bearing": False, "design_direction_approved": False}
|
|
f_pass_approved = {"tier": "heavy", "_ui_bearing": True, "design_direction_approved": True}
|
|
check("gate blocks UI+standard w/o approval", not pred(f_block)[0])
|
|
check("gate passes UI+light", pred(f_pass_light)[0])
|
|
check("gate passes non-UI", pred(f_pass_nonui)[0])
|
|
check("gate passes UI+heavy with approval", pred(f_pass_approved)[0])
|
|
check("protected fact _ui_bearing", "_ui_bearing" in SE._PROTECTED_FACTS)
|
|
|
|
# _facts does not infer UI-bearing from build family.
|
|
_f_ui = SE._facts("wf-ui-bearing-t13", {"build-families": ["FAM-ENG-FRONTEND"], "tier": "standard"}, None)
|
|
check("_facts does not derive _ui_bearing from FE build-family", _f_ui.get("_ui_bearing") is False)
|
|
_f_nonui = SE._facts("wf-non-ui-t13", {"build-families": ["FAM-ENG-BACKEND"], "tier": "standard"}, None)
|
|
check("_facts derives _ui_bearing False for non-FE build-family", _f_nonui.get("_ui_bearing") is False)
|
|
|
|
# design->spec transition now requires design-direction-gate-satisfied in addition to design-accepted
|
|
_ddg_trans = [t for t in trans if t.get("from") == "design" and t.get("to") == "spec"]
|
|
check("design->spec transition exists", len(_ddg_trans) == 1)
|
|
check("design->spec required-conditions include design-direction-gate-satisfied",
|
|
"design-direction-gate-satisfied" in (_ddg_trans[0].get("required-conditions") or []))
|
|
check("design->spec required-conditions still include design-accepted",
|
|
"design-accepted" in (_ddg_trans[0].get("required-conditions") or []))
|
|
|
|
# --- find-child-direction CLI (dedup gap closed for /design) ---
|
|
with isolated_ws() as ws:
|
|
_se("init", "--workflow", "pc13", ws=ws)
|
|
check("setup(t13): acceptance event appended (pc13/PD13)",
|
|
AL.append_event(AL.build_event("PD13", "accepted", workflow="pc13", role="HUMAN-001")))
|
|
wf13 = f"dd-{uuid.uuid4().hex[:8]}"
|
|
r_init13 = _se("init", "--workflow", wf13, "--plan", "design-direction", "--parent-workflow", "pc13",
|
|
"--product-decision", "PD13", "--direction-input-brief", _IB_PATH, ws=ws)
|
|
check("setup(t13): child design-direction init -> accept(exit 0)", r_init13.returncode == 0)
|
|
_abs_ib13 = _IB_PATH if os.path.isabs(_IB_PATH) else os.path.join(ROOT, _IB_PATH)
|
|
_brief_sha13 = _sha(_abs_ib13)
|
|
r_find = _se("find-child-direction", "--parent-workflow", "pc13", "--product-decision", "PD13",
|
|
"--direction-input-brief-sha256", _brief_sha13, ws=ws)
|
|
check("find-child-direction CLI -> exit 0 when found", r_find.returncode == 0)
|
|
check("find-child-direction CLI -> stdout mentions found child workflow-id", wf13 in r_find.stdout)
|
|
check("find-child-direction CLI -> stdout reports stale=False(matching sha)",
|
|
"false" in r_find.stdout.lower())
|
|
|
|
r_find_none = _se("find-child-direction", "--parent-workflow", "no-such-parent-t13",
|
|
"--product-decision", "PD-X13", "--direction-input-brief-sha256", "deadbeef", ws=ws)
|
|
check("find-child-direction CLI -> exit 0 when nothing found", r_find_none.returncode == 0)
|
|
check("find-child-direction CLI -> null/empty when nothing found",
|
|
"null" in r_find_none.stdout.lower() or not r_find_none.stdout.strip()
|
|
or r_find_none.stdout.strip() == "{}")
|
|
|
|
# --- Task 15 item 7: selected-direction acceptance not substitutable by a different report's
|
|
# acceptance(bundle+cycle). design-direction.md §3 명시: "selected-direction-accepted 는 bundle
|
|
# lint 통과 + 이 accepted 이벤트 둘 다 요구" — "이" accepted 이벤트란 selected-direction 자신의
|
|
# report-id 에 대한 것이어야 한다. 예전엔 `_al_has_accepted(wf)`(이 workflow 에 아무 report나
|
|
# accepted 됐는지만 확인)를 썼는데, 이는 이 workflow 에서 우연히/별도로 accepted 된 무관한 report로도
|
|
# 이 게이트를 대체 통과시킬 수 있었다(위조 경로 — bundle+cycle 결속 없이 아무 accepted 이벤트나
|
|
# substitutable). `_selected_direction_accepted_ok`(state_engine.py)가 `_al_accepted_ids(wf)`
|
|
# exact report-id 매칭으로 이를 닫는다(fix 적용됨, 아래가 회귀 테스트). ---
|
|
with isolated_ws() as ws:
|
|
dsp7 = _ds_file(("DIR-1", "DIR-2", "DIR-3"))
|
|
selp7 = _sel_file(dsp7) # selected-direction-id=DIR-2, 유효 bundle
|
|
wf7 = f"wf-item7-{uuid.uuid4().hex[:6]}"
|
|
led7 = {"design-direction-active": {"cycle-id": "C7", "selected-direction-report-id": "SD-7",
|
|
"direction-set-report-id": "DS-7"},
|
|
"artifacts": [
|
|
{"design-type": "direction-set", "report-id": "DS-7", "path": dsp7},
|
|
{"artifact-kind": "selected-direction", "design-type": "selected-direction",
|
|
"report-id": "SD-7", "path": selp7, "artifact-sha256": _sha(selp7)},
|
|
]}
|
|
check("item7: bundle lints clean but no acceptance at all -> selected_direction_accepted False",
|
|
not SE._facts(wf7, led7, None).get("selected_direction_accepted"))
|
|
check("item7: accept an UNRELATED report for the SAME workflow -> still False(not substitutable)",
|
|
AL.append_event(AL.build_event("SOME-UNRELATED-REPORT", "accepted", workflow=wf7, role="HUMAN-001")))
|
|
check("item7: unrelated report's acceptance does not satisfy selected_direction_accepted",
|
|
not SE._facts(wf7, led7, None).get("selected_direction_accepted"))
|
|
check("item7: now accept the selected-direction's OWN report-id(SD-7)",
|
|
_raw_append_event(AL.build_event("SD-7", "accepted", workflow=wf7, role="EXEC-CPO",
|
|
report_sha256=_sha(selp7))))
|
|
check("item7: the correct(exact) report-id's acceptance -> selected_direction_accepted True",
|
|
SE._facts(wf7, led7, None).get("selected_direction_accepted") is True)
|
|
|
|
# --- Task 15 item 8: child receipt not reusable as a DIFFERENT(e.g. parent) workflow's receipt —
|
|
# direct-function coverage for `_direction_approval_receipt_ok`'s exact workflow-id match, on top
|
|
# of the existing integration coverage(wrongwf/nonewf tests above, via `_has_direction_approval`). ---
|
|
check("item8: _direction_approval_receipt_ok with no matching event at all -> False",
|
|
not SE._direction_approval_receipt_ok(f"ghost-child-{uuid.uuid4().hex[:6]}", "r.yaml", "deadbeefSHA"))
|
|
with isolated_ws() as ws:
|
|
_child8, _parent8 = f"child8-{uuid.uuid4().hex[:6]}", f"parent8-{uuid.uuid4().hex[:6]}"
|
|
_report8 = _tmp({"approved-direction": {"marker": "item8"}})
|
|
_sha8 = _sha(_report8)
|
|
_dest8 = os.path.join(ws, "completion-records", _child8)
|
|
os.makedirs(_dest8, exist_ok=True)
|
|
shutil.copyfile(_report8, os.path.join(_dest8, "AD-8.report.yaml"))
|
|
check("item8 setup: accepted receipt appended, bound to CHILD workflow-id(real report file+hash)",
|
|
AL.append_event(AL.build_event("AD-8", "accepted", workflow=_child8, role="DES-DIRECTOR", report_sha256=_sha8)))
|
|
check("item8: receipt issued under the CHILD workflow-id is NOT reusable when queried "
|
|
"under a DIFFERENT(e.g. parent) workflow-id, even with the same report-sha256",
|
|
not SE._direction_approval_receipt_ok(_parent8, _report8, _sha8))
|
|
check("item8: (sanity) the SAME receipt IS recognized when queried under its own exact workflow-id",
|
|
SE._direction_approval_receipt_ok(_child8, _report8, _sha8))
|
|
|
|
# --- Task 15 item 9: gallery/self-reported preview alone does NOT pass critique — a REAL
|
|
# evidence-ledger preview_ui receipt(exit 0) is required in addition to the panel's own structural
|
|
# validity. `led_c2_ok`(defined above) already makes `_critique_panel_ok` True via a disjoint
|
|
# producer/reviewer set + pass verdict + resolvable direction-set — but that direction-set's
|
|
# self-reported `comparison-preview: {receipt-ref: "r", gallery-path: "g"}` is a *document claim*,
|
|
# not a verified receipt. `direction_critique_passed` must not be satisfied by that claim alone. ---
|
|
with isolated_ws() as ws:
|
|
wf9 = f"wf-item9-{uuid.uuid4().hex[:6]}"
|
|
check("item9: panel is structurally valid(_critique_panel_ok) even before any real preview receipt",
|
|
SE._critique_panel_ok(led_c2_ok, led_c2_ok["artifacts"]) is True)
|
|
check("item9: no real evidence-ledger preview_ui receipt planted yet -> direction_critique_passed "
|
|
"False(self-reported comparison-preview/gallery-path alone is not sufficient)",
|
|
not SE._facts(wf9, led_c2_ok, None).get("direction_critique_passed"))
|
|
import _workspace as W9 # noqa: E402
|
|
_ed9 = W9.evidence_dir()
|
|
os.makedirs(_ed9, exist_ok=True)
|
|
_preview9 = os.path.join(ws, "preview.png")
|
|
with open(_preview9, "wb") as _png9:
|
|
_png9.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 2000)
|
|
with open(os.path.join(_ed9, "ledger.jsonl"), "a", encoding="utf-8") as fh9:
|
|
fh9.write(json.dumps({"tool_use_id": "item9-preview", "tool_name": "Bash",
|
|
"command": f"python3 .claude/hooks/preview_ui.py {ws} --out {_preview9} --viewports 360 --check-css",
|
|
"exit_code": 0, "workflow_id": wf9, "session_id": "item9-session",
|
|
"agent_id": "item9-agent"}) + "\n")
|
|
check("item9: real preview_ui receipt(exit 0, bound to this workflow) planted -> "
|
|
"direction_critique_passed now True(winner-prototype preview genuinely required)",
|
|
SE._facts(wf9, led_c2_ok, None).get("direction_critique_passed") is True)
|
|
|
|
# --- Task 15 item 14: critique reverse-transition FACTS for all 3 verdict routes. minor-revision/
|
|
# concept-flaw covered here via synthetic ledgers(brief's own pattern); pass->direction_critique_passed
|
|
# is proven positively by item9 above and by the FULL e2e's real critique->finalize transition below.
|
|
# Also pins the transition rules' required-conditions so the 3 routes stay wired to these exact facts. ---
|
|
for _v14, _key14 in [("minor-revision", "critique_revision_requested"), ("concept-flaw", "concept_rejection_recorded")]:
|
|
_led14 = {"design-direction-active": {"cycle-id": "C14", "review-panel-report-id": "RP14"},
|
|
"artifacts": [{"design-type": "design-review-panel", "report-id": "RP14", "direction-cycle-id": "C14",
|
|
"design-review-panel": {"synthesis": {"verdict": _v14, "role-id": "DES-DIRECTOR"}, "reviews": []}}]}
|
|
check(f"item14: verdict={_v14} -> {_key14} True(routes critique -> the matching reverse/forward stage)",
|
|
bool(SE._facts(f"wf-item14-{_v14}", _led14, None).get(_key14)))
|
|
check("item14: critique->prototype gated exactly on [critique-revision-requested](minor-revision route)",
|
|
any(t.get("from") == "design-direction-critique" and t.get("to") == "design-direction-prototype"
|
|
and t.get("required-conditions") == ["critique-revision-requested"] for t in trans))
|
|
check("item14: critique->divergence gated exactly on [concept-rejection-recorded](concept-flaw route)",
|
|
any(t.get("from") == "design-direction-critique" and t.get("to") == "design-direction-divergence"
|
|
and t.get("required-conditions") == ["concept-rejection-recorded"] for t in trans))
|
|
check("item14: critique->finalize gated exactly on [direction-critique-passed](pass route)",
|
|
any(t.get("from") == "design-direction-critique" and t.get("to") == "design-direction-finalize"
|
|
and t.get("required-conditions") == ["direction-critique-passed"] for t in trans))
|
|
|
|
# --- Task 15 item 10: finalize artifacts absent(no register-direction-approval, no accepted
|
|
# approved-direction report — nothing at all) -> the REAL terminal transition(subprocess CLI,
|
|
# not a force-write) is rejected. ---
|
|
with isolated_ws() as ws:
|
|
s10 = _setup_dd_scenario(ws, "item10", final_stage="design-direction-finalize")
|
|
r10 = _se("transition", "--workflow", s10["child"], "--to", "design-direction-approved",
|
|
"--actor", "OPS-ORCH", ws=ws)
|
|
check("item10: finalize artifacts entirely absent -> REAL terminal transition rejected(non-zero exit)",
|
|
r10.returncode != 0)
|
|
check("item10: child stage unchanged(still design-direction-finalize) after the rejected transition",
|
|
SE._load_ledger_safe(s10["child"]).get("stage") == "design-direction-finalize")
|
|
|
|
# --- Task 15 item 11: approved-direction report IS accepted(child-side receipt exists), but
|
|
# `register-direction-approval` was NEVER called(parent-approval-link absent) -> the REAL terminal
|
|
# transition is still rejected — accepting the report is not itself the parent-side registration. ---
|
|
with isolated_ws() as ws:
|
|
s11 = _setup_dd_scenario(ws, "item11", final_stage="design-direction-finalize")
|
|
check("item11 setup: acceptance receipt for approved-direction report appended(register NOT called)",
|
|
_accept_report(ws, s11))
|
|
r11 = _se("transition", "--workflow", s11["child"], "--to", "design-direction-approved",
|
|
"--actor", "OPS-ORCH", ws=ws)
|
|
check("item11: accepted report exists but parent-approval-link never registered -> "
|
|
"REAL terminal transition rejected(non-zero exit)", r11.returncode != 0)
|
|
check("item11: child stage unchanged(still design-direction-finalize) after the rejected transition",
|
|
SE._load_ledger_safe(s11["child"]).get("stage") == "design-direction-finalize")
|
|
|
|
|
|
# ============================================================================================
|
|
# FULL END-TO-END: drive a design-direction child through EVERY stage via REAL state_engine.py
|
|
# subprocess transitions, from a genuine parent cascade workflow with a real accepted
|
|
# product-decision and a frozen direction-input-brief, all the way to design-direction-approved —
|
|
# then assert the PARENT cascade's check-direction-approved flips to YES, and that a UI-bearing
|
|
# standard-tier parent can now pass the design->spec gate.
|
|
#
|
|
# Regression: this E2E drives every design-direction stage using the public typed submit/review
|
|
# APIs. Artifact content is read only from immutable, hash-checked workflow-artifact envelopes;
|
|
# neither direct ledger mutation nor the removed generic `record` command participates.
|
|
# ============================================================================================
|
|
|
|
with isolated_ws() as ws:
|
|
_e2e_tmpdirs = []
|
|
parent = f"p-e2e-{uuid.uuid4().hex[:8]}"
|
|
pd_id = f"PD-e2e-{uuid.uuid4().hex[:8]}"
|
|
r = _se("init", "--workflow", parent, ws=ws)
|
|
check("FULL-E2E: real parent cascade workflow init -> exit 0", r.returncode == 0)
|
|
check("FULL-E2E: real accepted product-decision appended for parent",
|
|
AL.append_event(AL.build_event(pd_id, "accepted", workflow=parent, role="HUMAN-001")))
|
|
|
|
ib_path = _tmp(dict(_IB_REQ)) # frozen(valid) direction-input-brief
|
|
ib_sha = _sha(ib_path)
|
|
|
|
child = f"dd-e2e-{uuid.uuid4().hex[:8]}"
|
|
r = _se("init", "--workflow", child, "--plan", "design-direction", "--parent-workflow", parent,
|
|
"--product-decision", pd_id, "--direction-input-brief", ib_path, ws=ws)
|
|
check("FULL-E2E: real child design-direction init(bound to real parent+accepted PD) -> exit 0",
|
|
r.returncode == 0)
|
|
check("FULL-E2E: child starts at design-direction-intake",
|
|
_se("current", "--workflow", child, ws=ws).stdout.strip() == "design-direction-intake")
|
|
|
|
# --- intake -> discovery ---
|
|
check("FULL-E2E: guard intake->discovery(parent-binding-present+direction-input-brief-valid) -> exit 0",
|
|
_se("guard", "--workflow", child, "--to", "design-direction-discovery", ws=ws).returncode == 0)
|
|
r = _se("transition", "--workflow", child, "--to", "design-direction-discovery", "--actor", "OPS-ORCH", ws=ws)
|
|
check("FULL-E2E: REAL transition intake->discovery -> exit 0", r.returncode == 0)
|
|
disc_path = _tmp({"direction-input-brief-sha256": ib_sha, "findings": ["f1", "f2"],
|
|
"constraints-restated": ["c1"], "opportunity-notes": ["o1"]})
|
|
check("FULL-E2E: record direction-discovery -> exit 0",
|
|
_submit_raw_as_artifact(ws, child, "E2E-DISC", "direction-discovery",
|
|
disc_path).returncode == 0)
|
|
_axis_names = ["layout-topology", "navigation-model", "typography-voice",
|
|
"imagery-strategy", "motion-model", "dominant-primitives"]
|
|
charter_content = {
|
|
"direction-cycle-id": "E2E-C1",
|
|
"representative-screen": {"id": "S1", "kind": "core-task", "description": "same semantic task"},
|
|
"directions": [
|
|
{"id": "E2E-D1", "design-question": "editorial causality",
|
|
"layout-topology": "asymmetric editorial spread", "navigation-model": "scroll annotations",
|
|
"typography-voice": "serif display plus grotesk", "imagery-strategy": "data annotations",
|
|
"motion-model": "reading cursor reveal", "dominant-primitives": ["margin-note", "rule-line"],
|
|
"exclusive-primitives": ["folio-index", "pull-quote"],
|
|
"forbidden-primitives": ["centered-card", "progress-rail"]},
|
|
{"id": "E2E-D2", "design-question": "spatial direct manipulation",
|
|
"layout-topology": "full-bleed spatial stage", "navigation-model": "direct manipulation",
|
|
"typography-voice": "compact industrial sans", "imagery-strategy": "physical session objects",
|
|
"motion-model": "drag and collision", "dominant-primitives": ["session-object", "connection-path"],
|
|
"exclusive-primitives": ["spatial-canvas", "drag-handle"],
|
|
"forbidden-primitives": ["pill-cta", "white-app-shell"]},
|
|
{"id": "E2E-D3", "design-question": "sequential illustrated explanation",
|
|
"layout-topology": "vertical storyboard", "navigation-model": "chapter paging",
|
|
"typography-voice": "handwritten captions", "imagery-strategy": "bespoke narrative panels",
|
|
"motion-model": "panel-to-panel transition", "dominant-primitives": ["story-panel", "caption-balloon"],
|
|
"exclusive-primitives": ["character-scene", "chapter-marker"],
|
|
"forbidden-primitives": ["dashboard-grid", "session-tile"]},
|
|
],
|
|
"pairwise-separation": [
|
|
{"directions": [a, b], "differing-axes": _axis_names, "allowed-overlap": "semantic task only"}
|
|
for a, b in (("E2E-D1", "E2E-D2"), ("E2E-D1", "E2E-D3"), ("E2E-D2", "E2E-D3"))
|
|
],
|
|
}
|
|
charter_path = _tmp(charter_content)
|
|
check("FULL-E2E: submit divergence-charter(3 pairwise-separated territories)",
|
|
_submit_raw_as_artifact(ws, child, "E2E-CHARTER", "divergence-charter",
|
|
charter_path).returncode == 0)
|
|
check("FULL-E2E: accept divergence-charter exact revision",
|
|
_se("review-artifact", "--workflow", child,
|
|
"--report", _submitted_paths[(child, "E2E-CHARTER")], "--decision", "accepted",
|
|
"--reviewer", "EXEC-CPO", ws=ws).returncode == 0)
|
|
# --- discovery -> divergence ---
|
|
check("FULL-E2E: guard discovery->divergence(discovery+charter) -> exit 0",
|
|
_se("guard", "--workflow", child, "--to", "design-direction-divergence", ws=ws).returncode == 0)
|
|
r = _se("transition", "--workflow", child, "--to", "design-direction-divergence", "--actor", "OPS-ORCH", ws=ws)
|
|
check("FULL-E2E: REAL transition discovery->divergence -> exit 0", r.returncode == 0)
|
|
|
|
# --- divergence: 3 independently-run directions, distinct producer-run-ids, real coded-slice,
|
|
# comparison-preview(gallery) metadata ---
|
|
slice_dir = tempfile.mkdtemp(prefix="dd-e2e-slice-"); _e2e_tmpdirs.append(slice_dir)
|
|
slice_path = os.path.join(slice_dir, "slice.tsx")
|
|
open(slice_path, "w").write("export const Slice = () => null;\n")
|
|
slice_sha = _sha(slice_path)
|
|
board_paths, preview_paths = {}, {}
|
|
for i in (1, 2, 3):
|
|
board_paths[i] = os.path.join(slice_dir, f"reference-board-{i}.png")
|
|
preview_paths[i] = os.path.join(slice_dir, f"full-size-preview-{i}.png")
|
|
open(board_paths[i], "wb").write(b"PNG synthetic reference board " + str(i).encode())
|
|
open(preview_paths[i], "wb").write(b"PNG synthetic full preview " + str(i).encode())
|
|
ds_content = {
|
|
"direction-cycle-id": "E2E-C1",
|
|
"divergence-charter-ref": _submitted_paths[(child, "E2E-CHARTER")],
|
|
"divergence-charter-sha256": _sha(_submitted_paths[(child, "E2E-CHARTER")]),
|
|
"representative-screen": {"id": "S1", "kind": "core-task", "description": "same semantic task"},
|
|
"directions": [
|
|
{"id": f"E2E-D{i}", "producer-run-id": f"E2E-RUN-{i}", "context-package-id": f"E2E-CP{i}",
|
|
"producer-role-id": "DES-VISUAL", "concept-artifact": f"concept-{i}",
|
|
"reference-cluster": [
|
|
{"name": f"unique-ref-{i}-{n}", "signal": f"signal-{n}", "why-relevant": "charter fit"}
|
|
for n in (1, 2, 3)
|
|
],
|
|
"visual-thesis": f"thesis-{i}", "layout-grammar": f"layout-{i}",
|
|
"interaction-grammar": f"interaction-{i}", "typography-token-direction": f"type-{i}",
|
|
"primitive-inventory": [f"primitive-{i}-a", f"primitive-{i}-b"],
|
|
"reference-board-ref": board_paths[i], "reference-board-sha256": _sha(board_paths[i]),
|
|
"full-size-preview-ref": preview_paths[i], "full-size-preview-sha256": _sha(preview_paths[i]),
|
|
"coded-slice": slice_path, "coded-slice-sha256": slice_sha}
|
|
for i in (1, 2, 3)
|
|
],
|
|
"comparison-preview": {"receipt-ref": "e2e-gallery-receipt", "receipt-sha256": "e2e-gallery-sha",
|
|
"gallery-path": os.path.join(slice_dir, "gallery"), "representative-screen-id": "S1"},
|
|
}
|
|
ds_path = _tmp(ds_content)
|
|
check("FULL-E2E: record direction-set(3 distinct producer-run-ids, matching coded-slice sha) -> exit 0",
|
|
_submit_raw_as_artifact(ws, child, "E2E-DS", "direction-set", ds_path).returncode == 0)
|
|
audit_content = {
|
|
"direction-cycle-id": "E2E-C1",
|
|
"divergence-charter-ref": _submitted_paths[(child, "E2E-CHARTER")],
|
|
"divergence-charter-sha256": _sha(_submitted_paths[(child, "E2E-CHARTER")]),
|
|
"direction-set-ref": _submitted_paths[(child, "E2E-DS")],
|
|
"direction-set-sha256": _sha(_submitted_paths[(child, "E2E-DS")]),
|
|
"reviewer-role-id": "DES-VISUAL", "reviewer-run-id": "E2E-COMPARE-RUN",
|
|
"verdict": "pass", "blocking-findings": [],
|
|
"pairwise-comparisons": [
|
|
{"directions": [a, b], "differing-axes": _axis_names, "primitive-collisions": []}
|
|
for a, b in (("E2E-D1", "E2E-D2"), ("E2E-D1", "E2E-D3"), ("E2E-D2", "E2E-D3"))
|
|
],
|
|
"full-size-previews": [
|
|
{"direction-id": f"E2E-D{i}", "ref": preview_paths[i], "sha256": _sha(preview_paths[i])}
|
|
for i in (1, 2, 3)
|
|
],
|
|
}
|
|
audit_path = _tmp(audit_content)
|
|
check("FULL-E2E: submit comparative-divergence-audit(all pairs, no collision/blocker)",
|
|
_submit_raw_as_artifact(ws, child, "E2E-AUDIT", "comparative-divergence-audit",
|
|
audit_path).returncode == 0)
|
|
check("FULL-E2E: accept comparative audit exact revision",
|
|
_se("review-artifact", "--workflow", child,
|
|
"--report", _submitted_paths[(child, "E2E-AUDIT")], "--decision", "accepted",
|
|
"--reviewer", "DES-DIRECTOR", ws=ws).returncode == 0)
|
|
r_guard_div = _se("guard", "--workflow", child, "--to", "design-direction-decision", ws=ws)
|
|
check(f"FULL-E2E: guard divergence->decision(diverged+comparative audit pass) -> exit 0 "
|
|
f"(stderr: {r_guard_div.stderr.strip()[:200]!r})", r_guard_div.returncode == 0)
|
|
r = _se("transition", "--workflow", child, "--to", "design-direction-decision", "--actor", "OPS-ORCH", ws=ws)
|
|
check("FULL-E2E: REAL transition divergence->decision -> exit 0", r.returncode == 0)
|
|
|
|
# --- decision: single convergence(no averaging), selected-direction bundle + its own acceptance ---
|
|
sel_content = {
|
|
"direction-set-ref": _submitted_paths[(child, "E2E-DS")],
|
|
"direction-set-sha256": _sha(_submitted_paths[(child, "E2E-DS")]),
|
|
"selected-direction-id": "E2E-D2",
|
|
"rejected-directions": [{"id": "E2E-D1", "reason": "저밀도"}, {"id": "E2E-D3", "reason": "클리셰"}],
|
|
"locked-invariants": ["invariant-a", "invariant-b", "invariant-c"],
|
|
"parent-workflow-id": parent, "product-decision-id": pd_id,
|
|
"direction-input-brief-sha256": ib_sha, "selection-acceptance-receipt": "e2e-selection-receipt",
|
|
}
|
|
sel_path = _tmp(sel_content)
|
|
check("FULL-E2E: record selected-direction -> exit 0",
|
|
_submit_raw_as_artifact(ws, child, "E2E-SD", "selected-direction",
|
|
sel_path).returncode == 0)
|
|
check("FULL-E2E: non-human design approver cannot accept selected-direction",
|
|
_se("review-artifact", "--workflow", child,
|
|
"--report", _submitted_paths[(child, "E2E-SD")], "--decision", "accepted",
|
|
"--reviewer", "EXEC-CPO", ws=ws).returncode != 0)
|
|
check("FULL-E2E: human visual approval accepts selected-direction exact revision",
|
|
_se("review-artifact", "--workflow", child,
|
|
"--report", _submitted_paths[(child, "E2E-SD")], "--decision", "accepted",
|
|
"--reviewer", "HUMAN-001", ws=ws).returncode == 0)
|
|
r_guard_dec = _se("guard", "--workflow", child, "--to", "design-direction-prototype", ws=ws)
|
|
check(f"FULL-E2E: guard decision->prototype(selected-direction-accepted) -> exit 0 "
|
|
f"(stderr: {r_guard_dec.stderr.strip()[:200]!r})", r_guard_dec.returncode == 0)
|
|
r = _se("transition", "--workflow", child, "--to", "design-direction-prototype", "--actor", "OPS-ORCH", ws=ws)
|
|
check("FULL-E2E: REAL transition decision->prototype -> exit 0", r.returncode == 0)
|
|
|
|
# --- prototype: winner-prototype + a REAL preview_ui evidence-ledger receipt(item9: not the
|
|
# gallery/comparison-preview one from divergence — a genuine prototype render) ---
|
|
proto_dir = tempfile.mkdtemp(prefix="dd-e2e-proto-"); _e2e_tmpdirs.append(proto_dir)
|
|
proto_path = os.path.join(proto_dir, "prototype.tsx")
|
|
open(proto_path, "w").write("export const Prototype = () => null;\n")
|
|
proto_sha = _sha(proto_path)
|
|
win_content = {
|
|
"selected-direction-ref": _submitted_paths[(child, "E2E-SD")],
|
|
"selected-direction-sha256": _sha(_submitted_paths[(child, "E2E-SD")]),
|
|
# winner-prototype is a stage-synthesis artifact: bind its source to the
|
|
# exact trusted selected-direction revision instead of relying only on
|
|
# a path-shaped payload field.
|
|
"source-artifact-refs": [{
|
|
"artifact-id": "E2E-SD",
|
|
"artifact-sha256": _sha(_submitted_paths[(child, "E2E-SD")]),
|
|
}],
|
|
"prototype-path": proto_path, "prototype-sha256": proto_sha,
|
|
"preview-receipt-ref": "e2e-preview-receipt", "revision": 1,
|
|
}
|
|
win_path = _tmp(win_content)
|
|
check("FULL-E2E: record winner-prototype -> exit 0",
|
|
_submit_raw_as_artifact(ws, child, "E2E-WIN", "winner-prototype",
|
|
win_path).returncode == 0)
|
|
import _workspace as W_E2E # noqa: E402
|
|
_ed_e2e = W_E2E.evidence_dir()
|
|
os.makedirs(_ed_e2e, exist_ok=True)
|
|
with open(os.path.join(_ed_e2e, "ledger.jsonl"), "a", encoding="utf-8") as fh_e2e:
|
|
_preview_e2e = os.path.join(proto_dir, "preview.png")
|
|
with open(_preview_e2e, "wb") as _png_e2e:
|
|
_png_e2e.write(b"\x89PNG\r\n\x1a\n" + b"\x00" * 2000)
|
|
fh_e2e.write(json.dumps({"tool_use_id": "e2e-preview-receipt", "tool_name": "Bash",
|
|
"command": f"python3 .claude/hooks/preview_ui.py {proto_dir} --out {proto_dir}/preview.png "
|
|
"--viewports 360,768,1280 --check-css",
|
|
"exit_code": 0, "workflow_id": child, "session_id": "e2e-session",
|
|
"agent_id": "e2e-agent"}) + "\n")
|
|
r_guard_proto = _se("guard", "--workflow", child, "--to", "design-direction-critique", ws=ws)
|
|
check(f"FULL-E2E: guard prototype->critique(winner-prototype-present, real preview receipt) -> exit 0 "
|
|
f"(stderr: {r_guard_proto.stderr.strip()[:200]!r})", r_guard_proto.returncode == 0)
|
|
r = _se("transition", "--workflow", child, "--to", "design-direction-critique", "--actor", "OPS-ORCH", ws=ws)
|
|
check("FULL-E2E: REAL transition prototype->critique -> exit 0", r.returncode == 0)
|
|
|
|
# --- critique: /design-review panel — 7 lenses, producer-run-id != reviewer-run-id, pass ---
|
|
_e2e_lenses = ["product-fit", "usability", "distinctiveness", "systematizability",
|
|
"visual-craft", "market-memorability", "implementability"]
|
|
_e2e_roles = {"product-fit": "DES-PROD", "usability": "UX-RESEARCHER",
|
|
"distinctiveness": "DES-VISUAL", "visual-craft": "DES-VISUAL",
|
|
"systematizability": "DES-PLATFORM", "market-memorability": "GTM-PMM",
|
|
"implementability": "ENG-FE"}
|
|
_e2e_reviews = []
|
|
_winner_report_sha = _sha(_submitted_paths[(child, "E2E-WIN")])
|
|
for _lens in _e2e_lenses:
|
|
_review_id = f"E2E-LENS-{_lens}"
|
|
_run_id = f"E2E-REVIEWER-{_lens}"
|
|
_review_payload = {
|
|
"direction-cycle-id": "E2E-C1", "target-prototype-id": "E2E-WIN",
|
|
"target-prototype-sha256": _winner_report_sha,
|
|
"reviewer-role-id": _e2e_roles[_lens], "reviewer-run-id": _run_id,
|
|
"lens": _lens, "verdict": "pass", "findings": [],
|
|
}
|
|
_review_report = _write_contract_artifact(
|
|
ws, child, _review_id, "design-lens-review", _e2e_roles[_lens],
|
|
_review_payload, "design-direction-critique")
|
|
check(f"FULL-E2E: submit trusted lens review {_lens}",
|
|
_se("submit-artifact", "--workflow", child, "--report", _review_report,
|
|
"--actor", "OPS-ORCH", ws=ws).returncode == 0)
|
|
_e2e_reviews.append({
|
|
"report-id": _review_id, "report-ref": _review_report,
|
|
"report-sha256": _sha(_review_report), "lens": _lens,
|
|
"reviewer-role-id": _e2e_roles[_lens], "reviewer-run-id": _run_id,
|
|
"verdict": "pass",
|
|
})
|
|
panel_content = {
|
|
"direction-cycle-id": "E2E-C1",
|
|
"target-prototype-id": "E2E-WIN", "target-prototype-sha256": _winner_report_sha,
|
|
"reviews": _e2e_reviews,
|
|
"synthesis": {"verdict": "pass", "role-id": "DES-DIRECTOR", "unresolved-dissent": []},
|
|
}
|
|
panel_path = _tmp(panel_content)
|
|
check("FULL-E2E: record design-review-panel(7 lenses, pass, producer<>reviewer disjoint) -> exit 0",
|
|
_submit_raw_as_artifact(ws, child, "E2E-PANEL", "design-review-panel",
|
|
panel_path).returncode == 0)
|
|
r_guard_crit = _se("guard", "--workflow", child, "--to", "design-direction-finalize", ws=ws)
|
|
check(f"FULL-E2E: guard critique->finalize(direction-critique-passed) -> exit 0 "
|
|
f"(stderr: {r_guard_crit.stderr.strip()[:200]!r})", r_guard_crit.returncode == 0)
|
|
r = _se("transition", "--workflow", child, "--to", "design-direction-finalize", "--actor", "OPS-ORCH", ws=ws)
|
|
check("FULL-E2E: REAL transition critique->finalize -> exit 0", r.returncode == 0)
|
|
|
|
# --- finalize: approved-direction report + acceptance + parent-side registration ---
|
|
approved_payload = {
|
|
"parent-workflow-id": parent, "child-workflow-id": child, "product-decision-id": pd_id,
|
|
"direction-input-brief-sha256": ib_sha,
|
|
"selected-direction-ref": _submitted_paths[(child, "E2E-SD")],
|
|
"selected-direction-sha256": _sha(_submitted_paths[(child, "E2E-SD")]),
|
|
"winner-prototype-ref": _submitted_paths[(child, "E2E-WIN")],
|
|
"winner-prototype-sha256": _sha(_submitted_paths[(child, "E2E-WIN")]),
|
|
}
|
|
rid = f"E2E-APPROVED-{child}"
|
|
report_path = _write_contract_artifact(
|
|
ws, child, rid, "approved-direction", "DES-DIRECTOR", approved_payload,
|
|
"design-direction-finalize")
|
|
check("FULL-E2E: submit approved-direction through trusted artifact API",
|
|
_se("submit-artifact", "--workflow", child, "--report", report_path,
|
|
"--actor", "DES-DIRECTOR", ws=ws).returncode == 0)
|
|
report_sha = _sha(report_path)
|
|
check("FULL-E2E: accept approved-direction report(exact workflow+report-sha256 binding)",
|
|
_se("review-artifact", "--workflow", child, "--report", report_path,
|
|
"--decision", "accepted", "--reviewer", "EXEC-CPO", ws=ws).returncode == 0)
|
|
r_reg = _se("register-direction-approval", "--parent-workflow", parent, "--child-workflow", child,
|
|
"--report", report_path, "--report-sha256", report_sha, ws=ws)
|
|
check(f"FULL-E2E: register-direction-approval -> exit 0 (stderr: {r_reg.stderr.strip()[:200]!r})",
|
|
r_reg.returncode == 0)
|
|
check("FULL-E2E: child still at design-direction-finalize just before the terminal transition "
|
|
"(not force-written to approved)",
|
|
SE._load_ledger_safe(child).get("stage") == "design-direction-finalize")
|
|
r_guard_fin = _se("guard", "--workflow", child, "--to", "design-direction-approved", ws=ws)
|
|
check(f"FULL-E2E: guard finalize->approved(approved-direction-valid+approval-receipt-bound+"
|
|
f"parent-approval-link-recorded) -> exit 0 (stderr: {r_guard_fin.stderr.strip()[:200]!r})",
|
|
r_guard_fin.returncode == 0)
|
|
r_final = _se("transition", "--workflow", child, "--to", "design-direction-approved", "--actor", "OPS-ORCH", ws=ws)
|
|
check(f"FULL-E2E: REAL terminal transition finalize->approved -> exit 0 "
|
|
f"(stderr: {r_final.stderr.strip()[:200]!r})", r_final.returncode == 0)
|
|
check("FULL-E2E: child reaches design-direction-approved via REAL transitions end-to-end",
|
|
SE._load_ledger_safe(child).get("stage") == "design-direction-approved")
|
|
|
|
# --- assert the PARENT cascade sees the approval ---
|
|
r_chk = _se("check-direction-approved", "--workflow", parent, ws=ws)
|
|
check(f"FULL-E2E: parent check-direction-approved -> YES (stdout: {r_chk.stdout.strip()!r})",
|
|
r_chk.returncode == 0 and "YES" in r_chk.stdout)
|
|
|
|
# --- optional: a UI-bearing standard-tier parent can now pass the design->spec gate ---
|
|
parent_led = SE._load_ledger_safe(parent)
|
|
parent_led["build-families"] = ["FAM-ENG-FRONTEND"]
|
|
parent_led["tier"] = "standard"
|
|
SE._write_ledger(parent, parent_led)
|
|
_gate_pred = SE._PREDICATES["design-direction-gate-satisfied"]
|
|
_gate_facts = SE._facts(parent, SE._load_ledger_safe(parent), None)
|
|
check("FULL-E2E: UI-bearing+standard parent's design-direction-gate-satisfied -> True "
|
|
"now that the child is approved(design->spec no longer blocked)",
|
|
_gate_pred(_gate_facts)[0] is True)
|
|
|
|
for _d in _e2e_tmpdirs:
|
|
shutil.rmtree(_d, ignore_errors=True)
|
|
|
|
print(f"\n{passed} passed, {failed} failed"); sys.exit(1 if failed else 0)
|