//! Completed-source manifests and resumable staging. Legacy generations stay
//! searchable but cannot supply vectors without explicit completion evidence.
use super::*;

#[derive(Debug, Clone)]
pub(crate) struct SourceTask {
    pub key: String,
    pub path: String,
    /// `text` for one extracted passage, `media` for the entire media file.
    pub source_type: String,
    pub source_hash: String,
    pub title_hash: String,
    pub policy: String,
    pub passage_id: i64,
}

pub(super) fn migrate(connection: &Connection) -> Result<()> {
    for (table, name, declaration) in [
        (
            "semantic_generations",
            "incremental",
            "INTEGER NOT NULL DEFAULT 0",
        ),
        ("semantic_chunks", "task_key", "TEXT"),
        (
            "semantic_chunks",
            "file_change_revision",
            "INTEGER NOT NULL DEFAULT 0",
        ),
    ] {
        let exists: bool = connection.query_row(
            "SELECT EXISTS(SELECT 1 FROM pragma_table_info(?1) WHERE name=?2)",
            params![table, name],
            |r| r.get(0),
        )?;
        if !exists {
            connection.execute_batch(&format!(
                "ALTER TABLE {table} ADD COLUMN {name} {declaration}"
            ))?;
        }
    }
    connection.execute_batch("CREATE TABLE IF NOT EXISTS semantic_tasks(
        generation INTEGER NOT NULL REFERENCES semantic_generations(id) ON DELETE CASCADE,
        task_key TEXT NOT NULL,path TEXT NOT NULL,source_type TEXT NOT NULL,
        source_hash TEXT NOT NULL,title_hash TEXT NOT NULL,policy TEXT NOT NULL,
        passage_id INTEGER NOT NULL,state TEXT NOT NULL CHECK(state IN ('pending','running','done','failed')),
        attempts INTEGER NOT NULL DEFAULT 0,expected_chunks INTEGER,message TEXT,
        PRIMARY KEY(generation,task_key));
        CREATE INDEX IF NOT EXISTS semantic_task_reuse ON semantic_tasks(source_hash,title_hash,policy,source_type,state);
        CREATE INDEX IF NOT EXISTS semantic_chunk_task ON semantic_chunks(generation,task_key);
        UPDATE semantic_meta SET schema_version=8 WHERE singleton=1;")?;
    Ok(())
}

pub(super) fn validate_completion(connection: &Connection, generation: i64) -> Result<()> {
    let incomplete: bool = connection.query_row("SELECT EXISTS(SELECT 1 FROM semantic_tasks t WHERE generation=?1 AND
        (state NOT IN ('done','failed') OR (state='done' AND (expected_chunks IS NULL OR expected_chunks!=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key))) OR
        (state='failed' AND EXISTS(SELECT 1 FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key))))",[generation], |r|r.get(0))?;
    ensure!(
        !incomplete,
        "Semantic generation contains incomplete source tasks"
    );
    let unowned: bool = connection.query_row("SELECT EXISTS(SELECT 1 FROM semantic_chunks c LEFT JOIN semantic_tasks t ON t.generation=c.generation AND t.task_key=c.task_key WHERE c.generation=?1 AND (t.state IS NULL OR t.state!='done'))",[generation],|r|r.get(0))?;
    ensure!(
        !unowned,
        "Semantic generation contains chunks without completed sources"
    );
    Ok(())
}

impl Store {
    pub(crate) fn revision(&self) -> Result<i64> {
        Ok(self.connection.query_row(
            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
            [],
            |r| r.get(0),
        )?)
    }

    /// Caller must retain the OS lock through reconciliation and publication.
    pub(crate) fn start_or_resume(
        &mut self,
        fingerprint: &str,
        lock: &BuildLock,
    ) -> Result<(i64, bool)> {
        ensure!(
            lock.1 == self.lock_path,
            "Build lock belongs to a different index"
        );
        ensure!(
            !fingerprint.is_empty() && fingerprint.len() <= 4096,
            "Invalid generation fingerprint"
        );
        let tx = self
            .connection
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let generation: Option<i64>=tx.query_row("SELECT id FROM semantic_generations WHERE id!=COALESCE((SELECT active_generation FROM semantic_meta WHERE singleton=1),-1) AND fingerprint=?1 AND dimensions=?2 AND incremental=1 ORDER BY id DESC LIMIT 1",params![fingerprint,self.dimensions as i64],|r|r.get(0)).optional()?;
        let resumed = generation.is_some();
        let id = if let Some(id) = generation {
            id
        } else {
            tx.execute("INSERT INTO semantic_generations(fingerprint,dimensions,corpus_revision,incremental) SELECT ?1,?2,corpus_revision,1 FROM semantic_meta WHERE singleton=1",params![fingerprint,self.dimensions as i64])?;
            tx.last_insert_rowid()
        };
        tx.execute("DELETE FROM semantic_generations WHERE id!=?1 AND id!=COALESCE((SELECT active_generation FROM semantic_meta WHERE singleton=1),-1)",[id])?;
        // Untagged chunks are writes interrupted before the source transaction.
        tx.execute("UPDATE semantic_tasks SET state='pending',expected_chunks=NULL WHERE generation=?1 AND state='done' AND
            (expected_chunks IS NULL OR expected_chunks!=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=semantic_tasks.generation AND c.task_key=semantic_tasks.task_key))",[id])?;
        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND NOT EXISTS(SELECT 1 FROM semantic_tasks t WHERE t.generation=semantic_chunks.generation AND t.task_key=semantic_chunks.task_key AND t.state='done')",[id])?;
        tx.execute("UPDATE semantic_tasks SET state='pending',expected_chunks=NULL,message=NULL WHERE generation=?1 AND state!='done'",[id])?;
        tx.commit()?;
        Ok((id, resumed))
    }

    pub(crate) fn stage_task(&mut self, generation: i64, task: &SourceTask) -> Result<bool> {
        validate_task(task)?;
        let tx = self.connection.transaction()?;
        let same:bool=tx.query_row("SELECT EXISTS(SELECT 1 FROM semantic_tasks WHERE generation=?1 AND task_key=?2 AND source_hash=?3 AND title_hash=?4 AND policy=?5 AND source_type=?6 AND state='done')",params![generation,task.key,task.source_hash,task.title_hash,task.policy,task.source_type],|r|r.get(0))?;
        if !same {
            tx.execute(
                "DELETE FROM semantic_chunks WHERE generation=?1 AND task_key=?2",
                params![generation, task.key],
            )?;
            tx.execute("INSERT INTO semantic_tasks(generation,task_key,path,source_type,source_hash,title_hash,policy,passage_id,state,attempts)
                VALUES(?1,?2,?3,?4,?5,?6,?7,?8,'running',1)
                ON CONFLICT(generation,task_key) DO UPDATE SET path=excluded.path,source_type=excluded.source_type,source_hash=excluded.source_hash,title_hash=excluded.title_hash,policy=excluded.policy,passage_id=excluded.passage_id,state='running',attempts=semantic_tasks.attempts+1,expected_chunks=NULL,message=NULL",params![generation,task.key,task.path,task.source_type,task.source_hash,task.title_hash,task.policy,task.passage_id])?;
        }
        tx.commit()?;
        Ok(same)
    }

    pub(crate) fn complete_source(
        &mut self,
        generation: i64,
        task: &SourceTask,
        expected_chunks: usize,
    ) -> Result<()> {
        ensure!(
            !task.source_hash.is_empty(),
            "Cannot complete a source without verified identity"
        );
        ensure!(
            expected_chunks <= MAX_SCAN_CHUNKS,
            "Source exceeds semantic chunk budget"
        );
        let tx = self
            .connection
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let count:i64=tx.query_row("SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND path=?2 AND ((?3='text' AND source_type='text' AND passage_id=?4) OR (?3='media' AND source_type!='text'))",params![generation,task.path,task.source_type,task.passage_id],|r|r.get(0))?;
        ensure!(
            usize::try_from(count)? == expected_chunks,
            "Incomplete source chunk membership"
        );
        let total: usize = tx.query_row(
            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
            [generation],
            |r| r.get(0),
        )?;
        ensure!(
            total <= MAX_SCAN_CHUNKS,
            "Generation exceeds semantic chunk budget"
        );
        tx.execute("UPDATE semantic_chunks SET task_key=?5 WHERE generation=?1 AND path=?2 AND ((?3='text' AND source_type='text' AND passage_id=?4) OR (?3='media' AND source_type!='text'))",params![generation,task.path,task.source_type,task.passage_id,task.key])?;
        let changed=tx.execute("UPDATE semantic_tasks SET state='done',expected_chunks=?3,path=?4,passage_id=?5,message=NULL WHERE generation=?1 AND task_key=?2 AND source_hash=?6 AND title_hash=?7 AND policy=?8 AND source_type=?9",params![generation,task.key,count,task.path,task.passage_id,task.source_hash,task.title_hash,task.policy,task.source_type])?;
        ensure!(changed == 1, "Missing matching source task");
        if task.source_type == "media" {
            tx.execute(
                "DELETE FROM semantic_media_errors WHERE generation=?1 AND path=?2",
                params![generation, task.path],
            )?;
        }
        tx.commit()?;
        Ok(())
    }

    pub(crate) fn fail_source(
        &mut self,
        generation: i64,
        task: &SourceTask,
        message: &str,
    ) -> Result<()> {
        let tx = self.connection.transaction()?;
        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND (task_key=?2 OR (path=?3 AND ((?4='text' AND source_type='text' AND passage_id=?5) OR (?4='media' AND source_type!='text'))))",params![generation,task.key,task.path,task.source_type,task.passage_id])?;
        tx.execute("UPDATE semantic_tasks SET state='failed',expected_chunks=NULL,message=?3 WHERE generation=?1 AND task_key=?2",params![generation,task.key,message.chars().take(2048).collect::<String>()])?;
        tx.commit()?;
        Ok(())
    }

    /// Reconcile completed staging groups before any encoder call. A bounded
    /// SQL snapshot preserves renamed donors (including swaps and duplicates)
    /// while rebuilding only current compatible memberships. Active vectors
    /// are never touched. Full range/vector validation still happens on reuse.
    pub(crate) fn prune_before_inference(
        &mut self,
        generation: i64,
        tasks: &[SourceTask],
        revision: i64,
    ) -> Result<()> {
        ensure!(
            tasks.len() <= MAX_SCAN_CHUNKS,
            "Generation exceeds source task budget"
        );
        for task in tasks {
            validate_task(task)?;
        }
        // The bounded snapshot can still contain 100K x 768d vectors. Keep it
        // SQLite-backed on disk rather than turning that into a Rust/heap copy.
        self.connection.pragma_update(None, "temp_store", "FILE")?;
        let tx = self
            .connection
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let current: i64 = tx.query_row(
            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
            [],
            |r| r.get(0),
        )?;
        ensure!(
            current == revision,
            "Indexed corpus changed during semantic discovery; retry required"
        );
        let active: bool = tx.query_row(
            "SELECT active_generation=?1 FROM semantic_meta WHERE singleton=1",
            [generation],
            |r| r.get::<_, Option<bool>>(0).map(|v| v.unwrap_or(false)),
        )?;
        ensure!(!active, "Cannot prune the active semantic generation");
        let existing: usize = tx.query_row(
            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
            [generation],
            |r| r.get(0),
        )?;
        ensure!(
            existing <= MAX_SCAN_CHUNKS,
            "Staging generation exceeds semantic chunk budget"
        );
        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_discovery_tasks(task_key TEXT PRIMARY KEY,path TEXT NOT NULL,source_type TEXT NOT NULL,source_hash TEXT NOT NULL,title_hash TEXT NOT NULL,policy TEXT NOT NULL,passage_id INTEGER NOT NULL);
            DELETE FROM semantic_discovery_tasks;
            CREATE TEMP TABLE IF NOT EXISTS semantic_prune_tasks AS SELECT * FROM semantic_tasks WHERE 0;
            CREATE TEMP TABLE IF NOT EXISTS semantic_prune_chunks AS SELECT * FROM semantic_chunks WHERE 0;
            DELETE FROM semantic_prune_tasks; DELETE FROM semantic_prune_chunks;")?;
        for task in tasks {
            tx.execute(
                "INSERT INTO semantic_discovery_tasks VALUES(?1,?2,?3,?4,?5,?6,?7)",
                params![
                    task.key,
                    task.path,
                    task.source_type,
                    task.source_hash,
                    task.title_hash,
                    task.policy,
                    task.passage_id
                ],
            )?;
        }
        tx.execute("INSERT INTO semantic_prune_tasks SELECT * FROM semantic_tasks t WHERE generation=?1 AND state='done' AND source_hash!='' AND expected_chunks IS NOT NULL AND expected_chunks=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key)",[generation])?;
        tx.execute(
            "INSERT INTO semantic_prune_chunks SELECT * FROM semantic_chunks WHERE generation=?1",
            [generation],
        )?;
        tx.execute_batch("CREATE UNIQUE INDEX IF NOT EXISTS temp.semantic_prune_task_key ON semantic_prune_tasks(task_key);
            CREATE INDEX IF NOT EXISTS temp.semantic_prune_reuse ON semantic_prune_tasks(source_hash,title_hash,policy,source_type);
            CREATE INDEX IF NOT EXISTS temp.semantic_prune_membership ON semantic_prune_chunks(task_key);")?;
        let mut mappings = Vec::new();
        {
            let mut statement = tx.prepare("SELECT d.task_key,COALESCE((SELECT old.task_key FROM semantic_prune_tasks old WHERE old.task_key=d.task_key AND old.source_hash=d.source_hash AND old.title_hash=d.title_hash AND old.policy=d.policy AND old.source_type=d.source_type),(SELECT old.task_key FROM semantic_prune_tasks old WHERE old.source_hash=d.source_hash AND old.title_hash=d.title_hash AND old.policy=d.policy AND old.source_type=d.source_type ORDER BY old.task_key LIMIT 1)) FROM semantic_discovery_tasks d")?;
            let rows = statement.query_map([], |r| {
                Ok((r.get::<_, String>(0)?, r.get::<_, Option<String>>(1)?))
            })?;
            for row in rows {
                let (key, donor) = row?;
                if let Some(donor) = donor {
                    mappings.push((key, donor));
                }
            }
        }
        let mut total = 0usize;
        for (_, donor) in &mappings {
            let count: usize = tx.query_row(
                "SELECT expected_chunks FROM semantic_prune_tasks WHERE task_key=?1",
                [donor],
                |r| r.get(0),
            )?;
            total = total
                .checked_add(count)
                .context("Semantic chunk count overflow")?;
            ensure!(
                total <= MAX_SCAN_CHUNKS,
                "Current completed sources exceed semantic chunk budget"
            );
        }
        tx.execute(
            "DELETE FROM semantic_chunks WHERE generation=?1",
            [generation],
        )?;
        tx.execute(
            "DELETE FROM semantic_tasks WHERE generation=?1",
            [generation],
        )?;
        for (key, donor) in mappings {
            tx.execute("INSERT INTO semantic_tasks(generation,task_key,path,source_type,source_hash,title_hash,policy,passage_id,state,attempts,expected_chunks,message)
                SELECT ?1,d.task_key,d.path,d.source_type,d.source_hash,d.title_hash,d.policy,d.passage_id,'done',old.attempts,old.expected_chunks,NULL FROM semantic_discovery_tasks d JOIN semantic_prune_tasks old ON old.task_key=?3 WHERE d.task_key=?2",params![generation,key,donor])?;
            tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key,file_change_revision)
                SELECT ?1,d.passage_id,c.ordinal,d.path,c.kind,c.number,c.content_hash,c.start_byte,c.end_byte,c.snippet,c.vector,c.source_type,c.file_size,c.file_mtime,d.task_key,COALESCE((SELECT change_revision FROM files WHERE path=d.path),c.file_change_revision) FROM semantic_prune_chunks c JOIN semantic_discovery_tasks d ON d.task_key=?2 WHERE c.task_key=?3",params![generation,key,donor])?;
        }
        tx.execute(
            "DELETE FROM semantic_media_errors WHERE generation=?1",
            [generation],
        )?;
        tx.execute_batch("DELETE FROM semantic_prune_tasks; DELETE FROM semantic_prune_chunks; DELETE FROM semantic_discovery_tasks;")?;
        tx.commit()?;
        Ok(())
    }

    pub(crate) fn reconcile_tasks(
        &mut self,
        generation: i64,
        keys: &[String],
        revision: i64,
    ) -> Result<()> {
        ensure!(
            keys.len() <= MAX_SCAN_CHUNKS,
            "Generation exceeds source task budget"
        );
        let tx = self
            .connection
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let current: i64 = tx.query_row(
            "SELECT corpus_revision FROM semantic_meta WHERE singleton=1",
            [],
            |r| r.get(0),
        )?;
        ensure!(
            current == revision,
            "Indexed corpus changed during semantic reconciliation; retry required"
        );
        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_current_tasks(task_key TEXT PRIMARY KEY); DELETE FROM semantic_current_tasks;")?;
        for key in keys {
            tx.execute(
                "INSERT OR IGNORE INTO semantic_current_tasks VALUES(?1)",
                [key],
            )?;
        }
        tx.execute("DELETE FROM semantic_chunks WHERE generation=?1 AND (task_key IS NULL OR task_key NOT IN (SELECT task_key FROM semantic_current_tasks))",[generation])?;
        tx.execute("DELETE FROM semantic_tasks WHERE generation=?1 AND task_key NOT IN (SELECT task_key FROM semantic_current_tasks)",[generation])?;
        tx.execute("DELETE FROM semantic_media_errors WHERE generation=?1 AND path NOT IN (SELECT path FROM semantic_tasks WHERE generation=?1)",[generation])?;
        tx.execute(
            "UPDATE semantic_generations SET corpus_revision=?2 WHERE id=?1",
            params![generation, revision],
        )?;
        tx.commit()?;
        Ok(())
    }

    #[cfg(test)]
    pub(crate) fn activate_checked(&mut self, generation: i64, revision: i64) -> Result<()> {
        self.activate_checked_with_validation(generation, revision, || Ok(()))
    }

    /// Validate live sources after graph preparation, outside the SQLite write
    /// transaction. Rejection preserves both the active reader and staging.
    pub(crate) fn activate_checked_with_validation(
        &mut self,
        generation: i64,
        revision: i64,
        validate_sources: impl FnOnce() -> Result<()>,
    ) -> Result<()> {
        validate_completion(&self.connection, generation)?;
        self.activate_revision(generation, Some(revision), validate_sources)
    }
}

fn validate_task(task: &SourceTask) -> Result<()> {
    ensure!(
        !task.key.is_empty()
            && task.key.len() <= 256
            && !task.path.is_empty()
            && task.path.len() <= 32768,
        "Invalid source task identity"
    );
    ensure!(
        matches!(task.source_type.as_str(), "text" | "media"),
        "Invalid source task type"
    );
    ensure!(
        (!task.source_hash.is_empty() || task.source_type == "media")
            && task.source_hash.len() <= 256
            && task.title_hash.len() <= 256
            && !task.policy.is_empty()
            && task.policy.len() <= 4096,
        "Invalid source completion policy"
    );
    Ok(())
}

impl Store {
    fn reusable_task(
        &self,
        generation: i64,
        task: &SourceTask,
    ) -> Result<Option<(i64, String, usize)>> {
        Ok(self.connection.query_row("SELECT t.generation,t.task_key,t.expected_chunks FROM semantic_tasks t JOIN semantic_generations g ON g.id=t.generation JOIN semantic_generations destination ON destination.id=?1
            WHERE t.source_hash=?2 AND t.title_hash=?3 AND t.policy=?4 AND t.source_type=?5 AND t.state='done' AND t.expected_chunks IS NOT NULL
            AND g.fingerprint=destination.fingerprint AND g.dimensions=destination.dimensions
            AND t.expected_chunks=(SELECT count(*) FROM semantic_chunks c WHERE c.generation=t.generation AND c.task_key=t.task_key)
            ORDER BY (t.generation=?1) DESC,t.generation DESC LIMIT 1",params![generation,task.source_hash,task.title_hash,task.policy,task.source_type],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?)
    }

    /// No tokenizer or neural call: only a completed, policy-compatible group
    /// whose persisted ranges still equal the current extracted bytes qualifies.
    pub(crate) fn reuse_text_source(
        &mut self,
        generation: i64,
        task: &SourceTask,
        passage: &IndexedPassage,
    ) -> Result<Option<usize>> {
        ensure!(
            task.source_type == "text"
                && task.passage_id == passage.id
                && task.path == passage.path,
            "Text task mapping differs from current source"
        );
        ensure!(
            task.source_hash == content_hash(&passage.text),
            "Text task hash differs from source bytes"
        );
        let Some((source, key, count)) = self.reusable_task(generation, task)? else {
            return Ok(None);
        };
        ensure!(
            count <= MAX_SCAN_CHUNKS,
            "Reusable source exceeds chunk budget"
        );
        {
            let mut stmt=self.connection.prepare("SELECT start_byte,end_byte,snippet,vector,source_type,ordinal,content_hash FROM semantic_chunks WHERE generation=?1 AND task_key=?2 ORDER BY ordinal")?;
            let mut rows = stmt.query(params![source, key])?;
            let mut seen = 0usize;
            while let Some(row) = rows.next()? {
                let start: usize = row.get(0)?;
                let end: usize = row.get(1)?;
                let snippet: String = row.get(2)?;
                ensure!(
                    passage.text.get(start..end) == Some(snippet.as_str()),
                    "Reusable text range differs from current source"
                );
                ensure!(
                    row.get::<_, String>(4)? == "text"
                        && row.get::<_, usize>(5)? == seen
                        && row.get::<_, String>(6)? == task.source_hash,
                    "Corrupt completed text source membership"
                );
                validate_vector(&row.get::<_, Vec<u8>>(3)?, self.dimensions)?;
                seen += 1;
            }
            ensure!(seen == count, "Reusable text completion count differs");
        }
        self.copy_completed(
            generation,
            task,
            source,
            &key,
            count,
            Some(&passage.source),
            None,
        )?;
        Ok(Some(count))
    }

    pub(crate) fn reuse_media_source(
        &mut self,
        generation: i64,
        task: &SourceTask,
        file_id: i64,
        identity: &searchi_media::MediaIdentity,
    ) -> Result<Option<usize>> {
        ensure!(
            task.source_type == "media"
                && task.passage_id == -file_id
                && task.path == identity.path.to_string_lossy()
                && task.source_hash == identity.sha256,
            "Media task mapping differs from current source"
        );
        let Some((source, key, count)) = self.reusable_task(generation, task)? else {
            return Ok(None);
        };
        ensure!(
            count <= MAX_SCAN_CHUNKS,
            "Reusable source exceeds chunk budget"
        );
        {
            let mut stmt=self.connection.prepare("SELECT vector,source_type,ordinal,content_hash FROM semantic_chunks WHERE generation=?1 AND task_key=?2 ORDER BY ordinal")?;
            let mut rows = stmt.query(params![source, key])?;
            let mut seen = 0usize;
            while let Some(row) = rows.next()? {
                ensure!(
                    matches!(
                        row.get::<_, String>(1)?.as_str(),
                        "image" | "audio" | "video"
                    ) && row.get::<_, usize>(2)? == seen
                        && row.get::<_, String>(3)? == task.source_hash,
                    "Corrupt completed media source membership"
                );
                validate_vector(&row.get::<_, Vec<u8>>(0)?, self.dimensions)?;
                seen += 1;
            }
            ensure!(seen == count, "Reusable media completion count differs");
        }
        self.copy_completed(generation, task, source, &key, count, None, Some(identity))?;
        Ok(Some(count))
    }

    #[allow(clippy::too_many_arguments)]
    fn copy_completed(
        &mut self,
        generation: i64,
        task: &SourceTask,
        source: i64,
        key: &str,
        count: usize,
        location: Option<&SourceLocation>,
        identity: Option<&searchi_media::MediaIdentity>,
    ) -> Result<()> {
        let tx = self
            .connection
            .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
        let total:usize=tx.query_row("SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND (task_key IS NULL OR task_key!=?2)",params![generation,task.key],|r|r.get(0))?;
        ensure!(
            total.saturating_add(count) <= MAX_SCAN_CHUNKS,
            "Generation exceeds semantic chunk budget"
        );
        // A temporary SQL table allows remapping a completed staging task onto
        // itself without retaining its old citations or allocating all vectors.
        tx.execute_batch("CREATE TEMP TABLE IF NOT EXISTS semantic_reuse_chunks AS SELECT * FROM semantic_chunks WHERE 0; DELETE FROM semantic_reuse_chunks;")?;
        tx.execute("INSERT INTO semantic_reuse_chunks SELECT * FROM semantic_chunks WHERE generation=?1 AND task_key=?2",params![source,key])?;
        tx.execute(
            "DELETE FROM semantic_chunks WHERE generation=?1 AND task_key=?2",
            params![generation, task.key],
        )?;
        tx.execute("INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key,file_change_revision)
            SELECT ?1,?2,ordinal,?3,COALESCE(?4,kind),COALESCE(?5,number),content_hash,start_byte,end_byte,snippet,vector,source_type,COALESCE(?6,file_size),COALESCE(?7,file_mtime),?8,COALESCE((SELECT change_revision FROM files WHERE path=?3),file_change_revision) FROM semantic_reuse_chunks",params![generation,task.passage_id,task.path,location.map(|s|s.kind.as_str()),location.map(|s|s.number),identity.map(|i|i.bytes as i64),identity.map(|i|i.modified_ns),task.key])?;
        let changed=tx.execute("UPDATE semantic_tasks SET state='done',expected_chunks=?3,path=?4,passage_id=?5,message=NULL WHERE generation=?1 AND task_key=?2 AND source_hash=?6 AND title_hash=?7 AND policy=?8 AND source_type=?9",params![generation,task.key,count as i64,task.path,task.passage_id,task.source_hash,task.title_hash,task.policy,task.source_type])?;
        ensure!(changed == 1, "Missing source task during reuse");
        if task.source_type == "media" {
            tx.execute(
                "DELETE FROM semantic_media_errors WHERE generation=?1 AND path=?2",
                params![generation, task.path],
            )?;
        }
        tx.execute_batch("DELETE FROM semantic_reuse_chunks;")?;
        tx.commit()?;
        Ok(())
    }
}

fn validate_vector(bytes: &[u8], dimensions: usize) -> Result<()> {
    ensure!(
        bytes.len() == dimensions * 4,
        "Corrupt reusable vector length"
    );
    let mut norm = 0.0f32;
    for value in bytes.as_chunks::<4>().0 {
        let value = f32::from_le_bytes(*value);
        ensure!(value.is_finite(), "Corrupt reusable vector value");
        norm += value * value;
    }
    ensure!(
        norm.is_finite() && (norm.sqrt() - 1.0).abs() < 1e-3,
        "Reusable vector is not normalized"
    );
    Ok(())
}
#[cfg(test)]
mod tests {
    use super::*;
    use searchi_core::SearchIndex;

    // Storage-only vectors exercise durability and membership. They are not
    // simulated production inference or evidence of semantic relevance.
    fn vector() -> Vec<f32> {
        vec![1.0, 0.0]
    }
    fn task(p: &IndexedPassage) -> SourceTask {
        SourceTask {
            key: format!("text-{}-{}", p.path, p.source.number),
            path: p.path.clone(),
            source_type: "text".into(),
            source_hash: content_hash(&p.text),
            title_hash: content_hash(&p.filename),
            policy: "storage-test-window-policy".into(),
            passage_id: p.id,
        }
    }
    fn insert_complete(
        store: &mut Store,
        g: i64,
        p: &IndexedPassage,
        t: &SourceTask,
    ) -> Result<()> {
        store.stage_task(g, t)?;
        store.insert(
            g,
            p,
            0,
            &[TextWindow {
                text: p.text.clone(),
                start_byte: 0,
                end_byte: p.text.len(),
            }],
            &[vector()],
        )?;
        store.complete_source(g, t, 1)
    }

    #[test]
    fn pre_inference_pruning_preserves_renamed_donors_at_the_chunk_ceiling() -> Result<()> {
        // Actual 100K membership boundary; vectors are storage-only fixtures.
        let dir = tempfile::tempdir()?;
        let root = dir.path().join("docs");
        std::fs::create_dir(&root)?;
        std::fs::write(root.join("notes.txt"), "complete source")?;
        let db = dir.path().join("index.db");
        let mut lexical = SearchIndex::open(&db)?;
        lexical.index_folder(&root)?;
        let p = lexical.passage_batch(0, 1)?.remove(0);
        let mut store = Store::open(&db, 2)?;
        let lock = store.build_lock()?;
        let active = store.start_or_resume("prune-policy", &lock)?.0;
        insert_complete(&mut store, active, &p, &task(&p))?;
        store.activate_checked(active, store.revision()?)?;
        let staging = store.start_or_resume("prune-policy", &lock)?.0;
        let original = task(&p);
        insert_complete(&mut store, staging, &p, &original)?;
        store.connection.execute("WITH RECURSIVE ord(n) AS (SELECT 1 UNION ALL SELECT n+1 FROM ord WHERE n+1<?2)
            INSERT INTO semantic_chunks(generation,passage_id,ordinal,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key)
            SELECT generation,passage_id,n,path,kind,number,content_hash,start_byte,end_byte,snippet,vector,source_type,file_size,file_mtime,task_key FROM semantic_chunks CROSS JOIN ord WHERE generation=?1 AND ordinal=0",params![staging,MAX_SCAN_CHUNKS as i64])?;
        store.complete_source(staging, &original, MAX_SCAN_CHUNKS)?;
        let moved_dir = root.join("moved");
        std::fs::create_dir(&moved_dir)?;
        std::fs::rename(root.join("notes.txt"), moved_dir.join("notes.txt"))?;
        lexical.index_folder(&root)?;
        let moved = lexical.passage_batch(0, 1)?.remove(0);
        let moved_task = task(&moved);
        let revision = store.revision()?;
        assert!(store
            .prune_before_inference(staging, std::slice::from_ref(&moved_task), revision - 1)
            .is_err());
        let old_count: usize = store.connection.query_row(
            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
            [staging],
            |r| r.get(0),
        )?;
        assert_eq!(
            old_count, MAX_SCAN_CHUNKS,
            "Revision failure must preserve staging"
        );
        store.prune_before_inference(staging, std::slice::from_ref(&moved_task), revision)?;
        store.stage_task(staging, &moved_task)?;
        assert_eq!(
            store.reuse_text_source(staging, &moved_task, &moved)?,
            Some(MAX_SCAN_CHUNKS)
        );
        let (count, old_paths): (usize, usize) = store.connection.query_row(
            "SELECT count(*),sum(path!=?2) FROM semantic_chunks WHERE generation=?1",
            params![staging, moved.path],
            |r| Ok((r.get(0)?, r.get(1)?)),
        )?;
        assert_eq!((count, old_paths), (MAX_SCAN_CHUNKS, 0));
        let mut duplicate = moved_task.clone();
        duplicate.key.push_str("-another-path");
        duplicate.path.push_str("-another-path");
        assert!(
            store
                .prune_before_inference(staging, &[moved_task.clone(), duplicate], revision)
                .is_err(),
            "True current membership over budget must reject atomically"
        );
        let count: usize = store.connection.query_row(
            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
            [staging],
            |r| r.get(0),
        )?;
        assert_eq!(count, MAX_SCAN_CHUNKS);
        std::fs::write(
            moved_dir.join("notes.txt"),
            "replacement source with new content",
        )?;
        lexical.index_folder(&root)?;
        let replacement = lexical.passage_batch(0, 1)?.remove(0);
        let replacement_task = task(&replacement);
        store.prune_before_inference(
            staging,
            std::slice::from_ref(&replacement_task),
            store.revision()?,
        )?;
        let count: usize = store.connection.query_row(
            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
            [staging],
            |r| r.get(0),
        )?;
        assert_eq!(
            count, 0,
            "Obsolete ceiling-sized group must be removed before replacement inference"
        );
        insert_complete(&mut store, staging, &replacement, &replacement_task)?;
        assert_eq!(store.active_fingerprint()?.as_deref(), Some("prune-policy"));
        let active_count: usize = store.connection.query_row(
            "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
            [active],
            |r| r.get(0),
        )?;
        assert_eq!(active_count, 1, "Pruning must never remove active vectors");
        Ok(())
    }

    #[test]
    fn resume_retains_completed_groups_and_discards_unfinished_vectors() -> Result<()> {
        let dir = tempfile::tempdir()?;
        let root = dir.path().join("docs");
        std::fs::create_dir(&root)?;
        std::fs::write(root.join("first.txt"), "complete source")?;
        std::fs::write(root.join("second.txt"), "interrupted source")?;
        let db = dir.path().join("index.db");
        let mut lexical = SearchIndex::open(&db)?;
        lexical.index_folder(&root)?;
        let passages = lexical.passage_batch(0, 64)?;
        let first = task(&passages[0]);
        let second = task(&passages[1]);
        let generation;
        {
            let mut store = Store::open(&db, 2)?;
            let lock = store.build_lock()?;
            generation = store.start_or_resume("storage-only-model", &lock)?.0;
            insert_complete(&mut store, generation, &passages[0], &first)?;
            store.stage_task(generation, &second)?;
            store.insert(
                generation,
                &passages[1],
                0,
                &[TextWindow {
                    text: passages[1].text.clone(),
                    start_byte: 0,
                    end_byte: passages[1].text.len(),
                }],
                &[vector()],
            )?;
            assert!(store
                .activate_checked(generation, store.revision()?)
                .is_err());
        }
        let mut store = Store::open(&db, 2)?;
        let lock = store.build_lock()?;
        assert_eq!(
            store.start_or_resume("storage-only-model", &lock)?,
            (generation, true)
        );
        assert_eq!(
            store.connection.query_row(
                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
                [generation],
                |r| r.get::<_, usize>(0)
            )?,
            1
        );
        store.stage_task(generation, &first)?;
        assert_eq!(
            store.reuse_text_source(generation, &first, &passages[0])?,
            Some(1)
        );
        assert!(store
            .reuse_text_source(generation, &second, &passages[1])?
            .is_none());
        assert!(store
            .activate_checked(generation, store.revision()?)
            .is_err());
        store.reconcile_tasks(generation, &[first.key], store.revision()?)?;
        store.activate_checked(generation, store.revision()?)?;
        Ok(())
    }

    #[test]
    fn legacy_vectors_remain_searchable_but_do_not_qualify_for_reuse() -> Result<()> {
        let dir = tempfile::tempdir()?;
        let root = dir.path().join("docs");
        std::fs::create_dir(&root)?;
        std::fs::write(root.join("notes.txt"), "legacy searchable source")?;
        let db = dir.path().join("index.db");
        let mut lexical = SearchIndex::open(&db)?;
        lexical.index_folder(&root)?;
        let p = lexical.passage_batch(0, 1)?.remove(0);
        let mut store = Store::open(&db, 2)?;
        let legacy = store.begin("legacy-policy")?;
        store.insert(
            legacy,
            &p,
            0,
            &[TextWindow {
                text: p.text.clone(),
                start_byte: 0,
                end_byte: p.text.len(),
            }],
            &[vector()],
        )?;
        store.activate(legacy)?;
        // Reopen a schema6 DB with legacy active chunks and no manifests.
        store.connection.execute_batch(
            "DROP INDEX semantic_chunk_task; ALTER TABLE semantic_chunks DROP COLUMN task_key;
            DROP TABLE semantic_tasks; ALTER TABLE semantic_generations DROP COLUMN incremental;
            UPDATE semantic_meta SET schema_version=6;",
        )?;
        drop(store);
        let mut store = Store::open(&db, 2)?;
        assert_eq!(
            store
                .search(&vector(), "legacy-policy", 1, &SearchOptions::default())?
                .len(),
            1
        );
        let lock = store.build_lock()?;
        let (g, _) = store.start_or_resume("legacy-policy", &lock)?;
        let t = task(&p);
        store.stage_task(g, &t)?;
        assert_eq!(store.reuse_text_source(g, &t, &p)?, None);
        assert_eq!(
            store.active_fingerprint()?.as_deref(),
            Some("legacy-policy")
        );
        Ok(())
    }

    #[test]
    fn late_source_validation_rejects_publication_without_holding_write_lock() -> Result<()> {
        let dir = tempfile::tempdir()?;
        let root = dir.path().join("docs");
        std::fs::create_dir(&root)?;
        std::fs::write(root.join("notes.txt"), "source identity before graph build")?;
        let db = dir.path().join("index.db");
        let mut lexical = SearchIndex::open(&db)?;
        lexical.index_folder(&root)?;
        let passage = lexical.passage_batch(0, 1)?.remove(0);
        let source = task(&passage);
        let mut store = Store::open(&db, 2)?;
        let lock = store.build_lock()?;
        let (active, _) = store.start_or_resume("old-policy", &lock)?;
        insert_complete(&mut store, active, &passage, &source)?;
        store.activate_checked(active, store.revision()?)?;
        let (staging, _) = store.start_or_resume("new-policy", &lock)?;
        insert_complete(&mut store, staging, &passage, &source)?;
        let revision = store.revision()?;
        let mut called = false;
        let error = store
            .activate_checked_with_validation(staging, revision, || {
                called = true;
                // An independent write succeeds only when validation runs outside
                // the publishing connection's Immediate transaction.
                let other = rusqlite::Connection::open(&db)?;
                other.busy_timeout(std::time::Duration::ZERO)?;
                other.execute_batch("BEGIN IMMEDIATE; COMMIT;")?;
                std::fs::write(root.join("notes.txt"), "direct filesystem mutation")?;
                anyhow::bail!("source changed after graph preparation")
            })
            .unwrap_err();
        assert!(called);
        assert!(error
            .to_string()
            .contains("source changed after graph preparation"));
        assert_eq!(store.active_fingerprint()?.as_deref(), Some("old-policy"));
        assert_eq!(
            store.ann_cache.as_ref().map(|(generation, _)| *generation),
            Some(active)
        );
        assert_eq!(store.start_or_resume("new-policy", &lock)?, (staging, true));
        assert_eq!(
            store.connection.query_row(
                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
                [staging],
                |row| row.get::<_, usize>(0),
            )?,
            1
        );
        Ok(())
    }

    #[test]
    fn revision_race_cannot_publish_and_failed_activation_keeps_staging() -> Result<()> {
        let dir = tempfile::tempdir()?;
        let root = dir.path().join("docs");
        std::fs::create_dir(&root)?;
        std::fs::write(root.join("notes.txt"), "source before change")?;
        let db = dir.path().join("index.db");
        let mut lexical = SearchIndex::open(&db)?;
        lexical.index_folder(&root)?;
        let p = lexical.passage_batch(0, 1)?.remove(0);
        let mut store = Store::open(&db, 2)?;
        let lock = store.build_lock()?;
        let (g, _) = store.start_or_resume("checked-policy", &lock)?;
        let t = task(&p);
        insert_complete(&mut store, g, &p, &t)?;
        let old_revision = store.revision()?;
        std::fs::write(root.join("notes.txt"), "source after change longer")?;
        lexical.index_folder(&root)?;
        assert!(store.activate_checked(g, old_revision).is_err());
        assert_eq!(store.active_fingerprint()?, None);
        assert_eq!(store.start_or_resume("checked-policy", &lock)?, (g, true));
        assert_eq!(
            store.connection.query_row(
                "SELECT count(*) FROM semantic_chunks WHERE generation=?1",
                [g],
                |r| r.get::<_, usize>(0)
            )?,
            1
        );
        Ok(())
    }
}

#[cfg(test)]
mod completion_failure_tests {
    use super::*;
    use searchi_core::SearchIndex;

    #[test]
    fn failed_completion_rolls_back_chunk_membership_and_malformed_ranges_never_reuse() -> Result<()>
    {
        // SQLite durability fixture, not neural inference evidence.
        let dir = tempfile::tempdir()?;
        let root = dir.path().join("docs");
        std::fs::create_dir(&root)?;
        std::fs::write(root.join("notes.txt"), "UTF-8 source with 中文 preserved")?;
        let db = dir.path().join("index.db");
        let mut lexical = SearchIndex::open(&db)?;
        lexical.index_folder(&root)?;
        let p = lexical.passage_batch(0, 1)?.remove(0);
        let t = SourceTask {
            key: "durability-task".into(),
            path: p.path.clone(),
            source_type: "text".into(),
            source_hash: content_hash(&p.text),
            title_hash: content_hash(&p.filename),
            policy: "exact-window-test".into(),
            passage_id: p.id,
        };
        let mut store = Store::open(&db, 2)?;
        let lock = store.build_lock()?;
        let (g, _) = store.start_or_resume("storage-geometry", &lock)?;
        store.stage_task(g, &t)?;
        store.insert(
            g,
            &p,
            0,
            &[TextWindow {
                text: p.text.clone(),
                start_byte: 0,
                end_byte: p.text.len(),
            }],
            &[vec![1.0, 0.0]],
        )?;
        assert!(store.complete_source(g, &t, 2).is_err());
        let (state,owned):(String,usize)=store.connection.query_row("SELECT state,(SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND task_key IS NOT NULL) FROM semantic_tasks WHERE generation=?1 AND task_key=?2",params![g,t.key],|r|Ok((r.get(0)?,r.get(1)?)))?;
        assert_eq!(state, "running");
        assert_eq!(owned, 0);
        assert!(store.activate_checked(g, store.revision()?).is_err());
        store.complete_source(g, &t, 1)?;
        store.connection.execute("UPDATE semantic_chunks SET start_byte=1,snippet='different snippet' WHERE generation=?1",[g])?;
        assert!(store.reuse_text_source(g, &t, &p).is_err());
        assert!(store.active_fingerprint()?.is_none());
        Ok(())
    }
}