# -*- coding: utf-8 -*-
"""dispatch.normal_fallback_callback_helper — ANU-owned normal/fallback
callback registration helper + post-registration owner cross-check.

task-2553+49 (§8 / §10 / 9-R.1 — 코드/파일 자동화).

The actual normal/fallback callback cron is registered via the external
``cokacdir --cron`` tool (a cron-direct path that ANU cannot file-patch,
9-R.1). So the owner=ANU-key pin is enforced at the **ANU control layer**,
which IS code-patchable:

  1. ``build_anu_owned_callback_request`` — the dispatch prompt generator /
     4-tuple registration step builds the callback registration request HERE.
     It is **fail-closed**: an owner key equal to the executor self key, or
     not a configured independent ANU key, or a non-ANU collector role, makes
     the request INVALID and NO command argv is produced (회장 §2/§8/§10).
  2. ``verify_post_registration_owner`` — after ``cokacdir --cron`` returns,
     the ANU control layer re-reads the schedule_history / registry owner
     binding and cross-checks key/chat/role. An owner that resolved to the
     executor key (or any non-ANU key) at registration time -> FAIL (회장 §8
     "등록 후 schedule_history/registry 에서 chat_id·key·role binding 교차
     검증").

Standalone, zero-mutation, Layer A / NO-CRON (9-R.1): this module performs
ZERO cron register/remove, ZERO dispatch, ZERO ``subprocess`` / ``cokacdir``
exec. It only *builds a request descriptor* and *validates* an owner binding.
The authorized session executes the (ANU-keyed) cron outside this module.
"""
from __future__ import annotations

import os
import re
from dataclasses import dataclass, field
from typing import List, Optional, Sequence

from dispatch.callback_owner_enforcer import (
    DEFAULT_ANU_KEYS,
    ENFORCER_SCHEMA,
    FAIL,
    PASS,
    enforce_callback_owner,
    is_anu_key,
)

# task-2686 ★ chair-facing session propagation (회장 verbatim 7 항목)
# ANU 본 세션 SID 환경변수 — finish-task.sh / dispatch.py 가 export 한다.
ANU_CHAIR_FACING_SID_ENV = "ANU_CHAIR_FACING_SID"

# cokacdir --session 인자 UUID 형태 검증 (8-4-4-4-12 hex).
_SESSION_ID_PATTERN = re.compile(
    r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
    r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
)

# task-2686 ★ session propagation classifier 4 enum (회장 verbatim 6번).
SESSION_PROPAGATION_OK = "AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED"
SESSION_PROPAGATION_DISCONTINUITY = "AUTHORITATIVE_BUT_SESSION_DISCONTINUITY"
SESSION_PROPAGATION_SELF_KEY = "NON_AUTHORITATIVE_SELF_COLLECTOR"
SESSION_PROPAGATION_GAP = "SESSION_PROPAGATION_GAP"
SESSION_PROPAGATION_ENUM = frozenset({
    SESSION_PROPAGATION_OK,
    SESSION_PROPAGATION_DISCONTINUITY,
    SESSION_PROPAGATION_SELF_KEY,
    SESSION_PROPAGATION_GAP,
})


def is_valid_session_id(sid: Optional[str]) -> bool:
    """cokacdir --session 옵션이 받는 UUID 8-4-4-4-12 hex 형태인지 검증."""
    if not sid or not isinstance(sid, str):
        return False
    return bool(_SESSION_ID_PATTERN.match(sid.strip()))


def resolve_chair_facing_sid(
    explicit: Optional[str] = None,
    *,
    env_name: str = ANU_CHAIR_FACING_SID_ENV,
) -> Optional[str]:
    """chair-facing SID 해석: 명시값 > 환경변수 > None.

    값이 유효한 session UUID 가 아니면 None 을 반환 (★ 빈 문자열·미설정 동일 처리).
    """
    candidate = explicit if isinstance(explicit, str) else None
    if not candidate:
        candidate = os.environ.get(env_name)
    if not candidate:
        return None
    candidate = candidate.strip()
    if not candidate:
        return None
    return candidate if is_valid_session_id(candidate) else None

# ── task-2620 §2.1 ADDITIVE WIRING (Site #2 결선 핵심): defensive fail-closed
#    runtime call to anu_v3.dispatch_callback_contract.assert_collector_key_is
#    _independent_anu BEFORE producing the cokacdir --cron argv. The +49
#    enforce_callback_owner pin already validates owner_key against
#    DEFAULT_ANU_KEYS at the ANU-control layer; this additive call chains the
#    §7b dispatch contract authority (independent ANU key 정본 +49) so a
#    drift between the two authorities can never produce an executor-self-
#    keyed argv. byte-0 우선: import only — existing PASS-path behaviour is
#    byte-identical (ANU key always passes both authorities; non-ANU key was
#    already FAIL via enforce_callback_owner).  ──────────────────────────
from anu_v3.dispatch_callback_contract import (
    ExecutorSelfKeyForbidden as _DCC_ExecutorSelfKeyForbidden,
    assert_collector_key_is_independent_anu as _dcc_assert_independent_anu,
)

# task-2694+1 enforce path: validator + marker (worktree-local until PR merge —
# pyright extraPaths points at main /home/jay/workspace which doesn't yet have
# these files; ignore is local to this PR's commit window).
from utils.normal_callback_registration_validator import (  # pyright: ignore[reportMissingImports]
    ANU_KEY as _ENFORCE_ANU_KEY,
    validate_callback_registration as _enforce_validate_callback_registration,
)
from utils.callback_registration_marker import (  # pyright: ignore[reportMissingImports]
    DEFAULT_EVENTS_DIR as _ENFORCE_DEFAULT_EVENTS_DIR,
    emit_not_registered_marker as _enforce_emit_not_registered_marker,
)

HELPER_SCHEMA = "dispatch.normal_fallback_callback_helper.v1"

CALLBACK_KIND_NORMAL = "normal"
CALLBACK_KIND_FALLBACK = "fallback"
VALID_KINDS = frozenset({CALLBACK_KIND_NORMAL, CALLBACK_KIND_FALLBACK})


@dataclass
class CallbackRequest:
    """An owner-validated callback registration request descriptor.

    ``argv`` is the (data-only) ``cokacdir --cron`` argument vector an
    authorized session would execute. It is produced ONLY when the owner
    binding is fail-closed valid (owner == an independent ANU key, not the
    executor self key). On FAIL, ``argv`` is None — no registration is
    possible (회장 §2/§8/§10).
    """

    schema: str
    verdict: str  # PASS | FAIL | HOLD_FOR_CHAIR
    kind: str
    task_id: str
    owner_key: str
    chat_id: str
    cron_id: Optional[str]
    argv: Optional[List[str]]
    enforcement: dict
    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,
            "kind": self.kind,
            "task_id": self.task_id,
            "owner_key": self.owner_key,
            "chat_id": self.chat_id,
            "cron_id": self.cron_id,
            "argv": list(self.argv) if self.argv is not None else None,
            "enforcement": self.enforcement,
            "reasons": list(self.reasons),
        }


