"""Python handlers for JS bridge calls from the WebView.

Each handler accepts a single args dict and returns a JSON-serialisable
value (or raises, which the bridge surfaces as a rejected JS Promise).
Handlers run on a background thread so a slow file read doesn't freeze
the WebView's run loop.

Wired into WebShell.bridge by BankWindow._activate_web_shell.
"""

from __future__ import annotations

import html as _html
import json
from datetime import datetime
from pathlib import Path
from typing import Any, Callable


# ── Meetings ──────────────────────────────────────────────────────────


def _meetings_root() -> Path:
    """Same path the meeting_recorder writes to."""
    return Path.home() / "Library" / "Application Support" / "Lumo" / "meetings"


# Meeting IDs are directory names produced by the recorder — pattern
# `YYYY-MM-DD-HHMM-<slug>`. Reject anything with `/`, `..`, control
# chars, or whitespace so the bridge handlers can never be coerced
# into reading or writing a path outside the meetings/ root via a
# crafted `id` argument. Audit #3 (2026-06-10).
import re as _re_module
_MEETING_ID_RE = _re_module.compile(r"^[A-Za-z0-9_\-.:]{1,128}$")


def _safe_meeting_dir(meeting_id) -> Path | None:
    """Validate the user-provided meeting_id and return the resolved dir
    if it's a real subdirectory of meetings_root. Returns None otherwise."""
    if not isinstance(meeting_id, str):
        return None
    if not _MEETING_ID_RE.match(meeting_id):
        return None
    root = _meetings_root().resolve()
    candidate = (root / meeting_id).resolve()
    # Belt-and-braces: even with the regex, ensure the resolved path is
    # actually under root (so a symlink trick can't escape).
    try:
        candidate.relative_to(root)
    except ValueError:
        return None
    return candidate if candidate.is_dir() else None


def _read_meta(meeting_dir: Path) -> dict:
    """Best-effort meta.json load — never raises, returns {} on failure
    so the UI can still show what it can about the meeting."""
    try:
        return json.loads((meeting_dir / "meta.json").read_text(encoding="utf-8"))
    except Exception:
        return {}


def _fmt_date(iso: str | None, fallback: str = "—") -> str:
    """Render meta.started_at as 'Jun 6, 02:22'. The design CSS expects
    no leading zero on the day and 24-hour HH:MM."""
    if not iso:
        return fallback
    try:
        d = datetime.fromisoformat(iso)
        return d.strftime("%b %-d, %H:%M")
    except Exception:
        return fallback


def _fmt_dur(seconds: float | int | None) -> str:
    s = int(seconds or 0)
    return f"{s // 60:02d}:{s - (s // 60) * 60:02d}"


def _status_for_ui(status: str) -> str:
    """Map meta.json `status` to the design-CSS status class. The CSS uses:
       summary / transcript / recording / transcribing / failed."""
    return {
        "recording": "recording",
        "finalising": "transcribing",
        "ready_for_transcription": "transcribing",
        "transcribed": "transcript",
        "transcribed_partial": "transcript",
        "summarised": "summary",
        "skipped_too_short": "transcript",  # transcript exists, just no summary
        "awaiting_summary_consent": "transcript",  # audit #5 — no consent
        "summary_failed": "transcript",  # CR #14 — surface as transcript, Retry available
        "failed": "failed",
    }.get(status or "", "summary")


def _title_from_summary(meeting_dir: Path) -> str | None:
    """First ### heading in summary.md, or None."""
    sm = meeting_dir / "summary.md"
    try:
        for line in sm.read_text(encoding="utf-8").splitlines():
            if line.startswith("### "):
                return line[4:].strip()
    except OSError:
        pass
    return None


def _preview_from_summary(meeting_dir: Path) -> str:
    """First non-heading paragraph from summary.md, used for the 2-line
    preview clamp in the meeting list rows."""
    sm = meeting_dir / "summary.md"
    try:
        lines = sm.read_text(encoding="utf-8").splitlines()
    except OSError:
        return ""
    seen_heading = False
    chunks: list[str] = []
    for line in lines:
        stripped = line.strip()
        if stripped.startswith("###"):
            seen_heading = True
            continue
        if not seen_heading:
            continue
        if not stripped:
            if chunks:
                break
            continue
        if stripped.startswith(("-", "**", ">", "#")):
            break
        # Strip inline markdown markers.
        chunks.append(
            stripped.replace("**", "").replace("`", "").replace("*", "").replace("_", "")
        )
        if sum(len(c) for c in chunks) > 220:
            break
    preview = " ".join(chunks)
    if len(preview) > 240:
        preview = preview[:237].rsplit(" ", 1)[0] + "…"
    return preview


def list_meetings(_args: dict) -> list[dict]:
    """Return all meetings newest-first as a list of dicts the design
    HTML's row template can consume directly."""
    root = _meetings_root()
    if not root.exists():
        return []
    out: list[dict] = []
    try:
        entries = sorted(
            (p for p in root.iterdir() if p.is_dir()),
            reverse=True,
        )
    except OSError:
        return []
    for d in entries:
        meta = _read_meta(d)
        # Archived meetings stay on disk but are hidden from the main list
        # (set via _on_web_archive_meeting in bank.py — non-destructive).
        if meta.get("archived"):
            continue
        title = (meta.get("title_override")
                 or _title_from_summary(d)
                 or "Untitled meeting")
        out.append({
            "id": d.name,
            "title": title,
            "date": _fmt_date(meta.get("started_at")),
            "dur": _fmt_dur(meta.get("duration_seconds")),
            "status": _status_for_ui(meta.get("status", "")),
            "preview": _preview_from_summary(d),
            "favorite": bool(meta.get("favorite")),
        })
    # Favorites bubble to the top, then chronological (which is already
    # newest-first because we sorted reverse=True above).
    out.sort(key=lambda m: (not m["favorite"], 0))
    return out


