Add local spoken passages with Google retrieval and isolated runtime setup

crates/searchi-service/src/store/incremental.rs

@@ -48,7 +48,7 @@ pub(super) fn migrate(connection: &Connection) -> Result<()> {
         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;")?;
+        UPDATE semantic_meta SET schema_version=9 WHERE singleton=1;")?;
     Ok(())
 }
 
@@ -146,7 +146,7 @@ impl Store {
         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))?;
+        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 IN ('image','video','audio')) OR (?3='speech' AND source_type='speech'))",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"
@@ -160,7 +160,7 @@ impl Store {
             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])?;
+        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 IN ('image','video','audio')) OR (?3='speech' AND source_type='speech'))",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" {
@@ -180,7 +180,7 @@ impl Store {
         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("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 IN ('image','video','audio')) OR (?4='speech' AND source_type='speech'))))",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(())
@@ -299,8 +299,8 @@ impl Store {
         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("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,speech_end_ms)
+                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),c.speech_end_ms 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",
@@ -378,11 +378,11 @@ fn validate_task(task: &SourceTask) -> Result<()> {
         "Invalid source task identity"
     );
     ensure!(
-        matches!(task.source_type.as_str(), "text" | "media"),
+        matches!(task.source_type.as_str(), "text" | "media" | "speech"),
         "Invalid source task type"
     );
     ensure!(
-        (!task.source_hash.is_empty() || task.source_type == "media")
+        (!task.source_hash.is_empty() || task.source_type != "text")
             && task.source_hash.len() <= 256
             && task.title_hash.len() <= 256
             && !task.policy.is_empty()
@@ -418,6 +418,13 @@ impl Store {
                 candidate?;
             let compatible = source_dimensions == destination_dimensions
                 && (source_policy == destination_policy
+                    || (task.source_type == "media"
+                        && super::super::raw_media_policy(&destination_policy) == task.policy
+                        && super::super::raw_media_policy(&source_policy) == task.policy)
+                    || (task.source_type == "speech"
+                        && super::super::speech_component_policy(&destination_policy)
+                            == task.policy
+                        && super::super::speech_component_policy(&source_policy) == task.policy)
                     || (task.source_type == "text"
                         && super::super::policy_base(&destination_policy) == task.policy
                         && super::super::reusable_policy(&task.policy, &destination_policy)
@@ -437,7 +444,7 @@ impl Store {
         let second = admitted.get(1).copied().unwrap_or(first);
         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 OR (?5='text' AND t.generation IN (?6,?7))) AND g.dimensions=destination.dimensions
+            AND (g.fingerprint=destination.fingerprint OR t.generation IN (?6,?7)) 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,first,second],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?)
     }
@@ -510,8 +517,8 @@ impl Store {
         identity: &searchi_media::MediaIdentity,
     ) -> Result<Option<usize>> {
         ensure!(
-            task.source_type == "media"
-                && task.passage_id == -file_id
+            matches!(task.source_type.as_str(), "media" | "speech")
+                && task.passage_id == media_passage_id(file_id, &task.source_type)?
                 && task.path == identity.path.to_string_lossy()
                 && task.source_hash == identity.sha256,
             "Media task mapping differs from current source"
@@ -524,18 +531,34 @@ impl Store {
             "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 stmt=self.connection.prepare("SELECT vector,source_type,ordinal,content_hash,speech_end_ms,number,snippet,kind 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
+                    (if task.source_type == "speech" {
+                        row.get::<_, String>(1)? == "speech"
+                    } else {
+                        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"
                 );
+                if task.source_type == "speech" {
+                    let start: u32 = row.get(5)?;
+                    let end: Option<u32> = row.get(4)?;
+                    let text: String = row.get(6)?;
+                    ensure!(
+                        end.is_some_and(|end| end >= start)
+                            && row.get::<_, String>(7)? == "timestamp_ms"
+                            && !text.trim().is_empty()
+                            && text.len() <= 4096,
+                        "Corrupt completed speech evidence"
+                    );
+                }
                 validate_vector(&row.get::<_, Vec<u8>>(0)?, self.dimensions)?;
                 seen += 1;
             }
@@ -572,8 +595,8 @@ impl Store {
             "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])?;
+        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,speech_end_ms)
+            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),speech_end_ms 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" {
@@ -605,6 +628,148 @@ fn validate_vector(bytes: &[u8], dimensions: usize) -> Result<()> {
     );
     Ok(())
 }
+#[cfg(test)]
+mod speech_lifecycle_tests {
+    use super::*;
+    use searchi_core::SearchIndex;
+    #[test]
+    fn speech_upgrade_and_failure_reuse_raw_work_and_verified_excerpts_reject_same_stat_changes(
+    ) -> Result<()> {
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("media");
+        std::fs::create_dir(&root)?;
+        let file = root.join("meeting.wav");
+        std::fs::write(&file, b"real source bytes for storage tests")?;
+        let db = dir.path().join("index.db");
+        let mut lexical = SearchIndex::open(&db)?;
+        lexical.index_folder(&root)?;
+        let mut store = Store::open(&db, 2)?;
+        let (id, path, _, _) = store.media_batch(0)?.remove(0);
+        let identity = searchi_media::MediaIdentity::read(Path::new(&path), 1024)?;
+        let revision =
+            lexical.media_change_revision(&path, identity.bytes, identity.modified_ns)?;
+        let base = "storage-google-base";
+        let raw = SourceTask {
+            key: "raw-key".into(),
+            path: path.clone(),
+            source_type: "media".into(),
+            source_hash: identity.sha256.clone(),
+            title_hash: String::new(),
+            policy: base.into(),
+            passage_id: -id,
+        };
+        let original = store.begin(base)?;
+        store.stage_task(original, &raw)?;
+        // Synthetic unit geometry tests membership, not production inference or relevance.
+        store.insert_media(original, id, 0, &identity, "audio", 0, &[1., 0.])?;
+        store.complete_source(original, &raw, 1)?;
+        store.activate(original)?;
+        let speech_policy = format!("{base}:speech-v1-{}", "a".repeat(64));
+        let first = store.begin(&speech_policy)?;
+        store.stage_task(first, &raw)?;
+        assert_eq!(
+            store.reuse_media_source(first, &raw, id, &identity)?,
+            Some(1)
+        );
+        let speech = SourceTask {
+            key: "speech-key".into(),
+            source_type: "speech".into(),
+            policy: speech_policy.clone(),
+            passage_id: media_passage_id(id, "speech")?,
+            ..raw.clone()
+        };
+        store.stage_task(first, &speech)?;
+        store.insert_speech(
+            first,
+            id,
+            0,
+            &identity,
+            revision,
+            100,
+            100,
+            "Recognized words",
+            &[0., 1.],
+        )?;
+        store.insert_speech(
+            first,
+            id,
+            1,
+            &identity,
+            revision,
+            100,
+            110,
+            "Different recognized words",
+            &[0., 1.],
+        )?;
+        store.complete_source(first, &speech, 2)?;
+        validate_completion(&store.connection, first)?;
+        store.activate(first)?;
+        let excerpt = store
+            .speech_excerpt(&path, 100, 4096)?
+            .expect("verified excerpt");
+        assert_eq!(excerpt.text, "Recognized words");
+        assert_eq!(excerpt.end_ms, 100);
+        assert_eq!(excerpt.provenance, "local_speech_recognition");
+        let exact = store
+            .speech_excerpt_matching(&path, 100, Some("Different recognized words"), 4096)?
+            .expect("chosen duplicate-time passage");
+        assert_eq!(exact.end_ms, 110);
+        assert_eq!(exact.text, "Different recognized words");
+        assert!(store
+            .speech_excerpt_matching(&path, 100, Some("not recognized"), 4096)?
+            .is_none());
+        let same = store.begin(&speech_policy)?;
+        store.stage_task(same, &raw)?;
+        store.stage_task(same, &speech)?;
+        assert_eq!(
+            store.reuse_media_source(same, &raw, id, &identity)?,
+            Some(1)
+        );
+        assert_eq!(
+            store.reuse_media_source(same, &speech, id, &identity)?,
+            Some(2)
+        );
+        store.activate(same)?;
+        let changed_policy = format!("{base}:speech-v1-{}", "b".repeat(64));
+        let changed = store.begin(&changed_policy)?;
+        store.stage_task(changed, &raw)?;
+        assert_eq!(
+            store.reuse_media_source(changed, &raw, id, &identity)?,
+            Some(1)
+        );
+        let updated_speech = SourceTask {
+            policy: changed_policy,
+            ..speech.clone()
+        };
+        store.stage_task(changed, &updated_speech)?;
+        assert_eq!(
+            store.reuse_media_source(changed, &updated_speech, id, &identity)?,
+            None
+        );
+        store.fail_source(changed, &updated_speech, "explicit recognition failure")?;
+        let raw_count: i64 = store.connection.query_row(
+            "SELECT count(*) FROM semantic_chunks WHERE generation=?1 AND source_type='audio'",
+            [changed],
+            |row| row.get(0),
+        )?;
+        assert_eq!(raw_count, 1);
+        validate_completion(&store.connection, changed)?;
+        // Same-stat byte mutations never qualify for agent evidence.
+        let modified = std::fs::metadata(&path)?.modified()?;
+        let replacement = vec![b'x'; identity.bytes as usize];
+        std::fs::write(&path, replacement)?;
+        std::fs::OpenOptions::new()
+            .write(true)
+            .open(&path)?
+            .set_times(std::fs::FileTimes::new().set_modified(modified))?;
+        assert!(store.speech_excerpt(&path, 100, 4096)?.is_none());
+        assert!(store
+            .speech_excerpt_matching(&path, 100, Some("Different recognized words"), 4096)?
+            .is_none());
+        Ok(())
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;