/**
 * Index engine: SQLite schema (WAL, external-content trigram FTS5), capped
 * incremental sync, and discovery search with one-hop lineage suppression.
 * No pi runtime imports — pure Node + node:sqlite so it is testable headless.
 */
import { DatabaseSync } from "node:sqlite";
import type { SQLOutputValue } from "node:sqlite";
import fs from "node:fs";
import path from "node:path";
import { buildFtsQueryPlan, buildLikeQueryPlan, foldCase, nearLike } from "./query.ts";
import { MAX_SESSION_FILE_BYTES, readTranscriptEntries } from "./transcript.ts";
import type { PreparationSessionRow, SearchHit, SessionRow, SyncResult } from "./types.ts";
export const DEFAULT_SYNC_CAP = 50;
const MAX_TEXT_CHARS = 20000;

const SCHEMA_SQL = `
-- ctime_ms is raw stat.ctimeMs stored as REAL (SQLite REALs are IEEE doubles,
-- so JS numbers round-trip losslessly): flooring would collapse sub-ms restores.
CREATE TABLE IF NOT EXISTS session_files (
  path TEXT PRIMARY KEY,
  size INTEGER NOT NULL,
  mtime_ms INTEGER NOT NULL,
  ctime_ms REAL NOT NULL
);
-- Retry markers affect ordering only: unchanged failures move behind fresh work,
-- but remain eligible on every pass once fresh work is drained.
CREATE TABLE IF NOT EXISTS session_failures (
  path TEXT PRIMARY KEY,
  size INTEGER NOT NULL,
  mtime_ms INTEGER NOT NULL,
  ctime_ms REAL NOT NULL,
  attempts INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS sessions (
  path TEXT PRIMARY KEY,
  cwd TEXT,
  name TEXT,
  started_at TEXT,
  preview TEXT,
  parent_session TEXT
);
-- Oversized messages split into head/tail columns: one row per source
-- message keeps AND matching across both retained regions (FTS5 implicit AND
-- spans columns) while phrases and NEAR cannot cross column boundaries, so
-- truncation never manufactures proximity matches over the elided middle.
CREATE TABLE IF NOT EXISTS messages (
  rowid INTEGER PRIMARY KEY,
  path TEXT NOT NULL,
  entry_id TEXT NOT NULL,
  role TEXT NOT NULL,
  timestamp TEXT,
  head TEXT NOT NULL,
  tail TEXT NOT NULL,
  UNIQUE(path, entry_id)
);
CREATE INDEX IF NOT EXISTS messages_path ON messages(path);
CREATE VIRTUAL TABLE IF NOT EXISTS session_fts USING fts5(
  head,
  tail,
  content='messages',
  content_rowid='rowid',
  tokenize='trigram'
);
CREATE TRIGGER IF NOT EXISTS messages_ai AFTER INSERT ON messages BEGIN
  INSERT INTO session_fts(rowid, head, tail) VALUES (new.rowid, new.head, new.tail);
END;
CREATE TRIGGER IF NOT EXISTS messages_ad AFTER DELETE ON messages BEGIN
  INSERT INTO session_fts(session_fts, rowid, head, tail) VALUES ('delete', old.rowid, old.head, old.tail);
END;
CREATE TRIGGER IF NOT EXISTS messages_au AFTER UPDATE ON messages BEGIN
  INSERT INTO session_fts(session_fts, rowid, head, tail) VALUES ('delete', old.rowid, old.head, old.tail);
  INSERT INTO session_fts(rowid, head, tail) VALUES (new.rowid, new.head, new.tail);
END;
CREATE TABLE IF NOT EXISTS meta (
  key TEXT PRIMARY KEY,
  value TEXT NOT NULL
);
`;

/** Owner-only modes for the index directory and database files. The
 *  extension-owned parent dir is the only node tightened; intermediate config
 *  dirs are created restrictive but never chmod'ed if they predate this fix. */
