zeph_memory/graph/store/

mod.rs

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

//! SQLite-backed knowledge graph store.
//!
//! [`GraphStore`] manages the `graph_entities`, `graph_edges`, `graph_entity_aliases`,
//! `graph_communities`, and `graph_episodes` tables. It is the persistence layer for
//! the MAGMA / GAAMA graph subsystem.

use std::collections::HashMap;
#[allow(unused_imports)]
use zeph_db::sql;

use futures::Stream;
use zeph_db::fts::sanitize_fts_query;
use zeph_db::{ActiveDialect, DbPool, numbered_placeholder, placeholder_list};

use crate::error::MemoryError;
use crate::graph::conflict::{ApexMetrics, SUPERSEDE_DEPTH_CAP};
use crate::types::{EntityId, MessageId};

use super::types::{
    Community, Edge, EdgeType, Entity, EntityAlias, EntityType, GraphProvenance, provenance_parts,
};

/// SQLite-backed persistence layer for the knowledge graph.
///
/// All graph mutations go through this type: entity upserts, edge creation,
/// alias recording, community assignment, and episode management.
///
/// Obtain an instance via [`GraphStore::new`] after running the `zeph-db` migrations.
pub struct GraphStore {
    pool: DbPool,
    /// Benna-Fusi fast-variable learning rate. Range: `(0.0, 1.0]`. Default: `0.5`.
    benna_fast_rate: f32,
    /// Benna-Fusi slow-variable learning rate. Range: `(0.0, 1.0]`. Default: `0.05`.
    benna_slow_rate: f32,
    /// When `false`, edges with imported (non-conversation) origins are excluded from
    /// `query_batch_edges` and `edges_for_entity` results (spec-067 FR-003, INV-3). Default: `true`.
    recall_include_imported: bool,
}

/// Safe chunk size for `SQLite` ID batches, staying under `SQLite`'s
/// `SQLITE_MAX_VARIABLE_NUMBER` limit (999 by default): `999 / 2 = 499`, rounded down to 490
/// for headroom. Sized for call sites that bind each id twice per row (e.g. an `IN (...)`
/// clause matched against both `source_entity_id` and `target_entity_id`); call sites that
/// only bind each id once reuse the same constant for a uniform, still-safe batch size.
const SQLITE_BATCH_LIMIT_2X: usize = 490;

impl GraphStore {
    /// Create a new `GraphStore` wrapping `pool`.
    ///
    /// The pool must come from a database with the `zeph-db` migrations applied.
    #[must_use]
    pub fn new(pool: DbPool) -> Self {
        Self {
            pool,
            benna_fast_rate: 0.5,
            benna_slow_rate: 0.05,
            recall_include_imported: true,
        }
    }

    /// Set Benna-Fusi learning rates for the multi-timescale synaptic update (#3709).
    ///
    /// Call this after [`Self::new`] when the rates come from `[memory.graph.spreading_activation]`.
    #[must_use]
    pub fn with_benna_rates(mut self, fast_rate: f32, slow_rate: f32) -> Self {
        self.benna_fast_rate = fast_rate;
        self.benna_slow_rate = slow_rate;
        self
    }

    /// Control whether imported (non-conversation) edges appear in recall (spec-067 FR-003).
    ///
    /// When `false`, edges with any non-conversation origin (e.g. `'ingest'`, `'subagent'`)
    /// are excluded from [`Self::query_batch_edges`] and [`Self::edges_for_entity`].
    /// Default: `true` (all edges included).
    #[must_use]
    pub fn with_recall_include_imported(mut self, include: bool) -> Self {
        self.recall_include_imported = include;
        self
    }

    /// Access the underlying database pool.
    #[must_use]
    pub fn pool(&self) -> &DbPool {
        &self.pool
    }

    // ── Entities ─────────────────────────────────────────────────────────────

    /// Insert or update an entity by `(canonical_name, entity_type)`.
    ///
    /// - `surface_name`: the original display form (e.g. `"Rust"`) — stored in the `name` column
    ///   so user-facing output preserves casing. Updated on every upsert to the latest seen form.
    /// - `canonical_name`: the stable normalized key (e.g. `"rust"`) — used for deduplication.
    /// - `summary`: pass `None` to preserve the existing summary; pass `Some("")` to blank it.
    /// - `provenance`: pass `None` for conversation-origin rows (the default); pass
    ///   `Some(prov)` from the ingest path to stamp `origin` and `import_batch_id`.
    ///   The `DO UPDATE` clause intentionally does NOT overwrite `origin` on conflict —
    ///   an entity first seen in conversation that reappears in an import stays `conversation`.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.upsert_entity", skip_all)]
    pub async fn upsert_entity(
        &self,
        surface_name: &str,
        canonical_name: &str,
        entity_type: EntityType,
        summary: Option<&str>,
        provenance: Option<&GraphProvenance>,
    ) -> Result<EntityId, MemoryError> {
        let type_str = entity_type.as_str();
        let (origin, batch_id, _source_uri) = provenance_parts(provenance);
        let id: i64 = zeph_db::query_scalar(sql!(
            "INSERT INTO graph_entities (name, canonical_name, entity_type, summary, origin, import_batch_id)
             VALUES (?, ?, ?, ?, ?, ?)
             ON CONFLICT(canonical_name, entity_type) DO UPDATE SET
               name = excluded.name,
               summary = COALESCE(excluded.summary, graph_entities.summary),
               last_seen_at = CURRENT_TIMESTAMP
             RETURNING id"
        ))
        .bind(surface_name)
        .bind(canonical_name)
        .bind(type_str)
        .bind(summary)
        .bind(origin)
        .bind(batch_id)
        .fetch_one(&self.pool)
        .await?;
        Ok(EntityId(id))
    }

