# -*- coding: utf-8 -*-
"""dispatch.executor_completion_contract — executor completion callback contract.

task-2553+32 — EXECUTOR COMPLETION CALLBACK MANDATORY RULE 복원 (코드/파일 자동화).

회장 정정 (memory/events/task-2553.normal-callback-mandatory-doctrine-correction
_260518.json):

    executor completion callback = MANDATORY lifecycle signal.
    NO-CRON ≠ executor completion callback 금지.

This standalone module encodes that rule as executable code so that ANY future
executor task cannot omit its normal completion callback (§1/§4).

Standalone, zero-mutation: imports/edits ZERO tracked module. The huge
dispatch/__init__.py body is NOT touched (§8 기존 산출물 수정 0). File-level
contract only — the same pattern as anu_v3.callback_4tuple_index /
anu_v3.result_ready_recovery (+29).

NO-CRON note (9-R.1): this module performs ZERO cron register/remove. It only
*declares and validates* the contract. The executor's normal completion
callback is a designed lifecycle signal — NOT an ad-hoc cron-add by the
registry/checkpoint, and NOT a cron-remove by +32 (회장 금지 준수).

Closeout note (9-R.2): requiring an executor's own normal-completion callback /
result.json / report / .done is a *lifecycle signal*, NOT an escalation of
repository/task-state finalization authority. The latter (closeout 확정 권한)
remains forbidden; the former is REQUIRED.

──────────────────────────────────────────────────────────────────────────────
L4 wiring (task-2630 — CALLBACK_RUNTIME_ENFORCEMENT L4): callback lifecycle
classifier 실결선. 이 모듈에 result.json fields 10~14(축A/B/C lifecycle 분류)를
**append-only** 로 확장하고, `memory/events/<task>.callback_lifecycle.json`
artifact writer 를 추가한다 (스펙
memory/specs/system_callback_lifecycle_state_schema_260522.md §8/§9).

zero-mutation 원칙 유지: 기존 9-field per-callback contract
(dispatch.normal_fallback_callback_helper._contract_fields, 회장 §10) 는 그대로
보존하고 위 5개 lifecycle field 만 덧붙인다(ANCHOR-2). 신규 의존은
`utils.callback_lifecycle_classifier` (순수 함수 · 파일/네트워크/subprocess I/O 0 ·
anu_v3 런타임 import 0) 하나뿐 — 기존 tracked module **본문**은 수정하지 않는다.
artifact writer 외 어떤 cron register/remove · callback 재발사 · subprocess 도
수행하지 않는다(§11 / ANCHOR-4: production enforcement 완료 판정은 L4 merge 후 별도).
"""
from __future__ import annotations

import json
import os
import re
import subprocess
import tempfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, List, Optional

from utils.callback_lifecycle_classifier import classify_callback_lifecycle
from utils.callback_lifecycle_states import DEFAULT_ANU_KEYS

CONTRACT_SCHEMA = "dispatch.executor_completion_contract.v1"

# ── classifications (mirror anu_v3.result_ready_recovery vocabulary) ──────────
NORMAL_COLLECTOR_COMPLETED = "NORMAL_COLLECTOR_COMPLETED"
# result.json + .done exist but NO normal completion callback was registered.
# §4.6 / §6.8 / §6.9 — this is a RECOVERY state, NOT a normal lifecycle complete.
RESULT_READY_NO_NORMAL_CALLBACK = "RESULT_READY_NO_NORMAL_CALLBACK"
DISPATCH_FAILED = "DISPATCH_FAILED"

# Classifications that are NOT an accepted normal lifecycle completion (§6.9).
NON_NORMAL_LIFECYCLE = frozenset(
    {RESULT_READY_NO_NORMAL_CALLBACK, DISPATCH_FAILED}
)

# ── NO-CRON definition correction (§4.2 / §6.3 / §6.5) ────────────────────────
NO_CRON_CORRECTED_DEFINITION = (
    "NO-CRON == registry/checkpoint(+29/+30/+31) MUST NOT arbitrarily "
    "add/remove cron. It does NOT mean the executor is forbidden from "
    "sending its normal completion callback. The executor normal completion "
    "callback is a MANDATORY lifecycle signal, not a registry-initiated "
    "ad-hoc cron, and is therefore explicitly exempt from the cron-add ban."
)


def is_executor_completion_callback_a_cron_violation() -> bool:
    """§6.5 — executor completion callback is NOT a 'cron 신규 등록 금지' breach.

    Always False: the normal completion callback is a designed lifecycle
    signal, distinct from a registry/checkpoint ad-hoc cron add.
    """
    return False


# ── callback 4-tuple (§4.5 / §6.12) ──────────────────────────────────────────
@dataclass(frozen=True)
class Callback4Tuple:
    """{task_id, dispatch_cron_id, normal_collector_cron_id*, fallback_cron_id}.

    normal_collector_cron_id is MANDATORY (§4.5). A None/empty value makes the
    contract invalid (§6.12) — it is NOT a valid NO-CRON degradation.
    """

    task_id: str
    dispatch_cron_id: str
    normal_collector_cron_id: Optional[str]
    fallback_callback_cron_id: str


def validate_4tuple(t: Callback4Tuple) -> List[str]:
    """Return invalidity reasons (empty == valid). §6.12."""
    reasons: List[str] = []
    if not t.task_id:
        reasons.append("task_id empty")
    if not t.dispatch_cron_id:
        reasons.append("dispatch_cron_id empty")
    if not t.normal_collector_cron_id:
        reasons.append(
            "normal_collector_cron_id missing — MANDATORY lifecycle signal "
            "(§4.5/§6.12). NO-CRON does NOT exempt this (§6.3/§6.5)."
        )
    if not t.fallback_callback_cron_id:
        reasons.append("fallback_callback_cron_id empty (safety path, §6.6)")
    return reasons


def tuple_is_valid(t: Callback4Tuple) -> bool:
    return not validate_4tuple(t)


