# -*- coding: utf-8 -*-
"""anu_v3.cancel_on_success_live_e2e — cancel-on-success live remove END-TO-END.

task-2553+48 (회장 결정, 코드/파일 자동화). Strict-additive 신규 orchestrator.

문제 (회장 §2 verbatim):
  +45 self-test 에서는 fallback binding 이 properly-bound 상태가 아니어서
  ROLE_MISMATCH 로 fallback 보존이 발생했다. 따라서 cancel-on-success live
  remove 는 아직 end-to-end 실증되지 않았다.

목표 (회장 §1/§3 verbatim):
  properly-bound 4-tuple 기반 cancel-on-success live remove end-to-end 실증.
    1. dispatch 시 +44 guard 가 normal_collector_cron_id +
       fallback_cron_id(role=fallback) 를 registry 에 정합 기록(REGISTERED),
       +47 ``write_back_completed`` 로 normal durable-success COMPLETED 갱신
       → live cron id == ledger id 일치 보장.
    2. 이 properly-bound 상태에서 +45
       ``run_cancel_on_success_live_wiring`` 실행 → +37→+25→+23 cancel seam
       1회 → live verifier 5조건 AND PASS → bound fallback cron 1건 실제
       제거 → cancel-audit JSON 생성.
    3. normal success 후 fallback 발화 0(실 제거 입증) end-to-end.

결선 (9-R.1 2-layer):
  * Layer A — 실 운영 cron 무접촉 (§5 구속). 본 모듈은 cron 을 자체
    로직으로 추가/제거하지 않는다. ZERO ``cokacdir`` / ``subprocess`` exec.
    properly-bound 4-tuple 은 +44 append-only API + +47 write-back API
    (read-only/additive 재사용) 로만 구성하고, live cron 조회/제거는
    전적으로 주입 Fake/Spy DI(전용 격리 fixture cron 세계)로만 수행한다.
    실 운영 fallback cron 강제삭제 0.
  * Layer B — bound·verified single cancel (§3 목표 자체). properly-bound
    4-tuple(live==ledger·role=fallback) + +45 live verifier 5조건 AND PASS
    = 단일 bound·verified cancel 1건 = 회장 §1 목표(end-to-end 실증) 그
    자체. mismatch/missing/unverified = 보존·decouple(+45 보수 가드
    무회귀).

무변 의무 (회장 §4/§6):
  +44/+45/+46/+47 모듈·+37/+25/+23·frozen anchor·기존 산출물 byte-0
  /additive 무변. 본 모듈은 그것들을 read-only/additive 재사용하는
  orchestrator 일 뿐이다 (신규 cancel 로직 0 — +45 권위 그대로 경유).
"""
from __future__ import annotations

from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional, Union

# +44 durable append-only 4-tuple ledger (append API — properly-bound 구성).
from anu_v3.callback_4tuple_registry import (  # pyright: ignore[reportMissingImports]  # noqa: E501
    Callback4TupleRegistry,
    make_record,
)

# +47 normal durable-success registry write-back (COMPLETED 갱신, 1회).
from anu_v3.callback_event_trigger import (  # pyright: ignore[reportMissingImports]  # noqa: E501
    write_back_completed,
)

# +45 cancel-on-success live wiring 진입점 (PRIMARY — 신규 cancel 로직 0).
from anu_v3.cancel_on_success_live_wiring import (  # pyright: ignore[reportMissingImports]  # noqa: E501
    LOOKUP_VERIFIED,
    CancelOnSuccessLiveResult,
    run_cancel_on_success_live_wiring,
)

# +45 cancel-audit 빌더/보장 writer (Layer A — NO-CRON, 디커플 file IO).
from anu_v3.cancel_audit_writer import (  # pyright: ignore[reportMissingImports]  # noqa: E501
    guaranteed_write_audit,
)

# 소유권 하드 경계 (+23 verifier 와 동일값 — read-only).
from utils.live_cron_state_verifier import (  # pyright: ignore[reportMissingImports]  # noqa: E501
    ANU_CHAT_ID,
    FALLBACK_ROLE,
)