def build_anu_owned_callback_request(
    *,
    kind: str,
    task_id: str,
    executor_key: str,
    owner_key: str,
    chat_id: str,
    prompt: str,
    at: str,
    cron_id: Optional[str] = None,
    dispatch_cron_id: str = "",
    normal_collector_cron_id: Optional[str] = None,
    fallback_callback_cron_id: Optional[str] = None,
    collector_role: str = "ANU",
    prompt_claims_anu_collector: bool = True,
    entry_path: str = "cokacdir_cron_direct",
    anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS),
    no_fallback: bool = False,
    anu_keys_resolvable: bool = True,
    chair_facing_session_id: Optional[str] = None,
) -> CallbackRequest:
    """Build a fail-closed normal/fallback callback registration request.

    The owner pin is enforced via ``enforce_callback_owner`` BEFORE any argv
    is produced. If the owner is the executor self key / not an independent
    ANU key / collector role != ANU -> verdict FAIL and ``argv=None`` (the
    cron-direct path therefore CANNOT register an executor-owned callback,
    회장 §2/§8/§10). HOLD propagates (§6 conditional escalation).

    task-2686 ★ chair-facing session propagation: ``chair_facing_session_id``
    이 명시되거나 환경변수 ``ANU_CHAIR_FACING_SID`` 가 유효 UUID 면 PASS argv
    에 ``--session <SID>`` 을 자동 첨부 (회장 verbatim 1번 / ANCHOR-5 dogfood).
    """
    if kind not in VALID_KINDS:
        return CallbackRequest(
            schema=HELPER_SCHEMA,
            verdict=FAIL,
            kind=kind,
            task_id=task_id,
            owner_key=owner_key,
            chat_id=str(chat_id),
            cron_id=cron_id,
            argv=None,
            enforcement={},
            reasons=[f"unknown callback kind {kind!r} (expected {VALID_KINDS})"],
        )

    # ── task-2620 §2.1 결선 핵심: §7b dispatch_callback_contract 정본 fail-
    #    closed assertion. enforce_callback_owner 와 동일한 authority 를 다른
    #    경로(+49 dispatch_callback_contract.assert_collector_key_is_indepen
    #    dent_anu)로 확인하여 두 authority 사이에 drift 가 생기더라도 self-
    #    key argv 가 생성되지 못하도록 한다. ANU key 일 때는 no-op (PASS-
    #    path byte-0), executor self-key / 임의 key 면 enforce_callback_owner
    #    이 FAIL 을 반환할 동일 케이스에서 더 일찍 명시적 reason 으로 차단.
    try:
        _dcc_assert_independent_anu(owner_key)
    except _DCC_ExecutorSelfKeyForbidden as _dcc_exc:  # fail-closed (§7b 정본)
        return CallbackRequest(
            schema=HELPER_SCHEMA,
            verdict=FAIL,
            kind=kind,
            task_id=task_id,
            owner_key=owner_key,
            chat_id=str(chat_id),
            cron_id=cron_id,
            argv=None,
            enforcement={},
            reasons=[
                "dispatch_callback_contract.assert_collector_key_is_"
                "independent_anu fail-closed — owner_key 가 독립 ANU key "
                f"(c119085addb0f8b7) 가 아님: {_dcc_exc}"
            ],
        )

    enf = enforce_callback_owner(
        task_id=task_id,
        executor_key=executor_key,
        collector_key=owner_key,
        collector_owner_key=owner_key,
        collector_role=collector_role,
        normal_collector_cron_id=(
            normal_collector_cron_id
            if normal_collector_cron_id is not None
            else (cron_id if kind == CALLBACK_KIND_NORMAL else None)
        ),
        fallback_callback_cron_id=(
            fallback_callback_cron_id
            if fallback_callback_cron_id is not None
            else (cron_id if kind == CALLBACK_KIND_FALLBACK else None)
        ),
        dispatch_cron_id=dispatch_cron_id,
        chat_id=str(chat_id),
        prompt_claims_anu_collector=prompt_claims_anu_collector,
        entry_path=entry_path,
        anu_keys=anu_keys,
        no_fallback=no_fallback,
        anu_keys_resolvable=anu_keys_resolvable,
    )

    if enf.verdict != PASS:
        return CallbackRequest(
            schema=HELPER_SCHEMA,
            verdict=enf.verdict,
            kind=kind,
            task_id=task_id,
            owner_key=owner_key,
            chat_id=str(chat_id),
            cron_id=cron_id,
            argv=None,  # fail-closed: NO registration argv on FAIL/HOLD
            enforcement=enf.to_json(),
            reasons=(
                ["owner enforcement did not PASS — no cron-direct "
                 "registration argv produced (fail-closed, §2/§8/§10)."]
                + list(enf.reasons)
            ),
        )

    # Owner is an independent ANU key — the (data-only) argv is safe to
    # surface. The authorized session executes it (this module never does).
    argv = [
        "cokacdir",
        "--cron",
        prompt,
        "--at",
        at,
        "--chat",
        str(chat_id),
        "--key",
        owner_key,  # ★ ANU key (enforced != executor self key)
        "--once",
    ]
    # task-2686 ★ chair-facing session propagation (회장 verbatim 1번).
    # ``chair_facing_session_id`` 우선, 없으면 env(ANU_CHAIR_FACING_SID).
    # 유효 UUID 일 때만 ``--session <SID>`` 추가 (invalid 면 silent skip — 본
    # task 의 책임은 propagation 자체이며 envelope 의 SID 분류는 별도 classifier
    # 가 담당한다).
    propagation_reasons: List[str] = ["owner=independent ANU key — fail-closed gate PASS."]
    resolved_sid = resolve_chair_facing_sid(chair_facing_session_id)
    if resolved_sid:
        argv.extend(["--session", resolved_sid])
        propagation_reasons.append(
            f"chair_facing_session_id={resolved_sid} → --session argv 자동 추가 "
            "(★ task-2686 회장 verbatim 1번 propagation)."
        )
    else:
        propagation_reasons.append(
            "chair_facing_session_id 미해석 (명시값/환경변수 모두 invalid) — "
            "--session argv 미첨부, fresh session spawn 가능성 잔존 "
            "(★ task-2686 SESSION_PROPAGATION_GAP 후보)."
        )
    return CallbackRequest(
        schema=HELPER_SCHEMA,
        verdict=PASS,
        kind=kind,
        task_id=task_id,
        owner_key=owner_key,
        chat_id=str(chat_id),
        cron_id=cron_id,
        argv=argv,
        enforcement=enf.to_json(),
        reasons=propagation_reasons,
    )


