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

task-2553+58 (TRACK C) — FALLBACK_ACCEPTANCE_CRITERION_FOR_NEXT_PILOT.

Spec: memory/tasks/task-2553+58.md
(sha256 976d904dbe05971fad694ca0409d0cdbcd6cdc31434ac156371770ed9ea40f71).

회장 §2 verbatim 기준:

  normal callback durable-success 이후 bound fallback 은
    (a) cancel-on-success 로 제거되거나
    (b) registry 에 NON_BLOCKING 으로 명시 마크되어야 한다.
  아무 마크 없이 fallback 이 발화한 뒤 DUPLICATE_CALLBACK_IGNORED 로만
  처리되는 것은 안전성 OK 지만 운영 품질 PASS 로 보지 않는다.

이 모듈은 위 기준의 **실 entrypoint** 다 (문서-only 금지 — §3/§8).

  * ``evaluate_fallback_acceptance``      — fallback disposition observation
                                            1건을 기준 a/b/anti-pattern 으로
                                            판정 (real decision logic).
  * ``validate_non_blocking_mark``        — registry NON_BLOCKING 마크를
                                            schemas/non_blocking_fallback_schema.json
                                            (draft-07) 으로 검증.
  * ``check_criteria_schema_coherence``   — criteria JSON ↔ schema 정합
                                            (criteria/schema 정합, §3).
  * ``main``                              — 실 산출물(criteria·schema)에 대해
                                            정합 검사 + self-sample 판정을
                                            stdout JSON 으로 출력. **파일 write 0**
                                            (read-only — §4 expected_files
                                            allowlist 외 write 0).

순수/오프라인: network·git mutation·cron·dispatch·cokacdir·subprocess·
파일 write 0. fallback/dead-man/fixed-time 진행 트리거 0.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import sys
from pathlib import Path
from typing import Any

from jsonschema import Draft7Validator  # pyright: ignore[reportMissingImports]

WORKSPACE = Path(__file__).resolve().parent.parent
CRITERIA_PATH = WORKSPACE / "memory/events/fallback_acceptance_criteria.json"
SCHEMA_PATH = WORKSPACE / "schemas/non_blocking_fallback_schema.json"

ANU_KEY = "c119085addb0f8b7"
ANU_CHAT_ID = 6937032012

# ── verdicts ────────────────────────────────────────────────────────────────
OPERATIONAL_PASS = "OPERATIONAL_PASS"
OPERATIONAL_QUALITY_FAIL = "OPERATIONAL_QUALITY_FAIL"
NOT_APPLICABLE = "NOT_APPLICABLE"
SAFETY_FAIL = "SAFETY_FAIL"

VERDICTS = (
    OPERATIONAL_PASS,
    OPERATIONAL_QUALITY_FAIL,
    NOT_APPLICABLE,
    SAFETY_FAIL,
)

_IDLE_HANDLING = {"DUPLICATE_CALLBACK_IGNORED", "NO-ACTION", "NO_ACTION"}


# ── io helpers ──────────────────────────────────────────────────────────────
def load_json(path: str | Path) -> Any:
    """Read-only JSON load. No write side effects anywhere in this module."""
    return json.loads(Path(path).read_text(encoding="utf-8"))


def load_criteria(path: str | Path | None = None) -> dict:
    return load_json(path or CRITERIA_PATH)


def load_schema(path: str | Path | None = None) -> dict:
    return load_json(path or SCHEMA_PATH)


def sha256_of(path: str | Path) -> str:
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


# ── schema validation (criterion b machinery) ───────────────────────────────
def validate_non_blocking_mark(
    mark: Any, schema: dict | None = None
) -> tuple[bool, list[str]]:
    """Validate a registry NON_BLOCKING mark against the draft-07 schema.

    Returns ``(ok, errors)``. ``ok`` is True only when the mark is a dict
    that fully validates (additionalProperties:false, const owner/chat = ANU,
    classification ∈ {NON_BLOCKING, LEGACY_PENDING}, the decouple/trigger
    invariants). A malformed / non-ANU / non-dict mark → ok=False.
    """
    if schema is None:
        schema = load_schema()
    if not isinstance(mark, dict):
        return False, ["mark is not an object"]
    validator = Draft7Validator(schema)
    errors = sorted(
        validator.iter_errors(mark), key=lambda e: list(e.absolute_path)
    )
    msgs = [
        f"{'/'.join(str(p) for p in e.absolute_path) or '<root>'}: {e.message}"
        for e in errors
    ]
    return (not msgs), msgs