const DB_FILE_MODE = 0o600;
const DB_DIR_MODE = 0o700;

/** Create or tighten the index's private directory and owner-only database
 *  file BEFORE SQLite writes any transcript content into it. Descriptor-based
 *  O_NOFOLLOW|O_NONBLOCK open + fstat rejects symlinks and other non-regular
 *  nodes instead of following them; fchmod tightens an existing permissive db.
 *  WAL/SHM sidecars inherit the main db mode from SQLite; legacy sidecars left
 *  behind by older builds are tightened here too. Always closes the fd. */
function secureIndexNode(dbPath: string): void {
	const posix = process.platform !== "win32";
	const dir = path.dirname(dbPath);
	fs.mkdirSync(dir, { recursive: true, mode: DB_DIR_MODE });
	// O_NOFOLLOW on the db path does not protect parent components; reject a
	// symlinked extension directory before any chmod or database open follows it.
	if (!fs.lstatSync(dir).isDirectory()) throw new Error(`index directory is not a real directory: ${dir}`);
	if (posix) {
		// Tighten an existing permissive dir (e.g. created under umask 022).
		if ((fs.statSync(dir).mode & 0o777) !== DB_DIR_MODE) {
			fs.chmodSync(dir, DB_DIR_MODE);
		}
	}
	let fd: number | undefined;
	try {
		fd = fs.openSync(
			dbPath,
			(fs.constants.O_RDWR | fs.constants.O_CREAT | (fs.constants.O_NONBLOCK ?? 0) | (fs.constants.O_NOFOLLOW ?? 0)),
			DB_FILE_MODE,
		);
		const st = fs.fstatSync(fd);
		if (!st.isFile()) throw new Error("index database path is not a regular file");
		if (posix && (st.mode & 0o777) !== DB_FILE_MODE) fs.fchmodSync(fd, DB_FILE_MODE);
	} finally {
		if (fd !== undefined) fs.closeSync(fd);
	}
	if (!posix) return;
	for (const suffix of ["-wal", "-shm"]) {
		// Same descriptor no-follow safety as the db path: chmodSync(path) would
		// follow a swapped-in sidecar symlink and retarget its referent's mode.
		let fd: number | undefined;
		try {
			fd = fs.openSync(
				dbPath + suffix,
				fs.constants.O_RDWR | (fs.constants.O_NONBLOCK ?? 0) | (fs.constants.O_NOFOLLOW ?? 0),
			);
			const st = fs.fstatSync(fd);
			if (!st.isFile()) throw new Error(`index ${suffix} sidecar is not a regular file`);
			if ((st.mode & 0o777) !== DB_FILE_MODE) fs.fchmodSync(fd, DB_FILE_MODE);
		} catch (err) {
			// Sidecars are created by SQLite; a missing legacy sidecar stays harmless.
			if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err;
		} finally {
			if (fd !== undefined) fs.closeSync(fd);
		}
	}
}

