"""Runner-owned delivery queue for cron + background subagent results.

Crash-safe, append-only JSONL (``~/.mugi/delivery_queue.jsonl``) with
idempotent ``delivery_id``, exponential backoff, dead-lettering, and FIFO
ordering within a tick. See plan §4.

The actual delivery execution is owned by ``mugi_runner``'s dedicated 5s
drain thread; this module only persists state and returns entries to drain.
"""

from __future__ import annotations

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

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

_log = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------

SCHEMA_VERSION = 1

KIND_THREAD = "thread"
KIND_FILE = "file"
KIND_NOTIFICATION = "notification"
KIND_WEBHOOK = "webhook"
KIND_DISCORD = "discord"
KIND_SLACK = "slack"
_VALID_KINDS = frozenset({KIND_THREAD, KIND_FILE, KIND_NOTIFICATION, KIND_WEBHOOK, KIND_DISCORD, KIND_SLACK})

STATUS_PENDING = "pending"
STATUS_DELIVERED = "delivered"
STATUS_FAILED = "failed"
STATUS_DEAD = "dead"
STATUS_CLAIMED = "claimed"  # in-flight: a drainer owns this row (claim-before-execute)

# Backoff schedule in seconds: 30s, 2m, 10m, 1h, then dead on attempt 5.
BACKOFF_SECONDS = [30, 120, 600, 3600]
DEFAULT_MAX_ATTEMPTS = 5

# Stale-claim recovery: a row left CLAIMED longer than this (owner crashed mid-execute)
# becomes claimable again. Keep comfortably above the longest single delivery.
CLAIM_TTL_SECONDS = 300

# File bounds.
_MAX_PAYLOAD_BYTES = 32 * 1024

_lock = threading.Lock()
_wake_event: threading.Event | None = None


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


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


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


QUEUE_PATH = _path  # callable for tests


def set_wake_event(event: threading.Event) -> None:
    """Register an event to poke when a new entry is enqueued. Set by mugi_runner."""
    global _wake_event
    _wake_event = event


# ---------------------------------------------------------------------------
# File I/O helpers (flock-coordinated)
# ---------------------------------------------------------------------------

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 is responsible for any flock/threading serialization."""
    content = ("\n".join(lines) + "\n") if lines else ""
    tmp = path + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        f.write(content)
    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 parse → transform → write so a cross-process read-modify-write
    (e.g. claim-before-execute) is atomic.

    ``transform(entries)`` must return ``(result, new_entries)``; the file is rewritten only
    when ``new_entries`` is not None. The read happens *inside* the same flock as the write —
    this is the whole point: ``_flock_rewrite`` locks only the write, so two processes that
    each read-then-write (the old claim pattern) could both select the same rows. Corrupt
    lines are dead-lettered (parity with ``_read_all``) before the rewrite drops them.

    NB: callers must NOT also call ``_flock_rewrite`` while inside ``transform`` — it re-opens
    the same ``.lock`` path and a second ``LOCK_EX`` would deadlock the process.
    """
    path = _path()

    def _run() -> Any:
        entries, corrupt_lines = _parse_queue_lines()
        if corrupt_lines:
            for raw in corrupt_lines:
                _deadletter_corrupt_line(raw)
            _log.warning("delivery_queue skipped %d corrupt JSONL line(s)", len(corrupt_lines))
        result, new_entries = transform(entries)
        if new_entries is not None:
            _write_lines_atomic(
                path, [json.dumps(r, ensure_ascii=True, default=str) for r in new_entries]
            )
        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 _load_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 _save_meta(meta: dict[str, Any]) -> None:
    os.makedirs(_dir(), exist_ok=True)
    tmp = _meta_path() + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(meta, f, ensure_ascii=True)
    os.replace(tmp, _meta_path())


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


def _append_corrupt_line(raw_line: str) -> None:
    sidecar = _corrupt_sidecar_path()
    os.makedirs(_dir(), exist_ok=True)
    record = json.dumps(
        {"ts": _now_iso(), "line": raw_line[:4096]},
        ensure_ascii=True,
    )
    _flock_append(sidecar, record + "\n")


def _deadletter_corrupt_line(raw_line: str) -> None:
    """Append corrupt queue line to sidecar once (dedupe by content hash)."""
    digest = hashlib.sha256(raw_line.encode("utf-8")).hexdigest()[:24]
    meta = _load_meta()
    seen = meta.get("corrupt_line_hashes")
    if not isinstance(seen, list):
        seen = []
    if digest in seen:
        return
    seen.append(digest)
    meta["corrupt_line_hashes"] = seen[-512:]
    try:
        _save_meta(meta)
        _append_corrupt_line(raw_line)
    except Exception:
        _log.debug("delivery_queue corrupt sidecar append failed", exc_info=True)


def _parse_queue_lines() -> tuple[list[dict[str, Any]], list[str]]:
    """Parse queue file into valid rows and raw corrupt lines."""
    p = _path()
    if not os.path.isfile(p):
        return [], []
    entries: list[dict[str, Any]] = []
    corrupt_lines: list[str] = []
    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:
                    corrupt_lines.append(line)
                    continue
                if isinstance(row, dict):
                    entries.append(row)
    except OSError as e:
        _log.debug("delivery_queue read failed: %s", e)
    return entries, corrupt_lines


def _compact_queue_remove_corrupt() -> None:
    """Rewrite queue file without corrupt lines. Caller must not hold ``_lock``."""
    with _lock:
        entries, corrupt_lines = _parse_queue_lines()
        if not corrupt_lines:
            return
        new_lines = [json.dumps(row, ensure_ascii=True, default=str) for row in entries]
        _flock_rewrite(_path(), new_lines)


def _read_all(*, compact_corrupt: bool = True) -> list[dict[str, Any]]:
    entries, corrupt_lines = _parse_queue_lines()
    if corrupt_lines:
        for raw in corrupt_lines:
            _deadletter_corrupt_line(raw)
        _log.warning("delivery_queue skipped %d corrupt JSONL line(s)", len(corrupt_lines))
        if compact_corrupt:
            _compact_queue_remove_corrupt()
    return entries


def corrupt_sidecar_count() -> int:
    """Return number of dead-lettered corrupt lines (test/health helper)."""
    sidecar = _corrupt_sidecar_path()
    if not os.path.isfile(sidecar):
        return 0
    count = 0
    try:
        with open(sidecar, "r", encoding="utf-8") as f:
            for line in f:
                if line.strip():
                    count += 1
    except OSError:
        return 0
    return count


# ---------------------------------------------------------------------------
# Identity + time
# ---------------------------------------------------------------------------

def _now_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_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 compute_delivery_id(run_id: str, kind: str, target: str) -> str:
    """Idempotent delivery id. Re-enqueue with same tuple collapses."""
    target_hash = hashlib.sha256(str(target or "").encode("utf-8")).hexdigest()[:16]
    raw = f"{run_id or ''}|{kind}|{target_hash}"
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24]


# ---------------------------------------------------------------------------
# enqueue / claim / mark helpers
# ---------------------------------------------------------------------------

def enqueue(
    kind: str,
    payload: dict[str, Any],
    *,
    delivery_id: str | None = None,
    run_id: str | None = None,
    origin_thread_id: str | None = None,
) -> str:
    """Append a pending delivery. Returns the delivery_id (new or supplied)."""
    if kind not in _VALID_KINDS:
        raise ValueError(f"unknown delivery kind: {kind!r}")
    if not delivery_id:
        target = str(payload.get("target") or payload.get("thread_id") or payload.get("url") or payload.get("filename") or "")
        delivery_id = compute_delivery_id(run_id or uuid.uuid4().hex, kind, target)

    try:
        encoded_payload = json.dumps(payload, ensure_ascii=True, default=str)
    except Exception as e:
        raise ValueError(f"payload not json-serialisable: {e}")
    if len(encoded_payload.encode("utf-8")) > _MAX_PAYLOAD_BYTES:
        raise ValueError("delivery payload exceeds 32 KB cap")

    os.makedirs(_dir(), exist_ok=True, mode=0o700)

    # Short-circuit if an identical delivery_id already exists and is pending/delivered.
    existing = _read_all()
    for row in existing:
        if row.get("delivery_id") == delivery_id and row.get("status") in (
            STATUS_PENDING,
            STATUS_DELIVERED,
            STATUS_CLAIMED,
        ):
            return delivery_id

    row = {
        "schema_version": SCHEMA_VERSION,
        "delivery_id": delivery_id,
        "run_id": run_id,
        "origin_thread_id": origin_thread_id,
        "kind": kind,
        "payload": payload,
        "created_at": _now_iso(),
        "next_attempt_at": _now_iso(),
        "attempts": 0,
        "status": STATUS_PENDING,
        "last_error": None,
    }
    with _lock:
        _flock_append(_path(), json.dumps(row, ensure_ascii=True, default=str) + "\n")
    try:
        os.chmod(_path(), 0o600)
    except OSError:
        pass
    if _wake_event is not None:
        try:
            _wake_event.set()
        except Exception:
            pass
    return delivery_id


def claim_due(now_iso: str | None = None, *, limit: int = 10) -> list[dict[str, Any]]:
    """Return up to *limit* pending entries whose next_attempt_at <= now, FIFO."""
    now_dt = _parse_iso(now_iso) if now_iso else datetime.now(timezone.utc)
    entries = _read_all()
    due: list[dict[str, Any]] = []
    for row in entries:
        if row.get("status") != STATUS_PENDING:
            continue
        next_dt = _parse_iso(str(row.get("next_attempt_at") or ""))
        if next_dt is None or next_dt <= now_dt:
            due.append(row)

    def _sort_key(r):
        na = _parse_iso(str(r.get("next_attempt_at") or ""))
        ca = _parse_iso(str(r.get("created_at") or ""))
        return (na or datetime.min.replace(tzinfo=timezone.utc), ca or datetime.min.replace(tzinfo=timezone.utc))

    due.sort(key=_sort_key)
    return due[: max(1, int(limit))]


def claim_batch(now_iso: str | None = None, *, limit: int = 10) -> list[dict[str, Any]]:
    """Atomically claim up to *limit* due deliveries, 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 reading
    the queue therefore cannot re-select the same row, so each delivery executes once.
    ``mark_delivered`` / ``mark_failed`` move the row out of CLAIMED afterward.

    ``mugi_runner``'s single-instance lock already prevents two runners; this is
    defense-in-depth for a transient overlap window (manual second runner, restart) where
    the old read-only ``claim_due`` + execute pattern could double-deliver.
    """
    now_dt = _parse_iso(now_iso) if now_iso else datetime.now(timezone.utc)
    stale_cutoff = now_dt.timestamp() - CLAIM_TTL_SECONDS
    pid = os.getpid()

    def _is_claimable(row: dict[str, Any]) -> bool:
        status = row.get("status")
        if status == STATUS_PENDING:
            next_dt = _parse_iso(str(row.get("next_attempt_at") or ""))
            return next_dt is None or next_dt <= now_dt
        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 _sort_key(r):
        na = _parse_iso(str(r.get("next_attempt_at") or ""))
        ca = _parse_iso(str(r.get("created_at") or ""))
        return (
            na or datetime.min.replace(tzinfo=timezone.utc),
            ca or datetime.min.replace(tzinfo=timezone.utc),
        )

    def _transform(entries: list[dict[str, Any]]):
        due = sorted((r for r in entries if _is_claimable(r)), key=_sort_key)[: max(1, int(limit))]
        if not due:
            return [], None  # nothing to claim → skip the rewrite
        due_ids = {id(r) for r in due}
        claimed_at = _now_iso()
        claimed_rows: list[dict[str, Any]] = []
        for row in entries:
            if id(row) in due_ids:
                row["status"] = STATUS_CLAIMED
                row["claimed_at"] = claimed_at
                row["claimed_by"] = pid
                claimed_rows.append(row)
        return claimed_rows, entries

    # Hold the threading lock (intra-process) AND the flock across read+write (cross-process)
    # so the claim is atomic on both axes — otherwise two drainers select the same rows.
    with _lock:
        return _flock_read_modify_write(_transform)


def _mutate_row(delivery_id: str, mutator) -> dict[str, Any] | None:
    """Locate row by delivery_id, apply *mutator*, rewrite file. Returns the updated row or None."""
    with _lock:
        entries = _read_all(compact_corrupt=False)
        target = None
        new_lines: list[str] = []
        for row in entries:
            if row.get("delivery_id") == delivery_id and target is None:
                mutator(row)
                target = row
            new_lines.append(json.dumps(row, ensure_ascii=True, default=str))
        if target is None:
            return None
        _flock_rewrite(_path(), new_lines)
    return target


def mark_delivered(delivery_id: str) -> dict[str, Any] | None:
    return _mutate_row(delivery_id, lambda r: r.update({
        "status": STATUS_DELIVERED,
        "delivered_at": _now_iso(),
        "last_error": None,
    }))


def mark_failed(
    delivery_id: str,
    error: str,
    *,
    max_attempts: int = DEFAULT_MAX_ATTEMPTS,
    terminal: bool = False,
) -> dict[str, Any] | None:
    """Increment attempts and set backoff; terminal=True sends straight to dead."""
    def _mut(row: dict[str, Any]) -> None:
        attempts = int(row.get("attempts") or 0) + 1
        row["attempts"] = attempts
        row["last_error"] = str(error)[:400]
        if terminal or attempts >= max_attempts:
            row["status"] = STATUS_DEAD
            row["dead_at"] = _now_iso()
        else:
            backoff_idx = min(attempts - 1, len(BACKOFF_SECONDS) - 1)
            backoff_s = BACKOFF_SECONDS[backoff_idx]
            next_dt = datetime.now(timezone.utc).timestamp() + backoff_s
            row["next_attempt_at"] = datetime.fromtimestamp(next_dt, tz=timezone.utc).strftime(
                "%Y-%m-%dT%H:%M:%S.%f"
            )[:-3] + "Z"
            row["status"] = STATUS_PENDING

    return _mutate_row(delivery_id, _mut)


def sweep_dead(max_attempts: int = DEFAULT_MAX_ATTEMPTS) -> int:
    """Promote any row with attempts >= max_attempts to ``dead``. Returns count promoted."""
    promoted = 0
    with _lock:
        entries = _read_all(compact_corrupt=False)
        new_lines: list[str] = []
        for row in entries:
            if (
                row.get("status") == STATUS_PENDING
                and int(row.get("attempts") or 0) >= max_attempts
            ):
                row["status"] = STATUS_DEAD
                row["dead_at"] = _now_iso()
                promoted += 1
            new_lines.append(json.dumps(row, ensure_ascii=True, default=str))
        _flock_rewrite(_path(), new_lines)
    return promoted


def pending(kind: str | None = None, limit: int = 200) -> list[dict[str, Any]]:
    """Return pending+failed entries for debug/MCP visibility. Newest first."""
    entries = _read_all()
    out: list[dict[str, Any]] = []
    for row in entries:
        if row.get("status") not in (STATUS_PENDING, STATUS_FAILED):
            continue
        if kind and row.get("kind") != kind:
            continue
        out.append(row)
    out.sort(key=lambda r: str(r.get("created_at") or ""), reverse=True)
    return out[: max(1, int(limit))]


def purge_delivered(max_age_seconds: int = 7 * 24 * 3600) -> int:
    """Remove delivered+dead rows older than *max_age_seconds*. Returns count removed."""
    cutoff = datetime.now(timezone.utc).timestamp() - max_age_seconds
    removed = 0
    with _lock:
        entries = _read_all(compact_corrupt=False)
        keep: list[str] = []
        for row in entries:
            status = row.get("status")
            if status in (STATUS_DELIVERED, STATUS_DEAD):
                stamp_field = "delivered_at" if status == STATUS_DELIVERED else "dead_at"
                ts = _parse_iso(str(row.get(stamp_field) or row.get("created_at") or ""))
                if ts is not None and ts.timestamp() < cutoff:
                    removed += 1
                    continue
            keep.append(json.dumps(row, ensure_ascii=True, default=str))
        _flock_rewrite(_path(), keep)
    return removed


# ---------------------------------------------------------------------------
# Executor (invoked by the runner drain thread)
# ---------------------------------------------------------------------------

def _classify_error(err: Exception) -> tuple[bool, str]:
    """Return (terminal, short_reason). Terminal → no retry, straight to dead."""
    msg = str(err).lower()
    if "row-too-large" in msg or "schema" in msg and "auth" in msg:
        return True, str(err)[:120]
    return False, str(err)[:120]


def execute_one(row: dict[str, Any]) -> tuple[bool, str]:
    """Dispatch a delivery row to the correct delivery primitive.

    Returns (ok, error_summary). Uses delivery helpers from cron_scheduler
    so SSRF validation for webhook URLs is preserved.
    """
    kind = str(row.get("kind") or "")
    payload = row.get("payload") or {}
    if not isinstance(payload, dict):
        return False, "bad payload"
    try:
        from cron_scheduler import (
            _deliver_to_file,
            _deliver_to_thread,
            _deliver_notification,
            _deliver_webhook,
        )
    except Exception as e:
        return False, f"cron_scheduler import failed: {e}"

    try:
        if kind == KIND_THREAD:
            kanban_event = str(payload.get("kanban_event") or "").strip()
            thread_id = str(payload.get("thread_id") or "").strip()
            if kanban_event and thread_id:
                from utils.kanban_coordinator_delivery import (
                    deliver_coordinator_continuation,
                    wake_coordinator_thread,
                )

                delivered = deliver_coordinator_continuation(
                    thread_id,
                    banner=str(payload.get("banner") or "Kanban update"),
                    body=str(payload.get("result") or ""),
                )
                if delivered:
                    wake_coordinator_thread(
                        thread_id,
                        reason=kanban_event,
                    )
            else:
                banner = str(payload.get("banner") or "Scheduled task result")
                _deliver_to_thread(
                    str(payload.get("result") or ""),
                    str(payload.get("thread_id") or ""),
                    banner,
                )
        elif kind == KIND_FILE:
            _deliver_to_file(
                str(payload.get("job_id") or ""),
                str(payload.get("prompt") or ""),
                str(payload.get("result") or ""),
            )
        elif kind == KIND_NOTIFICATION:
            _deliver_notification(
                str(payload.get("prompt") or ""),
                str(payload.get("result") or ""),
            )
        elif kind == KIND_WEBHOOK:
            _deliver_webhook(
                str(payload.get("url") or ""),
                str(payload.get("job_id") or ""),
                str(payload.get("prompt") or ""),
                str(payload.get("result") or ""),
            )
        elif kind in (KIND_DISCORD, KIND_SLACK):
            from utils.webhook_formatters import deliver_chat_webhook
            deliver_chat_webhook(
                channel_kind=kind,
                url=str(payload.get("url") or ""),
                banner=str(payload.get("banner") or payload.get("prompt") or ""),
                body=str(payload.get("result") or payload.get("body") or ""),
                task_id=str(payload.get("task_id") or ""),
                event=str(payload.get("kanban_event") or ""),
            )
        else:
            return False, f"unknown kind: {kind}"
        return True, ""
    except Exception as e:
        return False, str(e)[:400]