from utils.anu_delegation_completion_callback import (  # pyright: ignore[reportMissingImports]  # noqa: E501
    CallbackInput,
    Classification,
)

TASK_MARKER_PREFIX = "task-2553+48"
E2E_AUDIT_SCHEMA = "task-2553+48.cancel-on-success-e2e-audit_v1"

# ── e2e verdicts ─────────────────────────────────────────────────────────────
E2E_PASS = "E2E_PASS"          # properly-bound + 5조건 AND PASS + cron-remove + fire 0
E2E_PRESERVED = "E2E_PRESERVED"  # mismatch/missing/unverified → 보존·decouple
E2E_DECOUPLED = "E2E_DECOUPLED"  # cron-remove 실패 → normal success decouple 유지


# ─────────────────────────────────────────────────────────────────────────────
# 전용 격리 fixture cron 세계 (Layer A — 실 운영 cron 무접촉).
#   순수 in-memory dict. cokacdir/subprocess exec 0. SpyRemover 는 이
#   세계만 변형하며 실 cron 을 절대 삭제하지 않는다(회장 §5/9-R.1 Layer A).
# ─────────────────────────────────────────────────────────────────────────────
class FakeCronWorld:
    """전용 격리 fixture cron 세계 (실 운영 cron 무접촉, NO-CRON).

    ``entries`` 는 ``{id, task_id, chat_id, role, fired, removed}`` dict
    리스트. 실 cokacdir/subprocess 를 절대 호출하지 않는다 — 본 task 의
    e2e 실증은 이 격리 세계로만 수행한다(회장 §5 "live cron 강제삭제"
    준수, Layer A).
    """

    def __init__(self, entries: List[dict]) -> None:
        # 깊은 복사 — 외부 fixture dict 원본 무변(부작용 0).
        self.entries: List[dict] = [dict(e) for e in entries]
        self.list_calls = 0
        self.fire_attempts = 0

    def _find(self, cron_id: str) -> Optional[dict]:
        for e in self.entries:
            if e.get("id") == cron_id:
                return e
        return None

    def lister(self) -> dict:
        """+23 verifier 가 호출하는 격리 cron 조회 (실 조회 0)."""
        self.list_calls += 1
        return {
            "status": "ok",
            "entries": [dict(e) for e in self.entries],
            "raw": {"fake_isolated_world": True},
        }

    def mark_removed(self, cron_id: str) -> bool:
        e = self._find(cron_id)
        if e is None:
            return False
        e["removed"] = True
        return True

    def attempt_fallback_fire(self, cron_id: str) -> bool:
        """normal success **후** fallback 발화 시도 시뮬레이션.

        실제 cron 이 제거됐으면(또는 부재) 발화 불가 → False.
        return True 는 fallback 이 실제로 발화함을 의미(회장 §3: end-to-end
        에서 0 이어야 함). 실 cron 무접촉 — 격리 세계 상태만 평가.
        """
        self.fire_attempts += 1
        e = self._find(cron_id)
        if e is None or e.get("removed") or e.get("fired"):
            return False
        # 격리 세계에서 미제거 fallback 은 발화로 간주(재현 경로 입증용).
        e["fired"] = True
        return True


class WorldSpyRemover:
    """주입 fake remover — 실 cron 삭제 0, FakeCronWorld 상태만 변형.

    +45→+37→+25→+23→+9a 경로가 operational=True 로 호출(dry_run=False).
    호출 인자를 spy 기록하고, 격리 세계의 해당 entry 만 removed 표시한다
    (실 운영 cron 무접촉 — 회장 §5/9-R.1 Layer A).
    """

    def __init__(self, world: FakeCronWorld, *, status: str = "removed"):
        self.world = world
        self.status = status
        self.calls: List[dict] = []

    def __call__(self, cron_id: str, *, dry_run: bool = True):
        # RemoverResult 는 +45 와 동일 계약 — 지연 import(순환 회피).
        from utils.completion_callback_fallback_cancel import (  # pyright: ignore[reportMissingImports]  # noqa: E501
            RemoverResult,
        )

        self.calls.append({"cron_id": cron_id, "dry_run": dry_run})
        if self.status == "removed" and not dry_run:
            self.world.mark_removed(cron_id)
        return RemoverResult(
            status=self.status, detail=f"fake-isolated:{self.status}"
        )


