/**
 * SessionDB — Persistent per-project SQLite database for session events.
 *
 * Stores raw events captured by hooks during a Claude Code session,
 * session metadata, and resume snapshots. Extends SQLiteBase from
 * the shared package.
 */

import { SQLiteBase, defaultDBPath } from "../db-base.js";
import type { PreparedStatement } from "../db-base.js";
import type { SessionEvent } from "../types.js";
import type { ProjectAttribution } from "./project-attribution.js";
import { createHash } from "node:crypto";
import { execFileSync } from "node:child_process";
import { accessSync, constants, existsSync, mkdirSync, realpathSync, renameSync } from "node:fs";
import { homedir } from "node:os";
import { dirname, isAbsolute, join, resolve } from "node:path";

// ─────────────────────────────────────────────────────────
// Storage root resolution
// ─────────────────────────────────────────────────────────
//
// This lives beside the session DB path helpers because packaged hooks and the
// statusline already consume `hooks/session-db.bundle.mjs` as their no-build
// runtime bridge. Keeping the storage resolver here avoids adding a second
// generated hook bundle just to share QUIET_CONTEXT_DIR behavior.

const STORAGE_ROOT_ENV = "QUIET_CONTEXT_DIR" as const;
const STORAGE_SESSIONS_SUBDIR = "sessions";
const STORAGE_CONTENT_SUBDIR = "content";

export type StorageDirectoryKind = "session" | "content" | "stats";
export type StorageOverrideEnvVar = typeof STORAGE_ROOT_ENV;
export type StorageDirectorySource = "default" | "override";
export type IgnoredStorageOverrideReason = "empty";

export interface ResolvedStorageDir {
  kind: StorageDirectoryKind;
  path: string;
  envVar: StorageOverrideEnvVar | null;
  source: StorageDirectorySource;
  ignoredEnvVar?: StorageOverrideEnvVar;
  ignoredReason?: IgnoredStorageOverrideReason;
}

export class StorageDirectoryError extends Error {
  readonly kind: StorageDirectoryKind;
  readonly path: string;
  readonly overrideEnvVar: StorageOverrideEnvVar;
  readonly ignoredEnvVar?: StorageOverrideEnvVar;
  readonly ignoredReason?: IgnoredStorageOverrideReason;

  constructor(
    kind: StorageDirectoryKind,
    path: string,
    overrideEnvVar: StorageOverrideEnvVar = STORAGE_ROOT_ENV,
    cause?: unknown,
    message?: string,
    metadata: Pick<ResolvedStorageDir, "ignoredEnvVar" | "ignoredReason"> = {},
  ) {
    super(message ?? storageDirectoryErrorMessage(kind, path, metadata), { cause });
    this.name = "StorageDirectoryError";
    this.kind = kind;
    this.path = path;
    this.overrideEnvVar = overrideEnvVar;
    this.ignoredEnvVar = metadata.ignoredEnvVar;
    this.ignoredReason = metadata.ignoredReason;
  }
}

type OverrideRoot =
  | { kind: "unset" }
  | { kind: "ignored-empty"; ignoredEnvVar: StorageOverrideEnvVar; ignoredReason: IgnoredStorageOverrideReason }
  | { kind: "override"; root: string };

const writableStorageCache = new Map<string, string | StorageDirectoryError>();

export interface DefaultSessionDirOptions {
  configDir: string;
  configDirEnv?: string;
  legacySessionDirEnv?: string;
  onLegacySessionDir?: (envVar: string, dir: string) => void;
  env?: NodeJS.ProcessEnv;
}

export function resolveDefaultSessionDir(opts: DefaultSessionDirOptions): string {
  const env = opts.env ?? process.env;
  const legacyEnvVar = opts.legacySessionDirEnv;
  const legacy = legacyEnvVar ? env[legacyEnvVar]?.trim() : undefined;
  if (legacy && legacyEnvVar) {
    opts.onLegacySessionDir?.(legacyEnvVar, legacy);
    return legacy;
  }

  return join(resolveConfigDirForDefaultSession(opts.configDir, opts.configDirEnv, env), "context-mode", "sessions");
}

function resolveConfigDirForDefaultSession(
  configDir: string,
  configDirEnv: string | undefined,
  env: NodeJS.ProcessEnv,
): string {
  const envValue = configDirEnv ? env[configDirEnv] : undefined;
  if (envValue && envValue.trim() !== "") {
    return resolveConfigDirValue(envValue.trim());
  }
  return resolveConfigDirValue(configDir, homedir());
}

function resolveConfigDirValue(value: string, baseDir?: string): string {
  if (value.startsWith("~")) return resolve(homedir(), value.replace(/^~[/\\]?/, ""));
  if (isAbsolute(value)) return resolve(value);
  return baseDir ? resolve(baseDir, value) : resolve(value);
}

function invalidStorageOverride(kind: StorageDirectoryKind, path: string, detail: string): StorageDirectoryError {
  return new StorageDirectoryError(
    kind,
    path,
    STORAGE_ROOT_ENV,
    undefined,
    [`Invalid ${STORAGE_ROOT_ENV} for context-mode ${kind} directory: ${detail}`, storageDirectoryHint()].join("\n"),
  );
}

function storageOverrideRoot(kind: StorageDirectoryKind): OverrideRoot {
  const raw = process.env[STORAGE_ROOT_ENV];
  if (raw === undefined) return { kind: "unset" };

  const trimmed = raw.trim();
  if (!trimmed) {
    return { kind: "ignored-empty", ignoredEnvVar: STORAGE_ROOT_ENV, ignoredReason: "empty" };
  }
  if (!isAbsolute(trimmed)) {
    throw invalidStorageOverride(kind, trimmed, `${STORAGE_ROOT_ENV} must be an absolute path.`);
  }

  return { kind: "override", root: resolve(trimmed) };
}

function ignoredStorageMetadata(root: OverrideRoot): Pick<ResolvedStorageDir, "ignoredEnvVar" | "ignoredReason"> {
  return root.kind === "ignored-empty"
    ? { ignoredEnvVar: root.ignoredEnvVar, ignoredReason: root.ignoredReason }
    : {};
}

function overrideStorageDir(kind: StorageDirectoryKind, subdir: string): ResolvedStorageDir | null {
  const root = storageOverrideRoot(kind);
  if (root.kind !== "override") return null;

  return {
    kind,
    path: join(root.root, subdir),
    envVar: STORAGE_ROOT_ENV,
    source: "override",
  };
}

function defaultStorageDir(
  kind: StorageDirectoryKind,
  getDefaultDir: () => string,
  metadata: Pick<ResolvedStorageDir, "ignoredEnvVar" | "ignoredReason">,
): ResolvedStorageDir {
  return {
    kind,
    path: resolve(getDefaultDir()),
    envVar: null,
    source: "default",
    ...metadata,
  };
}

export function resolveSessionStorageDir(getDefaultDir: () => string): ResolvedStorageDir {
  const root = storageOverrideRoot("session");
  if (root.kind === "override") {
    return {
      kind: "session",
      path: join(root.root, STORAGE_SESSIONS_SUBDIR),
      envVar: STORAGE_ROOT_ENV,
      source: "override",
    };
  }

  return defaultStorageDir("session", getDefaultDir, ignoredStorageMetadata(root));
}

export function resolveContentStorageDir(getSessionDir: () => string): ResolvedStorageDir {
  const override = overrideStorageDir("content", STORAGE_CONTENT_SUBDIR);
  if (override) return override;

  const session = resolveSessionStorageDir(getSessionDir);
  return {
    kind: "content",
    path: join(dirname(session.path), STORAGE_CONTENT_SUBDIR),
    envVar: session.envVar,
    source: session.source,
    ignoredEnvVar: session.ignoredEnvVar,
    ignoredReason: session.ignoredReason,
  };
}

export function resolveStatsStorageDir(getDefaultSessionDir: () => string): ResolvedStorageDir {
  const override = overrideStorageDir("stats", STORAGE_SESSIONS_SUBDIR);
  if (override) return override;

  const session = resolveSessionStorageDir(getDefaultSessionDir);
  return {
    kind: "stats",
    path: session.path,
    envVar: session.envVar,
    source: session.source,
    ignoredEnvVar: session.ignoredEnvVar,
    ignoredReason: session.ignoredReason,
  };
}

export function formatStorageDirectoryError(err: StorageDirectoryError): string {
  return err.message;
}