# ── executor closeout checklist (§4.4 / 9-R.2) ───────────────────────────────
# Each item is a lifecycle-signal requirement for the executor's OWN task.
# Requiring these is NOT a finalization-authority escalation (9-R.2).
EXECUTOR_CLOSEOUT_CHECKLIST = (
    "result_json_present",
    "done_marker_present",
    "report_present",
    "normal_callback_registration_evidence",  # ★ §4.4 mandatory
)


@dataclass
class ExecutorCloseoutEvidence:
    result_json_present: bool = False
    done_marker_present: bool = False
    report_present: bool = False
    # ★ §4.4 — proof the executor registered its normal completion callback
    # (e.g. the normal_collector_cron_id + cron-add ack/marker).
    normal_callback_registration_evidence: bool = False
    normal_collector_cron_id: Optional[str] = None
    # 9-R.2 guard: this evidence proves a lifecycle signal, never a claim of
    # repository/task-state finalization authority.
    claims_finalization_authority: bool = False
    notes: List[str] = field(default_factory=list)


def validate_closeout_evidence(ev: ExecutorCloseoutEvidence) -> List[str]:
    """FAIL reasons for an executor closeout (§4.4 / §4.6 / 9-R.2).

    Missing normal callback registration evidence == FAIL (not a valid
    closeout) even when result.json/.done/report all exist.
    """
    reasons: List[str] = []
    if not ev.result_json_present:
        reasons.append("result.json missing")
    if not ev.done_marker_present:
        reasons.append(".done marker missing")
    if not ev.report_present:
        reasons.append("report missing")
    if not ev.normal_callback_registration_evidence:
        reasons.append(
            "normal callback registration evidence missing — executor "
            "normal completion callback is MANDATORY (§4.4/§4.6). Not a "
            "valid normal lifecycle completion."
        )
    if not ev.normal_collector_cron_id:
        reasons.append(
            "normal_collector_cron_id absent in closeout evidence (§4.5)"
        )
    if ev.claims_finalization_authority:
        reasons.append(
            "closeout evidence claims repository/task-state finalization "
            "authority — forbidden (9-R.2/§6.15). Lifecycle signal only."
        )
    return reasons


def closeout_is_valid(ev: ExecutorCloseoutEvidence) -> bool:
    return not validate_closeout_evidence(ev)


def classify_completion(
    *,
    dispatch_ok: bool,
    result_present: bool,
    done_present: bool,
    normal_callback_registered: bool,
) -> str:
    """§4.6 / §6.8 / §6.9 completion classifier.

    result.json + .done exist but normal callback never registered ->
    RESULT_READY_NO_NORMAL_CALLBACK (a recovery state, NOT a normal
    lifecycle complete).
    """
    if not dispatch_ok:
        return DISPATCH_FAILED
    has_result = result_present or done_present
    if normal_callback_registered:
        return NORMAL_COLLECTOR_COMPLETED
    if has_result:
        return RESULT_READY_NO_NORMAL_CALLBACK
    return DISPATCH_FAILED


def is_accepted_normal_lifecycle(classification: str) -> bool:
    """§6.9 — RESULT_READY_NO_NORMAL_CALLBACK is NOT a normal completion."""
    return classification == NORMAL_COLLECTOR_COMPLETED


def is_recovery_state(classification: str) -> bool:
    """§6.6/§6.9 — recovery target, not a task failure."""
    return classification == RESULT_READY_NO_NORMAL_CALLBACK


# ═════════════════════════════════════════════════════════════════════════════
# L4 wiring — callback lifecycle classifier 실결선 (task-2630)
# 스펙: memory/specs/system_callback_lifecycle_state_schema_260522.md §8/§9
# ═════════════════════════════════════════════════════════════════════════════

# result.json fields 10~14 (축A/B/C lifecycle 분류) — 9-field contract 위에
# **append-only** 로 덧붙이는 키. (스펙 §8 · ANCHOR-1/ANCHOR-2)
#   (10) delivery_outcome          — 축A: 최종 어떻게 수집됐나
#   (11) normal_callback_miss_cause— 축B: 왜 normal 이 안 떴나
#   (12) root_cause_tags           — 축C: evidence-derivable 다중 태그
#   (13) lifecycle_state_evidence  — 판정 근거 소스 dict
#   (14) classified_by · applied_count
LIFECYCLE_RESULT_FIELDS = (
    "delivery_outcome",
    "normal_callback_miss_cause",
    "root_cause_tags",
    "lifecycle_state_evidence",
    "classified_by",
    "applied_count",
)

# 보존 대상 per-callback contract 9 fields (회장 §10) — 단일소스는
# dispatch.normal_fallback_callback_helper._contract_fields. 여기 manifest 는
# append-only 보존 검증용 name list 이며 taxonomy 신설이 아니다(§8 단일소스 유지).
CALLBACK_CONTRACT_9_FIELDS = (
    "callback_prompt_utf8_bytes",
    "callback_prompt_chars",
    "callback_cron_id",
    "callback_registration_status",
    "callback_role",
    "envelope_only_compliance",
    "fallback_prompt_utf8_bytes",
    "fallback_safety_net_registered",
    "fallback_safety_net_role_single_purpose",
)

CALLBACK_LIFECYCLE_ARTIFACT_SCHEMA = "dispatch.callback_lifecycle_artifact.v1"
# fallback collector artifact 와 구분하기 위한 종류 표식 (§6 / 필수구현 6).
# fallback collector artifact = <task>.independent-anu-collector.result.json /
# <task>.fallback_collector_applied.json (별도 파일). lifecycle classifier
# artifact = <task>.callback_lifecycle.json (이 writer 가 생성하는 유일 파일).
CALLBACK_LIFECYCLE_ARTIFACT_KIND = "callback_lifecycle_classifier"
CALLBACK_LIFECYCLE_ARTIFACT_SUFFIX = ".callback_lifecycle.json"


