"""Unified background-task ledger (time-series + retention).

Single append-only JSONL log of every Mugi background run: heartbeat,
idle_tick, cron, curiosity, dreaming, active_memory, subagent, webhook,
delivery, btw. Plaintext (matches ``memory/agent_turns.log`` precedent),
mode 0o600, never backed up via the encrypted chat-backup path.

Public surface:

    record_run(kind, task_name, ..., **meta) -> run_id
    recent(limit=200, kind=None, status=None, since_ts=None) -> list[dict]
    prune(retention_days=7, max_entries=20_000) -> int
    LEDGER_PATH (constant)

See plan §3 ("Unified background-task ledger").
"""

from __future__ import annotations

import json
import logging
import os
import threading
import time
import uuid
from datetime import datetime, timedelta, timezone
from typing import Any, Iterable

try:
    import fcntl
    _HAS_FLOCK = True
except ImportError:
    _HAS_FLOCK = False

from utils.redaction import cap_str, redact_value

_log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Schema constants & file paths
# ---------------------------------------------------------------------------

SCHEMA_VERSION = 1

# Valid kinds — extend with care; tests assert membership.
KIND_HEARTBEAT = "heartbeat"
KIND_IDLE_TICK = "idle_tick"
KIND_CRON = "cron"
KIND_CURIOSITY = "curiosity"
KIND_DREAMING = "dreaming"
KIND_ACTIVE_MEMORY = "active_memory"
KIND_SUBAGENT = "subagent"
KIND_WEBHOOK = "webhook"
KIND_DELIVERY = "delivery"
KIND_BTW = "btw"
KIND_TIMER = "timer"
KIND_FILE_EVENT = "file_event"
KIND_CALENDAR = "calendar"
KIND_OUTBOUND_WEBHOOK = "outbound_webhook"
KIND_REFLECTION = "reflection"
KIND_BROWSER = "browser"
_VALID_KINDS = frozenset({
    KIND_HEARTBEAT, KIND_IDLE_TICK, KIND_CRON, KIND_CURIOSITY,
    KIND_DREAMING, KIND_ACTIVE_MEMORY, KIND_SUBAGENT, KIND_WEBHOOK,
    KIND_DELIVERY, KIND_BTW,
    KIND_TIMER, KIND_FILE_EVENT, KIND_CALENDAR, KIND_OUTBOUND_WEBHOOK, KIND_REFLECTION,
    KIND_BROWSER,
})

STATUS_OK = "ok"
STATUS_SKIPPED = "skipped"
STATUS_ERROR = "error"
STATUS_TIMEOUT = "timeout"
STATUS_CANCELLED = "cancelled"
_VALID_STATUSES = frozenset({STATUS_OK, STATUS_SKIPPED, STATUS_ERROR, STATUS_TIMEOUT, STATUS_CANCELLED})

# Field caps (per plan §3 review addendum).
_MAX_TASK_NAME = 120
_MAX_THREAD_ID = 64
_MAX_SKIP_REASON = 120
_MAX_ERROR_SUMMARY = 400
_MAX_META_BYTES = 2 * 1024
_MAX_ROW_BYTES = 4 * 1024

# Prune triggers.
_PRUNE_FILE_BYTES = 4 * 1024 * 1024
_PRUNE_INTERVAL_S = 24 * 3600
_PRUNE_CHECK_EVERY_N_WRITES = 50
_DEFAULT_RETENTION_DAYS = 7
_DEFAULT_MAX_ENTRIES = 20_000

# Read caps.
MAX_RECENT_LIMIT = 500

_lock = threading.Lock()
_write_counter = 0


def _ledger_dir() -> str:
    from persistence import MUGI_DIR
    return os.path.expanduser(MUGI_DIR)


def _ledger_path() -> str:
    return os.path.join(_ledger_dir(), "task_ledger.jsonl")


def _meta_path() -> str:
    return os.path.join(_ledger_dir(), "task_ledger_meta.json")


# Backwards-compat constant for tests. Resolves at access time via _ledger_path().
class _PathProxy:
    def __str__(self) -> str:  # pragma: no cover - thin shim
        return _ledger_path()

    def __fspath__(self) -> str:
        return _ledger_path()


LEDGER_PATH = _PathProxy()


# ---------------------------------------------------------------------------
# Time helpers
# ---------------------------------------------------------------------------

def _utcnow_iso() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.") + f"{datetime.now(timezone.utc).microsecond // 1000:03d}Z"


def _parse_ts(ts: str) -> datetime | None:
    if not ts:
        return None
    try:
        s = ts.replace("Z", "+00:00") if ts.endswith("Z") else ts
        dt = datetime.fromisoformat(s)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=timezone.utc)
        return dt
    except Exception:
        return None


# ---------------------------------------------------------------------------
# Append (record_run)
# ---------------------------------------------------------------------------

def _flock_append(path: str, line: str) -> None:
    if _HAS_FLOCK:
        lock_path = path + ".lock"
        fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
        try:
            fcntl.flock(fd, fcntl.LOCK_EX)
            with open(path, "a", encoding="utf-8") as f:
                f.write(line)
        finally:
            fcntl.flock(fd, fcntl.LOCK_UN)
            os.close(fd)
    else:
        with open(path, "a", encoding="utf-8") as f:
            f.write(line)