export function describeStorageDirectorySource(dir: ResolvedStorageDir): string {
  if (dir.source === "override" && dir.envVar) return `via ${dir.envVar}`;
  if (dir.ignoredEnvVar && dir.ignoredReason === "empty") return `default; ignored empty ${dir.ignoredEnvVar}`;
  return "default";
}

export function clearStorageDirectoryCheckCacheForTests(): void {
  writableStorageCache.clear();
}

export function ensureWritableStorageDir(dir: ResolvedStorageDir): string {
  const key = [
    dir.kind,
    dir.path,
    dir.source,
    dir.envVar ?? "",
    dir.ignoredEnvVar ?? "",
    dir.ignoredReason ?? "",
  ].join("\0");
  const cached = writableStorageCache.get(key);
  if (cached instanceof StorageDirectoryError) throw cached;
  if (cached === dir.path) return cached;

  try {
    mkdirSync(dir.path, { recursive: true });
    accessSync(dir.path, constants.W_OK);
    writableStorageCache.set(key, dir.path);
    return dir.path;
  } catch (err) {
    const storageErr = new StorageDirectoryError(
      dir.kind,
      pathFromStorageError(err) ?? dir.path,
      STORAGE_ROOT_ENV,
      err,
      undefined,
      { ignoredEnvVar: dir.ignoredEnvVar, ignoredReason: dir.ignoredReason },
    );
    writableStorageCache.set(key, storageErr);
    throw storageErr;
  }
}

function storageDirectoryErrorMessage(
  kind: StorageDirectoryKind,
  path: string,
  metadata: Pick<ResolvedStorageDir, "ignoredEnvVar" | "ignoredReason"> = {},
): string {
  return [
    `context-mode ${kind} directory is not writable: ${path}`,
    ignoredStorageOverrideHint(metadata),
    storageDirectoryHint(),
  ].filter(Boolean).join("\n");
}

function ignoredStorageOverrideHint(metadata: Pick<ResolvedStorageDir, "ignoredEnvVar" | "ignoredReason">): string | null {
  if (metadata.ignoredEnvVar && metadata.ignoredReason === "empty") {
    return `Ignored empty ${metadata.ignoredEnvVar}; using adapter default.`;
  }
  return null;
}

function storageDirectoryHint(): string {
  return `Set ${STORAGE_ROOT_ENV} to a writable absolute path.`;
}

function pathFromStorageError(err: unknown): string | null {
  if (!err || typeof err !== "object") return null;
  const path = (err as { path?: unknown }).path;
  return typeof path === "string" && path.length > 0 ? path : null;
}

// ─────────────────────────────────────────────────────────
// Worktree isolation
// ─────────────────────────────────────────────────────────

/**
 * Returns the worktree suffix to append to session identifiers.
 * Returns empty string when running in the main working tree.
 *
 * Set QUIET_CONTEXT_SESSION_SUFFIX to an explicit value to override
 * (useful in CI environments or when git is unavailable).
 * Set to empty string to disable isolation entirely.
 */
// Memoized per (projectDir, env override) — recomputing on every tool call cost
// ~12ms (git worktree list subprocess fork) on macOS, 50ms+ on Windows.
// Key by projectDir so callers can pass the actual workspace even when the
// MCP server has chdir'd into the installed package directory.
let _wtCache: { projectDir: string; envSuffix: string | undefined; suffix: string } | undefined;

export function normalizeWorktreePath(path: string): string {
  const normalized = path.replace(/\\/g, "/");
  if (/^\/+$/.test(normalized)) return "/";
  if (/^[A-Za-z]:\/+$/.test(normalized)) return `${normalized.slice(0, 2)}/`;
  return normalized.replace(/\/+$/, "");
}

// Case-insensitive filesystems (macOS HFS+/APFS default, Windows NTFS default)
// can report `currentRoot` and `mainRoot` with different casing for the same
// physical directory — git itself sometimes preserves the on-disk casing while
// user-supplied paths use a different casing. Compare canonically by resolving
// symlinks via realpath and case-folding on these platforms. POSIX/Linux is
// strictly case-sensitive so this is a no-op there.
function canonicalizeForCompare(root: string): string {
  let resolved = root;
  try {
    resolved = realpathSync.native(root);
  } catch {
    // Path may not exist (test fixtures, deleted dirs); fall back to as-given.
  }
  const normalized = normalizeWorktreePath(resolved);
  if (process.platform === "win32" || process.platform === "darwin") {
    return normalized.toLowerCase();
  }
  return normalized;
}

function gitOutput(projectDir: string, args: string[]): string {
  return execFileSync(
    "git",
    ["-C", projectDir, ...args],
    {
      encoding: "utf-8",
      timeout: 2000,
      stdio: ["ignore", "pipe", "ignore"],
    },
  ).trim();
}

function getCurrentWorktreeRoot(projectDir: string): string | null {
  const root = gitOutput(projectDir, ["rev-parse", "--show-toplevel"]);
  return root.length > 0 ? normalizeWorktreePath(root) : null;
}

function getMainWorktreeRoot(projectDir: string): string | null {
  const root = gitOutput(projectDir, ["worktree", "list", "--porcelain"])
    .split(/\r?\n/)
    .find((line) => line.startsWith("worktree "))
    ?.replace("worktree ", "")
    ?.trim();
  return root ? normalizeWorktreePath(root) : null;
}

export function getWorktreeSuffix(projectDir = process.cwd()): string {
  const envSuffix = process.env.QUIET_CONTEXT_SESSION_SUFFIX;
  if (_wtCache && _wtCache.projectDir === projectDir && _wtCache.envSuffix === envSuffix) {
    return _wtCache.suffix;
  }

  let suffix = "";
  if (envSuffix !== undefined) {
    suffix = envSuffix ? `__${envSuffix}` : "";
  } else {
    try {
      const currentRoot = getCurrentWorktreeRoot(projectDir);
      const mainRoot = getMainWorktreeRoot(projectDir);
      if (currentRoot && mainRoot) {
        // Use the canonicalized currentRoot for BOTH the comparison and the
        // hash so the suffix DB filename stays stable across casing-variant
        // calls on the same machine (round-5 finding). Previously the hash
        // ate raw casing, so the same linked worktree could land at two
        // different `__<8-hex>` files depending on which casing the caller
        // passed in.
        const canonicalCurrent = canonicalizeForCompare(currentRoot);
        const canonicalMain = canonicalizeForCompare(mainRoot);
        if (canonicalCurrent !== canonicalMain) {
          suffix = `__${createHash("sha256").update(canonicalCurrent).digest("hex").slice(0, 8)}`;
        }
      }
    } catch {
      // git not available or not a git repo — no suffix
    }
  }

  _wtCache = { projectDir, envSuffix, suffix };
  return suffix;
}

// Test-only helper: clear the memoization between cases.
export function _resetWorktreeSuffixCacheForTests(): void {
  _wtCache = undefined;
}

// ─────────────────────────────────────────────────────────
// SessionDB path resolution + case-fold migration
// ─────────────────────────────────────────────────────────

/**
 * Hash a project directory the way the deployed code (≤ v1.0.111) did:
 * normalize slashes only, preserve raw casing. Kept exported so the
 * migration helper can locate pre-fix DB files for one-shot rename.
 *
 * Do NOT call this for new code paths — use {@link hashProjectDirCanonical}.
 */
export function hashProjectDirLegacy(projectDir: string): string {
  return createHash("sha256")
    .update(normalizeWorktreePath(projectDir))
    .digest("hex")
    .slice(0, 16);
}

/**
 * Hash a project directory case-stably. On case-insensitive filesystems
 * (macOS HFS+/APFS, Windows NTFS) the path is lowercased so that
 * `/Users/Mert/proj` and `/users/mert/proj` resolve to the same DB file.
 * On Linux (case-sensitive) casing is preserved.
 *
 * Used as the base half of the SessionDB filename:
 *   <baseHash><worktreeSuffix>.db
 */
export function hashProjectDirCanonical(projectDir: string): string {
  const normalized = normalizeWorktreePath(projectDir);
  const folded = (process.platform === "darwin" || process.platform === "win32")
    ? normalized.toLowerCase()
    : normalized;
  return createHash("sha256").update(folded).digest("hex").slice(0, 16);
}