# ─────────────────────────────────────────────────────────────────────────────
# properly-bound 4-tuple 구성 (+44 guard 정합 기록 + +47 write-back COMPLETED).
# ─────────────────────────────────────────────────────────────────────────────
@dataclass
class ProperlyBound4Tuple:
    """dispatch 시 +44 guard 가 기록할 4-tuple (role=fallback 정합)."""

    task_id: str
    dispatch_id: str
    dispatch_cron_id: str
    executor: str
    chat_id: str
    normal_collector_cron_id: str
    fallback_callback_cron_id: str
    role: str = FALLBACK_ROLE


@dataclass
class BindingConstructionEvidence:
    """properly-bound 4-tuple 구성 증거 (+44 guard + +47 write-back)."""

    guard_registered_appended: bool
    guard_record: dict
    writeback_status: str
    writeback_appended: bool
    completed_record: Optional[dict]
    ledger_path: str
    reasons: List[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "schema": "task-2553+48.binding_construction_v1",
            "guard_registered_appended": self.guard_registered_appended,
            "guard_record": self.guard_record,
            "writeback_status": self.writeback_status,
            "writeback_appended": self.writeback_appended,
            "completed_record": self.completed_record,
            "ledger_path": self.ledger_path,
            "reasons": list(self.reasons),
        }


@dataclass
class BindingIntegrity:
    """properly-bound 무결성 (live cron id == ledger id · role=fallback)."""

    ok: bool
    completed_present: bool
    role_is_fallback: bool
    chat_owned: bool
    live_eq_ledger: bool
    marker_eq_ledger: bool
    ledger_fallback_cron_id: Optional[str]
    live_cron_id: Optional[str]
    marker_fallback_cron_id: Optional[str]
    single_live_fallback: bool = False
    reasons: List[str] = field(default_factory=list)

    def to_dict(self) -> dict:
        return {
            "schema": "task-2553+48.binding_integrity_v1",
            "ok": self.ok,
            "completed_present": self.completed_present,
            "role_is_fallback": self.role_is_fallback,
            "chat_owned": self.chat_owned,
            "single_live_fallback": self.single_live_fallback,
            "live_eq_ledger": self.live_eq_ledger,
            "marker_eq_ledger": self.marker_eq_ledger,
            "ledger_fallback_cron_id": self.ledger_fallback_cron_id,
            "live_cron_id": self.live_cron_id,
            "marker_fallback_cron_id": self.marker_fallback_cron_id,
            "reasons": list(self.reasons),
        }


@dataclass
class FallbackFireProof:
    """normal success 후 fallback 발화 0 end-to-end 입증."""

    cron_removed: bool
    post_success_fire_attempts: int
    fallback_fired_count: int
    note: str = ""

    def to_dict(self) -> dict:
        return {
            "schema": "task-2553+48.fallback_fire_proof_v1",
            "cron_removed": self.cron_removed,
            "post_success_fire_attempts": self.post_success_fire_attempts,
            "fallback_fired_count": self.fallback_fired_count,
            "note": self.note,
        }


@dataclass
class CancelOnSuccessE2EResult:
    """cancel-on-success live remove end-to-end 1회 실증 결과."""

    verdict: str
    properly_bound: bool
    binding_construction: BindingConstructionEvidence
    binding_integrity: BindingIntegrity
    durable_success: bool
    seam_invoked: bool
    cron_remove_invoked: bool
    fallback_cancelled: bool
    fallback_preserved: bool
    normal_success_unchanged: bool
    fallback_fire_proof: FallbackFireProof
    live_result: CancelOnSuccessLiveResult
    e2e_audit: dict
    e2e_audit_path: Optional[str]
    notes: List[str] = field(default_factory=list)