def classify_completion_lifecycle(
    evidence: Dict, *, anu_keys=DEFAULT_ANU_KEYS
) -> Dict:
    """callback_lifecycle_classifier 결선 진입점 (필수구현 2).

    evidence snapshot(dict) → classifier(순수 함수) 호출 → 분류 결과(dict).
    추정 금지: evidence 결핍 시 classifier 가 UNKNOWN/INSUFFICIENT_EVIDENCE 로
    판정한다(필수구현 7). CALLBACK_DELIVERY_GAP residual-only ·
    SELF_KEY_FAIL_CLOSED vs SELF_KEY_FIRED_NON_AUTHORITATIVE 분리는 classifier
    가 보존(필수구현 8/9 · ANCHOR-3). 이 함수는 I/O 0 · cron 0 · 발사 0.
    """
    return classify_callback_lifecycle(evidence, anu_keys=anu_keys)


def callback_stage_separation(lifecycle_state_evidence: Dict) -> Dict:
    """§0 핵심구분 3단계를 분리 기록 (필수구현 5).

    callback **gate PASS ≠ callback fired ≠ collector received**. classifier 가
    이미 산출한 lifecycle_state_evidence 신호에서 결정적으로 파생한다(추정 0).
    """
    lse = lifecycle_state_evidence or {}
    # fail-closed (Gemini medium·추정 금지): notification_sent/collector_received 와
    # 일관되게, 증거가 명시적으로 '차단 아님(False)'일 때만 gate PASS 로 본다.
    gate_pass = (lse.get("git_gate_blocked") is False) and (lse.get("contract_violation") is False)
    return {
        # callback gate(GIT-GATE/contract) 가 callback 단계로 진행을 허용했나
        "callback_gate_pass": bool(gate_pass),
        # cron 이 실제 발사됐나 (gate PASS 라도 미발사일 수 있음)
        "notification_sent": bool(lse.get("normal_callback_fired")),
        # authoritative collector 가 수집했나 (발사됐어도 미수집일 수 있음)
        "collector_received": bool(lse.get("authoritative_cron_collection")),
    }


def append_lifecycle_fields(
    contract_9_fields: Dict, evidence: Dict, *, anu_keys=DEFAULT_ANU_KEYS
) -> Dict:
    """9-field callback contract 위에 fields 10~14 를 append-only 로 확장
    (필수구현 1/4 · ANCHOR-2).

    - 입력 dict 는 변형하지 않는다(shallow copy 후 확장).
    - 9-field 와 lifecycle field 키가 충돌하면 ValueError (append-only 보증 —
      기존 field 를 절대 덮어쓰지 않는다).
    반환: 9 fields + fields 10~14 가 동시 존재하는 result.json closeout dict.
    """
    base = dict(contract_9_fields or {})
    result = classify_completion_lifecycle(evidence, anu_keys=anu_keys)
    appended = {
        "delivery_outcome": result["delivery_outcome"],
        "normal_callback_miss_cause": result["normal_callback_miss_cause"],
        "root_cause_tags": result["root_cause_tags"],
        "lifecycle_state_evidence": result["lifecycle_state_evidence"],
        "classified_by": result["classified_by"],
        # None-guard (Gemini HIGH): callback_stage_separation 과 동일하게 lse=None 방어
        "applied_count": (result["lifecycle_state_evidence"] or {}).get("applied_count", 0),
    }
    overlap = sorted(set(base) & set(appended))
    if overlap:
        raise ValueError(
            "append-only violation — lifecycle fields 10~14 가 기존 callback "
            f"contract field 를 덮어쓰려 함: {overlap} (ANCHOR-2)"
        )
    base.update(appended)
    return base


def build_callback_lifecycle_artifact(
    task_id: str, evidence: Dict, *, anu_keys=DEFAULT_ANU_KEYS
) -> Dict:
    """`<task>.callback_lifecycle.json` artifact 내용(dict) 구성 — 순수·결정적.

    동일 입력 → 동일 출력 (I/O 0). fallback collector artifact 와 구분되도록
    artifact_kind 를 명시한다(필수구현 6).
    """
    result = classify_completion_lifecycle(evidence, anu_keys=anu_keys)
    lse = result["lifecycle_state_evidence"]
    return {
        "schema": CALLBACK_LIFECYCLE_ARTIFACT_SCHEMA,
        "artifact_kind": CALLBACK_LIFECYCLE_ARTIFACT_KIND,
        "task_id": task_id,
        "delivery_outcome": result["delivery_outcome"],
        "normal_callback_miss_cause": result["normal_callback_miss_cause"],
        "root_cause_tags": result["root_cause_tags"],
        "evidence_completeness": result["evidence_completeness"],
        "missing_evidence_sources": result["missing_evidence_sources"],
        "classification": result["classification"],
        "lifecycle_state_evidence": lse,
        "callback_stage_separation": callback_stage_separation(lse),
        "classified_by": result["classified_by"],
        # None-guard (Gemini HIGH): callback_stage_separation(lse) 와 일관되게 방어
        "applied_count": (lse or {}).get("applied_count", 0),
    }


def _serialize_lifecycle_artifact(artifact: Dict) -> str:
    """결정적 직렬화 — sort_keys + 고정 indent + trailing newline.

    동일 dict → byte-identical 문자열 (idempotent artifact 보증).
    """
    return json.dumps(artifact, ensure_ascii=False, sort_keys=True, indent=2) + "\n"


def default_events_dir() -> str:
    """artifact 기본 디렉터리 = <workspace>/memory/events.

    WORKSPACE_ROOT env override 우선(테스트 live-workspace 의존 0 보장), 없으면
    이 파일(dispatch/) 기준 repo 상대 경로로 해석.
    """
    root = os.environ.get("WORKSPACE_ROOT")
    if root:
        return os.path.join(root, "memory", "events")
    here = os.path.dirname(os.path.abspath(__file__))
    return os.path.join(os.path.dirname(here), "memory", "events")


def callback_lifecycle_artifact_path(task_id: str, events_dir: Optional[str] = None) -> str:
    """artifact 파일 경로. events_dir 미지정 시 default_events_dir()."""
    base = events_dir if events_dir is not None else default_events_dir()
    return os.path.join(base, f"{task_id}{CALLBACK_LIFECYCLE_ARTIFACT_SUFFIX}")


def write_callback_lifecycle_artifact(
    task_id: str,
    evidence: Dict,
    *,
    events_dir: Optional[str] = None,
    anu_keys=DEFAULT_ANU_KEYS,
) -> str:
    """`memory/events/<task>.callback_lifecycle.json` 기록 (idempotent · 필수구현 3).

    - 동일 입력 2회 실행 → byte-identical (결정적 직렬화).
    - 기존 파일 내용이 이미 동일하면 재기록하지 않는다(중복 write 0 · mtime 보존).
    - cron 0 · callback 재발사 0 · subprocess 0 — 오직 lifecycle artifact 파일
      1개만 생성/갱신한다(필수구현 6: fallback collector artifact 미접촉).
    반환: artifact 파일 절대/상대 경로.
    """
    artifact = build_callback_lifecycle_artifact(task_id, evidence, anu_keys=anu_keys)
    payload = _serialize_lifecycle_artifact(artifact)
    path = callback_lifecycle_artifact_path(task_id, events_dir)

    existing: Optional[str] = None
    # TOCTOU 제거 (Gemini medium): exists() 체크-후-open 대신 직접 open 시도.
    try:
        with open(path, "r", encoding="utf-8") as fh:
            existing = fh.read()
    except (FileNotFoundError, IsADirectoryError, PermissionError):
        pass
    if existing != payload:
        dirpath = os.path.dirname(path) or "."
        os.makedirs(dirpath, exist_ok=True)
        # atomic write (Gemini medium): 같은 디렉토리 temp 파일에 쓰고 os.replace 로
        # 원자적 교체 — 쓰기 도중 중단/오류에도 artifact 가 손상/불완전 상태로 남지 않는다.
        fd, tmp = tempfile.mkstemp(dir=dirpath, prefix=".callback_lifecycle.", suffix=".tmp")
        try:
            with os.fdopen(fd, "w", encoding="utf-8") as fh:
                fh.write(payload)
            os.replace(tmp, path)
        except BaseException:
            try:
                os.unlink(tmp)
            except OSError:
                pass
            raise
    return path


# ═════════════════════════════════════════════════════════════════════════════
# task-2739 — executor result file-contract 강제 + ANU-owned pickup scan-path 결선
#
# 문제(전수조사 §0): executor result.json 이 세션 workspace 에만 생성되고,
# finish-task.sh writer path(memory/events 직속) 와 pickup runner scan path
# (memory/events/p0b_inbox) 가 영원히 단절 → ANU-owned pickup 불가. 또한 schema
# 결손 + 미작성/불완전 FAIL 로직 부재로 "PR 만 만들고 result.json 없이 끝나도 성공"
# 오인 발생.
#
# 본 섹션은 (1) writer path 를 canonical p0b_inbox(memory/events/p0b_inbox/
# task-*.result.json) 로 일치, (2) 12-field schema 강제, (3) 미작성/불완전 시
# NON_AUTHORITATIVE_INCOMPLETE_RESULT 기록 + FAIL(비-0 exit) 을 코드로 강제한다.
#
# ★ ANU key literal 0: 이 writer 는 ANU key 를 전혀 참조하지 않는다(§4.4). executor 는
#   result.json 만 남기고 owner proof 는 ANU-key runner(anu_pickup_driver) 만 채운다.
#   result 에는 raw key 대신 owner_key_proof_present=False marker 만 둔다(§8).
# ★ NO-CRON / self-fire 0: cron register/remove 0, callback fire 0, sleep/polling 0.
#   systemd/P0B 실가동/driver loop/ACTIVE=true 와 무관(wired-candidate, §7).
# ★ core 로직(build/validate/write)은 subprocess/네트워크 I/O 0 — 순수 함수.
#   git-derivation(subprocess)은 CLI main() 경계에서만 호출된다.
# ═════════════════════════════════════════════════════════════════════════════

# writer↔runner path 단일 일치 지점(§4.1/§4.2). 아래 두 상수는
# dispatch.anu_pickup_driver.INBOX_DIR_REL / RESULT_GLOB 과 **동일 문자열**이어야
# 하며, 전용 회귀 테스트(§6.6)가 두 모듈 상수의 런타임 동등성을 assert 한다.
EXECUTOR_RESULT_INBOX_DIR_REL = "memory/events/p0b_inbox"
EXECUTOR_RESULT_GLOB = "task-*.result.json"

EXECUTOR_RESULT_SCHEMA = "dispatch.executor_completion_contract.result.v1"

# §3 — result.json 최소 12 필드(single source).
REQUIRED_RESULT_FIELDS = (
    "task_id",
    "terminal_state",
    "executor_id",
    "authoritative",
    "callback_requested",
    "owner_pickup_required",
    "changed_files",
    "head_sha",
    "pr_number",
    "created_at",
    "workspace",
    "branch",
)

# 키 존재만으로 부족하고 비어있으면 안 되는(non-empty) 필드. pr_number 는 무-PR 시
# null 허용(키는 반드시 존재). authoritative/owner_pickup_required 는 고정 bool 이라
# 별도 강제. changed_files 는 빈 리스트 허용(변경 0 일 수 있음).
NON_EMPTY_RESULT_FIELDS = (
    "task_id",
    "terminal_state",
    "executor_id",
    "head_sha",
    "created_at",
    "workspace",
    "branch",
)

EXECUTOR_RESULT_WRITTEN = "EXECUTOR_RESULT_WRITTEN"
# §3 — 미작성·불완전 시 기록되는 비권위 terminal state(완료 인정 0).
NON_AUTHORITATIVE_INCOMPLETE_RESULT = "NON_AUTHORITATIVE_INCOMPLETE_RESULT"
# incomplete marker 는 절대 'task-*.result.json' glob 에 매칭되지 않는 suffix 를
# 쓴다(pickup driver scan 대상 0) + 전용 inbox 밖(memory/events 직속)에 둔다.
INCOMPLETE_MARKER_SUFFIX = ".incomplete-result.json"


