"""spawn_safety_governor.py — task-2774: 무한 spawn 방지 게이트(governor).

stdlib only, 외부 의존 0. 순수 함수 + ledger I/O.
governor 자체는 절대 새 프로세스/재귀를 생성하지 않는다.
"""
from __future__ import annotations

import json
import os
import re
import tempfile
from datetime import datetime, timezone
from typing import Optional

# ── config 상수 (운영 활성화 X, 값 정의만) ────────────────────────────────────
RATE_PER_MIN = 1          # 분당 허용 spawn 최대치
RATE_PER_HOUR = 5         # 시간당 허용 spawn 최대치
LOOP_PER_TASK = 3         # task_id 누적 ALLOW spawn 한도
LOOP_PER_FAMILY = 5       # task family 누적 ALLOW spawn 한도
CIRCUIT_WINDOW_MIN = 10   # circuit-breaker 윈도우(분)
CIRCUIT_MAX_SPAWN = 10    # 윈도우 내 최대 spawn 수
CIRCUIT_FAIL_RATE = 0.5   # 윈도우 내 실패율 임계

# ── terminal 상태 집합 ─────────────────────────────────────────────────────────
TERMINAL_STATES = ("completed", "failed", "blocked", "crash")

# ── decision 값 ───────────────────────────────────────────────────────────────
ALLOW = "ALLOW"
BLOCK = "BLOCK"
QUEUE = "QUEUE"
TRIP  = "TRIP"

# ── reason 값 ─────────────────────────────────────────────────────────────────
REASON_ALLOW             = "ALLOW"
REASON_NON_TERMINAL      = "NON_TERMINAL"
REASON_DEDUP_HIT         = "DEDUP_HIT"
REASON_SINGLEFLIGHT_LIVE = "SINGLEFLIGHT_LIVE"
REASON_RATE_EXCEEDED     = "RATE_EXCEEDED"
REASON_LOOP_EXCEEDED     = "LOOP_EXCEEDED"
REASON_CIRCUIT_TRIP      = "CIRCUIT_TRIP"
# fail-closed reasons (task-2774+2)
REASON_COUNTERS_UNREADABLE  = "COUNTERS_UNREADABLE"
REASON_COUNTERS_SAVE_FAILED = "COUNTERS_SAVE_FAILED"
REASON_LEDGER_UNREADABLE    = "LEDGER_UNREADABLE"

# ── 기본 경로 상수 ─────────────────────────────────────────────────────────────
CANONICAL_ROOT = "/home/jay/workspace"
DEFAULT_DEDUP_LEDGER    = os.path.join(CANONICAL_ROOT, "memory", "events", "callback_4tuple_index.jsonl")
DEFAULT_DECISIONS_LOG   = os.path.join(CANONICAL_ROOT, "memory", "state", "spawn_decisions.jsonl")
DEFAULT_COUNTERS_PATH   = os.path.join(CANONICAL_ROOT, "memory", "state", "spawn_counters.json")
DEFAULT_LIVE_LOCK       = os.path.join(CANONICAL_ROOT, "memory", "state", "anu_session_alive.lock")
DEFAULT_CIRCUIT_MARKER  = os.path.join(CANONICAL_ROOT, "memory", "state", "spawn_circuit_tripped.marker")
DEFAULT_P0B_FLAG        = os.path.join(CANONICAL_ROOT, "memory", "state", "p0b_driver_enabled")

__all__ = [
    "evaluate_spawn",
    "_task_family",
    # config 상수
    "RATE_PER_MIN", "RATE_PER_HOUR", "LOOP_PER_TASK", "LOOP_PER_FAMILY",
    "CIRCUIT_WINDOW_MIN", "CIRCUIT_MAX_SPAWN", "CIRCUIT_FAIL_RATE",
    # terminal
    "TERMINAL_STATES",
    # decision 값
    "ALLOW", "BLOCK", "QUEUE", "TRIP",
    # reason 값
    "REASON_ALLOW", "REASON_NON_TERMINAL", "REASON_DEDUP_HIT",
    "REASON_SINGLEFLIGHT_LIVE", "REASON_RATE_EXCEEDED",
    "REASON_LOOP_EXCEEDED", "REASON_CIRCUIT_TRIP",
    "REASON_COUNTERS_UNREADABLE", "REASON_COUNTERS_SAVE_FAILED",
    "REASON_LEDGER_UNREADABLE",
    # 경로 상수
    "CANONICAL_ROOT",
    "DEFAULT_DEDUP_LEDGER", "DEFAULT_DECISIONS_LOG", "DEFAULT_COUNTERS_PATH",
    "DEFAULT_LIVE_LOCK", "DEFAULT_CIRCUIT_MARKER", "DEFAULT_P0B_FLAG",
]