def build_properly_bound_4tuple(
    registry: Callback4TupleRegistry,
    tup: ProperlyBound4Tuple,
    *,
    ts_kst: str = "",
) -> BindingConstructionEvidence:
    """dispatch 시 +44 guard 정합 기록 + +47 write-back COMPLETED 갱신.

    절차 (회장 §3.1):
      1. +44 ``make_record`` 로 REGISTERED 4-tuple(role=fallback,
         normal_collector_cron_id + fallback_callback_cron_id 정합) 을
         append (durable guard 기록 — +44 append-only API only, byte-0).
      2. +47 ``write_back_completed`` 로 normal durable-success COMPLETED
         line append (verified normal-collector identity 바인딩 — naive
         mark_completed 의 stale role/null copy 결함 회피).

    → ledger 최신 record = COMPLETED·role=fallback·fallback_callback_cron_id
    bound (live cron id == ledger id 일치 보장의 ledger 측 근거).
    """
    reasons: List[str] = []

    # ── 1. +44 guard 정합 기록 (REGISTERED, role=fallback) ────────────────
    guard = make_record(
        task_id=tup.task_id,
        dispatch_id=tup.dispatch_id,
        dispatch_cron_id=tup.dispatch_cron_id,
        executor=tup.executor,
        chat_id=str(tup.chat_id),
        normal_collector_cron_id=tup.normal_collector_cron_id,
        fallback_callback_cron_id=tup.fallback_callback_cron_id,
        role=tup.role,
        status="REGISTERED",
        no_fallback=False,
        ts_kst=ts_kst,
    )
    registry.append(guard)
    reasons.append(
        "+44 guard: REGISTERED 4-tuple append (role=fallback, "
        "normal_collector+fallback 정합) — append-only API only (+44 byte-0)"
    )

    # ── 2. +47 write_back_completed (COMPLETED, verified binding) ─────────
    wb = write_back_completed(
        registry,
        task_id=tup.task_id,
        dispatch_id=tup.dispatch_id,
        dispatch_cron_id=tup.dispatch_cron_id,
        executor=tup.executor,
        chat_id=str(tup.chat_id),
        normal_collector_cron_id=tup.normal_collector_cron_id,
        fallback_callback_cron_id=tup.fallback_callback_cron_id,
        role=tup.role,
        no_fallback=False,
        ts_kst=ts_kst,
    )
    reasons.append(
        f"+47 write_back_completed → {wb.status} "
        f"(appended={wb.appended}) — normal durable-success COMPLETED 갱신, "
        "naive mark_completed stale role/null copy 결함 회피"
    )

    completed = registry.latest_for(tup.task_id)
    return BindingConstructionEvidence(
        guard_registered_appended=True,
        guard_record=guard.to_json(),
        writeback_status=wb.status,
        writeback_appended=wb.appended,
        completed_record=completed.to_json() if completed else None,
        ledger_path=str(registry.ledger_path),
        reasons=reasons,
    )


