diff --git a/.superpowers/sdd/evidence-task-5d-report.md b/.superpowers/sdd/evidence-task-5d-report.md new file mode 100644 index 00000000..a46ac58c --- /dev/null +++ b/.superpowers/sdd/evidence-task-5d-report.md @@ -0,0 +1,42 @@ +# Evidence Task 5D — Real pgvector lifecycle gate + +## Status + +Complete. The Docker-backed L0 gate uses one persistent `pgvector/pgvector:pg16` +database and the production migrations, direct reader/writer `PgVectorStore`, +`CorpusStore`, `CorpusPipeline.run_as_job`/JobRunner, ACTIVE Evidence retrieval, +search-pack fusion, owned session artifact copy, retention, and explicit GC. + +## Lifecycle covered + +- Four real corpus publications with retention set to two generations. +- A higher-similarity stale vector proves ACTIVE metadata filtering happens before LIMIT + for normal Evidence retrieval and the search-pack fusion path. +- A removed source is absent from ACTIVE retrieval and cannot be copied to a session. +- An injected process death occurs after one real committed vector upsert. Resume uses the + real run ID, preserves that record, fills the missing records, and produces no duplicate keys. +- Database engines and direct store objects are disposed/recreated before persisted ACTIVE + retrieval is checked again. +- An exact canonical vector-only orphan generation is discovered and removed by explicit GC. +- Filesystem and vector inventories converge exactly to ACTIVE plus one rollback; a second GC + is a no-op. +- Owned session artifact bytes and SHA-256 match the ACTIVE canonical document. + +## Production bug found and fixed + +Production migration `003_roles.sql` intentionally restricted `vector_writer`, but omitted +the privileges used by the production generation lifecycle: `SELECT(metadata)` for inventory +and `DELETE` for cleanup on `vectors.evidence`. Consequently a real job published successfully +and then failed in `retention_cleanup` on its first run. + +Added versioned migration `004_evidence_generation_gc.sql` granting only those two Evidence +generation-management privileges. Runtime application code was not redesigned. + +## Verification + +- Target lifecycle: `1 passed` (Docker-backed). +- Full harness: `681 passed, 5 deselected`. +- Scoped Ruff: passed. +- `git diff --check`: passed. + +The existing Pydantic serialization and legacy-workspace deprecation warnings remain unchanged. diff --git a/harness/tests/l0/test_pgvector_corpus_lifecycle.py b/harness/tests/l0/test_pgvector_corpus_lifecycle.py new file mode 100644 index 00000000..5c63555d --- /dev/null +++ b/harness/tests/l0/test_pgvector_corpus_lifecycle.py @@ -0,0 +1,236 @@ +"""L0 gate for the complete durable Evidence/pgvector lifecycle.""" + +import hashlib +import json + +import pytest +from sqlalchemy import create_engine, text +from testcontainers.postgres import PostgresContainer + +from tht.adapters.evidence import FilesystemEvidenceSource +from tht.adapters.vector.pgvector import PgVectorStore +from tht.cli.vector_migrate_cmd import migrate +from tht.config import DatabaseConfig +from tht.corpus.chunk import ChunkPolicy +from tht.corpus.pipeline import CorpusPipeline +from tht.corpus.store import CorpusStore +from tht.ports.vector import VectorRecord, VectorWriteRecord +from tht.search import combined_search +from tht.search.evidence import ActiveEvidenceSearcher, resolve_evidence_file + + +DIMENSIONS = 768 + + +class DeterministicEmbedder: + def embed_documents(self, texts): + return [self.embed_query(text) for text in texts] + + def embed_query(self, text): + vector = [0.0] * DIMENSIONS + vector[0] = 0.8 + vector[1] = 0.6 + return vector + + +class EvidenceDelegate: + """Adapt the real multi-collection port to the runtime search protocol.""" + + def __init__(self, store): + self.store = store + + def search(self, embedding, top_n=10, kinds=None, metadata_filter=None): + return self.store.search( + ["evidence"], embedding, limit=top_n, kinds=kinds, + metadata_filter=metadata_filter, + ) + + +class InterruptAfterRealPartialUpsert: + """Crash after a committed real row, as a process death would.""" + + def __init__(self, store): + self.store = store + self.interrupt = True + + def __getattr__(self, name): + return getattr(self.store, name) + + def upsert(self, collection, records): + if self.interrupt and len(records) > 1: + self.interrupt = False + self.store.upsert(collection, records[:1]) + raise KeyboardInterrupt("injected process death after committed vector row") + return self.store.upsert(collection, records) + + +@pytest.fixture(scope="module") +def persistent_pgvector(): + with PostgresContainer("pgvector/pgvector:pg16") as postgres: + migrate(postgres.get_connection_url()) + admin = create_engine(postgres.get_connection_url()) + with admin.begin() as connection: + connection.exec_driver_sql( + "ALTER ROLE vector_reader LOGIN PASSWORD 'reader-lifecycle'" + ) + connection.exec_driver_sql( + "ALTER ROLE vector_writer LOGIN PASSWORD 'writer-lifecycle'" + ) + url = admin.url + common = dict( + host=url.host, port=url.port, database=url.database, schema="vectors" + ) + reader = DatabaseConfig( + **common, user="vector_reader", password="reader-lifecycle" + ) + writer = DatabaseConfig( + **common, user="vector_writer", password="writer-lifecycle" + ) + yield postgres, admin, reader, writer + admin.dispose() + + +def _pipeline(root, source_root, vectors): + return CorpusPipeline( + store=CorpusStore(root / "corpus"), + sources=[FilesystemEvidenceSource(source_root)], + embedder=DeterministicEmbedder(), + vector_store=vectors, + embedding_model="deterministic-l0", + embedding_dimensions=DIMENSIONS, + chunk_policy=ChunkPolicy(version="lifecycle-v1", max_chars=48), + pipeline_version="evidence-v1", + retain_published_generations=2, + ) + + +def _publish(pipeline, root, serial): + return pipeline.run_as_job( + workspace_id="pgvector-lifecycle", + workspace_root=root, + config_fingerprint="sha256:" + "1" * 64, + input_fingerprint="sha256:" + f"{serial:x}" * 64, + ) + + +@pytest.mark.l0 +def test_real_pgvector_corpus_job_lifecycle(tmp_path, persistent_pgvector): + postgres, admin, reader_config, writer_config = persistent_pgvector + source_root = tmp_path / "sources" + source_root.mkdir() + kept = source_root / "kept.md" + removed = source_root / "removed.md" + removed.write_text("removed evidence generation zero", encoding="utf-8") + + vectors = PgVectorStore(reader_config, writer_config, expected_dimension=DIMENSIONS) + generations = [] + for serial in range(3): + kept.write_text(f"active evidence generation {serial}", encoding="utf-8") + result = _publish(_pipeline(tmp_path, source_root, vectors), tmp_path, serial + 1) + assert result.status == "succeeded" + generations.append(result.generation) + + # A stale, closer row must not consume LIMIT before ACTIVE filtering. + stale_generation = generations[-2] + stale = VectorWriteRecord( + record=VectorRecord( + id="chunk:stale-perfect-match", kind="evidence", ref="doc:stale", + title="stale forbidden", content="stale forbidden", + metadata={ + "document_id": "doc:stale", + "vector_generation": stale_generation, + }, + ), + embedding=[1.0] + [0.0] * (DIMENSIONS - 1), + content_hash="sha256:" + "a" * 64, + ) + vectors.upsert("evidence", [stale]) + runtime = ActiveEvidenceSearcher(CorpusStore(tmp_path / "corpus"), EvidenceDelegate(vectors)) + query = DeterministicEmbedder().embed_query("active") + hits = runtime.search(query, top_n=1, kinds=["evidence"]) + assert len(hits) == 1 and hits[0].title != "stale forbidden" + packed = combined_search( + "active", lsh_hits=None, store=runtime, embedder=DeterministicEmbedder(), + top=1, rrf_k=60, kinds=["evidence"], query_vec=query, + ) + assert len(packed) == 1 and packed[0].label != "stale forbidden" + + # Fourth publication removes a document and creates multiple chunks for crash recovery. + removed.unlink() + kept.write_text("active fourth generation " * 8, encoding="utf-8") + crashing = InterruptAfterRealPartialUpsert(vectors) + candidate = _pipeline(tmp_path, source_root, crashing) + with pytest.raises(KeyboardInterrupt, match="injected process death"): + _publish(candidate, tmp_path, 4) + runs = tmp_path / ".tht-jobs" / "evidence" / "runs" + crashed_run = max(runs.iterdir(), key=lambda path: path.stat().st_mtime_ns).name + before = vectors.existing_hashes("evidence", ["evidence"]) + intent = json.loads( + (runs / crashed_run / "artifacts" / "vector-intent.json").read_text() + )["records"] + already_present = set(intent) & set(before) + assert len(already_present) == 1 + resumed = candidate.run_as_job( + workspace_id="pgvector-lifecycle", workspace_root=tmp_path, + config_fingerprint="sha256:" + "1" * 64, + input_fingerprint="sha256:" + "4" * 64, + resume_run_id=crashed_run, + ) + assert resumed.status == "succeeded" and resumed.resumed_from == crashed_run + generations.append(resumed.generation) + after = vectors.existing_hashes("evidence", ["evidence"]) + assert {key: after[key] for key in already_present} == { + key: before[key] for key in already_present + } + assert set(intent).issubset(after) + with admin.connect() as connection: + duplicate_count = connection.execute(text( + "SELECT count(*) - count(DISTINCT record_key) FROM vectors.evidence" + )).scalar_one() + assert duplicate_count == 0 + + runtime = ActiveEvidenceSearcher(CorpusStore(tmp_path / "corpus"), EvidenceDelegate(vectors)) + assert all(hit.ref != "doc:stale" for hit in runtime.search(query, top_n=20, kinds=["evidence"])) + manifest = CorpusStore(tmp_path / "corpus").active_manifest() + removed_document = next( + doc for doc in _pipeline(tmp_path, source_root, vectors).store.manifest(generations[-2]).documents + if "removed.md" in doc.source_uri + ) + assert resolve_evidence_file( + CorpusStore(tmp_path / "corpus"), removed_document.document_id, + materialized_root=tmp_path / "session", + ) == "" + + active_document = manifest.documents[0] + owned = CorpusStore(tmp_path / "corpus").materialize_document( + active_document.document_id, tmp_path / "session" / "active-evidence.md" + ) + assert owned.read_bytes() == active_document.content.encode() + assert hashlib.sha256(owned.read_bytes()).hexdigest() == active_document.content_hash[7:] + + # Recreate engines and stores against the same persisted database. + vectors._reader.dispose() + vectors._writer.dispose() + recreated = PgVectorStore(reader_config, writer_config, expected_dimension=DIMENSIONS) + assert recreated.health().ok is True + 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) + + orphan = "gen:" + "f" * 32 + recreated.upsert("evidence", [VectorWriteRecord( + record=VectorRecord( + id="chunk:exact-vector-orphan", kind="evidence", ref="doc:orphan", + title="orphan", content="orphan", + metadata={"document_id": "doc:orphan", "vector_generation": orphan}, + ), + embedding=query, content_hash="sha256:" + "f" * 64, + )]) + 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 + assert final_pipeline.gc(workspace_root=tmp_path)["evicted"] == [] diff --git a/harness/tests/l0/test_vector_migrations.py b/harness/tests/l0/test_vector_migrations.py index b5f660ea..72640003 100644 --- a/harness/tests/l0/test_vector_migrations.py +++ b/harness/tests/l0/test_vector_migrations.py @@ -22,7 +22,7 @@ def test_migrations_are_clean_and_idempotent(database_url): from tht.cli.vector_migrate_cmd import migrate, migration_status before = migration_status(database_url) - assert [item.version for item in before.pending] == ["001", "002", "003"] + assert [item.version for item in before.pending] == ["001", "002", "003", "004"] migrate(database_url) migrate(database_url) @@ -30,7 +30,7 @@ def test_migrations_are_clean_and_idempotent(database_url): status = migration_status(database_url) assert status.pending == () assert status.drifted == () - assert [item.version for item in status.applied] == ["001", "002", "003"] + assert [item.version for item in status.applied] == ["001", "002", "003", "004"] def test_schema_matches_direct_adapter_contract(database_url): @@ -156,7 +156,7 @@ def test_status_json_is_pristine(database_url, monkeypatch): assert result.exit_code == 0, result.output assert json.loads(result.stdout) == { - "applied": ["001", "002", "003"], + "applied": ["001", "002", "003", "004"], "drifted": [], "pending": [], } @@ -249,7 +249,7 @@ def test_hostile_admin_search_path_cannot_shadow_migration_objects(database_url) with verification.connect() as connection: assert connection.execute( text("SELECT count(*) FROM public.tht_vector_migrations") - ).scalar_one() == 3 + ).scalar_one() == 4 assert connection.execute( text("SELECT count(*) FROM shadow.tht_vector_migrations") ).scalar_one() == 0 @@ -279,10 +279,10 @@ def test_failed_batch_rolls_back_schema_and_ledger(database_url, tmp_path): from tht.cli.vector_migrate_cmd import MigrationError, migrate, migration_status migrations = _copy_migrations(tmp_path) - (migrations / "004_first.sql").write_text("CREATE TABLE public.must_rollback (id int);\n") - (migrations / "005_broken.sql").write_text("THIS IS NOT SQL;\n") + (migrations / "005_first.sql").write_text("CREATE TABLE public.must_rollback (id int);\n") + (migrations / "006_broken.sql").write_text("THIS IS NOT SQL;\n") - with pytest.raises(MigrationError, match="005_broken.sql"): + with pytest.raises(MigrationError, match="006_broken.sql"): migrate(database_url, migrations) engine = create_engine(database_url) @@ -290,8 +290,8 @@ def test_failed_batch_rolls_back_schema_and_ledger(database_url, tmp_path): assert connection.execute(text("SELECT to_regclass('public.must_rollback')")).scalar() is None engine.dispose() status = migration_status(database_url, migrations) - assert [item.version for item in status.applied] == ["001", "002", "003"] - assert [item.version for item in status.pending] == ["004", "005"] + assert [item.version for item in status.applied] == ["001", "002", "003", "004"] + assert [item.version for item in status.pending] == ["005", "006"] def _copy_migrations(tmp_path: Path) -> Path: diff --git a/harness/tht/migrations/vector/004_evidence_generation_gc.sql b/harness/tht/migrations/vector/004_evidence_generation_gc.sql new file mode 100644 index 00000000..94d4a4bb --- /dev/null +++ b/harness/tht/migrations/vector/004_evidence_generation_gc.sql @@ -0,0 +1,3 @@ +-- The writer owns derived-generation reconciliation but not runtime similarity reads. +GRANT SELECT (metadata) ON vectors.evidence TO vector_writer; +GRANT DELETE ON vectors.evidence TO vector_writer;