def _safe_meta(meta: dict[str, Any] | None) -> dict[str, Any]:
    if not meta:
        return {}
    redacted = redact_value(meta)
    if not isinstance(redacted, dict):
        return {}
    encoded = json.dumps(redacted, ensure_ascii=True, default=str)
    if len(encoded.encode("utf-8")) <= _MAX_META_BYTES:
        return redacted
    out: dict[str, Any] = {"_truncated": True}
    budget = _MAX_META_BYTES - len(json.dumps(out, ensure_ascii=True).encode("utf-8")) - 32
    for k, v in redacted.items():
        candidate = {**out, str(k)[:64]: v}
        try:
            if len(json.dumps(candidate, ensure_ascii=True, default=str).encode("utf-8")) <= _MAX_META_BYTES:
                out = candidate
                continue
        except Exception:
            continue
        sval = cap_str(v, max(32, budget))
        out[str(k)[:64]] = sval
        break
    return out


def record_run(
    kind: str,
    task_name: str,
    *,
    thread_id: str | None = None,
    parent_run_id: str | None = None,
    status: str = STATUS_OK,
    skip_reason: str | None = None,
    duration_ms: int | float = 0,
    error_summary: str | None = None,
    ts_start: str | None = None,
    ts_end: str | None = None,
    meta: dict[str, Any] | None = None,
) -> str:
    """Append one run record. Returns the run_id (always a string).

    All string fields are redacted + capped. Meta is recursively redacted
    and capped at ~2 KB. Total row hard-capped at 4 KB. Best-effort: any
    write failure is logged and swallowed (callers should never fail
    because the ledger is unavailable).
    """
    run_id = uuid.uuid4().hex
    if kind not in _VALID_KINDS:
        _log.warning("task_ledger: unknown kind %r; coercing to subagent", kind)
        kind = KIND_SUBAGENT
    if status not in _VALID_STATUSES:
        status = STATUS_OK

    now = _utcnow_iso()
    row = {
        "schema_version": SCHEMA_VERSION,
        "run_id": run_id,
        "ts_start": ts_start or now,
        "ts_end": ts_end or now,
        "kind": kind,
        "task_name": cap_str(task_name, _MAX_TASK_NAME),
        "thread_id": cap_str(thread_id, _MAX_THREAD_ID) if thread_id else None,
        "parent_run_id": cap_str(parent_run_id, 64) if parent_run_id else None,
        "status": status,
        "skip_reason": cap_str(skip_reason, _MAX_SKIP_REASON) if skip_reason else None,
        "duration_ms": int(max(0, float(duration_ms or 0))),
        "error_summary": cap_str(error_summary, _MAX_ERROR_SUMMARY) if error_summary else None,
        "meta": _safe_meta(meta),
    }

    encoded = json.dumps(row, ensure_ascii=True, default=str)
    if len(encoded.encode("utf-8")) > _MAX_ROW_BYTES:
        row["meta"] = {"_truncated": True}
        encoded = json.dumps(row, ensure_ascii=True, default=str)
        if len(encoded.encode("utf-8")) > _MAX_ROW_BYTES:
            row["error_summary"] = (row.get("error_summary") or "")[:120] + "[TRUNCATED]"
            encoded = json.dumps(row, ensure_ascii=True, default=str)

    try:
        os.makedirs(_ledger_dir(), exist_ok=True, mode=0o700)
    except OSError:
        pass

    path = _ledger_path()
    try:
        with _lock:
            _flock_append(path, encoded + "\n")
        try:
            os.chmod(path, 0o600)
        except OSError:
            pass
    except Exception as e:
        _log.debug("task_ledger append failed: %s", e)
        return run_id

    global _write_counter
    _write_counter += 1
    if _write_counter % _PRUNE_CHECK_EVERY_N_WRITES == 0:
        try:
            _maybe_prune()
        except Exception:
            _log.debug("task_ledger maybe_prune failed", exc_info=True)
    return run_id


# ---------------------------------------------------------------------------
# Read (recent)
# ---------------------------------------------------------------------------

def _iter_lines_reverse(path: str, limit: int) -> Iterable[str]:
    """Best-effort newest-first line iterator. Loads file then reverses for simplicity."""
    if not os.path.isfile(path):
        return []
    try:
        with open(path, "r", encoding="utf-8") as f:
            lines = f.readlines()
    except OSError:
        return []
    return reversed(lines[-(limit * 4):]) if limit > 0 else reversed(lines)