def verify_4tuple_binding_integrity(
    registry: Callback4TupleRegistry,
    *,
    task_id: str,
    expected_chat_id: Union[int, str] = ANU_CHAT_ID,
    expected_role: str = FALLBACK_ROLE,
    live_cron_entries: List[dict],
    dispatch_marker: dict,
) -> BindingIntegrity:
    """properly-bound 무결성 검증 — live cron id == ledger id · role=fallback.

    회장 §3.1 "live cron id == ledger id 일치 보장" 의 코드 검증:
      * ledger 최신 record = COMPLETED (+47 write-back 적용됨)
      * record.role == fallback, record.chat_id == ANU
      * record.fallback_callback_cron_id == live cron entry.id  (live==ledger)
      * record.fallback_callback_cron_id == marker.callback_policy_a.
        fallback_callback_cron_id                                (==marker)

    하나라도 불만족 → ``ok=False`` → +45 보수 가드 경로(보존·decouple).
    """
    reasons: List[str] = []
    rec = registry.latest_for(task_id)
    completed_present = bool(rec is not None and rec.status == "COMPLETED")
    if not completed_present:
        reasons.append(
            "ledger 최신 record 가 COMPLETED 아님 → +47 write-back 미적용"
        )
    role_is_fallback = bool(rec is not None and (rec.role or "") == expected_role)
    if not role_is_fallback:
        reasons.append(
            f"ledger role != {expected_role!r} → properly-bound 아님"
        )
    chat_owned = bool(
        rec is not None and str(rec.chat_id) == str(expected_chat_id)
    )
    if not chat_owned:
        reasons.append("ledger chat_id 가 ANU 소유 아님")

    ledger_fid = rec.fallback_callback_cron_id if rec is not None else None

    # Codex F1 (HIGH) 수용: task/role 매칭 fallback live entry 가 정확히
    # 1건이어야 properly-bound. 2건 이상이면 한 건을 제거해도 sibling
    # fallback 이 armed 로 남아 single bound·verified 보장이 깨진다
    # (회장 §1 "bound·verified single cancel 1건"·Layer B). 모호하면
    # ok=False → 보존·decouple (+45 보수 가드 경로).
    matches = [
        e
        for e in live_cron_entries
        if e.get("task_id") == task_id and e.get("role") == expected_role
    ]
    single_live_fallback = len(matches) == 1
    live_id = matches[0].get("id") if matches else None
    if not single_live_fallback:
        reasons.append(
            f"task/role 매칭 live fallback {len(matches)}건 — 정확히 1건이어야 "
            "properly-bound (모호 binding → 보존, Codex F1 수용)"
        )
    live_eq_ledger = bool(
        single_live_fallback and ledger_fid and live_id and live_id == ledger_fid
    )
    if not live_eq_ledger and single_live_fallback:
        reasons.append(
            f"live cron id({live_id!r}) != ledger id({ledger_fid!r}) "
            "→ properly-bound 4-tuple 불성립"
        )

    pol = (
        dispatch_marker.get("callback_policy_a")
        if isinstance(dispatch_marker, dict)
        else None
    )
    marker_fid = (
        pol.get("fallback_callback_cron_id")
        if isinstance(pol, dict)
        else None
    )
    marker_eq_ledger = bool(
        ledger_fid and marker_fid and marker_fid == ledger_fid
    )
    if not marker_eq_ledger:
        reasons.append(
            f"marker fallback id({marker_fid!r}) != ledger id"
            f"({ledger_fid!r}) → 4-tuple 정합 실패"
        )

    ok = (
        completed_present
        and role_is_fallback
        and chat_owned
        and single_live_fallback
        and live_eq_ledger
        and marker_eq_ledger
    )
    return BindingIntegrity(
        ok=ok,
        single_live_fallback=single_live_fallback,
        completed_present=completed_present,
        role_is_fallback=role_is_fallback,
        chat_owned=chat_owned,
        live_eq_ledger=live_eq_ledger,
        marker_eq_ledger=marker_eq_ledger,
        ledger_fallback_cron_id=ledger_fid,
        live_cron_id=live_id,
        marker_fallback_cron_id=marker_fid,
        reasons=reasons or ["properly-bound 4-tuple 무결성 전수 통과"],
    )


def _five_conditions(live_result: CancelOnSuccessLiveResult) -> Optional[dict]:
    wr = live_result.wired_result
    so = wr.wiring_result.seam_outcome if (wr and wr.wiring_result) else None
    if so is None:
        return None
    lv = so.live_verification or {}
    return lv.get("checks")


