#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""scripts/anu_control_status.py — read-only ANU 자동화 조종석 (ACTIVE=false).

task-2777. 회장이 ANU 자동화 상태를 **한눈에 read-only로 관측**하고 다음에 무엇을
승인해도 되는지 판단할 수 있는 조종석. **자동화 실행이 아니다 — 순수 관측 도구.**

설계 단일소스:
  - memory/reports/control_panel_status_command_spec_260628.md   (§2 소스 매핑 · §3 level)
  - memory/reports/capability_matrix_update_plan_260628.md       (WIRED≠ACTIVE 분리)

★ read-only 절대 불변 (코드 차원):
  flag 생성/수정 0 · state 파일 수정 0 · evidence append 0 · driver 실행 0 ·
  fixture 생성 0 · callback 발사 0 · real fire 0 · systemd 조작 0 · cron 등록 0 ·
  archive 이동 0 · key 출력 0 · key rotation 0 · git checkout/switch/reset/clean 0.
  → 이 모듈 어디에도 open(..., 'w'/'a'/'x') · os.remove/rename/replace ·
    subprocess write/mutation 명령이 없다. subprocess 는 read-only 조회만:
    `git rev-parse/branch/status(--porcelain)` · `systemctl --user is-active/is-enabled`.

★ WIRED ≠ ACTIVE: gated 축(callback real launcher 등)은 "WIRED(gated, ACTIVE=false)"
  로만 표기한다. 어떤 축도 active 처럼 보이게 쓰지 않는다.

★ cron count — raw key 금지: cron-list 호출에는 key 가 필요하나, sealed loader 기반
  read-only helper 가 없으면 raw key 를 만들지 않고 cron_count="UNKNOWN_KEY_REQUIRED"
  로 둔다. raw key literal 을 code/test/log/output 어디에도 남기지 않는다.
