lantern 0.3.0

//! Provenance-preserving entity layer.
//!
//! Extracts deterministic signals from chunk text — URLs, repository slugs,
//! domain names (derived from URLs and emails), email addresses,
//! backtick-wrapped file paths, and @-mentions — and stores them in a shape
//! that does not require another migration for each new kind. The schema
//! lives in `store::SCHEMA_V10`: `entities` is keyed by `UNIQUE(kind, value)`
//! and `chunk_entities` joins each entity back to the chunks it appeared in.
//!
//! Extraction stays regex-free and dependency-free. URLs scan for the
//! `http(s)://` prefix, take everything up to the next whitespace character,
//! and trim a small set of trailing punctuation. Emails scan around `@`, use
//! simple ASCII boundary rules, and validate a sane domain/TLD shape. File
//! paths are intentionally narrow: backtick-wrapped tokens are accepted when
//! they contain a `/` or look like a filename with an extension, and plain
//! whitespace-delimited path tokens are also picked up when they have a clear
//! path-like shape (slash plus extension or an absolute/relative prefix).
//! This keeps prose mentions like `` `src/main.rs` `` and `src/main.rs` while
//! rejecting the much larger pool of backtick-wrapped identifiers (`Vec`,
//! `String`, …). @-mentions look for `@` that is NOT preceded by an email-local
//! character (so `me@example.com` stays exclusively an email), require at
//! least two body characters with at least one ASCII letter (so dates and pure
//! digit runs do not slip through), and store the captured handle with the
//! leading `@` retained for readability. #-hashtags use the same body /
//! minimum-letter rules as mentions but skip any `#` that is preceded by an
//! alphanumeric or underscore — the URL extractor already owns trailing
//! `#anchor` fragments on URLs, and that preceded-by guard keeps URL fragments
//! from double-extracting as tags.
//! Wrong calls here are recoverable — entities can be re-extracted on reindex
//! without touching the authoritative chunk text.

use anyhow::{Context, Result};
use rusqlite::{Connection, params};
use serde::Serialize;
use sha2::{Digest, Sha256};
use std::collections::HashSet;

/// Discriminator for an extracted entity. Stored as the `kind` column in the
/// `entities` table; the string form is the on-disk representation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum EntityKind {
    Url,
    Repo,
    Domain,
    Email,
    FilePath,
    Mention,
    Hashtag,
}

impl EntityKind {
    pub fn as_str(self) -> &'static str {
        match self {
            EntityKind::Url => "url",
            EntityKind::Repo => "repo",
            EntityKind::Domain => "domain",
            EntityKind::Email => "email",
            EntityKind::FilePath => "filepath",
            EntityKind::Mention => "mention",
            EntityKind::Hashtag => "hashtag",
        }
    }
}

/// Typed relationship label for an entity-graph edge. Stays a closed enum so
/// new edge kinds (e.g. `mentions`) get a single source of truth for both the
/// JSON wire format and the human-readable column. The `snake_case` serde
/// rename keeps the on-the-wire string identical to the previous
/// `String`-typed `edge_kind`, so callers (CLI / MCP / external JSON consumers)
/// see no behavior change.
///
/// `CoOccursWith` is the chunk-level edge: two entities share at least one
/// chunk. `SameSessionAs` is a strictly looser, session-level edge: two
/// entities appear in at least one of the same sessions, even if they never
/// share a chunk. Surfacing both lets callers tell tightly-co-mentioned pairs
/// apart from entities that merely co-inhabit the same session.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum EdgeKind {
    CoOccursWith,
    SameSessionAs,
}

impl EdgeKind {
    pub fn as_str(self) -> &'static str {
        match self {
            EdgeKind::CoOccursWith => "co_occurs_with",
            EdgeKind::SameSessionAs => "same_session_as",
        }
    }
}

/// A single (kind, value) extracted from chunk text.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct Entity {
    pub kind: EntityKind,
    pub value: String,
}

/// Deterministic ID for an entity. Same (kind, value) on any machine yields the
/// same id, which lines up with the `UNIQUE(kind, value)` constraint and lets
/// `INSERT OR IGNORE` dedup without a follow-up SELECT.
pub fn entity_id(kind: EntityKind, value: &str) -> String {
    let mut h = Sha256::new();
    h.update(kind.as_str().as_bytes());
    h.update(b":");
    h.update(value.as_bytes());
    hex::encode(&h.finalize()[..16])
}

/// Trailing characters trimmed from a URL after greedy whitespace-bounded
/// capture. These are sentence punctuation that is almost always part of the
/// surrounding prose, not the URL itself. We do NOT try to balance parens
/// inside the URL — Wikipedia-style links lose their trailing `)` as a
/// trade-off for keeping extraction simple and predictable.
const URL_TRAILING_TRIM: &[char] = &[
    ',', '.', ';', ':', '!', '?', '\'', '"', ')', ']', '}', '>', '`',
];

const EMAIL_TRAILING_TRIM: &[char] = &[
    ',', '.', ';', ':', '!', '?', '\'', '"', ')', ']', '}', '>', '`',
];

fn push_entity(out: &mut Vec<Entity>, seen: &mut HashSet<String>, kind: EntityKind, value: String) {
    let key = format!("{}:{}", kind.as_str(), value);
    if seen.insert(key) {
        out.push(Entity { kind, value });
    }
}

fn looks_like_file_path_literal(candidate: &str) -> bool {
    if candidate.is_empty() || candidate.chars().any(|c| c.is_whitespace()) {
        return false;
    }
    if candidate.contains("://") {
        return false;
    }
    if candidate.contains('/') {
        return true;
    }
    match candidate.rsplit_once('.') {
        Some((stem, ext)) => {
            !stem.is_empty()
                && !ext.is_empty()
                && stem.chars().any(|c| c.is_ascii_alphanumeric())
                && ext.chars().any(|c| c.is_ascii_alphabetic())
                && stem
                    .chars()
                    .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
                && ext.chars().all(|c| c.is_ascii_alphanumeric())
        }
        None => false,
    }
}

fn push_file_path_entities(text: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    let mut cursor = 0usize;
    while let Some(rel_start) = text[cursor..].find('`') {
        let start = cursor + rel_start + 1;
        let rest = &text[start..];
        let Some(rel_end) = rest.find('`') else {
            break;
        };
        let candidate = rest[..rel_end].trim();
        if looks_like_file_path_literal(candidate) {
            push_entity(out, seen, EntityKind::FilePath, candidate.to_string());
        }
        cursor = start + rel_end + 1;
    }
}

const PATH_TOKEN_LEADING_TRIM: &[char] = &['(', '[', '{', '<', '\'', '"', '`'];
const PATH_TOKEN_TRAILING_TRIM: &[char] = &[
    ',', '.', ';', ':', '!', '?', '\'', '"', ')', ']', '}', '>', '`',
];

fn looks_like_plain_file_path_token(candidate: &str) -> bool {
    if candidate.is_empty() || candidate.chars().any(|c| c.is_whitespace()) {
        return false;
    }
    if candidate.contains("://") {
        return false;
    }
    if !(candidate.starts_with("./") || candidate.starts_with("../") || candidate.starts_with('/'))
    {
        return false;
    }
    looks_like_file_path_literal(candidate)
}

fn push_plain_file_path_entities(text: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    for token in text.split_whitespace() {
        let candidate = token
            .trim_start_matches(PATH_TOKEN_LEADING_TRIM)
            .trim_end_matches(PATH_TOKEN_TRAILING_TRIM);
        if looks_like_plain_file_path_token(candidate) {
            push_entity(out, seen, EntityKind::FilePath, candidate.to_string());
        }
    }
}

fn is_email_local_char(byte: u8) -> bool {
    byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'%' | b'+' | b'-')
}

fn is_email_domain_char(byte: u8) -> bool {
    byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-')
}

fn is_valid_email(candidate: &str) -> bool {
    let (local, domain) = match candidate.split_once('@') {
        Some(parts) => parts,
        None => return false,
    };
    if local.is_empty() || domain.is_empty() {
        return false;
    }
    if local.starts_with('.') || local.ends_with('.') {
        return false;
    }
    if domain.starts_with('.') || domain.ends_with('.') {
        return false;
    }
    if !local.as_bytes().iter().copied().all(is_email_local_char) {
        return false;
    }
    if !domain.as_bytes().iter().copied().all(is_email_domain_char) {
        return false;
    }
    let labels: Vec<&str> = domain.split('.').collect();
    if labels.len() < 2 || labels.iter().any(|label| label.is_empty()) {
        return false;
    }
    if labels
        .iter()
        .take(labels.len() - 1)
        .any(|label| label.starts_with('-') || label.ends_with('-'))
    {
        return false;
    }
    let tld = labels.last().unwrap();
    tld.len() >= 2 && tld.chars().all(|c| c.is_ascii_alphabetic())
}

fn push_url_entities(text: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    for prefix in ["https://", "http://"] {
        let mut cursor = 0usize;
        while let Some(rel) = text[cursor..].find(prefix) {
            let start = cursor + rel;
            let rest = &text[start..];
            let end = rest.find(|c: char| c.is_whitespace()).unwrap_or(rest.len());
            let raw = &rest[..end];
            let trimmed = raw.trim_end_matches(URL_TRAILING_TRIM);
            // Require at least one character past the scheme; "http://" alone
            // is not a URL worth indexing.
            if trimmed.len() > prefix.len() {
                push_entity(out, seen, EntityKind::Url, trimmed.to_string());
                push_repo_entity_from_url(trimmed, out, seen);
                push_domain_entity_from_url(trimmed, out, seen);
            }
            cursor = start + prefix.len();
        }
    }
}

/// Pull the host out of a URL's authority component and, if it normalizes to a
/// sensible domain, push it as a `Domain` entity. Strips an optional `userinfo@`
/// prefix and the trailing `:port`, then normalizes the host to lowercase.
/// IPv6 literals (bracket-wrapped) are intentionally skipped — the conservative
/// host shape used by [`looks_like_domain_host`] keeps the entity layer free of
/// brackets and other authority shapes that would not round-trip nicely.
fn push_domain_entity_from_url(url: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    let stripped = url
        .strip_prefix("https://")
        .or_else(|| url.strip_prefix("http://"))
        .unwrap_or(url);
    let authority = stripped
        .split(|c: char| matches!(c, '/' | '?' | '#'))
        .next()
        .unwrap_or(stripped);
    if authority.is_empty() {
        return;
    }
    // Drop optional `userinfo@` prefix; a host can't legally contain `@` itself,
    // so taking the part after the last `@` is enough.
    let host_with_port = authority
        .rsplit_once('@')
        .map(|(_, h)| h)
        .unwrap_or(authority);
    // Strip an explicit port. Leave IPv6 literals alone — the bracket guard in
    // `looks_like_domain_host` will reject them anyway.
    let host = host_with_port
        .rsplit_once(':')
        .map(|(h, _)| h)
        .unwrap_or(host_with_port);
    if let Some(domain) = normalize_domain(host) {
        push_entity(out, seen, EntityKind::Domain, domain);
    }
}

/// Validate and normalize a candidate domain/host string. Accepts conservative
/// DNS labels (alphanumeric + `.` + `-`) and dotted IPv4 literals, both of
/// which round-trip cleanly into the entity store. Lowercases the result so
/// `Example.COM` and `example.com` collapse to the same `Domain` entity.
fn normalize_domain(host: &str) -> Option<String> {
    if host.is_empty() {
        return None;
    }
    if !host
        .as_bytes()
        .iter()
        .copied()
        .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-'))
    {
        return None;
    }
    if host.starts_with('.') || host.ends_with('.') {
        return None;
    }
    let labels: Vec<&str> = host.split('.').collect();
    if labels.len() < 2 || labels.iter().any(|l| l.is_empty()) {
        return None;
    }
    if labels
        .iter()
        .any(|l| l.starts_with('-') || l.ends_with('-'))
    {
        return None;
    }
    Some(host.to_ascii_lowercase())
}

fn looks_like_repo_slug_segment(segment: &str) -> bool {
    if segment.is_empty() {
        return false;
    }
    let mut has_letter = false;
    for ch in segment.chars() {
        if ch.is_ascii_alphabetic() {
            has_letter = true;
        } else if !(ch.is_ascii_digit() || matches!(ch, '.' | '_' | '-')) {
            return false;
        }
    }
    has_letter
}

fn push_repo_entity_from_url(url: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    let stripped = url
        .strip_prefix("https://")
        .or_else(|| url.strip_prefix("http://"))
        .unwrap_or(url);
    let Some((_, path)) = stripped.split_once('/') else {
        return;
    };
    let path = path.split(['?', '#']).next().unwrap_or(path);
    let mut segments = path.split('/').filter(|segment| !segment.is_empty());
    let owner = match segments.next() {
        Some(segment) => segment,
        None => return,
    };
    let repo = match segments.next() {
        Some(segment) => segment.strip_suffix(".git").unwrap_or(segment),
        None => return,
    };
    if segments.next().is_some() {
        return;
    }
    if looks_like_repo_slug_segment(owner) && looks_like_repo_slug_segment(repo) {
        push_entity(out, seen, EntityKind::Repo, format!("{owner}/{repo}"));
    }
}

// Body characters allowed inside an @-mention handle. Conservative on purpose:
// alphanumeric plus a small set of internal separators that show up in real
// handles (`first.last`, `first-last`, `first_last`). Anything else ends the
// scan, so trailing punctuation like `?`, `,`, `)` does not get folded in.
fn is_mention_body_char(byte: u8) -> bool {
    byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.')
}

// Trailing characters trimmed off a mention body once the greedy scan stops.
// Includes the internal separators because handles do not end on `.`/`-`/`_`
// in any of the conventions we want to support, but a stray `bob.` is common
// at sentence ends.
const MENTION_TRAILING_TRIM: &[char] = &['.', '-', '_'];

fn push_mention_entities(text: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    let bytes = text.as_bytes();
    for (idx, byte) in bytes.iter().enumerate() {
        if *byte != b'@' {
            continue;
        }
        // If the `@` is preceded by an email-local character, the email
        // extractor already owns this match. Skipping here prevents a single
        // `me@example.com` from being recorded as both an email and a mention.
        if idx > 0 && is_email_local_char(bytes[idx - 1]) {
            continue;
        }
        let body_start = idx + 1;
        let mut end = body_start;
        while end < bytes.len() && is_mention_body_char(bytes[end]) {
            end += 1;
        }
        let raw = &text[body_start..end];
        let trimmed = raw.trim_end_matches(MENTION_TRAILING_TRIM);
        if trimmed.len() < 2 {
            continue;
        }
        let first = trimmed.as_bytes()[0];
        if !first.is_ascii_alphanumeric() && first != b'_' {
            continue;
        }
        // Conservative: require at least one ASCII letter so we do not pull in
        // pure digit/date-shaped runs like `@2024-01-15`.
        if !trimmed.bytes().any(|b| b.is_ascii_alphabetic()) {
            continue;
        }
        push_entity(out, seen, EntityKind::Mention, format!("@{trimmed}"));
    }
}

// Body characters allowed inside a #-hashtag. Matches the mention rules so the
// two entity kinds stay visually consistent, but we keep a separate entry point
// because the leading `#` guard differs from `@`.
fn is_hashtag_body_char(byte: u8) -> bool {
    byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-' | b'.')
}

fn push_hashtag_entities(text: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    let bytes = text.as_bytes();
    for (idx, byte) in bytes.iter().enumerate() {
        if *byte != b'#' {
            continue;
        }
        // URL anchors and other inline fragments usually come after an
        // alphanumeric or underscore. Skip those so `https://example.com/#x`
        // stays a URL-only extraction instead of becoming both a URL and tag.
        if idx > 0 && (bytes[idx - 1].is_ascii_alphanumeric() || bytes[idx - 1] == b'_') {
            continue;
        }
        let token_start = bytes[..idx]
            .iter()
            .rposition(|b| b.is_ascii_whitespace())
            .map(|i| i + 1)
            .unwrap_or(0);
        if text[token_start..idx].contains("://") {
            continue;
        }
        let body_start = idx + 1;
        let mut end = body_start;
        while end < bytes.len() && is_hashtag_body_char(bytes[end]) {
            end += 1;
        }
        let raw = &text[body_start..end];
        let trimmed = raw.trim_end_matches(MENTION_TRAILING_TRIM);
        if trimmed.len() < 2 {
            continue;
        }
        let first = trimmed.as_bytes()[0];
        if !first.is_ascii_alphanumeric() && first != b'_' {
            continue;
        }
        if !trimmed.bytes().any(|b| b.is_ascii_alphabetic()) {
            continue;
        }
        push_entity(out, seen, EntityKind::Hashtag, format!("#{trimmed}"));
    }
}

fn push_email_entities(text: &str, out: &mut Vec<Entity>, seen: &mut HashSet<String>) {
    let bytes = text.as_bytes();
    for (idx, byte) in bytes.iter().enumerate() {
        if *byte != b'@' {
            continue;
        }
        let mut start = idx;
        while start > 0 && is_email_local_char(bytes[start - 1]) {
            start -= 1;
        }
        let mut end = idx + 1;
        while end < bytes.len() && is_email_domain_char(bytes[end]) {
            end += 1;
        }
        if start < idx && end > idx + 1 {
            let candidate = text[start..end].trim_end_matches(EMAIL_TRAILING_TRIM);
            if is_valid_email(candidate) {
                push_entity(out, seen, EntityKind::Email, candidate.to_string());
                if let Some((_, domain)) = candidate.split_once('@') {
                    if let Some(normalized) = normalize_domain(domain) {
                        push_entity(out, seen, EntityKind::Domain, normalized);
                    }
                }
            }
        }
    }
}

/// Extract distinct URL, email, mention, and path-literal entities from
/// `text`. Order is the order of first appearance within each kind;
/// duplicates within the same chunk collapse to one entry so the caller does
/// not need to dedup before inserting. Email extraction runs before mentions
/// so the email-local-char guard inside `push_mention_entities` only needs to
/// inspect the byte directly preceding `@` to know an email already claims
/// that match.
pub fn extract_entities(text: &str) -> Vec<Entity> {
    let mut out: Vec<Entity> = Vec::new();
    let mut seen: HashSet<String> = HashSet::new();
    push_url_entities(text, &mut out, &mut seen);
    push_email_entities(text, &mut out, &mut seen);
    push_mention_entities(text, &mut out, &mut seen);
    push_hashtag_entities(text, &mut out, &mut seen);
    push_file_path_entities(text, &mut out, &mut seen);
    push_plain_file_path_entities(text, &mut out, &mut seen);
    out
}

/// Parse a CLI/MCP-supplied kind string back into [`EntityKind`].
///
/// Mirrors the `as_str` mapping above; case-insensitive so `URL` and `url` both
/// parse. Returns `Err` instead of silently coercing so the caller can surface
/// the bad input verbatim.
pub fn entity_kind_from_str(raw: &str) -> Result<EntityKind> {
    match raw.to_ascii_lowercase().as_str() {
        "url" => Ok(EntityKind::Url),
        "repo" => Ok(EntityKind::Repo),
        "domain" => Ok(EntityKind::Domain),
        "email" => Ok(EntityKind::Email),
        "filepath" => Ok(EntityKind::FilePath),
        "mention" => Ok(EntityKind::Mention),
        "hashtag" => Ok(EntityKind::Hashtag),
        other => {
            anyhow::bail!(
                "unknown entity kind '{other}' (expected: url, repo, domain, email, filepath, mention, hashtag)"
            )
        }
    }
}

/// Filter / paging options for [`list_entities`].
///
/// `value_contains` is a plain substring match (case-sensitive) so URLs and
/// file paths stay grep-able without escaping. `limit` caps how many rows are
/// returned, but `EntityListReport::total_matched` always reflects the unbounded
/// match count so callers can tell when results were truncated.
///
/// `with_chunks` opts into the graph-edge view: when set to `Some(n)`, each
/// returned entry carries up to `n` `EntityChunkRef` rows naming the chunks
/// that linked to the entity. When `None` (the default), `chunks` stays
/// `None` and no extra queries run, preserving the cheap flat listing.
///
/// `session_id` narrows the listing to entities that appear in chunks tagged
/// with the given `chunks.session_id`. When set, `chunk_count` reflects the
/// within-session count (so ordering surfaces entities most relevant to that
/// session, not globally popular ones), and any sampled `with_chunks` refs
/// are also session-scoped. Chunks whose `session_id IS NULL` are excluded.
///
/// `with_sessions` opts into shared-session evidence: when set to `Some(n > 0)`,
/// each returned entry carries up to `n` concrete `session_id` values that
/// reference the entity, ordered ascending. When `None` (the default), or
/// `Some(0)`, no evidence is loaded and `session_ids` stays `None`, preserving
/// the cheap default listing. Sessions with `NULL` ids are always excluded;
/// under a `session_id` filter the sample collapses to that one id.
///
/// `with_projects` mirrors `with_sessions` at the project grouping level
/// introduced by schema v14: when set to `Some(n > 0)`, each returned entry
/// carries up to `n` distinct non-NULL `chunks.project` values that reference
/// the entity, ordered ascending. Independent from the `session_id` filter:
/// a session-scoped listing may still legitimately span multiple projects, so
/// the sample is scoped to chunks inside that session but does NOT collapse
/// to one project.
///
/// `with_users` mirrors `with_projects` at the user grouping level introduced
/// by schema v15: when set to `Some(n > 0)`, each returned entry carries up to
/// `n` distinct non-NULL `chunks.user` values that reference the entity,
/// ordered ascending. Independent from the `session_id` filter: a single
/// session can legitimately span multiple users (e.g. a shared agent thread),
/// so the sample is scoped to that session's chunks but does NOT collapse to
/// one user.
///
/// `with_topics` mirrors `with_users` at the topic grouping level introduced
/// by schema v16: when set to `Some(n > 0)`, each returned entry carries up to
/// `n` distinct non-NULL `chunks.topic` values that reference the entity,
/// ordered ascending. Independent from the `session_id` filter: a single
/// session can legitimately span multiple topics, so the sample is scoped to
/// that session's chunks but does NOT collapse to one topic.
#[derive(Debug, Clone, Default)]
pub struct EntityListOptions {
    pub kind: Option<EntityKind>,
    pub value_contains: Option<String>,
    pub limit: Option<usize>,
    pub with_chunks: Option<usize>,
    pub session_id: Option<String>,
    pub with_sessions: Option<usize>,
    pub with_projects: Option<usize>,
    pub with_users: Option<usize>,
    pub with_topics: Option<usize>,
}

/// One row in the entity listing — the entity plus its chunk-reference count.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct EntityListEntry {
    pub id: String,
    pub kind: EntityKind,
    pub value: String,
    pub chunk_count: i64,
    /// Distinct sessions whose chunks reference this entity (chunks with
    /// `session_id IS NULL` are excluded). Tells "spammed in one session"
    /// (`session_count == 1`) apart from "spread across many sessions".
    /// In session-scoped listings this collapses to 1 by construction.
    pub session_count: i64,
    /// Distinct upstream projects whose chunks reference this entity (chunks
    /// with `project IS NULL` are excluded). Mirrors `session_count` at the
    /// project grouping level introduced by schema v14, telling "tied to one
    /// project" (`project_count == 1`) apart from "spread across many
    /// projects" (`project_count > 1`) or "no project metadata"
    /// (`project_count == 0`). Independent from the session scope filter: a
    /// `session_id` filter does not collapse this to 1.
    pub project_count: i64,
    /// Distinct upstream users whose chunks reference this entity (chunks with
    /// `user IS NULL` are excluded). Mirrors `project_count` at the user
    /// grouping level introduced by schema v15, telling "tied to one user"
    /// (`user_count == 1`) apart from "spread across users"
    /// (`user_count > 1`) or "no user metadata" (`user_count == 0`).
    /// Independent from the session scope filter: a `session_id` filter does
    /// not collapse this to 1 (one session can legitimately span multiple
    /// users — e.g. a shared agent thread).
    pub user_count: i64,
    /// Distinct upstream topics whose chunks reference this entity (chunks
    /// with `topic IS NULL` are excluded). Mirrors `project_count` /
    /// `user_count` at the topic grouping level introduced by schema v16,
    /// telling "tied to one topic" (`topic_count == 1`) apart from "spread
    /// across topics" (`topic_count > 1`) or "no topic metadata"
    /// (`topic_count == 0`). Independent from the session scope filter: a
    /// `session_id` filter does not collapse this to 1 (one session can
    /// legitimately span multiple topics).
    pub topic_count: i64,
    pub created_at: i64,
    /// Sample of chunks that reference this entity. `Some(_)` only when
    /// `EntityListOptions::with_chunks` was set; `None` keeps the flat
    /// listing's wire shape unchanged for callers that did not opt in.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub chunks: Option<Vec<EntityChunkRef>>,
    /// Sample of distinct `session_id` values whose chunks reference this
    /// entity. `Some(_)` only when `EntityListOptions::with_sessions` was set
    /// to `Some(n > 0)`; `None` keeps the default wire shape unchanged for
    /// callers that did not opt in. Ordered ascending by `session_id` so the
    /// output is deterministic across runs. Sessions with `NULL` ids are
    /// excluded; under a `session_id` filter this collapses to at most one id.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_ids: Option<Vec<String>>,
    /// Sample of distinct non-NULL `chunks.project` values that reference this
    /// entity. `Some(_)` only when `EntityListOptions::with_projects` was set
    /// to `Some(n > 0)`; `None` keeps the default wire shape unchanged for
    /// callers that did not opt in. Ordered ascending by `project` so the
    /// output is deterministic across runs. Chunks with `project IS NULL` are
    /// excluded. Under a `session_id` filter the sample is scoped to that
    /// session's chunks but does NOT collapse to one project — a session may
    /// legitimately span multiple projects.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub projects: Option<Vec<String>>,
    /// Sample of distinct non-NULL `chunks.user` values that reference this
    /// entity. `Some(_)` only when `EntityListOptions::with_users` was set to
    /// `Some(n > 0)`; `None` keeps the default wire shape unchanged for
    /// callers that did not opt in. Ordered ascending by `user` so the output
    /// is deterministic across runs. Chunks with `user IS NULL` are excluded.
    /// Under a `session_id` filter the sample is scoped to that session's
    /// chunks but does NOT collapse to one user — a single session can
    /// legitimately span multiple users.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub users: Option<Vec<String>>,
    /// Sample of distinct non-NULL `chunks.topic` values that reference this
    /// entity. `Some(_)` only when `EntityListOptions::with_topics` was set to
    /// `Some(n > 0)`; `None` keeps the default wire shape unchanged for
    /// callers that did not opt in. Ordered ascending by `topic` so the output
    /// is deterministic across runs. Chunks with `topic IS NULL` are excluded.
    /// Under a `session_id` filter the sample is scoped to that session's
    /// chunks but does NOT collapse to one topic — a single session can
    /// legitimately span multiple topics.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub topics: Option<Vec<String>>,
}

/// Minimal chunk reference returned with each entity when callers opt in via
/// `EntityListOptions::with_chunks`. The snippet is a short prefix of the
/// linked chunk's text — enough to recognize a hit, never the whole body —
/// so this stays cheap to render even for popular entities.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct EntityChunkRef {
    pub chunk_id: String,
    pub source_id: String,
    pub source_uri: String,
    pub snippet: String,
}

/// Maximum characters of chunk text included in `EntityChunkRef::snippet`.
/// Long enough to identify the hit by eye, short enough that listing many
/// entities with `with_chunks` does not dump kilobytes of body text.
const CHUNK_SNIPPET_CHARS: usize = 160;

/// Listing result: the (possibly truncated) entries and the total matched.
#[derive(Debug, Clone, Serialize)]
pub struct EntityListReport {
    pub entries: Vec<EntityListEntry>,
    pub total_matched: i64,
    pub kind_filter: Option<EntityKind>,
    pub value_contains: Option<String>,
    /// Echoes the `session_id` filter (if any) so JSON consumers can confirm
    /// the listing is session-scoped without re-deriving from `chunk_count`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
}

/// List entities, optionally filtered by kind and value substring.
///
/// Ordered by chunk count (descending) then value (ascending) so the most-
/// referenced entries surface first while still being deterministic for tests.
pub fn list_entities(conn: &Connection, opts: &EntityListOptions) -> Result<EntityListReport> {
    let mut where_clauses: Vec<&str> = Vec::new();
    let kind_str = opts.kind.map(|k| k.as_str().to_string());
    let like_pattern = opts.value_contains.as_deref().map(|s| {
        format!(
            "%{}%",
            s.replace('\\', "\\\\")
                .replace('%', "\\%")
                .replace('_', "\\_")
        )
    });
    let session_id = opts.session_id.as_deref();

    if kind_str.is_some() {
        where_clauses.push("e.kind = ?");
    }
    if like_pattern.is_some() {
        where_clauses.push("e.value LIKE ? ESCAPE '\\'");
    }
    // The session filter is the only where-clause that requires a JOIN onto
    // `chunks`. We push it onto the same WHERE list so kind / value_contains
    // still compose normally; the JOIN itself is added below only when the
    // filter is set, keeping the unfiltered path's plan unchanged.
    if session_id.is_some() {
        where_clauses.push("c.session_id = ?");
    }
    let where_sql = if where_clauses.is_empty() {
        String::new()
    } else {
        format!("WHERE {}", where_clauses.join(" AND "))
    };

    let (total_sql, list_sql) = if session_id.is_some() {
        // Session-scoped: INNER JOIN through chunk_entities + chunks so the
        // entity is only listed when it has at least one matching chunk, and
        // chunk_count reflects the within-session count rather than a global
        // tally. That keeps ordering aligned with "most relevant to this
        // session" instead of letting globally popular entities float to the
        // top regardless of session relevance.
        (
            format!(
                "SELECT COUNT(*) FROM (
                     SELECT e.id
                     FROM entities e
                     JOIN chunk_entities ce ON ce.entity_id = e.id
                     JOIN chunks c           ON c.id        = ce.chunk_id
                     {where_sql}
                     GROUP BY e.id
                 )"
            ),
            format!(
                "SELECT e.id, e.kind, e.value, e.created_at,
                        COUNT(DISTINCT ce.chunk_id)   AS chunk_count,
                        COUNT(DISTINCT c.session_id)  AS session_count,
                        COUNT(DISTINCT c.project)     AS project_count,
                        COUNT(DISTINCT c.user)        AS user_count,
                        COUNT(DISTINCT c.topic)       AS topic_count
                 FROM entities e
                 JOIN chunk_entities ce ON ce.entity_id = e.id
                 JOIN chunks c           ON c.id        = ce.chunk_id
                 {where_sql}
                 GROUP BY e.id, e.kind, e.value, e.created_at
                 ORDER BY chunk_count DESC, e.value ASC, e.id ASC
                 LIMIT ?"
            ),
        )
    } else {
        (
            format!("SELECT COUNT(*) FROM entities e {where_sql}"),
            format!(
                "SELECT e.id, e.kind, e.value, e.created_at,
                        (SELECT COUNT(*) FROM chunk_entities ce WHERE ce.entity_id = e.id) AS chunk_count,
                        (SELECT COUNT(DISTINCT c.session_id)
                         FROM chunk_entities ce
                         JOIN chunks c ON c.id = ce.chunk_id
                         WHERE ce.entity_id = e.id
                           AND c.session_id IS NOT NULL)                                  AS session_count,
                        (SELECT COUNT(DISTINCT c.project)
                         FROM chunk_entities ce
                         JOIN chunks c ON c.id = ce.chunk_id
                         WHERE ce.entity_id = e.id
                           AND c.project IS NOT NULL)                                     AS project_count,
                        (SELECT COUNT(DISTINCT c.user)
                         FROM chunk_entities ce
                         JOIN chunks c ON c.id = ce.chunk_id
                         WHERE ce.entity_id = e.id
                           AND c.user IS NOT NULL)                                        AS user_count,
                        (SELECT COUNT(DISTINCT c.topic)
                         FROM chunk_entities ce
                         JOIN chunks c ON c.id = ce.chunk_id
                         WHERE ce.entity_id = e.id
                           AND c.topic IS NOT NULL)                                       AS topic_count
                 FROM entities e
                 {where_sql}
                 ORDER BY chunk_count DESC, e.value ASC, e.id ASC
                 LIMIT ?"
            ),
        )
    };

    let mut total_params: Vec<rusqlite::types::Value> = Vec::new();
    if let Some(k) = &kind_str {
        total_params.push(k.clone().into());
    }
    if let Some(p) = &like_pattern {
        total_params.push(p.clone().into());
    }
    if let Some(s) = session_id {
        total_params.push(s.to_string().into());
    }
    let total_param_refs: Vec<&dyn rusqlite::ToSql> = total_params
        .iter()
        .map(|v| v as &dyn rusqlite::ToSql)
        .collect();
    let total_matched: i64 = conn
        .query_row(&total_sql, total_param_refs.as_slice(), |row| row.get(0))
        .with_context(|| format!("counting entities matching {opts:?}"))?;

    let limit = opts.limit.unwrap_or(usize::MAX).min(i64::MAX as usize) as i64;
    let mut list_params = total_params.clone();
    list_params.push(limit.into());
    let list_param_refs: Vec<&dyn rusqlite::ToSql> = list_params
        .iter()
        .map(|v| v as &dyn rusqlite::ToSql)
        .collect();

    let mut stmt = conn
        .prepare(&list_sql)
        .context("preparing list_entities query")?;
    let rows = stmt
        .query_map(list_param_refs.as_slice(), |row| {
            let kind_raw: String = row.get(1)?;
            let kind = entity_kind_from_str(&kind_raw).map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    1,
                    rusqlite::types::Type::Text,
                    Box::new(std::io::Error::other(e.to_string())),
                )
            })?;
            Ok(EntityListEntry {
                id: row.get(0)?,
                kind,
                value: row.get(2)?,
                created_at: row.get(3)?,
                chunk_count: row.get(4)?,
                session_count: row.get(5)?,
                project_count: row.get(6)?,
                user_count: row.get(7)?,
                topic_count: row.get(8)?,
                chunks: None,
                session_ids: None,
                projects: None,
                users: None,
                topics: None,
            })
        })
        .context("running list_entities query")?;
    let mut entries = rows.collect::<Result<Vec<_>, _>>()?;

    if let Some(per_entity_limit) = opts.with_chunks {
        for entry in &mut entries {
            entry.chunks = Some(load_chunk_refs(
                conn,
                &entry.id,
                per_entity_limit,
                session_id,
            )?);
        }
    }

    if let Some(per_entity_limit) = opts.with_sessions.filter(|n| *n > 0) {
        for entry in &mut entries {
            entry.session_ids = Some(load_entity_session_ids(
                conn,
                &entry.id,
                per_entity_limit,
                session_id,
            )?);
        }
    }

    if let Some(per_entity_limit) = opts.with_projects.filter(|n| *n > 0) {
        for entry in &mut entries {
            entry.projects = Some(load_entity_projects(
                conn,
                &entry.id,
                per_entity_limit,
                session_id,
            )?);
        }
    }

    if let Some(per_entity_limit) = opts.with_users.filter(|n| *n > 0) {
        for entry in &mut entries {
            entry.users = Some(load_entity_users(
                conn,
                &entry.id,
                per_entity_limit,
                session_id,
            )?);
        }
    }

    if let Some(per_entity_limit) = opts.with_topics.filter(|n| *n > 0) {
        for entry in &mut entries {
            entry.topics = Some(load_entity_topics(
                conn,
                &entry.id,
                per_entity_limit,
                session_id,
            )?);
        }
    }

    Ok(EntityListReport {
        entries,
        total_matched,
        kind_filter: opts.kind,
        value_contains: opts.value_contains.clone(),
        session_id: opts.session_id.clone(),
    })
}

fn load_chunk_refs(
    conn: &Connection,
    entity_id: &str,
    limit: usize,
    session_id: Option<&str>,
) -> Result<Vec<EntityChunkRef>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    // Order by chunk created_at then id so the output is deterministic across
    // runs even when several chunks of the same source linked to the entity in
    // the same insert batch (and thus share a created_at). When `session_id`
    // is set we narrow the sample to chunks within that session, keeping the
    // chunk-ref view consistent with the listing's session-scoped count.
    if let Some(session) = session_id {
        let mut stmt = conn
            .prepare(
                "SELECT c.id, c.source_id, s.uri, c.text
                 FROM chunk_entities ce
                 JOIN chunks c  ON c.id = ce.chunk_id
                 JOIN sources s ON s.id = c.source_id
                 WHERE ce.entity_id = ?1
                   AND c.session_id = ?2
                 ORDER BY c.created_at ASC, c.id ASC
                 LIMIT ?3",
            )
            .context("preparing session-scoped chunks-for-entity query")?;
        let rows = stmt
            .query_map(params![entity_id, session, limit as i64], |row| {
                build_chunk_ref(row)
            })
            .context("running session-scoped chunks-for-entity query")?;
        return Ok(rows.collect::<Result<Vec<_>, _>>()?);
    }
    let mut stmt = conn
        .prepare(
            "SELECT c.id, c.source_id, s.uri, c.text
             FROM chunk_entities ce
             JOIN chunks c  ON c.id = ce.chunk_id
             JOIN sources s ON s.id = c.source_id
             WHERE ce.entity_id = ?1
             ORDER BY c.created_at ASC, c.id ASC
             LIMIT ?2",
        )
        .context("preparing chunks-for-entity query")?;
    let rows = stmt
        .query_map(params![entity_id, limit as i64], |row| build_chunk_ref(row))
        .context("running chunks-for-entity query")?;
    let refs = rows.collect::<Result<Vec<_>, _>>()?;
    Ok(refs)
}

/// A single entity referenced from a specific chunk, returned in the order
/// `load_entities_for_chunk` produces. The (`kind`, `value`) pair matches the
/// `entities` row; `id` is the deterministic hash so callers can join straight
/// back to the entity graph (e.g. into `entity-neighbors`) without re-deriving
/// it.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct ChunkEntity {
    pub id: String,
    pub kind: EntityKind,
    pub value: String,
}

/// Load the entities referenced by a single chunk.
///
/// Ordered by (kind, value, id) so the output is deterministic across runs and
/// independent of insert order. `limit == 0` short-circuits to an empty vec so
/// opt-in callers can stay on the cheap path; otherwise the caller-supplied
/// limit caps how many entities are returned. The `entities.kind` column is
/// validated through `entity_kind_from_str` to keep this path aligned with
/// `list_entities` instead of silently surfacing unknown kinds.
pub fn load_entities_for_chunk(
    conn: &Connection,
    chunk_id: &str,
    limit: usize,
) -> Result<Vec<ChunkEntity>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let mut stmt = conn
        .prepare(
            "SELECT e.id, e.kind, e.value
             FROM chunk_entities ce
             JOIN entities e ON e.id = ce.entity_id
             WHERE ce.chunk_id = ?1
             ORDER BY e.kind ASC, e.value ASC, e.id ASC
             LIMIT ?2",
        )
        .context("preparing entities-for-chunk query")?;
    let rows = stmt
        .query_map(params![chunk_id, limit as i64], |row| {
            let id: String = row.get(0)?;
            let kind_raw: String = row.get(1)?;
            let value: String = row.get(2)?;
            let kind = entity_kind_from_str(&kind_raw).map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    1,
                    rusqlite::types::Type::Text,
                    Box::new(std::io::Error::other(e.to_string())),
                )
            })?;
            Ok(ChunkEntity { id, kind, value })
        })
        .context("running entities-for-chunk query")?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

fn load_shared_chunk_refs(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_id: Option<&str>,
) -> Result<Vec<EntityChunkRef>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    if let Some(session) = session_id {
        let mut stmt = conn
            .prepare(
                "SELECT c.id, c.source_id, s.uri, c.text
                 FROM chunk_entities ce_source
                 JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
                 JOIN chunks c ON c.id = ce_source.chunk_id
                 JOIN sources s ON s.id = c.source_id
                 WHERE ce_source.entity_id = ?1
                   AND ce_neighbor.entity_id = ?2
                   AND c.session_id = ?3
                 ORDER BY c.created_at ASC, c.id ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-chunks-for-neighbors query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, neighbor_entity_id, session, limit as i64],
                |row| build_chunk_ref(row),
            )
            .context("running session-scoped shared-chunks-for-neighbors query")?;
        return Ok(rows.collect::<Result<Vec<_>, _>>()?);
    }
    let mut stmt = conn
        .prepare(
            "SELECT c.id, c.source_id, s.uri, c.text
             FROM chunk_entities ce_source
             JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
             JOIN chunks c ON c.id = ce_source.chunk_id
             JOIN sources s ON s.id = c.source_id
             WHERE ce_source.entity_id = ?1
               AND ce_neighbor.entity_id = ?2
             ORDER BY c.created_at ASC, c.id ASC
             LIMIT ?3",
        )
        .context("preparing shared-chunks-for-neighbors query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit as i64],
            |row| build_chunk_ref(row),
        )
        .context("running shared-chunks-for-neighbors query")?;
    let refs = rows.collect::<Result<Vec<_>, _>>()?;
    Ok(refs)
}