def run_cancel_on_success_live_e2e(
    inp: CallbackInput,
    ack_path: Union[str, Path],
    tup: ProperlyBound4Tuple,
    *,
    dispatch_fired_marker_path: Union[str, Path],
    result_json_path: Union[str, Path],
    report_path: Union[str, Path],
    collector_result_marker_path: Union[str, Path],
    claim_dir: Union[str, Path],
    ledger_path: Union[str, Path],
    dispatch_marker: dict,
    world: FakeCronWorld,
    remover: WorldSpyRemover,
    e2e_audit_path: Optional[Union[str, Path]] = None,
    fallback_cancelled_marker_path: Optional[Union[str, Path]] = None,
    cancel_lock_path: Optional[Union[str, Path]] = None,
    seam_audit_path: Optional[Union[str, Path]] = None,
    callback_contract: Optional[dict] = None,
    expected_chat_id: Union[int, str] = ANU_CHAT_ID,
    expected_role: str = FALLBACK_ROLE,
    post_success_fire_attempts: int = 3,
    skip_binding_construction: bool = False,
    ts_kst: str = "",
) -> CancelOnSuccessE2EResult:
    """properly-bound 4-tuple 기반 cancel-on-success live remove end-to-end.

    절차 (회장 §3):
      1. +44 guard 정합 기록 + +47 write-back COMPLETED → properly-bound
         4-tuple 구성 (``skip_binding_construction`` 시 외부 ledger 상태
         그대로 — mismatch/missing 보수 가드 무회귀 재현용).
      2. properly-bound 무결성 검증 (live cron id == ledger id·role=
         fallback). 불만족이어도 +45 가 보수적으로 보존하는지(무회귀)
         그대로 진행 — 본 모듈이 cancel 을 강제하지 않는다.
      3. +45 ``run_cancel_on_success_live_wiring`` **경유** (PRIMARY —
         신규 cancel 로직 0). +37→+25→+23 seam 1회 → live verifier 5조건
         AND → bound fallback 1건 cron-remove(주입 WorldSpyRemover, 실
         운영 cron 무접촉).
      4. normal success 후 fallback 발화 0 입증: 제거 후 격리 세계에서
         fallback 발화를 ``post_success_fire_attempts`` 회 시도 → 전부
         미발화(0) 여야 end-to-end 성립.
      5. e2e cancel-audit JSON 생성·보장 기록. seam/cron-remove 의 어떤
         예외·skip·실패도 normal collector success 를 뒤집지 않는다(디커플
         — +45 권위 그대로, normal_success_unchanged=True 불변).
    """
    registry = Callback4TupleRegistry(Path(ledger_path))

    # ── 1. properly-bound 4-tuple 구성 ────────────────────────────────────
    if skip_binding_construction:
        rec = registry.latest_for(inp.task_id)
        construction = BindingConstructionEvidence(
            guard_registered_appended=False,
            guard_record={},
            writeback_status="SKIPPED_EXTERNAL_LEDGER",
            writeback_appended=False,
            completed_record=rec.to_json() if rec else None,
            ledger_path=str(registry.ledger_path),
            reasons=[
                "binding 구성 skip — 외부 ledger 상태 그대로 (mismatch/"
                "missing 보수 가드 무회귀 재현)"
            ],
        )
    else:
        construction = build_properly_bound_4tuple(
            registry, tup, ts_kst=ts_kst
        )

    # ── 2. properly-bound 무결성 검증 ─────────────────────────────────────
    integrity = verify_4tuple_binding_integrity(
        registry,
        task_id=inp.task_id,
        expected_chat_id=expected_chat_id,
        expected_role=expected_role,
        live_cron_entries=world.entries,
        dispatch_marker=dispatch_marker,
    )

    # ── 3. +45 cancel-on-success live wiring 경유 (PRIMARY) ───────────────
    live_result = run_cancel_on_success_live_wiring(
        inp,
        ack_path,
        dispatch_fired_marker_path=dispatch_fired_marker_path,
        result_json_path=result_json_path,
        report_path=report_path,
        collector_result_marker_path=collector_result_marker_path,
        claim_dir=claim_dir,
        ledger_path=ledger_path,
        cron_lister=world.lister,
        remover=remover,
        fallback_cancelled_marker_path=fallback_cancelled_marker_path,
        cancel_lock_path=cancel_lock_path,
        seam_audit_path=seam_audit_path,
        callback_contract=callback_contract,
        expected_chat_id=expected_chat_id,
        expected_role=expected_role,
    )

    so = (
        live_result.wired_result.wiring_result.seam_outcome
        if (
            live_result.wired_result
            and live_result.wired_result.wiring_result
        )
        else None
    )
    fallback_cancelled = bool(so.fallback_cancelled) if so else False

    # ── 4. normal success 후 fallback 발화 0 입증 (end-to-end) ────────────
    target = tup.fallback_callback_cron_id
    cron_removed = any(
        e.get("id") == target and e.get("removed")
        for e in world.entries
    )
    # Codex F1 (HIGH) 수용: bound target 1건만이 아니라 task 의 **모든**
    # fallback live entry 에 대해 발화를 시도한다. sibling fallback 이
    # armed 로 남아있으면 발화하여 fire count > 0 → E2E_PASS 차단(회장 §1
    # bound·verified single cancel 보장 — 한 건만 지워도 다른 fallback 이
    # 발화하면 end-to-end 실증 실패).
    fallback_ids: List[str] = [
        str(e.get("id"))
        for e in world.entries
        if e.get("task_id") == inp.task_id
        and e.get("role") == expected_role
        and e.get("id")
    ]
    if target not in fallback_ids:
        fallback_ids.append(target)
    fired = 0
    for _ in range(max(0, post_success_fire_attempts)):
        for fid in fallback_ids:
            if world.attempt_fallback_fire(fid):
                fired += 1
    fire_proof = FallbackFireProof(
        cron_removed=cron_removed,
        post_success_fire_attempts=post_success_fire_attempts,
        fallback_fired_count=fired,
        note=(
            "normal durable-success 후 bound fallback cron 이 실제 제거되어 "
            "이후 발화 시도 전부 미발화(0) — end-to-end 입증 (회장 §3)"
            if cron_removed and fired == 0
            else (
                "fallback 미제거(보존) — properly-bound 불성립/5조건 미충족 "
                "보수 가드 경로. normal success 는 디커플로 무변(+45 무회귀)"
            )
        ),
    )

    durable_success = bool(live_result.durable_success)

    # ── 5. verdict ───────────────────────────────────────────────────────
    if (
        not skip_binding_construction
        and integrity.ok
        and live_result.lookup.status == LOOKUP_VERIFIED
        and live_result.seam_invoked
        and live_result.cron_remove_invoked
        and fallback_cancelled
        and cron_removed
        and fired == 0
        and durable_success
        and live_result.normal_success_unchanged
    ):
        verdict = E2E_PASS
    elif (
        live_result.seam_invoked
        and live_result.cron_remove_invoked
        and not fallback_cancelled
    ):
        # seam 진입·remove 시도했으나 remover 실패 — normal success decouple.
        verdict = E2E_DECOUPLED
    else:
        verdict = E2E_PRESERVED

    e2e_audit = _build_e2e_audit(
        live_result=live_result,
        construction=construction,
        integrity=integrity,
        fire_proof=fire_proof,
        verdict=verdict,
        properly_bound=integrity.ok and not skip_binding_construction,
        target_cron_id=target,
    )
    audit_written = _write_e2e_audit(
        e2e_audit,
        claim_dir=claim_dir,
        e2e_audit_path=e2e_audit_path,
        event_id=live_result.event_id,
    )

    return CancelOnSuccessE2EResult(
        verdict=verdict,
        properly_bound=integrity.ok and not skip_binding_construction,
        binding_construction=construction,
        binding_integrity=integrity,
        durable_success=durable_success,
        seam_invoked=live_result.seam_invoked,
        cron_remove_invoked=live_result.cron_remove_invoked,
        fallback_cancelled=fallback_cancelled,
        fallback_preserved=live_result.fallback_preserved,
        normal_success_unchanged=live_result.normal_success_unchanged,
        fallback_fire_proof=fire_proof,
        live_result=live_result,
        e2e_audit=e2e_audit,
        e2e_audit_path=audit_written,
        notes=[
            "properly-bound 4-tuple = +44 guard 정합 기록 + +47 "
            "write_back_completed COMPLETED (live cron id == ledger id)",
            "+45 run_cancel_on_success_live_wiring 경유 — 신규 cancel "
            "로직 0, +37→+25→+23 seam·5조건 verifier 권위 그대로",
            "Layer A: 실 운영 cron 무접촉(격리 FakeCronWorld/Spy) / "
            "Layer B: bound·verified single cancel 1건 (회장 §1 목표 자체)",
            "디커플: normal_success_unchanged 절대불변 — cancel 실패/skip/"
            "exception 이 normal success 를 뒤집지 않음 (+45 권위 상속)",
        ],
    )