/**
 * Resolve the per-project FTS5 content store DB path, performing a one-shot
 * migration from a legacy raw-casing filename to the canonical one when only
 * the legacy file (with optional `-wal` / `-shm` SQLite sidecars) exists.
 *
 * Same dual-hash safety contract as {@link resolveSessionDbPath}:
 *   - Linux: canonical hash equals legacy hash → no migration attempted.
 *   - Mac/Win: rename legacy → canonical when canonical missing.
 *   - Both exist: leave legacy alone (data-loss safety). Caller picks
 *     canonical; reconciliation is a manual operation.
 *
 * Differs from `resolveSessionDbPath` in two ways:
 *   1. No worktree suffix — the FTS5 store is per-project, not per-worktree.
 *   2. The `-wal` / `-shm` sidecars travel with the main `.db` during
 *      migration so an active SQLite WAL checkpoint is not stranded behind.
 */
export function resolveContentStorePath(opts: {
  projectDir: string;
  contentDir: string;
}): string {
  const { projectDir, contentDir } = opts;
  const canonicalHash = hashProjectDirCanonical(projectDir);
  const canonicalPath = join(contentDir, `${canonicalHash}.db`);
  if (existsSync(canonicalPath)) return canonicalPath;

  const legacyHash = hashProjectDirLegacy(projectDir);
  if (legacyHash === canonicalHash) return canonicalPath; // Linux short-circuit

  const legacyPath = join(contentDir, `${legacyHash}.db`);
  if (existsSync(legacyPath)) {
    try {
      renameSync(legacyPath, canonicalPath);
      // Travel the SQLite sidecars too so an active WAL is not orphaned.
      for (const suffix of ["-wal", "-shm"]) {
        try { renameSync(legacyPath + suffix, canonicalPath + suffix); } catch { /* sidecar may not exist */ }
      }
    } catch {
      // Race or permission issue — caller will create canonicalPath fresh.
    }
  }
  return canonicalPath;
}

/**
 * Resolve the SessionDB file path for a project, performing a one-shot
 * migration from legacy raw-casing filenames to canonical ones when only
 * the legacy file exists.
 *
 * Migration rules:
 *   - Linux: `legacyHash === canonicalHash` so the resolver short-circuits;
 *     no migration ever runs (case-sensitive FS, never any drift).
 *   - macOS / Windows: if the canonical path does not exist but a legacy
 *     path does, rename in place. This preserves the user's session
 *     history across the casing-fix upgrade.
 *   - When BOTH paths exist (rare — usually only if the user previously
 *     ran two terminals with different casing) the legacy file is left
 *     UNTOUCHED. The canonical path wins; manual reconciliation needed.
 *     Avoiding the rename here is the data-loss safety guarantee.
 *
 * Worktree separation is preserved: each call only ever migrates the ONE
 * legacy file matching THIS projectDir's hash. Different worktrees have
 * different physical paths → different hashes → different DB files; the
 * migration cannot collapse worktrees.
 */
export function resolveSessionDbPath(opts: {
  projectDir: string;
  sessionsDir: string;
}): string {
  return resolveSessionPath({ ...opts, ext: ".db" });
}

/**
 * Generalized resolver: same case-fold + one-shot legacy-rename semantics
 * as {@link resolveSessionDbPath}, parameterised on the file extension so
 * the SAME logic powers `.db`, `-events.md`, and `.cleanup` paths.
 *
 * Source of truth for hooks: `hooks/session-helpers.mjs` imports this
 * function from the bundled output (`hooks/session-db.bundle.mjs`) so the
 * JS hooks and the TS server can never drift again on hash, suffix, or
 * migration policy.
 *
 * Optional `suffix` lets the hook layer inject its cross-process cached
 * worktree suffix (the marker-file optimisation that amortises the
 * `git worktree list` cost across hook forks). When omitted, falls back
 * to {@link getWorktreeSuffix} which uses an in-process cache only.
 */
export function resolveSessionPath(opts: {
  projectDir: string;
  sessionsDir: string;
  ext: string;
  suffix?: string;
}): string {
  const { projectDir, sessionsDir, ext } = opts;
  const suffix = opts.suffix ?? getWorktreeSuffix(projectDir);
  const canonicalHash = hashProjectDirCanonical(projectDir);
  const canonicalPath = join(sessionsDir, `${canonicalHash}${suffix}${ext}`);

  if (existsSync(canonicalPath)) return canonicalPath;

  const legacyHash = hashProjectDirLegacy(projectDir);
  if (legacyHash === canonicalHash) return canonicalPath; // Linux or already canonical

  const legacyPath = join(sessionsDir, `${legacyHash}${suffix}${ext}`);
  if (existsSync(legacyPath)) {
    try {
      renameSync(legacyPath, canonicalPath);
    } catch {
      // Race or permission issue — caller will create canonicalPath on first
      // write. Better to lose this rename than to throw and break ctx_stats.
    }
  }
  return canonicalPath;
}

// ─────────────────────────────────────────────────────────
// Types
// ─────────────────────────────────────────────────────────

/** A stored event row from the session_events table. */
export interface StoredEvent {
  id: number;
  session_id: string;
  type: string;
  category: string;
  priority: number;
  data: string;
  project_dir: string;
  attribution_source: string;
  attribution_confidence: number;
  bytes_avoided: number;
  bytes_returned: number;
  source_hook: string;
  created_at: string;
  data_hash: string;
}

/** Optional per-event byte accounting passed to {@link SessionDB.insertEvent}. */
export interface EventBytes {
  /** Bytes context-mode prevented from entering the model context window. */
  bytesAvoided?: number;
  /** Bytes context-mode actually returned to the model. */
  bytesReturned?: number;
}

/** Session metadata row from the session_meta table. */
export interface SessionMeta {
  session_id: string;
  project_dir: string;
  started_at: string;
  last_event_at: string | null;
  event_count: number;
  compact_count: number;
}

/**
 * Session rollup snapshot (seed-parity aggregate).
 *
 * 12 fields that mirror the platform's `session_summary` + `session_metadata`
 * stamps from src/routes/seed.ts. Each outgoing canonical event carries
 * this snapshot computed at the moment of forward so the analytics engine
 * can run its SUM/AVG/MAX rollups across per-event rows.
 */
export interface SessionRollup {
  tool_calls: number;
  errors: number;
  unique_tools: number;
  unique_files: number;
  max_file_edits: number;
  has_commit: 0 | 1;
  // v1.0.161 (Bug 2): latest commit subject from this session's type='git_commit'
  // events — stamped onto every outgoing event via the rollup spread so
  // has_commit=1 rows always carry a meaningful commit_message. Empty string
  // when the session has no commit events yet.
  commit_message: string;
  edit_test_cycles: number;
  duration_min: number;
  compact_count: number;
  sources_indexed: number;
  total_chunks: number;
  search_queries: number;
}

/** Resume snapshot row from the session_resume table. */
export interface ResumeRow {
  snapshot: string;
  event_count: number;
  consumed: number;
}

/** Aggregated tool-call stats for a single session. */
export interface ToolCallStats {
  totalCalls: number;
  totalBytesReturned: number;
  byTool: Record<string, { calls: number; bytesReturned: number }>;
}

// ─────────────────────────────────────────────────────────
// Constants
// ─────────────────────────────────────────────────────────

/** Maximum events per session before FIFO eviction kicks in. */
const MAX_EVENTS_PER_SESSION = 1000;

/** Number of recent events to check for deduplication. */
const DEDUP_WINDOW = 5;

/**
 * Coerce an arbitrary input to a non-negative integer suitable for
 * SQLite's INTEGER column. Accepts undefined / null / NaN / floats
 * and returns 0 for invalid inputs so the column never violates its
 * NOT NULL DEFAULT 0 contract.
 */
function clampNonNegativeInt(value: unknown): number {
  const n = Number(value);
  if (!Number.isFinite(n) || n <= 0) return 0;
  return Math.floor(n);
}

// ─────────────────────────────────────────────────────────
// Statement keys (typed enum to avoid string typos)
// ─────────────────────────────────────────────────────────

