#!/usr/bin/env python3
"""summon — Claude Desktop session toolbox: cross-account transfer + recover/rebind/doctor.

Usage:   summon [MODE] [ID] [OPTIONS]
Input:   optional MODE positional (rebind|pick|recover|doctor; omit for transfer),
         ID = sessionId/cliSessionId prefix for rebind/recover; picker reads stdin
Output:  transfer/pick/doctor render TTY panels; recover/pick emit a paste-ready
         handover on stdout — a Sonnet-distilled brief (Goal / What landed /
         Unfinished / Open decisions / Key context) + transcript pointer, cached
         at <transcript>.handover.md; falls back to the plain pointer prompt when
         the `claude` CLI is unavailable or --no-distill is set. doctor --json
         emits {"data": [...], "meta": {"schema": "claude-mods.summon.doctor/v1"}};
         pick --json emits the session inventory as
         {"data": [...], "meta": {"schema": "claude-mods.summon.pick/v1"}};
         widget emits FINISHED, self-contained card-picker HTML on stdout (pass
         it straight to show_widget) — the rich inventory pre-trimmed and
         injected into assets/picker-widget.html under a hard byte budget, one
         session object per line; a copy is written to --out (default
         <temp>/claude/summon-widget.html)
Stderr:  context panels for recover/pick, distillation progress, warnings, errors
Exit:    0 ok (including the non-distilled fallback — worker unavailability is
         advisory, never fatal), 2 usage/ambiguous id, 3 session or path not
         found, 10 doctor found broken sessions

Examples:
  summon --to other-account                   # transfer: push sessions to next account
  summon pick                                 # fzf/numbered picker -> distilled handover
  summon pick --json | jq '.data[]'           # machine-readable session inventory
  summon widget --days 30                      # finished in-chat card-picker HTML on stdout
  summon recover 6577b24c                     # distilled handover brief for one session
  summon recover 6577b24c --refresh           # ignore cached brief, re-distill
  summon recover 6577b24c --no-distill        # plain pointer prompt, no LLM call
  summon recover 6577b24c --model haiku       # distill with a different model
  summon rebind 6577b24c --cwd D:\\archive\\myapp\\.claude\\worktrees\\funny-hypatia-5e54f7
  summon doctor                               # scan all sessions for broken cwd bindings
  summon doctor --json | jq '.data[]'

Transfer mode (no positional) is documented in SKILL.md: copy by default, --move
to relocate, destination auto-detected as the most-recently-active account.

Output rendering follows docs/TERMINAL-DESIGN.md (Terminal Panel Design System).

Deliberately single-file: skill scripts ship as self-contained portable units
(docs/SKILL-RESOURCE-PROTOCOL.md) — do not split into a package. Navigate by
the `# ===` section headers — Sections: DESIGN(term) · Path discovery ·
Account discovery · Sessions · Grouping · Listing · Picker · Workspace
selection · Operate · Peek · Transcript/Distill · Modes (transfer / pick /
recover / rebind / doctor) · Widget (card-picker builder) · CLI entry.
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import sys
import tempfile
import time
import uuid as uuidlib
from collections import OrderedDict
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Iterable


# ============================================================
#  DESIGN: terminal panel rendering (per docs/TERMINAL-DESIGN.md)
# ============================================================

def _stdout_supports_unicode() -> bool:
    enc = (getattr(sys.stdout, "encoding", "") or "").lower()
    return "utf" in enc or "cp65001" in enc


class Term:
    WIDTH = 80
    USE_ASCII = (
        os.environ.get("TERM_ASCII") == "1"
        or os.environ.get("TERM") == "dumb"
        or not _stdout_supports_unicode()
    )
    USE_COLOR = (
        sys.stdout.isatty()
        and os.environ.get("NO_COLOR") is None
        and os.environ.get("TERM") != "dumb"
        and os.environ.get("FORCE_COLOR") != "0"
    )

    @classmethod
    def g(cls, uni: str, asc: str) -> str:
        return asc if cls.USE_ASCII else uni

    @classmethod
    def color(cls, token: str, text: str) -> str:
        if not cls.USE_COLOR:
            return text
        codes = {
            "accent": "36",   # cyan
            "ok": "32",       # green
            "warn": "33",     # yellow
            "alarm": "31",    # red
            "tag": "35",      # magenta
            "meta": "2",      # dim
            "dim": "2",
            "default": "",
        }
        c = codes.get(token, "")
        if not c:
            return text
        return f"\033[{c}m{text}\033[0m"


_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")


def vlen(s: str) -> int:
    """Visible length, excluding ANSI escape codes."""
    return len(_ANSI_RE.sub("", s))


def trunc(s: str, width: int) -> str:
    """Truncate with ellipsis if too long."""
    if vlen(s) <= width:
        return s
    ell = Term.g("…", "...")
    return s[: width - vlen(ell)] + ell


# Brand emoji for summon (not in TERMINAL-DESIGN.md registry yet — registering here)
BRAND_EMOJI = "🪄"
BRAND_ASCII = "[S]"


def panel_open(name: str, indicator: str = "") -> str:
    """╭── 🪄 summon ─────────────────  indicator ───●"""
    em = Term.g(BRAND_EMOJI, BRAND_ASCII)
    tl = Term.g("╭", "+")
    h = Term.g("─", "-")
    term = Term.g("●", "*")
    left = f"{tl}{h}{h} {em} {Term.color('accent', name)} "
    if indicator:
        right = f" {Term.color('meta', indicator)} {h}{h}{h}{term}"
    else:
        right = f" {h}{h}{h}{term}"
    fill_count = max(2, Term.WIDTH - vlen(left) - vlen(right))
    return left + (h * fill_count) + right


def panel_close(hotkeys: list[tuple[str, str]] | None = None,
                healths: list[tuple[str, str]] | None = None) -> str:
    """╰── y confirm · n cancel ───── • 5 ready ───●"""
    bl = Term.g("╰", "+")
    h = Term.g("─", "-")
    term = Term.g("●", "*")
    bullet = Term.g("•", "(+)")
    hotkeys = hotkeys or []
    healths = healths or []

    sep = Term.g(" · ", " | ")
    hot_str = sep.join(f"{Term.color('accent', k)} {v}" for k, v in hotkeys)
    health_str = "  ".join(f"{Term.color(c, bullet)} {v}" for c, v in healths)

    left = f"{bl}{h}{h} {hot_str}" if hot_str else f"{bl}{h}{h}"
    if hot_str:
        left += " "
    right = f" {health_str} {h}{h}{h}{term}" if health_str else f" {h}{h}{h}{term}"
    fill = max(2, Term.WIDTH - vlen(left) - vlen(right))
    return left + (h * fill) + right


def panel_blank() -> str:
    return Term.g("│", "|")


def section(label: str, count: int = -1, color_token: str = "accent") -> str:
    """├── LABEL (count)"""
    tee = Term.g("├", "+")
    h = Term.g("─", "-")
    parens = f" ({count})" if count >= 0 else ""
    return f"{tee}{h}{h} {Term.color(color_token, label)}{Term.color('meta', parens)}"


def sub_section(label: str, count: int = -1, color_token: str = "default") -> str:
    """│   ├── LABEL (count)   — second-level grouping"""
    pipe = Term.g("│", "|")
    tee = Term.g("├", "+")
    h = Term.g("─", "-")
    parens = f" ({count})" if count >= 0 else ""
    return f"{pipe}   {tee}{h}{h} {Term.color(color_token, label)}{Term.color('meta', parens)}"


def sub_section_last(label: str, count: int = -1, color_token: str = "default") -> str:
    """│   └── LABEL (count)   — last sub-section"""
    pipe = Term.g("│", "|")
    corner = Term.g("└", "`")
    h = Term.g("─", "-")
    parens = f" ({count})" if count >= 0 else ""
    return f"{pipe}   {corner}{h}{h} {Term.color(color_token, label)}{Term.color('meta', parens)}"


def leaf(num: int, name: str, *, meta: str = "", age: str = "",
         last: bool = False, depth: int = 2,
         parent_last: bool = False,
         meta_color: str = "meta", age_color: str = "meta") -> str:
    """│   │   ├──  3. session-name        meta      age

    parent_last: when at depth 2 inside a last-sub-section, drop the inner
    pipe so it reads as siblings of the corner `└──` rather than continuing.
    """
    pipe = Term.g("│", "|")
    h = Term.g("─", "-")
    conn = Term.g("└", "`") if last else Term.g("├", "+")

    if depth == 1:
        prefix = f"{pipe}   "
    elif depth == 2:
        inner = "    " if parent_last else f"{pipe}   "
        prefix = f"{pipe}   {inner}"
    else:
        prefix = pipe + ("   " * depth)

    num_str = f"{num:>2}." if num else "   "
    name_field = trunc(name, 32).ljust(32)
    # Tight meta column — turn count only (e.g. "30t").
    meta_width = 8
    meta_visible = vlen(meta)
    if meta_visible <= meta_width:
        pad = " " * (meta_width - meta_visible)
        meta_field = Term.color(meta_color, meta) + pad
    else:
        meta_field = Term.color(meta_color, meta)
    age_field = Term.color(age_color, age).rjust(6) if age else " " * 6

    return f"{prefix}{conn}{h}{h} {num_str} {name_field}  {meta_field}  {age_field}"


def summary_line(text: str) -> str:
    """├── 4 lanes · 3 active   (dim)"""
    tee = Term.g("├", "+")
    h = Term.g("─", "-")
    return f"{tee}{h}{h} {Term.color('meta', text)}"


# Hint registry — each entry has a `when` predicate (over a context dict) and
# a `text` template (str.format-able). Predicates returning True make the hint
# eligible; one is picked at random.
HINTS: list[dict] = [
    # --- Conditional ---
    {
        "id": "density",
        "when": lambda c: c["count"] > 30,
        "text": "{count} sessions — narrow with --cwd <pat> or --title <pat>, "
                "or shorten the window with --1d/--3d/--7d",
    },
    {
        "id": "generic-titles",
        "when": lambda c: c["generic_count"] >= 3,
        "text": "{generic_count} sessions have generic titles (dev, general, untitled). "
                "Use `summon --peek <id>` to preview the last messages before pulling",
    },
    {
        "id": "default-window",
        "when": lambda c: c["window_days"] == 3 and c["count"] >= 5,
        "text": "default window is 3 days — `--all` to see everything, "
                "`--1d` for just today, `--7d` for a week, or `--days N` for custom",
    },
    {
        "id": "remote-skipped",
        "when": lambda c: c["remote_count"] > 0,
        "text": "{remote_count} remote-VM session(s) auto-skipped — they have no "
                "local transcript to bridge, so cross-account transfer isn't possible",
    },
    # --- Always-eligible (rotate as background tips) ---
    {
        "id": "peek",
        "when": lambda _: True,
        "text": "preview a session's last messages with `summon --peek <id>` — handy "
                "when titles like 'dev' don't tell you which one is which",
    },
    {
        "id": "copy-vs-move",
        "when": lambda _: True,
        "text": "default is copy (visible from both accounts) — pass `--move` "
                "to delete the source for lean cleanup",
    },
    {
        "id": "logout-login",
        "when": lambda _: True,
        "text": "Desktop only loads sessions at login — Cowork/Code toggle, Ctrl+R, "
                "and tab clicks won't rescan. Plan for Logout/Login when you switch",
    },
    {
        "id": "proactive",
        "when": lambda _: True,
        "text": "best run BEFORE switching accounts: copy sessions to the next "
                "account first, then Logout/Login (the switch you were doing anyway)",
    },
    {
        "id": "dry-run",
        "when": lambda _: True,
        "text": "`--dry-run` previews a move without touching files — pair it with "
                "`--pick` to rehearse the picker without committing",
    },
]


def _pick_hint(context: dict) -> str:
    """Pick one hint from HINTS whose predicate matches the context, or '' if none."""
    import random
    eligible = [h for h in HINTS if _hint_safe(h["when"], context)]
    if not eligible:
        return ""
    chosen = random.choice(eligible)
    try:
        return chosen["text"].format(**context)
    except (KeyError, ValueError):
        return chosen["text"]


def _hint_safe(predicate, context) -> bool:
    try:
        return bool(predicate(context))
    except Exception:
        return False


def hint(text: str, width: int = 70) -> str:
    """│   💡  text — tip riding the panel rail.

    Continuation lines wrap under the text, not under the icon, so the eye
    follows the message rather than re-finding column alignment.
    """
    pipe = Term.g("│", "|")
    bulb = Term.g("💡", "(i)")
    # Visual cells: pipe(1) + 3sp + bulb(2 if emoji, 3 if ASCII) + 2sp
    bulb_cells = 3 if Term.USE_ASCII else 2
    indent_after_pipe = 3 + bulb_cells + 2  # spaces between pipe and text
    cont_pad = " " * indent_after_pipe

    # Word-wrap to `width` chars per content line.
    words = text.split(" ")
    lines: list[str] = []
    current = ""
    for w in words:
        candidate = f"{current} {w}".strip()
        if len(candidate) <= width:
            current = candidate
        else:
            if current:
                lines.append(current)
            current = w
    if current:
        lines.append(current)
    if not lines:
        return ""

    out = [f"{pipe}   {bulb}  {Term.color('meta', lines[0])}"]
    for line in lines[1:]:
        out.append(f"{pipe}{cont_pad}{Term.color('meta', line)}")
    return "\n".join(out)


def _print_safe(line: str = "") -> None:
    """print() that survives narrow stdout encodings (e.g. Windows cp1252).

    Session titles carry arbitrary Unicode (e.g. '→') that the console
    encoding may not represent; degrade those chars to '?' instead of
    crashing with UnicodeEncodeError.
    """
    try:
        print(line)
    except UnicodeEncodeError:
        enc = getattr(sys.stdout, "encoding", None) or "ascii"
        print(str(line).encode(enc, errors="replace").decode(enc, errors="replace"))


def echo(*lines):
    if not lines:
        _print_safe()
        return
    for line in lines:
        _print_safe(line)


# ============================================================
#  Path discovery
# ============================================================

def appdata_claude() -> Path:
    plat = str(sys.platform)
    if plat == "win32":
        appdata = os.environ.get("APPDATA")
        if not appdata:
            sys.exit("APPDATA env var not set; can't locate Claude Desktop dir")
        return Path(appdata) / "Claude"
    if plat == "darwin":
        return Path.home() / "Library/Application Support/Claude"
    return Path.home() / ".config/Claude"


def cli_jsonl_root() -> Path:
    return Path.home() / ".claude" / "projects"


def encode_cwd(cwd: str) -> str:
    """Convert cwd to ~/.claude/projects/ subdir name.

    Each ':', '\\', '/', '.' becomes '-'; consecutive separators stay consecutive.
    'D:\\code\\myapp\\.claude\\worktrees\\foo' -> 'D--code-myapp--claude-worktrees-foo'
    """
    return (cwd
            .replace(":", "-")
            .replace("\\", "-")
            .replace("/", "-")
            .replace(".", "-"))


# ============================================================
#  Account discovery
# ============================================================

@dataclass
class Account:
    uuid: str
    sessions_dir: Path
    email: str = ""
    last_activity: float = 0.0
    session_count: int = 0

    @property
    def short(self) -> str:
        return self.uuid[:8]

    @property
    def label(self) -> str:
        sep = Term.g("·", "|")
        return f"{self.email or '(unknown)'} {sep} {self.short}"


def _iter_session_files(account_dir: Path) -> Iterable[Path]:
    for ws in account_dir.iterdir():
        if not ws.is_dir():
            continue
        yield from ws.glob("local_*.json")


def _find_account_email(agent_root: Path, account_uuid: str) -> str:
    acct_dir = agent_root / account_uuid
    if not acct_dir.is_dir():
        return ""
    for ws in acct_dir.iterdir():
        if not ws.is_dir():
            continue
        for f in ws.glob("local_*.json"):
            try:
                d = json.loads(f.read_text(encoding="utf-8"))
                email = d.get("emailAddress", "")
                if email:
                    return email
            except (json.JSONDecodeError, OSError):
                continue
    return ""


def discover_accounts(claude_dir: Path) -> list[Account]:
    sessions_root = claude_dir / "claude-code-sessions"
    if not sessions_root.is_dir():
        return []
    agent_root = claude_dir / "local-agent-mode-sessions"
    accounts: list[Account] = []
    for acct_dir in sessions_root.iterdir():
        if not acct_dir.is_dir():
            continue
        sessions = list(_iter_session_files(acct_dir))
        if not sessions:
            continue
        last = max((s.stat().st_mtime for s in sessions), default=0.0)
        accounts.append(Account(
            uuid=acct_dir.name,
            sessions_dir=acct_dir,
            email=_find_account_email(agent_root, acct_dir.name),
            last_activity=last,
            session_count=len(sessions),
        ))
    return sorted(accounts, key=lambda a: -a.last_activity)


def detect_destination(accounts: list[Account]) -> Account | None:
    return accounts[0] if accounts else None


def resolve_account(query: str, accounts: list[Account]) -> Account | None:
    q = query.lower()
    for a in accounts:
        if a.uuid == query:
            return a
    for a in accounts:
        if a.uuid.startswith(query):
            return a
    for a in accounts:
        if q in a.email.lower():
            return a
    return None


# ============================================================
#  Sessions
# ============================================================

@dataclass
class Session:
    path: Path
    data: dict
    account: Account

    @property
    def sid(self) -> str:
        return self.data.get("sessionId", "")

    @property
    def cli_id(self) -> str:
        return self.data.get("cliSessionId", "")

    @property
    def cwd(self) -> str:
        return self.data.get("cwd", "")

    @property
    def title(self) -> str:
        return self.data.get("title", "(untitled)")

    @property
    def turns(self) -> int:
        return int(self.data.get("completedTurns", 0))

    @property
    def last_activity_ms(self) -> int:
        return int(self.data.get("lastActivityAt", 0))

    @property
    def is_remote(self) -> bool:
        return self.cwd.startswith("/sessions/")

    def transcript_path(self) -> Path | None:
        if not self.cli_id or not self.cwd:
            return None
        return cli_jsonl_root() / encode_cwd(self.cwd) / f"{self.cli_id}.jsonl"


def load_sessions(account: Account) -> list[Session]:
    out: list[Session] = []
    for f in _iter_session_files(account.sessions_dir):
        try:
            data = json.loads(f.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            continue
        out.append(Session(path=f, data=data, account=account))
    return out


def filter_sessions(
    sessions: list[Session],
    *,
    days: int | None,
    cwd_pattern: str = "",
    title_pattern: str = "",
) -> list[Session]:
    now_ms = int(time.time() * 1000)
    cutoff_ms = now_ms - (days * 86_400_000) if days is not None else 0
    out = []
    for s in sessions:
        if s.is_remote:
            continue
        if days is not None and s.last_activity_ms < cutoff_ms:
            continue
        if cwd_pattern and cwd_pattern.lower() not in s.cwd.lower():
            continue
        if title_pattern and title_pattern.lower() not in s.title.lower():
            continue
        out.append(s)
    return sorted(out, key=lambda s: -s.last_activity_ms)


# ============================================================
#  Grouping
# ============================================================

_WORKTREE_MARKERS = (
    "\\.claude\\worktrees\\",
    "/.claude/worktrees/",
)


def project_root(cwd: str) -> str:
    for marker in _WORKTREE_MARKERS:
        if marker in cwd:
            return cwd.split(marker)[0]
    return cwd


def relative_under_root(cwd: str, root: str) -> str:
    if cwd == root:
        return ""
    if cwd.startswith(root):
        return cwd[len(root):].lstrip("\\/")
    return cwd


def worktree_name(cwd: str) -> str:
    """If cwd is inside a `.claude/worktrees/<name>/...` path, return <name>; else ''."""
    for marker in _WORKTREE_MARKERS:
        if marker in cwd:
            tail = cwd.split(marker, 1)[1]
            # First path segment is the worktree name; strip any deeper subpath.
            return tail.split("\\", 1)[0].split("/", 1)[0]
    return ""


# ============================================================
#  Listing
# ============================================================

def render_hierarchy(sessions: list[Session], *, grouped: bool) -> dict[int, Session]:
    """Print sessions; return {1-based-index: session}."""
    if grouped:
        return _render_grouped(sessions)
    index_map: dict[int, Session] = {}
    for n, s in enumerate(sessions, 1):
        index_map[n] = s
        ago = _ago(s.last_activity_ms)
        meta = f"{s.turns} turns"
        display = f"{s.title}  ({s.cwd})"
        echo(leaf(n, display, meta=meta, age=ago, depth=1))
    return index_map


def _render_grouped(sessions: list[Session]) -> dict[int, Session]:
    """3-level hierarchy: Account -> Project -> Session."""
    index_map: dict[int, Session] = {}

    by_account: "OrderedDict[str, list[Session]]" = OrderedDict()
    for s in sessions:
        by_account.setdefault(s.account.uuid, []).append(s)

    n = 0
    for _, acct_sessions in by_account.items():
        acct = acct_sessions[0].account

        # Group within account by project root
        by_project: "OrderedDict[str, list[Session]]" = OrderedDict()
        for s in acct_sessions:
            by_project.setdefault(project_root(s.cwd), []).append(s)

        # Account header
        echo(panel_blank())
        echo(section(acct.email or "(unknown)", len(acct_sessions), color_token="accent"))

        proj_items = list(by_project.items())
        for pi, (root, members) in enumerate(proj_items):
            is_last_proj = pi == len(proj_items) - 1
            sub_func = sub_section_last if is_last_proj else sub_section
            echo(sub_func(root, len(members), color_token="default"))

            for li, s in enumerate(members):
                n += 1
                index_map[n] = s
                is_last_session = li == len(members) - 1
                ago = _ago(s.last_activity_ms)
                meta = f"{s.turns}t"
                echo(leaf(n, s.title, meta=meta, age=ago,
                          last=is_last_session, depth=2,
                          parent_last=is_last_proj))

    echo(panel_blank())
    return index_map


def _window_label(days: int | None) -> str:
    """Render the active time-window filter label."""
    if days is None:
        return "all time"
    if days <= 1:
        return "last 24h"
    return f"last {days}d"


def _ago(ms: int) -> str:
    if ms == 0:
        return "?"
    delta_s = max(0, int(time.time()) - (ms // 1000))
    if delta_s < 60:
        return f"{delta_s}s"
    if delta_s < 3600:
        return f"{delta_s // 60}m"
    if delta_s < 86400:
        return f"{delta_s // 3600}h"
    return f"{delta_s // 86400}d"


# ============================================================
#  Picker
# ============================================================

def interactive_pick(sessions: list[Session], *, grouped: bool) -> list[Session]:
    if not sessions:
        return []
    index_map = _render_grouped(sessions) if grouped else render_hierarchy(sessions, grouped=False)
    print()
    raw = input(Term.color("accent", "select> ")
                + "(numbers like '3,5,7', 'a' for all, blank to cancel): ").strip()
    if not raw:
        return []
    if raw.lower() == "a":
        return sessions
    picks = []
    for tok in raw.split(","):
        tok = tok.strip()
        if not tok:
            continue
        try:
            i = int(tok)
            if i in index_map:
                picks.append(index_map[i])
        except ValueError:
            continue
    return picks


# ============================================================
#  Workspace selection
# ============================================================

def pick_destination_workspace(account: Account) -> Path:
    workspaces = [w for w in account.sessions_dir.iterdir() if w.is_dir()]
    if not workspaces:
        new_ws = account.sessions_dir / str(uuidlib.uuid4())
        new_ws.mkdir(parents=True)
        return new_ws
    workspaces.sort(key=lambda w: -w.stat().st_mtime)
    return workspaces[0]


# ============================================================
#  Operate
# ============================================================

def summon_session(s: Session, dest_workspace: Path, *, move: bool, dry_run: bool) -> str:
    target = dest_workspace / s.path.name
    if target.exists():
        return "skip (already there)"
    if not s.cli_id:
        return "skip (no cliSessionId)"
    transcript = s.transcript_path()
    if transcript and not transcript.exists():
        return "skip (transcript missing)"
    if dry_run:
        return "would " + ("move" if move else "copy")
    op = shutil.move if move else shutil.copy2
    op(str(s.path), str(target))
    return "moved" if move else "copied"


def nudge_watcher(workspace_dir: Path, moved_files: list[Path] | None = None) -> None:
    """Force fs.watch to fire on the destination workspace dir.

    Desktop's fs.watch is finicky — sometimes it picks up move-in events
    immediately, sometimes it doesn't. We throw the kitchen sink at it:

      1. mtime update on each moved file (write event)
      2. Rename ping-pong on each moved file (move-out + move-in events)
      3. Sentinel create+delete in workspace dir (dir-mod event)
      4. Sentinel create+delete in account dir (parent dir-mod event)
      5. mtime update on workspace dir (dir-mod event)
      6. mtime update on account dir (parent dir-mod event)

    All paths are tried; failures are silent.

    Empirically: even with all of these, Desktop's renderer may still
    require a Logout -> Login cycle to refresh the sidebar. That's
    documented in SKILL.md as the canonical fallback.
    """
    now = time.time()
    account_dir = workspace_dir.parent

    # 1. mtime update on moved files
    for f in (moved_files or []):
        try:
            os.utime(f, (now, now))
        except OSError:
            pass

    # 2. Rename ping-pong on moved files
    for f in (moved_files or []):
        if not f.exists():
            continue
        tmp = f.with_name(f.name + ".summon-tmp")
        try:
            f.rename(tmp)
            tmp.rename(f)
        except OSError:
            try:
                if tmp.exists():
                    tmp.rename(f)
            except OSError:
                pass

    # 3 + 4. Sentinel pings at workspace AND account level
    for parent in (workspace_dir, account_dir):
        sentinel = parent / f".summon-nudge-{uuidlib.uuid4().hex[:8]}"
        try:
            sentinel.touch()
            sentinel.unlink()
        except OSError:
            pass

    # 5 + 6. mtime touch on workspace and account dirs
    for d in (workspace_dir, account_dir):
        try:
            os.utime(d, (now, now))
        except OSError:
            pass


# ============================================================
#  Peek
# ============================================================

def find_session_by_id(query: str, accounts: list[Account]) -> Session | None:
    q = query.lower().removeprefix("local_")
    for acct in accounts:
        for s in load_sessions(acct):
            sid = s.sid.lower().removeprefix("local_")
            cli = s.cli_id.lower()
            if sid == q or cli == query.lower():
                return s
            if sid.startswith(q) or cli.startswith(q):
                return s
    return None


def peek_session(query: str, accounts: list[Account], turns: int = 3) -> int:
    s = find_session_by_id(query, accounts)
    if not s:
        echo(panel_open(f"summon {Term.g('·', '|')} peek", indicator="not found"))
        echo(panel_blank())
        echo(f"   no session matching: {Term.color('alarm', query)}")
        echo(panel_blank())
        echo(panel_close())
        return 1
    transcript = s.transcript_path()
    if not transcript or not transcript.exists():
        echo(panel_open(f"summon {Term.g('·', '|')} peek", indicator=f"{s.account.short} {Term.g('·', '|')} {s.title}"))
        echo(panel_blank())
        echo(f"   transcript missing: {Term.color('alarm', str(transcript))}")
        echo(panel_blank())
        echo(panel_close())
        return 2

    exchanges: list[tuple[str, str]] = []
    try:
        with transcript.open("r", encoding="utf-8") as f:
            for line in f:
                try:
                    rec = json.loads(line)
                except json.JSONDecodeError:
                    continue
                t = rec.get("type")
                if t not in ("user", "assistant"):
                    continue
                msg = rec.get("message", {})
                text = _extract_text(msg.get("content"))
                if text:
                    exchanges.append((t, text))
    except OSError as e:
        echo(panel_open(f"summon {Term.g('·', '|')} peek", indicator="read error"))
        echo(panel_blank())
        echo(f"   {Term.color('alarm', str(e))}")
        echo(panel_blank())
        echo(panel_close())
        return 2

    indicator = f"{s.account.email or s.account.short}"
    echo(panel_open(f"summon {Term.g('·', '|')} peek", indicator=indicator))
    echo(panel_blank())
    sep = Term.g("·", "|")
    echo(summary_line(f"{s.title!r}   {s.cwd}"))
    echo(summary_line(f"{s.turns} turns {sep} last activity {_ago(s.last_activity_ms)}"))
    echo(panel_blank())

    if not exchanges:
        echo(f"   {Term.color('meta', '(transcript has no readable user/assistant messages)')}")
        echo(panel_blank())
        echo(panel_close())
        return 0

    tail = exchanges[-(turns * 2):]
    echo(section(f"last {len(tail)} message(s)", color_token="accent"))
    for role, text in tail:
        marker = Term.color("accent", ">>") if role == "user" else Term.color("ok", "<<")
        snippet = text.strip().replace("\n", " ")
        if len(snippet) > 600:
            snippet = snippet[:597] + "..."
        echo(panel_blank())
        # Wrap to 70 chars per line
        words = snippet.split(" ")
        line_width = 70
        line = ""
        first = True
        for w in words:
            candidate = (line + " " + w) if line else w
            if len(candidate) <= line_width:
                line = candidate
            else:
                pipe = Term.g("│", "|")
                lead = f"{pipe}   {marker} " if first else f"{pipe}      "
                echo(f"{lead}{line}")
                line = w
                first = False
        if line:
            pipe = Term.g("│", "|")
            lead = f"{pipe}   {marker} " if first else f"{pipe}      "
            echo(f"{lead}{line}")

    echo(panel_blank())
    echo(panel_close(hotkeys=[("q", "quit")]))
    return 0


def _extract_text(content) -> str:
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        chunks = []
        for block in content:
            if isinstance(block, dict):
                t = block.get("type")
                if t == "text":
                    chunks.append(block.get("text", ""))
                elif t == "tool_use":
                    chunks.append(f"[tool_use: {block.get('name', '?')}]")
                elif t == "tool_result":
                    chunks.append("[tool_result]")
        return " ".join(chunks)
    return ""


# ============================================================
#  Toolbox modes: rebind / pick / recover / doctor
# ============================================================

def eecho(*lines):
    """echo() to stderr — context/panels for modes whose stdout is the data product."""
    if not lines:
        lines = ("",)
    for line in lines:
        try:
            print(line, file=sys.stderr)
        except UnicodeEncodeError:
            enc = getattr(sys.stderr, "encoding", None) or "ascii"
            print(str(line).encode(enc, errors="replace").decode(enc, errors="replace"),
                  file=sys.stderr)


def resolve_transcript(s: Session) -> tuple[Path | None, str]:
    """Locate a session's transcript JSONL.

    Returns (path, how) with how in:
      "expected"  — at ~/.claude/projects/<enc(cwd)>/<cliSessionId>.jsonl
      "scanned"   — found by scanning all project dirs for <cliSessionId>.jsonl
                    (the wrapper-uuid != transcript-filename trap: the transcript
                    is named by cliSessionId, and may live under a munged dir that
                    doesn't derive from the wrapper's recorded cwd)
      ""          — not found anywhere (path is None)
    """
    if not s.cli_id:
        return None, ""
    expected = s.transcript_path()
    if expected and expected.exists():
        return expected, "expected"
    hits = sorted(cli_jsonl_root().glob(f"*/{s.cli_id}.jsonl"))
    if hits:
        return hits[0], "scanned"
    return None, ""


# Boilerplate the first-ask sniffer must skip: slash-command echoes, skill
# preambles, interrupt markers — none of which are the session's real opening ask.
_ASK_SKIP_MARKERS = (
    "command-name", "local-command", "Sync - Session Bootstrap",
    "Base directory for this skill", "[Request interrupted",
)


def _ts_ms(value: str) -> int | None:
    try:
        return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() * 1000)
    except (ValueError, AttributeError):
        return None


def analyze_transcript(path: Path | None, buckets: int = 24) -> dict:
    """Stream a transcript JSONL once and derive picker display metrics.

    Returns events / toolCalls / density buckets / durationMin / sizeKB /
    ctxTokens (last-turn context occupancy, matching Claude Code's live meter) /
    ctxPeak (max occupancy before any auto-compaction) / firstAsk. Every field
    degrades to a zero/empty default on a missing or unreadable transcript, so
    the picker still renders — just without the extras. One pass, so a large
    transcript costs one linear read, never a re-scan.
    """
    out = {"events": 0, "toolCalls": 0, "buckets": [0] * buckets,
           "durationMin": 0, "sizeKB": 0, "ctxTokens": 0, "ctxPeak": 0,
           "firstAsk": ""}
    if not path or not path.exists():
        return out
    try:
        out["sizeKB"] = round(path.stat().st_size / 1024)
    except OSError:
        pass
    stamps: list[int] = []
    last_ctx = peak_ctx = 0
    try:
        with path.open(encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if not line:
                    continue
                out["events"] += 1
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError:
                    continue
                stamp = obj.get("timestamp")
                if stamp:
                    ms = _ts_ms(stamp)
                    if ms:
                        stamps.append(ms)
                msg = obj.get("message")
                msg = msg if isinstance(msg, dict) else {}
                usage = msg.get("usage")
                if isinstance(usage, dict):
                    occ = (usage.get("input_tokens", 0)
                           + usage.get("cache_creation_input_tokens", 0)
                           + usage.get("cache_read_input_tokens", 0)
                           + usage.get("output_tokens", 0))
                    if occ:
                        last_ctx = occ
                        peak_ctx = max(peak_ctx, occ)
                typ = obj.get("type")
                if typ == "assistant":
                    content = msg.get("content")
                    if isinstance(content, list):
                        out["toolCalls"] += sum(
                            1 for p in content
                            if isinstance(p, dict) and p.get("type") == "tool_use")
                elif typ == "user" and not out["firstAsk"]:
                    content = msg.get("content")
                    if isinstance(content, str):
                        txt = content
                    elif isinstance(content, list):
                        txt = "".join(p.get("text", "") for p in content
                                      if isinstance(p, dict) and p.get("type") == "text")
                    else:
                        txt = ""
                    txt = txt.strip()
                    if txt and not txt.startswith("<") and not any(
                            m in txt for m in _ASK_SKIP_MARKERS):
                        out["firstAsk"] = txt[:280]
    except OSError:
        return out
    out["ctxTokens"] = last_ctx
    out["ctxPeak"] = peak_ctx
    if stamps:
        lo, hi = min(stamps), max(stamps)
        span = hi - lo
        out["durationMin"] = round(span / 60000)
        counts = [0] * buckets
        for ms in stamps:
            idx = 0 if span <= 0 else min(buckets - 1, int((ms - lo) / span * buckets))
            counts[idx] += 1
        out["buckets"] = counts
    return out


def find_wrappers_by_id(query: str, accounts: list[Account]) -> tuple[list[Session], bool]:
    """All wrapper files for one logical session (copies may exist in several accounts).

    Matches sessionId or cliSessionId, exact first, then prefix. Returns
    (matches, ambiguous): ambiguous=True when a prefix hits MORE than one
    distinct logical session.
    """
    q = query.lower().removeprefix("local_")
    exact: list[Session] = []
    prefix: list[Session] = []
    for acct in accounts:
        for s in load_sessions(acct):
            sid = s.sid.lower().removeprefix("local_")
            cli = s.cli_id.lower()
            if sid == q or cli == q:
                exact.append(s)
            elif sid.startswith(q) or cli.startswith(q):
                prefix.append(s)
    matches = exact or prefix
    logical = {s.sid for s in matches}
    return matches, len(logical) > 1


def _rebase_path(old_val: str, old_cwd: str, new_cwd: str) -> str:
    """Rebase a sibling path field (originCwd/worktreePath) after cwd moves.

    Worktree sessions record originCwd as the project ROOT while cwd is the
    worktree path under it — so a plain prefix replace isn't enough:
      old_val == old_cwd                      -> new_cwd
      old_cwd == old_val + suffix (root case) -> strip the same suffix off new_cwd
      old_val == old_cwd + suffix (child)     -> new_cwd + suffix
    Anything else is left untouched.
    """
    if not old_val:
        return old_val
    if old_val == old_cwd:
        return new_cwd
    if old_cwd.startswith(old_val):
        suffix = old_cwd[len(old_val):]
        if suffix and new_cwd.endswith(suffix):
            return new_cwd[: len(new_cwd) - len(suffix)]
    if old_val.startswith(old_cwd):
        return new_cwd + old_val[len(old_cwd):]
    return old_val


def mode_rebind(args, accounts: list[Account]) -> int:
    """Fix a session's recorded cwd after a folder move.

    Backs up each wrapper OUTSIDE the live store, atomically rewrites
    cwd/originCwd/worktreePath, bridges the transcript into the new munged
    project dir (Desktop resolves it via enc(cwd)), and verifies by re-read.
    """
    if not args.target:
        eecho("usage: summon rebind <id> --cwd <newpath>")
        return 2
    if not args.cwd:
        eecho("rebind needs --cwd <newpath> — the folder's new location")
        return 2

    new_path = Path(args.cwd)
    if new_path.exists():
        new_cwd = str(new_path.resolve())
    elif args.force:
        new_cwd = str(new_path)
    else:
        eecho(f"new cwd does not exist on disk: {args.cwd}",
              "(pass --force to rebind to a not-yet-existing path)")
        return 3

    matches, ambiguous = find_wrappers_by_id(args.target, accounts)
    if not matches:
        eecho(f"no session matching: {args.target}")
        return 3
    if ambiguous:
        eecho(f"'{args.target}' matches {len({m.sid for m in matches})} different sessions — be more specific:")
        for m in matches[:8]:
            eecho(f"  {m.sid}  {m.title!r}  {m.cwd}")
        return 2

    stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime())
    backup_root = Path.home() / ".claude" / "summon-backups" / stamp

    echo(panel_open(f"summon {Term.g('·', '|')} rebind",
                    indicator="dry-run" if args.dry_run else matches[0].sid[:14]))
    echo(panel_blank())
    old_cwd = matches[0].cwd
    echo(summary_line(f"{old_cwd}"))
    echo(summary_line(f"{Term.g('→', '->')} {new_cwd}"))
    echo(panel_blank())

    problems = 0
    transcript_note = ""
    for i, s in enumerate(matches):
        is_last = i == len(matches) - 1
        label = f"{s.account.short}/{s.path.name}"

        if args.dry_run:
            echo(leaf(0, label, meta=Term.color("meta", "would rebind"),
                      last=is_last, depth=1))
            continue

        # 1. Backup outside the live store
        backup_root.mkdir(parents=True, exist_ok=True)
        backup = backup_root / f"{s.account.uuid}__{s.path.parent.name}__{s.path.name}"
        shutil.copy2(s.path, backup)

        # 2. Atomic rewrite
        data = dict(s.data)
        data["cwd"] = new_cwd
        for field in ("originCwd", "worktreePath"):
            if field in data:
                data[field] = _rebase_path(str(data[field] or ""), s.cwd, new_cwd)
        tmp = s.path.with_name(s.path.name + ".tmp")
        tmp.write_text(json.dumps(data, indent=2), encoding="utf-8")
        os.replace(tmp, s.path)

        # 3. Verify by re-read
        try:
            reread = json.loads(s.path.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            reread = {}
        if reread.get("cwd") != new_cwd:
            problems += 1
            shutil.copy2(backup, s.path)  # restore from backup
            echo(leaf(0, label, meta=Term.color("alarm", "verify FAILED — restored"),
                      last=is_last, depth=1))
            continue

        echo(leaf(0, label, meta=Term.color("ok", "rebound"), last=is_last, depth=1))

    # 4. Transcript bridge — Desktop looks in enc(new cwd) after the rebind
    s0 = matches[0]
    if s0.cli_id:
        old_transcript, how = resolve_transcript(s0)  # s0.data still holds OLD cwd
        new_dir = cli_jsonl_root() / encode_cwd(new_cwd)
        new_transcript = new_dir / f"{s0.cli_id}.jsonl"
        if new_transcript.exists():
            transcript_note = "transcript already at new path"
        elif not old_transcript:
            transcript_note = "transcript missing everywhere — session may not reopen"
            problems += 1
        elif args.no_transcript:
            transcript_note = f"transcript NOT copied (--no-transcript): {old_transcript}"
        elif args.dry_run:
            transcript_note = f"would copy transcript {Term.g('→', '->')} {new_transcript}"
        else:
            new_dir.mkdir(parents=True, exist_ok=True)
            shutil.copy2(old_transcript, new_transcript)
            transcript_note = f"transcript copied ({how}) {Term.g('→', '->')} {new_transcript}"

    echo(panel_blank())
    if transcript_note:
        echo(summary_line(transcript_note))
    if not args.dry_run:
        echo(summary_line(f"backup: {backup_root}"))
    echo(panel_blank())
    healths = [("ok", f"{len(matches)} wrapper(s)")]
    if problems:
        healths.append(("alarm", f"{problems} problem(s)"))
    echo(panel_close(healths=healths))
    if not args.dry_run and not problems:
        echo()
        echo(Term.color("warn",
             "next: restart Desktop (or Logout/Login) so the sidebar re-reads the wrapper."))
        if any(m in new_cwd for m in _WORKTREE_MARKERS):
            echo(Term.color("meta",
                 "  git worktree links break on folder moves — run "
                 f"`git worktree repair {new_cwd}` from the repo root."))
    return 1 if problems else 0


_RECOVERY_INSTRUCTION = (
    "First read only the TAIL of the transcript (last ~150-200 lines) to see where "
    "it left off — do not ingest the whole file; read earlier chunks selectively "
    "only if something is unclear. Then: summarize the session state in a few "
    "bullets (goal, what's done, what was in flight, blockers), and continue the "
    "work from there in the current directory."
)


# --- Distilled handover: extract -> distill -> cache -> emit -----------------

EXTRACT_BUDGET_DEFAULT = 120_000   # chars of conversation fed to the distiller
VERBATIM_TAIL_TURNS = 15           # final turns always included in full
HEAD_TURN_CAP = 4_000              # per-turn cap for pre-tail turns (giant pastes)
DISTILL_TIMEOUT_S = 60
DISTILL_MODEL_DEFAULT = "sonnet"

_DISTILL_INSTRUCTION = """\
You are writing a HANDOVER BRIEF so a fresh Claude session can continue an \
interrupted one. Below is the previous session's conversation with tool noise \
removed; the final turns are verbatim.

Produce ONLY the brief, in markdown, with exactly these sections:

## Goal
## What landed
## Unfinished
## Open decisions
## Key context

Rules:
- "What landed" names the branch and any commits/hashes if mentioned.
- Be specific: file paths, branch names, commands, decisions.
- Do not invent anything not present in the conversation.
- No preamble, no closing remarks, no tool use.
- Keep the entire brief under about 1000 words.
"""


def _text_only(content) -> str:
    """Conversational text only — tool_use inputs and tool_result blobs are
    skipped entirely (they are most of a transcript's bytes)."""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        chunks = []
        for block in content:
            if isinstance(block, dict) and block.get("type") == "text":
                t = block.get("text", "")
                if t:
                    chunks.append(t)
        return "\n".join(chunks)
    return ""


def extract_conversation(transcript: Path, budget: int = EXTRACT_BUDGET_DEFAULT) -> str:
    """Build the distillation input from a transcript JSONL — in-script, no LLM.

    User/assistant text turns only. The final VERBATIM_TAIL_TURNS turns are
    always included in full; earlier turns fill the remaining budget from the
    START (so the goal statement survives), with the middle elided when the
    session is too long to fit.
    """
    turns: list[tuple[str, str]] = []
    try:
        with transcript.open("r", encoding="utf-8", errors="replace") as f:
            for line in f:
                try:
                    rec = json.loads(line)
                except json.JSONDecodeError:
                    continue
                role = rec.get("type")
                if role not in ("user", "assistant"):
                    continue
                text = _text_only(rec.get("message", {}).get("content")).strip()
                if text:
                    turns.append((role, text))
    except OSError:
        return ""
    if not turns:
        return ""

    def fmt(role: str, text: str) -> str:
        return f"{role.upper()}: {text}"

    tail = turns[-VERBATIM_TAIL_TURNS:]
    head = turns[:-VERBATIM_TAIL_TURNS]
    tail_block = "\n\n".join(fmt(r, t) for r, t in tail)
    if len(tail_block) >= budget:
        return tail_block[-budget:]  # most recent state wins

    remaining = budget - len(tail_block)
    head_parts: list[str] = []
    elided = not head
    for r, t in head:
        if len(t) > HEAD_TURN_CAP:
            t = t[:HEAD_TURN_CAP] + " …[turn truncated]"
        piece = fmt(r, t)
        cost = len(piece) + 2
        if cost > remaining:
            elided = True
            break
        head_parts.append(piece)
        remaining -= cost

    parts = list(head_parts)
    if elided and head_parts:
        parts.append("[… middle of session elided to fit extraction budget …]")
    parts.append(tail_block)
    return "\n\n".join(parts)


def handover_cache_path(transcript: Path) -> Path:
    return transcript.with_name(transcript.name + ".handover.md")


def load_cached_brief(transcript: Path, *, refresh: bool) -> str | None:
    """Cached brief, valid only when newer than the transcript itself."""
    if refresh:
        return None
    cache = handover_cache_path(transcript)
    try:
        if cache.exists() and cache.stat().st_mtime > transcript.stat().st_mtime:
            text = cache.read_text(encoding="utf-8").strip()
            return text or None
    except OSError:
        pass
    return None


def store_brief(transcript: Path, brief: str) -> None:
    cache = handover_cache_path(transcript)
    tmp = cache.with_name(cache.name + ".tmp")
    try:
        tmp.write_text(brief + "\n", encoding="utf-8")
        os.replace(tmp, cache)
    except OSError as e:
        eecho(f"warning: could not cache handover brief at {cache}: {e}")


def distill_brief(extraction: str, s: Session, model: str) -> str | None:
    """One-shot tool-less `claude -p` summarisation (gated child: dontAsk,
    no allowlist — per loop-engineering, never bypassPermissions).

    Returns None on any worker unavailability — absent CLI, non-zero exit,
    timeout, empty output — with a stderr warning. Advisory, never fatal.
    """
    claude_bin = shutil.which("claude")
    if not claude_bin:
        eecho("warning: `claude` CLI not on PATH — emitting non-distilled pointer prompt")
        return None
    branch = str(s.data.get("branch") or "")
    payload = (
        _DISTILL_INSTRUCTION
        + "\n--- SESSION METADATA ---\n"
        + f"Title: {s.title}\nBranch: {branch or '(none)'}\nCwd: {s.cwd}\n"
        + "\n--- CONVERSATION EXTRACTION ---\n"
        + extraction
    )
    import subprocess
    eecho(f"distilling handover brief via `claude -p --model {model}` "
          f"({len(extraction)} chars in, ~{DISTILL_TIMEOUT_S}s timeout)…")
    try:
        r = subprocess.run(
            [claude_bin, "-p", "--model", model, "--permission-mode", "dontAsk"],
            input=payload.encode("utf-8"),
            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            timeout=DISTILL_TIMEOUT_S)
    except subprocess.TimeoutExpired:
        eecho(f"warning: distillation timed out after {DISTILL_TIMEOUT_S}s — "
              "emitting non-distilled pointer prompt")
        return None
    except OSError as e:
        eecho(f"warning: could not run `claude`: {e} — emitting non-distilled pointer prompt")
        return None
    if r.returncode != 0:
        tail = r.stderr.decode("utf-8", "replace").strip()[-200:]
        eecho(f"warning: `claude -p` exited {r.returncode}"
              + (f" ({tail})" if tail else "")
              + " — emitting non-distilled pointer prompt")
        return None
    brief = r.stdout.decode("utf-8", "replace").strip()
    if not brief:
        eecho("warning: distillation produced no output — emitting non-distilled pointer prompt")
        return None
    return brief


def _pointer_clause(s: Session, transcript: Path) -> str:
    branch = str(s.data.get("branch") or "")
    ident = s.sid.removeprefix("local_") or s.cli_id
    branch_part = f", branch {branch}" if branch else ""
    return (f"Full transcript at {transcript} (session {ident}{branch_part}); "
            "consult it only if something specific is missing.")


def emit_recovery_prompt(s: Session, args) -> int:
    """Print a paste-ready recovery prompt for a new session (stdout = the prompt).

    Default flow: extract conversation -> distill via `claude -p` -> cache the
    brief at <transcript>.handover.md -> emit brief + pointer clause. Falls
    back to the plain pointer prompt when distillation is unavailable/disabled.
    """
    transcript, how = resolve_transcript(s)

    sep = Term.g("·", "|")
    eecho(panel_open(f"summon {Term.g('·', '|')} recover", indicator=s.account.short))
    eecho(panel_blank())
    eecho(summary_line(f"{s.title!r}   {s.turns}t {sep} {_ago(s.last_activity_ms)}"))
    if how == "scanned":
        eecho(summary_line("transcript found by scan — wrapper cwd does not derive its location"))
    eecho(panel_blank())
    eecho(panel_close(healths=[("ok", "prompt on stdout")] if transcript
                      else [("alarm", "no transcript")]))

    if not transcript:
        eecho(f"no transcript found for cliSessionId {s.cli_id or '(none)'} — cannot recover")
        return 3

    branch = str(s.data.get("branch") or "")

    # --- Distilled handover path ---
    if not getattr(args, "no_distill", False):
        brief = load_cached_brief(transcript, refresh=getattr(args, "refresh", False))
        if brief:
            eecho(f"reusing cached handover brief: {handover_cache_path(transcript)} "
                  "(--refresh to re-distill)")
        else:
            extraction = extract_conversation(
                transcript, getattr(args, "budget", EXTRACT_BUDGET_DEFAULT))
            if extraction:
                brief = distill_brief(extraction, s,
                                      getattr(args, "model", DISTILL_MODEL_DEFAULT))
                if brief:
                    store_brief(transcript, brief)
            else:
                eecho("warning: transcript has no conversational text — "
                      "emitting non-distilled pointer prompt")
        if brief:
            lines = [f"Continue a previous Claude session: {s.title!r}."]
            if branch:
                lines.append(f"Branch: {branch}")
            lines += ["", brief, "", _pointer_clause(s, transcript)]
            for line in lines:
                _print_safe(line)
            return 0

    # --- Fallback: non-distilled pointer prompt ---
    cwd_note = ""
    if s.cwd and not s.cwd.startswith("/sessions/") and not Path(s.cwd).exists():
        cwd_note = "  (missing on disk — folder moved?)"

    lines = [
        "Continue a previous Claude session.",
        "",
        f"Title:      {s.title}",
    ]
    if branch:
        lines.append(f"Branch:     {branch}")
    lines.append(f"Orig cwd:   {s.cwd}{cwd_note}")
    lines.append(f"Transcript: {transcript}")
    lines += ["", _RECOVERY_INSTRUCTION]
    for line in lines:
        _print_safe(line)
    return 0


def mode_recover(args, accounts: list[Account]) -> int:
    if not args.target:
        eecho("usage: summon recover <id>   (sessionId or cliSessionId, prefix ok)")
        return 2
    matches, ambiguous = find_wrappers_by_id(args.target, accounts)
    if not matches:
        eecho(f"no session matching: {args.target}")
        return 3
    if ambiguous:
        eecho(f"'{args.target}' matches {len({m.sid for m in matches})} different sessions — be more specific:")
        for m in matches[:8]:
            eecho(f"  {m.sid}  {m.title!r}  {m.cwd}")
        return 2
    return emit_recovery_prompt(matches[0], args)


def _is_live(s: Session, now_ms: int) -> bool:
    """Heuristic 'running state': wrapper touched within the last 10 minutes."""
    recent = now_ms - 10 * 60_000
    if s.last_activity_ms >= recent:
        return True
    try:
        return int(s.path.stat().st_mtime * 1000) >= recent
    except OSError:
        return False


def _cwd_broken(s: Session) -> bool:
    """Doctor's broken-binding check: recorded cwd no longer exists on disk."""
    return not (bool(s.cwd) and Path(s.cwd).exists())


def _iso_utc(ms: int) -> str:
    """Epoch-ms -> ISO-8601 Z, '' for the unset 0."""
    if not ms:
        return ""
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(ms / 1000))


def session_rows(candidates: list[Session], now_ms: int, rich: bool = False) -> list[dict]:
    """Build the pick inventory rows for a set of sessions (in-process).

    The single source of truth for a session's machine-readable shape — shared
    by ``pick --json`` (which wraps these in an envelope) and ``widget`` (which
    trims them and injects them into the card-picker template). Keeping it one
    function means the widget builder never has to shell out to ``pick`` or
    re-read a transcript that ``pick`` already read.

    With ``rich`` each row gains transcript-derived display metrics (context
    occupancy, activity density, event/tool counts, size, duration, opening
    ask) at the cost of one linear transcript read per session; without it the
    rows are metadata-only and instant.
    """
    rows = []
    for s in candidates:
        transcript = resolve_transcript(s)[0]
        row = {
            "id": s.sid.removeprefix("local_")[:8],
            "sessionId": s.sid,
            "cliSessionId": s.cli_id,
            "title": s.title,
            "cwd": s.cwd,
            "projectRoot": project_root(s.cwd),
            "worktree": worktree_name(s.cwd),
            "branch": str(s.data.get("branch") or ""),
            "model": str(s.data.get("model") or ""),
            "effort": str(s.data.get("effort") or ""),
            "turns": s.turns,
            "isArchived": bool(s.data.get("isArchived")),
            "isRunning": _is_live(s, now_ms),
            "brokenCwd": _cwd_broken(s),
            "lastActivityAt": _iso_utc(s.last_activity_ms),
            "account": s.account.short,
            "accountEmail": s.account.email,
            "transcriptPath": str(transcript) if transcript else None,
        }
        if rich:
            m = analyze_transcript(transcript)
            window = 1_000_000 if m["ctxPeak"] > 200_000 else 200_000
            row.update({
                "events": m["events"],
                "toolCalls": m["toolCalls"],
                "densityBuckets": m["buckets"],
                "durationMin": m["durationMin"],
                "sizeKB": m["sizeKB"],
                "ctxTokens": m["ctxTokens"],
                "ctxPeak": m["ctxPeak"],
                "ctxWindow": window,
                "ctxPct": round(min(100, m["ctxTokens"] / window * 100)) if window else 0,
                "ctxPeakPct": round(min(100, m["ctxPeak"] / window * 100)) if window else 0,
                "firstAsk": m["firstAsk"],
            })
        rows.append(row)
    return rows


def pick_json(candidates: list[Session], now_ms: int, rich: bool = False) -> int:
    """`pick --json` — the session inventory as a claude-mods.summon.pick
    envelope on stdout (JSON only; panels never touch stdout on this path).

    Feeds the in-chat visual card picker (assets/picker-widget.html) and any
    other scripted caller. An empty inventory is valid data, not an error:
    exit 0.

    With ``rich`` (``--rich``) the schema advances to pick/v2 and each row
    gains transcript-derived display metrics (see ``session_rows``).
    """
    rows = session_rows(candidates, now_ms, rich)
    print(json.dumps({
        "data": rows,
        "meta": {"count": len(rows),
                 "schema": "claude-mods.summon.pick/v2" if rich else
                           "claude-mods.summon.pick/v1"},
    }, indent=2))
    return 0


def mode_pick(args, accounts: list[Account]) -> int:
    """Interactive picker over the whole session store -> recovery prompt."""
    sessions: list[Session] = []
    for acct in

#!/usr/bin/env bash
# Self-test for the summon skill (scripts/summon.py).
#
# Offline-deterministic: builds a throwaway Claude Desktop dir tree in a temp
# sandbox (HOME/USERPROFILE/APPDATA redirected), so no real account data is
# read or written. Covers the selection/confirmation flow (--yes, --select,
# piped stdin), the cp1252 UnicodeEncodeError regression, the toolbox modes
# (rebind/recover/pick/doctor, incl. the worktree-repair hint), and the
# distilled-handover flow (extraction skips tool blobs, cache hit/miss on
# mtime, --no-distill, degrade paths via a PATH-shimmed fake `claude` —
# no real LLM call is ever made by this suite), the pick --json inventory
# envelope, and the in-chat picker asset (present + cited from SKILL.md).
#
# The behavioural checks live in test_summon.py — its pass/fail summary is the
# primary signal. One shell-level check also runs after it (below): a
# section-map drift gate that pins the docstring 'Sections:' list against the
# body's `# ===` banner headers, so the deliberately-single-file script's map
# cannot silently rot as it grows.
#
# Usage:   bash tests/run.sh
# Exit:    0 all pass, 1 one or more failures

set -uo pipefail

HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

# Pick a python that actually executes — skips the Windows Store python3 stub.
PYTHON=""
for c in python python3 py; do
  if command -v "$c" >/dev/null 2>&1 && "$c" -c "" >/dev/null 2>&1; then PYTHON="$c"; break; fi
done
[[ -z "$PYTHON" ]] && { echo "no working python found" >&2; exit 1; }

# Run the full behavioural suite, then fall through to the shell-level
# section-map drift gate (we deliberately do NOT `exec` the python here — the
# gate must run afterwards and contribute to the combined exit code).
SUMMON_PY_RC=0
"$PYTHON" "$HERE/test_summon.py" || SUMMON_PY_RC=$?

# --- section-map drift gate (summon.py docstring 'Sections:' ↔ # === banners) ---
# summon.py is deliberately a single multi-thousand-line file
# (docs/SKILL-RESOURCE-PROTOCOL.md: skill scripts ship as self-contained
# portable units — do not split). Its module docstring carries a `Sections:`
# map of the `# ===` banner headers so the file stays navigable. This gate
# pins BOTH the docstring section count and the body banner count, so a
# section added or removed on either side fails the build until the map is
# reconciled. An empty parse on either side is a hard FAIL (never a silent
# pass) — that is the rot mode this guard exists to catch: a docstring/map
# format change that yields zero names.
#
# Matching is STRICT on alias-resolved first-token keys (upgraded from a
# count gate per adversarial review — counts stay 14/13 under a rename, so
# structural drift slipped by). The docstring and banners carry different
# labels for a few sections ("DESIGN(term)"↔"DESIGN:", "Modes (…)"↔"Toolbox
# modes:", "CLI entry"↔"Main"), so those are an explicit alias table, and
# "Transcript/Distill" (implemented inline, banner-less) is an explicit
# exception. Anything else that diverges — including a rename on either
# side — is drift.
GP=0; GF=0
ok(){ GP=$((GP+1)); printf '  PASS  %s\n' "$1"; }
no(){ GF=$((GF+1)); printf '  FAIL  %s\n' "$1"; }

SRC="$HERE/../scripts/summon.py"

# docstring section names: from the first `Sections:` line to the next `"""`.
doc_sections="$(awk '
  !done && /Sections:/ { cap=1; sub(/.*Sections:[[:space:]]*/,"",$0); blob=blob $0 " "; next }
  cap { if (/"""/) { done=1; cap=0; next } blob=blob $0 " " }
  END { gsub(/·/,"\n",blob); n=split(blob,a,"\n");
        for (i=1;i<=n;i++){ s=a[i]; sub(/^[[:space:]]+/,"",s); sub(/[[:space:]]+$/,"",s); sub(/\.$/,"",s); if (s!="") print s } }
' "$SRC")"
dc="$(printf '%s\n' "$doc_sections" | grep -c . || true)"

# body banner sections: the `# ===…===` header pairs — the name line that
# sits between each opening banner and its closing banner.
ban_sections="$(awk '
  /^# ={20,}$/ { saw=1; next }
  saw && /^#  / { t=$0; sub(/^# +/,"",t); sub(/[[:space:]]+$/,"",t); print t }
  { saw=0 }
' "$SRC")"
bc="$(printf '%s\n' "$ban_sections" | grep -c . || true)"

# Normalize a section label to its comparison key: first token, with any
# "(...)" suffix and trailing ":" stripped ("DESIGN(term)" -> "DESIGN",
# "Toolbox modes: rebind…" -> "Toolbox").
norm_key() { awk '{ t=$1; sub(/\(.*$/,"",t); sub(/:$/,"",t); print t }'; }
doc_keys="$(printf '%s\n' "$doc_sections" | norm_key | sed \
  -e 's/^Modes$/Toolbox/' \
  -e 's/^CLI$/Main/' \
  | grep -v '^Transcript' | sort -u)"
ban_keys="$(printf '%s\n' "$ban_sections" | norm_key | sort -u)"

if [[ "$dc" -eq 0 ]]; then
  no "section-map (docstring) EMPTY PARSE: 0 sections — 'Sections:' line missing or unparseable"
elif [[ "$bc" -eq 0 ]]; then
  no "section-map (body) EMPTY PARSE: 0 banners — banner format changed"
elif [[ "$doc_keys" == "$ban_keys" ]]; then
  ok "section-map: docstring ($dc) <-> banners ($bc) match after alias resolution"
else
  diverged="$(comm -3 <(printf '%s\n' "$doc_keys") <(printf '%s\n' "$ban_keys") | tr -d '\t' | paste -sd', ' -)"
  no "section-map DRIFT — docstring and banners disagree on: ${diverged:-<unknown>}"
fi

# Boxed-banner parity: each banner is a 3-line box (=== / title / ===); an
# odd ruler-line count means a box's closing line was deleted.
rulers="$(grep -cE '^# ={20,}$' "$SRC" || true)"
if (( rulers > 0 && rulers % 2 == 0 )); then
  ok "section-map: $rulers banner ruler lines (even — boxes intact)"
else
  no "section-map: ruler-line count $rulers (odd or zero) — a banner box is broken"
fi

echo "=== section-map drift gate: $GP passed, $GF failed ==="

# Combine: the Python behavioural suite AND the shell section-map gate pass.
[[ "$SUMMON_PY_RC" -eq 0 && "$GF" -eq 0 ]] || exit 1
exit 0

#!/usr/bin/env python3
"""Regression tests for summon.py — selection/confirmation flow + encoding safety.

Hermetic: builds a throwaway Claude Desktop directory tree in a temp sandbox
and points the script at it via HOME/USERPROFILE/APPDATA. No network, no real
account data touched.

Covers:
  1. --yes does NOT bypass the selection picker (piped picks are honoured)
  2. --select answers the picker non-interactively
  3. --select all selects everything
  4. Without --yes, declining the confirmation cancels (nothing copied)
  5. Without --yes, confirming 'y' proceeds
  6. --yes with empty stdin cancels instead of auto-selecting all
  7. Non-cp1252 chars in session titles don't crash on cp1252 stdout

Toolbox modes (rebind / recover / pick / doctor):
  8.  rebind rewrites cwd + rebases originCwd/worktreePath, backs up outside
      the store, bridges the transcript into the new munged dir, exits 0
  9.  rebind --dry-run touches nothing
  10. rebind with unknown id exits 3; ambiguous prefix exits 2
  11. rebind to a nonexistent path exits 3 without --force, 0 with it
  12. doctor flags the broken-cwd session (exit 10, --json envelope), and
      goes healthy (exit 0) after the rebind
  13. recover --no-distill emits the plain pointer prompt on stdout; transcript
      found via the scan fallback when the munged dir doesn't derive from the
      recorded cwd; no handover cache is written
  14. recover with unknown id exits 3
  15. pick --select N emits the recovery prompt for the Nth candidate
  16. rebind into a .claude/worktrees/ path prints the `git worktree repair` hint

Distilled handover (recover -> extract -> distill -> cache -> emit):
  17. extract_conversation keeps user/assistant text, skips tool_use inputs and
      tool_result blobs (fixture JSONL); respects the char budget with the
      verbatim tail winning
  18. recover distills via a PATH-shimmed fake `claude`, emits brief + pointer
      clause, caches at <transcript>.handover.md
  19. cache hit: unchanged transcript reuses the cached brief (no re-distill);
      --refresh forces re-distillation; a newer transcript mtime busts the cache
  20. degrade: `claude` absent from PATH -> plain pointer prompt, exit 0,
      stderr warning; failing `claude` (exit 1) -> same advisory fallback

Pick --json (the in-chat visual picker's data feed):
  21. pick --json emits a parseable claude-mods.summon.pick/v1 envelope on a
      stdout free of panel glyphs (pure-ASCII JSON, nothing before/after it)
  22. worktree cwds are split into projectRoot + worktree; non-worktree cwds
      get worktree ""; brokenCwd flags the session whose recorded cwd is gone;
      transcriptPath is resolved (scan fallback included) and ISO timestamps
      parse

In-chat picker asset:
  23. assets/picker-widget.html exists, carries the <script id="D"> injection
      block + sendPrompt wiring, and is cited from SKILL.md

Widget builder (summon widget -> finished card-picker HTML):
  24. replaces the <script id="D"> payload, drops 0-turn stubs by default
      (--include-stubs keeps them), honours --limit, keeps only whitelisted
      keys, emits one session object per line, downsamples density 24->12
      (--full-density keeps 24), sets the window <select>, and holds the HTML
      under the byte budget (capping with a stderr note under a tight --max-kb)
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import time
import shutil
from pathlib import Path

HERE = Path(__file__).resolve().parent
SCRIPT = HERE.parent / "scripts" / "summon.py"

SRC_UUID = "aaaaaaaa-1111-4111-8111-111111111111"
DEST_UUID = "bbbbbbbb-2222-4222-8222-222222222222"

PASS = 0
FAIL = 0


def ok(name: str) -> None:
    global PASS
    PASS += 1
    print(f"  PASS  {name}")


def no(name: str, detail: str = "") -> None:
    global FAIL
    FAIL += 1
    print(f"  FAIL  {name}" + (f" — {detail}" if detail else ""))


def claude_dir(env: dict) -> Path:
    if sys.platform == "win32":
        return Path(env["APPDATA"]) / "Claude"
    if sys.platform == "darwin":
        return Path(env["HOME"]) / "Library/Application Support/Claude"
    return Path(env["HOME"]) / ".config/Claude"


def build_sandbox(tmp: Path, titles: list[str]) -> tuple[dict, Path, Path]:
    """Create src account (len(titles) sessions, newest first = #1) + dest account."""
    home = tmp / "home"
    appdata = home / "AppData" / "Roaming"
    env = os.environ.copy()
    env["HOME"] = str(home)
    env["USERPROFILE"] = str(home)
    env["APPDATA"] = str(appdata)
    env.pop("PYTHONIOENCODING", None)
    env.pop("TERM_ASCII", None)

    cdir = claude_dir(env)
    now_ms = int(time.time() * 1000)

    src_ws = cdir / "claude-code-sessions" / SRC_UUID / "11111111-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
    src_ws.mkdir(parents=True)
    for i, title in enumerate(titles):
        sid = f"src-sess-{i}"
        (src_ws / f"local_{sid}.json").write_text(json.dumps({
            "sessionId": f"local_{sid}",
            "cliSessionId": f"cli-{sid}",
            "title": title,
            "cwd": "",  # no cwd -> no transcript lookup -> copy proceeds
            "lastActivityAt": now_ms - (i + 1) * 60_000,  # newest first
            "completedTurns": 5 + i,
        }), encoding="utf-8")

    dest_ws = cdir / "claude-code-sessions" / DEST_UUID / "22222222-bbbb-4bbb-8bbb-bbbbbbbbbbbb"
    dest_ws.mkdir(parents=True)
    (dest_ws / "local_dest-own.json").write_text(json.dumps({
        "sessionId": "local_dest-own",
        "cliSessionId": "cli-dest-own",
        "title": "dest resident",
        "cwd": "",
        "lastActivityAt": now_ms,
        "completedTurns": 1,
    }), encoding="utf-8")

    return env, src_ws, dest_ws


def run_summon(env: dict, extra_args: list[str], stdin_text: str = "") -> tuple[int, str]:
    cmd = [sys.executable, str(SCRIPT),
           "--to", DEST_UUID[:8], "--from", SRC_UUID[:8], "--days", "3"] + extra_args
    r = subprocess.run(cmd, input=stdin_text.encode("utf-8"),
                       env=env, capture_output=True, timeout=60)
    out = (r.stdout.decode("utf-8", "replace")
           + r.stderr.decode("utf-8", "replace"))
    return r.returncode, out


def copied_titles(dest_ws: Path) -> set[str]:
    titles = set()
    for f in dest_ws.glob("local_src-sess-*.json"):
        titles.add(json.loads(f.read_text(encoding="utf-8"))["title"])
    return titles


def with_sandbox(titles=("alpha", "beta", "gamma")):
    tmp = Path(tempfile.mkdtemp(prefix="summon-test-"))
    env, src_ws, dest_ws = build_sandbox(tmp, list(titles))
    return tmp, env, src_ws, dest_ws


def encode_cwd(cwd: str) -> str:
    """Mirror of summon.py's munging: cwd -> ~/.claude/projects/ subdir name."""
    return (cwd.replace(":", "-").replace("\\", "-")
               .replace("/", "-").replace(".", "-"))


def build_toolbox_sandbox(tmp: Path) -> dict:
    """One account, three sessions exercising the toolbox modes.

    sess-a  'moved'   worktree session; recorded cwd no longer exists (project
                      moved old-root -> new-root); transcript under enc(old cwd)
    sess-b  'healthy' cwd exists, transcript at the expected munged path
    sess-c  'mismatch' cwd exists, but the transcript lives under a munged dir
                      that does NOT derive from the recorded cwd (scan-fallback)
    """
    home = tmp / "home"
    appdata = home / "AppData" / "Roaming"
    env = os.environ.copy()
    env["HOME"] = str(home)
    env["USERPROFILE"] = str(home)
    env["APPDATA"] = str(appdata)
    env.pop("PYTHONIOENCODING", None)
    env.pop("TERM_ASCII", None)

    cdir = claude_dir(env)
    projects = home / ".claude" / "projects"
    now_ms = int(time.time() * 1000)

    old_root = tmp / "proj-old"                    # moved away -> missing
    new_root = tmp / "proj-new"
    old_wt = old_root / ".claude" / "worktrees" / "wt1"
    new_wt = new_root / ".claude" / "worktrees" / "wt1"
    new_wt.mkdir(parents=True)
    good = tmp / "proj-good"
    good.mkdir()

    ws = cdir / "claude-code-sessions" / SRC_UUID / "11111111-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
    ws.mkdir(parents=True)

    def wrapper(sid, cli, title, cwd, age_min, **extra):
        d = {"sessionId": f"local_{sid}", "cliSessionId": cli, "title": title,
             "cwd": cwd, "lastActivityAt": now_ms - age_min * 60_000,
             "completedTurns": 7}
        d.update(extra)
        (ws / f"local_{sid}.json").write_text(json.dumps(d), encoding="utf-8")

    wrapper("aaaa-moved", "cli-moved", "moved session", str(old_wt), 30,
            originCwd=str(old_root), worktreePath=str(old_wt),
            branch="claude/wt1")
    wrapper("bbbb-healthy", "cli-healthy", "healthy session", str(good), 60)
    wrapper("cccc-mismatch", "cli-mismatch", "mismatch session", str(good), 90)

    for enc_dir, cli in ((encode_cwd(str(old_wt)), "cli-moved"),
                         (encode_cwd(str(good)), "cli-healthy"),
                         ("X--some-unrelated-munged-dir", "cli-mismatch")):
        d = projects / enc_dir
        d.mkdir(parents=True, exist_ok=True)
        (d / f"{cli}.jsonl").write_text(
            '{"type":"user","message":{"content":"hello"}}\n', encoding="utf-8")

    return {"env": env, "ws": ws, "projects": projects,
            "old_wt": old_wt, "new_wt": new_wt, "new_root": new_root,
            "old_root": old_root, "home": home}


def run_mode(env: dict, argv: list[str], stdin_text: str = "") -> tuple[int, str, str]:
    r = subprocess.run([sys.executable, str(SCRIPT)] + argv,
                       input=stdin_text.encode("utf-8"),
                       env=env, capture_output=True, timeout=60)
    return (r.returncode,
            r.stdout.decode("utf-8", "replace"),
            r.stderr.decode("utf-8", "replace"))


def toolbox_tests() -> None:
    tmp = Path(tempfile.mkdtemp(prefix="summon-toolbox-"))
    try:
        sb = build_toolbox_sandbox(tmp)
        env, ws = sb["env"], sb["ws"]
        new_wt, new_root = sb["new_wt"], sb["new_root"]
        wrapper_a = ws / "local_aaaa-moved.json"

        # 9. dry-run first (order matters: before the real rebind)
        rc, out, err = run_mode(env, ["rebind", "aaaa-moved", "--cwd", str(new_wt),
                                      "--dry-run"])
        data = json.loads(wrapper_a.read_text(encoding="utf-8"))
        backups = sb["home"] / ".claude" / "summon-backups"
        if rc == 0 and data["cwd"] == str(sb["old_wt"]) and not backups.exists():
            ok("rebind --dry-run touches nothing")
        else:
            no("rebind --dry-run touches nothing",
               f"rc={rc} cwd={data['cwd']!r} backups={backups.exists()}")

        # 10a. unknown id
        rc, _, _ = run_mode(env, ["rebind", "zzzz-nope", "--cwd", str(new_wt)])
        if rc == 3:
            ok("rebind unknown id exits 3")
        else:
            no("rebind unknown id exits 3", f"rc={rc}")

        # 10b. ambiguous prefix: 'cli-' hits several cliSessionIds... use
        # wrapper-id ambiguity instead — 'aaaa' vs 'aaaa-moved' is unique, so
        # craft the collision on the shared '' prefix? No: use 'cli-m' which
        # prefixes cli-moved and cli-mismatch — two distinct sessions.
        rc, _, err = run_mode(env, ["rebind", "cli-m", "--cwd", str(new_wt)])
        if rc == 2 and "different sessions" in err:
            ok("rebind ambiguous prefix exits 2")
        else:
            no("rebind ambiguous prefix exits 2", f"rc={rc} err-tail={err[-200:]!r}")

        # 11. nonexistent target path
        ghost = tmp / "not-there-yet"
        rc, _, _ = run_mode(env, ["rebind", "aaaa-moved", "--cwd", str(ghost)])
        if rc == 3:
            ok("rebind to nonexistent path exits 3 without --force")
        else:
            no("rebind to nonexistent path exits 3 without --force", f"rc={rc}")

        # 12a. doctor pre-rebind: flags exactly the moved session, exit 10
        rc, out, _ = run_mode(env, ["doctor", "--json"])
        try:
            env_doc = json.loads(out)
        except json.JSONDecodeError:
            env_doc = {"data": [], "meta": {}}
        ids = [f["sessionId"] for f in env_doc.get("data", [])]
        if (rc == 10 and ids == ["local_aaaa-moved"]
                and env_doc["meta"].get("checked") == 3
                and env_doc["meta"].get("schema") == "claude-mods.summon.doctor/v1"):
            ok("doctor flags the broken cwd (exit 10, --json envelope)")
        else:
            no("doctor flags the broken cwd (exit 10, --json envelope)",
               f"rc={rc} ids={ids} meta={env_doc.get('meta')}")

        # 8. real rebind: wrapper rewritten, siblings rebased, backup taken,
        #    transcript bridged into enc(new cwd)
        rc, out, err = run_mode(env, ["rebind", "aaaa-moved", "--cwd", str(new_wt)])
        data = json.loads(wrapper_a.read_text(encoding="utf-8"))
        resolved_new_wt = str(new_wt.resolve())
        resolved_new_root = str(new_root.resolve())
        bridged = (sb["projects"] / encode_cwd(resolved_new_wt) / "cli-moved.jsonl")
        original = (sb["projects"] / encode_cwd(str(sb["old_wt"])) / "cli-moved.jsonl")
        backup_files = list(backups.rglob("*.json")) if backups.exists() else []
        checks = {
            "rc": rc == 0,
            "cwd": data["cwd"] == resolved_new_wt,
            "originCwd": data["originCwd"] == resolved_new_root,
            "worktreePath": data["worktreePath"] == resolved_new_wt,
            "backup": len(backup_files) == 1,
            "bridged": bridged.exists(),
            "copy-not-move": original.exists(),
        }
        if all(checks.values()):
            ok("rebind rewrites wrapper + backup + transcript bridge")
        else:
            no("rebind rewrites wrapper + backup + transcript bridge",
               f"failed={[k for k, v in checks.items() if not v]} rc={rc}")

        # 16. rebind into a worktree path prints the `git worktree repair` hint
        if "git worktree repair" in out and str(new_wt.resolve()) in out:
            ok("rebind worktree hint present (git worktree repair)")
        else:
            no("rebind worktree hint present (git worktree repair)",
               f"out-tail={out[-300:]!r}")

        # 12b. doctor post-rebind: healthy, exit 0
        rc, out, _ = run_mode(env, ["doctor", "--json"])
        try:
            env_doc = json.loads(out)
        except json.JSONDecodeError:
            env_doc = {"data": ["unparsed"]}
        if rc == 0 and env_doc.get("data") == []:
            ok("doctor healthy after rebind (exit 0)")
        else:
            no("doctor healthy after rebind (exit 0)",
               f"rc={rc} data={env_doc.get('data')}")

        # 11b. --force allows a not-yet-existing path (rebind back and forth)
        rc, _, _ = run_mode(env, ["rebind", "aaaa-moved", "--cwd", str(ghost), "--force"])
        data = json.loads(wrapper_a.read_text(encoding="utf-8"))
        if rc == 0 and data["cwd"] == str(ghost):
            ok("rebind --force accepts a nonexistent path")
        else:
            no("rebind --force accepts a nonexistent path", f"rc={rc} cwd={data['cwd']!r}")

        # 13. recover --no-distill: plain pointer prompt on stdout, scan
        #     fallback for the mismatched dir, no handover cache written
        rc, out, err = run_mode(env, ["recover", "cccc-mismatch", "--no-distill"])
        scan_path = sb["projects"] / "X--some-unrelated-munged-dir" / "cli-mismatch.jsonl"
        cache = scan_path.with_name(scan_path.name + ".handover.md")
        if (rc == 0 and str(scan_path) in out
                and "Continue a previous Claude session" in out
                and "TAIL of the transcript" in out
                and "Continue a previous" not in err
                and not cache.exists()):
            ok("recover --no-distill emits pointer prompt; scan fallback; no cache")
        else:
            no("recover --no-distill emits pointer prompt; scan fallback; no cache",
               f"rc={rc} cache={cache.exists()} out-head={out[:200]!r}")

        # 14. recover unknown id
        rc, _, _ = run_mode(env, ["recover", "zzzz-nope"])
        if rc == 3:
            ok("recover unknown id exits 3")
        else:
            no("recover unknown id exits 3", f"rc={rc}")

        # 15. pick --select 2 -> second-newest candidate (healthy session)
        rc, out, _ = run_mode(env, ["pick", "--select", "2", "--no-distill"])
        if rc == 0 and "healthy session" in out and "cli-healthy.jsonl" in out:
            ok("pick --select 2 recovers the 2nd candidate")
        else:
            no("pick --select 2 recovers the 2nd candidate",
               f"rc={rc} out-head={out[:200]!r}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def _load_summon_module():
    """Import summon.py as a module for unit-testing extraction (no side effects
    at import time beyond terminal-capability sniffing)."""
    import importlib.util
    spec = importlib.util.spec_from_file_location("summon_under_test", SCRIPT)
    assert spec is not None and spec.loader is not None
    mod = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = mod  # dataclass needs the module registered
    spec.loader.exec_module(mod)
    return mod


def extraction_tests() -> None:
    """17. extract_conversation: tool blobs skipped, budget respected."""
    mod = _load_summon_module()
    tmp = Path(tempfile.mkdtemp(prefix="summon-extract-"))
    try:
        tr = tmp / "fixture.jsonl"
        recs = [
            {"type": "user", "message": {"content": "Build the frobnicator widget"}},
            {"type": "assistant", "message": {"content": [
                {"type": "text", "text": "Plan: refactor the gadget first"},
                {"type": "tool_use", "name": "Bash",
                 "input": {"command": "echo TOOL-INPUT-NOISE && make all"}},
            ]}},
            {"type": "user", "message": {"content": [
                {"type": "tool_result", "tool_use_id": "t1",
                 "content": "TOOL-RESULT-BLOB " * 200},
            ]}},
            {"type": "summary", "summary": "not a conversation turn"},
            {"type": "assistant", "message": {"content": [
                {"type": "text", "text": "Done; committed abc1234 on lane/foo"},
            ]}},
        ]
        tr.write_text("\n".join(json.dumps(r) for r in recs) + "\n", encoding="utf-8")
        out = mod.extract_conversation(tr, 120_000)
        checks = {
            "user-text": "frobnicator" in out,
            "assistant-text": "abc1234" in out,
            "roles": "USER:" in out and "ASSISTANT:" in out,
            "no-tool-input": "TOOL-INPUT-NOISE" not in out,
            "no-tool-result": "TOOL-RESULT-BLOB" not in out,
        }
        if all(checks.values()):
            ok("extraction keeps text turns, skips tool_use/tool_result blobs")
        else:
            no("extraction keeps text turns, skips tool_use/tool_result blobs",
               f"failed={[k for k, v in checks.items() if not v]}")

        # Budget: 40 turns x ~1KB, budget 5000 -> verbatim tail wins, earliest
        # turns dropped, output within budget.
        tr2 = tmp / "long.jsonl"
        recs2 = [{"type": "user" if i % 2 == 0 else "assistant",
                  "message": {"content": f"turn-{i} " + "x" * 1000}}
                 for i in range(40)]
        tr2.write_text("\n".join(json.dumps(r) for r in recs2) + "\n", encoding="utf-8")
        out2 = mod.extract_conversation(tr2, 5_000)
        if len(out2) <= 5_000 and "turn-39" in out2 and "turn-0 " not in out2:
            ok("extraction respects char budget; verbatim tail wins")
        else:
            no("extraction respects char budget; verbatim tail wins",
               f"len={len(out2)} tail-present={'turn-39' in out2}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def make_claude_shim(shim_dir: Path, brief_file: Path | None, fail: bool = False) -> None:
    """Drop a fake `claude` onto PATH: prints brief_file's content, or exits 1."""
    if sys.platform == "win32":
        bat = shim_dir / "claude.bat"
        if fail:
            bat.write_text("@echo off\r\nexit /b 1\r\n", encoding="ascii")
        else:
            bat.write_text(f'@echo off\r\ntype "{brief_file}"\r\n', encoding="ascii")
    else:
        sh = shim_dir / "claude"
        if fail:
            sh.write_text("#!/bin/sh\ncat >/dev/null\nexit 1\n", encoding="ascii")
        else:
            sh.write_text(f'#!/bin/sh\ncat >/dev/null\ncat "{brief_file}"\n',
                          encoding="ascii")
        sh.chmod(0o755)


def distill_tests() -> None:
    """18-20. recover distillation: shim claude, cache lifecycle, degrade paths."""
    tmp = Path(tempfile.mkdtemp(prefix="summon-distill-"))
    try:
        sb = build_toolbox_sandbox(tmp)
        env = sb["env"]
        good = tmp / "proj-good"
        transcript = sb["projects"] / encode_cwd(str(good)) / "cli-healthy.jsonl"
        cache = transcript.with_name(transcript.name + ".handover.md")

        shim = tmp / "shim"
        shim.mkdir()
        brief_file = tmp / "brief.txt"
        brief_file.write_text("## Goal\nBRIEF-ONE\n", encoding="utf-8")
        make_claude_shim(shim, brief_file)
        env_shim = dict(env)
        env_shim["PATH"] = str(shim) + os.pathsep + env.get("PATH", "")

        # 18. distill + cache write + brief-and-pointer emission
        rc, out, err = run_mode(env_shim, ["recover", "bbbb-healthy"])
        if (rc == 0 and "BRIEF-ONE" in out
                and "consult it only if something specific is missing" in out
                and str(transcript) in out
                and cache.exists() and "BRIEF-ONE" in cache.read_text(encoding="utf-8")
                and "BRIEF-ONE" not in err):
            ok("recover distills via shim claude; brief + pointer; cache written")
        else:
            no("recover distills via shim claude; brief + pointer; cache written",
               f"rc={rc} cache={cache.exists()} out-head={out[:200]!r} err-tail={err[-200:]!r}")

        # 19a. cache hit: shim now yields BRIEF-TWO but the cache must win
        brief_file.write_text("## Goal\nBRIEF-TWO\n", encoding="utf-8")
        rc, out, err = run_mode(env_shim, ["recover", "bbbb-healthy"])
        if rc == 0 and "BRIEF-ONE" in out and "BRIEF-TWO" not in out and "cached" in err:
            ok("cache hit: unchanged transcript reuses cached brief")
        else:
            no("cache hit: unchanged transcript reuses cached brief",
               f"rc={rc} out-head={out[:200]!r}")

        # 19b. --refresh forces re-distillation
        rc, out, _ = run_mode(env_shim, ["recover", "bbbb-healthy", "--refresh"])
        if rc == 0 and "BRIEF-TWO" in out:
            ok("--refresh forces re-distillation")
        else:
            no("--refresh forces re-distillation", f"rc={rc} out-head={out[:200]!r}")

        # 19c. newer transcript mtime busts the cache
        brief_file.write_text("## Goal\nBRIEF-THREE\n", encoding="utf-8")
        newer = cache.stat().st_mtime + 10
        os.utime(transcript, (newer, newer))
        rc, out, _ = run_mode(env_shim, ["recover", "bbbb-healthy"])
        if rc == 0 and "BRIEF-THREE" in out:
            ok("newer transcript mtime busts the cache")
        else:
            no("newer transcript mtime busts the cache", f"rc={rc} out-head={out[:200]!r}")

        # 20a. degrade: claude absent from PATH -> pointer prompt, exit 0, warning
        emptybin = tmp / "emptybin"
        emptybin.mkdir()
        env_absent = dict(env)
        env_absent["PATH"] = str(emptybin)
        rc, out, err = run_mode(env_absent, ["recover", "cccc-mismatch"])
        if (rc == 0 and "TAIL of the transcript" in out
                and "not on PATH" in err):
            ok("degrade: absent claude falls back to pointer prompt (exit 0)")
        else:
            no("degrade: absent claude falls back to pointer prompt (exit 0)",
               f"rc={rc} err-tail={err[-300:]!r}")

        # 20b. degrade: failing claude (exit 1) -> same advisory fallback
        shim_fail = tmp / "shim-fail"
        shim_fail.mkdir()
        make_claude_shim(shim_fail, None, fail=True)
        env_fail = dict(env)
        env_fail["PATH"] = str(shim_fail)
        rc, out, err = run_mode(env_fail, ["recover", "cccc-mismatch"])
        if (rc == 0 and "TAIL of the transcript" in out
                and "exited 1" in err):
            ok("degrade: failing claude falls back to pointer prompt (exit 0)")
        else:
            no("degrade: failing claude falls back to pointer prompt (exit 0)",
               f"rc={rc} err-tail={err[-300:]!r}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def pick_json_tests() -> None:
    """21-22. pick --json: envelope shape, clean stdout, worktree split, brokenCwd."""
    import re
    tmp = Path(tempfile.mkdtemp(prefix="summon-pickjson-"))
    try:
        sb = build_toolbox_sandbox(tmp)
        env = sb["env"]

        # _is_live also checks wrapper mtime — the fixtures were just written,
        # so backdate them past the 10-minute liveness window.
        stale = time.time() - 3600
        for f in sb["ws"].glob("local_*.json"):
            os.utime(f, (stale, stale))

        # 21. parseable envelope, schema match, stdout clean of panel glyphs
        rc, out, _ = run_mode(env, ["pick", "--json"])
        try:
            envlp = json.loads(out)
        except json.JSONDecodeError:
            envlp = {}
        pure_ascii = all(ord(c) < 128 for c in out)
        checks = {
            "rc": rc == 0,
            "parses": bool(envlp),
            "schema": envlp.get("meta", {}).get("schema") == "claude-mods.summon.pick/v1",
            "count": envlp.get("meta", {}).get("count") == 3 == len(envlp.get("data", [])),
            "stdout-json-only": out.lstrip().startswith("{") and pure_ascii,
            "no-glyphs": not any(g in out for g in ("╭", "│", "╰", "├", "\x1b[")),
        }
        if all(checks.values()):
            ok("pick --json emits parseable pick/v1 envelope; stdout clean")
        else:
            no("pick --json emits parseable pick/v1 envelope; stdout clean",
               f"failed={[k for k, v in checks.items() if not v]} rc={rc} out-head={out[:200]!r}")
            return

        rows = {r["sessionId"]: r for r in envlp["data"]}
        moved = rows.get("local_aaaa-moved", {})
        healthy = rows.get("local_bbbb-healthy", {})
        mismatch = rows.get("local_cccc-mismatch", {})

        # 22a. worktree cwd split into projectRoot + worktree
        if (moved.get("projectRoot") == str(sb["old_root"])
                and moved.get("worktree") == "wt1"
                and healthy.get("worktree") == ""
                and healthy.get("projectRoot") == healthy.get("cwd")):
            ok("pick --json splits worktree cwd into projectRoot + worktree")
        else:
            no("pick --json splits worktree cwd into projectRoot + worktree",
               f"moved-root={moved.get('projectRoot')!r} wt={moved.get('worktree')!r}")

        # 22b. brokenCwd flags the moved session only
        if (moved.get("brokenCwd") is True and healthy.get("brokenCwd") is False
                and mismatch.get("brokenCwd") is False):
            ok("pick --json flags brokenCwd for the missing-cwd session")
        else:
            no("pick --json flags brokenCwd for the missing-cwd session",
               f"moved={moved.get('brokenCwd')} healthy={healthy.get('brokenCwd')}")

        # 22c. transcriptPath resolved (incl. scan fallback), booleans + ISO stamps
        scan_path = sb["projects"] / "X--some-unrelated-munged-dir" / "cli-mismatch.jsonl"
        iso_re = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$")
        checks = {
            "scan-fallback": mismatch.get("transcriptPath") == str(scan_path),
            "expected-path": str(healthy.get("transcriptPath") or "").endswith("cli-healthy.jsonl"),
            "branch": moved.get("branch") == "claude/wt1",
            "not-running": healthy.get("isRunning") is False,
            "not-archived": healthy.get("isArchived") is False,
            "iso": all(iso_re.match(r["lastActivityAt"]) for r in envlp["data"]),
            "short-id": moved.get("id") == "aaaa-mov",
        }
        if all(checks.values()):
            ok("pick --json resolves transcriptPath; ISO stamps + field types sane")
        else:
            no("pick --json resolves transcriptPath; ISO stamps + field types sane",
               f"failed={[k for k, v in checks.items() if not v]}")

        # 22d. --rich advances the schema to pick/v2 and adds display metrics
        rc, out, _ = run_mode(env, ["pick", "--json", "--rich"])
        try:
            rich = json.loads(out)
        except json.JSONDecodeError:
            rich = {}
        rrows = rich.get("data", [])
        checks = {
            "rc": rc == 0,
            "schema": rich.get("meta", {}).get("schema") == "claude-mods.summon.pick/v2",
            "count": len(rrows) == 3,
            "metric-keys": all(
                all(k in r for k in ("events", "toolCalls", "densityBuckets",
                                     "durationMin", "sizeKB", "ctxTokens", "ctxPeak",
                                     "ctxWindow", "ctxPct", "ctxPeakPct", "firstAsk",
                                     "model", "effort"))
                for r in rrows),
            "buckets-24": all(isinstance(r.get("densityBuckets"), list)
                              and len(r["densityBuckets"]) == 24 for r in rrows),
            "window-standard": all(r.get("ctxWindow") in (200000, 1000000) for r in rrows),
            "pct-bounded": all(0 <= r.get("ctxPct", -1) <= 100 for r in rrows),
        }
        if all(checks.values()):
            ok("pick --json --rich emits pick/v2 with per-session display metrics")
        else:
            no("pick --json --rich emits pick/v2 with per-session display metrics",
               f"failed={[k for k, v in checks.items() if not v]} rc={rc}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def build_widget_sandbox(tmp: Path) -> dict:
    """One account, three real sessions + one 0-turn stub, all stale (not
    running), transcripts present — the fixture for the widget builder tests."""
    home = tmp / "home"
    appdata = home / "AppData" / "Roaming"
    env = os.environ.copy()
    env["HOME"] = str(home)
    env["USERPROFILE"] = str(home)
    env["APPDATA"] = str(appdata)
    env.pop("PYTHONIOENCODING", None)
    env.pop("TERM_ASCII", None)

    cdir = claude_dir(env)
    projects = home / ".claude" / "projects"
    now_ms = int(time.time() * 1000)

    ws = cdir / "claude-code-sessions" / SRC_UUID / "11111111-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
    ws.mkdir(parents=True)
    proj = tmp / "proj"
    proj.mkdir()

    # (sid, cli, title, turns, age_min) — newest first; ages > 10m so none live.
    specs = [
        ("w0", "cli-w0", "widget alpha", 5, 20),
        ("w1", "cli-w1", "widget beta", 3, 30),
        ("w2", "cli-w2", "widget gamma", 9, 40),
        ("wstub", "cli-wstub", "empty stub", 0, 50),  # 0-turn -> dropped by default
    ]
    for sid, cli, title, turns, age in specs:
        (ws / f"local_{sid}.json").write_text(json.dumps({
            "sessionId": f"local_{sid}", "cliSessionId": cli, "title": title,
            "cwd": str(proj), "lastActivityAt": now_ms - age * 60_000,
            "completedTurns": turns, "model": "claude-opus-4-8", "effort": "high",
        }), encoding="utf-8")

    enc_dir = projects / encode_cwd(str(proj))
    enc_dir.mkdir(parents=True)
    for _, cli, *_ in specs:
        (enc_dir / f"{cli}.jsonl").write_text(
            '{"type":"user","message":{"content":"hello there"},"timestamp":"2026-07-01T00:00:00Z"}\n'
            '{"type":"assistant","message":{"content":[{"type":"text","text":"hi"}],'
            '"usage":{"input_tokens":100,"output_tokens":50}},"timestamp":"2026-07-01T00:05:00Z"}\n',
            encoding="utf-8")

    # Backdate wrapper mtimes past the 10-minute liveness window (the stub must
    # read as NOT running so it is dropped by default).
    stale = time.time() - 3600
    for f in ws.glob("local_*.json"):
        os.utime(f, (stale, stale))

    return {"env": env, "ws": ws, "proj": proj, "projects": projects}


def _widget_data_array(html: str):
    """Extract + parse the injected <script id="D"> JSON array; also return the
    raw block so line-formatting can be asserted."""
    start = '<script id="D" type="application/json">'
    i = html.find(start)
    j = html.find("</script>", i + len(start)) if i >= 0 else -1
    if i < 0 or j < 0:
        return "", None
    block = html[i + len(start):j]
    try:
        return block, json.loads(block)
    except json.JSONDecodeError:
        return block, None


def widget_tests() -> None:
    """24. summon widget: one-shot finished card-picker HTML.

    Asserts the builder replaces the <script id="D"> payload, drops 0-turn
    stubs by default (--include-stubs keeps them), honours --limit, keeps only
    the whitelisted keys, emits one session object per line, sets the window
    <select>, and holds the assembled HTML under the byte budget (capping with
    a stderr note when it can't).
    """
    mod = _load_summon_module()
    allowed = set(mod.WIDGET_KEYS)
    tmp = Path(tempfile.mkdtemp(prefix="summon-widget-"))
    try:
        sb = build_widget_sandbox(tmp)
        env = sb["env"]
        out_html = tmp / "w.html"

        # A) default run: placeholder replaced, stub dropped, keys whitelisted,
        #    one-object-per-line, window set, under budget, mirror written.
        rc, out, err = run_mode(env, ["widget", "--out", str(out_html)])
        block, arr = _widget_data_array(out)
        ids = [r["sessionId"] for r in arr] if arr else []
        data_lines = [ln for ln in block.splitlines() if ln.strip().startswith("{")]
        checks = {
            "rc": rc == 0,
            "parses": arr is not None,
            "placeholder-gone": "local_0000" not in out,
            "payload-present": "widget alpha" in out,
            "stub-dropped": "local_wstub" not in ids,
            "three-real": ids == ["local_w0", "local_w1", "local_w2"],
            "keys-whitelisted": arr is not None and all(set(r).issubset(allowed) for r in arr),
            "dead-keys-gone": arr is not None and all(
                "cliSessionId" not in r and "transcriptPath" not in r
                and "branch" not in r for r in arr),
            "one-per-line": arr is not None and len(data_lines) == len(arr),
            "window-30d": '<option value="720" selected>' in out,
            "under-budget": len(out.encode("utf-8")) <= 28 * 1024,
            "mirror-written": out_html.is_file() and "widget alpha" in out_html.read_text(encoding="utf-8"),
            "stdout-not-stderr": "widget alpha" not in err,
        }
        if all(checks.values()):
            ok("widget builds finished HTML: stub-drop, whitelist, 1/line, budget")
        else:
            no("widget builds finished HTML: stub-drop, whitelist, 1/line, budget",
               f"failed={[k for k, v in checks.items() if not v]} rc={rc} err-tail={err[-200:]!r}")

        # B) --limit caps the session count.
        rc, out, _ = run_mode(env, ["widget", "--limit", "2", "--out", str(out_html)])
        _, arr = _widget_data_array(out)
        if rc == 0 and arr is not None and len(arr) == 2 and [r["sessionId"] for r in arr] == ["local_w0", "local_w1"]:
            ok("widget --limit caps to the N most-recently-active sessions")
        else:
            no("widget --limit caps to the N most-recently-active sessions",
               f"rc={rc} n={len(arr) if arr else None}")

        # C) --include-stubs keeps the 0-turn session.
        rc, out, _ = run_mode(env, ["widget", "--include-stubs", "--out", str(out_html)])
        _, arr = _widget_data_array(out)
        if rc == 0 and arr is not None and "local_wstub" in [r["sessionId"] for r in arr]:
            ok("widget --include-stubs keeps 0-turn sessions")
        else:
            no("widget --include-stubs keeps 0-turn sessions",
               f"rc={rc} ids={[r['sessionId'] for r in arr] if arr else None}")

        # D) density downsampled to <=12 by default, full 24 with --full-density.
        rc, out, _ = run_mode(env, ["widget", "--out", str(out_html)])
        _, arr = _widget_data_array(out)
        default_ds = len(arr[0]["densityBuckets"]) if arr else -1
        rc2, out2, _ = run_mode(env, ["widget", "--full-density", "--out", str(out_html)])
        _, arr2 = _widget_data_array(out2)
        full_ds = len(arr2[0]["densityBuckets"]) if arr2 else -1
        if rc == 0 and rc2 == 0 and default_ds == 12 and full_ds == 24:
            ok("widget downsamples density 24->12 (default); --full-density keeps 24")
        else:
            no("widget downsamples density 24->12 (default); --full-density keeps 24",
               f"default={default_ds} full={full_ds}")

        # E) a tight budget forces capping with a stderr note (never spools).
        rc, out, err = run_mode(env, ["widget", "--max-kb", "0.001", "--out", str(out_html)])
        _, arr = _widget_data_array(out)
        if rc == 0 and arr is not None and len(arr) == 1 and "capped" in err:
            ok("widget honours --max-kb: caps to fit + notes the cap on stderr")
        else:
            no("widget honours --max-kb: caps to fit + notes the cap on stderr",
               f"rc={rc} n={len(arr) if arr else None} err-tail={err[-200:]!r}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)


def asset_tests() -> None:
    """In-chat picker asset: present, injectable, and cited from SKILL.md.

    The injection marker is the `<script id="D">` JSON block (the original
    `>>> INJECT <<<` comment was replaced when the card picker was rebuilt
    in 551292e) — keep this assertion in sync with SKILL.md's render step.
    """
    name = "picker-widget.html asset present + <script id=\"D\"> block + cited from SKILL.md"
    asset = HERE.parent / "assets" / "picker-widget.html"
    skill = HERE.parent / "SKILL.md"
    checks = {}
    checks["asset exists"] = asset.is_file()
    if checks["asset exists"]:
        html = asset.read_text(encoding="utf-8")
        checks["script id=D block"] = '<script id="D"' in html
        checks["sendPrompt wiring"] = "sendPrompt" in html
    checks["cited from SKILL.md"] = (
        skill.is_file() and "assets/picker-widget.html" in skill.read_text(encoding="utf-8")
    )
    if all(checks.values()):
        ok(name)
    else:
        no(name, f"failed={[k for k, v in checks.items() if not v]}")


def main() -> int:
    # 1. Piped selection honoured even with --yes (the original bug: --yes
    #    used to select ALL candidates, ignoring the piped picks).
    tmp, env, _, dest_ws = with_sandbox()
    try:
        rc, out = run_summon(env, ["--yes"], stdin_text="1,3\n")
        got = copied_titles(dest_ws)
        if rc == 0 and got == {"alpha", "gamma"}:
            ok("--yes honours piped selection (copies 2 of 3)")
        else:
            no("--yes honours piped selection", f"rc={rc} copied={sorted(got)}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    # 2. --select answers the picker without stdin.
    tmp, env, _, dest_ws = with_sandbox()
    try:
        rc, out = run_summon(env, ["--select", "2", "--yes"])
        got = copied_titles(dest_ws)
        if rc == 0 and got == {"beta"}:
            ok("--select 2 copies exactly session #2")
        else:
            no("--select 2 copies exactly session #2", f"rc={rc} copied={sorted(got)}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    # 3. --select all selects everything.
    tmp, env, _, dest_ws = with_sandbox()
    try:
        rc, out = run_summon(env, ["--select", "all", "--yes"])
        got = copied_titles(dest_ws)
        if rc == 0 and got == {"alpha", "beta", "gamma"}:
            ok("--select all copies all 3")
        else:
            no("--select all copies all 3", f"rc={rc} copied={sorted(got)}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    # 4. Without --yes, declining the confirmation cancels.
    tmp, env, _, dest_ws = with_sandbox()
    try:
        rc, out = run_summon(env, [], stdin_text="1,2\nn\n")
        got = copied_titles(dest_ws)
        if rc == 0 and not got and "cancelled" in out:
            ok("confirmation 'n' cancels, nothing copied")
        else:
            no("confirmation 'n' cancels, nothing copied",
               f"rc={rc} copied={sorted(got)}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    # 5. Without --yes, confirming 'y' proceeds.
    tmp, env, _, dest_ws = with_sandbox()
    try:
        rc, out = run_summon(env, [], stdin_text="1\ny\n")
        got = copied_titles(dest_ws)
        if rc == 0 and got == {"alpha"}:
            ok("confirmation 'y' proceeds with the pick")
        else:
            no("confirmation 'y' proceeds with the pick",
               f"rc={rc} copied={sorted(got)}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    # 6. --yes with empty stdin cancels — must NOT fall back to select-all.
    tmp, env, _, dest_ws = with_sandbox()
    try:
        rc, out = run_summon(env, ["--yes"], stdin_text="")
        got = copied_titles(dest_ws)
        if rc == 0 and not got and "cancelled" in out:
            ok("--yes with no selection cancels (no auto-all)")
        else:
            no("--yes with no selection cancels (no auto-all)",
               f"rc={rc} copied={sorted(got)}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    # 7. Unicode title on cp1252 stdout must not crash (UnicodeEncodeError
    #    regression: U+2192 in a title with Windows cp1252 console encoding).
    tmp, env, _, dest_ws = with_sandbox(
        titles=("Update docs: Dagu → process-compose gallery", "beta", "gamma"))
    try:
        env_cp = dict(env)
        env_cp["PYTHONIOENCODING"] = "cp1252"
        rc, out = run_summon(env_cp, ["--select", "all", "--yes", "--dry-run"])
        if rc == 0 and "would copy" in out:
            ok("cp1252 stdout survives U+2192 in session title")
        else:
            no("cp1252 stdout survives U+2192 in session title",
               f"rc={rc} out-tail={out[-300:]!r}")
    finally:
        shutil.rmtree(tmp, ignore_errors=True)

    # 8-16. Toolbox modes: rebind / recover / pick / doctor
    toolbox_tests()

    # 17. Extraction (in-process unit tests)
    extraction_tests()

    # 18-20. Distilled handover: shim claude, cache lifecycle, degrade paths
    distill_tests()

    # 21-22. pick --json: envelope, clean stdout, worktree split, brokenCwd
    pick_json_tests()

    # 23. In-chat picker asset: present + injectable + cited from SKILL.md
    asset_tests()

    # 24. summon widget: one-shot finished card-picker HTML builder
    widget_tests()

    print(f"\nsummon tests: {PASS} passed, {FAIL} failed")
    return 1 if FAIL else 0


if __name__ == "__main__":
    sys.exit(main())