fix: stabilize runtime workspace identity

harness/tests/test_config_resources.py

@@ -95,6 +95,7 @@ def test_runtime_handoff_preserves_canonical_identity_and_durable_roots(monkeypa
 runtime_identity:
   workspace_id: psd-clinical
   workspace_revision: {'a' * 40}
+  source_identity: workspace://psd-clinical
 dwh:
   type: postgres_direct
   connection: {{database: analytics, schema: mart, user: reader, password: secret}}
@@ -113,6 +114,7 @@ embeddings: {{base_url: http://embedding.invalid, model: embed, dim: 768}}
 
     assert cfg._workspace_id == "psd-clinical"
     assert cfg._workspace_revision == "a" * 40
+    assert cfg._config_source == "workspace://psd-clinical"
     assert cfg.paths.sessions == runtime_root / "sessions"
     assert cfg.paths.artifacts == runtime_root / "artifacts"
     assert cfg.paths.indexes == runtime_root / "indexes"
@@ -135,6 +137,23 @@ roots: {{sessions: sessions, artifacts: artifacts, indexes: indexes}}
     assert workspace_id_for_config(cfg, workspace) == "psd-clinical"
 
 
+def test_runtime_identity_rejects_a_source_for_another_workspace(tmp_path):
+    workspace = tmp_path / "runtime-random-uuid.yaml"
+    workspace.write_text(f"""
+runtime_identity:
+  workspace_id: psd-clinical
+  workspace_revision: {'a' * 40}
+  source_identity: workspace://another-workspace
+dwh:
+  type: postgres_direct
+  connection: {{database: analytics, schema: mart, user: reader, password: secret}}
+roots: {{sessions: sessions, artifacts: artifacts, indexes: indexes}}
+""")
+
+    with pytest.raises(ConfigError, match="source_identity"):
+        load_config(workspace)
+
+
 def test_load_config_rejects_executable_yaml_tags_without_running_them(tmp_path):
     marker = tmp_path / "must-not-exist"
     workspace = tmp_path / "workspace.yaml"

harness/tests/test_dwh_preprocess_job.py

@@ -5,8 +5,9 @@ from pathlib import Path
 from typer.testing import CliRunner
 
 from tht.cli import app
+from tht.config import load_config
 from tht.jobs.dwh_pipeline import DwhPreprocessPipeline
-from tht.jobs.dwh_pipeline import active_generation_dir, config_dwh_binding
+from tht.jobs.dwh_pipeline import active_generation_dir, config_dwh_binding, fingerprint
 from tht.jobs.dwh_pipeline import resolve_dwh_snapshot
 from tht.jobs.dwh_pipeline import lease_dwh_snapshot
 from tht.jobs.locking import _lock_name
@@ -15,6 +16,54 @@ from tht.jobs.locking import _lock_name
 FP = "sha256:" + hashlib.sha256(b"test").hexdigest()
 
 
+def _runtime_config(tmp_path, filename, revision, database="warehouse"):
+    path = tmp_path / filename
+    path.write_text(f"""
+runtime_identity:
+  workspace_id: demo
+  workspace_revision: {revision}
+  source_identity: workspace://demo
+dwh:
+  type: postgres_direct
+  connection: {{database: {database}, schema: analytics, user: reader, password: secret}}
+roots:
+  sessions: {tmp_path / 'sessions'}
+  artifacts: {tmp_path / 'artifacts'}
+  indexes: {tmp_path / 'indexes'}
+""")
+    return path
+
+
+def test_runtime_lease_path_and_revision_metadata_do_not_change_dwh_binding(tmp_path, monkeypatch):
+    monkeypatch.delenv("THT_HOME", raising=False)
+    monkeypatch.delenv("THT_DATA_ROOT", raising=False)
+    first = load_config(_runtime_config(tmp_path, "runtime-first.yaml", "a" * 40))
+    second = load_config(_runtime_config(tmp_path, "runtime-second.yaml", "a" * 40))
+    metadata_only_revision = load_config(
+        _runtime_config(tmp_path, "runtime-third.yaml", "b" * 40)
+    )
+
+    expected = config_dwh_binding(first)
+
+    assert expected == config_dwh_binding(second)
+    assert expected == config_dwh_binding(metadata_only_revision)
+    assert expected["input_fingerprint"] == fingerprint("workspace://demo")
+
+
+def test_effective_dwh_change_invalidates_runtime_binding(tmp_path, monkeypatch):
+    monkeypatch.delenv("THT_HOME", raising=False)
+    monkeypatch.delenv("THT_DATA_ROOT", raising=False)
+    original = load_config(_runtime_config(tmp_path, "runtime-first.yaml", "a" * 40))
+    changed = load_config(
+        _runtime_config(tmp_path, "runtime-second.yaml", "b" * 40, database="warehouse_v2")
+    )
+
+    assert (
+        config_dwh_binding(original)["config_fingerprint"]
+        != config_dwh_binding(changed)["config_fingerprint"]
+    )
+
+
 def snapshot_config(tmp_path, workspace_id="demo"):
     from types import SimpleNamespace
 

harness/tht/config.py

@@ -165,6 +165,19 @@ class PathsConfig(BaseModel):
 class RuntimeIdentityConfig(BaseModel):
     workspace_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,62}$")
     workspace_revision: str = Field(pattern=r"^[0-9a-f]{40}$")
+    source_identity: str | None = Field(
+        default=None,
+        pattern=r"^workspace://[a-z][a-z0-9-]{2,62}$",
+    )
+
+    @model_validator(mode="after")
+    def source_matches_workspace(self):
+        expected = f"workspace://{self.workspace_id}"
+        if self.source_identity is None:
+            self.source_identity = expected
+        elif self.source_identity != expected:
+            raise ValueError("source_identity must match workspace_id")
+        return self
 
 
 class WorkspaceRoots(PathsConfig):
@@ -430,7 +443,11 @@ def load_config(path: Path) -> Config:
         if cfg.runtime_identity is not None
         else None
     )
-    cfg._config_source = path.resolve().as_posix()
+    cfg._config_source = (
+        cfg.runtime_identity.source_identity
+        if cfg.runtime_identity is not None
+        else path.resolve().as_posix()
+    )
     return cfg
 
 

harness/tht/jobs/dwh_pipeline.py

@@ -56,6 +56,9 @@ def config_dwh_binding(cfg) -> dict[str, str]:
         # 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)
+        # Git revision and logical source identify the runtime handoff, not the effective DWH
+        # or preprocessing configuration. They must not invalidate reusable DWH generations.
+        payload.pop("runtime_identity", None)
         config_fingerprint = fingerprint(json.dumps(payload, separators=(",", ":"), ensure_ascii=False))
     else:
         # Lightweight test doubles predating Pydantic's model_dump() retain the legacy seam.