"""Agent-accessible short timers (Phase 2b §1).

Append-only JSONL store at ``~/.mugi/timers.jsonl`` (0o600, flock-coordinated).
Each row is a one-shot reminder that fires into the delivery queue at a wall-clock
instant. Timers bypass ``activeHours`` (user-requested one-shots must ring on
schedule) but still respect the ``timers.enabled`` master pref.

Public surface
--------------

    schedule(delay_or_at, message, *, thread_id, origin, kind="reminder")  -> row
    cancel(timer_id)                                                       -> bool
    list_pending(thread_id=None)                                           -> list[dict]
    claim_due(now=None, *, limit=25)                                       -> list[dict]
    mark_fired(timer_id)                                                   -> dict | None
    mark_failed(timer_id, error)                                           -> dict | None

Rows are never deleted in-place; ``cancel``/``mark_fired`` flip ``status`` and
the file is rewritten atomically. The runner's drain tick (see ``mugi_runner``)
is responsible for claiming due rows and handing off to ``utils.delivery_queue``.
"""

from __future__ import annotations

import json
import logging
import os
import re
import secrets
import threading
from datetime import datetime, timedelta, timezone
from typing import Any

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

_log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Schema + limits
# ---------------------------------------------------------------------------

SCHEMA_VERSION = 1

STATUS_PENDING = "pending"
STATUS_FIRED = "fired"
STATUS_CANCELLED = "cancelled"
STATUS_FAILED = "failed"
STATUS_CLAIMED = "claimed"  # in-flight: a drainer owns this row (claim-before-fire)
_VALID_STATUSES = frozenset(
    {STATUS_PENDING, STATUS_FIRED, STATUS_CANCELLED, STATUS_FAILED, STATUS_CLAIMED}
)

# Stale-claim recovery: a row left CLAIMED longer than this (owner crashed mid-fire)
# becomes claimable again so the timer still fires (late) rather than never.
CLAIM_TTL_SECONDS = 300

ID_PREFIX = "tmr_"

_MIN_DELAY_SECONDS = 60
_MAX_DELAY_SECONDS = 30 * 24 * 3600  # 30 days
_MAX_MESSAGE_BYTES = 4 * 1024
_MAX_PENDING_PER_THREAD = 25
_MAX_PENDING_PER_DAY = 100
_ROW_HARD_CAP_BYTES = 6 * 1024

_DELAY_RE = re.compile(
    r"^\s*(\d+)\s*(s|sec|secs|second|seconds|m|min|mins|minute|minutes|"
    r"h|hr|hrs|hour|hours|d|day|days)\s*$",
    re.IGNORECASE,
)

_lock = threading.Lock()


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


def _path() -> str:
    return os.path.join(_dir(), "timers.jsonl")


TIMERS_PATH = _path  # callable accessor for tests


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _iso(dt: datetime) -> str:
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    s = dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
    return s


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


def _sanitize_message(raw: str) -> str:
    """Drop control chars (except tab/newline), cap to _MAX_MESSAGE_BYTES."""
    if raw is None:
        raw = ""
    cleaned = "".join(c if (c.isprintable() or c in ("\t", "\n")) else " " for c in str(raw))
    encoded = cleaned.encode("utf-8")
    if len(encoded) > _MAX_MESSAGE_BYTES:
        cleaned = encoded[: _MAX_MESSAGE_BYTES - 1].decode("utf-8", errors="ignore") + "…"
    return cleaned


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 _write_lines_atomic(path: str, lines: list[str]) -> None:
    """tmp+rename write. Caller owns any flock/threading serialization."""
    body = ("\n".join(lines) + "\n") if lines else ""
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(body)
    os.replace(tmp, path)
    try:
        os.chmod(path, 0o600)
    except OSError:
        pass