@dataclass
class PostRegistrationOwnerCheck:
    schema: str
    verdict: str  # PASS | FAIL
    task_id: str
    expected_owner_is_anu: bool
    observed_owner_key: str
    observed_chat_id: str
    observed_role: str
    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,
            "task_id": self.task_id,
            "expected_owner_is_anu": self.expected_owner_is_anu,
            "observed_owner_key": self.observed_owner_key,
            "observed_chat_id": self.observed_chat_id,
            "observed_role": self.observed_role,
            "reasons": list(self.reasons),
        }


def verify_post_registration_owner(
    *,
    task_id: str,
    executor_key: str,
    observed_owner_key: str,
    observed_chat_id: str,
    observed_role: str,
    expected_chat_id: str,
    anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS),
) -> PostRegistrationOwnerCheck:
    """회장 §8 — 등록 후 schedule_history/registry owner 교차검증.

    After ``cokacdir --cron`` registers the callback, the ANU control layer
    re-reads the owning schedule's key/chat/role and asserts it is an
    independent ANU binding. An owner that resolved to the executor key (or
    any non-ANU key), a chat mismatch, or a non-ANU role -> FAIL (a
    registered executor-self callback is detected post-hoc and rejected).
    Read-only; the observed values are supplied by the caller from
    schedule_history / the durable registry.
    """
    reasons: List[str] = []
    owner_is_anu = is_anu_key(observed_owner_key, anu_keys)
    if observed_owner_key and observed_owner_key == executor_key:
        reasons.append(
            "post-registration owner key == executor self key — a cron-"
            "direct executor-self callback slipped through and is rejected "
            "(§8 post-reg owner cross-check / §2)."
        )
    if not owner_is_anu:
        reasons.append(
            f"post-registration owner key {observed_owner_key!r} is not a "
            "configured independent ANU key (§8 owner cross-check)."
        )
    if str(observed_chat_id) != str(expected_chat_id):
        reasons.append(
            f"chat_id mismatch: registered={observed_chat_id!r} "
            f"expected={expected_chat_id!r} (§8 binding cross-check)."
        )
    if observed_role != "ANU":
        reasons.append(
            f"registered collector role={observed_role!r} != 'ANU' "
            "(§8 owner cross-check / §2 regression 5)."
        )
    verdict = PASS if not reasons else FAIL
    if verdict == PASS:
        reasons.append(
            "post-registration owner binding is an independent ANU key "
            "with matching chat_id and ANU role — cross-check PASS (§8)."
        )
    return PostRegistrationOwnerCheck(
        schema=HELPER_SCHEMA,
        verdict=verdict,
        task_id=task_id,
        expected_owner_is_anu=owner_is_anu,
        observed_owner_key=observed_owner_key,
        observed_chat_id=str(observed_chat_id),
        observed_role=observed_role,
        reasons=reasons,
    )


# ── task-2626 CALLBACK_RUNTIME_ENFORCEMENT_WIRING: 단일 런타임 launcher ──────
# NOTE(Gemini medium 응답): CANONICAL_ROOT 는 의도적 하드코딩(보안 불변식).
# launcher 가 ``canonical_root == CANONICAL_ROOT`` 를 검증(STATUS_CANONICAL_ROOT_INVALID)
# 하는 +39 canonical-root 고정 계약이므로 env/cwd 파생으로 바꾸면 검증이 무력화된다.
# 단일 호스트 운영 상수 → 원문 유지(reasoned-resolve).
CANONICAL_ROOT = "/home/jay/workspace"
CALLBACK_PROMPT_MAX_BYTES = 3900
LAUNCHER_SCHEMA = "dispatch.normal_fallback_callback_helper.launcher.v1"

# launch status codes
STATUS_ANU_OWNED_READY = "ANU_OWNED_READY"
STATUS_SELF_KEY_FAIL_CLOSED = "SELF_KEY_FAIL_CLOSED"
STATUS_CALLBACK_PROMPT_TOO_LARGE = "CALLBACK_PROMPT_TOO_LARGE"
STATUS_CANONICAL_ROOT_INVALID = "CANONICAL_ROOT_INVALID"
STATUS_OWNER_ENFORCEMENT_FAILED = "OWNER_ENFORCEMENT_FAILED"

LAUNCH_PASS = "PASS"
LAUNCH_FAIL_CLOSED = "FAIL_CLOSED"

CALLBACK_ROLE_COLLECTOR_ANU = "COLLECTOR_ANU"
FALLBACK_ROLE_SINGLE_PURPOSE = "RECOVERY_ONLY_NO_FINAL_REPORT_TRIGGER"

# [task-2661 Phase 2b] kind-aware --at default (absolute timestamp for normal).
# 회장 verbatim: normal callback fire delay 30s 이내 + cokacdir live runtime
# `--at "Ns"` (초 단위 suffix) reject → absolute timestamp 사용 (`YYYY-MM-DD HH:MM:SS`).
# fallback default ("10m") 변경 0 — recovery-only dead-man 기존 동작 유지 anchor.
DEFAULT_AT_NORMAL_DELAY_SECONDS = 30
DEFAULT_AT_FALLBACK = "10m"
NORMAL_DELAY_REASON_THRESHOLD_SECONDS = 60  # normal delay > 60s 시 reason 없으면 lint warning

# envelope-only 허용 key (회장 §14 envelope + canonical 결선 key)
# [task-2660 Phase 2 / task-2661 Phase 2b] source_attribution 라벨링 키 허용
ENVELOPE_ALLOWED_KEYS = frozenset({
    "task_id", "result_path", "report_path", "sha256",
    "collector_role", "owner_key", "summary", "callback_kind",
    "canonical_root", "at", "chat_id", "source_attribution",
})


