From 92b1d3ccc328034bb255a5f5d0e65eb1530966eb Mon Sep 17 00:00:00 2001 From: mptyl Date: Sun, 12 Jul 2026 05:34:47 +0200 Subject: [PATCH] fix(evidence): retain referenced vector generations --- .superpowers/sdd/evidence-task-5c-report.md | 19 +++++++++++ .../l0/test_pgvector_corpus_lifecycle.py | 14 +++++--- harness/tests/test_corpus_pipeline.py | 30 ++++++++++++++++ harness/tests/test_corpus_publish.py | 27 +++++++++++++++ harness/tests/test_search_pack.py | 4 ++- harness/tht/corpus/pipeline.py | 34 +++++++++++++------ harness/tht/search/evidence.py | 27 ++++++++------- 7 files changed, 127 insertions(+), 28 deletions(-) diff --git a/.superpowers/sdd/evidence-task-5c-report.md b/.superpowers/sdd/evidence-task-5c-report.md index 476e6786..c2d6c7b9 100644 --- a/.superpowers/sdd/evidence-task-5c-report.md +++ b/.superpowers/sdd/evidence-task-5c-report.md @@ -68,3 +68,22 @@ changed-file Ruff and `git diff --check` clean. Final fresh evidence: Docker pgvector/HTTP/migration suites `48 passed`; full harness `680 passed, 5 external L2 deselected`; changed-file Ruff and `git diff --check` clean. + +## Integrated Task 5 dependency fixes + +- GC now distinguishes filesystem retention from vector dependencies. ACTIVE and the newest + `N-1` published manifests keep their directories; every exact generation in their + `document_generations` maps remains vector-protected even after its old publication directory is + evicted. Job-protected manifests receive the same dependency treatment. +- The real four-publication Docker lifecycle now includes an unchanged document whose vectors come + from the first generation. With retention `N=2`, only the final two publication directories remain + while the first generation's vectors remain searchable from ACTIVE and survive restart/explicit GC. +- Evidence lookup is always wrapped by the ACTIVE-aware searcher. With no corpus/ACTIVE, Evidence + returns no rows and search packs cannot expose legacy vectors; non-Evidence kinds are unchanged. +- Session artifact resolution holds the corpus writer lock, snapshots the active manifest once, and + materializes bytes using that exact `manifest_id`, preventing a concurrent publish/retain-1 GC from + changing or deleting the selected source generation. + +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. diff --git a/harness/tests/l0/test_pgvector_corpus_lifecycle.py b/harness/tests/l0/test_pgvector_corpus_lifecycle.py index d2dec02d..bc988a32 100644 --- a/harness/tests/l0/test_pgvector_corpus_lifecycle.py +++ b/harness/tests/l0/test_pgvector_corpus_lifecycle.py @@ -119,6 +119,8 @@ def test_real_pgvector_corpus_job_lifecycle(tmp_path, persistent_pgvector): source_root = tmp_path / "sources" source_root.mkdir() kept = source_root / "kept.md" + stable = source_root / "stable.md" + stable.write_text("unchanged dependency evidence", encoding="utf-8") removed = source_root / "removed.md" removed.write_text("removed evidence generation zero", encoding="utf-8") @@ -208,6 +210,7 @@ def test_real_pgvector_corpus_job_lifecycle(tmp_path, persistent_pgvector): assert active_pack assert any("active fourth generation" in result.content for result in active_pack) assert all(removed_document.content not in result.content for result in active_pack) + assert any("unchanged dependency evidence" in result.content for result in active_pack) manifest = CorpusStore(tmp_path / "corpus").active_manifest() assert resolve_evidence_file( CorpusStore(tmp_path / "corpus"), removed_document_id, @@ -229,7 +232,9 @@ def test_real_pgvector_corpus_job_lifecycle(tmp_path, persistent_pgvector): recreated_hits = ActiveEvidenceSearcher( CorpusStore(tmp_path / "corpus"), EvidenceDelegate(recreated) ).search(query, top_n=2, kinds=["evidence"]) - assert recreated_hits and all(hit.metadata["vector_generation"] == resumed.generation for hit in recreated_hits) + active_dependencies = set(manifest.metadata["document_generations"].values()) + assert recreated_hits + assert all(hit.metadata["vector_generation"] in active_dependencies for hit in recreated_hits) orphan = "gen:" + "f" * 32 recreated.upsert("evidence", [VectorWriteRecord( @@ -243,7 +248,8 @@ def test_real_pgvector_corpus_job_lifecycle(tmp_path, persistent_pgvector): final_pipeline = _pipeline(tmp_path, source_root, recreated) report = final_pipeline.gc(workspace_root=tmp_path) assert report["evicted"] == [orphan] - expected = set(generations[-2:]) - assert set(CorpusStore(tmp_path / "corpus").list_generations()) == expected - assert set(recreated.list_evidence_generations("evidence")) == expected + expected_fs = set(generations[-2:]) + assert set(CorpusStore(tmp_path / "corpus").list_generations()) == expected_fs + expected_vectors = expected_fs | {generations[0]} + assert set(recreated.list_evidence_generations("evidence")) == expected_vectors assert final_pipeline.gc(workspace_root=tmp_path)["evicted"] == [] diff --git a/harness/tests/test_corpus_pipeline.py b/harness/tests/test_corpus_pipeline.py index f4a7eb6f..fdfa1ae2 100644 --- a/harness/tests/test_corpus_pipeline.py +++ b/harness/tests/test_corpus_pipeline.py @@ -220,6 +220,36 @@ def test_explicit_gc_blocks_while_job_holds_corpus_writer_lock(tmp_path): assert candidate.store.active_generation() is not None +def test_gc_preserves_vector_dependencies_of_retained_manifests(tmp_path): + vectors = Vectors() + one = item("one", "a") + first = pipeline(tmp_path, Source([(one, "stable")]), vectors=vectors, retain=2).run().generation + second = pipeline( + tmp_path, Source([(one, "stable"), (item("two", "b"), "two")]), + vectors=vectors, retain=2, + ).run().generation + third = pipeline( + tmp_path, Source([(one, "stable"), (item("two", "c"), "changed")]), + vectors=vectors, retain=2, + ).run().generation + assert CorpusStore(tmp_path / "corpus").list_generations() == [second, third] + assert first in vectors.list_evidence_generations("evidence") + + +def test_active_searcher_without_active_fails_closed_for_evidence(tmp_path): + from types import SimpleNamespace + from tht.search.evidence import active_searcher + + class Delegate: + def search(self, embedding, top_n=10, kinds=None, metadata_filter=None): + return ["legacy"] + + cfg = SimpleNamespace(paths=SimpleNamespace(artifacts=tmp_path / "artifacts")) + wrapped = active_searcher(cfg, Delegate()) + assert wrapped.search([1.0], kinds=["evidence"]) == [] + assert wrapped.search([1.0], kinds=["memory"]) == ["legacy"] + + 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/tests/test_corpus_publish.py b/harness/tests/test_corpus_publish.py index 1dcca4f1..0596bd3a 100644 --- a/harness/tests/test_corpus_publish.py +++ b/harness/tests/test_corpus_publish.py @@ -137,3 +137,30 @@ def test_owned_copy_uses_validated_descriptor_bytes_when_source_is_replaced(tmp_ owned = store.materialize_document(document.document_id, tmp_path / "session" / "evidence.md") assert owned.read_text() == content assert hashlib.sha256(owned.read_bytes()).hexdigest() == document.content_hash.removeprefix("sha256:") + + +def test_materialized_snapshot_uses_identified_manifest_when_active_changes(tmp_path): + from tht.corpus.models import CanonicalDocument + import hashlib + + def doc(content, fingerprint): + return CanonicalDocument( + document_id="doc:" + hashlib.sha256(content.encode()).hexdigest(), + source_id="fs:item", source_uri="file:///item", + source_fingerprint="sha256:" + fingerprint * 64, + content_hash="sha256:" + hashlib.sha256(content.encode()).hexdigest(), + content=content, pipeline_version="evidence-v1", + ) + + store = CorpusStore(tmp_path / "corpus") + old = doc("old", "a") + old_generation = store.stage(CorpusManifest(documents=(old,)), {old.document_id: old.content}) + store.publish(old_generation) + snapshot = store.active_manifest() + new = doc("new", "b") + new_generation = store.stage(CorpusManifest(documents=(new,)), {new.document_id: new.content}) + store.publish(new_generation) + path = store.materialize_document( + snapshot.documents[0].document_id, tmp_path / "owned.md", generation=snapshot.manifest_id, + ) + assert path.read_text() == "old" diff --git a/harness/tests/test_search_pack.py b/harness/tests/test_search_pack.py index 5f246219..86174342 100644 --- a/harness/tests/test_search_pack.py +++ b/harness/tests/test_search_pack.py @@ -81,7 +81,9 @@ def test_pack_single_embed_and_sections(tmp_path, monkeypatch): assert res.exit_code == 0, res.output assert emb.calls == 1 # UN solo embedding per le tre ricerche assert "fact_ablazione" in res.output and "Ablazioni" in res.output - assert "Dominio ablazione" in res.output + # Evidence is fail-closed until an ACTIVE corpus exists; legacy vector rows + # must not leak into a new search pack. + assert "Dominio ablazione" not in res.output assert "SELECT 1" in res.output diff --git a/harness/tht/corpus/pipeline.py b/harness/tht/corpus/pipeline.py index e0f06beb..4c068a7c 100644 --- a/harness/tht/corpus/pipeline.py +++ b/harness/tht/corpus/pipeline.py @@ -96,33 +96,47 @@ class CorpusPipeline: list_vectors = getattr(self.vector_store, "list_evidence_generations", None) vector_generations = set(list_vectors("evidence")) if list_vectors else set() generations = sorted(set(published) | vector_generations) - protected = self._protected_generations(workspace_root) + job_protected = self._protected_generations(workspace_root) active = self.store.active_generation() rollback_count = self.retain_published_generations - 1 rollback = [generation for generation in published if generation != active] keep = ({active} if active else set()) | set(rollback[-rollback_count:] if rollback_count else ()) - keep |= protected + fs_keep = keep | job_protected + vector_protected = set(fs_keep) + for generation in fs_keep: + try: + manifest = self.store.manifest(generation) + except (OSError, ValueError): + continue + vector_protected.update( + value for value in manifest.metadata.get("document_generations", {}).values() + if isinstance(value, str) + ) evicted, failures = [], [] + filesystem_generations = set(self.store.list_generations()) for generation in generations: - if generation in keep: + purge_vector = generation not in vector_protected + purge_filesystem = generation in filesystem_generations and generation not in fs_keep + if not purge_vector and not purge_filesystem: continue if dry_run: evicted.append(generation) continue + if purge_vector: + try: + self.vector_store.delete_generation("evidence", generation) + except Exception: + failures.append({"generation": generation, "error": "vector cleanup failed"}) + continue try: - self.vector_store.delete_generation("evidence", generation) - except Exception: - failures.append({"generation": generation, "error": "vector cleanup failed"}) - continue - try: - if generation in self.store.list_generations(): + if purge_filesystem: self.store.discard(generation) evicted.append(generation) except Exception: failures.append({"generation": generation, "error": "filesystem cleanup failed"}) return {"status": "partial" if failures else "succeeded", "dry_run": dry_run, "active_generation": self.store.active_generation(), "evicted": evicted, - "protected": sorted(protected), "failures": failures} + "protected": sorted(vector_protected), "failures": failures} def _discover(self) -> list[tuple[EvidenceSource, SourceObject]]: discovered = [] diff --git a/harness/tht/search/evidence.py b/harness/tht/search/evidence.py index ed2a1473..d04f5291 100644 --- a/harness/tht/search/evidence.py +++ b/harness/tht/search/evidence.py @@ -36,8 +36,6 @@ class ActiveEvidenceSearcher: def active_searcher(cfg, delegate): corpus_root = cfg.paths.artifacts.parent / "corpus" - if not corpus_root.exists(): - return delegate return ActiveEvidenceSearcher(CorpusStore(corpus_root), delegate) @@ -59,15 +57,18 @@ def active_evidence_hits(store: CorpusStore, vector_store, embedding, *, limit: def resolve_evidence_file( store: CorpusStore, evidence_id: str, *, materialized_root=None, ) -> str: - manifest = store.active_manifest() - if manifest is None: - return "" - for document in manifest.documents: - frontmatter = document.metadata.get("frontmatter", {}) - identifiers = {document.document_id, document.source_id, str(frontmatter.get("id", ""))} - if evidence_id in identifiers: - root = materialized_root or (store.root / "runtime") - filename = document.document_id.removeprefix("doc:") + ".md" - path = store.materialize_document(document.document_id, root / filename) - return str(path) if path else "" + with store.writer_lock(): + manifest = store.active_manifest() + if manifest is None: + return "" + for document in manifest.documents: + frontmatter = document.metadata.get("frontmatter", {}) + identifiers = {document.document_id, document.source_id, str(frontmatter.get("id", ""))} + if evidence_id in identifiers: + root = materialized_root or (store.root / "runtime") + filename = document.document_id.removeprefix("doc:") + ".md" + path = store.materialize_document( + document.document_id, root / filename, generation=manifest.manifest_id, + ) + return str(path) if path else "" return ""