def drain(*, limit: int = 10, max_attempts: int = DEFAULT_MAX_ATTEMPTS) -> dict[str, int]:
    """Drain one tick of due deliveries. Returns counters.

    The runner calls this on its 5s cadence (or on wake_event).
    Each delivered/failed row also records a ledger row.
    """
    due = claim_batch(limit=limit)
    counters = {"delivered": 0, "failed": 0, "dead": 0, "attempted": 0}
    for row in due:
        delivery_id = str(row.get("delivery_id") or "")
        counters["attempted"] += 1
        ok, err = execute_one(row)
        if ok:
            mark_delivered(delivery_id)
            counters["delivered"] += 1
            try:
                from utils import task_ledger as _tl
                _tl.record_run(
                    _tl.KIND_DELIVERY,
                    f"delivery_{row.get('kind')}",
                    thread_id=str(row.get("origin_thread_id") or ""),
                    parent_run_id=str(row.get("run_id") or ""),
                    status=_tl.STATUS_OK,
                    meta={"delivery_id": delivery_id, "attempts": int(row.get("attempts") or 0) + 1},
                )
            except Exception:
                _log.debug("ledger record (delivery ok) failed", exc_info=True)
        else:
            terminal, _ = _classify_error(Exception(err))
            updated = mark_failed(delivery_id, err, max_attempts=max_attempts, terminal=terminal)
            status = (updated or {}).get("status")
            if status == STATUS_DEAD:
                counters["dead"] += 1
                _surface_dead_letter(updated)
            else:
                counters["failed"] += 1
            try:
                from utils import task_ledger as _tl
                _tl.record_run(
                    _tl.KIND_DELIVERY,
                    f"delivery_{row.get('kind')}",
                    thread_id=str(row.get("origin_thread_id") or ""),
                    parent_run_id=str(row.get("run_id") or ""),
                    status=_tl.STATUS_ERROR if status == STATUS_DEAD else _tl.STATUS_ERROR,
                    error_summary=err,
                    meta={"delivery_id": delivery_id, "attempts": int((updated or {}).get("attempts") or 0), "dead": status == STATUS_DEAD},
                )
            except Exception:
                _log.debug("ledger record (delivery fail) failed", exc_info=True)
    return counters


def _surface_dead_letter(row: dict[str, Any] | None) -> None:
    """Emit a one-line system message + macOS notification so dead letters are visible."""
    if not row:
        return
    payload = row.get("payload") or {}
    attempts = int(row.get("attempts") or 0)
    err = str(row.get("last_error") or "")[:200]
    kind = str(row.get("kind") or "")
    origin = str(row.get("origin_thread_id") or "")

    if kind == KIND_THREAD and origin:
        try:
            from chat_storage import append_messages
            append_messages(origin, [{
                "role": "assistant",
                "content": f"[Mugi] Scheduled task delivery failed after {attempts} attempts: {err}",
            }])
        except Exception:
            _log.debug("dead-letter thread append failed", exc_info=True)

    try:
        from utils.notifications import send_mac_notification
        send_mac_notification(
            title="Mugi delivery failed",
            subtitle=kind,
            body=(err or "delivery gave up")[:200],
        )
    except Exception:
        _log.debug("dead-letter notification failed", exc_info=True)


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

def _reset_for_tests() -> None:
    global _wake_event
    _wake_event = None
    for path in (_path(), _corrupt_sidecar_path(), _meta_path()):
        try:
            if os.path.isfile(path):
                os.remove(path)
        except OSError:
            pass