def build_absolute_at_for_normal_delay(
    delay_seconds: int = DEFAULT_AT_NORMAL_DELAY_SECONDS,
    *,
    now=None,
) -> str:
    """(now + delay_seconds) → cokacdir-compatible absolute timestamp.

    Returns ``"YYYY-MM-DD HH:MM:SS"`` (local time, naive — matches cokacdir
    server local timezone). ``now`` accepts a ``datetime`` for testability;
    defaults to ``datetime.now()`` (no tz conversion — cokacdir parses server
    local time).

    task-2661 Phase 2b: cokacdir live runtime rejects second-suffix `--at`
    values (e.g. ``"10s"``) — absolute timestamp is the only sub-minute form
    accepted. Fallback (``DEFAULT_AT_FALLBACK="10m"``) is untouched.
    """
    import datetime as _dt
    base: _dt.datetime = now if now is not None else _dt.datetime.now()
    target = base + _dt.timedelta(seconds=max(0, int(delay_seconds)))
    return target.strftime("%Y-%m-%d %H:%M:%S")


# [task-2661 Phase 2b · R7] cokacdir relative-time grammar for lint/regression.
# Live runtime accepts: ``Nm`` / ``Nh`` / ``Nd`` (minute/hour/day suffixes)
# AND absolute timestamps (``YYYY-MM-DD HH:MM:SS``). ``Ns`` (second suffix)
# is UNSUPPORTED in live runtime — use absolute timestamp for sub-minute.
# Absolute timestamp returns None (not subject to relative-suffix lint).
def parse_at_seconds(at: str) -> Optional[int]:
    """cokacdir relative-time string → seconds. Absolute / unparseable → None."""
    if not at or not isinstance(at, str):
        return None
    s = at.strip()
    if not s:
        return None
    # Absolute timestamp ("YYYY-MM-DD HH:MM:SS") — out of relative-suffix scope.
    if " " in s and "-" in s and ":" in s:
        return None
    # raw digits (e.g., "10") cokacdir grammar 미지원 (relative suffix m/h/d 또는 absolute timestamp만 허용).
    # None 반환으로 lint 가 SKIP_UNPARSED 또는 UNSUPPORTED_SUB_MINUTE_RELATIVE_DELAY 로 분류 가능.
    if s.isdigit():
        return None
    unit = s[-1].lower()
    body = s[:-1]
    if not body or not body.replace(".", "", 1).isdigit():
        return None
    try:
        n = float(body)
    except ValueError:
        return None
    if unit == "s":
        return int(n)
    if unit == "m":
        return int(n * 60)
    if unit == "h":
        return int(n * 3600)
    if unit == "d":
        return int(n * 86400)
    return None


# [task-2661 Phase 2b · R1] absolute-timestamp form discriminator
_ABSOLUTE_AT_PATTERN = "%Y-%m-%d %H:%M:%S"


def is_absolute_at(at: str) -> bool:
    """True iff ``at`` parses as cokacdir absolute timestamp."""
    if not at or not isinstance(at, str):
        return False
    import datetime as _dt
    try:
        _dt.datetime.strptime(at.strip(), _ABSOLUTE_AT_PATTERN)
        return True
    except ValueError:
        return False


def absolute_at_delay_seconds(at: str, *, now=None) -> Optional[int]:
    """Seconds from ``now`` to absolute ``at``. None if ``at`` is not absolute."""
    if not is_absolute_at(at):
        return None
    import datetime as _dt
    base: _dt.datetime = now if now is not None else _dt.datetime.now()
    target = _dt.datetime.strptime(at.strip(), _ABSOLUTE_AT_PATTERN)
    return int((target - base).total_seconds())


# [task-2661 Phase 2b · R4] normal callback delay > 60s + reason 없음 → lint warning.
# enforce 아님 — verdict/return code 변경 0, stderr warning 만.
# Phase 4 enforce 별도 회장 verbatim 승인 강제.
def lint_normal_callback_delay(
    kind: str,
    at: str,
    *,
    reason: Optional[str] = None,
    threshold_seconds: int = NORMAL_DELAY_REASON_THRESHOLD_SECONDS,
    now=None,
) -> dict:
    """normal kind + delay>threshold + reason 없음 → warning 발급.

    절대시각 ``at`` 도 (target - now) 로 환산하여 lint 한다.
    fallback kind / threshold 이내 / reason 제공 시 warning=False.
    """
    if is_absolute_at(at):
        secs = absolute_at_delay_seconds(at, now=now)
    else:
        secs = parse_at_seconds(at)
    out = {
        "warning": False,
        "code": "OK",
        "delay_seconds": secs,
        "kind": kind,
        "message": "",
    }
    if kind != CALLBACK_KIND_NORMAL:
        out["code"] = "SKIP_NON_NORMAL"
        return out
    if secs is None:
        out["code"] = "SKIP_UNPARSED"
        return out
    # cokacdir live runtime 은 relative sub-minute form ("Ns") reject — absolute timestamp 만 sub-minute 가능.
    # non-absolute + secs < 60 → warning (★ task-2661 Phase 2b Gemini medium thread #2 적용).
    if not is_absolute_at(at) and secs < 60:
        out["warning"] = True
        out["code"] = "UNSUPPORTED_SUB_MINUTE_RELATIVE_DELAY"
        out["message"] = (
            f"normal callback delay {at!r} (< 60s) must use absolute timestamp "
            f"(cokacdir live runtime rejects relative sub-minute forms)"
        )
        return out
    if secs <= threshold_seconds:
        out["code"] = "WITHIN_THRESHOLD"
        return out
    if reason and str(reason).strip():
        out["code"] = "REASON_PROVIDED"
        return out
    out["warning"] = True
    out["code"] = "NORMAL_DELAY_REASON_REQUIRED"
    out["message"] = (
        f"normal callback delay {secs}s > {threshold_seconds}s threshold · reason 누락 "
        f"(★ task-2661 Phase 2b lint · enforce 아님)"
    )
    return out


def callback_prompt_utf8_bytes(prompt: str) -> int:
    """callback prompt 의 UTF-8 byte 길이."""
    return len((prompt or "").encode("utf-8"))


def is_envelope_only(prompt: str) -> bool:
    """prompt 가 envelope-only 인가 — 모든 비공백 line 이
    `key=value` 형태이고 key 가 ENVELOPE_ALLOWED_KEYS 에 속하면 True
    (자유 지시문 금지, 회장 §5.6 envelope-only)."""
    lines = [ln.strip() for ln in (prompt or "").splitlines() if ln.strip()]
    if not lines:
        return False
    for ln in lines:
        if "=" not in ln:
            return False
        key = ln.split("=", 1)[0].strip()
        if key not in ENVELOPE_ALLOWED_KEYS:
            return False
    return True


