solo-storage 0.7.1

// SPDX-License-Identifier: Apache-2.0

//! `WriterActor`, `WriteCommand`, `WriteHandle` — single-writer actor on a
//! dedicated OS thread. See ADR-0003 §"Trait shapes" and §"Operational
//! invariants".
//!
//! ## Why a dedicated OS thread (not a tokio task)
//!
//! `rusqlite::Connection::execute` blocks the current thread. If the actor
//! ran on a tokio worker, every write would block one worker for the
//! duration of the SQL — under burst load the runtime would starve and
//! every other task (HTTP handlers, MCP handlers, the snapshot timer) would
//! stall. The dedicated thread isolates that blocking from the runtime.
//!
//! Inside `run()` we use `mpsc::Receiver::blocking_recv()`; from outside the
//! actor, `WriteHandle::send().await` is async, so callers don't block on
//! sends.
//!
//! ## Reply-before-drain (ADR-0003 §P8-E)
//!
//! `Remember` carries a oneshot reply channel. The reply is sent **after**
//! the SQL transaction commits and `hnsw.add` succeeds, but **before** the
//! `pending_index` row is drained. Callers see "Ok = durable AND searchable"
//! without waiting on cleanup. If the drain itself fails, the row replays
//! on next startup — same end state.
//!
//! ## What's stubbed in commit 1.2
//!
//! `handle_forget`, `handle_consolidate`, `handle_reembed`,
//! `handle_save_snapshot` return `Error::Other("not yet implemented (commit
//! 1.x)")`. They're plumbed through the dispatch so callers can wire the
//! API today; the bodies fill in as the relevant commits land.

use std::path::PathBuf;
use std::str::FromStr;
use std::sync::Arc;

use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params, params_from_iter};
use solo_core::{Embedder, Embedding, Episode, Error, MemoryId, Result, Tier, VectorIndex};
use tokio::runtime::Handle;
use tokio::sync::{mpsc, oneshot};

use crate::backup::backup_from_connection;
use crate::hnsw_id::{chunk_hnsw_id, episode_hnsw_id};
use crate::key_material::KeyMaterial;

/// Default mpsc channel capacity. ADR-0003 §"Channel capacity": 1024 lets a
/// 1000-write consolidation burst land without backpressure-blocking the
/// regular write path.
pub const DEFAULT_CHANNEL_CAPACITY: usize = 1024;

/// Filter + flags for `WriteCommand::Consolidate`.
///
/// Implements `Deserialize` so HTTP / future MCP transports can
/// build it from a JSON request body without extra plumbing.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct ConsolidationScope {
    /// If `Some(N)`, only consolidate episodes with `ts_ms` in the last
    /// N days. `None` = walk all `tier='hot' AND status='active'` rows.
    /// Bounded windows are typical for the daemon's nightly timer;
    /// unbounded for a one-shot bulk run.
    pub window_days: Option<i64>,
    /// When `true`, run the merge + regen passes even if there are
    /// zero unclustered episodes to feed to `cluster_episodes`. The
    /// usual flow short-circuits on empty candidates because there's
    /// no fresh work — but pre-existing clusters can drift across
    /// runs and should occasionally coalesce regardless. `--force-merge`
    /// on `solo consolidate` (or `force_merge: true` in the HTTP
    /// JSON body) opts into that drift catch-up.
    #[serde(default)]
    pub force_merge: bool,
}

/// What `WriteCommand::Consolidate` returns to the caller. v0.2.0
/// covers the SWS-equivalent clustering pass; abstraction +
/// contradiction counts will populate in later commits.
///
/// `Serialize` so HTTP responses can ship it directly as JSON.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct ConsolidationReport {
    /// Distinct episodes the candidate query returned.
    pub episodes_seen: usize,
    /// Clusters that survived the size + threshold filter and were
    /// persisted to the `clusters` table.
    pub clusters_built: usize,
    /// Sum of episode_ids across all built clusters (i.e. how many
    /// `cluster_episodes` rows were inserted).
    pub episodes_clustered: usize,
    /// Number of clusters absorbed into a survivor by the centroid-
    /// merge pass (`solo_steward::cluster::merge_clusters_by_centroid`).
    /// Closes the cross-UTC-day-bucket case where conversations
    /// straddling midnight produce two clusters with similar centroids.
    /// 0 when no merges were possible. Counts losers, not survivors —
    /// `clusters_built` reflects the post-merge count.
    pub clusters_merged: usize,
    /// Number of freshly-built clusters absorbed into pre-existing
    /// DB clusters via
    /// `solo_steward::cluster::absorb_into_existing` (cross-run
    /// re-consolidation). The absorbed cluster never gets its own
    /// `clusters` row; its episodes link under the existing
    /// cluster_id, and the existing cluster's centroid + coherence
    /// refresh. Counts new-side clusters; `clusters_built` reflects
    /// the post-absorb count of brand-new clusters that survived to
    /// be inserted.
    pub clusters_absorbed: usize,
    /// Number of pre-existing clusters absorbed into another
    /// pre-existing cluster by the existing-vs-existing merge pass
    /// (`solo_steward::cluster::plan_existing_merges`). Closes the
    /// long-tail case where two clusters drift toward each other
    /// over time via repeated absorbs and should now coalesce.
    /// Counts losers, not survivors. Each loser's
    /// `cluster_episodes` rows reassign to the survivor's
    /// `cluster_id`, then the loser row is DELETEd (cascading via
    /// the 0001 + 0002 FKs through `cluster_episodes` (already
    /// empty after the UPDATE), `semantic_abstractions`, and
    /// `triples`). Survivor's stale abstraction is then regenerated
    /// by the same regen pass that handles cross-run absorb
    /// modifications.
    pub existing_clusters_merged: usize,
    /// Number of pre-existing clusters whose stale
    /// `semantic_abstractions` + linked `triples` were dropped and
    /// regenerated as a follow-on to the cross-run absorb pass.
    /// Equal to the count of distinct existing cluster_ids that
    /// absorbed at least one new cluster, **as long as** the
    /// regenerate-abstract LLM call succeeds; per-cluster failures
    /// are logged + skipped (the cluster row + its absorbed episodes
    /// stay; the abstraction row stays empty until the next run).
    /// 0 when no LLM steward is wired or no absorptions happened.
    pub abstractions_regenerated: usize,
    /// Number of `semantic_abstractions` rows successfully persisted
    /// (Y.3.3). 0 when the writer was spawned without a `Steward` —
    /// the prod default until a real `LlmClient` ships.
    pub abstractions_built: usize,
    /// Number of `triples` rows persisted alongside the abstractions
    /// (Y.3.3). Each abstraction can produce 0..N triples; the LLM
    /// is asked to extract them but may legitimately return none.
    pub triples_built: usize,
    /// Reserved for Y.4. Always 0 in this commit.
    pub contradictions_found: usize,
}

/// Filter + flags for `WriteCommand::Reembed`.
#[derive(Debug, Clone, Default)]
pub struct ReembedScope {
    /// If `Some((name, version))`, only reembed memories whose existing
    /// embedding row was produced by this embedder identity. If `None`,
    /// every memory whose embedding's `embedder_id` differs from the
    /// writer's current `embedder_id` is a candidate.
    pub from: Option<(String, String)>,
    /// Walk + count only; write nothing.
    pub dry_run: bool,
    /// After re-embedding each touched memory, DELETE the prior
    /// `embeddings` rows for that memory whose `embedder_id` differs
    /// from the current. Without this flag, stale rows are retained
    /// for forensics or rollback.
    pub gc: bool,
}

#[derive(Debug, Clone, Default)]
pub struct ReembedReport {
    /// Distinct memory_ids that matched the candidate query.
    pub rows_seen: usize,
    /// Memories whose new embedding was successfully written.
    pub rows_reembedded: usize,
    /// Memories that hit an error during embed or insert.
    pub rows_failed: usize,
    /// Number of stale `embeddings` rows DELETED via `--gc`.
    pub rows_gc_deleted: usize,
    /// Mirrors the scope flag so callers can format output without
    /// retaining the original scope.
    pub dry_run: bool,
}

/// What `WriteCommand::NormalizeSubjects` returns to the caller.
///
/// Opt-in backfill tool: rewrites historical `triples.subject_id` and
/// `triples.object_id` values according to a caller-supplied alias map.
/// The companion to v0.5.0's read-path alias resolution
/// (`IdentityConfig.user_aliases`) — that bridges queries transparently
/// against existing rows, while this rewrites the underlying data so
/// downstream consumers (third-party tools, exports) see the canonical
/// identity. See `docs/dev-log/0071-v0.5.x-roadmap.md` Priority 10.
#[derive(Debug, Clone, Default, serde::Serialize)]
pub struct NormalizeReport {
    /// Number of `(from, to)` alias pairs processed (== `scope.aliases.len()`).
    pub aliases_processed: usize,
    /// Rows whose `subject_id` was rewritten (summed across all pairs).
    /// In `dry_run` mode this is the count that *would* be rewritten;
    /// the transaction is rolled back before the change is persisted.
    pub subject_rows_updated: usize,
    /// Rows whose `object_id` was rewritten (summed across all pairs).
    /// Same dry-run semantics as `subject_rows_updated`.
    pub object_rows_updated: usize,
    /// Mirrors the scope flag so callers can format output without
    /// retaining the original scope.
    pub dry_run: bool,
}

/// Default per-file ingest size cap when `SOLO_INGEST_MAX_BYTES` is not
/// set in the environment. 50 MB matches the value advertised in
/// `docs/releases/v0.7.0.md` §"`SOLO_INGEST_MAX_BYTES`".
///
/// Above the cap, the writer returns an error before `parse_file` opens
/// the file, so the SQL and HNSW state are untouched. Set
/// `SOLO_INGEST_MAX_BYTES=0` to disable the cap entirely (caller-managed
/// resource bound).
pub const DEFAULT_INGEST_MAX_BYTES: u64 = 50 * 1024 * 1024;
pub(crate) const SOLO_INGEST_MAX_BYTES_ENV: &str = "SOLO_INGEST_MAX_BYTES";

/// Effective per-file ingest cap.
///
/// Returns:
///
///   - `Some(n)` — enforce a cap of `n` bytes.
///   - `None`    — cap disabled (env var explicitly set to `0`).
///
/// Resolution order:
///
///   1. If `SOLO_INGEST_MAX_BYTES` is unset → `Some(DEFAULT_INGEST_MAX_BYTES)`.
///   2. If `SOLO_INGEST_MAX_BYTES=0` → `None` (disabled).
///   3. If `SOLO_INGEST_MAX_BYTES=<positive integer>` → `Some(n)`.
///   4. If the env var is set but unparseable (negative, garbage, whitespace,
///      etc.) → `Some(DEFAULT_INGEST_MAX_BYTES)` plus a `tracing::warn!`. The
///      conservative fallback is "use the default" rather than "disable cap"
///      so a typo can't silently turn off the safety net.
pub fn resolve_ingest_max_bytes() -> Option<u64> {
    match std::env::var(SOLO_INGEST_MAX_BYTES_ENV) {
        Err(_) => Some(DEFAULT_INGEST_MAX_BYTES),
        Ok(raw) => {
            let trimmed = raw.trim();
            match trimmed.parse::<u64>() {
                Ok(0) => None,
                Ok(n) => Some(n),
                Err(_) => {
                    tracing::warn!(
                        value = %raw,
                        env = SOLO_INGEST_MAX_BYTES_ENV,
                        default_bytes = DEFAULT_INGEST_MAX_BYTES,
                        "unparseable SOLO_INGEST_MAX_BYTES; falling back to default"
                    );
                    Some(DEFAULT_INGEST_MAX_BYTES)
                }
            }
        }
    }
}

/// What `WriteCommand::IngestDocument` returns to the caller. New in
/// v0.7.0 (RAG / document-memory). See `docs/dev-log/0083-v0.7.0-
/// implementation-plan.md` §2 P3.
///
/// `deduped == true` indicates the same content_hash was already present
/// in `documents`; the returned `doc_id` is the pre-existing document's
/// id and `chunks_persisted` is zero (no new chunks were written, no
/// embeddings were called). Forgotten documents still participate in
/// dedup — re-ingesting the same text after `forget_document` returns
/// the forgotten doc_id unchanged (callers can re-activate via a
/// future `restore` command, or simply ingest under a different source
/// path if they want a fresh active doc).
#[derive(Debug, Clone, serde::Serialize)]
pub struct IngestReport {
    pub doc_id: solo_core::DocumentId,
    pub chunks_persisted: u32,
    pub bytes_ingested: u64,
    pub deduped: bool,
}

/// What `WriteCommand::ForgetDocument` returns to the caller. New in
/// v0.7.0.
///
/// `chunks_tombstoned` counts the `document_chunks` rows whose HNSW
/// rowid was tombstoned (so `index.len()` no longer counts them and
/// `detect_drift` stays clean). The chunk rows themselves are NOT
/// deleted from SQL — `documents.status='forgotten'` is the soft-delete
/// marker; chunks survive for forensic value (same pattern as episodes'
/// soft-delete via `episodes.status='forgotten'`).
#[derive(Debug, Clone, serde::Serialize)]
pub struct ForgetDocumentReport {
    pub doc_id: solo_core::DocumentId,
    pub chunks_tombstoned: u32,
}

/// All write operations go through this enum. Each variant carries a
/// oneshot reply channel.
#[derive(Debug)]
pub enum WriteCommand {
    Remember {
        episode: Episode,
        embedding: Embedding,
        reply: oneshot::Sender<Result<MemoryId>>,
    },
    Forget {
        memory_id: MemoryId,
        reason: String,
        reply: oneshot::Sender<Result<()>>,
    },
    /// Ingest a document from `path` into the documents / document_chunks
    /// tables, embedding each chunk via the writer's configured Embedder.
    /// Same outbox-via-`pending_index` discipline as `Remember`: BEGIN
    /// IMMEDIATE → INSERT documents → INSERT document_chunks → INSERT
    /// pending_index (kind='chunk') → COMMIT → hnsw.add per chunk →
    /// DELETE pending_index rows. Content-hash dedup short-circuits
    /// re-ingest of the same normalized text.
    ///
    /// Available only when the writer was spawned with an active embedder
    /// (the `spawn_full_with_embedder*` variants). Other spawn paths get
    /// a clear "not configured" error — same pattern as `Reembed`.
    IngestDocument {
        path: std::path::PathBuf,
        chunk_config: crate::document::ChunkConfig,
        reply: oneshot::Sender<Result<IngestReport>>,
    },
    /// Soft-delete a document: set `documents.status='forgotten'` and
    /// tombstone every chunk's HNSW rowid. Chunks remain in SQL for
    /// forensic value; queries that JOIN through `documents` filter
    /// `status='active'`. Forgotten docs survive content-hash dedup —
    /// re-ingesting the same content returns the forgotten doc_id.
    ForgetDocument {
        doc_id: solo_core::DocumentId,
        reply: oneshot::Sender<Result<ForgetDocumentReport>>,
    },
    Consolidate {
        scope: ConsolidationScope,
        reply: oneshot::Sender<Result<ConsolidationReport>>,
    },
    Reembed {
        scope: ReembedScope,
        reply: oneshot::Sender<Result<ReembedReport>>,
    },
    SaveSnapshot {
        reply: oneshot::Sender<Result<()>>,
    },
    /// Online encrypted backup of the writer's source database to
    /// `dest_path`. The destination is created with PRAGMA key bound to
    /// the same raw key the writer holds, so the backup file restores
    /// under the same passphrase + salt as the source.
    ///
    /// Available only when the writer was spawned with a `KeyMaterial`
    /// (the `spawn_full_with_key_and_optional_steward` variant). Other
    /// spawn paths get a clear "not configured" error.
    Backup {
        dest_path: PathBuf,
        reply: oneshot::Sender<Result<()>>,
    },
    /// Backfill: rewrite historical `triples.subject_id` and
    /// `triples.object_id` values per a caller-supplied alias map.
    /// Each `(from, to)` pair is applied to **both** the subject and
    /// object columns (a name appearing in either position should
    /// normalize identically).
    ///
    /// Opt-in: read-path alias resolution (v0.5.0 P1) already covers
    /// query-time bridging without touching stored rows. This command
    /// is for users who want the underlying data to match the canonical
    /// identity (e.g., when exporting to a system that won't honor
    /// `IdentityConfig.user_aliases`). See `docs/dev-log/0071-v0.5.x-roadmap.md`
    /// Priority 10.
    NormalizeSubjects {
        /// `(from_id, to_id)` pairs — e.g. `[("alex", "user"),
        /// ("bob", "user")]`. Each pair is applied as
        /// `UPDATE triples SET subject_id = to WHERE subject_id = from`
        /// and the symmetric object update.
        aliases: Vec<(String, String)>,
        /// When true, run the UPDATEs inside a transaction, count the
        /// affected rows, then `ROLLBACK` instead of committing. The
        /// returned report's row counts reflect what *would* have been
        /// rewritten.
        dry_run: bool,
        reply: oneshot::Sender<Result<NormalizeReport>>,
    },
}

/// Cheaply cloneable handle. Pass clones to every task that needs to write.
/// Dropping the *last* clone closes the channel and triggers actor shutdown.
#[derive(Clone, Debug)]
pub struct WriteHandle {
    tx: mpsc::Sender<WriteCommand>,
}

impl WriteHandle {
    pub async fn remember(
        &self,
        episode: Episode,
        embedding: Embedding,
    ) -> Result<MemoryId> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::Remember {
                episode,
                embedding,
                reply: reply_tx,
            })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    pub async fn forget(&self, memory_id: MemoryId, reason: String) -> Result<()> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::Forget {
                memory_id,
                reason,
                reply: reply_tx,
            })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    /// Ingest a document at `path` using `chunk_config` for splitting.
    /// See [`WriteCommand::IngestDocument`] for the persistence pipeline.
    pub async fn ingest_document(
        &self,
        path: std::path::PathBuf,
        chunk_config: crate::document::ChunkConfig,
    ) -> Result<IngestReport> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::IngestDocument {
                path,
                chunk_config,
                reply: reply_tx,
            })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    /// Soft-delete a document and tombstone its chunks' HNSW rowids.
    /// See [`WriteCommand::ForgetDocument`] for semantics.
    pub async fn forget_document(
        &self,
        doc_id: solo_core::DocumentId,
    ) -> Result<ForgetDocumentReport> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::ForgetDocument {
                doc_id,
                reply: reply_tx,
            })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    pub async fn consolidate(&self, scope: ConsolidationScope) -> Result<ConsolidationReport> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::Consolidate { scope, reply: reply_tx })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    pub async fn reembed(&self, scope: ReembedScope) -> Result<ReembedReport> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::Reembed { scope, reply: reply_tx })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    /// Run an online encrypted backup of the writer's source database
    /// to `dest_path`. Available only when the writer was spawned with
    /// a `KeyMaterial` (see [`WriterActor::spawn_full_with_key_and_optional_steward`]).
    pub async fn backup(&self, dest_path: PathBuf) -> Result<()> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::Backup {
                dest_path,
                reply: reply_tx,
            })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    pub async fn save_snapshot(&self) -> Result<()> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::SaveSnapshot { reply: reply_tx })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }

    /// Rewrite historical `triples.subject_id` / `triples.object_id`
    /// values for each `(from, to)` pair in `aliases`. See
    /// [`WriteCommand::NormalizeSubjects`] for semantics.
    pub async fn normalize_subjects(
        &self,
        aliases: Vec<(String, String)>,
        dry_run: bool,
    ) -> Result<NormalizeReport> {
        let (reply_tx, reply_rx) = oneshot::channel();
        self.tx
            .send(WriteCommand::NormalizeSubjects {
                aliases,
                dry_run,
                reply: reply_tx,
            })
            .await
            .map_err(|_| Error::storage("writer task gone (channel closed)"))?;
        reply_rx
            .await
            .map_err(|_| Error::storage("writer dropped reply channel"))?
    }
}

/// The writer actor.
pub struct WriterActor {
    conn: Connection,
    hnsw: Arc<dyn VectorIndex + Send + Sync>,
    rx: mpsc::Receiver<WriteCommand>,
    /// Directory for HNSW snapshot save. `None` means `SaveSnapshot` returns
    /// an error (used in unit tests that don't exercise the snapshot path).
    /// The daemon main (commit 1.5) sets this to the data dir.
    snapshot_dir: Option<PathBuf>,
    /// Resolved `embedders.embedder_id` for the active embedder. Set once
    /// at startup (`solo_storage::startup::run` resolves via
    /// `get_or_insert_embedder_id`) and cached for every `INSERT INTO
    /// embeddings` row. `None` means the writer was spawned in a test
    /// context that doesn't exercise the embeddings-table-write path —
    /// `dispatch_remember` falls back to skipping the embeddings INSERT
    /// and just writes pending_index.
    embedder_id: Option<i64>,
    /// The active embedder. Required by `handle_reembed` to regenerate
    /// vectors for episodes whose embeddings were produced by a different
    /// (older) embedder. The regular `Remember` path does NOT call this —
    /// callers pass pre-computed `Embedding` objects in `WriteCommand::
    /// Remember`. `None` here means `solo reembed` is not available on
    /// this writer (unit tests, daemon paths that opt out).
    embedder: Option<Arc<dyn Embedder>>,
    /// Handle to the tokio runtime that constructed this actor. Captured
    /// at `spawn_full_with_embedder` time so the dedicated writer thread
    /// can `block_on` async embedder calls during reembed. `None` matches
    /// `embedder == None`.
    runtime_handle: Option<Handle>,
    /// The Steward (clustering + LLM-driven abstraction). Required by
    /// `handle_consolidate`'s abstraction step (Y.3.3); the cheap
    /// clustering step (Y.2) always runs and uses
    /// `StewardConfig::default()` even when the steward is `None`. So:
    /// - `None` → `consolidate` runs cluster persistence only,
    ///   `abstractions_built` stays 0. This is the default for prod
    ///   today since no real `LlmClient` ships.
    /// - `Some` → after cluster persistence, the actor walks each
    ///   cluster, calls `steward.abstract_cluster`, persists the
    ///   `SemanticAbstraction`. Failures per cluster are logged +
    ///   counted; clusters themselves are already-persisted ground
    ///   truth and never roll back.
    steward: Option<Arc<solo_steward::Steward>>,
    /// Raw SQLCipher key used to open the source connection. Required
    /// by `handle_backup` so it can encrypt the destination connection
    /// with the same key. `None` means the writer was spawned without
    /// key material (test paths, the legacy spawn variants); `WriteCommand::
    /// Backup` returns a clear "not configured" error in that case.
    key: Option<KeyMaterial>,
}

