fix: prevent P2 from owning Qdrant lifecycle

harness/tests/test_registry_evidence_config.py

@@ -10,6 +10,7 @@ from tht.adapters.evidence import HttpManifestEvidenceSource
 from tht.adapters.factory import build_evidence_sources
 from tht.cli import app
 from tht.config import ConfigError, load_config
+from tht.jobs.dwh_pipeline import config_dwh_binding
 
 SIGNED_CANARY = "SIGNED-CANARY-QUERY"
 ACCESS_CANARY = "ACCESS-CANARY"
@@ -350,3 +351,59 @@ def test_validation_repr_cli_and_exception_output_never_disclose_transport_secre
     assert result.exit_code == 0
     assert_no_canaries(result.stdout)
     assert_no_canaries(result.stderr)
+
+
+def _qdrant_config_yaml(tmp_path, *, registry: bool) -> dict:
+    value = {
+        "dwh": {
+            "type": "postgres_direct",
+            "connection": {
+                "database": "analytics", "schema": "public", "user": "reader",
+                "password": "not-a-canary",
+            },
+        },
+        "vectors": {
+            "type": "qdrant", "base_url": "http://localhost:6333",
+            "collection": "workspace-semantic",
+        },
+        "embeddings": {
+            "base_url": "http://localhost:11434", "model": "qwen3-embedding:0.6b", "dim": 1024,
+        },
+        "evidence": {"source_root": str(tmp_path / "evidence")},
+    }
+    if registry:
+        value["runtime_identity"] = {
+            "workspace_id": "demo-workspace", "workspace_revision": "a" * 40,
+        }
+    return value
+
+
+@pytest.mark.parametrize("mode", ["session", "maintenance"])
+def test_registry_runtime_configs_require_existing_qdrant_collection(tmp_path, mode):
+    path = tmp_path / f"{mode}.yaml"
+    path.write_text(yaml.safe_dump(_qdrant_config_yaml(tmp_path, registry=True)))
+
+    cfg = load_config(path)
+
+    assert cfg.vectors.collection_lifecycle == "require_existing"
+
+
+def test_legacy_runtime_config_keeps_create_capable_qdrant_default(tmp_path):
+    path = tmp_path / "legacy.yaml"
+    path.write_text(yaml.safe_dump(_qdrant_config_yaml(tmp_path, registry=False)))
+
+    cfg = load_config(path)
+
+    assert cfg.vectors.collection_lifecycle == "create_if_missing"
+
+
+def test_registry_session_and_maintenance_bindings_are_equal(tmp_path):
+    values = _qdrant_config_yaml(tmp_path, registry=True)
+    session_path = tmp_path / "session.yaml"
+    maintenance_path = tmp_path / "maintenance.yaml"
+    session_path.write_text(yaml.safe_dump(values))
+    maintenance_path.write_text(yaml.safe_dump(values))
+
+    assert config_dwh_binding(load_config(session_path)) == config_dwh_binding(
+        load_config(maintenance_path)
+    )