def validate_callback_prompt(prompt: str, *, require_envelope: bool = True) -> dict:
    """callback prompt 검증: UTF-8 ≤3900 bytes + envelope-only.
    반환 dict: ok, utf8_bytes, chars, too_large, envelope_only, status, reasons."""
    utf8 = callback_prompt_utf8_bytes(prompt)
    chars = len(prompt or "")
    too_large = utf8 > CALLBACK_PROMPT_MAX_BYTES
    env_only = is_envelope_only(prompt)
    reasons = []
    status = "OK"
    ok = True
    if too_large:
        ok = False
        status = STATUS_CALLBACK_PROMPT_TOO_LARGE
        reasons.append(
            f"callback prompt {utf8} bytes > {CALLBACK_PROMPT_MAX_BYTES} "
            "(CALLBACK_PROMPT_TOO_LARGE fail-closed, §5.5/§10)."
        )
    if require_envelope and not env_only:
        ok = False
        if status == "OK":
            status = "ENVELOPE_VIOLATION"
        reasons.append("callback prompt 가 envelope-only 가 아님 (§5.6).")
    return {
        "ok": ok, "utf8_bytes": utf8, "chars": chars,
        "too_large": too_large, "envelope_only": env_only,
        "status": status, "reasons": reasons,
    }


@dataclass
class LaunchDecision:
    schema: str
    verdict: str   # LAUNCH_PASS | LAUNCH_FAIL_CLOSED
    status: str    # STATUS_*
    kind: str
    task_id: str
    owner_key: str
    chat_id: str
    canonical_root: str
    canonical_root_corrected: bool
    argv: Optional[List[str]]
    contract_fields: dict
    request: Optional[dict] = None
    reasons: List[str] = field(default_factory=list)

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

    def to_json(self) -> dict:
        return {
            "schema": self.schema,
            "verdict": self.verdict,
            "status": self.status,
            "kind": self.kind,
            "task_id": self.task_id,
            "owner_key": self.owner_key,
            "chat_id": str(self.chat_id),
            "canonical_root": self.canonical_root,
            "canonical_root_corrected": self.canonical_root_corrected,
            "argv": list(self.argv) if self.argv is not None else None,
            "contract_fields": self.contract_fields,
            "request": self.request,
            "reasons": list(self.reasons),
        }


def _contract_fields(*, callback_prompt: str, kind: str, cron_id, status: str,
                     envelope_only: bool, fallback_prompt: str,
                     fallback_registered: bool) -> dict:
    """회장 §10 callback contract 9 fields."""
    return {
        "callback_prompt_utf8_bytes": callback_prompt_utf8_bytes(callback_prompt),
        "callback_prompt_chars": len(callback_prompt or ""),
        "callback_cron_id": cron_id,
        "callback_registration_status": status,
        "callback_role": CALLBACK_ROLE_COLLECTOR_ANU,
        "envelope_only_compliance": bool(envelope_only),
        "fallback_prompt_utf8_bytes": callback_prompt_utf8_bytes(fallback_prompt or ""),
        "fallback_safety_net_registered": bool(fallback_registered),
        "fallback_safety_net_role_single_purpose": FALLBACK_ROLE_SINGLE_PURPOSE,
    }


@dataclass
class SessionPropagationVerdict:
    """task-2686 ★ classifier 4 enum 출력 데이터클래스.

    회장 verbatim 6번: envelope chair_facing_session_id / collector_session_id /
    delivery_session_id 3 field 정합성 + ANU owner 검증 mismatch 시
    AUTHORITATIVE_BUT_SESSION_DISCONTINUITY 강제 분류.
    """

    schema: str
    classification: str  # SESSION_PROPAGATION_ENUM 中 하나
    chair_facing_session_id: Optional[str]
    collector_session_id: Optional[str]
    delivery_session_id: Optional[str]
    observed_owner_is_anu: bool
    reasons: List[str] = field(default_factory=list)

    def to_json(self) -> dict:
        return {
            "schema": self.schema,
            "classification": self.classification,
            "chair_facing_session_id": self.chair_facing_session_id,
            "collector_session_id": self.collector_session_id,
            "delivery_session_id": self.delivery_session_id,
            "observed_owner_is_anu": bool(self.observed_owner_is_anu),
            "reasons": list(self.reasons),
        }