def _md_to_html(md_text: str) -> str:
    """Tiny markdown→HTML converter scoped to the subset our LLM writes.
    Mirrors `.md` rules in lumo.css: H2, H3, paragraphs, bullets,
    blockquote, code, hr. Action items use `.task` cards with `.box`."""
    out: list[str] = []
    lines = md_text.replace("\r\n", "\n").split("\n")
    i = 0
    in_list = False

    def close_list() -> None:
        nonlocal in_list
        if in_list:
            out.append("</ul>")
            in_list = False

    def render_inline(s: str) -> str:
        # bold + italic + code + link — minimal set, regex below.
        import re as _re
        s = _html.escape(s)
        # code first (so ** inside `code` isn't bold)
        s = _re.sub(r"`([^`\n]+?)`", r"<code>\1</code>", s)
        # bold
        s = _re.sub(r"\*\*([^*\n]+?)\*\*", r"<strong>\1</strong>", s)
        # italic (single * not surrounded by word chars)
        s = _re.sub(
            r"(?<![*\w])\*([^*\n]+?)\*(?!\w)", r"<em>\1</em>", s,
        )
        # links — restrict to http/https/mailto; everything else
        # (javascript:, file:, data:, vbscript:, custom schemes used by
        # other apps for handoff, etc.) renders as bare text instead of
        # an anchor. Audit #3 (2026-06-10): without this allow-list a
        # prompt-injected `[click](javascript:lumo.call(...))` in a
        # meeting summary would execute as soon as the user clicks.
        _ALLOWED_SCHEMES = ("http:", "https:", "mailto:")
        def _link_sub(m):
            text, url = m.group(1), m.group(2)
            url_lc = url.strip().lower()
            if any(url_lc.startswith(p) for p in _ALLOWED_SCHEMES):
                return f'<a href="{url}">{text}</a>'
            return text  # drop the href, keep the label as plain text
        s = _re.sub(r"\[([^\]\n]+)\]\(([^)\n]+)\)", _link_sub, s)
        return s

    while i < len(lines):
        line = lines[i].rstrip()
        stripped = line.strip()

        if not stripped:
            close_list()
            i += 1
            continue

        # Heading (## / ###)
        if stripped.startswith("### "):
            close_list()
            out.append(f"<h3>{render_inline(stripped[4:].strip())}</h3>")
            i += 1
            continue
        if stripped.startswith("## "):
            close_list()
            out.append(f"<h2>{render_inline(stripped[3:].strip())}</h2>")
            i += 1
            continue
        if stripped.startswith("# "):
            close_list()
            out.append(f"<h1>{render_inline(stripped[2:].strip())}</h1>")
            i += 1
            continue

        # Horizontal rule
        if stripped in ("---", "***", "___"):
            close_list()
            out.append("<hr>")
            i += 1
            continue

        # Task (action item) — checkbox style
        if stripped.startswith(("- [ ] ", "- [x] ", "- [X] ", "* [ ] ", "* [x] ", "* [X] ")):
            close_list()
            checked = "x" in stripped[2:6].lower()
            text = stripped.split("] ", 1)[1] if "] " in stripped else stripped
            check_glyph = (
                '<svg width="13" height="13" viewBox="0 0 24 24" fill="none">'
                '<path d="M5 12l4 4 10-10" stroke="currentColor" '
                'stroke-width="2.4" stroke-linecap="round" stroke-linejoin="round"/></svg>'
                if checked else ""
            )
            klass = "task done" if checked else "task"
            out.append(
                f'<div class="{klass}"><div class="box">{check_glyph}</div>'
                f'<div class="txt">{render_inline(text)}</div></div>'
            )
            i += 1
            continue

        # Bullet list
        if stripped.startswith(("- ", "* ")):
            if not in_list:
                out.append("<ul>")
                in_list = True
            out.append(f"<li>{render_inline(stripped[2:])}</li>")
            i += 1
            continue

        # Blockquote
        if stripped.startswith("> "):
            close_list()
            # Collect consecutive blockquote lines.
            qlines = [stripped[2:]]
            j = i + 1
            while j < len(lines) and lines[j].strip().startswith("> "):
                qlines.append(lines[j].strip()[2:])
                j += 1
            out.append(
                "<blockquote>" + render_inline(" ".join(qlines)) + "</blockquote>"
            )
            i = j
            continue

        # Paragraph — collect consecutive non-special lines.
        close_list()
        plines = [stripped]
        j = i + 1
        while j < len(lines):
            nxt = lines[j].strip()
            if (not nxt
                    or nxt.startswith("#")
                    or nxt.startswith("- ")
                    or nxt.startswith("* ")
                    or nxt.startswith("> ")
                    or nxt in ("---", "***", "___")):
                break
            plines.append(nxt)
            j += 1
        out.append(f"<p>{render_inline(' '.join(plines))}</p>")
        i = j

    close_list()
    return "".join(out)


def _transcript_to_turns_html(transcript_md: str) -> str:
    """Convert `[MM:SS] Speaker: text` lines into `.turn` blocks that
    match the design CSS."""
    out: list[str] = []
    for raw in transcript_md.splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or line == "## Transcript":
            continue
        # Expected shape: [12:34] Speaker: text
        if not line.startswith("["):
            continue
        try:
            close_bracket = line.index("]")
            ts = line[1:close_bracket]
            rest = line[close_bracket + 1:].lstrip()
            if ":" not in rest:
                continue
            who, said = rest.split(":", 1)
            out.append(
                '<div class="turn">'
                f'<div class="ts" title="Jump to this moment">{_html.escape(ts)}</div>'
                '<div class="speech">'
                f'<div class="who">{_html.escape(who.strip())}</div>'
                f'<div class="said">{_html.escape(said.strip())}</div>'
                "</div>"
                "</div>"
            )
        except (ValueError, IndexError):
            continue
    if not out:
        return '<p class="lead">No transcript content.</p>'
    return "".join(out)