function openDb(dbPath: string): DatabaseSync {
	secureIndexNode(dbPath);
	const db = new DatabaseSync(dbPath);
	try {
		db.exec("PRAGMA journal_mode = WAL");
		db.exec("PRAGMA busy_timeout = 5000");
		// The index is disposable derived state with no migration path. Only the
		// current schema is defined; an incompatible existing index fails on use
		// and is never deleted or rewritten here — remove it manually to rebuild.
		// SQLite's LIKE folds ASCII only, missing é/É, Greek, Cyrillic, …; every
		// LIKE below wraps columns in ulower and binds pre-folded operands so
		// both sides case-fold through the same foldCase helper.
		db.function("ulower", (s: SQLOutputValue): string => typeof s === "string" ? foldCase(s) : "");
		db.function("unear", nearLike);
		// Creating a missing external-content FTS table after messages and their
		// watermarks persist leaves it blank: unchanged files never replay inserts.
		// The disposable index must be deleted and rebuilt instead of repaired here.
		const hasFts = db.prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'session_fts'").get() !== undefined;
		const hasPersistedSchema = db.prepare("SELECT 1 FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' LIMIT 1").get() !== undefined;
		if (!hasFts && hasPersistedSchema) {
			throw new Error("index schema incompatible: session_fts missing — delete the index to rebuild");
		}
		db.exec(SCHEMA_SQL);
		// CREATE TABLE IF NOT EXISTS cannot repair a predecessor index whose
		// tables exist but lack current watermark columns (e.g. no ctime_ms): its
		// search rows would silently serve stale hits while change detection
		// breaks. Fail visibly; the index is disposable — remove it to rebuild.
		for (const [table, required] of [
			["session_files", ["path", "size", "mtime_ms", "ctime_ms"]],
			["session_failures", ["path", "size", "mtime_ms", "ctime_ms", "attempts"]],
		] as const) {
			const have = new Set((db.prepare(`PRAGMA table_info(${table})`).all() as any[]).map((c) => c.name as string));
			for (const col of required) {
				if (!have.has(col)) throw new Error(`index schema incompatible: ${table}.${col} missing — delete the index to rebuild`);
			}
		}
	} catch (err) {
		db.close();
		throw err;
	}
	return db;
}

/** Replacement/edit-sensitive file fingerprint from one fs.Stats. Size plus a
 *  floored mtime_ms misses same-length rewrites whose millisecond mtime is
 *  restored (backup restores, redactions). Exact ctime closes that: POSIX
 *  utimensat can set mtime but never ctime, so every restore bumps it. */
function fileFingerprint(stat: fs.Stats): { size: number; mtimeMs: number; ctimeMs: number } {
	return { size: stat.size, mtimeMs: Math.floor(stat.mtimeMs), ctimeMs: stat.ctimeMs };
}

// --- Parsing ---

interface ParsedMessage {
	entryId: string;
	role: string;
	timestamp: string | null;
	head: string;
	tail: string;
}

interface ParsedFile {
	cwd: string | null;
	name: string | null;
	startedAt: string | null;
	parentSession: string | null;
	preview: string | null;
	messages: ParsedMessage[];
}

/** Middle-out truncation into SEPARATE head/tail index columns: concatenating
 *  them would make boundary terms adjacent and manufacture phrase/proximity
 *  matches across the elided middle. Raw text only — no synthetic notices:
 *  these strings are FTS-indexed, so injected terms would pollute search
 *  results and snippets. */
function truncateRegions(text: string, max = MAX_TEXT_CHARS): { head: string; tail: string } | null {
	if (text.length <= max) return null;
	const kept = max - 1;
	return {
		head: text.slice(0, Math.ceil(kept / 2)),
		tail: text.slice(-Math.floor(kept / 2)),
	};
}

/** Cap an untrusted string at a parse boundary so malformed JSONL cannot
 *  produce unbounded metadata downstream. */
function capStr(value: unknown, max: number): string | null {
	return typeof value === "string" ? value.slice(0, max) : null;
}

/** Narrow ISO timestamp shape; captures wall-clock fields for validation. */
const ISO_TIMESTAMP_RE = /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$/;

/** Canonicalize a session-header timestamp to UTC ISO so its lexicographic
 *  order matches chronological order (browse sorts started_at DESC as TEXT).
 *  Anything else is stored null so it sorts behind every valid timestamp. */
function normalizeTimestamp(value: unknown): string | null {
	if (typeof value !== "string") return null;
	const m = ISO_TIMESTAMP_RE.exec(value);
	if (!m) return null;
	const [, y, mo, d, h] = m;
	if (+h > 23) return null;
	const probe = new Date(Date.UTC(+y, +mo - 1, +d));
	// Date parsing rejects invalid time fields, but silently rolls impossible
	// calendar dates such as February 30.
	if (probe.getUTCMonth() !== +mo - 1 || probe.getUTCDate() !== +d) return null;
	const date = new Date(value);
	return Number.isNaN(date.getTime()) ? null : date.toISOString();
}

