test(evidence): harden generation lifecycle boundaries

harness/tests/l0/test_vector_adapter_parity.py

@@ -264,3 +264,24 @@ def test_http_list_evidence_generations_exact_rpc_and_legacy_fail_closed(monkeyp
     assert client.list_evidence_generations("evidence") == ["gen:" + "a" * 32]
     assert calls[0][0].endswith("/rpc/list_evidence_generations")
     assert calls[0][1] == {"table_name": "evidence", "kind": "evidence"}
+
+
+@pytest.mark.parametrize("generation", ["gen:a", "gen:" + "A" * 32, "gen:" + "a" * 33])
+def test_http_generation_operations_reject_noncanonical_values(monkeypatch, generation):
+    monkeypatch.setattr(
+        "tht.vectorstore.rest_client.requests.post",
+        lambda *args, **kwargs: pytest.fail("invalid generation reached transport"),
+    )
+    client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer"))
+    with pytest.raises(ValueError, match="canonical"):
+        client.delete_generation("evidence", generation)
+
+
+def test_http_inventory_rejects_malformed_rpc_output(monkeypatch):
+    monkeypatch.setattr(
+        "tht.vectorstore.rest_client.requests.post",
+        lambda *args, **kwargs: Response([{"generation": "gen:../escape"}]),
+    )
+    client = VectorRestClient(RestConfig(base_url="https://vectors.test", api_key="writer"))
+    with pytest.raises(VectorRestError, match="malformed"):
+        client.list_evidence_generations("evidence")

harness/tests/test_corpus_pipeline.py

@@ -3,6 +3,7 @@ import pytest
 from tht.corpus.chunk import ChunkPolicy
 from tht.corpus.pipeline import CorpusPipeline, PipelineError
 from tht.corpus.store import CorpusStore
+from tht.corpus.models import CorpusManifest
 from tht.ports.evidence import AcquiredDocument, SourceObject
 from tht.ports.vector import VectorCapabilities
 
@@ -162,6 +163,61 @@ def test_gc_reconciles_vector_only_generation(tmp_path):
     report = candidate.gc(workspace_root=tmp_path)
     assert report["evicted"] == [orphan]
     assert vectors.list_evidence_generations("evidence") == []
+    assert candidate.gc(workspace_root=tmp_path)["evicted"] == []
+
+
+@pytest.mark.parametrize("status", ["running", "failed"])
+def test_gc_protects_generations_referenced_by_resumable_checkpoints(tmp_path, status):
+    generation = "gen:" + "e" * 32
+    store = CorpusStore(tmp_path / "corpus")
+    store.stage(CorpusManifest(), {}, generation=generation)
+    run = tmp_path / ".tht-jobs" / "evidence" / "runs" / ("a" * 32)
+    (run / "artifacts").mkdir(parents=True)
+    (run / "checkpoint.json").write_text(__import__("json").dumps({"status": status}))
+    (run / "artifacts" / "plan.json").write_text(
+        __import__("json").dumps({"generation": generation})
+    )
+    candidate = pipeline(tmp_path, Source([]), vectors=Vectors(), retain=1)
+    report = candidate.gc(workspace_root=tmp_path)
+    assert generation in report["protected"]
+    assert store.generation_path(generation).exists()
+
+
+def test_explicit_gc_blocks_while_job_holds_corpus_writer_lock(tmp_path):
+    import threading
+
+    candidate = pipeline(tmp_path, Source([(item("one", "a"), "one")]), vectors=Vectors())
+    entered = threading.Event()
+    release = threading.Event()
+    gc_finished = threading.Event()
+
+    def pause(_context, stage):
+        if stage == "discover":
+            entered.set()
+            assert release.wait(5)
+
+    job = threading.Thread(target=lambda: candidate.run_as_job(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint="sha256:" + "1" * 64,
+        input_fingerprint="sha256:" + "2" * 64,
+        after_stage_return=pause,
+    ))
+    job.start()
+    assert entered.wait(5)
+
+    def collect():
+        with candidate.store.writer_lock():
+            candidate.gc(workspace_root=tmp_path)
+        gc_finished.set()
+
+    gc_thread = threading.Thread(target=collect)
+    gc_thread.start()
+    assert not gc_finished.wait(0.1)
+    release.set()
+    job.join(5)
+    gc_thread.join(5)
+    assert gc_finished.is_set()
+    assert candidate.store.active_generation() is not None
 
 
 def test_unchanged_documents_skip_acquire_normalize_chunk_and_embed(tmp_path):

harness/tests/test_corpus_publish.py

@@ -108,3 +108,32 @@ def test_published_inventory_excludes_staged_and_invalid_newer_directories(tmp_p
     invalid.mkdir()
     (invalid / "PUBLISHED").write_text("2026-01-01T00:00:00Z\n")
     assert store.published_generations() == [first]
+
+
+def test_owned_copy_uses_validated_descriptor_bytes_when_source_is_replaced(tmp_path, monkeypatch):
+    from tht.corpus.models import CanonicalDocument
+    import hashlib
+
+    content = "active bytes"
+    document = CanonicalDocument(
+        document_id="doc:" + "c" * 64, source_id="fs:copy", source_uri="file:///copy",
+        source_fingerprint="sha256:" + "d" * 64,
+        content_hash="sha256:" + hashlib.sha256(content.encode()).hexdigest(),
+        content=content, pipeline_version="evidence-v1",
+    )
+    store = CorpusStore(tmp_path / "corpus")
+    generation = store.stage(CorpusManifest(documents=(document,)), {document.document_id: content})
+    store.publish(generation)
+    source = store.resolve_document(document.document_id)
+    real_read = os.read
+
+    def replace_after_read(fd, size):
+        payload = real_read(fd, size)
+        source.unlink()
+        source.write_text("replacement")
+        return payload
+
+    monkeypatch.setattr(os, "read", replace_after_read)
+    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:")