/// What `WriterActor::spawn*` returns. The daemon needs the
/// `std::thread::JoinHandle<()>` so it can wait for the writer's
/// `shutdown()` (`PRAGMA wal_checkpoint(TRUNCATE)`, final HNSW save)
/// to complete after the last `WriteHandle` is dropped. Without this,
/// the OS reaps the writer thread when `main` returns, possibly
/// mid-checkpoint.
///
/// Tests that don't care about clean shutdown timing can simply drop
/// the JoinHandle along with the WriteHandle.
pub struct WriterSpawn {
    pub handle: WriteHandle,
    pub join: std::thread::JoinHandle<()>,
}

impl WriterSpawn {
    /// Drop the WriteHandle (closing the mpsc) and block until the writer
    /// thread finishes its `shutdown()` and exits. No timeout — production
    /// supervisors decide when to force-kill via SIGKILL.
    pub fn shutdown_blocking(self) {
        drop(self.handle);
        if let Err(panic) = self.join.join() {
            tracing::error!(?panic, "solo-writer thread panicked during shutdown");
        }
    }
}

impl WriterActor {
    pub fn spawn(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
    ) -> WriterSpawn {
        Self::spawn_with_capacity(conn, hnsw, DEFAULT_CHANNEL_CAPACITY)
    }

    pub fn spawn_with_capacity(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
        capacity: usize,
    ) -> WriterSpawn {
        Self::spawn_internal(conn, hnsw, capacity, None, None, None, None, None, None)
    }

    /// Spawn with a snapshot directory wired up. The daemon main calls this
    /// path so `WriteCommand::SaveSnapshot` can reach disk.
    pub fn spawn_with_snapshot_dir(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
        snapshot_dir: PathBuf,
    ) -> WriterSpawn {
        Self::spawn_internal(
            conn,
            hnsw,
            DEFAULT_CHANNEL_CAPACITY,
            Some(snapshot_dir),
            None,
            None,
            None,
            None,
            None,
        )
    }

    /// Spawn with both snapshot dir + cached embedder_id. The daemon
    /// main calls this so every `remember` also INSERTs into the
    /// `embeddings` table for durability + future `solo reembed`.
    pub fn spawn_full(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
        snapshot_dir: PathBuf,
        embedder_id: i64,
    ) -> WriterSpawn {
        Self::spawn_internal(
            conn,
            hnsw,
            DEFAULT_CHANNEL_CAPACITY,
            Some(snapshot_dir),
            Some(embedder_id),
            None,
            None,
            None,
            None,
        )
    }

    /// Spawn with snapshot dir + embedder_id + the active embedder. Use
    /// this from any path that may invoke `WriteCommand::Reembed` — i.e.
    /// the `solo reembed` one-shot. Captures `Handle::current()` to bridge
    /// the async embedder API onto the writer's blocking thread.
    ///
    /// **Requires a multi-thread tokio runtime.** Panics if called outside
    /// any runtime context. On a `current_thread` runtime it would NOT
    /// panic, but `handle_reembed`'s `runtime.block_on(embedder.embed(...))`
    /// from the writer thread would deadlock — the runtime's only worker
    /// would be the test's outer thread, already blocked awaiting the
    /// reembed reply. Production callers run inside `#[tokio::main]`
    /// (multi-thread by default); tests use `rt_multi(N)` with N >= 1
    /// worker independent of the test's calling thread.
    pub fn spawn_full_with_embedder(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
        snapshot_dir: PathBuf,
        embedder_id: i64,
        embedder: Arc<dyn Embedder>,
    ) -> WriterSpawn {
        Self::spawn_full_with_embedder_and_optional_steward(
            conn,
            hnsw,
            snapshot_dir,
            embedder_id,
            embedder,
            None,
        )
    }

    /// The full surface: snapshot + embedder + steward. Use this from
    /// the `solo consolidate` one-shot path or from a prod daemon
    /// that ships with a real `LlmClient` configured. The steward's
    /// `Arc<dyn LlmClient>` powers `handle_consolidate`'s abstraction
    /// step (Y.3.3); without a steward, consolidate runs the
    /// clustering pass only.
    pub fn spawn_full_with_embedder_and_optional_steward(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
        snapshot_dir: PathBuf,
        embedder_id: i64,
        embedder: Arc<dyn Embedder>,
        steward: Option<Arc<solo_steward::Steward>>,
    ) -> WriterSpawn {
        let handle = Handle::current();
        Self::spawn_internal(
            conn,
            hnsw,
            DEFAULT_CHANNEL_CAPACITY,
            Some(snapshot_dir),
            Some(embedder_id),
            Some(embedder),
            Some(handle),
            steward,
            None,
        )
    }

    /// Like [`Self::spawn_full_with_embedder_and_optional_steward`] but
    /// also captures `key` so the writer can serve `WriteCommand::Backup`.
    /// The daemon (and one-shot paths that want HTTP-side backup) use
    /// this variant; pure-test spawn paths can use the no-key variant.
    pub fn spawn_full_with_key_and_optional_steward(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
        snapshot_dir: PathBuf,
        embedder_id: i64,
        embedder: Arc<dyn Embedder>,
        steward: Option<Arc<solo_steward::Steward>>,
        key: KeyMaterial,
    ) -> WriterSpawn {
        let handle = Handle::current();
        Self::spawn_internal(
            conn,
            hnsw,
            DEFAULT_CHANNEL_CAPACITY,
            Some(snapshot_dir),
            Some(embedder_id),
            Some(embedder),
            Some(handle),
            steward,
            Some(key),
        )
    }

    fn spawn_internal(
        conn: Connection,
        hnsw: Arc<dyn VectorIndex + Send + Sync>,
        capacity: usize,
        snapshot_dir: Option<PathBuf>,
        embedder_id: Option<i64>,
        embedder: Option<Arc<dyn Embedder>>,
        runtime_handle: Option<Handle>,
        steward: Option<Arc<solo_steward::Steward>>,
        key: Option<KeyMaterial>,
    ) -> WriterSpawn {
        let (tx, rx) = mpsc::channel(capacity);
        let actor = Self {
            conn,
            hnsw,
            rx,
            snapshot_dir,
            embedder_id,
            embedder,
            runtime_handle,
            steward,
            key,
        };
        let join = std::thread::Builder::new()
            .name("solo-writer".into())
            .spawn(move || actor.run())
            .expect("spawn solo-writer thread");
        WriterSpawn {
            handle: WriteHandle { tx },
            join,
        }
    }

    fn run(mut self) {
        while let Some(cmd) = self.rx.blocking_recv() {
            self.dispatch(cmd);
        }
        self.shutdown();
    }

    fn dispatch(&mut self, cmd: WriteCommand) {
        match cmd {
            WriteCommand::Remember {
                episode,
                embedding,
                reply,
            } => self.dispatch_remember(episode, embedding, reply),
            WriteCommand::Forget {
                memory_id,
                reason,
                reply,
            } => {
                let _ = reply.send(self.handle_forget(memory_id, reason));
            }
            WriteCommand::IngestDocument {
                path,
                chunk_config,
                reply,
            } => {
                self.dispatch_ingest_document(path, chunk_config, reply);
            }
            WriteCommand::ForgetDocument { doc_id, reply } => {
                let _ = reply.send(self.handle_forget_document(doc_id));
            }
            WriteCommand::Consolidate { scope, reply } => {
                let _ = reply.send(self.handle_consolidate(scope));
            }
            WriteCommand::Reembed { scope, reply } => {
                let _ = reply.send(self.handle_reembed(scope));
            }
            WriteCommand::SaveSnapshot { reply } => {
                let _ = reply.send(self.handle_save_snapshot());
            }
            WriteCommand::Backup { dest_path, reply } => {
                let _ = reply.send(self.handle_backup(&dest_path));
            }
            WriteCommand::NormalizeSubjects {
                aliases,
                dry_run,
                reply,
            } => {
                let _ = reply.send(self.handle_normalize_subjects(aliases, dry_run));
            }
        }
    }

    fn dispatch_remember(
        &mut self,
        episode: Episode,
        embedding: Embedding,
        reply: oneshot::Sender<Result<MemoryId>>,
    ) {
        let memory_id = episode.memory_id;
        let result = self.handle_remember_durable(episode, embedding);
        let durable_ok = result.is_ok();
        let _ = reply.send(result);

        if durable_ok {
            if let Err(e) = self.conn.execute(
                "DELETE FROM pending_index WHERE memory_id = ?",
                params![memory_id.to_string()],
            ) {
                tracing::warn!(
                    error = %e,
                    %memory_id,
                    "pending_index drain failed; will replay on next startup"
                );
            }
        }
    }

    fn handle_remember_durable(
        &mut self,
        episode: Episode,
        embedding: Embedding,
    ) -> Result<MemoryId> {
        embedding.validate()?;
        let memory_id = episode.memory_id;

        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(|e| Error::storage(format!("BEGIN IMMEDIATE for remember: {e}")))?;

        let now_ms = chrono::Utc::now().timestamp_millis();
        let encoding_ctx = serde_json::to_string(&episode.encoding_context)
            .map_err(|e| Error::storage(format!("serialize encoding_context: {e}")))?;
        let provenance_json = match &episode.provenance {
            Some(p) => Some(
                serde_json::to_string(p)
                    .map_err(|e| Error::storage(format!("serialize provenance: {e}")))?,
            ),
            None => None,
        };
        let tier_str = match episode.tier {
            Tier::Hot => "hot",
            Tier::Warm => "warm",
            Tier::Cold => "cold",
        };

        tx.execute(
            "INSERT INTO episodes (
                memory_id, ts_ms, source_type, source_id, content,
                encoding_context_json, provenance_json, confidence,
                strength, salience, tier, created_at_ms, updated_at_ms
             ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
            params![
                memory_id.to_string(),
                episode.ts_ms,
                episode.source_type,
                episode.source_id,
                episode.content,
                encoding_ctx,
                provenance_json,
                episode.confidence.0,
                episode.strength,
                episode.salience,
                tier_str,
                now_ms,
                now_ms,
            ],
        )
        .map_err(|e| Error::storage(format!("INSERT episode: {e}")))?;

        let rowid = tx.last_insert_rowid();

        // Persist the embedding to the `embeddings` table when
        // we have a cached embedder_id. Without this row, `solo
        // reembed` (post-v0.1) wouldn't know what vector this episode
        // had under the previous model, and HNSW rebuild from SQL
        // (also post-v0.1) couldn't repopulate the graph.
        //
        // Skipped when embedder_id is None — only happens in unit-test
        // setups that didn't run the embedder-registry resolution
        // step. The pending_index INSERT below still happens, so
        // recovery on restart still works for tests that exercise
        // the replay path.
        if let Some(eid) = self.embedder_id {
            let dtype_str = match embedding.dtype {
                solo_core::EmbeddingDtype::F32 => "f32",
                solo_core::EmbeddingDtype::F16 => "f16",
                solo_core::EmbeddingDtype::I8 => "i8",
                solo_core::EmbeddingDtype::Binary => "binary",
            };
            tx.execute(
                "INSERT INTO embeddings (
                    memory_id, embedder_id, dtype, dim, vector, created_at_ms
                 ) VALUES (?, ?, ?, ?, ?, ?)",
                params![
                    memory_id.to_string(),
                    eid,
                    dtype_str,
                    embedding.dim as i64,
                    &embedding.data[..],
                    now_ms,
                ],
            )
            .map_err(|e| Error::storage(format!("INSERT embeddings: {e}")))?;
        }

