#!/usr/bin/env python3
"""Trusted workflow state engine.

``workflow-contracts.yaml`` owns artifact vocabulary and stage bundles. Runtime truth is append-only:
``workflow-events.jsonl``, ``artifact-events.jsonl``, ``acceptance-events.jsonl`` and
``human-signoff.jsonl``. ``workflow.yaml`` is a disposable materialized view; caller-supplied gate
facts, artifact kinds, option counts and evidence grades are never trusted.

The normal lifecycle is ``init-workflow -> complete-stage -> enter-stage -> ...``. Reports use the
``workflow-artifact`` envelope and are registered/reviewed as exact immutable id+sha256 snapshots.
``transition`` remains only as a guarded compatibility advance (complete + enter).
"""
import hashlib
import json
import os
import re
import shlex
import sys
import uuid
from contextlib import contextmanager
from datetime import datetime, timezone

try:
    from orgos.planning.role_selector import resolve_family as _planned_resolve_family
except Exception:
    _planned_resolve_family = None

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
ROOT = os.environ.get("CLAUDE_PROJECT_DIR") or os.path.dirname(os.path.dirname(HERE))

from orgos.state import event_store as _event_store  # noqa: E402
from orgos.state import materializer as _materializer  # noqa: E402
from orgos.state import transition_engine as _transition_engine  # noqa: E402

try:
    import yaml
except Exception:  # pragma: no cover - yaml 은 저장소 전반에서 사용됨
    yaml = None

RULES_PATH = os.path.join(ROOT, "org-os", "00-role-registry", "state-transition-rules.yaml")
PLANS_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "execution-plans.yaml")
CMAP_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "collaboration-map.yaml")
TIERS_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "governance-tiers.yaml")
CONTRACTS_PATH = os.path.join(ROOT, "org-os", "06-agent-work", "workflow-contracts.yaml")
ROLES_PATH = os.path.join(ROOT, "org-os", "00-role-registry", "roles.yaml")
FAMILIES_PATH = os.path.join(ROOT, "org-os", "00-role-registry", "capability-families.yaml")

INITIAL_STAGE = "intake"
DEFAULT_PLAN = "cascade"
DEFAULT_TIER = "standard"
DEFAULT_MODE = "converge"
VALID_TIERS = ("light", "standard", "heavy")
VALID_MODES = ("converge", "divergent")
EVIDENCE_CONTRACT_VERSION = 2


def _log(msg):
    try:
        sys.stderr.write(f"[state_engine] {msg}\n")
    except Exception:
        pass


def _now():
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _stamp():
    return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")


# ---------------------------------------------------------------- SSOT loaders

def _load_yaml(path):
    if not yaml or not os.path.exists(path):
        return {}
    try:
        with open(path, encoding="utf-8") as fh:
            data = yaml.safe_load(fh)
        return data if isinstance(data, dict) else {}
    except Exception as e:
        _log(f"YAML 로드 실패({os.path.basename(path)}): {e}")
        return {}


def load_rules():
    return _load_yaml(RULES_PATH).get("state-transition-rules", {}) or {}


def load_plans():
    return _load_yaml(PLANS_PATH).get("execution-plans", {}) or {}


def load_cmap():
    return _load_yaml(CMAP_PATH).get("collaboration-map", {}) or {}


def load_tiers():
    return _load_yaml(TIERS_PATH).get("governance-tiers", {}) or {}


def load_contracts():
    return _load_yaml(CONTRACTS_PATH).get("workflow-contracts", {}) or {}


def _role_registry():
    """Concrete role registry. Placeholder strings and invented roles never enter it."""
    roles = {}
    try:
        registry = _load_yaml(ROLES_PATH).get("role-registry", {}) or {}
        for role in registry.get("roles", []) or []:
            if isinstance(role, dict) and role.get("role-id"):
                roles[str(role["role-id"])] = role
    except Exception:
        pass
    roles["HUMAN-001"] = {
        "role-id": "HUMAN-001", "role-type": "human", "is-decision-maker": True,
    }
    return roles


def _role_has_capability(role_id, capability):
    allowed = (load_contracts().get("role-capabilities", {}) or {}).get(capability, []) or []
    return str(role_id or "").upper() in {str(value).upper() for value in allowed}


def resolve_family(family_id, signals=None):
    """Resolve family metadata to concrete workers.

    Family is metadata, never an actor.  The executable planner treats its members
    as a candidate pool and returns the minimum sufficient concrete set.  The local
    implementation below is only a compatibility fallback for incomplete installs.
    """
    if _planned_resolve_family is not None:
        try:
            planned = _planned_resolve_family(family_id, signals=signals, tier="standard")
            if planned:
                return planned
        except Exception as exc:
            _log(f"role planner fallback for {family_id}: {exc}")
    signal_set = {str(value).strip().lower() for value in (signals or []) if str(value).strip()}
    families = (_load_yaml(FAMILIES_PATH).get("capability-families", {}) or {}).get("families", []) or []
    for family in families:
        if isinstance(family, dict) and family.get("family-id") == family_id:
            members = [role for role in family.get("member-role-ids", []) or []
                       if role in _role_registry()]
            collaboration = family.get("collaboration-default")
            if collaboration == "collapse":
                primary = family.get("primary-role-id")
                reason = "default-primary"
                for route in family.get("collapse-routes", []) or []:
                    route_signals = {str(value).strip().lower()
                                     for value in route.get("when-any", []) or []}
                    if signal_set.intersection(route_signals):
                        primary = route.get("role-id")
                        reason = "matched:" + ",".join(sorted(signal_set.intersection(route_signals)))
                        break
                if primary not in members:
                    return None
                return {
                    "requested-family": family_id,
                    "resolved-workers": [primary],
                    "primary-worker": primary,
                    "available-workers": members,
                    "routing-reason": reason,
                    "collaboration-default": collaboration,
                }
            primary = members[0] if members else None
            if not primary:
                return None
            return {
                "requested-family": family_id,
                "resolved-workers": [primary],
                "primary-worker": primary,
                "available-workers": members,
                "routing-reason": "compatibility-minimum-primary",
                "collaboration-default": collaboration,
            }
    return None


def _ws_transitions():
    """Derive runtime transitions from workflow-contracts.yaml (single contract SSOT)."""
    try:
        contract = load_contracts()
        transitions = []
        for workflow in (contract.get("workflows", {}) or {}).values():
            if not isinstance(workflow, dict):
                continue
            stages = workflow.get("stages", {}) or {}
            for stage, definition in stages.items():
                if not isinstance(definition, dict) or not definition.get("next"):
                    continue
                destinations = definition.get("next")
                if not isinstance(destinations, list):
                    destinations = [destinations]
                for destination in destinations:
                    transitions.append({
                        "from": stage,
                        "to": destination,
                        "allowed-by": definition.get("actor") or {"executor": ["OPS-ORCH"]},
                        "required-conditions": list(definition.get("exit-gate") or []),
                    })
            transitions.extend(workflow.get("additional-transitions") or [])
        transitions.extend(contract.get("side-transitions") or [])
        return transitions
    except Exception:
        return []


def _plan_stages(plan):
    try:
        contract_plan = (load_contracts().get("workflows", {}) or {}).get(plan, {}) or {}
        if isinstance(contract_plan.get("stages"), dict):
            return list(contract_plan["stages"].keys())
        p = (load_plans().get("plans", {}) or {}).get(plan, {})
        return list(p.get("stages", []) or [])
    except Exception:
        return []


def _initial_stage(plan):
    """plan 의 첫 stage. namespaced plan(design-direction)은 자기 첫 stage 에서 시작."""
    stages = _plan_stages(plan)
    return stages[0] if stages else INITIAL_STAGE


def _plan_terminal(plan):
    """plan 의 종단 stage(execution-plans terminal-stage). 없으면 None."""
    try:
        contract_plan = (load_contracts().get("workflows", {}) or {}).get(plan, {}) or {}
        if contract_plan.get("terminal-stage"):
            return contract_plan.get("terminal-stage")
        p = (load_plans().get("plans", {}) or {}).get(plan, {})
        return p.get("terminal-stage")
    except Exception:
        return None


def _must_read_map():
    """{build-family: [must-read-design, ...]} — collaboration-map design-to-build-contract."""
    out = {}
    try:
        for m in load_cmap().get("design-to-build-contract", {}).get("mappings", []) or []:
            fam = m.get("build-family")
            if fam:
                out[fam] = list(m.get("must-read-designs", []) or [])
    except Exception:
        pass
    return out


_EGRADE = {"E0": 0, "E1": 1, "E2": 2, "E3": 3, "E4": 4, "E5": 5}


def _tier_evidence_min(tier):
    """tier 최소 증거등급('E2'..). governance-tiers 우선, state-transition-rules 폴백."""
    try:
        t = (load_tiers().get("tiers", {}) or {}).get(tier, {})
        v = (t.get("converge", {}) or {}).get("evidence-grade-min")
        if v:
            return v
    except Exception:
        pass
    try:
        tm = load_rules().get("tier-modifiers", {}) or {}
        return (tm.get(tier, {}) or {}).get("evidence-grade-min")
    except Exception:
        return None


def _tier_human_gate_required(tier):
    """acceptance→released 에서 사람 승인이 필수인 tier 인가(heavy)."""
    return str(tier) == "heavy"


def _derived_risk_tier(profile):
    """Apply governance-tiers risk/reversibility/blast rules deterministically."""
    risk = profile.get("risk") if isinstance(profile, dict) else {}
    order = {"light": 0, "standard": 1, "heavy": 2}
    base = {"Low": 0, "Med": 1, "High": 2, "Critical": 2}.get(risk.get("risk-level"), 1)
    if risk.get("reversibility") == "one-way-door":
        base += 1
    if risk.get("blast-radius") == "cross-team":
        base += 1
    if risk.get("blast-radius") == "production-customer-revenue":
        base = 2
    if any(risk.get(key) for key in (
            "security-bearing", "data-migration", "external-side-effect",
            "privacy", "regulatory", "slo-impact", "pii", "data-residency")):
        base = max(base, order["standard"])
    return VALID_TIERS[min(base, 2)]


# ---------------------------------------------------------------- ledger I/O

def _state_dir(create=False):
    try:
        import _workspace as W  # noqa: E402
        sd = W.state_dir()
    except Exception as e:  # WorkspaceNotSetError 포함
        _log(f"workspace 미해석 — 원장 접근 스킵: {e}")
        return None
    if create:
        try:
            os.makedirs(sd, exist_ok=True)
        except Exception as e:
            _log(f"state_dir 생성 실패: {e}")
            return None
    return sd


def _wf_dir(wf, create=False):
    sd = _state_dir(create=create)
    if not sd:
        return None
    d = os.path.join(sd, wf)
    if create:
        try:
            os.makedirs(d, exist_ok=True)
        except Exception as e:
            _log(f"wf-dir 생성 실패: {e}")
            return None
    return d


def _ledger_path(wf, create=False):
    d = _wf_dir(wf, create=create)
    return os.path.join(d, "workflow.yaml") if d else None


def _events_path(wf, create=False):
    d = _wf_dir(wf, create=create)
    return os.path.join(d, "state-events.jsonl") if d else None


def _workflow_events_path(wf, create=False):
    d = _wf_dir(wf, create=create)
    return os.path.join(d, "workflow-events.jsonl") if d else None


def _artifact_events_path(create=False):
    sd = _state_dir(create=create)
    return os.path.join(sd, "artifact-events.jsonl") if sd else None


def _read_jsonl(path):
    return _event_store.read_jsonl(path, on_error=_log)


@contextmanager
def _workflow_lock(wf):
    """Serialize event append + projection writes for a workflow."""
    d = _wf_dir(wf, create=True)
    if not d:
        raise OSError("workspace 미설정")
    lock_path = os.path.join(d, ".workflow.lock")
    with open(lock_path, "a+", encoding="utf-8") as lock:
        try:
            import fcntl
            fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        except Exception:
            pass
        try:
            yield
        finally:
            try:
                import fcntl
                fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
            except Exception:
                pass


def _default_ledger(wf, plan=DEFAULT_PLAN, tier=DEFAULT_TIER, mode=DEFAULT_MODE):
    return {
        "workflow-id": wf,
        "stage": _initial_stage(plan),
        "stage-status": "running",
        "last-completed-stage": None,
        "completed-for-next-stage": None,
        "plan": plan,
        "tier": tier,
        "mode": mode,
        "evidence-contract-version": EVIDENCE_CONTRACT_VERSION,
        "artifacts": [],
        "progress": {},
        "created-at": _now(),
        "last-updated-at": _now(),
    }


def read_ledger(wf):
    """원장 dict 또는 None(부재/파싱불가). 예외를 던지지 않는다."""
    p = _ledger_path(wf, create=False)
    if not p or not os.path.exists(p):
        return None
    try:
        with open(p, encoding="utf-8") as fh:
            data = yaml.safe_load(fh) if yaml else None
        return _project_ledger(wf, data) if isinstance(data, dict) else None
    except Exception as e:
        _log(f"원장 읽기 실패({wf}): {e}")
        return None


def _load_ledger_safe(wf):
    """원장이 없으면 기본 원장(stage=intake)을 메모리로 반환(파일 생성 안 함)."""
    return read_ledger(wf) or _default_ledger(wf)


def _write_ledger(wf, data):
    p = _ledger_path(wf, create=True)
    if not p:
        return False
    try:
        # atomic write(temp + os.replace): 병렬 agent 가 원장을 동시에 읽을 때 torn/partial
        # read 를 막는다(재리뷰 지적). POSIX 에서 os.replace 는 원자적 rename.
        tmp = f"{p}.tmp.{os.getpid()}"
        with open(tmp, "w", encoding="utf-8") as fh:
            yaml.safe_dump(data, fh, allow_unicode=True, sort_keys=False)
        os.replace(tmp, p)
        return True
    except Exception as e:
        _log(f"원장 기록 실패({wf}): {e}")
        try:
            if os.path.exists(tmp):
                os.remove(tmp)
        except Exception:
            pass
        return False


def init_ledger(wf, plan=DEFAULT_PLAN, tier=DEFAULT_TIER, mode=DEFAULT_MODE, overwrite=False,
                parent_workflow=None, product_decision=None, direction_input_brief=None):
    """워크플로 원장을 생성(stage=intake). 이미 있으면 overwrite=False 시 그대로 반환.

    plan == "design-direction" 인 경우에만: 순환 정합을 위해 부모 workflow(product-decision) 로
    반드시 바인딩해야 한다 — parent_workflow/product_decision/direction_input_brief 3종 모두 필수,
    부모 원장 **실존**을 파일 존재로 직접 검증(주의: `_load_ledger_safe` 는 원장이 없어도 기본 원장을
    메모리로 반환하므로, 그 반환값의 진위 여부만으로는 부재를 구분할 수 없다 — 그래서 여기선 그 함수를
    쓰지 않고 `_ledger_path` + `os.path.exists` 로 먼저 실존을 확인한 다음에만 로드한다), 그리고
    부모에 **정확히 이 product_decision** 이 accepted 된 적이 있는지 `_al_accepted_ids(parent)` 로
    검증한다(부모가 다른 무언가를 accepted 했다는 사실만으로는 통과하지 않으며, 원장 문자열 부분일치
    fallback 도 없다 — 위조 id·substring 우회 모두 차단, 재리뷰 Critical fix). direction-input-brief
    는 부모가 S1 에서 이미 freeze 했어야 하므로 그 경로가 실존하지 않으면 즉시 거부한다. 다른 plan 은
    기존 동작 그대로."""
    if tier not in VALID_TIERS:
        raise ValueError(f"tier는 {VALID_TIERS} 중 하나여야 한다(got {tier!r})")
    if mode not in VALID_MODES:
        raise ValueError(f"mode는 {VALID_MODES} 중 하나여야 한다(got {mode!r})")
    if plan not in (load_contracts().get("workflows", {}) or {}):
        raise ValueError(f"등록되지 않은 workflow plan: {plan}")
    existing = read_ledger(wf)
    if existing and not overwrite:
        return existing
    led = _default_ledger(wf, plan, tier, mode)
    if plan in ("design-direction", "experience-foundation"):
        required = "--parent-workflow --product-decision"
        if plan == "design-direction":
            required += " --direction-input-brief"
        if not (parent_workflow and product_decision) or (
                plan == "design-direction" and not direction_input_brief):
            raise ValueError(
                f"{plan}: {required} 필수"
            )
        parent_path = _ledger_path(parent_workflow, create=False)
        if not parent_path or not os.path.exists(parent_path):
            raise ValueError(f"{plan}: 부모 workflow '{parent_workflow}' 원장 없음")
        # finding(Critical, re-review): _al_has_accepted(parent) 는 부모가 "아무 report 나" accepted
        # 했으면 True 였다(이 product_decision 자체를 검증하지 않음) — 위조 id 로 편승 가능했다.
        # 그리고 substring fallback(`product_decision not in str(parent_led)`)은 원장의 고정 스키마
        # 키("workflow-id" 등) 때문에 사실상 항상 통과했다(예: "workflow" 는 어떤 부모에도 매치).
        # 이제 부모의 **실제 accepted report-id 집합**(_al_accepted_ids)에 이 product_decision 이
        # 정확히 있는지만으로 검증한다 — substring fallback 은 완전히 제거.
        parent_arts = (_load_ledger_safe(parent_workflow).get("artifacts") or [])
        if product_decision not in _al_accepted_ids(parent_workflow, parent_arts):
            raise ValueError(
                f"{plan}: 부모 '{parent_workflow}'에 accepted product-decision '{product_decision}' 없음"
            )
        led["parent-workflow-id"] = parent_workflow
        led["product-decision-id"] = product_decision
        if (plan == "experience-foundation" and tier == "light"
                and _experience_foundation_required(parent_workflow)):
            raise ValueError(
                "experience-foundation: 공개 웹/신규 제품/대규모 리디자인은 tier standard 이상 필수"
            )
    if plan == "design-direction":
        parent_led = _load_ledger_safe(parent_workflow)
        if _experience_foundation_required(parent_workflow) and not _has_experience_foundation(
                parent_workflow, parent_led):
            raise ValueError(
                "design-direction: 공개 웹/신규 제품/대규모 리디자인은 approved experience-foundation 선행 필수"
            )
        led["direction-input-brief-ref"] = direction_input_brief
        # finding(Minor): direction-input-brief 는 S1 에서 부모가 이미 freeze 한 것이어야 한다 —
        # 자식 init 시점에 파일이 없으면 sha 없이 조용히 바인딩하지 말고 즉시 거부한다.
        _brief_path = (
            direction_input_brief if os.path.isabs(direction_input_brief)
            else os.path.join(ROOT, direction_input_brief)
        )
        if not os.path.exists(_brief_path):
            raise ValueError(f"design-direction: direction-input-brief 파일 없음: {direction_input_brief}")
        if (_experience_foundation_required(parent_workflow)
                and not _direction_brief_foundation_refs_ok(parent_workflow, direction_input_brief)):
            raise ValueError(
                "design-direction: direction-input-brief의 benchmark/blueprint/wireframe exact ref+SHA가 "
                "부모 approved experience-foundation과 불일치"
            )
        brief_sha = _current_input_brief_sha(led)
        if brief_sha:
            led["direction-input-brief-sha256"] = brief_sha
    if not _write_ledger(wf, led):
        raise OSError(f"workflow projection 생성 실패: {wf}")
    init_event = {
        "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
        "event-type": "workflow-initialized", "workflow-id": wf,
        "stage": led.get("stage"), "plan": plan, "tier": tier, "mode": mode,
        "evidence-contract-version": EVIDENCE_CONTRACT_VERSION,
        "actor": "OPS-ORCH", "effective-at": _now(),
    }
    for key in ("parent-workflow-id", "product-decision-id", "direction-input-brief-ref",
                "direction-input-brief-sha256"):
        if led.get(key) is not None:
            init_event[key] = led.get(key)
    committed, error = _atomic_event_transaction(wf, workflow_event=init_event)
    if not committed:
        raise OSError(error or f"workflow init event 기록 실패: {wf}")
    return read_ledger(wf) or led


def find_child_direction_workflow(parent_workflow, product_decision, brief_sha):
    """(parent_workflow, product_decision) 로 바인딩된 design-direction 자식 workflow 를 찾는다(dedup —
    같은 product-decision 에 대해 새 design-direction 사이클을 중복으로 열지 않기 위함). 없으면 None.
    찾으면 {"workflow-id", "stage", "stale"} — stale 은 자식이 바인딩 당시 읽은 input-brief 해시가
    호출측이 지금 들고 있는 brief_sha 와 다른지(즉 그 사이 input-brief 가 바뀌었는지)를 알려준다."""
    base = _state_dir(create=False)
    if not base or not os.path.isdir(base):
        return None
    for wf in os.listdir(base):
        led = _load_ledger_safe(wf)
        if led.get("plan") != "design-direction":
            continue
        if led.get("parent-workflow-id") == parent_workflow and led.get("product-decision-id") == product_decision:
            return {
                "workflow-id": wf,
                "stage": led.get("stage"),
                "stale": led.get("direction-input-brief-sha256") != brief_sha,
            }
    return None


def find_child_experience_workflow(parent_workflow, product_decision):
    """Return the single experience-foundation child bound to a parent decision."""
    base = _state_dir(create=False)
    if not base or not os.path.isdir(base):
        return None
    for wf in os.listdir(base):
        led = _load_ledger_safe(wf)
        if (led.get("plan") == "experience-foundation"
                and led.get("parent-workflow-id") == parent_workflow
                and led.get("product-decision-id") == product_decision):
            return {"workflow-id": wf, "stage": led.get("stage")}
    return None


def init_workflow(wf, **kwargs):
    """Canonical API name for workflow initialization."""
    return init_ledger(wf, **kwargs)


def _append_state_event(wf, event):
    """Append a canonical workflow event. workflow.yaml is only its projection."""
    p = _workflow_events_path(wf, create=True)
    return _event_store.append_jsonl(p, event, on_error=_log)


def _append_artifact_event(event):
    p = _artifact_events_path(create=True)
    # artifact-events.jsonl is global, so it needs its own file lock.
    return _event_store.append_jsonl(p, event, file_lock=True, on_error=_log)


def read_workflow_events(wf):
    return _read_jsonl(_workflow_events_path(wf, create=False))


def read_artifact_events(wf=None):
    events = _read_jsonl(_artifact_events_path(create=False))
    if wf is None:
        return events
    return [event for event in events if event.get("workflow-id") == wf]


def _stage_epoch(wf, stage=None):
    """Return the append-only event that opened the current/requested stage."""
    current = None
    for event in read_workflow_events(wf):
        if event.get("event-type") == "workflow-initialized":
            current = {
                "event-id": event.get("workflow-event-id"),
                "stage": event.get("stage") or _initial_stage(event.get("plan", DEFAULT_PLAN)),
                "effective-at": event.get("effective-at"),
            }
        elif event.get("event-type") == "state-transition":
            current = {
                "event-id": event.get("state-event-id") or event.get("workflow-event-id"),
                "stage": event.get("to"), "effective-at": event.get("effective-at"),
            }
    if current and (stage is None or current.get("stage") == stage):
        return current
    return None