def recent(
    limit: int = 200,
    *,
    kind: str | None = None,
    status: str | None = None,
    since_ts: str | None = None,
) -> list[dict[str, Any]]:
    """Return up to *limit* most recent rows (newest first), filtered."""
    limit = max(1, min(int(limit or 200), MAX_RECENT_LIMIT))
    since_dt = _parse_ts(since_ts) if since_ts else None
    out: list[dict[str, Any]] = []
    for raw in _iter_lines_reverse(_ledger_path(), limit):
        if len(out) >= limit:
            break
        line = (raw or "").strip()
        if not line:
            continue
        try:
            row = json.loads(line)
        except Exception:
            continue
        if not isinstance(row, dict):
            continue
        if kind and row.get("kind") != kind:
            continue
        if status and row.get("status") != status:
            continue
        if since_dt:
            row_dt = _parse_ts(str(row.get("ts_start") or ""))
            if row_dt is None or row_dt < since_dt:
                continue
        # Strip raw webhook bodies if any caller stuffed them in meta.raw.
        meta = row.get("meta") or {}
        if isinstance(meta, dict) and "raw" in meta:
            meta = {**meta, "raw": "[REDACTED]"}
            row["meta"] = meta
        out.append(row)
    return out


# ---------------------------------------------------------------------------
# Prune
# ---------------------------------------------------------------------------

def _read_meta() -> dict[str, Any]:
    p = _meta_path()
    if not os.path.isfile(p):
        return {}
    try:
        with open(p, "r", encoding="utf-8") as f:
            data = json.load(f)
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _write_meta(meta: dict[str, Any]) -> None:
    try:
        from persistence import secure_atomic_write
        secure_atomic_write(_meta_path(), json.dumps(meta, ensure_ascii=True))
        os.chmod(_meta_path(), 0o600)
    except Exception:
        _log.debug("task_ledger meta write failed", exc_info=True)


def _maybe_prune() -> None:
    """Prune when (a) file > 4MB, OR (b) last prune > 24h ago, OR (c) explicit force."""
    path = _ledger_path()
    if not os.path.isfile(path):
        return
    try:
        size = os.path.getsize(path)
    except OSError:
        return
    meta = _read_meta()
    last_prune_ts = meta.get("last_prune_ts")
    last_dt = _parse_ts(str(last_prune_ts) if last_prune_ts else "")
    age_s = (datetime.now(timezone.utc) - last_dt).total_seconds() if last_dt else float("inf")
    if size < _PRUNE_FILE_BYTES and age_s < _PRUNE_INTERVAL_S:
        return
    prune()


def prune(
    retention_days: int = _DEFAULT_RETENTION_DAYS,
    max_entries: int = _DEFAULT_MAX_ENTRIES,
) -> int:
    """Atomically rewrite the ledger keeping recent rows. Returns count kept."""
    path = _ledger_path()
    if not os.path.isfile(path):
        return 0
    cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, int(retention_days)))
    future_cutoff = datetime.now(timezone.utc) + timedelta(hours=1)
    keep: list[str] = []
    try:
        if _HAS_FLOCK:
            lock_path = path + ".lock"
            fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
            try:
                if fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) == -1:
                    return 0
            except BlockingIOError:
                os.close(fd)
                return 0
            try:
                with open(path, "r", encoding="utf-8") as f:
                    for raw in f:
                        line = raw.strip()
                        if not line:
                            continue
                        try:
                            row = json.loads(line)
                        except Exception:
                            continue
                        if not isinstance(row, dict):
                            continue
                        ts = _parse_ts(str(row.get("ts_start") or ""))
                        if ts is None:
                            continue
                        if ts > future_cutoff:
                            continue
                        if ts < cutoff:
                            continue
                        keep.append(line)
                if len(keep) > max_entries:
                    keep = keep[-max_entries:]
                tmp_path = path + ".tmp"
                with open(tmp_path, "w", encoding="utf-8") as f:
                    if keep:
                        f.write("\n".join(keep) + "\n")
                os.replace(tmp_path, path)
                try:
                    os.chmod(path, 0o600)
                except OSError:
                    pass
            finally:
                fcntl.flock(fd, fcntl.LOCK_UN)
                os.close(fd)
        else:
            with open(path, "r", encoding="utf-8") as f:
                for raw in f:
                    line = raw.strip()
                    if not line:
                        continue
                    try:
                        row = json.loads(line)
                    except Exception:
                        continue
                    ts = _parse_ts(str(row.get("ts_start") or ""))
                    if ts is None or ts > future_cutoff or ts < cutoff:
                        continue
                    keep.append(line)
            if len(keep) > max_entries:
                keep = keep[-max_entries:]
            tmp_path = path + ".tmp"
            with open(tmp_path, "w", encoding="utf-8") as f:
                if keep:
                    f.write("\n".join(keep) + "\n")
            os.replace(tmp_path, path)
    except Exception as e:
        _log.warning("task_ledger prune failed: %s", e)
        return 0

    meta = _read_meta()
    meta["last_prune_ts"] = _utcnow_iso()
    meta["last_prune_kept"] = len(keep)
    _write_meta(meta)
    return len(keep)


# ---------------------------------------------------------------------------
# Test helpers
# ---------------------------------------------------------------------------

def _reset_for_tests() -> None:
    """Reset in-process counters/locks. Tests only."""
    global _write_counter
    _write_counter = 0


def _force_prune_check() -> None:
    """Trigger the size/age check without waiting for 50 writes. Tests only."""
    _maybe_prune()