def _build_e2e_audit(
    *,
    live_result: CancelOnSuccessLiveResult,
    construction: BindingConstructionEvidence,
    integrity: BindingIntegrity,
    fire_proof: FallbackFireProof,
    verdict: str,
    properly_bound: bool,
    target_cron_id: str,
) -> dict:
    """e2e cancel-audit 단일 레코드 (schema 정합).

    ``normal_success_unchanged`` 는 디커플 절대불변 — 무엇이 와도 True
    (회장 §3: cancel 실패/skip/exception 이 normal success 를 뒤집지 않음).
    """
    ca = live_result.cancel_audit if isinstance(
        live_result.cancel_audit, dict
    ) else {}
    return {
        "schema": E2E_AUDIT_SCHEMA,
        "event_id": live_result.event_id or f"{live_result.lookup.task_id}|e2e",
        "task_id": live_result.lookup.task_id,
        "target_cron_id": target_cron_id or "<UNBOUND>",
        "verdict": verdict,
        "properly_bound": bool(properly_bound),
        "lookup_status": live_result.lookup.status,
        "binding_construction": construction.to_dict(),
        "binding_integrity": integrity.to_dict(),
        "cancel_audit": ca,
        "five_condition_results": _five_conditions(live_result),
        "seam_invoked": bool(live_result.seam_invoked),
        "cron_remove_invoked": bool(live_result.cron_remove_invoked),
        "fallback_cancelled": bool(
            ca.get("fallback_cancelled", False)
        ),
        "fallback_preserved": bool(live_result.fallback_preserved),
        "fallback_fire_proof": fire_proof.to_dict(),
        "wired_via_operational_collector_wiring": bool(
            live_result.wired_via_operational_collector_wiring
        ),
        # Codex F2 (HIGH) 수용: 하드코딩 True 가 아닌 +45 가 관측한 실제
        # 값을 그대로 직렬화한다. +45 구조적 디커플 보장으로 항상 True 가
        # 기대되며, 만약 False 면 schema(const true) 검증이 큰 소리로
        # 실패하여 디커플 회귀를 은폐가 아닌 표면화한다(검증 게이트).
        "normal_success_unchanged": bool(
            live_result.normal_success_unchanged
        ),
        "canonical_root": live_result.lookup.canonical_root,
        "ledger_path": live_result.lookup.ledger_path,
        "notes": [
            "task-2553+48 cancel-on-success live remove END-TO-END 실증",
            "properly-bound 4-tuple(+44 guard + +47 write-back) → +45 "
            "wiring → +37→+25→+23 seam → 5조건 AND → bound fallback "
            "cron-remove → normal success 후 fallback 발화 0",
        ],
        "ts_utc": ca.get("ts_utc"),
    }