        tx.execute(
            "INSERT INTO pending_index (
                memory_id, embedding, embedding_dim, enqueued_at
             ) VALUES (?, ?, ?, ?)",
            params![
                memory_id.to_string(),
                &embedding.data[..],
                embedding.dim as i64,
                now_ms,
            ],
        )
        .map_err(|e| Error::storage(format!("INSERT pending_index: {e}")))?;

        tx.commit()
            .map_err(|e| Error::storage(format!("COMMIT remember: {e}")))?;

        let f32_slice = embedding.as_f32_slice().ok_or_else(|| {
            Error::embedder("HNSW expects F32 embeddings; convert dtype upstream")
        })?;
        // Encode the rowid in the shared HNSW namespace (high bit clear
        // for episodes). See `crate::hnsw_id` for the encoding rationale.
        self.hnsw.add(episode_hnsw_id(rowid), f32_slice)?;

        Ok(memory_id)
    }

    fn handle_forget(&mut self, memory_id: MemoryId, reason: String) -> Result<()> {
        // Soft-delete: set status='forgotten' on the episode. Per ADR-0003
        // Forget reply timing semantics, the HNSW vector is NOT removed
        // from the underlying graph — recall paths exclude
        // `status='forgotten'` rows by SQL filter, and the architecture
        // preserves silent traces for forensics + consolidation.
        //
        // BUT: we DO mark the rowid in the in-memory tombstone set
        // (`HnswIndex::tombstones`). Without this, `index.len()` keeps
        // counting the forgotten vector and `detect_drift` spuriously
        // warns at runtime. Post-restart `rebuild_tombstones_from_sql`
        // does the same thing; runtime tombstoning matches that behaviour.
        //
        // The `reason` parameter is logged but not persisted. A future
        // schema (memory_revisions or forget_log) can record it; v0.1
        // surfaces it through tracing only.
        let now_ms = chrono::Utc::now().timestamp_millis();
        let id_str = memory_id.to_string();

        // Look up the rowid first so we can tombstone the HNSW even on
        // the already-forgotten / not-found paths. The query is cheap
        // (UNIQUE index on memory_id).
        let rowid: Option<i64> = self
            .conn
            .query_row(
                "SELECT rowid FROM episodes WHERE memory_id = ?",
                params![&id_str],
                |r| r.get::<_, i64>(0),
            )
            .optional()
            .map_err(|e| Error::storage(format!("lookup rowid for forget: {e}")))?;
        let Some(rowid) = rowid else {
            return Err(Error::not_found(format!(
                "memory_id {memory_id} not found in episodes"
            )));
        };

        let updated = self
            .conn
            .execute(
                "UPDATE episodes
                    SET status = 'forgotten', updated_at_ms = ?
                  WHERE memory_id = ? AND status <> 'forgotten'",
                params![now_ms, &id_str],
            )
            .map_err(|e| Error::storage(format!("UPDATE episodes for forget: {e}")))?;

        // Tombstone the HNSW. Idempotent for already-tombstoned rowids
        // (HashSet::insert just no-ops). The encoded id MUST match the
        // one passed at insert time (see `dispatch_remember`).
        if let Err(e) = self.hnsw.remove(episode_hnsw_id(rowid)) {
            tracing::warn!(
                error = %e,
                %memory_id,
                "hnsw.remove during forget failed (non-fatal; SQL filter still hides the row)"
            );
        }

        if updated == 0 {
            // Already forgotten → idempotent success.
            tracing::debug!(%memory_id, "forget called on already-forgotten episode (idempotent)");
            return Ok(());
        }
        tracing::info!(%memory_id, %reason, "episode soft-deleted (status=forgotten)");
        Ok(())
    }

    // --------------------------------------------------------------------
    // v0.7.0 — document ingest + forget
    //
    // Same outbox discipline as `dispatch_remember`: the SQL transaction
    // commits BEFORE we touch HNSW, the reply goes BEFORE we drain the
    // pending_index outbox. If hnsw.add fails the row stays in
    // pending_index and replay picks it up next startup; if the DELETE
    // drain fails the row stays and replay is idempotent.
    //
    // See `docs/dev-log/0083-v0.7.0-implementation-plan.md` §2 P3.
    // --------------------------------------------------------------------

    /// Mirror of `dispatch_remember`: send the reply (durable + HNSW-
    /// resident report) BEFORE draining the pending_index outbox rows.
    /// Drain failure logs + leaves the rows for next-startup replay.
    fn dispatch_ingest_document(
        &mut self,
        path: std::path::PathBuf,
        chunk_config: crate::document::ChunkConfig,
        reply: oneshot::Sender<Result<IngestReport>>,
    ) {
        // Capture the chunk_ids inserted so we can drain only those rows
        // (NOT a blanket `DELETE FROM pending_index WHERE kind='chunk'`,
        // which would clobber concurrent ingests' in-flight rows).
        let (result, drained_chunks) = self.handle_ingest_document_durable(path, chunk_config);
        let durable_ok = result.is_ok();
        let _ = reply.send(result);

        if durable_ok && !drained_chunks.is_empty() {
            for chunk_id in &drained_chunks {
                if let Err(e) = self.conn.execute(
                    "DELETE FROM pending_index WHERE kind = 'chunk' AND chunk_id = ?",
                    params![chunk_id.to_string()],
                ) {
                    tracing::warn!(
                        error = %e,
                        %chunk_id,
                        "pending_index drain (chunk) failed; will replay on next startup"
                    );
                }
            }
        }
    }

    /// The body of `IngestDocument`. Returns the report plus the list of
    /// chunk_ids whose pending_index rows want draining (empty on the
    /// dedup short-circuit and on any error).
    ///
    /// Pipeline:
    ///   1. Parse the file (off-tx).
    ///   2. Chunk the text (off-tx).
    ///   3. Compute content_hash; check `documents.content_hash` for dedup.
    ///   4. Embed all chunks via `embedder.embed_batch` (off-tx).
    ///   5. BEGIN IMMEDIATE.
    ///   6. INSERT documents row.
    ///   7. For each chunk: INSERT document_chunks → INSERT chunk_embeddings
    ///      → INSERT pending_index (kind='chunk').
    ///   8. COMMIT.
    ///   9. hnsw.add(chunk_rowid, embedding) for each chunk.
    ///  10. Caller drains pending_index rows.
    fn handle_ingest_document_durable(
        &mut self,
        path: std::path::PathBuf,
        chunk_config: crate::document::ChunkConfig,
    ) -> (Result<IngestReport>, Vec<solo_core::ChunkId>) {
        // -------- Step 0: SOLO_INGEST_MAX_BYTES guardrail --------
        //
        // Read the file's size on disk and reject before paying any parse /
        // chunk / embed cost when it exceeds the configured cap. The cap is
        // a per-file precheck so the writer never holds an oversized
        // document in RAM. `SOLO_INGEST_MAX_BYTES=0` disables the cap
        // entirely; an unparseable env var falls back to the default with
        // a single WARN line.
        //
        // We deliberately consult `std::fs::metadata(path).len()` rather
        // than reading the file twice — `parse_file` will read it again
        // immediately after, and the OS page cache makes the second read
        // free. Using `metadata().len()` for the precheck also means we
        // can reject a multi-GB PDF without ever asking `pdf-extract` to
        // allocate.
        if let Some(cap) = resolve_ingest_max_bytes() {
            match std::fs::metadata(&path) {
                Ok(meta) => {
                    let size = meta.len();
                    if size > cap {
                        return (
                            Err(Error::storage(format!(
                                "ingest_document: file {} is {size} bytes, exceeds \
                                 SOLO_INGEST_MAX_BYTES cap of {cap} bytes. Set \
                                 SOLO_INGEST_MAX_BYTES=<larger> to override, or \
                                 SOLO_INGEST_MAX_BYTES=0 to disable the cap.",
                                path.display()
                            ))),
                            Vec::new(),
                        );
                    }
                }
                Err(e) => {
                    return (
                        Err(Error::storage(format!(
                            "ingest_document: stat {}: {e}",
                            path.display()
                        ))),
                        Vec::new(),
                    );
                }
            }
        }

        // -------- Steps 1 + 2: parse + chunk (pure, off-tx) --------
        let parsed = match crate::document::parse_file(&path) {
            Ok(p) => p,
            Err(e) => {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: parse {}: {e}",
                        path.display()
                    ))),
                    Vec::new(),
                );
            }
        };
        let chunks = crate::document::chunk_text(&parsed.text, &chunk_config);

        // -------- Step 3: content_hash + dedup pre-check --------
        let content_hash = {
            use sha2::{Digest, Sha256};
            let mut hasher = Sha256::new();
            hasher.update(parsed.text.as_bytes());
            hex::encode(hasher.finalize())
        };

        let existing_doc: Option<String> = match self
            .conn
            .query_row(
                "SELECT doc_id FROM documents WHERE content_hash = ? LIMIT 1",
                params![&content_hash],
                |r| r.get::<_, String>(0),
            )
            .optional()
        {
            Ok(v) => v,
            Err(e) => {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: dedup lookup: {e}"
                    ))),
                    Vec::new(),
                );
            }
        };
        if let Some(doc_id_s) = existing_doc {
            let doc_id = match solo_core::DocumentId::from_str(&doc_id_s) {
                Ok(id) => id,
                Err(e) => {
                    return (
                        Err(Error::storage(format!(
                            "ingest_document: parse existing doc_id `{doc_id_s}`: {e}"
                        ))),
                        Vec::new(),
                    );
                }
            };
            tracing::info!(
                %doc_id,
                content_hash = %content_hash,
                "ingest_document: dedup hit; returning existing doc_id"
            );
            return (
                Ok(IngestReport {
                    doc_id,
                    chunks_persisted: 0,
                    bytes_ingested: parsed.byte_size,
                    deduped: true,
                }),
                Vec::new(),
            );
        }

        // Need an embedder + runtime to embed chunks. Pure dedup hit above
        // doesn't need either — every other path does.
        let embedder = match self.embedder.clone() {
            Some(e) => e,
            None => {
                return (
                    Err(Error::Other(
                        "ingest_document: writer has no embedder \
                         (use spawn_full_with_embedder)"
                            .into(),
                    )),
                    Vec::new(),
                );
            }
        };
        let runtime = match self.runtime_handle.clone() {
            Some(r) => r,
            None => {
                return (
                    Err(Error::Other(
                        "ingest_document: writer has no runtime handle \
                         (use spawn_full_with_embedder)"
                            .into(),
                    )),
                    Vec::new(),
                );
            }
        };
        let embedder_id = match self.embedder_id {
            Some(id) => id,
            None => {
                return (
                    Err(Error::Other(
                        "ingest_document: writer has no embedder_id \
                         (use spawn_full_with_embedder)"
                            .into(),
                    )),
                    Vec::new(),
                );
            }
        };

        // Empty document (e.g. only-whitespace would have returned
        // ParseError::Empty already, but defensive): nothing to embed.
        if chunks.is_empty() {
            return (
                Err(Error::storage(
                    "ingest_document: parser returned text but chunker produced 0 chunks",
                )),
                Vec::new(),
            );
        }

        // -------- Step 4: embed batch BEFORE the transaction --------
        // Embed-before-tx is the same risk-of-stale-tx pattern called out in
        // ADR-0003 §"Reembed batch ordering" — if the embedder fails or
        // hangs, no SQL state has changed and the writer is still free to
        // serve other commands.
        let texts: Vec<&str> = chunks.iter().map(|c| c.content.as_str()).collect();
        let embeddings = match runtime.block_on(embedder.embed_batch(&texts)) {
            Ok(v) => v,
            Err(e) => {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: embed_batch failed: {e}"
                    ))),
                    Vec::new(),
                );
            }
        };
        if embeddings.len() != chunks.len() {
            return (
                Err(Error::storage(format!(
                    "ingest_document: embed_batch returned {} embeddings for {} chunks",
                    embeddings.len(),
                    chunks.len()
                ))),
                Vec::new(),
            );
        }
        // Validate every embedding's (dtype, dim, data.len()) before SQL.
        for (i, emb) in embeddings.iter().enumerate() {
            if let Err(e) = emb.validate() {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: chunk {i} embedding invalid: {e}"
                    ))),
                    Vec::new(),
                );
            }
        }

        // Cache the dtype string for INSERT INTO chunk_embeddings.
        let dtype_str = match embedder.dtype() {
            solo_core::EmbeddingDtype::F32 => "f32",
            solo_core::EmbeddingDtype::F16 => "f16",
            solo_core::EmbeddingDtype::I8 => "i8",
            solo_core::EmbeddingDtype::Binary => "binary",
        };

        // -------- Step 5: allocate ids + open transaction --------
        let doc_id = solo_core::DocumentId::new();
        let now_ms = chrono::Utc::now().timestamp_millis();
        let modified_at_ms: Option<i64> = std::fs::metadata(&path)
            .ok()
            .and_then(|m| m.modified().ok())
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_millis() as i64);

        // Title heuristic: first markdown-style heading (`# ...`) on the
        // first 64 lines, else the file stem. Filename is more useful than
        // "(untitled)" in the typical case.
        let title: String = derive_document_title(&parsed.text, &path);
        let source: String = path.to_string_lossy().to_string();

        let chunk_count = chunks.len() as u32;

        // Collect chunk_ids + their assigned rowids for the post-COMMIT
        // hnsw.add + drain. We have to pull rowid out of `last_insert_rowid`
        // inside the tx; collect the (chunk_id, rowid, embedding) tuples
        // for use AFTER the commit.
        let mut chunk_records: Vec<(solo_core::ChunkId, i64, solo_core::Embedding)> =
            Vec::with_capacity(chunks.len());

        let tx = match self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)
        {
            Ok(t) => t,
            Err(e) => {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: BEGIN IMMEDIATE: {e}"
                    ))),
                    Vec::new(),
                );
            }
        };

        // -------- Step 6: INSERT documents row --------
        if let Err(e) = tx.execute(
            "INSERT INTO documents (
                doc_id, source, title, mime_type,
                ingested_at_ms, modified_at_ms, status,
                chunk_count, content_hash, byte_size
             ) VALUES (?, ?, ?, ?, ?, ?, 'active', ?, ?, ?)",
            params![
                doc_id.to_string(),
                source,
                title,
                parsed.mime_type,
                now_ms,
                modified_at_ms,
                chunk_count as i64,
                content_hash,
                parsed.byte_size as i64,
            ],
        ) {
            return (
                Err(Error::storage(format!(
                    "ingest_document: INSERT documents: {e}"
                ))),
                Vec::new(),
            );
        }

        // -------- Step 7: INSERT chunk + embedding + pending_index rows --------
        for (idx, (spec, embedding)) in chunks.iter().zip(embeddings.iter()).enumerate() {
            let chunk_id = solo_core::ChunkId::new();
            if let Err(e) = tx.execute(
                "INSERT INTO document_chunks (
                    chunk_id, doc_id, chunk_index, content,
                    token_count, start_offset, end_offset, created_at_ms
                 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                params![
                    chunk_id.to_string(),
                    doc_id.to_string(),
                    idx as i64,
                    spec.content,
                    spec.token_count as i64,
                    spec.start_offset as i64,
                    spec.end_offset as i64,
                    now_ms,
                ],
            ) {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: INSERT document_chunks (idx {idx}): {e}"
                    ))),
                    Vec::new(),
                );
            }
            let rowid = tx.last_insert_rowid();

            if let Err(e) = tx.execute(
                "INSERT INTO chunk_embeddings (
                    chunk_id, embedder_id, dtype, dim, vector, created_at_ms
                 ) VALUES (?, ?, ?, ?, ?, ?)",
                params![
                    chunk_id.to_string(),
                    embedder_id,
                    dtype_str,
                    embedding.dim as i64,
                    &embedding.data[..],
                    now_ms,
                ],
            ) {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: INSERT chunk_embeddings (idx {idx}): {e}"
                    ))),
                    Vec::new(),
                );
            }

            if let Err(e) = tx.execute(
                "INSERT INTO pending_index (
                    kind, chunk_id, embedding, embedding_dim, enqueued_at
                 ) VALUES ('chunk', ?, ?, ?, ?)",
                params![
                    chunk_id.to_string(),
                    &embedding.data[..],
                    embedding.dim as i64,
                    now_ms,
                ],
            ) {
                return (
                    Err(Error::storage(format!(
                        "ingest_document: INSERT pending_index (idx {idx}): {e}"
                    ))),
                    Vec::new(),
                );
            }

            chunk_records.push((chunk_id, rowid, embedding.clone()));
        }

        // -------- Step 8: COMMIT --------
        if let Err(e) = tx.commit() {
            return (
                Err(Error::storage(format!(
                    "ingest_document: COMMIT: {e}"
                ))),
                Vec::new(),
            );
        }

        // -------- Step 9: hnsw.add per chunk --------
        // Failure here is non-fatal — the row in pending_index will replay
        // on next startup. We log and continue so partial-success-with-
        // recoverable-replay matches the steady-state ordering of
        // dispatch_remember.
        let mut drained: Vec<solo_core::ChunkId> = Vec::with_capacity(chunk_records.len());
        for (chunk_id, rowid, embedding) in &chunk_records {
            let f32_slice = match embedding.as_f32_slice() {
                Some(s) => s,
                None => {
                    tracing::warn!(
                        %chunk_id,
                        "ingest_document: chunk embedding is not F32; HNSW add skipped \
                         (pending_index row will be replayed)"
                    );
                    continue;
                }
            };
            // Chunks encode their rowid with the chunk bit set so they
            // share the HNSW namespace with episodes without collision.
            // See `crate::hnsw_id` for the encoding rationale.
            match self.hnsw.add(chunk_hnsw_id(*rowid), f32_slice) {
                Ok(_) => drained.push(*chunk_id),
                Err(e) => {
                    tracing::warn!(
                        %chunk_id,
                        error = %e,
                        "ingest_document: hnsw.add failed; pending_index row left for replay"
                    );
                }
            }
        }

        tracing::info!(
            %doc_id,
            chunks = chunk_records.len(),
            indexed = drained.len(),
            bytes = parsed.byte_size,
            "ingest_document complete"
        );

        (
            Ok(IngestReport {
                doc_id,
                chunks_persisted: chunk_count,
                bytes_ingested: parsed.byte_size,
                deduped: false,
            }),
            drained,
        )
    }

    /// Soft-delete a document. See [`WriteCommand::ForgetDocument`]:
    ///
    ///   1. UPDATE documents SET status='forgotten' WHERE doc_id = ?
    ///      → 0 rows == Err::NotFound
    ///   2. SELECT every chunk's rowid for this document.
    ///   3. COMMIT (single-statement UPDATE is effectively atomic).
    ///   4. hnsw.remove(rowid) for each chunk — tombstones the in-memory
    ///      bitmap so `index.len()` accounting stays clean. Failures here
    ///      are logged + skipped; the SQL `status='forgotten'` filter is
    ///      the source of truth for queries anyway.
    ///
    /// Idempotent: re-forgetting an already-forgotten doc UPDATEs zero
    /// rows but still returns Ok (chunks_tombstoned reflects the chunks
    /// found by the SELECT; on second call those are the same chunks
    /// whose rowids are already tombstoned — `hnsw.remove` is a HashSet
    /// insert so no-ops on re-entry). Returning NotFound only when the
    /// doc_id was never in the table.
    fn handle_forget_document(
        &mut self,
        doc_id: solo_core::DocumentId,
    ) -> Result<ForgetDocumentReport> {
        let id_str = doc_id.to_string();

        // Confirm the doc exists; this is also the NotFound check.
        let exists: Option<String> = self
            .conn
            .query_row(
                "SELECT status FROM documents WHERE doc_id = ?",
                params![&id_str],
                |r| r.get::<_, String>(0),
            )
            .optional()
            .map_err(|e| {
                Error::storage(format!("forget_document: lookup status: {e}"))
            })?;
        let Some(prior_status) = exists else {
            return Err(Error::not_found(format!(
                "doc_id {doc_id} not found in documents"
            )));
        };

        // UPDATE to forgotten. Idempotent — UPDATEs zero rows when already
        // forgotten but we still proceed to tombstone the HNSW.
        self.conn
            .execute(
                "UPDATE documents
                    SET status = 'forgotten'
                  WHERE doc_id = ? AND status <> 'forgotten'",
                params![&id_str],
            )
            .map_err(|e| {
                Error::storage(format!("forget_document: UPDATE status: {e}"))
            })?;

        // Collect chunk rowids for HNSW tombstone.
        let mut stmt = self
            .conn
            .prepare("SELECT rowid FROM document_chunks WHERE doc_id = ?")
            .map_err(|e| {
                Error::storage(format!(
                    "forget_document: prepare chunk-rowid select: {e}"
                ))
            })?;
        let rowids: Vec<i64> = stmt
            .query_map(params![&id_str], |r| r.get::<_, i64>(0))
            .map_err(|e| {
                Error::storage(format!(
                    "forget_document: query chunk rowids: {e}"
                ))
            })?
            .filter_map(|r| r.ok())
            .collect();

        let chunks_tombstoned = rowids.len() as u32;
        for rowid in rowids {
            // The encoded id MUST match the one passed at ingest time
            // (see `dispatch_ingest_document`).
            if let Err(e) = self.hnsw.remove(chunk_hnsw_id(rowid)) {
                tracing::warn!(
                    rowid,
                    %doc_id,
                    error = %e,
                    "forget_document: hnsw.remove failed (non-fatal; SQL filter still hides chunk)"
                );
            }
        }

        if prior_status == "forgotten" {
            tracing::debug!(
                %doc_id,
                "forget_document called on already-forgotten doc (idempotent)"
            );
        } else {
            tracing::info!(
                %doc_id,
                chunks_tombstoned,
                "document soft-deleted (status=forgotten)"
            );
        }
        Ok(ForgetDocumentReport {
            doc_id,
            chunks_tombstoned,
        })
    }

    fn handle_consolidate(
        &mut self,
        scope: ConsolidationScope,
    ) -> Result<ConsolidationReport> {
        // v0.2.0 implements only the SWS-equivalent clustering pass.
        // Abstraction + contradiction-detection (Y.3+) require the
        // LLM client; both fields stay 0 in the report.
        //
        // Discipline mirrors `handle_reembed`: validate state, build
        // candidates from SQL, run the pure-deterministic algorithm,
        // persist the output in one transaction. Mid-run failures
        // bubble up — there's nothing to "skip and continue" inside
        // the storage step (we either commit all clusters or none).
        let current_id = self.embedder_id.ok_or_else(|| {
            Error::Other(
                "consolidate: writer has no current embedder_id (use spawn_full_with_embedder)"
                    .into(),
            )
        })?;

        // Optional time window. Computed BEFORE the SELECT so a slow
        // `now_ms` clock doesn't drift candidate selection.
        let now_ms = chrono::Utc::now().timestamp_millis();
        let cutoff_ms: Option<i64> = scope.window_days.and_then(|days| {
            const MS_PER_DAY: i64 = 86_400_000;
            days.checked_mul(MS_PER_DAY).map(|w| now_ms - w)
        });

        // Build candidates: (Episode, Embedding) tuples for active+hot
        // rows whose embedding row matches the current embedder. The
        // shape mirrors what `solo_steward::cluster::cluster_episodes`
        // expects, but we only populate the fields the algorithm reads
        // (memory_id, ts_ms) plus those that round-trip cheaply via SQL
        // (content). The defaulted fields (provenance, encoding_context,
        // confidence/strength/salience, source_id) are not used by the
        // SWS pass; Y.3's REM pass will need them and we'll widen this
        // SELECT then.
        // Idempotency: exclude memories that are already part of a
        // cluster. Without this, re-running consolidate on the same
        // data set creates duplicate clusters with different UUID v7
        // cluster_ids — same shape but two rows per memory in
        // `cluster_episodes`, plus wasted LLM `abstract_cluster` calls
        // in Y.3. Trade-off: a memory can never be re-clustered once
        // it's been placed; cluster-merging across consolidation
        // windows is a v0.3 feature ("re-consolidation"). For v0.2.0,
        // first-write wins.
        let candidates: Vec<(Episode, Embedding)> = {
            let (sql, params): (&str, Vec<rusqlite::types::Value>) = match cutoff_ms {
                Some(cutoff) => (
                    "SELECT e.memory_id, e.ts_ms, e.source_type, e.content,
                            e.confidence, e.strength, e.salience,
                            em.dtype, em.dim, em.vector
                     FROM episodes e
                     JOIN embeddings em ON em.memory_id = e.memory_id
                     WHERE em.embedder_id = ?1
                       AND e.status = 'active'
                       AND e.tier = 'hot'
                       AND e.ts_ms >= ?2
                       AND e.memory_id NOT IN (SELECT memory_id FROM cluster_episodes)
                     ORDER BY e.ts_ms, e.rowid",
                    vec![current_id.into(), cutoff.into()],
                ),
                None => (
                    "SELECT e.memory_id, e.ts_ms, e.source_type, e.content,
                            e.confidence, e.strength, e.salience,
                            em.dtype, em.dim, em.vector
                     FROM episodes e
                     JOIN embeddings em ON em.memory_id = e.memory_id
                     WHERE em.embedder_id = ?1
                       AND e.status = 'active'
                       AND e.tier = 'hot'
                       AND e.memory_id NOT IN (SELECT memory_id FROM cluster_episodes)
                     ORDER BY e.ts_ms, e.rowid",
                    vec![current_id.into()],
                ),
            };

            let mut stmt = self
                .conn
                .prepare(sql)
                .map_err(|e| Error::storage(format!("prepare consolidate select: {e}")))?;
            let rows = stmt
                .query_map(params_from_iter(&params), |r| {
                    let memory_id: String = r.get(0)?;
                    let ts_ms: i64 = r.get(1)?;
                    let source_type: String = r.get(2)?;
                    let content: String = r.get(3)?;
                    let confidence_f: f32 = r.get(4)?;
                    let strength: f32 = r.get(5)?;
                    let salience: f32 = r.get(6)?;
                    let dtype_str: String = r.get(7)?;
                    let dim: i64 = r.get(8)?;
                    let vector: Vec<u8> = r.get(9)?;
                    Ok((
                        memory_id,
                        ts_ms,
                        source_type,
                        content,
                        confidence_f,
                        strength,
                        salience,
                        dtype_str,
                        dim,
                        vector,
                    ))
                })
                .map_err(|e| Error::storage(format!("query_map consolidate: {e}")))?;

            let mut out = Vec::new();
            for row in rows {
                let (memory_id, ts_ms, source_type, content, conf, strength, salience, dtype_str, dim, vector) =
                    row.map_err(|e| {
                        Error::storage(format!("consolidate row decode: {e}"))
                    })?;
                let mid = MemoryId::from_str(&memory_id).map_err(|e| {
                    Error::storage(format!("parse memory_id `{memory_id}`: {e}"))
                })?;
                let dtype = match dtype_str.as_str() {
                    "f32" => solo_core::EmbeddingDtype::F32,
                    "f16" => solo_core::EmbeddingDtype::F16,
                    "i8" => solo_core::EmbeddingDtype::I8,
                    "binary" => solo_core::EmbeddingDtype::Binary,
                    other => {
                        return Err(Error::storage(format!(
                            "unknown embeddings.dtype value `{other}`"
                        )));
                    }
                };
                let embedding = Embedding {
                    dtype,
                    dim: dim as usize,
                    data: vector,
                };
                let confidence = solo_core::Confidence::new(conf).map_err(|e| {
                    Error::storage(format!("invalid confidence in episodes row: {e}"))
                })?;
                let episode = Episode {
                    memory_id: mid,
                    ts_ms,
                    source_type,
                    source_id: None, // not selected; Y.3 will widen
                    content,
                    encoding_context: solo_core::EncodingContext::default(),
                    provenance: None,
                    confidence,
                    strength,
                    salience,
                    tier: Tier::Hot,
                };
                out.push((episode, embedding));
            }
            out
        };

        let mut report = ConsolidationReport {
            episodes_seen: candidates.len(),
            ..ConsolidationReport::default()
        };

        if candidates.is_empty() && !scope.force_merge {
            tracing::info!(seen = 0, "consolidate: no candidates");
            return Ok(report);
        }
        if candidates.is_empty() {
            tracing::info!(
                seen = 0,
                "consolidate: no candidates, but force_merge set; falling through to merge + regen"
            );
        }

        // Run the pure-deterministic clustering. Threshold + min-size
        // come from `StewardConfig::from_env()` so the operator-facing
        // `SOLO_CLUSTER_COSINE_THRESHOLD` takes effect uniformly across:
        //   - this clustering pass (new candidates),
        //   - the in-run `merge_clusters_by_centroid` call below,
        //   - the existing-vs-existing `plan_existing_merges` further
        //     down — all three read this same `config`.
        //
        // Daemon and `solo consolidate` validate the env var at startup
        // (their `from_env()?` fails fast on a malformed value); reading
        // again here is defensive — env vars typically don't change
        // mid-process, but the cost is essentially zero and it keeps
        // the writer free of a separate `cluster_config` field that
        // would touch every `spawn_full_with_*` caller. The
        // `count_existing_merge_candidates` doctor path takes the same
        // resolved config as a parameter — see `merge_candidates`'s
        // "Sync requirement" docstring for the SQL+threshold pair.
        let config = solo_steward::StewardConfig::from_env()?;
        let mut clusters = solo_steward::cluster::cluster_episodes(&candidates, &config)?;

        // Re-consolidation pass: fold clusters whose centroids are
        // above threshold into a single survivor. v0.3 MVP scope is
        // **just-built only** — the merge sees only the clusters
        // produced by this run, not pre-existing ones in the DB.
        // That closes the cross-UTC-day-bucket case (conversations
        // straddling midnight produce two same-themed clusters in
        // the per-day bucketing). Cross-run merge requires fetching
        // existing clusters + abstraction-regeneration plumbing —
        // separate iteration.
        let absorbed =
            solo_steward::cluster::merge_clusters_by_centroid(&mut clusters, &config)?;
        report.clusters_merged = absorbed;
        if absorbed > 0 {
            tracing::info!(
                absorbed,
                survivors = clusters.len(),
                "consolidate: centroid merge collapsed cross-bucket clusters"
            );
        }

        report.clusters_built = clusters.len();
        report.episodes_clustered = clusters.iter().map(|c| c.episode_ids.len()).sum();

        if clusters.is_empty() {
            tracing::info!(
                seen = report.episodes_seen,
                "consolidate: no new clusters formed; falling through to merge + regen"
            );
            // No early-return: the existing-vs-existing merge pass
            // and the regen pass downstream operate on pre-existing
            // DB clusters and can fire even when this run produced
            // no fresh clusters (e.g. drift catch-up). The absorb
            // pass naturally no-ops on an empty `clusters` Vec.
        }

        // -------- Cross-run absorb pass --------
        //
        // For each freshly-built cluster, decide whether it should
        // fold into a pre-existing DB cluster with a similar
        // centroid (within `cluster_cosine_threshold`). If so:
        //   - the new cluster gets NO row in `clusters`;
        //   - its `cluster_episodes` rows link under the existing
        //     cluster's id;
        //   - the existing cluster's centroid + coherence refresh
        //     to the post-absorb weighted mean;
        //   - the existing cluster's `semantic_abstractions` row
        //     is INTENTIONALLY left in place — abstraction
        //     regeneration on absorb is a separate iteration
        //     (needs a triples → cluster FK or status-flag to
        //     avoid duplicate triple_id rows when re-extracting).
        //
        // The expected_dim for the existing-cluster fetch comes from
        // the first candidate's embedding (every candidate uses the
        // current embedder's dim, enforced by the SELECT above).
        // expected_dim is normally read off the first candidate's
        // embedding; with `force_merge` and an empty candidate set
        // there's no candidate to read from, so fall back to the
        // current embedder's row in the `embedders` table.
        let expected_dim = if let Some(c) = candidates.first() {
            c.1.dim
        } else {
            self.conn
                .query_row(
                    "SELECT dim FROM embedders WHERE embedder_id = ?",
                    params![current_id],
                    |r| r.get::<_, i64>(0),
                )
                .map(|d| d as usize)
                .map_err(|e| {
                    Error::storage(format!(
                        "consolidate force_merge: lookup dim for embedder_id {current_id}: {e}"
                    ))
                })?
        };
        let existing_summaries =
            self.fetch_existing_cluster_summaries(cutoff_ms, expected_dim)?;
        let absorb_plan = if existing_summaries.is_empty() {
            solo_steward::cluster::AbsorbPlan::default()
        } else {
            solo_steward::cluster::absorb_into_existing(
                &clusters,
                &existing_summaries,
                &config,
            )?
        };
        report.clusters_absorbed = absorb_plan.absorptions.len();
        if !absorb_plan.absorptions.is_empty() {
            tracing::info!(
                absorbed = absorb_plan.absorptions.len(),
                existing_modified = absorb_plan.modified_existing_ids().len(),
                "consolidate: cross-run absorb folded clusters into existing"
            );
        }

        // Build a quick map: new_cluster_id → AbsorbedCluster, for
        // O(1) lookup during the persistence loop.
        let absorbed_by_new: std::collections::HashMap<
            MemoryId,
            &solo_steward::cluster::AbsorbedCluster,
        > = absorb_plan
            .absorptions
            .iter()
            .map(|a| (a.new_cluster_id, a))
            .collect();

        // `clusters_built` should reflect the count of brand-new
        // clusters that actually got persisted (non-absorbed). Update
        // now that we know the absorb count.
        report.clusters_built = clusters.len() - absorb_plan.absorptions.len();

        // Persist all clusters in ONE transaction. If any insert fails,
        // rollback — partial state would leave dangling cluster_episodes
        // referencing nonexistent clusters.
        let txn = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(|e| Error::storage(format!("BEGIN consolidate: {e}")))?;

        for cluster in &clusters {
            // Cross-run absorb: if this freshly-built cluster was
            // matched to an existing DB cluster, link its episodes
            // under that existing cluster_id and skip the INSERT
            // into `clusters`. The existing cluster's centroid +
            // coherence refresh happens after this loop in a single
            // batched UPDATE step.
            if let Some(absorbed) = absorbed_by_new.get(&cluster.cluster_id) {
                let target_id_s = absorbed.existing_cluster_id.to_string();
                for memid in &cluster.episode_ids {
                    txn.execute(
                        "INSERT INTO cluster_episodes (cluster_id, memory_id) VALUES (?, ?)",
                        params![target_id_s, memid.to_string()],
                    )
                    .map_err(|e| {
                        Error::storage(format!("INSERT cluster_episodes (absorbed): {e}"))
                    })?;
                }
                continue;
            }

            let centroid_dtype: Option<&'static str> = cluster.centroid.as_ref().map(|e| {
                match e.dtype {
                    solo_core::EmbeddingDtype::F32 => "f32",
                    solo_core::EmbeddingDtype::F16 => "f16",
                    solo_core::EmbeddingDtype::I8 => "i8",
                    solo_core::EmbeddingDtype::Binary => "binary",
                }
            });
            let centroid_dim: Option<i64> =
                cluster.centroid.as_ref().map(|e| e.dim as i64);
            let centroid_blob: Option<&[u8]> =
                cluster.centroid.as_ref().map(|e| e.data.as_slice());

            txn.execute(
                "INSERT INTO clusters (cluster_id, centroid, centroid_dtype, centroid_dim, coherence, created_at_ms)
                 VALUES (?, ?, ?, ?, ?, ?)",
                params![
                    cluster.cluster_id.to_string(),
                    centroid_blob,
                    centroid_dtype,
                    centroid_dim,
                    cluster.coherence as f64,
                    now_ms,
                ],
            )
            .map_err(|e| Error::storage(format!("INSERT cluster: {e}")))?;

            for memid in &cluster.episode_ids {
                txn.execute(
                    "INSERT INTO cluster_episodes (cluster_id, memory_id) VALUES (?, ?)",
                    params![cluster.cluster_id.to_string(), memid.to_string()],
                )
                .map_err(|e| Error::storage(format!("INSERT cluster_episodes: {e}")))?;
            }
        }

        // After the new INSERTs, refresh each modified existing
        // cluster's centroid + coherence to the post-absorb values.
        // Order: deterministic (sorted by cluster_id) so multi-
        // existing absorb runs commit in stable order.
        let mut modified_existing: Vec<&solo_steward::cluster::AbsorbedCluster> =
            absorb_plan.absorptions.iter().collect();
        modified_existing.sort_by(|a, b| a.existing_cluster_id.cmp(&b.existing_cluster_id));
        // Only the LAST absorption into a given existing cluster
        // carries the final centroid + coherence (the working state
        // accumulates step-wise inside `absorb_into_existing`).
        // Deduplicate to keep the most recent per existing_cluster_id.
        let mut last_per_existing: std::collections::HashMap<
            MemoryId,
            &solo_steward::cluster::AbsorbedCluster,
        > = std::collections::HashMap::new();
        for a in &absorb_plan.absorptions {
            last_per_existing.insert(a.existing_cluster_id, a);
        }
        let mut existing_ids_sorted: Vec<MemoryId> =
            last_per_existing.keys().copied().collect();
        existing_ids_sorted.sort();
        for existing_id in existing_ids_sorted {
            let absorbed = last_per_existing[&existing_id];
            txn.execute(
                "UPDATE clusters
                    SET centroid = ?, centroid_dtype = ?, centroid_dim = ?, coherence = ?
                  WHERE cluster_id = ?",
                params![
                    absorbed.merged_centroid.data.as_slice(),
                    "f32",
                    absorbed.merged_centroid.dim as i64,
                    absorbed.merged_coherence as f64,
                    existing_id.to_string(),
                ],
            )
            .map_err(|e| Error::storage(format!("UPDATE existing cluster centroid: {e}")))?;
        }

        txn.commit()
            .map_err(|e| Error::storage(format!("COMMIT consolidate: {e}")))?;

        // -------- Existing-vs-existing merge pass --------
        //
        // Independent of cross-run absorb: detects pre-existing
        // clusters whose centroids have drifted close enough to
        // coalesce. Runs AFTER absorb so post-absorb centroid
        // updates feed in (an absorbed-into existing cluster might
        // now be similar to another pre-existing cluster).
        //
        // For each MergeOp in the plan:
        //   1. UPDATE cluster_episodes SET cluster_id = survivor
        //      WHERE cluster_id IN (loser_ids) — episodes move.
        //   2. UPDATE clusters SET centroid + coherence on the
        //      survivor.
        //   3. DELETE FROM clusters WHERE cluster_id IN (loser_ids)
        //      — cascades drop loser's `cluster_episodes` (now
        //      empty), `semantic_abstractions`, and `triples`.
        //
        // Survivor's own stale `semantic_abstractions` + `triples`
        // are NOT dropped here — the regen pass below handles them
        // identically to absorb-modified survivors.
        //
        // Skipped if no LLM steward is wired (the regen pass that
        // would replace the dropped abstractions only runs with a
        // steward; running merge without regen would leave
        // survivors in a stale-abstraction state with no recovery
        // until the next consolidate). Without a steward the
        // existing v0.2-era posture (stale-but-readable
        // abstractions) is preserved.
        let merge_plan: solo_steward::cluster::MergePlan =
            if self.steward.is_some() {
                let existing_full =
                    self.fetch_existing_clusters_full(cutoff_ms, expected_dim)?;
                if existing_full.len() < 2 {
                    solo_steward::cluster::MergePlan::default()
                } else {
                    solo_steward::cluster::plan_existing_merges(
                        &existing_full,
                        &config,
                    )?
                }
            } else {
                solo_steward::cluster::MergePlan::default()
            };
        report.existing_clusters_merged = merge_plan.absorbed();

        if !merge_plan.merges.is_empty() {
            tracing::info!(
                merges = merge_plan.merges.len(),
                absorbed = merge_plan.absorbed(),
                "consolidate: existing-vs-existing merge applied"
            );
            let merge_txn = self
                .conn
                .transaction_with_behavior(TransactionBehavior::Immediate)
                .map_err(|e| {
                    Error::storage(format!("BEGIN existing-merge txn: {e}"))
                })?;
            for op in &merge_plan.merges {
                let survivor_str = op.survivor_id.to_string();
                // 1. Move episodes from each loser to the survivor.
                for loser_id in &op.loser_ids {
                    merge_txn
                        .execute(
                            "UPDATE cluster_episodes
                                SET cluster_id = ?1
                              WHERE cluster_id = ?2",
                            params![survivor_str, loser_id.to_string()],
                        )
                        .map_err(|e| {
                            Error::storage(format!(
                                "UPDATE cluster_episodes (existing-merge): {e}"
                            ))
                        })?;
                }
                // 2. Refresh survivor's centroid + coherence.
                merge_txn
                    .execute(
                        "UPDATE clusters
                            SET centroid = ?, centroid_dtype = ?, centroid_dim = ?, coherence = ?
                          WHERE cluster_id = ?",
                        params![
                            op.merged_centroid.data.as_slice(),
                            "f32",
                            op.merged_centroid.dim as i64,
                            op.merged_coherence as f64,
                            survivor_str,
                        ],
                    )
                    .map_err(|e| {
                        Error::storage(format!(
                            "UPDATE clusters (existing-merge): {e}"
                        ))
                    })?;
                // 3. DELETE losers — cascades clean their
                // `cluster_episodes` (already empty),
                // `semantic_abstractions`, and `triples`.
                for loser_id in &op.loser_ids {
                    merge_txn
                        .execute(
                            "DELETE FROM clusters WHERE cluster_id = ?",
                            params![loser_id.to_string()],
                        )
                        .map_err(|e| {
                            Error::storage(format!(
                                "DELETE clusters (existing-merge): {e}"
                            ))
                        })?;
                }
            }
            merge_txn
                .commit()
                .map_err(|e| Error::storage(format!("COMMIT existing-merge: {e}")))?;
        }

        // -------- Y.3.3: REM-equivalent abstraction step --------
        //
        // For each cluster we just persisted, ask the LLM (via the
        // Steward) to produce a SemanticAbstraction, then persist it
        // to `semantic_abstractions`. Per-cluster failures are
        // logged + counted; clusters themselves are already-
        // persisted ground truth and never roll back. The cluster
        // table stays consistent even if half of the abstraction
        // calls fail (network / model glitch).
        //
        // Skipped entirely when `self.steward` is None, which is
        // the prod default until a real LlmClient ships. The cheap
        // clustering pass above always runs.
        if let (Some(steward), Some(rt)) = (&self.steward, &self.runtime_handle) {
            // Build a quick lookup so we can hand each cluster only
            // its member episodes.
            let by_id: std::collections::HashMap<MemoryId, &Episode> = candidates
                .iter()
                .map(|(ep, _)| (ep.memory_id, ep))
                .collect();

            // Y.4.2: accumulate every triple that successfully
            // persists during the abstraction loop so the
            // contradiction sweep below can pair them against the
            // existing triples table.
            let mut new_triples: Vec<solo_core::Triple> = Vec::new();

            for cluster in &clusters {
                // Cross-run absorb: clusters folded into an existing
                // DB cluster have NO row in the `clusters` table —
                // their cluster_id never landed. Abstracting them
                // would produce a `semantic_abstractions` INSERT
                // that FK-violates against the missing parent row.
                // Skip. The existing cluster's abstraction stays
                // as-is (stale w.r.t. the absorbed episodes); this
                // is the documented v0.3 tradeoff. Re-abstraction
                // on absorb is a planned follow-up requiring
                // triples → cluster FK plumbing to avoid duplicate
                // facts.
                if absorbed_by_new.contains_key(&cluster.cluster_id) {
                    continue;
                }

                // Resolve the cluster's episode_ids → owned Episode
                // copies (abstract_cluster takes a slice).
                let mut member_eps: Vec<Episode> =
                    Vec::with_capacity(cluster.episode_ids.len());
                let mut missing = false;
                for memid in &cluster.episode_ids {
                    match by_id.get(memid) {
                        Some(ep) => member_eps.push((*ep).clone()),
                        None => {
                            tracing::warn!(
                                cluster_id = %cluster.cluster_id,
                                missing_memory_id = %memid,
                                "abstract step: cluster references episode not in candidate set"
                            );
                            missing = true;
                            break;
                        }
                    }
                }
                if missing {
                    continue;
                }

                // Drive the async LLM call from the writer thread via
                // the captured runtime handle (same pattern as reembed).
                let abs_res = rt.block_on(steward.abstract_cluster(cluster, &member_eps));
                let abstraction = match abs_res {
                    Ok(a) => a,
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %cluster.cluster_id,
                            error = %e,
                            "abstract_cluster failed; cluster persisted, abstraction skipped"
                        );
                        continue;
                    }
                };

                // Persist the abstraction. One transaction per cluster
                // — independent atomicity from the cluster batch.
                let prov_json = match serde_json::to_string(&abstraction.provenance) {
                    Ok(s) => s,
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %cluster.cluster_id,
                            error = %e,
                            "serialize provenance for abstraction failed; skipping"
                        );
                        continue;
                    }
                };
                let abs_txn = match self
                    .conn
                    .transaction_with_behavior(TransactionBehavior::Immediate)
                {
                    Ok(t) => t,
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %cluster.cluster_id,
                            error = %e,
                            "BEGIN abstraction txn failed; skipping"
                        );
                        continue;
                    }
                };
                let abs_insert_res = abs_txn.execute(
                    "INSERT INTO semantic_abstractions
                        (abstraction_id, cluster_id, content, provenance_json,
                         confidence, created_at_ms)
                     VALUES (?, ?, ?, ?, ?, ?)",
                    params![
                        abstraction.abstraction_id.to_string(),
                        abstraction.cluster_id.to_string(),
                        abstraction.content,
                        prov_json,
                        abstraction.confidence.0,
                        now_ms,
                    ],
                );

                // Persist the triples the LLM extracted alongside the
                // abstraction. Same atomicity as the abstraction row —
                // a partial state where the abstraction lands but its
                // triples don't would leak claims with no provenance
                // back-link.
                let triple_insert_res = abs_insert_res.and_then(|_| {
                    for triple in &abstraction.triples {
                        let tprov = serde_json::to_string(&triple.provenance)
                            .unwrap_or_else(|_| "{}".to_string());
                        let object_kind_str = match triple.object_kind {
                            solo_core::TripleObjectKind::Entity => "entity",
                            solo_core::TripleObjectKind::Literal => "literal",
                        };
                        abs_txn.execute(
                            "INSERT INTO triples
                                (triple_id, subject_id, predicate, object_id,
                                 object_kind, valid_from_ms, valid_to_ms,
                                 confidence, provenance_json,
                                 created_at_ms, updated_at_ms, cluster_id)
                             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                            params![
                                triple.triple_id.to_string(),
                                triple.subject_id,
                                triple.predicate,
                                triple.object_id,
                                object_kind_str,
                                triple.valid_from_ms,
                                triple.valid_to_ms,
                                triple.confidence.0,
                                tprov,
                                now_ms,
                                now_ms,
                                cluster.cluster_id.to_string(),
                            ],
                        )?;
                    }
                    Ok(())
                });

                match triple_insert_res.and_then(|_| abs_txn.commit()) {
                    Ok(()) => {
                        report.abstractions_built += 1;
                        report.triples_built += abstraction.triples.len();
                        // Track these for the contradiction sweep
                        // below; only the committed-cluster triples
                        // join the candidate set.
                        new_triples.extend(abstraction.triples);
                    }
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %cluster.cluster_id,
                            error = %e,
                            "INSERT semantic_abstractions / triples failed; skipping"
                        );
                    }
                }
            }

            // -------- Re-abstraction regen pass --------
            //
            // For each pre-existing cluster whose stale abstraction
            // was invalidated by this run's absorb pass OR this
            // run's existing-vs-existing merge pass, drop the
            // existing `semantic_abstractions` row + its linked
            // `triples` (FK ON DELETE CASCADE on cluster_id makes
            // this a single DELETE chain — see migration 0002),
            // then regenerate by feeding the cluster's full episode
            // set back through `steward.abstract_cluster`.
            //
            // Failure modes:
            //   - LLM call fails → log + skip; the cluster ends up
            //     without an abstraction until a subsequent run
            //     retries.
            //   - DELETE/INSERT fails → log + skip with the txn
            //     rolled back; abstraction stays whatever the
            //     pre-DELETE state was (intact stale row).
            //
            // Sources:
            //   - Absorb-modified: from `absorb_plan.absorptions`;
            //     centroid + coherence come from the last
            //     absorption per existing_cluster_id.
            //   - Merge-modified: from `merge_plan.merges`;
            //     centroid + coherence come from `merged_centroid`
            //     + `merged_coherence` on the MergeOp.
            //   - When a cluster appears in BOTH (absorb fired,
            //     then merge consumed the result), the **merge**
            //     entry wins because it captures the FINAL post-
            //     merge state. Absorb's centroid is intermediate.
            //
            // One transaction per cluster — independent atomicity.
            let last_per_existing_for_regen: std::collections::HashMap<
                MemoryId,
                &solo_steward::cluster::AbsorbedCluster,
            > = absorb_plan
                .absorptions
                .iter()
                .map(|a| (a.existing_cluster_id, a))
                .collect();
            let merge_lookup: std::collections::HashMap<
                MemoryId,
                &solo_steward::cluster::MergeOp,
            > = merge_plan
                .merges
                .iter()
                .map(|m| (m.survivor_id, m))
                .collect();
            // Build the union of regen targets in deterministic
            // (sorted) order so the LLM sees the same prompt
            // sequence across re-runs.
            let mut regen_targets: std::collections::BTreeSet<MemoryId> =
                std::collections::BTreeSet::new();
            regen_targets.extend(absorb_plan.modified_existing_ids());
            regen_targets.extend(merge_lookup.keys().copied());
            for existing_id in regen_targets {
                // Pick the source with the post-everything state.
                // Merge wins over absorb when both apply.
                let (regen_centroid, regen_coherence) =
                    if let Some(op) = merge_lookup.get(&existing_id) {
                        (op.merged_centroid.clone(), op.merged_coherence)
                    } else if let Some(absorbed) =
                        last_per_existing_for_regen.get(&existing_id)
                    {
                        (absorbed.merged_centroid.clone(), absorbed.merged_coherence)
                    } else {
                        continue; // shouldn't happen — target sourced from one of the two maps.
                    };
                let episodes = match self.fetch_episodes_for_cluster(&existing_id) {
                    Ok(eps) => eps,
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %existing_id,
                            error = %e,
                            "regen: fetch_episodes_for_cluster failed; skipping"
                        );
                        continue;
                    }
                };
                if episodes.is_empty() {
                    continue;
                }
                let cluster_shell = solo_core::Cluster {
                    cluster_id: existing_id,
                    episode_ids: episodes.iter().map(|e| e.memory_id).collect(),
                    centroid: Some(regen_centroid),
                    coherence: regen_coherence,
                };

                let abs_res = rt.block_on(steward.abstract_cluster(&cluster_shell, &episodes));
                let new_abstraction = match abs_res {
                    Ok(a) => a,
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %existing_id,
                            error = %e,
                            "regen: abstract_cluster failed; abstraction stays absent until next run"
                        );
                        continue;
                    }
                };

                let prov_json = match serde_json::to_string(&new_abstraction.provenance) {
                    Ok(s) => s,
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %existing_id,
                            error = %e,
                            "regen: serialize provenance failed; skipping"
                        );
                        continue;
                    }
                };

                let regen_txn = match self
                    .conn
                    .transaction_with_behavior(TransactionBehavior::Immediate)
                {
                    Ok(t) => t,
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %existing_id,
                            error = %e,
                            "regen: BEGIN txn failed; skipping"
                        );
                        continue;
                    }
                };

                let regen_writes: rusqlite::Result<()> = (|| {
                    regen_txn.execute(
                        "DELETE FROM semantic_abstractions WHERE cluster_id = ?",
                        params![existing_id.to_string()],
                    )?;
                    regen_txn.execute(
                        "DELETE FROM triples WHERE cluster_id = ?",
                        params![existing_id.to_string()],
                    )?;
                    regen_txn.execute(
                        "INSERT INTO semantic_abstractions
                            (abstraction_id, cluster_id, content, provenance_json,
                             confidence, created_at_ms)
                         VALUES (?, ?, ?, ?, ?, ?)",
                        params![
                            new_abstraction.abstraction_id.to_string(),
                            existing_id.to_string(),
                            new_abstraction.content,
                            prov_json,
                            new_abstraction.confidence.0,
                            now_ms,
                        ],
                    )?;
                    for triple in &new_abstraction.triples {
                        let tprov = serde_json::to_string(&triple.provenance)
                            .unwrap_or_else(|_| "{}".to_string());
                        let object_kind_str = match triple.object_kind {
                            solo_core::TripleObjectKind::Entity => "entity",
                            solo_core::TripleObjectKind::Literal => "literal",
                        };
                        regen_txn.execute(
                            "INSERT INTO triples
                                (triple_id, subject_id, predicate, object_id,
                                 object_kind, valid_from_ms, valid_to_ms,
                                 confidence, provenance_json,
                                 created_at_ms, updated_at_ms, cluster_id)
                             VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                            params![
                                triple.triple_id.to_string(),
                                triple.subject_id,
                                triple.predicate,
                                triple.object_id,
                                object_kind_str,
                                triple.valid_from_ms,
                                triple.valid_to_ms,
                                triple.confidence.0,
                                tprov,
                                now_ms,
                                now_ms,
                                existing_id.to_string(),
                            ],
                        )?;
                    }
                    Ok(())
                })();

                match regen_writes.and_then(|_| regen_txn.commit()) {
                    Ok(()) => {
                        report.abstractions_regenerated += 1;
                        report.triples_built += new_abstraction.triples.len();
                        new_triples.extend(new_abstraction.triples);
                    }
                    Err(e) => {
                        tracing::warn!(
                            cluster_id = %existing_id,
                            error = %e,
                            "regen: DELETE/INSERT failed; skipping"
                        );
                    }
                }
            }

            // -------- Y.4.2: contradiction sweep --------
            //
            // For every triple we just persisted, look for existing
            // triples in SQL that share `(subject_id, predicate)` —
            // the cheap pre-filter that mirrors
            // `contradiction::is_contradiction_candidate`'s subject
            // + predicate short-circuits. Plus pair every
            // newly-built triple against the others in this run
            // (cross-cluster contradictions surfaced in the same
            // consolidation cycle).
            //
            // Errors (LLM, SQL fetch, INSERT) log + skip — clusters
            // / abstractions / triples are already-persisted ground
            // truth and never roll back. UNIQUE(a, b, kind) +
            // INSERT OR IGNORE gives idempotent re-runs.
            //
            // v0.5.0 sub-step 2B: gated on `steward.has_llm()`. A
            // stub-only steward can build clusters and synthetic
            // abstractions, but the LLM judge can't faithfully
            // arbitrate two real triples it has never seen —
            // running the sweep against canned responses would
            // produce nonsense contradictions. We skip + log
            // instead so operators see why no contradictions get
            // flagged. The cluster + abstraction work above
            // already ran; this is a clean early-return on the
            // sweep alone.
            if !new_triples.is_empty() && !steward.has_llm() {
                tracing::warn!(
                    reason = "no LLM client configured",
                    "contradiction sweep skipped; set \
                     ANTHROPIC_API_KEY or OPENAI_API_KEY (or use \
                     --ollama-model) to enable"
                );
            } else if !new_triples.is_empty() {
                let new_ids: std::collections::HashSet<MemoryId> = new_triples
                    .iter()
                    .map(|t| t.triple_id)
                    .collect();

                let mut detected: Vec<solo_core::Contradiction> = Vec::new();

                // Observability counters — emitted as a single
                // INFO-level structured log at the end of the
                // sweep. `candidate_pair_count` is every pair we
                // tried; `judge_call_count` is the subset that
                // survived the cheap rule filter and actually
                // reached the LLM; `judge_rejection_count` is
                // the count that ended up classified as NOT a
                // contradiction (rule-filtered OR judge-said-no).
                // Closes the observability gap from the v0.4.1
                // thesis test, where `RUST_LOG=debug` couldn't
                // tell whether the judge ran at all.
                let mut candidate_pair_count: usize = 0;
                let mut judge_call_count: usize = 0;
                let mut judge_rejection_count: usize = 0;

                // (1) New × Existing — for each new triple, fetch
                // existing-DB candidates that match (subject, predicate)
                // but aren't part of this run's batch.
                for new in &new_triples {
                    let existing = match self.fetch_triples_for_pair(
                        &new.subject_id,
                        &new.predicate,
                        &new_ids,
                    ) {
                        Ok(v) => v,
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                subject = %new.subject_id,
                                predicate = %new.predicate,
                                "fetch existing triples failed; skipping"
                            );
                            continue;
                        }
                    };
                    for old in &existing {
                        candidate_pair_count += 1;
                        // Rule-filter the pair locally too so we
                        // can count judge-vs-filter separately.
                        // `detect_contradiction` redundantly runs
                        // the same check internally; cost is a
                        // handful of integer compares.
                        if solo_steward::contradiction::is_contradiction_candidate(new, old) {
                            judge_call_count += 1;
                        }
                        match rt.block_on(steward.detect_contradiction(new, old)) {
                            Ok(Some(c)) => detected.push(c),
                            Ok(None) => {
                                judge_rejection_count += 1;
                            }
                            Err(e) => {
                                tracing::warn!(
                                    error = %e,
                                    "detect_contradiction failed; skipping pair"
                                );
                            }
                        }
                    }
                }

                // (2) New × New — pairs within this run.
                for i in 0..new_triples.len() {
                    for j in (i + 1)..new_triples.len() {
                        let a = &new_triples[i];
                        let b = &new_triples[j];
                        candidate_pair_count += 1;
                        if solo_steward::contradiction::is_contradiction_candidate(a, b) {
                            judge_call_count += 1;
                        }
                        match rt.block_on(steward.detect_contradiction(a, b)) {
                            Ok(Some(c)) => detected.push(c),
                            Ok(None) => {
                                judge_rejection_count += 1;
                            }
                            Err(e) => {
                                tracing::warn!(
                                    error = %e,
                                    "detect_contradiction failed; skipping pair"
                                );
                            }
                        }
                    }
                }

                tracing::info!(
                    candidate_pair_count,
                    judge_call_count,
                    judge_rejection_count,
                    prompt_version_hash =
                        solo_steward::contradiction::prompt_version_hash(),
                    "contradiction sweep completed"
                );

                // Persist. UNIQUE(a, b, kind) requires consistent
                // (a, b) ordering for dedup across re-runs;
                // normalize so a_memory_id <= b_memory_id.
                for c in detected {
                    let (a_str, b_str) = if c.a <= c.b {
                        (c.a.to_string(), c.b.to_string())
                    } else {
                        (c.b.to_string(), c.a.to_string())
                    };
                    let kind_str = match c.kind {
                        solo_core::ContradictionKind::OverlappingSingleValuedPredicate => {
                            "overlapping_single_valued_predicate"
                        }
                        solo_core::ContradictionKind::DirectNegation => "direct_negation",
                        solo_core::ContradictionKind::NumericInconsistency => {
                            "numeric_inconsistency"
                        }
                        solo_core::ContradictionKind::Other => "other",
                    };
                    match self.conn.execute(
                        "INSERT OR IGNORE INTO contradictions
                            (a_memory_id, b_memory_id, kind, explanation, detected_at_ms)
                         VALUES (?, ?, ?, ?, ?)",
                        params![a_str, b_str, kind_str, c.explanation, now_ms],
                    ) {
                        Ok(rows_inserted) => {
                            // Only count rows that actually landed —
                            // duplicate (UNIQUE) rejects → 0.
                            report.contradictions_found += rows_inserted;
                        }
                        Err(e) => {
                            tracing::warn!(
                                error = %e,
                                a = %a_str,
                                b = %b_str,
                                "INSERT contradiction failed; skipping"
                            );
                        }
                    }
                }
            }
        }

        tracing::info!(
            seen = report.episodes_seen,
            clusters = report.clusters_built,
            episodes_clustered = report.episodes_clustered,
            abstractions = report.abstractions_built,
            triples = report.triples_built,
            contradictions = report.contradictions_found,
            "consolidate complete"
        );
        Ok(report)
    }

    /// Fetch every triple that shares `(subject_id, predicate)` with
    /// the new triple, excluding the new run's batch (passed via
    /// `exclude`). Used by `handle_consolidate`'s contradiction sweep
    /// to narrow LLM-judge candidates to plausible pairs only.
    ///
    /// Returns reassembled `Triple` structs. Provenance is parsed
    /// from the row's `provenance_json` column; on parse failure we
    /// substitute a placeholder `Provenance` (the triple is still
    /// usable for contradiction detection — the rule filter doesn't
    /// touch provenance, and the LLM judge prompt doesn't include
    /// it either).
    fn fetch_triples_for_pair(
        &self,
        subject_id: &str,
        predicate: &str,
        exclude: &std::collections::HashSet<MemoryId>,
    ) -> Result<Vec<solo_core::Triple>> {
        let mut stmt = self
            .conn
            .prepare(
                "SELECT triple_id, object_id, object_kind, valid_from_ms, valid_to_ms,
                        confidence, provenance_json
                 FROM triples
                 WHERE subject_id = ?1 AND predicate = ?2
                   AND status = 'active'",
            )
            .map_err(|e| Error::storage(format!("prepare fetch_triples_for_pair: {e}")))?;
        let rows = stmt
            .query_map(params![subject_id, predicate], |r| {
                Ok((
                    r.get::<_, String>(0)?,
                    r.get::<_, String>(1)?,
                    r.get::<_, String>(2)?,
                    r.get::<_, i64>(3)?,
                    r.get::<_, Option<i64>>(4)?,
                    r.get::<_, f32>(5)?,
                    r.get::<_, String>(6)?,
                ))
            })
            .map_err(|e| Error::storage(format!("query_map triples: {e}")))?;

        let mut out = Vec::new();
        for row in rows {
            let (triple_id_s, object_id, object_kind_s, valid_from_ms, valid_to_ms, conf, prov_s) =
                row.map_err(|e| Error::storage(format!("triples row decode: {e}")))?;
            let triple_id = MemoryId::from_str(&triple_id_s).map_err(|e| {
                Error::storage(format!("parse triple_id `{triple_id_s}`: {e}"))
            })?;
            if exclude.contains(&triple_id) {
                continue;
            }
            let object_kind = match object_kind_s.as_str() {
                "entity" => solo_core::TripleObjectKind::Entity,
                "literal" => solo_core::TripleObjectKind::Literal,
                other => {
                    return Err(Error::storage(format!(
                        "unknown object_kind value `{other}` in triples row"
                    )));
                }
            };
            let confidence = solo_core::Confidence::new(conf).map_err(|e| {
                Error::storage(format!("invalid confidence in triples row: {e}"))
            })?;
            let provenance: solo_core::Provenance = serde_json::from_str(&prov_s)
                .unwrap_or_else(|_| solo_core::Provenance {
                    derived_from: vec![],
                    derivation: "(unparseable)".into(),
                    by: "(unknown)".into(),
                    at_ms: 0,
                });
            out.push(solo_core::Triple {
                triple_id,
                subject_id: subject_id.to_string(),
                predicate: predicate.to_string(),
                object_id,
                object_kind,
                valid_from_ms,
                valid_to_ms,
                confidence,
                provenance,
            });
        }
        Ok(out)
    }

    /// Fetch compact summaries of every existing cluster within the
    /// consolidate window. Used by `handle_consolidate`'s cross-run
    /// absorb pass to decide which freshly-built clusters fold into
    /// pre-existing DB clusters with similar centroids.
    ///
    /// Filters:
    ///
    ///   - `cutoff_ms`: when `Some(ms)`, restrict to clusters with
    ///     `created_at_ms >= ms` (matches the candidate-episode
    ///     window). When `None`, all clusters in the table.
    ///   - Centroid must be present (non-null) and dim must equal
    ///     `expected_dim` — clusters built under a different
    ///     embedder are skipped (their centroids live in a
    ///     different vector space and absorb-cosine would be
    ///     meaningless).
    ///
    /// Returns one row per surviving cluster with its centroid +
    /// coherence + episode count. The episode_count is computed via
    /// a correlated `COUNT(*)` against `cluster_episodes` — cheap
    /// thanks to `idx_cluster_episodes_memory` (well, technically
    /// thanks to the `(cluster_id, memory_id)` PRIMARY KEY ordering).
    fn fetch_existing_cluster_summaries(
        &self,
        cutoff_ms: Option<i64>,
        expected_dim: usize,
    ) -> Result<Vec<solo_steward::cluster::ExistingClusterSummary>> {
        let (sql, params): (&str, Vec<rusqlite::types::Value>) = match cutoff_ms {
            Some(cutoff) => (
                "SELECT c.cluster_id, c.centroid, c.centroid_dtype, c.centroid_dim,
                        c.coherence,
                        (SELECT COUNT(*) FROM cluster_episodes ce
                         WHERE ce.cluster_id = c.cluster_id) AS episode_count
                 FROM clusters c
                 WHERE c.centroid IS NOT NULL
                   AND c.centroid_dtype = 'f32'
                   AND c.centroid_dim = ?1
                   AND c.created_at_ms >= ?2
                 ORDER BY c.cluster_id",
                vec![(expected_dim as i64).into(), cutoff.into()],
            ),
            None => (
                "SELECT c.cluster_id, c.centroid, c.centroid_dtype, c.centroid_dim,
                        c.coherence,
                        (SELECT COUNT(*) FROM cluster_episodes ce
                         WHERE ce.cluster_id = c.cluster_id) AS episode_count
                 FROM clusters c
                 WHERE c.centroid IS NOT NULL
                   AND c.centroid_dtype = 'f32'
                   AND c.centroid_dim = ?1
                 ORDER BY c.cluster_id",
                vec![(expected_dim as i64).into()],
            ),
        };

        let mut stmt = self
            .conn
            .prepare(sql)
            .map_err(|e| Error::storage(format!("prepare existing-cluster summaries: {e}")))?;
        let rows = stmt
            .query_map(params_from_iter(&params), |r| {
                Ok((
                    r.get::<_, String>(0)?,       // cluster_id
                    r.get::<_, Vec<u8>>(1)?,      // centroid blob
                    r.get::<_, String>(2)?,       // centroid_dtype
                    r.get::<_, i64>(3)?,          // centroid_dim
                    r.get::<_, f32>(4)?,          // coherence
                    r.get::<_, i64>(5)?,          // episode_count
                ))
            })
            .map_err(|e| Error::storage(format!("query_map existing clusters: {e}")))?;

        let mut out: Vec<solo_steward::cluster::ExistingClusterSummary> = Vec::new();
        for row in rows {
            let (cid_s, centroid_bytes, dtype_s, dim_i, coherence, count_i) =
                row.map_err(|e| Error::storage(format!("cluster row decode: {e}")))?;
            // Defensive: SQL's WHERE already filtered to f32 +
            // expected_dim; trust but verify the row contents.
            if dtype_s != "f32" || (dim_i as usize) != expected_dim {
                continue;
            }
            let cluster_id = match MemoryId::from_str(&cid_s) {
                Ok(id) => id,
                Err(e) => {
                    tracing::warn!(
                        cluster_id = %cid_s,
                        error = %e,
                        "skipping cluster with unparseable cluster_id"
                    );
                    continue;
                }
            };
            let centroid = solo_core::Embedding {
                dtype: solo_core::EmbeddingDtype::F32,
                dim: dim_i as usize,
                data: centroid_bytes,
            };
            // count_i is the exact integer from COUNT(*); negatives
            // can't happen, but `as usize` saturates safely on the
            // off-chance of an i64 overflow case.
            let episode_count = count_i.max(0) as usize;
            // Skip clusters with zero episodes (orphaned rows
            // shouldn't exist thanks to ON DELETE CASCADE, but
            // guarding here keeps absorb math defensible).
            if episode_count == 0 {
                continue;
            }
            out.push(solo_steward::cluster::ExistingClusterSummary {
                cluster_id,
                centroid,
                coherence,
                episode_count,
            });
        }
        Ok(out)
    }

    /// Load every existing cluster as a full [`Cluster`] struct
    /// (centroid + coherence + complete `episode_ids` Vec). Used by
    /// the existing-vs-existing merge pass to feed
    /// `solo_steward::cluster::plan_existing_merges`.
    ///
    /// One SELECT joins `clusters` with `cluster_episodes`; rows are
    /// aggregated by `cluster_id`. Filters mirror
    /// `fetch_existing_cluster_summaries` (centroid present, dtype +
    /// dim match the current embedder, optional `created_at_ms` cutoff).
    /// Clusters with zero linked episodes are skipped — those are
    /// orphan rows that shouldn't exist post-CASCADE invariants.
    fn fetch_existing_clusters_full(
        &self,
        cutoff_ms: Option<i64>,
        expected_dim: usize,
    ) -> Result<Vec<solo_core::Cluster>> {
        let (sql, params): (&str, Vec<rusqlite::types::Value>) = match cutoff_ms {
            Some(cutoff) => (
                "SELECT c.cluster_id, c.centroid, c.centroid_dtype, c.centroid_dim,
                        c.coherence, ce.memory_id
                 FROM clusters c
                 JOIN cluster_episodes ce ON ce.cluster_id = c.cluster_id
                 WHERE c.centroid IS NOT NULL
                   AND c.centroid_dtype = 'f32'
                   AND c.centroid_dim = ?1
                   AND c.created_at_ms >= ?2
                 ORDER BY c.cluster_id, ce.memory_id",
                vec![(expected_dim as i64).into(), cutoff.into()],
            ),
            None => (
                "SELECT c.cluster_id, c.centroid, c.centroid_dtype, c.centroid_dim,
                        c.coherence, ce.memory_id
                 FROM clusters c
                 JOIN cluster_episodes ce ON ce.cluster_id = c.cluster_id
                 WHERE c.centroid IS NOT NULL
                   AND c.centroid_dtype = 'f32'
                   AND c.centroid_dim = ?1
                 ORDER BY c.cluster_id, ce.memory_id",
                vec![(expected_dim as i64).into()],
            ),
        };

        let mut stmt = self
            .conn
            .prepare(sql)
            .map_err(|e| Error::storage(format!("prepare existing clusters full: {e}")))?;
        let rows = stmt
            .query_map(params_from_iter(&params), |r| {
                Ok((
                    r.get::<_, String>(0)?, // cluster_id
                    r.get::<_, Vec<u8>>(1)?, // centroid bytes
                    r.get::<_, String>(2)?, // dtype
                    r.get::<_, i64>(3)?,    // dim
                    r.get::<_, f32>(4)?,    // coherence
                    r.get::<_, String>(5)?, // memory_id
                ))
            })
            .map_err(|e| Error::storage(format!("query_map clusters full: {e}")))?;

        // Aggregate. Rows are ORDER BY cluster_id so we can build
        // the output as a single pass.
        let mut out: Vec<solo_core::Cluster> = Vec::new();
        for row in rows {
            let (cid_s, centroid_bytes, dtype_s, dim_i, coherence, memid_s) =
                row.map_err(|e| Error::storage(format!("clusters full row decode: {e}")))?;
            if dtype_s != "f32" || (dim_i as usize) != expected_dim {
                continue;
            }
            let cluster_id = match MemoryId::from_str(&cid_s) {
                Ok(id) => id,
                Err(e) => {
                    tracing::warn!(
                        cluster_id = %cid_s,
                        error = %e,
                        "skipping cluster with unparseable cluster_id"
                    );
                    continue;
                }
            };
            let memory_id = match MemoryId::from_str(&memid_s) {
                Ok(id) => id,
                Err(e) => {
                    tracing::warn!(
                        memory_id = %memid_s,
                        error = %e,
                        "skipping cluster_episodes row with unparseable memory_id"
                    );
                    continue;
                }
            };
            // Append to the in-progress cluster (last entry in
            // `out`) when cluster_id matches; otherwise start a new
            // entry.
            if out.last().map(|c| c.cluster_id) == Some(cluster_id) {
                out.last_mut().unwrap().episode_ids.push(memory_id);
            } else {
                let centroid = solo_core::Embedding {
                    dtype: solo_core::EmbeddingDtype::F32,
                    dim: dim_i as usize,
                    data: centroid_bytes,
                };
                out.push(solo_core::Cluster {
                    cluster_id,
                    episode_ids: vec![memory_id],
                    centroid: Some(centroid),
                    coherence,
                });
            }
        }
        // Drop empty clusters (defensive — shouldn't happen given
        // the JOIN requires at least one cluster_episodes row).
        out.retain(|c| !c.episode_ids.is_empty());
        Ok(out)
    }

    /// Load full [`Episode`] structs for every memory_id linked to
    /// `cluster_id` via `cluster_episodes`. Used by the absorb→regen
    /// path to feed `steward.abstract_cluster` for a cluster whose
    /// stale abstraction needs replacing.
    ///
    /// Filters: `episodes.status = 'active'` (forgotten episodes
    /// can't drive a fresh abstraction; they get skipped). The
    /// SELECT mirrors `handle_consolidate`'s candidate fetch but
    /// without the `embedder_id` filter — the regen pass operates
    /// on the cluster's full historical episode set, regardless of
    /// which embedder produced their vectors.
    ///
    /// Returns episodes ordered by `(ts_ms, rowid)` for
    /// deterministic prompts.
    fn fetch_episodes_for_cluster(
        &self,
        cluster_id: &MemoryId,
    ) -> Result<Vec<Episode>> {
        let mut stmt = self
            .conn
            .prepare(
                "SELECT e.memory_id, e.ts_ms, e.source_type, e.source_id,
                        e.content, e.encoding_context_json, e.provenance_json,
                        e.confidence, e.strength, e.salience, e.tier
                 FROM episodes e
                 JOIN cluster_episodes ce ON ce.memory_id = e.memory_id
                 WHERE ce.cluster_id = ?1
                   AND e.status = 'active'
                 ORDER BY e.ts_ms, e.rowid",
            )
            .map_err(|e| {
                Error::storage(format!("prepare fetch_episodes_for_cluster: {e}"))
            })?;
        let rows = stmt
            .query_map(params![cluster_id.to_string()], |r| {
                Ok((
                    r.get::<_, String>(0)?,             // memory_id
                    r.get::<_, i64>(1)?,                // ts_ms
                    r.get::<_, String>(2)?,             // source_type
                    r.get::<_, Option<String>>(3)?,     // source_id
                    r.get::<_, String>(4)?,             // content
                    r.get::<_, String>(5)?,             // encoding_context_json
                    r.get::<_, Option<String>>(6)?,     // provenance_json
                    r.get::<_, f32>(7)?,                // confidence
                    r.get::<_, f32>(8)?,                // strength
                    r.get::<_, f32>(9)?,                // salience
                    r.get::<_, String>(10)?,            // tier
                ))
            })
            .map_err(|e| Error::storage(format!("query_map cluster episodes: {e}")))?;

        let mut out: Vec<Episode> = Vec::new();
        for row in rows {
            let (
                mid_s,
                ts_ms,
                source_type,
                source_id,
                content,
                ctx_json,
                prov_json,
                conf,
                strength,
                salience,
                tier_s,
            ) = row.map_err(|e| Error::storage(format!("episode row decode: {e}")))?;
            let mid = MemoryId::from_str(&mid_s)
                .map_err(|e| Error::storage(format!("parse memory_id `{mid_s}`: {e}")))?;
            let confidence = solo_core::Confidence::new(conf).map_err(|e| {
                Error::storage(format!("invalid confidence in episode row: {e}"))
            })?;
            let encoding_context: solo_core::EncodingContext =
                serde_json::from_str(&ctx_json).unwrap_or_default();
            let provenance: Option<solo_core::Provenance> = prov_json
                .as_deref()
                .and_then(|s| serde_json::from_str(s).ok());
            let tier = match tier_s.as_str() {
                "hot" => Tier::Hot,
                "warm" => Tier::Warm,
                "cold" => Tier::Cold,
                other => {
                    return Err(Error::storage(format!(
                        "unknown tier value `{other}` in episodes row"
                    )));
                }
            };
            out.push(Episode {
                memory_id: mid,
                ts_ms,
                source_type,
                source_id,
                content,
                encoding_context,
                provenance,
                confidence,
                strength,
                salience,
                tier,
            });
        }
        Ok(out)
    }

    fn handle_reembed(&mut self, scope: ReembedScope) -> Result<ReembedReport> {
        // Reembed needs the writer to have been spawned with the active
        // embedder + a runtime handle (see `spawn_full_with_embedder`).
        // The plain `spawn_full` constructor leaves these `None`, which
        // is the correct posture for daemon paths that don't dispatch
        // Reembed. A clean error here beats a panic.
        let current_id = self.embedder_id.ok_or_else(|| {
            Error::Other(
                "reembed: writer has no current embedder_id (use spawn_full_with_embedder)"
                    .into(),
            )
        })?;
        let embedder = self.embedder.clone().ok_or_else(|| {
            Error::Other(
                "reembed: writer has no embedder (use spawn_full_with_embedder)".into(),
            )
        })?;
        let runtime = self.runtime_handle.clone().ok_or_else(|| {
            Error::Other(
                "reembed: writer has no runtime handle (use spawn_full_with_embedder)"
                    .into(),
            )
        })?;

        // Optional `from` filter → resolve to embedder_id once. Refuse if
        // the user asked to migrate from "current to current" (nothing
        // would happen) or if the from-embedder isn't registered.
        let from_id: Option<i64> = match &scope.from {
            None => None,
            Some((name, version)) => {
                let id: Option<i64> = self
                    .conn
                    .query_row(
                        "SELECT embedder_id FROM embedders WHERE name = ? AND version = ?",
                        params![name, version],
                        |r| r.get::<_, i64>(0),
                    )
                    .optional()
                    .map_err(|e| Error::storage(format!("lookup from embedder: {e}")))?;
                match id {
                    Some(id) if id == current_id => {
                        return Err(Error::Other(format!(
                            "reembed: from-embedder ({name}, {version}) IS the current \
                             embedder; nothing to do"
                        )));
                    }
                    Some(id) => Some(id),
                    None => {
                        return Err(Error::not_found(format!(
                            "reembed: from-embedder ({name}, {version}) not registered \
                             in `embedders` table"
                        )));
                    }
                }
            }
        };

        // Build the candidate set. DISTINCT — a memory may have multiple
        // stale rows (if the user has rolled through more than one prior
        // embedder); we only want to embed each content once.
        let candidates: Vec<(String, String)> = {
            let (sql, bound_id): (&str, i64) = match from_id {
                None => (
                    "SELECT DISTINCT e.memory_id, e.content
                     FROM episodes e
                     JOIN embeddings em ON em.memory_id = e.memory_id
                     WHERE em.embedder_id != ?1
                       AND e.status = 'active'
                     ORDER BY e.rowid",
                    current_id,
                ),
                Some(fid) => (
                    "SELECT DISTINCT e.memory_id, e.content
                     FROM episodes e
                     JOIN embeddings em ON em.memory_id = e.memory_id
                     WHERE em.embedder_id = ?1
                       AND e.status = 'active'
                     ORDER BY e.rowid",
                    fid,
                ),
            };
            let mut stmt = self
                .conn
                .prepare(sql)
                .map_err(|e| Error::storage(format!("prepare reembed select: {e}")))?;
            let rows = stmt
                .query_map(params![bound_id], |r| {
                    Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
                })
                .map_err(|e| Error::storage(format!("query_map reembed: {e}")))?;
            let mut out = Vec::new();
            for row in rows {
                out.push(
                    row.map_err(|e| Error::storage(format!("reembed row decode: {e}")))?,
                );
            }
            out
        };

        let mut report = ReembedReport {
            rows_seen: candidates.len(),
            rows_reembedded: 0,
            rows_failed: 0,
            rows_gc_deleted: 0,
            dry_run: scope.dry_run,
        };

        if scope.dry_run {
            tracing::info!(
                seen = report.rows_seen,
                "reembed --dry-run: would re-embed N memories"
            );
            return Ok(report);
        }

        // Cache the dtype string for the current embedder once.
        let dtype_str = match embedder.dtype() {
            solo_core::EmbeddingDtype::F32 => "f32",
            solo_core::EmbeddingDtype::F16 => "f16",
            solo_core::EmbeddingDtype::I8 => "i8",
            solo_core::EmbeddingDtype::Binary => "binary",
        };
        let now_ms = chrono::Utc::now().timestamp_millis();

        // Per-memory loop. Embed (async, off the writer thread via the
        // captured runtime handle), then atomically apply the SQL changes
        // for that memory. A failure on one memory does NOT abort the
        // run — partial progress is fine because the next reembed pass
        // picks up wherever we left off (the SELECT re-evaluates the
        // candidate set at start). The per-memory transaction means a
        // mid-run crash cannot leave a memory with two `current`
        // embedding rows or with the new row but missing the GC.
        for (memory_id, content) in candidates {
            let embedding_res = runtime.block_on(embedder.embed(&content));
            let new_embedding = match embedding_res {
                Ok(emb) => emb,
                Err(e) => {
                    tracing::warn!(%memory_id, error = %e, "reembed: embedder failed");
                    report.rows_failed += 1;
                    continue;
                }
            };
            if let Err(e) = new_embedding.validate() {
                tracing::warn!(%memory_id, error = %e, "reembed: embedding validate failed");
                report.rows_failed += 1;
                continue;
            }

            let txn = match self
                .conn
                .transaction_with_behavior(TransactionBehavior::Immediate)
            {
                Ok(t) => t,
                Err(e) => {
                    tracing::warn!(%memory_id, error = %e, "reembed: BEGIN failed");
                    report.rows_failed += 1;
                    continue;
                }
            };

            // INSERT ... ON CONFLICT(memory_id, embedder_id) DO UPDATE.
            // If a partial earlier reembed already wrote the current row
            // for this memory, refresh it with the freshly-computed
            // vector. (Same content + same embedder = same vector, so
            // this is a no-op semantically; we still bump created_at_ms.)
            let insert_res = txn.execute(
                "INSERT INTO embeddings (memory_id, embedder_id, dtype, dim, vector, created_at_ms)
                 VALUES (?, ?, ?, ?, ?, ?)
                 ON CONFLICT(memory_id, embedder_id) DO UPDATE SET
                    dtype = excluded.dtype,
                    dim = excluded.dim,
                    vector = excluded.vector,
                    created_at_ms = excluded.created_at_ms",
                params![
                    memory_id,
                    current_id,
                    dtype_str,
                    new_embedding.dim as i64,
                    &new_embedding.data[..],
                    now_ms,
                ],
            );
            if let Err(e) = insert_res {
                tracing::warn!(%memory_id, error = %e, "reembed: INSERT failed");
                report.rows_failed += 1;
                continue;
            }

            let gc_count = if scope.gc {
                match txn.execute(
                    "DELETE FROM embeddings
                     WHERE memory_id = ? AND embedder_id != ?",
                    params![memory_id, current_id],
                ) {
                    Ok(n) => n,
                    Err(e) => {
                        tracing::warn!(%memory_id, error = %e, "reembed: GC DELETE failed");
                        report.rows_failed += 1;
                        continue;
                    }
                }
            } else {
                0
            };

            if let Err(e) = txn.commit() {
                tracing::warn!(%memory_id, error = %e, "reembed: COMMIT failed");
                report.rows_failed += 1;
                continue;
            }
            report.rows_reembedded += 1;
            report.rows_gc_deleted += gc_count;
        }

        tracing::info!(
            seen = report.rows_seen,
            reembedded = report.rows_reembedded,
            failed = report.rows_failed,
            gc_deleted = report.rows_gc_deleted,
            "reembed complete"
        );
        Ok(report)
    }

    /// Implementation of `WriteCommand::NormalizeSubjects`.
    ///
    /// One transaction wraps every `(from, to)` pair so the entire
    /// backfill is atomic. For each pair: run the symmetric
    /// `subject_id` and `object_id` UPDATEs, accumulate row counts.
    /// In `dry_run` mode the transaction is `ROLLBACK`ed at the end;
    /// otherwise it commits.
    ///
    /// SQL uses parameterized statements (`?1`, `?2`) — alias strings
    /// never reach SQL via `format!`, so this is injection-safe even
    /// for adversarial alias values.
    ///
    /// A single triple where `subject_id == object_id == from` is
    /// counted as one subject row + one object row (count of 2 against
    /// that triple), matching what SQLite's `changes()` reports for the
    /// two separate UPDATEs.
    fn handle_normalize_subjects(
        &mut self,
        aliases: Vec<(String, String)>,
        dry_run: bool,
    ) -> Result<NormalizeReport> {
        let mut report = NormalizeReport {
            aliases_processed: aliases.len(),
            subject_rows_updated: 0,
            object_rows_updated: 0,
            dry_run,
        };

        // Empty alias list: nothing to do, no transaction needed.
        // Mirrors the "zero candidates" short-circuit in handle_reembed.
        if aliases.is_empty() {
            tracing::info!(dry_run, "normalize_subjects: empty alias list, no-op");
            return Ok(report);
        }

        let tx = self
            .conn
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(|e| {
                Error::storage(format!("BEGIN IMMEDIATE for normalize_subjects: {e}"))
            })?;

        for (from, to) in &aliases {
            let subj_rows = tx
                .execute(
                    "UPDATE triples SET subject_id = ?1, updated_at_ms = ?3 \
                     WHERE subject_id = ?2",
                    params![to, from, chrono::Utc::now().timestamp_millis()],
                )
                .map_err(|e| {
                    Error::storage(format!(
                        "normalize_subjects: UPDATE subject_id ({from} -> {to}): {e}"
                    ))
                })?;
            let obj_rows = tx
                .execute(
                    "UPDATE triples SET object_id = ?1, updated_at_ms = ?3 \
                     WHERE object_id = ?2",
                    params![to, from, chrono::Utc::now().timestamp_millis()],
                )
                .map_err(|e| {
                    Error::storage(format!(
                        "normalize_subjects: UPDATE object_id ({from} -> {to}): {e}"
                    ))
                })?;
            report.subject_rows_updated += subj_rows;
            report.object_rows_updated += obj_rows;
        }

        if dry_run {
            tx.rollback().map_err(|e| {
                Error::storage(format!(
                    "normalize_subjects: ROLLBACK after dry-run: {e}"
                ))
            })?;
            tracing::info!(
                aliases_processed = report.aliases_processed,
                subject_rows = report.subject_rows_updated,
                object_rows = report.object_rows_updated,
                "normalize_subjects --dry-run: rolled back (would have updated N rows)"
            );
        } else {
            tx.commit().map_err(|e| {
                Error::storage(format!("normalize_subjects: COMMIT: {e}"))
            })?;
            tracing::info!(
                aliases_processed = report.aliases_processed,
                subject_rows = report.subject_rows_updated,
                object_rows = report.object_rows_updated,
                "normalize_subjects complete"
            );
        }

        Ok(report)
    }

    fn handle_backup(&mut self, dest_path: &std::path::Path) -> Result<()> {
        let key = self.key.as_ref().ok_or_else(|| {
            Error::storage(
                "backup called but writer has no key material configured. \
                 Spawn the writer with `spawn_full_with_key_and_optional_steward` \
                 to enable WriteCommand::Backup.",
            )
        })?;
        // Important: route through the writer's existing source connection
        // so the backup runs against live in-flight WAL state via SQLite's
        // page-level snapshot. SQLite serialises page reads with concurrent
        // writes on the same connection, so this is safe even mid-burst.
        backup_from_connection(&self.conn, dest_path, key)
    }

    fn handle_save_snapshot(&mut self) -> Result<()> {
        let dir = self.snapshot_dir.as_ref().ok_or_else(|| {
            Error::storage("save_snapshot called but writer has no snapshot_dir configured")
        })?;
        // Delegates to the impl's `save` (HnswIndex routes to crate::snapshot).
        // The trait keeps us implementation-agnostic; the StubVectorIndex used
        // in unit tests just bumps a counter.
        let save_result = self.hnsw.save(dir);

        // Piggyback maintenance pragmas on the snapshot cadence (5 min by
        // default). ADR-0003 §O5 / §"Final consolidated action items" #9
        // call for hourly `PRAGMA optimize` + idle PASSIVE checkpoint.
        // Running them every 5 min instead of hourly is harmless (both are
        // cheap when there's nothing to do) and avoids a separate timer
        // task. Failures are logged but don't fail the save itself —
        // they're maintenance, not durability.
        self.run_idle_maintenance();

        save_result
    }

    /// Best-effort PRAGMA optimize + wal_checkpoint(PASSIVE). Safe to call
    /// on the writer's connection at any time.
    fn run_idle_maintenance(&mut self) {
        if let Err(e) = self.conn.execute_batch("PRAGMA optimize") {
            tracing::debug!(error = %e, "PRAGMA optimize failed (non-fatal)");
        }
        if let Err(e) = self.conn.execute_batch("PRAGMA wal_checkpoint(PASSIVE)") {
            tracing::debug!(error = %e, "PRAGMA wal_checkpoint(PASSIVE) failed (non-fatal)");
        }
    }

    fn shutdown(&mut self) {
        if let Err(e) = self
            .conn
            .pragma_update(None, "wal_checkpoint", "TRUNCATE")
        {
            tracing::warn!(error = %e, "wal_checkpoint(TRUNCATE) on shutdown failed");
        }
        tracing::info!("writer actor shutdown complete");
    }
}