"""
from __future__ import annotations

import argparse
import glob
import json
import math
import os
import subprocess
import sys
import time
from typing import Any, Dict, List, Optional, Tuple

# ── 상수: read-only 소스 상대경로 (spec §2 그대로) ────────────────────────────
REL_DRIVER_ENABLED = "memory/state/p0b_driver_enabled"
REL_EVENT_STRATEGY = "memory/state/p0b_event_strategy_enabled"
REL_CALLBACK_LAUNCH = "memory/state/p0b_callback_launch_enabled"
REL_ACTIVATION_EPOCH = "memory/state/p0b_activation_epoch"
REL_KILL = "memory/state/p0b_kill"
REL_DRIVER_RUNS = "memory/p0b_state/driver_runs.jsonl"
REL_CALLBACK_AUDIT = "memory/p0b_state/callback_launch_audit.jsonl"
REL_PROCESSED_LEDGER = "memory/events/p0b_processed_ledger.jsonl"  # spec §2: 부재→0
REL_LIVE_INBOX_GLOB = "memory/events/p0b_live_inbox/task-*.result.json"
REL_P0B_INBOX_GLOB = "memory/events/p0b_inbox/task-*.result.json"
REL_ARCHIVE_GLOB = "memory/reports/p0b_inbox_backlog_archive_260627/task-*.result.json"
REL_REPORTS_DIR = "memory/reports"

SYSTEMD_UNIT = "anu-pickup.path"

# systemd 안전 표기 — 실패/부재/DBus 불가/timeout/empty 는 'unavailable' 로만 표기.
SYSTEMD_STATUS_UNAVAILABLE = "unavailable"
SYSTEMD_WARN_MARKER = "SYSTEMD_STATUS_UNAVAILABLE"
# is-active / is-enabled 의 정상 인식 토큰(allowlist). exit code 가 nonzero(예:
# is-active inactive=3, is-enabled disabled=1)여도 stdout 토큰이 이 집합이면 신뢰한다.
_KNOWN_ACTIVE_TOKENS = frozenset({
    "active", "inactive", "activating", "deactivating",
    "reloading", "failed", "maintenance",
})
_KNOWN_ENABLED_TOKENS = frozenset({
    "enabled", "enabled-runtime", "disabled", "static", "masked",
    "masked-runtime", "alias", "indirect", "linked", "linked-runtime",
    "generated", "transient", "bad",
})

# activation_epoch tz 처리 warning 토큰 (naive/parse/future 안전 표기).
EPOCH_WARN_NAIVE = "UNKNOWN_NAIVE_DATETIME"
EPOCH_WARN_PARSE = "EPOCH_PARSE_UNAVAILABLE"
EPOCH_WARN_FUTURE = "EPOCH_IN_FUTURE"

# verdict / outcome 토큰 (grep 대체)
TOK_REAL_FIRED = "REAL_FIRED"
TOK_LIVE_PROCESSED = "LIVE_PROCESSED"
TOK_LIVE_PENDING = "LIVE_PENDING_OWNER_PROOF"

# level 라벨
LEVEL_L2_DONE_L3_BLOCK = "L2_DONE_L3_BLOCK_EVIDENCE_ACTIVE_FALSE"
LEVEL_L2_READY = "L0_L2_READY_ACTIVE_FALSE"
LEVEL_INDETERMINATE = "INDETERMINATE_REVIEW_REQUIRED"

# cron count 안전 표기 (raw key 미사용)
CRON_UNKNOWN_KEY_REQUIRED = "UNKNOWN_KEY_REQUIRED"
CRON_UNKNOWN_NO_KEY = "UNKNOWN_NO_KEY"
# sealed read-only cron helper 계약 모듈(실제 존재 시에만 사용; 부재면 UNKNOWN 유지).
CRON_HELPER_MODULE = "scripts.anu_cron_count_readonly"

# 정적 forbidden / pending / next (spec §4 + Framework) — read-only 안내용.
FORBIDDEN_ACTIONS = [
    "flag ON",
    "driver run",
    "real fire",
    "LIVE_PROCESSED execution",
    "callback launch",
    "systemd start/enable",
    "cron register",
    "archive move",
    "pending fixture move",
    "key rotation",
    "merge",
    "production activation",
]
NEXT_ALLOWED_ACTION = (
    "design-only (D: LIVE_PROCESSED surface-only preflight) or read-only audit"
)
PENDING_APPROVALS = [
    "read-only anu_control_status.py (task-2777 — MERGE_CANDIDATE, merge=ANU/회장)",
    "D: LIVE_PROCESSED surface-only preflight (design-only, 별도 승인 후 관측)",
]

# capability matrix gated 축 (capability_matrix_update_plan_260628.md §1) — 전부 ACTIVE=false.
WIRED_GATED_AXES = [
    "limited activation bound gate(kill/N/T)",
    "live_inbox evidence wiring(write_evidence=True)",
    "owner-proof gate(sealed ANU+freshness)",
    "event_strategy live path(scan_live_inbox_once)",
    "normal callback real launcher",
    "limited_runtime 신호",
]
WIRED_GATED_LABEL = "WIRED(gated, ACTIVE=false)"


# ── 저수준 read-only helper (write 0) ─────────────────────────────────────────
def _read_text(path: str) -> Optional[str]:
    """파일 내용을 strip 하여 반환. 부재/오류 → None (crash 0, write 0)."""
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            return f.read().strip()
    except OSError:
        return None


def _exists(path: str) -> bool:
    return os.path.exists(path)


def _count_token_lines(path: str, token: str) -> int:
    """파일에서 token 을 포함하는 줄 수(grep -c 대체). 부재/오류 → 0."""
    try:
        n = 0
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                if token in line:
                    n += 1
        return n
    except OSError:
        return 0


def _last_json_line(path: str) -> Optional[Dict[str, Any]]:
    """jsonl 마지막 비어있지 않은 줄을 dict 로 파싱. 부재/파싱실패 → None."""
    last = None
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as f:
            for line in f:
                s = line.strip()
                if s:
                    last = s
    except OSError:
        return None
    if last is None:
        return None
    try:
        obj = json.loads(last)
        return obj if isinstance(obj, dict) else None
    except (ValueError, TypeError):
        return None


def _glob_count(pattern: str) -> int:
    try:
        return len(glob.glob(pattern))
    except OSError:
        return 0


def _quarantine_count(reports_root: str) -> int:
    """find memory/reports -path '*pending_fixture_quarantine*' -name '*.result.json' 대체."""
    n = 0
    try:
        for dirpath, _, filenames in os.walk(reports_root):
            if "pending_fixture_quarantine" not in dirpath:
                continue
            for fn in filenames:
                if fn.endswith(".result.json"):
                    n += 1
    except OSError:
        return n
    return n


def _git(root: str, args: List[str]) -> Optional[str]:
    """read-only git 명령만 실행(rev-parse/branch/status). 실패 → None."""
    try:
        proc = subprocess.run(
            ["git", "-C", root, *args],
            shell=False,
            check=False,
            capture_output=True,
            text=True,
            timeout=15,
        )
    except (OSError, subprocess.SubprocessError):
        return None
    if proc.returncode != 0:
        return None
    return proc.stdout.strip()


def _classify_systemctl_stderr(stderr: str) -> str:
    """raw stderr 를 짧은 분류 토큰으로 요약(원문 길게 출력 0 · key/env 유출 0).

    DBus/session bus 불가 · command missing · unit 부재 · 권한 거부 등을 고정
    토큰으로만 분류한다. stderr 원문을 status 필드/출력에 절대 넣지 않는다.
    """
    low = (stderr or "").lower()
    if not low.strip():
        return "no_output"
    if "failed to connect to bus" in low or "d-bus" in low or "dbus" in low \
            or "no medium found" in low or "connection refused" in low:
        return "dbus_unavailable"
    if "command not found" in low or "no such file or directory" in low \
            or "not found" in low:
        return "command_missing"
    if "no such unit" in low or "could not be found" in low \
            or "not loaded" in low or "not-found" in low:
        return "unit_not_found"
    if "access denied" in low or "permission" in low or "not authorized" in low:
        return "permission_denied"
    return "systemctl_error"


def _systemd(unit: str) -> Tuple[str, str, List[str]]:
    """systemctl --user is-active/is-enabled (read-only · 안전 처리).

    nonzero+미인식 stdout · stderr 출력 · DBus/session bus unavailable ·
    command missing · timeout · stdout empty 는 모두 'unavailable' 로 **안전 표기**
    하고 원인을 분류 요약해 warnings(SYSTEMD_STATUS_UNAVAILABLE marker)로 돌려준다.

    - stdout 첫 토큰이 정상 status allowlist 면 신뢰한다(is-active inactive=exit3,
      is-enabled disabled=exit1 처럼 nonzero 가 정상인 케이스 포함).
    - 미인식/empty/예외 → 'unavailable' (실패를 inactive/disabled 로 확정 표기 0).
    - raw stderr 원문/길게 출력 0 · traceback 0 · key/env 유출 0 · crash 0.

    반환: (is_active, is_enabled, warnings)
    """
    warnings: List[str] = []

    def _probe(verb: str, known: "frozenset[str]") -> str:
        try:
            proc = subprocess.run(
                ["systemctl", "--user", verb, unit],
                shell=False,
                check=False,
                capture_output=True,
                text=True,
                timeout=15,
            )
        except FileNotFoundError:
            warnings.append(f"{SYSTEMD_WARN_MARKER}: {verb} reason=command_missing")
            return SYSTEMD_STATUS_UNAVAILABLE
        except subprocess.TimeoutExpired:
            warnings.append(f"{SYSTEMD_WARN_MARKER}: {verb} reason=timeout")
            return SYSTEMD_STATUS_UNAVAILABLE
        except (OSError, subprocess.SubprocessError):
            warnings.append(f"{SYSTEMD_WARN_MARKER}: {verb} reason=subprocess_error")
            return SYSTEMD_STATUS_UNAVAILABLE

        stdout_lines = [ln.strip() for ln in (proc.stdout or "").splitlines() if ln.strip()]
        token = stdout_lines[0].lower() if stdout_lines else ""
        if token in known:
            # 정상 인식된 status 토큰 — exit code 가 nonzero 여도 신뢰(systemctl 규약).
            return token
        # 미인식/empty → 실패로 간주(확정 표시 금지). 원인 분류(raw 길게 X).
        reason = _classify_systemctl_stderr(proc.stderr or "")
        warnings.append(
            f"{SYSTEMD_WARN_MARKER}: {verb} reason={reason} rc={proc.returncode}"
        )
        return SYSTEMD_STATUS_UNAVAILABLE

    active = _probe("is-active", _KNOWN_ACTIVE_TOKENS)
    enabled = _probe("is-enabled", _KNOWN_ENABLED_TOKENS)
    return active, enabled, warnings


# ── flag / epoch ─────────────────────────────────────────────────────────────
def _epoch_age_seconds(epoch_raw: Optional[str]) -> Tuple[Optional[int], Optional[str]]:
    """activation epoch(unix sec or ISO8601) → (age_seconds, warning_token).

    tz 안전 규칙(KST 환경에서도 안정):
      - 순수 숫자 → unix epoch 초로 보고 `time.time() - epoch` (tz 독립).
      - ISO8601 + tz offset(aware) → tz-aware 비교(`dt.timestamp()` 가 정확).
      - ISO8601 naive(offset 없음) → KST/local 임의 해석 **금지** → age=None +
        UNKNOWN_NAIVE_DATETIME warning (datetime.timestamp() 의 local 가정 회피).
      - 파싱 실패/비유한값 → (None, EPOCH_PARSE_UNAVAILABLE).
      - 극단 날짜(예: 0001-01-01, platform time_t 범위 밖)에서 timestamp() 가
        OverflowError/OSError → (None, EPOCH_PARSE_UNAVAILABLE), crash 0.
      - 음수 age(미래 epoch) → 0 으로 clamp + EPOCH_IN_FUTURE warning(crash 0).
    """
    if not epoch_raw:
        return None, None
    raw = epoch_raw.strip()
    if not raw:
        return None, None

    # 1) unix seconds (정수/실수) — tz 독립.
    try:
        epoch = float(raw)
    except ValueError:
        epoch = None
    if epoch is not None:
        if not math.isfinite(epoch):
            return None, EPOCH_WARN_PARSE
        delta = time.time() - epoch
        if delta < 0:
            return 0, EPOCH_WARN_FUTURE
        return int(delta), None

    # 2) ISO8601.
    try:
        from datetime import datetime

        dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except (ValueError, TypeError):
        return None, EPOCH_WARN_PARSE
    try:
        if dt.tzinfo is None:
            # naive — local(KST) 임의 해석 금지. UTC 강제 가정도 하지 않고 UNKNOWN 표기.
            return None, EPOCH_WARN_NAIVE
        ts = dt.timestamp()
    except (OverflowError, OSError):
        # 극단 날짜(예: 0001-01-01, platform time_t 범위 밖)에서 timestamp() 가
        # OverflowError/OSError 를 던질 수 있다 → crash 0 · raw 유출 0 으로 안전 처리.
        return None, EPOCH_WARN_PARSE
    delta = time.time() - ts
    if delta < 0:
        return 0, EPOCH_WARN_FUTURE
    return int(delta), None


def _human_age(seconds: Optional[int]) -> str:
    if seconds is None:
        return "none"
    if seconds < 60:
        return f"{seconds}s"
    if seconds < 3600:
        return f"{seconds // 60}m"
    if seconds < 86400:
        return f"{seconds // 3600}h"
    return f"{seconds // 86400}d"


# ── cron count (raw key 금지) ─────────────────────────────────────────────────
def _cron_count() -> Tuple[str, str]:
    """cron 등록 수 (raw key 금지). 반환: (value, reason).

    cron-list 조회에는 key 가 필요하다. raw key literal 을 만들지 않으며, sealed
    loader 기반 read-only helper(`CRON_HELPER_MODULE.cron_count_readonly`)가
    **실제로 존재하고** raw key 출력/기록 없이 cron 수(int)만 돌려줄 때에만 사용한다.

    helper 부재 · 계약 불명확 · 새 key path 신설 필요 → **사용 금지**,
    `UNKNOWN_KEY_REQUIRED` 유지하고 reason 에 왜 UNKNOWN 인지 명시한다.
    cron-list 를 직접 호출하지 않는다(등록/수정 0, 새 key path 0).
    """
    try:
        import importlib

        helper = importlib.import_module(CRON_HELPER_MODULE)
    except Exception:  # noqa: BLE001 — helper 부재 → 안전 UNKNOWN
        return (
            CRON_UNKNOWN_KEY_REQUIRED,
            "no sealed read-only cron helper module present; "
            "raw key required and forbidden",
        )
    fn = getattr(helper, "cron_count_readonly", None)
    if not callable(fn):
        return (
            CRON_UNKNOWN_KEY_REQUIRED,
            "cron helper present but lacks sealed read-only "
            "cron_count_readonly() contract",
        )
    try:
        result = fn()  # 계약: raw key 출력/기록 0, cron 수(int>=0)만 반환.
    except Exception:  # noqa: BLE001 — helper 실패 → 안전 UNKNOWN
        return (
            CRON_UNKNOWN_KEY_REQUIRED,
            "sealed cron helper raised during read-only query",
        )
    if isinstance(result, int) and not isinstance(result, bool) and result >= 0:
        return (str(result), "sealed read-only cron helper (no raw key)")
    return (
        CRON_UNKNOWN_KEY_REQUIRED,
        "sealed cron helper returned non-count result",
    )


# ── automation level 판정 (read-only 추론, spec §3 + task) ────────────────────
def compute_level(
    *,
    active: bool,
    flags: Dict[str, Any],
    counts: Dict[str, int],
) -> str:
    """현재 flag 스냅샷 + counts 기반 read-only level 판정.

    - 전 flag OFF + LIVE_PENDING≥1 + LIVE_PROCESSED 0 + ACTIVE=false
        → L2_DONE_L3_BLOCK_EVIDENCE_ACTIVE_FALSE.
    - LIVE_PROCESSED 0 이면 L3 PASS 로 올리지 않는다.
    - 전 flag OFF + pending 0 + processed 0 → L0/L2-ready (baseline).
    - 모호 → INDETERMINATE_REVIEW_REQUIRED.
    """
    driver_off = (flags.get("driver") != "enabled")
    es_off = not flags.get("event_strategy")
    cb_off = not flags.get("callback")
    epoch_off = not flags.get("epoch_present")
    kill_off = not flags.get("kill")
    all_off = driver_off and es_off and cb_off and epoch_off and kill_off

    live_pending = int(counts.get("LIVE_PENDING_OWNER_PROOF", 0))
    live_processed = int(counts.get("LIVE_PROCESSED", 0))

    # production activation 은 별도 승인 전 항상 false. active=True 면 read-only 로
    # 안전판정 불가 → 검토 필요.
    if active:
        return LEVEL_INDETERMINATE
    if all_off and live_pending >= 1 and live_processed == 0:
        return LEVEL_L2_DONE_L3_BLOCK
    if all_off and live_pending == 0 and live_processed == 0:
        return LEVEL_L2_READY
    return LEVEL_INDETERMINATE


# ── 수집 (collect) ────────────────────────────────────────────────────────────
def collect(root: str) -> Dict[str, Any]:
    """root(repo) 기준으로 모든 항목을 read-only 수집해 dict 반환."""
    def p(rel: str) -> str:
        return os.path.join(root, rel)

    # flags
    driver_text = _read_text(p(REL_DRIVER_ENABLED))
    driver_state = driver_text if driver_text else "disabled"  # absent→disabled
    active = (driver_text == "enabled")
    event_strategy = _exists(p(REL_EVENT_STRATEGY))
    callback = _exists(p(REL_CALLBACK_LAUNCH))
    epoch_raw = _read_text(p(REL_ACTIVATION_EPOCH))
    epoch_present = epoch_raw is not None
    epoch_age, epoch_warn = _epoch_age_seconds(epoch_raw)
    kill = _exists(p(REL_KILL))

    flags = {
        "driver": driver_state,
        "event_strategy": bool(event_strategy),
        "callback": bool(callback),
        "epoch_present": epoch_present,
        "epoch_age_seconds": epoch_age,
        "kill": bool(kill),
    }

    # last verdict / evidence (driver_runs.jsonl)
    last_obj = _last_json_line(p(REL_DRIVER_RUNS))
    last_verdict = (last_obj or {}).get("verdict") if last_obj else None
    # last evidence: 안전 요약(raw key 미포함 필드만). driver_runs 에는 key 없음.
    last_evidence: Optional[Dict[str, Any]] = None
    if last_obj:
        last_evidence = {
            "ts": last_obj.get("ts"),
            "verdict": last_obj.get("verdict"),
            "driver": last_obj.get("driver"),
            "activation": last_obj.get("activation"),
            "owner_key_class": last_obj.get("owner_key_class"),
        }

    # counts
    counts = {
        TOK_REAL_FIRED: _count_token_lines(p(REL_CALLBACK_AUDIT), TOK_REAL_FIRED),
        TOK_LIVE_PROCESSED: _count_token_lines(p(REL_PROCESSED_LEDGER), TOK_LIVE_PROCESSED),
        TOK_LIVE_PENDING: _count_token_lines(p(REL_DRIVER_RUNS), TOK_LIVE_PENDING),
        "live_inbox": _glob_count(p(REL_LIVE_INBOX_GLOB)),
        "p0b_inbox": _glob_count(p(REL_P0B_INBOX_GLOB)),
        "archive": _glob_count(p(REL_ARCHIVE_GLOB)),
        "pending_fixture_quarantine": _quarantine_count(p(REL_REPORTS_DIR)),
    }

    # inbox / archive (구조화)
    inbox = {"live_inbox": counts["live_inbox"], "p0b_inbox": counts["p0b_inbox"]}
    archive = {
        "archive": counts["archive"],
        "pending_fixture_quarantine": counts["pending_fixture_quarantine"],
    }

    # systemd (실패 안전 처리 — unavailable + 분류 warning, raw stderr 오염 0)
    sysd_active, sysd_enabled, sysd_warnings = _systemd(SYSTEMD_UNIT)
    systemd = {
        "unit": SYSTEMD_UNIT,
        "is_active": sysd_active,
        "is_enabled": sysd_enabled,
    }

    # cron (raw key 금지 → 안전 helper 있을 때만 사용, 아니면 UNKNOWN + reason)
    cron, cron_reason = _cron_count()

    # git (read-only)
    branch = _git(root, ["branch", "--show-current"])
    head = _git(root, ["rev-parse", "HEAD"])
    origin_main = _git(root, ["rev-parse", "origin/main"])
    porcelain = _git(root, ["status", "--porcelain=v1"])
    tracked_dirty = _tracked_dirty_count(porcelain)
    git = {
        "branch": branch if branch else "unknown",
        "head": head if head else "unknown",
        "origin_main": origin_main if origin_main else "unknown",
        "synced": bool(head and origin_main and head == origin_main),
        "tracked_dirty": tracked_dirty,
    }

    # level
    automation_level = compute_level(active=active, flags=flags, counts=counts)

    # warnings (동적)
    warnings: List[str] = []
    if active:
        warnings.append("ACTIVE=true observed — expected false (review required).")
    if flags["kill"]:
        warnings.append("p0b_kill present — kill switch engaged.")
    if git["tracked_dirty"] and isinstance(git["tracked_dirty"], int) and git["tracked_dirty"] > 0:
        warnings.append(f"tracked dirty files: {git['tracked_dirty']}.")
    if not git["synced"]:
        warnings.append("HEAD != origin/main (or git unavailable).")
    if automation_level == LEVEL_INDETERMINATE:
        warnings.append("automation level indeterminate — manual review required.")
    # systemd 실패 분류 warning (raw stderr 미포함 — 분류 토큰만)
    warnings.extend(sysd_warnings)
    # activation_epoch tz 안전 warning (naive/parse/future)
    if epoch_warn:
        warnings.append(f"activation_epoch: {epoch_warn}.")
    if str(cron).startswith("UNKNOWN"):
        warnings.append(f"cron count unavailable — {cron_reason}.")
    if counts[TOK_LIVE_PROCESSED] == 0 and counts[TOK_LIVE_PENDING] >= 1:
        warnings.append(
            "L3 owner-proof PASS (LIVE_PROCESSED) not reached — evidence/block layer only."
        )

    return {
        "schema": "anu_control_status.v1",
        "generated_at": _now_iso(),
        "root": root,
        "automation_level": automation_level,
        "active": active,
        "flags": flags,
        "counts": counts,
        "last_verdict": last_verdict,
        "last_evidence": last_evidence,
        "inbox": inbox,
        "archive": archive,
        "systemd": systemd,
        "cron": cron,
        "cron_reason": cron_reason,
        "git": git,
        "wired_gated_axes": {axis: WIRED_GATED_LABEL for axis in WIRED_GATED_AXES},
        "pending_approvals": list(PENDING_APPROVALS),
        "next_allowed_action": NEXT_ALLOWED_ACTION,
        "forbidden_actions": list(FORBIDDEN_ACTIONS),
        "warnings": warnings,
    }


def _tracked_dirty_count(porcelain: Optional[str]) -> Any:
    """git status --porcelain=v1 출력에서 tracked dirty(미스테이지 M/D) 줄 수.

    spec §2: `git status --porcelain=v1 | grep -cE '^ (M|D)'`.
    git 미사용/불가(tmp root 등) → 'unknown'.
    """
    if porcelain is None:
        return "unknown"
    n = 0
    for line in porcelain.splitlines():
        if len(line) >= 2 and line[0] == " " and line[1] in ("M", "D"):
            n += 1
    return n


def _now_iso() -> str:
    try:
        from datetime import datetime

        return datetime.now().astimezone().isoformat(timespec="seconds")
    except Exception:  # noqa: BLE001
        return ""


# ── 출력 포매터 ───────────────────────────────────────────────────────────────
def format_json(data: Dict[str, Any]) -> str:
    return json.dumps(data, ensure_ascii=False, indent=2, sort_keys=True)


def _yn(b: Any) -> str:
    return "true" if b else "false"


def format_human(data: Dict[str, Any]) -> str:
    f = data["flags"]
    c = data["counts"]
    g = data["git"]
    s = data["systemd"]
    lines: List[str] = []
    add = lines.append

    add("┌─ ANU 자동화 조종석 (read-only · 관측 전용) ──────────────────────────")
    add(f"│ generated_at : {data.get('generated_at', '')}")
    add(f"│ automation level : {data['automation_level']}")
    add(f"│ ACTIVE : {_yn(data['active'])}   (production activation: false, 별도 승인 전 항상 false)")
    add("├─ flags ─────────────────────────────────────────────────────────────")
    add(f"│ p0b_driver_enabled        : {f['driver']}")
    add(f"│ p0b_event_strategy_enabled: {_yn(f['event_strategy'])}")
    add(f"│ p0b_callback_launch_enabled: {_yn(f['callback'])}")
    add(f"│ p0b_activation_epoch      : present={_yn(f['epoch_present'])} age={_human_age(f['epoch_age_seconds'])}")
    add(f"│ p0b_kill                  : {_yn(f['kill'])}")
    add("├─ verdict / evidence ────────────────────────────────────────────────")
    add(f"│ last verdict  : {data['last_verdict']}")
    ev = data.get("last_evidence") or {}
    add(f"│ last evidence : ts={ev.get('ts')} verdict={ev.get('verdict')} "
        f"driver={ev.get('driver')} activation={ev.get('activation')}")
    add("├─ counts ────────────────────────────────────────────────────────────")
    add(f"│ REAL_FIRED total            : {c['REAL_FIRED']}")
    add(f"│ LIVE_PROCESSED total        : {c['LIVE_PROCESSED']}")
    add(f"│ LIVE_PENDING_OWNER_PROOF tot: {c['LIVE_PENDING_OWNER_PROOF']}")
    add(f"│ live_inbox count            : {c['live_inbox']}")
    add(f"│ p0b_inbox count             : {c['p0b_inbox']}")
    add(f"│ archive count               : {c['archive']}")
    add(f"│ pending fixture quarantine  : {c['pending_fixture_quarantine']}")
    add("├─ systemd / cron ────────────────────────────────────────────────────")
    add(f"│ systemd {s['unit']} : is-active={s['is_active']} is-enabled={s['is_enabled']}")
    add(f"│ cron count : {data['cron']}  (raw key 미사용 — {data.get('cron_reason', '')})")
    add("├─ git ───────────────────────────────────────────────────────────────")
    add(f"│ branch={g['branch']} HEAD={g['head']}")
    add(f"│ origin/main={g['origin_main']} synced={_yn(g['synced'])} tracked_dirty={g['tracked_dirty']}")
    add("├─ WIRED ≠ ACTIVE (gated 축은 WIRED·ACTIVE=false) ─────────────────────")
    for axis, label in data["wired_gated_axes"].items():
        add(f"│ - {axis}: {label}")
    add("├─ approvals / actions ───────────────────────────────────────────────")
    for pa in data["pending_approvals"]:
        add(f"│ pending approval : {pa}")
    add(f"│ next allowed action : {data['next_allowed_action']}")
    add(f"│ forbidden next actions : {', '.join(data['forbidden_actions'])}")
    add("├─ warnings ──────────────────────────────────────────────────────────")
    if data["warnings"]:
        for w in data["warnings"]:
            add(f"│ ⚠ {w}")
    else:
        add("│ (none)")
    add("└──────────────────────────────────────────────────────────────────────")
    return "\n".join(lines)


# ── 진입점 ────────────────────────────────────────────────────────────────────
def _default_root() -> str:
    """이 스크립트(scripts/anu_control_status.py) 기준 repo root."""
    return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


def build_status(root: Optional[str] = None) -> Dict[str, Any]:
    """공개 API: read-only 상태 dict 반환 (write 0)."""
    return collect(root or _default_root())


def main(argv: Optional[List[str]] = None) -> int:
    ap = argparse.ArgumentParser(
        prog="anu_control_status",
        description="read-only ANU 자동화 조종석 (관측 전용 · write 0).",
    )
    ap.add_argument("--json", action="store_true", help="JSON 출력")
    ap.add_argument(
        "--root",
        default=None,
        help="repo root override (테스트용 격리 tmp root). 기본=스크립트 기준 repo root.",
    )
    args = ap.parse_args(argv)

    data = build_status(args.root)
    if args.json:
        sys.stdout.write(format_json(data) + "\n")
    else:
        sys.stdout.write(format_human(data) + "\n")
    return 0


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