# ── core entrypoint: fallback acceptance decision ───────────────────────────
def evaluate_fallback_acceptance(
    observation: dict,
    *,
    criteria: dict | None = None,
    schema: dict | None = None,
) -> dict:
    """REAL entrypoint — 회장 §2 기준 판정 (mock 아님).

    ``observation`` keys:
      task_id (str)
      fallback_cron_id (str|None)
      fallback_bound (bool)
      normal_callback_durable_success (bool)
      normal_success_unchanged (bool)            -- safety invariant
      cancel_on_success_applied (bool)           -- criterion (a)
      fallback_fired (bool)
      fallback_handling (str|None)               -- DUPLICATE_CALLBACK_IGNORED…
      registry_non_blocking_mark (dict|None)     -- criterion (b)

    Returns a verdict dict (verdict ∈ VERDICTS).
    """
    if criteria is None:
        criteria = load_criteria()
    if schema is None:
        schema = load_schema()

    task_id = observation.get("task_id")
    fb_cron = observation.get("fallback_cron_id")

    def _v(verdict: str, **extra: Any) -> dict:
        out = {
            "schema": "task-2553+58.fallback_acceptance_verdict.v1",
            "verdict": verdict,
            "task_id": task_id,
            "fallback_cron_id": fb_cron,
            "satisfied_criterion": [],
            "criteria_schema": criteria.get("schema"),
            "nonblocking_schema_id": schema.get("$id"),
        }
        out.update(extra)
        return out

    # 1. safety gate first (fail-closed) — 기준은 '안전성 OK' 전제.
    if observation.get("normal_success_unchanged") is not True:
        return _v(
            SAFETY_FAIL,
            reason="NORMAL_SUCCESS_DECOUPLE_VIOLATED",
            detail="normal_success_unchanged != true — 디커플 절대불변 위반; "
            "기준 a/b 평가 이전 fail-closed (§5/safety_invariant).",
        )

    # 2. applicability — durable-success 선행 + bound fallback 존재.
    durable = observation.get("normal_callback_durable_success") is True
    bound = bool(fb_cron) and observation.get("fallback_bound") is True
    if not (durable and bound):
        return _v(
            NOT_APPLICABLE,
            reason="NO_DURABLE_SUCCESS_OR_NO_BOUND_FALLBACK",
            detail="durable-success 미선행 또는 bound fallback 부재 — "
            "정당 recovery / 적용 대상 아님 (applicability.out_of_scope).",
        )

    fired = observation.get("fallback_fired") is True

    # 3. criterion (a): cancel-on-success 로 제거되어 발화하지 않음.
    crit_a = (
        observation.get("cancel_on_success_applied") is True and not fired
    )

    # 4. criterion (b): registry NON_BLOCKING schema-valid + binding 일치.
    mark = observation.get("registry_non_blocking_mark")
    crit_b = False
    mark_errors: list[str] = []
    binding_ok = None
    if mark is not None:
        schema_ok, mark_errors = validate_non_blocking_mark(mark, schema)
        binding_ok = (
            isinstance(mark, dict)
            and mark.get("task_id") == task_id
            and mark.get("fallback_cron_id") == fb_cron
        )
        crit_b = bool(schema_ok and binding_ok)

    satisfied = []
    if crit_a:
        satisfied.append("a")
    if crit_b:
        satisfied.append("b")

    if satisfied:
        return _v(
            OPERATIONAL_PASS,
            satisfied_criterion=satisfied,
            criterion_a=crit_a,
            criterion_b=crit_b,
            mark_schema_errors=mark_errors,
            mark_binding_ok=binding_ok,
        )

    # 5. neither a nor b → operational quality FAIL (safety still OK).
    handling = observation.get("fallback_handling")
    if fired and handling in _IDLE_HANDLING:
        reason = "DUPLICATE_IGNORED_ONLY_NO_MARK"
        detail = (
            "마크 없이 fallback 발화 후 DUPLICATE_CALLBACK_IGNORED/NO-ACTION "
            "로만 처리 — 회장 §2 anti_pattern. 안전성 OK 이나 운영 품질 PASS 아님."
        )
    elif not fired:
        reason = "UNMARKED_NO_CANCEL_NO_NONBLOCK"
        detail = (
            "cancel-on-success 미적용·NON_BLOCKING 마크 부재 — 미발화여도 "
            "운영 품질 미충족(legacy 잔여)."
        )
    else:
        reason = "UNMARKED_FALLBACK_NONIDLE_DISPOSITION"
        detail = (
            "마크 없이 fallback 발화·idle 처리 아님 — 운영 품질 미충족 "
            "(마크/제거 부재)."
        )
    return _v(
        OPERATIONAL_QUALITY_FAIL,
        reason=reason,
        detail=detail,
        criterion_a=crit_a,
        criterion_b=crit_b,
        mark_schema_errors=mark_errors,
        mark_binding_ok=binding_ok,
        safety="OK (normal_success_unchanged=true) — 차단성 결함 아님",
    )


