"""tests/fixtures/cancel_on_success_live_observation_harness_2553plus41.py

task-2553+41 — TRACK D: +37 CANCEL-ON-SUCCESS LIVE OBSERVATION FIXTURE
(read-only observation only).

목적 (task-2553+41 §1 / §3.1, 회장 verbatim 의도):
  +37 wired entrypoint(``run_wired_normal_completion_callback_collector``)의
  cancel-on-success 동작을 **read-only 로 관측**하는 mock/fixture harness 를
  사전 구현한다. 향후 실 post-+37 task callback 사이클이 도래하면 본 harness
  로 6-step:

      ① normal success
      ② wired entrypoint
      ③ operational_collector_wiring (PRIMARY 경유)
      ④ run_operational_cancel_seam (operational=True)
      ⑤ live verifier 5조건 AND PASS
      ⑥ bound fallback cron-remove + cancel-audit

  을 결정적(passive)으로 검증할 수 있다. md 박제만 금지 — 실행 fixture·
  observation harness·regression 산출.

callback/collector 경로 추가 수정 0 (회장 §1 / §3.1 / §5):
  * 본 harness 는 +37 entrypoint·+25 wiring·+23 seam·+9a·verifier·frozen
    anchor 를 **read-only import / parse / reference only** 한다 (1 byte
    수정 0 — strict-additive 신규 모듈). 6-step 은 +37 공개 API
    ``run_wired_normal_completion_callback_collector`` 에 **주입 Fake/Spy +
    격리 sandbox FS** 만 물려 관측한다.

mock-only / 실 운영 무접촉 (§5 / §7 / 9-R):
  * 실 cron-list·실 cron-remove·실 schedule_history·실 callback 4-tuple·
    실 cron 발화 **무접촉**. contract/+37 산출은 read-only 참조만.
  * 격리 강제: ① subprocess 전면 차단 ② 실 cron API entrypoint
    (RealCokacdirCronLister/Remover) 차단 ③ ``/usr/local/bin/cokacdir``
    바이너리 접근 차단 ④ live schedule_history / live workspace mutation
    을 sandbox + read-only allowlist 로 제한.
  * exact-once = 명시 claim artifact/key (event_id 기반 격리 marker,
    ``O_CREAT|O_EXCL`` create-if-absent) + first-winner=1 invoke·
    second=즉시 no-op. 약한 in-memory mutex 대체 0.

본 모듈 자체는 실 cron/callback 을 1회도 등록·제거·발화하지 않는다. 실
task 적용 시에도 본 harness 는 passive observer — cron-list 1회(주입
fake)·schedule_history read-only(주입 spy) 만 수행하며 write/cron-remove/
dispatch 0. 실 cron-remove 는 결선완료·verifier 5조건 PASS 후 운영단계
동작이며 본 task 행위가 아니다 (+37 모듈 docstring §62~65 와 동형).
"""
from __future__ import annotations

import builtins
import json
import os
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

WORKSPACE = Path(__file__).resolve().parent.parent.parent
FROZEN_ANCHOR = WORKSPACE / "utils" / "anu_delegation_completion_callback.py"
FROZEN_SHA = (
    "83b3e307c8207c76a3e311c408aab4951373bd317896e51687d3007907b0c3d4"
)

#: read-only consume 대상 (+37 PRIMARY 결선 경로) — 무수정 검증용.
WIRED_ENTRYPOINT_SRC = (
    WORKSPACE / "utils" / "normal_completion_callback_collector_entrypoint.py"
)
PLUS25_WIRING_SRC = WORKSPACE / "utils" / "operational_collector_wiring.py"

#: 실 운영 경로 — 본 harness 는 이 경로들을 read·write·register·remove 0.
LIVE_SCHEDULE_HISTORY = Path("/home/jay/.cokacdir/schedule_history")
LIVE_COKACDIR_BIN = Path("/usr/local/bin/cokacdir")
ANU_CHAT_ID = 6937032012

#: 향후 실 task 적용 시 본 harness 가 산출하는 정상수렴 마커 네임스페이스.
#: (task-2553+41.* = git-untracked / zero-overlap, 실 cron 무접촉.)
TASK_MARKER_PREFIX = "task-2553+41"

# ── +37 wired entrypoint / +25 wiring / frozen anchor — READ-ONLY import ──────
#    (수정·결합 0. 본 harness 는 공개 API 만 호출하는 passive observer.)
from utils.anu_delegation_completion_callback import (  # noqa: E402
    CallbackInput,
    CallbackType,
    Classification,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    ANU_CHAT_ID as _FC_ANU_CHAT_ID,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    ANU_KEY as _FC_ANU_KEY,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    FALLBACK_ROLE as _FC_FALLBACK_ROLE,
)
from utils.completion_callback_fallback_cancel import (  # noqa: E402
    RemoverResult,
)
from utils.normal_completion_callback_collector_entrypoint import (  # noqa: E402,E501
    NormalCallbackBinding,
    WiredCollectorResult,
    run_wired_normal_completion_callback_collector,
    validate_normal_callback_binding,
)


# ════════════════════════════════════════════════════════════════════════════
# Fake / Spy — 주입 전용 (실 cron·실 schedule_history·실 4-tuple 무접촉)
# ════════════════════════════════════════════════════════════════════════════
class FakeCronLister:
    """주입 fake cron-list. 실 ``cokacdir --cron-list`` 무호출.

    ``entries`` = +23 live verifier 가 보는 정규화 목록. ``drop_id_after_first``
    True 면 한 번 lister 가 호출된 *뒤* 두 번째 조회부터 대상 id 부재로
    응답하여 cron-remove 후 cron-list 교차확인을 결정적으로 재현한다.
    """

    def __init__(self, entries, *, status="ok", drop_id_after_first=None):
        self.entries = [dict(e) for e in entries]
        self.status = status
        self.drop_id_after_first = drop_id_after_first
        self.calls = 0

    def __call__(self) -> dict:
        self.calls += 1
        if self.status != "ok":
            return {"status": self.status, "entries": [], "raw": {"fake": True}}
        cur = [dict(e) for e in self.entries]
        if self.drop_id_after_first is not None and self.calls > 1:
            cur = [e for e in cur if e.get("id") != self.drop_id_after_first]
        return {"status": "ok", "entries": cur, "raw": {"fake": True}}

    def id_present(self, cron_id: str) -> bool:
        snap = self()
        return any(e.get("id") == cron_id for e in snap.get("entries", []))


class SpyRemover:
    """주입 spy cron-remover. 실 cron 제거 0 — 호출만 기록."""

    def __init__(self, status: str = "removed"):
        self.status = status
        self.calls: list = []

    def __call__(self, cron_id: str, *, dry_run: bool = True) -> RemoverResult:
        self.calls.append({"cron_id": cron_id, "dry_run": dry_run})
        return RemoverResult(status=self.status, detail=f"fake:{self.status}")


class SpyScheduleHistory:
    """주입 spy schedule_history reader. 실 schedule_history 무접촉.

    실 운영에서 fallback 발화 확인은 ``<fallback_id>.log`` 무발화 검사다.
    본 spy 는 **격리 sandbox 내부** log 파일만 read-only 조회하며 (실
    ``/home/jay/.cokacdir/schedule_history`` 경로는 절대 열지 않음), 기본
    적으로 발화 0(빈 로그/파일부재) 을 반환한다.
    """

    def __init__(self, sandbox: Path, *, fired_ids=None):
        self.sandbox = Path(sandbox)
        self.fired_ids = set(fired_ids or [])
        self.reads: list = []

    def _log_path(self, fallback_id: str) -> Path:
        # 격리 sandbox 하위만 — 실 LIVE_SCHEDULE_HISTORY 절대 미사용.
        p = (
            self.sandbox / "schedule_history" / f"{fallback_id}.log"
        ).resolve()
        if not str(p).startswith(str(self.sandbox.resolve())):
            raise AssertionError(
                "schedule_history 경로가 sandbox 밖 (격리 위반)"
            )
        return p

    def fired_count(self, fallback_id: str) -> int:
        self.reads.append(fallback_id)
        if fallback_id in self.fired_ids:
            return 1
        lp = self._log_path(fallback_id)
        if not lp.exists():
            return 0
        return sum(
            1
            for ln in lp.read_text(encoding="utf-8").splitlines()
            if ln.strip()
        )

    def no_firing(self, fallback_id: str) -> bool:
        return self.fired_count(fallback_id) == 0


# ════════════════════════════════════════════════════════════════════════════
# 격리 강제 — subprocess / 실 cron API / callback 등록 / live-path 차단
# ════════════════════════════════════════════════════════════════════════════
def install_isolation_guards(monkeypatch, sandbox: Path) -> None:
    """harness 실행 전 모든 라이브 경로를 monkeypatch/allowlist 로 차단한다.

    ① subprocess.run/Popen/call/check_output/check_call + os.system →
       즉시 AssertionError
    ② 실 cron API entrypoint: live_cron_state_verifier.RealCokacdirCronLister
       / completion_callback_fallback_cancel.RealCokacdirCronRemover __call__
       → 즉시 AssertionError (주입 Fake/Spy 만 허용)
    ③ callback 등록 entrypoint: ``/usr/local/bin/cokacdir`` 접근 차단
    ④ live schedule_history / live workspace mutation:
       open()/os.open 을 sandbox + read-only allowlist 로 제한.
    """
    sandbox = Path(sandbox).resolve()

    def _boom(*a, **k):  # noqa: ANN001, ANN002, ANN003
        raise AssertionError(
            "실 subprocess/cron/callback 호출 금지 (§5 / 9-R) — "
            "주입 Fake/Spy·격리 sandbox 만 허용 (read-only observation)"
        )

    # ① subprocess 전면 차단
    monkeypatch.setattr(subprocess, "run", _boom)
    monkeypatch.setattr(subprocess, "Popen", _boom)
    monkeypatch.setattr(subprocess, "call", _boom)
    monkeypatch.setattr(subprocess, "check_output", _boom)
    monkeypatch.setattr(subprocess, "check_call", _boom)
    monkeypatch.setattr(os, "system", _boom)

    # ② 실 cron API entrypoint 차단 (allowlist = 주입 Fake/Spy 만)
    import utils.completion_callback_fallback_cancel as _fc
    import utils.live_cron_state_verifier as _lv

    monkeypatch.setattr(
        _lv.RealCokacdirCronLister, "__call__", _boom, raising=False
    )
    monkeypatch.setattr(
        _fc.RealCokacdirCronRemover, "__call__", _boom, raising=False
    )

    # ③+④ live schedule_history / live workspace / cokacdir 바이너리 차단.
    #     read-only allowlist = +37 entrypoint·+25 wiring 소스·frozen anchor·
    #     utils/tests 소스 + sandbox.
    real_open = builtins.open
    real_os_open = os.open
    allow_read_prefixes = (
        str(WORKSPACE / "utils"),
        str(WORKSPACE / "tests"),
        str(WORKSPACE / "memory" / "fixtures"),
        str(FROZEN_ANCHOR),
    )

    def _classify(path) -> str:
        try:
            ap = str(Path(os.fspath(path)).resolve())
        except (TypeError, ValueError):
            return "deny"
        if ap.startswith(str(sandbox)):
            return "rw"
        if ap.startswith(str(LIVE_SCHEDULE_HISTORY.resolve())):
            return "deny"
        if ap.startswith(str(LIVE_COKACDIR_BIN)):
            return "deny"
        if ap.startswith(allow_read_prefixes):
            return "ro"
        return "deny"

    def _is_write(mode: str) -> bool:
        return any(c in mode for c in ("w", "a", "x", "+"))

    def _guarded_open(file, mode="r", *a, **k):  # noqa: ANN001
        cls = _classify(file)
        if cls == "rw":
            return real_open(file, mode, *a, **k)
        if cls == "ro" and not _is_write(mode):
            return real_open(file, mode, *a, **k)
        raise AssertionError(
            f"live-path 차단 (§5 / 9-R): {file!r} mode={mode!r} — "
            "harness 는 격리 sandbox·read-only allowlist 만 접근"
        )

    def _guarded_os_open(path, flags, *a, **k):  # noqa: ANN001
        cls = _classify(path)
        wr = bool(
            flags & (os.O_WRONLY | os.O_RDWR | os.O_CREAT | os.O_APPEND)
        )
        if cls == "rw":
            return real_os_open(path, flags, *a, **k)
        if cls == "ro" and not wr:
            return real_os_open(path, flags, *a, **k)
        raise AssertionError(
            f"live-path 차단 (§5 / 9-R): os.open {path!r} flags={flags} — "
            "harness 는 격리 sandbox 만 write 가능"
        )

    monkeypatch.setattr(builtins, "open", _guarded_open)
    monkeypatch.setattr(os, "open", _guarded_os_open)