/// Pick a `documents.title` from the parsed text + source path. Prefer
/// the first Markdown-style `# heading` line in the first 64 lines of
/// the document; fall back to the file stem; fall back to "(untitled)".
fn derive_document_title(text: &str, path: &std::path::Path) -> String {
    for (i, line) in text.lines().enumerate() {
        if i >= 64 {
            break;
        }
        let trimmed = line.trim_start();
        if let Some(rest) = trimmed.strip_prefix('#') {
            // Consume any number of leading #s.
            let body = rest.trim_start_matches('#').trim();
            if !body.is_empty() {
                // Markdown headings can have a trailing # run; strip it.
                let clean = body.trim_end_matches('#').trim();
                if !clean.is_empty() {
                    return clean.to_string();
                }
            }
        }
    }
    path.file_stem()
        .and_then(|s| s.to_str())
        .map(|s| s.to_string())
        .unwrap_or_else(|| "(untitled)".to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::test_support::{StubVectorIndex, fixture_episode, fixture_embedding, open_test_db};
    use std::time::Duration;

    fn rt() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
    }

    #[test]
    fn remember_happy_path_round_trip() {
        let (conn, _tmp) = open_test_db();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let WriterSpawn { handle, join: _ } = WriterActor::spawn(conn, hnsw.clone());

        let episode = fixture_episode("test content");
        let embedding = fixture_embedding(4);
        let mid = rt()
            .block_on(handle.remember(episode.clone(), embedding))
            .unwrap();
        assert_eq!(mid, episode.memory_id);

        std::thread::sleep(Duration::from_millis(50));
        drop(handle);
        std::thread::sleep(Duration::from_millis(50));

        assert_eq!(hnsw.add_count(), 1);
        let added = hnsw.last_added().unwrap();
        assert_eq!(added.0, 1, "rowid should be 1 (first insert)");
        assert_eq!(added.1.len(), 4);
    }

    #[test]
    fn dispatch_remember_replies_before_drain() {
        let (conn, _tmp) = open_test_db();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let (_tx, rx) = mpsc::channel(1);
        let mut actor = WriterActor {
            conn,
            hnsw,
            rx,
            snapshot_dir: None,
            embedder_id: None,
            embedder: None,
            runtime_handle: None,
            steward: None,
            key: None,
        };
        let (reply_tx, reply_rx) = oneshot::channel();
        let episode = fixture_episode("ordering test");
        let embedding = fixture_embedding(4);

        actor.dispatch_remember(episode.clone(), embedding, reply_tx);

        let received = reply_rx.blocking_recv().unwrap();
        assert_eq!(received.unwrap(), episode.memory_id);

        let n: u32 = actor
            .conn
            .query_row("SELECT COUNT(*) FROM pending_index", [], |row| row.get(0))
            .unwrap();
        assert_eq!(n, 0);
    }

    #[test]
    fn forget_unknown_memory_id_returns_not_found() {
        let (conn, _tmp) = open_test_db();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let WriterSpawn { handle, join: _ } = WriterActor::spawn(conn, hnsw);

        let mid = MemoryId::new();
        let err = rt()
            .block_on(handle.forget(mid, "test".into()))
            .unwrap_err();
        assert!(err.to_string().contains("not found"), "got: {err}");
    }

    #[test]
    fn forget_marks_status_forgotten() {
        let (conn, _tmp) = open_test_db();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let WriterSpawn { handle, join: _ } = WriterActor::spawn(conn, hnsw.clone());

        let episode = fixture_episode("to be forgotten");
        let mid = rt()
            .block_on(handle.remember(episode.clone(), fixture_embedding(4)))
            .unwrap();
        rt().block_on(handle.forget(mid, "no longer relevant".into()))
            .unwrap();

        // Re-open the file (writer holds the DB connection; test-side reopen
        // gets read access via SQLite's WAL).
        // We can't peek into the writer's connection, so close everything and
        // re-open via open_test_db_at on the underlying path. But open_test_db
        // returned us an in-memory tmp; we need a different fixture.
        // Simpler: use the StubVectorIndex's add_count to verify the embed
        // happened, and trust that the handle.forget Ok return means the
        // UPDATE ran. The full SQL roundtrip is exercised by the
        // reader.rs::reader_sees_writes_committed_through_writer_actor test
        // pattern, which we don't replicate here.
        assert_eq!(hnsw.add_count(), 1);
        let _ = mid; // ensure we exercised the codepath; status check is in
                     // the reader-pool test below.
    }

    #[test]
    fn forget_is_idempotent_when_already_forgotten() {
        let (conn, _tmp) = open_test_db();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let WriterSpawn { handle, join: _ } = WriterActor::spawn(conn, hnsw);

        let episode = fixture_episode("forget twice");
        let mid = rt()
            .block_on(handle.remember(episode, fixture_embedding(4)))
            .unwrap();
        rt().block_on(handle.forget(mid, "first".into())).unwrap();
        // Second call: still Ok (idempotent), no error.
        rt().block_on(handle.forget(mid, "second".into())).unwrap();
    }

    #[test]
    fn many_concurrent_writes_serialize_correctly() {
        let (conn, _tmp) = open_test_db();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let WriterSpawn { handle, join: _ } = WriterActor::spawn(conn, hnsw.clone());

        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(4)
            .enable_all()
            .build()
            .unwrap();

        let results: Vec<Result<MemoryId>> = runtime.block_on(async {
            let mut tasks = Vec::new();
            for i in 0..50 {
                let h = handle.clone();
                let ep = fixture_episode(&format!("write {i}"));
                tasks.push(tokio::spawn(async move {
                    h.remember(ep, fixture_embedding(4)).await
                }));
            }
            let mut out = Vec::new();
            for t in tasks {
                out.push(t.await.unwrap());
            }
            out
        });

        let mut ids = std::collections::HashSet::new();
        for r in results {
            let mid = r.expect("remember must succeed");
            assert!(ids.insert(mid), "memory_ids must be unique");
        }
        assert_eq!(ids.len(), 50);
        assert_eq!(hnsw.add_count(), 50);
    }

    // -- normalize_subjects -------------------------------------------------
    //
    // Test pattern: drive the handle_normalize_subjects path **directly** on
    // a hand-built WriterActor. The plain `WriterActor::spawn` path owns its
    // connection on the writer thread, which would block read-back queries
    // in the same process. Direct invocation lets us:
    //
    //   1. Seed triples via the same connection the actor will mutate.
    //   2. Call `handle_normalize_subjects` on `&mut actor`.
    //   3. Query `actor.conn` afterwards to assert row contents.
    //
    // The dispatch arm itself is so thin (one match arm, one method call)
    // that exercising the method directly covers the same path the public
    // `WriteHandle::normalize_subjects` would take.

    /// Helper: seed a triple row with a given subject/object. Returns the
    /// triple_id so tests can read it back. `cluster_id` is left NULL —
    /// FK is `ON DELETE CASCADE` but NULLs don't reference anything.
    fn seed_triple(
        conn: &Connection,
        triple_id: &str,
        subject: &str,
        predicate: &str,
        object: &str,
        object_kind: &str,
    ) {
        let now_ms = chrono::Utc::now().timestamp_millis();
        conn.execute(
            "INSERT INTO triples (
                triple_id, subject_id, predicate, object_id, object_kind,
                valid_from_ms, valid_to_ms, confidence, provenance_json,
                created_at_ms, updated_at_ms
             ) VALUES (?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, ?)",
            params![
                triple_id,
                subject,
                predicate,
                object,
                object_kind,
                now_ms,
                0.9_f64,
                "{}",
                now_ms,
                now_ms,
            ],
        )
        .expect("seed triple");
    }

    /// Helper: read back `subject_id` for a known triple_id. Panics if not
    /// found — tests should always seed before reading.
    fn read_subject(conn: &Connection, triple_id: &str) -> String {
        conn.query_row(
            "SELECT subject_id FROM triples WHERE triple_id = ?",
            params![triple_id],
            |r| r.get::<_, String>(0),
        )
        .expect("read subject_id")
    }

    /// Helper: read back `object_id` for a known triple_id.
    fn read_object(conn: &Connection, triple_id: &str) -> String {
        conn.query_row(
            "SELECT object_id FROM triples WHERE triple_id = ?",
            params![triple_id],
            |r| r.get::<_, String>(0),
        )
        .expect("read object_id")
    }

    /// Helper: build a `WriterActor` directly (no spawned thread) so tests
    /// can call `handle_normalize_subjects` and then query the same
    /// connection in the same thread.
    fn build_actor_inline(conn: Connection) -> WriterActor {
        let (_tx, rx) = mpsc::channel(1);
        let hnsw = Arc::new(StubVectorIndex::new(4));
        WriterActor {
            conn,
            hnsw,
            rx,
            snapshot_dir: None,
            embedder_id: None,
            embedder: None,
            runtime_handle: None,
            steward: None,
            key: None,
        }
    }

    #[test]
    fn normalize_subjects_updates_subject_column() {
        let (conn, _tmp) = open_test_db();
        seed_triple(&conn, "t1", "alex", "uses", "rust", "literal");
        let mut actor = build_actor_inline(conn);

        let report = actor
            .handle_normalize_subjects(
                vec![("alex".into(), "user".into())],
                false,
            )
            .expect("normalize ok");

        assert_eq!(report.aliases_processed, 1);
        assert_eq!(report.subject_rows_updated, 1);
        assert_eq!(report.object_rows_updated, 0);
        assert!(!report.dry_run);
        assert_eq!(read_subject(&actor.conn, "t1"), "user");
        assert_eq!(read_object(&actor.conn, "t1"), "rust");
    }

    #[test]
    fn normalize_subjects_updates_object_column() {
        let (conn, _tmp) = open_test_db();
        // Object position: someone-uses-alex (object_kind=entity).
        seed_triple(&conn, "t1", "bob", "knows", "alex", "entity");
        let mut actor = build_actor_inline(conn);

        let report = actor
            .handle_normalize_subjects(
                vec![("alex".into(), "user".into())],
                false,
            )
            .expect("normalize ok");

        assert_eq!(report.subject_rows_updated, 0);
        assert_eq!(report.object_rows_updated, 1);
        assert_eq!(read_subject(&actor.conn, "t1"), "bob");
        assert_eq!(read_object(&actor.conn, "t1"), "user");
    }

    #[test]
    fn normalize_subjects_updates_both_when_subject_equals_object() {
        // Self-loop: subject == object == "alex". The two UPDATEs both fire
        // against the same row — count is 2 (one for each column rewrite),
        // matching what SQLite's `changes()` reports per statement.
        let (conn, _tmp) = open_test_db();
        seed_triple(&conn, "t1", "alex", "is", "alex", "entity");
        let mut actor = build_actor_inline(conn);

        let report = actor
            .handle_normalize_subjects(
                vec![("alex".into(), "user".into())],
                false,
            )
            .expect("normalize ok");

        assert_eq!(report.subject_rows_updated, 1);
        assert_eq!(report.object_rows_updated, 1);
        assert_eq!(read_subject(&actor.conn, "t1"), "user");
        assert_eq!(read_object(&actor.conn, "t1"), "user");
    }

    #[test]
    fn normalize_subjects_dry_run_rolls_back() {
        let (conn, _tmp) = open_test_db();
        seed_triple(&conn, "t1", "alex", "uses", "rust", "literal");
        seed_triple(&conn, "t2", "bob", "knows", "alex", "entity");
        let mut actor = build_actor_inline(conn);

        let report = actor
            .handle_normalize_subjects(
                vec![("alex".into(), "user".into())],
                true,
            )
            .expect("dry-run normalize ok");

        // Counts reflect would-have-been-updated:
        assert!(report.dry_run);
        assert_eq!(report.subject_rows_updated, 1);
        assert_eq!(report.object_rows_updated, 1);
        // But the rows are unchanged — transaction rolled back.
        assert_eq!(read_subject(&actor.conn, "t1"), "alex");
        assert_eq!(read_object(&actor.conn, "t1"), "rust");
        assert_eq!(read_subject(&actor.conn, "t2"), "bob");
        assert_eq!(read_object(&actor.conn, "t2"), "alex");
    }

    #[test]
    fn normalize_subjects_multiple_aliases() {
        let (conn, _tmp) = open_test_db();
        seed_triple(&conn, "t1", "alex", "uses", "rust", "literal");
        seed_triple(&conn, "t2", "bob", "uses", "python", "literal");
        seed_triple(&conn, "t3", "charlie", "knows", "alex", "entity");
        let mut actor = build_actor_inline(conn);

        let report = actor
            .handle_normalize_subjects(
                vec![
                    ("alex".into(), "user".into()),
                    ("bob".into(), "user".into()),
                ],
                false,
            )
            .expect("normalize ok");

        assert_eq!(report.aliases_processed, 2);
        // t1.subject (alex→user) + t2.subject (bob→user) = 2 subject rows
        assert_eq!(report.subject_rows_updated, 2);
        // t3.object (alex→user) = 1 object row
        assert_eq!(report.object_rows_updated, 1);

        assert_eq!(read_subject(&actor.conn, "t1"), "user");
        assert_eq!(read_subject(&actor.conn, "t2"), "user");
        assert_eq!(read_object(&actor.conn, "t3"), "user");
        // charlie (subject of t3) was not in the alias map.
        assert_eq!(read_subject(&actor.conn, "t3"), "charlie");
    }

    #[test]
    fn normalize_subjects_no_match_returns_zero_counts() {
        let (conn, _tmp) = open_test_db();
        seed_triple(&conn, "t1", "alex", "uses", "rust", "literal");
        let mut actor = build_actor_inline(conn);

        let report = actor
            .handle_normalize_subjects(
                vec![("nobody".into(), "user".into())],
                false,
            )
            .expect("normalize ok");

        assert_eq!(report.aliases_processed, 1);
        assert_eq!(report.subject_rows_updated, 0);
        assert_eq!(report.object_rows_updated, 0);
        // Existing rows untouched.
        assert_eq!(read_subject(&actor.conn, "t1"), "alex");
    }

    #[test]
    fn normalize_subjects_empty_alias_list_is_noop() {
        let (conn, _tmp) = open_test_db();
        seed_triple(&conn, "t1", "alex", "uses", "rust", "literal");
        let mut actor = build_actor_inline(conn);

        let report = actor
            .handle_normalize_subjects(vec![], false)
            .expect("normalize ok");

        assert_eq!(report.aliases_processed, 0);
        assert_eq!(report.subject_rows_updated, 0);
        assert_eq!(report.object_rows_updated, 0);
        assert_eq!(read_subject(&actor.conn, "t1"), "alex");
    }

    #[test]
    fn normalize_subjects_via_handle_round_trip() {
        // End-to-end: dispatch through `WriteHandle::normalize_subjects`
        // so we cover the variant + dispatch arm + handle method together.
        let (conn, tmp) = open_test_db();
        seed_triple(&conn, "t1", "alex", "uses", "rust", "literal");
        seed_triple(&conn, "t2", "bob", "knows", "alex", "entity");
        // Drop the seed connection so the writer's connection (opened from
        // the same path) has exclusive write access. We re-open at the end
        // to verify.
        drop(conn);
        let conn = crate::test_support::open_test_db_at(&tmp.path().join("test.db"));

        let hnsw = Arc::new(StubVectorIndex::new(4));
        let WriterSpawn { handle, join } = WriterActor::spawn(conn, hnsw);

        let report = rt()
            .block_on(handle.normalize_subjects(
                vec![("alex".into(), "user".into())],
                false,
            ))
            .expect("normalize via handle");
        assert_eq!(report.subject_rows_updated, 1);
        assert_eq!(report.object_rows_updated, 1);

        drop(handle);
        join.join().expect("writer thread joins");

        // Verify via a fresh read connection (writer's connection is now
        // closed because the actor's thread exited).
        let conn = crate::test_support::open_test_db_at(&tmp.path().join("test.db"));
        let subj: String = conn
            .query_row(
                "SELECT subject_id FROM triples WHERE triple_id = 't1'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        let obj: String = conn
            .query_row(
                "SELECT object_id FROM triples WHERE triple_id = 't2'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(subj, "user");
        assert_eq!(obj, "user");
    }

    // ====================================================================
    // v0.7.0 P3 — IngestDocument + ForgetDocument + recovery replay
    //
    // Test pattern: each test builds a WriterActor directly with a real
    // StubEmbedder + a fresh tokio runtime handle (the writer's
    // dispatch_ingest_document calls `runtime.block_on(embedder.embed_batch)`,
    // so it needs a Handle). For tests that don't call `handle.ingest_document`
    // on a separate writer thread, we drive the handler synchronously via
    // `actor.dispatch_ingest_document` after constructing the actor in-place
    // — matching the `normalize_subjects` pattern further up.
    // ====================================================================

    use crate::document::ChunkConfig;
    use crate::embedder::StubEmbedder;
    use crate::embedder_registry::{EmbedderIdentity, get_or_insert_embedder_id};
    use solo_core::{ChunkId, DocumentId, Embedder};

    /// Build an actor with a stub embedder wired up + the embedders row
    /// registered. Returns the actor plus a kept-alive runtime (drop the
    /// returned tuple together).
    fn build_ingest_actor(
        conn: Connection,
    ) -> (WriterActor, tokio::runtime::Runtime, Arc<StubVectorIndex>) {
        // Use a multi-thread runtime so the writer's block_on doesn't
        // deadlock when the calling thread happens to be a worker.
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .unwrap();
        let handle = runtime.handle().clone();
        let embedder: Arc<dyn Embedder> = Arc::new(StubEmbedder::new("stub", "v1", 4));
        let identity = EmbedderIdentity::from_embedder(embedder.as_ref());
        let embedder_id = get_or_insert_embedder_id(&conn, &identity).unwrap();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let (_tx, rx) = mpsc::channel(1);
        let actor = WriterActor {
            conn,
            hnsw: hnsw.clone(),
            rx,
            snapshot_dir: None,
            embedder_id: Some(embedder_id),
            embedder: Some(embedder),
            runtime_handle: Some(handle),
            steward: None,
            key: None,
        };
        (actor, runtime, hnsw)
    }

    /// Write a small markdown document under `tmp` and return its path.
    fn write_markdown(tmp: &tempfile::TempDir, name: &str, body: &str) -> std::path::PathBuf {
        let path = tmp.path().join(name);
        std::fs::write(&path, body).expect("write fixture");
        path
    }

    /// Embedder that always returns Err — used for the embed-failure rollback test.
    #[derive(Debug)]
    struct FailingEmbedder {
        dim: usize,
    }

    #[async_trait::async_trait]
    impl Embedder for FailingEmbedder {
        fn name(&self) -> &str {
            "fail"
        }
        fn version(&self) -> &str {
            "v1"
        }
        fn dim(&self) -> usize {
            self.dim
        }
        fn dtype(&self) -> solo_core::EmbeddingDtype {
            solo_core::EmbeddingDtype::F32
        }
        async fn embed_batch(
            &self,
            _texts: &[&str],
        ) -> Result<Vec<Embedding>> {
            Err(solo_core::Error::embedder("forced failure for test"))
        }
    }

    // ----- Ingest tests -----

    #[test]
    fn ingest_document_persists_doc_and_chunks() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(
            &docs_tmp,
            "intro.md",
            "# Intro\n\nFirst paragraph here.\n\nSecond paragraph here.\n",
        );

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path.clone(), ChunkConfig::default(), reply_tx);
        let report = reply_rx.blocking_recv().unwrap().expect("ingest ok");

        assert!(!report.deduped);
        assert_eq!(report.chunks_persisted, 1, "tiny doc → one chunk");
        assert!(report.bytes_ingested > 0);

        // documents row exists with status=active.
        let (status, title, chunk_count): (String, String, i64) = actor
            .conn
            .query_row(
                "SELECT status, title, chunk_count FROM documents WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
            )
            .unwrap();
        assert_eq!(status, "active");
        assert_eq!(title, "Intro", "first markdown heading becomes title");
        assert_eq!(chunk_count, 1);

        // document_chunks row count matches.
        let n_chunks: i64 = actor
            .conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_chunks, 1);

        // chunk_embeddings row count matches.
        let n_emb: i64 = actor
            .conn
            .query_row(
                "SELECT COUNT(*) FROM chunk_embeddings ce
                 JOIN document_chunks dc ON dc.chunk_id = ce.chunk_id
                 WHERE dc.doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_emb, 1);

        // HNSW got an add.
        assert_eq!(hnsw.add_count(), 1);
    }

    #[test]
    fn ingest_document_pending_index_drains_cleanly() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "doc.md", "# Doc\n\nBody text.\n");

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let _ = reply_rx.blocking_recv().unwrap().expect("ingest ok");

        let pending: i64 = actor
            .conn
            .query_row(
                "SELECT COUNT(*) FROM pending_index WHERE kind = 'chunk'",
                [],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(pending, 0, "pending_index chunk rows fully drained");
    }

    #[test]
    fn ingest_document_is_idempotent_by_content_hash() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "same.md", "# Same\n\nDeterministic body.\n");

        // First ingest.
        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path.clone(), ChunkConfig::default(), reply_tx);
        let report1 = reply_rx.blocking_recv().unwrap().unwrap();
        assert!(!report1.deduped);
        assert_eq!(report1.chunks_persisted, 1);
        let hnsw_after_first = hnsw.add_count();

        // Re-ingest — dedup, no new chunks, no HNSW add.
        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let report2 = reply_rx.blocking_recv().unwrap().unwrap();
        assert!(report2.deduped);
        assert_eq!(report2.doc_id, report1.doc_id);
        assert_eq!(report2.chunks_persisted, 0);
        assert_eq!(
            hnsw.add_count(),
            hnsw_after_first,
            "dedup hit must not embed or add to HNSW"
        );

        // Documents table still has exactly ONE row.
        let n_docs: i64 = actor
            .conn
            .query_row("SELECT COUNT(*) FROM documents", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n_docs, 1);
    }

    #[test]
    fn ingest_document_rolls_back_on_embed_failure() {
        // Build the actor manually with FailingEmbedder.
        let (conn, _tmp) = open_test_db();
        let runtime = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .unwrap();
        let handle = runtime.handle().clone();
        let embedder: Arc<dyn Embedder> = Arc::new(FailingEmbedder { dim: 4 });
        let identity = EmbedderIdentity::from_embedder(embedder.as_ref());
        let embedder_id = get_or_insert_embedder_id(&conn, &identity).unwrap();
        let hnsw = Arc::new(StubVectorIndex::new(4));
        let (_tx, rx) = mpsc::channel(1);
        let mut actor = WriterActor {
            conn,
            hnsw: hnsw.clone(),
            rx,
            snapshot_dir: None,
            embedder_id: Some(embedder_id),
            embedder: Some(embedder),
            runtime_handle: Some(handle),
            steward: None,
            key: None,
        };

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "fail.md", "# Fail\n\nBody.\n");

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let err = reply_rx.blocking_recv().unwrap().unwrap_err();
        assert!(err.to_string().contains("embed_batch"), "got: {err}");

        // No documents persisted (embed-before-tx ordering proves no SQL state).
        let n_docs: i64 = actor
            .conn
            .query_row("SELECT COUNT(*) FROM documents", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n_docs, 0);
        let n_chunks: i64 = actor
            .conn
            .query_row("SELECT COUNT(*) FROM document_chunks", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n_chunks, 0);
        let n_pending: i64 = actor
            .conn
            .query_row("SELECT COUNT(*) FROM pending_index", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n_pending, 0);
        // HNSW unchanged.
        assert_eq!(hnsw.add_count(), 0);
    }

    #[test]
    fn ingest_document_large_document() {
        // Force multi-chunk: ~10 paragraphs at ~50 chars each → ~500 chars
        // > target=80 → multiple chunks.
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let mut body = String::from("# Header\n\n");
        for i in 0..30 {
            body.push_str(&format!(
                "Paragraph number {i} with several words in it.\n\n"
            ));
        }
        let path = write_markdown(&docs_tmp, "big.md", &body);

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(
            path,
            ChunkConfig {
                target_tokens: 80,
                overlap_tokens: 10,
            },
            reply_tx,
        );
        let report = reply_rx.blocking_recv().unwrap().unwrap();

        assert!(
            report.chunks_persisted >= 2,
            "expected multi-chunk, got {}",
            report.chunks_persisted
        );
        let n_chunks: i64 = actor
            .conn
            .query_row(
                "SELECT COUNT(*) FROM document_chunks WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(n_chunks as u32, report.chunks_persisted);
        assert_eq!(hnsw.add_count() as u32, report.chunks_persisted);
    }

    #[test]
    fn ingest_document_uses_first_heading_as_title() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(
            &docs_tmp,
            "any_name.md",
            "Preamble line without heading.\n\n## Sub Section Title\n\nBody.\n",
        );

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let report = reply_rx.blocking_recv().unwrap().unwrap();

        let title: String = actor
            .conn
            .query_row(
                "SELECT title FROM documents WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(
            title, "Sub Section Title",
            "title comes from first heading line"
        );
    }

    #[test]
    fn ingest_document_records_file_mtime() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "mtime.md", "# T\n\nBody.\n");
        let fs_mtime_ms = std::fs::metadata(&path)
            .unwrap()
            .modified()
            .unwrap()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as i64;

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let report = reply_rx.blocking_recv().unwrap().unwrap();

        let modified_at_ms: Option<i64> = actor
            .conn
            .query_row(
                "SELECT modified_at_ms FROM documents WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        let m = modified_at_ms.expect("modified_at_ms must be set when file mtime is readable");
        // File-system mtime resolution varies; allow ±2 sec slack.
        assert!(
            (m - fs_mtime_ms).abs() < 2_000,
            "modified_at_ms drift: db={m} fs={fs_mtime_ms}"
        );
    }

    #[test]
    fn ingest_document_unsupported_extension_errors_cleanly() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = docs_tmp.path().join("blob.bin");
        std::fs::write(&path, b"\x00\x01\x02").unwrap();

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let err = reply_rx.blocking_recv().unwrap().unwrap_err();
        assert!(
            err.to_string().contains("parse") || err.to_string().contains("extension"),
            "unsupported extension should surface as a parse error: {err}"
        );
        // No SQL or HNSW state changed.
        let n_docs: i64 = actor
            .conn
            .query_row("SELECT COUNT(*) FROM documents", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n_docs, 0);
        assert_eq!(hnsw.add_count(), 0);
    }

    #[test]
    fn ingest_document_writes_embedding_dim_correctly() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "dim.md", "# Dim\n\nText.\n");

        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let report = reply_rx.blocking_recv().unwrap().unwrap();

        let dim: i64 = actor
            .conn
            .query_row(
                "SELECT ce.dim FROM chunk_embeddings ce
                 JOIN document_chunks dc ON dc.chunk_id = ce.chunk_id
                 WHERE dc.doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(dim, 4, "stub embedder dim is 4");

        let dtype: String = actor
            .conn
            .query_row(
                "SELECT ce.dtype FROM chunk_embeddings ce
                 JOIN document_chunks dc ON dc.chunk_id = ce.chunk_id
                 WHERE dc.doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(dtype, "f32");
    }

    // ----- Forget tests -----

    #[test]
    fn forget_document_sets_status_forgotten() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "f.md", "# F\n\nBody.\n");
        let (tx, rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), tx);
        let report = rx.blocking_recv().unwrap().unwrap();

        let forget_report = actor.handle_forget_document(report.doc_id).unwrap();
        assert_eq!(forget_report.doc_id, report.doc_id);
        assert_eq!(forget_report.chunks_tombstoned, report.chunks_persisted);

        let status: String = actor
            .conn
            .query_row(
                "SELECT status FROM documents WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(status, "forgotten");
    }

    #[test]
    fn forget_document_tombstones_hnsw_rowids() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "t.md", "# T\n\nBody.\n");
        let (tx, rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), tx);
        let report = rx.blocking_recv().unwrap().unwrap();

        let added_before = hnsw.add_count();
        let removed_before = hnsw.remove_count();

        let _ = actor.handle_forget_document(report.doc_id).unwrap();

        // remove_count should be at least the chunks_persisted, since one
        // hnsw.remove call per chunk fired.
        assert_eq!(
            hnsw.remove_count() - removed_before,
            report.chunks_persisted as usize
        );
        assert_eq!(hnsw.add_count(), added_before, "forget must not add");
    }

    #[test]
    fn forget_document_unknown_doc_id_returns_not_found() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

        let err = actor
            .handle_forget_document(DocumentId::new())
            .unwrap_err();
        assert!(err.to_string().contains("not found"), "got: {err}");
    }

    #[test]
    fn forget_document_idempotent() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "idem.md", "# Idem\n\nBody.\n");
        let (tx, rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), tx);
        let report = rx.blocking_recv().unwrap().unwrap();

        let r1 = actor.handle_forget_document(report.doc_id).unwrap();
        let r2 = actor.handle_forget_document(report.doc_id).unwrap();
        assert_eq!(r1.doc_id, r2.doc_id);
        assert_eq!(r1.chunks_tombstoned, r2.chunks_tombstoned);

        // Still forgotten.
        let status: String = actor
            .conn
            .query_row(
                "SELECT status FROM documents WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(status, "forgotten");
    }

    #[test]
    fn ingest_document_then_forget_then_reingest_same_content_hash_dedups_forgotten_doc() {
        // Document chosen behavior: a forgotten doc still wins content-hash
        // dedup. Re-ingest returns the SAME (forgotten) doc_id without
        // resurrecting it. Operators who want a fresh active doc must
        // ingest under a different content (or future `restore` command).
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(&docs_tmp, "fr.md", "# FR\n\nBody.\n");

        let (tx, rx) = oneshot::channel();
        actor.dispatch_ingest_document(path.clone(), ChunkConfig::default(), tx);
        let report1 = rx.blocking_recv().unwrap().unwrap();
        let _ = actor.handle_forget_document(report1.doc_id).unwrap();

        let adds_before = hnsw.add_count();
        let (tx, rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), tx);
        let report2 = rx.blocking_recv().unwrap().unwrap();

        assert!(report2.deduped, "forgotten doc still wins dedup");
        assert_eq!(report2.doc_id, report1.doc_id);
        assert_eq!(report2.chunks_persisted, 0);
        assert_eq!(
            hnsw.add_count(),
            adds_before,
            "dedup hit must not add (even though doc is forgotten)"
        );

        // Doc remains forgotten — re-ingest did NOT resurrect.
        let status: String = actor
            .conn
            .query_row(
                "SELECT status FROM documents WHERE doc_id = ?",
                params![report1.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(status, "forgotten");
    }

    // ----- Recovery replay tests -----

    /// Helper: insert a `documents` row + N `document_chunks` rows + N
    /// `pending_index` (kind='chunk') rows, and return the chunk rowids
    /// the test should expect after replay. The pending rows hold the
    /// chunks' embeddings; the chunk rows themselves carry no embedding
    /// (the chunk_embeddings table is empty in this helper — replay only
    /// reads pending_index).
    fn seed_pending_chunks(
        conn: &Connection,
        doc_id: &str,
        chunk_dim: usize,
        n: usize,
    ) -> Vec<i64> {
        let now_ms = chrono::Utc::now().timestamp_millis();
        conn.execute(
            "INSERT INTO documents (
                doc_id, source, title, mime_type,
                ingested_at_ms, modified_at_ms, status,
                chunk_count, content_hash, byte_size
             ) VALUES (?, ?, ?, ?, ?, NULL, 'active', ?, ?, ?)",
            params![
                doc_id,
                "test://source",
                "test",
                "text/plain",
                now_ms,
                n as i64,
                format!("{doc_id}_hash"),
                100i64,
            ],
        )
        .unwrap();

        let mut rowids = Vec::with_capacity(n);
        for i in 0..n {
            let chunk_id = ChunkId::new().to_string();
            conn.execute(
                "INSERT INTO document_chunks (
                    chunk_id, doc_id, chunk_index, content,
                    token_count, start_offset, end_offset, created_at_ms
                 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
                params![
                    chunk_id,
                    doc_id,
                    i as i64,
                    format!("chunk {i} text"),
                    3i64,
                    (i * 10) as i64,
                    ((i + 1) * 10) as i64,
                    now_ms,
                ],
            )
            .unwrap();
            let rowid = conn.last_insert_rowid();
            rowids.push(rowid);

            let zeros = vec![0u8; chunk_dim * 4];
            conn.execute(
                "INSERT INTO pending_index (
                    kind, chunk_id, embedding, embedding_dim, enqueued_at
                 ) VALUES ('chunk', ?, ?, ?, ?)",
                params![chunk_id, &zeros[..], chunk_dim as i64, now_ms + i as i64],
            )
            .unwrap();
        }
        rowids
    }

    #[test]
    fn recovery_replay_handles_chunk_pending_rows() {
        let (mut conn, _tmp) = open_test_db();
        let rowids = seed_pending_chunks(&conn, "11111111-1111-1111-1111-111111111111", 4, 3);

        let stub = StubVectorIndex::new(4);
        let report = crate::recovery::replay_pending_index(&mut conn, &stub).unwrap();
        assert_eq!(report.rows_seen, 3);
        assert_eq!(report.rows_replayed, 3);
        assert_eq!(report.rows_failed, 0);
        // All chunk rowids landed in HNSW, encoded with the chunk-kind
        // discriminator (high bit set) per `crate::hnsw_id`. The raw
        // rowids returned by SQL are translated through `chunk_hnsw_id`
        // by the recovery replay loop.
        let added: std::collections::HashSet<i64> =
            stub.entries().iter().map(|(r, _)| *r).collect();
        let expected: std::collections::HashSet<i64> = rowids
            .iter()
            .copied()
            .map(crate::hnsw_id::chunk_hnsw_id)
            .collect();
        assert_eq!(added, expected);
        // pending_index is drained.
        let n: i64 = conn
            .query_row("SELECT COUNT(*) FROM pending_index", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 0);
    }

    #[test]
    fn recovery_replay_handles_mixed_episode_and_chunk_rows() {
        let (mut conn, _tmp) = open_test_db();

        // Seed 2 episodes (with pending rows).
        let now_ms = chrono::Utc::now().timestamp_millis();
        let mut episode_rowids = Vec::new();
        for content in &["ep_a", "ep_b"] {
            let ep = fixture_episode(content);
            let mid = ep.memory_id.to_string();
            conn.execute(
                "INSERT INTO episodes (
                    memory_id, ts_ms, source_type, source_id, content,
                    encoding_context_json, provenance_json, confidence,
                    strength, salience, tier, created_at_ms, updated_at_ms
                 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                params![
                    mid,
                    ep.ts_ms,
                    ep.source_type,
                    ep.source_id,
                    ep.content,
                    "{}",
                    Option::<String>::None,
                    ep.confidence.0,
                    ep.strength,
                    ep.salience,
                    "hot",
                    now_ms,
                    now_ms,
                ],
            )
            .unwrap();
            episode_rowids.push(conn.last_insert_rowid());

            conn.execute(
                "INSERT INTO pending_index (kind, memory_id, embedding, embedding_dim, enqueued_at)
                 VALUES ('episode', ?, ?, ?, ?)",
                params![mid, &vec![0u8; 16][..], 4i64, now_ms],
            )
            .unwrap();
        }

        // Seed 2 chunks (with pending rows).
        let chunk_rowids =
            seed_pending_chunks(&conn, "22222222-2222-2222-2222-222222222222", 4, 2);

        let stub = StubVectorIndex::new(4);
        let report = crate::recovery::replay_pending_index(&mut conn, &stub).unwrap();
        assert_eq!(report.rows_seen, 4);
        assert_eq!(report.rows_replayed, 4);
        assert_eq!(report.rows_failed, 0);

        // Both classes of rowids landed in HNSW, each encoded with the
        // matching kind discriminator. This is the integration-level
        // anchor that recovery replay applies the correct encoder
        // per-kind: episodes via `episode_hnsw_id` (identity) and
        // chunks via `chunk_hnsw_id` (high bit set).
        let added: std::collections::HashSet<i64> =
            stub.entries().iter().map(|(r, _)| *r).collect();
        let mut expected: std::collections::HashSet<i64> = episode_rowids
            .iter()
            .copied()
            .map(crate::hnsw_id::episode_hnsw_id)
            .collect();
        expected.extend(
            chunk_rowids
                .iter()
                .copied()
                .map(crate::hnsw_id::chunk_hnsw_id),
        );
        assert_eq!(added, expected);

        // Critically: episode and chunk ids do NOT collide in the
        // HNSW namespace, even when their SQLite rowids happen to share
        // values. (In this test the AUTOINCREMENT sequences keep them
        // disjoint; the collision-free invariant comes from the
        // kind-discriminator encoding, not from rowid disjointness.)
        for r in &episode_rowids {
            for c in &chunk_rowids {
                let ep_id = crate::hnsw_id::episode_hnsw_id(*r);
                let chunk_id = crate::hnsw_id::chunk_hnsw_id(*c);
                assert_ne!(
                    ep_id, chunk_id,
                    "encoded episode and chunk ids must never collide"
                );
            }
        }

        let n: i64 = conn
            .query_row("SELECT COUNT(*) FROM pending_index", [], |r| r.get(0))
            .unwrap();
        assert_eq!(n, 0);
    }

    /// Critical regression test: an episode at `rowid=N` and a chunk at
    /// `rowid=N` (same numeric value!) must BOTH be retrievable from the
    /// shared HNSW. Without the kind-discriminated encoding, the second
    /// add would collide with the first; with `hnsw_rs` 0.3.4's silent-
    /// accept behavior the recall path would surface ambiguous results.
    ///
    /// This test simulates the production scenario by forcibly assigning
    /// chunk rowid=1 to coincide with episode rowid=1, then verifies
    /// that:
    ///   (1) Both vectors land in HNSW at distinct encoded ids.
    ///   (2) `recall` (episode side) returns the episode.
    ///   (3) `doc_search` (chunk side) returns the chunk.
    ///
    /// The simulation is done at the recovery layer rather than via
    /// AUTOINCREMENT (which in a fresh DB starts both sequences at 1
    /// anyway, so simply remembering one episode + ingesting one
    /// chunk reproduces the collision naturally).
    #[test]
    fn episode_and_chunk_with_same_rowid_coexist_in_hnsw() {
        let (conn, _tmp) = open_test_db();
        let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

        // Step 1: write one episode (assigned rowid=1 by AUTOINCREMENT).
        let ep = fixture_episode("episode body");
        let now_ms = chrono::Utc::now().timestamp_millis();
        // Use `handle_remember`-equivalent path via direct SQL + actor
        // dispatch. We hand-roll the SQL so we can assert the assigned
        // rowid is 1. (The actor's `dispatch_remember` does the same
        // INSERT under the hood.)
        let memory_id = ep.memory_id.to_string();
        actor
            .conn
            .execute(
                "INSERT INTO episodes (
                    memory_id, ts_ms, source_type, source_id, content,
                    encoding_context_json, provenance_json, confidence,
                    strength, salience, tier, created_at_ms, updated_at_ms
                 ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
                params![
                    memory_id,
                    ep.ts_ms,
                    ep.source_type,
                    ep.source_id,
                    ep.content,
                    "{}",
                    Option::<String>::None,
                    ep.confidence.0,
                    ep.strength,
                    ep.salience,
                    "hot",
                    now_ms,
                    now_ms,
                ],
            )
            .unwrap();
        let episode_rowid = actor.conn.last_insert_rowid();
        assert_eq!(episode_rowid, 1, "first episode insert must yield rowid=1");
        // Simulate the writer's HNSW add for this episode with a
        // distinctive vector.
        let ep_vec = vec![1.0f32, 0.0, 0.0, 0.0];
        hnsw.add(crate::hnsw_id::episode_hnsw_id(episode_rowid), &ep_vec)
            .unwrap();

        // Step 2: ingest a document. The first chunk gets
        // document_chunks.rowid = 1 (independent AUTOINCREMENT
        // sequence per ADR-0003 §shared-HNSW-namespace), colliding
        // numerically with the episode above.
        let docs_tmp = tempfile::TempDir::new().unwrap();
        let path = write_markdown(
            &docs_tmp,
            "doc.md",
            "# Doc\n\nSome chunk content.\n",
        );
        let (reply_tx, reply_rx) = oneshot::channel();
        actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
        let report = reply_rx.blocking_recv().unwrap().expect("ingest ok");
        assert_eq!(report.chunks_persisted, 1, "fixture produces one chunk");

        // The chunk's rowid should be 1 — same as the episode's.
        let chunk_rowid: i64 = actor
            .conn
            .query_row(
                "SELECT rowid FROM document_chunks WHERE doc_id = ?",
                params![report.doc_id.to_string()],
                |r| r.get(0),
            )
            .unwrap();
        assert_eq!(
            chunk_rowid, episode_rowid,
            "chunk rowid must collide numerically with episode rowid for this test (both AUTOINCREMENT sequences start at 1)"
        );

        // Step 3: assert HNSW carries BOTH vectors at DISTINCT encoded ids.
        let entries = hnsw.entries();
        let encoded_episode_id = crate::hnsw_id::episode_hnsw_id(episode_rowid);
        let encoded_chunk_id = crate::hnsw_id::chunk_hnsw_id(chunk_rowid);
        assert_ne!(
            encoded_episode_id, encoded_chunk_id,
            "encoded episode and chunk ids must differ even when raw rowids collide"
        );
        let ids: std::collections::HashSet<i64> =
            entries.iter().map(|(r, _)| *r).collect();
        assert!(
            ids.contains(&encoded_episode_id),
            "HNSW must carry episode at encoded id {encoded_episode_id}; entries: {entries:?}"
        );
        assert!(
            ids.contains(&encoded_chunk_id),
            "HNSW must carry chunk at encoded id {encoded_chunk_id}; entries: {entries:?}"
        );

        // Step 4: decode each entry and confirm kind/rowid round-trip.
        for (id, _) in &entries {
            let (kind, decoded) = crate::hnsw_id::decode_hnsw_id(*id);
            match kind {
                crate::hnsw_id::HnswIdKind::Episode => {
                    assert_eq!(decoded, episode_rowid);
                }
                crate::hnsw_id::HnswIdKind::Chunk => {
                    assert_eq!(decoded, chunk_rowid);
                }
            }
        }
    }

    // ----- v0.7.1: SOLO_INGEST_MAX_BYTES guardrail -----
    //
    // Process-global env-var manipulation is mutexed against itself; the
    // pattern matches `embedder::tests` so a single serialised lock keeps
    // parallel cargo-test workers from racing each other through the env.

    mod ingest_max_bytes {
        use super::*;
        use crate::writer::{
            DEFAULT_INGEST_MAX_BYTES, SOLO_INGEST_MAX_BYTES_ENV, resolve_ingest_max_bytes,
        };
        use std::sync::Mutex;

        static ENV_LOCK: Mutex<()> = Mutex::new(());

        /// Drop-guard that clears `SOLO_INGEST_MAX_BYTES` at end of scope so
        /// each test starts from the unset baseline regardless of order.
        struct EnvGuard;
        impl Drop for EnvGuard {
            fn drop(&mut self) {
                // SAFETY: caller holds `ENV_LOCK`; no concurrent env access.
                unsafe { std::env::remove_var(SOLO_INGEST_MAX_BYTES_ENV) };
            }
        }
        fn fresh_env() -> EnvGuard {
            // SAFETY: caller holds `ENV_LOCK`.
            unsafe { std::env::remove_var(SOLO_INGEST_MAX_BYTES_ENV) };
            EnvGuard
        }

        #[test]
        fn resolve_unset_returns_default() {
            let _lock = ENV_LOCK.lock().unwrap();
            let _g = fresh_env();
            assert_eq!(resolve_ingest_max_bytes(), Some(DEFAULT_INGEST_MAX_BYTES));
        }

        #[test]
        fn resolve_zero_disables_cap() {
            let _lock = ENV_LOCK.lock().unwrap();
            let _g = fresh_env();
            // SAFETY: ENV_LOCK held.
            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "0") };
            assert_eq!(resolve_ingest_max_bytes(), None);
        }

        #[test]
        fn resolve_positive_integer_uses_value() {
            let _lock = ENV_LOCK.lock().unwrap();
            let _g = fresh_env();
            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "1024") };
            assert_eq!(resolve_ingest_max_bytes(), Some(1024));
        }

        #[test]
        fn resolve_garbage_falls_back_to_default() {
            let _lock = ENV_LOCK.lock().unwrap();
            let _g = fresh_env();
            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "not-a-number") };
            assert_eq!(resolve_ingest_max_bytes(), Some(DEFAULT_INGEST_MAX_BYTES));

            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "-1") };
            assert_eq!(resolve_ingest_max_bytes(), Some(DEFAULT_INGEST_MAX_BYTES));

            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "  ") };
            assert_eq!(resolve_ingest_max_bytes(), Some(DEFAULT_INGEST_MAX_BYTES));
        }

        #[test]
        fn ingest_rejects_oversized_file_with_clear_error() {
            let _lock = ENV_LOCK.lock().unwrap();
            let _g = fresh_env();
            // 100-byte file but cap is 1 byte → reject.
            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "1") };

            let (conn, _tmp) = open_test_db();
            let (mut actor, _rt, hnsw) = build_ingest_actor(conn);

            let docs_tmp = tempfile::TempDir::new().unwrap();
            let path = write_markdown(
                &docs_tmp,
                "big.md",
                "# Big\n\nThis is well over a single byte of content text.\n",
            );

            let (reply_tx, reply_rx) = oneshot::channel();
            actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
            let err = reply_rx.blocking_recv().unwrap().unwrap_err();
            let msg = err.to_string();
            assert!(
                msg.contains("SOLO_INGEST_MAX_BYTES")
                    && msg.contains("exceeds")
                    && msg.contains("disable"),
                "rejection message must call out the env var, threshold, and disable hint; got: {msg}"
            );
            // Zero SQL state, zero HNSW state.
            let n_docs: i64 = actor
                .conn
                .query_row("SELECT COUNT(*) FROM documents", [], |r| r.get(0))
                .unwrap();
            assert_eq!(n_docs, 0);
            assert_eq!(hnsw.add_count(), 0);
        }

        #[test]
        fn ingest_allows_undersized_file_under_custom_cap() {
            let _lock = ENV_LOCK.lock().unwrap();
            let _g = fresh_env();
            // 4 KiB cap; tiny doc is well under.
            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "4096") };

            let (conn, _tmp) = open_test_db();
            let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

            let docs_tmp = tempfile::TempDir::new().unwrap();
            let path = write_markdown(&docs_tmp, "ok.md", "# OK\n\nShort body.\n");

            let (reply_tx, reply_rx) = oneshot::channel();
            actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
            let report = reply_rx.blocking_recv().unwrap().expect("ingest under cap must succeed");
            assert!(!report.deduped);
            assert_eq!(report.chunks_persisted, 1);
        }

        #[test]
        fn ingest_with_cap_zero_allows_any_size() {
            let _lock = ENV_LOCK.lock().unwrap();
            let _g = fresh_env();
            // Disable cap entirely.
            unsafe { std::env::set_var(SOLO_INGEST_MAX_BYTES_ENV, "0") };

            let (conn, _tmp) = open_test_db();
            let (mut actor, _rt, _hnsw) = build_ingest_actor(conn);

            let docs_tmp = tempfile::TempDir::new().unwrap();
            // Ingest succeeds even though body is non-trivial.
            let path = write_markdown(
                &docs_tmp,
                "any.md",
                "# Any\n\nWith SOLO_INGEST_MAX_BYTES=0 any size is allowed.\n",
            );

            let (reply_tx, reply_rx) = oneshot::channel();
            actor.dispatch_ingest_document(path, ChunkConfig::default(), reply_tx);
            let report = reply_rx.blocking_recv().unwrap().expect("cap=0 disables cap");
            assert_eq!(report.chunks_persisted, 1);
        }
    }
}