# ── criteria ↔ schema coherence (§3 'criteria/schema 정합') ─────────────────
def check_criteria_schema_coherence(
    criteria: dict | None = None, schema: dict | None = None
) -> dict:
    """criteria JSON 과 NON_BLOCKING schema 의 정합을 검사한다."""
    if criteria is None:
        criteria = load_criteria()
    if schema is None:
        schema = load_schema()

    checks: dict[str, bool] = {}

    # schema 자체가 valid draft-07.
    try:
        Draft7Validator.check_schema(schema)
        checks["schema_is_valid_draft07"] = True
    except Exception:  # noqa: BLE001
        checks["schema_is_valid_draft07"] = False

    checks["criteria_schema_tag"] = (
        criteria.get("schema") == "anu.fallback_acceptance_criteria.v1"
    )
    cb = criteria.get("criterion_b", {})
    checks["criterion_b_schema_ref_matches"] = (
        cb.get("schema_ref") == "schemas/non_blocking_fallback_schema.json"
        and schema.get("$id") == "schemas/non_blocking_fallback_schema.json"
    )
    checks["criterion_b_schema_title_matches"] = (
        cb.get("schema_title") == schema.get("title")
        == "task-2553+58.non_blocking_fallback_mark_v1"
    )
    ep = criteria.get("entrypoint", {})
    checks["entrypoint_points_to_this_module"] = (
        ep.get("module") == "scripts/validate_fallback_acceptance_2553plus58.py"
        and ep.get("function") == "evaluate_fallback_acceptance"
    )
    checks["verdict_enum_consistent"] = sorted(
        criteria.get("verdicts", [])
    ) == sorted(VERDICTS)

    # schema 가 ANU owner/chat 및 decouple/trigger 불변을 const 로 강제.
    props = schema.get("properties", {})
    checks["schema_owner_is_anu_const"] = (
        props.get("owner_key", {}).get("const") == ANU_KEY
    )
    checks["schema_chat_is_anu_const"] = (
        props.get("chat_id", {}).get("const") == ANU_CHAT_ID
    )
    checks["schema_decouple_const_true"] = (
        props.get("normal_success_unchanged", {}).get("const") is True
    )
    checks["schema_progress_trigger_const_false"] = (
        props.get("progress_trigger", {}).get("const") is False
    )
    checks["schema_cancel_applied_const_false"] = (
        props.get("cancel_on_success_applied", {}).get("const") is False
    )
    checks["criteria_anti_pattern_is_quality_fail"] = (
        criteria.get("anti_pattern", {}).get("operational_quality", "")
        .startswith("FAIL")
    )

    coherent = all(checks.values())
    return {
        "schema": "task-2553+58.criteria_schema_coherence.v1",
        "coherent": coherent,
        "checks": checks,
        "criteria_path": str(CRITERIA_PATH.relative_to(WORKSPACE)),
        "schema_path": str(SCHEMA_PATH.relative_to(WORKSPACE)),
    }


# ── CLI (read-only; stdout only — no file writes) ───────────────────────────
def _self_sample() -> list[dict]:
    """판정 로직이 실제로 동작함을 보이는 self-sample (mock 아님)."""
    base = {
        "task_id": "task-2553+58",
        "fallback_cron_id": "F0683510-FB",
        "fallback_bound": True,
        "normal_callback_durable_success": True,
        "normal_success_unchanged": True,
    }
    valid_mark = {
        "schema": "task-2553+58.non_blocking_fallback_mark_v1",
        "task_id": "task-2553+58",
        "fallback_cron_id": "F0683510-FB",
        "owner_key": ANU_KEY,
        "chat_id": ANU_CHAT_ID,
        "bound_after_normal_durable_success": True,
        "classification": "NON_BLOCKING",
        "marked_at_kst": "2026-05-18 22:10 KST",
        "marked_by_collector_role": "ANU",
        "basis": "self-sample: durable-success registry line present",
        "normal_success_unchanged": True,
        "expected_on_fire": "DUPLICATE_CALLBACK_IGNORED",
        "progress_trigger": False,
        "cancel_on_success_eligible": True,
        "cancel_on_success_applied": False,
    }
    return [
        evaluate_fallback_acceptance(
            {**base, "cancel_on_success_applied": True, "fallback_fired": False}
        ),
        evaluate_fallback_acceptance(
            {**base, "registry_non_blocking_mark": valid_mark}
        ),
        evaluate_fallback_acceptance(
            {
                **base,
                "fallback_fired": True,
                "fallback_handling": "DUPLICATE_CALLBACK_IGNORED",
            }
        ),
    ]


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(
        description="task-2553+58 fallback acceptance criterion validator "
        "(read-only; no file writes)."
    )
    ap.add_argument(
        "--observation",
        help="평가할 fallback disposition observation JSON 경로 (선택).",
    )
    args = ap.parse_args(argv)

    criteria = load_criteria()
    schema = load_schema()
    coherence = check_criteria_schema_coherence(criteria, schema)

    report = {
        "schema": "task-2553+58.validate_report.v1",
        "coherence": coherence,
        "self_sample_verdicts": [v["verdict"] for v in _self_sample()],
        "criteria_sha256": sha256_of(CRITERIA_PATH),
        "schema_sha256": sha256_of(SCHEMA_PATH),
        "writes_performed": 0,
    }
    if args.observation:
        report["observation_verdict"] = evaluate_fallback_acceptance(
            load_json(args.observation), criteria=criteria, schema=schema
        )

    print(json.dumps(report, ensure_ascii=False, indent=2))
    ok = coherence["coherent"] and report["self_sample_verdicts"] == [
        OPERATIONAL_PASS,
        OPERATIONAL_PASS,
        OPERATIONAL_QUALITY_FAIL,
    ]
    return 0 if ok else 1


if __name__ == "__main__":
    sys.exit(main())