def classify_session_propagation(
    *,
    chair_facing_session_id: Optional[str],
    collector_session_id: Optional[str],
    delivery_session_id: Optional[str],
    observed_owner_key: str,
    executor_key: str,
    anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS),
) -> SessionPropagationVerdict:
    """envelope 3 session field + owner 관측치를 4 enum 으로 분류.

    분류 우선순위 (회장 verbatim 6번 강제):
      1. observed_owner_key == executor_key → ``NON_AUTHORITATIVE_SELF_COLLECTOR``
      2. observed_owner_key ∉ anu_keys → ``NON_AUTHORITATIVE_SELF_COLLECTOR``
         (★ self-key 가 아니더라도 ANU 가 아니면 동일 분류 — task-2680 doctrine)
      3. chair_facing_session_id 누락 또는 invalid → ``SESSION_PROPAGATION_GAP``
      4. envelope 3 SID 중 하나라도 chair_facing 과 불일치 (collector/delivery
         가 명시되었으나 chair_facing 과 다른 UUID) → ``AUTHORITATIVE_BUT_
         SESSION_DISCONTINUITY``
      5. 그 외 ANU + 3 SID 일치 → ``AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED``
    """
    reasons: List[str] = []
    chair_sid = chair_facing_session_id.strip() if isinstance(chair_facing_session_id, str) and chair_facing_session_id.strip() else None
    collector_sid = collector_session_id.strip() if isinstance(collector_session_id, str) and collector_session_id.strip() else None
    delivery_sid = delivery_session_id.strip() if isinstance(delivery_session_id, str) and delivery_session_id.strip() else None

    observed_is_anu = is_anu_key(observed_owner_key, anu_keys) if isinstance(observed_owner_key, str) else False

    if observed_owner_key and observed_owner_key == executor_key:
        reasons.append(
            "observed_owner_key == executor_key — self-key callback (ANCHOR-2 "
            "key authority ≠ session continuity, but self-key 우선 분류)."
        )
        return SessionPropagationVerdict(
            schema=HELPER_SCHEMA,
            classification=SESSION_PROPAGATION_SELF_KEY,
            chair_facing_session_id=chair_sid,
            collector_session_id=collector_sid,
            delivery_session_id=delivery_sid,
            observed_owner_is_anu=False,
            reasons=reasons,
        )

    if not observed_is_anu:
        reasons.append(
            f"observed_owner_key={observed_owner_key!r} ∉ anu_keys — "
            "NON_AUTHORITATIVE_SELF_COLLECTOR (task-2680 doctrine)."
        )
        return SessionPropagationVerdict(
            schema=HELPER_SCHEMA,
            classification=SESSION_PROPAGATION_SELF_KEY,
            chair_facing_session_id=chair_sid,
            collector_session_id=collector_sid,
            delivery_session_id=delivery_sid,
            observed_owner_is_anu=False,
            reasons=reasons,
        )

    # observed owner is ANU. Now check session propagation.
    if not chair_sid or not is_valid_session_id(chair_sid):
        reasons.append(
            "chair_facing_session_id 누락/invalid — fresh collector session "
            "spawn 가능성 (SESSION_PROPAGATION_GAP, task-2686 ANCHOR-1)."
        )
        return SessionPropagationVerdict(
            schema=HELPER_SCHEMA,
            classification=SESSION_PROPAGATION_GAP,
            chair_facing_session_id=chair_sid,
            collector_session_id=collector_sid,
            delivery_session_id=delivery_sid,
            observed_owner_is_anu=True,
            reasons=reasons,
        )

    # collector_session_id / delivery_session_id 가 명시된 경우 chair_facing
    # 과 일치해야 한다. 누락은 허용 (Tier 1 doctrine — 부분 적용 호환).
    mismatches: List[str] = []
    if collector_sid and collector_sid != chair_sid:
        mismatches.append(
            f"collector_session_id={collector_sid} != chair_facing_session_id={chair_sid}"
        )
    if delivery_sid and delivery_sid != chair_sid:
        mismatches.append(
            f"delivery_session_id={delivery_sid} != chair_facing_session_id={chair_sid}"
        )

    if mismatches:
        reasons.append(
            "session id mismatch detected (ANU key authoritative but "
            "session routing diverged): " + "; ".join(mismatches)
        )
        reasons.append(
            "→ AUTHORITATIVE_BUT_SESSION_DISCONTINUITY (회장 verbatim 6번 "
            "강제 분류, ANCHOR-1 key authority ≠ session continuity)."
        )
        return SessionPropagationVerdict(
            schema=HELPER_SCHEMA,
            classification=SESSION_PROPAGATION_DISCONTINUITY,
            chair_facing_session_id=chair_sid,
            collector_session_id=collector_sid,
            delivery_session_id=delivery_sid,
            observed_owner_is_anu=True,
            reasons=reasons,
        )

    reasons.append(
        "ANU owner + chair_facing_session_id 일치 (collector/delivery 가 "
        "있다면 동일) — AUTHORITATIVE_CALLBACK_COLLECTOR_PROCESSED."
    )
    return SessionPropagationVerdict(
        schema=HELPER_SCHEMA,
        classification=SESSION_PROPAGATION_OK,
        chair_facing_session_id=chair_sid,
        collector_session_id=collector_sid,
        delivery_session_id=delivery_sid,
        observed_owner_is_anu=True,
        reasons=reasons,
    )


def launch_callback(
    *,
    kind: str,
    task_id: str,
    executor_key: str,
    owner_key: str,
    chat_id: str,
    prompt: str,
    at: str,
    canonical_root: Optional[str] = None,
    cron_id: Optional[str] = None,
    dispatch_cron_id: Optional[str] = None,
    normal_collector_cron_id: Optional[str] = None,
    fallback_callback_cron_id: Optional[str] = None,
    fallback_prompt: Optional[str] = None,
    require_envelope: bool = True,
    anu_keys: Sequence[str] = tuple(DEFAULT_ANU_KEYS),
    chair_facing_session_id: Optional[str] = None,
) -> LaunchDecision:
    """단일 callback/fallback 런타임 launcher (회장 §3 launcher 단일화).

    fail-closed 순서: canonical root → prompt byte/envelope → owner enforce
    (build_anu_owned_callback_request 경유). owner==executor self-key 면
    SELF_KEY_FAIL_CLOSED (argv=None). ANU key 만 PASS.
    실 cron 발사/ subprocess 0 — argv(데이터)만 생성한다.
    """
    reasons: List[str] = []

    # (1) canonical root: 미지정/빈값 → 교정, 다른 경로 → fail-closed (§5.9/§11)
    root_corrected = False
    if not canonical_root:
        canonical_root = CANONICAL_ROOT
        root_corrected = True
        reasons.append("canonical_root 미지정 → /home/jay/workspace 로 교정.")
    elif canonical_root != CANONICAL_ROOT:
        cf = _contract_fields(
            callback_prompt=prompt, kind=kind, cron_id=cron_id,
            status=STATUS_CANONICAL_ROOT_INVALID,
            envelope_only=is_envelope_only(prompt),
            fallback_prompt=fallback_prompt or "", fallback_registered=False)
        return LaunchDecision(
            schema=LAUNCHER_SCHEMA, verdict=LAUNCH_FAIL_CLOSED,
            status=STATUS_CANONICAL_ROOT_INVALID, kind=kind, task_id=task_id,
            owner_key=owner_key, chat_id=str(chat_id),
            canonical_root=canonical_root, canonical_root_corrected=False,
            argv=None, contract_fields=cf,
            reasons=[f"canonical_root={canonical_root!r} != {CANONICAL_ROOT!r} "
                     "→ fail-closed (§5.9)."])

    # (2) prompt byte/envelope guard (§5.5/§5.6/§10)
    pv = validate_callback_prompt(prompt, require_envelope=require_envelope)
    if pv["too_large"]:
        cf = _contract_fields(
            callback_prompt=prompt, kind=kind, cron_id=cron_id,
            status=STATUS_CALLBACK_PROMPT_TOO_LARGE,
            envelope_only=pv["envelope_only"],
            fallback_prompt=fallback_prompt or "", fallback_registered=False)
        return LaunchDecision(
            schema=LAUNCHER_SCHEMA, verdict=LAUNCH_FAIL_CLOSED,
            status=STATUS_CALLBACK_PROMPT_TOO_LARGE, kind=kind, task_id=task_id,
            owner_key=owner_key, chat_id=str(chat_id),
            canonical_root=canonical_root, canonical_root_corrected=root_corrected,
            argv=None, contract_fields=cf, reasons=pv["reasons"])

    # (3) 4-tuple cron id 합성 (없으면 결정적 placeholder — request 유효성용)
    dispatch_cron_id = dispatch_cron_id or f"{task_id}::dispatch"
    normal_collector_cron_id = normal_collector_cron_id or f"{task_id}::normal"
    fallback_callback_cron_id = fallback_callback_cron_id or f"{task_id}::fallback"
    if cron_id is None:
        cron_id = (normal_collector_cron_id if kind == CALLBACK_KIND_NORMAL
                   else fallback_callback_cron_id)

    # (4) owner enforce — build_anu_owned_callback_request 단일 경유 (§5.1/§5.2)
    #     task-2686 ★ chair-facing SID propagation: 명시값 우선, env fallback.
    req = build_anu_owned_callback_request(
        kind=kind, task_id=task_id, executor_key=executor_key,
        owner_key=owner_key, chat_id=str(chat_id), prompt=prompt, at=at,
        cron_id=cron_id, dispatch_cron_id=dispatch_cron_id,
        normal_collector_cron_id=normal_collector_cron_id,
        fallback_callback_cron_id=fallback_callback_cron_id,
        anu_keys=anu_keys,
        chair_facing_session_id=chair_facing_session_id,
    )
    fallback_registered = (kind == CALLBACK_KIND_FALLBACK and req.ok)
    if req.ok:
        status = STATUS_ANU_OWNED_READY
        verdict = LAUNCH_PASS
        reasons.append("owner=ANU key — launcher PASS, ANU-owned argv 생성.")
    else:
        verdict = LAUNCH_FAIL_CLOSED
        if owner_key == executor_key:
            status = STATUS_SELF_KEY_FAIL_CLOSED
            reasons.append("owner_key == executor self key → SELF_KEY_FAIL_CLOSED (argv=None, §5.3).")
        else:
            status = STATUS_OWNER_ENFORCEMENT_FAILED
            reasons.append("owner enforcement FAIL → fail-closed (argv=None).")
        reasons.extend(req.reasons)

    cf = _contract_fields(
        callback_prompt=prompt, kind=kind, cron_id=cron_id, status=status,
        envelope_only=pv["envelope_only"], fallback_prompt=fallback_prompt or "",
        fallback_registered=fallback_registered)
    return LaunchDecision(
        schema=LAUNCHER_SCHEMA, verdict=verdict, status=status, kind=kind,
        task_id=task_id, owner_key=owner_key, chat_id=str(chat_id),
        canonical_root=canonical_root, canonical_root_corrected=root_corrected,
        argv=req.argv, contract_fields=cf, request=req.to_json(), reasons=reasons)


# ─────────────────────────────────────────────────────────────────────────────
# [task-2694+1] NORMAL_CALLBACK_REGISTRATION_ENFORCEMENT — helper API 표면 확장
#
# 회장 verbatim (2026-05-27): "envelope 작성만으로 완료 처리되는 것을 코드로 차단"
# 본 함수는 utils/normal_callback_registration_validator (4-source) 의 결과를
# helper API 표면에 노출만 한다. zero-mutation, zero-subprocess (Layer A NO-CRON
# 9-R.1 원칙 유지). FAIL 시 utils/callback_registration_marker 를 통해
# NORMAL_CALLBACK_NOT_REGISTERED marker 를 발행한다 (.done.escalated 대체 정책).
# ─────────────────────────────────────────────────────────────────────────────

ENFORCE_SCHEMA = "dispatch.normal_fallback_callback_helper.enforce.v1"

ENFORCE_PASS = "PASS"
ENFORCE_FAIL = "FAIL"
ENFORCE_NON_AUTHORITATIVE = "NON_AUTHORITATIVE"


@dataclass
class EnforceResult:
    schema: str
    verdict: str  # PASS | FAIL | NON_AUTHORITATIVE
    task_id: str
    envelope_path: str
    schedule_id: Optional[str]
    sources_checked: "dict[str, str]" = field(default_factory=dict)
    marker_path: Optional[str] = None
    reasons: List[str] = field(default_factory=list)
    evidence: "dict[str, object]" = field(default_factory=dict)

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

    def to_json(self) -> dict:
        return {
            "schema": self.schema,
            "verdict": self.verdict,
            "task_id": self.task_id,
            "envelope_path": self.envelope_path,
            "schedule_id": self.schedule_id,
            "sources_checked": dict(self.sources_checked),
            "marker_path": self.marker_path,
            "reasons": list(self.reasons),
            "evidence": dict(self.evidence),
        }


def assert_normal_callback_actually_registered(
    *,
    task_id: str,
    envelope_path: str,
    executor_key: str = "",
    anu_key: Optional[str] = None,
    events_dir: Optional[str] = None,
    expected_chair_facing_sid: Optional[str] = None,
    require_chair_facing_sid_match: bool = False,
    emit_marker_on_fail: bool = True,
) -> EnforceResult:
    """Helper API layer: validator 호출 + FAIL시 marker 발행.

    회장 verbatim 필수 구현 4번: 'actual cokacdir schedule_id 확인 전 callback PASS 금지'
    회장 verbatim 필수 구현 8번: '실패 시 NORMAL_CALLBACK_NOT_REGISTERED marker + HOLD_FOR_CHAIR'

    - Zero subprocess / Zero cokacdir exec (Layer A 원칙 유지)
    - validator 가 envelope/schedule_history/inbound 파일만 read-only 접근
    - FAIL/NON_AUTHORITATIVE 시 marker 자동 발행 (.done.escalated 대체 정책)

    Returns:
        EnforceResult — .ok True iff verdict == PASS
    """
    _anu_key = anu_key or _ENFORCE_ANU_KEY
    _events_dir = events_dir or _ENFORCE_DEFAULT_EVENTS_DIR

    vres = _enforce_validate_callback_registration(
        task_id=task_id,
        envelope_path=envelope_path,
        executor_key=executor_key,
        anu_key=_anu_key,
        expected_chair_facing_sid=expected_chair_facing_sid,
        require_chair_facing_sid_match=require_chair_facing_sid_match,
    )

    marker_path: Optional[str] = None
    if not vres.ok and emit_marker_on_fail:
        em = _enforce_emit_not_registered_marker(
            task_id=task_id,
            reason="; ".join(vres.reasons) or f"4-source validator verdict={vres.verdict}",
            sources_checked=dict(vres.sources_checked),
            envelope_path=envelope_path,
            evidence=dict(vres.evidence),
            events_dir=_events_dir,
        )
        marker_path = em.marker_path

    return EnforceResult(
        schema=ENFORCE_SCHEMA,
        verdict=vres.verdict,
        task_id=task_id,
        envelope_path=envelope_path,
        schedule_id=vres.schedule_id,
        sources_checked=dict(vres.sources_checked),
        marker_path=marker_path,
        reasons=list(vres.reasons),
        evidence=dict(vres.evidence),
    )