/// Sample up to `limit` distinct non-NULL `session_id` values drawn from the
/// chunks shared by `source_entity_id` and `neighbor_entity_id`. Ordered
/// ascending so the evidence is deterministic across runs. When
/// `session_filter` is set the result collapses to at most one session id by
/// construction. `limit == 0` short-circuits so opt-in callers stay on the
/// cheap path.
fn load_shared_chunk_session_ids(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.session_id
                 FROM chunk_entities ce_source
                 JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
                 JOIN chunks c ON c.id = ce_source.chunk_id
                 WHERE ce_source.entity_id = ?1
                   AND ce_neighbor.entity_id = ?2
                   AND c.session_id = ?3
                 ORDER BY c.session_id ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-chunk-session-ids query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, neighbor_entity_id, session, limit],
                |row| row.get::<_, String>(0),
            )
            .context("running session-scoped shared-chunk-session-ids query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.session_id
             FROM chunk_entities ce_source
             JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
             JOIN chunks c ON c.id = ce_source.chunk_id
             WHERE ce_source.entity_id = ?1
               AND ce_neighbor.entity_id = ?2
               AND c.session_id IS NOT NULL
             ORDER BY c.session_id ASC
             LIMIT ?3",
        )
        .context("preparing shared-chunk-session-ids query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit],
            |row| row.get::<_, String>(0),
        )
        .context("running shared-chunk-session-ids query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` distinct non-NULL `chunks.project` values drawn from
/// the chunks shared by `source_entity_id` and `neighbor_entity_id`. Ordered
/// ascending so the evidence is deterministic across runs. When
/// `session_filter` is set the sample is scoped to that session's chunks but
/// does NOT collapse to one project — a single session can legitimately span
/// multiple projects. `limit == 0` short-circuits so opt-in callers stay on
/// the cheap path.
fn load_shared_chunk_projects(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.project
                 FROM chunk_entities ce_source
                 JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
                 JOIN chunks c ON c.id = ce_source.chunk_id
                 WHERE ce_source.entity_id = ?1
                   AND ce_neighbor.entity_id = ?2
                   AND c.session_id = ?3
                   AND c.project IS NOT NULL
                 ORDER BY c.project ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-chunk-projects query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, neighbor_entity_id, session, limit],
                |row| row.get::<_, String>(0),
            )
            .context("running session-scoped shared-chunk-projects query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.project
             FROM chunk_entities ce_source
             JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
             JOIN chunks c ON c.id = ce_source.chunk_id
             WHERE ce_source.entity_id = ?1
               AND ce_neighbor.entity_id = ?2
               AND c.project IS NOT NULL
             ORDER BY c.project ASC
             LIMIT ?3",
        )
        .context("preparing shared-chunk-projects query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit],
            |row| row.get::<_, String>(0),
        )
        .context("running shared-chunk-projects query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` distinct non-NULL `chunks.user` values drawn from the
/// chunks shared by `source_entity_id` and `neighbor_entity_id`. Ordered
/// ascending so the evidence is deterministic across runs. When
/// `session_filter` is set the sample is scoped to that session's chunks but
/// does NOT collapse to one user — a single session can legitimately span
/// multiple users (e.g. a shared agent thread). `limit == 0` short-circuits so
/// opt-in callers stay on the cheap path.
fn load_shared_chunk_users(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.user
                 FROM chunk_entities ce_source
                 JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
                 JOIN chunks c ON c.id = ce_source.chunk_id
                 WHERE ce_source.entity_id = ?1
                   AND ce_neighbor.entity_id = ?2
                   AND c.session_id = ?3
                   AND c.user IS NOT NULL
                 ORDER BY c.user ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-chunk-users query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, neighbor_entity_id, session, limit],
                |row| row.get::<_, String>(0),
            )
            .context("running session-scoped shared-chunk-users query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.user
             FROM chunk_entities ce_source
             JOIN chunk_entities ce_neighbor ON ce_neighbor.chunk_id = ce_source.chunk_id
             JOIN chunks c ON c.id = ce_source.chunk_id
             WHERE ce_source.entity_id = ?1
               AND ce_neighbor.entity_id = ?2
               AND c.user IS NOT NULL
             ORDER BY c.user ASC
             LIMIT ?3",
        )
        .context("preparing shared-chunk-users query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit],
            |row| row.get::<_, String>(0),
        )
        .context("running shared-chunk-users query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

fn build_chunk_ref(row: &rusqlite::Row<'_>) -> rusqlite::Result<EntityChunkRef> {
    let text: String = row.get(3)?;
    Ok(EntityChunkRef {
        chunk_id: row.get(0)?,
        source_id: row.get(1)?,
        source_uri: row.get(2)?,
        snippet: snippet_for(&text),
    })
}

fn snippet_for(text: &str) -> String {
    let collapsed: String = text.split_whitespace().collect::<Vec<_>>().join(" ");
    let mut out = String::new();
    for ch in collapsed.chars().take(CHUNK_SNIPPET_CHARS) {
        out.push(ch);
    }
    if collapsed.chars().count() > CHUNK_SNIPPET_CHARS {
        out.push('…');
    }
    out
}

/// Filter / paging options for [`entity_neighbors`].
///
/// `kind` restricts neighbors to a single entity kind (e.g. only co-occurring
/// URLs). `limit` caps how many neighbors are returned;
/// `EntityNeighborsReport::total_neighbors` always reflects the unbounded
/// count so callers can detect truncation. `with_chunks` optionally includes
/// up to N shared chunk references per neighbor, exposing the concrete
/// chunk-level evidence for the co-occurrence edge.
///
/// `session_id` narrows the traversal to chunks tagged with the given
/// `chunks.session_id`. When set, both `shared_chunks` and `chunk_count`
/// reflect within-session counts (so a neighbor only surfaces when it
/// co-occurs with the source *inside* that session) and any sampled
/// `with_chunks` evidence is also session-scoped. Chunks whose
/// `session_id IS NULL` are excluded.
///
/// `with_sessions` opts into session-id evidence sampling: when set to
/// `Some(n > 0)`, each neighbor carries up to `n` distinct `session_id` values
/// drawn from the chunks it shares with the source, ordered ascending for
/// determinism. When `None` or `Some(0)` the cheap default shape is preserved
/// and no extra queries run. When a `session_id` filter is set the sample
/// collapses to that single session id by construction.
///
/// `with_projects` mirrors `with_sessions` at the project grouping level
/// introduced by schema v14. When set to `Some(n > 0)`, each neighbor carries
/// up to `n` distinct non-NULL `chunks.project` values drawn from the chunks
/// it shares with the source, ordered ascending for determinism. When `None`
/// or `Some(0)` the cheap default shape is preserved and no extra queries
/// run. Independent of the `session_id` filter — a single session can still
/// span multiple projects, so the sample is scoped to chunks inside that
/// session but does NOT collapse to one project.
///
/// `with_users` mirrors `with_projects` at the user grouping level introduced
/// by schema v15. When set to `Some(n > 0)`, each neighbor carries up to `n`
/// distinct non-NULL `chunks.user` values drawn from the chunks it shares with
/// the source, ordered ascending for determinism. When `None` or `Some(0)` the
/// cheap default shape is preserved and no extra queries run. Independent of
/// the `session_id` filter — a single session can legitimately span multiple
/// users (e.g. a shared agent thread), so the sample is scoped to chunks
/// inside that session but does NOT collapse to one user.
#[derive(Debug, Clone, Default)]
pub struct EntityNeighborsOptions {
    pub kind: Option<EntityKind>,
    pub limit: Option<usize>,
    pub with_chunks: Option<usize>,
    pub session_id: Option<String>,
    pub with_sessions: Option<usize>,
    pub with_projects: Option<usize>,
    pub with_users: Option<usize>,
}

/// One neighbor of the source entity: a co-occurring entity, the typed edge
/// kind that links it to the source, the number of chunks shared with the
/// source, and that neighbor's overall chunk count.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct EntityNeighbor {
    pub id: String,
    pub kind: EntityKind,
    pub value: String,
    /// Stable relationship label for this graph edge.
    pub edge_kind: EdgeKind,
    /// Number of chunks the neighbor appears in *together with* the source
    /// entity. Always >= 1.
    pub shared_chunks: i64,
    /// Total chunks the neighbor appears in across the whole store. Useful
    /// for telling "specifically related" (low total, all shared) apart from
    /// "popular everywhere" (high total, mostly elsewhere).
    pub chunk_count: i64,
    /// Distinct sessions whose chunks reference this neighbor (chunks with
    /// `session_id IS NULL` excluded). Combined with `shared_chunks`, this
    /// lets callers tell "co-occurs in this session" apart from "spread
    /// across many sessions": a neighbor with `shared_chunks = 3` and
    /// `session_count = 1` is locally noisy, while the same `shared_chunks`
    /// with `session_count = 12` is a recurring partner across the store.
    /// Collapses to 0 or 1 under the `session_id` filter by construction.
    pub session_count: i64,
    /// Distinct upstream projects whose chunks reference this neighbor
    /// (chunks with `project IS NULL` excluded). Mirrors `session_count` at
    /// the project grouping level introduced by schema v14, telling "tied to
    /// one project" (`project_count == 1`) apart from "spread across
    /// projects" (`project_count > 1`) or "no project metadata"
    /// (`project_count == 0`). Follows the same chunk-scope rule as
    /// `chunk_count` rather than `session_count`: under a `session_id`
    /// filter the count reflects distinct projects among the neighbor's
    /// chunks *inside* that session and does NOT collapse to 1.
    pub project_count: i64,
    /// Distinct upstream users whose chunks reference this neighbor (chunks
    /// with `user IS NULL` excluded). Mirrors `project_count` at the user
    /// grouping level added in schema v15, telling "tied to one user"
    /// (`user_count == 1`) apart from "spread across users"
    /// (`user_count > 1`) or "no user metadata" (`user_count == 0`).
    /// Follows the same chunk-scope rule as `chunk_count` rather than
    /// `session_count`: under a `session_id` filter the count reflects
    /// distinct users among the neighbor's chunks *inside* that session
    /// and does NOT collapse to 1, since a single session may legitimately
    /// span multiple users (e.g. a shared agent thread).
    pub user_count: i64,
    /// Sample of chunks shared by the source entity and this neighbor.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub chunks: Option<Vec<EntityChunkRef>>,
    /// Sample of distinct `session_id` values drawn from the chunks shared by
    /// the source entity and this neighbor. `Some(_)` only when
    /// `EntityNeighborsOptions::with_sessions` was set to `Some(n > 0)`;
    /// `None` keeps the default wire shape unchanged for callers that did not
    /// opt in. Ordered ascending by `session_id` for determinism. Under a
    /// `session_id` filter the sample collapses to that single id by
    /// construction.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub shared_session_ids: Option<Vec<String>>,
    /// Sample of distinct non-NULL `chunks.project` values drawn from the
    /// chunks shared by the source entity and this neighbor. `Some(_)` only
    /// when `EntityNeighborsOptions::with_projects` was set to `Some(n > 0)`;
    /// `None` keeps the default wire shape unchanged for callers that did not
    /// opt in. Ordered ascending by `project` for determinism. Under a
    /// `session_id` filter the sample is scoped to that session's chunks but
    /// does NOT collapse to one project — a session may legitimately span
    /// multiple projects.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub shared_projects: Option<Vec<String>>,
    /// Sample of distinct non-NULL `chunks.user` values drawn from the chunks
    /// shared by the source entity and this neighbor. `Some(_)` only when
    /// `EntityNeighborsOptions::with_users` was set to `Some(n > 0)`; `None`
    /// keeps the default wire shape unchanged for callers that did not opt in.
    /// Ordered ascending by `user` for determinism. Under a `session_id`
    /// filter the sample is scoped to that session's chunks but does NOT
    /// collapse to one user — a session may legitimately span multiple users.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub shared_users: Option<Vec<String>>,
}

/// Result of an [`entity_neighbors`] query: the source entity's identity plus
/// the (possibly truncated) list of co-occurring entities.
#[derive(Debug, Clone, Serialize)]
pub struct EntityNeighborsReport {
    pub source_id: String,
    pub source_kind: EntityKind,
    pub source_value: String,
    pub neighbors: Vec<EntityNeighbor>,
    pub total_neighbors: i64,
    pub kind_filter: Option<EntityKind>,
    /// Echoes the `session_id` filter (if any) so JSON consumers can confirm
    /// the traversal is session-scoped without re-deriving from the counts.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
}

/// Find entities that share at least one chunk with `source_id`, ranked by the
/// number of chunks shared (descending), with `value` then `id` as deterministic
/// tie-breakers. The source entity itself is excluded from the results.
///
/// Returns `Err` when no entity with `source_id` exists, so callers (CLI / MCP)
/// can surface a clear "unknown entity" message instead of an empty result.
pub fn entity_neighbors(
    conn: &Connection,
    source_id: &str,
    opts: &EntityNeighborsOptions,
) -> Result<EntityNeighborsReport> {
    let (source_kind_raw, source_value) = conn
        .query_row(
            "SELECT kind, value FROM entities WHERE id = ?1",
            params![source_id],
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
        )
        .map_err(|err| match err {
            rusqlite::Error::QueryReturnedNoRows => {
                anyhow::anyhow!("no entity with id '{source_id}'")
            }
            other => anyhow::Error::from(other).context(format!("looking up entity {source_id}")),
        })?;
    let source_kind = entity_kind_from_str(&source_kind_raw)?;

    let kind_str = opts.kind.map(|k| k.as_str().to_string());
    let session_str = opts.session_id.as_deref();

    // Position assignments for `?N` placeholders. ?1 is always the source id.
    // session, kind, and limit slot in afterwards in this fixed order so the
    // chunk_count subquery can reuse the session position by index.
    let session_pos: Option<usize> = session_str.map(|_| 2);
    let kind_pos: Option<usize> = if kind_str.is_some() {
        Some(if session_pos.is_some() { 3 } else { 2 })
    } else {
        None
    };
    let limit_pos: usize = match (session_pos.is_some(), kind_pos.is_some()) {
        (true, true) => 4,
        (true, false) => 3,
        (false, true) => 3,
        (false, false) => 2,
    };

    let session_join = if session_pos.is_some() {
        " JOIN chunks c ON c.id = ce_other.chunk_id "
    } else {
        ""
    };
    let session_where = match session_pos {
        Some(p) => format!(" AND c.session_id = ?{p}"),
        None => String::new(),
    };
    let kind_clause = match kind_pos {
        Some(p) => format!(" AND e.kind = ?{p}"),
        None => String::new(),
    };
    // The neighbor's `chunk_count` follows the same scope as the listing: when
    // session-scoped it counts chunks of the neighbor *inside* that session, so
    // ranking and "specifically related vs popular everywhere" stays meaningful
    // within the narrowed view.
    let chunk_count_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT COUNT(*) FROM chunk_entities ce2
                JOIN chunks c2 ON c2.id = ce2.chunk_id
                WHERE ce2.entity_id = e.id AND c2.session_id = ?{p})"
        ),
        None => "(SELECT COUNT(*) FROM chunk_entities ce2 WHERE ce2.entity_id = e.id)".to_string(),
    };
    // The neighbor's `session_count` mirrors the chunk_count scoping rule and
    // matches `entity_session_neighbors`' semantics: count distinct non-NULL
    // session_ids the neighbor appears in. When session-scoped this collapses
    // to 1 by construction (every surviving neighbor appears in `?{p}`).
    // Chunks with NULL session_id are excluded so untagged ingest cannot
    // inflate the spread signal.
    let session_count_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT COUNT(DISTINCT c2.session_id) FROM chunk_entities ce2
                JOIN chunks c2 ON c2.id = ce2.chunk_id
                WHERE ce2.entity_id = e.id AND c2.session_id = ?{p})"
        ),
        None => "(SELECT COUNT(DISTINCT c2.session_id) FROM chunk_entities ce2
                JOIN chunks c2 ON c2.id = ce2.chunk_id
                WHERE ce2.entity_id = e.id AND c2.session_id IS NOT NULL)"
            .to_string(),
    };
    // The neighbor's `project_count` follows the chunk_count scoping rule
    // (not session_count's): under a `session_id` filter it still counts
    // distinct non-NULL projects, just restricted to the neighbor's chunks
    // *inside* that session. That keeps project grouping orthogonal to
    // session grouping — a session-scoped view can still span multiple
    // projects, so the count must not collapse to 1. Chunks with NULL
    // project are excluded so untagged ingest cannot inflate the spread.
    let project_count_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT COUNT(DISTINCT c2.project) FROM chunk_entities ce2
                JOIN chunks c2 ON c2.id = ce2.chunk_id
                WHERE ce2.entity_id = e.id AND c2.session_id = ?{p}
                  AND c2.project IS NOT NULL)"
        ),
        None => "(SELECT COUNT(DISTINCT c2.project) FROM chunk_entities ce2
                JOIN chunks c2 ON c2.id = ce2.chunk_id
                WHERE ce2.entity_id = e.id AND c2.project IS NOT NULL)"
            .to_string(),
    };
    // The neighbor's `user_count` mirrors `project_count`'s rule: orthogonal
    // to the session filter, so a session-scoped view can still surface a
    // neighbor spanning multiple users. Under `session_id` it counts distinct
    // non-NULL users among the neighbor's chunks *inside* that session
    // (does NOT collapse to 1). Chunks with NULL user are excluded so
    // untagged ingest cannot inflate the spread.
    let user_count_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT COUNT(DISTINCT c2.user) FROM chunk_entities ce2
                JOIN chunks c2 ON c2.id = ce2.chunk_id
                WHERE ce2.entity_id = e.id AND c2.session_id = ?{p}
                  AND c2.user IS NOT NULL)"
        ),
        None => "(SELECT COUNT(DISTINCT c2.user) FROM chunk_entities ce2
                JOIN chunks c2 ON c2.id = ce2.chunk_id
                WHERE ce2.entity_id = e.id AND c2.user IS NOT NULL)"
            .to_string(),
    };

    let total_sql = format!(
        "SELECT COUNT(*) FROM (
             SELECT e.id
             FROM entities e
             JOIN chunk_entities ce_other ON ce_other.entity_id = e.id
             JOIN chunk_entities ce_self  ON ce_self.chunk_id  = ce_other.chunk_id
             {session_join}
             WHERE ce_self.entity_id = ?1
               AND e.id != ?1{session_where}{kind_clause}
             GROUP BY e.id
         )"
    );
    let list_sql = format!(
        "SELECT e.id, e.kind, e.value,
                COUNT(DISTINCT ce_other.chunk_id) AS shared_chunks,
                {chunk_count_subquery} AS chunk_count,
                {session_count_subquery} AS session_count,
                {project_count_subquery} AS project_count,
                {user_count_subquery} AS user_count
         FROM entities e
         JOIN chunk_entities ce_other ON ce_other.entity_id = e.id
         JOIN chunk_entities ce_self  ON ce_self.chunk_id  = ce_other.chunk_id
         {session_join}
         WHERE ce_self.entity_id = ?1
           AND e.id != ?1{session_where}{kind_clause}
         GROUP BY e.id
         ORDER BY shared_chunks DESC, e.value ASC, e.id ASC
         LIMIT ?{limit_pos}"
    );

    let mut total_params: Vec<rusqlite::types::Value> = vec![source_id.to_string().into()];
    if let Some(s) = session_str {
        total_params.push(s.to_string().into());
    }
    if let Some(k) = &kind_str {
        total_params.push(k.clone().into());
    }
    let total_param_refs: Vec<&dyn rusqlite::ToSql> = total_params
        .iter()
        .map(|v| v as &dyn rusqlite::ToSql)
        .collect();
    let total_neighbors: i64 = conn
        .query_row(&total_sql, total_param_refs.as_slice(), |row| row.get(0))
        .with_context(|| format!("counting neighbors of {source_id}"))?;

    let limit = opts.limit.unwrap_or(usize::MAX).min(i64::MAX as usize) as i64;
    let mut list_params = total_params.clone();
    list_params.push(limit.into());
    let list_param_refs: Vec<&dyn rusqlite::ToSql> = list_params
        .iter()
        .map(|v| v as &dyn rusqlite::ToSql)
        .collect();

    let mut stmt = conn
        .prepare(&list_sql)
        .context("preparing entity_neighbors query")?;
    let rows = stmt
        .query_map(list_param_refs.as_slice(), |row| {
            let kind_raw: String = row.get(1)?;
            let kind = entity_kind_from_str(&kind_raw).map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    1,
                    rusqlite::types::Type::Text,
                    Box::new(std::io::Error::other(e.to_string())),
                )
            })?;
            Ok(EntityNeighbor {
                id: row.get(0)?,
                kind,
                value: row.get(2)?,
                edge_kind: EdgeKind::CoOccursWith,
                shared_chunks: row.get(3)?,
                chunk_count: row.get(4)?,
                session_count: row.get(5)?,
                project_count: row.get(6)?,
                user_count: row.get(7)?,
                chunks: None,
                shared_session_ids: None,
                shared_projects: None,
                shared_users: None,
            })
        })
        .context("running entity_neighbors query")?;
    let mut neighbors = rows.collect::<Result<Vec<_>, _>>()?;

    if let Some(per_neighbor_limit) = opts.with_chunks {
        for neighbor in &mut neighbors {
            neighbor.chunks = Some(load_shared_chunk_refs(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    if let Some(per_neighbor_limit) = opts.with_sessions.filter(|n| *n > 0) {
        for neighbor in &mut neighbors {
            neighbor.shared_session_ids = Some(load_shared_chunk_session_ids(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    if let Some(per_neighbor_limit) = opts.with_projects.filter(|n| *n > 0) {
        for neighbor in &mut neighbors {
            neighbor.shared_projects = Some(load_shared_chunk_projects(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    if let Some(per_neighbor_limit) = opts.with_users.filter(|n| *n > 0) {
        for neighbor in &mut neighbors {
            neighbor.shared_users = Some(load_shared_chunk_users(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    Ok(EntityNeighborsReport {
        source_id: source_id.to_string(),
        source_kind,
        source_value,
        neighbors,
        total_neighbors,
        kind_filter: opts.kind,
        session_id: opts.session_id.clone(),
    })
}

pub fn print_neighbors_text(report: &EntityNeighborsReport) {
    let mut scope_parts: Vec<String> = Vec::new();
    if let Some(k) = report.kind_filter {
        scope_parts.push(format!("kind={}", k.as_str()));
    }
    if let Some(s) = &report.session_id {
        scope_parts.push(format!("session_id={s}"));
    }
    let scope = if scope_parts.is_empty() {
        String::new()
    } else {
        format!(" ({})", scope_parts.join(", "))
    };
    println!(
        "{} ({}) — {} neighbor{}{}",
        report.source_value,
        report.source_kind.as_str(),
        report.total_neighbors,
        if report.total_neighbors == 1 { "" } else { "s" },
        scope,
    );
    if report.neighbors.is_empty() {
        println!("  (no co-occurring entities)");
        return;
    }
    for n in &report.neighbors {
        println!(
            "  {kind:<8} edge={edge:<16} shared={shared:<3} total={total:<4} sessions={sessions:<3} projects={projects:<3} users={users:<3} {value}",
            kind = n.kind.as_str(),
            edge = n.edge_kind.as_str(),
            shared = n.shared_chunks,
            total = n.chunk_count,
            sessions = n.session_count,
            projects = n.project_count,
            users = n.user_count,
            value = n.value,
        );
        if let Some(refs) = n.chunks.as_ref() {
            for chunk in refs {
                println!(
                    "    - {chunk_id} [{uri}] {snippet}",
                    chunk_id = chunk.chunk_id,
                    uri = chunk.source_uri,
                    snippet = chunk.snippet,
                );
            }
        }
        if let Some(sessions) = n.shared_session_ids.as_ref() {
            for session in sessions {
                println!("      session={session}");
            }
        }
        if let Some(projects) = n.shared_projects.as_ref() {
            for project in projects {
                println!("      project={project}");
            }
        }
        if let Some(users) = n.shared_users.as_ref() {
            for user in users {
                println!("      user={user}");
            }
        }
    }
}

pub fn print_neighbors_json(report: &EntityNeighborsReport) -> Result<()> {
    println!("{}", serde_json::to_string_pretty(report)?);
    Ok(())
}

/// Filter / paging options for [`entity_session_neighbors`].
///
/// `kind` restricts neighbors to a single entity kind. `limit` caps how many
/// neighbors are returned; `EntitySessionNeighborsReport::total_neighbors`
/// always reflects the unbounded count so callers can detect truncation.
///
/// `session_id` narrows the traversal to chunks tagged with the given
/// `chunks.session_id`. When set, the source must appear in that session for
/// any neighbors to surface, and every surviving neighbor must also appear in
/// that session. Both `source_session_count` and per-neighbor
/// `shared_sessions` / `session_count` collapse to 0 or 1 by construction
/// (the filter pins everything to a single session). Chunks whose
/// `session_id IS NULL` are excluded.
///
/// `with_sessions` opts into evidence sampling: when set to `Some(n > 0)`,
/// each neighbor carries up to `n` of the actual `session_id` values it shares
/// with the source, ordered ascending for determinism. When `None` or
/// `Some(0)` the cheap default shape is preserved and no extra queries run.
/// When a `session_id` filter is set, the sample collapses to that single
/// session id by construction.
///
/// `with_projects` mirrors `with_sessions` at the project grouping level:
/// when set to `Some(n > 0)`, each neighbor carries up to `n` distinct
/// non-NULL `chunks.project` values drawn from the chunks (containing the
/// source or the neighbor) within sessions shared by both, ordered ascending
/// for determinism. When `None` or `Some(0)` the cheap default shape is
/// preserved and no extra queries run. Independent of the `session_id`
/// filter — a single session can still span multiple projects, so the sample
/// is scoped to chunks inside that session but does NOT collapse to one
/// project.
///
/// `with_users` mirrors `with_projects` at the user grouping level introduced
/// by schema v15: when set to `Some(n > 0)`, each neighbor carries up to `n`
/// distinct non-NULL `chunks.user` values drawn from the chunks (containing
/// the source or the neighbor) within sessions shared by both, ordered
/// ascending for determinism. When `None` or `Some(0)` the cheap default
/// shape is preserved and no extra queries run. Independent of the
/// `session_id` filter — a single session can legitimately span multiple
/// users (e.g. a shared agent thread), so the sample is scoped to chunks
/// inside that session but does NOT collapse to one user.
///
/// `with_chunks` opts into chunk-level evidence for the same-session edge:
/// when set to `Some(n > 0)`, each neighbor carries up to `n`
/// `EntityChunkRef` rows naming concrete chunks that live inside a session
/// shared by both entities and reference either the source or the neighbor.
/// When `None` or `Some(0)` the cheap default shape is preserved and no
/// extra queries run. When a `session_id` filter is set the sample is
/// scoped to that single session's chunks. The chunk-level evidence is the
/// session-graph companion to chunk-level `entity_neighbors`'s
/// `with_chunks`, but the chunks need not contain *both* entities — they
/// only need to share a session — because the underlying edge is
/// `SameSessionAs`, not `CoOccursWith`.
#[derive(Debug, Clone, Default)]
pub struct EntitySessionNeighborsOptions {
    pub kind: Option<EntityKind>,
    pub limit: Option<usize>,
    pub session_id: Option<String>,
    pub with_sessions: Option<usize>,
    pub with_projects: Option<usize>,
    pub with_users: Option<usize>,
    pub with_chunks: Option<usize>,
}

/// One session-level neighbor of the source entity: an entity that appears in
/// at least one of the same sessions as the source, the typed edge kind that
/// links it (`SameSessionAs`), the number of distinct sessions shared with the
/// source, and that neighbor's overall session count.
#[derive(Debug, Clone, Serialize, PartialEq, Eq)]
pub struct EntitySessionNeighbor {
    pub id: String,
    pub kind: EntityKind,
    pub value: String,
    /// Stable relationship label for this graph edge.
    pub edge_kind: EdgeKind,
    /// Number of distinct sessions the neighbor appears in *together with* the
    /// source entity. Always >= 1.
    pub shared_sessions: i64,
    /// Total distinct sessions the neighbor appears in across the whole store
    /// (chunks with `session_id IS NULL` excluded). Useful for telling
    /// "specifically session-related" (low total, all shared) apart from
    /// "session-popular" (high total, mostly elsewhere).
    pub session_count: i64,
    /// Distinct upstream projects whose chunks reference this neighbor
    /// (chunks with `project IS NULL` excluded). Mirrors `project_count` on
    /// `lantern entities` and chunk-level `lantern entity-neighbors` at the
    /// session-graph layer. Independent of the `session_id` filter — a single
    /// session can still span multiple projects, so the count does NOT
    /// collapse to 1 under `--session-id`; it just narrows to the neighbor's
    /// chunks inside that session. `project_count == 0` distinguishes "no
    /// project metadata" from `project_count == 1` ("tied to one project")
    /// and `project_count > 1` ("spread across projects").
    pub project_count: i64,
    /// Distinct users whose chunks reference this neighbor (chunks with
    /// `user IS NULL` excluded). Mirrors `user_count` on `lantern entities`
    /// and chunk-level `lantern entity-neighbors` at the session-graph layer.
    /// Like `project_count`, it stays orthogonal to the `session_id` filter:
    /// a single session can legitimately span multiple users (e.g. a shared
    /// agent thread), so the count does NOT collapse to 1 under
    /// `--session-id`; it just narrows to the neighbor's chunks inside that
    /// session. `user_count == 0` distinguishes "no user metadata" from
    /// `user_count == 1` ("tied to one user") and `user_count > 1` ("spread
    /// across users").
    pub user_count: i64,
    /// Sample of `session_id` values this neighbor shares with the source
    /// entity. `Some(_)` only when `EntitySessionNeighborsOptions::with_sessions`
    /// was set to `Some(n > 0)`; `None` keeps the default wire shape unchanged
    /// for callers that did not opt in. Ordered ascending by `session_id`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub shared_session_ids: Option<Vec<String>>,
    /// Sample of distinct non-NULL `chunks.project` values drawn from the
    /// chunks (containing the source or the neighbor) within the sessions
    /// they share. `Some(_)` only when
    /// `EntitySessionNeighborsOptions::with_projects` was set to `Some(n > 0)`;
    /// `None` keeps the default wire shape unchanged for callers that did not
    /// opt in. Ordered ascending by `project` for determinism. Under a
    /// `session_id` filter the sample is scoped to that session's chunks but
    /// does NOT collapse to one project — a session may legitimately span
    /// multiple projects.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub shared_projects: Option<Vec<String>>,
    /// Sample of distinct non-NULL `chunks.user` values drawn from the chunks
    /// (containing the source or the neighbor) within the sessions they share.
    /// `Some(_)` only when `EntitySessionNeighborsOptions::with_users` was set
    /// to `Some(n > 0)`; `None` keeps the default wire shape unchanged for
    /// callers that did not opt in. Ordered ascending by `user` for
    /// determinism. Under a `session_id` filter the sample is scoped to that
    /// session's chunks but does NOT collapse to one user — a single session
    /// can legitimately span multiple users (e.g. a shared agent thread).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub shared_users: Option<Vec<String>>,
    /// Sample of `EntityChunkRef` rows drawn from chunks that live inside a
    /// session shared by both the source and this neighbor and that reference
    /// either entity. `Some(_)` only when
    /// `EntitySessionNeighborsOptions::with_chunks` was set to `Some(n > 0)`;
    /// `None` keeps the default wire shape unchanged for callers that did not
    /// opt in. Ordered by `chunk created_at ASC, id ASC` for determinism.
    /// Under a `session_id` filter the sample is scoped to that single
    /// session's chunks. Unlike chunk-level `entity_neighbors` evidence, a
    /// sampled chunk need not contain *both* entities — it only needs to live
    /// inside a shared session and reference at least one of them — because
    /// the underlying edge is `SameSessionAs`, not `CoOccursWith`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub chunks: Option<Vec<EntityChunkRef>>,
}

/// Result of an [`entity_session_neighbors`] query: the source entity's
/// identity plus the (possibly truncated) list of session-co-occurring entities.
#[derive(Debug, Clone, Serialize)]
pub struct EntitySessionNeighborsReport {
    pub source_id: String,
    pub source_kind: EntityKind,
    pub source_value: String,
    /// Number of distinct sessions the source entity itself appears in. When
    /// zero, no neighbors are returned by construction. Scoped to the
    /// `session_id` filter when set — i.e. 1 if the source appears in that
    /// session, else 0.
    pub source_session_count: i64,
    pub neighbors: Vec<EntitySessionNeighbor>,
    pub total_neighbors: i64,
    pub kind_filter: Option<EntityKind>,
    /// Echoes the `session_id` filter (if any) so JSON consumers can confirm
    /// the traversal is session-scoped without re-deriving from the counts.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub session_id: Option<String>,
}

/// Find entities that share at least one session with `source_id`, ranked by
/// the number of sessions shared (descending), with `value` then `id` as
/// deterministic tie-breakers. The source entity itself is excluded.
///
/// Unlike [`entity_neighbors`], two entities can be neighbors here without
/// ever sharing a chunk — it is enough that both appear somewhere inside the
/// same session. Chunks whose `session_id IS NULL` are skipped on both sides,
/// so untagged ingest cannot create phantom session edges.
///
/// Returns `Err` when no entity with `source_id` exists, so callers (CLI / MCP)
/// can surface a clear "unknown entity" message instead of an empty result.
pub fn entity_session_neighbors(
    conn: &Connection,
    source_id: &str,
    opts: &EntitySessionNeighborsOptions,
) -> Result<EntitySessionNeighborsReport> {
    let (source_kind_raw, source_value) = conn
        .query_row(
            "SELECT kind, value FROM entities WHERE id = ?1",
            params![source_id],
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
        )
        .map_err(|err| match err {
            rusqlite::Error::QueryReturnedNoRows => {
                anyhow::anyhow!("no entity with id '{source_id}'")
            }
            other => anyhow::Error::from(other).context(format!("looking up entity {source_id}")),
        })?;
    let source_kind = entity_kind_from_str(&source_kind_raw)?;

    let kind_str = opts.kind.map(|k| k.as_str().to_string());
    let session_str = opts.session_id.as_deref();

    // Distinct session_ids the source entity appears in. We compute this up
    // front so the count is available in the report even when no neighbors
    // are returned (e.g. the source only appears in NULL-session chunks).
    // When a `session_id` filter is set, the count is scoped to that session
    // and collapses to 0 or 1 — the rest of the traversal will be empty
    // unless the source actually appears in that session.
    let source_session_count: i64 = match session_str {
        Some(s) => conn
            .query_row(
                "SELECT COUNT(DISTINCT c.session_id)
                 FROM chunk_entities ce
                 JOIN chunks c ON c.id = ce.chunk_id
                 WHERE ce.entity_id = ?1 AND c.session_id = ?2",
                params![source_id, s],
                |row| row.get(0),
            )
            .with_context(|| format!("counting sessions for entity {source_id}"))?,
        None => conn
            .query_row(
                "SELECT COUNT(DISTINCT c.session_id)
                 FROM chunk_entities ce
                 JOIN chunks c ON c.id = ce.chunk_id
                 WHERE ce.entity_id = ?1 AND c.session_id IS NOT NULL",
                params![source_id],
                |row| row.get(0),
            )
            .with_context(|| format!("counting sessions for entity {source_id}"))?,
    };

    // Position assignments for `?N` placeholders. ?1 is always the source id.
    // session, kind, and limit slot in afterwards in this fixed order so the
    // session_count subquery can reuse the session position by index.
    let session_pos: Option<usize> = session_str.map(|_| 2);
    let kind_pos: Option<usize> = if kind_str.is_some() {
        Some(if session_pos.is_some() { 3 } else { 2 })
    } else {
        None
    };
    let limit_pos: usize = match (session_pos.is_some(), kind_pos.is_some()) {
        (true, true) => 4,
        (true, false) => 3,
        (false, true) => 3,
        (false, false) => 2,
    };

    let kind_clause = match kind_pos {
        Some(p) => format!(" AND e.kind = ?{p}"),
        None => String::new(),
    };

    // When a `session_id` filter is set, the source's "shared sessions" set
    // narrows to either {filter} (if the source is in it) or {} (if not).
    // Both `shared_sessions_subquery` and the outer scope inherit the same
    // restriction, so the traversal stays anchored to that single session.
    let shared_sessions_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT DISTINCT cs.session_id
              FROM chunks cs
              JOIN chunk_entities ces ON ces.chunk_id = cs.id
              WHERE ces.entity_id = ?1 AND cs.session_id = ?{p})"
        ),
        None => "(SELECT DISTINCT cs.session_id
              FROM chunks cs
              JOIN chunk_entities ces ON ces.chunk_id = cs.id
              WHERE ces.entity_id = ?1 AND cs.session_id IS NOT NULL)"
            .to_string(),
    };

    // The neighbor's `session_count` follows the same scope as the listing:
    // when session-scoped it counts distinct sessions of the neighbor *inside*
    // that one session, which collapses to 1 by construction. That keeps the
    // contract consistent — every count in the report reflects the requested
    // scope rather than a hidden global value.
    let session_count_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT COUNT(DISTINCT c2.session_id)
              FROM chunk_entities ce2
              JOIN chunks c2 ON c2.id = ce2.chunk_id
              WHERE ce2.entity_id = e.id AND c2.session_id = ?{p})"
        ),
        None => "(SELECT COUNT(DISTINCT c2.session_id)
              FROM chunk_entities ce2
              JOIN chunks c2 ON c2.id = ce2.chunk_id
              WHERE ce2.entity_id = e.id AND c2.session_id IS NOT NULL)"
            .to_string(),
    };
    // The neighbor's `project_count` stays orthogonal to the session scope:
    // under a `session_id` filter it still counts distinct non-NULL projects,
    // just restricted to the neighbor's chunks *inside* that session. A single
    // session can span multiple projects, so the spread signal must not
    // collapse to 1 the way `session_count` does. Chunks with NULL project
    // are excluded so untagged ingest cannot inflate the spread.
    let project_count_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT COUNT(DISTINCT c2.project)
              FROM chunk_entities ce2
              JOIN chunks c2 ON c2.id = ce2.chunk_id
              WHERE ce2.entity_id = e.id AND c2.session_id = ?{p}
                AND c2.project IS NOT NULL)"
        ),
        None => "(SELECT COUNT(DISTINCT c2.project)
              FROM chunk_entities ce2
              JOIN chunks c2 ON c2.id = ce2.chunk_id
              WHERE ce2.entity_id = e.id AND c2.project IS NOT NULL)"
            .to_string(),
    };
    // The neighbor's `user_count` mirrors `project_count`'s rule at the user
    // grouping level: orthogonal to the session filter, so a session-scoped
    // view can still surface a neighbor spanning multiple users. Under
    // `--session-id` it counts distinct non-NULL users among the neighbor's
    // chunks *inside* that session (does NOT collapse to 1). Chunks with NULL
    // user are excluded so untagged ingest cannot inflate the spread.
    let user_count_subquery = match session_pos {
        Some(p) => format!(
            "(SELECT COUNT(DISTINCT c2.user)
              FROM chunk_entities ce2
              JOIN chunks c2 ON c2.id = ce2.chunk_id
              WHERE ce2.entity_id = e.id AND c2.session_id = ?{p}
                AND c2.user IS NOT NULL)"
        ),
        None => "(SELECT COUNT(DISTINCT c2.user)
              FROM chunk_entities ce2
              JOIN chunks c2 ON c2.id = ce2.chunk_id
              WHERE ce2.entity_id = e.id AND c2.user IS NOT NULL)"
            .to_string(),
    };

    let session_where = match session_pos {
        Some(p) => format!(" AND c.session_id = ?{p}"),
        None => String::new(),
    };

    let total_sql = format!(
        "SELECT COUNT(*) FROM (
             SELECT e.id
             FROM entities e
             JOIN chunk_entities ce ON ce.entity_id = e.id
             JOIN chunks c ON c.id = ce.chunk_id
             WHERE e.id != ?1
               AND c.session_id IS NOT NULL{session_where}
               AND c.session_id IN {shared_sessions_subquery}{kind_clause}
             GROUP BY e.id
         )"
    );

    let list_sql = format!(
        "SELECT e.id, e.kind, e.value,
                COUNT(DISTINCT c.session_id) AS shared_sessions,
                {session_count_subquery} AS session_count,
                {project_count_subquery} AS project_count,
                {user_count_subquery} AS user_count
         FROM entities e
         JOIN chunk_entities ce ON ce.entity_id = e.id
         JOIN chunks c ON c.id = ce.chunk_id
         WHERE e.id != ?1
           AND c.session_id IS NOT NULL{session_where}
           AND c.session_id IN {shared_sessions_subquery}{kind_clause}
         GROUP BY e.id
         ORDER BY shared_sessions DESC, e.value ASC, e.id ASC
         LIMIT ?{limit_pos}"
    );

    let mut total_params: Vec<rusqlite::types::Value> = vec![source_id.to_string().into()];
    if let Some(s) = session_str {
        total_params.push(s.to_string().into());
    }
    if let Some(k) = &kind_str {
        total_params.push(k.clone().into());
    }
    let total_param_refs: Vec<&dyn rusqlite::ToSql> = total_params
        .iter()
        .map(|v| v as &dyn rusqlite::ToSql)
        .collect();
    let total_neighbors: i64 = conn
        .query_row(&total_sql, total_param_refs.as_slice(), |row| row.get(0))
        .with_context(|| format!("counting session-neighbors of {source_id}"))?;

    let limit = opts.limit.unwrap_or(usize::MAX).min(i64::MAX as usize) as i64;
    let mut list_params = total_params.clone();
    list_params.push(limit.into());
    let list_param_refs: Vec<&dyn rusqlite::ToSql> = list_params
        .iter()
        .map(|v| v as &dyn rusqlite::ToSql)
        .collect();

    let mut stmt = conn
        .prepare(&list_sql)
        .context("preparing entity_session_neighbors query")?;
    let rows = stmt
        .query_map(list_param_refs.as_slice(), |row| {
            let kind_raw: String = row.get(1)?;
            let kind = entity_kind_from_str(&kind_raw).map_err(|e| {
                rusqlite::Error::FromSqlConversionFailure(
                    1,
                    rusqlite::types::Type::Text,
                    Box::new(std::io::Error::other(e.to_string())),
                )
            })?;
            Ok(EntitySessionNeighbor {
                id: row.get(0)?,
                kind,
                value: row.get(2)?,
                edge_kind: EdgeKind::SameSessionAs,
                shared_sessions: row.get(3)?,
                session_count: row.get(4)?,
                project_count: row.get(5)?,
                user_count: row.get(6)?,
                shared_session_ids: None,
                shared_projects: None,
                shared_users: None,
                chunks: None,
            })
        })
        .context("running entity_session_neighbors query")?;
    let mut neighbors = rows.collect::<Result<Vec<_>, _>>()?;

    if let Some(per_neighbor_limit) = opts.with_sessions.filter(|n| *n > 0) {
        for neighbor in &mut neighbors {
            neighbor.shared_session_ids = Some(load_shared_session_ids(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    if let Some(per_neighbor_limit) = opts.with_projects.filter(|n| *n > 0) {
        for neighbor in &mut neighbors {
            neighbor.shared_projects = Some(load_shared_session_projects(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    if let Some(per_neighbor_limit) = opts.with_users.filter(|n| *n > 0) {
        for neighbor in &mut neighbors {
            neighbor.shared_users = Some(load_shared_session_users(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    if let Some(per_neighbor_limit) = opts.with_chunks.filter(|n| *n > 0) {
        for neighbor in &mut neighbors {
            neighbor.chunks = Some(load_shared_session_chunk_refs(
                conn,
                source_id,
                &neighbor.id,
                per_neighbor_limit,
                session_str,
            )?);
        }
    }

    Ok(EntitySessionNeighborsReport {
        source_id: source_id.to_string(),
        source_kind,
        source_value,
        source_session_count,
        neighbors,
        total_neighbors,
        kind_filter: opts.kind,
        session_id: opts.session_id.clone(),
    })
}

/// Sample up to `limit` distinct `session_id` values whose chunks reference
/// `entity_id`. Always orders ascending so the evidence is deterministic
/// across runs. Sessions with `NULL` ids are excluded; when `session_filter`
/// is set the result collapses to at most one session id by construction.
/// `limit == 0` short-circuits so opt-in callers stay on the cheap path.
/// Sample up to `limit` distinct non-NULL `chunks.project` values whose chunks
/// reference `entity_id`, ordered ascending. `limit == 0` short-circuits to an
/// empty vec so opt-in callers stay on the cheap path. When `session_filter`
/// is set, the sample is scoped to chunks inside that session but is NOT
/// collapsed to a single project — a session can legitimately span multiple
/// projects, mirroring `project_count`'s orthogonality to the session filter.
fn load_entity_projects(
    conn: &Connection,
    entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.project
                 FROM chunk_entities ce
                 JOIN chunks c ON c.id = ce.chunk_id
                 WHERE ce.entity_id = ?1
                   AND c.session_id = ?2
                   AND c.project IS NOT NULL
                 ORDER BY c.project ASC
                 LIMIT ?3",
            )
            .context("preparing session-scoped entity-projects query")?;
        let rows = stmt
            .query_map(params![entity_id, session, limit], |row| {
                row.get::<_, String>(0)
            })
            .context("running session-scoped entity-projects query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.project
             FROM chunk_entities ce
             JOIN chunks c ON c.id = ce.chunk_id
             WHERE ce.entity_id = ?1
               AND c.project IS NOT NULL
             ORDER BY c.project ASC
             LIMIT ?2",
        )
        .context("preparing entity-projects query")?;
    let rows = stmt
        .query_map(params![entity_id, limit], |row| row.get::<_, String>(0))
        .context("running entity-projects query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` distinct non-NULL `chunks.user` values whose chunks
/// reference `entity_id`, ordered ascending. `limit == 0` short-circuits to an
/// empty vec so opt-in callers stay on the cheap path. When `session_filter`
/// is set, the sample is scoped to chunks inside that session but is NOT
/// collapsed to a single user — a session can legitimately span multiple
/// users (e.g. a shared agent thread), mirroring `user_count`'s orthogonality
/// to the session filter.
fn load_entity_users(
    conn: &Connection,
    entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.user
                 FROM chunk_entities ce
                 JOIN chunks c ON c.id = ce.chunk_id
                 WHERE ce.entity_id = ?1
                   AND c.session_id = ?2
                   AND c.user IS NOT NULL
                 ORDER BY c.user ASC
                 LIMIT ?3",
            )
            .context("preparing session-scoped entity-users query")?;
        let rows = stmt
            .query_map(params![entity_id, session, limit], |row| {
                row.get::<_, String>(0)
            })
            .context("running session-scoped entity-users query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.user
             FROM chunk_entities ce
             JOIN chunks c ON c.id = ce.chunk_id
             WHERE ce.entity_id = ?1
               AND c.user IS NOT NULL
             ORDER BY c.user ASC
             LIMIT ?2",
        )
        .context("preparing entity-users query")?;
    let rows = stmt
        .query_map(params![entity_id, limit], |row| row.get::<_, String>(0))
        .context("running entity-users query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` distinct non-NULL `chunks.topic` values whose chunks
/// reference `entity_id`, ordered ascending. `limit == 0` short-circuits to an
/// empty vec so opt-in callers stay on the cheap path. When `session_filter`
/// is set, the sample is scoped to chunks inside that session but is NOT
/// collapsed to a single topic — a session can legitimately span multiple
/// topics, mirroring `topic_count`'s orthogonality to the session filter.
fn load_entity_topics(
    conn: &Connection,
    entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.topic
                 FROM chunk_entities ce
                 JOIN chunks c ON c.id = ce.chunk_id
                 WHERE ce.entity_id = ?1
                   AND c.session_id = ?2
                   AND c.topic IS NOT NULL
                 ORDER BY c.topic ASC
                 LIMIT ?3",
            )
            .context("preparing session-scoped entity-topics query")?;
        let rows = stmt
            .query_map(params![entity_id, session, limit], |row| {
                row.get::<_, String>(0)
            })
            .context("running session-scoped entity-topics query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.topic
             FROM chunk_entities ce
             JOIN chunks c ON c.id = ce.chunk_id
             WHERE ce.entity_id = ?1
               AND c.topic IS NOT NULL
             ORDER BY c.topic ASC
             LIMIT ?2",
        )
        .context("preparing entity-topics query")?;
    let rows = stmt
        .query_map(params![entity_id, limit], |row| row.get::<_, String>(0))
        .context("running entity-topics query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

fn load_entity_session_ids(
    conn: &Connection,
    entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.session_id
                 FROM chunk_entities ce
                 JOIN chunks c ON c.id = ce.chunk_id
                 WHERE ce.entity_id = ?1
                   AND c.session_id = ?2
                 ORDER BY c.session_id ASC
                 LIMIT ?3",
            )
            .context("preparing session-scoped entity-session-ids query")?;
        let rows = stmt
            .query_map(params![entity_id, session, limit], |row| {
                row.get::<_, String>(0)
            })
            .context("running session-scoped entity-session-ids query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.session_id
             FROM chunk_entities ce
             JOIN chunks c ON c.id = ce.chunk_id
             WHERE ce.entity_id = ?1
               AND c.session_id IS NOT NULL
             ORDER BY c.session_id ASC
             LIMIT ?2",
        )
        .context("preparing entity-session-ids query")?;
    let rows = stmt
        .query_map(params![entity_id, limit], |row| row.get::<_, String>(0))
        .context("running entity-session-ids query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` distinct `session_id` values shared by both the source
/// entity and a neighbor. Always orders ascending so the evidence is
/// deterministic across runs. When `session_id` is set, the result collapses
/// to at most one session id by construction. `limit == 0` short-circuits so
/// opt-in callers stay on the cheap path.
fn load_shared_session_ids(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT cs.session_id
                 FROM chunks cs
                 JOIN chunk_entities ces ON ces.chunk_id = cs.id
                 WHERE ces.entity_id = ?1
                   AND cs.session_id = ?2
                   AND cs.session_id IN (
                       SELECT DISTINCT cn.session_id
                       FROM chunks cn
                       JOIN chunk_entities cen ON cen.chunk_id = cn.id
                       WHERE cen.entity_id = ?3
                         AND cn.session_id = ?2
                   )
                 ORDER BY cs.session_id ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-session-ids query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, session, neighbor_entity_id, limit],
                |row| row.get::<_, String>(0),
            )
            .context("running session-scoped shared-session-ids query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT cs.session_id
             FROM chunks cs
             JOIN chunk_entities ces ON ces.chunk_id = cs.id
             WHERE ces.entity_id = ?1
               AND cs.session_id IS NOT NULL
               AND cs.session_id IN (
                   SELECT DISTINCT cn.session_id
                   FROM chunks cn
                   JOIN chunk_entities cen ON cen.chunk_id = cn.id
                   WHERE cen.entity_id = ?2
                     AND cn.session_id IS NOT NULL
               )
             ORDER BY cs.session_id ASC
             LIMIT ?3",
        )
        .context("preparing shared-session-ids query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit],
            |row| row.get::<_, String>(0),
        )
        .context("running shared-session-ids query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` distinct non-NULL `chunks.project` values drawn from
/// chunks (containing the source or the neighbor) that live inside sessions
/// shared by both entities. Always orders ascending so the evidence is
/// deterministic across runs. `limit == 0` short-circuits so opt-in callers
/// stay on the cheap path. When `session_filter` is set, the sample is scoped
/// to that session's chunks but is NOT collapsed to a single project — a
/// session may legitimately span multiple projects.
fn load_shared_session_projects(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.project
                 FROM chunks c
                 JOIN chunk_entities ce ON ce.chunk_id = c.id
                 WHERE ce.entity_id IN (?1, ?2)
                   AND c.session_id = ?3
                   AND c.project IS NOT NULL
                 ORDER BY c.project ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-session-projects query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, neighbor_entity_id, session, limit],
                |row| row.get::<_, String>(0),
            )
            .context("running session-scoped shared-session-projects query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.project
             FROM chunks c
             JOIN chunk_entities ce ON ce.chunk_id = c.id
             WHERE ce.entity_id IN (?1, ?2)
               AND c.session_id IS NOT NULL
               AND c.project IS NOT NULL
               AND c.session_id IN (
                   SELECT DISTINCT cs.session_id
                   FROM chunks cs
                   JOIN chunk_entities ces ON ces.chunk_id = cs.id
                   WHERE ces.entity_id = ?1 AND cs.session_id IS NOT NULL
                   INTERSECT
                   SELECT DISTINCT cn.session_id
                   FROM chunks cn
                   JOIN chunk_entities cen ON cen.chunk_id = cn.id
                   WHERE cen.entity_id = ?2 AND cn.session_id IS NOT NULL
               )
             ORDER BY c.project ASC
             LIMIT ?3",
        )
        .context("preparing shared-session-projects query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit],
            |row| row.get::<_, String>(0),
        )
        .context("running shared-session-projects query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` distinct non-NULL `chunks.user` values drawn from
/// chunks (containing the source or the neighbor) that live inside sessions
/// shared by both entities. Always orders ascending so the evidence is
/// deterministic across runs. `limit == 0` short-circuits so opt-in callers
/// stay on the cheap path. When `session_filter` is set, the sample is scoped
/// to that session's chunks but is NOT collapsed to a single user — a single
/// session can legitimately span multiple users (e.g. a shared agent thread),
/// mirroring `user_count`'s orthogonality to the session filter.
fn load_shared_session_users(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<String>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT DISTINCT c.user
                 FROM chunks c
                 JOIN chunk_entities ce ON ce.chunk_id = c.id
                 WHERE ce.entity_id IN (?1, ?2)
                   AND c.session_id = ?3
                   AND c.user IS NOT NULL
                 ORDER BY c.user ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-session-users query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, neighbor_entity_id, session, limit],
                |row| row.get::<_, String>(0),
            )
            .context("running session-scoped shared-session-users query")?;
        return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
    }
    let mut stmt = conn
        .prepare(
            "SELECT DISTINCT c.user
             FROM chunks c
             JOIN chunk_entities ce ON ce.chunk_id = c.id
             WHERE ce.entity_id IN (?1, ?2)
               AND c.session_id IS NOT NULL
               AND c.user IS NOT NULL
               AND c.session_id IN (
                   SELECT DISTINCT cs.session_id
                   FROM chunks cs
                   JOIN chunk_entities ces ON ces.chunk_id = cs.id
                   WHERE ces.entity_id = ?1 AND cs.session_id IS NOT NULL
                   INTERSECT
                   SELECT DISTINCT cn.session_id
                   FROM chunks cn
                   JOIN chunk_entities cen ON cen.chunk_id = cn.id
                   WHERE cen.entity_id = ?2 AND cn.session_id IS NOT NULL
               )
             ORDER BY c.user ASC
             LIMIT ?3",
        )
        .context("preparing shared-session-users query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit],
            |row| row.get::<_, String>(0),
        )
        .context("running shared-session-users query")?;
    rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
}

/// Sample up to `limit` chunk references drawn from chunks that (a) live inside
/// a session shared by both the source and a neighbor and (b) reference either
/// of the two entities. Mirrors the chunk-evidence shape used by
/// `entity_neighbors`'s `with_chunks`, but loosened so a sampled chunk need
/// not contain *both* entities — the underlying edge is `SameSessionAs`, not
/// `CoOccursWith`. Ordered by `chunks.created_at ASC, id ASC` for determinism.
/// When `session_filter` is set the sample is scoped to that one session;
/// otherwise the shared-session intersection of both entities narrows the
/// chunk pool. `limit == 0` short-circuits so opt-in callers stay on the
/// cheap path.
fn load_shared_session_chunk_refs(
    conn: &Connection,
    source_entity_id: &str,
    neighbor_entity_id: &str,
    limit: usize,
    session_filter: Option<&str>,
) -> Result<Vec<EntityChunkRef>> {
    if limit == 0 {
        return Ok(Vec::new());
    }
    let limit = limit.min(i64::MAX as usize) as i64;
    if let Some(session) = session_filter {
        let mut stmt = conn
            .prepare(
                "SELECT c.id, c.source_id, s.uri, c.text
                 FROM chunks c
                 JOIN sources s ON s.id = c.source_id
                 JOIN chunk_entities ce ON ce.chunk_id = c.id
                 WHERE ce.entity_id IN (?1, ?2)
                   AND c.session_id = ?3
                 GROUP BY c.id
                 ORDER BY c.created_at ASC, c.id ASC
                 LIMIT ?4",
            )
            .context("preparing session-scoped shared-session-chunks query")?;
        let rows = stmt
            .query_map(
                params![source_entity_id, neighbor_entity_id, session, limit],
                |row| build_chunk_ref(row),
            )
            .context("running session-scoped shared-session-chunks query")?;
        return Ok(rows.collect::<Result<Vec<_>, _>>()?);
    }
    let mut stmt = conn
        .prepare(
            "SELECT c.id, c.source_id, s.uri, c.text
             FROM chunks c
             JOIN sources s ON s.id = c.source_id
             JOIN chunk_entities ce ON ce.chunk_id = c.id
             WHERE ce.entity_id IN (?1, ?2)
               AND c.session_id IS NOT NULL
               AND c.session_id IN (
                   SELECT DISTINCT cs.session_id
                   FROM chunks cs
                   JOIN chunk_entities ces ON ces.chunk_id = cs.id
                   WHERE ces.entity_id = ?1 AND cs.session_id IS NOT NULL
                   INTERSECT
                   SELECT DISTINCT cn.session_id
                   FROM chunks cn
                   JOIN chunk_entities cen ON cen.chunk_id = cn.id
                   WHERE cen.entity_id = ?2 AND cn.session_id IS NOT NULL
               )
             GROUP BY c.id
             ORDER BY c.created_at ASC, c.id ASC
             LIMIT ?3",
        )
        .context("preparing shared-session-chunks query")?;
    let rows = stmt
        .query_map(
            params![source_entity_id, neighbor_entity_id, limit],
            |row| build_chunk_ref(row),
        )
        .context("running shared-session-chunks query")?;
    Ok(rows.collect::<Result<Vec<_>, _>>()?)
}

pub fn print_session_neighbors_text(report: &EntitySessionNeighborsReport) {
    let mut scope_parts: Vec<String> = Vec::new();
    if let Some(k) = report.kind_filter {
        scope_parts.push(format!("kind={}", k.as_str()));
    }
    if let Some(s) = &report.session_id {
        scope_parts.push(format!("session_id={s}"));
    }
    let scope = if scope_parts.is_empty() {
        String::new()
    } else {
        format!(" ({})", scope_parts.join(", "))
    };
    println!(
        "{} ({}) — {} session-neighbor{}, source spans {} session{}{scope}",
        report.source_value,
        report.source_kind.as_str(),
        report.total_neighbors,
        if report.total_neighbors == 1 { "" } else { "s" },
        report.source_session_count,
        if report.source_session_count == 1 {
            ""
        } else {
            "s"
        },
    );
    if report.neighbors.is_empty() {
        println!("  (no session-co-occurring entities)");
        return;
    }
    for n in &report.neighbors {
        println!(
            "  {kind:<8} edge={edge:<16} shared={shared:<3} total={total:<4} projects={projects:<3} users={users:<3} {value}",
            kind = n.kind.as_str(),
            edge = n.edge_kind.as_str(),
            shared = n.shared_sessions,
            total = n.session_count,
            projects = n.project_count,
            users = n.user_count,
            value = n.value,
        );
        if let Some(sessions) = &n.shared_session_ids {
            for session in sessions {
                println!("      session={session}");
            }
        }
        if let Some(projects) = &n.shared_projects {
            for project in projects {
                println!("      project={project}");
            }
        }
        if let Some(users) = &n.shared_users {
            for user in users {
                println!("      user={user}");
            }
        }
        if let Some(refs) = n.chunks.as_ref() {
            for chunk in refs {
                println!(
                    "    - {chunk_id} [{uri}] {snippet}",
                    chunk_id = chunk.chunk_id,
                    uri = chunk.source_uri,
                    snippet = chunk.snippet,
                );
            }
        }
    }
}

pub fn print_session_neighbors_json(report: &EntitySessionNeighborsReport) -> Result<()> {
    println!("{}", serde_json::to_string_pretty(report)?);
    Ok(())
}

pub fn print_text(report: &EntityListReport) {
    if report.entries.is_empty() {
        println!("no entities matched");
        if let Some(kind) = report.kind_filter {
            println!("  kind:  {}", kind.as_str());
        }
        if let Some(v) = &report.value_contains {
            println!("  value contains: {v}");
        }
        if let Some(session) = &report.session_id {
            println!("  session_id: {session}");
        }
        return;
    }
    println!(
        "{} entit{} ({} total matched){scope}",
        report.entries.len(),
        if report.entries.len() == 1 {
            "y"
        } else {
            "ies"
        },
        report.total_matched,
        scope = match &report.session_id {
            Some(s) => format!(" (session_id={s})"),
            None => String::new(),
        },
    );
    for entry in &report.entries {
        println!(
            "  {kind:<8} chunks={chunks:<4} sessions={sessions:<3} projects={projects:<3} users={users:<3} topics={topics:<3} {value}",
            kind = entry.kind.as_str(),
            chunks = entry.chunk_count,
            sessions = entry.session_count,
            projects = entry.project_count,
            users = entry.user_count,
            topics = entry.topic_count,
            value = entry.value,
        );
        if let Some(sessions) = entry.session_ids.as_ref() {
            for session in sessions {
                println!("    session={session}");
            }
        }
        if let Some(projects) = entry.projects.as_ref() {
            for project in projects {
                println!("    project={project}");
            }
        }
        if let Some(users) = entry.users.as_ref() {
            for user in users {
                println!("    user={user}");
            }
        }
        if let Some(topics) = entry.topics.as_ref() {
            for topic in topics {
                println!("    topic={topic}");
            }
        }
        if let Some(refs) = entry.chunks.as_ref() {
            for chunk in refs {
                println!(
                    "    - {chunk_id} [{uri}] {snippet}",
                    chunk_id = chunk.chunk_id,
                    uri = chunk.source_uri,
                    snippet = chunk.snippet,
                );
            }
        }
    }
}

pub fn print_json(report: &EntityListReport) -> Result<()> {
    println!("{}", serde_json::to_string_pretty(report)?);
    Ok(())
}

/// Upsert each entity and link it to `chunk_id`. Designed to be called inside
/// the same transaction that inserted the chunk, so a failed extraction does
/// not leave the entity tables out of sync with `chunks`.
///
/// Returns the number of new chunk -> entity edges actually created (existing
/// links are silently ignored, so re-running on the same chunk is a no-op).
pub fn record_chunk_entities(
    conn: &Connection,
    chunk_id: &str,
    entities: &[Entity],
    now: i64,
) -> Result<usize> {
    if entities.is_empty() {
        return Ok(0);
    }
    let mut up_stmt = conn
        .prepare_cached(
            "INSERT OR IGNORE INTO entities (id, kind, value, created_at) VALUES (?1, ?2, ?3, ?4)",
        )
        .context("preparing entity upsert")?;
    let mut link_stmt = conn
        .prepare_cached(
            "INSERT OR IGNORE INTO chunk_entities (chunk_id, entity_id) VALUES (?1, ?2)",
        )
        .context("preparing chunk_entities link")?;
    let mut linked = 0usize;
    for entity in entities {
        let id = entity_id(entity.kind, &entity.value);
        up_stmt
            .execute(params![id, entity.kind.as_str(), entity.value, now])
            .with_context(|| {
                format!("upserting entity {}={}", entity.kind.as_str(), entity.value)
            })?;
        let inserted = link_stmt
            .execute(params![chunk_id, id])
            .with_context(|| format!("linking chunk {chunk_id} to entity {id}"))?;
        linked += inserted;
    }
    Ok(linked)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::Store;
    use tempfile::tempdir;

    #[test]
    fn extracts_basic_https_url() {
        let entities = extract_entities("see https://example.com for details");
        let urls: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Url)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(urls, vec!["https://example.com"]);
    }

    #[test]
    fn extracts_both_schemes() {
        let entities = extract_entities("http://a.test and https://b.test/path");
        let urls: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Url)
            .map(|e| e.value.as_str())
            .collect();
        assert!(urls.contains(&"http://a.test"));
        assert!(urls.contains(&"https://b.test/path"));
    }

    #[test]
    fn extracts_repository_slugs_from_repository_urls() {
        let entities = extract_entities(
            "see https://github.com/diogenes/lantern and https://git.skylantix.com/diogenes/lantern/",
        );
        let repos: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Repo)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(repos, vec!["diogenes/lantern"]);
    }

    #[test]
    fn extracts_domain_from_basic_url() {
        let entities = extract_entities("see https://example.com/page for details");
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(domains, vec!["example.com"]);
    }

    #[test]
    fn url_domain_lowercases_the_host() {
        let entities = extract_entities("visit https://Example.COM/path");
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        // Lowercased so `Example.COM` and `example.com` collapse onto the same
        // entity instead of creating per-casing duplicates.
        assert_eq!(domains, vec!["example.com"]);
    }

    #[test]
    fn url_domain_strips_port_and_userinfo() {
        let entities =
            extract_entities("ping https://user:pass@api.example.com:8080/v1/probe please");
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        // Port and `user:pass@` should both be stripped — the entity value is
        // the bare host so different ports under the same host collapse to one.
        assert_eq!(domains, vec!["api.example.com"]);
    }

    #[test]
    fn url_domain_dedupes_across_paths_and_schemes() {
        let entities = extract_entities(
            "compare http://example.com/a, https://example.com/b, https://example.com/c",
        );
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        // Three URLs over the same host collapse to one Domain entity even
        // though scheme and path differ.
        assert_eq!(domains, vec!["example.com"]);
    }

    #[test]
    fn extracts_domain_from_email() {
        let entities = extract_entities("reach me at alice@example.com today");
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(domains, vec!["example.com"]);
    }

    #[test]
    fn url_and_email_domains_dedupe_on_same_host() {
        let entities = extract_entities("see https://example.com/page or email me@example.com");
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        // URL and email both contribute the same host — dedup collapses them
        // onto a single Domain entity so the graph node count stays sane.
        assert_eq!(domains, vec!["example.com"]);
    }

    #[test]
    fn ipv4_host_extracts_as_domain() {
        let entities = extract_entities("local server http://192.168.1.10:8080/health");
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(domains, vec!["192.168.1.10"]);
    }

    #[test]
    fn ipv6_bracket_host_does_not_extract_as_domain() {
        // Bracket-wrapped IPv6 literals fall outside the conservative
        // alphanumeric/`.`/`-` host shape, so they are intentionally skipped.
        // The URL itself still surfaces as a `Url` entity.
        let entities = extract_entities("ping https://[::1]:8080/health");
        let urls: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Url)
            .map(|e| e.value.as_str())
            .collect();
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(urls, vec!["https://[::1]:8080/health"]);
        assert!(
            domains.is_empty(),
            "ipv6 literal should not produce a domain entity, got: {domains:?}"
        );
    }

    #[test]
    fn single_label_host_does_not_extract_as_domain() {
        // `localhost` and other bare hostnames have no `.` separator. The
        // domain extractor requires at least two labels so we don't pollute
        // the entity store with bare-word hosts that look like every other
        // identifier.
        let entities = extract_entities("hit http://localhost:3000/health for status");
        let domains: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Domain)
            .map(|e| e.value.as_str())
            .collect();
        assert!(
            domains.is_empty(),
            "single-label host should not produce a domain entity, got: {domains:?}"
        );
    }

    #[test]
    fn trims_trailing_sentence_punctuation() {
        let entities = extract_entities("visit https://example.com/page, or https://other.test.");
        let urls: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Url)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(urls, vec!["https://example.com/page", "https://other.test"]);
    }

    #[test]
    fn dedups_within_same_text() {
        let entities =
            extract_entities("https://x.test and again https://x.test plus https://y.test");
        let urls: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Url)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(urls, vec!["https://x.test", "https://y.test"]);
    }

    #[test]
    fn ignores_bare_scheme_without_host() {
        let entities = extract_entities("not a url: http:// or https:// alone");
        assert!(entities.is_empty());
    }

    #[test]
    fn extracts_emails_and_trims_punctuation() {
        let entities =
            extract_entities("reach me at alice@example.com, or bob.smith+test@sub.example.co.uk.");
        let emails: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Email)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(
            emails,
            vec!["alice@example.com", "bob.smith+test@sub.example.co.uk",]
        );
    }

    #[test]
    fn extracts_backtick_wrapped_file_paths() {
        let entities = extract_entities(
            "look at `src/main.rs`, `Cargo.toml`, and `Vec` while ignoring `https://example.com`",
        );
        let values: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::FilePath)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(values, vec!["src/main.rs", "Cargo.toml"]);
    }

    #[test]
    fn extracts_plain_file_paths_with_relative_or_absolute_prefixes() {
        let entities = extract_entities(
            "check ./src/main.rs, /tmp/example.log, and ../notes/todo.md but not issue/123 or https://example.com/src/main.rs",
        );
        let values: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::FilePath)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(
            values,
            vec!["./src/main.rs", "/tmp/example.log", "../notes/todo.md"]
        );
    }

    #[test]
    fn entity_id_is_deterministic_and_kind_scoped() {
        let a = entity_id(EntityKind::Url, "https://example.com");
        let b = entity_id(EntityKind::Url, "https://example.com");
        let repo = entity_id(EntityKind::Repo, "https://example.com");
        let domain = entity_id(EntityKind::Domain, "https://example.com");
        let email = entity_id(EntityKind::Email, "https://example.com");
        let path = entity_id(EntityKind::FilePath, "https://example.com");
        let mention = entity_id(EntityKind::Mention, "https://example.com");
        let hashtag = entity_id(EntityKind::Hashtag, "https://example.com");
        assert_eq!(a, b);
        assert_ne!(a, repo);
        assert_ne!(a, domain);
        assert_ne!(a, email);
        assert_ne!(a, path);
        assert_ne!(a, mention);
        assert_ne!(a, hashtag);
        assert_ne!(email, mention);
        assert_ne!(domain, repo);
        assert_eq!(a.len(), 32);
    }

    #[test]
    fn extracts_basic_mentions() {
        let entities = extract_entities("Hi @alice and @bob_smith — see also @charlie.");
        let values: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(values, vec!["@alice", "@bob_smith", "@charlie"]);
    }

    #[test]
    fn mention_at_start_of_text() {
        let entities = extract_entities("@alice says hi");
        let mentions: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(mentions, vec!["@alice"]);
    }

    #[test]
    fn email_does_not_double_extract_as_mention() {
        let entities = extract_entities("ping me@example.com");
        let mention_count = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .count();
        assert_eq!(mention_count, 0);
        let emails: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Email)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(emails, vec!["me@example.com"]);
    }

    #[test]
    fn mention_and_email_can_coexist() {
        let entities = extract_entities("@alice emailed bob@example.com today");
        let mentions: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .map(|e| e.value.as_str())
            .collect();
        let emails: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Email)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(mentions, vec!["@alice"]);
        assert_eq!(emails, vec!["bob@example.com"]);
    }

    #[test]
    fn mention_dedupes_within_text() {
        let entities = extract_entities("@bob said hi, then @bob left, finally @carol");
        let mentions: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(mentions, vec!["@bob", "@carol"]);
    }

    #[test]
    fn mention_trims_trailing_separators() {
        // Trailing punctuation outside the body-char set ends the scan; trailing
        // `.`/`-`/`_` inside the scan get trimmed off by `MENTION_TRAILING_TRIM`.
        let entities = extract_entities("Ping @charlie! Then @dana. And @eve-");
        let mentions: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(mentions, vec!["@charlie", "@dana", "@eve"]);
    }

    #[test]
    fn mention_skips_short_or_empty_handles() {
        let entities = extract_entities("@ alone, @x too short, but @ab is fine");
        let mentions: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(mentions, vec!["@ab"]);
    }

    #[test]
    fn mention_skips_pure_digit_or_date_handles() {
        // Conservative: no ASCII letter in the body means we leave it alone,
        // so dates and version-shaped tokens do not become mentions.
        let entities = extract_entities("see @2024-01-15 or @1.2.3 vs @v1.2.3");
        let mentions: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Mention)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(mentions, vec!["@v1.2.3"]);
    }

    #[test]
    fn extracts_hashtags_and_ignores_url_fragments() {
        let entities = extract_entities(
            "follow #Lantern and #rust-lang, but keep https://example.com/#anchor and foo#bar as-is",
        );
        let tags: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Hashtag)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(tags, vec!["#Lantern", "#rust-lang"]);
    }

    #[test]
    fn hashtag_skips_short_or_digit_only_tags() {
        let entities = extract_entities("# alone, #x too short, #2024 no, but #v1 is fine");
        let tags: Vec<_> = entities
            .iter()
            .filter(|e| e.kind == EntityKind::Hashtag)
            .map(|e| e.value.as_str())
            .collect();
        assert_eq!(tags, vec!["#v1"]);
    }

    #[test]
    fn record_chunk_entities_dedups_across_chunks() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        // Seed a fake source + two chunks so the FK targets exist.
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('src1', 'mem://t', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (id, ord) in [("c1", 0), ("c2", 1)] {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, created_at)
                 VALUES (?1, 'src1', ?2, 0, 0, 0, '', '', 0)",
                params![id, ord],
            )
            .unwrap();
        }
        let shared = vec![Entity {
            kind: EntityKind::Url,
            value: "https://shared.test".into(),
        }];
        record_chunk_entities(&tx, "c1", &shared, 100).unwrap();
        record_chunk_entities(&tx, "c2", &shared, 200).unwrap();
        tx.commit().unwrap();

        let entity_count: i64 = store
            .conn()
            .query_row("SELECT COUNT(*) FROM entities", [], |row| row.get(0))
            .unwrap();
        assert_eq!(
            entity_count, 1,
            "shared URL must collapse to a single entity"
        );

        let edges: i64 = store
            .conn()
            .query_row("SELECT COUNT(*) FROM chunk_entities", [], |row| row.get(0))
            .unwrap();
        assert_eq!(edges, 2, "both chunks should link to the shared entity");
    }

    fn seed_chunks(store: &mut Store, ids: &[&str]) {
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, id) in ids.iter().enumerate() {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', 0)",
                params![id, ord as i64],
            )
            .unwrap();
        }
        tx.commit().unwrap();
    }

    /// Seed `(chunk_id, session_id_opt)` pairs. `None` leaves
    /// `chunks.session_id` as NULL so the listing can be tested with a mix of
    /// session-tagged and untagged chunks under one source.
    fn seed_chunks_with_sessions(store: &mut Store, rows: &[(&str, Option<&str>)]) {
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session)) in rows.iter().enumerate() {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, 0)",
                params![id, ord as i64, session],
            )
            .unwrap();
        }
        tx.commit().unwrap();
    }

    /// Seed `(chunk_id, project_opt)` pairs. `None` leaves `chunks.project`
    /// as NULL so the project_count code path can be exercised against a mix
    /// of tagged and untagged chunks under one source.
    fn seed_chunks_with_projects(store: &mut Store, rows: &[(&str, Option<&str>)]) {
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, project)) in rows.iter().enumerate() {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, project, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, 0)",
                params![id, ord as i64, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();
    }

    /// Seed `(chunk_id, user_opt)` pairs. `None` leaves `chunks.user` as NULL
    /// so the user_count code path can be exercised against a mix of tagged
    /// and untagged chunks under one source.
    fn seed_chunks_with_users(store: &mut Store, rows: &[(&str, Option<&str>)]) {
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, user)) in rows.iter().enumerate() {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, user, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, 0)",
                params![id, ord as i64, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();
    }

    /// Seed `(chunk_id, topic_opt)` pairs. `None` leaves `chunks.topic` as
    /// NULL so the topic_count code path can be exercised against a mix of
    /// tagged and untagged chunks under one source.
    fn seed_chunks_with_topics(store: &mut Store, rows: &[(&str, Option<&str>)]) {
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, topic)) in rows.iter().enumerate() {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, topic, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, 0)",
                params![id, ord as i64, topic],
            )
            .unwrap();
        }
        tx.commit().unwrap();
    }

    fn link(store: &mut Store, chunk_id: &str, entities: &[Entity], now: i64) {
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        record_chunk_entities(&tx, chunk_id, entities, now).unwrap();
        tx.commit().unwrap();
    }

    #[test]
    fn list_entities_orders_by_chunk_count_desc() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2", "c3"]);
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        let lonely = Entity {
            kind: EntityKind::Url,
            value: "https://lonely.test".into(),
        };
        link(&mut store, "c1", &[popular.clone(), lonely.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.total_matched, 2);
        assert_eq!(report.entries.len(), 2);
        assert_eq!(report.entries[0].value, "https://popular.test");
        assert_eq!(report.entries[0].chunk_count, 3);
        assert_eq!(report.entries[1].value, "https://lonely.test");
        assert_eq!(report.entries[1].chunk_count, 1);
    }

    #[test]
    fn list_entities_filters_by_kind() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1"]);
        link(
            &mut store,
            "c1",
            &[
                Entity {
                    kind: EntityKind::Url,
                    value: "https://x.test".into(),
                },
                Entity {
                    kind: EntityKind::Email,
                    value: "alice@x.test".into(),
                },
                Entity {
                    kind: EntityKind::Mention,
                    value: "@bob".into(),
                },
            ],
            100,
        );

        let opts = EntityListOptions {
            kind: Some(EntityKind::Email),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.total_matched, 1);
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].kind, EntityKind::Email);
        assert_eq!(report.entries[0].value, "alice@x.test");
    }

    #[test]
    fn list_entities_value_contains_treats_percent_as_literal() {
        // The SQL builder must escape `%` so users searching for a literal
        // percent sign do not accidentally turn it into a wildcard.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);
        link(
            &mut store,
            "c1",
            &[Entity {
                kind: EntityKind::Url,
                value: "https://example.test/a%20b".into(),
            }],
            100,
        );
        link(
            &mut store,
            "c2",
            &[Entity {
                kind: EntityKind::Url,
                value: "https://other.test/plain".into(),
            }],
            200,
        );

        // A literal `%` substring should match only the first URL — not act
        // as the LIKE wildcard.
        let opts = EntityListOptions {
            value_contains: Some("%20".into()),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.total_matched, 1);
        assert_eq!(report.entries[0].value, "https://example.test/a%20b");
    }

    #[test]
    fn list_entities_limit_truncates_but_total_matched_is_full() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1"]);
        let many = (0..5)
            .map(|i| Entity {
                kind: EntityKind::Url,
                value: format!("https://e{i}.test"),
            })
            .collect::<Vec<_>>();
        link(&mut store, "c1", &many, 100);

        let opts = EntityListOptions {
            limit: Some(2),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.entries.len(), 2);
        assert_eq!(report.total_matched, 5);
    }

    #[test]
    fn list_entities_with_chunks_includes_chunk_refs() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2", "c3"]);

        {
            let conn = store.conn_mut();
            let tx = conn.transaction().unwrap();
            tx.execute(
                "UPDATE chunks
                 SET text = CASE id
                     WHEN 'c1' THEN 'first reference chunk text'
                     WHEN 'c2' THEN 'second reference chunk text'
                     WHEN 'c3' THEN 'third reference chunk text'
                 END
                 WHERE id IN ('c1', 'c2', 'c3')",
                [],
            )
            .unwrap();
            tx.commit().unwrap();
        }

        let entity = Entity {
            kind: EntityKind::Url,
            value: "https://graph.test/edge".into(),
        };
        link(&mut store, "c2", &[entity.clone()], 100);
        link(&mut store, "c1", &[entity], 200);

        let opts = EntityListOptions {
            with_chunks: Some(2),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.total_matched, 1);
        let entry = &report.entries[0];
        let refs = entry
            .chunks
            .as_ref()
            .expect("chunk refs should be included");
        assert_eq!(refs.len(), 2);
        assert_eq!(refs[0].chunk_id, "c1");
        assert_eq!(refs[0].source_uri, "mem://list");
        assert_eq!(refs[0].snippet, "first reference chunk text");
        assert_eq!(refs[1].chunk_id, "c2");
        assert_eq!(refs[1].snippet, "second reference chunk text");
    }

    #[test]
    fn list_entities_session_evidence_is_opt_in_and_zero_keeps_default_shape() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(&mut store, &[("c1", Some("s1")), ("c2", Some("s2"))]);
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", &[mention], 200);

        let default_report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(default_report.entries.len(), 1);
        assert!(
            default_report.entries[0].session_ids.is_none(),
            "session evidence must be omitted by default"
        );

        let zeroed_report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_sessions: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(
            zeroed_report.entries[0].session_ids.is_none(),
            "with_sessions=0 must collapse to the cheap default path"
        );
    }

    #[test]
    fn list_entities_with_sessions_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("s3")),
                ("c2", Some("s1")),
                ("c3", Some("s2")),
                ("c4", None),
            ],
        );
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", std::slice::from_ref(&mention), 200);
        link(&mut store, "c3", std::slice::from_ref(&mention), 300);
        link(&mut store, "c4", &[mention], 400);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_sessions: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(
            report.entries[0]
                .session_ids
                .as_ref()
                .expect("opt-in session evidence should be present"),
            &vec!["s1".to_string(), "s2".to_string()]
        );
    }

    #[test]
    fn list_entities_with_sessions_collapses_under_session_filter() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(&mut store, &[("c1", Some("s1")), ("c2", Some("s2"))]);
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", &[mention], 200);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                session_id: Some("s2".into()),
                with_sessions: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s2"));
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].session_count, 1);
        assert_eq!(
            report.entries[0]
                .session_ids
                .as_ref()
                .expect("session-scoped evidence should be present"),
            &vec!["s2".to_string()]
        );
    }

    #[test]
    fn list_entities_project_evidence_is_opt_in_and_zero_keeps_default_shape() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_projects(&mut store, &[("c1", Some("alpha")), ("c2", Some("beta"))]);
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", &[mention], 200);

        let default_report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(default_report.entries.len(), 1);
        assert!(
            default_report.entries[0].projects.is_none(),
            "project evidence must be omitted by default"
        );

        let zeroed_report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_projects: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(
            zeroed_report.entries[0].projects.is_none(),
            "with_projects=0 must collapse to the cheap default path"
        );
    }

    #[test]
    fn list_entities_with_projects_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_projects(
            &mut store,
            &[
                ("c1", Some("gamma")),
                ("c2", Some("alpha")),
                ("c3", Some("beta")),
                ("c4", None),
            ],
        );
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", std::slice::from_ref(&mention), 200);
        link(&mut store, "c3", std::slice::from_ref(&mention), 300);
        link(&mut store, "c4", &[mention], 400);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_projects: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(
            report.entries[0]
                .projects
                .as_ref()
                .expect("opt-in project evidence should be present"),
            &vec!["alpha".to_string(), "beta".to_string()]
        );
    }

    #[test]
    fn list_entities_with_projects_honors_session_scope_without_collapsing() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, project)) in [
            ("c1", "s1", Some("alpha")),
            ("c2", "s1", Some("beta")),
            ("c3", "s2", Some("gamma")),
            ("c4", "s1", None),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", std::slice::from_ref(&mention), 200);
        link(&mut store, "c3", std::slice::from_ref(&mention), 300);
        link(&mut store, "c4", &[mention], 400);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                session_id: Some("s1".into()),
                with_projects: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s1"));
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].session_count, 1);
        assert_eq!(
            report.entries[0]
                .projects
                .as_ref()
                .expect("session-scoped project evidence should be present"),
            &vec!["alpha".to_string(), "beta".to_string()]
        );
    }

    #[test]
    fn list_entities_user_evidence_is_opt_in_and_zero_keeps_default_shape() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_users(&mut store, &[("c1", Some("alice")), ("c2", Some("bob"))]);
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", &[mention], 200);

        let default_report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(default_report.entries.len(), 1);
        assert!(
            default_report.entries[0].users.is_none(),
            "user evidence must be omitted by default"
        );

        let zeroed_report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_users: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(
            zeroed_report.entries[0].users.is_none(),
            "with_users=0 must collapse to the cheap default path"
        );
    }

    #[test]
    fn list_entities_with_users_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_users(
            &mut store,
            &[
                ("c1", Some("zoe")),
                ("c2", Some("alice")),
                ("c3", Some("bob")),
                ("c4", None),
            ],
        );
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", std::slice::from_ref(&mention), 200);
        link(&mut store, "c3", std::slice::from_ref(&mention), 300);
        link(&mut store, "c4", &[mention], 400);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_users: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(
            report.entries[0]
                .users
                .as_ref()
                .expect("opt-in user evidence should be present"),
            &vec!["alice".to_string(), "bob".to_string()]
        );
    }

    #[test]
    fn list_entities_with_users_honors_session_scope_without_collapsing() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, user)) in [
            ("c1", "s1", Some("alice")),
            ("c2", "s1", Some("bob")),
            ("c3", "s2", Some("carol")),
            ("c4", "s1", None),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", std::slice::from_ref(&mention), 200);
        link(&mut store, "c3", std::slice::from_ref(&mention), 300);
        link(&mut store, "c4", &[mention], 400);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                session_id: Some("s1".into()),
                with_users: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s1"));
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].session_count, 1);
        assert_eq!(
            report.entries[0]
                .users
                .as_ref()
                .expect("session-scoped user evidence should be present"),
            &vec!["alice".to_string(), "bob".to_string()]
        );
    }

    #[test]
    fn list_entities_topic_evidence_is_opt_in_and_zero_keeps_default_shape() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_topics(&mut store, &[("c1", Some("alpha")), ("c2", Some("beta"))]);
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", &[mention], 200);

        let default_report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(default_report.entries.len(), 1);
        assert!(
            default_report.entries[0].topics.is_none(),
            "topic evidence must be omitted by default"
        );

        let zeroed_report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_topics: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(
            zeroed_report.entries[0].topics.is_none(),
            "with_topics=0 must collapse to the cheap default path"
        );
    }

    #[test]
    fn list_entities_with_topics_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_topics(
            &mut store,
            &[
                ("c1", Some("zeta")),
                ("c2", Some("alpha")),
                ("c3", Some("beta")),
                ("c4", None),
            ],
        );
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", std::slice::from_ref(&mention), 200);
        link(&mut store, "c3", std::slice::from_ref(&mention), 300);
        link(&mut store, "c4", &[mention], 400);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                with_topics: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(
            report.entries[0]
                .topics
                .as_ref()
                .expect("opt-in topic evidence should be present"),
            &vec!["alpha".to_string(), "beta".to_string()]
        );
    }

    #[test]
    fn list_entities_with_topics_honors_session_scope_without_collapsing() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, topic)) in [
            ("c1", "s1", Some("alpha")),
            ("c2", "s1", Some("beta")),
            ("c3", "s2", Some("gamma")),
            ("c4", "s1", None),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, topic, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, topic],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@topic".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&mention), 100);
        link(&mut store, "c2", std::slice::from_ref(&mention), 200);
        link(&mut store, "c3", std::slice::from_ref(&mention), 300);
        link(&mut store, "c4", &[mention], 400);

        let report = list_entities(
            store.conn(),
            &EntityListOptions {
                session_id: Some("s1".into()),
                with_topics: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s1"));
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].session_count, 1);
        assert_eq!(
            report.entries[0]
                .topics
                .as_ref()
                .expect("session-scoped topic evidence should be present"),
            &vec!["alpha".to_string(), "beta".to_string()]
        );
    }

    #[test]
    fn list_entities_empty_store_returns_empty_report() {
        let dir = tempdir().unwrap();
        let store = Store::initialize(&dir.path().join("store")).unwrap();
        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert!(report.entries.is_empty());
        assert_eq!(report.total_matched, 0);
    }

    #[test]
    fn list_entities_session_count_is_zero_for_chunks_without_session_id() {
        // The default seed leaves `chunks.session_id` NULL, so an entity that
        // only ever links to those chunks must surface `session_count = 0`.
        // Guards against a future refactor that would, for example, count the
        // raw chunk rows instead of the distinct non-NULL session ids.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);
        let url = Entity {
            kind: EntityKind::Url,
            value: "https://untagged.test".into(),
        };
        link(&mut store, "c1", &[url.clone()], 100);
        link(&mut store, "c2", &[url], 200);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].chunk_count, 2);
        assert_eq!(report.entries[0].session_count, 0);
    }

    #[test]
    fn list_entities_session_count_counts_distinct_sessions() {
        // Three chunks across two sessions plus one untagged chunk. Entity
        // `popular` appears in all four → session_count == 2 (NULL excluded);
        // entity `narrow` only appears in session "a" → session_count == 1.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("a")),
                ("c2", Some("a")),
                ("c3", Some("b")),
                ("c4", None),
            ],
        );
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        let narrow = Entity {
            kind: EntityKind::Url,
            value: "https://narrow.test".into(),
        };
        link(&mut store, "c1", &[popular.clone(), narrow.clone()], 100);
        link(&mut store, "c2", &[popular.clone(), narrow.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 2);
        // Ordered by chunk_count desc then value asc: popular (4 chunks), narrow (2).
        assert_eq!(report.entries[0].value, "https://popular.test");
        assert_eq!(report.entries[0].chunk_count, 4);
        assert_eq!(
            report.entries[0].session_count, 2,
            "popular spans sessions a + b; the untagged chunk must not count",
        );
        assert_eq!(report.entries[1].value, "https://narrow.test");
        assert_eq!(report.entries[1].chunk_count, 2);
        assert_eq!(report.entries[1].session_count, 1);
    }

    #[test]
    fn list_entities_project_count_is_zero_for_chunks_without_project() {
        // The default seed leaves `chunks.project` NULL, so an entity that
        // only ever links to those chunks must surface `project_count = 0`.
        // Mirrors the session_count NULL guard against an accidental rewrite
        // that would count raw chunk rows instead of distinct non-NULL
        // projects.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);
        let url = Entity {
            kind: EntityKind::Url,
            value: "https://untagged.test".into(),
        };
        link(&mut store, "c1", &[url.clone()], 100);
        link(&mut store, "c2", &[url], 200);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].chunk_count, 2);
        assert_eq!(report.entries[0].project_count, 0);
    }

    #[test]
    fn list_entities_project_count_counts_distinct_projects() {
        // Three project-tagged chunks across two projects plus one untagged
        // chunk. Entity `popular` appears in all four → project_count == 2
        // (NULL excluded); entity `narrow` only appears in project "alpha" →
        // project_count == 1.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_projects(
            &mut store,
            &[
                ("c1", Some("alpha")),
                ("c2", Some("alpha")),
                ("c3", Some("beta")),
                ("c4", None),
            ],
        );
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        let narrow = Entity {
            kind: EntityKind::Url,
            value: "https://narrow.test".into(),
        };
        link(&mut store, "c1", &[popular.clone(), narrow.clone()], 100);
        link(&mut store, "c2", &[popular.clone(), narrow.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 2);
        assert_eq!(report.entries[0].value, "https://popular.test");
        assert_eq!(report.entries[0].chunk_count, 4);
        assert_eq!(
            report.entries[0].project_count, 2,
            "popular spans projects alpha + beta; the untagged chunk must not count",
        );
        assert_eq!(report.entries[1].value, "https://narrow.test");
        assert_eq!(report.entries[1].chunk_count, 2);
        assert_eq!(report.entries[1].project_count, 1);
    }

    #[test]
    fn list_entities_project_count_is_independent_of_session_filter() {
        // Project grouping lives on its own column; a `session_id` filter
        // must not collapse `project_count` to 1. Here every chunk is in one
        // session but spans two projects — the scoped listing should still
        // see `project_count == 2`. Pins the asymmetry so a future refactor
        // does not accidentally couple the two grouping fields.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, project)) in [("c1", "alpha"), ("c2", "alpha"), ("c3", "beta")]
            .iter()
            .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', 's1', ?3, 0)",
                params![id, ord as i64, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        link(&mut store, "c1", &[popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);

        let opts = EntityListOptions {
            session_id: Some("s1".into()),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].session_count, 1);
        assert_eq!(
            report.entries[0].project_count, 2,
            "project grouping must not collapse under a session_id filter",
        );
    }

    #[test]
    fn list_entities_user_count_is_zero_for_chunks_without_user() {
        // The default seed leaves `chunks.user` NULL, so an entity that only
        // ever links to those chunks must surface `user_count = 0`. Mirrors
        // the project_count NULL guard against an accidental rewrite that
        // would count raw chunk rows instead of distinct non-NULL users.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);
        let url = Entity {
            kind: EntityKind::Url,
            value: "https://untagged.test".into(),
        };
        link(&mut store, "c1", &[url.clone()], 100);
        link(&mut store, "c2", &[url], 200);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].chunk_count, 2);
        assert_eq!(report.entries[0].user_count, 0);
    }

    #[test]
    fn list_entities_user_count_counts_distinct_users() {
        // Three user-tagged chunks across two users plus one untagged chunk.
        // Entity `popular` appears in all four → user_count == 2 (NULL
        // excluded); entity `narrow` only appears under user "alice" →
        // user_count == 1.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_users(
            &mut store,
            &[
                ("c1", Some("alice")),
                ("c2", Some("alice")),
                ("c3", Some("bob")),
                ("c4", None),
            ],
        );
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        let narrow = Entity {
            kind: EntityKind::Url,
            value: "https://narrow.test".into(),
        };
        link(&mut store, "c1", &[popular.clone(), narrow.clone()], 100);
        link(&mut store, "c2", &[popular.clone(), narrow.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 2);
        assert_eq!(report.entries[0].value, "https://popular.test");
        assert_eq!(report.entries[0].chunk_count, 4);
        assert_eq!(
            report.entries[0].user_count, 2,
            "popular spans users alice + bob; the untagged chunk must not count",
        );
        assert_eq!(report.entries[1].value, "https://narrow.test");
        assert_eq!(report.entries[1].chunk_count, 2);
        assert_eq!(report.entries[1].user_count, 1);
    }

    #[test]
    fn list_entities_user_count_is_independent_of_session_filter() {
        // User grouping lives on its own column; a `session_id` filter must
        // not collapse `user_count` to 1. Here every chunk is in one session
        // but spans two users — the scoped listing should still see
        // `user_count == 2`. Pins the asymmetry so a future refactor does not
        // accidentally couple the two grouping fields.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, user)) in [("c1", "alice"), ("c2", "alice"), ("c3", "bob")]
            .iter()
            .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', 's1', ?3, 0)",
                params![id, ord as i64, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        link(&mut store, "c1", &[popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);

        let opts = EntityListOptions {
            session_id: Some("s1".into()),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].session_count, 1);
        assert_eq!(
            report.entries[0].user_count, 2,
            "user grouping must not collapse under a session_id filter",
        );
    }

    #[test]
    fn list_entities_topic_count_is_zero_for_chunks_without_topic() {
        // The default seed leaves `chunks.topic` NULL, so an entity that only
        // ever links to those chunks must surface `topic_count = 0`. Mirrors
        // the project_count / user_count NULL guard against an accidental
        // rewrite that would count raw chunk rows instead of distinct non-NULL
        // topics.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);
        let url = Entity {
            kind: EntityKind::Url,
            value: "https://untagged.test".into(),
        };
        link(&mut store, "c1", &[url.clone()], 100);
        link(&mut store, "c2", &[url], 200);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].chunk_count, 2);
        assert_eq!(report.entries[0].topic_count, 0);
    }

    #[test]
    fn list_entities_topic_count_counts_distinct_topics() {
        // Three topic-tagged chunks across two topics plus one untagged chunk.
        // Entity `popular` appears in all four → topic_count == 2 (NULL
        // excluded); entity `narrow` only appears under topic "alpha" →
        // topic_count == 1.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_topics(
            &mut store,
            &[
                ("c1", Some("alpha")),
                ("c2", Some("alpha")),
                ("c3", Some("beta")),
                ("c4", None),
            ],
        );
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        let narrow = Entity {
            kind: EntityKind::Url,
            value: "https://narrow.test".into(),
        };
        link(&mut store, "c1", &[popular.clone(), narrow.clone()], 100);
        link(&mut store, "c2", &[popular.clone(), narrow.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let report = list_entities(store.conn(), &EntityListOptions::default()).unwrap();
        assert_eq!(report.entries.len(), 2);
        assert_eq!(report.entries[0].value, "https://popular.test");
        assert_eq!(report.entries[0].chunk_count, 4);
        assert_eq!(
            report.entries[0].topic_count, 2,
            "popular spans topics alpha + beta; the untagged chunk must not count",
        );
        assert_eq!(report.entries[1].value, "https://narrow.test");
        assert_eq!(report.entries[1].chunk_count, 2);
        assert_eq!(report.entries[1].topic_count, 1);
    }

    #[test]
    fn list_entities_topic_count_is_independent_of_session_filter() {
        // Topic grouping lives on its own column; a `session_id` filter must
        // not collapse `topic_count` to 1. Here every chunk is in one session
        // but spans two topics — the scoped listing should still see
        // `topic_count == 2`. Pins the asymmetry so a future refactor does not
        // accidentally couple the two grouping fields.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, topic)) in [("c1", "alpha"), ("c2", "alpha"), ("c3", "beta")]
            .iter()
            .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, topic, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', 's1', ?3, 0)",
                params![id, ord as i64, topic],
            )
            .unwrap();
        }
        tx.commit().unwrap();
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        link(&mut store, "c1", &[popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);

        let opts = EntityListOptions {
            session_id: Some("s1".into()),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].session_count, 1);
        assert_eq!(
            report.entries[0].topic_count, 2,
            "topic grouping must not collapse under a session_id filter",
        );
    }

    #[test]
    fn list_entities_session_count_is_one_under_session_filter() {
        // When the listing is narrowed to a single `session_id`, every matched
        // entity is by construction within that session — `session_count`
        // collapses to 1. Pin it so the session-scoped path stays consistent
        // with the unfiltered path's distinct-sessions semantics.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[("c1", Some("a")), ("c2", Some("a")), ("c3", Some("b"))],
        );
        let popular = Entity {
            kind: EntityKind::Url,
            value: "https://popular.test".into(),
        };
        link(&mut store, "c1", &[popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);

        let opts = EntityListOptions {
            session_id: Some("a".into()),
            ..Default::default()
        };
        let report = list_entities(store.conn(), &opts).unwrap();
        assert_eq!(report.entries.len(), 1);
        assert_eq!(report.entries[0].chunk_count, 2);
        assert_eq!(report.entries[0].session_count, 1);
    }

    #[test]
    fn entity_neighbors_ranks_by_shared_chunk_count() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2", "c3"]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let frequent = Entity {
            kind: EntityKind::Mention,
            value: "@frequent".into(),
        };
        let sometimes = Entity {
            kind: EntityKind::Mention,
            value: "@sometimes".into(),
        };
        let unrelated = Entity {
            kind: EntityKind::Url,
            value: "https://unrelated.test".into(),
        };

        // c1: topic + frequent + sometimes
        // c2: topic + frequent
        // c3: unrelated only (must not surface as a neighbor of topic)
        link(
            &mut store,
            "c1",
            &[topic.clone(), frequent.clone(), sometimes.clone()],
            100,
        );
        link(&mut store, "c2", &[topic.clone(), frequent.clone()], 200);
        link(&mut store, "c3", &[unrelated], 300);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.source_value, "#topic");
        assert_eq!(report.total_neighbors, 2);
        assert_eq!(report.neighbors.len(), 2);
        assert_eq!(report.neighbors[0].value, "@frequent");
        assert_eq!(report.neighbors[0].edge_kind, EdgeKind::CoOccursWith);
        assert_eq!(report.neighbors[0].shared_chunks, 2);
        assert_eq!(report.neighbors[1].value, "@sometimes");
        assert_eq!(report.neighbors[1].edge_kind, EdgeKind::CoOccursWith);
        assert_eq!(report.neighbors[1].shared_chunks, 1);
        // The source entity must never appear in its own neighbor list.
        assert!(report.neighbors.iter().all(|n| n.id != topic_id));
    }

    #[test]
    fn entity_neighbors_filters_by_kind() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1"]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };
        let url = Entity {
            kind: EntityKind::Url,
            value: "https://example.test/x".into(),
        };
        link(
            &mut store,
            "c1",
            &[topic.clone(), mention.clone(), url.clone()],
            100,
        );

        let topic_id = entity_id(topic.kind, &topic.value);
        let opts = EntityNeighborsOptions {
            kind: Some(EntityKind::Url),
            ..Default::default()
        };
        let report = entity_neighbors(store.conn(), &topic_id, &opts).unwrap();
        assert_eq!(report.total_neighbors, 1);
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(report.neighbors[0].kind, EntityKind::Url);
        assert_eq!(report.neighbors[0].edge_kind, EdgeKind::CoOccursWith);
        assert_eq!(report.neighbors[0].value, "https://example.test/x");
    }

    #[test]
    fn entity_neighbors_limit_truncates_but_total_is_full() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1"]);

        let hub = Entity {
            kind: EntityKind::Hashtag,
            value: "#hub".into(),
        };
        let mut linked = vec![hub.clone()];
        for i in 0..4 {
            linked.push(Entity {
                kind: EntityKind::Url,
                value: format!("https://n{i}.test"),
            });
        }
        link(&mut store, "c1", &linked, 100);

        let hub_id = entity_id(hub.kind, &hub.value);
        let opts = EntityNeighborsOptions {
            limit: Some(2),
            ..Default::default()
        };
        let report = entity_neighbors(store.conn(), &hub_id, &opts).unwrap();
        assert_eq!(report.total_neighbors, 4);
        assert_eq!(report.neighbors.len(), 2);
    }

    #[test]
    fn entity_neighbors_returns_empty_when_no_co_occurrences() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1"]);
        let solo = Entity {
            kind: EntityKind::Url,
            value: "https://solo.test".into(),
        };
        link(&mut store, "c1", &[solo.clone()], 100);

        let solo_id = entity_id(solo.kind, &solo.value);
        let report =
            entity_neighbors(store.conn(), &solo_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.total_neighbors, 0);
        assert!(report.neighbors.is_empty());
    }

    #[test]
    fn entity_neighbors_with_chunks_includes_shared_chunk_refs() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2", "c3"]);

        {
            let conn = store.conn_mut();
            let tx = conn.transaction().unwrap();
            tx.execute(
                "UPDATE chunks
                 SET text = CASE id
                     WHEN 'c1' THEN '#topic and @friend in first shared chunk'
                     WHEN 'c2' THEN '#topic and @friend in second shared chunk'
                     WHEN 'c3' THEN '@friend only unrelated chunk'
                 END
                 WHERE id IN ('c1', 'c2', 'c3')",
                [],
            )
            .unwrap();
            tx.commit().unwrap();
        }

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let friend = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };

        link(&mut store, "c1", &[topic.clone(), friend.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), friend.clone()], 200);
        link(&mut store, "c3", &[friend], 300);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                with_chunks: Some(1),
                ..Default::default()
            },
        )
        .unwrap();

        assert_eq!(report.neighbors.len(), 1);
        let refs = report.neighbors[0]
            .chunks
            .as_ref()
            .expect("shared chunk refs should be included");
        assert_eq!(refs.len(), 1);
        assert_eq!(refs[0].chunk_id, "c1");
        assert_eq!(refs[0].source_uri, "mem://list");
        assert_eq!(refs[0].snippet, "#topic and @friend in first shared chunk");
    }

    #[test]
    fn entity_neighbors_errors_on_unknown_id() {
        let dir = tempdir().unwrap();
        let store = Store::initialize(&dir.path().join("store")).unwrap();
        let err =
            entity_neighbors(store.conn(), "nope", &EntityNeighborsOptions::default()).unwrap_err();
        assert!(err.to_string().contains("no entity with id"));
    }

    #[test]
    fn entity_neighbors_reports_neighbor_total_chunk_count() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2", "c3"]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        // Source-and-popular share c1 only; popular also appears alone in c2/c3.
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.edge_kind, EdgeKind::CoOccursWith);
        assert_eq!(n.shared_chunks, 1);
        assert_eq!(n.chunk_count, 3);
    }

    #[test]
    fn entity_neighbors_session_count_is_zero_for_chunks_without_session_id() {
        // When every linking chunk has NULL session_id, the neighbor's spread
        // signal must be 0 — guards against a future refactor that counts the
        // raw chunk_entities rows instead of distinct non-NULL session_ids.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let friend = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };
        link(&mut store, "c1", &[topic.clone(), friend.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), friend.clone()], 200);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_chunks, 2);
        assert_eq!(n.chunk_count, 2);
        assert_eq!(n.session_count, 0);
    }

    #[test]
    fn entity_neighbors_session_count_counts_distinct_sessions() {
        // `@popular` shares one chunk with `#topic` in session s1 but also
        // appears in s2 alone and a NULL-session chunk. session_count must
        // distinguish "locally noisy in one session" from "recurring across
        // many sessions" — here the spread is 2 (s1+s2), not 3 or 4.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("s1")),
                ("c2", Some("s2")),
                ("c3", Some("s2")),
                ("c4", None),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_chunks, 1);
        assert_eq!(n.chunk_count, 4);
        assert_eq!(
            n.session_count, 2,
            "popular spans sessions s1 + s2; the NULL-session chunk must not count",
        );
    }

    #[test]
    fn entity_neighbors_session_count_is_one_under_session_filter() {
        // Under `--session-id`, every surviving neighbor lives inside that
        // single session by construction, so the spread signal collapses to 1
        // even when the neighbor is globally spread across more sessions.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("s1")),
                ("c2", Some("s1")),
                ("c3", Some("s2")),
                ("c4", Some("s2")),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        // topic + popular co-occur in s1 (c1); popular also lives in s2 alone.
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[topic.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);

        // Global scope: spread is 2 sessions (s1+s2).
        let global =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(global.neighbors.len(), 1);
        assert_eq!(global.neighbors[0].session_count, 2);

        // Session-scoped: collapses to 1 by construction.
        let scoped = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                session_id: Some("s1".into()),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(scoped.neighbors.len(), 1);
        assert_eq!(scoped.neighbors[0].shared_chunks, 1);
        // chunk_count is already session-scoped: popular has c1 + c2 inside s1.
        assert_eq!(scoped.neighbors[0].chunk_count, 2);
        assert_eq!(scoped.neighbors[0].session_count, 1);
    }

    #[test]
    fn entity_neighbors_project_count_is_zero_for_chunks_without_project() {
        // When every linking chunk has NULL project, the neighbor's project
        // spread signal must be 0 — guards against a future refactor that
        // counts raw chunk_entities rows instead of distinct non-NULL
        // projects. Mirrors the session_count NULL guard.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let friend = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };
        link(&mut store, "c1", &[topic.clone(), friend.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), friend.clone()], 200);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_chunks, 2);
        assert_eq!(n.chunk_count, 2);
        assert_eq!(n.project_count, 0);
    }

    #[test]
    fn entity_neighbors_project_count_counts_distinct_projects() {
        // `@popular` shares one chunk with `#topic` in project alpha but also
        // appears in beta alone and a NULL-project chunk. project_count must
        // distinguish "tied to one project" from "spread across projects" —
        // here the spread is 2 (alpha+beta), not 3 or 4.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_projects(
            &mut store,
            &[
                ("c1", Some("alpha")),
                ("c2", Some("beta")),
                ("c3", Some("beta")),
                ("c4", None),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_chunks, 1);
        assert_eq!(n.chunk_count, 4);
        assert_eq!(
            n.project_count, 2,
            "popular spans projects alpha + beta; the NULL-project chunk must not count",
        );
    }

    #[test]
    fn entity_neighbors_project_count_is_independent_of_session_filter() {
        // Unlike `session_count`, `project_count` must NOT collapse to 1 under
        // a `session_id` filter. Project grouping is orthogonal: a single
        // session can still span multiple projects, so the count reflects
        // distinct non-NULL projects among the neighbor's chunks inside that
        // session. Pins the asymmetry against an accidental refactor that
        // would couple the two grouping fields.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        // c1+c2 live in session s1 / project alpha; c3 lives in s1 / project beta;
        // c4 lives in s2 / project gamma so we can verify the scope filter
        // restricts project counting to within-session chunks too.
        for (ord, (id, session, project)) in [
            ("c1", "s1", "alpha"),
            ("c2", "s1", "alpha"),
            ("c3", "s1", "beta"),
            ("c4", "s2", "gamma"),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        // topic + popular co-occur in every chunk so popular surfaces under
        // both global and session-scoped views.
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), popular.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), popular.clone()], 300);
        link(&mut store, "c4", &[topic.clone(), popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);

        // Global scope: popular spans 3 projects (alpha, beta, gamma).
        let global =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(global.neighbors.len(), 1);
        assert_eq!(global.neighbors[0].project_count, 3);
        assert_eq!(global.neighbors[0].session_count, 2);

        // Session-scoped to s1: session_count collapses to 1, but
        // project_count must stay at 2 (alpha + beta inside s1).
        let scoped = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                session_id: Some("s1".into()),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(scoped.neighbors.len(), 1);
        assert_eq!(scoped.neighbors[0].session_count, 1);
        assert_eq!(
            scoped.neighbors[0].project_count, 2,
            "project grouping must not collapse under a session_id filter",
        );
        // chunk_count is session-scoped (c1+c2+c3 inside s1), confirming the
        // project count uses the same scope as chunk_count.
        assert_eq!(scoped.neighbors[0].chunk_count, 3);
    }

    #[test]
    fn entity_neighbors_user_count_is_zero_for_chunks_without_user() {
        // When every linking chunk has NULL user, the neighbor's user spread
        // signal must be 0 — guards against a future refactor that counts raw
        // chunk_entities rows instead of distinct non-NULL users. Mirrors the
        // project_count NULL guard.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks(&mut store, &["c1", "c2"]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let friend = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };
        link(&mut store, "c1", &[topic.clone(), friend.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), friend.clone()], 200);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_chunks, 2);
        assert_eq!(n.chunk_count, 2);
        assert_eq!(n.user_count, 0);
    }

    #[test]
    fn entity_neighbors_user_count_counts_distinct_users() {
        // `@popular` shares one chunk with `#topic` under user alice but also
        // appears under bob alone and a NULL-user chunk. user_count must
        // distinguish "tied to one user" from "spread across users" — here
        // the spread is 2 (alice+bob), not 3 or 4.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_users(
            &mut store,
            &[
                ("c1", Some("alice")),
                ("c2", Some("bob")),
                ("c3", Some("bob")),
                ("c4", None),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[popular.clone()], 200);
        link(&mut store, "c3", &[popular.clone()], 300);
        link(&mut store, "c4", &[popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_chunks, 1);
        assert_eq!(n.chunk_count, 4);
        assert_eq!(
            n.user_count, 2,
            "popular spans users alice + bob; the NULL-user chunk must not count",
        );
    }

    #[test]
    fn entity_neighbors_user_count_is_independent_of_session_filter() {
        // Unlike `session_count`, `user_count` must NOT collapse to 1 under a
        // `session_id` filter. User grouping is orthogonal: one session can
        // legitimately span multiple users (e.g. a shared agent thread), so
        // the count reflects distinct non-NULL users among the neighbor's
        // chunks inside that session. Pins the asymmetry against an
        // accidental refactor that would couple the two grouping fields.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        // c1+c2 live in session s1 / user alice; c3 lives in s1 / user bob;
        // c4 lives in s2 / user carol so we can verify the scope filter
        // restricts user counting to within-session chunks too.
        for (ord, (id, session, user)) in [
            ("c1", "s1", "alice"),
            ("c2", "s1", "alice"),
            ("c3", "s1", "bob"),
            ("c4", "s2", "carol"),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        // topic + popular co-occur in every chunk so popular surfaces under
        // both global and session-scoped views.
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), popular.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), popular.clone()], 300);
        link(&mut store, "c4", &[topic.clone(), popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);

        // Global scope: popular spans 3 users (alice, bob, carol).
        let global =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(global.neighbors.len(), 1);
        assert_eq!(global.neighbors[0].user_count, 3);
        assert_eq!(global.neighbors[0].session_count, 2);

        // Session-scoped to s1: session_count collapses to 1, but user_count
        // must stay at 2 (alice + bob inside s1).
        let scoped = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                session_id: Some("s1".into()),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(scoped.neighbors.len(), 1);
        assert_eq!(scoped.neighbors[0].session_count, 1);
        assert_eq!(
            scoped.neighbors[0].user_count, 2,
            "user grouping must not collapse under a session_id filter",
        );
        // chunk_count is session-scoped (c1+c2+c3 inside s1), confirming the
        // user count uses the same scope as chunk_count.
        assert_eq!(scoped.neighbors[0].chunk_count, 3);
    }

    #[test]
    fn entity_neighbors_session_evidence_is_opt_in_and_zero_keeps_default_shape() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[("c1", Some("s1")), ("c2", Some("s2")), ("c3", Some("s3"))],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        let only_s2 = Entity {
            kind: EntityKind::Mention,
            value: "@only-s2".into(),
        };

        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(
            &mut store,
            "c2",
            &[topic.clone(), shared.clone(), only_s2.clone()],
            200,
        );
        link(&mut store, "c3", &[shared.clone()], 300);

        let topic_id = entity_id(topic.kind, &topic.value);
        let default_report =
            entity_neighbors(store.conn(), &topic_id, &EntityNeighborsOptions::default()).unwrap();
        assert_eq!(default_report.neighbors.len(), 2);
        assert!(
            default_report
                .neighbors
                .iter()
                .all(|n| n.shared_session_ids.is_none())
        );

        let zeroed = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                with_sessions: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(
            zeroed
                .neighbors
                .iter()
                .all(|n| n.shared_session_ids.is_none())
        );
    }

    #[test]
    fn entity_neighbors_session_evidence_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("s3")),
                ("c2", Some("s1")),
                ("c3", Some("s2")),
                ("c4", Some("s4")),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };

        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), shared.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), shared.clone()], 300);
        link(&mut store, "c4", &[shared.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_sessions: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_session_ids.as_ref(),
            Some(&vec!["s1".to_string(), "s2".to_string()]),
        );
    }

    #[test]
    fn entity_neighbors_session_evidence_collapses_under_session_filter() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[("c1", Some("s1")), ("c2", Some("s2")), ("c3", Some("s2"))],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        let only_s2 = Entity {
            kind: EntityKind::Mention,
            value: "@only-s2".into(),
        };

        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), shared.clone()], 200);
        link(&mut store, "c3", &[shared.clone(), only_s2.clone()], 300);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                session_id: Some("s2".into()),
                with_sessions: Some(3),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s2"));
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(report.neighbors[0].value, "@shared");
        assert_eq!(
            report.neighbors[0].shared_session_ids.as_ref(),
            Some(&vec!["s2".to_string()]),
        );
    }

    #[test]
    fn entity_neighbors_project_evidence_is_opt_in_and_zero_keeps_default_shape() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_projects(&mut store, &[("c1", Some("alpha")), ("c2", Some("beta"))]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic, shared], 200);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let default_report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(default_report.neighbors.len(), 1);
        assert!(default_report.neighbors[0].shared_projects.is_none());

        let zeroed = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_projects: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(zeroed.neighbors[0].shared_projects.is_none());
    }

    #[test]
    fn entity_neighbors_project_evidence_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_projects(
            &mut store,
            &[
                ("c1", Some("gamma")),
                ("c2", Some("alpha")),
                ("c3", Some("beta")),
                ("c4", None),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), shared.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), shared.clone()], 300);
        link(&mut store, "c4", &[shared], 400);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_projects: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_projects.as_ref(),
            Some(&vec!["alpha".to_string(), "beta".to_string()]),
        );
    }

    #[test]
    fn entity_neighbors_project_evidence_honors_session_scope_without_collapsing() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, project)) in [
            ("c1", Some("s1"), Some("alpha")),
            ("c2", Some("s1"), Some("beta")),
            ("c3", Some("s2"), Some("gamma")),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), shared.clone()], 200);
        link(&mut store, "c3", &[topic, shared], 300);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                session_id: Some("s1".into()),
                with_projects: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s1"));
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_projects.as_ref(),
            Some(&vec!["alpha".to_string(), "beta".to_string()]),
        );
    }

    #[test]
    fn entity_neighbors_user_evidence_is_opt_in_and_zero_keeps_default_shape() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_users(&mut store, &[("c1", Some("alice")), ("c2", Some("bob"))]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic, shared], 200);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let default_report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(default_report.neighbors.len(), 1);
        assert!(default_report.neighbors[0].shared_users.is_none());

        let zeroed = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_users: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(zeroed.neighbors[0].shared_users.is_none());
    }

    #[test]
    fn entity_neighbors_user_evidence_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_users(
            &mut store,
            &[
                ("c1", Some("carol")),
                ("c2", Some("alice")),
                ("c3", Some("bob")),
                ("c4", None),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), shared.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), shared.clone()], 300);
        link(&mut store, "c4", &[shared], 400);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_users: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_users.as_ref(),
            Some(&vec!["alice".to_string(), "bob".to_string()]),
        );
    }

    #[test]
    fn entity_neighbors_user_evidence_honors_session_scope_without_collapsing() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, user)) in [
            ("c1", Some("s1"), Some("alice")),
            ("c2", Some("s1"), Some("bob")),
            ("c3", Some("s2"), Some("carol")),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone(), shared.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), shared.clone()], 200);
        link(&mut store, "c3", &[topic, shared], 300);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_neighbors(
            store.conn(),
            &topic_id,
            &EntityNeighborsOptions {
                kind: Some(EntityKind::Mention),
                session_id: Some("s1".into()),
                with_users: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s1"));
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_users.as_ref(),
            Some(&vec!["alice".to_string(), "bob".to_string()]),
        );
    }

    #[test]
    fn edge_kind_serializes_to_co_occurs_with() {
        // Pin the JSON wire format: switching from `String` to a typed enum
        // must keep the externally-visible label identical so existing CLI/MCP
        // consumers do not see a behavior change.
        assert_eq!(EdgeKind::CoOccursWith.as_str(), "co_occurs_with");
        let json = serde_json::to_string(&EdgeKind::CoOccursWith).unwrap();
        assert_eq!(json, "\"co_occurs_with\"");
    }

    #[test]
    fn edge_kind_serializes_to_same_session_as() {
        assert_eq!(EdgeKind::SameSessionAs.as_str(), "same_session_as");
        let json = serde_json::to_string(&EdgeKind::SameSessionAs).unwrap();
        assert_eq!(json, "\"same_session_as\"");
    }

    #[test]
    fn entity_session_neighbors_project_evidence_is_omitted_by_default_and_zero() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('session_projects_default_src', 'mem://session-projects-default', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, project)) in [
            ("c1", Some("s1"), Some("alpha")),
            ("c2", Some("s1"), Some("beta")),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'session_projects_default_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone()], 100);
        link(&mut store, "c2", &[shared.clone()], 200);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let default_report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(default_report.neighbors.len(), 1);
        assert!(default_report.neighbors[0].shared_projects.is_none());

        let zeroed = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_projects: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(zeroed.neighbors[0].shared_projects.is_none());
    }

    #[test]
    fn entity_session_neighbors_project_evidence_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('session_projects_src', 'mem://session-projects', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, project)) in [
            ("c1", Some("s2"), Some("gamma")),
            ("c2", Some("s1"), Some("alpha")),
            ("c3", Some("s1"), Some("beta")),
            ("c4", Some("s3"), Some("delta")),
            ("c5", Some("s2"), None),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'session_projects_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone()], 100);
        link(&mut store, "c2", &[topic.clone()], 200);
        link(&mut store, "c3", &[shared.clone()], 300);
        link(&mut store, "c4", &[shared.clone()], 400);
        link(&mut store, "c5", &[shared], 500);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_projects: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_projects.as_ref(),
            Some(&vec!["alpha".to_string(), "beta".to_string()]),
        );
    }

    #[test]
    fn entity_session_neighbors_project_evidence_honors_session_scope_without_collapsing() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('session_scope_src', 'mem://session-scope', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, project)) in [
            ("c1", Some("s1"), Some("alpha")),
            ("c2", Some("s1"), Some("beta")),
            ("c3", Some("s2"), Some("gamma")),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'session_scope_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", &[topic.clone()], 100);
        link(&mut store, "c2", &[shared.clone()], 200);
        link(&mut store, "c3", &[topic, shared], 300);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                session_id: Some("s1".into()),
                with_projects: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s1"));
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_projects.as_ref(),
            Some(&vec!["alpha".to_string(), "beta".to_string()]),
        );
    }

    #[test]
    fn entity_session_neighbors_user_evidence_is_omitted_by_default_and_zero() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('session_users_default_src', 'mem://session-users-default', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, user)) in [
            ("c1", Some("s1"), Some("alice")),
            ("c2", Some("s1"), Some("bob")),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'session_users_default_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&topic), 100);
        link(&mut store, "c2", std::slice::from_ref(&shared), 200);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let omitted = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(omitted.neighbors.len(), 1);
        assert!(omitted.neighbors[0].shared_users.is_none());

        let zeroed = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_users: Some(0),
                ..Default::default()
            },
        )
        .unwrap();
        assert!(zeroed.neighbors[0].shared_users.is_none());
    }

    #[test]
    fn entity_session_neighbors_user_evidence_orders_and_truncates_deterministically() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('session_users_order_src', 'mem://session-users-order', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, user)) in [
            ("c1", Some("s1"), Some("carol")),
            ("c2", Some("s1"), Some("alice")),
            ("c3", Some("s2"), Some("bob")),
            ("c4", Some("s2"), None),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'session_users_order_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&topic), 100);
        link(&mut store, "c2", std::slice::from_ref(&shared), 200);
        link(&mut store, "c3", std::slice::from_ref(&topic), 300);
        link(&mut store, "c4", std::slice::from_ref(&shared), 400);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                with_users: Some(2),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_users.as_ref(),
            Some(&vec!["alice".to_string(), "bob".to_string()]),
        );
    }

    #[test]
    fn entity_session_neighbors_user_evidence_honors_session_scope_without_collapsing() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('session_users_scope_src', 'mem://session-users-scope', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, user)) in [
            ("c1", Some("s1"), Some("alice")),
            ("c2", Some("s1"), Some("bob")),
            ("c3", Some("s2"), Some("carol")),
            ("c4", Some("s2"), Some("dave")),
        ]
        .into_iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'session_users_scope_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let shared = Entity {
            kind: EntityKind::Mention,
            value: "@shared".into(),
        };
        link(&mut store, "c1", std::slice::from_ref(&topic), 100);
        link(&mut store, "c2", std::slice::from_ref(&shared), 200);
        link(&mut store, "c3", std::slice::from_ref(&topic), 300);
        link(&mut store, "c4", std::slice::from_ref(&shared), 400);

        let topic_id = entity_id(EntityKind::Hashtag, "#topic");
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                kind: Some(EntityKind::Mention),
                session_id: Some("s1".into()),
                with_users: Some(5),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(report.session_id.as_deref(), Some("s1"));
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(
            report.neighbors[0].shared_users.as_ref(),
            Some(&vec!["alice".to_string(), "bob".to_string()]),
        );
    }

    #[test]
    fn entity_session_neighbors_ranks_by_shared_session_count() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        // c1+c2 belong to session s1; c3+c4 to s2; c5 to s3; c6 NULL.
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("s1")),
                ("c2", Some("s1")),
                ("c3", Some("s2")),
                ("c4", Some("s2")),
                ("c5", Some("s3")),
                ("c6", None),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let frequent = Entity {
            kind: EntityKind::Mention,
            value: "@frequent".into(),
        };
        let sometimes = Entity {
            kind: EntityKind::Mention,
            value: "@sometimes".into(),
        };
        let unrelated = Entity {
            kind: EntityKind::Url,
            value: "https://unrelated.test".into(),
        };

        // topic in s1 (via c1) and s2 (via c3).
        link(&mut store, "c1", &[topic.clone()], 100);
        link(&mut store, "c3", &[topic.clone()], 200);
        // frequent in s1 (c2) AND s2 (c4) — shares 2 sessions with topic.
        link(&mut store, "c2", &[frequent.clone()], 150);
        link(&mut store, "c4", &[frequent.clone()], 250);
        // sometimes only in s2 (c4) — shares 1 session with topic.
        link(&mut store, "c4", &[sometimes.clone()], 260);
        // unrelated lives only in s3 and untagged c6 — must not surface.
        link(&mut store, "c5", &[unrelated.clone()], 300);
        link(&mut store, "c6", &[unrelated.clone()], 310);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(report.source_value, "#topic");
        assert_eq!(report.source_session_count, 2);
        assert_eq!(report.total_neighbors, 2);
        assert_eq!(report.neighbors.len(), 2);
        assert_eq!(report.neighbors[0].value, "@frequent");
        assert_eq!(report.neighbors[0].edge_kind, EdgeKind::SameSessionAs);
        assert_eq!(report.neighbors[0].shared_sessions, 2);
        // @frequent appears in s1 and s2 across the whole store.
        assert_eq!(report.neighbors[0].session_count, 2);
        assert_eq!(report.neighbors[1].value, "@sometimes");
        assert_eq!(report.neighbors[1].shared_sessions, 1);
        assert_eq!(report.neighbors[1].session_count, 1);
        // The source entity must never appear in its own neighbor list.
        assert!(report.neighbors.iter().all(|n| n.id != topic_id));
    }

    #[test]
    fn entity_session_neighbors_honors_session_id_scope() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("s1")),
                ("c2", Some("s1")),
                ("c3", Some("s2")),
                ("c4", Some("s2")),
            ],
        );

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let frequent = Entity {
            kind: EntityKind::Mention,
            value: "@frequent".into(),
        };
        let only_s2 = Entity {
            kind: EntityKind::Mention,
            value: "@only-s2".into(),
        };

        link(&mut store, "c1", &[topic.clone()], 100);
        link(&mut store, "c2", &[frequent.clone()], 110);
        link(&mut store, "c3", &[topic.clone()], 200);
        link(&mut store, "c4", &[frequent.clone(), only_s2.clone()], 210);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                session_id: Some("s2".into()),
                ..Default::default()
            },
        )
        .unwrap();

        assert_eq!(report.session_id.as_deref(), Some("s2"));
        assert_eq!(report.source_session_count, 1);
        assert_eq!(report.total_neighbors, 2);
        assert_eq!(report.neighbors.len(), 2);
        assert!(report.neighbors.iter().all(|n| n.shared_sessions == 1));
        assert!(report.neighbors.iter().all(|n| n.session_count == 1));
        let values: Vec<&str> = report.neighbors.iter().map(|n| n.value.as_str()).collect();
        assert_eq!(values, vec!["@frequent", "@only-s2"]);

        let missing = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                session_id: Some("missing".into()),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(missing.session_id.as_deref(), Some("missing"));
        assert_eq!(missing.source_session_count, 0);
        assert_eq!(missing.total_neighbors, 0);
        assert!(missing.neighbors.is_empty());
    }

    #[test]
    fn entity_session_neighbors_surfaces_session_overlap_without_chunk_overlap() {
        // Two entities never share a chunk, but both appear inside session s1
        // — chunk-level neighbors would miss this; session-level must catch it.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(&mut store, &[("c1", Some("s1")), ("c2", Some("s1"))]);

        let a = Entity {
            kind: EntityKind::Hashtag,
            value: "#a".into(),
        };
        let b = Entity {
            kind: EntityKind::Mention,
            value: "@b".into(),
        };
        link(&mut store, "c1", &[a.clone()], 100);
        link(&mut store, "c2", &[b.clone()], 200);

        let a_id = entity_id(a.kind, &a.value);
        let chunk_report =
            entity_neighbors(store.conn(), &a_id, &EntityNeighborsOptions::default()).unwrap();
        assert!(chunk_report.neighbors.is_empty(), "no chunk co-occurrence");

        let session_report = entity_session_neighbors(
            store.conn(),
            &a_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(session_report.neighbors.len(), 1);
        assert_eq!(session_report.neighbors[0].value, "@b");
        assert_eq!(session_report.neighbors[0].shared_sessions, 1);
        assert_eq!(
            session_report.neighbors[0].edge_kind,
            EdgeKind::SameSessionAs
        );
    }

    #[test]
    fn entity_session_neighbors_excludes_null_session_chunks() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(&mut store, &[("c1", None), ("c2", None)]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let other = Entity {
            kind: EntityKind::Mention,
            value: "@other".into(),
        };
        // Both entities live ONLY in untagged chunks — no session edges.
        link(&mut store, "c1", &[topic.clone(), other.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), other.clone()], 200);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(report.source_session_count, 0);
        assert_eq!(report.total_neighbors, 0);
        assert!(report.neighbors.is_empty());
    }

    #[test]
    fn entity_session_neighbors_filters_by_kind() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(&mut store, &[("c1", Some("s1")), ("c2", Some("s1"))]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let mention = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };
        let url = Entity {
            kind: EntityKind::Url,
            value: "https://example.test/x".into(),
        };
        link(&mut store, "c1", &[topic.clone()], 100);
        link(&mut store, "c2", &[mention.clone(), url.clone()], 200);

        let topic_id = entity_id(topic.kind, &topic.value);
        let opts = EntitySessionNeighborsOptions {
            kind: Some(EntityKind::Url),
            ..Default::default()
        };
        let report = entity_session_neighbors(store.conn(), &topic_id, &opts).unwrap();
        assert_eq!(report.total_neighbors, 1);
        assert_eq!(report.neighbors.len(), 1);
        assert_eq!(report.neighbors[0].kind, EntityKind::Url);
        assert_eq!(report.neighbors[0].value, "https://example.test/x");
        assert_eq!(report.kind_filter, Some(EntityKind::Url));
    }

    #[test]
    fn entity_session_neighbors_limit_truncates_but_total_is_full() {
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(
            &mut store,
            &[
                ("c1", Some("s1")),
                ("c2", Some("s1")),
                ("c3", Some("s1")),
                ("c4", Some("s1")),
                ("c5", Some("s1")),
            ],
        );

        let hub = Entity {
            kind: EntityKind::Hashtag,
            value: "#hub".into(),
        };
        link(&mut store, "c1", &[hub.clone()], 100);
        for i in 0..4 {
            let n = Entity {
                kind: EntityKind::Url,
                value: format!("https://n{i}.test"),
            };
            link(&mut store, &format!("c{}", i + 2), &[n], 200 + i as i64);
        }

        let hub_id = entity_id(hub.kind, &hub.value);
        let opts = EntitySessionNeighborsOptions {
            limit: Some(2),
            ..Default::default()
        };
        let report = entity_session_neighbors(store.conn(), &hub_id, &opts).unwrap();
        assert_eq!(report.total_neighbors, 4);
        assert_eq!(report.neighbors.len(), 2);
    }

    #[test]
    fn entity_session_neighbors_project_count_is_zero_for_chunks_without_project() {
        // Session-tagged chunks with NULL project must surface
        // `project_count = 0` for every neighbor — guards against a future
        // refactor that would count chunk_entities rows instead of distinct
        // non-NULL projects. Mirrors the existing session_count NULL guard at
        // the session-graph layer.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(&mut store, &[("c1", Some("s1")), ("c2", Some("s2"))]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let friend = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };
        link(&mut store, "c1", &[topic.clone(), friend.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), friend.clone()], 200);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_sessions, 2);
        assert_eq!(n.session_count, 2);
        assert_eq!(n.project_count, 0);
    }

    #[test]
    fn entity_session_neighbors_project_count_counts_distinct_projects_with_mixed_nulls() {
        // Three session-tagged chunks across two projects plus one
        // session-tagged chunk with NULL project. `@popular` co-occurs with
        // `#topic` everywhere; its project spread must distinguish "tied to
        // one project" from "spread across projects" — here the spread is 2
        // (alpha + beta), not 3 (the NULL-project chunk must not count).
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, project)) in [
            ("c1", "s1", Some("alpha")),
            ("c2", "s2", Some("beta")),
            ("c3", "s3", Some("beta")),
            ("c4", "s4", None),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), popular.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), popular.clone()], 300);
        link(&mut store, "c4", &[topic.clone(), popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_sessions, 4);
        assert_eq!(n.session_count, 4);
        assert_eq!(
            n.project_count, 2,
            "popular spans projects alpha + beta; the NULL-project chunk must not count",
        );
    }

    #[test]
    fn entity_session_neighbors_project_count_is_independent_of_session_filter() {
        // Mirrors the chunk-level entity-neighbors orthogonality contract at
        // the session-graph layer: `session_count` collapses to 1 under a
        // `session_id` filter, but `project_count` must NOT — a single
        // session can still span multiple projects. Pins the asymmetry
        // explicitly against an accidental refactor that would couple the
        // two grouping fields.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        // c1+c2 live in session s1 / project alpha; c3 lives in s1 / project
        // beta; c4 lives in s2 / project gamma so we can verify the scope
        // filter restricts project counting to within-session chunks too.
        for (ord, (id, session, project)) in [
            ("c1", "s1", "alpha"),
            ("c2", "s1", "alpha"),
            ("c3", "s1", "beta"),
            ("c4", "s2", "gamma"),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, project, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, project],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        // topic + popular co-occur in every chunk so popular surfaces under
        // both global and session-scoped views.
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), popular.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), popular.clone()], 300);
        link(&mut store, "c4", &[topic.clone(), popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);

        // Global scope: popular spans 3 projects (alpha, beta, gamma) and
        // 2 sessions.
        let global = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(global.neighbors.len(), 1);
        assert_eq!(global.neighbors[0].session_count, 2);
        assert_eq!(global.neighbors[0].project_count, 3);

        // Session-scoped to s1: session_count collapses to 1 by construction,
        // but project_count must stay at 2 (alpha + beta inside s1).
        let scoped = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                session_id: Some("s1".into()),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(scoped.neighbors.len(), 1);
        assert_eq!(scoped.neighbors[0].session_count, 1);
        assert_eq!(
            scoped.neighbors[0].project_count, 2,
            "project grouping must not collapse under a session_id filter",
        );
    }

    #[test]
    fn entity_session_neighbors_user_count_is_zero_for_chunks_without_user() {
        // Session-tagged chunks with NULL user must surface `user_count = 0`
        // for every neighbor — guards against a future refactor that would
        // count chunk_entities rows instead of distinct non-NULL users.
        // Mirrors the existing project_count NULL guard at the session-graph
        // layer.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        seed_chunks_with_sessions(&mut store, &[("c1", Some("s1")), ("c2", Some("s2"))]);

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let friend = Entity {
            kind: EntityKind::Mention,
            value: "@friend".into(),
        };
        link(&mut store, "c1", &[topic.clone(), friend.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), friend.clone()], 200);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_sessions, 2);
        assert_eq!(n.session_count, 2);
        assert_eq!(n.user_count, 0);
    }

    #[test]
    fn entity_session_neighbors_user_count_counts_distinct_users_with_mixed_nulls() {
        // Three session-tagged chunks across two users plus one session-tagged
        // chunk with NULL user. `@popular` co-occurs with `#topic` everywhere;
        // its user spread must distinguish "tied to one user" from "spread
        // across users" — here the spread is 2 (alice + bob), not 3 (the
        // NULL-user chunk must not count).
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        for (ord, (id, session, user)) in [
            ("c1", "s1", Some("alice")),
            ("c2", "s2", Some("bob")),
            ("c3", "s3", Some("bob")),
            ("c4", "s4", None),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), popular.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), popular.clone()], 300);
        link(&mut store, "c4", &[topic.clone(), popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);
        let report = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(report.neighbors.len(), 1);
        let n = &report.neighbors[0];
        assert_eq!(n.shared_sessions, 4);
        assert_eq!(n.session_count, 4);
        assert_eq!(
            n.user_count, 2,
            "popular spans users alice + bob; the NULL-user chunk must not count",
        );
    }

    #[test]
    fn entity_session_neighbors_user_count_is_independent_of_session_filter() {
        // Mirrors the chunk-level entity-neighbors orthogonality contract at
        // the session-graph layer: `session_count` collapses to 1 under a
        // `session_id` filter, but `user_count` must NOT — a single session
        // can legitimately span multiple users (e.g. a shared agent thread).
        // Pins the asymmetry explicitly against an accidental refactor that
        // would couple the two grouping fields.
        let dir = tempdir().unwrap();
        let mut store = Store::initialize(&dir.path().join("store")).unwrap();
        let conn = store.conn_mut();
        let tx = conn.transaction().unwrap();
        tx.execute(
            "INSERT INTO sources (id, uri, path, kind, bytes, content_sha256, mtime_unix, ingested_at)
             VALUES ('list_src', 'mem://list', NULL, 'text/plain', 0, 'deadbeef', NULL, 0)",
            [],
        )
        .unwrap();
        // c1+c2 live in session s1 / user alice; c3 lives in s1 / user bob;
        // c4 lives in s2 / user carol so we can verify the scope filter
        // restricts user counting to within-session chunks too.
        for (ord, (id, session, user)) in [
            ("c1", "s1", "alice"),
            ("c2", "s1", "alice"),
            ("c3", "s1", "bob"),
            ("c4", "s2", "carol"),
        ]
        .iter()
        .enumerate()
        {
            tx.execute(
                "INSERT INTO chunks (id, source_id, ordinal, byte_start, byte_end, char_count, text, sha256, session_id, user, created_at)
                 VALUES (?1, 'list_src', ?2, 0, 0, 0, '', '', ?3, ?4, 0)",
                params![id, ord as i64, session, user],
            )
            .unwrap();
        }
        tx.commit().unwrap();

        let topic = Entity {
            kind: EntityKind::Hashtag,
            value: "#topic".into(),
        };
        let popular = Entity {
            kind: EntityKind::Mention,
            value: "@popular".into(),
        };
        // topic + popular co-occur in every chunk so popular surfaces under
        // both global and session-scoped views.
        link(&mut store, "c1", &[topic.clone(), popular.clone()], 100);
        link(&mut store, "c2", &[topic.clone(), popular.clone()], 200);
        link(&mut store, "c3", &[topic.clone(), popular.clone()], 300);
        link(&mut store, "c4", &[topic.clone(), popular.clone()], 400);

        let topic_id = entity_id(topic.kind, &topic.value);

        // Global scope: popular spans 3 users (alice, bob, carol) and 2
        // sessions.
        let global = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap();
        assert_eq!(global.neighbors.len(), 1);
        assert_eq!(global.neighbors[0].session_count, 2);
        assert_eq!(global.neighbors[0].user_count, 3);

        // Session-scoped to s1: session_count collapses to 1 by construction,
        // but user_count must stay at 2 (alice + bob inside s1).
        let scoped = entity_session_neighbors(
            store.conn(),
            &topic_id,
            &EntitySessionNeighborsOptions {
                session_id: Some("s1".into()),
                ..Default::default()
            },
        )
        .unwrap();
        assert_eq!(scoped.neighbors.len(), 1);
        assert_eq!(scoped.neighbors[0].session_count, 1);
        assert_eq!(
            scoped.neighbors[0].user_count, 2,
            "user grouping must not collapse under a session_id filter",
        );
    }

    #[test]
    fn entity_session_neighbors_errors_on_unknown_id() {
        let dir = tempdir().unwrap();
        let store = Store::initialize(&dir.path().join("store")).unwrap();
        let err = entity_session_neighbors(
            store.conn(),
            "nope",
            &EntitySessionNeighborsOptions::default(),
        )
        .unwrap_err();
        assert!(err.to_string().contains("no entity with id"));
    }

    #[test]
    fn entity_kind_from_str_round_trips_known_kinds() {
        for k in [
            EntityKind::Url,
            EntityKind::Repo,
            EntityKind::Domain,
            EntityKind::Email,
            EntityKind::FilePath,
            EntityKind::Mention,
            EntityKind::Hashtag,
        ] {
            assert_eq!(entity_kind_from_str(k.as_str()).unwrap(), k);
        }
        // Case insensitive.
        assert_eq!(entity_kind_from_str("URL").unwrap(), EntityKind::Url);
        assert_eq!(entity_kind_from_str("Domain").unwrap(), EntityKind::Domain);
        assert!(entity_kind_from_str("nope").is_err());
    }
}