function extractText(content: unknown): string {
	if (typeof content === "string") return content;
	if (!Array.isArray(content)) return "";
	const parts: string[] = [];
	for (const block of content) {
		if (block && typeof block === "object" && (block as any).type === "text" && typeof (block as any).text === "string") {
			parts.push((block as any).text);
		}
		// thinking / toolUse / tool_result blocks are skipped entirely
	}
	return parts.join("\n").trim();
}

function parseSessionFile(filePath: string, maxBytes: number): ParsedFile {
	const parsed: ParsedFile = {
		cwd: null,
		name: null,
		startedAt: null,
		parentSession: null,
		preview: null,
		messages: [],
	};
	// Unreadable/oversized file must throw so the sync transaction rolls back
	// instead of wiping previously indexed rows and advancing the watermark over a hole.
	for (const { data: entryRaw, id } of readTranscriptEntries(filePath, maxBytes)) {
		const entry = entryRaw !== null && typeof entryRaw === "object" ? (entryRaw as Record<string, unknown>) : undefined;
		switch (entry?.type) {
			case "session":
				// Untrusted header strings are capped here so every consumer
				// (browse rows, discovery meta) inherits the bound.
				parsed.cwd = capStr(entry.cwd, 500) ?? parsed.cwd;
				parsed.startedAt = normalizeTimestamp(entry.timestamp) ?? parsed.startedAt;
				parsed.parentSession = capStr(entry.parentSession, 1024) ?? parsed.parentSession;
				break;
			case "session_info":
				if (typeof entry.name === "string") parsed.name = entry.name.slice(0, 500);
				break;
			case "message": {
				const msg = entry.message as { role?: unknown; content?: unknown } | undefined;
				const role = msg?.role;
				if (!msg || (role !== "user" && role !== "assistant")) break;
				const full = extractText(msg.content);
				if (!full || id === undefined) break;
				if (role === "user" && parsed.preview === null) {
					parsed.preview = full.slice(0, 200);
				}
				const regions = truncateRegions(full);
				parsed.messages.push({
					entryId: id,
					role,
					timestamp: capStr(entry.timestamp, 128),
					head: regions ? regions.head : full,
					tail: regions ? regions.tail : "",
				});
				break;
			}
			// compaction, branch_summary, custom, custom_message, label,
			// session_info (name handled above), model_change, … are skipped
		}
	}
	return parsed;
}

/** Walk result. `complete: false` means the tree could not be fully read
 *  (missing root, readdir/stat failure) — callers must not treat unseen
 *  indexed paths as deleted. */
interface WalkResult {
	files: Map<string, fs.Stats>;
	complete: boolean;
}

function walkJsonlFiles(sessionsDir: string): WalkResult {
	const files = new Map<string, fs.Stats>();
	const stack = [sessionsDir];
	let complete = true;
	while (stack.length > 0) {
		const dir = stack.pop()!;
		let entries: fs.Dirent[];
		try {
			entries = fs.readdirSync(dir, { withFileTypes: true });
		} catch {
			complete = false;
			continue;
		}
		for (const ent of entries) {
			const full = path.join(dir, ent.name);
			if (ent.isDirectory()) {
				// Junk dirs are pruned before descent so a large ignored tree never
				// costs a walk + per-file stats on every sync pass.
				if (ent.name.startsWith("--tmp-") || ent.name.startsWith("--private-tmp-")) continue;
				stack.push(full);
			} else if (ent.isFile() && ent.name.endsWith(".jsonl")) {
				try {
					files.set(full, fs.statSync(full));
				} catch {
					complete = false; // vanished or became unreadable mid-walk
				}
			}
		}
	}
	return { files, complete };
}

// --- Sync ---

