fix(preprocess): claim DWH roots atomically

harness/tests/test_dwh_preprocess_job.py

@@ -104,17 +104,87 @@ def test_shared_root_mismatch_fails_without_deadlock_while_owner_reader_is_activ
             target=lambda: (errors.append(_capture_error(contender.run)), finished.set())
         )
         thread.start()
-        assert finished.wait(2)
+        assert not finished.wait(0.1)
     thread.join(2)
+    assert finished.is_set()
     assert "different workspace configuration" in str(errors[0])
 
 
+def test_concurrent_brand_new_shared_root_has_one_atomic_owner_and_loser_never_builds(tmp_path):
+    import threading
+
+    calls = {"alpha": 0, "beta": 0}
+    results = []
+    barrier = threading.Barrier(2)
+
+    def run(workspace):
+        def introspect(output):
+            calls[workspace] += 1
+            output.write_text("catalog")
+
+        def build(physical, output):
+            calls[workspace] += 1
+            _write_lsh([], physical, output)
+
+        candidate = DwhPreprocessPipeline(
+            workspace_id=workspace, workspace_root=tmp_path,
+            config_fingerprint=FP, input_fingerprint=FP,
+            introspect=introspect, build_lsh=build,
+        )
+        barrier.wait()
+        results.append((workspace, _capture_error(candidate.run)))
+
+    threads = [threading.Thread(target=run, args=(name,)) for name in ("alpha", "beta")]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(5)
+    assert all(not thread.is_alive() for thread in threads)
+    winner = next(name for name, result in results if not isinstance(result, Exception))
+    loser = next(name for name, result in results if isinstance(result, Exception))
+    assert calls[winner] == 2
+    assert calls[loser] == 0
+
+
+def test_missing_active_with_generations_and_symlink_owner_marker_fail_closed(tmp_path):
+    import pytest
+
+    calls = []
+    owner = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: output.write_text("catalog"),
+        build_lsh=lambda physical, output: _write_lsh([], physical, output),
+    )
+    owner.run()
+    (tmp_path / ".tht-dwh" / "ACTIVE").unlink()
+    owner.introspect = lambda output: calls.append("called")
+    with pytest.raises(Exception, match="without a consistent ACTIVE"):
+        owner.run()
+    assert calls == []
+
+    other_root = tmp_path / "other"
+    marker_root = other_root / ".tht-dwh"
+    marker_root.mkdir(parents=True)
+    external = tmp_path / "external-owner"
+    external.write_text("foreign")
+    (marker_root / "OWNER.json").symlink_to(external)
+    contender = DwhPreprocessPipeline(
+        workspace_id="demo", workspace_root=other_root,
+        config_fingerprint=FP, input_fingerprint=FP,
+        introspect=lambda output: calls.append("symlink-called"),
+        build_lsh=lambda physical, output: None,
+    )
+    with pytest.raises(Exception, match="ownership marker"):
+        contender.run()
+    assert calls == []
+
+
 def _capture_error(operation):
     try:
-        operation()
+        return operation()
     except Exception as error:
         return error
-    raise AssertionError("operation unexpectedly succeeded")
 
 
 def _write_lsh(calls, physical: Path, output: Path):
@@ -430,10 +500,6 @@ def test_cleanup_never_follows_top_level_or_child_symlinks(tmp_path):
     victim = external / "victim"
     victim.write_text("safe")
 
-    generations = tmp_path / ".tht-dwh" / "generations"
-    generations.mkdir(parents=True)
-    (generations / ("a" * 32)).symlink_to(external, target_is_directory=True)
-
     def make(content, retain=1):
         return DwhPreprocessPipeline(
             workspace_id="demo", workspace_root=tmp_path,
@@ -446,6 +512,8 @@ def test_cleanup_never_follows_top_level_or_child_symlinks(tmp_path):
         )
 
     first = make("one", retain=2).run()
+    generations = tmp_path / ".tht-dwh" / "generations"
+    (generations / ("a" * 32)).symlink_to(external, target_is_directory=True)
     make("two", retain=2).run()
     old = generations / first.run_id
     old.chmod(0o700)