#!/usr/bin/env python3
"""
agsearch — global full-text search across all your coding agent sessions.

Every coding agent keeps its sessions on disk. Their native pickers search session
*metadata*: the title, the first prompt, the branch. This searches what was actually *said*,
across all of them at once, and drops you back into the session with that tool's own
resume command.

    claude       ~/.claude/projects        claude --resume
    codex        ~/.codex/sessions         codex resume
    cursor-cli   ~/.cursor/projects        cursor-agent --resume
    opencode     ~/.local/share/opencode   opencode --session
    gemini-cli   ~/.gemini/tmp             gemini --session-file

Usage:
    agsearch                       # interactive fuzzy TUI (needs fzf)
    agsearch "stripe tax id"       # open the TUI pre-filtered to this query
    agsearch -n "stripe tax id"    # non-interactive: ranked sessions as plain text, no fzf
    agsearch read <session-id>     # print a whole session, without resuming it
    agsearch --here "..."          # only sessions from the current directory's project
    agsearch --project myapp       # only sessions whose path matches 'myapp'
    agsearch --thinking            # also index assistant thinking blocks
    agsearch --reindex             # force a full rebuild of the cache
    agsearch --version             # print the installed version and exit
    agsearch _preview <sid> <seq> <thinking> <query...>   # (internal) fzf preview

`-n` prints one entry per session, led by its session id, and drops colour when it is
not writing to a terminal — so a script or a coding agent can search, then read a hit
with `agsearch read <session-id>`. Piped, ids shorten to a unique prefix, the columns
lose their padding and `read` caps its output; a terminal sees none of that.

Claude can drive that loop itself. `/plugin marketplace add devcodes9/agsearch` then
`/plugin install agsearch@agsearch` installs a skill that searches these sessions when
you refer to earlier work, instead of answering that it has no record of it.

In the TUI the right pane previews the matched session, auto-scrolled to your match
(marked ▶) with a "match N of M" header. Enter resumes the session (and copies your
query to the clipboard, so ⌘F → ⌘V → Enter jumps to it inside the replayed transcript).
Resume is id-based: a session whose project dir was deleted still resumes, from the nearest
surviving ancestor dir. Sessions that still look live are marked ● and confirm before resuming.

Warm runs are near-instant: parsed sessions are cached per-file and only re-parsed
when their .jsonl mtime changes.
"""

__version__ = "0.2.0"

import os
import re
import sys
import math
import json
import shlex
import time
import shutil
import hashlib
import subprocess

HOME = os.path.expanduser("~")
PROJECTS_DIR = os.path.join(HOME, ".claude", "projects")
CODEX_DIR = os.path.join(HOME, ".codex", "sessions")
GEMINI_DIR = os.path.join(HOME, ".gemini", "tmp")
CURSOR_DIR = os.path.join(HOME, ".cursor", "projects")
CURSOR_CHATS_DIR = os.path.join(HOME, ".cursor", "chats")   # titles only; see _cursor_titles
OPENCODE_DIR = os.path.join(HOME, ".local", "share", "opencode")
CACHE_DIR = os.path.join(os.environ.get("XDG_CACHE_HOME", os.path.join(HOME, ".cache")), "agsearch")
FRAG_DIR = os.path.join(CACHE_DIR, "frag")
META_PATH = os.path.join(CACHE_DIR, "meta.json")
SESSIONS_PATH = os.path.join(CACHE_DIR, "sessions.tsv")   # one line per session, for _filter
SUBMAP_PATH = os.path.join(CACHE_DIR, "submap.json")      # parent-sid -> [subagent file paths]
INDEX_PATH = os.path.join(CACHE_DIR, "index.json")        # sid -> {source, path} for preview/resume
FORKS_PATH = os.path.join(CACHE_DIR, "forks.json")        # forked sid -> {of, at}

CACHE_FMT = 7   # bump when the TSV column layout / keying changes, to invalidate old fragments

# TSV columns (tab-separated, one row per message):
#   0 session_id  1 cwd  2 gitBranch  3 iso_date  4 role  5 seq  6 title  7 text
SEP = "\t"

# Preview layout: this many header lines print before the first message, so a message
# with sequence `seq` lands on preview line HEADER_LINES + 1 + seq. Keep in sync with
# render_preview() — the fzf scroll offset is computed from it.
HEADER_LINES = 5

# A session whose transcript was appended to this recently is almost certainly still running
# (an agent mid-turn, or a CLI you have open in another tab). Marked ● live in the list.
ACTIVE_WINDOW_SEC = 180

# sessions.tsv columns (written by build_sessions, read by cmd_filter).
C_SID, C_CWD, C_DATE, C_SOURCE, C_KIND, C_TITLE, C_FIRST, C_BLOB = range(8)
SESSION_COLS = 8

# Ranking knobs — see rank_sessions(). k1/b are textbook BM25; b<1 keeps length normalization
# firm but not brutal, since a long session that genuinely discusses a term should still win
# over a short one that name-drops it. Fields are weighted: what you opened the session ASKING
# for (first prompt) beats what got said somewhere in hour three.
BM25_K1 = 1.2
BM25_B = 0.65
W_TITLE = 2.0
W_FIRST = 2.5
W_BODY = 1.0
# How much of each message is searchable. _single_line defaults to 400, which is right for a
# preview line and wrong for an index: 41% of messages are longer than that, and capping there
# left only 19% of transcript text searchable at all. A table or a summary a few hundred
# characters into a reply was simply absent from the index, so no amount of ranking could
# surface it. 4,000 recovers that; 20,000 measured no better and costs more to scan.
MSG_INDEX_CHARS = 4_000

RECENCY_HALFLIFE_DAYS = 45.0
RECENCY_W = 0.35          # max multiplier bump for a session from today
USAGE_W = 0.30            # max multiplier bump for a session you resume often


# ------------------------------------------------------------------ parsing

def _flatten_content(content):
    """Pull human-readable text out of a message.content (str or block list)."""
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for b in content:
            if isinstance(b, str):
                parts.append(b)
            elif isinstance(b, dict):
                t = b.get("type")
                if t == "text" and b.get("text"):
                    parts.append(b["text"])
                elif t == "tool_result":
                    parts.append(_flatten_content(b.get("content", "")))
        return "\n".join(p for p in parts if p)
    return ""


def _single_line(s, limit=400):
    return " ".join(s.split())[:limit]


def parse_session(path, include_thinking=False, limit=MSG_INDEX_CHARS):
    """Parse one .jsonl file into ordered index rows (8-field lists, seq assigned).

    Rows are keyed by each entry's `sessionId` field, not the filename. For normal sessions
    those are identical; for `agent-*.jsonl` subagent transcripts the sessionId points to the
    PARENT conversation, so subagent content folds into (and resumes) its parent session.
    """
    file_id = os.path.splitext(os.path.basename(path))[0]
    session_id = file_id
    title = ""
    cwd = ""
    branch = ""
    entry = ""
    rows = []   # each: [sid, cwd, branch, ts, role, title, text]
    try:
        fh = open(path, "r", errors="replace")
    except OSError:
        return session_id, []
    with fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                o = json.loads(line)
            except json.JSONDecodeError:
                continue
            t = o.get("type")
            if t == "ai-title" and o.get("aiTitle"):
                title = o["aiTitle"]
                continue
            if t not in ("user", "assistant"):
                continue
            # Claude Code flags injected content as meta: a skill's whole SKILL.md pasted in
            # as a user turn, hook notices, image-paste markers. It is filed under the user's
            # role but nobody typed it, and a single skill injection runs to 15k characters,
            # which buries the real prompt in the preview and the matched line.
            if o.get("isMeta"):
                continue
            session_id = o.get("sessionId") or session_id   # parent id for agent-* files
            cwd = o.get("cwd", cwd)
            branch = o.get("gitBranch", branch)
            entry = o.get("entrypoint") or entry            # cli vs sdk-py/sdk-ts
            msg = o.get("message", {}) or {}
            role = msg.get("role", t)
            # Claude Code files a tool's output under the user's role. Left as "user" the
            # preview tells the reader you said "The file has been updated successfully",
            # which is the one thing a transcript must not get wrong. Nine in ten user-role
            # entries are this. Still indexed: an error string a search has to find lives in
            # tool output, not in what anyone typed.
            content = msg.get("content")
            if role == "user" and isinstance(content, list) and any(
                    isinstance(b, dict) and b.get("type") == "tool_result" for b in content):
                role = "tool"
            ts = o.get("timestamp", "")

            if t == "assistant" and include_thinking:
                content = msg.get("content")
                if isinstance(content, list):
                    for b in content:
                        if isinstance(b, dict) and b.get("type") == "thinking" and b.get("thinking"):
                            rows.append([session_id, cwd, branch, ts, "thinking", title,
                                         _single_line(b["thinking"])])

            text = _single_line(_flatten_content(msg.get("content", "")), limit)
            if not text:
                continue
            rows.append([session_id, cwd, branch, ts, role, title, text])

    # Titles can appear after the messages they label; backfill, then stamp sequence.
    # `kind` marks who drove the session: your own CLI use vs a plugin/SDK-spawned run.
    kind = "auto" if entry.startswith("sdk") else "cli"
    final = []
    for i, r in enumerate(rows):
        final.append([r[0], r[1], r[2], r[3], r[4], str(i), title or r[5], r[6], kind])
    return session_id, final


_CODEX_NOISE = ("<environment_context>", "<permissions", "<multi_agent_mode>", "<user_instructions>")

# Codex user messages often lead with injected preambles instead of the real task. Pick the
# first message that's an actual request: strip a known security-preamble prefix, and skip
# messages that are pure wrappers (AGENTS.md dumps, XML blocks, leftover preamble).
_CODEX_PREAMBLE_ANCHOR = "repository code only."   # end of the injected security preamble
_CODEX_SKIP_PREFIXES = ("<", "# agents.md", "agents.md instructions", "important: do not")


def _codex_title(user_texts):
    for t in user_texts:
        t = t.strip()
        if _CODEX_PREAMBLE_ANCHOR in t:            # real task is appended after the preamble
            t = t.split(_CODEX_PREAMBLE_ANCHOR, 1)[1].strip()
        if not t or t.lower().startswith(_CODEX_SKIP_PREFIXES):
            continue                               # pure boilerplate → try the next message
        return t[:90]
    return ""


def parse_codex_session(path, include_thinking=False, limit=MSG_INDEX_CHARS):
    """Parse an OpenAI Codex CLI rollout file into the same 8-field row schema as Claude.

    Session id/cwd come from the `session_meta` entry; messages are `response_item` entries of
    payload type "message" (roles user/assistant; `developer`/`system` and injected context are
    dropped). Keyed by the meta `id` UUID, which is what `codex resume <id>` takes.
    """
    sid = os.path.splitext(os.path.basename(path))[0]
    cwd = branch = ts0 = ""
    rows = []
    try:
        fh = open(path, "r", errors="replace")
    except OSError:
        return sid, []
    with fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                o = json.loads(line)
            except json.JSONDecodeError:
                continue
            t = o.get("type")
            if t == "session_meta":
                p = o.get("payload", {}) or {}
                sid = p.get("id") or p.get("session_id") or sid
                cwd = p.get("cwd", cwd)
                ts0 = p.get("timestamp", ts0)
                g = p.get("git")
                if isinstance(g, dict):
                    branch = g.get("branch") or branch
                continue
            if t != "response_item":
                continue
            p = o.get("payload", {}) or {}
            if p.get("type") != "message" or p.get("role") not in ("user", "assistant"):
                continue
            c = p.get("content")
            if isinstance(c, str):
                text = c
            elif isinstance(c, list):
                text = " ".join(b.get("text", "") for b in c if isinstance(b, dict) and b.get("text"))
            else:
                text = ""
            text = _single_line(text, limit)
            if not text or text.lstrip().startswith(_CODEX_NOISE):
                continue
            rows.append([sid, cwd, branch, o.get("timestamp") or ts0, p["role"], "", "", text])

    title = _codex_title([r[7] for r in rows if r[4] == "user"])
    final = [[sid, r[1], r[2], r[3], r[4], str(i), title, r[7], "cli"] for i, r in enumerate(rows)]
    return sid, final


# ------------------------------------------------------------------ gemini cli

# Gemini writes a session as ONE json object, not jsonl, and tags every entry with a `type`
# rather than a role. Only the two conversational types are indexed: `info`, `error` and the
# rest are CLI chrome (auth prompts, update notices) that would match queries and mean nothing.
_GEMINI_ROLE = {"user": "user", "gemini": "assistant", "model": "assistant",
                "assistant": "assistant"}


def _gemini_cwd(path):
    """Gemini records a `projectHash`, never the directory it ran in.

    The transcript lives at ~/.gemini/tmp/<project>/chats/<session>.json, and the sibling
    `.project_root` file holds the real absolute path. Without it there is nothing to recover:
    the hash is a sha256 and the directory name is a basename, not a path.
    """
    proj = os.path.dirname(os.path.dirname(path))         # .../tmp/<project>
    try:
        with open(os.path.join(proj, ".project_root"), errors="replace") as fh:
            return fh.read().strip()
    except OSError:
        return ""


def parse_gemini_session(path, include_thinking=False, limit=MSG_INDEX_CHARS):
    """Parse one Gemini CLI chat json into the shared 9-field row schema.

    Keyed by the `sessionId` field. Resume is by file path (`gemini --session-file`), not by id,
    so the id here is for display and dedupe only.
    """
    sid = os.path.splitext(os.path.basename(path))[0]
    try:
        with open(path, errors="replace") as fh:
            doc = json.load(fh)
    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
        return sid, []
    if not isinstance(doc, dict):
        return sid, []

    sid = doc.get("sessionId") or sid
    cwd = _gemini_cwd(path)
    ts0 = doc.get("startTime") or doc.get("lastUpdated") or ""

    rows = []
    for m in doc.get("messages", []):
        if not isinstance(m, dict):
            continue
        role = _GEMINI_ROLE.get(m.get("type"))
        if not role:
            continue
        text = _single_line(_flatten_content(m.get("content", "")), limit)
        if not text:
            continue
        rows.append([sid, cwd, "", m.get("timestamp") or ts0, role, "", "", text])

    title = ""
    for r in rows:
        if r[4] == "user":
            title = r[7][:90]
            break
    return sid, [[sid, r[1], r[2], r[3], r[4], str(i), title, r[7], "cli"]
                 for i, r in enumerate(rows)]


# ------------------------------------------------------------------ cursor

# Cursor writes each session twice: a SQLite store under ~/.cursor/chats holding the raw API
# payloads, and a clean JSONL transcript under ~/.cursor/projects/<slug>/agent-transcripts/.
# The JSONL is the better source in every way that matters here. It covers more sessions (all
# of the SQLite ones and more), it is ordered, and it carries only the conversation: the
# SQLite copy also contains injected hook and environment context, which lands in the index as
# session titles like `<hooks_context description="...">`.

_CURSOR_QUERY = re.compile(r"<user_query>(.*?)</user_query>", re.DOTALL)
# Tags carry attributes (`<hooks_context description="...">`), so a bare-tag pattern misses them.
_CURSOR_TAG_BLOCK = re.compile(r"<([a-z_]+)(?:\s[^>]*)?>.*?</\1>", re.DOTALL)


def _cursor_text(text, role):
    """The words a human would recognise as the turn.

    Cursor wraps what the user typed in <user_query>, and surrounds it with attachments,
    timestamps and skill lists. Indexed whole, those swamp the actual prompt.
    """
    if role == "user":
        found = _CURSOR_QUERY.findall(text)
        if found:
            return " ".join(f.strip() for f in found)
    prev = None
    while prev != text:
        prev = text
        text = _CURSOR_TAG_BLOCK.sub(" ", text)
    return text


_UNSLUG_CACHE = {}


def _unslug(slug):
    """Turn a Cursor project directory name back into the path it was made from.

    The name is the path with every non-alphanumeric replaced by `-`, which is not reversible
    on its own: a real dash is indistinguishable from a separator. Resolve it against the
    filesystem instead, taking the longest prefix that exists at each step. Returns "" for a
    directory that is gone, which is a display gap, not a reason to skip the session.
    """
    if slug in _UNSLUG_CACHE:
        return _UNSLUG_CACHE[slug]
    parts = [p for p in slug.split("-") if p]
    path, i = "", 0
    while i < len(parts):
        for j in range(len(parts), i, -1):
            cand = path + os.sep + "-".join(parts[i:j])
            if os.path.isdir(cand):
                path, i = cand, j
                break
        else:
            path = ""
            break
    _UNSLUG_CACHE[slug] = path
    return path


_CURSOR_TITLES = None


def _cursor_titles():
    """Chat id -> title, from the SQLite side's meta.json files.

    The JSONL transcript has no title of its own, and Cursor's own titles read far better than
    a truncated first prompt. Read once: there is one small file per chat.
    """
    global _CURSOR_TITLES
    if _CURSOR_TITLES is None:
        _CURSOR_TITLES = {}
        for root, _dirs, files in os.walk(CURSOR_CHATS_DIR):
            if "meta.json" not in files:
                continue
            try:
                with open(os.path.join(root, "meta.json"), errors="replace") as fh:
                    meta = json.load(fh)
            except (OSError, json.JSONDecodeError, UnicodeDecodeError):
                continue
            if isinstance(meta, dict) and meta.get("title"):
                _CURSOR_TITLES[os.path.basename(root)] = _single_line(meta["title"], 90)
    return _CURSOR_TITLES


def parse_cursor_session(path, include_thinking=False, limit=MSG_INDEX_CHARS):
    """Parse one Cursor transcript into the shared 9-field row schema.

    Keyed by the file stem, which is the chat id `cursor-agent --resume <id>` takes. The
    transcript records no per-message time, so every row carries the file's mtime: that is when
    the session was last written to, which is what the date column and the recency boost want.
    """
    sid = os.path.splitext(os.path.basename(path))[0]
    proj = os.path.basename(os.path.dirname(os.path.dirname(os.path.dirname(path))))
    cwd = _unslug(proj)
    try:
        ts = time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(os.path.getmtime(path)))
    except (OSError, ValueError):
        ts = ""

    rows = []
    try:
        fh = open(path, errors="replace")
    except OSError:
        return sid, []
    with fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                o = json.loads(line)
            except json.JSONDecodeError:
                continue
            role = o.get("role")
            if role not in ("user", "assistant"):     # status and error entries are not turns
                continue
            msg = o.get("message")
            if not isinstance(msg, dict):
                continue
            text = _single_line(_cursor_text(_flatten_content(msg.get("content", "")), role),
                                limit)
            if not text:
                continue
            rows.append([sid, cwd, "", ts, role, "", "", text])

    title = _cursor_titles().get(sid, "")
    if not title:
        title = next((r[7][:90] for r in rows if r[4] == "user"), "")
    return sid, [[sid, r[1], r[2], r[3], r[4], str(i), title, r[7], "cli"]
                 for i, r in enumerate(rows)]


# ------------------------------------------------------------------ opencode

# opencode keeps every session in one SQLite database rather than a file per session, so this
# parser returns rows for all of them at once and the indexer registers each session it finds.
# Message text lives in `part`, one row per span, with the role on the parent `message`.
_OPENCODE_SQL = """
SELECT p.session_id, m.data, p.data
FROM part p JOIN message m ON p.message_id = m.id
ORDER BY m.time_created, p.time_created, p.id
"""
_OPENCODE_SESSIONS = "SELECT id, directory, title, time_updated FROM session"


def _sqlite_ro(path):
    """Open a harness database without ever taking a write lock on one it may still be using."""
    import sqlite3
    for uri in ("file:%s?mode=ro" % path, "file:%s?immutable=1" % path):
        try:
            return sqlite3.connect(uri, uri=True, timeout=1.0)
        except sqlite3.Error:
            continue
    return None


def _opencode_iso(ms):
    try:
        return time.strftime("%Y-%m-%dT%H:%M:%S", time.localtime(ms / 1000.0)) if ms else ""
    except (TypeError, ValueError, OSError):
        return ""


def parse_opencode_session(path, include_thinking=False, limit=MSG_INDEX_CHARS):
    """Parse the opencode database into the shared 9-field row schema, all sessions at once."""
    conn = _sqlite_ro(path)
    if conn is None:
        return "", []

    wanted = {"text", "reasoning"} if include_thinking else {"text"}
    per_session = {}
    try:
        meta = {}
        for sid, directory, title, updated in conn.execute(_OPENCODE_SESSIONS):
            meta[sid] = (directory or "", _single_line(title or "", 90), _opencode_iso(updated))
        for sid, mdata, pdata in conn.execute(_OPENCODE_SQL):
            if sid not in meta:
                continue
            try:
                part = json.loads(pdata)
                msg = json.loads(mdata)
            except (json.JSONDecodeError, TypeError, ValueError):
                continue
            if not isinstance(part, dict) or part.get("type") not in wanted:
                continue
            role = msg.get("role") if isinstance(msg, dict) else None
            if role not in ("user", "assistant"):
                continue
            text = _single_line(part.get("text") or "", limit)
            if not text:
                continue
            per_session.setdefault(sid, []).append((role, text))
    except Exception:            # a database mid-write is not worth crashing the whole index on
        pass
    finally:
        conn.close()

    rows = []
    for sid, turns in per_session.items():
        cwd, title, ts = meta.get(sid, ("", "", ""))
        if not title:
            title = next((t for r, t in turns if r == "user"), "")[:90]
        for i, (role, text) in enumerate(turns):
            rows.append([sid, cwd, "", ts, role, str(i), title, text, "cli"])
    return (rows[0][0] if rows else ""), rows

# ------------------------------------------------------------------ sources

def _is_jsonl(path):
    return path.endswith(".jsonl")


def _is_gemini_chat(path):
    name = os.path.basename(path)
    return name.startswith("session-") and name.endswith(".json")


def _is_cursor_transcript(path):
    """`<id>/<id>.jsonl` is a session; `<id>/subagents/<other>.jsonl` is one of its subagents.

    Only the filename used to be tested, and these two are indistinguishable by name alone.
    """
    if not path.endswith(".jsonl"):
        return False
    return os.path.splitext(os.path.basename(path))[0] == os.path.basename(os.path.dirname(path))


def _is_opencode_db(path):
    return os.path.basename(path) == "opencode.db"


# One record per harness, keyed by the source tag stored in index.json. Everything that used to
# be a `source == "codex"` ternary reads this table instead, so adding a harness is one entry
# plus a parser rather than an edit in five places that can silently disagree.
#
#   roots    directories to walk for transcripts
#   match    path predicate; harnesses agree on neither the extension nor the layout
#   parse    (path, include_thinking, limit) -> (sid, rows) in the shared 9-field schema
#   label    the harness name, used for the source column, assistant turns and the preview.
#            One whitespace-free token: the piped row is columns, and a name with a space in
#            it shifts every field after it for anything parsing them.
#   colour   SGR code for the source column
#   resume   ("id", argv) substitutes {sid}; ("path", argv) substitutes {path}
#   subagents   harness writes separate subagent transcripts that fold into the parent
#   launch_dir  resume is scoped to the directory the session was started in
SOURCES = {
    "cc": {
        "roots": [PROJECTS_DIR], "match": _is_jsonl, "parse": None,
        "label": "claude", "colour": "34",
        "resume": ("id", ["claude", "--resume", "{sid}"]),
        "subagents": True, "launch_dir": True,
    },
    "codex": {
        "roots": [CODEX_DIR], "match": _is_jsonl, "parse": None,
        "label": "codex", "colour": "35",
        "resume": ("id", ["codex", "resume", "{sid}"]),
        "subagents": False, "launch_dir": False,
    },
    "gemini": {
        "roots": [GEMINI_DIR], "match": _is_gemini_chat, "parse": None,
        "label": "gemini-cli", "colour": "36",
        # --resume takes a project-scoped index number, which is not a stable handle for a
        # session found by search. --session-file takes the transcript path, which is.
        "resume": ("path", ["gemini", "--session-file", "{path}"]),
        "subagents": False, "launch_dir": False,
    },
    "cursor": {
        "roots": [CURSOR_DIR], "match": _is_cursor_transcript, "parse": None,
        "label": "cursor-cli", "colour": "32",
        "resume": ("id", ["cursor-agent", "--resume", "{sid}"]),
        "subagents": False, "launch_dir": False,
    },
    "opencode": {
        "roots": [OPENCODE_DIR], "match": _is_opencode_db, "parse": None,
        "label": "opencode", "colour": "33",
        # `opencode run` is the non-interactive form and demands a message; the bare command
        # opens the TUI on that session, which is what resuming means here.
        "resume": ("id", ["opencode", "--session", "{sid}"]),
        "subagents": False, "launch_dir": False,
    },
}

SOURCES["cc"]["parse"] = parse_session
SOURCES["codex"]["parse"] = parse_codex_session
SOURCES["gemini"]["parse"] = parse_gemini_session
SOURCES["cursor"]["parse"] = parse_cursor_session
SOURCES["opencode"]["parse"] = parse_opencode_session

DEFAULT_SOURCE = "cc"


def _source(name):
    """The record for a source tag, falling back to Claude for an index written by an older
    version that did not know this harness."""
    return SOURCES.get(name) or SOURCES[DEFAULT_SOURCE]

# ------------------------------------------------------------------ forks

# Claude Code forks a session by copying the transcript so far into a new file under a new
# session id. Nothing in the format announces that: no parent field, no marker entry. The only
# trace is that the copied messages keep the uuids they had in the original, so two Claude
# sessions whose FIRST message carries the same uuid are one conversation branched in two.
#
# Worth saying out loud, because until now the list showed them as two unrelated sessions with
# the same title, the same project and the same opening prompt, and picking the wrong one
# resumes a branch that is missing everything you did after the fork.

FORK_FAMILY_MAX = 12      # a bigger "family" than this is a fingerprint collision, not a fork
FORK_SCAN_LINES = 4000    # how far into a file to look for its first real message


def _root_uuid(path, scan=FORK_SCAN_LINES):
    """uuid of a Claude session's first user/assistant entry: its fork fingerprint.

    Cheap on purpose — this runs per session file, and the answer never changes once a file
    exists, so build_index carries it forward instead of recomputing it.
    """
    try:
        fh = open(path, "r", errors="replace")
    except OSError:
        return ""
    with fh:
        for i, line in enumerate(fh):
            if i >= scan:
                break
            try:
                o = json.loads(line)
            except json.JSONDecodeError:
                continue
            if o.get("type") in ("user", "assistant") and o.get("uuid"):
                return o["uuid"]
    return ""


def _msg_uuids(path):
    """Ordered (uuid, timestamp) for a Claude session's user/assistant entries."""
    out = []
    try:
        fh = open(path, "r", errors="replace")
    except OSError:
        return out
    with fh:
        for line in fh:
            try:
                o = json.loads(line)
            except json.JSONDecodeError:
                continue
            if o.get("type") in ("user", "assistant") and o.get("uuid"):
                out.append((o["uuid"], o.get("timestamp", "")))
    return out


def _shared_prefix(a, b):
    n = 0
    while n < len(a) and n < len(b) and a[n][0] == b[n][0]:
        n += 1
    return n


def _older(a, b):
    """True if branch `a` is the one branch `b` grew out of, rather than the other way round.

    Two branches agree up to the message where they split, and whichever carried on FIRST at
    that point is the one that existed to be copied. If one of them runs out at the split it
    *is* the copied prefix: the branch somebody forked from and then stopped using, which is
    why length can never be the signal on its own — an abandoned original is usually the
    shorter of the two.
    """
    k = _shared_prefix(a, b)
    if k >= len(a) or k >= len(b):
        return len(a) <= len(b)
    return a[k][1] <= b[k][1]


def detect_forks(index):
    """{forked sid: {"of": original sid, "at": messages shared}} across the whole index.

    Which branch is the original is decided at the message where two branches stop agreeing,
    by _older(); each fork is then attributed to the closest earlier branch it shares a prefix
    with, so a fork of a fork points at the fork and not at the root.

    Only the handful of sessions that share a fingerprint are read here; everything else costs
    a dict lookup.
    """
    family = {}
    for sid, info in index.items():
        if info.get("source") == "cc" and info.get("root"):
            family.setdefault(info["root"], []).append(sid)

    forks = {}
    for sids in family.values():
        if not 2 <= len(sids) <= FORK_FAMILY_MAX:
            continue
        seq = {s: _msg_uuids(index[s]["path"]) for s in sids}
        ranked = []                               # oldest branch first, by insertion
        for s in sorted(sids):
            i = 0
            while i < len(ranked) and _older(seq[ranked[i]], seq[s]):
                i += 1
            ranked.insert(i, s)
        for i, sid in enumerate(ranked[1:], 1):
            of = max(ranked[:i], key=lambda p: _shared_prefix(seq[p], seq[sid]))
            at = _shared_prefix(seq[of], seq[sid])
            if at:
                forks[sid] = {"of": of, "at": at}
    return forks


def load_forks():
    """The fork map written at index time, or {} if it was never built."""
    try:
        return json.load(open(FORKS_PATH))
    except (OSError, json.JSONDecodeError):
        return {}


# ------------------------------------------------------------------ cache

def _frag_path(jsonl_path):
    key = hashlib.sha1(jsonl_path.encode()).hexdigest()[:16]
    return os.path.join(FRAG_DIR, key + ".tsv")


# A cold build over ~1,200 sessions takes several seconds. Silence for that long
# reads as "hung" and is the cheapest bounce in the product, so we narrate it —
# but only when there is real work and someone watching. Warm runs re-parse a
# handful of files and must stay silent; piped/redirected stderr must stay clean
# so `agsearch -n` output can be consumed by scripts.
PROGRESS_MIN_FILES = 25     # below this a rebuild is fast enough to need no narration


class _IndexProgress:
    """Single overwriting stderr line: 'indexing 412/1238 sessions...'."""

    def __init__(self, total, stream=None):
        self.total = total
        self.done = 0
        self.stream = stream if stream is not None else sys.stderr
        self.active = total >= PROGRESS_MIN_FILES and _isatty(self.stream)
        self._width = 0

    def tick(self):
        self.done += 1
        if not self.active:
            return
        # Repaint on the first file, then ~every 1%, so a big corpus does not
        # spend its time writing escape codes.
        step = max(1, self.total // 100)
        if self.done != 1 and self.done % step and self.done != self.total:
            return
        msg = f"indexing {self.done}/{self.total} sessions..."
        self._width = max(self._width, len(msg))
        self.stream.write("\r" + msg.ljust(self._width))
        self.stream.flush()

    def done_(self):
        """Erase the line so it leaves no residue above the TUI or the results."""
        if not self.active or not self._width:
            return
        self.stream.write("\r" + " " * self._width + "\r")
        self.stream.flush()
        self._width = 0


# Any CSI sequence, not just the colours we emit: transcripts quote terminal output, so a
# message can carry escapes of its own, and a program reading this wants none of them.
_ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[a-zA-Z]|\x1b\[?")   # trailing alt: a code the snippet cut in half


def _use_color():
    """Colour is for a human at a terminal. A pipe (an agent, a script, a test) wants text.

    The TUI paths never ask: fzf reads the escapes itself, and its preview and pager are pipes.
    """
    return _isatty(sys.stdout) and not os.environ.get("NO_COLOR")


def _emit(text, color=None):
    """Print a rendered block, dropping ANSI unless colour is wanted (auto-detect by default)."""
    sys.stdout.write(text if (_use_color() if color is None else color)
                     else _ANSI_RE.sub("", text))


def _isatty(stream):
    try:
        return bool(stream.isatty())
    except (AttributeError, ValueError):
        return False


def build_index(include_thinking=False, force=False):
    """Return the full index as a list of TSV strings, refreshing per-file caches."""
    os.makedirs(FRAG_DIR, exist_ok=True)
    meta = {}
    try:                                # roots are immutable, so carry them over cache hits
        old_index = json.load(open(INDEX_PATH))
    except (OSError, json.JSONDecodeError):
        old_index = {}
    if os.path.exists(META_PATH) and not force:
        try:
            meta = json.load(open(META_PATH))
        except (OSError, json.JSONDecodeError):
            meta = {}
    if meta.get("_thinking") != include_thinking or meta.get("_fmt") != CACHE_FMT:
        force = True          # thinking toggle or format change invalidates fragments
        meta = {}

    # Harnesses disagree on where transcripts live and what they are named, so both the roots
    # and the filename test come from the source record rather than being hardcoded here.
    files = []                          # (path, source, parser)
    for source, rec in SOURCES.items():
        match, parser = rec["match"], rec["parse"]
        for root in rec["roots"]:
            if not os.path.isdir(root):
                continue
            for r, _dirs, fs in os.walk(root):
                for fn in fs:
                    full = os.path.join(r, fn)
                    if match(full):
                        files.append((full, source, parser))

    # Stat every file once up front: the same mtimes decide cache hits below and
    # tell us how many sessions actually need parsing, which is what we report.
    mtimes = {}
    stale = 0
    for path, _src, _parser in files:
        try:
            mtimes[path] = os.path.getmtime(path)
        except OSError:
            continue
        if force or meta.get(path) != mtimes[path] or not os.path.exists(_frag_path(path)):
            stale += 1
    progress = _IndexProgress(stale)

    new_meta = {"_thinking": include_thinking, "_fmt": CACHE_FMT}
    live_frags = set()
    lines = []
    sub_map = {}                        # parent sid -> [subagent file paths] (Claude only)
    index = {}                          # sid -> {source, path} for preview + resume routing
    for path, source, parser in files:
        if path not in mtimes:              # vanished between the stat pass and now
            continue
        mtime = mtimes[path]
        frag = _frag_path(path)
        live_frags.add(os.path.basename(frag))
        if not force and meta.get(path) == mtime and os.path.exists(frag):
            with open(frag, errors="replace") as fh:
                frag_lines = fh.read().splitlines()
        else:
            _sid, rows = parser(path, include_thinking)
            frag_lines = [SEP.join(r) for r in rows]
            with open(frag, "w") as fh:
                fh.write("\n".join(frag_lines))
            progress.tick()
        new_meta[path] = mtime
        lines.extend(frag_lines)
        if not frag_lines:
            continue
        base = os.path.basename(path)
        if _source(source)["subagents"] and base.startswith("agent-"):
            sid0 = frag_lines[0].split(SEP, 1)[0]
            sub_map.setdefault(sid0, []).append(path)     # subagent folds into parent
            continue
        # Most harnesses write one file per session, but some keep every session in a single
        # database. Register whatever sessions the fragment actually contains rather than
        # assuming the first row speaks for the file.
        for sid0 in dict.fromkeys(l.split(SEP, 1)[0] for l in frag_lines):
            index[sid0] = {"source": source, "path": path}
            if source == "cc":
                root = (old_index.get(sid0) or {}).get("root")
                index[sid0]["root"] = root or _root_uuid(path)

    for fn in os.listdir(FRAG_DIR):     # drop fragments for deleted sessions
        if fn not in live_frags:
            try:
                os.remove(os.path.join(FRAG_DIR, fn))
            except OSError:
                pass

    progress.done_()

    json.dump(new_meta, open(META_PATH, "w"))
    json.dump(sub_map, open(SUBMAP_PATH, "w"))
    json.dump(index, open(INDEX_PATH, "w"))
    json.dump(detect_forks(index), open(FORKS_PATH, "w"))
    return lines


# ------------------------------------------------------------------ filtering

def apply_scope(lines, here=False, project=None):
    if here:
        cwd = os.getcwd()
        lines = [l for l in lines
                 if l.split(SEP, 2)[1] == cwd or l.split(SEP, 2)[1].startswith(cwd + os.sep)]
    if project:
        p = project.lower()
        lines = [l for l in lines if p in l.split(SEP, 2)[1].lower()]
    return lines


# ------------------------------------------------------------------ rendering

def short_proj(cwd):
    return os.path.basename(cwd.rstrip("/")) or cwd


def _highlight(text, terms, code="\033[1;30;43m"):
    """Bold-highlight each query term (case-insensitive, first occurrence per term)."""
    for t in terms:
        if not t:
            continue
        low = text.lower()
        idx = low.find(t)
        if idx >= 0:
            text = text[:idx] + code + text[idx:idx + len(t)] + "\033[0m" + text[idx + len(t):]
    return text


ROW_TEXT_WIDTH = 160        # the "why it matched" line under a result, however big the message was


def _agent_name(source):
    """Name the agent side of a session after the tool it came from.

    The session list marks the source with the same name, so a row or preview line that calls
    every assistant turn `claude` contradicts the column two inches to its left. Spelled out
    rather than abbreviated: two-letter codes were guessable with two harnesses and are not
    with five.
    """
    return _source(source)["label"]


AGENT_ID_MIN = 12       # git's short-hash rule; see _short_id_len for why 8 is not enough


def _short_id_len(sids):
    """Shortest prefix that still tells every indexed session apart, floored at AGENT_ID_MIN.

    Git's rule, for the same reason: the full id is the one field a reader has to copy, and a
    36-char uuid is the most expensive thing on the line for the one consumer that pays per
    character. The floor is not cosmetic. Codex writes uuidv7, whose leading bytes are a
    timestamp, so sessions recorded near each other share 8-char prefixes: on a 741-session
    corpus 8 collided 36 times and 12 collided none.
    """
    uniq = set(sids)
    for n in (13, 18, 23, 36):          # uuid group boundaries: a cut mid-group reads as noise
        if n >= AGENT_ID_MIN and len({s[:n] for s in uniq}) == len(uniq):
            return n
    return 36


def _match_entry(f, matched, total, texts, keys, idlen=36, pad=True):
    """One result: the session's fields on one line, then the line that matched, indented.

    The session id leads because it is the only field another program needs — it is the handle
    `agsearch read <sid>` takes.

    A terminal gets padded columns because eyes track a ragged left edge badly. A pipe gets
    neither the padding nor the full id: alignment buys an agent nothing and every run of
    spaces costs it a token.
    """
    tag = "auto" if f[C_KIND] == "auto" else _agent_name(f[C_SOURCE])
    badge = f"{matched}/{total}" if total else ""
    proj = short_proj(f[C_CWD])[:20]
    if pad:
        head = (f"{f[C_SID]}  {f[C_DATE]}  {tag:<{SOURCE_COL}} {badge:<4} "
                f"\033[36m{proj:<20}\033[0m  {f[C_TITLE][:60]}")
    else:
        head = (f"{f[C_SID][:idlen]} {f[C_DATE]} {tag} {badge} "
                f"\033[36m{proj}\033[0m {f[C_TITLE][:60]}")
    idx = best_matching(texts, keys)[0] if (keys and texts) else ()
    # Matched on the title alone (or no query at all): show what the session opened with.
    body = texts[idx[0]] if idx else f[C_FIRST]
    return head + "\n    " + _snippet(body, keys, ROW_TEXT_WIDTH)


def print_matches(lines, query, limit=20):
    """Non-interactive results: the same ranked sessions the TUI lists, one entry each.

    This is the surface a pipe reads — an agent, a script, an install without fzf — so it runs
    the same rank_sessions the list does instead of a filter of its own. It used to be an
    AND-of-substrings sorted by date over *message* rows, which meant no BM25, no stemming, no
    typo tier, no demotion of automation, and one session repeated once per matching message.
    """
    rows = build_sessions(lines)
    if not rows:
        print("No indexed sessions found.", file=sys.stderr)
        return 1
    qterms = parse_query(query) if query.strip() else []
    if qterms:
        hits = [(m, f) for _score, m, f in rank_sessions(rows, qterms, _usage_counts())]
    else:                              # no query: newest first, yours ahead of automation
        rows.sort(key=lambda f: f[C_DATE], reverse=True)
        rows.sort(key=lambda f: f[C_KIND] == "auto")
        hits = [(0, f) for f in rows]
    if not hits:
        print("No matches.", file=sys.stderr)
        return 1

    keys = query_keys(qterms)
    texts = {}
    for l in lines:                    # sid -> its messages, for the "why it matched" line
        f = l.split(SEP)
        texts.setdefault(f[0], []).append(f[7])
    pad = _isatty(sys.stdout)
    idlen = 36 if pad else _short_id_len([f[C_SID] for f in rows])
    _emit("\n".join(_match_entry(f, m, len(qterms), texts.get(f[C_SID], []), keys, idlen, pad)
                    for m, f in hits[:limit]) + "\n")
    extra = len(hits) - limit
    if extra > 0:
        print(f"... and {extra} more (narrow the query, or run agsearch for the TUI).",
              file=sys.stderr)
    if not pad:
        # Whoever is reading this is a program, and the next thing it wants is one of these
        # sessions. Naming the command here means it does not have to be told anywhere else.
        print("open one: agsearch read <id>  (the id is the first field above)")
    return 0


def resolve_sid(sid):
    """Full session id for `sid`, which may be any unambiguous prefix of one.

    `-n` prints shortened ids, so `read` has to accept what `-n` printed. Git again: a prefix
    is a handle until it stops being unique, and then the ambiguity is worth saying out loud
    rather than guessing at.

    Returns (sid, None) or (None, message).
    """
    sid = (sid or "").strip()
    try:
        index = json.load(open(INDEX_PATH))
    except (OSError, json.JSONDecodeError):
        return sid, None                    # no index to check against; let the caller try
    if sid in index:
        return sid, None
    hits = sorted(k for k in index if k.startswith(sid))
    if len(hits) == 1:
        return hits[0], None
    if not hits:
        return None, f"no session id starts with {sid!r}. Run a search first: agsearch -n \"...\""
    listed = ", ".join(hits[:5]) + (" ..." if len(hits) > 5 else "")
    return None, f"{sid!r} matches {len(hits)} sessions: {listed}. Use more characters."


def _session_path(sid):
    if os.path.isdir(PROJECTS_DIR):
        for root, _dirs, files in os.walk(PROJECTS_DIR):
            if sid + ".jsonl" in files:
                return os.path.join(root, sid + ".jsonl")
    return None


def _turn_header(role, source, is_sub):
    """Chat-style role gutter for a preview turn: '▌ you', '▌ claude', '▌ tool'.

    The agent side is named after the source so the preview mirrors the tool the session came
    from, the way you saw it in Claude Code or Codex.
    """
    agent = _source(source)["label"]
    name = {"user": "you", "assistant": agent, "thinking": "thinking",
            "tool": "tool"}.get(role, role or "?")
    if is_sub:
        return f"\033[35m▌ ⤷ {name}\033[0m"
    color = {"user": "36", "assistant": "32", "thinking": "90", "tool": "90"}.get(role, "37")
    return f"\033[{color}m▌ {name}\033[0m"


def _turn(row, is_sub, source, keys, width=200):
    """One conversation turn: role gutter line, then the message."""
    return [_turn_header(row[4], source, is_sub), _snippet(row[7], keys, width)]


# --- snippet cleanup: transcripts are full of payloads nobody wants to read in a result row.
_IMAGE_RE = re.compile(r"\[Image:[^\]]*\]")
_FENCE_RE = re.compile(r"```[a-zA-Z0-9_+.-]*.*?```", re.S)      # closed fence
_OPEN_FENCE_RE = re.compile(r"```[a-zA-Z0-9_+.-]*.*", re.S)     # fence cut off by the 400-char cap
_MD_LINK_RE = re.compile(r"!?\[([^\]\n]{1,80})\]\([^)\s]*\)")   # [text](url) -> text
_MD_CODESPAN_RE = re.compile(r"`+([^`]+)`+")
# Emphasis markers only count when they stand alone — otherwise a C-style /** comment */ or a
# glob pattern gets shredded into gibberish.
_MD_BOLD_RE = re.compile(r"(?<![\w/*])(\*\*|__)(?!\s)(.+?)(?<!\s)\1(?![\w/*])", re.S)
_MD_ITALIC_RE = re.compile(r"(?<![\w/*])\*(?!\s)([^*/\n]{1,160}?)(?<!\s)\*(?![\w/*])")
_MD_HEADING_RE = re.compile(r"(?:^|(?<= ))#{1,6}\s+")
# Newlines are already collapsed by index time, so a mid-string "- " is ambiguous with a dash in
# prose. Only strip one when it introduces a bold/heading run — the shape list items actually have.
_MD_BULLET_RE = re.compile(r"(?:^|(?<= ))[-*+•]\s+(?=\*\*|#{1,6}\s)")
_MD_LEAD_LIST_RE = re.compile(r"^\s*(?:[-*+•]|\d+\.|>)\s+")
_LONG_TOKEN_RE = re.compile(r"\S{45,}")


def _condense_json(s, minlen=40):
    """Replace embedded JSON objects/arrays with a [json] placeholder. Hand-scanned rather than
    regexed because these dumps nest, and they're routinely truncated mid-structure."""
    out = []
    i, n = 0, len(s)
    while i < n:
        c = s[i]
        nxt = s[i + 1:i + 3].lstrip()[:1]
        if c in "[{" and nxt in ('"', "{", "["):
            j, depth, in_str, esc = i, 0, False, False
            while j < n:
                ch = s[j]
                if in_str:
                    if esc:
                        esc = False
                    elif ch == "\\":
                        esc = True
                    elif ch == '"':
                        in_str = False
                elif ch == '"':
                    in_str = True
                elif ch in "[{":
                    depth += 1
                elif ch in "]}":
                    depth -= 1
                    if depth == 0:
                        j += 1
                        break
                j += 1
            if j - i >= minlen:                  # short inline objects read fine, leave them
                out.append("[json]")
                i = j
                continue
        out.append(c)
        i += 1
    return "".join(out)


def _condense_line_numbers(s, minrun=4):
    """`cat -n`-style file reads — "1 import x 2 import y 3 …" — are pure noise once newlines
    are gone. Collapse a run of ascending bare line numbers into a [file] placeholder. Bare
    digits only, so a "1. do this 2. do that" prose list is left alone."""
    toks = s.split(" ")
    out, i, n = [], 0, len(toks)
    while i < n:
        if toks[i].isdigit():
            last, chain = int(toks[i]), [i]
            for k in range(i + 1, n):
                if toks[k].isdigit():
                    v = int(toks[k])
                    if 0 < v - last <= 3:      # small gaps = blank lines in the file
                        last, _ = v, chain.append(k)
                    else:
                        break
            if len(chain) >= minrun:
                out.append("[file]")
                i = chain[-1] + 1
                continue
        out.append(toks[i])
        i += 1
    return " ".join(out)


def clean_snippet(text):
    """Turn one raw transcript message into readable prose for a result row.

    Strips markdown noise (fences, headings, bullets, emphasis, link syntax) and condenses
    payloads — JSON dumps, pasted images, long opaque tokens — into short placeholders, so a
    snippet shows what was said rather than what was dumped. Search still runs over the raw
    text; this is display only.
    """
    if not text:
        return ""
    t = _FENCE_RE.sub(" [code] ", text)
    t = _OPEN_FENCE_RE.sub(" [code] ", t)
    t = _IMAGE_RE.sub(" [image] ", t)
    t = _condense_json(t)
    t = _condense_line_numbers(t)
    t = _MD_LINK_RE.sub(r"\1", t)
    t = _MD_CODESPAN_RE.sub(r"\1", t)
    t = _MD_LEAD_LIST_RE.sub("", t)
    t = _MD_BULLET_RE.sub("", t)          # before emphasis: the bullet is spotted by the ** it leads
    t = _MD_BOLD_RE.sub(r"\2", t)
    t = _MD_ITALIC_RE.sub(r"\1", t)
    t = _MD_HEADING_RE.sub("", t)
    t = _LONG_TOKEN_RE.sub(lambda mo: mo.group(0)[:20] + "…", t)
    return " ".join(t.split())


def _snippet(text, terms, width=200):
    """A readable window of text centered on the first matching term, highlighted.

    Cleaned first, so the window shows prose instead of raw payload. If the match only exists
    in the part that got condensed, fall back to the raw text — a snippet without the term you
    searched for is worse than a noisy one.
    """
    live = [t for t in terms if t]
    clean = clean_snippet(text)
    if live and not any(t in clean.lower() for t in live):
        clean = _single_line(text, len(text))
    lo = clean.lower()
    pos = min([lo.find(t) for t in live if t in lo] or [0])
    start = max(0, pos - 40)
    frag = clean[start:start + width]
    if start:
        frag = "…" + frag
    if start + width < len(clean):
        frag += "…"
    return _highlight(frag, live)


def best_matching(texts, keys):
    """Indices of the messages containing the most query words, and how many that was.

    An exact AND when some message has them all, otherwise the closest ones. A blank card on
    the result you were just told is the best match reads as a broken search, and the session
    may well have been ranked there on its title or first prompt.
    """
    counts = [sum(1 for k in keys if _at_word_start(t.lower(), k)) for t in texts]
    best = max(counts, default=0)
    return [i for i, n in enumerate(counts) if n and n == best], best


def _arc_segments(tagged, source):
    """How the chat opened and where it ended, for when there is nothing matched to show."""
    segs = []
    idxs = []
    first_user = next((i for i, (r, _s) in enumerate(tagged) if r[4] == "user"), None)
    if first_user is not None:
        idxs.append(first_user)
        reply = next((i for i in range(first_user + 1, len(tagged))
                      if tagged[i][0][4] == "assistant"), None)
        if reply is not None:
            idxs.append(reply)
    for i in idxs:
        segs.append(_turn(*tagged[i][:1], tagged[i][1], source, []))
    last = len(tagged) - 1
    if last >= 0 and last not in idxs:
        if idxs and last > idxs[-1] + 1:
            segs.append(["\033[2m⋯\033[0m"])
        segs.append(_turn(tagged[last][0], tagged[last][1], source, []))
    return segs


# Six turns with a blank line between each ran 1.76x the old card's height, so a
# preview that used to fit began to scroll. Four turns and no blank lines lands at
# 1.05x: the gutter (`> you`) already separates turns, so the blank line was paying
# ~30% of the height for separation it was not providing.
PREVIEW_TURNS = 4


def _preview_lines(tagged, keys, source):
    """The preview body, rendered as a chat excerpt. Returns lines; kept pure so it is testable.

    Matching rules are the ranker's, not a second set: word-start keys, exact AND when some
    turn has every word, otherwise the closest turns. The chat shape is what changed — turns
    with a role gutter instead of isolated snippet lines, anchored by the opening prompt so
    what the chat was about stays visible even when the match is buried deep in it.
    """
    segs = []
    if keys:
        picked, best = best_matching([r[7] for r, _sub in tagged], keys)
        matches = [tagged[i] for i in picked]
        if best == len(keys):
            head = f"● {len(matches)} match(es)"
        elif best:
            head = f"● {len(matches)} partial · best turn has {best} of {len(keys)} words"
        else:
            head = "● matched on title or first prompt, not in any turn"
        segs.append([f"\033[1;33m{head}\033[0m"])

        shown = matches[:PREVIEW_TURNS]
        if shown:
            shown_rows = [r for r, _sub in shown]
            opening = next(((r, sub) for r, sub in tagged if r[4] == "user"), None)
            if opening and opening[0] not in shown_rows:
                segs.append(_turn(opening[0], opening[1], source, []))
                segs.append(["\033[2m⋯\033[0m"])
            for r, sub in shown:
                segs.append(_turn(r, sub, source, keys))
            if len(matches) > PREVIEW_TURNS:
                segs.append([f"\033[2m… {len(matches) - PREVIEW_TURNS} more (resume to read)\033[0m"])
        else:
            segs += _arc_segments(tagged, source)
    else:
        segs += _arc_segments(tagged, source)

    out = []
    for seg in segs:
        out += seg
    return out


def _flag(v):
    """A flag that arrived either as a Python bool or as an argv string from fzf."""
    return str(v).lower() in ("1", "true")


def fork_mark(sid):
    """The flag a fork puts ahead of its title, in a preview or a read.

    Same word in the same place as the list row the session was selected from. Carrying the
    fact only on the meta line put it in a different position in each view, and last on a line
    that leads with the project and the date, which is the wrong end for a flag.
    """
    return _FORK_MARK if sid in load_forks() else ""


def fork_line(sid):
    """The clause under that flag: which branch it came from, and where the two split.

    The original's id is spelled short because this sits inside a line that must not wrap; the
    prefix is enough to find it in the list, which is the only thing you want it for.
    """
    fork = load_forks().get(sid)
    if not fork:
        return ""
    return f" · fork of {fork['of'][:8]} at msg {fork['at']}"


def load_session_rows(sid, thinking=False, limit=MSG_INDEX_CHARS):
    """(source, [(row, is_subagent)]) for one session, chronological.

    Shared by the preview card and the transcript reader so both see exactly the
    same conversation, subagent turns folded in and all. `limit` is what separates
    them: the card reuses the index cap, the reader asks for the full text.
    """
    try:
        index = json.load(open(INDEX_PATH))
    except (OSError, json.JSONDecodeError):
        index = {}
    info = index.get(sid, {})
    source = info.get("source", "cc")
    path = info.get("path") or _session_path(sid)

    # Only some harnesses write separate subagent transcripts; the rest are a single file.
    rec = _source(source)
    tagged = []
    if path:
        _s, prows = rec["parse"](path, include_thinking=thinking, limit=limit)
        # A shared database hands back every session in the file; keep the one asked for.
        tagged += [(r, False) for r in prows if r[0] == sid]
    if rec["subagents"]:
        try:
            submap = json.load(open(SUBMAP_PATH))
        except (OSError, json.JSONDecodeError):
            submap = {}
        for sp in submap.get(sid, [])[:40]:
            _s, srows = parse_session(sp, include_thinking=thinking, limit=limit)
            tagged += [(r, True) for r in srows]
    tagged.sort(key=lambda x: x[0][3])          # chronological by timestamp
    return source, tagged


def resume_line(sid, cwd):
    """The paste-ready reattach line for one session id.

    Built on resume_plan so this lands in the directory the session was actually
    filed under. Reattaching from the cwd recorded on a message fails whenever
    that cwd is a subdirectory of the launch dir, which is common.
    """
    _source, _bin, argv, cwd, target, _exists = resume_plan(sid, cwd)
    return resume_command(target or cwd, argv)


AGENT_READ_CHARS = 12_000       # what a piped `read` spends before it starts eliding
OPENING_TURNS = 2               # kept whatever the budget: the opening states the goal


def _elide(blocks, budget, matched=()):
    """Fit a transcript into `budget` characters, keeping the turns worth keeping.

    A session is read to answer one of two questions: what were we doing, and where did we
    stop. Those live at the two ends, so the middle is what a too-long transcript can afford
    to lose. A query changes the question to a third one, what was said about this, and then
    the turns that matched outrank the ending: asking for matches and getting the last twenty
    messages instead is the wrong answer to the question that was asked.

    Returns (blocks, dropped).
    """
    if sum(len(b) for b in blocks) <= budget:
        return blocks, 0
    n = len(blocks)
    keep = set(range(min(OPENING_TURNS, n)))            # the opening states the goal
    spent = sum(len(blocks[i]) for i in keep)
    for i in list(matched) + list(range(n - 1, -1, -1)):  # matches, then the ending
        if i not in keep and spent + len(blocks[i]) <= budget:
            keep.add(i)
            spent += len(blocks[i])
    return [blocks[i] for i in sorted(keep)], n - len(keep)


def render_transcript(sid, thinking="0", query="", color=None, budget=None):
    """The whole conversation, readable, without resuming it.

    Resuming to read costs a CLI start, a context load, and a session you then
    have to leave. Usually you only wanted to check this is the right session or
    lift one answer out of it. Matched turns are marked with a bar so the pager
    can jump between them.

    `budget` caps the printed characters for a reader that pays for them. A terminal has a
    scrollback and a pager and so gets the whole thing; a pipe gets the opening and the
    ending, and a line saying what it did not get and how to ask for it.
    """
    # Full text, not the index cap: this view exists to be read.
    source, tagged = load_session_rows(sid, _flag(thinking), limit=100_000)
    if not tagged:
        print("(session not found)")      # stdout: the TUI reads this through a pager
        return 1
    rows = [r for r, _ in tagged]
    keys = query_keys(parse_query(query)) if query.strip() else []

    r0 = rows[0]
    title = r0[6] or next((r[7] for r, _ in tagged if r[4] == "user"), "") or "(untitled)"
    n_sub = sum(1 for _, sub in tagged if sub)
    out = [fork_mark(sid) + f"\033[1m{title[:80]}\033[0m",
           f"\033[2m{short_proj(r0[1])} · {r0[3][:10]} · {len(rows)} msgs"
           + (f" · {n_sub} subagent" if n_sub else "") + fork_line(sid) + "\033[0m",
           f"\033[2m{resume_line(sid, r0[1])}\033[0m\n"]

    blocks, matched = [], []
    for r, sub in tagged:
        hit = bool(keys) and all(k in r[7].lower() for k in keys)
        mark = "\033[1;33m▶\033[0m " if hit else "  "
        if hit:
            matched.append(len(blocks))
        blocks.append(mark + _turn_header(r[4], source, sub) + "\n"
                      + f"\033[2m{r[3][11:16]}\033[0m  " + _snippet(r[7], keys, 100_000) + "\n")

    total, dropped = len(blocks), 0
    if budget:
        blocks, dropped = _elide(blocks, budget, matched)
    if dropped:
        kept = ("the opening, every turn matching your query that fits, and the ending"
                if matched else "the opening and the ending")
        note = (f"[{dropped} of {total} messages elided; kept {kept}. "
                f"Whole transcript: agsearch read {sid[:13]} --full]")
        blocks.insert(min(OPENING_TURNS, len(blocks)), "\033[2m" + note + "\033[0m\n")
    _emit("\n".join(out + blocks) + "\n", color)
    return 0


def render_preview(sid, thinking, query):
    """Compact preview card for one session (no full transcript — that's what resume is for).

    With a query: the matched lines only, highlighted. Without one: the bookends (first prompt
    + last message) so you know what it was about. Merges subagent transcripts (tagged ⤷) so
    their content is previewable too. Re-runs per keystroke via fzf's {q}.
    """
    thinking = str(thinking) == "1"
    source, tagged = load_session_rows(sid, thinking)
    if not tagged:
        print("(session not found)")
        return
    tagged.sort(key=lambda x: x[0][3])   # chronological by timestamp
    rows = [r for r, _ in tagged]

    keys = query_keys(parse_query(query)) if query.strip() else []
    r0 = rows[0]
    n_sub = sum(1 for _, sub in tagged if sub)
    disp_title = r0[6] or next((r[7] for r, _ in tagged if r[4] == "user"), "") or "(untitled)"
    print(fork_mark(sid) + f"\033[1m{disp_title[:80]}\033[0m")
    gone = " · orig dir gone" if r0[1] and not os.path.isdir(r0[1]) else ""
    print(f"\033[2m{short_proj(r0[1])} · {r0[3][:10]} · {len(rows)} msgs"
          + (f" · {n_sub} subagent" if n_sub else "") + gone + fork_line(sid) + "\033[0m")

    body = _preview_lines(tagged, keys, source)
    if body:
        print()
        print("\n".join(body))


# ------------------------------------------------------------------ resume

def _die(msg):
    """Show an error and hold the window open — otherwise the popup just vanishes."""
    print(f"\n\033[31m{msg}\033[0m\n", file=sys.stderr)
    try:
        input("press enter to close ")
    except (EOFError, KeyboardInterrupt):
        pass
    sys.exit(1)


def _launch_dir(session_path, cwd):
    """The directory Claude actually filed this session under, or "" if undetermined.

    Claude scopes `--resume <id>` by the directory it was launched from, storing the transcript
    in ~/.claude/projects/<slug>/ where slug = the launch dir with non-alphanumerics replaced by
    `-`. The `cwd` recorded on messages can be a SUBDIRECTORY of that launch dir, and resuming
    from the subdirectory makes Claude look in the wrong project ("No conversation found").
    Slugs can't be decoded back to a path unambiguously (real dashes are indistinguishable from
    separators), so walk cwd's ancestors and take the one whose slug matches.
    """
    if not session_path or not cwd:
        return ""
    slug = os.path.basename(os.path.dirname(session_path))
    d = os.path.abspath(os.path.expanduser(cwd))
    while d and d != os.sep:
        if re.sub(r"[^A-Za-z0-9]", "-", d) == slug:
            return d
        parent = os.path.dirname(d)
        if parent == d:
            break
        d = parent
    return ""


def _nearest_existing_dir(path):
    """Closest existing ancestor of `path` (the path itself if it exists), else $HOME.

    Worktrees get deleted, but `claude --resume <id>` / `codex resume <id>` are id-based and
    don't need the original directory — so a dead cwd is a reason to relocate, not to abort.
    """
    p = os.path.abspath(os.path.expanduser(path)) if path else ""
    while p and p != os.sep:
        if os.path.isdir(p):
            return p
        parent = os.path.dirname(p)
        if parent == p:
            break
        p = parent
    if os.path.isdir(os.sep) and not os.path.isdir(HOME):
        return os.sep
    return HOME


def _confirm_active(sid, bin_):
    """Warn that a session still looks live, and ask before attaching. Never hard-blocks:
    with no tty to prompt on (fzf popup, piped run) it warns and proceeds."""
    warn = (f"\033[33m⚠ this session looks active (written to in the last "
            f"{ACTIVE_WINDOW_SEC}s) — resuming may collide with the running {bin_}.\033[0m")
    print(warn, file=sys.stderr)
    if not sys.stdin.isatty():
        return True
    try:
        ok = input("resume anyway? [Y/n] ").strip().lower() in ("", "y", "yes")
    except (EOFError, KeyboardInterrupt):
        return True
    if not ok:
        print("aborted.", file=sys.stderr)
    return ok


def resume_plan(sid, cwd):
    """Where to resume from and what to run: (source, bin, argv, cwd, target, cwd_exists).

    Shared by the launcher and by `--no-resume`, so the command printed for you to run by hand
    can never drift from the one agsearch would have run itself.
    """
    try:
        with open(INDEX_PATH) as fh:
            info = json.load(fh).get(sid, {})
    except (OSError, json.JSONDecodeError):
        info = {}
    source = info.get("source", DEFAULT_SOURCE)
    rec = _source(source)
    kind, template = rec["resume"]
    handle = info.get("path", "") if kind == "path" else sid
    argv = [a.replace("{sid}", sid).replace("{path}", handle) for a in template]
    bin_ = argv[0]

    # Claude looks for the session in the project of whatever directory it starts in, so resume
    # from the dir it was launched in, not the `cwd` on the messages, which may be a subdir.
    if rec["launch_dir"]:
        cwd = _launch_dir(info.get("path", ""), cwd) or cwd

    # The recorded worktree may be long gone. Resume is id-based, so relocate to the nearest
    # surviving ancestor (or $HOME) instead of refusing to launch.
    cwd_exists = bool(cwd) and os.path.isdir(cwd)
    target = cwd if cwd_exists else (_nearest_existing_dir(cwd) if cwd else "")
    return source, bin_, argv, cwd, target, cwd_exists


def resume_command(target, argv):
    """A shell line you can paste. Worktree paths contain spaces often enough to quote."""
    cmd = " ".join(argv)
    return f"cd {shlex.quote(target)} && {cmd}" if target else cmd


# Clipboard tools in preference order: macOS, then Wayland, then the two common X11 ones.
CLIPBOARD_CMDS = (["pbcopy"], ["wl-copy"], ["xclip", "-selection", "clipboard"],
                  ["xsel", "--clipboard", "--input"])


def copy_to_clipboard(text, which=None, run=None):
    """Put `text` on the clipboard with whatever tool exists. Returns the tool used, or "".

    Best effort on purpose: the copy is a convenience (⌘F/^F straight to your query inside the
    resumed session), so a machine with no clipboard tool should resume normally rather than
    fail. Injectable which/run so the fallback order is testable without installing anything.
    """
    which = which or shutil.which
    run = run or subprocess.run
    for cmd in CLIPBOARD_CMDS:
        if not which(cmd[0]):
            continue
        try:
            run(cmd, input=text, text=True, check=False)
            return cmd[0]
        except OSError:
            continue
    return ""


def resume(sid, cwd, query="", active=False):
    if query:
        copy_to_clipboard(query)                    # so ⌘F finds it inside the resumed session
    source, bin_, argv, cwd, target, cwd_exists = resume_plan(sid, cwd)

    # Leave a trace: if the launched CLI dies instantly the popup window vanishes with it,
    # so this log is the only way to see what was attempted.
    note = resume_command(target or cwd, argv)
    try:
        with open(os.path.join(CACHE_DIR, "last-resume.log"), "a") as fh:
            fh.write(f"{note}   [source={source} cwd_exists={cwd_exists}"
                     f" orig_cwd={cwd or '-'} fallback={'-' if cwd_exists else (target or '-')}"
                     f" active={bool(active)}]\n")
    except OSError:
        pass

    if not shutil.which(bin_):
        _die(f"{bin_} CLI not found on PATH.\nRun manually:\n  {note}")
    if cwd and not cwd_exists:
        print(f"\033[33moriginal dir gone ({cwd}), resuming from {target}\033[0m", file=sys.stderr)
    if active and not _confirm_active(sid, bin_):
        return
    if target:
        try:
            os.chdir(target)
        except OSError:                              # raced away between check and chdir
            os.chdir(HOME)
    os.execvp(bin_, argv)                            # replace this process


# ------------------------------------------------------------------ fzf TUI

def group_sessions(lines):
    """Collapse per-message index rows into one entry per session, newest first."""
    by = {}
    for l in lines:
        f = (l.split(SEP) + [""] * 9)[:9]
        sid, cwd, _branch, ts, _role, _seq, title, text, kind = f
        g = by.get(sid)
        if g is None:
            g = by[sid] = {"sid": sid, "cwd": cwd, "date": ts, "title": title,
                           "first_user": "", "kind": kind or "cli", "texts": []}
        if ts > g["date"]:
            g["date"] = ts
        if cwd:
            g["cwd"] = cwd
        if kind == "cli":
            g["kind"] = "cli"                # a real CLI turn outranks folded-in agent rows
        if title and not g["title"]:
            g["title"] = title
        if _role == "user" and not g["first_user"]:
            g["first_user"] = text
        g["texts"].append(text)
    sessions = list(by.values())
    for s in sessions:                       # fall back to the first prompt when untitled
        if not s["title"]:
            s["title"] = s["first_user"] or "(untitled)"
    sessions.sort(key=lambda s: s["date"], reverse=True)
    return sessions


def build_sessions(lines):
    """Write one line per session to SESSIONS_PATH: sid, cwd, date, source, kind, title, first
    prompt, blob. The first prompt is stored separately from the blob so ranking can weight it
    as its own field — it's the strongest single signal of what a session is about."""
    try:
        index = json.load(open(INDEX_PATH))
    except (OSError, json.JSONDecodeError):
        index = {}
    out = []
    for s in group_sessions(lines):
        source = index.get(s["sid"], {}).get("source", "cc")
        title = _single_line(s["title"] or "(untitled)", 90)
        first = _single_line(s["first_user"], 400)
        # High cap so full sessions (incl. folded-in subagent content) stay searchable.
        # Stored lowercased: ranking is the only reader, and it would otherwise re-lower the
        # whole corpus on every keystroke. Nothing displays this column.
        blob = _single_line(" · ".join(s["texts"]), 2_000_000).lower()
        out.append([s["sid"], s["cwd"], s["date"][:10], source, s["kind"], title, first, blob])
    with open(SESSIONS_PATH, "w") as fh:                 # what _filter reads, one line each
        fh.write("\n".join(SEP.join(r) for r in out))
    return out


# Common words that add noise, not signal, to a search ("migration OF the database").
_STOP = {"of", "the", "a", "an", "to", "for", "in", "on", "and", "or", "is", "it", "this",
         "that", "with", "from", "by", "at", "as", "be", "are", "was", "were", "how", "do",
         "i", "my", "me", "we", "you", "about", "into", "using", "use", "some", "any"}

def _stem(w):
    """Conservative inflectional stem: strip one reliable suffix only if a solid (>=5 char)
    root remains, so migration/migrate/migrating -> 'migrat' while running stays 'running'
    (better to under-stem than to produce junk roots like 'oper' or 'runn')."""
    for suf in ("ing", "ion", "ed", "es", "e", "s"):
        if w.endswith(suf) and len(w) - len(suf) >= 5:
            return w[:-len(suf)]
    return w


def _fuzzy_span(hay, term):
    """Greedy first subsequence match; returns its character span, or None (last-resort typo tier)."""
    i = start = 0
    for ci, c in enumerate(hay):
        if c == term[i]:
            if i == 0:
                start = ci
            i += 1
            if i == len(term):
                return ci - start + 1
    return None


# Derived from SOURCES so the column and the assistant-turn label can never disagree about
# what a harness is called. They used to be written out separately, and had already drifted.
# Width of the source column, from the longest harness name, so adding one cannot misalign
# every row below it.
SOURCE_COL = max([len("auto")] + [len(r["label"]) for r in SOURCES.values()])
_SRC_MARK = {name: "\033[%sm%-*s\033[0m" % (rec["colour"], SOURCE_COL, rec["label"])
             for name, rec in SOURCES.items()}
_AUTO_MARK = "\033[90m%-*s\033[0m" % (SOURCE_COL, "auto")   # plugin/SDK-spawned, not your typing
_LIVE_MARK = "\033[1;31m●\033[0m "           # session still being written to → probably running
# Informational only: the session still resumes (from the nearest surviving ancestor dir),
# so this is muted enough to read as a footnote rather than a warning.
_GONE_MARK = "  \033[2morig dir gone\033[0m"
# Leads the title rather than trailing it. The list pane is a fraction of the terminal, so
# anything parked after the title is the first thing truncated away — exactly on the rows that
# need it, since a fork carries the same long title as the session it was forked from. Leading
# it costs nothing on the other 99% of rows, and every title starts in the same column, so the
# marks still line up to be scanned. A word, not a glyph: ⑂ and ⋔ are unreadable at 14px.
_FORK_MARK = "\033[2mfork\033[0m "


def _active_sids(sids):
    """Subset of `sids` whose transcript — or one of its subagent transcripts — was appended to
    within ACTIVE_WINDOW_SEC, i.e. the session still looks live."""
    try:
        index = json.load(open(INDEX_PATH))
    except (OSError, json.JSONDecodeError):
        index = {}
    try:
        submap = json.load(open(SUBMAP_PATH))
    except (OSError, json.JSONDecodeError):
        submap = {}
    now = time.time()
    live = set()
    for sid in sids:
        for p in [index.get(sid, {}).get("path")] + submap.get(sid, [])[:40]:
            try:
                if p and now - os.path.getmtime(p) <= ACTIVE_WINDOW_SEC:
                    live.add(sid)
                    break
            except OSError:
                continue
    return live


def _missing_dirs(cwds):
    """Which of these recorded working directories no longer exist. Deduped before stat'ing,
    since a project's sessions all share one dir."""
    gone = set()
    for cwd in set(cwds):
        if cwd and not os.path.isdir(cwd):
            gone.add(cwd)
    return gone


def _row(sid, cwd, date, source, kind, title, badge, active=False, dir_gone=False,
         forked=False):
    mark = _AUTO_MARK if kind == "auto" else _SRC_MARK.get(source, " " * SOURCE_COL)
    live = _LIVE_MARK if active else "  "
    forkm = _FORK_MARK if forked else ""
    tail = _GONE_MARK if dir_gone else ""
    body = (f"{date}  {mark}  \033[36m{short_proj(cwd)[:15]:<15}\033[0m  "
            f"{badge} {live}{forkm}{title[:64]}{tail}")
    if kind == "auto":
        body = (f"\033[2m{date}  \033[0m{_AUTO_MARK}\033[2m  {short_proj(cwd)[:15]:<15}  "
                f"{badge} \033[0m{live}{forkm}\033[2m{title[:64]}\033[0m{tail}")
    return SEP.join([sid, cwd, body, "1" if active else "0"])


def _bm25_tf(tf, dl, avgdl):
    """BM25 saturated term frequency with length normalization: a term mentioned twice counts
    for much less than twice once, and a hit in a huge transcript counts for less than the same
    hit in a short, on-point one."""
    if not tf:
        return 0.0
    return tf * (BM25_K1 + 1) / (tf + BM25_K1 * (1 - BM25_B + BM25_B * dl / (avgdl or 1)))


def _age_days(date_str, now=None):
    try:
        t = time.mktime(time.strptime(date_str[:10], "%Y-%m-%d"))
    except (ValueError, OverflowError, TypeError):
        return 3650.0                                # undated → treat as ancient, never boosted
    return max(0.0, ((now if now is not None else time.time()) - t) / 86400.0)


_RESUME_SID_RE = re.compile(r"(?:--resume|resume)\s+([0-9a-fA-F][0-9a-fA-F-]{7,})")


def _usage_counts(path=None):
    """How often each session was resumed, read back off the last-resume.log breadcrumb.
    That log is the only record of which sessions you actually return to — the sessions you
    keep reopening are the ones you most likely mean next time."""
    counts = {}
    try:
        with open(path or os.path.join(CACHE_DIR, "last-resume.log"), errors="replace") as fh:
            recent = fh.readlines()[-2000:]
    except OSError:
        return counts
    for line in recent:
        m = _RESUME_SID_RE.search(line)
        if m:
            counts[m.group(1)] = counts.get(m.group(1), 0) + 1
    return counts


def _boost(f, usage, now):
    """Multiplier for how likely this session is the one you want, independent of the query:
    recent sessions and ones you resume often. Bounded, so it re-orders near-ties without ever
    floating an irrelevant session above a real match."""
    recency = 0.5 ** (_age_days(f[C_DATE], now) / RECENCY_HALFLIFE_DAYS)
    used = usage.get(f[C_SID], 0)
    return 1.0 + RECENCY_W * recency + USAGE_W * min(1.0, math.log1p(used) / math.log(6))


def _is_word_char(c):
    return c.isalnum() or c == "_"


def _at_word_start(hay, key, whole=False):
    """Does `key` occur at the start of a word in `hay`? With `whole`, as a complete word.

    str.find is C-fast and this returns on the first real hit, so it stays cheap on a huge
    transcript. Prefix matching exists to undo stemming (migrat -> migration/migrate), so it
    is applied only to keys that were actually stemmed. An unstemmed key is the whole word you
    typed: `pr` should not match `print`, `previously` or `prisma`.
    """
    n = len(key)
    i = hay.find(key)
    while i != -1:
        if i == 0 or not _is_word_char(hay[i - 1]):
            end = i + n
            if not whole or end >= len(hay) or not _is_word_char(hay[end]):
                return True
        i = hay.find(key, i + 1)
    return False


def _key_probe(key, whole=False):
    """Term frequency of `key`, but zero unless it appears at a word start somewhere.

    Plain `str.count` counts substrings, and that is what poisons ranking: `pr` is a substring
    of 700/718 sessions but a word in 182, so its idf collapses to nothing and the coverage
    tiebreaker stops discriminating. Presence is the part that has to be exact; the count
    itself can stay a substring count because BM25 saturates tf anyway.
    """
    def probe(hay):
        if not _at_word_start(hay, key, whole):
            return 0
        return hay.count(key)
    return probe


def query_keys(qterms):
    """Ranking/preview keys: stem when it is long enough, else the raw term."""
    return [stem if len(stem) >= 3 else term for term, stem in qterms]


def rank_sessions(rows, qterms, usage=None, now=None):
    """Rank sessions for a query. Returns [(score, matched, row)] best-first.

    BM25 over three weighted fields — title+project, first prompt, full conversation — with
    length normalization, so a sprawling session no longer outranks a short exact match, and
    what a session was *opened to do* outweighs a passing mention buried in it. The result is
    then nudged by recency and how often you've resumed that session. A term that exists almost
    nowhere as text (a typo like 'conection') falls back to subsequence matching.
    """
    usage = usage or {}
    n = len(rows) or 1
    keys = query_keys(qterms)

    # Pass 1: per-field term frequencies, plus document frequency per term for idf.
    # A key equal to the word typed was never stemmed, so match it whole.
    probes = [_key_probe(k, whole=(k == t)) for (t, _st), k in zip(qterms, keys)]
    data = []
    df = dict.fromkeys(keys, 0)
    for f in rows:
        title_hay = (f[C_TITLE] + " " + short_proj(f[C_CWD])).lower()
        first_hay = f[C_FIRST].lower()
        body_hay = f[C_BLOB]                     # already lowercased at index time
        rec = []
        for key, probe in zip(keys, probes):
            tf = probe(body_hay)
            tf_first = probe(first_hay)
            it = probe(title_hay) > 0
            if tf or tf_first or it:
                df[key] += 1
            rec.append((tf, tf_first, it))
        data.append((f, title_hay, first_hay, body_hay, rec))

    avg_body = sum(len(d[3]) for d in data) / len(data) if data else 1
    avg_first = sum(len(d[2]) for d in data) / len(data) if data else 1
    # Standard BM25 idf, shifted to stay positive even for terms in most documents.
    idf = {k: math.log(1 + (n - df[k] + 0.5) / (df[k] + 0.5)) for k in df}
    # Barely-there word → probably a typo, so fall back to subsequence matching. Only for words
    # long enough that an in-order subsequence is evidence of anything: `p...r` within six
    # characters is satisfied by almost any English text, so short terms would fuzzy-match
    # every session they are genuinely absent from.
    typo = {k for k in df if df[k] <= 2 and len(k) >= 5}

    scored = []
    for f, title_hay, first_hay, body_hay, rec in data:
        score = 0.0
        matched = 0        # concepts covered, including typo-resolved ones — this is the badge
        strong = 0         # concepts the session actually contains — this is what ranks
        for (term, _stem_unused), key, (tf, tf_first, it) in zip(qterms, keys, rec):
            if tf or tf_first or it:
                matched += 1
                strong += 1
                score += idf[key] * (W_TITLE * it
                                     + W_FIRST * _bm25_tf(tf_first, len(first_hay), avg_first)
                                     + W_BODY * _bm25_tf(tf, len(body_hay), avg_body))
            elif key in typo:
                if _fuzzy_span(title_hay, term) is not None:
                    matched += 1
                    score += 1.0
                else:
                    sp = _fuzzy_span(body_hay, term)
                    if sp is not None and sp <= len(term) * 3:
                        matched += 1
                        score += 0.3
        if matched:
            scored.append((score * _boost(f, usage, now), matched, strong, f))
    # Categorise first: your own sessions always outrank plugin/SDK-spawned runs. Then coverage,
    # counting only words the session really contains, then relevance. Automation is demoted,
    # never hidden.
    #
    # Coverage deliberately ignores typo-resolved terms. A subsequence hit is a guess, and
    # letting a guess count toward coverage let a session that merely contains `b i l l i n g`
    # spread across a sentence outrank one genuinely about billing at three times the score.
    # The guess still raises `score`, so a good fuzzy match can win on relevance; it just
    # cannot win on breadth.
    scored.sort(key=lambda x: (x[3][C_KIND] == "auto", -x[2], -x[0]))
    return [(sc, m, f) for sc, m, _st, f in scored]


def _smart_rows(rows, qterms, live=frozenset(), usage=None, gone=frozenset(),
                forks=frozenset()):
    """Render the ranked sessions as fzf rows. Badge = matched/total query terms."""
    total = len(qterms)
    return [_row(f[C_SID], f[C_CWD], f[C_DATE], f[C_SOURCE], f[C_KIND], f[C_TITLE],
                 f"\033[33m{m}/{total}\033[0m", f[C_SID] in live, f[C_CWD] in gone,
                 f[C_SID] in forks)
            for _score, m, f in rank_sessions(rows, qterms, usage)[:200]]


def parse_query(query):
    """Query string → [(word, stem)]: stopwords dropped, words conservatively stemmed."""
    words = re.findall(r"[a-z0-9]+", query.lower())
    qwords = [w for w in words if w not in _STOP] or words   # keep stopwords if that's all
    return [(w, _stem(w)) for w in qwords]


def cmd_filter(argv):
    """fzf reload target: print CLEAN session rows ranked for the live query.

    Search happens here (fzf runs --disabled), so the list shows only `date · project · N/T ·
    title` while full conversation text is searched. Smart ranking: stopwords dropped, words
    stemmed (migration≈migrate), matched per-concept (exact→stem→typo), scored by weighted
    BM25 over title / first prompt / transcript, then nudged by recency and resume count.
    """
    qterms = parse_query(" ".join(argv[1:]) if argv else "")
    try:
        raw = open(SESSIONS_PATH, errors="replace").read().splitlines()
    except OSError:
        return
    rows = [line.split(SEP) for line in raw if line.count(SEP) >= SESSION_COLS - 1]

    live = _active_sids([f[C_SID] for f in rows])
    gone = _missing_dirs([f[C_CWD] for f in rows])
    forks = load_forks()
    if not qterms:                                  # initial list: yours first, then automation
        rows = sorted(rows, key=lambda f: f[C_KIND] == "auto")
        out = [_row(f[C_SID], f[C_CWD], f[C_DATE], f[C_SOURCE], f[C_KIND], f[C_TITLE], "    ",
                    f[C_SID] in live, f[C_CWD] in gone, f[C_SID] in forks) for f in rows]
    else:
        out = _smart_rows(rows, qterms, live, _usage_counts(), gone, forks)
    sys.stdout.write("\n".join(out))


def run_fzf(lines, query, thinking=False, no_resume=False, fuzzy=False):
    if not shutil.which("fzf"):
        print("fzf not installed. Falling back to non-interactive output.\n"
              "Install fzf 0.35+: https://github.com/junegunn/fzf#installation\n",
              file=sys.stderr)
        return print_matches(lines, query)

    self = os.path.abspath(__file__)
    build_sessions(lines)                # refresh the per-session cache _filter reads
    thi = "1" if thinking else "0"
    fz = "1" if fuzzy else "0"
    filter_cmd = "python3 {} _filter {} {{q}}".format(shlex.quote(self), fz)
    preview = "python3 {} _preview {{1}} {} {{q}}".format(shlex.quote(self), thi)
    pager = os.environ.get("AGSEARCH_PAGER") or os.environ.get("PAGER") or "less -R"
    transcript = "python3 {} _transcript {{1}} {} {{q}} | {}".format(
        shlex.quote(self), thi, pager)
    copy_cmd = "python3 {} _copy {{1}} {{2}}".format(shlex.quote(self))
    args = [
        "fzf", "--ansi", "--delimiter", SEP, "--with-nth", "3", "--disabled",
        "--print-query",                 # so the clipboard gets what you actually typed
        "--bind", "start:reload:" + filter_cmd,
        "--bind", "change:reload:" + filter_cmd,
        "--preview", preview,
        "--preview-window", "right,58%,wrap",
        "--header", "type to search all sessions · enter: resume (copies query for ⌘F) · "
                    "ctrl-o: read · ctrl-y: copy cmd · ctrl-/: preview · ● = live session",
        "--bind", "ctrl-/:toggle-preview",
        # ctrl-o reads the whole conversation in a pager: no resume, no CLI start,
        # no tokens. ctrl-y puts the reattach line on the clipboard.
        #
        # Only these two, and only on keys fzf leaves free. ctrl-u and ctrl-d are
        # fzf defaults (unix-line-discard and delete-char/eof); rebinding them
        # would take away "clear the query", which in a search box is the edit
        # people reach for most.
        "--bind", "ctrl-o:execute(" + transcript + ")",
        "--bind", "ctrl-y:execute-silent(" + copy_cmd + ")+bell",
    ]
    if query:
        args += ["--query", query]
    proc = subprocess.run(args, input="", text=True, capture_output=True)
    out = proc.stdout.split("\n")
    typed = out[0] if out else query     # --print-query puts the final query on line 1
    sel = next((l for l in out[1:] if l.strip()), "")
    if not sel:
        return 0
    f = (sel.split(SEP) + [""] * 4)[:4]
    sid, cwd, active = f[0], f[1], f[3] == "1"
    if no_resume:
        _source, _bin, argv, _cwd, target, _exists = resume_plan(sid, cwd)
        print(resume_command(target or cwd, argv))
        return 0
    resume(sid, cwd, typed, active=active)
    return 0


# ------------------------------------------------------------------ main

def main(argv):
    if argv and argv[0] == "_preview":
        sid = argv[1] if len(argv) > 1 else ""
        thinking = argv[2] if len(argv) > 2 else "0"
        query = " ".join(argv[3:])
        render_preview(sid, thinking, query)
        return 0
    if argv and argv[0] == "_filter":
        cmd_filter(argv[1:])
        return 0
    if argv and argv[0] == "_transcript":    # fzf ctrl-o, piped to a pager: keep the colour
        render_transcript(argv[1] if len(argv) > 1 else "",
                          argv[2] if len(argv) > 2 else "0", " ".join(argv[3:]), color=True)
        return 0
    if argv and argv[0] == "read":
        if len(argv) < 2 or not argv[1].strip():
            print("usage: agsearch read <session-id> [query]", file=sys.stderr)
            return 1
        rest = [a for a in argv[2:] if a != "--full"]
        full = "--full" in argv[2:] or _isatty(sys.stdout)
        sid, err = resolve_sid(argv[1])
        if err:
            print(err, file=sys.stderr)
            return 1
        return render_transcript(sid, "0", " ".join(rest),
                                 budget=None if full else AGENT_READ_CHARS)
    if argv and argv[0] == "_copy":
        sid = argv[1] if len(argv) > 1 else ""
        cwd = argv[2] if len(argv) > 2 else ""
        if sid:
            copy_to_clipboard(resume_line(sid, cwd))
        return 0

    no_fzf = no_resume = here = thinking = reindex = fuzzy = False
    project = None
    query_parts = []
    i = 0
    while i < len(argv):
        a = argv[i]
        if a in ("-n", "--no-fzf"):
            no_fzf = True
        elif a == "--no-resume":
            no_resume = True
        elif a == "--here":
            here = True
        elif a == "--thinking":
            thinking = True
        elif a == "--fuzzy":
            fuzzy = True
        elif a == "--reindex":
            reindex = True
        elif a in ("-p", "--project"):
            i += 1
            project = argv[i] if i < len(argv) else None
        elif a in ("-h", "--help"):
            print(__doc__)
            return 0
        elif a in ("-V", "--version"):
            print("agsearch " + __version__)
            return 0
        else:
            query_parts.append(a)
        i += 1

    query = " ".join(query_parts)
    lines = build_index(include_thinking=thinking, force=reindex)
    lines = apply_scope(lines, here=here, project=project)
    if not lines:
        print("No indexed sessions found.", file=sys.stderr)
        return 1

    if no_fzf:
        return print_matches(lines, query)
    return run_fzf(lines, query, thinking=thinking, no_resume=no_resume, fuzzy=fuzzy)


def _entry():
    """Console-script entry point.

    A [project.scripts] entry point is called with no arguments, while main()
    takes argv — so the two cannot be wired directly. This wrapper is what
    `pipx`/`uvx`/`pip install agsearch` invoke, and running the file directly
    goes through it too, so both paths share one error-handling path.
    """
    try:
        sys.exit(main(sys.argv[1:]))
    except KeyboardInterrupt:
        sys.exit(130)
    except SystemExit:
        raise
    except Exception:                       # never let the popup vanish without a reason
        import traceback
        _die("agsearch crashed:\n\n" + traceback.format_exc())


if __name__ == "__main__":
    _entry()