export function syncSessions(
	sessionsDir: string,
	dbPath: string,
	opts?: { cap?: number; maxFileBytes?: number },
): SyncResult {
	const cap = opts?.cap ?? DEFAULT_SYNC_CAP;
	const db = openDb(dbPath);
	try {

	// Junk dirs never reach the walk result: pruned during descent.
	const walk = walkJsonlFiles(sessionsDir);
	const all = [...walk.files];
	const watermarks = new Map(
		(db.prepare("SELECT path, size, mtime_ms, ctime_ms FROM session_files").all() as any[]).map((r) => [r.path, r]),
	);
	const failures = new Map(
		(db.prepare("SELECT path, size, mtime_ms, ctime_ms, attempts FROM session_failures").all() as any[]).map((r) => [r.path, r]),
	);
	// An incomplete traversal proves nothing about absence: purging here would
	// wipe healthy sessions over a transient readdir/stat failure. Changed/new
	// files already discovered still process normally.
	const discovered = new Set(all.map(([fp]) => fp));
	const deleted = walk.complete ? [...watermarks.keys()].filter((p) => !discovered.has(p)) : [];
	const deletedFailures = walk.complete ? [...failures.keys()].filter((p) => !discovered.has(p)) : [];

	const changed: { path: string; stat: fs.Stats }[] = [];
	for (const [p, stat] of all) {
		const fp = fileFingerprint(stat);
		const wm = watermarks.get(p);
		if (wm && wm.size === fp.size && wm.mtime_ms === fp.mtimeMs && wm.ctime_ms === fp.ctimeMs) continue;
		changed.push({ path: p, stat });
	}

	// Newest-first, but an unchanged prior failure moves behind untouched work.
	// Among retries, fewest attempts first so one persistent failure cannot
	// monopolize a small cap forever.
	const retryAttempts = (p: string, stat: fs.Stats): number | null => {
		const f = failures.get(p);
		if (!f) return null;
		const fp = fileFingerprint(stat);
		return f.size === fp.size && f.mtime_ms === fp.mtimeMs && f.ctime_ms === fp.ctimeMs ? f.attempts : null;
	};
	changed.sort((a, b) => {
		const aRetry = retryAttempts(a.path, a.stat);
		const bRetry = retryAttempts(b.path, b.stat);
		if (aRetry !== null && bRetry !== null) return aRetry - bRetry || b.stat.mtimeMs - a.stat.mtimeMs;
		if (aRetry !== null) return 1;
		if (bRetry !== null) return -1;
		return b.stat.mtimeMs - a.stat.mtimeMs;
	});
	deleted.sort((a, b) => {
		const ma = watermarks.get(a)!.mtime_ms;
		const mb = watermarks.get(b)!.mtime_ms;
		return mb - ma;
	});

	const delStmt = db.prepare("DELETE FROM messages WHERE path = ?");
	const delSession = db.prepare("DELETE FROM sessions WHERE path = ?");
	const delFile = db.prepare("DELETE FROM session_files WHERE path = ?");
	const delFailure = db.prepare("DELETE FROM session_failures WHERE path = ?");
	const upsertFailure = db.prepare(
		"INSERT INTO session_failures(path, size, mtime_ms, ctime_ms, attempts) VALUES (?, ?, ?, ?, 1) ON CONFLICT(path) DO UPDATE SET size=excluded.size, mtime_ms=excluded.mtime_ms, ctime_ms=excluded.ctime_ms, attempts=CASE WHEN size=excluded.size AND mtime_ms=excluded.mtime_ms AND ctime_ms=excluded.ctime_ms THEN attempts+1 ELSE 1 END",
	);
	const upsertFile = db.prepare(
		"INSERT INTO session_files(path, size, mtime_ms, ctime_ms) VALUES (?, ?, ?, ?) ON CONFLICT(path) DO UPDATE SET size=excluded.size, mtime_ms=excluded.mtime_ms, ctime_ms=excluded.ctime_ms",
	);
	const upsertSession = db.prepare(
		`INSERT INTO sessions(path, cwd, name, started_at, preview, parent_session) VALUES (?, ?, ?, ?, ?, ?)
		 ON CONFLICT(path) DO UPDATE SET cwd=excluded.cwd, name=excluded.name, started_at=excluded.started_at,
		 preview=excluded.preview, parent_session=excluded.parent_session`,
	);
	const insertMsg = db.prepare(
		"INSERT INTO messages(path, entry_id, role, timestamp, head, tail) VALUES (?, ?, ?, ?, ?, ?)",
	);

	const maxFileBytes = opts?.maxFileBytes ?? MAX_SESSION_FILE_BYTES;

	const tx = (filePath: string, stat: fs.Stats): void => {
		db.exec("BEGIN");
		try {
			const parsed = parseSessionFile(filePath, maxFileBytes);
			delStmt.run(filePath);
			upsertSession.run(filePath, parsed.cwd, parsed.name, parsed.startedAt, parsed.preview, parsed.parentSession);
			for (const msg of parsed.messages) {
				insertMsg.run(filePath, msg.entryId, msg.role, msg.timestamp, msg.head, msg.tail);
			}
			{
				const fp = fileFingerprint(stat);
				upsertFile.run(filePath, fp.size, fp.mtimeMs, fp.ctimeMs);
			}
			delFailure.run(filePath);
			db.exec("COMMIT");
		} catch (err) {
			db.exec("ROLLBACK");
			throw err;
		}
	};

	let filesProcessed = 0;

	// The cap bounds total indexing attempts per pass, not successes.
	const batch = changed.slice(0, cap);
	for (const { path: p, stat } of batch) {
		try {
			tx(p, stat);
			filesProcessed++;
		} catch {
			// Keep the file retryable, but put this unchanged fingerprint behind
			// untouched work on the next pass.
			{
				const fp = fileFingerprint(stat);
				upsertFailure.run(p, fp.size, fp.mtimeMs, fp.ctimeMs);
			}
		}
	}

	const deletedRemaining = Math.max(0, deleted.length - cap);
	for (const p of deleted.slice(0, cap)) {
		delStmt.run(p);
		delSession.run(p);
		delFile.run(p);
		delFailure.run(p);
	}
	for (const p of deletedFailures) delFailure.run(p);

	// Attempted failures remain unsynced and therefore remain in the backlog.
	const backlog = changed.length - filesProcessed + deletedRemaining;
	db.prepare("INSERT INTO meta(key, value) VALUES ('backlog', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value").run(String(backlog));

	return { filesProcessed, backlogRemaining: backlog, walkComplete: walk.complete };
	} finally {
		db.close();
	}
}