const S = {
  insertEvent: "insertEvent",
  getEvents: "getEvents",
  getEventsByType: "getEventsByType",
  getEventsByPriority: "getEventsByPriority",
  getEventsByTypeAndPriority: "getEventsByTypeAndPriority",
  getEventCount: "getEventCount",
  getLatestAttributedProject: "getLatestAttributedProject",
  checkDuplicate: "checkDuplicate",
  evictLowestPriority: "evictLowestPriority",
  updateMetaLastEvent: "updateMetaLastEvent",
  ensureSession: "ensureSession",
  getSessionStats: "getSessionStats",
  getSessionRollup: "getSessionRollup",
  getMaxFileEdits: "getMaxFileEdits",
  getLatestCommitMessage: "getLatestCommitMessage",
  incrementCompactCount: "incrementCompactCount",
  getUsageCursor: "getUsageCursor",
  setUsageCursor: "setUsageCursor",
  upsertResume: "upsertResume",
  getResume: "getResume",
  markResumeConsumed: "markResumeConsumed",
  claimLatestUnconsumedResume: "claimLatestUnconsumedResume",
  deleteEvents: "deleteEvents",
  deleteMeta: "deleteMeta",
  deleteResume: "deleteResume",
  getOldSessions: "getOldSessions",
  searchEvents: "searchEvents",
  incrementToolCall: "incrementToolCall",
  getToolCallTotals: "getToolCallTotals",
  getToolCallByTool: "getToolCallByTool",
  getEventBytesSummary: "getEventBytesSummary",
} as const;

// ─────────────────────────────────────────────────────────
// Schema migration helpers (shared with the analytics aggregator)
// ─────────────────────────────────────────────────────────

/**
 * Columns that the current `session_events` schema requires but earlier
 * versions of context-mode did not write. Older DBs on disk are missing
 * these — the analytics aggregator opens every DB it finds across all
 * adapters, so without an in-place migration the SUM queries below fail
 * the entire DB (the catch at the top of the read loop swallows the
 * "no such column" error and the DB contributes zero to every column,
 * not just the new ones). v1.0.148 hotfix.
 */
const SESSION_EVENTS_REQUIRED_COLUMNS: ReadonlyArray<readonly [string, string]> = [
  ["project_dir", "TEXT NOT NULL DEFAULT ''"],
  ["attribution_source", "TEXT NOT NULL DEFAULT 'unknown'"],
  ["attribution_confidence", "REAL NOT NULL DEFAULT 0"],
  ["bytes_avoided", "INTEGER NOT NULL DEFAULT 0"],
  ["bytes_returned", "INTEGER NOT NULL DEFAULT 0"],
];

/**
 * Apply any missing post-v1.0.130 `session_events` columns to an already-
 * open writable database handle. Idempotent — each ALTER is guarded by a
 * PRAGMA table_xinfo check, and the project_dir index is created only
 * when a migration actually ran. Returns true if any column was added.
 *
 * Used by both the SessionDB constructor (for the active DB) and the
 * analytics aggregator (for the 100+ historical DBs that never get
 * opened through SessionDB). ADR-0001 compatible: no EXCLUSIVE pragma,
 * no acquireDbLock — relies on the SQLite busy_timeout + WAL semantics
 * already provided by SQLiteBase.
 */
export function applyMissingSessionEventsColumns(db: {
  pragma: (q: string) => Array<{ name: string }>;
  exec: (sql: string) => void;
}): boolean {
  const colInfo = db.pragma("table_xinfo(session_events)") as Array<{ name: string }>;
  const cols = new Set(colInfo.map((c) => c.name));
  let changed = false;
  for (const [name, spec] of SESSION_EVENTS_REQUIRED_COLUMNS) {
    if (!cols.has(name)) {
      db.exec(`ALTER TABLE session_events ADD COLUMN ${name} ${spec}`);
      changed = true;
    }
  }
  if (changed) {
    db.exec(
      "CREATE INDEX IF NOT EXISTS idx_session_events_project ON session_events(session_id, project_dir)",
    );
  }
  return changed;
}

/**
 * Open a session DB file briefly, run any missing schema migrations,
 * and close. Best-effort: missing tables, file-locks, corrupt files,
 * and any DatabaseCtor error are swallowed silently — the caller
 * (analytics aggregator) handles the readonly query that follows and
 * will skip the DB if it remains unreadable.
 *
 * Lazy migration entry point for the analytics aggregator, which would
 * otherwise read 100+ historical DBs with the old (pre-v1.0.130) schema
 * and lose every signal (not just bytes_avoided) because the SELECT
 * statement references columns that don't exist on legacy schemas.
 *
 * Two open/close cycles in the worst case (one readonly probe to detect
 * legacy schema, one writable to migrate). For already-migrated DBs
 * (the common case after first read), this opens writable once and
 * exits without writing — cheaper than always-writable.
 */
export function ensureSessionEventsSchema(
  dbPath: string,
  DatabaseCtor: new (path: string, opts?: { readonly?: boolean }) => {
    pragma: (q: string) => Array<{ name: string }>;
    exec: (sql: string) => void;
    close: () => void;
  },
): void {
  let db: { pragma: (q: string) => Array<{ name: string }>; exec: (sql: string) => void; close: () => void } | null = null;
  let primaryFailure: unknown;
  let closeFailure: unknown;
  try {
    db = new DatabaseCtor(dbPath);
    applyMissingSessionEventsColumns(db);
  } catch (error) {
    primaryFailure = error;
  } finally {
    try {
      db?.close();
    } catch (error) {
      closeFailure = error;
    }
  }
  if (!closeFailure) return;
  if (primaryFailure) {
    throw new AggregateError(
      [primaryFailure, closeFailure],
      primaryFailure instanceof Error ? primaryFailure.message : "Session-events migration and cleanup failed",
    );
  }
  throw closeFailure;
}

// ─────────────────────────────────────────────────────────
// SessionDB
// ─────────────────────────────────────────────────────────

export class SessionDB extends SQLiteBase {
  /**
   * Cached prepared statements. Stored in a Map to avoid the JS private-field
   * inheritance issue where `#field` declarations in a subclass are not
   * accessible during base-class constructor calls.
   *
   * `declare` ensures TypeScript does NOT emit a field initializer at runtime.
   * Without `declare`, even `stmts!: Map<...>` emits `this.stmts = undefined`
   * after super() returns, wiping what prepareStatements() stored. The Map
   * is created inside prepareStatements() instead.
   */
  private declare stmts: Map<string, PreparedStatement>;

  constructor(opts?: { dbPath?: string }) {
    super(opts?.dbPath ?? defaultDBPath("session"));
  }

  /** Shorthand to retrieve a cached statement. */
  private stmt(key: string): PreparedStatement {
    return this.stmts.get(key)!;
  }

  // ── Schema ──