def _utc_now_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _sanitize_basename_component(value) -> str:
    """basename 안전화 — path traversal/구분자 제거."""
    s = os.path.basename(str(value or "").strip())
    s = re.sub(r"[^A-Za-z0-9._+-]", "_", s)
    return s or "unknown"


def _task_prefixed(task_id: str) -> str:
    safe = _sanitize_basename_component(task_id)
    return safe if safe.startswith("task-") else "task-" + safe


def normalize_result_basename(task_id: str) -> str:
    """pickup runner glob(task-*.result.json) 호환 basename 보장(§4.1).

    task_id 가 'task-' 로 시작하지 않으면 prefix 를 붙여 writer↔runner glob 일치를
    보증한다 → 'task-<id>.result.json'.
    """
    return f"{_task_prefixed(task_id)}.result.json"


def canonical_inbox_dir(workspace_root: Optional[str] = None) -> str:
    """canonical p0b_inbox 절대경로. workspace_root 미지정 시 WORKSPACE_ROOT env →
    repo-relative 로 해석(default_events_dir() 과 동일 루트 규칙)."""
    if workspace_root is None:
        events = default_events_dir()          # <root>/memory/events
        root = os.path.dirname(os.path.dirname(events))
    else:
        root = workspace_root
    return os.path.join(root, *EXECUTOR_RESULT_INBOX_DIR_REL.split("/"))


# ═════════════════════════════════════════════════════════════════════════════
# task-2740 — canonical p0b_inbox 귀착 (A: freshness fix)
#
# 문제(task-2740 §0): result writer 의 canonical root 가 worktree CWD/`--workspace`
# 에 종속 → worktree 실행 시 result.json 이 worktree 의 memory/events/p0b_inbox
# 에 작성된다. worktree p0b_inbox 는 git-untracked 이고 merge 에 반영되지 않으므로
# canonical(메인 워크트리) p0b_inbox 는 영원히 STALE → pickup runner(canonical scan)
# 가 stale 만 보게 된다.
#
# fix(§2): result.json(pickup 대상)만 canonical workspace 로 귀착시킨다. canonical
# root resolver 우선순위(고정):
#   1. 명시 CANONICAL_WORKSPACE env(또는 동등한 명시 설정) → 우선 사용.
#   2. 없으면 `git rev-parse --git-common-dir` 의 부모 = 메인 워크트리 root.
#      (linked worktree 의 common dir 은 항상 공용 `.git` 을 가리키므로 그 부모가
#       canonical root 이다 — worktree CWD/`--workspace` 와 무관.)
#   3. 그래도 확정 불가하거나 결과가 `.worktrees` 내부면 → fail-closed(예외) 로 중단.
#      ★ worktree 안으로 떨어뜨려 p0b_inbox 가 `.worktrees` 내부에 생기는 것을 금지.
#
# ★ subprocess 0(순수 함수): resolver 는 git 을 직접 호출하지 않고 git_common_dir 을
#   주입받는다. git rev-parse(subprocess)는 CLI main() 경계에서만 수행된다.
# ★ hardcoded `/home/jay/workspace` literal 0 — resolver 로 계산한다.
# ═════════════════════════════════════════════════════════════════════════════

# §2.1 — 명시 canonical workspace override env(우선순위 1). 값이 있으면 git common
# dir 해석보다 우선한다. raw key/literal 0 — 단순 경로 문자열.
CANONICAL_WORKSPACE_ENV = "CANONICAL_WORKSPACE"

# §2.2 — git worktree 디렉토리 마커(이 컴포넌트가 경로에 있으면 canonical 아님).
_WORKTREE_DIR_COMPONENT = ".worktrees"


class CanonicalWorkspaceUnresolved(Exception):
    """canonical(메인 워크트리) workspace root 확정 불가 — fail-closed(§2.3).

    worktree CWD/`--workspace` 로 떨어뜨려 p0b_inbox 가 `.worktrees` 내부에 생기는
    것을 막기 위해, 확정 불가 시 result.json 을 쓰지 않고 명시적으로 중단한다."""


def _path_is_inside_worktree(path: str) -> bool:
    """경로가 git worktree 디렉토리(`.worktrees/`) 내부면 True → canonical 아님."""
    norm = os.path.normpath(os.path.abspath(path))
    return _WORKTREE_DIR_COMPONENT in norm.split(os.sep)


def resolve_canonical_workspace_root(
    *,
    explicit_canonical: Optional[str] = None,
    git_common_dir: Optional[str] = None,
    git_common_dir_relative_to: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
) -> str:
    """canonical(메인 워크트리) workspace root 를 결정한다(§2 resolver 우선순위, 고정).

    1. explicit_canonical(또는 CANONICAL_WORKSPACE env) 가 있으면 **우선 사용**.
    2. 없으면 git_common_dir(=`git rev-parse --git-common-dir` 결과)의 부모 디렉토리
       = 메인 워크트리 root. relative 인 경우 git_common_dir_relative_to 기준으로 해석.
    3. 그래도 확정 불가하거나 결과가 `.worktrees` 내부면 CanonicalWorkspaceUnresolved
       (fail-closed — worktree 로 떨어뜨리지 않는다).

    순수 함수: subprocess 0(git 호출은 CLI 경계에서 git_common_dir 로 주입)."""
    environ = os.environ if env is None else env
    cand = (explicit_canonical or environ.get(CANONICAL_WORKSPACE_ENV) or "").strip()
    if cand:
        root = os.path.normpath(os.path.abspath(cand))
        if _path_is_inside_worktree(root):
            raise CanonicalWorkspaceUnresolved(
                f"명시 canonical workspace 가 .worktrees 내부(canonical 아님): {root}"
            )
        return root
    gcd = (git_common_dir or "").strip()
    if gcd:
        if not os.path.isabs(gcd) and git_common_dir_relative_to:
            gcd = os.path.join(git_common_dir_relative_to, gcd)
        gcd = os.path.normpath(os.path.abspath(gcd))
        # `--git-common-dir` 은 (linked worktree 에서도) 항상 공용 `.git` 디렉토리를
        # 가리킨다 → 그 부모가 메인 워크트리 root.
        if os.path.basename(gcd) == ".git":
            root = os.path.dirname(gcd)
            if root and not _path_is_inside_worktree(root):
                return root
    raise CanonicalWorkspaceUnresolved(
        "canonical workspace root 확정 불가(CANONICAL_WORKSPACE env / git common dir "
        "모두 결정 불가) — fail-closed: worktree 로 떨어뜨리지 않고 중단."
    )