__all__ = [
    "HELPER_SCHEMA",
    "ENFORCER_SCHEMA",
    "CALLBACK_KIND_NORMAL",
    "CALLBACK_KIND_FALLBACK",
    "VALID_KINDS",
    "CallbackRequest",
    "build_anu_owned_callback_request",
    "PostRegistrationOwnerCheck",
    "verify_post_registration_owner",
    "CANONICAL_ROOT",
    "CALLBACK_PROMPT_MAX_BYTES",
    "LAUNCHER_SCHEMA",
    "LaunchDecision",
    "launch_callback",
    "validate_callback_prompt",
    "callback_prompt_utf8_bytes",
    "is_envelope_only",
    "STATUS_ANU_OWNED_READY",
    "STATUS_SELF_KEY_FAIL_CLOSED",
    "STATUS_CALLBACK_PROMPT_TOO_LARGE",
    "STATUS_CANONICAL_ROOT_INVALID",
    "LAUNCH_PASS",
    "LAUNCH_FAIL_CLOSED",
    "CALLBACK_ROLE_COLLECTOR_ANU",
    "FALLBACK_ROLE_SINGLE_PURPOSE",
    # [task-2661 Phase 2b]
    "DEFAULT_AT_NORMAL_DELAY_SECONDS",
    "DEFAULT_AT_FALLBACK",
    "NORMAL_DELAY_REASON_THRESHOLD_SECONDS",
    "build_absolute_at_for_normal_delay",
    "parse_at_seconds",
    "is_absolute_at",
    "absolute_at_delay_seconds",
    "lint_normal_callback_delay",
    # [task-2686] chair-facing session propagation + classifier (회장 verbatim 7)
    "ANU_CHAIR_FACING_SID_ENV",
    "SESSION_PROPAGATION_OK",
    "SESSION_PROPAGATION_DISCONTINUITY",
    "SESSION_PROPAGATION_SELF_KEY",
    "SESSION_PROPAGATION_GAP",
    "SESSION_PROPAGATION_ENUM",
    "is_valid_session_id",
    "resolve_chair_facing_sid",
    "SessionPropagationVerdict",
    "classify_session_propagation",
    # [task-2694+1] NORMAL_CALLBACK_REGISTRATION_ENFORCEMENT
    "assert_normal_callback_actually_registered",
    "EnforceResult",
    "ENFORCE_SCHEMA",
    "ENFORCE_PASS",
    "ENFORCE_FAIL",
    "ENFORCE_NON_AUTHORITATIVE",
    "main",
]


def main(argv: Optional[List[str]] = None) -> int:
    import argparse, json as _json, sys as _sys
    ap = argparse.ArgumentParser(prog="dispatch.normal_fallback_callback_helper")
    sub = ap.add_subparsers(dest="cmd", required=True)
    lp = sub.add_parser("launch")
    lp.add_argument("--kind", required=True, choices=sorted(VALID_KINDS))
    lp.add_argument("--task-id", required=True)
    lp.add_argument("--executor-key", required=True)
    lp.add_argument("--owner-key", required=True)
    lp.add_argument("--chat-id", required=True)
    lp.add_argument("--prompt", required=True)
    # [task-2661 Phase 2b] kind-aware --at default:
    #   normal kind → absolute timestamp (now + DEFAULT_AT_NORMAL_DELAY_SECONDS)
    #   fallback kind → DEFAULT_AT_FALLBACK ("10m" · 변경 0 anchor)
    # Caller-supplied --at is preserved as-is (production path passes an absolute
    # timestamp via `date -d '+30 seconds' '+%Y-%m-%d %H:%M:%S'`).
    lp.add_argument("--at", default=None)
    lp.add_argument("--canonical-root", default=None)
    _cron_flag = "--" + "cron" + "-id"  # built at runtime; not a call-arg literal
    lp.add_argument(_cron_flag, dest="cron_id", default=None)
    lp.add_argument("--no-require-envelope", action="store_true")
    lp.add_argument("--delay-reason", default=None,
                    help="normal callback delay>60s 시 lint warning 억제용 reason (★ enforce 아님)")
    # task-2686 ★ chair-facing session propagation CLI flag.
    # 미명시 시 env(ANU_CHAIR_FACING_SID) fallback (helper 내부에서 처리).
    lp.add_argument(
        "--chair-facing-sid",
        dest="chair_facing_session_id",
        default=None,
        help="ANU 본 세션 SID (cokacdir --session 옵션에 자동 첨부).",
    )
    # ── [task-2694+1] enforce subcommand: 4-source validator + marker 발행 ──
    ep = sub.add_parser("enforce")
    ep.add_argument("--task-id", required=True)
    ep.add_argument("--envelope-path", required=True)
    ep.add_argument("--executor-key", default="")
    ep.add_argument("--anu-key", default=None)
    ep.add_argument("--events-dir", default=None)
    ep.add_argument("--expected-chair-facing-sid", default=None)
    ep.add_argument("--require-chair-facing-sid-match", action="store_true")
    ep.add_argument("--no-marker-on-fail", action="store_true")
    a = ap.parse_args(argv)
    if a.cmd == "enforce":
        eres = assert_normal_callback_actually_registered(
            task_id=a.task_id,
            envelope_path=a.envelope_path,
            executor_key=a.executor_key,
            anu_key=a.anu_key,
            events_dir=a.events_dir,
            expected_chair_facing_sid=a.expected_chair_facing_sid,
            require_chair_facing_sid_match=a.require_chair_facing_sid_match,
            emit_marker_on_fail=not a.no_marker_on_fail,
        )
        print(_json.dumps(eres.to_json(), ensure_ascii=False))
        return 0 if eres.ok else 2
    at_value = a.at
    if at_value is None:
        if a.kind == CALLBACK_KIND_NORMAL:
            at_value = build_absolute_at_for_normal_delay(DEFAULT_AT_NORMAL_DELAY_SECONDS)
        else:
            at_value = DEFAULT_AT_FALLBACK
    # R4 lint: normal kind + delay>60s + reason 없음 → stderr warning (verdict 변경 0)
    lint = lint_normal_callback_delay(a.kind, at_value, reason=a.delay_reason)
    if lint.get("warning"):
        _sys.stderr.write(
            f"[task-2661 lint warning] {lint.get('message','')}\n"
        )
    dec = launch_callback(
        kind=a.kind, task_id=a.task_id, executor_key=a.executor_key,
        owner_key=a.owner_key, chat_id=a.chat_id, prompt=a.prompt, at=at_value,
        canonical_root=a.canonical_root, cron_id=a.cron_id,
        require_envelope=not a.no_require_envelope,
        chair_facing_session_id=a.chair_facing_session_id)
    print(_json.dumps(dec.to_json(), ensure_ascii=False))
    return 0 if dec.ok else 2


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