  protected initSchema(): void {
    // ── Migration: fix data_hash generated column from older schema ──
    // Old schema had data_hash as GENERATED ALWAYS AS — new schema uses explicit INSERT.
    // Detect and recreate table if needed (session data is ephemeral, safe to drop).
    try {
      const colInfo = this.db.pragma("table_xinfo(session_events)") as Array<{ name: string; hidden: number }>;
      const hashCol = colInfo.find((c) => c.name === "data_hash");
      if (hashCol && hashCol.hidden !== 0) {
        // hidden != 0 means generated column — must recreate
        this.db.exec("DROP TABLE session_events");
      }
    } catch { /* table doesn't exist yet — fine */ }

    this.db.exec(`
      CREATE TABLE IF NOT EXISTS session_events (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id TEXT NOT NULL,
        type TEXT NOT NULL,
        category TEXT NOT NULL,
        priority INTEGER NOT NULL DEFAULT 2,
        data TEXT NOT NULL,
        project_dir TEXT NOT NULL DEFAULT '',
        attribution_source TEXT NOT NULL DEFAULT 'unknown',
        attribution_confidence REAL NOT NULL DEFAULT 0,
        bytes_avoided INTEGER NOT NULL DEFAULT 0,
        bytes_returned INTEGER NOT NULL DEFAULT 0,
        source_hook TEXT NOT NULL,
        created_at TEXT NOT NULL DEFAULT (datetime('now')),
        data_hash TEXT NOT NULL DEFAULT ''
      );

      CREATE INDEX IF NOT EXISTS idx_session_events_session ON session_events(session_id);
      CREATE INDEX IF NOT EXISTS idx_session_events_type ON session_events(session_id, type);
      CREATE INDEX IF NOT EXISTS idx_session_events_priority ON session_events(session_id, priority);

      CREATE TABLE IF NOT EXISTS session_meta (
        session_id TEXT PRIMARY KEY,
        project_dir TEXT NOT NULL,
        started_at TEXT NOT NULL DEFAULT (datetime('now')),
        last_event_at TEXT,
        event_count INTEGER NOT NULL DEFAULT 0,
        compact_count INTEGER NOT NULL DEFAULT 0
      );

      CREATE TABLE IF NOT EXISTS session_resume (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id TEXT NOT NULL UNIQUE,
        snapshot TEXT NOT NULL,
        event_count INTEGER NOT NULL,
        created_at TEXT NOT NULL DEFAULT (datetime('now')),
        consumed INTEGER NOT NULL DEFAULT 0
      );

      CREATE TABLE IF NOT EXISTS tool_calls (
        session_id TEXT NOT NULL,
        tool TEXT NOT NULL,
        calls INTEGER NOT NULL DEFAULT 0,
        bytes_returned INTEGER NOT NULL DEFAULT 0,
        updated_at TEXT NOT NULL DEFAULT (datetime('now')),
        PRIMARY KEY (session_id, tool)
      );

      CREATE INDEX IF NOT EXISTS idx_tool_calls_session ON tool_calls(session_id);
    `);

    // Migration: add per-event attribution columns for existing DBs.
    // Shared helper — the analytics aggregator (analytics.ts) runs the
    // SAME migration against every historical DB it scans, so the column
    // list lives in one place at the top of this module.
    try {
      applyMissingSessionEventsColumns(this.db as unknown as {
        pragma: (q: string) => Array<{ name: string }>;
        exec: (sql: string) => void;
      });
    } catch {
      // best-effort migration only
    }

    // Migration: per-session usage high-water cursor for the Stop hook's
    // cursor-aware main-turn capture (extractTranscriptUsageSince). Stores the
    // uuid of the last assistant turn already emitted so the next Stop forwards
    // only NEW spend. Idempotent — guarded by a table_xinfo column check.
    try {
      const metaCols = this.db.pragma("table_xinfo(session_meta)") as Array<{ name: string }>;
      if (!metaCols.some((c) => c.name === "usage_cursor")) {
        this.db.exec("ALTER TABLE session_meta ADD COLUMN usage_cursor TEXT");
      }
    } catch {
      // best-effort migration only
    }

  }

  protected prepareStatements(): void {
    this.stmts = new Map<string, PreparedStatement>();

    const p = (key: string, sql: string) => {
      this.stmts.set(key, this.db.prepare(sql) as PreparedStatement);
    };

    // ── Events ──
    p(S.insertEvent,
      `INSERT INTO session_events (
         session_id, type, category, priority, data,
         project_dir, attribution_source, attribution_confidence,
         bytes_avoided, bytes_returned,
         source_hook, data_hash
       )
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`);

    p(S.getEvents,
      `SELECT id, session_id, type, category, priority, data,
              project_dir, attribution_source, attribution_confidence,
              bytes_avoided, bytes_returned,
              source_hook, created_at, data_hash
       FROM session_events WHERE session_id = ? ORDER BY id ASC LIMIT ?`);

    p(S.getEventsByType,
      `SELECT id, session_id, type, category, priority, data,
              project_dir, attribution_source, attribution_confidence,
              bytes_avoided, bytes_returned,
              source_hook, created_at, data_hash
       FROM session_events WHERE session_id = ? AND type = ? ORDER BY id ASC LIMIT ?`);

    p(S.getEventsByPriority,
      `SELECT id, session_id, type, category, priority, data,
              project_dir, attribution_source, attribution_confidence,
              bytes_avoided, bytes_returned,
              source_hook, created_at, data_hash
       FROM session_events WHERE session_id = ? AND priority >= ? ORDER BY id ASC LIMIT ?`);

    p(S.getEventsByTypeAndPriority,
      `SELECT id, session_id, type, category, priority, data,
              project_dir, attribution_source, attribution_confidence,
              bytes_avoided, bytes_returned,
              source_hook, created_at, data_hash
       FROM session_events WHERE session_id = ? AND type = ? AND priority >= ? ORDER BY id ASC LIMIT ?`);

    p(S.getEventCount,
      `SELECT COUNT(*) AS cnt FROM session_events WHERE session_id = ?`);

    p(S.getLatestAttributedProject,
      `SELECT project_dir
       FROM session_events
       WHERE session_id = ? AND project_dir != ''
       ORDER BY id DESC
       LIMIT 1`);

    p(S.checkDuplicate,
      `SELECT 1 FROM (
         SELECT type, data_hash FROM session_events
         WHERE session_id = ? ORDER BY id DESC LIMIT ?
       ) AS recent
       WHERE recent.type = ? AND recent.data_hash = ?
       LIMIT 1`);

    p(S.evictLowestPriority,
      `DELETE FROM session_events WHERE id = (
         SELECT id FROM session_events WHERE session_id = ?
         ORDER BY priority ASC, id ASC LIMIT 1
       )`);

    p(S.updateMetaLastEvent,
      `UPDATE session_meta
       SET last_event_at = datetime('now'), event_count = event_count + 1
       WHERE session_id = ?`);

    // ── Meta ──
    p(S.ensureSession,
      `INSERT OR IGNORE INTO session_meta (session_id, project_dir) VALUES (?, ?)`);

    p(S.getSessionStats,
      `SELECT session_id, project_dir, started_at, last_event_at, event_count, compact_count
       FROM session_meta WHERE session_id = ?`);

    // ── Session rollup (seed-parity aggregator) ────────────────────────
    // Single query producing 9 of the 12 platform-side session_summary +
    // session_metadata fields. Computed against the local SessionDB
    // session_events table at forward time so every outgoing canonical
    // event carries a session-wide snapshot at that moment — matches the
    // seed.ts shape where each event row has tool_calls/errors/etc. stamped.
    // max_file_edits and edit_test_cycles need separate GROUP BY queries
    // (below). compact_count is read from session_meta (already in getSessionStats).
    p(S.getSessionRollup,
      `SELECT
         COUNT(*) AS tool_calls,
         COALESCE(SUM(CASE WHEN category = 'error' THEN 1 ELSE 0 END), 0) AS errors,
         COUNT(DISTINCT type) AS unique_tools,
         COUNT(DISTINCT CASE WHEN category = 'file' THEN data END) AS unique_files,
         CASE WHEN SUM(CASE WHEN type = 'git_commit' THEN 1 ELSE 0 END) > 0 THEN 1 ELSE 0 END AS has_commit,
         CAST(COALESCE((MAX(strftime('%s', created_at)) - MIN(strftime('%s', created_at))) / 60.0, 0) AS INTEGER) AS duration_min,
         COALESCE(SUM(CASE WHEN type = 'external_ref' THEN 1 ELSE 0 END), 0) AS sources_indexed,
         CAST(COALESCE(SUM(bytes_avoided) / 1024.0, 0) AS INTEGER) AS total_chunks,
         COALESCE(SUM(CASE WHEN type IN ('file_search', 'file_glob') THEN 1 ELSE 0 END), 0) AS search_queries
       FROM session_events
       WHERE session_id = ?`);

    // max_file_edits: max edits on any single file path in the session.
    // Two-level aggregation — GROUP BY data first, then MAX of those counts.
    p(S.getMaxFileEdits,
      `SELECT COALESCE(MAX(c), 0) AS max_file_edits
       FROM (
         SELECT COUNT(*) AS c
         FROM session_events
         WHERE session_id = ? AND category = 'file' AND type IN ('file_edit', 'file_write')
         GROUP BY data
       )`);

    // v1.0.161 (Bug 2): latest commit message from session's type='git_commit'
    // events. Used by rollup spread to stamp commit_message symmetric with
    // has_commit on every outgoing event. Separate prepared statement (vs.
    // sub-select in getSessionRollup) keeps the binding shape uniform — every
    // rollup query takes a single sessionId parameter.
    p(S.getLatestCommitMessage,
      `SELECT data
       FROM session_events
       WHERE session_id = ? AND type = 'git_commit'
       ORDER BY id DESC
       LIMIT 1`);

    p(S.incrementCompactCount,
      `UPDATE session_meta SET compact_count = compact_count + 1 WHERE session_id = ?`);

    p(S.getUsageCursor,
      `SELECT usage_cursor FROM session_meta WHERE session_id = ?`);

    p(S.setUsageCursor,
      `UPDATE session_meta SET usage_cursor = ? WHERE session_id = ?`);

    // ── Resume ──
    p(S.upsertResume,
      `INSERT INTO session_resume (session_id, snapshot, event_count)
       VALUES (?, ?, ?)
       ON CONFLICT(session_id) DO UPDATE SET
         snapshot = excluded.snapshot,
         event_count = excluded.event_count,
         created_at = datetime('now'),
         consumed = 0`);

    p(S.getResume,
      `SELECT snapshot, event_count, consumed FROM session_resume WHERE session_id = ?`);

    p(S.markResumeConsumed,
      `UPDATE session_resume SET consumed = 1 WHERE session_id = ?`);

    // Atomic "pick newest unconsumed snapshot AND mark it consumed in one
    // statement". Required for race-safe cross-session resume injection
    // (Mickey / PR #376) — two parallel chat-turn hooks must not both read
    // the same row before either one writes consumed=1.
    //
    // The `session_id != ?` clause prevents self-injection (v1.0.106): when
    // Session B compacts mid-flight and produces its own row, B's next chat
    // turn must NOT claim that row back into its own prompt — that's wasted
    // tokens and steals the snapshot meant for the next fresh session.
    p(S.claimLatestUnconsumedResume,
      `UPDATE session_resume
       SET consumed = 1
       WHERE id = (
         SELECT id FROM session_resume
         WHERE consumed = 0
           AND session_id != ?
         ORDER BY created_at DESC, id DESC
         LIMIT 1
       )
       RETURNING session_id, snapshot`);

    // ── Delete ──
    p(S.deleteEvents, `DELETE FROM session_events WHERE session_id = ?`);
    p(S.deleteMeta, `DELETE FROM session_meta WHERE session_id = ?`);
    p(S.deleteResume, `DELETE FROM session_resume WHERE session_id = ?`);

    // ── Search ──
    p(S.searchEvents,
      `SELECT id, session_id, category, type, data, created_at
       FROM session_events
       WHERE (project_dir = ? OR project_dir = '')
         AND (data LIKE '%' || ? || '%' ESCAPE '\\' OR category LIKE '%' || ? || '%' ESCAPE '\\')
         AND (? IS NULL OR category = ?)
       ORDER BY id ASC
       LIMIT ?`);

    // ── Cleanup ──
    p(S.getOldSessions,
      `SELECT session_id FROM session_meta WHERE started_at < datetime('now', ? || ' days')`);

    // ── Tool calls (persistent counter) ──
    p(S.incrementToolCall,
      `INSERT INTO tool_calls (session_id, tool, calls, bytes_returned)
       VALUES (?, ?, 1, ?)
       ON CONFLICT(session_id, tool) DO UPDATE SET
         calls = calls + 1,
         bytes_returned = bytes_returned + excluded.bytes_returned,
         updated_at = datetime('now')`);

    p(S.getToolCallTotals,
      `SELECT COALESCE(SUM(calls), 0) AS calls,
              COALESCE(SUM(bytes_returned), 0) AS bytes_returned
       FROM tool_calls WHERE session_id = ?`);

    p(S.getToolCallByTool,
      `SELECT tool, calls, bytes_returned
       FROM tool_calls WHERE session_id = ? ORDER BY calls DESC`);

    // ── Event-level byte accounting (D2 PRD Phase 2) ──
    p(S.getEventBytesSummary,
      `SELECT COALESCE(SUM(bytes_avoided), 0) AS bytes_avoided,
              COALESCE(SUM(bytes_returned), 0) AS bytes_returned
       FROM session_events WHERE session_id = ?`);
  }