def _write_e2e_audit(
    audit: dict,
    *,
    claim_dir: Union[str, Path],
    e2e_audit_path: Optional[Union[str, Path]],
    event_id: str,
) -> Optional[str]:
    """e2e cancel-audit 다중 후보 보장 기록 (Layer A — NO-CRON, 디커플 IO)."""
    primary = (
        Path(e2e_audit_path)
        if e2e_audit_path is not None
        else Path(claim_dir) / f"{TASK_MARKER_PREFIX}.cancel-audit.json"
    )
    safe_eid = "".join(c if c.isalnum() else "_" for c in (event_id or ""))[:80]
    return guaranteed_write_audit(
        audit,
        primary=primary,
        fallbacks=[
            Path(claim_dir)
            / f"{TASK_MARKER_PREFIX}.cancel-audit.{safe_eid}.json"
        ],
    )


__all__ = [
    "TASK_MARKER_PREFIX",
    "E2E_AUDIT_SCHEMA",
    "E2E_PASS",
    "E2E_PRESERVED",
    "E2E_DECOUPLED",
    "FakeCronWorld",
    "WorldSpyRemover",
    "ProperlyBound4Tuple",
    "BindingConstructionEvidence",
    "BindingIntegrity",
    "FallbackFireProof",
    "CancelOnSuccessE2EResult",
    "build_properly_bound_4tuple",
    "verify_4tuple_binding_integrity",
    "run_cancel_on_success_live_e2e",
]

# Classification re-export 회피 — +45/+37/+25 권위 그대로 사용 (lint 정합).
_ = Classification