function getBacklog(db: DatabaseSync): number {
	const row = db.prepare("SELECT value FROM meta WHERE key = 'backlog'").get() as { value: string } | undefined;
	return row ? Number(row.value) : 0;
}

// --- Search ---

export interface SearchOptions {
	limit?: number;
	/** Path of the current session file — the whole file is skipped. */
	currentSessionPath?: string;
}

// Current-session exclusion happens INSIDE SQL (before ROW_NUMBER/caps), so
// the excluded file cannot consume the per-file rank, lineage, or scan limit.
const CURRENT_SESSION_FILTER_SQL = "(? IS NULL OR m.path <> ?)";
const LINEAGE_FILTER_SQL = `NOT EXISTS (
  SELECT 1 FROM ranked parent
  WHERE parent.path = r.parent_session
    AND parent.path <> r.path
    AND parent.rn <= 1
    -- Direct mutual cycle (A<->B) would suppress both sides; keep the
    -- lexicographically smaller path deterministically.
    AND NOT (parent.parent_session IS r.path AND r.path < parent.path)
)`;
const BASE_SELECT = `
WITH matches AS (
  SELECT m.path, m.entry_id, m.role, m.timestamp, s.cwd, s.name, s.started_at,
         s.parent_session, m.rowid AS rid,
         snippet(session_fts, -1, '[', ']', '…', 16) AS snip,
         bm25(session_fts) AS score
  FROM session_fts JOIN messages m ON m.rowid = session_fts.rowid
  LEFT JOIN sessions s ON s.path = m.path
  WHERE session_fts MATCH ? AND ${CURRENT_SESSION_FILTER_SQL}
), ranked AS (
  SELECT *, ROW_NUMBER() OVER (PARTITION BY path ORDER BY score, rid) AS rn
  FROM matches
)
SELECT path, entry_id, role, timestamp, snip, cwd, name, started_at
FROM ranked r
WHERE rn <= 1
  AND ${LINEAGE_FILTER_SQL}
ORDER BY score, rid
LIMIT ?`;