  // ═══════════════════════════════════════════
  // Events
  // ═══════════════════════════════════════════

  /**
   * Insert a session event with deduplication and FIFO eviction.
   *
   * Deduplication: skips if the same type + data_hash appears in the
   * last DEDUP_WINDOW events for this session.
   *
   * Eviction: if session exceeds MAX_EVENTS_PER_SESSION, evicts the
   * lowest-priority (then oldest) event.
   */
  insertEvent(
    sessionId: string,
    event: Omit<SessionEvent, "data_hash"> & { data_hash?: string },
    sourceHook: string = "PostToolUse",
    attribution?: Partial<ProjectAttribution>,
    bytes?: EventBytes,
  ): void {
    // SHA256-based dedup hash (first 16 hex chars = 8 bytes of entropy)
    const dataHash = createHash("sha256")
      .update(event.data)
      .digest("hex")
      .slice(0, 16)
      .toUpperCase();
    const projectDir = String(
      attribution?.projectDir
      ?? event.project_dir
      ?? this._getSessionProjectDir(sessionId),
    ).trim();
    const attributionSource = String(
      attribution?.source
      ?? event.attribution_source
      ?? "unknown",
    );
    const rawConfidence = Number(
      attribution?.confidence
      ?? event.attribution_confidence
      ?? 0,
    );
    const attributionConfidence = Number.isFinite(rawConfidence)
      ? Math.max(0, Math.min(1, rawConfidence))
      : 0;
    const bytesAvoided = clampNonNegativeInt(bytes?.bytesAvoided);
    const bytesReturned = clampNonNegativeInt(bytes?.bytesReturned);

    // Atomic: dedup check + eviction + insert in a single transaction
    // to prevent race conditions from concurrent hook calls.
    const transaction = this.db.transaction(() => {
      // Deduplication check: same type + data_hash in last N events
      const dup = this.stmt(S.checkDuplicate).get(sessionId, DEDUP_WINDOW, event.type, dataHash);
      if (dup) return;

      // Enforce max events with FIFO eviction of lowest priority
      const countRow = this.stmt(S.getEventCount).get(sessionId) as { cnt: number };
      if (countRow.cnt >= MAX_EVENTS_PER_SESSION) {
        this.stmt(S.evictLowestPriority).run(sessionId);
      }

      // Insert the event
      this.stmt(S.insertEvent).run(
        sessionId,
        event.type,
        event.category,
        event.priority,
        event.data,
        projectDir,
        attributionSource,
        attributionConfidence,
        bytesAvoided,
        bytesReturned,
        sourceHook,
        dataHash,
      );

      // Update meta if session exists
      this.stmt(S.updateMetaLastEvent).run(sessionId);
    });

    this.withRetry(() => transaction());
  }

  /**
   * Bulk-insert N events in a SINGLE transaction.
   *
   * PostToolUse hooks emit 5–15 events per tool call. Calling insertEvent()
   * in a loop runs N transactions = N WAL commits = N fsync candidates,
   * which is painful on Windows NTFS where commit latency dominates.
   * One transaction = one commit, dedup/evict checks reuse cached statements.
   *
   * Cross-platform: uses the same WAL-mode transaction primitive as
   * insertEvent — behavior identical on macOS / Linux / Windows.
   */
  bulkInsertEvents(
    sessionId: string,
    events: SessionEvent[],
    sourceHook: string = "PostToolUse",
    attributions?: Array<Partial<ProjectAttribution> | undefined>,
    bytesList?: Array<EventBytes | undefined>,
  ): void {
    if (!events || events.length === 0) return;
    if (events.length === 1) {
      // Cheaper to fall through to insertEvent (its own dedicated transaction).
      this.insertEvent(sessionId, events[0], sourceHook, attributions?.[0], bytesList?.[0]);
      return;
    }

    // Pre-compute hashes + normalized attribution outside the transaction
    // so the SQL transaction holds only DB work (shorter lock window).
    const prepared = events.map((event, i) => {
      const dataHash = createHash("sha256")
        .update(event.data)
        .digest("hex")
        .slice(0, 16)
        .toUpperCase();
      const attribution = attributions?.[i];
      // #827: store project_dir in canonical path shape so the search-time
      // allow-set lookup (getSessionIdsForProject) matches regardless of the
      // separator / trailing-slash form the host adapter happened to emit.
      // normalizeWorktreePath is the same rule used for project-hash stability.
      const rawProjectDir = String(
        attribution?.projectDir ?? event.project_dir ?? this._getSessionProjectDir(sessionId) ?? "",
      ).trim();
      const projectDir = rawProjectDir === "" ? "" : normalizeWorktreePath(rawProjectDir);
      const attributionSource = String(
        attribution?.source ?? event.attribution_source ?? "unknown",
      );
      const rawConfidence = Number(
        attribution?.confidence ?? event.attribution_confidence ?? 0,
      );
      const attributionConfidence = Number.isFinite(rawConfidence)
        ? Math.max(0, Math.min(1, rawConfidence))
        : 0;
      const eventBytes = bytesList?.[i];
      const bytesAvoided = clampNonNegativeInt(eventBytes?.bytesAvoided);
      const bytesReturned = clampNonNegativeInt(eventBytes?.bytesReturned);
      return {
        event,
        dataHash,
        projectDir,
        attributionSource,
        attributionConfidence,
        bytesAvoided,
        bytesReturned,
      };
    });

    const transaction = this.db.transaction(() => {
      let cnt = (this.stmt(S.getEventCount).get(sessionId) as { cnt: number }).cnt;
      for (const row of prepared) {
        const dup = this.stmt(S.checkDuplicate).get(
          sessionId, DEDUP_WINDOW, row.event.type, row.dataHash,
        );
        if (dup) continue;
        if (cnt >= MAX_EVENTS_PER_SESSION) {
          this.stmt(S.evictLowestPriority).run(sessionId);
        } else {
          cnt++;
        }
        this.stmt(S.insertEvent).run(
          sessionId,
          row.event.type,
          row.event.category,
          row.event.priority,
          row.event.data,
          row.projectDir,
          row.attributionSource,
          row.attributionConfidence,
          row.bytesAvoided,
          row.bytesReturned,
          sourceHook,
          row.dataHash,
        );
      }
      this.stmt(S.updateMetaLastEvent).run(sessionId);
    });

    this.withRetry(() => transaction());
  }

  /**
   * Retrieve events for a session with optional filtering.
   */
  getEvents(
    sessionId: string,
    opts?: { type?: string; minPriority?: number; limit?: number },
  ): StoredEvent[] {
    const limit = opts?.limit ?? 1000;
    const type = opts?.type;
    const minPriority = opts?.minPriority;

    if (type && minPriority !== undefined) {
      return this.stmt(S.getEventsByTypeAndPriority).all(sessionId, type, minPriority, limit) as StoredEvent[];
    }
    if (type) {
      return this.stmt(S.getEventsByType).all(sessionId, type, limit) as StoredEvent[];
    }
    if (minPriority !== undefined) {
      return this.stmt(S.getEventsByPriority).all(sessionId, minPriority, limit) as StoredEvent[];
    }
    return this.stmt(S.getEvents).all(sessionId, limit) as StoredEvent[];
  }

  /**
   * Get the total event count for a session.
   */
  getEventCount(sessionId: string): number {
    const row = this.stmt(S.getEventCount).get(sessionId) as { cnt: number };
    return row.cnt;
  }

  /**
   * Aggregate per-event byte accounting for a session.
   *
   * Returns the total bytes context-mode kept OUT of the model context
   * window (`bytesAvoided`) and the total it actually returned to the
   * model (`bytesReturned`). Both default to 0 for unknown sessions.
   *
   * Used by the Insight dashboard to render the "saved vs returned"
   * panel without scanning every event row in JS.
   */
  getEventBytesSummary(sessionId: string): { bytesAvoided: number; bytesReturned: number } {
    const row = this.stmt(S.getEventBytesSummary).get(sessionId) as
      | { bytes_avoided: number | null; bytes_returned: number | null }
      | undefined;
    return {
      bytesAvoided: Number(row?.bytes_avoided ?? 0),
      bytesReturned: Number(row?.bytes_returned ?? 0),
    };
  }

  /**
   * Return the most recently attributed project dir for a session.
   */
  getLatestAttributedProjectDir(sessionId: string): string | null {
    const row = this.stmt(S.getLatestAttributedProject).get(sessionId) as { project_dir: string } | undefined;
    return row?.project_dir || null;
  }

  /**
   * Look up the project_dir from session_meta as a last-resort fallback
   * for event attribution. Prevents project_dir='' orphans when the caller
   * (e.g. pi adapter) omits the attribution parameter.
   */
  _getSessionProjectDir(sessionId: string): string {
    try {
      const row = this.db.prepare("SELECT project_dir FROM session_meta WHERE session_id = ?").get(sessionId) as { project_dir: string } | undefined;
      return row?.project_dir || "";
    } catch {
      return "";
    }
  }

  /**
   * Search events by text query scoped to a project directory.
   *
   * Performs a case-insensitive LIKE search across the `data` and `category`
   * columns. An optional `source` parameter filters by exact category match.
   * Returns results ordered by monotonic id (chronological).
   *
   * Best-effort: returns empty array on any error.
   */
  searchEvents(
    query: string,
    limit: number,
    projectDir: string,
    source?: string,
  ): Array<{
    id: number;
    session_id: string;
    category: string;
    type: string;
    data: string;
    created_at: string;
  }> {
    try {
      const escapedQuery = query.replace(/[%_]/g, (char) => "\\" + char);
      const sourceParam = source ?? null;
      return this.stmt(S.searchEvents).all(
        projectDir,
        escapedQuery,
        escapedQuery,
        sourceParam,
        sourceParam,
        limit,
      ) as Array<{
        id: number;
        session_id: string;
        category: string;
        type: string;
        data: string;
        created_at: string;
      }>;
    } catch {
      return [];
    }
  }

  /**
   * Return the distinct list of session ids whose events were attributed
   * to a given `project_dir`. Powers the ctx_search `project:` filter
   * (#737) via the 2-step IN-clause strategy — ATTACH DATABASE is avoided
   * because SQLite's WAL + ATTACH combination has known correctness
   * trade-offs flagged in the upstream docs.
   *
   * Backed by the `idx_session_events_project(session_id, project_dir)`
   * composite index, so 1000-session lookups complete in single-digit
   * milliseconds. Best-effort: returns `[]` on any error.
   */
  getSessionIdsForProject(projectDir: string): string[] {
    try {
      // #827: match by canonical path shape, not raw bytes. The host adapter
      // may store `project_dir` in a different separator / trailing-slash
      // shape than the search path resolves the scope in — most visibly on
      // Windows, where attribution often carries `C:\Users\me\proj` while the
      // server resolves `C:/Users/me/proj`. An exact `project_dir = ?` match
      // then returned an EMPTY allow-set and ctx_search reported "No results
      // found" even though the content was present. We fold BOTH sides through
      // the same canonical rule used for project-hash stability
      // (normalizeWorktreePath): backslash → forward slash, then strip the
      // trailing slash. Normalizing in SQL (RTRIM(REPLACE(...))) covers rows
      // already written un-normalized without a migration, while the JS-side
      // normalize keeps the bound parameter in the identical shape. This
      // preserves the #737 project scope — distinct directories still differ
      // after normalization, so cross-project isolation is intact.
      const normalized = normalizeWorktreePath(projectDir);
      const rows = this.db
        .prepare(
          `SELECT DISTINCT session_id
             FROM session_events
            WHERE RTRIM(REPLACE(project_dir, '\\', '/'), '/') = ?`,
        )
        .all(normalized) as Array<{ session_id: string }>;
      return rows.map((r) => r.session_id);
    } catch {
      return [];
    }
  }

  // ═══════════════════════════════════════════
  // Meta
  // ═══════════════════════════════════════════

  /**
   * Ensure a session metadata entry exists. Idempotent (INSERT OR IGNORE).
   * `projectDir` is the session origin directory, not per-event attribution.
   */
  ensureSession(sessionId: string, projectDir: string): void {
    this.stmt(S.ensureSession).run(sessionId, projectDir);
  }

  /**
   * Get session statistics/metadata.
   */
  getSessionStats(sessionId: string): SessionMeta | null {
    const row = this.stmt(S.getSessionStats).get(sessionId) as SessionMeta | undefined;
    return row ?? null;
  }