def _flock_rewrite(path: str, lines: list[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)
            _write_lines_atomic(path, lines)
        finally:
            fcntl.flock(fd, fcntl.LOCK_UN)
            os.close(fd)
    else:
        _write_lines_atomic(path, lines)


def _flock_read_modify_write(transform):
    """Hold ``LOCK_EX`` across read → transform → write so a cross-process read-modify-write
    (claim-before-fire) is atomic. ``transform(rows)`` returns ``(result, new_rows)``; the
    file is rewritten only when ``new_rows`` is not None.

    Reading inside the same flock as the write is the point — ``_flock_rewrite`` locks only
    the write, so two drainers that each read-then-write could both claim the same timer.
    Callers must not call ``_flock_rewrite`` inside ``transform`` (same ``.lock`` path would
    deadlock on a second ``LOCK_EX``).
    """
    path = _path()

    def _run() -> Any:
        rows = _read_all()
        result, new_rows = transform(rows)
        if new_rows is not None:
            _write_lines_atomic(
                path, [json.dumps(r, ensure_ascii=True, default=str) for r in new_rows]
            )
        return result

    if not _HAS_FLOCK:
        return _run()
    lock_path = path + ".lock"
    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
    try:
        fcntl.flock(fd, fcntl.LOCK_EX)
        return _run()
    finally:
        fcntl.flock(fd, fcntl.LOCK_UN)
        os.close(fd)


def _read_all() -> list[dict[str, Any]]:
    p = _path()
    if not os.path.isfile(p):
        return []
    out: list[dict[str, Any]] = []
    try:
        with open(p, "r", encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    row = json.loads(line)
                except Exception:
                    continue
                if isinstance(row, dict) and row.get("id"):
                    out.append(row)
    except OSError as e:
        _log.debug("timers read failed: %s", e)
    return out


# ---------------------------------------------------------------------------
# Delay parsing
# ---------------------------------------------------------------------------

_UNIT_SECONDS = {
    "s": 1, "sec": 1, "secs": 1, "second": 1, "seconds": 1,
    "m": 60, "min": 60, "mins": 60, "minute": 60, "minutes": 60,
    "h": 3600, "hr": 3600, "hrs": 3600, "hour": 3600, "hours": 3600,
    "d": 86400, "day": 86400, "days": 86400,
}


def parse_delay(delay_or_at: str | int | float) -> tuple[datetime | None, str]:
    """Resolve a delay spec to an absolute UTC fire time.

    Accepts integer/float seconds, "30m"/"2h"/"1d" shorthand, or an ISO-8601
    timestamp. Returns (fire_at, error_message). Enforces the min/max window.
    """
    now = _now()
    fire_at: datetime | None = None

    if isinstance(delay_or_at, (int, float)):
        fire_at = now + timedelta(seconds=float(delay_or_at))
    elif isinstance(delay_or_at, str):
        raw = delay_or_at.strip()
        if not raw:
            return None, "empty delay"
        m = _DELAY_RE.match(raw)
        if m:
            n = int(m.group(1))
            unit = m.group(2).lower()
            fire_at = now + timedelta(seconds=n * _UNIT_SECONDS[unit])
        else:
            try:
                n = float(raw)
                fire_at = now + timedelta(seconds=n)
            except ValueError:
                dt = _parse_iso(raw)
                if dt is not None:
                    fire_at = dt
    else:
        return None, "unsupported delay type"

    if fire_at is None:
        return None, (
            "could not parse delay; use '30m', '2h', '1d', an integer of seconds, "
            "or an ISO-8601 timestamp"
        )

    if fire_at.tzinfo is None:
        fire_at = fire_at.replace(tzinfo=timezone.utc)

    delta = (fire_at - now).total_seconds()
    if delta < _MIN_DELAY_SECONDS:
        return None, f"timer delay must be at least {_MIN_DELAY_SECONDS}s in the future"
    if delta > _MAX_DELAY_SECONDS:
        return None, f"timer delay must be within {_MAX_DELAY_SECONDS // 86400} days"
    return fire_at, ""


# ---------------------------------------------------------------------------
# Enabled gate
# ---------------------------------------------------------------------------

def is_enabled() -> bool:
    """Master switch. Returns False when ``timers.enabled`` is off."""
    try:
        from core.config import get_timers_config
        return bool((get_timers_config() or {}).get("enabled", False))
    except Exception:
        return False


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def _mint_id() -> str:
    return ID_PREFIX + secrets.token_hex(8)


def schedule(
    delay_or_at: str | int | float,
    message: str,
    *,
    thread_id: str | None,
    origin: str = "agent",
    kind: str = "reminder",
) -> dict[str, Any]:
    """Create a new pending timer. Raises ValueError on validation failure."""
    fire_at, err = parse_delay(delay_or_at)
    if err or fire_at is None:
        raise ValueError(err or "invalid delay")

    sanitized = _sanitize_message(message).strip()
    if not sanitized:
        raise ValueError("timer message cannot be empty")

    tid = str(thread_id or "").strip()[:64] or None

    rows = _read_all()
    if tid:
        pending_for_thread = sum(
            1 for r in rows
            if r.get("status") == STATUS_PENDING and r.get("thread_id") == tid
        )
        if pending_for_thread >= _MAX_PENDING_PER_THREAD:
            raise ValueError(
                f"already {_MAX_PENDING_PER_THREAD} pending timers for this thread; "
                "cancel one before scheduling another"
            )

    cutoff = _now() - timedelta(days=1)
    recent = 0
    for r in rows:
        created = _parse_iso(str(r.get("created_at") or ""))
        if created and created >= cutoff:
            recent += 1
    if recent >= _MAX_PENDING_PER_DAY:
        raise ValueError(
            f"timer rate limit ({_MAX_PENDING_PER_DAY} per 24h) reached"
        )

    timer_id = _mint_id()
    # Vanishingly small chance of collision; loop until unique.
    seen_ids = {r.get("id") for r in rows}
    while timer_id in seen_ids:
        timer_id = _mint_id()

    row = {
        "schema_version": SCHEMA_VERSION,
        "id": timer_id,
        "fire_at": _iso(fire_at),
        "created_at": _iso(_now()),
        "thread_id": tid,
        "message": sanitized,
        "kind": str(kind or "reminder")[:32],
        "origin": str(origin or "agent")[:32],
        "status": STATUS_PENDING,
        "retry_count": 0,
    }

    encoded = json.dumps(row, ensure_ascii=True, default=str)
    if len(encoded.encode("utf-8")) > _ROW_HARD_CAP_BYTES:
        raise ValueError("timer row exceeds size cap")

    os.makedirs(_dir(), exist_ok=True, mode=0o700)
    with _lock:
        _flock_append(_path(), encoded + "\n")
    try:
        os.chmod(_path(), 0o600)
    except OSError:
        pass

    try:
        from utils import task_ledger as _tl
        _tl.record_run(
            _tl.KIND_TIMER if hasattr(_tl, "KIND_TIMER") else "subagent",
            f"timer_scheduled_{row['kind']}",
            thread_id=tid,
            status=_tl.STATUS_OK,
            meta={"timer_id": timer_id, "fire_at": row["fire_at"], "origin": row["origin"]},
        )
    except Exception:
        _log.debug("ledger (timer scheduled) failed", exc_info=True)

    try:
        from utils import delivery_queue as _dq
        ev = getattr(_dq, "_wake_event", None)
        if ev is not None:
            ev.set()
    except Exception:
        pass
    return row


def cancel(timer_id: str) -> bool:
    """Flip status → cancelled. Returns True if a pending row was found."""
    if not timer_id:
        return False
    found = False
    with _lock:
        rows = _read_all()
        new_lines: list[str] = []
        for r in rows:
            if r.get("id") == timer_id and r.get("status") == STATUS_PENDING:
                r["status"] = STATUS_CANCELLED
                r["cancelled_at"] = _iso(_now())
                found = True
            new_lines.append(json.dumps(r, ensure_ascii=True, default=str))
        if found:
            _flock_rewrite(_path(), new_lines)
    return found


def list_pending(thread_id: str | None = None) -> list[dict[str, Any]]:
    """Return pending rows, optionally filtered by thread_id."""
    out: list[dict[str, Any]] = []
    for row in _read_all():
        if row.get("status") != STATUS_PENDING:
            continue
        if thread_id is not None and row.get("thread_id") != thread_id:
            continue
        out.append(row)
    out.sort(key=lambda r: str(r.get("fire_at") or ""))
    return out


def claim_due(now: datetime | None = None, *, limit: int = 25) -> list[dict[str, Any]]:
    """Return up to *limit* pending rows whose fire_at <= now, oldest first."""
    ref = (now or _now())
    due: list[dict[str, Any]] = []
    for row in _read_all():
        if row.get("status") != STATUS_PENDING:
            continue
        fire_at = _parse_iso(str(row.get("fire_at") or ""))
        if fire_at is None:
            continue
        if fire_at <= ref:
            due.append(row)
    due.sort(key=lambda r: str(r.get("fire_at") or ""))
    return due[: max(1, int(limit))]


def claim_batch(now: datetime | None = None, *, limit: int = 25) -> list[dict[str, Any]]:
    """Atomically claim up to *limit* due timers, returning the claimed rows.

    Under the file lock, flips selected PENDING-due rows (plus stale CLAIMED rows whose
    owner died more than ``CLAIM_TTL_SECONDS`` ago) to ``STATUS_CLAIMED`` stamped with the
    claiming pid, then rewrites the file in one shot. A second concurrent drainer cannot
    re-select the same row, so each timer fires once. ``mark_fired`` / ``mark_failed`` move
    the row out of CLAIMED.

    Defense-in-depth alongside ``mugi_runner``'s single-instance lock: the prior read-only
    ``claim_due`` + fire pattern could double-fire a timer during a transient runner overlap.
    """
    ref = now or _now()
    stale_cutoff = ref.timestamp() - CLAIM_TTL_SECONDS
    pid = os.getpid()

    def _is_claimable(row: dict[str, Any]) -> bool:
        status = row.get("status")
        if status == STATUS_PENDING:
            fire_at = _parse_iso(str(row.get("fire_at") or ""))
            return fire_at is not None and fire_at <= ref
        if status == STATUS_CLAIMED:
            cl = _parse_iso(str(row.get("claimed_at") or ""))
            return cl is None or cl.timestamp() <= stale_cutoff
        return False

    def _transform(rows: list[dict[str, Any]]):
        due = sorted(
            (r for r in rows if _is_claimable(r)),
            key=lambda r: str(r.get("fire_at") or ""),
        )[: max(1, int(limit))]
        if not due:
            return [], None  # nothing due → skip the rewrite
        due_ids = {id(r) for r in due}
        claimed_at = _iso(_now())
        claimed_rows: list[dict[str, Any]] = []
        for r in rows:
            if id(r) in due_ids:
                r["status"] = STATUS_CLAIMED
                r["claimed_at"] = claimed_at
                r["claimed_by"] = pid
                claimed_rows.append(r)
        return claimed_rows, rows

    # Threading lock (intra-process) + flock across read+write (cross-process): the claim is
    # only single-owner if the selection and the rewrite happen under one held file lock.
    with _lock:
        return _flock_read_modify_write(_transform)


def _mutate(timer_id: str, mutator) -> dict[str, Any] | None:
    target = None
    with _lock:
        rows = _read_all()
        new_lines: list[str] = []
        for r in rows:
            if r.get("id") == timer_id and target is None:
                mutator(r)
                target = r
            new_lines.append(json.dumps(r, ensure_ascii=True, default=str))
        if target is None:
            return None
        _flock_rewrite(_path(), new_lines)
    return target


def mark_fired(timer_id: str) -> dict[str, Any] | None:
    return _mutate(timer_id, lambda r: r.update({
        "status": STATUS_FIRED,
        "fired_at": _iso(_now()),
    }))


def mark_failed(timer_id: str, error: str) -> dict[str, Any] | None:
    def _m(r: dict[str, Any]) -> None:
        r["status"] = STATUS_FAILED
        r["failed_at"] = _iso(_now())
        r["last_error"] = str(error or "")[:400]
        r["retry_count"] = int(r.get("retry_count") or 0) + 1
    return _mutate(timer_id, _m)


def purge_completed(max_age_days: int = 7) -> int:
    """Remove fired/cancelled/failed rows older than max_age_days."""
    cutoff = _now() - timedelta(days=max(1, int(max_age_days)))
    removed = 0
    with _lock:
        rows = _read_all()
        keep_lines: list[str] = []
        for r in rows:
            status = r.get("status")
            if status in (STATUS_FIRED, STATUS_CANCELLED, STATUS_FAILED):
                stamp = (
                    r.get("fired_at") or r.get("cancelled_at")
                    or r.get("failed_at") or r.get("created_at")
                )
                ts = _parse_iso(str(stamp or ""))
                if ts is not None and ts < cutoff:
                    removed += 1
                    continue
            keep_lines.append(json.dumps(r, ensure_ascii=True, default=str))
        _flock_rewrite(_path(), keep_lines)
    return removed


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

def _reset_for_tests() -> None:
    pass