    /// Find an entity by exact canonical name and type.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.find_entity", skip_all)]
    pub async fn find_entity(
        &self,
        canonical_name: &str,
        entity_type: EntityType,
    ) -> Result<Option<Entity>, MemoryError> {
        let type_str = entity_type.as_str();
        let raw = format!(
            "SELECT {} FROM graph_entities WHERE canonical_name = ? AND entity_type = ?",
            entity_select_cols()
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let row: Option<EntityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(canonical_name)
            .bind(type_str)
            .fetch_optional(&self.pool)
            .await?;
        row.map(entity_from_row).transpose()
    }

    /// Find an entity by its numeric ID.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.find_entity_by_id", skip_all)]
    pub async fn find_entity_by_id(&self, entity_id: i64) -> Result<Option<Entity>, MemoryError> {
        let raw = format!(
            "SELECT {} FROM graph_entities WHERE id = ?",
            entity_select_cols()
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let row: Option<EntityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(entity_id)
            .fetch_optional(&self.pool)
            .await?;
        row.map(entity_from_row).transpose()
    }

    /// Resolve entity IDs to their canonical names in a single batched query.
    ///
    /// Ids with no matching row are silently omitted from the result map. Requests are
    /// chunked at 490 ids to stay under the `SQLite` bind-parameter limit, mirroring the
    /// chunking used by [`Self::entity_ids_in`] and [`Self::qdrant_point_ids_for_entities`].
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.resolve_entity_names", skip_all, fields(count = ids.len()))]
    pub async fn resolve_entity_names(
        &self,
        ids: &[i64],
    ) -> Result<HashMap<i64, String>, MemoryError> {
        let mut result: HashMap<i64, String> = HashMap::with_capacity(ids.len());
        if ids.is_empty() {
            return Ok(result);
        }
        for chunk in ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let placeholders = placeholder_list(1, chunk.len());
            let sql = format!(
                "SELECT id, canonical_name FROM graph_entities WHERE id IN ({placeholders})"
            );
            let mut q = zeph_db::query_as::<_, (i64, String)>(sqlx::AssertSqlSafe(sql));
            for id in chunk {
                q = q.bind(*id);
            }
            let rows: Vec<(i64, String)> = q.fetch_all(&self.pool).await?;
            for (id, name) in rows {
                result.insert(id, name);
            }
        }
        Ok(result)
    }

    /// Update the `qdrant_point_id` for an entity.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.set_entity_qdrant_point_id", skip_all)]
    pub async fn set_entity_qdrant_point_id(
        &self,
        entity_id: i64,
        point_id: &str,
    ) -> Result<(), MemoryError> {
        zeph_db::query(sql!(
            "UPDATE graph_entities SET qdrant_point_id = ? WHERE id = ?"
        ))
        .bind(point_id)
        .bind(entity_id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Find entities matching `query` in name, summary, or aliases, up to `limit` results, ranked by relevance.
    ///
    /// Uses FTS5 MATCH with prefix wildcards (`token*`) and bm25 ranking. Name matches are
    /// weighted 10x higher than summary matches. Also searches `graph_entity_aliases` for
    /// alias matches via a UNION query.
    ///
    /// # Behavioral note
    ///
    /// This replaces the previous `LIKE '%query%'` implementation. FTS5 prefix matching differs
    /// from substring matching: searching "SQL" will match "`SQLite`" (prefix) but NOT "`GraphQL`"
    /// (substring). Entity names are indexed as single tokens by the unicode61 tokenizer, so
    /// mid-word substrings are not matched. This is a known trade-off for index performance.
    ///
    /// Single-character queries (e.g., "a") are allowed and produce a broad prefix match ("a*").
    /// The `limit` parameter caps the result set. No minimum query length is enforced; if this
    /// causes noise in practice, add a minimum length guard at the call site.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.find_entities_fuzzy", skip_all)]
    pub async fn find_entities_fuzzy(
        &self,
        query: &str,
        limit: usize,
    ) -> Result<Vec<Entity>, MemoryError> {
        // FTS5 boolean operator keywords (case-sensitive uppercase). Filtering these
        // prevents syntax errors when user input contains them as literal search terms
        // (e.g., "graph OR unrelated" must not produce "graph* OR* unrelated*").
        const FTS5_OPERATORS: &[&str] = &["AND", "OR", "NOT", "NEAR"];
        let query = &query[..query.floor_char_boundary(512)];
        // Sanitize input: split on non-alphanumeric characters, filter empty tokens,
        // append '*' to each token for FTS5 prefix matching ("graph" -> "graph*").
        let sanitized = sanitize_fts_query(query);
        if sanitized.is_empty() {
            return Ok(vec![]);
        }
        let fts_query: String = sanitized
            .split_whitespace()
            .filter(|t| !FTS5_OPERATORS.contains(t))
            .map(|t| format!("{t}*"))
            .collect::<Vec<_>>()
            .join(" ");
        if fts_query.is_empty() {
            return Ok(vec![]);
        }

        let limit = i64::try_from(limit)?;
        // bm25(graph_entities_fts, 10.0, 1.0): name column weighted 10x over summary.
        // bm25() returns negative values; ORDER BY ASC puts best matches first.
        let search_sql = format!(
            "SELECT DISTINCT e.id, e.name, e.canonical_name, e.entity_type, e.summary, \
                    e.first_seen_at, e.last_seen_at, e.qdrant_point_id \
             FROM graph_entities_fts fts \
             JOIN graph_entities e ON e.id = fts.rowid \
             WHERE graph_entities_fts MATCH ? \
             UNION \
             SELECT e.id, e.name, e.canonical_name, e.entity_type, e.summary, \
                    e.first_seen_at, e.last_seen_at, e.qdrant_point_id \
             FROM graph_entity_aliases a \
             JOIN graph_entities e ON e.id = a.entity_id \
             WHERE a.alias_name LIKE ? ESCAPE '\\' {} \
             LIMIT ?",
            <ActiveDialect as zeph_db::dialect::Dialect>::COLLATE_NOCASE,
        );
        let rows: Vec<EntityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(search_sql))
            .bind(&fts_query)
            .bind(format!(
                "%{}%",
                query
                    .trim()
                    .replace('\\', "\\\\")
                    .replace('%', "\\%")
                    .replace('_', "\\_")
            ))
            .bind(limit)
            .fetch_all(&self.pool)
            .await?;
        rows.into_iter()
            .map(entity_from_row)
            .collect::<Result<Vec<_>, _>>()
    }

    /// Flush the `SQLite` WAL to the main database file.
    ///
    /// Runs `PRAGMA wal_checkpoint(PASSIVE)`. Safe to call at any time; does not block active
    /// readers or writers. Call after bulk entity inserts to ensure FTS5 shadow table writes are
    /// visible to connections opened in future sessions.
    ///
    /// # Errors
    ///
    /// Returns an error if the PRAGMA execution fails.
    #[cfg(feature = "sqlite")]
    #[tracing::instrument(name = "memory.graph.store.checkpoint_wal", skip_all)]
    pub async fn checkpoint_wal(&self) -> Result<(), MemoryError> {
        zeph_db::query("PRAGMA wal_checkpoint(PASSIVE)")
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// No-op on `PostgreSQL` (WAL management is handled by the server).
    ///
    /// # Errors
    ///
    /// Never returns an error.
    #[cfg(feature = "postgres")]
    #[allow(clippy::unused_async)]
    #[tracing::instrument(name = "memory.graph.store.checkpoint_wal", skip_all)]
    pub async fn checkpoint_wal(&self) -> Result<(), MemoryError> {
        Ok(())
    }

    /// Stream all entities from the database incrementally (true cursor, no full-table load).
    pub fn all_entities_stream(&self) -> impl Stream<Item = Result<Entity, MemoryError>> + '_ {
        use futures::StreamExt as _;
        let raw = format!(
            "SELECT {} FROM graph_entities ORDER BY id ASC",
            entity_select_cols()
        );
        zeph_db::query_as::<_, EntityRow>(sqlx::AssertSqlSafe(raw))
            .fetch(&self.pool)
            .map(|r: Result<EntityRow, zeph_db::SqlxError>| {
                r.map_err(MemoryError::from).and_then(entity_from_row)
            })
    }

    // ── Alias methods ─────────────────────────────────────────────────────────

    /// Insert an alias for an entity (idempotent: duplicate alias is silently ignored via UNIQUE constraint).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.add_alias", skip_all)]
    pub async fn add_alias(&self, entity_id: i64, alias_name: &str) -> Result<(), MemoryError> {
        let insert_alias_sql = zeph_db::rewrite_placeholders(&format!(
            "{} INTO graph_entity_aliases (entity_id, alias_name) VALUES (?, ?){}",
            <ActiveDialect as zeph_db::dialect::Dialect>::INSERT_IGNORE,
            <ActiveDialect as zeph_db::dialect::Dialect>::CONFLICT_NOTHING,
        ));
        zeph_db::query(sqlx::AssertSqlSafe(insert_alias_sql))
            .bind(entity_id)
            .bind(alias_name)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// Find an entity by alias name and entity type (case-insensitive).
    ///
    /// Filters by `entity_type` to avoid cross-type alias collisions (S2 fix).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.find_entity_by_alias", skip_all)]
    pub async fn find_entity_by_alias(
        &self,
        alias_name: &str,
        entity_type: EntityType,
    ) -> Result<Option<Entity>, MemoryError> {
        let type_str = entity_type.as_str();
        let first_seen_sel =
            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("e.first_seen_at");
        let last_seen_sel =
            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("e.last_seen_at");
        let alias_typed_sql = zeph_db::rewrite_placeholders(&format!(
            "SELECT e.id, e.name, e.canonical_name, e.entity_type, e.summary, \
                    {first_seen_sel} AS first_seen_at, {last_seen_sel} AS last_seen_at, \
                    e.qdrant_point_id \
             FROM graph_entity_aliases a \
             JOIN graph_entities e ON e.id = a.entity_id \
             WHERE a.alias_name = ? {} \
               AND e.entity_type = ? \
             ORDER BY e.id ASC \
             LIMIT 1",
            <ActiveDialect as zeph_db::dialect::Dialect>::COLLATE_NOCASE,
        ));
        let row: Option<EntityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(alias_typed_sql))
            .bind(alias_name)
            .bind(type_str)
            .fetch_optional(&self.pool)
            .await?;
        row.map(entity_from_row).transpose()
    }

    /// Get all aliases for an entity.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.aliases_for_entity", skip_all)]
    pub async fn aliases_for_entity(
        &self,
        entity_id: i64,
    ) -> Result<Vec<EntityAlias>, MemoryError> {
        let sql = zeph_db::rewrite_placeholders(&format!(
            "SELECT {} FROM graph_entity_aliases WHERE entity_id = ? ORDER BY id ASC",
            alias_select_cols()
        ));
        let rows: Vec<AliasRow> = zeph_db::query_as(sqlx::AssertSqlSafe(sql))
            .bind(entity_id)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(alias_from_row).collect())
    }

    /// Collect all entities into a Vec.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or `entity_type` parsing fails.
    #[tracing::instrument(name = "memory.graph.store.all_entities", skip_all)]
    pub async fn all_entities(&self) -> Result<Vec<Entity>, MemoryError> {
        use futures::TryStreamExt as _;
        self.all_entities_stream().try_collect().await
    }

    /// Count the total number of entities.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.entity_count", skip_all)]
    pub async fn entity_count(&self) -> Result<i64, MemoryError> {
        let count: i64 = zeph_db::query_scalar(sql!("SELECT COUNT(*) FROM graph_entities"))
            .fetch_one(&self.pool)
            .await?;
        Ok(count)
    }

    // ── Edges ─────────────────────────────────────────────────────────────────

    /// Insert a new edge between two entities, or update the existing active edge.
    ///
    /// An active edge is identified by `(source_entity_id, target_entity_id, relation, edge_type)`
    /// with `valid_to IS NULL`. If such an edge already exists, its `confidence` is updated to the
    /// maximum of the stored and incoming values, and the existing id is returned. This prevents
    /// duplicate edges from repeated extraction of the same context messages.
    ///
    /// The dedup key includes `edge_type` (critic mitigation): the same `(source, target, relation)`
    /// triple can legitimately exist with different edge types (e.g., `depends_on` can be both
    /// Semantic and Causal). Without `edge_type` in the key, the second insertion would silently
    /// update the first and lose the type classification.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[allow(clippy::too_many_arguments)]
    #[tracing::instrument(name = "memory.graph.store.insert_edge", skip_all)]
    pub async fn insert_edge(
        &self,
        source_entity_id: i64,
        target_entity_id: i64,
        relation: &str,
        fact: &str,
        confidence: f32,
        episode_id: Option<MessageId>,
        provenance: Option<&GraphProvenance>,
    ) -> Result<i64, MemoryError> {
        self.insert_edge_typed(
            source_entity_id,
            target_entity_id,
            relation,
            fact,
            confidence,
            episode_id,
            EdgeType::Semantic,
            None,
            provenance,
        )
        .await
    }

    /// Insert a typed edge between two entities, or update the existing active edge of the same type.
    ///
    /// Identical semantics to [`Self::insert_edge`] but with an explicit `edge_type` parameter.
    /// The dedup key is `(source_entity_id, target_entity_id, relation, edge_type, valid_to IS NULL)`.
    ///
    /// `turn_index` (#5784) is stored only on the newly inserted row; the dedup UPDATE branch
    /// (re-encountering an identical active edge) does not overwrite it, matching
    /// [`Self::insert_or_supersede_with_turn_index_and_metrics`]'s reassertion semantics.
    ///
    /// The `provenance` parameter stamps `origin`, `import_batch_id`, and `source_uri` on the new
    /// row. Pass `None` for conversation-origin edges (the default). The dedup UPDATE branch does
    /// NOT overwrite provenance — an existing edge keeps its original origin on re-encounter.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[allow(clippy::too_many_arguments)]
    // function with many required inputs; a *Params struct would be more verbose without simplifying the call site
    #[tracing::instrument(name = "memory.graph.store.insert_edge_typed", skip_all)]
    pub async fn insert_edge_typed(
        &self,
        source_entity_id: i64,
        target_entity_id: i64,
        relation: &str,
        fact: &str,
        confidence: f32,
        episode_id: Option<MessageId>,
        edge_type: EdgeType,
        turn_index: Option<u32>,
        provenance: Option<&GraphProvenance>,
    ) -> Result<i64, MemoryError> {
        if source_entity_id == target_entity_id {
            return Err(MemoryError::InvalidInput(format!(
                "self-loop edge rejected: source and target are the same entity (id={source_entity_id})"
            )));
        }
        let confidence = confidence.clamp(0.0, 1.0);
        let edge_type_str = edge_type.as_str();
        let (origin, batch_id, source_uri) = provenance_parts(provenance);

        // Wrap SELECT + INSERT/UPDATE in a single transaction to eliminate the race window
        // between existence check and write. The unique partial index uq_graph_edges_active
        // covers (source, target, relation, edge_type) WHERE valid_to IS NULL; SQLite does not
        // support ON CONFLICT DO UPDATE against partial indexes, so we keep two statements.
        let mut tx = zeph_db::begin(&self.pool).await?;

        let existing: Option<(i64, f64, f64, f64)> = zeph_db::query_as(sql!(
            "SELECT id, confidence, CAST(confidence_fast AS DOUBLE PRECISION),
                    CAST(confidence_slow AS DOUBLE PRECISION) FROM graph_edges
             WHERE source_entity_id = ?
               AND target_entity_id = ?
               AND relation = ?
               AND edge_type = ?
               AND valid_to IS NULL
             LIMIT 1"
        ))
        .bind(source_entity_id)
        .bind(target_entity_id)
        .bind(relation)
        .bind(edge_type_str)
        .fetch_optional(&mut *tx)
        .await?;

        if let Some((existing_id, stored_conf, stored_fast, stored_slow)) = existing {
            let updated_conf = f64::from(confidence).max(stored_conf);
            // Benna-Fusi multi-timescale update (#3709):
            // fast tracks incoming confidence at high plasticity; slow integrates fast at high retention.
            // Clamp DB-read values to [0, 1] as defense-in-depth against corrupted rows.
            #[allow(clippy::cast_possible_truncation)]
            let stored_fast_f32 = (stored_fast as f32).clamp(0.0, 1.0);
            #[allow(clippy::cast_possible_truncation)]
            let stored_slow_f32 = (stored_slow as f32).clamp(0.0, 1.0);
            let new_fast = stored_fast_f32 + self.benna_fast_rate * (confidence - stored_fast_f32);
            let new_slow = stored_slow_f32 + self.benna_slow_rate * (new_fast - stored_slow_f32);
            // DO NOT update origin/import_batch_id/source_uri — the existing row keeps its origin.
            zeph_db::query(sql!(
                "UPDATE graph_edges SET confidence = ?, confidence_fast = ?, confidence_slow = ? WHERE id = ?"
            ))
            .bind(updated_conf)
            .bind(f64::from(new_fast))
            .bind(f64::from(new_slow))
            .bind(existing_id)
            .execute(&mut *tx)
            .await?;
            tx.commit().await?;
            return Ok(existing_id);
        }

        let episode_raw: Option<i64> = episode_id.map(|m| m.0);
        let turn_index_raw: Option<i64> = turn_index.map(i64::from);
        let id: i64 = zeph_db::query_scalar(sql!(
            "INSERT INTO graph_edges
             (source_entity_id, target_entity_id, relation, fact, confidence,
              confidence_fast, confidence_slow, episode_id, edge_type, turn_index,
              origin, import_batch_id, source_uri)
             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
             RETURNING id"
        ))
        .bind(source_entity_id)
        .bind(target_entity_id)
        .bind(relation)
        .bind(fact)
        .bind(f64::from(confidence))
        .bind(f64::from(confidence))
        .bind(f64::from(confidence))
        .bind(episode_raw)
        .bind(edge_type_str)
        .bind(turn_index_raw)
        .bind(origin)
        .bind(batch_id)
        .bind(source_uri)
        .fetch_one(&mut *tx)
        .await?;
        tx.commit().await?;
        Ok(id)
    }

    /// Mark an edge as invalid (set `valid_to` and `expired_at` to now).
    ///
    /// # Errors
    ///
    /// Returns an error if the database update fails.
    #[tracing::instrument(name = "memory.graph.store.invalidate_edge", skip_all)]
    pub async fn invalidate_edge(&self, edge_id: i64) -> Result<(), MemoryError> {
        zeph_db::query(sql!(
            "UPDATE graph_edges SET valid_to = CURRENT_TIMESTAMP, expired_at = CURRENT_TIMESTAMP
             WHERE id = ?"
        ))
        .bind(edge_id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Invalidate an edge and record the supersession pointer for Kumiho belief revision audit trail.
    ///
    /// Sets `valid_to`, `expired_at`, and `superseded_by` on the old edge to link it to its replacement.
    ///
    /// # Errors
    ///
    /// Returns an error if the database update fails.
    #[tracing::instrument(
        name = "memory.graph.store.invalidate_edge_with_supersession",
        skip_all
    )]
    pub async fn invalidate_edge_with_supersession(
        &self,
        old_edge_id: i64,
        new_edge_id: i64,
    ) -> Result<(), MemoryError> {
        zeph_db::query(sql!(
            "UPDATE graph_edges
             SET valid_to = CURRENT_TIMESTAMP,
                 expired_at = CURRENT_TIMESTAMP,
                 superseded_by = ?
             WHERE id = ?"
        ))
        .bind(new_edge_id)
        .bind(old_edge_id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Get all active edges for a batch of entity IDs, with optional MAGMA edge type filtering.
    ///
    /// Fetches all currently-active edges (`valid_to IS NULL`) where either endpoint
    /// is in `entity_ids`. Traversal is always current-time only (no `at_timestamp` support
    /// in v1 — see `bfs_at_timestamp` for historical traversal).
    ///
    /// # `SQLite` bind limit safety
    ///
    /// `SQLite` limits the number of bind parameters to `SQLITE_MAX_VARIABLE_NUMBER` (999 by
    /// default). Each entity ID requires two bind slots (source OR target), so batches are
    /// chunked at [`SQLITE_BATCH_LIMIT_2X`] to stay safely under the limit regardless of
    /// compile-time `SQLite` configuration.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.edges_for_entities", skip_all, fields(count = entity_ids.len()))]
    pub async fn edges_for_entities(
        &self,
        entity_ids: &[i64],
        edge_types: &[super::types::EdgeType],
    ) -> Result<Vec<Edge>, MemoryError> {
        let mut all_edges: Vec<Edge> = Vec::new();

        for chunk in entity_ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let edges = self.query_batch_edges(chunk, edge_types).await?;
            all_edges.extend(edges);
        }

        Ok(all_edges)
    }

    /// Query active edges for a single chunk of entity IDs (internal helper).
    ///
    /// Caller is responsible for ensuring `entity_ids.len() <= SQLITE_BATCH_LIMIT_2X`.
    ///
    /// # Errors
    ///
    /// Returns an error if any database query fails.
    #[tracing::instrument(name = "memory.graph.store.query_batch_edges", skip_all)]
    async fn query_batch_edges(
        &self,
        entity_ids: &[i64],
        edge_types: &[super::types::EdgeType],
    ) -> Result<Vec<Edge>, MemoryError> {
        if entity_ids.is_empty() {
            return Ok(Vec::new());
        }

        // Build a parameterized IN clause with backend-appropriate placeholders.
        // We cannot use the sql! macro here because the placeholder count is dynamic.
        let n_ids = entity_ids.len();
        let n_types = edge_types.len();

        // Append recall_include_imported filter when ingest-origin edges should be excluded.
        let origin_filter = if self.recall_include_imported {
            String::new()
        } else {
            " AND origin = 'conversation'".to_owned()
        };

        let edge_cols = edge_select_cols("");
        let sql = if n_types == 0 {
            // placeholders used twice (source IN and target IN)
            let placeholders = placeholder_list(1, n_ids);
            let placeholders2 = placeholder_list(n_ids + 1, n_ids);
            format!(
                "SELECT {edge_cols}
                 FROM graph_edges
                 WHERE valid_to IS NULL{origin_filter}
                   AND (source_entity_id IN ({placeholders}) OR target_entity_id IN ({placeholders2}))"
            )
        } else {
            let placeholders = placeholder_list(1, n_ids);
            let placeholders2 = placeholder_list(n_ids + 1, n_ids);
            let type_placeholders = placeholder_list(n_ids * 2 + 1, n_types);
            format!(
                "SELECT {edge_cols}
                 FROM graph_edges
                 WHERE valid_to IS NULL{origin_filter}
                   AND (source_entity_id IN ({placeholders}) OR target_entity_id IN ({placeholders2}))
                   AND edge_type IN ({type_placeholders})"
            )
        };

        // Bind entity IDs twice (source IN and target IN clauses) then edge types.
        let mut query = zeph_db::query_as::<_, EdgeRow>(sqlx::AssertSqlSafe(sql));
        for id in entity_ids {
            query = query.bind(*id);
        }
        for id in entity_ids {
            query = query.bind(*id);
        }
        for et in edge_types {
            query = query.bind(et.as_str());
        }

        // Bound worst-case per-chunk blocking on pool.acquire() + query execution: a
        // stuck pool turns into a typed MemoryError::Timeout instead of an indefinite
        // hang, so the caller (SemanticMemory recall) can fail fast and proceed.
        let rows: Vec<EdgeRow> = tokio::time::timeout(
            std::time::Duration::from_millis(500),
            query.fetch_all(&self.pool),
        )
        .await
        .map_err(|_| MemoryError::Timeout("graph pool acquire timed out after 500ms".into()))??;
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    /// Get all active edges where entity is source or target.
    ///
    /// When `recall_include_imported` is `false`, imported (non-conversation) edges are excluded.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.edges_for_entity", skip_all)]
    pub async fn edges_for_entity(&self, entity_id: i64) -> Result<Vec<Edge>, MemoryError> {
        let origin_filter = if self.recall_include_imported {
            ""
        } else {
            " AND origin = 'conversation'"
        };
        let sql = format!(
            "SELECT {} \
             FROM graph_edges \
             WHERE valid_to IS NULL{origin_filter} \
               AND (source_entity_id = ? OR target_entity_id = ?)",
            edge_select_cols("")
        );
        let query_sql = zeph_db::rewrite_placeholders(&sql);
        let rows: Vec<EdgeRow> = zeph_db::query_as::<_, EdgeRow>(sqlx::AssertSqlSafe(query_sql))
            .bind(entity_id)
            .bind(entity_id)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    /// Get all edges (active and expired) where entity is source or target, ordered by
    /// `valid_from DESC`. Used by the `/graph history <name>` slash command.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or if `limit` overflows `i64`.
    // Raw access — does not apply recall_include_imported filter; for filtered recall use query_batch_edges / edges_for_entity.
    #[tracing::instrument(name = "memory.graph.store.edge_history_for_entity", skip_all)]
    pub async fn edge_history_for_entity(
        &self,
        entity_id: i64,
        limit: usize,
    ) -> Result<Vec<Edge>, MemoryError> {
        let limit = i64::try_from(limit)?;
        let raw = format!(
            "SELECT {} \
             FROM graph_edges \
             WHERE source_entity_id = ? OR target_entity_id = ? \
             ORDER BY graph_edges.valid_from DESC \
             LIMIT ?",
            edge_select_cols("")
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let rows: Vec<EdgeRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(entity_id)
            .bind(entity_id)
            .bind(limit)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    /// Get all active edges between two entities (both directions).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    // Raw access — does not apply recall_include_imported filter; for filtered recall use query_batch_edges / edges_for_entity.
    #[tracing::instrument(name = "memory.graph.store.edges_between", skip_all)]
    pub async fn edges_between(
        &self,
        entity_a: i64,
        entity_b: i64,
    ) -> Result<Vec<Edge>, MemoryError> {
        let raw = format!(
            "SELECT {} \
             FROM graph_edges \
             WHERE valid_to IS NULL \
               AND ((source_entity_id = ? AND target_entity_id = ?) \
                 OR (source_entity_id = ? AND target_entity_id = ?))",
            edge_select_cols("")
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let rows: Vec<EdgeRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(entity_a)
            .bind(entity_b)
            .bind(entity_b)
            .bind(entity_a)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    /// Get active edges from `source` to `target` in the exact direction (no reverse).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    // Raw access — does not apply recall_include_imported filter; for filtered recall use query_batch_edges / edges_for_entity.
    #[tracing::instrument(name = "memory.graph.store.edges_exact", skip_all)]
    pub async fn edges_exact(
        &self,
        source_entity_id: i64,
        target_entity_id: i64,
    ) -> Result<Vec<Edge>, MemoryError> {
        let raw = format!(
            "SELECT {} \
             FROM graph_edges \
             WHERE valid_to IS NULL \
               AND source_entity_id = ? \
               AND target_entity_id = ?",
            edge_select_cols("")
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let rows: Vec<EdgeRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(source_entity_id)
            .bind(target_entity_id)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    /// Count active (non-invalidated) edges.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.active_edge_count", skip_all)]
    pub async fn active_edge_count(&self) -> Result<i64, MemoryError> {
        let count: i64 = zeph_db::query_scalar(sql!(
            "SELECT COUNT(*) FROM graph_edges WHERE valid_to IS NULL"
        ))
        .fetch_one(&self.pool)
        .await?;
        Ok(count)
    }

    /// Return per-type active edge counts as `(edge_type, count)` pairs.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.edge_type_distribution", skip_all)]
    pub async fn edge_type_distribution(&self) -> Result<Vec<(String, i64)>, MemoryError> {
        let rows: Vec<(String, i64)> = zeph_db::query_as(
            sql!("SELECT edge_type, COUNT(*) FROM graph_edges WHERE valid_to IS NULL GROUP BY edge_type ORDER BY edge_type"),
        )
        .fetch_all(&self.pool)
        .await?;
        Ok(rows)
    }

    // ── Communities ───────────────────────────────────────────────────────────

    /// Insert or update a community by name.
    ///
    /// `fingerprint` is a BLAKE3 hex string computed from sorted entity IDs and
    /// intra-community edge IDs. Pass `None` to leave the fingerprint unchanged (e.g. when
    /// `assign_to_community` adds an entity without a full re-detection pass).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or JSON serialization fails.
    #[tracing::instrument(name = "memory.graph.store.upsert_community", skip_all)]
    pub async fn upsert_community(
        &self,
        name: &str,
        summary: &str,
        entity_ids: &[i64],
        fingerprint: Option<&str>,
    ) -> Result<i64, MemoryError> {
        let entity_ids_json = serde_json::to_string(entity_ids)?;
        let id: i64 = zeph_db::query_scalar(sql!(
            "INSERT INTO graph_communities (name, summary, entity_ids, fingerprint)
             VALUES (?, ?, ?, ?)
             ON CONFLICT(name) DO UPDATE SET
               summary = excluded.summary,
               entity_ids = excluded.entity_ids,
               fingerprint = COALESCE(excluded.fingerprint, graph_communities.fingerprint),
               updated_at = CURRENT_TIMESTAMP
             RETURNING id"
        ))
        .bind(name)
        .bind(summary)
        .bind(entity_ids_json)
        .bind(fingerprint)
        .fetch_one(&self.pool)
        .await?;
        Ok(id)
    }

    /// Return a map of `fingerprint -> community_id` for all communities with a non-NULL
    /// fingerprint. Used by `detect_communities` to skip unchanged partitions.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.community_fingerprints", skip_all)]
    pub async fn community_fingerprints(&self) -> Result<HashMap<String, i64>, MemoryError> {
        let rows: Vec<(String, i64)> = zeph_db::query_as(sql!(
            "SELECT fingerprint, id FROM graph_communities WHERE fingerprint IS NOT NULL"
        ))
        .fetch_all(&self.pool)
        .await?;
        Ok(rows.into_iter().collect())
    }

    /// Delete a single community by its primary key.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.delete_community_by_id", skip_all)]
    pub async fn delete_community_by_id(&self, id: i64) -> Result<(), MemoryError> {
        zeph_db::query(sql!("DELETE FROM graph_communities WHERE id = ?"))
            .bind(id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    /// Set the fingerprint of a community to `NULL`, invalidating the incremental cache.
    ///
    /// Used by `assign_to_community` when an entity is added without a full re-detection pass,
    /// ensuring the next `detect_communities` run re-summarizes the affected community.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.clear_community_fingerprint", skip_all)]
    pub async fn clear_community_fingerprint(&self, id: i64) -> Result<(), MemoryError> {
        zeph_db::query(sql!(
            "UPDATE graph_communities SET fingerprint = NULL WHERE id = ?"
        ))
        .bind(id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Find the first community that contains the given `entity_id`.
    ///
    /// Uses `json_each()` to push the membership search into `SQLite`, avoiding a full
    /// table scan with per-row JSON parsing.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or JSON parsing fails.
    #[tracing::instrument(name = "memory.graph.store.community_for_entity", skip_all)]
    pub async fn community_for_entity(
        &self,
        entity_id: i64,
    ) -> Result<Option<Community>, MemoryError> {
        // NOTE: `json_each()` is `SQLite`-only syntax; this query is not yet dialect-gated for
        // Postgres (see `community_ids_sql` for the established sqlite/postgres split pattern
        // used elsewhere) — pre-existing gap, out of scope for the `TIMESTAMPTZ` decode fix here.
        let raw = format!(
            "SELECT {} \
             FROM graph_communities c, json_each(c.entity_ids) j \
             WHERE CAST(j.value AS INTEGER) = ? \
             LIMIT 1",
            community_select_cols("c.")
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let row: Option<CommunityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(entity_id)
            .fetch_optional(&self.pool)
            .await?;
        row.map(community_from_row).transpose()
    }

    /// Get all communities.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or JSON parsing fails.
    #[tracing::instrument(name = "memory.graph.store.all_communities", skip_all)]
    pub async fn all_communities(&self) -> Result<Vec<Community>, MemoryError> {
        let raw = format!(
            "SELECT {} FROM graph_communities ORDER BY id ASC",
            community_select_cols("")
        );
        let rows: Vec<CommunityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(raw))
            .fetch_all(&self.pool)
            .await?;

        rows.into_iter().map(community_from_row).collect()
    }

    /// Count the total number of communities.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.community_count", skip_all)]
    pub async fn community_count(&self) -> Result<i64, MemoryError> {
        let count: i64 = zeph_db::query_scalar(sql!("SELECT COUNT(*) FROM graph_communities"))
            .fetch_one(&self.pool)
            .await?;
        Ok(count)
    }

    // ── Metadata ──────────────────────────────────────────────────────────────

    /// Get a metadata value by key.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.get_metadata", skip_all)]
    pub async fn get_metadata(&self, key: &str) -> Result<Option<String>, MemoryError> {
        let val: Option<String> =
            zeph_db::query_scalar(sql!("SELECT value FROM graph_metadata WHERE key = ?"))
                .bind(key)
                .fetch_optional(&self.pool)
                .await?;
        Ok(val)
    }

    /// Set a metadata value by key (upsert).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.set_metadata", skip_all)]
    pub async fn set_metadata(&self, key: &str, value: &str) -> Result<(), MemoryError> {
        zeph_db::query(sql!(
            "INSERT INTO graph_metadata (key, value) VALUES (?, ?)
             ON CONFLICT(key) DO UPDATE SET value = excluded.value"
        ))
        .bind(key)
        .bind(value)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Get the current extraction count from metadata.
    ///
    /// Returns 0 if the counter has not been initialized.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.extraction_count", skip_all)]
    pub async fn extraction_count(&self) -> Result<i64, MemoryError> {
        let val = self.get_metadata("extraction_count").await?;
        Ok(val.and_then(|v| v.parse::<i64>().ok()).unwrap_or(0))
    }

    /// Stream all active (non-invalidated) edges.
    pub fn all_active_edges_stream(&self) -> impl Stream<Item = Result<Edge, MemoryError>> + '_ {
        use futures::StreamExt as _;
        let raw = format!(
            "SELECT {} FROM graph_edges WHERE valid_to IS NULL ORDER BY id ASC",
            edge_select_cols("")
        );
        zeph_db::query_as::<_, EdgeRow>(sqlx::AssertSqlSafe(raw))
            .fetch(&self.pool)
            .map(|r| r.map_err(MemoryError::from).map(edge_from_row))
    }

    /// Fetch a chunk of active edges using keyset pagination.
    ///
    /// Returns edges with `id > after_id` in ascending order, up to `limit` rows.
    /// Starting with `after_id = 0` returns the first chunk. Pass the last `id` from
    /// the returned chunk as `after_id` for the next page. An empty result means all
    /// edges have been consumed.
    ///
    /// Keyset pagination is O(1) per page (index seek on `id`) vs OFFSET which is O(N).
    /// It is also stable under concurrent inserts: new edges get monotonically higher IDs
    /// and will appear in subsequent chunks or after the last chunk, never causing
    /// duplicates. Concurrent invalidations (setting `valid_to`) may cause a single edge
    /// to be skipped, which is acceptable — LPA operates on an eventual-consistency snapshot.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.edges_after_id", skip_all)]
    pub async fn edges_after_id(
        &self,
        after_id: i64,
        limit: i64,
    ) -> Result<Vec<Edge>, MemoryError> {
        let raw = format!(
            "SELECT {} \
             FROM graph_edges \
             WHERE valid_to IS NULL AND id > ? \
             ORDER BY id ASC \
             LIMIT ?",
            edge_select_cols("")
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let rows: Vec<EdgeRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(after_id)
            .bind(limit)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    /// Find a community by its primary key.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or JSON parsing fails.
    #[tracing::instrument(name = "memory.graph.store.find_community_by_id", skip_all)]
    pub async fn find_community_by_id(&self, id: i64) -> Result<Option<Community>, MemoryError> {
        let raw = format!(
            "SELECT {} FROM graph_communities WHERE id = ?",
            community_select_cols("")
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let row: Option<CommunityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(id)
            .fetch_optional(&self.pool)
            .await?;
        row.map(community_from_row).transpose()
    }

    /// Delete all communities (full rebuild before upsert).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.delete_all_communities", skip_all)]
    pub async fn delete_all_communities(&self) -> Result<(), MemoryError> {
        zeph_db::query(sql!("DELETE FROM graph_communities"))
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    // ── A-MEM Retrieval Tracking ──────────────────────────────────────────────

    /// Find entities matching `query` and return them with normalized FTS5 scores.
    ///
    /// Returns `Vec<(Entity, fts_score)>` where `fts_score` is normalized to `[0.0, 1.0]`
    /// by dividing each negated BM25 value by the maximum in the result set.
    /// Alias matches receive a fixed score of `0.5` (relative to FTS matches before normalization).
    ///
    /// Uses `UNION ALL` with outer `ORDER BY` to preserve FTS5 ordering through the LIMIT.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.find_entities_ranked", skip_all)]
    pub async fn find_entities_ranked(
        &self,
        query: &str,
        limit: usize,
    ) -> Result<Vec<(Entity, f32)>, MemoryError> {
        let query = &query[..query.floor_char_boundary(512)];
        let Some(fts_query) = build_fts_query(query) else {
            return Ok(vec![]);
        };

        let limit_i64 = i64::try_from(limit)?;
        let ranked_fts_sql = build_ranked_fts_sql();
        let rows: Vec<EntityFtsRow> = zeph_db::query_as(sqlx::AssertSqlSafe(ranked_fts_sql))
            .bind(&fts_query)
            .bind(format!(
                "%{}%",
                query
                    .trim()
                    .replace('\\', "\\\\")
                    .replace('%', "\\%")
                    .replace('_', "\\_")
            ))
            .bind(limit_i64)
            .fetch_all(&self.pool)
            .await?;

        if rows.is_empty() {
            return Ok(vec![]);
        }

        Ok(normalize_and_dedup(rows))
    }

    /// Compute structural scores (degree + edge type diversity) for a batch of entity IDs.
    ///
    /// Returns `HashMap<entity_id, structural_score>` where score is in `[0.0, 1.0]`.
    /// Formula: `0.6 * (degree / max_degree) + 0.4 * (type_diversity / 4.0)`.
    /// Entities with no edges receive score `0.0`.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.entity_structural_scores", skip_all, fields(count = entity_ids.len()))]
    pub async fn entity_structural_scores(
        &self,
        entity_ids: &[i64],
    ) -> Result<HashMap<i64, f32>, MemoryError> {
        // Each query binds entity_ids three times (three IN clauses).
        // Stay safely under SQLite 999-variable limit: 999 / 3 = 333, use 163 for headroom.
        const MAX_BATCH: usize = 163;

        if entity_ids.is_empty() {
            return Ok(HashMap::new());
        }

        let mut all_rows: Vec<(i64, i64, i64)> = Vec::new();
        for chunk in entity_ids.chunks(MAX_BATCH) {
            let n = chunk.len();
            // Three copies of chunk IDs: positions 1..n, n+1..2n, 2n+1..3n
            let ph1 = placeholder_list(1, n);
            let ph2 = placeholder_list(n + 1, n);
            let ph3 = placeholder_list(n * 2 + 1, n);

            // Build query: count degree and distinct edge types for each entity.
            let sql = format!(
                "SELECT entity_id,
                        COUNT(*) AS degree,
                        COUNT(DISTINCT edge_type) AS type_diversity
                 FROM (
                     SELECT source_entity_id AS entity_id, edge_type
                     FROM graph_edges
                     WHERE valid_to IS NULL AND source_entity_id IN ({ph1})
                     UNION ALL
                     SELECT target_entity_id AS entity_id, edge_type
                     FROM graph_edges
                     WHERE valid_to IS NULL AND target_entity_id IN ({ph2})
                 )
                 WHERE entity_id IN ({ph3})
                 GROUP BY entity_id"
            );

            let mut query = zeph_db::query_as::<_, (i64, i64, i64)>(sqlx::AssertSqlSafe(sql));
            // Bind chunk three times (three IN clauses)
            for id in chunk {
                query = query.bind(*id);
            }
            for id in chunk {
                query = query.bind(*id);
            }
            for id in chunk {
                query = query.bind(*id);
            }

            let chunk_rows: Vec<(i64, i64, i64)> = query.fetch_all(&self.pool).await?;
            all_rows.extend(chunk_rows);
        }

        if all_rows.is_empty() {
            return Ok(entity_ids.iter().map(|&id| (id, 0.0_f32)).collect());
        }

        let max_degree = all_rows
            .iter()
            .map(|(_, d, _)| *d)
            .max()
            .unwrap_or(1)
            .max(1);

        let mut scores: HashMap<i64, f32> = entity_ids.iter().map(|&id| (id, 0.0_f32)).collect();
        for (entity_id, degree, type_diversity) in all_rows {
            #[allow(clippy::cast_precision_loss)]
            let norm_degree = degree as f32 / max_degree as f32;
            #[allow(clippy::cast_precision_loss)]
            let norm_diversity = (type_diversity as f32 / 4.0).min(1.0);
            let score = 0.6 * norm_degree + 0.4 * norm_diversity;
            scores.insert(entity_id, score);
        }

        Ok(scores)
    }

    /// Look up community IDs for a batch of entity IDs.
    ///
    /// Returns `HashMap<entity_id, community_id>`. Entities not assigned to any community
    /// are absent from the map (treated as `None` by callers — no community cap applied).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[cfg(any(feature = "sqlite", feature = "postgres"))]
    #[tracing::instrument(name = "memory.graph.store.entity_community_ids", skip_all, fields(count = entity_ids.len()))]
    pub async fn entity_community_ids(
        &self,
        entity_ids: &[i64],
    ) -> Result<HashMap<i64, i64>, MemoryError> {
        if entity_ids.is_empty() {
            return Ok(HashMap::new());
        }

        let mut result: HashMap<i64, i64> = HashMap::new();
        for chunk in entity_ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let placeholders = placeholder_list(1, chunk.len());

            let community_sql = community_ids_sql(&placeholders);
            let mut query = zeph_db::query_as::<_, (i64, i64)>(sqlx::AssertSqlSafe(community_sql));
            for id in chunk {
                query = query.bind(*id);
            }

            let rows: Vec<(i64, i64)> = query.fetch_all(&self.pool).await?;
            result.extend(rows);
        }

        Ok(result)
    }

    /// Increment `retrieval_count` and set `last_retrieved_at` for a batch of edge IDs.
    ///
    /// Fire-and-forget: errors are logged but not propagated. Caller should log the warning.
    /// Batched with [`SQLITE_BATCH_LIMIT_2X`] to stay safely under `SQLite` bind variable limit.
    /// Each chunk's write is bounded by a 500ms timeout (matching
    /// [`Self::qdrant_point_ids_for_entities`]) so a stuck pool surfaces as a typed
    /// [`MemoryError::Timeout`] instead of hanging the `graph_recall_astar` hot path.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails or a chunk write times out.
    #[tracing::instrument(name = "memory.graph.store.record_edge_retrieval", skip_all, fields(count = edge_ids.len()))]
    pub async fn record_edge_retrieval(&self, edge_ids: &[i64]) -> Result<(), MemoryError> {
        let epoch_now = <ActiveDialect as zeph_db::dialect::Dialect>::EPOCH_NOW;
        for chunk in edge_ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let edge_placeholders = placeholder_list(1, chunk.len());
            let retrieval_sql = format!(
                "UPDATE graph_edges \
                 SET retrieval_count = retrieval_count + 1, \
                     last_retrieved_at = {epoch_now} \
                 WHERE id IN ({edge_placeholders})"
            );
            let mut q = zeph_db::query(sqlx::AssertSqlSafe(retrieval_sql));
            for id in chunk {
                q = q.bind(*id);
            }
            tokio::time::timeout(std::time::Duration::from_millis(500), q.execute(&self.pool))
                .await
                .map_err(|_| {
                    MemoryError::Timeout(
                        "record_edge_retrieval: write timed out after 500ms".into(),
                    )
                })??;
        }
        Ok(())
    }

    /// Increment `weight` on the set of edges traversed during the current recall (HL-F2, #3344).
    ///
    /// Mirrors [`Self::record_edge_retrieval`] in shape: same [`SQLITE_BATCH_LIMIT_2X`] chunking,
    /// same `WHERE id IN (…) AND valid_to IS NULL` filter (defensive — traversed edges should
    /// already be active, but this prevents reinforcing tombstoned edges).
    ///
    /// No-op when `edge_ids` is empty or `delta == 0.0`.
    ///
    /// Each chunk's write is bounded by a 500ms timeout (matching
    /// [`Self::qdrant_point_ids_for_entities`]) so a stuck pool surfaces as a typed
    /// [`MemoryError::Timeout`] instead of hanging the `graph_recall_astar` hot path.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying `UPDATE` fails or a chunk write times out.
    #[tracing::instrument(
        name = "memory.graph.hebbian_increment",
        skip_all,
        fields(edge_count = edge_ids.len())
    )]
    pub async fn apply_hebbian_increment(
        &self,
        edge_ids: &[i64],
        delta: f32,
    ) -> Result<(), MemoryError> {
        // SQLITE_BATCH_LIMIT_2X chosen to stay under SQLite's 999 host-parameter limit
        // ($1 = delta, $2..$N+1 = edge ids — 1 + 490 = 491 params per chunk).
        if edge_ids.is_empty() || delta == 0.0 {
            return Ok(());
        }
        for chunk in edge_ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let edge_placeholders = placeholder_list(2, chunk.len());
            let sql = format!(
                "UPDATE graph_edges \
                 SET weight = weight + $1 \
                 WHERE id IN ({edge_placeholders}) \
                   AND valid_to IS NULL"
            );
            let mut q = zeph_db::query(sqlx::AssertSqlSafe(sql));
            q = q.bind(f64::from(delta));
            for id in chunk {
                q = q.bind(*id);
            }
            tokio::time::timeout(std::time::Duration::from_millis(500), q.execute(&self.pool))
                .await
                .map_err(|_| {
                    MemoryError::Timeout(
                        "apply_hebbian_increment: write timed out after 500ms".into(),
                    )
                })??;
        }
        Ok(())
    }

    /// Return the subset of `ids` that exist in `graph_entities`.
    ///
    /// Useful for cross-referencing Qdrant-side entity IDs against the `SQLite` truth.
    /// Processes in chunks of [`SQLITE_BATCH_LIMIT_2X`] to stay under the `SQLite` variable
    /// limit (~32 k).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.entity_ids_in", skip_all, fields(count = ids.len()))]
    pub async fn entity_ids_in(&self, ids: &[i64]) -> Result<Vec<i64>, MemoryError> {
        if ids.is_empty() {
            return Ok(Vec::new());
        }
        // TODO: chunk when ids.len() > 5_000 to guard against extreme cases
        let mut result = Vec::with_capacity(ids.len());
        for chunk in ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let placeholders = zeph_db::placeholder_list(1, chunk.len());
            let sql = format!("SELECT id FROM graph_entities WHERE id IN ({placeholders})");
            let mut q = zeph_db::query_as::<_, (i64,)>(sqlx::AssertSqlSafe(sql));
            for id in chunk {
                q = q.bind(*id);
            }
            let rows = q.fetch_all(&self.pool).await?;
            for (id,) in rows {
                result.push(id);
            }
        }
        Ok(result)
    }

    /// Resolve entity IDs to their Qdrant point IDs in a single batched `SELECT` (HL-F5, #3346).
    ///
    /// Entities without a `qdrant_point_id` are silently omitted from the result.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.qdrant_point_ids_for_entities", skip_all, fields(count = entity_ids.len()))]
    pub async fn qdrant_point_ids_for_entities(
        &self,
        entity_ids: &[i64],
    ) -> Result<HashMap<i64, String>, MemoryError> {
        // Chunk to stay under SQLite variable limit.
        if entity_ids.is_empty() {
            return Ok(HashMap::new());
        }
        let mut result: HashMap<i64, String> = HashMap::with_capacity(entity_ids.len());
        for chunk in entity_ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let placeholders = zeph_db::placeholder_list(1, chunk.len());
            let sql = format!(
                "SELECT id, qdrant_point_id \
                 FROM graph_entities \
                 WHERE id IN ({placeholders}) \
                   AND qdrant_point_id IS NOT NULL"
            );
            let mut q = zeph_db::query_as::<_, (i64, String)>(sqlx::AssertSqlSafe(sql));
            for id in chunk {
                q = q.bind(*id);
            }
            // Bound worst-case per-chunk blocking on pool.acquire() + query execution: a
            // stuck pool turns into a typed MemoryError::Timeout instead of an indefinite
            // hang, so callers (e.g. hela_spreading_recall's per-step circuit breaker) can
            // fail fast and proceed.
            let rows: Vec<(i64, String)> = tokio::time::timeout(
                std::time::Duration::from_millis(500),
                q.fetch_all(&self.pool),
            )
            .await
            .map_err(|_| MemoryError::Timeout("qdrant_point_ids_for_entities".into()))??;
            for (entity_id, point_id) in rows {
                result.insert(entity_id, point_id);
            }
        }
        Ok(result)
    }

    /// Apply multiplicative decay to `retrieval_count` for un-retrieved active edges.
    ///
    /// Only edges with `retrieval_count > 0` and `last_retrieved_at < (now - interval_secs)`
    /// are updated. Returns the number of rows affected.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.decay_edge_retrieval_counts", skip_all)]
    pub async fn decay_edge_retrieval_counts(
        &self,
        decay_lambda: f64,
        interval_secs: u64,
    ) -> Result<usize, MemoryError> {
        let epoch_now_decay = <ActiveDialect as zeph_db::dialect::Dialect>::EPOCH_NOW;
        let greatest_fn = <ActiveDialect as zeph_db::dialect::Dialect>::GREATEST_FN;
        let decay_raw = format!(
            "UPDATE graph_edges \
             SET retrieval_count = {greatest_fn}(CAST(retrieval_count * ? AS INTEGER), 0) \
             WHERE valid_to IS NULL \
               AND retrieval_count > 0 \
               AND (last_retrieved_at IS NULL OR last_retrieved_at < {epoch_now_decay} - ?)"
        );
        let decay_sql = zeph_db::rewrite_placeholders(&decay_raw);
        let result = zeph_db::query(sqlx::AssertSqlSafe(decay_sql))
            .bind(decay_lambda)
            .bind(i64::try_from(interval_secs).unwrap_or(i64::MAX))
            .execute(&self.pool)
            .await?;
        Ok(usize::try_from(result.rows_affected())?)
    }

    /// Delete expired edges older than `retention_days` and return count deleted.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.delete_expired_edges", skip_all)]
    pub async fn delete_expired_edges(&self, retention_days: u32) -> Result<usize, MemoryError> {
        let retention_secs = i64::from(retention_days) * 86400;
        let epoch_now = <ActiveDialect as zeph_db::dialect::Dialect>::EPOCH_NOW;
        let expired_at_epoch =
            <ActiveDialect as zeph_db::dialect::Dialect>::epoch_from_col("expired_at");
        let raw = format!(
            "DELETE FROM graph_edges
             WHERE expired_at IS NOT NULL
               AND {expired_at_epoch} < {epoch_now} - ?"
        );
        let sql = zeph_db::rewrite_placeholders(&raw);
        let result = zeph_db::query(sqlx::AssertSqlSafe(sql))
            .bind(retention_secs)
            .execute(&self.pool)
            .await?;
        Ok(usize::try_from(result.rows_affected())?)
    }

    /// Delete orphan entities (no active edges, last seen more than `retention_days` ago).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.delete_orphan_entities", skip_all)]
    pub async fn delete_orphan_entities(&self, retention_days: u32) -> Result<usize, MemoryError> {
        let retention_secs = i64::from(retention_days) * 86400;
        let epoch_now = <ActiveDialect as zeph_db::dialect::Dialect>::EPOCH_NOW;
        let last_seen_at_epoch =
            <ActiveDialect as zeph_db::dialect::Dialect>::epoch_from_col("last_seen_at");
        let raw = format!(
            "DELETE FROM graph_entities
             WHERE id NOT IN (
                 SELECT DISTINCT source_entity_id FROM graph_edges WHERE valid_to IS NULL
                 UNION
                 SELECT DISTINCT target_entity_id FROM graph_edges WHERE valid_to IS NULL
             )
             AND {last_seen_at_epoch} < {epoch_now} - ?"
        );
        let sql = zeph_db::rewrite_placeholders(&raw);
        let result = zeph_db::query(sqlx::AssertSqlSafe(sql))
            .bind(retention_secs)
            .execute(&self.pool)
            .await?;
        Ok(usize::try_from(result.rows_affected())?)
    }

    /// Delete all edges and orphaned entities stamped with `import_batch_id`.
    ///
    /// Sequence:
    /// 1. Delete all `graph_edges` where `import_batch_id = batch_id`.
    /// 2. Delete `graph_entities` where `import_batch_id = batch_id` AND they are
    ///    now orphaned (no remaining edges reference them).
    ///
    /// Rows with `origin = 'conversation'` are never touched by this method since
    /// they carry a `NULL` or different `import_batch_id`.
    /// Returns `(edges_deleted, entities_deleted)`.
    ///
    /// # Errors
    ///
    /// Returns [`MemoryError::Sqlx`] on database failure.
    #[tracing::instrument(name = "memory.graph.store.delete_batch", skip_all)]
    pub async fn delete_batch(&self, batch_id: &str) -> Result<(u64, u64), MemoryError> {
        let edge_result = zeph_db::query(sql!("DELETE FROM graph_edges WHERE import_batch_id = ?"))
            .bind(batch_id)
            .execute(&self.pool)
            .await?;
        let edges_deleted = edge_result.rows_affected();

        let entity_result = zeph_db::query(sql!(
            "DELETE FROM graph_entities
             WHERE import_batch_id = ?
               AND id NOT IN (
                   SELECT DISTINCT source_entity_id FROM graph_edges
                   UNION
                   SELECT DISTINCT target_entity_id FROM graph_edges
               )"
        ))
        .bind(batch_id)
        .execute(&self.pool)
        .await?;
        let entities_deleted = entity_result.rows_affected();

        Ok((edges_deleted, entities_deleted))
    }

    /// Delete all graph edges and orphaned entities for `batch_id` within a caller-provided
    /// transaction.
    ///
    /// Executes the same two-step DELETE as [`Self::delete_batch`] but uses `tx` so the
    /// caller can combine it with other writes in a single atomic unit.
    /// Returns `(edges_deleted, entities_deleted)`.
    ///
    /// # Errors
    ///
    /// Returns [`MemoryError::Sqlx`] on database failure.
    #[tracing::instrument(name = "memory.graph.store.delete_batch_in_tx", skip_all)]
    pub async fn delete_batch_in_tx(
        &self,
        batch_id: &str,
        tx: &mut zeph_db::DbTransaction<'_>,
    ) -> Result<(u64, u64), MemoryError> {
        let edges_deleted =
            zeph_db::query(sql!("DELETE FROM graph_edges WHERE import_batch_id = ?"))
                .bind(batch_id)
                .execute(&mut **tx)
                .await?
                .rows_affected();

        let entities_deleted = zeph_db::query(sql!(
            "DELETE FROM graph_entities
             WHERE import_batch_id = ?
               AND id NOT IN (
                   SELECT DISTINCT source_entity_id FROM graph_edges
                   UNION
                   SELECT DISTINCT target_entity_id FROM graph_edges
               )"
        ))
        .bind(batch_id)
        .execute(&mut **tx)
        .await?
        .rows_affected();

        Ok((edges_deleted, entities_deleted))
    }

    /// Delete the oldest excess entities when count exceeds `max_entities`.
    ///
    /// Entities are ranked by ascending edge count, then ascending `last_seen_at` (LRU).
    /// Only deletes when `entity_count() > max_entities`.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.cap_entities", skip_all)]
    pub async fn cap_entities(&self, max_entities: usize) -> Result<usize, MemoryError> {
        let current = self.entity_count().await?;
        let max = i64::try_from(max_entities)?;
        if current <= max {
            return Ok(0);
        }
        let excess = current - max;
        let result = zeph_db::query(sql!(
            "DELETE FROM graph_entities
             WHERE id IN (
                 SELECT e.id
                 FROM graph_entities e
                 LEFT JOIN (
                     SELECT source_entity_id AS eid, COUNT(*) AS cnt
                     FROM graph_edges WHERE valid_to IS NULL GROUP BY source_entity_id
                     UNION ALL
                     SELECT target_entity_id AS eid, COUNT(*) AS cnt
                     FROM graph_edges WHERE valid_to IS NULL GROUP BY target_entity_id
                 ) edge_counts ON e.id = edge_counts.eid
                 ORDER BY COALESCE(edge_counts.cnt, 0) ASC, e.last_seen_at ASC
                 LIMIT ?
             )"
        ))
        .bind(excess)
        .execute(&self.pool)
        .await?;
        Ok(usize::try_from(result.rows_affected())?)
    }

    // ── Temporal Edge Queries ─────────────────────────────────────────────────

    /// Return all edges for `entity_id` (as source or target) that were valid at `timestamp`.
    ///
    /// An edge is valid at `timestamp` when:
    /// - `valid_from <= timestamp`, AND
    /// - `valid_to IS NULL` (open-ended) OR `valid_to > timestamp`.
    ///
    /// `timestamp` must be a `SQLite` datetime string: `"YYYY-MM-DD HH:MM:SS"`.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.edges_at_timestamp", skip_all)]
    pub async fn edges_at_timestamp(
        &self,
        entity_id: i64,
        timestamp: &str,
    ) -> Result<Vec<Edge>, MemoryError> {
        // Split into two UNIONed branches to leverage the partial indexes from migration 030:
        //   Branch 1 (active edges):     idx_graph_edges_valid + idx_graph_edges_source/target
        //   Branch 2 (historical edges): idx_graph_edges_src_temporal / idx_graph_edges_tgt_temporal
        //
        // `timestamp` is a Rust-formatted string bound against `valid_from`/`valid_to`
        // (`TIMESTAMPTZ` on Postgres); Postgres has no implicit text->timestamptz cast for a
        // bound parameter, so an explicit `TIMESTAMPTZ_CAST` is required on each placeholder —
        // mirrors `agent_sessions.rs`'s fix for the same class of mismatch.
        let ts_cast = <ActiveDialect as zeph_db::dialect::Dialect>::TIMESTAMPTZ_CAST;
        let edge_cols = edge_select_cols("");
        let raw = format!(
            "SELECT {edge_cols}
             FROM graph_edges
             WHERE valid_to IS NULL
               AND valid_from <= ?{ts_cast}
               AND (source_entity_id = ? OR target_entity_id = ?)
             UNION ALL
             SELECT {edge_cols}
             FROM graph_edges
             WHERE valid_to IS NOT NULL
               AND valid_from <= ?{ts_cast}
               AND valid_to > ?{ts_cast}
               AND (source_entity_id = ? OR target_entity_id = ?)"
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let rows: Vec<EdgeRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(timestamp)
            .bind(entity_id)
            .bind(entity_id)
            .bind(timestamp)
            .bind(timestamp)
            .bind(entity_id)
            .bind(entity_id)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    /// Return all edge versions (active and expired) for the given `(source, predicate)` pair.
    ///
    /// The optional `relation` filter restricts results to a specific relation label.
    /// Results are ordered by `valid_from DESC` (most recent first).
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.edge_history", skip_all)]
    pub async fn edge_history(
        &self,
        source_entity_id: i64,
        predicate: &str,
        relation: Option<&str>,
        limit: usize,
    ) -> Result<Vec<Edge>, MemoryError> {
        // Escape LIKE wildcards so `%` and `_` in the predicate are treated as literals.
        let escaped = predicate
            .replace('\\', "\\\\")
            .replace('%', "\\%")
            .replace('_', "\\_");
        let like_pattern = format!("%{escaped}%");
        let limit = i64::try_from(limit)?;
        let edge_cols = edge_select_cols("");
        let rows: Vec<EdgeRow> = if let Some(rel) = relation {
            let raw = format!(
                "SELECT {edge_cols}
                 FROM graph_edges
                 WHERE source_entity_id = ?
                   AND fact LIKE ? ESCAPE '\\'
                   AND relation = ?
                 ORDER BY graph_edges.valid_from DESC
                 LIMIT ?"
            );
            let query_sql = zeph_db::rewrite_placeholders(&raw);
            zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
                .bind(source_entity_id)
                .bind(&like_pattern)
                .bind(rel)
                .bind(limit)
                .fetch_all(&self.pool)
                .await?
        } else {
            let raw = format!(
                "SELECT {edge_cols}
                 FROM graph_edges
                 WHERE source_entity_id = ?
                   AND fact LIKE ? ESCAPE '\\'
                 ORDER BY graph_edges.valid_from DESC
                 LIMIT ?"
            );
            let query_sql = zeph_db::rewrite_placeholders(&raw);
            zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
                .bind(source_entity_id)
                .bind(&like_pattern)
                .bind(limit)
                .fetch_all(&self.pool)
                .await?
        };
        Ok(rows.into_iter().map(edge_from_row).collect())
    }

    // ── BFS Traversal ─────────────────────────────────────────────────────────

    /// Breadth-first traversal from `start_entity_id` up to `max_hops` hops.
    ///
    /// Returns all reachable entities and the active edges connecting them.
    /// Implements BFS iteratively in Rust to guarantee cycle safety regardless
    /// of `SQLite` CTE limitations.
    ///
    /// **`SQLite` bind parameter limit**: each BFS hop binds the frontier IDs three times in the
    /// neighbour query. At ~300+ frontier entities per hop, the IN clause may approach `SQLite`'s
    /// default `SQLITE_MAX_VARIABLE_NUMBER` limit of 999. Acceptable for Phase 1 (small graphs,
    /// `max_hops` typically 2–3). For large graphs, consider batching or a temp-table approach.
    ///
    /// # Errors
    ///
    /// Returns an error if any database query fails.
    #[tracing::instrument(name = "memory.graph.store.bfs", skip_all)]
    pub async fn bfs(
        &self,
        start_entity_id: i64,
        max_hops: u32,
    ) -> Result<(Vec<Entity>, Vec<Edge>), MemoryError> {
        self.bfs_with_depth(start_entity_id, max_hops)
            .await
            .map(|(e, ed, _)| (e, ed))
    }

    /// BFS traversal returning entities, edges, and a depth map (`entity_id` → hop distance).
    ///
    /// The depth map records the minimum hop distance from `start_entity_id` to each visited
    /// entity. The start entity itself has depth 0.
    ///
    /// **`SQLite` bind parameter limit**: see [`Self::bfs`] for notes on frontier size limits.
    ///
    /// # Errors
    ///
    /// Returns an error if any database query fails.
    #[allow(clippy::type_complexity)]
    #[tracing::instrument(name = "memory.graph.store.bfs_with_depth", skip_all)]
    pub async fn bfs_with_depth(
        &self,
        start_entity_id: i64,
        max_hops: u32,
    ) -> Result<(Vec<Entity>, Vec<Edge>, std::collections::HashMap<i64, u32>), MemoryError> {
        self.bfs_core(start_entity_id, max_hops, None, &[]).await
    }

    /// BFS traversal considering only edges that were valid at `timestamp`.
    ///
    /// Equivalent to [`Self::bfs_with_depth`] but replaces the `valid_to IS NULL` filter with
    /// the temporal range predicate `valid_from <= ts AND (valid_to IS NULL OR valid_to > ts)`.
    ///
    /// `timestamp` must be a `SQLite` datetime string: `"YYYY-MM-DD HH:MM:SS"`.
    ///
    /// # Errors
    ///
    /// Returns an error if any database query fails.
    #[allow(clippy::type_complexity)]
    #[tracing::instrument(name = "memory.graph.store.bfs_at_timestamp", skip_all)]
    pub async fn bfs_at_timestamp(
        &self,
        start_entity_id: i64,
        max_hops: u32,
        timestamp: &str,
    ) -> Result<(Vec<Entity>, Vec<Edge>, std::collections::HashMap<i64, u32>), MemoryError> {
        self.bfs_core(start_entity_id, max_hops, Some(timestamp), &[])
            .await
    }

    /// BFS traversal scoped to specific MAGMA edge types.
    ///
    /// When `edge_types` is empty, behaves identically to [`Self::bfs_with_depth`] (traverses all
    /// active edges). When `edge_types` is non-empty, only traverses edges whose `edge_type`
    /// matches one of the provided types.
    ///
    /// This enables subgraph-scoped retrieval: a causal query traverses only causal + semantic
    /// edges, a temporal query only temporal + semantic edges, etc.
    ///
    /// Note: Semantic is typically included in `edge_types` by the caller to ensure recall is
    /// never worse than the untyped BFS. See `classify_graph_subgraph` in `router.rs`.
    ///
    /// # Errors
    ///
    /// Returns an error if any database query fails.
    #[allow(clippy::type_complexity)]
    #[tracing::instrument(name = "memory.graph.store.bfs_typed", skip_all)]
    pub async fn bfs_typed(
        &self,
        start_entity_id: i64,
        max_hops: u32,
        edge_types: &[EdgeType],
    ) -> Result<(Vec<Entity>, Vec<Edge>, std::collections::HashMap<i64, u32>), MemoryError> {
        self.bfs_core(start_entity_id, max_hops, None, edge_types)
            .await
    }

    /// Shared BFS implementation, optionally scoped to specific edge types.
    ///
    /// When `at_timestamp` is `None`, only active edges (`valid_to IS NULL`) are traversed.
    /// When `at_timestamp` is `Some(ts)`, edges valid at `ts` are traversed (temporal BFS).
    /// When `edge_types` is empty, all edge types are traversed; otherwise the `edge_type IN
    /// (...)` clause restricts traversal to the given types (`EdgeType::as_str()` values —
    /// no user input reaches SQL).
    ///
    /// All IDs used in dynamic SQL come from our own database — no user input reaches the
    /// format string, so there is no SQL injection risk.
    #[allow(clippy::type_complexity)]
    #[tracing::instrument(name = "memory.graph.store.bfs_core", skip_all)]
    async fn bfs_core(
        &self,
        start_entity_id: i64,
        max_hops: u32,
        at_timestamp: Option<&str>,
        edge_types: &[EdgeType],
    ) -> Result<(Vec<Entity>, Vec<Edge>, std::collections::HashMap<i64, u32>), MemoryError> {
        use std::collections::HashMap;

        // SQLite binds frontier IDs 3× per hop; at >333 IDs the IN clause exceeds
        // SQLITE_MAX_VARIABLE_NUMBER (999). Cap to 300 to stay safely within the limit.
        const MAX_FRONTIER: usize = 300;

        let type_strs: Vec<&str> = edge_types.iter().map(|t| t.as_str()).collect();
        let n_types = type_strs.len();
        // type_in is constant for the entire BFS — positions 1..=n_types never change.
        let type_in = (n_types > 0).then(|| placeholder_list(1, n_types));
        let id_start = n_types + 1;

        let mut depth_map: HashMap<i64, u32> = HashMap::new();
        let mut frontier: Vec<i64> = vec![start_entity_id];
        depth_map.insert(start_entity_id, 0);

        for hop in 0..max_hops {
            if frontier.is_empty() {
                break;
            }
            frontier.truncate(MAX_FRONTIER);
            // IDs come from our own DB — no user input, no injection risk.
            // Positions: types first (if any), then 3 copies of frontier IDs, then an
            // optional timestamp.
            let n_frontier = frontier.len();
            let fp1 = placeholder_list(id_start, n_frontier);
            let fp2 = placeholder_list(id_start + n_frontier, n_frontier);
            let fp3 = placeholder_list(id_start + n_frontier * 2, n_frontier);

            let type_filter = type_in
                .as_ref()
                .map(|type_in| format!("edge_type IN ({type_in}) AND "))
                .unwrap_or_default();
            let edge_filter = if at_timestamp.is_some() {
                let ts_pos = id_start + n_frontier * 3;
                format!(
                    "{type_filter}valid_from <= {ts} AND (valid_to IS NULL OR valid_to > {ts})",
                    ts = numbered_placeholder(ts_pos),
                )
            } else {
                format!("{type_filter}valid_to IS NULL")
            };
            let neighbour_sql = format!(
                "SELECT DISTINCT CASE
                     WHEN source_entity_id IN ({fp1}) THEN target_entity_id
                     ELSE source_entity_id
                 END as neighbour_id
                 FROM graph_edges
                 WHERE {edge_filter}
                   AND (source_entity_id IN ({fp2}) OR target_entity_id IN ({fp3}))"
            );
            let mut q = zeph_db::query_scalar::<_, i64>(sqlx::AssertSqlSafe(neighbour_sql));
            for t in &type_strs {
                q = q.bind(*t);
            }
            for id in &frontier {
                q = q.bind(*id);
            }
            for id in &frontier {
                q = q.bind(*id);
            }
            for id in &frontier {
                q = q.bind(*id);
            }
            if let Some(ts) = at_timestamp {
                q = q.bind(ts);
            }
            let neighbours: Vec<i64> = q.fetch_all(&self.pool).await?;
            let mut next_frontier: Vec<i64> = Vec::new();
            for nbr in neighbours {
                if let std::collections::hash_map::Entry::Vacant(e) = depth_map.entry(nbr) {
                    e.insert(hop + 1);
                    next_frontier.push(nbr);
                }
            }
            frontier = next_frontier;
        }

        self.bfs_fetch_results(depth_map, at_timestamp, &type_strs)
            .await
    }

    /// Fetch entities and edges for a completed BFS depth map, optionally filtered by
    /// `edge_type` (empty `type_strs` = no filter, all types included).
    ///
    /// # Errors
    ///
    /// Returns an error if any database query fails.
    #[allow(clippy::type_complexity)]
    #[tracing::instrument(name = "memory.graph.store.bfs_fetch_results", skip_all)]
    async fn bfs_fetch_results(
        &self,
        depth_map: std::collections::HashMap<i64, u32>,
        at_timestamp: Option<&str>,
        type_strs: &[&str],
    ) -> Result<(Vec<Entity>, Vec<Edge>, std::collections::HashMap<i64, u32>), MemoryError> {
        let mut visited_ids: Vec<i64> = depth_map.keys().copied().collect();
        if visited_ids.is_empty() {
            return Ok((Vec::new(), Vec::new(), depth_map));
        }
        // Edge query binds visited_ids twice — cap at 499 to stay under SQLite 999 limit.
        if visited_ids.len() > 499 {
            tracing::warn!(
                total = visited_ids.len(),
                retained = 499,
                "bfs_fetch_results: visited entity set truncated to 499 to stay within SQLite bind limit; \
                 some reachable entities will be dropped from results"
            );
            visited_ids.truncate(499);
        }

        let n_types = type_strs.len();
        let n_visited = visited_ids.len();

        // Bind order: types first (if any), then visited_ids twice, then optional timestamp.
        let type_in = (n_types > 0).then(|| placeholder_list(1, n_types));
        let id_start = n_types + 1;
        let ph_ids1 = placeholder_list(id_start, n_visited);
        let ph_ids2 = placeholder_list(id_start + n_visited, n_visited);

        let type_filter = type_in
            .as_ref()
            .map(|type_in| format!("edge_type IN ({type_in}) AND "))
            .unwrap_or_default();
        let edge_filter = if at_timestamp.is_some() {
            let ts_pos = id_start + n_visited * 2;
            format!(
                "{type_filter}valid_from <= {ts} AND (valid_to IS NULL OR valid_to > {ts})",
                ts = numbered_placeholder(ts_pos),
            )
        } else {
            format!("{type_filter}valid_to IS NULL")
        };

        let edge_sql = format!(
            "SELECT {edge_cols}
             FROM graph_edges
             WHERE {edge_filter}
               AND source_entity_id IN ({ph_ids1})
               AND target_entity_id IN ({ph_ids2})",
            edge_cols = edge_select_cols(""),
        );
        let mut edge_query = zeph_db::query_as::<_, EdgeRow>(sqlx::AssertSqlSafe(edge_sql));
        for t in type_strs {
            edge_query = edge_query.bind(*t);
        }
        for id in &visited_ids {
            edge_query = edge_query.bind(*id);
        }
        for id in &visited_ids {
            edge_query = edge_query.bind(*id);
        }
        if let Some(ts) = at_timestamp {
            edge_query = edge_query.bind(ts);
        }
        let edge_rows: Vec<EdgeRow> = edge_query.fetch_all(&self.pool).await?;

        let entity_sql = format!(
            "SELECT {entity_cols} FROM graph_entities WHERE id IN ({ph})",
            entity_cols = entity_select_cols(),
            ph = placeholder_list(1, n_visited),
        );
        let mut entity_query = zeph_db::query_as::<_, EntityRow>(sqlx::AssertSqlSafe(entity_sql));
        for id in &visited_ids {
            entity_query = entity_query.bind(*id);
        }
        let entity_rows: Vec<EntityRow> = entity_query.fetch_all(&self.pool).await?;

        let entities: Vec<Entity> = entity_rows
            .into_iter()
            .map(entity_from_row)
            .collect::<Result<Vec<_>, _>>()?;
        let edges: Vec<Edge> = edge_rows.into_iter().map(edge_from_row).collect();

        Ok((entities, edges, depth_map))
    }

    // ── Backfill helpers ──────────────────────────────────────────────────────

    /// Find an entity by name only (no type filter).
    ///
    /// Uses a two-phase lookup to ensure exact name matches are always prioritised:
    /// 1. Exact case-insensitive match on `name` or `canonical_name`.
    /// 2. If no exact match found, falls back to FTS5 prefix search (see `find_entities_fuzzy`).
    ///
    /// This prevents FTS5 from returning a different entity whose *summary* mentions the
    /// searched name (e.g. searching "Alice" returning "Google" because Google's summary
    /// contains "Alice").
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.find_entity_by_name", skip_all)]
    pub async fn find_entity_by_name(&self, name: &str) -> Result<Vec<Entity>, MemoryError> {
        let find_by_name_sql = format!(
            "SELECT {} \
             FROM graph_entities \
             WHERE name = ? {cn} OR canonical_name = ? {cn} \
             LIMIT 5",
            entity_select_cols(),
            cn = <ActiveDialect as zeph_db::dialect::Dialect>::COLLATE_NOCASE,
        );
        let query_sql = zeph_db::rewrite_placeholders(&find_by_name_sql);
        let rows: Vec<EntityRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(name)
            .bind(name)
            .fetch_all(&self.pool)
            .await?;

        if !rows.is_empty() {
            return rows.into_iter().map(entity_from_row).collect();
        }

        self.find_entities_fuzzy(name, 5).await
    }

    /// Return up to `limit` messages that have not yet been processed by graph extraction.
    ///
    /// Reads the `graph_processed` column added by migration 021.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(
        name = "memory.graph.store.unprocessed_messages_for_backfill",
        skip_all
    )]
    pub async fn unprocessed_messages_for_backfill(
        &self,
        limit: usize,
    ) -> Result<Vec<(crate::types::MessageId, String)>, MemoryError> {
        let limit = i64::try_from(limit)?;
        let rows: Vec<(i64, String)> = zeph_db::query_as(sql!(
            "SELECT id, content FROM messages
             WHERE graph_processed = FALSE
             ORDER BY id ASC
             LIMIT ?"
        ))
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;
        Ok(rows
            .into_iter()
            .map(|(id, content)| (crate::types::MessageId(id), content))
            .collect())
    }

    /// Return the count of messages not yet processed by graph extraction.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.unprocessed_message_count", skip_all)]
    pub async fn unprocessed_message_count(&self) -> Result<i64, MemoryError> {
        let count: i64 = zeph_db::query_scalar(sql!(
            "SELECT COUNT(*) FROM messages WHERE graph_processed = FALSE"
        ))
        .fetch_one(&self.pool)
        .await?;
        Ok(count)
    }

    /// Mark a batch of messages as graph-processed.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.mark_messages_graph_processed", skip_all, fields(count = ids.len()))]
    pub async fn mark_messages_graph_processed(
        &self,
        ids: &[crate::types::MessageId],
    ) -> Result<(), MemoryError> {
        if ids.is_empty() {
            return Ok(());
        }
        for chunk in ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let placeholders = placeholder_list(1, chunk.len());
            let sql =
                format!("UPDATE messages SET graph_processed = TRUE WHERE id IN ({placeholders})");
            let mut query = zeph_db::query(sqlx::AssertSqlSafe(sql));
            for id in chunk {
                query = query.bind(id.0);
            }
            query.execute(&self.pool).await?;
        }
        Ok(())
    }
}

// ── Dialect helpers ───────────────────────────────────────────────────────────

#[cfg(feature = "sqlite")]
fn community_ids_sql(placeholders: &str) -> String {
    format!(
        "SELECT CAST(j.value AS INTEGER) AS entity_id, c.id AS community_id
         FROM graph_communities c, json_each(c.entity_ids) j
         WHERE CAST(j.value AS INTEGER) IN ({placeholders})"
    )
}

#[cfg(feature = "postgres")]
fn community_ids_sql(placeholders: &str) -> String {
    format!(
        "SELECT (j.value)::bigint AS entity_id, c.id AS community_id
         FROM graph_communities c,
              jsonb_array_elements_text(c.entity_ids::jsonb) j(value)
         WHERE (j.value)::bigint IN ({placeholders})"
    )
}

// ── Row types for zeph_db::query_as ─────────────────────────────────────────────

#[derive(zeph_db::FromRow)]
struct EntityRow {
    id: i64,
    name: String,
    canonical_name: String,
    entity_type: String,
    summary: Option<String>,
    first_seen_at: String,
    last_seen_at: String,
    qdrant_point_id: Option<String>,
}

/// `graph_entities` projection for [`EntityRow`], dialect-cast for `TIMESTAMPTZ` columns.
///
/// `first_seen_at`/`last_seen_at` are `TIMESTAMPTZ` on Postgres (`TEXT` on `SQLite`); both are
/// projected through `Dialect::select_as_text` and aliased back to their original names so
/// `#[derive(FromRow)]` still binds them into `EntityRow`'s `String` fields.
fn entity_select_cols() -> String {
    let first_seen_sel =
        <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("first_seen_at");
    let last_seen_sel =
        <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("last_seen_at");
    format!(
        "id, name, canonical_name, entity_type, summary, \
         {first_seen_sel} AS first_seen_at, {last_seen_sel} AS last_seen_at, qdrant_point_id"
    )
}

fn entity_from_row(row: EntityRow) -> Result<Entity, MemoryError> {
    let entity_type = row
        .entity_type
        .parse::<EntityType>()
        .map_err(MemoryError::GraphStore)?;
    Ok(Entity {
        id: EntityId(row.id),
        name: row.name,
        canonical_name: row.canonical_name,
        entity_type,
        summary: row.summary,
        first_seen_at: row.first_seen_at,
        last_seen_at: row.last_seen_at,
        qdrant_point_id: row.qdrant_point_id,
    })
}

#[derive(zeph_db::FromRow)]
struct AliasRow {
    id: i64,
    entity_id: i64,
    alias_name: String,
    created_at: String,
}

/// `graph_entity_aliases` projection for [`AliasRow`], dialect-cast for `TIMESTAMPTZ` columns.
///
/// `created_at` is `TIMESTAMPTZ` on Postgres — see [`entity_select_cols`].
fn alias_select_cols() -> String {
    let created_at_sel = <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("created_at");
    format!("id, entity_id, alias_name, {created_at_sel} AS created_at")
}

fn alias_from_row(row: AliasRow) -> EntityAlias {
    EntityAlias {
        id: row.id,
        entity_id: EntityId(row.entity_id),
        alias_name: row.alias_name,
        created_at: row.created_at,
    }
}

#[derive(zeph_db::FromRow)]
struct EdgeRow {
    id: i64,
    source_entity_id: i64,
    target_entity_id: i64,
    relation: String,
    fact: String,
    confidence: f64,
    valid_from: String,
    valid_to: Option<String>,
    created_at: String,
    expired_at: Option<String>,
    #[sqlx(rename = "episode_id")]
    source_message_id: Option<i64>,
    qdrant_point_id: Option<String>,
    edge_type: String,
    retrieval_count: i32,
    last_retrieved_at: Option<i64>,
    // `INTEGER` (`INT4`) on Postgres, so it decodes as `Option<i32>`, not `Option<i64>`.
    superseded_by: Option<i32>,
    canonical_relation: Option<String>,
    supersedes: Option<i64>,
    // Hebbian reinforcement weight (HL-F1, #3344). SQLite REAL maps to f64 via sqlx.
    weight: f64,
    // SYNAPSE multi-timescale variables (#3709).
    confidence_fast: f64,
    confidence_slow: f64,
    // `INTEGER` (`INT4`) on Postgres, so it decodes as `Option<i32>`, not `Option<i64>`.
    turn_index: Option<i32>,
}

/// `graph_edges` projection for [`EdgeRow`], dialect-cast for `TIMESTAMPTZ` columns.
///
/// `valid_from`/`valid_to`/`created_at`/`expired_at` are `TIMESTAMPTZ` on Postgres (`TEXT` on
/// `SQLite`); each is projected through `Dialect::select_as_text` and aliased back to its
/// original name so `#[derive(FromRow)]` still binds them into `EdgeRow`'s `String`/
/// `Option<String>` fields. `table_prefix` (e.g. `"ge."` or `""`) is prepended to each source
/// column reference so this helper works both in plain `FROM graph_edges` queries and in joins
/// with an alias — see [`community_select_cols`].
fn edge_select_cols(table_prefix: &str) -> String {
    let valid_from_sel = <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text(&format!(
        "{table_prefix}valid_from"
    ));
    let valid_to_sel = <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text(&format!(
        "{table_prefix}valid_to"
    ));
    let created_at_sel = <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text(&format!(
        "{table_prefix}created_at"
    ));
    let expired_at_sel = <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text(&format!(
        "{table_prefix}expired_at"
    ));
    format!(
        "{table_prefix}id, {table_prefix}source_entity_id, {table_prefix}target_entity_id, \
         {table_prefix}relation, {table_prefix}fact, {table_prefix}confidence, \
         {valid_from_sel} AS valid_from, {valid_to_sel} AS valid_to, \
         {created_at_sel} AS created_at, {expired_at_sel} AS expired_at, \
         {table_prefix}episode_id, {table_prefix}qdrant_point_id, {table_prefix}edge_type, \
         {table_prefix}retrieval_count, {table_prefix}last_retrieved_at, \
         {table_prefix}superseded_by, {table_prefix}canonical_relation, {table_prefix}supersedes, \
         CAST({table_prefix}weight AS DOUBLE PRECISION) AS weight, \
         CAST({table_prefix}confidence_fast AS DOUBLE PRECISION) AS confidence_fast, \
         CAST({table_prefix}confidence_slow AS DOUBLE PRECISION) AS confidence_slow, \
         {table_prefix}turn_index"
    )
}

fn edge_from_row(row: EdgeRow) -> Edge {
    let edge_type = row
        .edge_type
        .parse::<EdgeType>()
        .unwrap_or(EdgeType::Semantic);
    let canonical_relation = row
        .canonical_relation
        .unwrap_or_else(|| row.relation.clone());
    Edge {
        id: row.id,
        source_entity_id: row.source_entity_id,
        target_entity_id: row.target_entity_id,
        canonical_relation,
        relation: row.relation,
        fact: row.fact,
        #[allow(clippy::cast_possible_truncation)]
        confidence: row.confidence as f32,
        valid_from: row.valid_from,
        valid_to: row.valid_to,
        created_at: row.created_at,
        expired_at: row.expired_at,
        source_message_id: row.source_message_id.map(MessageId),
        qdrant_point_id: row.qdrant_point_id,
        edge_type,
        retrieval_count: row.retrieval_count,
        last_retrieved_at: row.last_retrieved_at,
        superseded_by: row.superseded_by.map(i64::from),
        supersedes: row.supersedes,
        #[allow(clippy::cast_possible_truncation)]
        weight: row.weight as f32,
        #[allow(clippy::cast_possible_truncation)]
        confidence_fast: row.confidence_fast as f32,
        #[allow(clippy::cast_possible_truncation)]
        confidence_slow: row.confidence_slow as f32,
        #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
        turn_index: row.turn_index.map(|v| v as u32),
    }
}

#[derive(zeph_db::FromRow)]
struct CommunityRow {
    id: i64,
    name: String,
    summary: String,
    entity_ids: String,
    fingerprint: Option<String>,
    created_at: String,
    updated_at: String,
}

/// `graph_communities` projection for [`CommunityRow`], dialect-cast for `TIMESTAMPTZ` columns.
///
/// `created_at`/`updated_at` are `TIMESTAMPTZ` on Postgres — see [`entity_select_cols`].
/// `table_prefix` (e.g. `"c."` or `""`) is prepended to each source column reference so this
/// helper works both in plain `FROM graph_communities` queries and in joins with an alias.
fn community_select_cols(table_prefix: &str) -> String {
    let created_at_sel = <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text(&format!(
        "{table_prefix}created_at"
    ));
    let updated_at_sel = <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text(&format!(
        "{table_prefix}updated_at"
    ));
    format!(
        "{table_prefix}id, {table_prefix}name, {table_prefix}summary, {table_prefix}entity_ids, \
         {table_prefix}fingerprint, {created_at_sel} AS created_at, {updated_at_sel} AS updated_at"
    )
}

fn community_from_row(row: CommunityRow) -> Result<Community, MemoryError> {
    let raw_ids: Vec<i64> = serde_json::from_str(&row.entity_ids)?;
    let entity_ids = raw_ids.into_iter().map(EntityId).collect();
    Ok(Community {
        id: row.id,
        name: row.name,
        summary: row.summary,
        entity_ids,
        fingerprint: row.fingerprint,
        created_at: row.created_at,
        updated_at: row.updated_at,
    })
}

// ── GAAMA Episode methods ──────────────────────────────────────────────────────

impl GraphStore {
    /// Ensure a GAAMA episode exists for this conversation, returning its ID.
    ///
    /// Idempotent: inserts on first call, returns existing ID on subsequent calls.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.ensure_episode", skip_all)]
    pub async fn ensure_episode(&self, conversation_id: i64) -> Result<i64, MemoryError> {
        // Ensure the conversation row exists before inserting into graph_episodes,
        // which has a FK referencing conversations(id). On a fresh database the agent
        // may run graph extraction before the conversation row is committed.
        zeph_db::query(sql!(
            "INSERT INTO conversations (id) VALUES (?)
             ON CONFLICT (id) DO NOTHING"
        ))
        .bind(conversation_id)
        .execute(&self.pool)
        .await?;

        let id: i64 = zeph_db::query_scalar(sql!(
            "INSERT INTO graph_episodes (conversation_id)
             VALUES (?)
             ON CONFLICT(conversation_id) DO UPDATE SET conversation_id = excluded.conversation_id
             RETURNING id"
        ))
        .bind(conversation_id)
        .fetch_one(&self.pool)
        .await?;
        Ok(id)
    }

    /// Record that an entity was observed in an episode.
    ///
    /// Idempotent: does nothing if the link already exists.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.link_entity_to_episode", skip_all)]
    pub async fn link_entity_to_episode(
        &self,
        episode_id: i64,
        entity_id: i64,
    ) -> Result<(), MemoryError> {
        zeph_db::query(sql!(
            "INSERT INTO graph_episode_entities (episode_id, entity_id)
             VALUES (?, ?)
             ON CONFLICT (episode_id, entity_id) DO NOTHING"
        ))
        .bind(episode_id)
        .bind(entity_id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    /// Return all episodes in which an entity appears.
    ///
    /// # Errors
    ///
    /// Returns an error if the database query fails.
    #[tracing::instrument(name = "memory.graph.store.episodes_for_entity", skip_all)]
    pub async fn episodes_for_entity(
        &self,
        entity_id: i64,
    ) -> Result<Vec<super::types::Episode>, MemoryError> {
        #[derive(zeph_db::FromRow)]
        struct EpisodeRow {
            id: i64,
            conversation_id: i64,
            created_at: String,
            closed_at: Option<String>,
        }
        // `created_at`/`closed_at` are `TIMESTAMPTZ` on Postgres (`TEXT` on SQLite); project
        // both through `Dialect::select_as_text`, aliased back to their original names so
        // `#[derive(FromRow)]` still binds them into `EpisodeRow`'s `String`/`Option<String>`
        // fields.
        let created_at_sel =
            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("e.created_at");
        let closed_at_sel =
            <ActiveDialect as zeph_db::dialect::Dialect>::select_as_text("e.closed_at");
        let raw = format!(
            "SELECT e.id, e.conversation_id, {created_at_sel} AS created_at, \
                    {closed_at_sel} AS closed_at
             FROM graph_episodes e
             JOIN graph_episode_entities ee ON ee.episode_id = e.id
             WHERE ee.entity_id = ?"
        );
        let query_sql = zeph_db::rewrite_placeholders(&raw);
        let rows: Vec<EpisodeRow> = zeph_db::query_as(sqlx::AssertSqlSafe(query_sql))
            .bind(entity_id)
            .fetch_all(&self.pool)
            .await?;
        Ok(rows
            .into_iter()
            .map(|r| super::types::Episode {
                id: r.id,
                conversation_id: r.conversation_id,
                created_at: r.created_at,
                closed_at: r.closed_at,
            })
            .collect())
    }

    /// Insert a new edge using APEX-MEM append-only supersede semantics (FR-001).
    ///
    /// If a byte-identical active edge exists for `(src, tgt, canonical_relation, edge_type)`,
    /// records a reassertion in `edge_reassertions` and returns the existing edge id (FR-015).
    ///
    /// If a different active edge exists for the same `(src, canonical_relation, edge_type)` key,
    /// invalidates it with a supersession pointer, inserts the new edge, and sets `supersedes` on
    /// the new row to link the chain. Checks that the resulting chain depth would not exceed
    /// [`SUPERSEDE_DEPTH_CAP`] and that no cycle would be introduced before committing.
    ///
    /// Optionally increments [`ApexMetrics::supersedes_total`] when `metrics` is provided and a
    /// supersession actually occurs.
    ///
    /// # Errors
    ///
    /// - [`MemoryError::SupersedeCycle`] — the new edge would create a supersede cycle.
    /// - [`MemoryError::SupersedeDepthExceeded`] — chain depth cap would be exceeded.
    /// - [`MemoryError::Sqlx`] / [`MemoryError::Db`] — database errors.
    #[allow(clippy::too_many_arguments)]
    // TODO(B3): refactor into a builder or config struct to reduce argument count
    // function with many required inputs; a *Params struct would be more verbose without simplifying the call site
    #[tracing::instrument(name = "memory.graph.insert_or_supersede", skip_all)]
    pub async fn insert_or_supersede_with_metrics(
        &self,
        source_entity_id: i64,
        target_entity_id: i64,
        relation: &str,
        canonical_relation: &str,
        fact: &str,
        confidence: f32,
        episode_id: Option<MessageId>,
        edge_type: EdgeType,
        set_supersedes: bool,
        metrics: Option<&ApexMetrics>,
    ) -> Result<i64, MemoryError> {
        self.insert_or_supersede_with_turn_index_and_metrics(
            source_entity_id,
            target_entity_id,
            relation,
            canonical_relation,
            fact,
            confidence,
            episode_id,
            edge_type,
            set_supersedes,
            metrics,
            None,
            None,
        )
        .await
    }

    /// Insert or supersede an edge, recording turn-level provenance (#3710).
    ///
    /// `turn_index` is stored in the new edge row when a supersession chain is started.
    /// Pass `None` to omit provenance (equivalent to [`Self::insert_or_supersede_with_metrics`]).
    ///
    /// `provenance` stamps `origin`, `import_batch_id`, and `source_uri` on the newly inserted
    /// row. Pass `None` for conversation-origin edges (the default).
    ///
    /// # Errors
    ///
    /// Returns an error if the database transaction fails.
    #[allow(clippy::too_many_arguments)]
    #[tracing::instrument(name = "memory.graph.store.insert_or_supersede", skip_all)]
    pub async fn insert_or_supersede_with_turn_index_and_metrics(
        &self,
        source_entity_id: i64,
        target_entity_id: i64,
        relation: &str,
        canonical_relation: &str,
        fact: &str,
        confidence: f32,
        episode_id: Option<MessageId>,
        edge_type: EdgeType,
        set_supersedes: bool,
        metrics: Option<&ApexMetrics>,
        turn_index: Option<u32>,
        provenance: Option<&GraphProvenance>,
    ) -> Result<i64, MemoryError> {
        if source_entity_id == target_entity_id {
            return Err(MemoryError::InvalidInput(format!(
                "self-loop edge rejected: source and target are the same entity (id={source_entity_id})"
            )));
        }
        let confidence = confidence.clamp(0.0, 1.0);
        let edge_type_str = edge_type.as_str();
        let episode_raw: Option<i64> = episode_id.map(|m| m.0);

        let mut tx = zeph_db::begin(&self.pool).await?;

        if let Some(existing_id) = find_identical_active_edge(
            &mut tx,
            source_entity_id,
            target_entity_id,
            canonical_relation,
            edge_type_str,
            fact,
        )
        .await?
        {
            record_reassertion(
                &mut tx,
                existing_id,
                episode_raw,
                confidence,
                self.benna_fast_rate,
                self.benna_slow_rate,
            )
            .await?;
            tx.commit().await?;
            return Ok(existing_id);
        }

        let prior_head =
            find_prior_active_head(&mut tx, source_entity_id, canonical_relation, edge_type_str)
                .await?;

        // Cycle guard: depth cap also bounds cycles. Run inside the transaction using sqlx
        // directly to avoid a second pool acquire (SQLite pool is typically size 1 in tests).
        if let Some(head_id) = prior_head {
            check_supersede_depth_in_tx(&mut tx, head_id).await?;
        }

        // Expire the prior head BEFORE inserting the new row so that the partial unique index
        // uq_graph_edges_active_head is not violated. SQLite enforces unique indexes at statement
        // level, not at transaction commit, so the deactivation must precede the INSERT.
        if let Some(head_id) = prior_head {
            expire_prior_head(&mut tx, head_id).await?;
        }

        let supersedes_val: Option<i64> = if set_supersedes { prior_head } else { None };
        let turn_index_raw: Option<i64> = turn_index.map(i64::from);
        let (prov_origin, prov_batch, prov_uri) = provenance_parts(provenance);
        let new_id = insert_new_edge(
            &mut tx,
            source_entity_id,
            target_entity_id,
            relation,
            canonical_relation,
            fact,
            confidence,
            episode_raw,
            edge_type_str,
            supersedes_val,
            turn_index_raw,
            prov_origin,
            prov_batch,
            prov_uri,
        )
        .await?;

        if let Some(head_id) = prior_head {
            set_superseded_by(&mut tx, head_id, new_id).await?;
            if let Some(m) = metrics {
                m.supersedes_total
                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
            }
        }

        tx.commit().await?;
        Ok(new_id)
    }

    /// Convenience wrapper: calls [`Self::insert_or_supersede_with_metrics`] with `metrics = None`.
    ///
    /// # Errors
    ///
    /// Propagates errors from [`Self::insert_or_supersede_with_metrics`].
    #[allow(clippy::too_many_arguments)]
    // function with many required inputs; a *Params struct would be more verbose without simplifying the call site
    #[tracing::instrument(name = "memory.graph.store.insert_or_supersede", skip_all)]
    pub async fn insert_or_supersede(
        &self,
        source_entity_id: i64,
        target_entity_id: i64,
        relation: &str,
        canonical_relation: &str,
        fact: &str,
        confidence: f32,
        episode_id: Option<MessageId>,
        edge_type: EdgeType,
        set_supersedes: bool,
    ) -> Result<i64, MemoryError> {
        self.insert_or_supersede_with_metrics(
            source_entity_id,
            target_entity_id,
            relation,
            canonical_relation,
            fact,
            confidence,
            episode_id,
            edge_type,
            set_supersedes,
            None,
        )
        .await
    }

    /// Insert or supersede an edge, then run implicit conflict detection if a detector is provided.
    ///
    /// Calls [`Self::insert_or_supersede_with_metrics`] first, then — when `detector` is `Some` and
    /// the ontology confirms the predicate is cardinality-1 — queries active edges on the same source
    /// entity, detects conflict candidates, and stages them in `implicit_conflict_candidates`.
    ///
    /// Conflict detection failures are logged and swallowed so they never block the write path.
    ///
    /// # Errors
    ///
    /// Propagates errors from [`Self::insert_or_supersede_with_metrics`].
    #[allow(clippy::too_many_arguments)]
    // function with many required inputs; a *Params struct would be more verbose without simplifying the call site
    #[tracing::instrument(name = "memory.graph.store.insert_or_supersede_conflict", skip_all)]
    pub async fn insert_or_supersede_with_conflict_detection(
        &self,
        source_entity_id: i64,
        target_entity_id: i64,
        relation: &str,
        canonical_relation: &str,
        fact: &str,
        confidence: f32,
        episode_id: Option<MessageId>,
        edge_type: EdgeType,
        set_supersedes: bool,
        metrics: Option<&ApexMetrics>,
        detector: Option<&crate::graph::implicit_conflict::ImplicitConflictDetector>,
        ontology: Option<&crate::graph::ontology::OntologyTable>,
    ) -> Result<i64, MemoryError> {
        let new_id = self
            .insert_or_supersede_with_metrics(
                source_entity_id,
                target_entity_id,
                relation,
                canonical_relation,
                fact,
                confidence,
                episode_id,
                edge_type,
                set_supersedes,
                metrics,
            )
            .await?;

        if let (Some(det), Some(onto)) = (detector, ontology)
            && det.is_enabled()
        {
            let is_cardinality_n =
                onto.cardinality(canonical_relation) == crate::graph::ontology::Cardinality::Many;

            // Query existing active edges for the same source entity (excluding the new one).
            let existing_raw = zeph_db::query(sql!(
                "SELECT id, canonical_relation FROM graph_edges
                     WHERE source_entity_id = ?
                       AND id != ?
                       AND expired_at IS NULL"
            ))
            .bind(source_entity_id)
            .bind(new_id)
            .fetch_all(&self.pool)
            .await;

            match existing_raw {
                Ok(rows) => {
                    use sqlx::Row as _;
                    let existing: Vec<(i64, String)> = rows
                        .into_iter()
                        .map(|r| {
                            let id: i64 = r.try_get("id").unwrap_or(0);
                            let rel: String = r.try_get("canonical_relation").unwrap_or_default();
                            (id, rel)
                        })
                        .collect();
                    let existing_refs: Vec<(i64, &str)> = existing
                        .iter()
                        .map(|(id, rel)| (*id, rel.as_str()))
                        .collect();

                    let candidates = det.detect_candidates(
                        new_id,
                        canonical_relation,
                        &existing_refs,
                        is_cardinality_n,
                    );

                    if !candidates.is_empty() {
                        let ttl = det.candidate_ttl_days();
                        match zeph_db::begin(&self.pool).await {
                            Ok(mut tx) => {
                                match det.stage_candidates(&candidates, &mut tx, ttl).await {
                                    Ok(()) => {
                                        if let Err(e) = tx.commit().await {
                                            tracing::warn!(
                                                error = %e,
                                                "implicit conflict tx commit failed (non-fatal)"
                                            );
                                        }
                                    }
                                    Err(e) => {
                                        tracing::warn!(
                                            error = %e,
                                            "implicit conflict staging failed (non-fatal)"
                                        );
                                    }
                                }
                            }
                            Err(e) => {
                                tracing::warn!(
                                    error = %e,
                                    "implicit conflict: tx begin failed (non-fatal)"
                                );
                            }
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "implicit conflict: failed to query existing edges (non-fatal)"
                    );
                }
            }
        }

        Ok(new_id)
    }

    /// Walk the `supersedes` chain from `head_id` using a single recursive CTE and return its depth.
    ///
    /// Returns `0` when the edge has no `supersedes` pointer (it is the root).
    /// The CTE is capped at `SUPERSEDE_DEPTH_CAP + 1` to prevent unbounded recursion.
    ///
    /// # Errors
    ///
    /// Returns [`MemoryError::SupersedeCycle`] when the CTE detects a cycle (depth exceeds cap
    /// but the chain has not terminated), or a database error on query failure.
    #[tracing::instrument(name = "memory.graph.store.check_supersede_depth", skip_all)]
    pub async fn check_supersede_depth(&self, head_id: i64) -> Result<usize, MemoryError> {
        Self::check_supersede_depth_with_pool(&self.pool, head_id).await
    }

    #[tracing::instrument(name = "memory.graph.store.check_supersede_depth_pool", skip_all)]
    async fn check_supersede_depth_with_pool(
        pool: &zeph_db::DbPool,
        head_id: i64,
    ) -> Result<usize, MemoryError> {
        let cap = i64::try_from(SUPERSEDE_DEPTH_CAP + 1).unwrap_or(i64::MAX);
        // CTE walks the supersedes chain starting at head_id (depth=0).
        // Each hop to a superseded ancestor increments depth. NULL supersedes terminates the walk.
        let depth: Option<i64> = zeph_db::query_scalar(sql!(
            "WITH RECURSIVE chain(id, depth) AS (
               SELECT id, 0 FROM graph_edges WHERE id = ?
               UNION ALL
               SELECT e.supersedes, c.depth + 1
               FROM graph_edges e JOIN chain c ON e.id = c.id
               WHERE e.supersedes IS NOT NULL AND c.depth < ?
             )
             SELECT MAX(depth) FROM chain"
        ))
        .bind(head_id)
        .bind(cap)
        .fetch_optional(pool)
        .await?
        .flatten();

        #[allow(clippy::cast_sign_loss, clippy::cast_possible_truncation)]
        let d = depth.unwrap_or(0) as usize;
        if d > SUPERSEDE_DEPTH_CAP {
            return Err(MemoryError::SupersedeCycle(head_id));
        }
        Ok(d)
    }

    // ── MemCoT provenance helpers (issue #3574 / #3575) ───────────────────────

    /// Fetch `(edge_id, source_message_id)` pairs for a batch of edge ids.
    ///
    /// Returns only rows where `source_message_id` (`episode_id` column) is not NULL.
    /// Called from [`crate::semantic::SemanticMemory::recall_graph_view`] for Zoom-In provenance.
    ///
    /// # Errors
    ///
    /// Returns [`crate::error::MemoryError`] if the SQL query fails.
    #[tracing::instrument(name = "memory.graph.store.source_message_ids_for_edges", skip_all, fields(count = edge_ids.len()))]
    pub async fn source_message_ids_for_edges(
        &self,
        edge_ids: &[i64],
    ) -> Result<Vec<(i64, crate::types::MessageId)>, crate::error::MemoryError> {
        if edge_ids.is_empty() {
            return Ok(Vec::new());
        }
        let placeholders = placeholder_list(1, edge_ids.len());
        let sql = format!(
            "SELECT id, episode_id FROM graph_edges \
             WHERE id IN ({placeholders}) AND episode_id IS NOT NULL"
        );
        let mut q =
            zeph_db::query_as::<_, (i64, crate::types::MessageId)>(sqlx::AssertSqlSafe(sql));
        for &eid in edge_ids {
            q = q.bind(eid);
        }
        let rows = q.fetch_all(&self.pool).await?;
        Ok(rows)
    }

    /// Return the `source_entity_id` for a given edge id, or `None` if not found.
    ///
    /// Used by `recall_graph_view` with `ZoomOut` to seed the 1-hop BFS.
    ///
    /// # Errors
    ///
    /// Returns [`crate::error::MemoryError`] if the SQL query fails.
    #[tracing::instrument(name = "memory.graph.store.source_entity_id_for_edge", skip_all)]
    pub async fn source_entity_id_for_edge(
        &self,
        edge_id: i64,
    ) -> Result<Option<i64>, crate::error::MemoryError> {
        let row: Option<i64> = zeph_db::query_scalar(sql!(
            "SELECT source_entity_id FROM graph_edges WHERE id = ? AND valid_to IS NULL LIMIT 1"
        ))
        .bind(edge_id)
        .fetch_optional(&self.pool)
        .await?;
        Ok(row)
    }

    /// Fetch all active edges exactly 1 hop away from `entity_id`, filtered by `edge_types`.
    ///
    /// Returns a flat list of [`crate::recall_view::RecalledFact`] wrapping `GraphFact` values
    /// with `hop_distance = 1`. Used by `recall_graph_view` for `ZoomOut` neighbor expansion.
    ///
    /// # Errors
    ///
    /// Returns [`crate::error::MemoryError`] if the SQL query or entity name lookup fails.
    #[tracing::instrument(name = "memory.graph.store.bfs_edges_at_depth", skip_all)]
    pub async fn bfs_edges_at_depth(
        &self,
        entity_id: i64,
        _depth: u32,
        edge_types: &[crate::graph::types::EdgeType],
    ) -> Result<Vec<crate::recall_view::RecalledFact>, crate::error::MemoryError> {
        let type_strs: Vec<&str> = if edge_types.is_empty() {
            vec!["semantic", "temporal", "causal", "entity"]
        } else {
            edge_types.iter().map(|et| et.as_str()).collect()
        };

        let ph = placeholder_list(1, type_strs.len());
        let src_pos = type_strs.len() + 1;
        let tgt_pos = src_pos + 1;
        let src_ph = numbered_placeholder(src_pos);
        let tgt_ph = numbered_placeholder(tgt_pos);

        let sql = format!(
            "SELECT {edge_cols}
             FROM graph_edges ge
             WHERE ge.edge_type IN ({ph})
               AND ge.valid_to IS NULL
               AND (ge.source_entity_id = {src_ph} OR ge.target_entity_id = {tgt_ph})
             LIMIT 200",
            edge_cols = edge_select_cols("ge.")
        );

        let mut q = zeph_db::query_as::<_, EdgeRow>(sqlx::AssertSqlSafe(sql));
        for t in &type_strs {
            q = q.bind(*t);
        }
        q = q.bind(entity_id).bind(entity_id);

        let rows = q.fetch_all(&self.pool).await?;

        // Resolve entity names from ids.
        let all_ids: Vec<i64> = rows
            .iter()
            .flat_map(|r| [r.source_entity_id, r.target_entity_id])
            .collect::<std::collections::HashSet<_>>()
            .into_iter()
            .collect();

        let mut name_map: std::collections::HashMap<i64, String> = std::collections::HashMap::new();
        for chunk in all_ids.chunks(SQLITE_BATCH_LIMIT_2X) {
            let ph2 = placeholder_list(1, chunk.len());
            let name_sql =
                format!("SELECT id, canonical_name FROM graph_entities WHERE id IN ({ph2})");
            let mut nq = zeph_db::query_as::<_, (i64, String)>(sqlx::AssertSqlSafe(name_sql));
            for &id in chunk {
                nq = nq.bind(id);
            }
            let name_rows = nq.fetch_all(&self.pool).await?;
            for (id, name) in name_rows {
                name_map.insert(id, name);
            }
        }

        let facts: Vec<crate::recall_view::RecalledFact> = rows
            .into_iter()
            .filter_map(|row| {
                let edge = edge_from_row(row);
                let entity_name = name_map.get(&edge.source_entity_id).cloned()?;
                let target_name = name_map.get(&edge.target_entity_id).cloned()?;
                if entity_name.is_empty() || target_name.is_empty() {
                    return None;
                }
                let fact = crate::graph::types::GraphFact {
                    entity_name,
                    relation: edge.canonical_relation.clone(),
                    target_name,
                    fact: edge.fact.clone(),
                    entity_match_score: 0.5,
                    hop_distance: 1,
                    confidence: edge.confidence,
                    valid_from: if edge.valid_from.is_empty() {
                        None
                    } else {
                        Some(edge.valid_from.clone())
                    },
                    edge_type: edge.edge_type,
                    retrieval_count: edge.retrieval_count,
                    edge_id: Some(edge.id),
                };
                Some(crate::recall_view::RecalledFact::from_graph_fact(fact))
            })
            .collect();

        Ok(facts)
    }
}

// ── insert_or_supersede helpers ───────────────────────────────────────────────
// All helpers take `&mut zeph_db::DbTransaction` so they run inside the caller's transaction
// without acquiring a second connection (SQLite pool is typically size 1 in tests).

type Tx<'a> = zeph_db::DbTransaction<'a>;

/// Look up a byte-identical active edge (FR-015 reassertion path).
///
/// Returns the existing edge ID when all columns match exactly, or `None` otherwise.
#[tracing::instrument(name = "memory.graph.store.find_identical_active_edge", skip_all)]
async fn find_identical_active_edge(
    tx: &mut Tx<'_>,
    src: i64,
    tgt: i64,
    canon: &str,
    edge_type_str: &str,
    fact: &str,
) -> Result<Option<i64>, MemoryError> {
    Ok(zeph_db::query_scalar(sql!(
        "SELECT id FROM graph_edges
         WHERE source_entity_id = ?
           AND target_entity_id = ?
           AND canonical_relation = ?
           AND edge_type = ?
           AND fact = ?
           AND valid_to IS NULL
           AND expired_at IS NULL
         LIMIT 1"
    ))
    .bind(src)
    .bind(tgt)
    .bind(canon)
    .bind(edge_type_str)
    .bind(fact)
    .fetch_optional(&mut **tx)
    .await?)
}

/// Record a reassertion event for an existing edge (FR-015) and apply the Benna-Fusi
/// multi-timescale update to `confidence_fast`/`confidence_slow` in `graph_edges` (#3709).
///
/// This ensures SYNAPSE synaptic variables are updated on every reassertion, including the
/// APEX append-only path where identical edges are not superseded but re-confirmed.
#[tracing::instrument(name = "memory.graph.store.record_reassertion", skip_all)]
async fn record_reassertion(
    tx: &mut Tx<'_>,
    head_id: i64,
    episode_raw: Option<i64>,
    confidence: f32,
    benna_fast_rate: f32,
    benna_slow_rate: f32,
) -> Result<(), MemoryError> {
    #[allow(clippy::cast_possible_wrap)]
    let asserted_at = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs() as i64;
    zeph_db::query(sql!(
        "INSERT INTO edge_reassertions (head_edge_id, asserted_at, episode_id, confidence)
         VALUES (?, ?, ?, ?)"
    ))
    .bind(head_id)
    .bind(asserted_at)
    .bind(episode_raw)
    .bind(f64::from(confidence))
    .execute(&mut **tx)
    .await?;

    // Apply Benna-Fusi update to the head edge's synaptic variables.
    let existing: Option<(f64, f64)> = zeph_db::query_as(sql!(
        "SELECT CAST(confidence_fast AS DOUBLE PRECISION), \
                CAST(confidence_slow AS DOUBLE PRECISION) FROM graph_edges WHERE id = ? LIMIT 1"
    ))
    .bind(head_id)
    .fetch_optional(&mut **tx)
    .await?;

    if let Some((stored_fast, stored_slow)) = existing {
        #[allow(clippy::cast_possible_truncation)]
        let stored_fast_f32 = (stored_fast as f32).clamp(0.0, 1.0);
        #[allow(clippy::cast_possible_truncation)]
        let stored_slow_f32 = (stored_slow as f32).clamp(0.0, 1.0);
        let new_fast = stored_fast_f32 + benna_fast_rate * (confidence - stored_fast_f32);
        let new_slow = stored_slow_f32 + benna_slow_rate * (new_fast - stored_slow_f32);
        zeph_db::query(sql!(
            "UPDATE graph_edges SET confidence_fast = ?, confidence_slow = ? WHERE id = ?"
        ))
        .bind(f64::from(new_fast))
        .bind(f64::from(new_slow))
        .bind(head_id)
        .execute(&mut **tx)
        .await?;
    }

    Ok(())
}

/// Find the current active head edge for `(src, canonical_relation, edge_type)`.
#[tracing::instrument(name = "memory.graph.store.find_prior_active_head", skip_all)]
async fn find_prior_active_head(
    tx: &mut Tx<'_>,
    src: i64,
    canon: &str,
    edge_type_str: &str,
) -> Result<Option<i64>, MemoryError> {
    Ok(zeph_db::query_scalar(sql!(
        "SELECT id FROM graph_edges
         WHERE source_entity_id = ?
           AND canonical_relation = ?
           AND edge_type = ?
           AND valid_to IS NULL
           AND expired_at IS NULL
         ORDER BY created_at DESC
         LIMIT 1"
    ))
    .bind(src)
    .bind(canon)
    .bind(edge_type_str)
    .fetch_optional(&mut **tx)
    .await?)
}

/// Verify the supersede chain depth for `head_id` does not exceed [`SUPERSEDE_DEPTH_CAP`].
///
/// Returns `Err(MemoryError::SupersedeDepthExceeded)` when the cap would be exceeded.
#[tracing::instrument(name = "memory.graph.store.check_supersede_depth_in_tx", skip_all)]
async fn check_supersede_depth_in_tx(tx: &mut Tx<'_>, head_id: i64) -> Result<(), MemoryError> {
    let cap = i64::try_from(SUPERSEDE_DEPTH_CAP + 1).unwrap_or(i64::MAX);
    let depth: Option<i64> = zeph_db::query_scalar(sql!(
        "WITH RECURSIVE chain(id, depth) AS (
           SELECT supersedes, 1 FROM graph_edges WHERE id = ? AND supersedes IS NOT NULL
           UNION ALL
           SELECT e.supersedes, c.depth + 1
           FROM graph_edges e JOIN chain c ON e.id = c.id
           WHERE e.supersedes IS NOT NULL AND c.depth < ?
         )
         SELECT MAX(depth) FROM chain"
    ))
    .bind(head_id)
    .bind(cap)
    .fetch_optional(&mut **tx)
    .await?
    .flatten();
    let d = usize::try_from(depth.unwrap_or(0)).unwrap_or(usize::MAX);
    if d > SUPERSEDE_DEPTH_CAP {
        return Err(MemoryError::SupersedeDepthExceeded(head_id));
    }
    Ok(())
}

/// Insert a new edge row and return its generated ID.
#[allow(clippy::too_many_arguments)]
#[tracing::instrument(name = "memory.graph.store.insert_new_edge", skip_all)]
async fn insert_new_edge(
    tx: &mut Tx<'_>,
    src: i64,
    tgt: i64,
    relation: &str,
    canonical_relation: &str,
    fact: &str,
    confidence: f32,
    episode_raw: Option<i64>,
    edge_type_str: &str,
    supersedes_val: Option<i64>,
    turn_index: Option<i64>,
    origin: &str,
    import_batch_id: Option<&str>,
    source_uri: Option<&str>,
) -> Result<i64, MemoryError> {
    Ok(zeph_db::query_scalar(sql!(
        "INSERT INTO graph_edges
         (source_entity_id, target_entity_id, relation, canonical_relation, fact,
          confidence, confidence_fast, confidence_slow, episode_id, edge_type, supersedes, turn_index,
          origin, import_batch_id, source_uri)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         RETURNING id"
    ))
    .bind(src)
    .bind(tgt)
    .bind(relation)
    .bind(canonical_relation)
    .bind(fact)
    .bind(f64::from(confidence))
    .bind(f64::from(confidence))
    .bind(f64::from(confidence))
    .bind(episode_raw)
    .bind(edge_type_str)
    .bind(supersedes_val)
    .bind(turn_index)
    .bind(origin)
    .bind(import_batch_id)
    .bind(source_uri)
    .fetch_one(&mut **tx)
    .await?)
}

/// Deactivate the prior head by setting `valid_to` and `expired_at`, leaving `superseded_by`
/// unset. Must be called **before** inserting the new row to avoid violating
/// `uq_graph_edges_active_head` — the unique index is enforced at statement level, not at commit.
#[tracing::instrument(name = "memory.graph.store.expire_prior_head", skip_all)]
async fn expire_prior_head(tx: &mut Tx<'_>, head_id: i64) -> Result<(), MemoryError> {
    zeph_db::query(sql!(
        "UPDATE graph_edges
         SET valid_to = CURRENT_TIMESTAMP,
             expired_at = CURRENT_TIMESTAMP
         WHERE id = ?"
    ))
    .bind(head_id)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

/// Back-fill `superseded_by` on the already-expired prior head after the new row has been
/// inserted and its ID is known.
#[tracing::instrument(name = "memory.graph.store.set_superseded_by", skip_all)]
async fn set_superseded_by(tx: &mut Tx<'_>, head_id: i64, new_id: i64) -> Result<(), MemoryError> {
    zeph_db::query(sql!(
        "UPDATE graph_edges SET superseded_by = ? WHERE id = ?"
    ))
    .bind(new_id)
    .bind(head_id)
    .execute(&mut **tx)
    .await?;
    Ok(())
}

// ── FTS helpers ───────────────────────────────────────────────────────────────

/// Row type for the UNION ALL FTS5 query in `find_entities_ranked`.
type EntityFtsRow = (
    i64,
    String,
    String,
    String,
    Option<String>,
    String,
    String,
    Option<String>,
    f64,
);

/// Sanitize and tokenize `query` into an FTS5 query string.
///
/// Returns `None` when the input is empty or contains only FTS5 operator tokens,
/// indicating no search should be attempted.
fn build_fts_query(query: &str) -> Option<String> {
    const FTS5_OPERATORS: &[&str] = &["AND", "OR", "NOT", "NEAR"];
    let sanitized = sanitize_fts_query(query);
    if sanitized.is_empty() {
        return None;
    }
    let fts_query: String = sanitized
        .split_whitespace()
        .filter(|t| !FTS5_OPERATORS.contains(t))
        .map(|t| format!("{t}*"))
        .collect::<Vec<_>>()
        .join(" ");
    if fts_query.is_empty() {
        None
    } else {
        Some(fts_query)
    }
}

/// Build the UNION ALL FTS5 SQL for entity ranked search.
///
/// The SQL selects entities by direct FTS5 match (negated BM25) and by alias LIKE match
/// (fixed score 0.5), then orders by score descending with a LIMIT applied outside.
fn build_ranked_fts_sql() -> String {
    format!(
        "SELECT * FROM ( \
             SELECT e.id, e.name, e.canonical_name, e.entity_type, e.summary, \
                    e.first_seen_at, e.last_seen_at, e.qdrant_point_id, \
                    -bm25(graph_entities_fts, 10.0, 1.0) AS fts_rank \
             FROM graph_entities_fts fts \
             JOIN graph_entities e ON e.id = fts.rowid \
             WHERE graph_entities_fts MATCH ? \
             UNION ALL \
             SELECT e.id, e.name, e.canonical_name, e.entity_type, e.summary, \
                    e.first_seen_at, e.last_seen_at, e.qdrant_point_id, \
                    0.5 AS fts_rank \
             FROM graph_entity_aliases a \
             JOIN graph_entities e ON e.id = a.entity_id \
             WHERE a.alias_name LIKE ? ESCAPE '\\' {} \
         ) \
         ORDER BY fts_rank DESC \
         LIMIT ?",
        <ActiveDialect as zeph_db::dialect::Dialect>::COLLATE_NOCASE,
    )
}

/// Normalize FTS scores to `[0, 1]` and deduplicate by entity ID.
///
/// Deduplication keeps the first (highest-ranked) occurrence of each entity ID.
fn normalize_and_dedup(rows: Vec<EntityFtsRow>) -> Vec<(Entity, f32)> {
    // Guard against div-by-zero when all scores are 0.
    let max_score: f64 = rows.iter().map(|r| r.8).fold(0.0_f64, f64::max);
    let max_score = if max_score <= 0.0 { 1.0 } else { max_score };

    let mut seen_ids: std::collections::HashSet<i64> = std::collections::HashSet::new();
    let mut result: Vec<(Entity, f32)> = Vec::with_capacity(rows.len());
    for (
        id,
        name,
        canonical_name,
        entity_type_str,
        summary,
        first_seen_at,
        last_seen_at,
        qdrant_point_id,
        raw_score,
    ) in rows
    {
        if !seen_ids.insert(id) {
            continue;
        }
        let entity_type = entity_type_str
            .parse()
            .unwrap_or(super::types::EntityType::Concept);
        let entity = Entity {
            id: EntityId(id),
            name,
            canonical_name,
            entity_type,
            summary,
            first_seen_at,
            last_seen_at,
            qdrant_point_id,
        };
        #[allow(clippy::cast_possible_truncation)]
        let normalized = (raw_score / max_score).clamp(0.0, 1.0) as f32;
        result.push((entity, normalized));
    }
    result
}

// ── Tests ─────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests;