#!/usr/bin/env python3
"""scripts/run_cancel_on_success_live_adoption_2604.py

task-2604 Track A — CANCEL_ON_SUCCESS LIVE ADOPTION 실 entrypoint
(메커니즘 단독 호출점). 문서-only 금지 — 본 CLI 가 실 결선 진입점이다.

read-only consume: +9a/+23/+25/+44/+45/+48 frozen 모듈을 read-only import.
``utils/anu_delegation_completion_callback.py`` 등 frozen anchor byte-0
무수정. callback_4tuple_index.jsonl(공유 durable·track 미소유)은 절대
직접 write 0 — operational 모드는 read-only consume, self-proof 모드는
**격리 tmp ledger**(실 shared index 무접촉)만 +44/+47 API 로 구성한다.

모드:
  --mode self-proof   (기본·deliverable) 격리 tmp ledger 에 properly-bound
                       task-2604 4-tuple 구성(+48 build_properly_bound_4tuple
                       = +44 guard + +47 write-back, read-only consume) +
                       전용 격리 FakeCronWorld/WorldSpyRemover 주입 → 실
                       모듈 run_cancel_on_success_live_adoption end-to-end
                       → CANCEL_ADOPTED 실증. 실 운영 cron·실 shared
                       callback_4tuple_index.jsonl 무접촉(byte-0).
  --mode operational   실 durable registry(default canonical ledger)
                       read-only consume + 실 lister/remover(dry-run 안전
                       기본 — 실 subprocess 0). task-2604 record 부재 →
                       NON_BLOCKING(보존, 정상 안전 흐름).

산출(§4 allowlist):
  memory/events/task-2604.cancel-adoption-audit.json
  memory/events/task-2604.decision.json
  memory/events/task-2604.result.json
"""
from __future__ import annotations

import argparse
import json
import sys
import tempfile
from pathlib import Path

WORKSPACE = Path(__file__).resolve().parent.parent
if str(WORKSPACE) not in sys.path:
    sys.path.insert(0, str(WORKSPACE))

from anu_v3.cancel_on_success_live_adoption_2604 import (  # noqa: E402
    ADOPT_CANCEL_ADOPTED,
    CancelOnSuccessAdoptionResult,
    run_cancel_on_success_live_adoption,
)
from anu_v3.callback_4tuple_registry import (  # noqa: E402
    Callback4TupleRegistry,
)
from anu_v3.cancel_on_success_live_e2e import (  # noqa: E402  (+48 read-only)
    FakeCronWorld,
    ProperlyBound4Tuple,
    WorldSpyRemover,
    build_properly_bound_4tuple,
)
from utils.anu_delegation_completion_callback import (  # noqa: E402
    CallbackInput,
    CallbackType,
)

EVENTS = WORKSPACE / "memory" / "events"
ADOPTION_AUDIT_OUT = EVENTS / "task-2604.cancel-adoption-audit.json"
DECISION_OUT = EVENTS / "task-2604.decision.json"
RESULT_OUT = EVENTS / "task-2604.result.json"

# self-proof 격리 fixture (실 shared 산출물·실 cron 무접촉).
PROOF_TASK_ID = "task-2604-adoption-proof"
PROOF_FALLBACK_CRON = "T2604FALLBACK"
PROOF_MARKER = {
    "ts_utc": "2026-05-19T00:00:00Z",
    "marker_kind": "dispatch fired",
    "dispatch_cron_id": "DISP2604",
    "callback_policy_a": {
        "standardized": True,
        "normal_callback_cron_id": "NORM2604",
        "fallback_callback_cron_id": PROOF_FALLBACK_CRON,
        "fallback_role": "fallback",
        "chat_id": 6937032012,
        # ANU callback-owner key (독립 ANU·executor self key 아님 — +9a
        # c3_ownership 게이트가 요구하는 정당 소유권 토큰. dispatch-fired
        # marker 에 박힌 owner 식별자일 뿐 본 모듈이 cron 등록에 사용 0).
        "anu_key": "c119085addb0f8b7",
    },
    "task_id": PROOF_TASK_ID,
}


def _inp(task_id: str) -> CallbackInput:
    return CallbackInput(
        task_id=task_id,
        executor="dev-sim",
        dispatch_cron_id="DISP2604",
        callback_type=CallbackType.NORMAL,
        callback_cron_id="NORM2604",
        cron_status="ok",
        task_status="completed",
        required_closeout_markers={"result_json": True, "report": True},
        preservation_anchors={"frozen_anchor": "match"},
        dev_sunset=True,
    )


def _materialize(tmp: Path, task_id: str) -> dict:
    tmp.mkdir(parents=True, exist_ok=True)
    dfm = tmp / f"{task_id}.dispatch-fired.json"
    marker = dict(PROOF_MARKER)
    marker["task_id"] = task_id
    marker["callback_policy_a"] = dict(PROOF_MARKER["callback_policy_a"])
    dfm.write_text(json.dumps(marker), encoding="utf-8")
    rj = tmp / 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 = tmp / f"{task_id}.report.md"
    rep.write_text(
        f"# {task_id} report\n\nnormal completion callback PASS — "
        "properly-bound 4-tuple, durable-success.\n",
        encoding="utf-8",
    )
    crm = tmp / 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",
    )
    return {
        "dispatch_fired_marker_path": dfm,
        "result_json_path": rj,
        "report_path": rep,
        "collector_result_marker_path": crm,
        "fallback_cancelled_marker_path": tmp / f"{task_id}.fb-cancelled.json",
        "cancel_lock_path": tmp / f"{task_id}.cancel.lock",
        "seam_audit_path": tmp / f"{task_id}.plus23-cancel-audit.json",
        "marker": marker,
    }


def _run_self_proof(out_audit: Path) -> CancelOnSuccessAdoptionResult:
    """격리 tmp ledger + FakeCronWorld/WorldSpyRemover 로 실 모듈
    end-to-end (CANCEL_ADOPTED 실증). 실 shared index·실 cron byte-0."""
    tmpdir = Path(tempfile.mkdtemp(prefix="task2604-adoption-proof-"))
    paths = _materialize(tmpdir / "claim", PROOF_TASK_ID)
    ledger = tmpdir / "isolated_callback_4tuple_index.jsonl"

    # properly-bound 4-tuple 을 **격리 tmp ledger** 에만 구성 (+44 guard +
    # +47 write-back — read-only consume API; 실 shared index 무접촉).
    registry = Callback4TupleRegistry(ledger)
    tup = ProperlyBound4Tuple(
        task_id=PROOF_TASK_ID,
        dispatch_id="DSP-2604-PB",
        dispatch_cron_id="DISP2604",
        executor="dev-sim",
        chat_id="6937032012",
        normal_collector_cron_id="NORM2604",
        fallback_callback_cron_id=PROOF_FALLBACK_CRON,
        role="fallback",
    )
    build_properly_bound_4tuple(registry, tup, ts_kst="2026-05-19 09:00 KST")

    world = FakeCronWorld(
        [
            {
                "id": PROOF_FALLBACK_CRON,
                "task_id": PROOF_TASK_ID,
                "chat_id": 6937032012,
                "role": "fallback",
                "fired": False,
                "removed": False,
            }
        ]
    )
    spy = WorldSpyRemover(world, status="removed")

    result = run_cancel_on_success_live_adoption(
        _inp(PROOF_TASK_ID),
        tmpdir / "ack.marker",
        dispatch_fired_marker_path=paths["dispatch_fired_marker_path"],
        result_json_path=paths["result_json_path"],
        report_path=paths["report_path"],
        collector_result_marker_path=paths["collector_result_marker_path"],
        claim_dir=tmpdir / "claim",
        ledger_path=ledger,
        adoption_audit_path=out_audit,
        cron_lister=world.lister,
        remover=spy,
        fallback_cancelled_marker_path=paths["fallback_cancelled_marker_path"],
        cancel_lock_path=paths["cancel_lock_path"],
        seam_audit_path=paths["seam_audit_path"],
        callback_contract={
            "schema": "completion_callback_contract_v1",
            "fallback_callback_cron_id": PROOF_FALLBACK_CRON,
        },
    )
    # 실증 강제: 격리 세계에서 bound fallback 이 실제 제거됐는가.
    cron_removed = any(
        e.get("id") == PROOF_FALLBACK_CRON and e.get("removed")
        for e in world.entries
    )
    result.notes.append(
        f"self-proof: spy_calls={len(spy.calls)} cron_removed={cron_removed} "
        f"isolated_ledger={ledger} (실 shared index·실 cron 무접촉)"
    )
    return result


def _run_operational(task_id: str, out_audit: Path) -> CancelOnSuccessAdoptionResult:
    """실 durable registry read-only consume + 실 lister/remover(dry-run
    안전 기본 — 실 subprocess 0). task-2604 record 부재 → NON_BLOCKING."""
    tmpdir = Path(tempfile.mkdtemp(prefix="task2604-adoption-op-"))
    paths = _materialize(tmpdir / "claim", task_id)
    return run_cancel_on_success_live_adoption(
        _inp(task_id),
        tmpdir / "ack.marker",
        dispatch_fired_marker_path=paths["dispatch_fired_marker_path"],
        result_json_path=paths["result_json_path"],
        report_path=paths["report_path"],
        collector_result_marker_path=paths["collector_result_marker_path"],
        claim_dir=tmpdir / "claim",
        ledger_path=None,  # default canonical — read-only consume
        adoption_audit_path=out_audit,
        cron_lister=None,  # RealCokacdirCronLister (운영 collector 전용)
        remover=None,  # RealCokacdirCronRemover (dry-run 기본 — 실 호출 0)
        fallback_cancelled_marker_path=paths["fallback_cancelled_marker_path"],
        cancel_lock_path=paths["cancel_lock_path"],
        seam_audit_path=paths["seam_audit_path"],
    )


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="task-2604 cancel-on-success live adoption"
    )
    p.add_argument(
        "--mode",
        choices=["self-proof", "operational"],
        default="self-proof",
    )
    p.add_argument("--task-id", default=PROOF_TASK_ID)
    p.add_argument("--out", default=str(ADOPTION_AUDIT_OUT))
    return p


def main(argv: "list[str] | None" = None) -> int:
    args = build_parser().parse_args(argv)
    out_audit = Path(args.out)

    if args.mode == "self-proof":
        result = _run_self_proof(out_audit)
    else:
        result = _run_operational(args.task_id, out_audit)

    audit = result.adoption_audit
    EVENTS.mkdir(parents=True, exist_ok=True)
    ADOPTION_AUDIT_OUT.write_text(
        json.dumps(audit, ensure_ascii=False, indent=2), encoding="utf-8"
    )

    decision = {
        "schema": "task-2604.decision_v1",
        "task_id": audit["task_id"],
        "mode": args.mode,
        "adoption_classification": result.adoption_classification,
        "cancel_adopted": result.cancel_adopted,
        "cancel_failed_classified": result.cancel_failed_classified,
        "non_blocking_marked": result.non_blocking_marked,
        "normal_callback_durable_success_identified": (
            result.identification.identified
        ),
        "lookup_status": result.lookup_status,
        "seam_invoked": result.seam_invoked,
        "seam_classification": result.seam_classification,
        "cron_remove_invoked": result.cron_remove_invoked,
        "fallback_cancelled": result.fallback_cancelled,
        "remove_result": result.remove_result,
        "normal_success_unchanged": result.normal_success_unchanged,
        "decouple_invariant_held": result.decouple_invariant_held,
        "durable_success": result.durable_success,
        "target_cron_id": result.target_cron_id,
        "event_id": result.event_id,
        "adoption_audit_path": str(ADOPTION_AUDIT_OUT),
        "skip_reason": result.skip_reason,
        "ts_utc": result.ts_utc,
    }
    DECISION_OUT.write_text(
        json.dumps(decision, ensure_ascii=False, indent=2), encoding="utf-8"
    )

    final_status = (
        "ok"
        if (
            result.normal_success_unchanged
            and result.decouple_invariant_held
            and result.adoption_classification
            in (
                ADOPT_CANCEL_ADOPTED,
                "CANCEL_FAILED_CLASSIFIED",
                "NON_BLOCKING",
            )
        )
        else "error"
    )
    out_result = {
        "schema": "task-2604.result_v1",
        "task_id": audit["task_id"],
        "status": final_status,
        "classification": result.adoption_classification,
        "mode": args.mode,
        "normal_success_unchanged": result.normal_success_unchanged,
        "decouple_invariant_held": result.decouple_invariant_held,
        "cancel_adopted": result.cancel_adopted,
        "cancel_failed_classified": result.cancel_failed_classified,
        "non_blocking_marked": result.non_blocking_marked,
        "adoption_audit_path": str(ADOPTION_AUDIT_OUT),
        "decision_path": str(DECISION_OUT),
        "notes": result.notes,
        "ts_utc": result.ts_utc,
    }
    RESULT_OUT.write_text(
        json.dumps(out_result, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )
    print(json.dumps(out_result, ensure_ascii=False))
    # 채택 분류(보존/실패 안전 분류 포함)는 정상 흐름 → exit 0.
    return 0


if __name__ == "__main__":
    raise SystemExit(main())