/** Bounded excerpt around the first matched term for the LIKE fallback,
 *  searching head then tail. */
function likeSnippet(head: string, tail: string, terms: string[]): string {
	for (const text of [head, tail]) {
		const lower = foldCase(text);
		let at = -1;
		for (const t of terms) {
			const i = lower.indexOf(foldCase(t));
			if (i >= 0 && (at < 0 || i < at)) at = i;
		}
		if (at >= 0) {
			const start = Math.max(0, at - 60);
			return `${start > 0 ? "…" : ""}${text.slice(start, start + 120)}…`;
		}
	}
	return head.slice(0, 120);
}

export function searchIndex(
	dbPath: string,
	query: string,
	opts?: SearchOptions,
): { hits: SearchHit[]; backlogRemaining: number } {
	const limit = opts?.limit ?? 3;
	const db = openDb(dbPath);
	try {
		const currentSessionPath = opts?.currentSessionPath ?? null;
		const plan = buildFtsQueryPlan(query);
		if (plan.ftsCandidates.length === 0 && !plan.forceLike) {
			return { hits: [], backlogRemaining: getBacklog(db) };
		}

		interface RawRow {
			path: string;
			entry_id: string;
			role: string;
			timestamp: string | null;
			snip: string;
			cwd: string | null;
			name: string | null;
			started_at: string | null;
		}
		let rows: Omit<RawRow, "snip">[] = [];
		let usedLike = plan.forceLike;

		if (!plan.forceLike) {
			// Recovery candidates run only after an FTS5 PARSE error. A candidate
			// that parses defines the result even with zero rows (`a AND b` with
			// no co-occurrence stays empty instead of degrading to OR/LIKE).
			let parseFailed = false;
			for (const cand of plan.ftsCandidates) {
				try {
					rows = db.prepare(BASE_SELECT).all(cand, currentSessionPath, currentSessionPath, limit) as any;
					// First/raw success defines the result even with zero rows. After a
					// parse error, an empty recovery candidate keeps trying later ones.
					if (!parseFailed || rows.length > 0) {
						usedLike = false;
						break;
					}
					usedLike = true;
				} catch {
					parseFailed = true;
					usedLike = true; // malformed syntax → try next candidate
				}
			}
		}

		if (usedLike) {
			const likePlan = buildLikeQueryPlan(query);
			if (likePlan === null) return { hits: [], backlogRemaining: getBacklog(db) };
			rows = db.prepare(`WITH ranked AS (
			            SELECT m.path, m.entry_id, m.role, m.timestamp, m.head, m.tail, s.cwd, s.name, s.started_at,
			                   s.parent_session,
			                   ROW_NUMBER() OVER (PARTITION BY m.path ORDER BY m.rowid DESC) AS rn,
			                   COUNT(*) OVER (PARTITION BY m.path) AS matches
			             FROM messages m LEFT JOIN sessions s ON s.path = m.path
			             WHERE ${likePlan.where} AND ${CURRENT_SESSION_FILTER_SQL}
			            )
			            SELECT path, entry_id, role, timestamp, head, tail, cwd, name, started_at
			            FROM ranked r
			            WHERE rn <= 1
			              AND ${LINEAGE_FILTER_SQL}
			            ORDER BY matches DESC, started_at DESC, path
			            LIMIT ?`).all(...likePlan.params, currentSessionPath, currentSessionPath, limit) as any;
			// Snippets anchor on operand terms only — operator words like OR would
			// otherwise match common substrings and hide the real match.
			for (const r of rows as any[]) r.snip = likeSnippet(r.head, r.tail, likePlan.terms);
		}

		// Current-file and one-hop lineage suppression already happened in SQL,
		// before the LIMIT, so excluded or fork rows cannot starve matches.
		const hits: SearchHit[] = (rows as RawRow[]).map((r, rank) => ({
			path: r.path,
			snippet: r.snip ?? "",
			rank,
			matchMessageId: r.entry_id,
			role: r.role,
			timestamp: r.timestamp ?? "",
			cwd: r.cwd ?? undefined,
			name: r.name ?? undefined,
			startedAt: r.started_at ?? undefined,
		}));
		return { hits, backlogRemaining: getBacklog(db) };
	} finally {
		db.close();
	}
}