# ── Transcripts (dictation history) ───────────────────────────────────


def _transcripts_dir() -> Path:
    """Same path the dictation store writes to (src/store.py)."""
    return Path.home() / "Library" / "Application Support" / "Lumo" / "transcripts"


_BADGE_FOR_LEVEL = {
    "raw": "Raw",
    "light": "Light",
    "standard": "Standard",
    "polished": "Polished",
}


def _fmt_transcript_date(iso: str) -> str:
    """Render `2026-06-07T02:10:19` as `Jun 7, 02:10` — same style the
    design uses for the card date footer."""
    try:
        d = datetime.fromisoformat(iso)
        return d.strftime("%b %-d, %H:%M")
    except Exception:
        return iso or "—"


def list_transcripts(args: dict) -> list[dict]:
    """Return all dictation transcripts newest-first. Optional filter:
       args.level — 'raw' | 'light' | 'standard' | 'polished' | 'all'."""
    idx = _transcripts_dir() / "index.json"
    if not idx.exists():
        return []
    try:
        entries = json.loads(idx.read_text(encoding="utf-8"))
    except Exception:
        return []
    level_filter = (args or {}).get("level") or "all"
    if not isinstance(entries, list):
        return []
    out: list[dict] = []
    for e in entries:
        if not isinstance(e, dict):
            continue
        cleanup = (e.get("cleanup") or "raw").lower()
        if level_filter != "all" and cleanup != level_filter:
            continue
        text = e.get("text") or e.get("text_raw") or ""
        out.append({
            "id": e.get("id") or "",
            "text": text,
            "date": _fmt_transcript_date(e.get("ts") or ""),
            "cleanup": cleanup,
            "badge_label": _BADGE_FOR_LEVEL.get(cleanup, "Raw"),
            "lang": e.get("lang") or "",
        })
    # index.json is already stored newest-first (src/store.py prepends),
    # so no reverse — preserve the on-disk order.
    return out


# ── Account ────────────────────────────────────────────────────────────


def get_account_info(_args: dict) -> dict:
    """Surface the bits of license + session needed by the Account pane:
    email, plan name, valid-until window, this device's name. None of
    this leaves the machine — it's just splicing local state into the
    web view."""
    # CR (2026-07-01) — the source of truth for "is this Mac signed in"
    # is session.json (bearer token). license.json holds the last cert
    # we saw, which after sign_out becomes stale — email + plan are
    # still on disk from a previous activation, so a naive read here
    # showed the Account pane as logged-in with the old email even
    # though sign_out had just cleared the session. Now we early-out
    # to `signed_in=False` when session.json is missing.
    sess_path = (Path.home() / "Library" / "Application Support"
                 / "Lumo" / "session.json")
    signed_in = sess_path.exists()
    info: dict = {
        "email": "",
        "plan": "free",
        "plan_label": "Free",
        "valid_until_iso": "",
        "valid_until_human": "",
        "device_name": "this Mac",
        "active": False,
        "signed_in": signed_in,
        "devices": [],          # CR (2026-06-23) — list of {machine_id, name, last_seen, ...}
        "device_cap": 0,        # capacity from plan
        "this_machine_id": "",  # used by JS to badge the row that's "this Mac"
    }
    if not signed_in:
        return info
    try:
        from . import licensing
        lic = licensing.load_license()
        info["email"] = lic.email or ""
        info["plan"] = lic.plan
        info["plan_label"] = {
            "free": "Free",
            "lite": "Lite",
            "plus": "Plus",
            "pro": "Pro",
        }.get(lic.plan, lic.plan.title())
        if lic.subscription_valid_until:
            info["valid_until_iso"] = lic.subscription_valid_until.isoformat()
            info["valid_until_human"] = lic.subscription_valid_until.strftime(
                "%b %-d, %Y"
            )
        # LicenseState.is_active is a @property — calling it as a method
        # was raising TypeError, swallowed by the outer try, so
        # info["active"] silently stayed False forever. CR #7.
        info["active"] = bool(lic.is_active)
        info["device_name"] = licensing.machine_name()
        try:
            info["this_machine_id"] = licensing.machine_id()
        except Exception:
            pass
    except Exception as e:
        info["error"] = f"{type(e).__name__}: {e}"
    # CR (2026-06-23) — pull the authoritative device list from the
    # license server's /me. Failure is non-fatal: the Account pane
    # falls back to a single "this Mac" row, which is what we used to
    # show unconditionally.
    try:
        from . import licensing as _lic
        ok, payload = _lic.fetch_account_state()
        if ok and isinstance(payload, dict):
            devs = payload.get("devices") or []
            if isinstance(devs, list):
                info["devices"] = devs
            cap = payload.get("device_cap")
            if isinstance(cap, int):
                info["device_cap"] = cap
    except Exception as e:
        print(f"[web_handlers] fetch_account_state failed (non-fatal): {e}",
              flush=True)
    # Session is the auth state — show it as the email fallback if license
    # has no email yet (i.e. activation pending).
    if not info["email"]:
        try:
            sess_path = (Path.home() / "Library" / "Application Support"
                         / "Lumo" / "session.json")
            sess = json.loads(sess_path.read_text(encoding="utf-8"))
            info["email"] = sess.get("email", "")
        except Exception:
            pass
    return info


def sign_out_everywhere(_args: dict) -> dict:
    """CR (2026-07-02) — POST /me/sign-out-everywhere so every session
    and every device row for this email is killed server-side, then
    delete local session.json so THIS Mac lands back on the activation
    flow. Other Macs will hit 401 next time they call /me and drop
    into the sign-in state (v0.1.22 shows this correctly)."""
    sess_path = (Path.home() / "Library" / "Application Support"
                 / "Lumo" / "session.json")
    try:
        if not sess_path.exists():
            return {"ok": False, "error": "not signed in"}
        try:
            sess = json.loads(sess_path.read_text(encoding="utf-8"))
            token = sess.get("token") or ""
        except Exception:
            token = ""
        if not token:
            return {"ok": False, "error": "session token missing"}
        try:
            import urllib.request
            from . import licensing as _lic
            req = urllib.request.Request(
                f"{_lic.LICENSE_SERVER_URL.rstrip('/')}/me/sign-out-everywhere",
                data=b"",
                headers={
                    "Authorization": f"Bearer {token}",
                    "Content-Type": "application/json",
                },
                method="POST",
            )
            with urllib.request.urlopen(req, timeout=10) as resp:
                body = json.loads(resp.read().decode("utf-8"))
        except Exception as e:  # noqa: BLE001
            return {"ok": False, "error": f"server: {e}"}
        sess_path.unlink()
        return {"ok": True, **body}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def remove_device(args: dict) -> dict:
    """CR (2026-07-02) — DELETE /me/devices/{machine_id}. Used by the
    per-row Remove button in Account → Devices. Doesn't touch local
    session.json — this is for de-authorising a DIFFERENT Mac from
    the list, not for signing out the current one."""
    mid = ((args or {}).get("machine_id") or "").strip()
    if not mid:
        return {"ok": False, "error": "machine_id required"}
    sess_path = (Path.home() / "Library" / "Application Support"
                 / "Lumo" / "session.json")
    try:
        if not sess_path.exists():
            return {"ok": False, "error": "not signed in"}
        try:
            sess = json.loads(sess_path.read_text(encoding="utf-8"))
            token = sess.get("token") or ""
        except Exception:
            token = ""
        if not token:
            return {"ok": False, "error": "session token missing"}
        try:
            import urllib.request
            from . import licensing as _lic
            req = urllib.request.Request(
                f"{_lic.LICENSE_SERVER_URL.rstrip('/')}/me/devices/{mid}",
                headers={"Authorization": f"Bearer {token}"},
                method="DELETE",
            )
            with urllib.request.urlopen(req, timeout=10) as resp:
                resp.read()
            return {"ok": True}
        except Exception as e:  # noqa: BLE001
            return {"ok": False, "error": f"server: {e}"}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def sign_out(_args: dict) -> dict:
    """Sign out THIS Mac only.
       1. POST /auth/logout to the license server so the bearer token
          is invalidated server-side too — without this we leave a
          "zombie" session that's gone locally but still valid on the
          backend, which broke device-list consistency across Macs.
       2. Delete local session.json so next launch shows activation.
       Does NOT touch transcripts/meetings/audio cache, license cert,
       or anything on the OTHER devices the user is signed in on.
       CR (2026-06-28)."""
    sess_path = (Path.home() / "Library" / "Application Support"
                 / "Lumo" / "session.json")
    server_logout_ok = True
    server_logout_err = None
    try:
        if sess_path.exists():
            try:
                sess = json.loads(sess_path.read_text(encoding="utf-8"))
                token = sess.get("token") or ""
            except Exception:
                token = ""
            if token:
                try:
                    import urllib.request
                    import urllib.error
                    from . import licensing as _lic
                    req = urllib.request.Request(
                        f"{_lic.LICENSE_SERVER_URL.rstrip('/')}/auth/logout",
                        data=b"",
                        headers={
                            "Authorization": f"Bearer {token}",
                            "Content-Type": "application/json",
                        },
                        method="POST",
                    )
                    with urllib.request.urlopen(req, timeout=5) as resp:
                        resp.read()
                except Exception as e:  # noqa: BLE001
                    server_logout_ok = False
                    server_logout_err = f"{type(e).__name__}: {e}"
                    print(f"[web_handlers] /auth/logout failed (continuing "
                          f"with local sign-out): {e}", flush=True)
            sess_path.unlink()
        return {"ok": True, "server_logout_ok": server_logout_ok,
                "server_logout_err": server_logout_err}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


# ── Settings ───────────────────────────────────────────────────────────


_SETTINGS_SCHEMA = {
    # path-in-config.json  →  allowed values (list = enum, None = freeform)
    "cleanup.level": ["raw", "light", "standard", "polished"],
    "hotkey.mode":   ["right_option", "left_option", "fn"],
    "whisper.model": None,
    "whisper.language": None,
    "ui.show_overlay": [True, False],
    "ui.show_notification_on_paste": [True, False],
    "dictation.hands_free_pause_ms": [1200, 2000, 3500],
    # CR (2026-06-23) — user-facing toggle for cloud vs local dictation.
    # True = cloud-first with local fallback (default; faster, network-bound).
    # False = stay local always (offline, slower on first run, mic never leaves).
    "dictation.cloud_mode": [True, False],
}


def _config_path() -> Path:
    """User-config path (NOT the bundled defaults). CR #3 — writes go to
    Application Support; reads use load_config() so they see the merged
    view (defaults + user overrides + type-stripped comments)."""
    from .app import user_config_path
    return user_config_path()


def _read_config() -> dict:
    """Always return the merged defaults+user view so update_setting
    operates on the same dict structure load_config() produces at boot."""
    try:
        from .app import load_config
        return load_config()
    except Exception:
        return {}


def _dig(d: dict, dotted: str):
    cur: Any = d
    for part in dotted.split("."):
        if not isinstance(cur, dict):
            return None
        cur = cur.get(part)
    return cur


def get_settings(_args: dict) -> dict:
    cfg = _read_config()
    return {
        "cleanup_level":   _dig(cfg, "cleanup.level") or "raw",
        "hotkey_mode":     _dig(cfg, "hotkey.mode") or "right_option",
        "whisper_model":   _dig(cfg, "whisper.model") or "",
        "whisper_language": _dig(cfg, "whisper.language") or "auto",
        "show_overlay":    bool(_dig(cfg, "ui.show_overlay")),
        "hands_free_pause_ms": int(_dig(cfg, "dictation.hands_free_pause_ms") or 2000),
        # When the key is absent we default to True (cloud-first) since
        # that's the shipped default for fresh installs; users who never
        # touched the toggle still see "On" in the UI.
        "dictation_cloud_mode":
            True if _dig(cfg, "dictation.cloud_mode") is None
            else bool(_dig(cfg, "dictation.cloud_mode")),
    }


def update_setting(args: dict) -> dict:
    """Write a single setting into config.json. Some keys (hotkey/whisper)
    require an app restart to take effect; the UI surfaces that hint —
    we don't bounce the process automatically because that would lose
    any active recording."""
    key = (args or {}).get("key") or ""
    value = (args or {}).get("value")
    if key not in _SETTINGS_SCHEMA:
        return {"ok": False, "error": f"unknown setting: {key}"}
    allowed = _SETTINGS_SCHEMA[key]
    if isinstance(allowed, list):
        # If the allowed list is all-int, coerce numeric strings — JS seg
        # buttons round-trip values through dataset which stringifies.
        if all(isinstance(a, int) and not isinstance(a, bool) for a in allowed):
            try:
                value = int(value)
            except (TypeError, ValueError):
                pass
        # CR (2026-06-23) — same shape, but for bool enums. JS dataset
        # values arrive as the literal strings "true"/"false"; without
        # coercion `dictation.cloud_mode` clicks would always reject.
        elif all(isinstance(a, bool) for a in allowed):
            if isinstance(value, str):
                lower = value.strip().lower()
                if lower == "true":
                    value = True
                elif lower == "false":
                    value = False
        if value not in allowed:
            return {"ok": False, "error": f"invalid value: {value}"}
    else:
        # CR #15 — freeform string keys (whisper.model/whisper.language)
        # used to accept anything JS sent, including a number or a list,
        # which then crashed make_transcriber on next launch. Lock these
        # down to strings here so a UI bug can't brick the app.
        if not isinstance(value, str) or not value.strip():
            return {"ok": False, "error":
                    f"{key} must be a non-empty string, got {type(value).__name__}"}

    # CR #3 — read ONLY the existing user-config (not the merged view) so
    # the write doesn't silently persist every bundled default into the
    # user file (which would defeat the point of layering — future bundle
    # updates would no longer be visible through them).
    cfg_path = _config_path()
    user_cfg: dict = {}
    if cfg_path.exists():
        try:
            user_cfg = json.loads(cfg_path.read_text(encoding="utf-8"))
            if not isinstance(user_cfg, dict):
                user_cfg = {}
        except Exception:
            user_cfg = {}

    parts = key.split(".")
    cur = user_cfg
    for p in parts[:-1]:
        if p not in cur or not isinstance(cur[p], dict):
            cur[p] = {}
        cur = cur[p]
    cur[parts[-1]] = value

    # Atomic write via mkstemp+os.replace — same pattern store._index_save
    # uses. A crash / disk-full mid-write can no longer leave a truncated
    # config.json that bricks the next launch.
    import os as _os
    import tempfile as _tf
    try:
        cfg_path.parent.mkdir(parents=True, exist_ok=True)
        fd, tmp = _tf.mkstemp(dir=str(cfg_path.parent),
                              prefix=".config.", suffix=".json")
        try:
            with _os.fdopen(fd, "w", encoding="utf-8") as f:
                json.dump(user_cfg, f, indent=2, ensure_ascii=False)
            _os.replace(tmp, str(cfg_path))
        except Exception:
            if _os.path.exists(tmp):
                _os.unlink(tmp)
            raise
        try:
            cfg_path.chmod(0o600)
        except OSError:
            pass
        return {
            "ok": True,
            "needs_restart": key in ("hotkey.mode", "whisper.model",
                                     "whisper.language"),
        }
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def get_summary_consent(_args: dict) -> dict:
    """Audit #5. Returns the current state of the off-device summary
    consent flag so the bank UI can decide whether to show the modal."""
    p = (Path.home() / "Library" / "Application Support" / "Lumo"
         / "summary_consent.json")
    if not p.exists():
        return {"granted": None}   # None = never asked
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
        return {"granted": bool(data.get("granted")), "at": data.get("at")}
    except Exception:
        return {"granted": None}


def set_summary_consent(args: dict) -> dict:
    """Bank UI writes the user's decision here. `granted` must be a real
    bool to prevent JS from accidentally tri-stating us back to 'never
    asked'."""
    granted = (args or {}).get("granted")
    if granted is not True and granted is not False:
        return {"ok": False, "error": "granted must be true or false"}
    from datetime import datetime as _dt
    p = (Path.home() / "Library" / "Application Support" / "Lumo"
         / "summary_consent.json")
    try:
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(json.dumps({
            "granted": bool(granted),
            "at": _dt.now().isoformat(timespec="seconds"),
        }, ensure_ascii=False, indent=2), encoding="utf-8")
        try:
            p.chmod(0o600)
        except OSError:
            pass
        return {"ok": True, "granted": bool(granted)}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def get_user_cleanup_level(_args: dict) -> str:
    """Return the user's configured default cleanup level (raw/light/
    standard/polished). Reads through the merged-view loader (CR #3) so
    bundled defaults are honoured when the user-config doesn't override."""
    try:
        cfg = _read_config()
        level = (cfg.get("cleanup") or {}).get("level") or "raw"
        if level in ("raw", "light", "standard", "polished"):
            return level
    except Exception:
        pass
    return "raw"


def retry_summaries_for_awaiting(_args: dict) -> dict:
    """CR #8 — after the user enables cloud summaries, kick a summary
    pass on every meeting whose status is 'awaiting_summary_consent'.
    Returns {ok, kicked} so the UI can confirm count."""
    root = _meetings_root()
    if not root.exists():
        return {"ok": True, "kicked": 0}
    kicked = 0
    try:
        for d in sorted(root.iterdir(), reverse=True):
            if not d.is_dir():
                continue
            try:
                meta = json.loads((d / "meta.json").read_text(encoding="utf-8"))
            except Exception:
                continue
            if meta.get("status") != "awaiting_summary_consent":
                continue
            try:
                # Demote status so summarize_meeting doesn't short-circuit.
                meta["status"] = "transcribed"
                meta.pop("summary_skipped_reason", None)
                (d / "meta.json").write_text(
                    json.dumps(meta, ensure_ascii=False, indent=2),
                    encoding="utf-8")
                from .meeting_summarize import summarize_meeting_async
                summarize_meeting_async(d, on_done=None)
                kicked += 1
            except Exception as e:
                print(f"[summarize] retry-awaiting failed for {d.name}: {e}",
                      flush=True)
    except OSError:
        pass
    return {"ok": True, "kicked": kicked}


def retry_meeting_transcribe(args: dict) -> dict:
    """Kick the transcribe pipeline for a meeting whose transcript is
    missing/empty. Useful for old meetings that finished with
    status='ready_for_transcription' when live transcription wasn't
    available, or for ones that crashed mid-way. Returns {ok, kicked}.

    Path: load LirelleApp.transcriber via the recently-bridged app handle,
    then dispatch transcribe_meeting_async on the recovered meeting dir.
    Idempotent — if transcript.md already has content, we skip."""
    mdir = _safe_meeting_dir((args or {}).get("id"))
    if mdir is None:
        return {"ok": False, "error": "invalid id"}
    transcript_path = mdir / "transcript.md"
    if transcript_path.exists():
        # Don't overwrite if there's real content. The marker "## Transcript"
        # header is just preamble; check for actual lines below it.
        try:
            content = transcript_path.read_text(encoding="utf-8")
            has_lines = any(
                ln.strip().startswith("[") for ln in content.splitlines()
            )
            if has_lines:
                return {"ok": False, "error": "transcript already exists"}
        except Exception:
            pass
    # Reset status so the transcribe path picks it up.
    try:
        meta_path = mdir / "meta.json"
        meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.exists() else {}
        meta["status"] = "ready_for_transcription"
        meta.pop("summary_skipped_reason", None)
        meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2),
                             encoding="utf-8")
    except Exception:
        pass
    # Run the transcribe + summary pipeline. The transcriber comes from
    # the live LirelleApp instance, registered into web_handlers at bridge
    # setup time (bank.py passes a provider lambda).
    try:
        tr = _live_transcriber_provider() if _live_transcriber_provider else None
        if tr is None:
            return {"ok": False, "error": "transcriber not ready (try again in a few seconds)"}
        from .meeting_transcribe import transcribe_meeting_async
        # on_done can chain into summary kickoff; existing
        # MeetingRecorder.on_transcribed handles that path but isn't
        # wired here. Kick summary explicitly after transcribe finishes.
        def _then_summarize(_path):
            try:
                from .meeting_summarize import summarize_meeting_async
                summarize_meeting_async(mdir, on_done=None)
            except Exception as e:
                print(f"[web_handlers] summary kickoff after transcribe failed: {e}",
                      flush=True)
        transcribe_meeting_async(mdir, tr, on_done=_then_summarize)
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def update_meeting_title(args: dict) -> dict:
    """Save a user-supplied title for a meeting. Stored in
    meta.json:title_override; list_meetings + get_meeting prefer this
    over the heading-from-summary fallback. Empty string clears the
    override (back to auto-derived title)."""
    mdir = _safe_meeting_dir((args or {}).get("id"))
    if mdir is None:
        return {"ok": False, "error": "invalid id"}
    title = ((args or {}).get("title") or "").strip()
    try:
        meta_path = mdir / "meta.json"
        meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.exists() else {}
        if title:
            meta["title_override"] = title
        else:
            meta.pop("title_override", None)
        meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2),
                             encoding="utf-8")
        return {"ok": True, "title": title}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


# Bridge handle for the live transcriber. bank.py sets this at WebShell
# init so retry_meeting_transcribe can reach the warmed-up MLX instance
# without re-importing src.app (which would re-load the whole module).
_live_transcriber_provider: Callable[[], object] | None = None


def set_live_transcriber_provider(fn: Callable[[], object]) -> None:
    global _live_transcriber_provider
    _live_transcriber_provider = fn


# CR (2026-06-27) — open an external URL in the user's default browser.
# Used by Account → Manage subscription / Upgrade buttons to send the
# user to https://lirelle.us/cabinet for plan management + payment.
# Allow-list scheme=https only so a misbehaving page can't fire
# `file://` or `app-bundle:` URLs through the bridge.
def open_external_url(args: dict) -> dict:
    url = (args or {}).get("url") or ""
    if not url.startswith("https://"):
        return {"ok": False, "error": "only https URLs allowed"}
    try:
        from Foundation import NSURL          # type: ignore
        from AppKit import NSWorkspace        # type: ignore
        from PyObjCTools import AppHelper     # type: ignore
        ns_url = NSURL.URLWithString_(url)
        if ns_url is None:
            return {"ok": False, "error": "invalid URL"}
        # CR (2026-06-28) — AppKit/Foundation are NOT thread-safe and
        # PyObjC class lookups can crash with NSClassicMapTable race
        # under JS-bridge dispatch on background threads (saw this in
        # the v0.1.16 crash log). Force the actual openURL_ onto the
        # main thread so the bridge handler returns immediately and
        # the AppKit call happens where Cocoa expects it.
        AppHelper.callAfter(
            lambda: NSWorkspace.sharedWorkspace().openURL_(ns_url))
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


# CR (2026-06-23) — Settings pane "Offline Model" surface.
# `get_transcriber_status` reports the currently-active backend (cloud vs.
# local kind) + whether the local model is on disk, so the UI can show
# "Active: Cloud" / "Offline model: not downloaded — [Download]".
# `download_offline_model` kicks `ensure_model_ready` on a background
# thread + reports progress via a bridge push event.

_offline_download_thread = None  # type: ignore[var-annotated]
_offline_download_state = {"running": False, "done": False, "error": None,
                           "bytes_done": 0, "bytes_total": 0, "msg": ""}


def get_transcriber_status(_args: dict) -> dict:
    """Snapshot for the Settings UI:
       backend = "cloud" | "mlx" | "whisper_cpp" | "unknown"
       cloud_available = True if cloud_dictator is importable + bundled
       local_ready = True if the offline whisper model is on disk
       model_repo = HF repo id (mlx) or weights path (whisper.cpp)
    The "active" backend is whatever the running dictation transcriber
    instance is — that's the only authoritative answer because users can
    flip dictation.cloud_mode in config.json mid-session."""
    out = {
        "backend": "unknown",
        "active_label": "Unknown",
        "cloud_available": False,
        "local_ready": False,
        "model_repo": "",
        "downloading": bool(_offline_download_state.get("running")),
        "download_progress": dict(_offline_download_state),
    }
    try:
        tr = _live_transcriber_provider() if _live_transcriber_provider else None
        active_is_local = False
        if tr is not None:
            cls = type(tr).__name__
            if cls == "CloudDictator":
                out["backend"] = "cloud"
                out["active_label"] = "Cloud"
            elif cls == "MLXTranscriber":
                out["backend"] = "mlx"
                out["active_label"] = "Local — MLX Whisper"
                active_is_local = True
            elif cls == "WhisperCppTranscriber":
                out["backend"] = "whisper_cpp"
                out["active_label"] = "Local — whisper.cpp"
                active_is_local = True
            # Pull model_repo / model_path off the active instance when it's
            # a real local backend; for CloudDictator both attrs are absent
            # (no model to point at).
            if active_is_local:
                for attr in ("model_repo", "model_path"):
                    v = getattr(tr, attr, None)
                    if v:
                        out["model_repo"] = str(v)
                        break
                try:
                    out["local_ready"] = bool(tr.is_model_ready())
                except Exception:
                    pass
        # CR (2026-06-23) — when Cloud Mode is active, the live transcriber
        # is CloudDictator whose `is_model_ready()` is True by definition
        # (nothing to download). Probing it would lie: Settings would show
        # "Offline Whisper model: ready on disk" on a machine that has
        # never downloaded Whisper. So when active is Cloud (or there's
        # no active instance at all), build a fresh LOCAL transcriber off
        # config.json and ask IT whether the model is cached.
        if not active_is_local:
            try:
                from .app import load_config as _load_config
                from . import transcriber as _trx
                cfg = _load_config()
                local_probe = _trx.make_transcriber(cfg.get("whisper") or {})
                out["local_ready"] = bool(local_probe.is_model_ready())
                for attr in ("model_repo", "model_path"):
                    v = getattr(local_probe, attr, None)
                    if v:
                        out["model_repo"] = str(v)
                        break
            except Exception as e:
                print(f"[web_handlers] local-model probe failed: {e}",
                      flush=True)
                out["local_ready"] = False
    except Exception as e:
        out["error"] = f"{type(e).__name__}: {e}"
    try:
        from . import cloud_dictator  # noqa: F401
        out["cloud_available"] = True
    except Exception:
        pass
    return out


def download_offline_model(_args: dict) -> dict:
    """Kick `ensure_model_ready()` on the currently-built local transcriber
    in a background thread. UI polls via `requestTranscriberStatus`. We
    don't push progress events from inside huggingface_hub (their stream
    is its own thread) — JS polls every 1s while `downloading == True`.

    If the user is in Cloud Mode, the active transcriber is the cloud
    dictator (which has no model to download) — in that case we fall
    back to building a fresh local instance off config.json so the
    user can download the offline backup while still using Cloud."""
    global _offline_download_thread, _offline_download_state
    if _offline_download_state.get("running"):
        return {"ok": False, "error": "already running"}

    def _resolve_local_transcriber():
        # Active instance is the simplest path — works for both MLX and
        # whisper.cpp when Cloud Mode is off.
        tr = _live_transcriber_provider() if _live_transcriber_provider else None
        if tr is not None and type(tr).__name__ != "CloudDictator":
            return tr
        # Cloud Mode is on (or no instance) → build a fresh local one off
        # config.json so the user can pre-cache the offline fallback.
        try:
            from .app import load_config as _load_config
            from . import transcriber as _trx
            cfg = _load_config()
            return _trx.make_transcriber(cfg.get("whisper") or {})
        except Exception as e:
            print(f"[web_handlers] couldn't build local transcriber: {e}",
                  flush=True)
            return None

    target = _resolve_local_transcriber()
    if target is None:
        return {"ok": False, "error": "no local transcriber available"}

    def _progress_cb(done_bytes, total_bytes, msg):
        _offline_download_state["bytes_done"] = int(done_bytes or 0)
        _offline_download_state["bytes_total"] = int(total_bytes or 0)
        _offline_download_state["msg"] = str(msg or "")

    def _worker():
        global _offline_download_state
        _offline_download_state.update({
            "running": True, "done": False, "error": None,
            "bytes_done": 0, "bytes_total": 0, "msg": "Starting…",
        })
        try:
            target.ensure_model_ready(_progress_cb)
            _offline_download_state.update({
                "running": False, "done": True,
                "msg": "Model ready",
            })
        except Exception as e:  # noqa: BLE001
            _offline_download_state.update({
                "running": False, "done": False,
                "error": f"{type(e).__name__}: {e}",
                "msg": "Download failed",
            })
            print(f"[web_handlers] offline model download failed: {e}",
                  flush=True)

    import threading as _t
    _offline_download_thread = _t.Thread(target=_worker, daemon=True)
    _offline_download_thread.start()
    return {"ok": True}


def retry_meeting_summary(args: dict) -> dict:
    """CR #14 — kick summarize_meeting again for a meeting that failed
    last time. Idempotent: if summary.md already exists, the inner
    function short-circuits."""
    mdir = _safe_meeting_dir((args or {}).get("id"))
    if mdir is None:
        return {"ok": False, "error": "invalid id"}
    # Clear the failure flag so summarize_meeting doesn't short-circuit
    # on the "already failed" status; the upstream function rewrites
    # status either way.
    try:
        meta_path = mdir / "meta.json"
        meta = json.loads(meta_path.read_text(encoding="utf-8"))
        if meta.get("status") == "summary_failed":
            meta["status"] = "transcribed"
            meta.pop("summary_error", None)
            meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2),
                                 encoding="utf-8")
    except Exception:
        pass
    try:
        from .meeting_summarize import summarize_meeting_async
        summarize_meeting_async(mdir, on_done=None)
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def copy_transcript_to_clipboard(args: dict) -> dict:
    """Copy a single transcript's text to NSPasteboard. Used by the
    Transcripts pane's copy button. Returns {ok, error?}."""
    text = (args or {}).get("text") or ""
    if not text:
        # The JS sends the text directly so we don't have to look it up;
        # if it's empty the user clicked a card with no text — no-op.
        return {"ok": False, "error": "empty"}
    try:
        from AppKit import NSPasteboard, NSPasteboardTypeString  # type: ignore
        pb = NSPasteboard.generalPasteboard()
        pb.clearContents()
        pb.setString_forType_(str(text), NSPasteboardTypeString)
        return {"ok": True}
    except Exception as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}


def get_meeting(args: dict) -> dict | None:
    """Full payload for a single meeting — title, meta, summary (as HTML),
    transcript (as HTML turns). Returned dict matches what the Meetings
    page's JS data binder will splice into the design template."""
    mdir = _safe_meeting_dir(args.get("id"))
    if mdir is None:
        return None
    meta = _read_meta(mdir)
    title = (meta.get("title_override")
             or _title_from_summary(mdir)
             or "Untitled meeting")
    summary_md = ""
    transcript_md = ""
    try:
        summary_md = (mdir / "summary.md").read_text(encoding="utf-8")
    except OSError:
        pass
    try:
        transcript_md = (mdir / "transcript.md").read_text(encoding="utf-8")
    except OSError:
        pass

    # If summary was skipped because the meeting was too short, show a
    # friendly placeholder instead of an empty Summary tab.
    summary_html = _md_to_html(summary_md) if summary_md.strip() else ""
    status_raw = meta.get("status")
    if not summary_html and status_raw == "skipped_too_short":
        reason = meta.get("summary_skipped_reason") or "meeting was too short"
        summary_html = (
            f'<p class="lead">No summary — {_html.escape(str(reason))}.</p>'
            '<p class="lead" style="margin-top:6px;">The full transcript is '
            'on the Transcript tab.</p>'
        )
    # Audit #5 — explicit "consent not granted yet" message + jump-to-Settings link.
    if not summary_html and status_raw == "awaiting_summary_consent":
        summary_html = (
            '<p class="lead">No summary yet — cloud summaries are turned off.</p>'
            '<p class="lead" style="margin-top:6px;">Enable them in '
            '<a href="#" data-jump-settings="cloud-summary">Settings</a> '
            'if you want this meeting summarised. The audio + transcript '
            'stay on this Mac either way.</p>'
        )
    # CR #14 — surface failure with a Retry button (wired by meetings-bind.js).
    if not summary_html and status_raw == "summary_failed":
        err = meta.get("summary_error") or "unknown error"
        summary_html = (
            f'<p class="lead">Summary failed — {_html.escape(str(err))}.</p>'
            '<p class="lead" style="margin-top:6px;">'
            '<button class="btn btn-primary" data-action="retry-summary">'
            'Retry summary</button></p>'
        )

    # Transcript-retry affordance: if transcript is missing/empty
    # (status implies it never landed), surface a button.
    transcript_html = (
        _transcript_to_turns_html(transcript_md) if transcript_md.strip() else ""
    )
    # Detect "no actual content" — the transcript_to_turns_html fallback
    # ends up as "No transcript content." when no [MM:SS] lines parse.
    transcript_has_content = (
        transcript_md.strip()
        and any(ln.strip().startswith("[")
                for ln in transcript_md.splitlines())
    )
    if not transcript_has_content:
        transcript_html = (
            '<p class="lead">Transcript is missing — the recording '
            'finished without a transcript pass.</p>'
            '<p class="lead" style="margin-top:6px;">'
            '<button class="btn btn-primary" data-action="retry-transcribe">'
            'Generate transcript now</button></p>'
        )

    return {
        "id": mdir.name,
        "title": title,
        "date": _fmt_date(meta.get("started_at")),
        "dur": _fmt_dur(meta.get("duration_seconds")),
        "status": _status_for_ui(meta.get("status", "")),
        "summary_html": summary_html,
        "transcript_html": transcript_html,
        "folder": str(mdir),
    }