def canonical_inbox_dir_resolved(
    *,
    explicit_canonical: Optional[str] = None,
    git_common_dir: Optional[str] = None,
    git_common_dir_relative_to: Optional[str] = None,
    env: Optional[Dict[str, str]] = None,
) -> str:
    """resolver 로 canonical workspace root 를 구해 canonical p0b_inbox 절대경로 반환.

    확정 불가 시 CanonicalWorkspaceUnresolved 전파(fail-closed)."""
    root = resolve_canonical_workspace_root(
        explicit_canonical=explicit_canonical,
        git_common_dir=git_common_dir,
        git_common_dir_relative_to=git_common_dir_relative_to,
        env=env,
    )
    return canonical_inbox_dir(root)


def canonical_result_path(
    task_id: str,
    *,
    inbox_dir: Optional[str] = None,
    workspace_root: Optional[str] = None,
) -> str:
    """canonical result.json 절대경로(memory/events/p0b_inbox/task-*.result.json)."""
    base = inbox_dir if inbox_dir is not None else canonical_inbox_dir(workspace_root)
    return os.path.join(base, normalize_result_basename(task_id))


def build_result_record(
    *,
    task_id: str,
    terminal_state: str,
    executor_id: str,
    callback_requested: bool,
    changed_files,
    head_sha: str,
    pr_number,
    workspace: str,
    branch: str,
    created_at: Optional[str] = None,
    report_path: str = "",
) -> Dict:
    """12-field result.json 레코드 구성(순수·결정적, I/O 0).

    authoritative=False / owner_pickup_required=True 는 입력과 무관하게 **강제
    고정**한다(§2/§3 — executor 는 절대 self-authoritative 가 아니며 owner pickup
    이 필수). raw ANU key 대신 owner_key_proof_present=False marker 만 둔다(§8 —
    실제 owner proof 는 ANU runner 가 채움).
    """
    return {
        "schema": EXECUTOR_RESULT_SCHEMA,
        "task_id": str(task_id or ""),
        "terminal_state": str(terminal_state or ""),
        "executor_id": str(executor_id or ""),
        # ★ 강제 고정(입력 무시) — self-collector/self-authoritative 통제.
        "authoritative": False,
        "callback_requested": bool(callback_requested),
        "owner_pickup_required": True,
        "changed_files": list(changed_files or []),
        "head_sha": str(head_sha or ""),
        "pr_number": pr_number,                 # int 또는 None(무-PR)
        "created_at": created_at or _utc_now_iso(),
        "workspace": str(workspace or ""),
        "branch": str(branch or ""),
        # ── self-collector 통제 marker (§8 — raw key 기록 0) ──
        "owner_key_proof_present": False,
        "schedule_owner_proof": "PENDING_ANU",
        # executor 가 절대 하지 않는 것(자기-증명) — schedule/콜백 미발사.
        "schedule_created_by_executor": False,
        "callback_fired_by_executor": False,
        "report_path": str(report_path or ""),
    }


def validate_result_schema(record: Dict) -> List[str]:
    """12-field 완전성 FAIL 사유 목록(빈 리스트 == 완전). §3/§6.4.

    - 12 필수 키가 모두 존재(누락 → FAIL).
    - NON_EMPTY_RESULT_FIELDS 는 빈 문자열/None 불가.
    - authoritative 는 반드시 False, owner_pickup_required 는 반드시 True.
    - changed_files 는 list 이며 모든 요소가 str(파일경로), callback_requested 는 bool.
    """
    reasons: List[str] = []
    rec = record or {}
    for key in REQUIRED_RESULT_FIELDS:
        if key not in rec:
            reasons.append(f"필수 schema 필드 누락: {key}")
    for key in NON_EMPTY_RESULT_FIELDS:
        if key in rec and (rec.get(key) is None or str(rec.get(key)).strip() == ""):
            reasons.append(f"필수 필드 비어있음: {key}")
    if rec.get("authoritative", None) is not False:
        reasons.append("authoritative 는 반드시 False (executor self-authoritative 0)")
    if rec.get("owner_pickup_required", None) is not True:
        reasons.append("owner_pickup_required 는 반드시 True (ANU-owned pickup 필수)")
    if "changed_files" in rec:
        cf = rec.get("changed_files")
        if not isinstance(cf, list):
            reasons.append("changed_files 는 list 여야 함")
        elif not all(isinstance(x, str) for x in cf):
            reasons.append("changed_files 의 모든 요소는 str(파일경로) 여야 함")
    if "callback_requested" in rec and not isinstance(rec.get("callback_requested"), bool):
        reasons.append("callback_requested 는 bool(true/false) 여야 함")
    return reasons


def result_schema_complete(record: Dict) -> bool:
    return not validate_result_schema(record)


