From 6656a69630ecd4efe9e63164c4375a586d4911ff Mon Sep 17 00:00:00 2001 From: mptyl Date: Sun, 12 Jul 2026 05:40:45 +0200 Subject: [PATCH] fix(evidence): filter active data in every search --- .superpowers/sdd/evidence-task-5c-report.md | 15 +++++ harness/tests/test_corpus_pipeline.py | 69 +++++++++++++++++++++ harness/tht/search/evidence.py | 69 ++++++++++----------- 3 files changed, 118 insertions(+), 35 deletions(-) diff --git a/.superpowers/sdd/evidence-task-5c-report.md b/.superpowers/sdd/evidence-task-5c-report.md index c2d6c7b9..45f3c021 100644 --- a/.superpowers/sdd/evidence-task-5c-report.md +++ b/.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. diff --git a/harness/tests/test_corpus_pipeline.py b/harness/tests/test_corpus_pipeline.py index fdfa1ae2..92fb12f1 100644 --- a/harness/tests/test_corpus_pipeline.py +++ b/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")]) diff --git a/harness/tht/search/evidence.py b/harness/tht/search/evidence.py index d04f5291..a2d24f93 100644 --- a/harness/tht/search/evidence.py +++ b/harness/tht/search/evidence.py @@ -11,26 +11,40 @@ 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) - manifest = self.corpus.active_manifest() - if manifest is None: - return [] - 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(): - hits.extend(self.delegate.search( - embedding, top_n=top_n, kinds=["evidence"], - metadata_filter={ - "vector_generation": generation, - "document_ids": document_ids, - }, - )) + 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 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) + 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": 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: