from __future__ import annotations

import sqlite3
from dataclasses import dataclass
from pathlib import Path
from typing import TypeAlias

from .jsonio import as_map, iter_jsonl, parse_json_text, text
from .timeparse import file_time, unix_millis, unix_seconds
from .transcript import content_text, existing, flat_parallel, merge_usage, recent
from .types import JsonMap, Session

SqlValue: TypeAlias = str | int | float | bytes | None
SqlRow: TypeAlias = tuple[SqlValue, ...]


@dataclass(frozen=True, slots=True)
class _StateRow:
    parent_id: str | None
    title: str | None
    cwd: str | None
    provider: str | None
    model: str | None
    created_at: str | None
    updated_at: str | None


def scan_aside(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".aside"], extra_roots, (".aside",))
    sessions: list[Session] = []
    for user_dir in _user_dirs(roots):
        index = _state_index(user_dir / "state.db")
        paths = list((user_dir / "sessions").glob("*/messages.jsonl"))
        sessions.extend(flat_parallel(recent(paths), workers, lambda path, rows=index: [_aside_session(path, rows)]))
    return sessions


def _user_dirs(roots: list[Path]) -> list[Path]:
    dirs: list[Path] = []
    for root in roots:
        users = sorted(path for path in (root / "u").glob("*") if path.is_dir())
        if users:
            dirs.extend(users)
        elif (root / "sessions").exists():
            dirs.append(root)
    return dirs


def _state_index(db_path: Path) -> dict[str, _StateRow]:
    if not db_path.exists():
        return {}
    try:
        with sqlite3.connect(f"file:{db_path}?mode=ro", uri=True) as conn:
            rows: list[SqlRow] = conn.execute("select id, parent_id, title, cwd, model, created_at, updated_at from sessions").fetchall()
    except sqlite3.Error:
        return {}
    index: dict[str, _StateRow] = {}
    for row in rows:
        sid = _cell_text(row, 0)
        if sid is None:
            continue
        provider, model = _model_fields(_cell_text(row, 4))
        index[sid] = _StateRow(
            _cell_text(row, 1),
            _cell_text(row, 2),
            _cell_text(row, 3),
            provider,
            model,
            unix_seconds(_cell_number(row, 5)),
            unix_seconds(_cell_number(row, 6)),
        )
    return index


def _cell_text(row: SqlRow, index: int) -> str | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, str) else None


def _cell_number(row: SqlRow, index: int) -> int | float | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, int | float) else None


def _model_fields(model_json: str | None) -> tuple[str | None, str | None]:
    data = as_map(parse_json_text(model_json)) if model_json else None
    if data is None:
        return None, None
    return text(data.get("provider")), text(data.get("modelId")) or text(data.get("model"))


def _aside_session(path: Path, index: dict[str, _StateRow]) -> Session:
    sid = _dir_session_id(path.parent.name)
    row = index.get(sid)
    first_user = last_user = ""
    provider = model = None
    created = updated = None
    usage: JsonMap = {}
    for data in iter_jsonl(path):
        stamp = data.get("timestamp")
        moment = unix_millis(int(stamp)) if isinstance(stamp, int | float) else None
        created = created or moment
        updated = moment or updated
        role = text(data.get("role"))
        if role == "user":
            prompt = content_text(data.get("content"))
            if prompt:
                first_user = first_user or prompt
                last_user = prompt
        elif role == "assistant":
            provider = provider or text(data.get("provider"))
            model = model or text(data.get("model"))
            merge_usage(usage, as_map(data.get("usage")))
    return Session(
        "aside",
        sid,
        str(path),
        row.cwd if row is not None else None,
        (row.created_at if row is not None else None) or created or file_time(path),
        (row.updated_at if row is not None else None) or updated or created or file_time(path),
        (row.provider if row is not None else None) or provider,
        (row.model if row is not None else None) or model,
        first_user,
        usage,
        row.parent_id if row is not None else None,
        row.title if row is not None and row.parent_id is not None else None,
        last_user,
    )


def _dir_session_id(name: str) -> str:
    return name.split("_", 1)[-1]


def _roots(defaults: list[Path], extra_roots: tuple[Path, ...], children: tuple[str, ...]) -> list[Path]:
    candidates = [*defaults]
    for root in extra_roots:
        candidates.append(root)
        candidates.extend(root / child for child in children)
    return existing(candidates)

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from .jsonio import as_map, iter_jsonl, read_json, text
from .timeparse import file_time
from .transcript import content_text, env_path, existing, jsonl_parallel, recent
from .types import Session


def scan_claude(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    appdata = env_path("APPDATA")
    roots = existing([Path.home() / ".claude", *(path / "Claude" for path in (appdata,) if path is not None), *extra_roots])
    mains: list[Path] = []
    subagents: list[Path] = []
    for root in roots:
        mains.extend((root / "transcripts").glob("*.jsonl"))
        mains.extend((root / "projects").glob("*/*.jsonl"))
        mains.extend((root / "pre-compact-session-histories").glob("*.jsonl"))
        subagents.extend((root / "projects").glob("*/*/subagents/**/agent-*.jsonl"))
    sessions = jsonl_parallel(recent(mains), workers, "claude", lambda path: path.stem)
    sessions.extend(_subagent_parallel(recent(subagents), workers))
    return sessions


def _subagent_parallel(paths: list[Path], workers: int) -> list[Session]:
    if not paths:
        return []
    sessions: list[Session] = []
    with ThreadPoolExecutor(max_workers=min(workers, max(len(paths), 1))) as pool:
        futures = [pool.submit(_subagent_session, path) for path in paths]
        for future in as_completed(futures):
            session = future.result()
            if session is not None:
                sessions.append(session)
    return sessions


def _subagent_session(path: Path) -> Session | None:
    parts = path.parts
    if "subagents" not in parts:
        return None
    parent_sid = parts[parts.index("subagents") - 1]
    meta = as_map(read_json(path.with_name(path.stem + ".meta.json"))) or {}
    first = next(iter_jsonl(path), None) or {}
    message = as_map(first.get("message")) or {}
    description = text(meta.get("description")) or ""
    task = content_text(message.get("content"))
    prompt = "\n".join(part for part in (description, task) if part)
    created = text(first.get("timestamp")) or file_time(path)
    return Session(
        "claude",
        path.stem.removeprefix("agent-"),
        str(path),
        text(first.get("cwd")),
        created,
        file_time(path) or created,
        None,
        None,
        prompt,
        {},
        parent_sid,
        text(meta.get("agentType")),
        prompt,
    )

from __future__ import annotations

import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

from .jsonio import as_map, dumps, iter_jsonl
from .scanners import DEFAULT_PLATFORMS, scan
from .timeparse import date_bound, parse_stamp
from .transcript import user_text
from .types import Json, JsonMap, Options, Session


def main() -> int:
    command, opts, rest = _parse()
    sessions = sorted(scan(opts.platforms, opts.roots, opts.workers), key=lambda item: item.created_at or "", reverse=True)
    if command == "list":
        _emit(_list_payload(_filter(sessions, opts), sessions, opts.limit, include_subagents=opts.include_subagents))
        return 0
    if command == "search":
        queries = _queries(opts, rest)
        _require(list(queries), "search requires a query")
        _emit(_search_payload(_filter(sessions, opts), sessions, queries, opts.limit, opts.workers, include_subagents=opts.include_subagents))
        return 0
    if command == "get":
        _require(rest, "read requires at least one session id")
        _emit(_get_payload(sessions, rest))
        return 0
    raise SystemExit(f"unknown command: {command}")


def _parse() -> tuple[str, Options, list[str]]:
    import sys

    args = sys.argv[1:]
    if not args or args[0] in {"-h", "--help"}:
        usage = (
            "Usage: find-agent-sessions.py list|find|search|read|get [query|ids...] [--query TEXT ...] "
            "[--platform NAME ...] [--root PATH] [--from DATE] [--to DATE] [--cwd TEXT] "
            "[--model TEXT] [--limit N] [--workers N] [--include-subagents]"
        )
        print(usage)
        raise SystemExit(0)
    command = args.pop(0)
    if command == "find":
        command = "search"
    elif command == "read":
        command = "get"
    roots: list[Path] = []
    queries: list[str] = []
    platforms: list[str] = []
    date_from = date_to = cwd = model = None
    limit = 20
    workers = _default_workers()
    include_subagents = False
    rest: list[str] = []
    index = 0
    while index < len(args):
        arg = args[index]
        if arg == "--root":
            roots.append(Path(args[index + 1]).expanduser())
            index += 2
        elif arg == "--query":
            queries.append(args[index + 1])
            index += 2
        elif arg == "--platform":
            platform = args[index + 1].strip().lower()
            if "," in platform:
                raise SystemExit("Use repeated --platform flags, for example: --platform senpi --platform opencode")
            platforms.append(platform)
            index += 2
        elif arg == "--from":
            date_from = args[index + 1]
            index += 2
        elif arg == "--to":
            date_to = args[index + 1]
            index += 2
        elif arg == "--cwd":
            cwd = args[index + 1].lower()
            index += 2
        elif arg == "--model":
            model = args[index + 1].lower()
            index += 2
        elif arg == "--limit":
            limit = int(args[index + 1])
            index += 2
        elif arg == "--workers":
            workers = max(int(args[index + 1]), 1)
            index += 2
        elif arg == "--include-subagents":
            include_subagents = True
            index += 1
        else:
            rest.append(arg)
            index += 1
    selected_platforms = frozenset(platforms) if platforms else DEFAULT_PLATFORMS
    return command, Options(selected_platforms, tuple(roots), tuple(queries), date_from, date_to, cwd, model, limit, workers, include_subagents), rest


def _filter(sessions: list[Session], opts: Options) -> list[Session]:
    start = date_bound(opts.date_from)
    end = date_bound(opts.date_to, end=True)
    result: list[Session] = []
    for item in sessions:
        stamp = parse_stamp(item.created_at)
        if start is not None and stamp is not None and stamp < start:
            continue
        if end is not None and stamp is not None and stamp >= end:
            continue
        if opts.cwd is not None and opts.cwd not in (item.cwd or "").lower():
            continue
        if opts.model is not None and opts.model not in (item.model or "").lower():
            continue
        result.append(item)
    return result


def _child_counts(sessions: list[Session]) -> dict[tuple[str, str], int]:
    counts: dict[tuple[str, str], int] = {}
    for item in sessions:
        if item.parent_id is not None:
            key = (item.platform, item.parent_id)
            counts[key] = counts.get(key, 0) + 1
    return counts


def _annotate(item: Session, counts: dict[tuple[str, str], int]) -> JsonMap:
    data = item.to_json()
    data["subagent_count"] = counts.get((item.platform, item.id), 0)
    data["detail_hint"] = _detail_hint(item)
    return data


def _list_payload(filtered: list[Session], all_sessions: list[Session], limit: int, include_subagents: bool = False) -> JsonMap:
    counts = _child_counts(all_sessions)
    candidates = filtered if include_subagents else [item for item in filtered if item.parent_id is None]
    results: list[Json] = [_annotate(item, counts) for item in candidates[:limit]]
    return {"count": len(results), "results": results}


def _search_payload(filtered: list[Session], all_sessions: list[Session], queries: tuple[str, ...], limit: int, workers: int, include_subagents: bool = False) -> JsonMap:
    counts = _child_counts(all_sessions)
    candidates = filtered if include_subagents else [item for item in filtered if item.parent_id is None]
    per_query: list[Json] = []
    merged: dict[tuple[str, str], tuple[Session, list[JsonMap]]] = {}
    with ThreadPoolExecutor(max_workers=min(workers, max(len(queries), 1))) as pool:
        futures = [pool.submit(_search_one, candidates, query, limit) for query in queries]
        for future in as_completed(futures):
            query, matches = future.result()
            per_query.append({"query": query, "count": len(matches), "results": [_annotate_search(item, counts, reasons) for item, reasons in matches]})
            for item, reasons in matches:
                key = (item.platform, item.id)
                if key not in merged:
                    merged[key] = (item, reasons)
    results: list[Json] = [_annotate_search(item, counts, reasons) for item, reasons in list(merged.values())[:limit]]
    return {"count": len(results), "queries": per_query, "results": results}


def _get_payload(sessions: list[Session], ids: list[str]) -> JsonMap:
    counts = _child_counts(sessions)
    results: list[Json] = []
    for item in sessions:
        if item.id not in ids and not any(item.id.startswith(prefix) for prefix in ids):
            continue
        events = _events(item)
        first_prompt, last_prompt = _prompt_edges(item, events)
        session = _annotate(item, counts)
        session["first_user_message"] = first_prompt[:300]
        session["last_user_message"] = last_prompt[:300]
        children = sorted(
            (child for child in sessions if child.platform == item.platform and child.parent_id == item.id),
            key=lambda child: child.created_at or "",
        )
        results.append(
            {
                "session": session,
                "prompts": {"first_user_message": first_prompt, "last_user_message": last_prompt},
                "events": events,
                "subagents": [_annotate(child, counts) for child in children],
                "detail_hint": _detail_hint(item),
            }
        )
    return {"count": len(results), "results": results}


def _annotate_search(item: Session, counts: dict[tuple[str, str], int], reasons: list[JsonMap]) -> JsonMap:
    data = _annotate(item, counts)
    reason_rows: list[Json] = []
    reason_rows.extend(reasons)
    data["match_reasons"] = reason_rows
    return data


def _detail_hint(item: Session) -> str:
    return f"python3 scripts/find-agent-sessions.py read {item.id} --platform {item.platform}"


def _events(item: Session) -> list[Json]:
    if item.path.endswith(".jsonl"):
        return list(iter_jsonl(Path(item.path)))
    events: list[Json] = []
    if item.first_user_message:
        events.append({"type": "message", "message": {"role": "user", "content": item.first_user_message}})
    if item.last_user_message and item.last_user_message != item.first_user_message:
        events.append({"type": "message", "message": {"role": "user", "content": item.last_user_message}})
    return events


def _prompt_edges(item: Session, events: list[Json]) -> tuple[str, str]:
    first_prompt = item.first_user_message
    last_prompt = item.last_user_message or item.first_user_message
    for event in events:
        if not isinstance(event, dict):
            continue
        message = as_map(event.get("message")) or as_map(event.get("payload")) or {}
        prompt = user_text(event, message)
        if prompt:
            first_prompt = first_prompt or prompt
            last_prompt = prompt
    return first_prompt, last_prompt


def _match_reasons(item: Session, query: str) -> list[JsonMap]:
    needle = query.lower()
    reasons: list[JsonMap] = []
    for field, value in _search_fields(item):
        if needle in value.lower():
            reasons.append({"query": query, "platform": item.platform, "field": field, "snippet": _snippet(value, needle)})
    return reasons


def _search_fields(item: Session) -> tuple[tuple[str, str], ...]:
    return (
        ("platform", item.platform),
        ("id", item.id),
        ("path", item.path),
        ("cwd", item.cwd or ""),
        ("provider", item.provider or ""),
        ("model", item.model or ""),
        ("agent", item.agent or ""),
        ("first_user_message", item.first_user_message),
        ("last_user_message", item.last_user_message),
    )


def _snippet(value: str, needle: str) -> str:
    start = max(value.lower().find(needle) - 60, 0)
    end = min(start + 160, len(value))
    return value[start:end]


def _queries(opts: Options, rest: list[str]) -> tuple[str, ...]:
    if opts.queries:
        return opts.queries
    joined = " ".join(rest).strip()
    return (joined,) if joined else ()


def _search_one(sessions: list[Session], query: str, limit: int) -> tuple[str, list[tuple[Session, list[JsonMap]]]]:
    matches: list[tuple[Session, list[JsonMap]]] = []
    for item in sessions:
        reasons = _match_reasons(item, query)
        if reasons:
            matches.append((item, reasons))
    return query, matches[:limit]


def _emit(value: JsonMap) -> None:
    print(dumps(value))


def _require(values: list[str], message: str) -> None:
    if not values:
        raise SystemExit(message)


def _default_workers() -> int:
    cpu = os.cpu_count() or 4
    return min(max(cpu * 4, 8), 64)


if __name__ == "__main__":
    raise SystemExit(main())

from __future__ import annotations

import sqlite3
from pathlib import Path
from typing import TypeAlias

from .timeparse import unix_seconds
from .transcript import env_path, existing, flat_parallel, jsonl_parallel, nick_role, recent, spawn_info, stem_id
from .types import Json, Session

THREADS_SQL = (
    "SELECT id, rollout_path, cwd, created_at, updated_at, model_provider, model, first_user_message, tokens_used, "
    "source, agent_nickname, agent_role FROM threads"
)
LEGACY_THREADS_SQL = (
    "SELECT id, rollout_path, cwd, created_at, updated_at, model_provider, model, first_user_message, tokens_used FROM threads"
)
SPAWN_EDGES_SQL = "SELECT child_thread_id, parent_thread_id FROM thread_spawn_edges"

SqliteScalar: TypeAlias = str | int | float | bytes | None
CodexRow: TypeAlias = tuple[SqliteScalar, ...]


def scan_codex(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = existing([*(path for path in (env_path("CODEX_HOME"),) if path is not None), Path.home() / ".codex", *extra_roots])
    db_paths: list[Path] = []
    rollouts: list[Path] = []
    for root in roots:
        db_paths.extend(root.glob("state_*.sqlite"))
        rollouts.extend((root / "sessions").rglob("rollout-*.jsonl"))
        rollouts.extend((root / "archived_sessions").glob("rollout-*.jsonl"))
    sessions = flat_parallel(db_paths, workers, _codex_db)
    sessions.extend(jsonl_parallel(recent(rollouts), workers, "codex", lambda path: stem_id(path, "rollout-")))
    return sessions


def _codex_db(path: Path) -> list[Session]:
    conn: sqlite3.Connection | None = None
    try:
        conn = sqlite3.connect(str(path), timeout=3)
        edges = _spawn_edges(conn)
        rows = _thread_rows(conn)
    except sqlite3.Error:
        return []
    finally:
        if conn is not None:
            conn.close()
    return [_codex_row(path, row, edges) for row in rows]


def _thread_rows(conn: sqlite3.Connection) -> list[CodexRow]:
    try:
        primary_rows: list[CodexRow] = conn.execute(THREADS_SQL).fetchall()
        return primary_rows
    except sqlite3.Error:
        legacy_rows: list[CodexRow] = conn.execute(LEGACY_THREADS_SQL).fetchall()
        return legacy_rows


def _spawn_edges(conn: sqlite3.Connection) -> dict[str, str]:
    try:
        rows: list[CodexRow] = conn.execute(SPAWN_EDGES_SQL).fetchall()
    except sqlite3.Error:
        return {}
    edges: dict[str, str] = {}
    for row in rows:
        child = _row_text(_row_value(row, 0))
        parent = _row_text(_row_value(row, 1))
        if child is not None and parent is not None:
            edges[child] = parent
    return edges


def _codex_row(path: Path, row: CodexRow, edges: dict[str, str]) -> Session:
    thread_id = str(_row_value(row, 0))
    source_parent, source_agent = spawn_info(_row_text(_row_value(row, 9)))
    return Session(
        "codex",
        thread_id,
        str(_row_value(row, 1) or path),
        _row_text(_row_value(row, 2)),
        unix_seconds(_row_number(_row_value(row, 3))),
        unix_seconds(_row_number(_row_value(row, 4))),
        _row_text(_row_value(row, 5)),
        _row_text(_row_value(row, 6)),
        _row_text(_row_value(row, 7)) or "",
        {"total_tokens": _row_number(_row_value(row, 8))},
        edges.get(thread_id) or source_parent,
        nick_role(_row_text(_row_value(row, 10)), _row_text(_row_value(row, 11))) or source_agent,
    )


def _row_text(value: Json | int | float | None) -> str | None:
    return value if isinstance(value, str) else None


def _row_number(value: Json | int | float | None) -> int | float | None:
    return value if isinstance(value, int | float) else None


def _row_value(row: CodexRow, index: int) -> Json | int | float | None:
    value = row[index] if index < len(row) else None
    if value is None or isinstance(value, str | int | float | bool):
        return value
    return None

from __future__ import annotations

from pathlib import Path
from urllib.parse import unquote, urlparse

from .jsonio import as_map, int_value, iter_jsonl, parse_json_text, read_json, text
from .timeparse import file_time, unix_millis, unix_seconds
from .transcript import content_text, env_path, existing, flat_parallel, jsonl_parallel, recent
from .types import Json, JsonMap, Session


def scan_openclaw(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".openclaw"], extra_roots, (".openclaw",))
    paths: list[Path] = []
    for root in roots:
        paths.extend((root / "agents").glob("*/sessions/*.jsonl"))
        paths.extend((root / "sessions").glob("*.jsonl"))
        paths.extend((root / "session-backups").glob("*/*.jsonl"))
        paths.extend((root / "session-backups").glob("*/sessions/*.jsonl"))
    return jsonl_parallel(recent(paths), workers, "openclaw", lambda path: path.stem)


def scan_droid(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".factory"], extra_roots, (".factory",))
    paths = [path for root in roots for path in (root / "sessions").glob("*/*.jsonl")]
    return flat_parallel(recent(paths), workers, lambda path: [_droid_session(path)])