# ════════════════════════════════════════════════════════════════════════════
# exact-once — 명시 claim artifact/key, O_EXCL create-if-absent mirror
# ════════════════════════════════════════════════════════════════════════════
def claim_key(event_id: str) -> str:
    """exact-once 명시 claim key (event_id 기반 격리 marker 명)."""
    return f"{TASK_MARKER_PREFIX}.seam-claim.{event_id}.json"


def attempt_exact_once_claim(claim_dir: Path, event_id: str) -> bool:
    """``O_CREAT|O_EXCL|O_WRONLY`` 원자 선점 (실 FS, in-memory mutex 아님).

    +25 ``utils.operational_collector_wiring`` claim 의 의미를 충실 mirror:
    True = first-winner(seam 1회 진입 허용), False = 이미 선점
    (retries/duplicate/concurrent) → 즉시 no-op. claim artifact 는 격리
    sandbox 내부 ``task-2553+41.*`` 네임스페이스.
    """
    claim_dir = Path(claim_dir)
    claim_dir.mkdir(parents=True, exist_ok=True)
    cp = claim_dir / claim_key(event_id)
    try:
        fd = os.open(str(cp), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    except FileExistsError:
        return False
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(
            {
                "schema": "task-2553+41.seam-claim_v1",
                "event_id": event_id,
                "atomic_create_method": "O_CREAT|O_EXCL",
            },
            f,
            ensure_ascii=False,
        )
    return True


def simulate_concurrent_double(claim_dir: Path, event_id: str) -> dict:
    """동시 2호출 emulation — first-winner=1 invoke, second=즉시 no-op.

    각 호출이 동일 event_id 로 ``attempt_exact_once_claim`` 을 시도한다.
    O_EXCL 의미상 정확히 1개만 winner. invoke 카운터로 seam 1회·no-op 1회
    를 단언 가능하게 반환한다 (약한 mutex 대체 아님).
    """
    invokes = []

    def _one(tag: str):
        claimed = attempt_exact_once_claim(claim_dir, event_id)
        if claimed:
            invokes.append(tag)  # winner 만 seam 진입 (1 invoke)
        return claimed

    first = _one("first")
    second = _one("second")
    return {
        "first_claimed": first,
        "second_claimed": second,
        "seam_invoke_count": len(invokes),
        "winner": invokes[0] if invokes else None,
        "second_is_noop": (second is False),
        "exact_once_ok": (
            first is True and second is False and len(invokes) == 1
        ),
    }


# ════════════════════════════════════════════════════════════════════════════
# 격리 fixture materialize — sandbox 내부, task-2553+41.* 네임스페이스
# ════════════════════════════════════════════════════════════════════════════
def pass_callback_input(task_id: str) -> CallbackInput:
    """frozen collector 가 PASS(durable-success) 를 산출하는 최소 입력.

    +37 regression ``_pass_input`` 과 동형 (frozen 공개 API). 실 4-tuple
    무사용 — 합성 격리 입력만.
    """
    return CallbackInput(
        task_id=task_id,
        executor="dev5-sim-2553p41",
        dispatch_cron_id="DISP2553P41",
        callback_type=CallbackType.NORMAL,
        callback_cron_id="NORM2553P41",
        cron_status="ok",
        task_status="completed",
        required_closeout_markers={"result_json": True, "report": True},
        preservation_anchors={"frozen_anchor": "match"},
        dev_sunset=True,
    )


def make_binding(
    task_id: str,
    target_cron_id: str,
    *,
    dispatch_cron_id: str = "DISP2553P41",
    normal_collector_cron_id: str = "NORM2553P41",
    fallback_cron_id: str = "__bound__",
    chat_id: int = ANU_CHAT_ID,
) -> NormalCallbackBinding:
    """+37 NormalCallbackBinding 합성 (fallback_cron_id 필수 bound).

    ``fallback_cron_id == "__bound__"`` 면 ``target_cron_id`` 로 바인드되어
    dispatch-fired marker 와 일치(binding-valid) 한다. None/"" 또는 다른 값
    이면 binding-invalid (missing/marker mismatch) 경로를 결정적으로 재현.
    """
    fid = target_cron_id if fallback_cron_id == "__bound__" else fallback_cron_id
    return NormalCallbackBinding(
        task_id=task_id,
        dispatch_cron_id=dispatch_cron_id,
        normal_collector_cron_id=normal_collector_cron_id,
        fallback_cron_id=fid,
        chat_id=chat_id,
    )


def materialize_scenario(
    sandbox: Path,
    *,
    task_id: str = "task-2553+41-obs",
    target_cron_id: str = "FBOBS2553P41",
    verifier_entries=None,
    drop_id_after_first: Optional[str] = None,
) -> dict:
    """격리 sandbox 에 +37 6-step 관측용 fixture 파일을 만든다.

    실 schedule_history / 실 4-tuple / 실 cron 무사용 — 전부 sandbox 내부.
    dispatch-fired marker 의 ``callback_policy_a.fallback_callback_cron_id``
    = ``target_cron_id`` 로, binding-valid 경로에서 +37 binding cross-check
    가 통과하도록 합성한다 (전부 격리 mock — 실 callback 4-tuple 무사용).
    """
    sandbox = Path(sandbox)
    sandbox.mkdir(parents=True, exist_ok=True)

    dfm = sandbox / f"{task_id}.dispatch-fired.json"
    dfm.write_text(
        json.dumps(
            {
                "schema": "dispatch_fired_v1",
                "marker_kind": "dispatch fired",
                "dispatch_cron_id": "DISP2553P41",
                "task_id": task_id,
                # +37 binding cross-check + +23 verifier c4 + +9a safe-remove
                # 5조건 교차입력 (callback_policy_a 단일 권위). 전부 격리
                # 합성 mock — 실 callback 4-tuple/실 cron 무사용.
                "callback_policy_a": {
                    "standardized": True,
                    "normal_callback_cron_id": "NORM2553P41",
                    "fallback_callback_cron_id": target_cron_id,
                    "fallback_role": _FC_FALLBACK_ROLE,
                    "chat_id": _FC_ANU_CHAT_ID,
                    "anu_key": _FC_ANU_KEY,
                },
            }
        ),
        encoding="utf-8",
    )
    rj = sandbox / f"{task_id}.result.json"
    rj.write_text(
        json.dumps(
            {
                "schema": "anu_delegation_result_v1",
                "task_id": task_id,
                "status": "completed",
                "classification": "PASS",
            }
        ),
        encoding="utf-8",
    )
    rep = sandbox / f"{task_id}.report.md"
    rep.write_text(
        f"# {task_id} report\n\nnormal collector PASS — durable-success, "
        "wired via operational_collector_wiring (read-only observation).\n",
        encoding="utf-8",
    )
    crm = sandbox / f"{task_id}.collector-result.json"
    crm.write_text(
        json.dumps(
            {
                "schema": "callback_ack_dedupe_v1",
                "task_id": task_id,
                "classification": "PASS",
                "ack_acquired": True,
            }
        ),
        encoding="utf-8",
    )

    if verifier_entries is None:
        verifier_entries = [
            {
                "id": target_cron_id,
                "task_id": task_id,
                "chat_id": ANU_CHAT_ID,
                "role": "fallback",
                "fired": False,
                "removed": False,
            }
        ]

    return {
        "task_id": task_id,
        "target_cron_id": target_cron_id,
        "sandbox": sandbox,
        "dispatch_fired_marker_path": dfm,
        "result_json_path": rj,
        "report_path": rep,
        "collector_result_marker_path": crm,
        "fallback_cancelled_marker_path": sandbox
        / f"{task_id}.fallback-cancelled.json",
        "cancel_lock_path": sandbox / f"{task_id}.cancel.lock",
        "seam_audit_path": sandbox / f"{task_id}.plus23-cancel-audit.json",
        "cancel_audit_path": sandbox / f"{task_id}.plus41-cancel-audit.json",
        "claim_dir": sandbox / "claims",
        "ack_path": sandbox / f"{task_id}.ack.json",
        "callback_contract": {
            "schema": "completion_callback_contract_v1",
            "fallback_callback_cron_id": target_cron_id,
        },
        "verifier_entries": verifier_entries,
        "drop_id_after_first": drop_id_after_first,
    }


# ════════════════════════════════════════════════════════════════════════════
# 6-step 관측 — §3.1 6-step ↔ +37 wired entrypoint 매핑 (passive, mock-only)
# ════════════════════════════════════════════════════════════════════════════
@dataclass
class SixStepObservation:
    """§3.1 6-step 관측 결과 (전부 mock/fixture 유래, +37 entrypoint 경유)."""

    step1_normal_success: bool
    step2_wired_entrypoint: bool
    step3_operational_collector_wiring: bool
    step4_run_operational_cancel_seam: bool
    step5_live_verifier_five_and_pass: bool
    step6_bound_fallback_remove_and_audit: bool
    all_pass: bool
    seam_invoke_count: int
    remove_call_count: int
    binding_valid: bool
    cancel_audit: dict = field(default_factory=dict)
    mock_only: bool = True
    notes: list = field(default_factory=list)

    def step_map(self) -> dict:
        return {
            "step1": self.step1_normal_success,
            "step2": self.step2_wired_entrypoint,
            "step3": self.step3_operational_collector_wiring,
            "step4": self.step4_run_operational_cancel_seam,
            "step5": self.step5_live_verifier_five_and_pass,
            "step6": self.step6_bound_fallback_remove_and_audit,
        }


_AUDIT_REQUIRED_KEYS = (
    "schema",
    "event_id",
    "five_condition_results",
    "remove_attempted",
    "remove_result",
    "skip_reason",
    "already_removed_or_missing",
    "normal_success_unchanged",
)
_FIVE_KEYS = (
    "c1_task_id_match",
    "c2_chat_id_owned",
    "c3_role_fallback",
    "c4_marker_id_crosscheck",
    "c5_pending_not_fired_not_removed",
)


def _audit_schema_ok(audit: dict) -> bool:
    if not all(k in audit for k in _AUDIT_REQUIRED_KEYS):
        return False
    five = audit.get("five_condition_results") or {}
    if not all(k in five for k in _FIVE_KEYS):
        return False
    return audit.get("normal_success_unchanged") is True


def observe_six_step(
    scenario: dict,
    *,
    binding: NormalCallbackBinding,
    lister: FakeCronLister,
    remover: SpyRemover,
    history: SpyScheduleHistory,
) -> SixStepObservation:
    """§3.1 6-step 을 결정적으로 passive 관측한다 (mock-only).

    +37 표준 wired entrypoint ``run_wired_normal_completion_callback_collector``
    에 주입 Fake/Spy + 격리 FS 만 물려 1회 실행한 뒤, 그 공개 결과를
    §3.1 6-step 으로 매핑한다. 실 cron-list/cron-remove/schedule_history/
    4-tuple 무접촉, +37/+25/+23/frozen 무수정 (read-only consume).
    """
    res: WiredCollectorResult = run_wired_normal_completion_callback_collector(
        pass_callback_input(scenario["task_id"]),
        scenario["ack_path"],
        binding=binding,
        dispatch_fired_marker_path=scenario["dispatch_fired_marker_path"],
        result_json_path=scenario["result_json_path"],
        report_path=scenario["report_path"],
        collector_result_marker_path=scenario[
            "collector_result_marker_path"
        ],
        claim_dir=scenario["claim_dir"],
        cancel_audit_path=scenario["cancel_audit_path"],
        cron_lister=lister,
        remover=remover,
        fallback_cancelled_marker_path=scenario[
            "fallback_cancelled_marker_path"
        ],
        cancel_lock_path=scenario["cancel_lock_path"],
        seam_audit_path=scenario["seam_audit_path"],
        callback_contract=scenario["callback_contract"],
    )
    audit = res.cancel_audit or {}
    five = audit.get("five_condition_results") or {}

    # ① normal success — frozen collector durable-success (디커플 불변).
    s1 = (
        res.collector_result.classification == Classification.PASS
        and bool(res.durable_success)
    )
    # ② wired entrypoint — +37 표준 entrypoint 가 binding-valid 로 실행됨.
    s2 = bool(res.binding_valid) and not res.binding_invalid_reasons
    # ③ operational_collector_wiring — PRIMARY 경유 (+25 wiring 결과 보유).
    s3 = (
        bool(res.wired_via_operational_collector_wiring)
        and res.wiring_result is not None
    )
    # ④ run_operational_cancel_seam(operational=True) — seam 1회 + claim 1개.
    claim_dir = scenario["claim_dir"]
    claim_files = (
        sorted(p.name for p in Path(claim_dir).glob("*.seam-claim.*.json"))
        if Path(claim_dir).exists()
        else []
    )
    s4 = bool(res.seam_invoked) and len(claim_files) == 1
    # ⑤ live verifier 5조건 AND PASS.
    s5 = bool(five) and all(five.get(k) is True for k in _FIVE_KEYS)
    # ⑥ bound fallback cron-remove + cancel-audit.
    remove_calls = list(remover.calls)
    removed_ok = (
        len(remove_calls) == 1
        and remove_calls[0]["cron_id"] == scenario["target_cron_id"]
        and remove_calls[0]["dry_run"] is False  # operational=True
        and audit.get("remove_result") == "CANCELLED"
    )
    crosscheck_absent = not lister.id_present(scenario["target_cron_id"])
    s6 = (
        bool(res.cron_remove_invoked)
        and res.fallback_preserved is False
        and removed_ok
        and crosscheck_absent
        and _audit_schema_ok(audit)
    )
    # passive 부수확인 — fallback 발화 0 (격리 schedule_history spy).
    no_fire = history.no_firing(scenario["target_cron_id"])

    all_pass = all((s1, s2, s3, s4, s5, s6)) and no_fire
    return SixStepObservation(
        step1_normal_success=s1,
        step2_wired_entrypoint=s2,
        step3_operational_collector_wiring=s3,
        step4_run_operational_cancel_seam=s4,
        step5_live_verifier_five_and_pass=s5,
        step6_bound_fallback_remove_and_audit=s6,
        all_pass=all_pass,
        seam_invoke_count=len(claim_files),
        remove_call_count=len(remove_calls),
        binding_valid=bool(res.binding_valid),
        cancel_audit=audit,
        mock_only=True,
        notes=[
            "+37 wired entrypoint(run_wired_normal_completion_callback_"
            "collector) 공개 API 에 주입 Fake/Spy — read-only consume, "
            "1 byte 수정 0",
            "PRIMARY 경유: entrypoint → operational_collector_wiring(+25) → "
            "run_operational_cancel_seam(+23) → live verifier 5조건 AND",
            "실 cron-list/cron-remove/schedule_history/4-tuple 무접촉, "
            f"fallback 발화 0 (spy reads={history.reads})",
            f"claim artifact(O_EXCL) {claim_files} — exact-once 명시 key",
            "디커플: collector_result 는 seam/cron-remove 결과와 무관 — "
            f"normal_success_unchanged={audit.get('normal_success_unchanged')}",
        ],
    )


def observe_binding_invalid(
    scenario: dict,
    *,
    binding: NormalCallbackBinding,
    lister: FakeCronLister,
    remover: SpyRemover,
) -> WiredCollectorResult:
    """binding-invalid (missing/marker mismatch) 경로 passive 관측.

    +37 §2.5 — seam 미진입·cron-remove 0·fallback 보존·BINDING_* cancel-
    audit·디커플(collector_result 그대로) 을 read-only 로 확인하기 위한
    헬퍼. 실 cron 무접촉 (주입 Fake/Spy).
    """
    return run_wired_normal_completion_callback_collector(
        pass_callback_input(scenario["task_id"]),
        scenario["ack_path"],
        binding=binding,
        dispatch_fired_marker_path=scenario["dispatch_fired_marker_path"],
        result_json_path=scenario["result_json_path"],
        report_path=scenario["report_path"],
        collector_result_marker_path=scenario[
            "collector_result_marker_path"
        ],
        claim_dir=scenario["claim_dir"],
        cancel_audit_path=scenario["cancel_audit_path"],
        cron_lister=lister,
        remover=remover,
        fallback_cancelled_marker_path=scenario[
            "fallback_cancelled_marker_path"
        ],
        cancel_lock_path=scenario["cancel_lock_path"],
        seam_audit_path=scenario["seam_audit_path"],
    )


def binding_invalid_reasons(
    binding: NormalCallbackBinding,
    *,
    dispatch_fired_marker_path: Path,
) -> list:
    """+37 ``validate_normal_callback_binding`` read-only 호출 (무수정).

    harness 가 binding 무결성을 사전 점검하는 passive reference 경로.
    """
    return validate_normal_callback_binding(
        binding, dispatch_fired_marker_path=Path(dispatch_fired_marker_path)
    )


def emit_normal_convergence_markers(
    events_dir: Path, *, result_packet: dict
) -> dict:
    """정상종료 시 result.json + .done 만 산출 (9-R / callback (a) +41 변형).

    executor 는 실 normal-collector cron 자가등록 0 — 회수는 ANU-registered
    fallback callback / normal completion callback 이 result.json/.done
    존재로 정상수렴 판정한다. 본 함수는 화이트리스트 경로
    (``task-2553+41.result.json`` / ``task-2553+41.done``) 만 기록하며
    실 cron 등록 0.
    """
    events_dir = Path(events_dir)
    events_dir.mkdir(parents=True, exist_ok=True)
    rj = events_dir / f"{TASK_MARKER_PREFIX}.result.json"
    dn = events_dir / f"{TASK_MARKER_PREFIX}.done"
    rj.write_text(
        json.dumps(result_packet, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    dn.write_text(
        f"{TASK_MARKER_PREFIX} DONE — normal convergence marker "
        "(real cron self-register 0; ANU normal/fallback callback recovers "
        "via result.json/.done presence)\n",
        encoding="utf-8",
    )
    return {"result_json": str(rj), "done": str(dn)}


__all__ = [
    "WORKSPACE",
    "FROZEN_ANCHOR",
    "FROZEN_SHA",
    "WIRED_ENTRYPOINT_SRC",
    "PLUS25_WIRING_SRC",
    "LIVE_SCHEDULE_HISTORY",
    "LIVE_COKACDIR_BIN",
    "TASK_MARKER_PREFIX",
    "ANU_CHAT_ID",
    "FakeCronLister",
    "SpyRemover",
    "SpyScheduleHistory",
    "install_isolation_guards",
    "claim_key",
    "attempt_exact_once_claim",
    "simulate_concurrent_double",
    "pass_callback_input",
    "make_binding",
    "materialize_scenario",
    "SixStepObservation",
    "observe_six_step",
    "observe_binding_invalid",
    "binding_invalid_reasons",
    "emit_normal_convergence_markers",
]