// --- Preparation / browse ---

export interface PreparationRowOptions {
	limit: number;
	/** Canonical repository root. Omit to sample all indexed sessions. */
	repositoryRoot?: string;
	currentSessionPath?: string;
}

/** Select deterministic, repository-scoped pattern-miner candidates from the
 * existing index. Scope, current-session exclusion, and one-hop lineage
 * collapse all happen before the sample limit. */
export function getPreparationRows(
	dbPath: string,
	opts: PreparationRowOptions,
): PreparationSessionRow[] {
	const clauses: string[] = [];
	const params: string[] = [];
	if (opts.repositoryRoot !== undefined) {
		const prefix = opts.repositoryRoot.endsWith(path.sep)
			? opts.repositoryRoot
			: opts.repositoryRoot + path.sep;
		clauses.push("(s.cwd = ? OR substr(s.cwd, 1, length(?)) = ?)");
		params.push(opts.repositoryRoot, prefix, prefix);
	}
	if (opts.currentSessionPath !== undefined) {
		clauses.push("s.path <> ?");
		params.push(opts.currentSessionPath);
	}
	const where = clauses.length > 0 ? `WHERE ${clauses.join(" AND ")}` : "";
	const db = openDb(dbPath);
	try {
		const rows = db.prepare(`
			WITH eligible AS (
				SELECT s.path, s.cwd, s.name, s.started_at, s.preview,
					CASE
						WHEN s.parent_session IS NULL THEN s.path
						WHEN parent.path <> s.path AND parent.parent_session = s.path
							THEN min(s.path, parent.path)
						ELSE s.parent_session
					END AS lineage_id
				FROM sessions s
				LEFT JOIN sessions parent ON parent.path = s.parent_session
				${where}
			), ranked AS (
				SELECT *, ROW_NUMBER() OVER (
					PARTITION BY lineage_id
					ORDER BY CASE WHEN path = lineage_id THEN 0 ELSE 1 END,
						started_at DESC, path
				) AS rn
				FROM eligible
			)
			SELECT path, cwd, name, started_at, preview, lineage_id
			FROM ranked
			WHERE rn = 1
			ORDER BY started_at DESC, path
			LIMIT ?
		`).all(...params, opts.limit) as any[];
		return rows.map((r) => ({
			path: r.path,
			cwd: r.cwd ?? "",
			name: r.name ?? undefined,
			startedAt: r.started_at ?? undefined,
			preview: r.preview ?? undefined,
			lineageId: r.lineage_id,
		}));
	} finally {
		db.close();
	}
}

export function getSessionRows(dbPath: string, limit: number): SessionRow[] {
	const db = openDb(dbPath);
	try {
		const rows = db
			.prepare("SELECT path, cwd, name, started_at, preview FROM sessions ORDER BY started_at DESC LIMIT ?")
			.all(limit) as any[];
		return rows.map((r) => ({ path: r.path, cwd: r.cwd ?? "", name: r.name ?? undefined, startedAt: r.started_at ?? undefined, preview: r.preview ?? undefined }));
	} finally {
		db.close();
	}
}