Prepare text search during media setup and retain query sessions through promotion

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

@@ -398,11 +398,48 @@ impl Store {
         generation: i64,
         task: &SourceTask,
     ) -> Result<Option<(i64, String, usize)>> {
+        // Cross-coverage reuse is admitted only for the active generation,
+        // with at most two metadata identities (active/current staging). Exact
+        // generation-policy reuse retains the existing all-generation lookup.
+        // Validate before LIMIT, not by scanning/rejecting matching task rows.
+        let mut metadata=self.connection.prepare("SELECT g.id,g.fingerprint,g.dimensions,d.fingerprint,d.dimensions FROM semantic_generations d JOIN semantic_generations g ON (g.id=d.id OR g.id=(SELECT active_generation FROM semantic_meta WHERE singleton=1)) WHERE d.id=?1")?;
+        let generations = metadata.query_map([generation], |row| {
+            Ok((
+                row.get::<_, i64>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, usize>(2)?,
+                row.get::<_, String>(3)?,
+                row.get::<_, usize>(4)?,
+            ))
+        })?;
+        let mut admitted = Vec::with_capacity(2);
+        for candidate in generations {
+            let (id, source_policy, source_dimensions, destination_policy, destination_dimensions) =
+                candidate?;
+            let compatible = source_dimensions == destination_dimensions
+                && (source_policy == destination_policy
+                    || (task.source_type == "text"
+                        && super::super::policy_base(&destination_policy) == task.policy
+                        && super::super::reusable_policy(&task.policy, &destination_policy)
+                        && super::super::reusable_policy(&task.policy, &source_policy)));
+            if compatible {
+                admitted.push(id);
+            }
+        }
+        ensure!(
+            admitted.len() <= 2,
+            "Donor generation admission exceeds resource budget"
+        );
+        if admitted.is_empty() {
+            return Ok(None);
+        }
+        let first = admitted[0];
+        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 AND g.dimensions=destination.dimensions
+            AND (g.fingerprint=destination.fingerprint OR (?5='text' AND 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],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?))).optional()?)
+            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()?)
     }
 
     /// No tokenizer or neural call: only a completed, policy-compatible group
@@ -910,6 +947,102 @@ mod tests {
         );
         Ok(())
     }
+    #[test]
+    fn text_donor_coverage_admission_preserves_strict_identity_and_exact_legacy_lookup(
+    ) -> Result<()> {
+        // Storage geometry only; actual Google promotion is tested separately.
+        let dir = tempfile::tempdir()?;
+        let root = dir.path().join("docs");
+        std::fs::create_dir(&root)?;
+        std::fs::write(
+            root.join("notes.txt"),
+            "A real source with stable exact bytes.",
+        )?;
+        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 mut source_task = task(&passage);
+        let base = "storage-model:window-v1";
+        source_task.policy = base.into();
+        let mut store = Store::open(&db, 2)?;
+        let lock = store.build_lock()?;
+        let source = store.begin(&format!("{base}:text-only"))?;
+        insert_complete(&mut store, source, &passage, &source_task)?;
+        store.activate(source)?;
+        let destination = store.start_or_resume(base, &lock)?.0;
+        store.stage_task(destination, &source_task)?;
+        assert_eq!(
+            store.reusable_task(destination, &source_task)?.unwrap().0,
+            source
+        );
+        for wrong in [
+            "different-model:window-v1:text-only",
+            "storage-model:window-v2:text-only",
+            "storage-model:window-v1:custom-unknown",
+        ] {
+            store.connection.execute(
+                "UPDATE semantic_generations SET fingerprint=?1 WHERE id=?2",
+                params![wrong, source],
+            )?;
+            assert!(
+                store.reusable_task(destination, &source_task)?.is_none(),
+                "Forged identical task policy must not admit generation {wrong}"
+            );
+        }
+        let custom = format!("{base}:custom-240-2000-120-10000-2000-1800-14400000-17179869184");
+        store.connection.execute(
+            "UPDATE semantic_generations SET fingerprint=?1 WHERE id=?2",
+            params![custom, source],
+        )?;
+        assert_eq!(
+            store.reusable_task(destination, &source_task)?.unwrap().0,
+            source
+        );
+        store.connection.execute(
+            "UPDATE semantic_generations SET dimensions=3 WHERE id=?1",
+            [source],
+        )?;
+        assert!(store.reusable_task(destination, &source_task)?.is_none());
+        store.connection.execute(
+            "UPDATE semantic_generations SET dimensions=2,fingerprint=?1 WHERE id=?2",
+            params![format!("{base}:text-only"), source],
+        )?;
+        let mut media_task = source_task.clone();
+        media_task.key = "media-control".into();
+        media_task.source_type = "media".into();
+        media_task.passage_id = -1;
+        store.stage_task(source, &media_task)?;
+        store.complete_source(source, &media_task, 0)?;
+        assert!(
+            store.reusable_task(destination, &media_task)?.is_none(),
+            "Media donors still require exact generation policy"
+        );
+        // Exact-policy donors outside the active/current pair remain eligible.
+        store.connection.execute("INSERT INTO semantic_generations(fingerprint,dimensions,corpus_revision) SELECT ?1,2,corpus_revision FROM semantic_meta WHERE singleton=1",[base])?;
+        let older = store.connection.last_insert_rowid();
+        insert_complete(&mut store, older, &passage, &source_task)?;
+        assert_eq!(
+            store.reusable_task(destination, &source_task)?.unwrap().0,
+            older
+        );
+        assert_eq!(
+            store.reuse_text_source(destination, &source_task, &passage)?,
+            Some(1)
+        );
+        let before: Vec<u8> = store.connection.query_row(
+            "SELECT vector FROM semantic_chunks WHERE generation=?1",
+            [older],
+            |row| row.get(0),
+        )?;
+        let after: Vec<u8> = store.connection.query_row(
+            "SELECT vector FROM semantic_chunks WHERE generation=?1",
+            [destination],
+            |row| row.get(0),
+        )?;
+        assert_eq!(before, after);
+        Ok(())
+    }
 }
 
 #[cfg(test)]