# -*- coding: utf-8 -*-
"""dispatch.cron_dispatch_guard — cron-direct path callback fail-closed guard.

task-2553+44 (구현목표 A). Standalone, pure stdlib + same-package imports of
the existing (byte-0) executor_completion_contract / spec_template_validator.

Problem (회장 §2.3 / diagnosis report_items.7/9): the +32 mandatory
executor-completion-callback contract exists in code, but the ACTUAL dispatch
entry used for +41/+42/+43 is the ``cokacdir --cron`` direct path
(dispatch.py shim / dispatch.core.main is bypassed). The contract was never
called there — §8 held only by task-md text + bot self-compliance, NOT by a
code-level fail-closed gate.

This guard is the fail-closed gate for BOTH paths:
  * ``dispatch.py`` / dispatch.core.main, and
  * the ``cokacdir --cron`` direct path (legacy/bot-driven).

It validates the mandatory normal-completion-callback clause + the durable
4-tuple, records the 4-tuple to the durable ledger, and returns
PASS / FAIL / HOLD_FOR_CHAIR. A cron-direct dispatch that bypasses the
contract -> FAIL (regression 6).

Layer A / NO-CRON (9-R.1): the guard VALIDATES, RECORDS (append-only ledger),
and FAILs/HOLDs only. It performs ZERO cron register/remove, ZERO dispatch,
ZERO ``cokacdir``/``subprocess`` exec. The executor's own normal completion
callback (Layer B, §10) is a designed lifecycle signal — NOT a registry-added
ad-hoc cron — and is therefore explicitly NOT a "cron 신규 등록" breach.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import List, Optional

from dispatch.executor_completion_contract import (
    Callback4Tuple,
    validate_4tuple,
)
from dispatch.spec_template_validator import (
    FAIL,
    HOLD,
    PASS,
    spec_has_normal_callback_clause,
)

GUARD_SCHEMA = "dispatch.cron_dispatch_guard_result.v1"

# Entry-path identifiers (§3.A — both must be gated, regression 6).
PATH_DISPATCH_PY = "dispatch.core.main"
PATH_CRON_DIRECT = "cokacdir_cron_direct"
KNOWN_PATHS = frozenset({PATH_DISPATCH_PY, PATH_CRON_DIRECT})


@dataclass
class CronDispatchGuardResult:
    schema: str
    verdict: str  # PASS | FAIL | HOLD_FOR_CHAIR
    entry_path: str
    clause_present: bool
    tuple_valid: bool
    no_fallback_contract: bool
    recorded_to_ledger: bool
    reasons: List[str] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return self.verdict == PASS

    def to_json(self) -> dict:
        return {
            "schema": self.schema,
            "verdict": self.verdict,
            "entry_path": self.entry_path,
            "clause_present": self.clause_present,
            "tuple_valid": self.tuple_valid,
            "no_fallback_contract": self.no_fallback_contract,
            "recorded_to_ledger": self.recorded_to_ledger,
            "reasons": list(self.reasons),
        }


def is_no_cron_registry_checkpoint_task(spec_text: str) -> bool:
    """A NO-CRON label applies to registry/checkpoint cron-ban ONLY.

    It does NOT exempt the executor's own normal completion callback
    (regression 5). This helper exists only to make the carve-out explicit;
    the clause requirement is enforced regardless of NO-CRON labelling.
    """
    return "NO-CRON" in (spec_text or "").upper()


def _reconcile_ledger_record(
    ledger_record, tuple_: Optional[Callback4Tuple]
) -> List[str]:
    """Validate the durable record and reconcile it with the 4-tuple.

    Returns FAIL reasons (empty == consistent). Pure check: ZERO cron /
    process / dispatch (Layer A). The registry's own ``validate_record`` is
    imported lazily so this module keeps zero module-level anu_v3 coupling.
    """
    reasons: List[str] = []
    # 1. the record must be self-valid (normal_collector_cron_id mandatory,
    #    fallback present unless explicit no_fallback) — registry's rule.
    #    FAIL-CLOSED (Codex [2] re-adjudication): if registry validation
    #    cannot be performed, REFUSE the durable write rather than record
    #    unverified state. The lazy import is pure-Python (no cron/process).
    try:
        from anu_v3.callback_4tuple_registry import (  # noqa: WPS433
            validate_record as _vr,
        )
    except ImportError:
        return reasons + [
            "registry validate_record unavailable — cannot fail-closed-"
            "verify ledger record; refusing durable write (fail-closed)."
        ]
    try:
        rr = _vr(ledger_record)
    except Exception as exc:  # malformed record -> treat as invalid
        return reasons + [
            f"ledger record validation raised {type(exc).__name__} — "
            f"treated as invalid, refusing durable write (fail-closed)."
        ]
    if rr:
        reasons.extend(f"ledger record invalid: {x}" for x in rr)
    # 2. identity must reconcile with the already-validated 4-tuple.
    if tuple_ is not None:
        try:
            ident = ledger_record.identity()
        except Exception:
            return reasons + ["ledger record exposes no identity()"]
        expected = {
            "task_id": tuple_.task_id,
            "dispatch_cron_id": tuple_.dispatch_cron_id,
            "normal_collector_cron_id": tuple_.normal_collector_cron_id,
            "fallback_callback_cron_id": tuple_.fallback_callback_cron_id,
        }
        for k, v in expected.items():
            if ident.get(k) != v:
                reasons.append(
                    f"ledger/4-tuple mismatch on {k}: "
                    f"record={ident.get(k)!r} tuple={v!r} (fail-closed)."
                )
    return reasons


def guard_dispatch(
    *,
    spec_text: str,
    entry_path: str,
    tuple_: Optional[Callback4Tuple],
    no_fallback_contract: bool = False,
    spec_location_known: bool = True,
    ledger=None,
    ledger_record=None,
) -> CronDispatchGuardResult:
    """Fail-closed gate for an executor dispatch (any entry path).

    Verdict logic (§3.A):
      * spec/template location unresolvable        -> HOLD_FOR_CHAIR (§8)
      * unknown / unrecognised entry path          -> FAIL (cron-direct
                                                       bypass, regression 6)
      * normal completion callback clause missing  -> FAIL (regression 2)
      * 4-tuple absent                             -> FAIL
      * normal_collector_cron_id missing           -> FAIL (regression 3)
      * fallback_callback_cron_id missing AND no
        explicit no-fallback contract              -> FAIL (regression 4)
      * else                                       -> PASS, and the 4-tuple
                                                       is appended to the
                                                       durable ledger.

    ``ledger`` (Callback4TupleRegistry) + ``ledger_record``
    (Callback4TupleRecord) are optional; when both are supplied a PASS
    appends the durable record (§3.A / §3.B). The guard NEVER touches cron.
    """
    reasons: List[str] = []

    if not spec_location_known:
        return CronDispatchGuardResult(
            schema=GUARD_SCHEMA,
            verdict=HOLD,
            entry_path=entry_path,
            clause_present=False,
            tuple_valid=False,
            no_fallback_contract=no_fallback_contract,
            recorded_to_ledger=False,
            reasons=[
                "dispatch/spec template location unresolvable — cannot "
                "code-enforce the cron-direct callback contract "
                "(§8 HOLD_FOR_CHAIR)."
            ],
        )

    if entry_path not in KNOWN_PATHS:
        reasons.append(
            f"entry_path {entry_path!r} not a recognised gated path — a "
            "cron-direct dispatch that bypasses the executor completion "
            "contract is FAIL (§3.A / regression 6)."
        )

    clause_present = spec_has_normal_callback_clause(spec_text)
    if not clause_present:
        reasons.append(
            "executor normal completion callback clause MISSING — every "
            "executor dispatch (incl. the cokacdir --cron direct path) MUST "
            "carry it. NO-CRON does not exempt this (registry/checkpoint "
            "cron-ban only, regression 2/5)."
        )

    tuple_valid = False
    if tuple_ is None:
        reasons.append("callback 4-tuple absent (§3.A).")
    else:
        t_reasons = validate_4tuple(tuple_)
        # An explicit no-fallback contract neutralises ONLY the fallback
        # reason (regression 4); normal_collector_cron_id stays MANDATORY.
        if no_fallback_contract:
            t_reasons = [
                r for r in t_reasons
                if "fallback_callback_cron_id" not in r
            ]
        if t_reasons:
            reasons.extend(t_reasons)
        else:
            tuple_valid = True

    verdict = PASS if (
        not reasons and clause_present and tuple_valid
    ) else FAIL

    # Fail-closed ledger integrity (Codex [2]): on PASS, the durable record
    # MUST itself be valid AND its identity MUST reconcile with the already
    # validated 4-tuple — otherwise the guard would durably persist
    # malformed/mismatched callback state while reporting PASS.
    recorded = False
    if verdict == PASS and ledger is not None and ledger_record is not None:
        rec_reasons = _reconcile_ledger_record(ledger_record, tuple_)
        if rec_reasons:
            reasons.extend(rec_reasons)
            verdict = FAIL  # fail-closed: do NOT record inconsistent state
        else:
            ledger.append(ledger_record)
            recorded = True

    return CronDispatchGuardResult(
        schema=GUARD_SCHEMA,
        verdict=verdict,
        entry_path=entry_path,
        clause_present=clause_present,
        tuple_valid=tuple_valid,
        no_fallback_contract=no_fallback_contract,
        recorded_to_ledger=recorded,
        reasons=reasons,
    )


# ── task-2553+49 ADDITIVE PATCH (byte-0 우선; existing signatures/behaviour
#    above are byte-identical — only NEW symbols are appended, 9-R.1) ─────────
#
# The +44 ``guard_dispatch`` validates the mandatory normal-callback clause +
# the durable 4-tuple but does NOT pin the callback OWNER to an independent
# ANU key. task-2553+49 §2/§8 requires that executor self-collector /
# self-dispatch / self-adjudication be structurally impossible. This additive
# composite gate chains the existing guard with the standalone
# ``dispatch.callback_owner_enforcer`` (registry/ECC stay byte-0). It is a
# NEW function — ``guard_dispatch`` is untouched (회장 §5 byte-0 우선).


def guard_dispatch_with_owner(
    *,
    spec_text: str,
    entry_path: str,
    tuple_: Optional[Callback4Tuple],
    executor_key: str,
    collector_key: str,
    collector_owner_key: Optional[str],
    collector_role: str,
    dispatch_cron_id: str,
    normal_collector_cron_id: Optional[str],
    fallback_callback_cron_id: Optional[str],
    chat_id: str = "",
    prompt_claims_anu_collector: bool = False,
    no_fallback_contract: bool = False,
    spec_location_known: bool = True,
    anu_keys=None,
    anu_keys_resolvable: bool = True,
    ledger=None,
    ledger_record=None,
):
    """task-2553+49 §2/§8 composite fail-closed gate.

    Runs the byte-0 ``guard_dispatch`` (clause + 4-tuple + ledger integrity)
    AND the ``callback_owner_enforcer`` owner pin (owner == independent ANU
    key, collector_role == ANU, executor_key != collector_key, prompt-says-
    ANU-but-owner-is-executor, 4-tuple owner-valid). The composite verdict is
    the strictest of the two: any FAIL -> FAIL, any HOLD with no FAIL -> HOLD,
    else PASS. Returns ``(guard_result, owner_result, composite_verdict)``.

    Layer A: ZERO cron/dispatch/subprocess (both components are pure
    validation; the only optional write is the +44 append-only ledger via the
    existing guard, unchanged).
    """
    from dispatch.callback_owner_enforcer import (  # noqa: WPS433
        FAIL as OWNER_FAIL,
        HOLD as OWNER_HOLD,
        PASS as OWNER_PASS,
        enforce_callback_owner,
    )
    from dispatch.callback_owner_enforcer import (  # noqa: WPS433
        DEFAULT_ANU_KEYS,
    )

    base = guard_dispatch(
        spec_text=spec_text,
        entry_path=entry_path,
        tuple_=tuple_,
        no_fallback_contract=no_fallback_contract,
        spec_location_known=spec_location_known,
        ledger=ledger,
        ledger_record=ledger_record,
    )
    owner = enforce_callback_owner(
        task_id=(tuple_.task_id if tuple_ is not None else ""),
        executor_key=executor_key,
        collector_key=collector_key,
        collector_owner_key=collector_owner_key,
        collector_role=collector_role,
        normal_collector_cron_id=normal_collector_cron_id,
        fallback_callback_cron_id=fallback_callback_cron_id,
        dispatch_cron_id=dispatch_cron_id,
        chat_id=chat_id,
        prompt_claims_anu_collector=prompt_claims_anu_collector,
        entry_path=entry_path,
        anu_keys=(
            tuple(anu_keys) if anu_keys is not None
            else tuple(DEFAULT_ANU_KEYS)
        ),
        no_fallback=no_fallback_contract,
        anu_keys_resolvable=anu_keys_resolvable,
    )
    if base.verdict == FAIL or owner.verdict == OWNER_FAIL:
        composite = FAIL
    elif HOLD in (base.verdict, owner.verdict) or owner.verdict == OWNER_HOLD:
        composite = HOLD
    elif base.verdict == PASS and owner.verdict == OWNER_PASS:
        composite = PASS
    else:  # pragma: no cover - defensive: any unknown -> fail-closed
        composite = FAIL
    return base, owner, composite


# ── task-2553+49 AUTHORITATIVE ADDITIVE PATCH ────────────────────────────────
#    (byte-0 우선; ALL signatures/behaviour above are byte-identical — only
#    NEW symbols are appended. This is the REAL dispatch-path결선 of the
#    anu_v3 runtime guards: callback registration helper guard +
#    self-collector/self-dispatch runtime guard + authoritative verdict
#    selector. narrow +49 wired the owner enforcer only; §4 우선순위 2~5 요구
#    하는 registration-helper / 4-tuple / authoritative-selector 결선을 본
#    절이 실 entrypoint 에 추가한다. 9-R.1: cokacdir 외부경로는 mediated-call
#    (이 helper 경유) + acceptance-side (selector 영구 비권위) 로 충족.)


def guard_callback_registration(
    *,
    task_id: str,
    executor_key: str,
    collector_key: str,
    collector_owner_key: Optional[str],
    collector_role: str,
    dispatch_cron_id: str,
    normal_collector_cron_id: Optional[str],
    fallback_callback_cron_id: Optional[str],
    chat_id: str = "",
    prompt_claims_anu_collector: bool = False,
    entry_path: str = PATH_CRON_DIRECT,
    anu_keys=None,
    no_fallback: bool = False,
    anu_keys_resolvable: bool = True,
    actor_key: Optional[str] = None,
    is_executor_self_session: Optional[bool] = None,
):
    """§5.C callback registration helper guard — 등록 직전 fail-closed.

    The normal/fallback callback registration path MUST route through this
    gate before any ``cokacdir --cron`` is issued (9-R.1 mediated-call). It
    chains the anu_v3 runtime owner validator (owner/key/role) AND the
    self-collector runtime guard (executor self-session owning its own
    collector). On a non-PASS verdict it BOTH returns blocked AND raises
    ``CallbackRegistrationBlocked`` via ``assert_registration_permitted`` so a
    caller that ignores the return value still cannot register a self-owned
    callback (§5.C "mismatch -> 등록 안 함" — structural, not advisory).

    Returns ``(validation_result, self_guard_result, registration_allowed)``.
    Layer A: ZERO cron/dispatch/subprocess.
    """
    from anu_v3.callback_owner_validator import (  # noqa: WPS433
        assert_registration_permitted,
        validate_callback_owner_runtime,
    )
    from anu_v3.self_collector_guard import guard_self_collector_session
    from dispatch.callback_owner_enforcer import DEFAULT_ANU_KEYS

    keys = tuple(anu_keys) if anu_keys is not None else tuple(DEFAULT_ANU_KEYS)
    val = validate_callback_owner_runtime(
        task_id=task_id,
        executor_key=executor_key,
        collector_key=collector_key,
        collector_owner_key=collector_owner_key,
        collector_role=collector_role,
        normal_collector_cron_id=normal_collector_cron_id,
        fallback_callback_cron_id=fallback_callback_cron_id,
        dispatch_cron_id=dispatch_cron_id,
        chat_id=chat_id,
        prompt_claims_anu_collector=prompt_claims_anu_collector,
        entry_path=entry_path,
        anu_keys=keys,
        no_fallback=no_fallback,
        anu_keys_resolvable=anu_keys_resolvable,
    )
    self_guard = guard_self_collector_session(
        executor_key=executor_key,
        collector_key=collector_key,
        actor_key=actor_key,
        collector_role=collector_role,
        is_executor_self_session=is_executor_self_session,
    )
    registration_allowed = (
        val.verdict == PASS and self_guard.verdict == PASS
    )
    # fail-closed structural block: raise unless BOTH PASS (§5.C).
    if val.verdict == PASS and self_guard.verdict != PASS:
        # surface the self-collector failure through the same hard gate.
        from anu_v3.callback_owner_validator import (  # noqa: WPS433
            CallbackRegistrationBlocked,
        )
        if not registration_allowed:
            raise CallbackRegistrationBlocked(
                f"callback registration blocked by self-collector guard: "
                f"{self_guard.classification}"
            )
    assert_registration_permitted(val)
    return val, self_guard, registration_allowed


def select_runtime_authoritative_verdict(
    records,
    *,
    task_id: str,
    anu_keys=None,
    anu_keys_resolvable: bool = True,
):
    """§5.D authoritative verdict selector 실 runtime 결선.

    Collector 회수단계가 verdict 를 확정하기 직전 반드시 경유한다. self-chain
    verdict 는 QUARANTINED·영구 비권위, independent ANU verdict 만
    authoritative (회장 §2.5/§2.6/§5.D). Returns the
    ``AuthoritativeSelectionResult``. Layer A: ZERO cron/dispatch/subprocess.
    """
    from anu_v3.authoritative_verdict_selector import (  # noqa: WPS433
        select_authoritative_verdict,
    )
    from dispatch.callback_owner_enforcer import DEFAULT_ANU_KEYS

    keys = tuple(anu_keys) if anu_keys is not None else tuple(DEFAULT_ANU_KEYS)
    return select_authoritative_verdict(
        records,
        task_id=task_id,
        anu_keys=keys,
        anu_keys_resolvable=anu_keys_resolvable,
    )


def guard_dispatch_runtime_authoritative(
    *,
    spec_text: str,
    entry_path: str,
    tuple_,
    executor_key: str,
    collector_key: str,
    collector_owner_key: Optional[str],
    collector_role: str,
    dispatch_cron_id: str,
    normal_collector_cron_id: Optional[str],
    fallback_callback_cron_id: Optional[str],
    chat_id: str = "",
    prompt_claims_anu_collector: bool = False,
    no_fallback_contract: bool = False,
    spec_location_known: bool = True,
    anu_keys=None,
    anu_keys_resolvable: bool = True,
    ledger=None,
    ledger_record=None,
    actor_key: Optional[str] = None,
    is_executor_self_session: Optional[bool] = None,
    is_followup_dispatch: bool = False,
):
    """§4 우선순위 2~5 통합 실 dispatch-path runtime gate.

    Chains: byte-0 ``guard_dispatch`` (clause+4-tuple+ledger) ->
    ``guard_dispatch_with_owner`` owner pin -> anu_v3 self-collector +
    self-dispatch runtime guards. Any FAIL -> FAIL; any HOLD w/o FAIL ->
    HOLD; else PASS. Returns
    ``(base, owner, self_collector, self_dispatch, composite)``.
    """
    from anu_v3.self_collector_guard import (  # noqa: WPS433
        guard_self_collector_session,
        guard_self_dispatch,
    )

    base, owner, composite = guard_dispatch_with_owner(
        spec_text=spec_text,
        entry_path=entry_path,
        tuple_=tuple_,
        executor_key=executor_key,
        collector_key=collector_key,
        collector_owner_key=collector_owner_key,
        collector_role=collector_role,
        dispatch_cron_id=dispatch_cron_id,
        normal_collector_cron_id=normal_collector_cron_id,
        fallback_callback_cron_id=fallback_callback_cron_id,
        chat_id=chat_id,
        prompt_claims_anu_collector=prompt_claims_anu_collector,
        no_fallback_contract=no_fallback_contract,
        spec_location_known=spec_location_known,
        anu_keys=anu_keys,
        anu_keys_resolvable=anu_keys_resolvable,
        ledger=ledger,
        ledger_record=ledger_record,
    )
    sc = guard_self_collector_session(
        executor_key=executor_key,
        collector_key=collector_key,
        actor_key=actor_key,
        collector_role=collector_role,
        is_executor_self_session=is_executor_self_session,
    )
    sd = guard_self_dispatch(
        executor_key=executor_key,
        actor_key=actor_key,
        is_followup_dispatch=is_followup_dispatch,
        is_executor_self_session=is_executor_self_session,
    )
    if FAIL in (composite, sc.verdict, sd.verdict):
        final = FAIL
    elif composite == HOLD:
        final = HOLD
    elif composite == PASS and sc.verdict == PASS and sd.verdict == PASS:
        final = PASS
    else:  # pragma: no cover - defensive fail-closed
        final = FAIL
    return base, owner, sc, sd, final