def _latest_artifact_of_kind(artifacts, kind):
    return next((artifact for artifact in reversed(artifacts or [])
                 if isinstance(artifact, dict) and artifact.get("artifact-kind") == kind), None)


def _event_artifact_valid(event):
    if event.get("event-type") != "artifact-submitted":
        return False
    path = event.get("path")
    try:
        import artifact_contract as AC
        ap = AC.absolute_path(path)
        return bool(ap and os.path.isfile(ap)
                    and AC.sha256_file(ap) == event.get("artifact-sha256"))
    except Exception:
        return False


def _trusted_artifacts(wf):
    """Active immutable artifact revisions reconstructed from canonical events."""
    active = {}
    order = []
    for event in read_artifact_events(wf):
        if event.get("event-type") != "artifact-submitted":
            continue
        artifact_id = event.get("artifact-id")
        if not artifact_id:
            continue
        if artifact_id not in active:
            order.append(artifact_id)
        active[artifact_id] = dict(event)
    return [active[artifact_id] for artifact_id in order
            if _event_artifact_valid(active[artifact_id])]


def _project_ledger(wf, ledger):
    """Rebuild protected materialized fields from append-only events on every read."""
    return _materializer.project_workflow(
        ledger,
        trusted_artifacts=_trusted_artifacts(wf),
        workflow_events=read_workflow_events(wf),
        initial_stage=_initial_stage,
        quality_panel_unmet=_quality_panel_unmet,
        default_plan=DEFAULT_PLAN,
    )


# ---------------------------------------------------------------- human signoff (P0-4)
def _human_signoff_path(wf, create=False):
    d = _wf_dir(wf, create=create)
    return os.path.join(d, "human-signoff.jsonl") if d else None


def _has_human_signoff(wf, stage):
    """guard 보호 signoff 파일에 이 stage(또는 '*')에 대한 사람(HUMAN-*) 승인 항목이 있으면 True.

    finding P0-4: human-gate 를 원장 자기신고 플래그가 아니라 이 파일로 파생한다. guard_tools 가
    에이전트의 이 파일 쓰기와 `state_engine.py signoff` 호출을 모두 막으므로, 위조하려면 guard 를
    우회해야 한다(문서화된 soft-boundary — 하네스는 사람을 인증할 수 없다)."""
    p = _human_signoff_path(wf, create=False)
    if not p or not os.path.exists(p):
        return False
    try:
        with open(p, encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    ev = json.loads(line)
                except Exception:
                    continue
                if not isinstance(ev, dict):
                    continue
                by = str(ev.get("by") or "")
                st = str(ev.get("stage") or "")
                role = _role_registry().get(by)
                if (role and role.get("role-type") == "human"
                        and ev.get("workflow-id") == wf and (st == str(stage) or st == "*")):
                    return True
    except Exception:
        return False
    return False


def _human_role(actor_id):
    role = _role_registry().get(str(actor_id or "")) or {}
    return role.get("role-type") == "human"


def _has_exact_human_approval(wf, stage):
    """Return whether the gate's canonical decision artifact was accepted by a human.

    A decision-packet acceptance and a release decision are already typed, immutable,
    id+sha-bound human decisions. Requiring a second stage signoff for the same decision
    creates duplicate authority and contradictory audit histories.
    """
    trusted = _trusted_artifacts(wf)
    if stage == "decide":
        packet = _latest_artifact_of_kind(trusted, "executive-decision-packet")
        if not packet:
            return False
        try:
            import acceptance_log as AL
            for event in reversed(AL.read_events()):
                if (event.get("workflow-id") != wf
                        or event.get("report-id") != packet.get("artifact-id")
                        or event.get("artifact-sha256") != packet.get("artifact-sha256")):
                    continue
                reviewer = event.get("reviewer") or {}
                reviewer_id = reviewer.get("actor-id") or reviewer.get("role-id") or event.get("role-id")
                return event.get("decision") == "accepted" and _human_role(reviewer_id)
        except Exception:
            return False
        return False
    if stage == "acceptance":
        trusted_revisions = {
            (item.get("artifact-id"), item.get("artifact-sha256"))
            for item in trusted if item.get("artifact-kind") == "release-decision"
        }
        for event in reversed(read_workflow_events(wf)):
            if event.get("event-type") != "release-decision-recorded":
                continue
            revision = (event.get("decision-artifact-id"), event.get("decision-artifact-sha256"))
            if revision not in trusted_revisions:
                continue
            return event.get("status") == "Approved" and _human_role(event.get("actor"))
    return False


def _human_gate_satisfied(wf, stage):
    """Canonical exact approval first; legacy stage signoff remains a narrow fallback."""
    return _has_exact_human_approval(wf, stage) or _has_human_signoff(wf, stage)


def record_signoff(wf, stage, by):
    """사람 승인(human-signoff)을 append. **사람이 세션 밖에서** 호출해야 한다 — guard_tools 가
    에이전트의 이 CLI 호출을 차단한다(P0-4 soft-boundary). by 는 HUMAN-* 여야 유효."""
    human = _role_registry().get(str(by or ""))
    if not human or human.get("role-type") != "human":
        return False, "signoff --by 는 등록된 human role이어야 한다."
    if read_ledger(wf) is None:
        return False, f"workflow 원장 없음: {wf}"
    p = _human_signoff_path(wf, create=True)
    if not p:
        return False, "workspace 미설정 — signoff 기록 불가."
    try:
        with _workflow_lock(wf), open(p, "a", encoding="utf-8") as fh:
            fh.write(json.dumps({
                "human-signoff-event-id": f"hse-{_stamp()}-{uuid.uuid4().hex[:8]}",
                "workflow-id": wf, "stage": stage, "by": by, "at": _now(),
            }, ensure_ascii=False) + "\n")
            fh.flush()
            os.fsync(fh.fileno())
        return True, None
    except Exception as e:
        return False, f"signoff 기록 실패: {e}"


def record_human_signoff(wf, stage, by):
    """Canonical API name; authentication is enforced by the human-only CLI boundary."""
    return record_signoff(wf, stage, by)


def _prepare_submit_event(wf, report_path, actor):
    ledger = read_ledger(wf)
    if ledger is None:
        return False, f"workflow 원장 없음: {wf} (init-workflow 먼저 실행)"
    roles = _role_registry()
    actor_id = str(actor or "").strip()
    if actor_id not in roles:
        return False, f"등록되지 않은 submit actor: {actor_id}"
    try:
        import artifact_contract as AC
        artifact = AC.validate_snapshot(report_path, expected_workflow=wf)
    except Exception as exc:
        return False, str(exc)
    producer = artifact.get("producer-role-id")
    if producer not in roles:
        return False, f"등록되지 않은 producer-role-id: {producer}"
    if not AC.producer_allowed(artifact.get("artifact-kind"), producer):
        return False, f"producer '{producer}'는 artifact-kind={artifact.get('artifact-kind')} 생산 권한이 없다"
    if actor_id != producer and not _role_has_capability(actor_id, "transition-executor"):
        return False, f"submit actor '{actor_id}'는 producer '{producer}'도 transition-executor도 아니다"
    try:
        import method_contracts as MC
        _ap, raw_report = AC.load_report(report_path)
        raw_report = dict(raw_report)
        raw_report.setdefault("role-id", producer)
        raw_report.setdefault("workflow-id", wf)
        declared_tier = raw_report.get("tier")
        if declared_tier is not None and declared_tier != ledger.get("tier"):
            return False, (f"report tier가 canonical workflow tier와 불일치"
                           f"(report={declared_tier}, workflow={ledger.get('tier')})")
        trusted = _trusted_artifacts(wf)

        def _resolve_method_ref(ref):
            return next((item for item in trusted
                         if item.get("artifact-id") == ref.get("report-id")
                         and item.get("artifact-sha256") == ref.get("sha256")), None)

        method_errors = MC.validate_method_execution(
            raw_report, enforced_tier=ledger.get("tier"), artifact_resolver=_resolve_method_ref,
            current_artifact=artifact)
        if method_errors:
            return False, "method execution 계약 위반:\n  - " + "\n  - ".join(method_errors)
    except Exception as exc:
        if ledger.get("tier") in ("standard", "heavy"):
            return False, f"method execution policy 평가 실패(fail-closed): {exc}"
    body = artifact.get("payload") or {}
    artifact_definition = (AC.load_contract().get("artifact-kinds", {}) or {}).get(
        artifact.get("artifact-kind"), {}) or {}
    method_binding = artifact_definition.get("method-binding") or {}
    if method_binding.get("mode") == "stage-synthesis":
        source_refs = body.get("source-artifact-refs") or []
        seen_source_refs = set()
        for index, ref in enumerate(source_refs):
            if not isinstance(ref, dict):
                return False, f"stage-synthesis source-artifact-refs[{index}] object 필요"
            key = (str(ref.get("artifact-id") or ""), str(ref.get("artifact-sha256") or ""))
            if not key[0] or not re.fullmatch(r"[0-9a-f]{64}", key[1]):
                return False, (f"stage-synthesis source-artifact-refs[{index}]는 "
                               "artifact-id + 64-hex artifact-sha256 필수")
            if key in seen_source_refs:
                return False, "stage-synthesis source-artifact-refs 중복"
            seen_source_refs.add(key)
            if not any(item.get("artifact-id") == key[0]
                       and item.get("artifact-sha256") == key[1] for item in trusted):
                return False, ("stage-synthesis source artifact가 현재 workflow trusted registry에 없음: "
                               f"{key[0]}@{key[1][:12]}")
    basis_id = body.get("basis-artifact-id")
    basis_sha = body.get("basis-artifact-sha256")
    if basis_id is not None or basis_sha is not None:
        if not (basis_id and basis_sha):
            return False, "basis artifact binding은 id와 sha256을 함께 선언해야 한다"
        if not any(item.get("artifact-id") == basis_id
                   and item.get("artifact-sha256") == basis_sha for item in trusted):
            return False, "basis artifact id+sha256가 현재 workflow의 trusted revision과 불일치"
    if artifact.get("artifact-kind") == "compatibility-review":
        if str(body.get("reviewer-role-id") or "") != producer:
            return False, "compatibility-review reviewer-role-id는 artifact producer와 같아야 한다"
        endpoint_producers = set()
        for side in ("left", "right"):
            endpoint = body.get(side) or {}
            target = next((item for item in trusted
                           if item.get("artifact-kind") == endpoint.get("artifact-kind")
                           and item.get("artifact-id") == endpoint.get("artifact-id")
                           and item.get("artifact-sha256") == endpoint.get("artifact-sha256")), None)
            if not target:
                return False, f"compatibility-review {side} exact trusted artifact binding 불일치"
            endpoint_producers.add(str(target.get("producer-role-id") or ""))
        if producer in endpoint_producers:
            return False, "compatibility-review는 양쪽 산출물 producer와 독립이어야 한다"
    if artifact.get("artifact-kind") == "method-judgment-review":
        if str(body.get("reviewer-role-id") or "") != producer:
            return False, "method-judgment-review reviewer-role-id는 artifact producer와 같아야 한다"
        target = next((item for item in trusted
                       if item.get("artifact-id") == body.get("reviewed-artifact-id")
                       and item.get("artifact-sha256") == body.get("reviewed-artifact-sha256")), None)
        if not target:
            return False, "method-judgment-review reviewed artifact exact id+sha binding 불일치"
        if producer == target.get("producer-role-id"):
            return False, "method-judgment-review self-review 금지"
    if artifact.get("artifact-kind") == "decision-brief" and body.get("tier") != ledger.get("tier"):
        return False, "decision-brief.payload.tier는 canonical workflow tier와 같아야 한다"
    if artifact.get("artifact-kind") == "workload-profile":
        required_tier = _derived_risk_tier(body)
        if ledger.get("tier") not in VALID_TIERS:
            return False, f"canonical workflow tier가 미등록 값이다: {ledger.get('tier')!r}"
        if VALID_TIERS.index(ledger.get("tier")) < VALID_TIERS.index(required_tier):
            return False, (f"workflow tier가 workload risk hard floor보다 낮다"
                           f"(workflow={ledger.get('tier')}, required={required_tier})")
    current_stage = ledger.get("stage")
    artifact_stage = artifact.get("stage")
    allowed_stages = {current_stage}
    if artifact.get("artifact-kind") == "blocked-report":
        allowed_stages.add("blocked")
    if artifact_stage not in allowed_stages:
        return False, (f"artifact stage는 현재 running stage와 같아야 한다"
                       f"(current={current_stage}, report={artifact_stage})")
    ap = AC.absolute_path(report_path)
    base = os.path.basename(ap)
    if base.endswith(".report.yaml"):
        file_id = base[:-len(".report.yaml")]
        if file_id != artifact.get("artifact-id"):
            return False, f"report-id/파일명 불일치: {artifact.get('artifact-id')} != {file_id}"
    existing = [event for event in read_artifact_events(wf)
                if event.get("artifact-id") == artifact.get("artifact-id")]
    if existing:
        if existing[-1].get("artifact-sha256") == artifact.get("artifact-sha256"):
            return True, {"event": existing[-1], "existing": True, "artifact": artifact}
        return False, f"불변 artifact-id 재사용 거부: {artifact.get('artifact-id')} (새 revision은 새 id 필요)"
    event = {
        "artifact-event-id": f"afe-{_stamp()}-{uuid.uuid4().hex[:8]}",
        "event-type": "artifact-submitted",
        "effective-at": _now(),
        "submitted-by": actor_id,
        **{key: value for key, value in artifact.items() if key != "payload"},
    }
    epoch = _stage_epoch(wf, current_stage)
    if epoch:
        event["stage-epoch-id"] = epoch.get("event-id")
        event["stage-entered-at"] = epoch.get("effective-at")
    body = artifact.get("payload") or {}
    profile = body if artifact.get("artifact-kind") == "workload-profile" else body.get("workload-profile")
    if isinstance(profile, dict):
        event["workload-profile"] = profile
    for key in ("basis-artifact-id", "basis-artifact-sha256"):
        if body.get(key) is not None:
            event[key] = body.get(key)
    return True, {"event": event, "existing": False, "artifact": artifact}


def _atomic_event_transaction(wf, artifact_event=None, workflow_event=None):
    """Atomically append the event pair; roll back both JSONL tails on failure."""
    paths = []
    if artifact_event:
        paths.append((_artifact_events_path(create=True), artifact_event))
    if workflow_event:
        paths.append((_workflow_events_path(wf, create=True), workflow_event))
    try:
        _event_store.atomic_append(paths, transaction_lock=_workflow_lock(wf))
        with _workflow_lock(wf):
            led = read_ledger(wf) or _default_ledger(wf)
            led["artifacts"] = _trusted_artifacts(wf)
            led["last-updated-at"] = _now()
            _write_ledger(wf, led)
        return True, None
    except Exception as exc:
        return False, f"event transaction 실패: {exc}"


def submit_report(wf, report_path, actor):
    """Validate and atomically register an immutable report snapshot."""
    ok, prepared = _prepare_submit_event(wf, report_path, actor)
    if not ok:
        return False, prepared
    event = prepared["event"]
    if prepared["existing"]:
        return True, event
    ok, error = _atomic_event_transaction(wf, artifact_event=event)
    return (True, event) if ok else (False, error)


def submit_artifact(wf, report_path, actor):
    """Canonical API name; ``submit_report`` remains a compatibility alias."""
    return submit_report(wf, report_path, actor)


def record_artifact(wf, design_type=None, report_id=None, path=None, option_count=None,
                    evidence_grade=None):
    """Removed unsafe compatibility API.

    Callers cannot provide derived gate fields.  Kept only so old imports fail
    explicitly instead of silently minting a trusted artifact.
    """
    _log("record_artifact 제거됨: submit_report(wf, report_path, actor)를 사용하라")
    return False, read_ledger(wf) or _default_ledger(wf)


def _method_judgment_unmet(wf, artifact):
    """Independent method judgments are post-submit, pre-acceptance exact reviews."""
    if artifact.get("artifact-kind") in {"method-judgment-review", "compatibility-review", "quality-gate-review"}:
        return []
    ledger = read_ledger(wf) or {}
    if ledger.get("tier") not in ("standard", "heavy"):
        return []
    try:
        import artifact_contract as AC
        import method_contracts as MC
        _path, raw = AC.load_report(artifact.get("path"))
        producer = str(artifact.get("producer-role-id") or "").upper()
        execution = raw.get("method-execution") or {}
        profile = MC.resolve_method_profile(producer, execution.get("method-id"))
        if not profile:
            return []
        completed = {item.get("step-id") for item in execution.get("step-results", []) or []
                     if item.get("status") == "completed"}
        requirements = []
        for step in profile.get("workflow", []) or []:
            if step.get("step-id") not in completed:
                continue
            for gate in ((step.get("completion-gates") or {}).get("judgment") or []):
                reviewer = str(gate.get("reviewer-role") or "").upper()
                if reviewer and reviewer != producer:
                    requirements.append((step.get("step-id"), gate.get("gate-id"), reviewer))
        if not requirements:
            return []
        reviews = [item for item in _trusted_artifacts(wf)
                   if item.get("artifact-kind") == "method-judgment-review"]
        unmet = []
        for step_id, gate_id, reviewer in requirements:
            matched = False
            for review in reversed(reviews):
                body = _artifact_content(review, "method-judgment-review") or {}
                if (body.get("method-role-id") == producer
                        and body.get("method-id") == execution.get("method-id")
                        and body.get("step-id") == step_id and body.get("gate-id") == gate_id
                        and body.get("reviewed-artifact-id") == artifact.get("artifact-id")
                        and body.get("reviewed-artifact-sha256") == artifact.get("artifact-sha256")
                        and str(body.get("reviewer-role-id") or "").upper() == reviewer
                        and str(review.get("producer-role-id") or "").upper() == reviewer
                        and body.get("verdict") == "Passed"):
                    matched = True
                    break
            if not matched:
                unmet.append(f"{step_id}/{gate_id}: independent reviewer {reviewer}")
        return unmet
    except Exception as exc:
        return [f"method judgment evaluation failed: {exc}"]


def _record_internal_artifact(wf, kind, path, actor="OPS-ORCH"):
    """Narrow non-report writer for the company-context committer only."""
    if kind != "company-context" or actor != "OPS-ORCH":
        return False, "internal artifact writer는 OPS-ORCH company-context만 허용"
    try:
        import artifact_contract as AC
        ap = AC.absolute_path(path)
        if not ap or not os.path.isfile(ap):
            return False, f"artifact 파일 없음: {path}"
        event = {
            "artifact-event-id": f"afe-{_stamp()}-{uuid.uuid4().hex[:8]}",
            "event-type": "artifact-submitted", "effective-at": _now(),
            "submitted-by": actor, "workflow-id": wf,
            "artifact-id": f"company-context-{AC.sha256_file(ap)[:12]}",
            "report-id": f"company-context-{AC.sha256_file(ap)[:12]}",
            "artifact-kind": kind, "design-type": kind,
            "artifact-version": 1, "producer-role-id": actor,
            "path": ap, "artifact-sha256": AC.sha256_file(ap),
            "report-sha256": AC.sha256_file(ap),
        }
        with _workflow_lock(wf):
            if not _append_artifact_event(event):
                return False, "artifact event append 실패"
            led = read_ledger(wf) or _default_ledger(wf)
            led["artifacts"] = _trusted_artifacts(wf)
            _write_ledger(wf, led)
        return True, event
    except Exception as exc:
        return False, str(exc)


def _artifact_by_snapshot(wf, report_path):
    try:
        import artifact_contract as AC
        ap, report = AC.load_report(report_path)
        ident = AC.identity(report)
        sha = AC.sha256_file(ap)
    except Exception as exc:
        return None, str(exc)
    for artifact in _trusted_artifacts(wf):
        if (artifact.get("artifact-id") == str(ident.get("artifact-id"))
                and artifact.get("artifact-sha256") == sha):
            return artifact, None
    return None, "submit-report로 등록된 정확한 artifact revision(id+sha256)이 아니다"


def review_artifact(wf, report_path, decision, reviewer, supersedes=None):
    """Append an authorized decision for one exact artifact revision."""
    normalized = str(decision or "").strip().lower()
    if normalized not in ("accepted", "changes-requested", "blocked"):
        return False, "decision은 accepted|changes-requested|blocked 중 하나여야 한다"
    reviewer_id = str(reviewer or "").strip()
    if reviewer_id not in _role_registry():
        return False, f"등록되지 않은 reviewer: {reviewer_id}"
    artifact, error = _artifact_by_snapshot(wf, report_path)
    if not artifact:
        return False, error
    producer = artifact.get("producer-role-id")
    if reviewer_id == producer:
        return False, f"self-review 금지: reviewer={reviewer_id}, producer={producer}"
    if normalized == "accepted":
        judgment_unmet = _method_judgment_unmet(wf, artifact)
        if judgment_unmet:
            return False, "independent method judgment 미충족: " + "; ".join(judgment_unmet)
    try:
        import artifact_contract as AC
        capability = AC.reviewer_capability(artifact.get("artifact-kind"))
        definition = ((AC.load_contract().get("artifact-kinds") or {})
                      .get(artifact.get("artifact-kind")) or {})
    except Exception:
        capability = "artifact-reviewer"
        definition = {}
    if not _role_has_capability(reviewer_id, capability):
        return False, f"reviewer '{reviewer_id}'에 필요한 capability '{capability}'가 없다"
    required_reviewers = {str(r).upper() for r in (definition.get("required-reviewer-roles") or [])}
    if required_reviewers and reviewer_id.upper() not in required_reviewers:
        return False, (f"artifact-kind={artifact.get('artifact-kind')}는 reviewer가 "
                       f"{sorted(required_reviewers)} 중 하나여야 한다(got {reviewer_id})")
    if supersedes:
        prior = [a for a in _trusted_artifacts(wf) if a.get("artifact-id") == supersedes]
        if not prior:
            return False, f"supersedes 대상 artifact 없음: {supersedes}"
    try:
        import acceptance_log as AL
        event = AL.build_event(
            artifact.get("artifact-id"), normalized, workflow=wf,
            role=reviewer_id, supersedes=supersedes,
            report_sha256=artifact.get("artifact-sha256"),
            artifact_kind=artifact.get("artifact-kind"),
            producer_role_id=producer,
            reviewer={"actor-id": reviewer_id, "role-id": reviewer_id},
            authorization={
                "expected-reviewer-capability": capability,
                "producer-role-id": producer,
                "self-review": False,
            },
        )
        errors = AL.validate(event)
        if errors:
            return False, "; ".join(errors)
        with _workflow_lock(wf):
            if not AL.append_event(event):
                return False, "acceptance event append 실패"
        return True, event
    except Exception as exc:
        return False, f"review-artifact 실패: {exc}"


def _report_payload(report_path):
    import artifact_contract as AC
    _ap, report = AC.load_report(report_path)
    return AC.payload(report), AC.artifact_kind(report)


def record_quality_gate(wf, review_path, actor):
    actor_id = str(actor or "").strip()
    if not (_role_has_capability(actor_id, "quality-auditor")
            or _role_has_capability(actor_id, "data-quality-auditor")):
        return False, f"actor '{actor_id}'에 quality-auditor/data-quality-auditor capability가 없다"
    ledger = read_ledger(wf)
    if not ledger or ledger.get("stage") != "verification":
        return False, "quality gate는 현재 running stage가 verification일 때만 기록할 수 있다"
    ok, prepared = _prepare_submit_event(wf, review_path, actor_id)
    if not ok:
        return False, prepared
    try:
        body, kind = _report_payload(review_path)
        if kind != "quality-gate-review":
            return False, f"record-quality-gate는 artifact-kind=quality-gate-review만 허용(got {kind})"
        gate = body.get("quality-gate")
        status = gate.get("status") if isinstance(gate, dict) else gate
        if status not in ("Passed", "Failed"):
            return False, "quality-gate.status는 Passed|Failed여야 한다"
        blocker_open = body.get("blocker-open")
        if not isinstance(blocker_open, bool):
            return False, "blocker-open은 boolean이어야 한다"
        target_id = body.get("reviewed-artifact-id")
        target_sha = body.get("reviewed-artifact-sha256")
        trusted = _trusted_artifacts(wf)
        target = next((a for a in trusted
                       if a.get("artifact-id") == target_id
                       and a.get("artifact-sha256") == target_sha), None)
        if not target:
            return False, "quality review 대상 artifact id+sha256가 등록 revision과 불일치"
        if target.get("producer-role-id") == actor_id:
            return False, "quality gate self-review 금지"
        latest_completion = _latest_artifact_of_kind(trusted, "completion-record")
        if not latest_completion or target.get("artifact-kind") != "completion-record":
            return False, "quality review 대상은 completion-record여야 한다"
        if (target_id, target_sha) != (
                latest_completion.get("artifact-id"), latest_completion.get("artifact-sha256")):
            return False, "quality review 대상은 현재 최신 completion-record exact revision이어야 한다"
        checks = body.get("checks") or []
        findings = body.get("findings") or []
        check_ids = [check.get("check-id") for check in checks if isinstance(check, dict)]
        if len(check_ids) != len(set(check_ids)):
            return False, "quality checks[].check-id는 중복될 수 없다"
        derived_blocker = any(
            finding.get("blocking") is True and finding.get("resolved") is not True
            for finding in findings if isinstance(finding, dict)
        )
        derived_status = "Passed" if checks and all(
            check.get("status") == "Passed" for check in checks if isinstance(check, dict)
        ) and not derived_blocker else "Failed"
        if status != derived_status:
            return False, f"quality-gate.status는 checks/findings에서 파생해야 한다(derived={derived_status})"
        if blocker_open != derived_blocker:
            return False, f"blocker-open은 unresolved blocking findings에서 파생해야 한다(derived={derived_blocker})"
        import artifact_contract as AC
        epoch = _stage_epoch(wf, "verification") or {}
        strict_receipts = int(ledger.get("evidence-contract-version") or 1) >= 2
        completion_source_sha = None
        if (strict_receipts and ledger.get("tier") in ("standard", "heavy")
                and any(check.get("status") == "Passed" for check in checks)):
            try:
                completion_body, completion_kind = _report_payload(latest_completion.get("path"))
            except Exception as exc:
                return False, ("standard/heavy Passed quality gate는 최신 completion-record의 "
                               f"source revision을 읽을 수 있어야 한다: {exc}")
            source_revision = completion_body.get("source-revision") or {}
            completion_source_sha = str(source_revision.get("sha256") or "")
            if (completion_kind != "completion-record"
                    or not re.fullmatch(r"[0-9a-f]{64}", completion_source_sha)):
                return False, ("standard/heavy Passed quality gate는 최신 completion-record "
                               "payload.source-revision.sha256의 64-hex 결속이 필요하다")
        passed_receipt_owner = {}
        for check in checks:
            receipt_errors, receipts = AC.validate_receipt_ids(
                review_path, wf, check.get("evidence-receipt-ids") or [],
                require_success=check.get("status") == "Passed",
                since=epoch.get("effective-at"), require_context=True,
            )
            if receipt_errors:
                return False, f"quality check {check.get('check-id')}: " + "; ".join(receipt_errors)
            if strict_receipts:
                expected_assertion = "passed" if check.get("status") == "Passed" else "failed"
                for receipt in receipts:
                    if receipt.get("receipt_type") != "verification-run":
                        return False, (f"quality check {check.get('check-id')}: receipt "
                                       f"{AC.receipt_id(receipt)}는 verify_run.py가 발급한 "
                                       "verification-run이 아님")
                    if receipt.get("verification_category") != check.get("category"):
                        return False, (f"quality check {check.get('check-id')}: receipt category "
                                       f"{receipt.get('verification_category')!r} != "
                                       f"check category {check.get('category')!r}")
                    if receipt.get("assertion_status") != expected_assertion:
                        return False, (f"quality check {check.get('check-id')}: assertion_status "
                                       f"{receipt.get('assertion_status')!r} != {expected_assertion!r}")
                    if check.get("status") != "Passed":
                        # A failure receipt records what broke, not proof that a
                        # completion revision satisfied a particular criterion.
                        # Keep that legacy diagnostic path permissive.
                        continue
                    check_id = str(check.get("check-id") or "")
                    rid = str(AC.receipt_id(receipt) or "")
                    prior_owner = passed_receipt_owner.get(rid)
                    if prior_owner is not None and prior_owner != check_id:
                        return False, (f"quality receipt {rid} 재사용 금지: Passed checks "
                                       f"{prior_owner!r}, {check_id!r}")
                    passed_receipt_owner[rid] = check_id
                    if str(receipt.get("verification_subject") or "") != check_id:
                        return False, (f"quality check {check_id}: receipt {rid}의 "
                                       f"verification_subject {receipt.get('verification_subject')!r}가 "
                                       "check-id와 불일치")
                    if (ledger.get("tier") in ("standard", "heavy")
                            and str(receipt.get("source_revision_sha256") or "")
                            != completion_source_sha):
                        return False, (f"quality check {check_id}: standard/heavy Passed receipt "
                                       "source_revision_sha256가 최신 completion-record "
                                       f"revision과 불일치(got="
                                       f"{receipt.get('source_revision_sha256')!r}, "
                                       f"expected={completion_source_sha!r})")
        event = {
            "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
            "event-type": "quality-gate-recorded", "workflow-id": wf,
            "status": status, "blocker-open": blocker_open,
            "review-artifact-id": prepared["event"].get("artifact-id"),
            "review-artifact-sha256": prepared["event"].get("artifact-sha256"),
            "reviewed-artifact-id": target_id, "reviewed-artifact-sha256": target_sha,
            "check-count": len(checks),
            "check-categories": sorted({check.get("category") for check in checks}),
            "evidence-receipt-ids": sorted({rid for check in checks
                                             for rid in check.get("evidence-receipt-ids", [])}),
            "actor": actor_id, "effective-at": _now(),
        }
        artifact_event = None if prepared["existing"] else prepared["event"]
        committed, error = _atomic_event_transaction(
            wf, artifact_event=artifact_event, workflow_event=event)
        if not committed:
            return False, error
        return True, event
    except Exception as exc:
        return False, f"record-quality-gate 실패: {exc}"


def record_release_decision(wf, report_path, actor):
    actor_id = str(actor or "").strip()
    if not _role_has_capability(actor_id, "release-decider"):
        return False, f"actor '{actor_id}'에 release-decider capability가 없다"
    ledger = read_ledger(wf)
    if not ledger or ledger.get("stage") != "acceptance":
        return False, "release decision은 현재 running stage가 acceptance일 때만 기록할 수 있다"
    ok, prepared = _prepare_submit_event(wf, report_path, actor_id)
    if not ok:
        return False, prepared
    try:
        body, kind = _report_payload(report_path)
        if kind != "release-decision":
            return False, f"record-release-decision은 artifact-kind=release-decision만 허용(got {kind})"
        decision = body.get("release-decision")
        status = decision.get("status") if isinstance(decision, dict) else decision
        if status not in ("Approved", "Held", "Rejected"):
            return False, "release-decision.status는 Approved|Held|Rejected여야 한다"
        unresolved = body.get("unresolved-critical-risks")
        if not isinstance(unresolved, bool):
            return False, "unresolved-critical-risks는 boolean이어야 한다"
        completion_binding = (
            body.get("reviewed-completion-artifact-id"),
            body.get("reviewed-completion-artifact-sha256"),
        )
        current_completion = (
            ledger.get("current-completion-artifact-id"),
            ledger.get("current-completion-artifact-sha256"),
        )
        if not all(current_completion) or completion_binding != current_completion:
            return False, "release decision의 completion binding이 현재 latest completion과 불일치"
        if (body.get("reviewed-quality-event-id") != ledger.get("current-quality-event-id")
                or body.get("reviewed-quality-artifact-id") != ledger.get("current-quality-artifact-id")
                or body.get("reviewed-quality-artifact-sha256") != ledger.get("current-quality-artifact-sha256")):
            return False, "release decision의 quality binding이 현재 quality review와 불일치"
        if status == "Approved" and (
                ledger.get("quality_gate_status") != "Passed"
                or ledger.get("blocker-open") or unresolved):
            return False, "Approved는 현재 quality Passed, blocker 없음, unresolved critical risk 없음에서만 파생 가능"
        event = {
            "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
            "event-type": "release-decision-recorded", "workflow-id": wf,
            "status": status, "unresolved-critical-risks": unresolved,
            "decision-artifact-id": prepared["event"].get("artifact-id"),
            "decision-artifact-sha256": prepared["event"].get("artifact-sha256"),
            "reviewed-completion-artifact-id": completion_binding[0],
            "reviewed-completion-artifact-sha256": completion_binding[1],
            "reviewed-quality-event-id": body.get("reviewed-quality-event-id"),
            "reviewed-quality-artifact-id": body.get("reviewed-quality-artifact-id"),
            "reviewed-quality-artifact-sha256": body.get("reviewed-quality-artifact-sha256"),
            "quality-event-set": list(ledger.get("current-quality-event-ids") or []),
            "actor": actor_id, "effective-at": _now(),
        }
        artifact_event = None if prepared["existing"] else prepared["event"]
        committed, error = _atomic_event_transaction(
            wf, artifact_event=artifact_event, workflow_event=event)
        if not committed:
            return False, error
        return True, event
    except Exception as exc:
        return False, f"record-release-decision 실패: {exc}"


def block_workflow(wf, report_path, actor="OPS-ORCH"):
    if not _role_has_capability(actor, "transition-executor"):
        return False, "block command는 transition-executor만 실행 가능"
    ok, prepared = _prepare_submit_event(wf, report_path, actor)
    if not ok:
        return False, prepared
    body, kind = _report_payload(report_path)
    if kind != "blocked-report":
        return False, "block command는 artifact-kind=blocked-report가 필요"
    led = read_ledger(wf)
    if not led or led.get("stage") == "blocked":
        return False, "workflow가 없거나 이미 blocked 상태다"
    event = {
        "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
        "event-type": "workflow-blocked", "workflow-id": wf,
        "blocked-report-id": prepared["event"].get("artifact-id"),
        "blocked-report-sha256": prepared["event"].get("artifact-sha256"),
        "blocked-from": led.get("stage"),
        "resume-condition": body.get("resume-condition"),
        "actor": actor, "effective-at": _now(),
    }
    artifact_event = None if prepared["existing"] else prepared["event"]
    committed, error = _atomic_event_transaction(
        wf, artifact_event=artifact_event, workflow_event=event)
    if not committed:
        return False, error
    return True, event


def resume_workflow(wf, evidence_path, actor="OPS-ORCH"):
    if not _role_has_capability(actor, "transition-executor"):
        return False, "resume command는 transition-executor만 실행 가능"
    led = read_ledger(wf)
    if not led or led.get("stage") != "blocked" or not led.get("blocked-from"):
        return False, "blocked workflow가 아니거나 blocked-from이 없다"
    ok, prepared = _prepare_submit_event(wf, evidence_path, actor)
    if not ok:
        return False, prepared
    body, kind = _report_payload(evidence_path)
    if kind != "resume-evidence" or body.get("resume-condition-satisfied") is not True:
        return False, "resume-evidence와 resume-condition-satisfied:true가 필요"
    event = {
        "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
        "event-type": "workflow-resumed", "workflow-id": wf,
        "evidence-artifact-id": prepared["event"].get("artifact-id"),
        "evidence-artifact-sha256": prepared["event"].get("artifact-sha256"),
        "to": led.get("blocked-from"),
        "actor": actor, "effective-at": _now(),
    }
    artifact_event = None if prepared["existing"] else prepared["event"]
    committed, error = _atomic_event_transaction(
        wf, artifact_event=artifact_event, workflow_event=event)
    if not committed:
        return False, error
    return True, event


def read_state_events(wf):
    canonical = [event for event in read_workflow_events(wf)
                 if event.get("event-type") == "state-transition"]
    legacy = _read_jsonl(_events_path(wf, create=False))
    return legacy + canonical


# ---------------------------------------------------------------- fact derivation

def _report_id_of(art):
    rid = art.get("report-id")
    if rid:
        return rid
    path = art.get("path")
    if path:
        base = os.path.basename(str(path))
        if base.endswith(".report.yaml"):
            return base[: -len(".report.yaml")]
        return base
    return None


def _al_accepted_ids(wf, arts=None):
    """Exact, current, sha-bound accepted artifact ids."""
    try:
        import acceptance_log as AL  # noqa: E402
        ids = set()
        if arts is not None:
            for artifact in arts or []:
                if not isinstance(artifact, dict):
                    continue
                rid = _report_id_of(artifact)
                sha = artifact.get("artifact-sha256") or artifact.get("report-sha256")
                if rid and sha and AL.is_effectively_accepted(wf, rid, sha):
                    ids.add(rid)
            return ids
        for ev in AL.read_events():
            if ev.get("decision") != "accepted" or ev.get("workflow-id") != wf:
                continue
            rid = ev.get("accepted-report-id") or ev.get("report-id")
            sha = ev.get("artifact-sha256") or ev.get("report-sha256")
            if rid and AL.is_effectively_accepted(wf, rid, sha):
                ids.add(rid)
        return ids
    except Exception:
        return set()


def _accepted_design_types(wf, arts):
    """acceptance_log 에서 accepted 된 산출물의 design-type 집합.

    finding P0-4: 예전엔 원장 아티팩트의 자기신고 `review-state: Accepted` 만으로도 accepted 로
    쳤다 — 에이전트가 아티팩트를 등록하며 스스로 '승인됨'이라 적으면 상태 게이트를 통과할 수 있었다.
    이제 승인은 **오직 acceptance_log** 를 통한다(acceptance_log append 는 실존·validate 통과한
    report 만 인정 — P0-4c). 원장 아티팩트는 (design-type, report-id) 만 제공하고, 그 report-id 가
    acceptance_log 의 accepted 집합에 있을 때만 accepted 로 파생한다."""
    # One active revision per artifact-kind: submitting a newer revision makes the
    # older kind stale for gates until the newer exact id+sha is accepted.
    latest_by_kind = {}
    for artifact in arts or []:
        if isinstance(artifact, dict) and artifact.get("artifact-kind"):
            latest_by_kind[artifact.get("artifact-kind")] = artifact
    types = set()
    try:
        import acceptance_log as AL  # noqa: E402
        for kind, artifact in latest_by_kind.items():
            rid = _report_id_of(artifact)
            sha = artifact.get("artifact-sha256") or artifact.get("report-sha256")
            if rid and sha and AL.is_effectively_accepted(wf, rid, sha):
                types.add(kind)
    except Exception:
        return set()
    return types


def _latest_accepted_artifact(wf, arts, kind):
    artifact = _latest_artifact_of_kind(arts, kind)
    if not artifact:
        return None
    rid = _report_id_of(artifact)
    return artifact if rid and rid in _al_accepted_ids(wf, arts) else None


def _experience_foundation_required(parent_wf):
    """Typed workload predicate for the front-of-funnel experience gate."""
    profile = _workload_profile(_trusted_artifacts(parent_wf))
    surfaces = profile.get("surfaces") if isinstance(profile.get("surfaces"), dict) else {}
    if not surfaces.get("ui"):
        return False
    return (
        profile.get("surface-archetype") in ("public-website", "interactive-learning")
        or profile.get("experience-change") in ("new-product", "major-redesign")
    )


def _same_snapshot_ref(ref, sha, artifact):
    if not (ref and sha and artifact and sha == artifact.get("artifact-sha256")):
        return False
    try:
        import artifact_contract as AC
        return os.path.abspath(AC.absolute_path(ref)) == os.path.abspath(
            AC.absolute_path(artifact.get("path")))
    except Exception:
        return False


def _accepted_experience_feasibility(child_wf, parent_wf, product_decision, strategy, kind):
    artifact = _latest_accepted_artifact(child_wf, _trusted_artifacts(child_wf), kind)
    body = _artifact_content(artifact, kind) if artifact else None
    if not isinstance(body, dict):
        return None
    if (body.get("parent-workflow-id") != parent_wf
            or body.get("product-decision-id") != product_decision
            or body.get("verdict") != "feasible"):
        return None
    if not _same_snapshot_ref(body.get("strategy-ref"), body.get("strategy-sha256"), strategy):
        return None
    return artifact


def _experience_artifact_bundle(child_wf, parent_wf):
    """Return six accepted, exact, cross-bound foundation artifacts or None."""
    led = _load_ledger_safe(child_wf)
    if (led.get("plan") != "experience-foundation"
            or led.get("parent-workflow-id") != parent_wf
            or led.get("stage") not in ("wireframes", "foundation-approved")):
        return None
    product_decision = led.get("product-decision-id")
    if not _product_decision_current(parent_wf, product_decision):
        return None
    arts = _trusted_artifacts(child_wf)
    bundle = {
        "benchmark": _latest_accepted_artifact(child_wf, arts, "competitive-experience-benchmark"),
        "strategy": _latest_accepted_artifact(child_wf, arts, "experience-strategy"),
    }
    if any(value is None for value in bundle.values()):
        return None
    bundle["technical"] = _accepted_experience_feasibility(
        child_wf, parent_wf, product_decision, bundle["strategy"], "experience-technical-feasibility")
    bundle["operational"] = _accepted_experience_feasibility(
        child_wf, parent_wf, product_decision, bundle["strategy"], "experience-operational-feasibility")
    bundle["blueprint"] = _latest_accepted_artifact(child_wf, arts, "experience-blueprint")
    bundle["wireframe"] = _latest_accepted_artifact(child_wf, arts, "wireframe-set")
    if any(value is None for value in bundle.values()):
        return None
    bodies = {key: _artifact_content(value, value.get("artifact-kind"))
              for key, value in bundle.items()}
    if any(not isinstance(value, dict) for value in bodies.values()):
        return None
    for body in bodies.values():
        if (body.get("parent-workflow-id") != parent_wf
                or body.get("product-decision-id") != product_decision):
            return None
    if bodies["strategy"].get("decision") != "proceed":
        return None
    if not _same_snapshot_ref(bodies["strategy"].get("benchmark-ref"),
                              bodies["strategy"].get("benchmark-sha256"), bundle["benchmark"]):
        return None
    if not (_same_snapshot_ref(bodies["blueprint"].get("benchmark-ref"),
                               bodies["blueprint"].get("benchmark-sha256"), bundle["benchmark"])
            and _same_snapshot_ref(bodies["blueprint"].get("strategy-ref"),
                                   bodies["blueprint"].get("strategy-sha256"), bundle["strategy"])):
        return None
    if not _same_snapshot_ref(bodies["wireframe"].get("blueprint-ref"),
                              bodies["wireframe"].get("blueprint-sha256"), bundle["blueprint"]):
        return None
    return bundle


def _has_experience_foundation(wf, led):
    """Validate the parent link and every accepted foundation snapshot fail-closed."""
    approval = led.get("experience-foundation-approval")
    if approval:
        parent_wf, parent_mode = wf, True
    elif led.get("plan") == "experience-foundation" and led.get("parent-workflow-id"):
        parent_wf, parent_mode = led.get("parent-workflow-id"), False
        approval = (_load_ledger_safe(parent_wf).get("experience-foundation-approval") or {})
        if approval.get("child-workflow-id") != wf:
            return False
    else:
        return False
    child = approval.get("child-workflow-id")
    if not child:
        return False
    child_led = _load_ledger_safe(child)
    if parent_mode and child_led.get("stage") != "foundation-approved":
        return False
    if not parent_mode and child_led.get("stage") not in ("wireframes", "foundation-approved"):
        return False
    bundle = _experience_artifact_bundle(child, parent_wf)
    if not bundle or approval.get("product-decision-id") != child_led.get("product-decision-id"):
        return False
    for key, artifact in bundle.items():
        if (approval.get(f"{key}-id") != artifact.get("artifact-id")
                or approval.get(f"{key}-sha256") != artifact.get("artifact-sha256")
                or not _same_snapshot_ref(approval.get(f"{key}-ref"),
                                          approval.get(f"{key}-sha256"), artifact)):
            return False
    return True


def register_experience_foundation(parent, child):
    """Atomically bind an accepted foundation bundle to its parent workflow."""
    parent_path = _ledger_path(parent, create=False)
    child_path = _ledger_path(child, create=False)
    if not parent_path or not os.path.exists(parent_path):
        raise ValueError(f"parent workflow '{parent}' 원장 없음")
    if not child_path or not os.path.exists(child_path):
        raise ValueError(f"child workflow '{child}' 원장 없음")
    bundle = _experience_artifact_bundle(child, parent)
    if not bundle:
        raise ValueError("accepted benchmark/strategy/technical/operational/blueprint/wireframe exact bundle 또는 cross-reference 불충족")
    child_led = _load_ledger_safe(child)
    existing = (_load_ledger_safe(parent).get("experience-foundation-approval") or {})
    if existing.get("child-workflow-id") not in (None, child):
        raise ValueError(f"기존 active experience-foundation child={existing.get('child-workflow-id')} 와 충돌")
    event = {
        "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
        "event-type": "experience-foundation-registered", "workflow-id": parent,
        "child-workflow-id": child, "product-decision-id": child_led.get("product-decision-id"),
        "actor": "OPS-ORCH", "effective-at": _now(),
    }
    for key, artifact in bundle.items():
        event[f"{key}-id"] = artifact.get("artifact-id")
        event[f"{key}-ref"] = artifact.get("path")
        event[f"{key}-sha256"] = artifact.get("artifact-sha256")
    committed, error = _atomic_event_transaction(parent, workflow_event=event)
    if not committed:
        raise ValueError(error or "부모 experience-foundation event 기록 실패")
    return True


def _ui_design_release_binding_ok(arts):
    ui = _latest_artifact_of_kind(arts, "ui-design")
    body = _artifact_content(ui, "ui-design") if ui else None
    if not isinstance(body, dict):
        return False
    binding = next((item for item in body.get("design-system-bindings", [])
                    if isinstance(item, dict) and item.get("release-id")), None)
    if not binding:
        return False
    try:
        import artifact_contract as AC
        path = AC.absolute_path(binding.get("release-ref"))
        if (not path or not os.path.isfile(path)
                or AC.sha256_file(path) != binding.get("release-sha256")):
            return False
        release = _load_yaml(path).get("design-system-release", {})
        if (release.get("release-id") != binding.get("release-id")
                or release.get("state") not in ("candidate", "stable")):
            return False
        if not set(binding.get("component-ids") or []).issubset(set(release.get("components") or [])):
            return False
        delta = binding.get("delta") or {}
        return isinstance(delta.get("tokens"), list) and isinstance(delta.get("components"), list)
    except Exception:
        return False


def _direction_brief_foundation_refs_ok(parent_wf, brief_ref):
    approval = (_load_ledger_safe(parent_wf).get("experience-foundation-approval") or {})
    path = brief_ref if os.path.isabs(str(brief_ref or "")) else os.path.join(ROOT, str(brief_ref or ""))
    brief = _load_yaml(path)
    names = {
        "benchmark": "competitive-experience-benchmark",
        "blueprint": "experience-blueprint",
        "wireframe": "wireframe-set",
    }
    for approval_key, brief_key in names.items():
        ref = brief.get(f"{brief_key}-ref")
        sha = brief.get(f"{brief_key}-sha256")
        if sha != approval.get(f"{approval_key}-sha256"):
            return False
        try:
            import artifact_contract as AC
            if os.path.abspath(AC.absolute_path(ref)) != os.path.abspath(
                    AC.absolute_path(approval.get(f"{approval_key}-ref"))):
                return False
        except Exception:
            return False
    return True


def _artifact_present(arts, type_aliases):
    for a in arts or []:
        if isinstance(a, dict) and a.get("design-type") in type_aliases:
            return True
    return False


# ---------------------------------------------------------------- preview_ui gate (항목3)
def _evidence_receipts():
    """evidence-ledger(ledger.jsonl)의 receipt 리스트. degrade -> []. 예외 없음.

    PostToolUse evidence_ledger 가 쓴 Claude-Code 공급 receipt(command·exit_code·workflow_id)를
    읽는다 — 이것이 위조 불가한 실행 증거의 원천(P0-6)."""
    try:
        import _workspace as W  # noqa: E402
        ed = W.evidence_dir()
    except Exception:
        return []
    p = os.path.join(ed, "ledger.jsonl")
    if not os.path.exists(p):
        return []
    out = []
    try:
        with open(p, encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                    if isinstance(obj, dict):
                        out.append(obj)
                except Exception:
                    continue
    except Exception:
        return []
    return out


_PNG_SIG = b"\x89PNG\r\n\x1a\n"


def _abs_path(p):
    """상대경로를 ROOT(CLAUDE_PROJECT_DIR) 기준 절대경로로. $VAR 는 export 된 것만 확장."""
    if not p:
        return p
    p = os.path.expandvars(p)
    if os.path.isabs(p):
        return p
    for base in (ROOT, os.getcwd()):
        if base:
            cand = os.path.join(base, p)
            if os.path.exists(cand):
                return cand
    return os.path.join(ROOT or os.getcwd(), p)


def _valid_png(path):
    """실제 렌더된 PNG 인가 — 유효 시그니처(\\x89PNG..) + 비자명 크기(>1000B). 위조 저항 근거."""
    try:
        if not os.path.isfile(path) or os.path.getsize(path) <= 1000:
            return False
        with open(path, "rb") as fh:
            return fh.read(8) == _PNG_SIG
    except Exception:
        return False


def _render_png_ok(cmd):
    """finding F9: exit-code 를 못 읽는 환경(모든 receipt exit_code=None)에서도 렌더 게이트를
    **실제 실행 산출물**에 접지한다. preview_ui.py 호출 command 에서 출력 위치(--out 값의 형제
    스크린샷들·positional project_dir)를 뽑아, 유효한 PNG 스크린샷이 실재하는지 검증한다.

    exit_code=0 자기신고가 아니라 실물 스크린샷 파일(시그니처+크기)을 확인하므로 위조 저항이
    오히려 exit-code 보다 높다 — preview_ui 는 렌더 실패 시 shot 을 쓰기 전에 die() 하므로
    유효 PNG 존재 == 실제 DOM 렌더 성공. P0-6 유지: receipt 자체는 여전히 PostToolUse hook 이
    실행맥락(session/tool_use_id/cwd)에 결속하며, 임의 명령의 성공을 위장할 수는 없다."""
    try:
        toks = shlex.split(cmd)
    except Exception:
        toks = cmd.split()
    files, dirs = [], []
    i = 0
    while i < len(toks):
        t = toks[i]
        if t in ("--out", "--out=") and i + 1 < len(toks):
            files.append(toks[i + 1])
            i += 2
            continue
        if t.startswith("--out="):
            files.append(t.split("=", 1)[1])
        elif t.lower().endswith(".png"):
            files.append(t)
        elif t.endswith("preview_ui.py") and i + 1 < len(toks) and not toks[i + 1].startswith("-"):
            dirs.append(toks[i + 1])  # positional project_dir
        i += 1
    # --out X/foo.png 는 foo.png 자체가 아니라 foo.w1280.png / foo.state-*.png 형제로 저장된다.
    for f in files:
        ap = _abs_path(f)
        d = os.path.dirname(ap)
        stem = os.path.splitext(os.path.basename(ap))[0]
        if _valid_png(ap):
            return True
        try:
            for name in os.listdir(d):
                if name.lower().endswith(".png") and name.startswith(stem):
                    if _valid_png(os.path.join(d, name)):
                        return True
        except Exception:
            continue
    for dd in dirs:
        ad = _abs_path(dd)
        try:
            for name in os.listdir(ad):
                if name.lower().endswith(".png") and _valid_png(os.path.join(ad, name)):
                    return True
        except Exception:
            continue
    return False


def _has_preview_receipt(wf, prototype=None):
    """항목3: 이 workflow 에 **통과한 preview_ui 렌더 게이트 receipt** 가 있는가.

    design-system 산출물이 실제로 렌더·품질검증(preview_ui: #root 비어있지 않음·WCAG 대비·포커스·
    반응형)을 통과했음을 evidence-ledger 의 실제 실행 receipt(command=preview_ui.py, exit_code=0)로
    확인한다 — 산문 문서만으로 'design-system Accepted' 를 위장하지 못하게 한다. receipt 가
    workflow_id 를 가지면 wf 와 일치할 때만 인정, 없으면(느슨한 결속 환경) command·exit 만으로 인정."""
    for r in _evidence_receipts():
        if not isinstance(r, dict):
            continue
        # Typed ``verify_run`` receipts preserve argv as a list so shell quoting
        # cannot change the evidenced command.  Legacy hook receipts used one
        # ``command`` string.  Accept both representations, but always rebuild a
        # quoted command from the trusted argv list rather than ignoring the new
        # receipt format (which previously made every typed preview invisible).
        argv = r.get("command_argv")
        # A typed receipt's argv is the canonical command representation.  Do
        # not let an optional legacy ``command`` string override it: accepting
        # two disagreeing command sources makes the supposedly typed receipt
        # ambiguous.  Legacy receipts without argv still use ``command``.
        if isinstance(argv, list) and argv and all(isinstance(token, str) for token in argv):
            cmd = shlex.join(argv)
        else:
            cmd = str(r.get("command") or "")
        # 실제 preview_ui.py **호출**만 인정 — grep/find/cat/echo 가 "preview_ui" 를 단순
        # 언급하는 receipt(예: `grep -rn "preview_ui" ...`)는 렌더가 아니므로 제외(강화).
        if not re.search(r"python[0-9]*\s+\S*preview_ui\.py", cmd):
            continue
        # --contrast-only 만 돌린 정적 체크는 렌더 게이트가 아니다(빌드·DOM 검증 없음).
        if "--contrast-only" in cmd:
            continue
        # PNG files can survive a failed run (for example CSS health may fail
        # after responsive screenshots were written).  Whenever an exit code is
        # available it is authoritative; the PNG fallback is only for legacy
        # hook environments that genuinely omit exit status.
        exit_code = r.get("exit_code")
        if exit_code is not None and exit_code != 0:
            continue
        if (r.get("receipt_type") == "verification-run"
                and r.get("assertion_status") != "passed"):
            continue
        rwf = r.get("workflow_id")
        if wf is not None and rwf != wf:
            continue
        if not r.get("session_id") or not r.get("agent_id"):
            continue
        if isinstance(prototype, dict):
            expected_receipt = prototype.get("preview-receipt-ref")
            if not expected_receipt or str(r.get("tool_use_id") or r.get("receipt_id") or "") != str(expected_receipt):
                continue
            prototype_path = _abs_path(prototype.get("prototype-path"))
            if not prototype_path or not os.path.isfile(prototype_path):
                continue
            if _sha256_of(prototype_path) != prototype.get("prototype-sha256"):
                continue
            try:
                tokens = [os.path.abspath(_abs_path(token)) for token in shlex.split(cmd)
                          if token and not token.startswith("-")]
            except Exception:
                tokens = []
            if os.path.dirname(prototype_path) not in tokens and prototype_path not in tokens:
                continue
        # 접지(F9): exit-code 를 읽을 수 있으면(exit_code==0) 그대로 인정하고, 못 읽는 환경
        # (이 harness 처럼 모든 receipt exit_code=None)에서는 **실제 렌더된 PNG 산출물**로 접지한다.
        # 사용자 명시 승인(P0-6 렌더 게이트 검증방식 변경) — self-report 가 아니라 실물 스크린샷을
        # 검증하므로 현행보다 위조 저항이 높다. P0-6 의 임의-명령-성공-위장 방지는 유지된다.
        if _render_png_ok(cmd):
            return True
    return False


_UI_KIND_TOKENS = {"screen", "frontend", "app", "webapp"}


def _is_ui_bearing(led, wf=None):
    """Return the sole UI-bearing signal: trusted workload-profile.payload.surfaces.ui."""
    arts = led.get("artifacts") or []
    profile = _workload_profile(arts)
    surfaces = profile.get("surfaces") if isinstance(profile.get("surfaces"), dict) else {}
    return bool(surfaces.get("ui"))


_FOUNDER_CTX = os.path.join(ROOT, "org-os", "01-company", "founder-context.yaml")
_COMPANY_CTX_PATH = os.path.join(ROOT, "org-os", "01-company", "company-context.yaml")


def _founder_context_filled():
    try:
        return str(_load_yaml(_FOUNDER_CTX).get("status", "")).strip().lower() == "filled"
    except Exception:
        return False


def _company_ctx_status():
    try:
        return str(_load_yaml(_COMPANY_CTX_PATH).get("status", "")).strip().lower()
    except Exception:
        return ""


def _company_ctx_lint_ok():
    try:
        import lint_company_context as L
        hard, _ = L.lint_file(_COMPANY_CTX_PATH, is_candidate=False)
        return not hard
    except Exception:
        return False


def _company_ctx_artifact_recorded(arts):
    return any(isinstance(a, dict) and a.get("design-type") == "company-context" for a in (arts or []))


def _venture_decision_receipt_ok(wf, expected_decision_ids=None):
    """HUMAN-001 accepted 이벤트가 venture-decision report 의 현재 해시와 일치 바인딩(§9.4).
    이벤트에 report-sha256 이 있고(Task 15), report 파일이 실존하며 해시 일치할 때만 True.
    candidate commit 때 expected_decision_ids를 주면 그 source-decision-id와도 정확히 결속한다."""
    try:
        import acceptance_log as AL, hashlib
        expected = {str(value) for value in (expected_decision_ids or []) if str(value).strip()}
        for ev in reversed(list(AL.read_events())):
            if ev.get("decision") != "accepted":
                continue
            if ev.get("workflow-id") not in (None, wf):
                continue
            if str(ev.get("role-id", "")).upper() != "HUMAN-001".upper():
                continue
            if ev.get("artifact-kind") != "venture-decision":
                continue
            if str(ev.get("producer-role-id", "")).upper() != "EXEC-CEO":
                continue
            sha = ev.get("report-sha256")
            rid = ev.get("accepted-report-id") or ev.get("report-id")
            if not (sha and rid):
                continue
            if expected and str(rid) not in expected:
                continue
            path = AL._resolve_report_path(rid, wf)
            if not path or not os.path.exists(path):
                continue
            actual = hashlib.sha256(open(path, "rb").read()).hexdigest()
            if actual == sha and AL.is_effectively_accepted(wf, rid, sha):
                return True
        return False
    except Exception:
        return False


def _distinct_opportunity_cluster_count(artifacts):
    """Trusted opportunity-cluster snapshots의 서로 다른 payload.id 개수."""
    cluster_ids = set()
    for artifact in artifacts or []:
        if not isinstance(artifact, dict) or artifact.get("design-type") != "opportunity-cluster":
            continue
        body = _artifact_content(artifact, "opportunity-cluster") or {}
        cluster_id = str(body.get("id") or "").strip()
        if cluster_id:
            cluster_ids.add(cluster_id)
    return len(cluster_ids)


def _dd_active(led):
    return led.get("design-direction-active") or {}


def _active_artifact(led, arts, design_type, id_key):
    """active cycle pointer 가 가리키는 report-id 의 아티팩트만 반환(오래된 것 무시, Blocker 6).
    want 이 있으면 정확히 일치하는 것만(없으면 None, fail closed) — 오래된 아티팩트로 대체되는 것을 막는다.
    want 이 없으면(포인터 미명시) 최신(마지막) 아티팩트를 사용."""
    want = _dd_active(led).get(id_key)
    match = None
    for a in (arts or []):
        if isinstance(a, dict) and a.get("design-type") == design_type:
            if want is not None:
                if a.get("report-id") == want:
                    return a
            else:
                match = a  # 포인터가 이 아티팩트 report-id를 명시 안 하면 최신(마지막) 사용
    return None if want is not None else match


def _artifact_content(a, design_type):
    """Load the immutable submitted snapshot payload (legacy embedded form is read-compatible)."""
    if not isinstance(a, dict):
        return None
    embedded = a.get(design_type)
    if isinstance(embedded, dict):
        return embedded
    path = a.get("path")
    if path:
        try:
            import artifact_contract as AC
            ap = AC.absolute_path(path)
        except Exception:
            ap = path if os.path.isabs(path) else os.path.join(ROOT, path)
        if os.path.exists(ap):
            try:
                doc = _load_yaml(ap)
                if doc.get("report-type") == "workflow-artifact" and isinstance(doc.get("payload"), dict):
                    doc = doc["payload"]
                # 파일이 {design_type: {...}} 래퍼든 평문 dict 든 모두 지원
                if isinstance(doc, dict):
                    return doc.get(design_type) if isinstance(doc.get(design_type), dict) else doc
            except Exception:
                return None
    return None


def _dd_input_brief_ref(led):
    return led.get("direction-input-brief-ref") or (led.get("facts") or {}).get("direction-input-brief-ref")


def _current_input_brief_sha(led):
    ref = _dd_input_brief_ref(led)
    if not ref:
        return None
    p = ref if os.path.isabs(ref) else os.path.join(ROOT, ref)
    return hashlib.sha256(open(p, "rb").read()).hexdigest() if os.path.exists(p) else None


def _direction_experience_inputs_ok(led, direction_set, audit=None):
    parent = led.get("parent-workflow-id")
    if not parent or not _experience_foundation_required(parent):
        return True
    approval = (_load_ledger_safe(parent).get("experience-foundation-approval") or {})
    if not approval:
        return False
    checks = (("experience-blueprint", "blueprint"), ("wireframe-set", "wireframe"))
    for document_key, approval_key in checks:
        if (direction_set.get(f"{document_key}-sha256") != approval.get(f"{approval_key}-sha256")
                or not _direction_brief_foundation_refs_ok(parent, _dd_input_brief_ref(led))):
            return False
        try:
            import artifact_contract as AC
            if os.path.abspath(AC.absolute_path(direction_set.get(f"{document_key}-ref"))) != os.path.abspath(
                    AC.absolute_path(approval.get(f"{approval_key}-ref"))):
                return False
        except Exception:
            return False
    wireframe_sha = approval.get("wireframe-sha256")
    if any(item.get("content-contract-sha256") != wireframe_sha
           for item in (direction_set.get("directions") or []) if isinstance(item, dict)):
        return False
    if audit is not None:
        if audit.get("competitive-experience-benchmark-sha256") != approval.get("benchmark-sha256"):
            return False
        try:
            import artifact_contract as AC
            if os.path.abspath(AC.absolute_path(audit.get("competitive-experience-benchmark-ref"))) != os.path.abspath(
                    AC.absolute_path(approval.get("benchmark-ref"))):
                return False
        except Exception:
            return False
        findings = audit.get("benchmark-relative-findings") or []
        if not isinstance(findings, list) or len(findings) < 3:
            return False
    return True


def _directions_diverged(led, arts):
    a = _active_artifact(led, arts, "direction-set", "direction-set-report-id")
    ds = _artifact_content(a, "direction-set")
    if not isinstance(ds, dict):
        return False
    if not _direction_experience_inputs_ok(led, ds):
        return False
    dirs = ds.get("directions") or []
    if len(dirs) < 3:
        return False
    runs = [d.get("producer-run-id") for d in dirs]
    pkgs = [d.get("context-package-id") for d in dirs]
    if None in runs or len(set(runs)) != len(runs):
        return False
    if None in pkgs or len(set(pkgs)) != len(pkgs):
        return False
    if not (ds.get("representative-screen") or {}).get("id"):
        return False
    cp = ds.get("comparison-preview") or {}
    if not (cp.get("receipt-ref") and cp.get("gallery-path")):
        return False
    for d in dirs:
        cs = d.get("coded-slice")
        if not cs:
            return False
        ap = cs if os.path.isabs(cs) else os.path.join(ROOT, cs)
        if not os.path.isfile(ap):
            return False   # 실제 픽셀 = 파일(디렉터리면 open() 크래시 대신 무효 처리, F6)
        csha = d.get("coded-slice-sha256")
        if not csha or hashlib.sha256(open(ap, "rb").read()).hexdigest() != csha:
            return False   # 실존만으로 부족 — 내용까지 대조(state-transition-rules.yaml:120 실존/hash); 필드 생략도 차단(우회방지)
    return True


def _divergence_charter_ok(led, arts):
    """Active charter must be a real, lint-clean submitted snapshot."""
    artifact = _active_artifact(led, arts, "divergence-charter", "divergence-charter-report-id")
    if not (artifact and artifact.get("path")):
        return False
    try:
        import artifact_contract as AC
        import lint_design_direction as LDD
        path = AC.absolute_path(artifact["path"])
        hard, _warn = LDD.lint_file(path, "divergence-charter")
        return not hard
    except Exception:
        return False


def _divergence_audit_ok(wf, led, arts):
    """Selection is impossible until a comparative, sibling-visible audit passes.

    This deliberately differs from divergence worker isolation: the audit must bind and
    compare all three originals, cover every pair, validate full-size preview hashes, and
    have no primitive collision or blocking finding.  The charter and audit themselves
    must be accepted exact revisions; a bare synthesis string cannot substitute for them.
    """
    charter = _active_artifact(led, arts, "divergence-charter", "divergence-charter-report-id")
    dset = _active_artifact(led, arts, "direction-set", "direction-set-report-id")
    audit = _active_artifact(led, arts, "comparative-divergence-audit", "divergence-audit-report-id")
    if not all(a and a.get("path") for a in (charter, dset, audit)):
        return False
    accepted = _al_accepted_ids(wf, arts)
    if _report_id_of(charter) not in accepted or _report_id_of(audit) not in accepted:
        return False
    try:
        import artifact_contract as AC
        import lint_design_direction as LDD
        hard, _warn = LDD.lint_divergence_bundle(
            AC.absolute_path(audit["path"]), AC.absolute_path(dset["path"]),
            AC.absolute_path(charter["path"]))
        if hard:
            return False
        audit_doc = _artifact_content(audit, "comparative-divergence-audit") or {}
        ds_doc = _artifact_content(dset, "direction-set") or {}
        if not _direction_experience_inputs_ok(led, ds_doc, audit_doc):
            return False
        producer_runs = {d.get("producer-run-id") for d in (ds_doc.get("directions") or [])
                         if isinstance(d, dict) and d.get("producer-run-id")}
        return (str(audit_doc.get("reviewer-role-id") or "").upper() == "DES-VISUAL"
                and audit_doc.get("reviewer-run-id") not in producer_runs)
    except Exception:
        return False


def _selected_direction_ok(led, arts):
    sd = _active_artifact(led, arts, "selected-direction", "selected-direction-report-id")
    dset = _active_artifact(led, arts, "direction-set", "direction-set-report-id")
    if not (sd and dset and sd.get("path") and dset.get("path")):
        return False
    try:
        import artifact_contract as AC
        import lint_design_direction as LDD
        hard, _ = LDD.lint_selected_direction(
            AC.absolute_path(sd["path"]), AC.absolute_path(dset["path"]))
        return not hard
    except Exception:
        return False


def _selected_direction_accepted_ok(wf, led, arts):
    """Task 15 item 7(리뷰): `_selected_direction_ok`(bundle lint) 만으로는 부족하다 — design-direction.md
    §3이 명시하는 계약("selected-direction-accepted 는 bundle lint 통과 **+** 이 accepted 이벤트 둘 다
    요구", 즉 selected-direction 자신의 report-id 에 대한 accepted 이벤트)을 정확히 지켜야 한다.
    예전엔 `_al_has_accepted(wf)`(이 workflow 에 **아무 report나** accepted 됐는지만 확인)를 썼는데,
    이는 이 workflow 에서 우연히·별도로 accepted 된 무관한 report(예: 다른 design-type 산출물)로도
    이 게이트를 통과시킬 수 있었다(대체 가능 = 위조 경로). `_al_accepted_ids(wf)`(exact report-id
    matching — `_product_decision_current`/`_accepted_design_types`와 동일 패턴)로 active
    selected-direction 아티팩트의 정확한 report-id 가 accepted 됐는지만 인정한다."""
    if not _selected_direction_ok(led, arts):
        return False
    sd = _active_artifact(led, arts, "selected-direction", "selected-direction-report-id")
    body = _artifact_content(sd, "selected-direction") or {}
    if body.get("selection-decision") == "none-of-the-above":
        return False
    rid = _report_id_of(sd) if sd else None
    return bool(rid) and rid in _al_accepted_ids(wf, arts)


def _none_of_the_above_recorded(wf, led, arts):
    sd = _active_artifact(led, arts, "selected-direction", "selected-direction-report-id")
    if not sd or not _selected_direction_ok(led, arts):
        return False
    body = _artifact_content(sd, "selected-direction") or {}
    rid = _report_id_of(sd)
    return (body.get("selection-decision") == "none-of-the-above"
            and bool(rid) and rid in _al_accepted_ids(wf, arts))


def _critique_panel_ok(led, arts):
    a = _active_artifact(led, arts, "design-review-panel", "review-panel-report-id")
    panel = _artifact_content(a, "design-review-panel")
    if not isinstance(panel, dict):
        return False
    required = {"product-fit", "usability", "distinctiveness", "visual-craft",
                "systematizability", "market-memorability", "implementability"}
    reviews = panel.get("reviews") or []
    review_lenses = [r.get("lens") for r in reviews if isinstance(r, dict)]
    if set(review_lenses) != required or len(review_lenses) != len(required):
        return False
    lens_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",
    }
    if any(str(review.get("reviewer-role-id") or "").upper() != lens_roles.get(review.get("lens"))
           for review in reviews):
        return False
    reviewer_runs = [review.get("reviewer-run-id") for review in reviews]
    if None in reviewer_runs or len(set(reviewer_runs)) != len(reviewer_runs):
        return False
    trusted_panel = bool(a.get("artifact-sha256"))
    winner = _active_artifact(led, arts, "winner-prototype", "prototype-report-id")
    if trusted_panel:
        if not winner:
            return False
        if (panel.get("target-prototype-id"), panel.get("target-prototype-sha256")) != (
                _report_id_of(winner), winner.get("artifact-sha256")):
            return False
    # Fix B(최종리뷰): producer≠reviewer 를 run-id 문자열 자기신고만으로 확인하던 것을 강화한다
    # (design-direction-design.md §8 "각 review report hash 검증"). 각 review 가 가리키는
    # report-ref 파일이 실존하고, 그 **라이브 sha256** 이 report-sha256 과 일치할 때만 그 review 를
    # 진짜로 인정한다 — `_directions_diverged`의 coded-slice hash 대조, `_artifact_content`/
    # `_selected_direction_ok`와 동일한 ROOT-relative 경로 해석. 참조·해시 누락/파일 부재/불일치는
    # 전부 fail closed(위조·스테일 review 로 패널을 통과시키는 것을 차단).
    def _normalized_verdict(value):
        value = str(value or "").strip().lower()
        return {"pass": "pass", "passed": "pass",
                "concerns": "revise", "minor-revision": "revise", "revise": "revise",
                "blocking": "blocking", "concept-flaw": "blocking", "failed": "blocking"}.get(value)

    for r in reviews:
        ref = r.get("report-ref")
        sha = r.get("report-sha256")
        if not (ref and sha and r.get("reviewer-role-id") and r.get("reviewer-run-id")
                and r.get("lens") and r.get("verdict")):
            return False
        ap = ref if os.path.isabs(ref) else os.path.join(ROOT, ref)
        if not os.path.exists(ap):
            return False
        if _sha256_of(ap) != sha:
            return False
        if trusted_panel:
            try:
                import artifact_contract as AC
            except Exception:
                return False
            submitted = next((item for item in arts or []
                              if item.get("artifact-kind") == "design-lens-review"
                              and _report_id_of(item) == r.get("report-id")
                              and item.get("artifact-sha256") == sha
                              and os.path.abspath(AC.absolute_path(item.get("path"))) == os.path.abspath(ap)), None)
            if not submitted:
                return False
        # The panel summary cannot misrepresent the hash-bound source review.
        try:
            envelope = _load_yaml(ap)
            source = (envelope.get("payload")
                      if envelope.get("report-type") == "workflow-artifact"
                      and isinstance(envelope.get("payload"), dict) else envelope)
        except Exception:
            return False
        for key in ("lens", "reviewer-role-id", "reviewer-run-id"):
            if str(source.get(key) or "") != str(r.get(key) or ""):
                return False
        if trusted_panel:
            ident = envelope.get("identity") if isinstance(envelope.get("identity"), dict) else {}
            payload_source = source
            if str(ident.get("producer-role-id") or "").upper() != str(r.get("reviewer-role-id") or "").upper():
                return False
            if (payload_source.get("target-prototype-id"), payload_source.get("target-prototype-sha256")) != (
                    _report_id_of(winner), winner.get("artifact-sha256")):
                return False
        if _normalized_verdict(source.get("verdict")) != _normalized_verdict(r.get("verdict")):
            return False
        # A pass synthesis requires every lens to pass. Concerns are revision work,
        # never non-blocking prose that the synthesis lead may silently override.
        if _normalized_verdict(r.get("verdict")) != "pass":
            return False
        findings = source.get("findings") or []
        for finding in findings if isinstance(findings, list) else []:
            if not isinstance(finding, dict):
                continue
            severity = str(finding.get("severity") or "").strip().lower()
            if severity in ("blocking", "critical") or finding.get("blocking") is True:
                return False
    ds = _active_artifact(led, arts, "direction-set", "direction-set-report-id")
    if not ds:
        return False  # 직물 검증 불가 -> fail closed
    ds_content = _artifact_content(ds, "direction-set") or {}
    producer_runs = {d.get("producer-run-id") for d in (ds_content.get("directions") or []) if d.get("producer-run-id")}
    if not producer_runs:
        return False  # 대조할 producer 없음(전부 None 포함) -> fail closed
    if any(r.get("reviewer-run-id") in producer_runs for r in reviews):
        return False
    syn = panel.get("synthesis") or {}
    dissent = syn.get("unresolved-dissent")
    return (syn.get("verdict") == "pass"
            and str(syn.get("role-id", "")).upper() == "DES-DIRECTOR"
            and isinstance(dissent, list) and not dissent)


def _panel_verdict(led, arts):
    a = _active_artifact(led, arts, "design-review-panel", "review-panel-report-id")
    panel = _artifact_content(a, "design-review-panel")
    return (panel or {}).get("synthesis", {}).get("verdict") if a else None


def _approval_brief_matches(approved_doc, current_sha):
    """staleness helper. current_sha 가 없으면(파일 미해석/부재) fail-closed(False) —
    '측정 불가'를 통과로 치지 않는다. 있으면 승인문서에 박제된 direction-input-brief-sha256 과
    정확히 일치할 때만 stale 아님(True)."""
    return bool(current_sha) and approved_doc.get("direction-input-brief-sha256") == current_sha


def _dd_ws_path(path):
    """report-ref/selected-direction-ref/winner-prototype-ref 를 **workspace root**(W.work_root())
    기준 상대경로로 해석한다(이 참조들은 org-os SSOT 가 아니라 워크스페이스 산출물이므로
    _current_input_brief_sha 가 쓰는 레포 ROOT 기준과 다르다). 절대경로면 그대로. 실존하지
    않으면 None(호출부가 fail-closed 처리)."""
    if not path:
        return None
    if os.path.isabs(path):
        ap = path
    else:
        try:
            import _workspace as W  # noqa: E402
            ap = os.path.join(W.work_root(), path)
        except Exception:
            return None
    return ap if os.path.exists(ap) else None


def _sha256_of(path):
    try:
        return hashlib.sha256(open(path, "rb").read()).hexdigest()
    except Exception:
        return None


def _product_decision_current(parent_wf, pd_id):
    """product-decision(pd_id) 이 parent_wf 에서 여전히 CURRENT(비-superseded) accepted 인가.

    1) `_al_accepted_ids(parent_wf)` 로 **정확히 이 id** 가 accepted 된 적 있는지 확인한다
       (init_ledger 의 기존 위조방지 패턴 재사용 — 부모가 '아무거나' accepted 했다는 사실만으로
       통과하지 않는다).
    2) acceptance_log 이벤트 중 이 pd_id 를 `supersedes-report-id` 로 지목한 이벤트가(그 부모
       스코프에서) 하나라도 있으면, 더 새로운 결정이 이를 대체했다는 뜻이므로 False(supersede
       를 무시하면 폐기된 옛 product-decision 으로 direction-approval 이 영구히 유효해진다)."""
    if not pd_id:
        return False
    try:
        parent_arts = (_load_ledger_safe(parent_wf).get("artifacts") or [])
        if pd_id not in _al_accepted_ids(parent_wf, parent_arts):
            return False
        import acceptance_log as AL  # noqa: E402
        for ev in AL.read_events():
            if ev.get("workflow-id") not in (None, parent_wf):
                continue
            if ev.get("supersedes-report-id") == pd_id:
                return False
        return True
    except Exception:
        return False


def _direction_approval_receipt_ok(child_wf, report_ref, report_sha):
    """acceptance_log 에 이 approved-direction report 를 accepted 로 정확히 바인딩한 이벤트가
    있는가. workflow-id 는 child_wf 와 **EXACT** 일치해야 한다(None 은 불허 — 느슨한 매칭으로
    아무 child 나 편승하는 것을 차단, receipt 자기신고 위조 방지). report-sha256 도 정확히
    일치해야 한다(다른 리포트의 accepted 이벤트를 재사용하는 것을 차단).

    Important fix(리뷰, `_venture_decision_receipt_ok` 와 동일 패턴): report-sha256 **필드값**만
    대조하면, 이벤트가 실존하지 않거나 무관한 report-id 를 달고도 그 필드에 approved-direction 의
    해시를 그대로 베껴 자기신고할 수 있다(값은 맞는데 가리키는 파일은 다른 상황). 그래서 이벤트가
    가리키는 report-id 를 실제 completion-records 경로로 재해석(`AL._resolve_report_path`)해
    파일이 실존하고 그 **라이브 sha256**도 report_sha 와 일치할 때만 인정한다 — accepted 로
    표시된 그 파일의 실제 내용이 승인된 direction 의 해시로 귀결돼야 진짜 바인딩이다."""
    try:
        import acceptance_log as AL  # noqa: E402
        for ev in AL.read_events():
            if ev.get("decision") != "accepted":
                continue
            if ev.get("workflow-id") != child_wf:  # None 불허 — 정확 일치만 인정
                continue
            if ev.get("report-sha256") != report_sha:
                continue
            rid = ev.get("accepted-report-id") or ev.get("report-id")
            if not rid:
                continue
            path = AL._resolve_report_path(rid, child_wf)
            if not path or not os.path.exists(path):
                continue
            if _sha256_of(path) != report_sha:
                continue
            if not AL.is_effectively_accepted(child_wf, rid, report_sha):
                continue
            return True
        return False
    except Exception:
        return False


def _has_direction_approval(wf, led):
    """§7 exact 8점 검증. fail-open 없음 — 실패 경로는 전부 False 를 반환한다.

    호출 모양(shape) 2종을 모두 지지한다. `_facts(wf, led)` 는 **평가 대상 워크플로 자신의**
    wf/led 를 그대로 이 함수에 넘기는데, 이 사실이 쓰이는 두 자리의 '자신'이 다르다:
      (a) 부모(cascade) 자신의 facts 평가 — 예: Phase F `check-direction-approved --workflow
          <parent>`, 상위 cascade 가 /design-system 진입 전 게이트로 확인할 때. 이때 led 에
          `design-direction-approval` 링크가 직접 있다.
      (b) design-direction **자식** 자신의 facts 평가 — 자식의 `design-direction-finalize ->
          design-direction-approved` 전이 자체가 이 사실을 조건으로 요구하므로(approved-direction
          -valid/approval-receipt-bound/parent-approval-link-recorded 3종 모두 이 사실 하나에서
          파생), `guard/transition --workflow <child>`가 호출하는 `_facts(child, child_led)` 도
          이 함수를 반드시 통과시켜야 한다. 이땐 led 에 `parent-workflow-id`만 있고 승인 링크
          자체는 **부모** 원장에 있으므로, 부모로 거슬러 올라가 그 링크가 정확히 이 자식(wf)을
          가리키는지부터 확인한다(가리키지 않으면 위조/오배선 -> False).
    두 경우 모두 아니면(예: 평범한 cascade 원장) 이 게이트와 무관 -> False.

    Critical fix(리뷰): (b) 모드는 자식의 `finalize -> approved` 전이 자체를 평가하는 도중에
    호출되므로, 평가 시점의 child stage 는 아직 `design-direction-finalize` 다 — 여기서
    `design-direction-approved` 를 요구하면(3번 체크) 그 전이가 만드는 바로 그 stage 를
    전이 성립 조건으로 요구하는 셈이라 전이가 영원히 발동할 수 없다(교착). 그래서 3번 체크는
    `parent_mode` 로 분기한다: (a) parent-shape 는 자식이 이미 완전히 종료(approved)됐음을
    요구하고, (b) child-shape 는 전이가 떠나는 stage(finalize) 도 허용한다. 다른 체크(1,2,4~8)는
    두 모드 모두 동일하게(완화 없이) 수행한다."""
    approval = led.get("design-direction-approval")
    if approval:
        parent_wf = wf
        parent_mode = True
    elif led.get("parent-workflow-id"):
        parent_wf = led.get("parent-workflow-id")
        parent_path = _ledger_path(parent_wf, create=False)
        if not parent_path or not os.path.exists(parent_path):
            return False
        parent_led = _load_ledger_safe(parent_wf)
        approval = parent_led.get("design-direction-approval") or {}
        if approval.get("child-workflow-id") != wf:
            return False  # 부모 링크가 이 자식을 가리키지 않음
        parent_mode = False
    else:
        return False

    report_ref = approval.get("report-ref")
    report_sha = approval.get("report-sha256")
    child = approval.get("child-workflow-id")
    if not (report_ref and report_sha and child):
        return False  # 1) 부모 원장 approval 링크(report-ref+report-sha256+child-workflow-id) 불완전

    # 2) child 실존(파일 존재로 직접 확인 — _load_ledger_safe 는 부재도 기본원장으로 감춘다) + 관계
    child_path = _ledger_path(child, create=False)
    if not child_path or not os.path.exists(child_path):
        return False
    child_led = _load_ledger_safe(child)
    if child_led.get("parent-workflow-id") != parent_wf:
        return False

    # 3) child stage — mode-dependent(Critical fix, 교착 해소): parent-shape(a)는 자식이 이미
    # 종료(approved)됐음을 요구하고, child-shape(b)는 이 fact 자체가 만들어내는 stage(approved)를
    # 평가 시점(아직 finalize)에 요구하면 전이가 영원히 발동 못 하므로 finalize 도 허용한다.
    # 어느 모드든 그 외 stage(예: critique/prototype 등으로의 역행·오염)는 여전히 거부(fail-closed).
    if parent_mode:
        if child_led.get("stage") != "design-direction-approved":
            return False
    else:
        if child_led.get("stage") not in ("design-direction-finalize", "design-direction-approved"):
            return False

    # 4) approved-direction report 파일 실존 + hash 일치(workspace 상대경로)
    ap = _dd_ws_path(report_ref)
    if not ap:
        return False
    if _sha256_of(ap) != report_sha:
        return False
    raw = _load_yaml(ap)
    if raw.get("report-type") == "workflow-artifact" and isinstance(raw.get("payload"), dict):
        raw = raw["payload"]
    doc = raw.get("approved-direction") if isinstance(raw.get("approved-direction"), dict) else raw
    if not isinstance(doc, dict):
        return False

    # 5) doc 의 parent/child 가 정확히 이 쌍과 일치
    if doc.get("parent-workflow-id") != parent_wf or doc.get("child-workflow-id") != child:
        return False

    # 6) product-decision-id 일치 + 부모에서 현재(비-superseded) accepted
    pd_id = doc.get("product-decision-id")
    if not pd_id or pd_id != child_led.get("product-decision-id"):
        return False
    if not _product_decision_current(parent_wf, pd_id):
        return False

    # 7) staleness: 자식의 현재(live) input-brief hash == 승인문서에 박제된 hash
    current_sha = _current_input_brief_sha(child_led)
    if not _approval_brief_matches(doc, current_sha):
        return False

    # 8) 참조 아티팩트(selected-direction/winner-prototype) 실존+hash 일치 + acceptance receipt 정확 바인딩
    sd_ap = _dd_ws_path(doc.get("selected-direction-ref"))
    if not sd_ap or _sha256_of(sd_ap) != doc.get("selected-direction-sha256"):
        return False
    wp_ap = _dd_ws_path(doc.get("winner-prototype-ref"))
    if not wp_ap or _sha256_of(wp_ap) != doc.get("winner-prototype-sha256"):
        return False
    if not _direction_approval_receipt_ok(child, report_ref, report_sha):
        return False

    return True


def register_direction_approval(parent, child, report, report_sha256):
    """trusted CLI 백엔드 — 전 검증 통과 후에만 부모 원장에 `design-direction-approval` 을
    원자적으로 기록한다(guard_tools 가 직접 YAML 편집을 막으므로 이 CLI 가 유일한 등록 경로).
    검증 실패 시 ValueError 를 던진다(원장은 건드리지 않는다) — 호출측(CLI)이 non-zero exit 로
    변환한다."""
    child_path = _ledger_path(child, create=False)
    if not child_path or not os.path.exists(child_path):
        raise ValueError(f"child workflow '{child}' 원장 없음")
    child_led = _load_ledger_safe(child)
    if child_led.get("parent-workflow-id") != parent:
        raise ValueError(f"child '{child}' 가 parent '{parent}' 의 자식이 아님")
    if child_led.get("stage") not in ("design-direction-finalize", "design-direction-approved"):
        raise ValueError(f"child stage '{child_led.get('stage')}' 는 finalize/approved 가 아님")
    ap = _dd_ws_path(report)
    if not ap:
        raise ValueError(f"report 경로 실존하지 않음: {report}")
    if _sha256_of(ap) != report_sha256:
        raise ValueError("report sha256 불일치")
    artifact, artifact_error = _artifact_by_snapshot(child, ap)
    if not artifact or artifact.get("artifact-kind") != "approved-direction":
        raise ValueError(artifact_error or "submit-artifact된 approved-direction exact revision이 아니다")
    parent_path = _ledger_path(parent, create=False)
    if not parent_path or not os.path.exists(parent_path):
        raise ValueError(f"parent workflow '{parent}' 원장 없음")
    parent_led = _load_ledger_safe(parent)
    existing = parent_led.get("design-direction-approval") or {}
    existing_child = existing.get("child-workflow-id")
    if existing_child and existing_child != child:
        raise ValueError(f"기존 active approval(child={existing_child}) 과 충돌")
    event = {
        "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
        "event-type": "direction-approval-registered", "workflow-id": parent,
        "child-workflow-id": child, "report-ref": report,
        "report-sha256": report_sha256, "actor": "OPS-ORCH", "effective-at": _now(),
    }
    committed, error = _atomic_event_transaction(parent, workflow_event=event)
    if not committed:
        raise ValueError(error or "부모 direction approval event 기록 실패")
    return True


def _present(led, arts, type_aliases, flag_key):
    # Gate presence is derived only from trusted artifact events.  Explicit
    # ledger flags were the old bypass (`facts.*-present: true`).
    return _artifact_present(arts, set(type_aliases))


def _workload_profile(arts):
    profile = {}
    for artifact in arts or []:
        if not isinstance(artifact, dict):
            continue
        if artifact.get("artifact-kind") in ("decision-brief", "workload-profile"):
            value = artifact.get("workload-profile")
            if isinstance(value, dict):
                profile = value
    return profile


def _competitive_market_grounding_required(wf):
    """Public/new/major work needs market grounding during /ground, not later design research."""
    profile = _workload_profile(_trusted_artifacts(wf))
    return (
        profile.get("surface-archetype") == "public-website"
        or profile.get("experience-change") in ("new-product", "major-redesign")
    )


def _role_agent_id(value):
    return str(value or "").strip().upper().replace("_", "-")


def _grounding_lens_coverage(wf, led, arts):
    """Verify discovery contributions from immutable reports and context packages.

    Grounding-package summaries are projections only.  Every covered lens is
    re-derived from the exact report revision, its registered producer role and
    the immutable context package that assigned the lens.
    """
    package = _latest_artifact_of_kind(arts, "grounding-package")
    body = _artifact_content(package, "grounding-package") if package else None
    if not isinstance(body, dict):
        return False, "grounding-package 없음"
    contributions = body.get("source-contributions") or []
    if not isinstance(contributions, list):
        return False, "grounding-package.source-contributions array 필요"

    try:
        import acceptance_log as AL
        import artifact_contract as AC
        from orgos.planning.lens_policy import (
            divergent_policy,
            family_for_role,
            required_lenses,
            role_can_carry_lens,
        )
    except Exception as exc:
        return False, f"grounding lens policy 로드 실패(fail-closed): {exc}"

    trusted_by_revision = {
        (str(item.get("artifact-id") or ""), str(item.get("artifact-sha256") or "")): item
        for item in arts if isinstance(item, dict)
    }
    errors = []
    seen_reports, seen_packages, seen_runs = set(), set(), set()
    covered = set()
    contrarian_reports = []
    resolved = {}
    package_epoch = package.get("stage-epoch-id")
    package_producer_family, _ = family_for_role(package.get("producer-role-id"))
    brief = _latest_artifact_of_kind(arts, "decision-brief")
    brief_body = _artifact_content(brief, "decision-brief") or {}
    candidate_families = {str(value).upper()
                          for value in brief_body.get("candidate-families", []) or []}
    if not candidate_families:
        errors.append("현재 Decision Brief의 non-empty candidate-families 없음")

    for index, ref in enumerate(contributions):
        prefix = f"source-contributions[{index}]"
        if not isinstance(ref, dict):
            errors.append(f"{prefix}: object 필요")
            continue
        report_id = str(ref.get("report-id") or "")
        report_sha = str(ref.get("report-sha256") or "")
        report_key = (report_id, report_sha)
        context_key = (str(ref.get("context-package-ref") or ""),
                       str(ref.get("context-package-sha256") or ""))
        run_id = str(ref.get("producer-run-id") or "")
        lens = str(ref.get("assigned-lens") or "").upper()
        if report_key in seen_reports:
            errors.append(f"{prefix}: report id/SHA 중복")
        if context_key in seen_packages:
            errors.append(f"{prefix}: context package ref/SHA 중복")
        if run_id in seen_runs:
            errors.append(f"{prefix}: producer-run-id 중복")
        seen_reports.add(report_key)
        seen_packages.add(context_key)
        seen_runs.add(run_id)

        artifact = trusted_by_revision.get(report_key)
        if not artifact:
            errors.append(f"{prefix}: 현재 workflow trusted report id/SHA 불일치")
            continue
        if artifact.get("artifact-kind") not in ("grounding-contribution", "competitive-market-grounding"):
            errors.append(f"{prefix}: grounding contribution artifact-kind 아님")
            continue
        if artifact.get("workflow-id") != wf or artifact.get("stage") != "discovery":
            errors.append(f"{prefix}: 동일 workflow/discovery stage가 아님")
        if package_epoch and artifact.get("stage-epoch-id") != package_epoch:
            errors.append(f"{prefix}: 현재 discovery 실행 epoch와 불일치(stale)")
        if not _same_snapshot_ref(ref.get("report-ref"), report_sha, artifact):
            errors.append(f"{prefix}: report-ref/live SHA exact binding 불일치")

        producer = str(artifact.get("producer-role-id") or "").upper()
        if producer not in _role_registry():
            errors.append(f"{prefix}: 미등록 producer role {producer!r}")
        if str(ref.get("producer-role-id") or "").upper() != producer:
            errors.append(f"{prefix}: producer-role-id가 report identity와 불일치")
        if not role_can_carry_lens(producer, lens):
            errors.append(f"{prefix}: role {producer}는 registry상 {lens}를 carry할 수 없음")
        producer_family, _producer_family_doc = family_for_role(producer)
        if producer_family not in candidate_families:
            errors.append(f"{prefix}: producer family {producer_family}가 Decision Brief candidate-families 밖")

        contribution = _artifact_content(artifact, artifact.get("artifact-kind")) or {}
        if (str(contribution.get("assigned-lens") or "").upper() != lens
                or str(contribution.get("producer-run-id") or "") != run_id):
            errors.append(f"{prefix}: report payload lens/run이 source ref와 불일치")
        if (str(contribution.get("context-package-ref") or "") != context_key[0]
                or str(contribution.get("context-package-sha256") or "") != context_key[1]):
            errors.append(f"{prefix}: report payload context package binding 불일치")

        context_path = AC.absolute_path(context_key[0])
        if not context_path or not os.path.isfile(context_path):
            errors.append(f"{prefix}: context package 파일 없음")
        elif AC.sha256_file(context_path) != context_key[1]:
            errors.append(f"{prefix}: context package live SHA 불일치")
        else:
            context = _load_yaml(context_path)
            if context.get("workflow-id") != wf:
                errors.append(f"{prefix}: context package workflow-id 불일치")
            if context.get("mode") != "divergent":
                errors.append(f"{prefix}: context package mode=divergent 필요")
            if str(context.get("tier") or "").lower() != str(led.get("tier") or "").lower():
                errors.append(f"{prefix}: context package tier 불일치")
            if str(context.get("assigned-lens") or "").upper() != lens:
                errors.append(f"{prefix}: context package assigned-lens 불일치")
            if _role_agent_id(context.get("target-role-agent")) != producer:
                errors.append(f"{prefix}: context package target-role-agent와 producer 불일치")

        decision = AL.effective_decision(wf, report_id, report_sha)
        if decision in ("Superseded", "ChangesRequested", "Blocked"):
            errors.append(f"{prefix}: stale/rejected contribution({decision})")
        covered.add(lens)
        resolved[report_id] = artifact
        if lens == "LENS-CONTRARIAN":
            contrarian_reports.append(report_id)
            contrarian_family, _family = family_for_role(producer)
            if contrarian_family == package_producer_family:
                errors.append(f"{prefix}: contrarian producer family가 grounding author family와 같음")

    tier = str(led.get("tier") or DEFAULT_TIER).lower()
    policy = divergent_policy(tier)
    minimum = policy.get("min-distinct-lenses")
    required = required_lenses(sorted(candidate_families), tier=tier, mode="divergent")
    if minimum == "all-relevant":
        if not required or not required.issubset(covered):
            errors.append(f"heavy all-relevant 렌즈 누락: {sorted(required - covered)}")
    elif len(covered) < int(minimum or 0):
        errors.append(f"distinct lens 부족: {len(covered)} < tier {tier} 최소 {minimum}")
    if policy.get("contrarian-required") and len(contrarian_reports) != 1:
        errors.append("standard/heavy는 정확히 하나의 LENS-CONTRARIAN contribution 필요")

    declared = body.get("lens-coverage") or {}
    if declared.get("required-min") != minimum:
        errors.append("lens-coverage.required-min이 governance tier 정책과 불일치")
    if set(declared.get("covered") or []) != covered:
        errors.append("lens-coverage.covered가 검증된 source contribution 렌즈와 불일치")
    expected_contrarian = contrarian_reports[0] if len(contrarian_reports) == 1 else None
    if declared.get("contrarian-report-id") != expected_contrarian:
        errors.append("lens-coverage.contrarian-report-id exact binding 불일치")

    workload = _workload_profile(arts)
    required_capabilities = {str(value).strip().lower()
                             for value in workload.get("required-capabilities", []) or []}
    if ("competitive-intelligence" in required_capabilities
            and not any(str(item.get("producer-role-id") or "").upper() == "GTM-CI"
                        for item in resolved.values())):
        errors.append("required-capability competitive-intelligence는 실제 GTM-CI source contribution이 필요")

    if _competitive_market_grounding_required(wf):
        market_ref = body.get("competitive-market-grounding-ref") or {}
        market_key = (str(market_ref.get("report-id") or ""),
                      str(market_ref.get("report-sha256") or ""))
        market = trusted_by_revision.get(market_key)
        if (not market or market.get("artifact-kind") != "competitive-market-grounding"
                or str(market.get("producer-role-id") or "").upper() != "GTM-CI"):
            errors.append("공개형/신규/대규모 작업은 GTM-CI competitive-market-grounding exact report 필요")
        elif (market_key not in seen_reports
              or not _same_snapshot_ref(market_ref.get("report-ref"), market_key[1], market)):
            errors.append("competitive-market-grounding은 source-contributions에 포함되고 ref/SHA가 일치해야 함")

    return not errors, "; ".join(errors[:12]) if errors else None


def _quality_panel_unmet(tier, arts, events):
    """Return missing independent reviewer requirements for the trusted workload risk."""
    actors = {str(event.get("actor") or "").upper() for event in events}
    categories = {category for event in events for category in (event.get("check-categories") or [])}
    profile = _workload_profile(arts)
    surfaces = profile.get("surfaces") if isinstance(profile.get("surfaces"), dict) else {}
    risk = profile.get("risk") if isinstance(profile.get("risk"), dict) else {}
    unmet = []
    if not actors.intersection({"QA", "EXEC-VPENG", "HUMAN-001"}):
        unmet.append("independent general quality auditor(QA/EXEC-VPENG/HUMAN-001)")
    if risk.get("security-bearing"):
        if not any(actor.startswith("SEC-") for actor in actors):
            unmet.append("independent security auditor")
        if "security" not in categories:
            unmet.append("security check category")
    if risk.get("data-migration") or surfaces.get("persistence"):
        if "ARCH-DATA" not in actors:
            unmet.append("independent data auditor(ARCH-DATA)")
        if "data-quality" not in categories:
            unmet.append("data-quality check category")
    if risk.get("privacy") and "privacy" not in categories:
        unmet.append("privacy check category")
    if risk.get("slo-impact") and "reliability" not in categories:
        unmet.append("reliability check category")
    if tier == "heavy" and len(actors) < 3:
        unmet.append("heavy tier minimum 3 independent verifiers")
    return unmet


def _required_bundle_kinds(arts, bundle_name):
    bundle = (load_contracts().get("artifact-bundles", {}) or {}).get(bundle_name, {}) or {}
    required = set(bundle.get("always") or [])
    profile = _workload_profile(arts)
    surfaces = profile.get("surfaces") if isinstance(profile.get("surfaces"), dict) else {}
    risks = profile.get("risk") if isinstance(profile.get("risk"), dict) else {}
    signals = {
        "ui": bool(surfaces.get("ui")),
        "public-api": bool(surfaces.get("public-api")),
        "persistence": bool(surfaces.get("persistence")),
        "security-bearing": bool(risks.get("security-bearing")),
        "product-feature": profile.get("product-feature", True),
    }
    for condition in bundle.get("conditional") or []:
        if isinstance(condition, dict) and signals.get(condition.get("when")):
            required.update(condition.get("require") or [])
    return required


def _bundle_accepted(wf, arts, bundle_name):
    accepted = _accepted_design_types(wf, arts)
    required = _required_bundle_kinds(arts, bundle_name)
    if not required or not required.issubset(accepted):
        return False
    latest = {}
    for artifact in arts or []:
        if isinstance(artifact, dict) and artifact.get("artifact-kind"):
            latest[artifact.get("artifact-kind")] = artifact
    components = [latest.get(kind) for kind in required]
    if any(not component for component in components):
        return False
    bases = {(component.get("basis-artifact-id"), component.get("basis-artifact-sha256"))
             for component in components}
    if len(bases) != 1 or next(iter(bases))[0] in (None, "") or next(iter(bases))[1] in (None, ""):
        return False
    target_kind = {
        "design-bundle": "executive-decision-packet",
        "spec-bundle": "overall-design",
    }.get(bundle_name)
    target = latest.get(target_kind)
    if not target:
        return False
    target_identity = (_report_id_of(target), target.get("artifact-sha256") or target.get("report-sha256"))
    if next(iter(bases)) != target_identity:
        return False
    if not _compatibility_reviews_ok(wf, arts, required, latest):
        return False
    try:
        import acceptance_log as AL  # noqa: E402
        return AL.is_effectively_accepted(wf, target_identity[0], target_identity[1])
    except Exception:
        return False


def _compatibility_reviews_ok(wf, arts, required, latest=None):
    """Require an exact typed compatibility verdict for every active contract pair."""
    contracts = load_contracts().get("compatibility-contracts", []) or []
    active = [item for item in contracts
              if item.get("left") in required and item.get("right") in required]
    if not active:
        return True
    if latest is None:
        latest = {}
        for artifact in arts or []:
            if isinstance(artifact, dict) and artifact.get("artifact-kind"):
                latest[artifact.get("artifact-kind")] = artifact
    reviews = [artifact for artifact in arts or []
               if isinstance(artifact, dict) and artifact.get("artifact-kind") == "compatibility-review"]
    for contract in active:
        left = latest.get(contract.get("left"))
        right = latest.get(contract.get("right"))
        if not left or not right:
            return False
        expected = {
            contract.get("left"): (_report_id_of(left), left.get("artifact-sha256")),
            contract.get("right"): (_report_id_of(right), right.get("artifact-sha256")),
        }
        matched = False
        for review in reversed(reviews):
            content = _artifact_content(review, "compatibility-review") or {}
            endpoints = {}
            for side in ("left", "right"):
                value = content.get(side) or {}
                endpoints[value.get("artifact-kind")] = (
                    value.get("artifact-id"), value.get("artifact-sha256"))
            reviewer = str(content.get("reviewer-role-id") or "")
            if (endpoints == expected and content.get("verdict") == "Passed"
                    and set(content.get("dimensions") or []) >= set(contract.get("dimensions") or [])
                    and reviewer == str(review.get("producer-role-id") or "")
                    and reviewer not in {str(left.get("producer-role-id") or ""),
                                         str(right.get("producer-role-id") or "")}):
                matched = True
                break
        if not matched:
            return False
    return True


def _must_read_unmet(wf, led, arts):
    """spec→build gate derived from the canonical conditional bundles.

    The former family map used undeclared aliases such as ``design-system`` and
    ``service-boundary``.  Those values could never be produced through the
    trusted artifact API and also created a second workload classifier.  The
    accepted design/spec kinds now come solely from ``workflow-contracts`` and
    its trusted workload-profile conditions.
    """
    required = (_required_bundle_kinds(arts, "design-bundle")
                | _required_bundle_kinds(arts, "spec-bundle"))
    accepted = _accepted_design_types(wf, arts)
    unmet = []
    for kind in required:
        if kind not in accepted:
            unmet.append(kind)
        elif kind == "ui-design" and not _has_preview_receipt(wf):
            # UI design은 acceptance event가 있어도 실제 preview_ui 렌더 게이트
            # receipt(exit 0)가 evidence-ledger 에 있어야 충족한다 — 렌더된 적 없는(산문만) design-system
            # 을 Accepted 로 위장해 프론트 BUILD 를 여는 것을 차단(docs-but-no-pixels 구멍 봉인).
            unmet.append("ui-design (preview_ui 렌더 게이트 receipt 없음 — 실제 렌더·품질검증 필요)")
    return sorted(unmet)


def _method_handoff_unmet(wf, led, ctx):
    """P3-B: 전이 시 지정된 consumer profile 들의 required-inputs handoff 미충족이면 True(하드 게이트).

    opt-in: 전이 규칙/ctx 가 `handoff-check: [{role, method}]` 를 줄 때만 검사한다(무지정 → False,
    무회귀). both-active 엣지 위반만 차단(method_contracts.handoff_violations), 한쪽 draft 는 debt.
    수락 근사 = acceptance_log 의 from-role 최신 accepted 존재(Phase5 golden 에서 정밀화)."""
    checks = (ctx or {}).get("handoff-check") if isinstance(ctx, dict) else None
    if not checks:
        return False
    try:
        import method_contracts as _MC
        import acceptance_log as _AL
    except Exception:  # noqa: BLE001
        return False

    def _accepted(edge):
        frm = (edge.get("from") or {}).get("role-id")
        kind = edge.get("artifact-type")
        try:
            return bool(_AL.latest_accepted_artifact(
                wf, producer_role=frm, artifact_kind=kind))
        except Exception:  # noqa: BLE001
            return False

    unmet = False
    for c in checks:
        try:
            errs, debts = _MC.handoff_violations(c.get("role"), c.get("method"),
                                                 present=_accepted, accepted=_accepted, phase="transition")
        except Exception as exc:  # noqa: BLE001
            if led.get("tier") in ("standard", "heavy"):
                errs, debts = [f"method handoff policy 평가 실패(fail-closed): {exc}"], []
            else:
                errs, debts = [], []
        for d in debts:  # T4.4: draft 엣지 부채 기록(fold 로 중복 무해). 정보용.
            try:
                _MC.record_debt(d)
            except Exception:  # noqa: BLE001
                pass
        if errs:
            unmet = True
    return unmet


def _facts(wf, led, ctx=None):
    """원장 + acceptance_log + collaboration-map 에서 조건 평가용 사실 dict 를 파생한다.

    Protected gate facts always come from canonical events and live immutable snapshots. ``facts``/ctx
    overrides are retained only for non-security operational hints and cannot override protected facts.
    """
    arts = led.get("artifacts") or []
    explicit = dict(led.get("facts") or {})
    if isinstance(ctx, dict):
        explicit.update(ctx.get("facts") or {})

    f = {}
    f["tier"] = led.get("tier") or DEFAULT_TIER

    accepted_types = _accepted_design_types(wf, arts)

    # 발산 option-set은 grounding-package의 검증된 본문에서만 파생한다.
    opts = []
    for artifact in arts:
        if isinstance(artifact, dict) and artifact.get("artifact-kind") == "grounding-package":
            opts = artifact.get("option-set") or []
    f["option_set_count"] = len(opts) if isinstance(opts, list) else 0

    # blocker
    f["blocker_open"] = bool(led.get("blocker-open"))

    # These fields are projected exclusively from workflow events.
    f["quality_gate_status"] = led.get("quality_gate_status")
    f["quality_gate_failed"] = led.get("quality_gate_status") == "Failed"
    f["release_acceptance_status"] = led.get("release_acceptance_status")
    f["unresolved_critical_risks"] = bool(
        led.get("unresolved_critical_risks") or led.get("unresolved-critical-risks")
    )
    f["human_gate_approved"] = bool(_human_gate_satisfied(wf, led.get("stage")))
    f["evidence_grade"] = None
    latest_packet = next((artifact for artifact in reversed(arts)
                          if isinstance(artifact, dict)
                          and artifact.get("artifact-kind") == "executive-decision-packet"), None)
    if latest_packet and "executive-decision-packet" in accepted_types:
        grade = latest_packet.get("max-evidence-grade")
        if grade in _EGRADE:
            f["evidence_grade"] = grade

    # 산출물 존재 플래그(design-type 또는 원장 플래그)
    f["decision_brief_present"] = _present(led, arts, ["decision-brief"], "decision-brief-present")
    f["workload_profile_present"] = _present(led, arts, ["workload-profile"], "workload-profile-present")

    # --- venture-bootstrap facts (P1) ---
    f["founder_context_present"] = _founder_context_filled()
    # caller가 workflow.yaml/ctx에 리스트를 자기신고해도 게이트 사실이 되지 않는다. 전역 artifact
    # event 원장에서 live id+sha가 검증된 snapshot만 읽고, 같은 payload.id 복제도 한 개로 센다.
    f["opportunity_cluster_count"] = _distinct_opportunity_cluster_count(_trusted_artifacts(wf))
    f["venture_options_validated"] = bool(accepted_types & {"venture-validation", "venture-option"})
    f["venture_decision_accepted"] = bool(accepted_types & {"venture-decision", "ExecutiveDecisionPacket", "decision-packet"})
    f["human_acceptance_receipt_present"] = _venture_decision_receipt_ok(wf)
    _st = _company_ctx_status()
    f["company_context_provisional_committed"] = _st in ("provisional", "operating")
    f["company_context_lint_passed"] = _company_ctx_lint_ok()
    f["company_context_artifact_recorded"] = _company_ctx_artifact_recorded(arts)
    # A repository-scoped product change must not be forced through company formation.
    # The exception is narrow and trusted: only the canonical workload-profile may
    # declare context-scope=project, and it must also declare product-feature=true.
    # This opens delivery against the named project; it does not create company facts,
    # market validation, or an operating company context.
    workload = next((artifact.get("workload-profile") for artifact in reversed(arts)
                     if isinstance(artifact, dict)
                     and artifact.get("artifact-kind") == "workload-profile"
                     and isinstance(artifact.get("workload-profile"), dict)), None)
    f["project_context_scoped"] = bool(
        workload and workload.get("context-scope") == "project"
        and workload.get("product-feature") is True)
    f["company_context_ready"] = bool(
        (f.get("company_context_provisional_committed") or f.get("project_context_scoped"))
        and not f.get("blocker_open"))

    # --- experience-foundation facts (front-of-funnel experience grounding) ---
    f["experience_parent_binding_present"] = bool(
        led.get("plan") == "experience-foundation"
        and led.get("parent-workflow-id") and led.get("product-decision-id")
        and _product_decision_current(led.get("parent-workflow-id"), led.get("product-decision-id")))
    f["competitive_benchmark_accepted"] = bool(
        _latest_accepted_artifact(wf, arts, "competitive-experience-benchmark"))
    _strategy = _latest_accepted_artifact(wf, arts, "experience-strategy")
    f["experience_strategy_accepted"] = bool(
        _strategy and (_artifact_content(_strategy, "experience-strategy") or {}).get("decision") == "proceed")
    _experience_parent_id = led.get("parent-workflow-id") if led.get("plan") == "experience-foundation" else None
    _experience_pd = led.get("product-decision-id") if led.get("plan") == "experience-foundation" else None
    f["experience_technical_feasibility_accepted"] = bool(
        _strategy and _experience_parent_id and _accepted_experience_feasibility(
            wf, _experience_parent_id, _experience_pd, _strategy, "experience-technical-feasibility"))
    f["experience_operational_feasibility_accepted"] = bool(
        _strategy and _experience_parent_id and _accepted_experience_feasibility(
            wf, _experience_parent_id, _experience_pd, _strategy, "experience-operational-feasibility"))
    f["experience_blueprint_accepted"] = bool(
        _latest_accepted_artifact(wf, arts, "experience-blueprint"))
    f["wireframe_set_accepted"] = bool(_latest_accepted_artifact(wf, arts, "wireframe-set"))
    f["experience_foundation_link_recorded"] = bool(_has_experience_foundation(wf, led))
    _experience_parent = (led.get("parent-workflow-id")
                          if led.get("plan") == "design-direction" else wf)
    _experience_parent_led = (_load_ledger_safe(_experience_parent)
                              if _experience_parent != wf else led)
    f["experience_foundation_required"] = _experience_foundation_required(_experience_parent)
    f["experience_foundation_gate_satisfied"] = bool(
        not f["experience_foundation_required"]
        or _has_experience_foundation(_experience_parent, _experience_parent_led))

    # --- design-direction facts (P2) ---
    f["parent_binding_present"] = bool(led.get("parent-workflow-id") and led.get("product-decision-id") and _dd_input_brief_ref(led))
    try:
        import lint_design_direction as _LDD
        _ib = _dd_input_brief_ref(led)
        _ibp = _ib if (_ib and os.path.isabs(_ib)) else (os.path.join(ROOT, _ib) if _ib else None)
        f["direction_input_brief_valid"] = bool(_ibp and os.path.exists(_ibp) and not _LDD.lint_file(_ibp, "direction-input-brief")[0])
    except Exception:
        f["direction_input_brief_valid"] = False
    f["direction_discovery_present"] = bool(_active_artifact(led, arts, "direction-discovery", "direction-discovery-report-id"))
    f["divergence_charter_present"] = _divergence_charter_ok(led, arts)
    f["directions_diverged"] = _directions_diverged(led, arts)
    f["divergence_audit_passed"] = _divergence_audit_ok(wf, led, arts)
    f["selected_direction_accepted"] = _selected_direction_accepted_ok(wf, led, arts)
    f["none_of_the_above_recorded"] = _none_of_the_above_recorded(wf, led, arts)
    _wp = _active_artifact(led, arts, "winner-prototype", "prototype-report-id")
    _wp_content = _artifact_content(_wp, "winner-prototype")
    f["winner_prototype_present"] = bool(_wp and _wp_content and _wp_content.get("preview-receipt-ref")) and _has_preview_receipt(wf, _wp_content)
    _v = _panel_verdict(led, arts)
    f["critique_revision_requested"] = (_v == "minor-revision")
    f["concept_rejection_recorded"] = (_v == "concept-flaw")
    f["direction_critique_passed"] = _critique_panel_ok(led, arts) and _has_preview_receipt(wf, _wp_content)
    f["design_direction_approved"] = _has_direction_approval(wf, led)   # Task 11
    f["_ui_bearing"] = _is_ui_bearing(led, wf)   # Task 13/Fix A: 부모 cascade design→spec gate
    f["method_handoff_unmet"] = _method_handoff_unmet(wf, led, ctx)      # P3-B T4.3: 전이 handoff 게이트

    f["grounding_evidence_present"] = _present(led, arts, ["grounding-package"], "grounding-evidence-present")
    _grounding_ok, _grounding_reason = _grounding_lens_coverage(wf, led, arts)
    f["grounding_lens_coverage_satisfied"] = _grounding_ok
    f["_grounding_lens_coverage_reason"] = _grounding_reason
    f["wave_plan_present"] = _present(led, arts, ["wave-plan", "plan", "wave_plan"], "wave-plan-present")
    f["completion_record_present"] = _present(led, arts, ["completion-record"], "completion-record-present")
    f["blocked_report_present"] = _present(led, arts, ["blocked-report"], "blocked-report-present")

    # 승인(Accepted) 파생
    f["design_accepted"] = _bundle_accepted(wf, arts, "design-bundle")
    f["design_system_release_attached"] = bool(
        not _experience_foundation_required(wf) or _ui_design_release_binding_ok(arts))
    f["spec_accepted"] = _bundle_accepted(wf, arts, "spec-bundle")
    # No fallback: only the exact current executive-decision-packet revision.
    f["decision_packet_accepted"] = "executive-decision-packet" in accepted_types

    # spec→build 핵심 게이트
    f["_must_read_unmet"] = _must_read_unmet(wf, led, arts)

    # 재개(blocked)
    f["resume_condition_present"] = bool(
        led.get("resume-condition") or led.get("resume_condition_present")
        or led.get("resume-condition-satisfied") or led.get("resume_condition_satisfied")
    )
    f["resume_condition_satisfied"] = bool(
        led.get("resume-condition-satisfied") or led.get("resume_condition_satisfied")
    )
    f["human_instruction_needed"] = bool(led.get("human-instruction-needed") or led.get("human_instruction_needed"))
    f["human_instruction_applied"] = bool(led.get("human-instruction-applied") or led.get("human_instruction_applied"))

    # Magentic run 루프
    prog = led.get("progress") or {}
    limits = (load_tiers().get("governance-limits", {}) or {})
    max_rounds = limits.get("max-rounds", 12)
    max_stalls = limits.get("max-stalls", 3)
    rnd = prog.get("round", 0) or 0
    stalls = prog.get("stall_count", prog.get("stall-count", 0)) or 0
    ipbm = prog.get("is_progress_being_made", prog.get("is-progress-being-made", True))
    f["progress_ok"] = bool(ipbm) and rnd < max_rounds and stalls < max_stalls
    f["_progress_detail"] = f"round={rnd}/{max_rounds}, stalls={stalls}/{max_stalls}, progressing={ipbm}"

    # evidence-grade vs tier 최소치
    f["_tier_evidence_min"] = _tier_evidence_min(f["tier"])

    # 명시적 오버라이드 병합 — 단, **신뢰 게이트(PROTECTED)는 오버라이드 불가**(finding P0-4).
    # 이들은 실제 accepted 아티팩트(acceptance_log)·signoff 파일·산출물 실존에서만 파생한다 —
    # 원장 facts 에 손으로 적은 값이 게이트를 통과시키지 못하게 한다(원장은 guard 보호이지만
    # 심층방어로 파생값을 되살린다). 나머지 운영 facts(progress/tier 등)만 오버라이드 허용.
    for k, v in explicit.items():
        nk = _normkey(k)
        if nk in _PROTECTED_FACTS:
            continue
        f[nk] = v
    return f


# 신뢰 게이트에 쓰이는 파생 사실 — 원장 명시 facts 로 오버라이드 금지(P0-4).
_PROTECTED_FACTS = {
    "design_accepted", "design_system_release_attached", "spec_accepted", "decision_packet_accepted",
    "human_gate_approved", "completion_record_present", "quality_gate_status",
    "quality_gate_failed",
    "release_acceptance_status", "evidence_grade", "_must_read_unmet",
    "grounding_evidence_present", "wave_plan_present", "decision_brief_present",
    "grounding_lens_coverage_satisfied", "_grounding_lens_coverage_reason",
    "workload_profile_present",
    "blocker_open",
    "founder_context_present", "venture_options_validated",
    "venture_decision_accepted", "human_acceptance_receipt_present",
    "company_context_provisional_committed", "company_context_lint_passed",
    "company_context_artifact_recorded", "company_context_ready",
    "experience_parent_binding_present", "competitive_benchmark_accepted",
    "experience_strategy_accepted", "experience_technical_feasibility_accepted",
    "experience_operational_feasibility_accepted", "experience_blueprint_accepted",
    "wireframe_set_accepted", "experience_foundation_link_recorded",
    "experience_foundation_required", "experience_foundation_gate_satisfied",
    "parent_binding_present", "direction_input_brief_valid", "direction_discovery_present",
    "divergence_charter_present", "directions_diverged", "divergence_audit_passed",
    "selected_direction_accepted", "none_of_the_above_recorded", "winner_prototype_present",
    "critique_revision_requested", "concept_rejection_recorded", "direction_critique_passed",
    "design_direction_approved", "_ui_bearing",
    "method_handoff_unmet",
}


def _normkey(k):
    return str(k).replace("-", "_")


# ---------------------------------------------------------------- condition predicates

def _p_evidence_grade(f):
    grade = f.get("evidence_grade")
    minv = f.get("_tier_evidence_min")
    if not minv:
        return (True, None)  # tier 에 최소 증거등급 없음 -> 게이트 없음
    # finding P0-4: tier 가 최소치를 요구하는데 원장에 evidence-grade 가 없으면 **차단**한다.
    # 예전엔 미기재를 '측정 불가'로 통과시켜, standard+ tier 에서 증거등급 게이트가 무력화됐다.
    if not grade:
        return (False, f"승인된 executive-decision-packet의 max evidence grade 없음 — "
                       f"tier 최소 {minv} 충족을 증명할 수 없다(submit/review된 report evidence에서만 파생).")
    if _EGRADE.get(str(grade), -1) >= _EGRADE.get(str(minv), 99):
        return (True, None)
    return (False, f"증거등급 부족: {grade} < tier 최소 {minv}")


def _p_human_gate(f):
    if _tier_human_gate_required(f.get("tier")):
        if f.get("human_gate_approved"):
            return (True, None)
        return (False, "heavy tier: 사람 승인(human-gate) 필요 — human_gate_approved=false")
    return (True, None)


def _p_must_read(f):
    unmet = f.get("_must_read_unmet") or []
    if unmet:
        return (False, "must-read 설계 미승인(Accepted 아님): " + ", ".join(unmet))
    return (True, None)


def _p_progress(f):
    if f.get("progress_ok"):
        return (True, None)
    return (False, "run 루프 진전 없음/상한 초과: " + str(f.get("_progress_detail")))


def _p_human_instr(f):
    if f.get("human_instruction_needed") and not f.get("human_instruction_applied"):
        return (False, "사람 지시 필요하나 미적용(human-instruction-applied=false)")
    return (True, None)


_PREDICATES = {
    "decision-brief-present": lambda f: (bool(f.get("decision_brief_present")), "decision-brief(intake 산출물) 없음"),
    "workload-profile-present": lambda f: (bool(f.get("workload_profile_present")), "workload-profile(intake typed 산출물) 없음"),
    "grounding-evidence-present": lambda f: (bool(f.get("grounding_evidence_present")), "discovery 근거 접지 산출물 없음"),
    "option-set-present": lambda f: (f.get("option_set_count", 0) >= 2, f"option-set ≥2 필요(현재 {f.get('option_set_count', 0)})"),
    "grounding-lens-coverage-satisfied": lambda f: (
        bool(f.get("grounding_lens_coverage_satisfied")),
        f.get("_grounding_lens_coverage_reason") or "grounding lens/source binding 미충족",
    ),
    "decision-packet-accepted": lambda f: (bool(f.get("decision_packet_accepted")), "ExecutiveDecisionPacket 미승인(Accepted 아님)"),
    "evidence-grade-min": _p_evidence_grade,
    "design-accepted": lambda f: (bool(f.get("design_accepted")), "설계 산출물 미승인(Accepted 아님)"),
    "design-system-release-attached": lambda f: (
        bool(f.get("design_system_release_attached")),
        "공개/신규/대규모 UI의 ui-design에 candidate|stable 조직 design release exact ref/SHA+subset+delta 없음"),
    "spec-accepted": lambda f: (bool(f.get("spec_accepted")), "기능명세 미승인(Accepted 아님)"),
    "must-read-designs-accepted": _p_must_read,
    "completion-record-present": lambda f: (bool(f.get("completion_record_present")), "completion-record 없음"),
    "quality-gate-passed": lambda f: (f.get("quality_gate_status") == "Passed", f"quality_gate_status != Passed (현재 {f.get('quality_gate_status')})"),
    "quality-gate-failed": lambda f: (bool(f.get("quality_gate_failed")), f"quality_gate_status != Failed (현재 {f.get('quality_gate_status')})"),
    "blocker-open-false": lambda f: (not f.get("blocker_open"), "열린 blocker 존재"),
    "release-approved": lambda f: (f.get("release_acceptance_status") == "Approved", f"release_acceptance_status != Approved (현재 {f.get('release_acceptance_status')})"),
    "no-unresolved-critical-risks": lambda f: (not f.get("unresolved_critical_risks"), "미해결 Critical 리스크 존재"),
    "human-gate": _p_human_gate,
    "wave-plan-present": lambda f: (bool(f.get("wave_plan_present")), "wave plan 산출물 없음"),
    "progress-being-made": _p_progress,
    "blocked-report-present": lambda f: (bool(f.get("blocked_report_present")), "BlockedReport 없음"),
    "resume-condition-present": lambda f: (bool(f.get("resume_condition_present")), "재개 조건 미명시"),
    "resume-condition-satisfied": lambda f: (bool(f.get("resume_condition_satisfied")), "재개 조건 미충족"),
    "human-instruction-applied-if-needed": _p_human_instr,
    "founder-context-present": lambda f: (bool(f.get("founder_context_present")), "founder-context.yaml status != filled"),
    "opportunity-clusters-present": lambda f: (f.get("opportunity_cluster_count", 0) >= 2, f"opportunity-cluster ≥2 필요(현재 {f.get('opportunity_cluster_count', 0)})"),
    "venture-options-validated": lambda f: (bool(f.get("venture_options_validated")), "venture-validation accepted 산출물 없음(kill-criteria 포함 옵션 검증 필요)"),
    "venture-decision-accepted": lambda f: (bool(f.get("venture_decision_accepted")), "venture-decision accepted 산출물 없음"),
    "human-acceptance-receipt-present": lambda f: (bool(f.get("human_acceptance_receipt_present")), "HUMAN-001 accepted 이벤트(report-sha256 바인딩) 없음 — boolean 자기신고 불가"),
    "company-context-provisional-committed": lambda f: (bool(f.get("company_context_provisional_committed")), "공식 company-context.yaml status != provisional/operating"),
    "company-context-lint-passed": lambda f: (bool(f.get("company_context_lint_passed")), "company-context lint Hard Fail 존재"),
    "company-context-artifact-recorded": lambda f: (bool(f.get("company_context_artifact_recorded")), "company-context commit receipt(artifact) 없음"),
    "company-context-ready": lambda f: (
        bool(f.get("company_context_ready")),
        "company-context 미준비(status provisional/operating 또는 trusted project-scoped workload 필요; blocker 없어야 함)"),
    "experience-parent-binding-present": lambda f: (
        bool(f.get("experience_parent_binding_present")),
        "experience-foundation parent/product-decision exact binding 없음"),
    "competitive-benchmark-accepted": lambda f: (
        bool(f.get("competitive_benchmark_accepted")),
        "competitive-experience-benchmark exact revision 미승인"),
    "experience-strategy-accepted": lambda f: (
        bool(f.get("experience_strategy_accepted")),
        "experience-strategy proceed exact revision 미승인"),
    "experience-technical-feasibility-accepted": lambda f: (
        bool(f.get("experience_technical_feasibility_accepted")),
        "현재 strategy exact ref/SHA에 결속된 CTO/CPTO technical feasibility=feasible revision 미승인"),
    "experience-operational-feasibility-accepted": lambda f: (
        bool(f.get("experience_operational_feasibility_accepted")),
        "현재 strategy exact ref/SHA에 결속된 COO operational feasibility=feasible revision 미승인"),
    "experience-blueprint-accepted": lambda f: (
        bool(f.get("experience_blueprint_accepted")),
        "experience-blueprint exact revision 미승인"),
    "wireframe-set-accepted": lambda f: (
        bool(f.get("wireframe_set_accepted")),
        "wireframe-set exact revision 미승인"),
    "experience-foundation-link-recorded": lambda f: (
        bool(f.get("experience_foundation_link_recorded")),
        "부모 experience-foundation exact bundle 링크 미기록/스테일"),
    "experience-foundation-gate-satisfied": lambda f: (
        bool(f.get("experience_foundation_gate_satisfied")),
        "공개 웹/interactive-learning/신규 제품/대규모 리디자인: approved experience-foundation 선행 필요"),
    "parent-binding-present": lambda f: (bool(f.get("parent_binding_present")), "parent 바인딩 없음"),
    "direction-input-brief-valid": lambda f: (bool(f.get("direction_input_brief_valid")), "input-brief lint 실패(필수/금지)"),
    "direction-discovery-present": lambda f: (bool(f.get("direction_discovery_present")), "direction-discovery 아티팩트 없음"),
    "divergence-charter-present": lambda f: (bool(f.get("divergence_charter_present")), "divergence-charter 부재/직교 분할 lint 실패"),
    "directions-diverged": lambda f: (bool(f.get("directions_diverged")), "3안 독립성/실픽셀/비교렌더 실패"),
    "divergence-audit-passed": lambda f: (bool(f.get("divergence_audit_passed")), "선택 전 비교감사 미통과(쌍별 4축 차이/full-size preview/primitive collision/blocker/승인 확인)"),
    "selected-direction-accepted": lambda f: (bool(f.get("selected_direction_accepted")), "selected bundle lint 실패 또는 미승인"),
    "none-of-the-above-recorded": lambda f: (
        bool(f.get("none_of_the_above_recorded")),
        "HUMAN-001 none-of-the-above exact decision 없음"),
    "winner-prototype-present": lambda f: (bool(f.get("winner_prototype_present")), "winner-prototype/preview receipt 없음"),
    "critique-revision-requested": lambda f: (bool(f.get("critique_revision_requested")), "active verdict != minor-revision"),
    "concept-rejection-recorded": lambda f: (bool(f.get("concept_rejection_recorded")), "active verdict != concept-flaw"),
    "direction-critique-passed": lambda f: (bool(f.get("direction_critique_passed")), "패널 pass 미충족"),
    "approved-direction-valid": lambda f: (bool(f.get("design_direction_approved")), "approved-direction 8점 검증 실패(불변 report/schema/hash 불일치)"),
    "approval-receipt-bound": lambda f: (bool(f.get("design_direction_approved")), "acceptance receipt 가 child workflow+report-sha256 에 정확 바인딩되지 않음"),
    "parent-approval-link-recorded": lambda f: (bool(f.get("design_direction_approved")), "부모 원장 design-direction-approval(report-ref/sha256/child-workflow-id) 미기록"),
    "design-direction-approved": lambda f: (bool(f.get("design_direction_approved")), "approved-direction 8점/staleness 실패 — 자기신고 불가"),
    "design-direction-gate-satisfied": lambda f: (
        (not f.get("_ui_bearing")) or f.get("tier") not in ("standard", "heavy") or bool(f.get("design_direction_approved")),
        "UI-bearing standard/heavy: 승인된 design-direction 필요(우회 차단)"),
    "method-handoff-satisfied": lambda f: (
        not f.get("method_handoff_unmet"),
        "전이 handoff 미충족: consumer required-inputs(both-active) 부재/미수락(Accepted 필요)"),
}


def _cmp(actual, expected):
    if isinstance(expected, bool):
        return bool(actual) == expected
    if isinstance(expected, str) and expected.lower() in ("true", "false"):
        return bool(actual) == (expected.lower() == "true")
    return str(actual) == str(expected)


def _eval_condition(cond, facts):
    """조건 하나 평가 -> (ok, reason_if_false). 문자열(술어) 또는 {key: expected} 형식 지원."""
    if isinstance(cond, dict):
        for k, v in cond.items():
            actual = facts.get(_normkey(k))
            if not _cmp(actual, v):
                return (False, f"{k} != {v} (실제: {actual})")
        return (True, None)
    key = str(cond)
    pred = _PREDICATES.get(key)
    if pred is None:
        return (False, f"미지의 조건: {key}")
    try:
        return pred(facts)
    except Exception as e:
        return (False, f"조건 평가 오류({key}): {e}")


# ---------------------------------------------------------------- transition lookup

def _find_transition(frm, to):
    """(frm, to) 에 해당하는 전이 규칙. 와일드카드(*->blocked)·재개(blocked-><resume>) 해석 포함."""
    ws = _ws_transitions()
    for t in ws:
        if t.get("from") == frm and t.get("to") == to:
            return t
    if frm == "blocked":
        for t in ws:
            if t.get("from") == "blocked" and t.get("to") == "<resume>":
                return t
    if to == "blocked":
        for t in ws:
            if t.get("from") == "*" and t.get("to") == "blocked":
                return t
    return None


# ---------------------------------------------------------------- public API

def current_stage(wf):
    """현재 stage(원장 없으면 초기 stage intake). 부작용 없음. 예외 없음."""
    try:
        return _load_ledger_safe(wf).get("stage", INITIAL_STAGE)
    except Exception as e:
        _log(f"current_stage 오류: {e}")
        return INITIAL_STAGE


def allowed_next(wf):
    """현재 stage 에서 구조적으로 도달 가능한 다음 stage 목록(조건 미검사). 예외 없음."""
    try:
        led = _load_ledger_safe(wf)
        cur = led.get("stage", INITIAL_STAGE)
        plan = led.get("plan", DEFAULT_PLAN)
        stages = _plan_stages(plan)
        return _transition_engine.structural_next(
            cur, stages, _ws_transitions(), led.get("blocked-from")
        )
    except Exception as e:
        _log(f"allowed_next 오류: {e}")
        return []


def _actor_allowed(t, actor):
    """Require a concrete registered transition executor.

    Runtime placeholders are documentation bugs, not authorization grants.
    Decision authors/reviewers are authorized by review-artifact separately.
    """
    return _transition_engine.actor_allowed(t, actor, _role_registry())


def can_transition(wf, to, ctx=None, actor=None):
    """(ok, [unmet_reason]). required-conditions/forbidden-if 를 원장 사실로 검사. 예외 없음."""
    try:
        led = _load_ledger_safe(wf)
        frm = led.get("stage", INITIAL_STAGE)
        t = _find_transition(frm, to)
        if not t:
            return (False, [f"정의된 전이 규칙 없음: {frm} -> {to}"])
        plan = led.get("plan", DEFAULT_PLAN)
        plan_stages = set(_plan_stages(plan))
        facts = _facts(wf, led, ctx)
        reasons = _transition_engine.evaluate_transition(
            t,
            current=frm,
            destination=to,
            plan=plan,
            plan_stages=plan_stages,
            blocked_from=led.get("blocked-from"),
            facts=facts,
            condition_evaluator=_eval_condition,
            actor=actor,
            role_registry=_role_registry(),
        )
        return (len(reasons) == 0, reasons)
    except Exception as e:
        _log(f"can_transition 오류: {e}")
        return (False, [f"can_transition 내부 오류(degrade): {e}"])


def enter_stage(wf, to, evidence=None, actor=None, ctx=None):
    """완료된 current stage에서 ``to`` stage를 running으로 연다.

    실제 전이는 actor 를 명시해야 한다(권한 allowed-by 검사 + 감사). guard/check
    (전제조건 미리보기)와 달리 enter-stage는 actor 미지정 시 거부한다. blocked 진입/재개는
    작업 완료와 무관한 side-state이므로 stage-status 검사를 적용하지 않는다."""
    if not str(actor or "").strip():
        return (False, ["전이 주체(--actor) 미지정 — 전이는 actor 를 명시해야 한다(allowed-by 권한/감사, P0-4)."])
    ok, reasons = can_transition(wf, to, ctx, actor=actor)
    if not ok:
        return (False, reasons)
    try:
        with _workflow_lock(wf):
            led = read_ledger(wf)
            if led is None:
                led = _default_ledger(wf)
            frm = led.get("stage", INITIAL_STAGE)
            if (led.get("stage-status") == "running" and frm != "blocked"
                    and to != "blocked" and to != frm):
                return (False, [f"현재 stage '{frm}'가 running이다 — complete-stage 후 '{to}'에 진입하라"])
            intended = led.get("completed-for-next-stage")
            if (led.get("stage-status") == "completed" and intended and to != intended):
                return (False, [f"stage '{frm}'는 '{intended}' 진입용으로 완료됐다(요청: {to})"])
            event = {
                "state-event-id": f"se-{_stamp()}-{uuid.uuid4().hex[:8]}",
                "event-type": "state-transition", "workflow-id": wf,
                "from": frm, "to": to, "actor": actor,
                "evidence": evidence, "effective-at": _now(),
            }
            if not _append_state_event(wf, event):
                return (False, ["workflow event append 실패"])
            led["stage"] = to
            led["stage-status"] = "running"
            led["last-completed-stage"] = frm
            led["completed-for-next-stage"] = None
            if to == "blocked":
                led["blocked-from"] = frm
            elif frm == "blocked":
                led.pop("blocked-from", None)
            led["last-updated-at"] = _now()
            _write_ledger(wf, led)
        return (True, [])
    except Exception as e:
        _log(f"enter_stage 오류: {e}")
        return (False, [f"enter-stage 내부 오류(degrade): {e}"])


def complete_stage(wf, actor, evidence=None, to=None):
    """Validate the current stage's exit gate and mark it completed."""
    if actor != "OPS-ORCH" or not _role_has_capability(actor, "transition-executor"):
        return False, ["complete-stage는 OPS-ORCH transition-executor만 실행 가능"]
    led = read_ledger(wf)
    if not led:
        return False, [f"workflow 원장 없음: {wf}"]
    stage = led.get("stage")
    if led.get("stage-status") == "completed":
        return True, []
    plan = led.get("plan", DEFAULT_PLAN)
    stages = _plan_stages(plan)
    next_stage = str(to).strip() if to else None
    if next_stage and next_stage not in allowed_next(wf):
        return False, [f"현재 stage '{stage}'에서 완료 대상으로 '{next_stage}'를 선택할 수 없다"]
    if not next_stage and stage in stages:
        index = stages.index(stage)
        if index + 1 < len(stages):
            next_stage = stages[index + 1]
    if next_stage:
        ok, reasons = can_transition(wf, next_stage)
        if not ok:
            return False, reasons
    event = {
        "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
        "event-type": "stage-completed", "workflow-id": wf,
        "stage": stage, "actor": actor, "evidence": evidence,
        "intended-next-stage": next_stage,
        "effective-at": _now(),
    }
    committed, error = _atomic_event_transaction(wf, workflow_event=event)
    return (True, []) if committed else (False, [error])


def transition(wf, to, evidence=None, actor=None, ctx=None):
    """Deprecated compatibility advance: complete current, then enter ``to``.

    신규 command/runtime은 ``complete_stage``와 ``enter_stage``를 분리해 호출한다. 과거
    ``transition`` 호출은 gate fact를 우회하지 않도록 현재 exit gate를 먼저 검증·완료한 뒤
    같은 권한 검사로 다음 stage를 연다. blocked side-state는 완료를 요구하지 않는다.
    """
    led = read_ledger(wf)
    if not led:
        return False, [f"workflow 원장 없음: {wf}"]
    frm = led.get("stage", INITIAL_STAGE)
    if to == "blocked" or frm == "blocked" or to == frm:
        return enter_stage(wf, to, evidence=evidence, actor=actor, ctx=ctx)
    if led.get("stage-status", "running") != "completed":
        ok, reasons = complete_stage(wf, actor, evidence=evidence, to=to)
        if not ok:
            return ok, reasons
    return enter_stage(wf, to, evidence=evidence, actor=actor, ctx=ctx)


def update_progress(wf, updates, plan=None):
    """wave/light 의 Magentic progress 필드를 통합 원장 progress: 하위에 병합한다.

    E2 통합(#7): plan-wave/run-wave 가 별도 progress.yaml 대신 이 원장의 progress: 를 읽고 쓴다
    (하나의 wf-id, 하나의 원장). stage 는 바꾸지 않는다(전진은 transition 이 담당). 원장이 없으면
    생성(plan 기본값). (ok, progress_dict) 반환. 예외를 던지지 않는다(degrade-safe).

    keys 는 전달된 그대로 보존한다(round/stall_count/is_progress_being_made/next 등) — _facts 가
    underscore/hyphen 두 표기를 모두 읽으므로 정규화하지 않는다."""
    try:
        led = read_ledger(wf)
        if led is None:
            led = _default_ledger(wf, plan=plan or DEFAULT_PLAN)
        elif plan and not led.get("plan"):
            led["plan"] = plan
        prog = dict(led.get("progress") or {})
        if isinstance(updates, dict):
            for k, v in updates.items():
                if v is not None:
                    prog[k] = v
        led["progress"] = prog
        led["last-updated-at"] = _now()
        ok = _write_ledger(wf, led)
        return (ok, prog)
    except Exception as e:
        _log(f"update_progress 오류: {e}")
        return (False, {})


def read_progress(wf):
    """통합 원장의 progress: dict(없으면 {}). 예외 없음."""
    try:
        return dict((_load_ledger_safe(wf).get("progress") or {}))
    except Exception as e:
        _log(f"read_progress 오류: {e}")
        return {}


# ---------------------------------------------------------------- orchestrator (항목4)
# stage -> 그 stage 의 작업을 수행하는 cascade 커맨드. released/closed 는 종단(커맨드 없음).
_STAGE_COMMAND = {
    "intake": "/ceo-intake",
    "discovery": "/ground",
    "decide": "/decide",
    "design": "/design",
    "spec": "/spec",
    "build": "/build",
    "verification": "/review-output",
    "acceptance": "/release-check",
    "released": None,
    "closed": None,
    # wave/light 내부 stage
    "plan": "/plan-wave",
    "run": "/run-wave",
}


def _stage_command(plan, stage):
    definition = (((load_contracts().get("workflows", {}) or {}).get(plan, {}) or {})
                  .get("stages", {}).get(stage, {}))
    if isinstance(definition, dict) and "command" in definition:
        command = definition.get("command")
        return f"/{command}" if command else None
    return _STAGE_COMMAND.get(stage)

# 사람 결정(DRAI decider=human)이 개입하는 전이 조건 — /run-cascade 는 여기서 멈춰 사람 수용을 받는다.
_HUMAN_GATE_CONDS = {
    "human-gate": "릴리스 수용(release acceptance) — 사람 최종 승인",
    "release-approved": "릴리스 승인",
    "decision-packet-accepted": "go/no-go 의사결정 수용(방향 확정)",
}


def _human_gate_for(frm, to):
    """(frm->to) 전이가 사람 게이트를 요구하나 -> (required, approver, what|None).

    전이 규칙의 required-conditions 에 DRAI human-decider 조건(_HUMAN_GATE_CONDS)이 있으면 사람
    승인 지점이다. approver 는 governance-tiers plan-signoff/human-gate 기준 HUMAN-001(위임 시 EXEC-CEO).
    엔진의 하드 강제는 heavy tier 의 human-gate(signoff 파일)뿐이고, decide go/no-go 는 오케스트레이터가
    멈춰 사람 수용을 받는 pause 지점이다(자동 승인·자동 완주 금지 — 항목4 불변식)."""
    t = _find_transition(frm, to)
    if not t:
        return (False, None, None)
    hits = [c for c in (t.get("required-conditions") or [])
            if isinstance(c, str) and c in _HUMAN_GATE_CONDS]
    if not hits:
        return (False, None, None)
    what = "; ".join(_HUMAN_GATE_CONDS[c] for c in hits)
    return (True, "HUMAN-001 (위임 시 EXEC-CEO)", what)


def next_info(wf):
    """오케스트레이터(/run-cascade)용 결정론적 다음-스텝 계산. 예외 없음.

    state graph(can_transition/allowed_next)를 그대로 재사용한다 — 평행 엔진이 아니라 얇은 조회층.
    반환 dict:
      current-stage/current-command  : 현 stage 와 그 작업 커맨드(작업 미완이면 이걸 실행)
      next-stage/next-command        : plan 시퀀스상 다음 stage 와 커맨드
      advance{ok,reasons,human-gate} : cur->next 전이의 guard 결과 + 사람 게이트 여부/승인자/사유
      terminal                       : 종단 도달
    """
    try:
        led = _load_ledger_safe(wf)
        cur = led.get("stage", INITIAL_STAGE)
        plan = led.get("plan", DEFAULT_PLAN)
        tier = led.get("tier", DEFAULT_TIER)
        stage_status = led.get("stage-status", "running")
        stages = _plan_stages(plan)
        terminal = _plan_terminal(plan)
        nxt = None
        if cur == "blocked":
            nxt = led.get("blocked-from")
        elif cur in stages:
            i = stages.index(cur)
            if i + 1 < len(stages):
                nxt = stages[i + 1]
        terminal_stage = (cur == terminal) or (cur in ("released", "closed")) or (
            nxt is None and cur != "blocked")
        is_terminal = bool(terminal_stage and stage_status == "completed")
        info = {
            "workflow-id": wf,
            "plan": plan,
            "tier": tier,
            "current-stage": cur,
            "stage-status": stage_status,
            "last-completed-stage": led.get("last-completed-stage"),
            "current-command": _stage_command(plan, cur) if stage_status == "running" else None,
            "next-stage": nxt,
            "next-command": _stage_command(plan, nxt) if nxt else None,
            "terminal": bool(is_terminal),
        }
        if nxt:
            ok, reasons = can_transition(wf, nxt)
            hg_req, approver, what = _human_gate_for(cur, nxt)
            # 사람 signoff 가 이미 있으면(사람이 세션 밖에서 승인) 게이트 해제 → 재개 가능.
            if hg_req and _human_gate_satisfied(wf, cur):
                hg_req = False
            info["advance"] = {
                "ok": bool(ok),
                "reasons": reasons,
                "human-gate": {"required": bool(hg_req), "approver": approver, "what": what},
            }
        return info
    except Exception as e:
        _log(f"next_info 오류: {e}")
        return {"workflow-id": wf, "error": str(e)}


# ---------------------------------------------------------------- CLI

def _argval(args, flag, default=None):
    return args[args.index(flag) + 1] if flag in args and args.index(flag) + 1 < len(args) else default


def _coerce(s):
    """CLI 값 문자열을 bool/int 로 최대한 변환(progress 필드 타입 보존)."""
    if isinstance(s, str):
        low = s.strip().lower()
        if low in ("true", "false"):
            return low == "true"
        try:
            return int(s)
        except (TypeError, ValueError):
            return s
    return s


def _workspace_ok():
    return _state_dir(create=False) is not None


def main():
    args = sys.argv[1:]
    if not args:
        sys.stderr.write(__doc__)
        return 1
    cmd = args[0]
    wf = _argval(args, "--workflow")
    to = _argval(args, "--to")

    if cmd == "guard":
        if not wf or not to:
            sys.stderr.write("usage: state_engine.py guard --workflow WF --to STAGE\n")
            return 2
        if not _workspace_ok():
            # fail-CLOSED(finding P0-1): 상태 판별 불가면 통과시키지 않는다. 과거엔
            # degrade(allow)라서 workspace 한 줄만 비우면 상태 게이트가 fail-open 됐다.
            sys.stderr.write(
                "[state_engine] BLOCK guard: workspace 미설정 — 상태 검증 불가로 전이 거부(exit 2). "
                "ORGOS_WORKSPACE=<project> 를 설정하세요.\n")
            return 2
        ok, reasons = can_transition(wf, to)
        if ok:
            return 0
        cur = current_stage(wf)
        sys.stderr.write(f"[state_engine] BLOCK 전이 {cur} -> {to}: 선행조건 미충족\n")
        for r in reasons:
            sys.stderr.write(f"  - {r}\n")
        return 2

    if cmd == "check-company-context-ready":
        if not wf:
            sys.stderr.write("usage: state_engine.py check-company-context-ready --workflow WF\n")
            return 1
        f = _facts(wf, _load_ledger_safe(wf))
        ok, reason = _eval_condition("company-context-ready", f)
        if ok:
            print("[state_engine] company-context READY")
            return 0
        sys.stderr.write(f"[state_engine] NOT READY: {reason}\n")
        return 2

    if cmd == "next":
        # 항목4: /run-cascade 용 결정론적 다음-스텝 조회(정보용, exit 0). guard/transition 을
        # 재사용하는 얇은 조회층 — 사람 게이트(human-gate.required)는 오케스트레이터가 멈추는 지점.
        if not wf:
            sys.stderr.write("usage: state_engine.py next --workflow WF\n")
            return 1
        print(json.dumps(next_info(wf), ensure_ascii=False, indent=2))
        return 0

    if cmd == "current":
        if not wf:
            sys.stderr.write("usage: state_engine.py current --workflow WF\n")
            return 1
        print(current_stage(wf))
        return 0

    if cmd == "resolve-family":
        family = _argval(args, "--family")
        raw_signals = _argval(args, "--signals", "")
        signals = [value for value in str(raw_signals).split(",") if value.strip()]
        resolved = resolve_family(family, signals=signals)
        if not family or not resolved:
            sys.stderr.write(f"[state_engine] 등록 family 없음: {family}\n")
            return 2
        print(json.dumps(resolved, ensure_ascii=False, indent=2))
        return 0

    if cmd == "allowed":
        if not wf:
            sys.stderr.write("usage: state_engine.py allowed --workflow WF\n")
            return 1
        for s in allowed_next(wf):
            print(s)
        return 0

    if cmd == "check":
        if not wf or not to:
            sys.stderr.write("usage: state_engine.py check --workflow WF --to STAGE [--actor A]\n")
            return 1
        ok, reasons = can_transition(wf, to, actor=_argval(args, "--actor"))
        print("ALLOW" if ok else "BLOCK")
        for r in reasons:
            print(f"  - {r}")
        return 0 if ok else 1

    if cmd == "record":
        sys.stderr.write(
            "[state_engine] record 제거됨: caller-supplied --design-type/--report-id/"
            "--option-count/--evidence-grade는 신뢰할 수 없다. submit-report --report PATH --actor ROLE을 사용하라.\n")
        return 2

    if cmd in ("submit-report", "submit-artifact"):
        report = _argval(args, "--report")
        actor = _argval(args, "--actor")
        if not wf or not report or not actor:
            sys.stderr.write(
                f"usage: state_engine.py {cmd} --workflow WF --report PATH --actor ROLE\n"
            )
            return 2
        ok, result = submit_report(wf, report, actor)
        if not ok:
            sys.stderr.write(f"[state_engine] {cmd} 거부: {result}\n")
            return 2
        print(json.dumps(result, ensure_ascii=False))
        return 0

    if cmd == "review-artifact":
        report = _argval(args, "--report")
        reviewer = _argval(args, "--reviewer")
        decision = _argval(args, "--decision")
        if not wf or not report or not reviewer or not decision:
            sys.stderr.write("usage: state_engine.py review-artifact --workflow WF --report PATH "
                             "--decision accepted|changes-requested|blocked --reviewer ROLE [--supersedes ID]\n")
            return 2
        ok, result = review_artifact(wf, report, decision, reviewer, _argval(args, "--supersedes"))
        if not ok:
            sys.stderr.write(f"[state_engine] review-artifact 거부: {result}\n")
            return 2
        print(result["acceptance-event-id"])
        return 0

    if cmd == "record-quality-gate":
        review = _argval(args, "--review")
        actor = _argval(args, "--actor")
        if not wf or not review or not actor:
            sys.stderr.write("usage: state_engine.py record-quality-gate --workflow WF --review PATH --actor ROLE\n")
            return 2
        ok, result = record_quality_gate(wf, review, actor)
        if not ok:
            sys.stderr.write(f"[state_engine] record-quality-gate 거부: {result}\n")
            return 2
        print(result["workflow-event-id"])
        return 0

    if cmd == "record-release-decision":
        report = _argval(args, "--report")
        actor = _argval(args, "--actor")
        if not wf or not report or not actor:
            sys.stderr.write("usage: state_engine.py record-release-decision --workflow WF --report PATH --actor ROLE\n")
            return 2
        ok, result = record_release_decision(wf, report, actor)
        if not ok:
            sys.stderr.write(f"[state_engine] record-release-decision 거부: {result}\n")
            return 2
        print(result["workflow-event-id"])
        return 0

    if cmd in ("block", "block-workflow"):
        report = _argval(args, "--report")
        actor = _argval(args, "--actor", "OPS-ORCH")
        if not wf or not report:
            sys.stderr.write("usage: state_engine.py block --workflow WF --report PATH [--actor OPS-ORCH]\n")
            return 2
        ok, result = block_workflow(wf, report, actor)
        if not ok:
            sys.stderr.write(f"[state_engine] block 거부: {result}\n")
            return 2
        print(f"{wf}: -> blocked")
        return 0

    if cmd in ("resume", "resume-workflow"):
        evidence = _argval(args, "--evidence")
        actor = _argval(args, "--actor", "OPS-ORCH")
        if not wf or not evidence:
            sys.stderr.write("usage: state_engine.py resume --workflow WF --evidence PATH [--actor OPS-ORCH]\n")
            return 2
        ok, result = resume_workflow(wf, evidence, actor)
        if not ok:
            sys.stderr.write(f"[state_engine] resume 거부: {result}\n")
            return 2
        print(f"{wf}: resumed")
        return 0

    if cmd in ("signoff", "record-human-signoff"):
        # 사람 승인(human-gate). **사람이 세션 밖에서** 호출해야 한다 — guard_tools 가 에이전트의
        # 이 CLI 호출을 차단한다(P0-4 soft-boundary). --by 는 HUMAN-<id>.
        if not wf:
            sys.stderr.write("usage: state_engine.py signoff --workflow WF --stage STAGE --by HUMAN-<id>\n")
            return 1
        ok, err = record_signoff(wf, _argval(args, "--stage"), _argval(args, "--by"))
        if not ok:
            sys.stderr.write(f"[state_engine] signoff 거부: {err}\n")
            return 1
        print(f"{wf}: human-signoff recorded (stage={_argval(args, '--stage')})")
        return 0

    if cmd in ("transition", "enter-stage"):
        if not wf or not to:
            sys.stderr.write(
                f"usage: state_engine.py {cmd} --workflow WF --to STAGE "
                "[--actor A] [--evidence E]\n"
            )
            return 1
        advance = enter_stage if cmd == "enter-stage" else transition
        ok, reasons = advance(
            wf, to, evidence=_argval(args, "--evidence"), actor=_argval(args, "--actor")
        )
        if ok:
            print(f"{wf}: -> {to}")
            return 0
        sys.stderr.write(f"[state_engine] 전이 거부 -> {to}:\n")
        for r in reasons:
            sys.stderr.write(f"  - {r}\n")
        return 1

    if cmd == "complete-stage":
        actor = _argval(args, "--actor")
        if not wf or not actor:
            sys.stderr.write(
                "usage: state_engine.py complete-stage --workflow WF --actor OPS-ORCH "
                "[--to NEXT-STAGE] [--evidence PATH]\n"
            )
            return 2
        ok, reasons = complete_stage(
            wf, actor, evidence=_argval(args, "--evidence"), to=_argval(args, "--to")
        )
        if ok:
            print(f"{wf}: stage completed ({current_stage(wf)})")
            return 0
        sys.stderr.write("[state_engine] complete-stage 거부:\n")
        for reason in reasons:
            sys.stderr.write(f"  - {reason}\n")
        return 2

    if cmd in ("init", "init-workflow"):
        if not wf:
            sys.stderr.write(
                "usage: state_engine.py init --workflow WF [--plan P] [--tier T] [--mode M] "
                "[--parent-workflow WF] [--product-decision ID] [--direction-input-brief PATH]\n"
            )
            return 1
        try:
            led = init_ledger(
                wf,
                plan=_argval(args, "--plan", DEFAULT_PLAN),
                tier=_argval(args, "--tier", DEFAULT_TIER),
                mode=_argval(args, "--mode", DEFAULT_MODE),
                overwrite="--overwrite" in args,
                parent_workflow=_argval(args, "--parent-workflow"),
                product_decision=_argval(args, "--product-decision"),
                direction_input_brief=_argval(args, "--direction-input-brief"),
            )
        except ValueError as e:
            sys.stderr.write(f"[state_engine] init 거부: {e}\n")
            return 1
        p = _ledger_path(wf, create=False)
        print(os.path.relpath(p, ROOT) if p else f"{wf}: stage={led.get('stage')}")
        return 0

    if cmd == "register-direction-approval":
        # trusted CLI(P2/E — Task 11): 전 검증 통과 후에만 부모 원장에 design-direction-approval
        # 을 기록한다. ValueError -> non-zero exit(init 의 exit-1 관례와 동일).
        try:
            register_direction_approval(
                _argval(args, "--parent-workflow"),
                _argval(args, "--child-workflow"),
                _argval(args, "--report"),
                _argval(args, "--report-sha256"),
            )
        except Exception as e:
            sys.stderr.write(f"[state_engine] register-direction-approval 거부: {e}\n")
            return 1
        print("registered")
        return 0

    if cmd == "register-experience-foundation":
        try:
            register_experience_foundation(
                _argval(args, "--parent-workflow"),
                _argval(args, "--child-workflow"),
            )
        except Exception as e:
            sys.stderr.write(f"[state_engine] register-experience-foundation 거부: {e}\n")
            return 1
        print("registered")
        return 0

    if cmd == "find-child-experience":
        parent = _argval(args, "--parent-workflow")
        pd = _argval(args, "--product-decision")
        if not (parent and pd):
            sys.stderr.write(
                "usage: state_engine.py find-child-experience --parent-workflow WF --product-decision ID\n")
            return 1
        found = find_child_experience_workflow(parent, pd)
        print(json.dumps(found, ensure_ascii=False) if found else "null")
        return 0

    if cmd == "check-experience-foundation":
        if not wf:
            sys.stderr.write("usage: state_engine.py check-experience-foundation --workflow WF\n")
            return 1
        led = _load_ledger_safe(wf)
        ok = ((not _experience_foundation_required(wf))
              or _has_experience_foundation(wf, led))
        print(f"{wf}: experience-foundation={'YES' if ok else 'NO'}")
        return 0 if ok else 3

    if cmd == "find-child-direction":
        # dedup CLI(Task 13 — closes the gap that find_child_direction_workflow(Task 10) had no
        # CLI, so /design couldn't call it without an inline python -c). --parent-workflow +
        # --product-decision 필수, --direction-input-brief-sha256 은 staleness 판정용(생략 가능
        # — 그러면 stale 은 항상 True 로 보수적으로 보고된다, find_child_direction_workflow 참조).
        parent = _argval(args, "--parent-workflow")
        pd = _argval(args, "--product-decision")
        brief_sha = _argval(args, "--direction-input-brief-sha256")
        if not (parent and pd):
            sys.stderr.write(
                "usage: state_engine.py find-child-direction --parent-workflow WF --product-decision ID "
                "[--direction-input-brief-sha256 SHA]\n")
            return 1
        found = find_child_direction_workflow(parent, pd, brief_sha)
        print(json.dumps(found, ensure_ascii=False) if found else "null")
        return 0

    if cmd == "check-direction-approved":
        if not wf:
            sys.stderr.write("usage: state_engine.py check-direction-approved --workflow WF\n")
            return 1
        led = _load_ledger_safe(wf)
        ok = _has_direction_approval(wf, led)
        print(f"{wf}: direction-approved={'YES' if ok else 'NO'}")
        return 0 if ok else 3

    if cmd == "set-tier":
        # F1: 기존 워크플로 tier 승격(운영 fact, PROTECTED 아님). init --overwrite 는 이력 파괴,
        # 원장 직접편집은 guard_tools 차단 → 이 CLI 가 유일한 지원 경로. 다운그레이드는 금지한다
        # (heavy->light 로 human-gate 등 상위 게이트를 우회하는 것을 막는다).
        _order = {"light": 0, "standard": 1, "heavy": 2}
        new_tier = _argval(args, "--tier")
        if not wf or new_tier not in _order:
            sys.stderr.write("usage: state_engine.py set-tier --workflow WF --tier light|standard|heavy\n")
            return 1
        led = _load_ledger_safe(wf)
        if not led:
            sys.stderr.write(f"[state_engine] 원장 없음: {wf}\n")
            return 1
        old = led.get("tier") or DEFAULT_TIER
        if _order[new_tier] < _order.get(old, 1):
            sys.stderr.write(f"[state_engine] tier 다운그레이드 거부({old}->{new_tier}) — 상위 게이트 우회 방지. "
                             "낮은 tier 가 필요하면 새 워크플로로 시작하라.\n")
            return 2
        event = {
            "workflow-event-id": f"wfe-{_stamp()}-{uuid.uuid4().hex[:8]}",
            "event-type": "tier-escalated", "workflow-id": wf,
            "from-tier": old, "to-tier": new_tier,
            "actor": "OPS-ORCH", "effective-at": _now(),
        }
        committed, error = _atomic_event_transaction(wf, workflow_event=event)
        if not committed:
            sys.stderr.write(f"[state_engine] tier event 기록 실패: {error}\n")
            return 2
        print(f"{wf}: tier {old} -> {new_tier}")
        return 0

    if cmd == "progress":
        if not wf:
            sys.stderr.write("usage: state_engine.py progress --workflow WF "
                             "[--round N] [--stall N] [--next FAM] [--progressing true|false] "
                             "[--satisfied true|false] [--set key=value ...]\n")
            return 1
        updates = {}
        # 공통 wave 필드(Magentic 원장) — 지정된 것만 갱신.
        rnd = _argval(args, "--round")
        if rnd is not None:
            updates["round"] = _coerce(rnd)
        stall = _argval(args, "--stall")
        if stall is not None:
            updates["stall_count"] = _coerce(stall)
        nxt = _argval(args, "--next")
        if nxt is not None:
            updates["next"] = nxt
        prog = _argval(args, "--progressing")
        if prog is not None:
            updates["is_progress_being_made"] = _coerce(prog)
        sat = _argval(args, "--satisfied")
        if sat is not None:
            updates["is_request_satisfied"] = _coerce(sat)
        # 자유형 --set key=value (여러 번 가능)
        for i, a in enumerate(args):
            if a == "--set" and i + 1 < len(args) and "=" in args[i + 1]:
                k, v = args[i + 1].split("=", 1)
                updates[k.strip()] = _coerce(v.strip())
        ok, result = update_progress(wf, updates, plan=_argval(args, "--plan"))
        if not ok:
            sys.stderr.write("[state_engine] progress 기록 실패(degrade) — workspace 설정 확인\n")
        print(json.dumps(result, ensure_ascii=False, sort_keys=True))
        return 0 if ok else 1

    sys.stderr.write(__doc__)
    return 1


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as e:  # 최종 안전망 — 어떤 경우에도 크래시로 파이프라인을 막지 않는다
        _log(f"unexpected: {e}")
        sys.exit(0)