def _atomic_write_result_json(record: Dict, path: str) -> str:
    """result.json 을 atomic 하게 기록(tmp write → fsync → os.replace)."""
    payload = json.dumps(record, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
    dirpath = os.path.dirname(path) or "."
    os.makedirs(dirpath, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=dirpath, prefix=".result.", suffix=".tmp")
    # fd leak 보호 (Gemini #2): os.fdopen 이 fd 소유권을 가져가기 전에 예외가 나면
    # fd 가 누수된다. fdopen 성공 시에만 fd_owned=True 로 표시하고, 실패 시(except)에는
    # 직접 os.close(fd) 로 회수한다(fdopen 성공 후엔 with 가 닫으므로 이중 close 방지).
    fd_owned = False
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as fh:
            fd_owned = True
            fh.write(payload)
            fh.flush()
            os.fsync(fh.fileno())
        os.replace(tmp, path)
    except BaseException:
        if not fd_owned:
            try:
                os.close(fd)
            except OSError:
                pass
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise
    return path


def _incomplete_marker_dir(
    workspace_root: Optional[str],
    events_dir: Optional[str],
    inbox_dir: Optional[str] = None,
) -> str:
    """incomplete marker 디렉토리 결정(§2 — inbox_dir 일관성 fix).

    우선순위: events_dir → inbox_dir 의 부모 디렉토리 → workspace_root/memory/events
    → default_events_dir(). inbox_dir 이 명시(canonical p0b_inbox = <root>/memory/
    events/p0b_inbox)되면 그 부모(<root>/memory/events)에 marker 를 둔다 — events_dir
    /workspace_root 가 없다는 이유로 canonical default_events_dir() 로 떨어져 격리
    환경을 오염시키지 않는다(happy path result.json inbox_dir 사용과 일관).
    """
    if events_dir is not None:
        return events_dir
    if inbox_dir is not None:
        return os.path.dirname(os.path.abspath(inbox_dir))
    if workspace_root is not None:
        return os.path.join(workspace_root, "memory", "events")
    return default_events_dir()


def _write_incomplete_marker(
    task_id: str,
    reasons: List[str],
    record: Dict,
    *,
    events_dir: Optional[str] = None,
    workspace_root: Optional[str] = None,
    inbox_dir: Optional[str] = None,
) -> str:
    """NON_AUTHORITATIVE_INCOMPLETE_RESULT marker 기록(§3/§6.3).

    canonical result.json(p0b_inbox)은 작성하지 않고, pickup glob 에 매칭되지 않는
    suffix 로 memory/events 직속에 marker 만 남긴다 → pickup 대상 0, 완료 인정 0.
    """
    base = _incomplete_marker_dir(workspace_root, events_dir, inbox_dir)
    path = os.path.join(base, f"{_task_prefixed(task_id)}{INCOMPLETE_MARKER_SUFFIX}")
    marker = {
        "schema": EXECUTOR_RESULT_SCHEMA,
        "task_id": str(task_id),
        "terminal_state": NON_AUTHORITATIVE_INCOMPLETE_RESULT,
        "authoritative": False,
        "owner_pickup_required": True,
        "owner_key_proof_present": False,
        "created_at": _utc_now_iso(),
        "missing_or_invalid": list(reasons),
        "note": (
            "result.json incomplete/missing — canonical p0b_inbox result 미작성. "
            "PR 만 있고 result.json 없으면 성공 아님(§2.4). ANU 독립 검증 필요."
        ),
        "partial_record_keys": sorted(list((record or {}).keys())),
    }
    _atomic_write_result_json(marker, path)
    return path


@dataclass
class ExecutorResultWriteOutcome:
    """write_executor_result 결과. ok=False 면 canonical result.json 은 미작성이고
    incomplete marker 만 존재한다(완료 인정 0)."""

    ok: bool
    terminal_state: str
    result_path: Optional[str]
    incomplete_marker_path: Optional[str]
    reasons: List[str] = field(default_factory=list)

    def to_json(self) -> Dict:
        return {
            "ok": self.ok,
            "terminal_state": self.terminal_state,
            "result_path": self.result_path,
            "incomplete_marker_path": self.incomplete_marker_path,
            "reasons": list(self.reasons),
        }


def write_executor_result(
    record: Dict,
    *,
    inbox_dir: Optional[str] = None,
    workspace_root: Optional[str] = None,
    events_dir: Optional[str] = None,
) -> ExecutorResultWriteOutcome:
    """12-field 강제 후 canonical p0b_inbox 에 result.json 작성(§2/§3/§4).

    완전(schema-complete) → memory/events/p0b_inbox/task-*.result.json atomic 작성
      → ok=True(terminal_state = record 의 값, 기본 EXECUTOR_RESULT_WRITTEN).
    불완전/미작성 → canonical result.json 은 **작성하지 않고**(pickup 대상 0),
      NON_AUTHORITATIVE_INCOMPLETE_RESULT marker 만 기록 → ok=False.
      (PR-only without result.json → 성공 0 — §2.4/§6.10.)

    cron 0 · callback fire 0 · ANU key 참조 0 · subprocess 0.
    """
    reasons = validate_result_schema(record)
    task_id = str((record or {}).get("task_id") or "unknown")
    if reasons:
        marker = _write_incomplete_marker(
            task_id, reasons, record,
            events_dir=events_dir, workspace_root=workspace_root,
            inbox_dir=inbox_dir,
        )
        return ExecutorResultWriteOutcome(
            ok=False,
            terminal_state=NON_AUTHORITATIVE_INCOMPLETE_RESULT,
            result_path=None,
            incomplete_marker_path=marker,
            reasons=reasons,
        )
    path = canonical_result_path(
        task_id, inbox_dir=inbox_dir, workspace_root=workspace_root
    )
    _atomic_write_result_json(record, path)
    return ExecutorResultWriteOutcome(
        ok=True,
        terminal_state=str(record.get("terminal_state") or EXECUTOR_RESULT_WRITTEN),
        result_path=path,
        incomplete_marker_path=None,
        reasons=[],
    )


# ── CLI git-derivation 경계(subprocess) — core 로직 외부에서만 호출 ───────────────
def _git_capture(proj_dir: str, args: List[str], default: str = "") -> str:
    try:
        out = subprocess.run(
            ["git", "-C", proj_dir, *args],
            capture_output=True, text=True, timeout=20, check=False,
        )
        if out.returncode == 0:
            return out.stdout.strip()
    except (OSError, subprocess.SubprocessError):
        pass
    return default


def _derive_changed_files(proj_dir: str, base_ref: str = "") -> List[str]:
    files = set()
    if base_ref:
        d = _git_capture(proj_dir, ["diff", "--name-only", f"{base_ref}...HEAD"])
        files.update(f for f in d.splitlines() if f.strip())
    for extra in (["diff", "--name-only"], ["diff", "--cached", "--name-only"]):
        d = _git_capture(proj_dir, extra)
        files.update(f for f in d.splitlines() if f.strip())
    return sorted(files)


def main(argv: Optional[List[str]] = None) -> int:
    import argparse

    ap = argparse.ArgumentParser(
        prog="dispatch.executor_completion_contract",
        description="executor result file-contract (canonical p0b_inbox + 12-field).",
    )
    sub = ap.add_subparsers(dest="cmd", required=True)

    ew = sub.add_parser(
        "executor-write-result",
        help="canonical p0b_inbox 에 12-field result.json 작성(미작성/불완전 → FAIL).",
    )
    ew.add_argument("--task-id", required=True)
    ew.add_argument("--workspace", default=os.environ.get("WORKSPACE_ROOT", ""))
    ew.add_argument("--proj-dir", default="")
    ew.add_argument("--executor-id", default="")
    ew.add_argument("--terminal-state", default=EXECUTOR_RESULT_WRITTEN)
    ew.add_argument("--branch", default="")
    ew.add_argument("--head-sha", default="")
    ew.add_argument("--pr-number", default="")
    ew.add_argument("--base-ref", default="")
    ew.add_argument("--report-path", default="")
    ew.add_argument(
        "--callback-requested", default="true", choices=["true", "false"]
    )
    ew.add_argument("--inbox-dir", default="")
    # task-2740 §2.1 — 명시 canonical workspace override(우선순위 1). 미지정 시
    #   CANONICAL_WORKSPACE env, 그래도 없으면 git common dir 로 canonical root 해석.
    ew.add_argument(
        "--canonical-workspace",
        default=os.environ.get(CANONICAL_WORKSPACE_ENV, ""),
    )

    va = sub.add_parser(
        "validate", help="기존 result.json 의 12-field schema 검증(완전 → exit 0)."
    )
    va.add_argument("--result-json-path", required=True)

    a = ap.parse_args(argv)

    if a.cmd == "executor-write-result":
        proj = a.proj_dir or a.workspace or os.getcwd()
        head = a.head_sha or _git_capture(proj, ["rev-parse", "HEAD"])
        branch = a.branch or _git_capture(proj, ["rev-parse", "--abbrev-ref", "HEAD"])
        workspace = a.workspace or proj
        executor_id = (
            a.executor_id
            or os.environ.get("EXECUTOR_ID", "")
            or os.environ.get("BOT_NAME", "")
            or _git_capture(proj, ["config", "user.name"])
            or "unknown-executor"
        )
        pr_number = None
        if str(a.pr_number).strip():
            raw = str(a.pr_number).strip().lstrip("#")
            try:
                pr_number = int(raw)
            except ValueError:
                pr_number = raw
        changed = _derive_changed_files(proj, a.base_ref)
        record = build_result_record(
            task_id=a.task_id,
            terminal_state=a.terminal_state,
            executor_id=executor_id,
            callback_requested=(a.callback_requested == "true"),
            changed_files=changed,
            head_sha=head,
            pr_number=pr_number,
            workspace=workspace,
            branch=branch,
            report_path=a.report_path,
        )
        # ── task-2740 §2: canonical p0b_inbox 귀착 ───────────────────────────
        # result.json(pickup 대상)은 worktree CWD/`--workspace` 가 아니라 canonical
        # (메인 워크트리) workspace 의 p0b_inbox 에 써야 한다. `--inbox-dir` 명시
        # (테스트/override)는 그대로 존중하고, 없으면 resolver(CANONICAL_WORKSPACE
        # env → git common dir 부모 → fail-closed)로 canonical inbox 를 결정한다.
        # ★ git rev-parse(subprocess)는 여기(CLI 경계)에서만 호출 — core 는 순수.
        inbox_dir = (a.inbox_dir or "").strip()
        if not inbox_dir:
            git_common = _git_capture(proj, ["rev-parse", "--git-common-dir"])
            try:
                inbox_dir = canonical_inbox_dir_resolved(
                    explicit_canonical=(a.canonical_workspace or None),
                    git_common_dir=(git_common or None),
                    git_common_dir_relative_to=proj,
                )
            except CanonicalWorkspaceUnresolved as exc:
                # fail-closed: worktree p0b_inbox 로 떨어뜨리지 않는다. canonical 미확정
                # 이므로 진단용 incomplete marker(pickup glob 비매칭 suffix)만 실행
                # workspace 의 memory/events 직속에 남기고 FAIL(exit 1).
                reason = f"canonical workspace root 미확정(fail-closed): {exc}"
                marker = _write_incomplete_marker(
                    a.task_id, [reason], record, workspace_root=workspace
                )
                print(json.dumps(
                    {
                        "ok": False,
                        "terminal_state": NON_AUTHORITATIVE_INCOMPLETE_RESULT,
                        "result_path": None,
                        "incomplete_marker_path": marker,
                        "reasons": [reason],
                    },
                    ensure_ascii=False,
                ))
                return 1
        outcome = write_executor_result(record, inbox_dir=inbox_dir)
        print(json.dumps(outcome.to_json(), ensure_ascii=False))
        return 0 if outcome.ok else 1

    if a.cmd == "validate":
        try:
            with open(a.result_json_path, "r", encoding="utf-8") as fh:
                rec = json.load(fh)
        except (OSError, ValueError) as exc:
            print(json.dumps(
                {"ok": False, "reasons": [f"load 실패: {exc}"]},
                ensure_ascii=False,
            ))
            return 1
        reasons = validate_result_schema(rec)
        print(json.dumps({"ok": not reasons, "reasons": reasons}, ensure_ascii=False))
        return 0 if not reasons else 1

    return 2


if __name__ == "__main__":
    import sys as _sys

    raise SystemExit(main(_sys.argv[1:]))