# ──────────────────────────────────────────────────────────────────────────────
# 내부 헬퍼
# ──────────────────────────────────────────────────────────────────────────────

def _cand_get(candidate, name: str, default: str = "") -> str:
    """candidate dict 또는 attribute 접근 가능 객체에서 필드 값 반환."""
    if isinstance(candidate, dict):
        return str(candidate.get(name, default) or default)
    return str(getattr(candidate, name, default) or default)


def _task_family(task_id: str) -> str:
    """task_id에서 base 'task-N' 추출.

    예: task-2774+1 → task-2774, task-2774-r2 → task-2774, task-2774 → task-2774.
    """
    m = re.match(r"^(task-\d+)", task_id or "")
    return m.group(1) if m else task_id


def _now_utc() -> datetime:
    """현재 UTC datetime(timezone-aware) 반환."""
    return datetime.now(timezone.utc)


def _iso(dt: datetime) -> str:
    """datetime → ISO8601 UTC 문자열."""
    return dt.strftime("%Y-%m-%dT%H:%M:%SZ")


def _load_counters(counters_path: str) -> dict:
    """spawn_counters.json 로드.

    (a) 파일 부재(첫 실행, 정상) → 빈 dict 반환(ALLOW 경로 보존).
    (b) 파일 존재하나 손상/파싱오류/권한·IO 오류(위험) → 예외 전파 → 호출부에서
        fail-closed(COUNTERS_UNREADABLE → BLOCK) 처리. 빈 dict 로 무시 금지(fail-open 차단).
    """
    try:
        with open(counters_path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        return {}  # (a) 첫 실행: 정상 빈 카운터
    # (b) 손상/파싱/IO/권한 오류는 그대로 전파(fail-closed)
    if not isinstance(data, dict):
        # 파일은 존재하나 dict 가 아님 = 손상 → fail-closed
        raise ValueError("spawn_counters.json is not a dict (corrupted)")
    return data


def _save_counters(counters_path: str, counters: dict) -> None:
    """spawn_counters.json 원자적 저장 (tempfile + os.replace).

    저장 실패(디스크 풀/권한/IO 오류) 시 예외를 상위로 전파한다(silent ignore 금지).
    호출부는 저장 실패를 fail-closed(COUNTERS_SAVE_FAILED → BLOCK)로 처리하여
    count 미증가로 인한 무한 spawn 을 차단한다.
    """
    dir_ = os.path.dirname(counters_path) or "."
    os.makedirs(dir_, exist_ok=True)
    fd, tmp_path = tempfile.mkstemp(dir=dir_, suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as f:
            json.dump(counters, f, ensure_ascii=False)
        os.replace(tmp_path, counters_path)
    except Exception:
        try:
            os.unlink(tmp_path)
        except Exception:
            pass
        raise  # 저장 실패 전파 → 호출부 fail-closed


def _append_decision(decisions_path: str, record: dict) -> None:
    """spawn_decisions.jsonl에 결정 1줄 append (silent drop 0)."""
    try:
        dir_ = os.path.dirname(decisions_path) or "."
        os.makedirs(dir_, exist_ok=True)
        with open(decisions_path, "a", encoding="utf-8") as f:
            f.write(json.dumps(record, ensure_ascii=False) + "\n")
    except Exception:
        pass  # 파일 I/O 실패 시 governor는 계속 진행


def _iter_jsonl_lines(path: str):
    """JSONL 파일을 라인 단위로 스트리밍하며 파싱된 dict 를 yield (bounded read).

    전량 메모리 적재 금지 — 호출부는 일치 시 즉시 break 하여 조기탈출한다.
    (a) 파일 부재(FileNotFoundError) → 빈 제너레이터(정상 초기 상태).
    (b) 권한/IO 오류 등 기타 open 예외 → 상위로 전파 → 호출부 fail-closed
        (LEDGER_UNREADABLE → BLOCK). 손상된 개별 라인은 skip(append-only 부분쓰기 허용).
    """
    try:
        f = open(path, "r", encoding="utf-8")
    except FileNotFoundError:
        return  # (a) 빈 제너레이터
    # (b) PermissionError/IsADirectoryError/OSError 등은 전파(fail-closed)
    try:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except Exception:
                continue  # 손상 라인 skip
            if isinstance(obj, dict):
                yield obj
    finally:
        f.close()


def _get_event_epoch(e: dict) -> float:
    """이벤트의 epoch(float) 반환. 'epoch' 필드 우선, 없으면 'ts' 문자열 1회 파싱."""
    if "epoch" in e:
        try:
            return float(e["epoch"])
        except Exception:
            pass
    return _ts_to_epoch(str(e.get("ts", "")))


# ──────────────────────────────────────────────────────────────────────────────
# 핵심 함수
# ──────────────────────────────────────────────────────────────────────────────

def evaluate_spawn(
    candidate,
    *,
    counters_path: Optional[str] = None,
    ledger_path: Optional[str] = None,
    now: Optional[datetime] = None,
    live_lock_present: Optional[bool] = None,
    decisions_path: Optional[str] = None,
    circuit_marker_path: Optional[str] = None,
    p0b_flag_path: Optional[str] = None,
) -> dict:
    """spawn 가능 여부를 평가하고 Decision dict를 반환한다.

    차단 순서(첫 차단에서 멈춤, 모든 결정은 spawn_decisions.jsonl에 append):
      1. terminal-only HARD GATE: terminal_state가 TERMINAL_STATES에 없으면 BLOCK
      2. dedup: 동일 key가 이미 ALLOW로 기록돼 있으면 BLOCK
      3. single-flight: live_lock 존재 시 QUEUE
      4. rate-limit: 분당/시간당 초과 시 QUEUE
      5. loop-budget: task/family 누적 ALLOW 초과 시 BLOCK
      6. circuit-breaker: 윈도우 내 spawn/실패율 초과 시 TRIP
      7. 전부 통과 → ALLOW

    반환 구조: dict(decision, reason, task_id, key, ts)
    """
    # 경로 기본값 설정
    if counters_path is None:
        counters_path = DEFAULT_COUNTERS_PATH
    if ledger_path is None:
        ledger_path = DEFAULT_DEDUP_LEDGER
    if decisions_path is None:
        decisions_path = DEFAULT_DECISIONS_LOG
    if circuit_marker_path is None:
        circuit_marker_path = DEFAULT_CIRCUIT_MARKER
    if p0b_flag_path is None:
        p0b_flag_path = DEFAULT_P0B_FLAG

    # candidate 필드 추출
    task_id        = _cand_get(candidate, "task_id")
    head_sha       = _cand_get(candidate, "head_sha")
    terminal_state = _cand_get(candidate, "terminal_state")

    # key = head_sha + terminal_state 조합
    key = f"{head_sha}|{terminal_state}"

    # now 설정
    if now is None:
        now = _now_utc()
    ts_str = _iso(now)

    def _make_decision(decision: str, reason: str) -> dict:
        return dict(
            decision=decision,
            reason=reason,
            task_id=task_id,
            key=key,
            ts=ts_str,
        )

    def _record_and_return(d: dict) -> dict:
        """spawn_decisions.jsonl에 append 후 반환."""
        _append_decision(decisions_path, d)
        return d

    # ── 게이트 1: terminal-only HARD GATE ─────────────────────────────────────
    if terminal_state not in TERMINAL_STATES:
        return _record_and_return(_make_decision(BLOCK, REASON_NON_TERMINAL))

    # ── 게이트 2: dedup (스트리밍 bounded read + 일치 시 즉시 조기탈출) ─────────
    # 파일 부재=정상(빈 제너레이터). 권한/IO 오류=보수적 fail-closed(LEDGER_UNREADABLE).
    try:
        # (a) dedup ledger(callback_4tuple_index.jsonl)에서 task_id + head_sha 매칭
        for entry in _iter_jsonl_lines(ledger_path):
            if (str(entry.get("task_id", "")) == task_id
                    and str(entry.get("head_sha", "")) == head_sha):
                return _record_and_return(_make_decision(BLOCK, REASON_DEDUP_HIT))

        # (b) spawn_decisions.jsonl에서 동일 key + decision==ALLOW 존재 확인
        for entry in _iter_jsonl_lines(decisions_path):
            if (str(entry.get("key", "")) == key
                    and str(entry.get("decision", "")) == ALLOW):
                return _record_and_return(_make_decision(BLOCK, REASON_DEDUP_HIT))
    except Exception:
        # 손상/권한·IO 오류 → 보수적 차단(전량 적재·fail-open 금지)
        return _record_and_return(_make_decision(BLOCK, REASON_LEDGER_UNREADABLE))

    # ── 게이트 3: single-flight ────────────────────────────────────────────────
    if live_lock_present is None:
        live_lock_present = os.path.exists(DEFAULT_LIVE_LOCK)
    if live_lock_present:
        return _record_and_return(_make_decision(QUEUE, REASON_SINGLEFLIGHT_LIVE))

    # ── 카운터 로드 (rate/loop/circuit 판정 전 1회만) ─────────────────────────
    # 파일 부재=정상 빈 카운터(ALLOW 경로 보존). 손상/IO·권한 오류=fail-closed(COUNTERS_UNREADABLE).
    try:
        counters = _load_counters(counters_path)
    except Exception:
        return _record_and_return(_make_decision(BLOCK, REASON_COUNTERS_UNREADABLE))
    events: list = counters.get("events", [])
    if not isinstance(events, list):
        events = []

    now_ts = now.timestamp()

    # ── 게이트 4: rate-limit (현재 시도 포함 전 기준) ─────────────────────────
    events_1min  = [e for e in events if now_ts - _get_event_epoch(e) <= 60]
    events_1hour = [e for e in events if now_ts - _get_event_epoch(e) <= 3600]

    if len(events_1min) >= RATE_PER_MIN or len(events_1hour) >= RATE_PER_HOUR:
        return _record_and_return(_make_decision(QUEUE, REASON_RATE_EXCEEDED))

    # ── 게이트 5: loop-budget ──────────────────────────────────────────────────
    per_task: dict   = counters.get("per_task", {})
    per_family: dict = counters.get("per_family", {})
    if not isinstance(per_task, dict):
        per_task = {}
    if not isinstance(per_family, dict):
        per_family = {}

    family = _task_family(task_id)
    task_count   = int(per_task.get(task_id, 0))
    family_count = int(per_family.get(family, 0))

    if task_count >= LOOP_PER_TASK or family_count >= LOOP_PER_FAMILY:
        return _record_and_return(_make_decision(BLOCK, REASON_LOOP_EXCEEDED))

    # ── 게이트 6: circuit-breaker ──────────────────────────────────────────────
    window_sec = CIRCUIT_WINDOW_MIN * 60
    events_window = [e for e in events if now_ts - _get_event_epoch(e) <= window_sec]
    window_spawn_count = len(events_window)
    fail_count = sum(1 for e in events_window if str(e.get("outcome", "")) == "fail")
    fail_rate = (fail_count / window_spawn_count) if window_spawn_count > 0 else 0.0

    circuit_tripped = (
        window_spawn_count >= CIRCUIT_MAX_SPAWN
        or fail_rate >= CIRCUIT_FAIL_RATE
    )
    if circuit_tripped:
        # circuit_tripped.marker 파일 생성
        try:
            dir_ = os.path.dirname(circuit_marker_path) or "."
            os.makedirs(dir_, exist_ok=True)
            with open(circuit_marker_path, "w", encoding="utf-8") as f:
                f.write(ts_str + "\n")
        except Exception:
            pass

        # p0b_flag_path 파일이 존재하는 경우에만 'blocked' 쓰기
        if os.path.exists(p0b_flag_path):
            try:
                with open(p0b_flag_path, "w", encoding="utf-8") as f:
                    f.write("blocked")
            except Exception:
                pass
        # 회장보고 stub — 실제 발사 절대 0: marker/로그만
        return _record_and_return(_make_decision(TRIP, REASON_CIRCUIT_TRIP))

    # ── 게이트 전부 통과 → ALLOW (카운터 저장 성공해야 확정) ────────────────────
    # ALLOW 시에만 카운터 증가 — epoch 미리 저장(다음 호출 파싱 비용 제거).
    per_task[task_id]  = task_count + 1
    per_family[family] = family_count + 1
    events.append({"ts": ts_str, "epoch": now_ts, "task_id": task_id, "outcome": "allow"})

    counters["per_task"]   = per_task
    counters["per_family"] = per_family
    counters["events"]     = events

    # 저장 우선 — 저장 실패 시 count 미증가 → fail-closed BLOCK(무한 spawn 차단).
    # ALLOW 결정은 저장 성공 후에만 기록(저장 실패 시 dedup 오염 방지).
    try:
        _save_counters(counters_path, counters)
    except Exception:
        return _record_and_return(_make_decision(BLOCK, REASON_COUNTERS_SAVE_FAILED))

    decision_rec = _make_decision(ALLOW, REASON_ALLOW)
    return _record_and_return(decision_rec)


def _ts_to_epoch(ts_str: str) -> float:
    """ISO8601 UTC 문자열을 epoch(float)로 변환. 파싱 실패 시 0.0."""
    try:
        # '2026-06-24T12:34:56Z' 형식
        if ts_str.endswith("Z"):
            ts_str = ts_str[:-1] + "+00:00"
        dt = datetime.fromisoformat(ts_str)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.timestamp()
    except Exception:
        return 0.0