  /**
   * Session rollup snapshot — 12 aggregate fields the analytics platform
   * stamps onto every outgoing event row (seed.ts shape parity).
   *
   * Called from session-loaders BEFORE `maybeForward`; the snapshot is
   * computed against the LOCAL SessionDB and threaded into the canonical
   * event so the platform-side Zod schema receives the rich shape without
   * the bridge ever hand-mapping fields (PRD §5.4 ABI passthrough).
   *
   * Returns zeroed defaults for unknown sessions — callers MUST tolerate
   * a snapshot from an empty session (first event into a fresh DB).
   */
  getSessionRollup(sessionId: string): SessionRollup {
    const main = this.stmt(S.getSessionRollup).get(sessionId) as Partial<SessionRollup> | undefined;
    const maxRow = this.stmt(S.getMaxFileEdits).get(sessionId) as { max_file_edits?: number } | undefined;
    const commitRow = this.stmt(S.getLatestCommitMessage).get(sessionId) as { data?: string } | undefined;
    const meta = this.getSessionStats(sessionId);

    // edit_test_cycles: heuristic — min(file edits, errors) approximates
    // the number of edit-then-test attempts in a session. Exact pattern
    // detection (consecutive file_edit followed by error_tool) would need
    // a windowed query; this scalar pair under-counts but never overshoots.
    const fileEdits =
      ((main as { tool_calls?: number })?.tool_calls ?? 0) > 0
        ? ((main as { unique_files?: number })?.unique_files ?? 0)
        : 0;
    const errors = (main as { errors?: number })?.errors ?? 0;
    const editTestCycles = Math.min(fileEdits, errors);

    return {
      tool_calls: main?.tool_calls ?? 0,
      errors: main?.errors ?? 0,
      unique_tools: main?.unique_tools ?? 0,
      unique_files: main?.unique_files ?? 0,
      max_file_edits: maxRow?.max_file_edits ?? 0,
      has_commit: main?.has_commit ?? 0,
      commit_message: commitRow?.data ?? "",
      edit_test_cycles: editTestCycles,
      duration_min: main?.duration_min ?? 0,
      compact_count: meta?.compact_count ?? 0,
      sources_indexed: main?.sources_indexed ?? 0,
      total_chunks: main?.total_chunks ?? 0,
      search_queries: main?.search_queries ?? 0,
    };
  }

  /**
   * Increment the compact_count for a session (tracks snapshot rebuilds).
   */
  incrementCompactCount(sessionId: string): void {
    this.stmt(S.incrementCompactCount).run(sessionId);
  }

  /**
   * Read the per-session usage high-water cursor — the uuid of the last
   * assistant turn already emitted by the Stop hook's main-turn capture.
   * Returns null when unset (first Stop) or the session row is absent.
   */
  getUsageCursor(sessionId: string): string | null {
    const row = this.stmt(S.getUsageCursor).get(sessionId) as { usage_cursor: string | null } | undefined;
    return row?.usage_cursor ?? null;
  }

  /**
   * Advance the per-session usage high-water cursor to `uuid`. No-op when the
   * session_meta row does not exist yet (callers ensureSession first).
   */
  setUsageCursor(sessionId: string, uuid: string): void {
    this.stmt(S.setUsageCursor).run(uuid, sessionId);
  }

  // ═══════════════════════════════════════════
  // Resume
  // ═══════════════════════════════════════════

  /**
   * Upsert a resume snapshot for a session. Resets consumed flag on update.
   */
  upsertResume(sessionId: string, snapshot: string, eventCount?: number): void {
    this.stmt(S.upsertResume).run(sessionId, snapshot, eventCount ?? 0);
  }

  /**
   * Retrieve the resume snapshot for a session.
   */
  getResume(sessionId: string): ResumeRow | null {
    const row = this.stmt(S.getResume).get(sessionId) as ResumeRow | undefined;
    return row ?? null;
  }

  /**
   * Mark the resume snapshot as consumed (already injected into conversation).
   */
  markResumeConsumed(sessionId: string): void {
    this.stmt(S.markResumeConsumed).run(sessionId);
  }

  /**
   * Atomically claim the most recent unconsumed resume snapshot in this DB,
   * EXCLUDING any row that belongs to `currentSessionId`.
   *
   * `SessionDB` is sharded per project (see `resolveSessionDbPath` — SHA-256
   * of canonical project dir), so "this DB" already implies "this project".
   * The atomic
   * `UPDATE … RETURNING` ensures concurrent processes for the same project
   * cannot both inject the same snapshot (Mickey / PR #376 race).
   *
   * The `currentSessionId` parameter prevents self-injection: when a session
   * compacts mid-flight and produces its own row, that session's next chat
   * turn must NOT claim that row back (wasted tokens AND it would consume
   * the snapshot meant for the next fresh session).
   *
   * Pass an empty string to allow self-claim (legacy behaviour, only useful
   * in tests or one-off harnesses).
   *
   * Returns null when no unconsumed snapshot exists for any other session.
   */
  claimLatestUnconsumedResume(
    currentSessionId: string,
  ): { sessionId: string; snapshot: string } | null {
    const row = this.stmt(S.claimLatestUnconsumedResume).get(currentSessionId) as
      | { session_id: string; snapshot: string }
      | undefined;
    if (!row) return null;
    return { sessionId: row.session_id, snapshot: row.snapshot };
  }

  /**
   * Return the most recent session_id from session_meta, or null if none.
   * Used by the runtime to attach persistent counters to the right session
   * after a process restart.
   */
  getLatestSessionId(): string | null {
    try {
      const row = this.db.prepare(
        "SELECT session_id FROM session_meta ORDER BY started_at DESC LIMIT 1",
      ).get() as { session_id?: string } | undefined;
      return row?.session_id ?? null;
    } catch {
      return null;
    }
  }

  // ═══════════════════════════════════════════
  // Tool call counters (Bug #1 + #2 — survive restart, --continue, upgrade)
  // ═══════════════════════════════════════════

  /**
   * Increment the persistent tool-call counter for `tool` in `sessionId`.
   * Adds `bytesReturned` to the cumulative total. Idempotent across
   * SessionDB instances — counters survive process restart.
   */
  incrementToolCall(sessionId: string, tool: string, bytesReturned: number = 0): void {
    const safeBytes = Number.isFinite(bytesReturned) && bytesReturned > 0 ? Math.round(bytesReturned) : 0;
    try {
      this.stmt(S.incrementToolCall).run(sessionId, tool, safeBytes);
    } catch {
      // best-effort: counter must never throw and break the parent call
    }
  }

  /**
   * Get aggregated tool-call stats for `sessionId`. Returns zero-stats
   * when the session has no recorded calls.
   */
  getToolCallStats(sessionId: string): ToolCallStats {
    try {
      const totals = this.stmt(S.getToolCallTotals).get(sessionId) as
        | { calls: number; bytes_returned: number }
        | undefined;
      const rows = this.stmt(S.getToolCallByTool).all(sessionId) as Array<{
        tool: string;
        calls: number;
        bytes_returned: number;
      }>;

      const byTool: ToolCallStats["byTool"] = {};
      for (const row of rows) {
        byTool[row.tool] = {
          calls: row.calls,
          bytesReturned: row.bytes_returned,
        };
      }

      return {
        totalCalls: totals?.calls ?? 0,
        totalBytesReturned: totals?.bytes_returned ?? 0,
        byTool,
      };
    } catch {
      return { totalCalls: 0, totalBytesReturned: 0, byTool: {} };
    }
  }

  // ═══════════════════════════════════════════
  // Lifecycle
  // ═══════════════════════════════════════════

  /**
   * Delete all data for a session (events, meta, resume).
   */
  deleteSession(sessionId: string): void {
    this.db.transaction(() => {
      this.stmt(S.deleteEvents).run(sessionId);
      this.stmt(S.deleteResume).run(sessionId);
      this.stmt(S.deleteMeta).run(sessionId);
    })();
  }

  /**
   * Remove sessions older than maxAgeDays. Returns the count of deleted sessions.
   */
  cleanupOldSessions(maxAgeDays: number = 7): number {
    const negDays = `-${maxAgeDays}`;
    const oldSessions = this.stmt(S.getOldSessions).all(negDays) as Array<{ session_id: string }>;

    for (const { session_id } of oldSessions) {
      this.deleteSession(session_id);
    }

    return oldSessions.length;
  }

  /**
   * Delete event rows whose session_id has no matching session_meta row.
   *
   * Orphaned events accumulate when meta rows were aged out by an older
   * version of `cleanupOldSessions` but the matching events were left
   * behind (or when callers wrote events without a meta upsert). The Kimi
   * Code sessionstart hook calls this on every startup as a self-healing
   * step; surfacing it as a SessionDB method keeps the SQL definition in
   * one place instead of letting hook scripts reach through to
   * `db.db.exec(...)` and re-encode schema knowledge in mjs files.
   */
  pruneOrphanedEvents(): number {
    const result = this.db
      .prepare(
        `DELETE FROM session_events WHERE session_id NOT IN (SELECT session_id FROM session_meta)`,
      )
      .run();
    return Number(result.changes ?? 0);
  }
}