fix(evidence): filter active data in every search

.superpowers/sdd/evidence-task-5c-report.md

@@ -87,3 +87,18 @@ Final fresh evidence: Docker pgvector/HTTP/migration suites `48 passed`; full ha
 Focused unit tests, the updated real Docker lifecycle, changed-file Ruff, and `git diff --check` pass.
 The final full harness invocation completed with exit code 0, including the concurrently added DWH
 JobRunner tests.
+
+## Final ACTIVE search review fixes
+
+- `ActiveEvidenceSearcher` now treats default (`kinds=None`) and mixed-kind searches as explicit
+  split queries: non-Evidence kinds are queried separately, while Evidence is queried only with
+  ACTIVE manifest generation/document predicates applied server-side before every limit.
+- Results are merged deterministically by descending similarity then stable id and truncated once
+  to the caller's global `top_n`. Pure non-Evidence searches retain their original delegate path.
+- The corpus writer lock now covers manifest snapshot construction and all corresponding vector
+  queries, preventing retain-1 publication/GC from switching or deleting generations mid-search.
+- Removed the public post-LIMIT `active_evidence_hits` helper; no public Evidence path performs
+  client filtering after limit.
+
+Focused default/mixed/no-ACTIVE/search-pack tests pass, the real Docker pgvector lifecycle passes,
+and the final full harness plus scoped Ruff/diff invocation completed with exit code 0.

harness/tests/test_corpus_pipeline.py

@@ -250,6 +250,75 @@ def test_active_searcher_without_active_fails_closed_for_evidence(tmp_path):
     assert wrapped.search([1.0], kinds=["memory"]) == ["legacy"]
 
 
+def test_active_searcher_splits_default_and_mixed_kinds_before_global_limit(tmp_path):
+    from types import SimpleNamespace
+    from tht.search.evidence import ActiveEvidenceSearcher
+
+    store = CorpusStore(tmp_path / "corpus")
+    generation = store.stage(CorpusManifest(), {}, generation="gen:" + "a" * 32)
+    store.publish(generation)
+    calls = []
+
+    class Delegate:
+        def search(self, embedding, top_n=10, kinds=None, metadata_filter=None):
+            calls.append((kinds, metadata_filter))
+            if kinds == ["evidence"]:
+                return [SimpleNamespace(id="active", similarity=0.8)]
+            return [SimpleNamespace(id="memory", similarity=0.9)]
+
+    searcher = ActiveEvidenceSearcher(store, Delegate())
+    hits = searcher.search([1.0], top_n=1, kinds=["evidence", "memory"])
+    assert [hit.id for hit in hits] == ["memory"]
+    assert calls[0] == (["memory"], None)
+    # Empty manifest means no Evidence query, but the split remains explicit and safe.
+    assert all(call[0] != ["evidence"] for call in calls)
+    calls.clear()
+    searcher.search([1.0], top_n=1)
+    assert calls[0][0] == ["memory", "schema_column", "schema_table", "solved_question"]
+    assert all(call[0] is not None for call in calls)
+
+
+def test_active_evidence_query_holds_lock_against_publish(tmp_path):
+    import threading
+    from types import SimpleNamespace
+    from tht.search.evidence import ActiveEvidenceSearcher
+
+    first_pipeline = pipeline(tmp_path, Source([(item("one", "a"), "old")]), vectors=Vectors())
+    first_pipeline.run()
+    store = first_pipeline.store
+    entered = threading.Event()
+    release = threading.Event()
+    published = threading.Event()
+
+    class Delegate:
+        def search(self, embedding, top_n=10, kinds=None, metadata_filter=None):
+            entered.set()
+            assert release.wait(5)
+            return [SimpleNamespace(id="active", similarity=1.0)]
+
+    search = threading.Thread(
+        target=lambda: ActiveEvidenceSearcher(store, Delegate()).search(
+            [1.0], kinds=["evidence"]
+        )
+    )
+    search.start()
+    assert entered.wait(5)
+    next_generation = store.stage(CorpusManifest(), {})
+
+    def publish():
+        with store.writer_lock():
+            store.publish(next_generation)
+        published.set()
+
+    publisher = threading.Thread(target=publish)
+    publisher.start()
+    assert not published.wait(0.1)
+    release.set()
+    search.join(5)
+    publisher.join(5)
+    assert published.is_set()
+
+
 def test_unchanged_documents_skip_acquire_normalize_chunk_and_embed(tmp_path):
     one = item("one", "a")
     first_source = Source([(one, "hello")])

harness/tht/search/evidence.py

@@ -11,24 +11,38 @@ class ActiveEvidenceSearcher:
         self.delegate = delegate
 
     def search(self, embedding, top_n=10, kinds=None, metadata_filter=None):
-        if kinds != ["evidence"]:
-            return self.delegate.search(embedding, top_n=top_n, kinds=kinds)
+        requested = set(kinds) if kinds is not None else {
+            "schema_table", "schema_column", "evidence", "memory", "solved_question",
+        }
+        include_evidence = "evidence" in requested
+        other_kinds = sorted(requested - {"evidence"})
+        if not include_evidence:
+            kwargs = {"top_n": top_n, "kinds": kinds}
+            if metadata_filter is not None:
+                kwargs["metadata_filter"] = metadata_filter
+            return self.delegate.search(embedding, **kwargs)
+        with self.corpus.writer_lock():
+            hits = []
+            if other_kinds:
+                kwargs = {"top_n": top_n, "kinds": other_kinds}
+                if metadata_filter is not None:
+                    kwargs["metadata_filter"] = metadata_filter
+                hits.extend(self.delegate.search(embedding, **kwargs))
+            if include_evidence:
                 manifest = self.corpus.active_manifest()
-        if manifest is None:
-            return []
+                if manifest is not None:
                     by_generation: dict[str, list[str]] = {}
                     mapping = dict(manifest.metadata.get("document_generations", {}))
                     for document in manifest.documents:
                         generation = mapping.get(document.document_id, manifest.vector_generation)
                         if generation:
                             by_generation.setdefault(generation, []).append(document.document_id)
-        hits = []
-        for generation, document_ids in by_generation.items():
+                    for generation, document_ids in sorted(by_generation.items()):
                         hits.extend(self.delegate.search(
                             embedding, top_n=top_n, kinds=["evidence"],
                             metadata_filter={
                                 "vector_generation": generation,
-                    "document_ids": document_ids,
+                                "document_ids": sorted(document_ids),
                             },
                         ))
         return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:top_n]
@@ -39,21 +53,6 @@ def active_searcher(cfg, delegate):
     return ActiveEvidenceSearcher(CorpusStore(corpus_root), delegate)
 
 
-def active_evidence_hits(store: CorpusStore, vector_store, embedding, *, limit: int):
-    manifest = store.active_manifest()
-    if manifest is None or manifest.vector_generation is None:
-        return []
-    document_generations = dict(manifest.metadata.get("document_generations", {}))
-    active_documents = {document.document_id for document in manifest.documents}
-    hits = vector_store.search(["evidence"], embedding, limit=limit, kinds=["evidence"])
-    return [
-        hit for hit in hits
-        if hit.metadata.get("document_id") in active_documents
-        and hit.metadata.get("vector_generation")
-        == document_generations.get(hit.metadata.get("document_id"), manifest.vector_generation)
-    ]
-
-
 def resolve_evidence_file(
     store: CorpusStore, evidence_id: str, *, materialized_root=None,
 ) -> str: