fix: bootstrap user preferences without invalidating DWH cache

harness/tests/test_dwh_preprocess_job.py

@@ -31,6 +31,24 @@ def test_dwh_and_evidence_jobs_have_distinct_lock_names():
     assert _lock_name("demo", "dwh") != _lock_name("demo", "evidence")
 
 
+def test_session_storage_does_not_change_the_dwh_artifact_binding(tmp_path):
+    from types import SimpleNamespace
+
+    def config(session_storage=None):
+        payload = {"dwh": {"connection": {"database": "warehouse"}}}
+        if session_storage is not None:
+            payload["session_storage"] = session_storage
+        cfg = SimpleNamespace(_workspace_id="demo", _config_source="test")
+        cfg.model_dump = lambda **_kwargs: payload
+        cfg.model_dump_json = lambda: json.dumps(payload, separators=(",", ":"))
+        return cfg
+
+    without_session_storage = config()
+    with_session_storage = config({"type": "postgres_direct", "connection": {"database": "sessions"}})
+
+    assert config_dwh_binding(without_session_storage) == config_dwh_binding(with_session_storage)
+
+
 def test_unowned_reads_fail_closed_without_creating_any_files(tmp_path):
     import pytest
 

harness/tht/jobs/dwh_pipeline.py

@@ -48,9 +48,21 @@ def config_dwh_binding(cfg) -> dict[str, str]:
     config_source = getattr(cfg, "_config_source", None)
     if not isinstance(workspace_id, str) or not isinstance(config_source, str):
         raise CorruptCheckpointError("DWH workspace identity is unavailable; reload configuration")
+    model_dump = getattr(cfg, "model_dump", None)
+    if callable(model_dump):
+        payload = model_dump(mode="json")
+        if not isinstance(payload, dict):
+            raise CorruptCheckpointError("DWH workspace configuration is unavailable; reload configuration")
+        # Session persistence has no bearing on schema/LSH artifacts. Excluding it keeps an
+        # opt-in session-storage deployment from invalidating an otherwise identical DWH cache.
+        payload.pop("session_storage", None)
+        config_fingerprint = fingerprint(json.dumps(payload, separators=(",", ":"), ensure_ascii=False))
+    else:
+        # Lightweight test doubles predating Pydantic's model_dump() retain the legacy seam.
+        config_fingerprint = fingerprint(cfg.model_dump_json())
     return {
         "workspace_id": workspace_id,
-        "config_fingerprint": fingerprint(cfg.model_dump_json()),
+        "config_fingerprint": config_fingerprint,
         "input_fingerprint": fingerprint(config_source),
     }