def scan_amp(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".local" / "share" / "amp"], extra_roots, ("amp", ".local/share/amp"))
    paths = [path for root in roots for path in (root / "threads").glob("T-*.json")]
    return flat_parallel(recent(paths), workers, _amp_sessions)


def scan_gemini(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".gemini"], extra_roots, (".gemini",))
    paths = [path for root in roots for path in (root / "tmp").glob("*/chats/*.json")]
    return flat_parallel(recent(paths), workers, _gemini_sessions)


def scan_kimi(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".kimi"], extra_roots, (".kimi",))
    paths = [path for root in roots for path in (root / "sessions").glob("*/*/wire.jsonl")]
    return flat_parallel(recent(paths), workers, lambda path: [_kimi_session(path)])


def scan_qwen(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".qwen"], extra_roots, (".qwen",))
    paths = [path for root in roots for path in (root / "projects").glob("*/chats/*.jsonl")]
    return jsonl_parallel(recent(paths), workers, "qwen", lambda path: path.stem)


def scan_codebuff(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    defaults = [Path.home() / ".config" / name for name in ("manicode", "manicode-dev", "manicode-staging", "codebuff")]
    roots = _roots(defaults, extra_roots, ("manicode", "codebuff"))
    paths = [path for root in roots for path in (root / "projects").glob("*/chats/*/chat-messages.json")]
    return flat_parallel(recent(paths), workers, _codebuff_sessions)


def scan_roocode(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    return _scan_task_dirs("roo-code", "rooveterinaryinc.roo-cline", extra_roots, workers)


def scan_kilocode(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    return _scan_task_dirs("kilo-code", "kilocode.kilo-code", extra_roots, workers)


def scan_cline(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    return _scan_task_dirs("cline", "saoudrizwan.claude-dev", extra_roots, workers)


def scan_aider(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / "local-workspaces", Path.home() / "indent"], extra_roots, ("local-workspaces",))
    paths: list[Path] = []
    for root in roots:
        paths.extend(_bounded_named(root, ".aider.chat.history.md"))
    return flat_parallel(recent(paths), workers, _aider_sessions)


def _droid_session(path: Path) -> Session:
    sid = path.stem
    cwd = model = first_user = last_user = None
    created = updated = None
    settings = as_map(read_json(path.with_suffix(".settings.json"))) or {}
    for data in iter_jsonl(path):
        created = created or text(data.get("timestamp"))
        updated = text(data.get("timestamp")) or updated
        if data.get("type") == "session_start":
            sid = text(data.get("id")) or sid
            cwd = cwd or text(data.get("cwd"))
        message = as_map(data.get("message")) or {}
        if message.get("role") == "user":
            prompt = _without_system_reminders(message.get("content"))
            if prompt:
                first_user = first_user or prompt
                last_user = prompt
    return Session("droid", sid, str(path), cwd, created or file_time(path), updated or created or file_time(path), None, text(settings.get("model")) or model, first_user or "", _usage(settings), last_user_message=last_user or "")


def _amp_sessions(path: Path) -> list[Session]:
    data = as_map(read_json(path))
    if data is None:
        return []
    messages = _json_maps(data.get("messages"))
    first_user, last_user = _message_edges(messages)
    env = as_map(data.get("env")) or {}
    initial = as_map(env.get("initial")) or {}
    trees = initial.get("trees")
    tree = as_map(trees[0]) if isinstance(trees, list) and trees else None
    created = int_value(data.get("created"))
    return [
        Session(
            "amp",
            text(data.get("id")) or path.stem,
            str(path),
            _file_uri_path(text(tree.get("uri")) if tree is not None else None),
            unix_millis(created) or file_time(path),
            file_time(path),
            None,
            _tag_model(initial.get("tags")),
            first_user,
            {"message_count": len(messages)},
            last_user_message=last_user,
        )
    ]


def _gemini_sessions(path: Path) -> list[Session]:
    data = as_map(read_json(path))
    if data is None:
        return []
    messages = _json_maps(data.get("messages"))
    first_user, last_user = _message_edges(messages, role_keys=("user",))
    return [Session("gemini", text(data.get("sessionId")) or path.stem, str(path), text(data.get("projectHash")), text(data.get("startTime")) or file_time(path), text(data.get("lastUpdated")) or file_time(path), "google", None, first_user, {"message_count": len(messages)}, last_user_message=last_user)]


def _kimi_session(path: Path) -> Session:
    first_user = last_user = ""
    created = updated = None
    for data in iter_jsonl(path):
        stamp = data.get("timestamp")
        created = created or unix_seconds(stamp if isinstance(stamp, int | float) else None)
        updated = unix_seconds(stamp if isinstance(stamp, int | float) else None) or updated
        payload = as_map(data.get("payload")) or {}
        prompt = text(payload.get("user_input")) if data.get("type") == "TurnBegin" else None
        if prompt:
            first_user = first_user or prompt
            last_user = prompt
    return Session("kimi", path.parent.name, str(path), None, created or file_time(path), updated or file_time(path), "moonshot", None, first_user, {}, last_user_message=last_user)


def _codebuff_sessions(path: Path) -> list[Session]:
    data = read_json(path)
    rows = _json_maps(data)
    first_user, last_user = _message_edges(rows, role_keys=("user", "human"))
    return [Session("codebuff", path.parent.name, str(path), None, file_time(path), file_time(path), None, None, first_user, {"message_count": len(rows)}, last_user_message=last_user)] if rows else []


def _scan_task_dirs(platform: str, extension: str, extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    appdata = env_path("APPDATA")
    defaults = [
        Path.home() / "Library" / "Application Support" / "Code" / "User" / "globalStorage" / extension,
        Path.home() / ".config" / "Code" / "User" / "globalStorage" / extension,
        *(path / "Code" / "User" / "globalStorage" / extension for path in (appdata,) if path is not None),
    ]
    roots = _roots(defaults, extra_roots, (extension,))
    paths = [path for root in roots for path in (root / "tasks").glob("*/api_conversation_history.json")]
    return flat_parallel(recent(paths), workers, lambda path: _task_sessions(platform, path))


def _task_sessions(platform: str, path: Path) -> list[Session]:
    data = read_json(path)
    rows = _json_maps(data)
    first_user, last_user = _message_edges(rows, role_keys=("user",))
    return [Session(platform, path.parent.name, str(path), None, file_time(path), file_time(path), None, None, first_user, {"message_count": len(rows)}, last_user_message=last_user)] if rows else []


def _aider_sessions(path: Path) -> list[Session]:
    text_value = path.read_text(encoding="utf-8", errors="replace")
    sessions: list[Session] = []
    for block in text_value.split("# aider chat started at "):
        if not block.strip():
            continue
        first_line, _, body = block.partition("\n")
        prompt = _first_aider_prompt(body)
        stamp = first_line.strip()
        sid = f"{path.parent.name}-{stamp.replace(':', '-').replace(' ', '-')}"
        sessions.append(Session("aider", sid, str(path), str(path.parent), stamp, file_time(path), None, None, prompt, {}, last_user_message=prompt))
    return sessions


def _message_edges(messages: list[JsonMap], role_keys: tuple[str, ...] = ("user",)) -> tuple[str, str]:
    first_user = ""
    last_user = ""
    for message in messages:
        if text(message.get("role")) not in role_keys and text(message.get("type")) not in role_keys:
            continue
        prompt = _content(message.get("content"))
        if prompt:
            first_user = first_user or prompt
            last_user = prompt
    return first_user, last_user


def _json_maps(value: Json | None) -> list[JsonMap]:
    if not isinstance(value, list):
        return []
    return [item for item in value if isinstance(item, dict)]


def _content(value: Json | None) -> str:
    parsed = parse_json_text(value) if isinstance(value, str) else value
    return content_text(parsed) or (value if isinstance(value, str) else "")


def _without_system_reminders(value: Json | None) -> str:
    if not isinstance(value, list):
        return _content(value)
    parts: list[str] = []
    for item in value:
        if not isinstance(item, dict):
            continue
        value_text = text(item.get("text")) or text(item.get("content")) or ""
        if value_text.startswith("<system-reminder>"):
            continue
        parts.append(value_text)
    return "\n".join(part for part in parts if part)


def _usage(data: JsonMap) -> JsonMap:
    usage = as_map(data.get("tokenUsage")) or {}
    return {key: value for key, value in usage.items() if isinstance(value, int | float)}


def _roots(defaults: list[Path], extra_roots: tuple[Path, ...], children: tuple[str, ...]) -> list[Path]:
    candidates = [*defaults]
    for root in extra_roots:
        candidates.append(root)
        candidates.extend(root / child for child in children)
    return existing(candidates)


def _bounded_named(root: Path, name: str) -> list[Path]:
    return [path for pattern in (name, f"*/{name}", f"*/*/{name}") for path in root.glob(pattern)]


def _file_uri_path(value: str | None) -> str | None:
    if value is None:
        return None
    parsed = urlparse(value)
    if parsed.scheme != "file":
        return value
    path = unquote(parsed.path)
    if parsed.netloc and parsed.netloc != "localhost":
        return f"//{parsed.netloc}{path}"
    if len(path) >= 3 and path[0] == "/" and path[2] == ":" and path[1].isalpha():
        return path[1:]
    return path


def _tag_model(value: Json | None) -> str | None:
    if not isinstance(value, list):
        return None
    for item in value:
        tag = text(item)
        if tag is not None and tag.startswith("model:"):
            return tag.removeprefix("model:")
    return None


def _first_aider_prompt(body: str) -> str:
    for line in body.splitlines():
        if line.startswith("#### "):
            return line.removeprefix("#### ").strip()
    return ""

from __future__ import annotations

import json
from collections.abc import Iterator
from pathlib import Path
from typing import TYPE_CHECKING

from .types import Json, JsonMap

if TYPE_CHECKING:
    def _loads(_text: str) -> Json: ...
else:
    def _loads(text_value: str) -> Json:
        return json.loads(text_value)


def as_map(value: Json | None) -> JsonMap | None:
    return value if isinstance(value, dict) else None


def text(value: Json | None) -> str | None:
    return value if isinstance(value, str) else None


def int_value(value: Json | None) -> int | None:
    return value if isinstance(value, int) else None


def parse_json_text(value: str) -> Json | None:
    try:
        return _loads(value)
    except json.JSONDecodeError:
        return None


def read_json(path: Path) -> Json | None:
    try:
        return parse_json_text(path.read_text(encoding="utf-8", errors="replace"))
    except OSError:
        return None


def iter_jsonl(path: Path) -> Iterator[JsonMap]:
    try:
        with path.open(encoding="utf-8", errors="replace") as handle:
            for line in handle:
                value = parse_json_text(line)
                if isinstance(value, dict):
                    yield value
    except OSError:
        return


def dumps(value: JsonMap) -> str:
    return json.dumps(value, ensure_ascii=False, indent=2)

from __future__ import annotations

from pathlib import Path

from .jsonio import as_map, iter_jsonl, read_json, text
from .timeparse import file_time, unix_seconds
from .transcript import existing, flat_parallel, recent
from .types import Json, JsonMap, Session


def scan_kiro(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".kiro"], extra_roots, (".kiro", "kiro"))
    paths = [path for root in roots for path in (root / "sessions" / "cli").glob("*.json")]
    return flat_parallel(recent(paths), workers, _kiro_sessions)


def _kiro_sessions(path: Path) -> list[Session]:
    data = as_map(read_json(path))
    if data is None:
        return []
    state = as_map(data.get("session_state")) or {}
    rts_state = as_map(state.get("rts_model_state")) or {}
    model_info = as_map(rts_state.get("model_info")) or {}
    metadata = as_map(state.get("conversation_metadata")) or {}
    turns = metadata.get("user_turn_metadatas")
    first_user, last_user, created = _kiro_prompt_edges(path.with_suffix(".jsonl"))
    if not first_user:
        return []
    usage: JsonMap = {"turn_count": len(turns)} if isinstance(turns, list) else {}
    return [
        Session(
            "kiro",
            text(data.get("session_id")) or text(data.get("sessionId")) or path.stem,
            str(path),
            text(data.get("cwd")),
            created or file_time(path),
            file_time(path),
            "amazon-bedrock",
            text(model_info.get("model_id")),
            first_user,
            usage,
            last_user_message=last_user,
        )
    ]


def _kiro_prompt_edges(path: Path) -> tuple[str, str, str | None]:
    first_user = last_user = ""
    created = None
    for row in iter_jsonl(path):
        if row.get("kind") != "Prompt":
            continue
        data = as_map(row.get("data")) or {}
        prompt = _kiro_content(data.get("content"))
        if prompt:
            first_user = first_user or prompt
            last_user = prompt
        meta = as_map(data.get("meta")) or {}
        stamp = meta.get("timestamp")
        if created is None and isinstance(stamp, int | float):
            created = unix_seconds(stamp)
    return first_user, last_user, created


def _kiro_content(value: Json | None) -> str:
    if not isinstance(value, list):
        return text(value) or ""
    parts: list[str] = []
    for item in value:
        part = _kiro_content_part(item)
        if part:
            parts.append(part)
    return "\n".join(parts)


def _kiro_content_part(value: Json) -> str:
    if not isinstance(value, dict):
        return ""
    item: JsonMap = value
    kind = text(item.get("kind"))
    if kind is not None and kind != "text":
        return ""
    return text(item.get("data")) or text(item.get("text")) or text(item.get("content")) or ""


def _roots(defaults: list[Path], extra_roots: tuple[Path, ...], children: tuple[str, ...]) -> list[Path]:
    candidates = [*defaults]
    for root in extra_roots:
        candidates.append(root)
        candidates.extend(root / child for child in children)
    return existing(candidates)

from __future__ import annotations

import shutil
import sqlite3
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import TypeAlias

from .jsonio import as_map, int_value, parse_json_text, read_json, text
from .timeparse import unix_millis
from .transcript import env_path, existing
from .types import JsonMap, Session

MAX_OPENCODE_SESSIONS = 50000
MAX_OPENCODE_CLI_SESSIONS = 100
OPENCODE_TIMEOUT_SECONDS = 8
SESSION_SQL = (
    "select id, title, directory, time_created, time_updated, cost, tokens_input, tokens_output, "
    "tokens_reasoning, tokens_cache_read, tokens_cache_write, model, parent_id, agent "
    "from session where time_archived is null "
    f"order by time_updated desc limit {MAX_OPENCODE_SESSIONS}"
)
LEGACY_SESSION_SQL = (
    "select id, title, directory, time_created, time_updated, cost, tokens_input, tokens_output, "
    "tokens_reasoning, tokens_cache_read, tokens_cache_write, model "
    "from session where time_archived is null "
    "order by time_updated desc limit 2000"
)
SqlValue: TypeAlias = str | int | float | bytes | None
OpenCodeRow: TypeAlias = tuple[SqlValue, ...]


def scan_opencode(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    if not extra_roots:
        sessions = _db_sessions()
        if sessions:
            return sessions
        sessions = _cli_sessions()
        if sessions:
            return sessions
    appdata = env_path("APPDATA")
    roots = existing([
        *(path for path in (env_path("OPENCODE_HOME"),) if path is not None),
        Path.home() / ".opencode",
        Path.home() / ".local" / "share" / "opencode",
        *(path / "opencode" for path in (appdata,) if path is not None),
        *extra_roots,
    ])
    message_dirs = [message_dir for root in roots for message_dir in (root / "messages").glob("ses_*")]
    sessions: list[Session] = []
    if message_dirs:
        with ThreadPoolExecutor(max_workers=min(workers, len(message_dirs))) as pool:
            futures = [pool.submit(_file_session, message_dir.parents[1], message_dir) for message_dir in message_dirs]
            for future in as_completed(futures):
                session = future.result()
                if session is not None:
                    sessions.append(session)
    sessions.extend(_storage_sessions(roots))
    return sessions


def _storage_sessions(roots: list[Path]) -> list[Session]:
    sessions: list[Session] = []
    for root in roots:
        for path in (root / "storage" / "session").glob("*/ses_*.json"):
            info = as_map(read_json(path))
            if info is None:
                continue
            time_info = as_map(info.get("time")) or {}
            sessions.append(
                Session(
                    "opencode",
                    text(info.get("id")) or path.stem,
                    str(path),
                    text(info.get("directory")),
                    unix_millis(int_value(time_info.get("created"))),
                    unix_millis(int_value(time_info.get("updated"))),
                    None,
                    None,
                    text(info.get("title")) or "",
                    {},
                    text(info.get("parentID")),
                    text(info.get("agent")),
                )
            )
    return sessions


def _db_sessions() -> list[Session]:
    path = _db_path()
    if path is None:
        return []
    conn: sqlite3.Connection | None = None
    try:
        conn = sqlite3.connect(str(path), timeout=3)
        rows: list[OpenCodeRow] = _session_rows(conn)
    except sqlite3.Error:
        return []
    finally:
        if conn is not None:
            conn.close()
    return [_db_session(row) for row in rows]


def _session_rows(conn: sqlite3.Connection) -> list[OpenCodeRow]:
    try:
        return conn.execute(SESSION_SQL).fetchall()
    except sqlite3.Error:
        return conn.execute(LEGACY_SESSION_SQL).fetchall()


def _db_path() -> Path | None:
    data = _opencode_text(["db", "path"])
    candidates = [Path(data.strip()).expanduser()] if data else []
    appdata = env_path("APPDATA")
    candidates.extend([Path.home() / ".local" / "share" / "opencode" / "opencode.db"])
    candidates.extend(path / "opencode" / "opencode.db" for path in (appdata,) if path is not None)
    for path in candidates:
        if str(path) and path.exists():
            return path
    return None


def _db_session(row: OpenCodeRow) -> Session:
    session_id = _row_text(row, 0) or ""
    title = _row_text(row, 1) or ""
    model = as_map(parse_json_text(_row_text(row, 11) or ""))
    return Session(
        "opencode",
        session_id,
        f"opencode://{session_id}",
        _row_text(row, 2),
        unix_millis(_row_int(row, 3)),
        unix_millis(_row_int(row, 4)),
        text(model.get("providerID")) if model is not None else None,
        text(model.get("id")) if model is not None else None,
        title,
        _db_usage(row),
        _row_text(row, 12),
        _row_text(row, 13),
    )


def _db_usage(row: OpenCodeRow) -> JsonMap:
    usage: JsonMap = {}
    for index, key in ((5, "cost_total"), (6, "input"), (7, "output"), (8, "reasoning"), (9, "cacheRead"), (10, "cacheWrite")):
        value = _row_number(row, index)
        if value is not None:
            usage[key] = value
    return usage


def _row_text(row: OpenCodeRow, index: int) -> str | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, str) else None


def _row_number(row: OpenCodeRow, index: int) -> int | float | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, int | float) else None


def _row_int(row: OpenCodeRow, index: int) -> int | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, int) else None


def _cli_sessions() -> list[Session]:
    data = _opencode_json(["session", "list", "--format", "json", "--max-count", str(MAX_OPENCODE_CLI_SESSIONS)])
    if not isinstance(data, list):
        return []
    return [_cli_session(item) for item in data]


def _cli_session(item: JsonMap) -> Session:
    session_id = text(item.get("id")) or ""
    title = text(item.get("title")) or ""
    usage = _usage(item)
    return Session(
        "opencode",
        session_id,
        f"opencode://{session_id}",
        text(item.get("directory")),
        unix_millis(int_value(item.get("created"))),
        unix_millis(int_value(item.get("updated"))),
        None,
        None,
        title,
        usage,
    )


def _opencode_json(args: list[str]) -> JsonMap | list[JsonMap] | None:
    data = _opencode_text(args)
    return _json_result(data) if data is not None else None


def _opencode_text(args: list[str]) -> str | None:
    binary = shutil.which("opencode")
    if binary is None:
        return None
    try:
        proc = subprocess.run(
            [binary, *args],
            check=False,
            capture_output=True,
            text=True,
            timeout=OPENCODE_TIMEOUT_SECONDS,
        )
    except (OSError, subprocess.TimeoutExpired):
        return None
    if proc.returncode != 0:
        return None
    return proc.stdout


def _json_result(value: str) -> JsonMap | list[JsonMap] | None:
    data = parse_json_text(value)
    if isinstance(data, dict):
        return data
    if isinstance(data, list):
        rows = [item for item in data if isinstance(item, dict)]
        return rows
    return None


def _usage(item: JsonMap) -> JsonMap:
    usage: JsonMap = {}
    cost = item.get("cost")
    tokens = as_map(item.get("tokens"))
    if isinstance(cost, int | float):
        usage["cost_total"] = cost
    if tokens is not None:
        for key in ("input", "output", "reasoning", "cacheRead", "cacheWrite"):
            value = tokens.get(key)
            if isinstance(value, int | float):
                usage[key] = value
    return usage


def _file_session(root: Path, message_dir: Path) -> Session | None:
    messages = sorted(message_dir.glob("*.json"), key=lambda item: item.name)
    first = as_map(read_json(messages[0])) if messages else None
    if first is None:
        return None
    session_id = text(first.get("sessionID")) or message_dir.name
    model = as_map(first.get("model")) or {}
    path_info = as_map(first.get("path")) or {}
    stamps = [_message_millis(path) for path in messages]
    known = [stamp for stamp in stamps if stamp is not None]
    first_prompt, last_prompt = _prompt_edges(root, session_id, messages)
    return Session(
        "opencode",
        session_id,
        str(message_dir),
        text(path_info.get("cwd")),
        unix_millis(min(known) if known else None),
        unix_millis(max(known) if known else None),
        text(model.get("providerID")),
        text(model.get("modelID")),
        first_prompt,
        {"message_count": len(messages)},
        last_user_message=last_prompt,
    )


def _prompt_edges(root: Path, session_id: str, messages: list[Path]) -> tuple[str, str]:
    first_prompt = ""
    last_prompt = ""
    for path in messages:
        data = as_map(read_json(path))
        if data is None or data.get("role") != "user":
            continue
        message_id = text(data.get("id"))
        for part in (root / "parts" / (message_id or "")).glob("*.json"):
            part_data = as_map(read_json(part))
            if part_data is not None and part_data.get("sessionID") == session_id:
                value = text(part_data.get("text"))
                if value:
                    first_prompt = first_prompt or value
                    last_prompt = value
    return first_prompt, last_prompt


def _message_millis(path: Path) -> int | None:
    data = as_map(read_json(path))
    time_data = as_map(data.get("time")) if data is not None else None
    return int_value(time_data.get("created")) if time_data is not None else None

from __future__ import annotations

from pathlib import Path

from .transcript import env_path, existing, jsonl_parallel, recent, stem_id
from .types import Session

SENPI_CONFIG_DIRS = (".omo", ".senpi", ".pi")
OH_MY_PI_CONFIG_DIRS = (".omp",)
GAJAE_CODE_CONFIG_DIRS = (".gjc",)

__all__ = ["scan_gajae_code", "scan_oh_my_pi", "scan_senpi"]


def scan_senpi(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    return _scan_pi_family("senpi", SENPI_CONFIG_DIRS, "senpi", extra_roots, workers)


def scan_oh_my_pi(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    return _scan_pi_family("oh-my-pi", OH_MY_PI_CONFIG_DIRS, "omp", extra_roots, workers)


def scan_gajae_code(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    return _scan_pi_family("gajae-code", GAJAE_CODE_CONFIG_DIRS, "gjc", extra_roots, workers)


def _scan_pi_family(platform: str, config_dirs: tuple[str, ...], xdg_app: str, extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _pi_family_roots(config_dirs, xdg_app, extra_roots)
    paths = [path for root in roots for path in (root / "sessions").rglob("*.jsonl")]
    return jsonl_parallel(recent(paths), workers, platform, lambda path: stem_id(path, "_"))


def _pi_family_roots(config_dirs: tuple[str, ...], xdg_app: str, extra_roots: tuple[Path, ...]) -> list[Path]:
    home = Path.home()
    roots: list[Path] = []
    for config_dir in config_dirs:
        base = home / config_dir
        roots.append(base / "agent")
        roots.extend(sorted(base.glob("profiles/*/agent")))
    xdg_data = env_path("XDG_DATA_HOME")
    if xdg_data is not None:
        app_root = xdg_data / xdg_app
        roots.append(app_root)
        roots.extend(sorted(app_root.glob("profiles/*")))
    roots.extend(extra_roots)
    return existing(roots)

from __future__ import annotations

from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Callable, TypeAlias

from .aside_scanner import scan_aside
from .claude import scan_claude
from .codex import scan_codex
from .file_scanners import (
    scan_aider,
    scan_amp,
    scan_cline,
    scan_codebuff,
    scan_droid,
    scan_gemini,
    scan_kilocode,
    scan_kimi,
    scan_openclaw,
    scan_qwen,
    scan_roocode,
)
from .kiro_scanner import scan_kiro
from .opencode import scan_opencode
from .pi_family import scan_gajae_code, scan_oh_my_pi, scan_senpi
from .sqlite_optional_scanners import scan_crush, scan_goose, scan_hermes, scan_kilo_cli, scan_zed
from .sqlite_scanners import scan_cursor_cli, scan_kodu
from .types import Session

Scanner: TypeAlias = Callable[[tuple[Path, ...], int], list[Session]]

PLATFORM_SCANNERS: dict[str, Scanner] = {
    "codex": scan_codex,
    "claude": scan_claude,
    "senpi": scan_senpi,
    "oh-my-pi": scan_oh_my_pi,
    "gajae-code": scan_gajae_code,
    "opencode": scan_opencode,
    "openclaw": scan_openclaw,
    "droid": scan_droid,
    "amp": scan_amp,
    "gemini": scan_gemini,
    "kimi": scan_kimi,
    "qwen": scan_qwen,
    "codebuff": scan_codebuff,
    "roo-code": scan_roocode,
    "kilo-code": scan_kilocode,
    "cline": scan_cline,
    "kodu": scan_kodu,
    "cursor-cli": scan_cursor_cli,
    "aider": scan_aider,
    "kilo-cli": scan_kilo_cli,
    "hermes": scan_hermes,
    "goose": scan_goose,
    "crush": scan_crush,
    "zed": scan_zed,
    "kiro": scan_kiro,
    "aside": scan_aside,
}
DEFAULT_PLATFORMS = frozenset(PLATFORM_SCANNERS)
PLATFORM_ALIASES = {
    "cursor": "cursor-cli",
    "factory": "droid",
    "roo": "roo-code",
    "roocode": "roo-code",
    "kilocode": "kilo-code",
    "kilo": "kilo-cli",
    "omp": "oh-my-pi",
    "ohmypi": "oh-my-pi",
    "oh_my_pi": "oh-my-pi",
    "gjc": "gajae-code",
    "gajae": "gajae-code",
    "gajaecode": "gajae-code",
    "aside-browser": "aside",
}

__all__ = ["DEFAULT_PLATFORMS", "PLATFORM_SCANNERS", "scan", "scan_claude", "scan_codex", "scan_gajae_code", "scan_oh_my_pi", "scan_opencode", "scan_senpi"]


def scan(platforms: frozenset[str], roots: tuple[Path, ...], workers: int) -> list[Session]:
    selected = frozenset(PLATFORM_ALIASES.get(platform, platform) for platform in platforms)
    tasks = [task for platform, task in PLATFORM_SCANNERS.items() if platform in selected]
    if not tasks:
        return []
    sessions: list[Session] = []
    with ThreadPoolExecutor(max_workers=min(workers, max(len(tasks), 1))) as pool:
        futures = [pool.submit(task, roots, workers) for task in tasks]
        for future in as_completed(futures):
            sessions.extend(future.result())
    return _dedupe(sessions)


def _dedupe(sessions: list[Session]) -> list[Session]:
    found: dict[tuple[str, str], Session] = {}
    for session in sessions:
        key = (session.platform, session.id)
        current = found.get(key)
        if current is None or _linkage_score(session) > _linkage_score(current):
            found[key] = session
    return list(found.values())


def _linkage_score(session: Session) -> int:
    return int(session.parent_id is not None) + int(session.agent is not None)

from __future__ import annotations

import sqlite3
from collections.abc import Iterable
from pathlib import Path

from .jsonio import as_map, parse_json_text, text
from .timeparse import file_time, unix_millis
from .transcript import content_text, existing, flat_parallel, recent
from .types import JsonMap, Session

SqlValue = str | int | float | bytes | None
SqlRow = tuple[SqlValue, ...]


def scan_kilo_cli(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    paths = existing([Path.home() / ".local" / "share" / "kilo" / "kilo.db", *(root / "kilo.db" for root in extra_roots)])
    return flat_parallel(recent(paths), workers, lambda path: _message_table_sessions(path, "kilo-cli", "message"))


def scan_hermes(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    paths = existing([Path.home() / ".hermes" / "state.db", *(root / "state.db" for root in extra_roots)])
    return flat_parallel(recent(paths), workers, lambda path: _message_table_sessions(path, "hermes", "messages"))


def scan_goose(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    defaults = [
        Path.home() / ".local" / "share" / "goose" / "sessions" / "sessions.db",
        Path.home() / "Library" / "Application Support" / "goose" / "sessions" / "sessions.db",
    ]
    paths = existing([*defaults, *(root / "sessions.db" for root in extra_roots)])
    return flat_parallel(recent(paths), workers, lambda path: _message_table_sessions(path, "goose", "messages"))


def scan_crush(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    paths = existing([Path.home() / ".local" / "share" / "crush" / "crush.db", *(root / "crush.db" for root in extra_roots)])
    return flat_parallel(recent(paths), workers, _crush_sessions)


def scan_zed(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    paths = existing([Path.home() / "Library" / "Application Support" / "Zed" / "threads" / "threads.db", *(root / "threads.db" for root in extra_roots)])
    return flat_parallel(recent(paths), workers, _zed_sessions)


def _message_table_sessions(path: Path, platform: str, table: str) -> list[Session]:
    try:
        with sqlite3.connect(path) as conn:
            rows = _fetch_all(conn, f"select session_id, role, data, created_at from {table} order by created_at")
    except sqlite3.Error:
        return []
    return _group_message_rows(platform, path, rows)


def _crush_sessions(path: Path) -> list[Session]:
    try:
        with sqlite3.connect(path) as conn:
            rows = _fetch_all(conn, "select session_id, role, parts, created_at from messages order by created_at")
    except sqlite3.Error:
        return []
    return _group_message_rows("crush", path, rows)


def _zed_sessions(path: Path) -> list[Session]:
    try:
        with sqlite3.connect(path) as conn:
            rows = _fetch_all(conn, "select id, data_type, data, updated_at from threads")
    except sqlite3.Error:
        return []
    sessions: list[Session] = []
    for row in rows:
        if _row_text(row, 1) != "json":
            continue
        data = _json_blob(_row_bytes(row, 2))
        messages = _json_messages(data)
        first_user, last_user = _message_edges(messages)
        if not first_user:
            continue
        sid = _row_text(row, 0) or path.stem
        sessions.append(Session("zed", sid, str(path), None, _row_text(row, 3) or file_time(path), _row_text(row, 3) or file_time(path), "zed.dev", _zed_model(data), first_user, {"message_count": len(messages)}, last_user_message=last_user))
    return sessions


def _group_message_rows(platform: str, path: Path, rows: Iterable[SqlRow]) -> list[Session]:
    grouped: dict[str, list[JsonMap]] = {}
    stamps: dict[str, str] = {}
    for row in rows:
        sid = _row_text(row, 0)
        if sid is None:
            continue
        message = _message_json(row)
        if message is None:
            continue
        if sid not in grouped:
            grouped[sid] = []
        grouped[sid].append(message)
        if sid not in stamps:
            stamps[sid] = _stamp(row, 3) or file_time(path) or ""
    sessions: list[Session] = []
    for sid, messages in grouped.items():
        first_user, last_user = _message_edges(messages)
        if not first_user:
            continue
        sessions.append(Session(platform, sid, str(path), None, stamps.get(sid) or file_time(path), file_time(path), _provider(messages), _model(messages), first_user, {"message_count": len(messages)}, last_user_message=last_user))
    return sessions


def _message_json(row: SqlRow) -> JsonMap | None:
    role = _row_text(row, 1)
    raw = _row_json(row, 2)
    if raw is None:
        return {"role": role or "", "content": _row_text(row, 2) or ""}
    if "role" not in raw:
        raw["role"] = role or text(raw.get("role")) or ""
    return raw


def _message_edges(messages: list[JsonMap]) -> tuple[str, str]:
    first_user = last_user = ""
    for message in messages:
        if text(message.get("role")) != "user":
            continue
        prompt = _content(message)
        if prompt:
            first_user = first_user or prompt
            last_user = prompt
    return first_user, last_user


def _content(message: JsonMap) -> str:
    for key in ("content", "parts", "text"):
        value = message.get(key)
        parsed = parse_json_text(value) if isinstance(value, str) else value
        result = content_text(parsed) or (value if isinstance(value, str) else "")
        if result:
            return result
    return ""


def _json_messages(data: JsonMap | None) -> list[JsonMap]:
    if data is None:
        return []
    for key in ("messages", "turns", "entries"):
        value = data.get(key)
        if isinstance(value, list):
            return [item for item in value if isinstance(item, dict)]
    return []


def _model(messages: list[JsonMap]) -> str | None:
    for message in messages:
        model = text(message.get("modelID")) or text(message.get("model_id")) or text(message.get("model"))
        if model:
            return model
    return None


def _provider(messages: list[JsonMap]) -> str | None:
    for message in messages:
        provider = text(message.get("providerID")) or text(message.get("provider_id")) or text(message.get("provider"))
        if provider:
            return provider
    return None


def _zed_model(data: JsonMap | None) -> str | None:
    model = as_map(data.get("model")) if data is not None else None
    return text(model.get("model")) if model is not None else None


def _fetch_all(conn: sqlite3.Connection, sql: str) -> list[SqlRow]:
    rows: list[SqlRow] = conn.execute(sql).fetchall()
    return rows


def _row_json(row: SqlRow, index: int) -> JsonMap | None:
    value = _row_text(row, index)
    return as_map(parse_json_text(value)) if value is not None else None


def _json_blob(value: bytes | None) -> JsonMap | None:
    if value is None:
        return None
    try:
        return as_map(parse_json_text(value.decode()))
    except UnicodeDecodeError:
        return None


def _row_text(row: SqlRow, index: int) -> str | None:
    value = row[index] if index < len(row) else None
    return text(value) if not isinstance(value, bytes) else None


def _row_bytes(row: SqlRow, index: int) -> bytes | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, bytes) else None


def _stamp(row: SqlRow, index: int) -> str | None:
    value = row[index] if index < len(row) else None
    if isinstance(value, int):
        return unix_millis(value) if value > 10_000_000_000 else None
    return value if isinstance(value, str) else None

from __future__ import annotations

import sqlite3
from pathlib import Path
from typing import TypeAlias

from .jsonio import as_map, parse_json_text, text
from .timeparse import file_time, unix_millis
from .transcript import content_text, env_path, existing, flat_parallel, recent
from .types import Json, JsonMap, Session

SqlValue: TypeAlias = str | int | float | bytes | None
SqlRow: TypeAlias = tuple[SqlValue, ...]


def scan_kodu(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    appdata = env_path("APPDATA")
    roots = _roots(
        [
            Path.home() / "Library" / "Application Support" / "Code" / "User" / "globalStorage" / "kodu-ai.claude-dev-experimental",
            Path.home() / ".config" / "Code" / "User" / "globalStorage" / "kodu-ai.claude-dev-experimental",
            *(path / "Code" / "User" / "globalStorage" / "kodu-ai.claude-dev-experimental" for path in (appdata,) if path is not None),
        ],
        extra_roots,
        ("kodu-ai.claude-dev-experimental",),
    )
    paths = [root / "db" / "Azad.db" for root in roots if (root / "db" / "Azad.db").exists()]
    return flat_parallel(recent(paths), workers, _kodu_db)


def scan_cursor_cli(extra_roots: tuple[Path, ...], workers: int) -> list[Session]:
    roots = _roots([Path.home() / ".cursor"], extra_roots, (".cursor",))
    paths = [path for root in roots for path in (root / "chats").glob("*/*/store.db")]
    return flat_parallel(recent(paths), workers, _cursor_db)


def _kodu_db(path: Path) -> list[Session]:
    try:
        with sqlite3.connect(path) as conn:
            tasks = _fetch_all(
                conn,
                "select id, name, dir_absolute_path, created_at, updated_at, tokens_in, tokens_out, cache_reads, cache_writes, cost from tasks"
            )
            return [_kodu_session(path, conn, row) for row in tasks]
    except sqlite3.Error:
        return []


def _kodu_session(path: Path, conn: sqlite3.Connection, row: SqlRow) -> Session:
    sid = _row_text(row, 0) or ""
    messages = _fetch_all(
        conn,
        "select role, content, model_id, started_at, finished_at, tokens_in, tokens_out, cache_reads, cache_writes, cost from messages where task_id = ? order by started_at",
        (sid,),
    )
    first_user = last_user = ""
    model = None
    for message in messages:
        model = model or _row_text(message, 2)
        if _row_text(message, 0) == "user":
            prompt = _content(_row_text(message, 1))
            first_user = first_user or prompt
            last_user = prompt
    return Session(
        "kodu",
        sid,
        str(path),
        _row_text(row, 2),
        unix_millis(_row_int(row, 3)),
        unix_millis(_row_int(row, 4)) or file_time(path),
        None,
        model,
        first_user,
        _usage(row, (5, "input"), (6, "output"), (7, "cacheRead"), (8, "cacheWrite"), (9, "cost_total")),
        last_user_message=last_user,
    )


def _cursor_db(path: Path) -> list[Session]:
    try:
        with sqlite3.connect(path) as conn:
            rows = _fetch_all(conn, "select data from blobs")
    except sqlite3.Error:
        return []
    first_user = last_user = ""
    for row in rows:
        data = _json_blob(_row_bytes(row, 0))
        if data is None or data.get("role") != "user":
            continue
        prompt = _cursor_prompt(_content(data.get("content")))
        if prompt:
            first_user = first_user or prompt
            last_user = prompt
    if not first_user:
        return []
    return [
        Session(
            "cursor-cli",
            path.parent.name,
            str(path),
            None,
            file_time(path),
            file_time(path),
            "cursor",
            None,
            first_user,
            {"blob_count": len(rows)},
            last_user_message=last_user,
        )
    ]


def _content(value: Json | None) -> str:
    parsed = parse_json_text(value) if isinstance(value, str) else value
    return content_text(parsed) or (value if isinstance(value, str) else "")


def _json_blob(value: bytes | None) -> JsonMap | None:
    if value is None:
        return None
    try:
        decoded = value.decode()
    except UnicodeDecodeError:
        return None
    return as_map(parse_json_text(decoded))


def _cursor_prompt(value: str) -> str:
    start = value.find("<user_query>")
    end = value.find("</user_query>")
    if start >= 0 and end > start:
        return value[start + len("<user_query>") : end].strip()
    if value.lstrip().startswith("<user_info>"):
        return ""
    return value.strip()


def _usage(row: SqlRow, *fields: tuple[int, str]) -> JsonMap:
    result: JsonMap = {}
    for index, key in fields:
        value = _row_number(row, index)
        if value is not None:
            result[key] = value
    return result


def _fetch_all(conn: sqlite3.Connection, sql: str, params: tuple[str, ...] = ()) -> list[SqlRow]:
    rows: list[SqlRow] = conn.execute(sql, params).fetchall()
    return rows


def _row_text(row: SqlRow, index: int) -> str | None:
    value = row[index] if index < len(row) else None
    return text(value) if not isinstance(value, bytes) else None


def _row_bytes(row: SqlRow, index: int) -> bytes | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, bytes) else None


def _row_number(row: SqlRow, index: int) -> int | float | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, int | float) else None


def _row_int(row: SqlRow, index: int) -> int | None:
    value = row[index] if index < len(row) else None
    return value if isinstance(value, int) else None


def _roots(defaults: list[Path], extra_roots: tuple[Path, ...], children: tuple[str, ...]) -> list[Path]:
    candidates = [*defaults]
    for root in extra_roots:
        candidates.append(root)
        candidates.extend(root / child for child in children)
    return existing(candidates)

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from pathlib import Path


def date_bound(value: str | None, end: bool = False) -> datetime | None:
    if value is None:
        return None
    text = value.strip().lower()
    now = datetime.now(timezone.utc)
    if text == "today":
        base = now.replace(hour=0, minute=0, second=0, microsecond=0)
    elif text == "yesterday":
        base = (now - timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
    elif text.endswith("d") and text[:-1].isdigit():
        base = (now - timedelta(days=int(text[:-1]))).replace(hour=0, minute=0, second=0, microsecond=0)
    else:
        parts = [int(part) for part in text.split("-")]
        base = datetime(parts[0], parts[1] if len(parts) > 1 else 1, parts[2] if len(parts) > 2 else 1, tzinfo=timezone.utc)
    if not end:
        return base
    if text in {"today", "yesterday"} or text.endswith("d") or len(text.split("-")) == 3:
        return base + timedelta(days=1)
    if len(text.split("-")) == 2:
        return datetime(base.year + (base.month // 12), (base.month % 12) + 1, 1, tzinfo=timezone.utc)
    return datetime(base.year + 1, 1, 1, tzinfo=timezone.utc)


def parse_stamp(value: str | None) -> datetime | None:
    if value is None:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)
    except ValueError:
        return None


def unix_seconds(value: int | float | None) -> str | None:
    return datetime.fromtimestamp(value, timezone.utc).isoformat() if value is not None else None


def unix_millis(value: int | None) -> str | None:
    return datetime.fromtimestamp(value / 1000, timezone.utc).isoformat() if value is not None else None


def file_time(path: Path) -> str | None:
    try:
        return datetime.fromtimestamp(path.stat().st_mtime, timezone.utc).isoformat()
    except OSError:
        return None

from __future__ import annotations

import os
from concurrent.futures import ThreadPoolExecutor, as_completed
from heapq import nlargest
from pathlib import Path
from typing import Callable

from .jsonio import as_map, iter_jsonl, parse_json_text, text
from .timeparse import file_time
from .types import Json, JsonMap, Session

MAX_PLATFORM_FILES = 2000


def jsonl_parallel(paths: list[Path], workers: int, platform: str, fallback_id: Callable[[Path], str]) -> list[Session]:
    if not paths:
        return []
    sessions: list[Session] = []
    with ThreadPoolExecutor(max_workers=min(workers, max(len(paths), 1))) as pool:
        futures = [pool.submit(jsonl_session, platform, path, fallback_id(path)) for path in paths]
        for future in as_completed(futures):
            sessions.append(future.result())
    return sessions


def flat_parallel(paths: list[Path], workers: int, read: Callable[[Path], list[Session]]) -> list[Session]:
    if not paths:
        return []
    sessions: list[Session] = []
    with ThreadPoolExecutor(max_workers=min(workers, max(len(paths), 1))) as pool:
        futures = [pool.submit(read, path) for path in paths]
        for future in as_completed(futures):
            sessions.extend(future.result())
    return sessions


def jsonl_session(platform: str, path: Path, fallback_id: str) -> Session:
    sid = fallback_id
    cwd = provider = model = first_user = parent = agent = None
    last_user = ""
    created = updated = None
    usage: JsonMap = {}
    for data in iter_jsonl(path):
        event_type = data.get("type")
        session_line_id = text(data.get("id")) if event_type == "session" else None
        sid = text(data.get("sessionId")) or session_line_id or sid
        cwd = cwd or text(data.get("cwd"))
        created = created or text(data.get("timestamp"))
        updated = text(data.get("timestamp")) or updated
        provider = provider or text(data.get("provider"))
        model = model or text(data.get("modelId")) or text(data.get("model"))
        payload = as_map(data.get("payload"))
        if event_type == "session_meta" and payload is not None:
            sid = text(payload.get("id")) or sid
            cwd = cwd or text(payload.get("cwd"))
            provider = provider or text(payload.get("model_provider"))
            source_parent, source_agent = spawn_info(payload.get("source"))
            parent = parent or source_parent
            agent = agent or source_agent or nick_role(text(payload.get("agent_nickname")), text(payload.get("agent_role")))
        message = as_map(data.get("message")) or payload or {}
        provider = provider or text(message.get("provider"))
        model = model or text(message.get("model"))
        prompt = user_text(data, message)
        if prompt:
            first_user = first_user or prompt
            last_user = prompt
        merge_usage(usage, as_map(message.get("usage")) or as_map(data.get("usage")))
    return Session(platform, sid, str(path), cwd, created or file_time(path), updated or created or file_time(path), provider, model, first_user or "", usage, parent, agent, last_user)


def spawn_info(source: Json | None) -> tuple[str | None, str | None]:
    data = as_map(parse_json_text(source) if isinstance(source, str) else source)
    if data is None:
        return None, None
    subagent = data.get("subagent")
    if isinstance(subagent, str):
        return None, subagent
    subagent_map = as_map(subagent)
    spawn = as_map(subagent_map.get("thread_spawn")) if subagent_map is not None else None
    if spawn is None:
        return None, None
    return text(spawn.get("parent_thread_id")), nick_role(text(spawn.get("agent_nickname")), text(spawn.get("agent_role")))


def nick_role(nickname: str | None, role: str | None) -> str | None:
    if nickname and role:
        return f"{nickname} ({role})"
    return nickname or role


def user_text(data: JsonMap, message: JsonMap) -> str:
    if data.get("type") == "user":
        value = content_text(data.get("content"))
        if value:
            return value
    if message.get("role") == "user":
        return content_text(message.get("content"))
    return ""


def content_text(value: Json | None) -> str:
    if isinstance(value, str):
        return value
    if isinstance(value, list):
        parts = [text(item.get("text")) or text(item.get("content")) or "" for item in value if isinstance(item, dict)]
        return "\n".join(part for part in parts if part)
    return ""


def merge_usage(target: JsonMap, value: JsonMap | None) -> None:
    if value is None:
        return
    for key in ("totalTokens", "total_tokens", "input", "output", "cacheRead", "cacheWrite"):
        if key in value:
            target[key] = value[key]
    cost = as_map(value.get("cost"))
    if cost is not None and "total" in cost:
        target["cost_total"] = cost["total"]


def existing(paths: list[Path]) -> list[Path]:
    seen: set[Path] = set()
    result: list[Path] = []
    for path in paths:
        if str(path) and path.exists() and path not in seen:
            seen.add(path)
            result.append(path)
    return result


def recent(paths: list[Path]) -> list[Path]:
    return nlargest(MAX_PLATFORM_FILES, paths, key=modified_time)


def modified_time(path: Path) -> float:
    try:
        return path.stat().st_mtime
    except OSError:
        return 0.0


def stem_id(path: Path, marker: str) -> str:
    return path.stem.split(marker)[-1].split("_")[-1]


def env_path(name: str) -> Path | None:
    value = os.environ.get(name)
    return Path(value).expanduser() if value else None

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import TypeAlias


Json: TypeAlias = str | int | float | bool | None | list["Json"] | dict[str, "Json"]
JsonMap: TypeAlias = dict[str, Json]


@dataclass(frozen=True, slots=True)
class Options:
    platforms: frozenset[str]
    roots: tuple[Path, ...]
    queries: tuple[str, ...]
    date_from: str | None
    date_to: str | None
    cwd: str | None
    model: str | None
    limit: int
    workers: int
    include_subagents: bool


@dataclass(frozen=True, slots=True)
class Session:
    platform: str
    id: str
    path: str
    cwd: str | None
    created_at: str | None
    updated_at: str | None
    provider: str | None
    model: str | None
    first_user_message: str
    usage: JsonMap
    parent_id: str | None = None
    agent: str | None = None
    last_user_message: str = ""

    def to_json(self) -> JsonMap:
        return {
            "platform": self.platform,
            "id": self.id,
            "path": self.path,
            "cwd": self.cwd,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
            "provider": self.provider,
            "model": self.model,
            "first_user_message": self.first_user_message[:300],
            "last_user_message": (self.last_user_message or self.first_user_message)[:300],
            "usage": self.usage,
            "parent_id": self.parent_id,
            "agent": self.agent,
        }

from __future__ import annotations

# /// script
# requires-python = ">=3.11"
# dependencies = ["pytest"]
# ///
# --- How to run ---
# uv run --with pytest pytest scripts/tests/test_agent_sessions.py -v
# pyright: reportImplicitRelativeImport=false, reportMissingImports=false
from __future__ import annotations

import json
import os
import sqlite3
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from agent_sessions import cli, scanners
from agent_sessions.opencode import scan_opencode
from agent_sessions.scanners import scan_claude, scan_codex
from agent_sessions.transcript import MAX_PLATFORM_FILES, recent
from agent_sessions.types import Json, JsonMap, Session


def _map(value: Json) -> JsonMap:
    assert isinstance(value, dict)
    return value


def _rows(payload: JsonMap, key: str) -> list[JsonMap]:
    value = payload[key]
    assert isinstance(value, list)
    return [_map(item) for item in value]


def _claude_line(session_id: str, content: str, agent_id: str | None = None) -> str:
    data: JsonMap = {
        "sessionId": session_id,
        "type": "user",
        "timestamp": "2026-06-10T06:56:20.048Z",
        "cwd": "/tmp/work",
        "message": {"role": "user", "content": content},
    }
    if agent_id is not None:
        data["agentId"] = agent_id
        data["isSidechain"] = True
    return json.dumps(data)


def _session(platform: str, sid: str, parent_id: str | None = None, agent: str | None = None, path: str = "/tmp/x.x") -> Session:
    return Session(platform, sid, path, "/tmp/work", "2026-06-10T00:00:00+00:00", None, None, None, "hello world", {}, parent_id, agent)


@pytest.fixture
def claude_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    project = tmp_path / ".claude" / "projects" / "-tmp-work"
    sub = project / "main-sid" / "subagents"
    wf = sub / "workflows" / "wf_1"
    wf.mkdir(parents=True)
    (project / "main-sid.jsonl").write_text(_claude_line("main-sid", "build the feature") + "\n")
    (sub / "agent-abc.jsonl").write_text(_claude_line("main-sid", "audit repos", "abc") + "\n")
    (sub / "agent-abc.meta.json").write_text(json.dumps({"agentType": "general-purpose", "description": "Security sweep", "toolUseId": "toolu_1"}))
    (wf / "agent-def.jsonl").write_text(_claude_line("main-sid", "verify finding", "def") + "\n")
    (wf / "agent-def.meta.json").write_text(json.dumps({"agentType": "Explore", "description": "Verify finding"}))
    (wf / "journal.jsonl").write_text(json.dumps({"type": "journal", "sessionId": "main-sid"}) + "\n")
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.setenv("APPDATA", "")
    return tmp_path


def test_claude_main_session_keeps_its_own_transcript_path(claude_home: Path) -> None:
    sessions = scan_claude((), 4)

    mains = [item for item in sessions if item.id == "main-sid"]
    assert len(mains) == 1, f"expected exactly one main-sid session, got {mains}"
    assert mains[0].path.endswith("main-sid.jsonl"), f"main session path hijacked: {mains[0].path}"
    assert mains[0].parent_id is None


def test_claude_subagent_transcripts_become_child_sessions(claude_home: Path) -> None:
    sessions = scan_claude((), 4)

    by_id = {item.id: item for item in sessions}
    assert "abc" in by_id, f"task subagent missing from {sorted(by_id)}"
    assert "def" in by_id, f"workflow subagent missing from {sorted(by_id)}"
    assert by_id["abc"].parent_id == "main-sid"
    assert by_id["def"].parent_id == "main-sid"
    assert by_id["abc"].agent == "general-purpose"
    assert "audit repos" in by_id["abc"].first_user_message
    assert "Security sweep" in by_id["abc"].first_user_message, "meta description must be searchable"


def test_claude_workflow_journal_is_not_a_session(claude_home: Path) -> None:
    sessions = scan_claude((), 4)

    assert [item for item in sessions if item.path.endswith("journal.jsonl")] == [], "journal.jsonl must not be scanned as a session"


CODEX_NEW_SCHEMA = (
    "CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, created_at INTEGER NOT NULL, "
    "updated_at INTEGER NOT NULL, source TEXT NOT NULL, model_provider TEXT NOT NULL, cwd TEXT NOT NULL, "
    "model TEXT, first_user_message TEXT NOT NULL DEFAULT '', tokens_used INTEGER NOT NULL DEFAULT 0, "
    "agent_nickname TEXT, agent_role TEXT)"
)
CODEX_OLD_SCHEMA = (
    "CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, created_at INTEGER NOT NULL, "
    "updated_at INTEGER NOT NULL, model_provider TEXT NOT NULL, cwd TEXT NOT NULL, "
    "model TEXT, first_user_message TEXT NOT NULL DEFAULT '', tokens_used INTEGER NOT NULL DEFAULT 0)"
)
SPAWN_SOURCE = json.dumps({"subagent": {"thread_spawn": {"parent_thread_id": "parent-1", "depth": 1, "agent_nickname": "Mencius", "agent_role": "worker"}}})


@pytest.fixture
def codex_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
    codex = tmp_path / ".codex"
    codex.mkdir(parents=True)
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.setenv("CODEX_HOME", str(codex))
    return codex


def test_codex_spawn_edges_link_child_to_parent(codex_home: Path) -> None:
    with sqlite3.connect(codex_home / "state_9.sqlite") as conn:
        conn.execute(CODEX_NEW_SCHEMA)
        conn.execute("CREATE TABLE thread_spawn_edges (parent_thread_id TEXT NOT NULL, child_thread_id TEXT NOT NULL PRIMARY KEY, status TEXT NOT NULL)")
        conn.execute("INSERT INTO threads VALUES ('parent-1', '/tmp/p.jsonl', 100, 200, 'cli', 'openai', '/tmp/work', 'gpt-5', 'do it', 9, NULL, NULL)")
        conn.execute(
            "INSERT INTO threads VALUES ('child-1', '/tmp/c.jsonl', 150, 210, ?, 'openai', '/tmp/work', 'gpt-5', 'sub task', 3, 'Mencius', 'worker')",
            (SPAWN_SOURCE,),
        )
        conn.execute("INSERT INTO thread_spawn_edges VALUES ('parent-1', 'child-1', 'closed')")

    sessions = {item.id: item for item in scan_codex((), 4)}

    assert sessions["parent-1"].parent_id is None
    assert sessions["child-1"].parent_id == "parent-1"
    assert sessions["child-1"].agent == "Mencius (worker)"


def test_codex_old_schema_still_lists_threads(codex_home: Path) -> None:
    with sqlite3.connect(codex_home / "state_5.sqlite") as conn:
        conn.execute(CODEX_OLD_SCHEMA)
        conn.execute("INSERT INTO threads VALUES ('old-1', '/tmp/o.jsonl', 100, 200, 'openai', '/tmp/work', 'gpt-5', 'legacy', 1)")

    sessions = {item.id: item for item in scan_codex((), 4)}

    assert "old-1" in sessions, "old-schema codex db must still be scanned"
    assert sessions["old-1"].parent_id is None


def test_codex_rollout_session_meta_recovers_id_and_parent(codex_home: Path) -> None:
    day = codex_home / "sessions" / "2026" / "06" / "01"
    day.mkdir(parents=True)
    meta: JsonMap = {
        "timestamp": "2026-06-01T00:00:00.000Z",
        "type": "session_meta",
        "payload": {
            "id": "child-9",
            "timestamp": "2026-06-01T00:00:00.000Z",
            "cwd": "/tmp/work",
            "model_provider": "openai",
            "source": {"subagent": {"thread_spawn": {"parent_thread_id": "parent-9", "depth": 1, "agent_nickname": "Tesla", "agent_role": "explorer"}}},
        },
    }
    user: JsonMap = {
        "timestamp": "2026-06-01T00:00:01.000Z",
        "type": "response_item",
        "payload": {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "spawned task prompt"}]},
    }
    (day / "rollout-2026-06-01T00-00-00-child-9.jsonl").write_text(json.dumps(meta) + "\n" + json.dumps(user) + "\n")

    sessions = {item.id: item for item in scan_codex((), 4)}

    assert "child-9" in sessions, f"payload id not recovered: {sorted(sessions)}"
    assert sessions["child-9"].parent_id == "parent-9"
    assert sessions["child-9"].agent == "Tesla (explorer)"
    assert sessions["child-9"].cwd == "/tmp/work"
    assert "spawned task prompt" in sessions["child-9"].first_user_message


OPENCODE_SCHEMA = (
    "CREATE TABLE session (id TEXT PRIMARY KEY, parent_id TEXT, directory TEXT NOT NULL, title TEXT NOT NULL, "
    "agent TEXT, model TEXT, cost REAL DEFAULT 0, tokens_input INTEGER DEFAULT 0, tokens_output INTEGER DEFAULT 0, "
    "tokens_reasoning INTEGER DEFAULT 0, tokens_cache_read INTEGER DEFAULT 0, tokens_cache_write INTEGER DEFAULT 0, "
    "time_created INTEGER NOT NULL, time_updated INTEGER NOT NULL, time_archived INTEGER)"
)


def test_opencode_db_children_carry_parent_and_agent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    db = tmp_path / "opencode.db"
    with sqlite3.connect(db) as conn:
        conn.execute(OPENCODE_SCHEMA)
        conn.execute(
            "INSERT INTO session (id, parent_id, directory, title, agent, model, time_created, time_updated) "
            "VALUES ('ses_main', NULL, '/tmp/work', 'main work', NULL, '{\"providerID\":\"anthropic\",\"id\":\"claude\"}', 1000, 2000)"
        )
        conn.execute(
            "INSERT INTO session (id, parent_id, directory, title, agent, model, time_created, time_updated) "
            "VALUES ('ses_child', 'ses_main', '/tmp/work', 'explore docs (@explore subagent)', 'explore', NULL, 1100, 1900)"
        )
    monkeypatch.setattr("agent_sessions.opencode._db_path", lambda: db)

    sessions = {item.id: item for item in scan_opencode((), 4)}

    assert "ses_child" in sessions, f"child sessions missing from db scan: {sorted(sessions)}"
    assert sessions["ses_child"].parent_id == "ses_main"
    assert sessions["ses_child"].agent == "explore"
    assert sessions["ses_main"].parent_id is None


def test_opencode_storage_fallback_reads_parent_id(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    root = tmp_path / "opencode"
    store = root / "storage" / "session" / "hash1"
    store.mkdir(parents=True)
    (store / "ses_main.json").write_text(json.dumps({"id": "ses_main", "title": "main", "directory": "/tmp/work", "time": {"created": 1000, "updated": 2000}}))
    (store / "ses_child.json").write_text(json.dumps({"id": "ses_child", "parentID": "ses_main", "title": "child task", "directory": "/tmp/work", "time": {"created": 1100, "updated": 1900}}))
    monkeypatch.setattr(Path, "home", lambda: tmp_path / "nohome")

    sessions = {item.id: item for item in scan_opencode((root,), 4)}

    assert "ses_child" in sessions, f"storage fallback missing children: {sorted(sessions)}"
    assert sessions["ses_child"].parent_id == "ses_main"
    assert sessions["ses_main"].parent_id is None


def test_empty_env_roots_do_not_scan_cwd(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.chdir(tmp_path)
    monkeypatch.setattr(Path, "home", lambda: tmp_path / "home")
    monkeypatch.setenv("APPDATA", "")
    monkeypatch.setenv("CODEX_HOME", "")
    monkeypatch.setenv("OPENCODE_HOME", "")
    monkeypatch.setattr("agent_sessions.opencode._db_path", lambda: None)
    monkeypatch.setattr("agent_sessions.opencode._opencode_json", lambda _args: [])

    with sqlite3.connect(tmp_path / "state_leak.sqlite") as conn:
        conn.execute(CODEX_OLD_SCHEMA)
        conn.execute("INSERT INTO threads VALUES ('cwd-codex', '/tmp/o.jsonl', 100, 200, 'openai', '/tmp/work', 'gpt-5', 'cwd leak', 1)")
    (tmp_path / "Claude" / "transcripts").mkdir(parents=True)
    (tmp_path / "Claude" / "transcripts" / "cwd-claude.jsonl").write_text(_claude_line("cwd-claude", "cwd leak") + "\n")
    store = tmp_path / "storage" / "session" / "hash1"
    store.mkdir(parents=True)
    (store / "ses_cwd.json").write_text(json.dumps({"id": "ses_cwd", "title": "cwd leak", "directory": "/tmp/work", "time": {"created": 1000, "updated": 2000}}))

    assert scan_codex((), 4) == []
    assert scan_claude((), 4) == []
    assert scan_opencode((), 4) == []


def test_recent_keeps_newest_platform_file_cap(tmp_path: Path) -> None:
    paths: list[Path] = []
    for index in range(MAX_PLATFORM_FILES + 3):
        path = tmp_path / f"session-{index}.jsonl"
        path.write_text("{}\n")
        os.utime(path, (index, index))
        paths.append(path)

    missing = tmp_path / "missing.jsonl"
    selected = recent([missing, *paths])

    assert len(selected) == MAX_PLATFORM_FILES
    assert [path.name for path in selected[:3]] == ["session-2002.jsonl", "session-2001.jsonl", "session-2000.jsonl"]
    assert "session-0.jsonl" not in {path.name for path in selected}
    assert missing not in selected


@pytest.fixture
def family() -> list[Session]:
    return [
        _session("opencode", "ses_main", path="/tmp/main.jsonl"),
        _session("opencode", "ses_child1", parent_id="ses_main", agent="explore"),
        _session("opencode", "ses_child2", parent_id="ses_main", agent="plan"),
    ]


def test_cli_list_hides_children_and_counts_them(family: list[Session]) -> None:
    payload = cli._list_payload(family, family, 10, include_subagents=False)

    results = _rows(payload, "results")
    ids = [item["id"] for item in results]
    assert ids == ["ses_main"], f"children leaked into default list: {ids}"
    assert results[0]["subagent_count"] == 2


def test_cli_list_include_subagents_keeps_children(family: list[Session]) -> None:
    payload = cli._list_payload(family, family, 10, include_subagents=True)

    assert {item["id"] for item in _rows(payload, "results") if isinstance(item["id"], str)} == {"ses_main", "ses_child1", "ses_child2"}


def test_cli_get_main_includes_children(family: list[Session]) -> None:
    payload = cli._get_payload(family, ["ses_main"])

    assert payload["count"] == 1
    result = _rows(payload, "results")[0]
    assert sorted(item["id"] for item in _rows(result, "subagents") if isinstance(item["id"], str)) == ["ses_child1", "ses_child2"]
    assert _map(result["session"])["subagent_count"] == 2


def test_cli_get_child_by_id_still_works(family: list[Session]) -> None:
    payload = cli._get_payload(family, ["ses_child1"])

    assert payload["count"] == 1
    assert _map(_rows(payload, "results")[0]["session"])["parent_id"] == "ses_main"


def test_dedupe_prefers_linked_session_regardless_of_order() -> None:
    bare = _session("codex", "t1")
    linked = _session("codex", "t1", parent_id="t0", agent="Tesla (explorer)")

    for ordering in ([bare, linked], [linked, bare]):
        result = scanners._dedupe(ordering)

        assert len(result) == 1
        assert result[0].parent_id == "t0", f"dedupe dropped spawn linkage for ordering starting with {ordering[0]}"


def test_cli_search_matches_agent_name(family: list[Session]) -> None:
    payload = cli._search_payload(family, family, ("explore",), 10, 2, include_subagents=True)

    assert "ses_child1" in {item["id"] for item in _rows(payload, "results") if isinstance(item["id"], str)}, "agent label must be searchable"

# /// script
# requires-python = ">=3.11"
# dependencies = ["pytest"]
# ///
# --- How to run ---
# uv run --with pytest pytest scripts/tests/test_aside_scanner.py -v
# pyright: reportImplicitRelativeImport=false, reportMissingImports=false
from __future__ import annotations

import json
import sqlite3
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from agent_sessions.aside_scanner import scan_aside
from agent_sessions.types import JsonMap, Session


def _write_jsonl(path: Path, rows: list[JsonMap]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    _ = path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")


def _write_state_db(path: Path, rows: list[tuple[str, str | None, str, str, str, int, int]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(path) as conn:
        _ = conn.execute("CREATE TABLE sessions (id TEXT PRIMARY KEY, parent_id TEXT, title TEXT NOT NULL, cwd TEXT NOT NULL, model TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL)")
        _ = conn.executemany("INSERT INTO sessions VALUES (?, ?, ?, ?, ?, ?, ?)", rows)


def _by_id(sessions: list[Session]) -> dict[str, Session]:
    return {item.id: item for item in sessions}


def test_scan_aside_reads_messages_and_state_db_index(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    # given: a main session, a state.db-linked child, and a hardlink-mirror agents dir
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    user = tmp_path / ".aside" / "u" / "0"
    model = json.dumps({"provider": "quotio", "modelId": "gpt-5.6-sol", "thinkingLevel": "medium"})
    _write_state_db(
        user / "state.db",
        [
            ("main1", None, "Ambassador research", "/tmp/aside-main", model, 1785295598, 1785377046),
            ("child1", "main1", "Check settings context", "/tmp/aside-main", model, 1785295600, 1785295700),
        ],
    )
    _write_jsonl(
        user / "sessions" / "2026-07-29_main1" / "messages.jsonl",
        [
            {"role": "system-message", "content": "skill docs", "timestamp": 1785295598327},
            {"role": "user", "content": "find the ambassador benefits", "timestamp": 1785295598204},
            {"role": "assistant", "content": [{"type": "text", "text": "on it"}], "provider": "apitopia", "model": "kimi-k3", "usage": {"input": 28019, "output": 406, "totalTokens": 28425, "cost": {"total": 0.090147}}, "timestamp": 1785295598449},
            {"role": "user", "content": "now summarize it", "timestamp": 1785295599000},
        ],
    )
    _write_jsonl(
        user / "sessions" / "2026-07-29_child1" / "messages.jsonl",
        [{"role": "user", "content": "verify the settings page", "timestamp": 1785295600000}],
    )
    _write_jsonl(
        user / "agents" / "main" / "sessions" / "2026-07-29_main1" / "messages.jsonl",
        [{"role": "user", "content": "mirror copy that must not be scanned", "timestamp": 1785295601000}],
    )

    # when
    sessions = _by_id(scan_aside((), 4))

    # then: state.db metadata wins, transcript supplies prompts and usage
    assert len(sessions) == 2
    main = sessions["main1"]
    assert main.platform == "aside"
    assert main.agent is None
    assert "/agents/" not in main.path
    assert main.first_user_message == "find the ambassador benefits"
    assert main.last_user_message == "now summarize it"
    assert main.cwd == "/tmp/aside-main"
    assert main.provider == "quotio"
    assert main.model == "gpt-5.6-sol"
    assert main.parent_id is None
    assert main.usage["totalTokens"] == 28425
    assert main.usage["cost_total"] == 0.090147
    assert (main.created_at or "").startswith("2026-07-29")

    child = sessions["child1"]
    assert child.parent_id == "main1"
    assert child.agent == "Check settings context"
    assert child.first_user_message == "verify the settings page"


def test_scan_aside_without_state_db_falls_back_to_transcript(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    # given: only messages.jsonl, no state.db
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    user = tmp_path / ".aside" / "u" / "1"
    _write_jsonl(
        user / "sessions" / "2026-07-30_solo1" / "messages.jsonl",
        [
            {"role": "user", "content": "bare prompt", "timestamp": 1785295598204},
            {"role": "assistant", "content": [{"type": "text", "text": "ok"}], "provider": "apitopia", "model": "kimi-k3", "timestamp": 1785295598449},
        ],
    )

    # when
    sessions = _by_id(scan_aside((), 4))

    # then
    solo = sessions["solo1"]
    assert solo.first_user_message == "bare prompt"
    assert solo.provider == "apitopia"
    assert solo.model == "kimi-k3"
    assert solo.cwd is None
    assert (solo.created_at or "").startswith("2026-07-29")


def test_scan_aside_accepts_root_pointing_at_user_dir(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    # given: an extra root that is itself a user dir (sessions/ directly inside)
    monkeypatch.setattr(Path, "home", lambda: tmp_path / "nowhere")
    user = tmp_path / "exported-aside-user"
    _write_jsonl(
        user / "sessions" / "2026-07-30_rooted1" / "messages.jsonl",
        [{"role": "user", "content": "rooted prompt", "timestamp": 1785295598204}],
    )

    # when
    sessions = _by_id(scan_aside((user,), 4))

    # then
    assert sessions["rooted1"].first_user_message == "rooted prompt"

# /// script
# requires-python = ">=3.11"
# dependencies = ["pytest"]
# ///
# --- How to run ---
# uv run --with pytest pytest scripts/tests/test_cli_contract.py -v
# pyright: reportImplicitRelativeImport=false, reportMissingImports=false
from __future__ import annotations

import json
import os
import sqlite3
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from agent_sessions.jsonio import as_map, parse_json_text
from agent_sessions.types import Json, JsonMap

SKILL_ROOT = Path(__file__).resolve().parents[2]


def _map(value: Json) -> JsonMap:
    assert isinstance(value, dict)
    return value


def _rows(payload: JsonMap, key: str) -> list[JsonMap]:
    value = payload[key]
    assert isinstance(value, list)
    return [_map(item) for item in value]


def _payload(text_value: str) -> JsonMap:
    value = as_map(parse_json_text(text_value))
    assert value is not None
    return value


def _run(root: Path, *args: str) -> JsonMap:
    env = os.environ.copy()
    env["APPDATA"] = str(root / "appdata")
    env["CODEX_HOME"] = str(root)
    env["HOME"] = str(root / "home")
    env["OPENCODE_HOME"] = str(root / "opencode-home")
    proc = subprocess.run(
        [sys.executable, "scripts/find-agent-sessions.py", *args, "--root", str(root)],
        cwd=SKILL_ROOT,
        env=env,
        check=True,
        capture_output=True,
        text=True,
    )
    return _payload(proc.stdout)


def _write_jsonl(path: Path, rows: list[JsonMap]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    _ = path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")


def _fixture_root(tmp_path: Path) -> Path:
    root = tmp_path / "agents"
    root.mkdir(parents=True)
    with sqlite3.connect(root / "state_test.sqlite") as conn:
        schema = (
            "CREATE TABLE threads (id TEXT PRIMARY KEY, rollout_path TEXT NOT NULL, created_at INTEGER NOT NULL, "
            + "updated_at INTEGER NOT NULL, source TEXT NOT NULL, model_provider TEXT NOT NULL, cwd TEXT NOT NULL, "
            + "model TEXT, first_user_message TEXT NOT NULL DEFAULT '', tokens_used INTEGER NOT NULL DEFAULT 0, "
            + "agent_nickname TEXT, agent_role TEXT)"
        )
        row = (
            "INSERT INTO threads VALUES ('codex-alpha', '/tmp/codex-alpha.jsonl', 100, 200, 'cli', 'openai', "
            + "'/tmp/work', 'gpt-5', 'alpha rollout fix', 9, NULL, NULL)"
        )
        _ = conn.execute(schema)
        _ = conn.execute(row)
    _write_jsonl(
        root / "transcripts" / "claude-beta.jsonl",
        [
            {"sessionId": "claude-beta", "type": "user", "timestamp": "2026-06-10T00:00:00Z", "cwd": "/tmp/work", "content": "unrelated"},
            {"sessionId": "claude-beta", "type": "user", "timestamp": "2026-06-10T00:00:03Z", "cwd": "/tmp/work", "content": "alpha review notes"},
        ],
    )
    return root


def test_find_searches_all_platforms_and_explains_matches(tmp_path: Path) -> None:
    payload = _run(_fixture_root(tmp_path), "find", "alpha", "--limit", "10")

    results = _rows(payload, "results")
    platforms: set[str] = set()
    for item in results:
        platform = item["platform"]
        assert isinstance(platform, str)
        platforms.add(platform)
    assert platforms == {"codex", "claude"}
    for item in results:
        reasons = _rows(item, "match_reasons")
        assert reasons, f"missing match reasons for {item}"
        assert reasons[0]["query"] == "alpha"
        assert reasons[0]["platform"] == item["platform"]
        assert isinstance(reasons[0]["snippet"], str) and "alpha" in reasons[0]["snippet"].lower()
        assert item["detail_hint"] == f"python3 scripts/find-agent-sessions.py read {item['id']} --platform {item['platform']}"


def test_platform_filter_narrows_find_results(tmp_path: Path) -> None:
    payload = _run(_fixture_root(tmp_path), "find", "alpha", "--platform", "codex")

    results = _rows(payload, "results")
    assert len(results) == 1
    assert results[0]["platform"] == "codex"


def test_read_summarizes_first_and_last_user_prompts(tmp_path: Path) -> None:
    payload = _run(_fixture_root(tmp_path), "read", "claude-beta", "--platform", "claude")

    result = _rows(payload, "results")[0]
    prompts = _map(result["prompts"])
    session = _map(result["session"])
    assert prompts["first_user_message"] == "unrelated"
    assert prompts["last_user_message"] == "alpha review notes"
    assert session["last_user_message"] == "alpha review notes"
    assert result["detail_hint"] == "python3 scripts/find-agent-sessions.py read claude-beta --platform claude"

# /// script
# requires-python = ">=3.11"
# dependencies = ["pytest"]
# ///
# --- How to run ---
# uv run --with pytest pytest scripts/tests/test_extended_scanners.py -v
# pyright: reportImplicitRelativeImport=false, reportMissingImports=false
from __future__ import annotations

import json
import sqlite3
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from agent_sessions import scanners
from agent_sessions.cli import _get_payload
from agent_sessions.file_scanners import _file_uri_path
from agent_sessions.sqlite_scanners import scan_kodu
from agent_sessions.types import JsonMap, Session


def _write_jsonl(path: Path, rows: list[JsonMap]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    _ = path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")


def _sessions_by_platform(items: list[Session]) -> dict[str, Session]:
    return {item.platform: item for item in items}


def test_default_platforms_are_canonical_transcript_sources() -> None:
    required = {
        "codex",
        "claude",
        "senpi",
        "oh-my-pi",
        "gajae-code",
        "opencode",
        "openclaw",
        "droid",
        "amp",
        "gemini",
        "kimi",
        "qwen",
        "codebuff",
        "roo-code",
        "kilo-code",
        "cline",
        "kodu",
        "cursor-cli",
        "aider",
        "kilo-cli",
        "hermes",
        "goose",
        "crush",
        "zed",
        "kiro",
        "aside",
    }
    forbidden = {"copilot", "mux", "antigravity", "synthetic", "cursor"}

    assert scanners.DEFAULT_PLATFORMS == required
    assert not forbidden & scanners.DEFAULT_PLATFORMS
    assert scanners.PLATFORM_ALIASES["roocode"] == "roo-code"
    assert scanners.PLATFORM_ALIASES["kilocode"] == "kilo-code"
    assert scanners.PLATFORM_ALIASES["kilo"] == "kilo-cli"


def test_extended_default_scanners_find_transcript_rich_stores(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.setenv("APPDATA", str(tmp_path / "appdata"))

    _write_jsonl(
        tmp_path / ".openclaw" / "agents" / "main" / "sessions" / "openclaw-1.jsonl",
        [
            {"type": "session", "id": "openclaw-1", "timestamp": "2026-06-10T00:00:00Z", "cwd": "/tmp/openclaw"},
            {"type": "message", "timestamp": "2026-06-10T00:00:01Z", "message": {"role": "user", "content": [{"type": "text", "text": "openclaw build"}]}},
        ],
    )
    _write_jsonl(
        tmp_path / ".factory" / "sessions" / "-tmp-droid" / "droid-1.jsonl",
        [
            {"type": "session_start", "id": "droid-1", "timestamp": "2026-06-10T00:00:00Z", "cwd": "/tmp/droid"},
            {"type": "message", "timestamp": "2026-06-10T00:00:01Z", "message": {"role": "user", "content": [{"type": "text", "text": "<system-reminder>noise</system-reminder>"}, {"type": "text", "text": "droid real prompt"}]}},
        ],
    )
    amp = tmp_path / ".local" / "share" / "amp" / "threads"
    amp.mkdir(parents=True)
    (amp / "T-amp-1.json").write_text(
        json.dumps(
            {
                "id": "T-amp-1",
                "created": 1771748797025,
                "env": {"initial": {"trees": [{"uri": "file:///tmp/amp"}], "tags": ["model:claude-opus-4-6"]}},
                "messages": [{"role": "user", "content": [{"type": "text", "text": "amp prompt"}], "meta": {"sentAt": 1771748810692}}],
            }
        )
    )

    sessions = _sessions_by_platform(scanners.scan(scanners.DEFAULT_PLATFORMS, (), 4))

    assert sessions["openclaw"].first_user_message == "openclaw build"
    assert sessions["droid"].first_user_message == "droid real prompt"
    assert sessions["amp"].first_user_message == "amp prompt"


def test_sqlite_and_repo_scanners_use_bounded_default_roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.setenv("APPDATA", str(tmp_path / "appdata"))

    kodu_dir = tmp_path / "Library" / "Application Support" / "Code" / "User" / "globalStorage" / "kodu-ai.claude-dev-experimental" / "db"
    kodu_dir.mkdir(parents=True)
    with sqlite3.connect(kodu_dir / "Azad.db") as conn:
        conn.execute("CREATE TABLE tasks (id TEXT PRIMARY KEY, created_at INTEGER, updated_at INTEGER, name TEXT, dir_absolute_path TEXT, tokens_in INTEGER, tokens_out INTEGER, cache_writes INTEGER, cache_reads INTEGER, cost INTEGER)")
        conn.execute("CREATE TABLE messages (id TEXT PRIMARY KEY, task_id TEXT, role TEXT, content TEXT, model_id TEXT, started_at INTEGER, finished_at INTEGER, tokens_in INTEGER, tokens_out INTEGER, cache_writes INTEGER, cache_reads INTEGER, cost INTEGER)")
        conn.execute("INSERT INTO tasks VALUES ('kodu-task', 1000, 3000, 'kodu title', '/tmp/kodu', 1, 2, 3, 4, 5)")
        conn.execute("INSERT INTO messages VALUES ('m1', 'kodu-task', 'user', '[{\"type\":\"text\",\"text\":\"kodu prompt\"}]', 'claude', 1100, 1200, 1, 0, 0, 0, 0)")
        conn.execute("INSERT INTO messages VALUES ('m2', 'kodu-task', 'assistant', 'answer', 'claude', 1300, 1400, 0, 2, 0, 0, 5)")

    cursor_dir = tmp_path / ".cursor" / "chats" / "hash" / "cursor-session"
    cursor_dir.mkdir(parents=True)
    with sqlite3.connect(cursor_dir / "store.db") as conn:
        conn.execute("CREATE TABLE blobs (id TEXT PRIMARY KEY, data BLOB)")
        conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)")
        conn.execute("INSERT INTO blobs VALUES ('b1', ?)", (json.dumps({"role": "user", "content": "<user_info>metadata</user_info>"}).encode(),))
        conn.execute("INSERT INTO blobs VALUES ('b2', ?)", (json.dumps({"role": "user", "content": [{"type": "text", "text": "<user_query>\ncursor prompt\n</user_query>"}]}).encode(),))
    (tmp_path / ".cursor" / "prompt_history.json").write_text(json.dumps(["cursor prompt"]))

    aider = tmp_path / "local-workspaces" / "repo"
    aider.mkdir(parents=True)
    (aider / ".aider.chat.history.md").write_text("# aider chat started at 2026-06-10 00:00:00\n\n#### aider prompt\n\nassistant reply\n")

    kiro = tmp_path / ".kiro" / "sessions" / "cli"
    kiro.mkdir(parents=True)
    (kiro / "kiro-session.json").write_text(
        json.dumps(
            {
                "session_id": "kiro-session",
                "cwd": "/tmp/kiro",
                "session_state": {
                    "rts_model_state": {"model_info": {"model_id": "claude-sonnet-4-5"}},
                    "conversation_metadata": {"user_turn_metadatas": [{"message_ids": ["prompt-1", "assistant-1"]}]},
                },
            }
        )
    )
    _write_jsonl(
        kiro / "kiro-session.jsonl",
        [
            {"version": "v1", "kind": "Prompt", "data": {"message_id": "prompt-1", "content": [{"kind": "text", "data": "kiro prompt"}], "meta": {"timestamp": 1770983426.42}}},
            {"version": "v1", "kind": "AssistantMessage", "data": {"message_id": "assistant-1", "content": [{"kind": "text", "data": "kiro answer"}]}},
        ],
    )

    sessions = {(item.platform, item.id): item for item in scanners.scan(scanners.DEFAULT_PLATFORMS, (), 4)}

    assert sessions[("kodu", "kodu-task")].first_user_message == "kodu prompt"
    assert sessions[("cursor-cli", "cursor-session")].first_user_message == "cursor prompt"
    assert sessions[("aider", "repo-2026-06-10-00-00-00")].first_user_message == "aider prompt"
    assert sessions[("kiro", "kiro-session")].first_user_message == "kiro prompt"
    assert sessions[("kiro", "kiro-session")].model == "claude-sonnet-4-5"


def test_vscode_extension_scanners_include_windows_appdata_roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path / "home")
    monkeypatch.setenv("APPDATA", str(tmp_path / "AppData" / "Roaming"))

    roo_history = (
        tmp_path
        / "AppData"
        / "Roaming"
        / "Code"
        / "User"
        / "globalStorage"
        / "rooveterinaryinc.roo-cline"
        / "tasks"
        / "roo-task"
        / "api_conversation_history.json"
    )
    roo_history.parent.mkdir(parents=True)
    roo_history.write_text(json.dumps([{"role": "user", "content": [{"type": "text", "text": "windows roo prompt"}]}]))

    kodu_dir = (
        tmp_path
        / "AppData"
        / "Roaming"
        / "Code"
        / "User"
        / "globalStorage"
        / "kodu-ai.claude-dev-experimental"
        / "db"
    )
    kodu_dir.mkdir(parents=True)
    with sqlite3.connect(kodu_dir / "Azad.db") as conn:
        conn.execute("CREATE TABLE tasks (id TEXT PRIMARY KEY, created_at INTEGER, updated_at INTEGER, name TEXT, dir_absolute_path TEXT, tokens_in INTEGER, tokens_out INTEGER, cache_writes INTEGER, cache_reads INTEGER, cost INTEGER)")
        conn.execute("CREATE TABLE messages (id TEXT PRIMARY KEY, task_id TEXT, role TEXT, content TEXT, model_id TEXT, started_at INTEGER, finished_at INTEGER, tokens_in INTEGER, tokens_out INTEGER, cache_writes INTEGER, cache_reads INTEGER, cost INTEGER)")
        conn.execute("INSERT INTO tasks VALUES ('kodu-windows', 1000, 3000, 'kodu title', 'C:/repo', 1, 2, 3, 4, 5)")
        conn.execute("INSERT INTO messages VALUES ('m1', 'kodu-windows', 'user', '[{\"type\":\"text\",\"text\":\"windows kodu prompt\"}]', 'claude', 1100, 1200, 1, 0, 0, 0, 0)")

    roo_sessions = scanners.scan(frozenset({"roo-code"}), (), 4)
    kodu_sessions = scan_kodu((), 4)

    assert [item.first_user_message for item in roo_sessions] == ["windows roo prompt"]
    assert [item.first_user_message for item in kodu_sessions] == ["windows kodu prompt"]


def test_file_uri_path_preserves_windows_drive_and_unc_paths() -> None:
    assert _file_uri_path("file:///tmp/amp") == "/tmp/amp"
    assert _file_uri_path("file:///C:/Users/yeongyu/project") == "C:/Users/yeongyu/project"
    assert _file_uri_path("file://server/share/project") == "//server/share/project"


def test_kiro_scanner_skips_metadata_only_sessions(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)

    kiro = tmp_path / ".kiro" / "sessions" / "cli"
    kiro.mkdir(parents=True)
    (kiro / "metadata-only.json").write_text(
        json.dumps(
            {
                "session_id": "metadata-only",
                "cwd": "/tmp/kiro",
                "session_state": {
                    "rts_model_state": {"model_info": {"model_id": "claude-sonnet-4-5"}},
                    "conversation_metadata": {"user_turn_metadatas": [{"message_ids": ["prompt-1"]}]},
                },
            }
        )
    )

    sessions = scanners.scan(frozenset({"kiro"}), (), 4)

    assert ("kiro", "metadata-only") not in {(item.platform, item.id) for item in sessions}


def test_get_payload_reconstructs_events_for_non_jsonl_sessions() -> None:
    session = Session(
        "amp",
        "T-amp-1",
        "/tmp/T-amp-1.json",
        "/tmp/amp",
        "2026-06-10T00:00:00+00:00",
        "2026-06-10T00:00:01+00:00",
        None,
        "claude-opus-4-6",
        "first prompt",
        {},
        last_user_message="last prompt",
    )

    payload = _get_payload([session], ["T-amp-1"])

    result = payload["results"][0]
    assert isinstance(result, dict)
    assert result["events"] == [
        {"type": "message", "message": {"role": "user", "content": "first prompt"}},
        {"type": "message", "message": {"role": "user", "content": "last prompt"}},
    ]

# /// script
# requires-python = ">=3.11"
# dependencies = ["pytest"]
# ///
# --- How to run ---
# uv run --with pytest pytest scripts/tests/test_optional_sqlite_scanners.py -v
# pyright: reportImplicitRelativeImport=false, reportMissingImports=false
from __future__ import annotations

import json
import sqlite3
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from agent_sessions.sqlite_optional_scanners import scan_crush, scan_goose, scan_hermes, scan_kilo_cli, scan_zed


def _message_db(path: Path, table: str, prompt: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(path) as conn:
        _ = conn.execute(f"CREATE TABLE {table} (session_id TEXT, role TEXT, data TEXT, created_at INTEGER)")
        _ = conn.execute(
            f"INSERT INTO {table} VALUES ('s1', 'user', ?, 1770983426420)",
            (json.dumps({"role": "user", "content": [{"type": "text", "text": prompt}]}),),
        )
        _ = conn.execute(
            f"INSERT INTO {table} VALUES ('s1', 'assistant', ?, 1770983427420)",
            (json.dumps({"role": "assistant", "modelID": "claude-sonnet-4-5", "providerID": "anthropic", "content": "answer"}),),
        )


def test_optional_sqlite_scanners_reconstruct_present_prompt_stores(tmp_path: Path) -> None:
    _message_db(tmp_path / "kilo" / "kilo.db", "message", "kilo prompt")
    _message_db(tmp_path / "hermes" / "state.db", "messages", "hermes prompt")
    _message_db(tmp_path / "goose" / "sessions.db", "messages", "goose prompt")

    (tmp_path / "crush").mkdir()
    with sqlite3.connect(tmp_path / "crush" / "crush.db") as conn:
        _ = conn.execute("CREATE TABLE messages (session_id TEXT, role TEXT, parts TEXT, created_at INTEGER)")
        _ = conn.execute("INSERT INTO messages VALUES ('s1', 'user', ?, 1770983426420)", (json.dumps([{"type": "text", "text": "crush prompt"}]),))

    (tmp_path / "zed").mkdir()
    with sqlite3.connect(tmp_path / "zed" / "threads.db") as conn:
        _ = conn.execute("CREATE TABLE threads (id TEXT, data_type TEXT, data BLOB, updated_at TEXT)")
        _ = conn.execute(
            "INSERT INTO threads VALUES ('t1', 'json', ?, '2026-06-10T00:00:00Z')",
            (json.dumps({"model": {"provider": "zed.dev", "model": "claude"}, "messages": [{"role": "user", "content": "zed prompt"}]}).encode(),),
        )

    sessions = [
        *scan_kilo_cli((tmp_path / "kilo",), 4),
        *scan_hermes((tmp_path / "hermes",), 4),
        *scan_goose((tmp_path / "goose",), 4),
        *scan_crush((tmp_path / "crush",), 4),
        *scan_zed((tmp_path / "zed",), 4),
    ]
    prompts = {item.platform: item.first_user_message for item in sessions}

    assert prompts == {
        "kilo-cli": "kilo prompt",
        "hermes": "hermes prompt",
        "goose": "goose prompt",
        "crush": "crush prompt",
        "zed": "zed prompt",
    }


def test_optional_sqlite_scanners_skip_unsupported_schemas(tmp_path: Path) -> None:
    for name in ("kilo.db", "state.db", "sessions.db", "crush.db", "threads.db"):
        with sqlite3.connect(tmp_path / name) as conn:
            _ = conn.execute("CREATE TABLE unrelated (id TEXT)")

    assert scan_kilo_cli((tmp_path,), 4) == []
    assert scan_hermes((tmp_path,), 4) == []
    assert scan_goose((tmp_path,), 4) == []
    assert scan_crush((tmp_path,), 4) == []
    assert scan_zed((tmp_path,), 4) == []

# /// script
# requires-python = ">=3.11"
# dependencies = ["pytest"]
# ///
# --- How to run ---
# uv run --with pytest pytest scripts/tests/test_pi_family_scanners.py -v
# pyright: reportImplicitRelativeImport=false, reportMissingImports=false
from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from agent_sessions import scanners
from agent_sessions.types import JsonMap, Session

PI_FAMILY = frozenset({"senpi", "oh-my-pi", "gajae-code"})


def _write_session(path: Path, session_id: str, cwd: str, prompt: str, agent: str | None = None) -> None:
    rows: list[JsonMap] = [
        {"type": "session", "version": 3, "id": session_id, "timestamp": "2026-07-20T00:00:00.000Z", "cwd": cwd},
        {"type": "model_change", "id": "m1", "timestamp": "2026-07-20T00:00:00.100Z", "model": "anthropic/claude-fable-5"},
        {
            "type": "message",
            "id": "e1",
            "timestamp": "2026-07-20T00:00:01.000Z",
            "message": {"role": "user", "content": [{"type": "text", "text": prompt}]},
        },
    ]
    if agent is not None:
        rows.insert(
            1,
            {
                "type": "session_meta",
                "timestamp": "2026-07-20T00:00:00.050Z",
                "payload": {"id": session_id, "source": {"subagent": agent}},
            },
        )
    path.parent.mkdir(parents=True, exist_ok=True)
    _ = path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")


def _by_platform(items: list[Session]) -> dict[str, Session]:
    return {item.platform: item for item in items}


def test_pi_family_platforms_and_aliases_are_registered() -> None:
    assert PI_FAMILY <= scanners.DEFAULT_PLATFORMS
    assert scanners.PLATFORM_ALIASES["omp"] == "oh-my-pi"
    assert scanners.PLATFORM_ALIASES["ohmypi"] == "oh-my-pi"
    assert scanners.PLATFORM_ALIASES["gjc"] == "gajae-code"
    assert scanners.PLATFORM_ALIASES["gajae"] == "gajae-code"


def test_pi_family_scanners_read_each_product_home_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.delenv("XDG_DATA_HOME", raising=False)

    _write_session(
        tmp_path / ".senpi" / "agent" / "sessions" / "-tmp-senpi" / "2026-07-20T00-00-00-000Z_senpi-1.jsonl",
        "senpi-1",
        "/tmp/senpi",
        "senpi prompt",
    )
    _write_session(
        tmp_path / ".omp" / "agent" / "sessions" / "-tmp-omp" / "2026-07-20T00-00-00-000Z_omp-1.jsonl",
        "omp-1",
        "/tmp/omp",
        "oh-my-pi prompt",
    )
    _write_session(
        tmp_path / ".gjc" / "agent" / "sessions" / "-tmp-gjc" / "2026-07-20T00-00-00-000Z_gjc-1.jsonl",
        "gjc-1",
        "/tmp/gjc",
        "gajae-code prompt",
    )

    sessions = _by_platform(scanners.scan(PI_FAMILY, (), 4))

    assert sessions["senpi"].first_user_message == "senpi prompt"
    assert sessions["oh-my-pi"].id == "omp-1"
    assert sessions["oh-my-pi"].first_user_message == "oh-my-pi prompt"
    assert sessions["oh-my-pi"].cwd == "/tmp/omp"
    assert sessions["oh-my-pi"].model == "anthropic/claude-fable-5"
    assert sessions["gajae-code"].id == "gjc-1"
    assert sessions["gajae-code"].first_user_message == "gajae-code prompt"


def test_senpi_scanner_reads_omo_and_legacy_home_stores(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.delenv("XDG_DATA_HOME", raising=False)

    _write_session(
        tmp_path / ".omo" / "agent" / "sessions" / "-tmp-omo" / "2026-07-20T00-00-00-000Z_omo-current.jsonl",
        "omo-current",
        "/tmp/omo",
        "omo current prompt",
    )
    _write_session(
        tmp_path / ".senpi" / "agent" / "sessions" / "-tmp-senpi" / "2026-07-20T00-00-00-000Z_senpi-legacy.jsonl",
        "senpi-legacy",
        "/tmp/senpi",
        "senpi legacy prompt",
    )

    sessions = {item.id: item for item in scanners.scan(frozenset({"senpi"}), (), 4)}

    assert set(sessions) == {"omo-current", "senpi-legacy"}
    assert sessions["omo-current"].platform == "senpi"
    assert sessions["omo-current"].cwd == "/tmp/omo"
    assert sessions["senpi-legacy"].first_user_message == "senpi legacy prompt"


def test_senpi_scanner_deduplicates_overlapping_omo_and_legacy_sessions(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.delenv("XDG_DATA_HOME", raising=False)

    _write_session(
        tmp_path / ".omo" / "agent" / "sessions" / "-tmp-overlap" / "2026-07-20T00-00-00-000Z_shared.jsonl",
        "shared",
        "/tmp/omo",
        "omo linked prompt",
        agent="sisyphus",
    )
    _write_session(
        tmp_path / ".senpi" / "agent" / "sessions" / "-tmp-overlap" / "2026-07-20T00-00-00-000Z_shared.jsonl",
        "shared",
        "/tmp/senpi",
        "legacy prompt",
    )

    sessions = scanners.scan(frozenset({"senpi"}), (), 4)

    assert len(sessions) == 1
    assert sessions[0].id == "shared"
    assert sessions[0].agent == "sisyphus"
    assert sessions[0].cwd == "/tmp/omo"


def test_pi_family_scanners_cover_profile_and_xdg_roots(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "xdg-data"))

    _write_session(
        tmp_path / ".omp" / "profiles" / "work" / "agent" / "sessions" / "-repo" / "2026-07-20T00-00-00-000Z_omp-work.jsonl",
        "omp-work",
        "/repo",
        "omp profile prompt",
    )
    _write_session(
        tmp_path / "xdg-data" / "gjc" / "sessions" / "-repo" / "2026-07-20T00-00-00-000Z_gjc-xdg.jsonl",
        "gjc-xdg",
        "/repo",
        "gjc xdg prompt",
    )
    _write_session(
        tmp_path / "xdg-data" / "omp" / "profiles" / "work" / "sessions" / "-repo" / "2026-07-20T00-00-00-000Z_omp-xdg.jsonl",
        "omp-xdg",
        "/repo",
        "omp xdg profile prompt",
    )

    found = {(item.platform, item.id) for item in scanners.scan(PI_FAMILY, (), 4)}

    assert ("oh-my-pi", "omp-work") in found
    assert ("gajae-code", "gjc-xdg") in found
    assert ("oh-my-pi", "omp-xdg") in found


def test_pi_family_stores_do_not_leak_across_platform_keys(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(Path, "home", lambda: tmp_path)
    monkeypatch.delenv("XDG_DATA_HOME", raising=False)

    _write_session(
        tmp_path / ".omp" / "agent" / "sessions" / "-tmp-omp" / "2026-07-20T00-00-00-000Z_omp-only.jsonl",
        "omp-only",
        "/tmp/omp",
        "oh-my-pi prompt",
    )

    senpi_sessions = scanners.scan(frozenset({"senpi"}), (), 4)
    omp_sessions = scanners.scan(frozenset({"omp"}), (), 4)

    assert senpi_sessions == []
    assert [(item.platform, item.id) for item in omp_sessions] == [("oh-my-pi", "omp-only")]