import hashlib
import json
from pathlib import Path

from typer.testing import CliRunner

from tht.cli import app
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 resolve_dwh_snapshot
from tht.jobs.dwh_pipeline import lease_dwh_snapshot
from tht.jobs.locking import _lock_name


FP = "sha256:" + hashlib.sha256(b"test").hexdigest()


def snapshot_config(tmp_path, workspace_id="demo"):
    from types import SimpleNamespace

    cfg = SimpleNamespace(
        paths=SimpleNamespace(artifacts=tmp_path / "artifacts", indexes=tmp_path / "indexes"),
        _workspace_id=workspace_id,
        _config_source="test",
    )
    cfg.model_dump_json = lambda: "test"
    return cfg


def test_dwh_and_evidence_jobs_have_distinct_lock_names():
    assert _lock_name("demo", "dwh") != _lock_name("demo", "evidence")


def test_unowned_reads_fail_closed_without_creating_any_files(tmp_path):
    import pytest

    cfg = snapshot_config(tmp_path)
    with pytest.raises(Exception, match="not initialized"):
        resolve_dwh_snapshot(cfg)
    with pytest.raises(Exception, match="not initialized"):
        with lease_dwh_snapshot(cfg):
            pass
    assert not (tmp_path / ".tht-dwh").exists()


def test_writer_claim_allows_only_lock_and_empty_generations(tmp_path):
    import pytest

    for name, make_entry in (
        ("unexpected", lambda root: (root / "unexpected").write_text("x")),
        ("stale-temp", lambda root: (root / ".OWNER.json.stale.tmp").write_text("x")),
        ("unexpected-dir", lambda root: (root / "other").mkdir()),
    ):
        root = tmp_path / name / ".tht-dwh"
        root.mkdir(parents=True, mode=0o700)
        make_entry(root)
        calls = []
        pipeline = DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=root.parent,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output: calls.append("called"),
            build_lsh=lambda physical, output: None,
        )
        with pytest.raises(Exception, match="unbound"):
            pipeline.run()
        assert calls == []
        assert not (root / "OWNER.json").exists()

    allowed = tmp_path / "allowed"
    (allowed / ".tht-dwh" / "generations").mkdir(parents=True, mode=0o700)
    report = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=allowed,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("catalog"),
        build_lsh=lambda physical, output: _write_lsh([], physical, output),
    ).run()
    assert report.status == "succeeded"


def test_writer_rejects_unbound_legacy_artifacts_before_building(tmp_path):
    import pytest

    legacy = tmp_path / "artifacts" / "mschema" / "physical.yaml"
    legacy.parent.mkdir(parents=True)
    legacy.write_text("legacy")
    calls = []
    pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: calls.append("called"),
        build_lsh=lambda physical, output: None,
        current_physical=legacy,
    )
    with pytest.raises(Exception, match="legacy artifacts are unbound"):
        pipeline.run()
    assert calls == []
    assert not (tmp_path / ".tht-dwh" / "OWNER.json").exists()


def test_writer_rejects_dangling_legacy_symlinks_before_claim_or_callback(tmp_path):
    import pytest

    legacy = tmp_path / "artifacts" / "mschema" / "physical.yaml"
    legacy.parent.mkdir(parents=True)
    legacy.symlink_to(tmp_path / "missing-catalog")
    calls = []
    pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: calls.append("called"),
        build_lsh=lambda physical, output: None,
        current_physical=legacy,
    )
    with pytest.raises(Exception, match="legacy artifacts are unbound"):
        pipeline.run()
    assert calls == []
    assert not (tmp_path / ".tht-dwh" / "OWNER.json").exists()


def test_owner_publication_remains_on_locked_root_when_path_is_swapped(
    monkeypatch, tmp_path,
):
    import pytest
    import tht.jobs.dwh_pipeline as module

    real_replace = module.os.replace
    moved = tmp_path / "locked-root"
    replacement = tmp_path / ".tht-dwh"
    swapped = False

    def swapping_replace(source, destination, *args, **kwargs):
        nonlocal swapped
        if destination == "OWNER.json" and kwargs.get("dst_dir_fd") is not None:
            swapped = True
            replacement.rename(moved)
            replacement.mkdir(mode=0o700)
            (moved / "generation.lock").rename(replacement / "generation.lock")
        return real_replace(source, destination, *args, **kwargs)

    monkeypatch.setattr(module.os, "replace", swapping_replace)
    pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: (_ for _ in ()).throw(AssertionError("callback called")),
        build_lsh=lambda physical, output: None,
    )
    with pytest.raises(Exception):
        pipeline.run()
    assert swapped
    assert (moved / "OWNER.json").is_file()
    assert not (replacement / "OWNER.json").exists()
    assert (replacement / "generation.lock").is_file()


def test_owner_requires_exact_read_only_owner_mode_and_active_requires_binding(tmp_path):
    import pytest

    pipeline = 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),
    )
    pipeline.run()
    cfg = snapshot_config(tmp_path)
    binding = config_dwh_binding(cfg)
    assert active_generation_dir(tmp_path, binding) is not None
    with pytest.raises(Exception, match="different workspace configuration"):
        active_generation_dir(tmp_path, {**binding, "workspace_id": "other"})

    marker = tmp_path / ".tht-dwh" / "OWNER.json"
    marker.chmod(0o440)
    with pytest.raises(Exception, match="ownership marker"):
        resolve_dwh_snapshot(cfg)


def test_selected_dwh_stages_run_in_declared_order(tmp_path):
    calls = []
    pipeline = DwhPreprocessPipeline(
        workspace_id="demo",
        workspace_root=tmp_path,
        config_fingerprint=FP,
        input_fingerprint=FP,
        introspect=lambda output: (calls.append("introspect"), output.write_text("catalog")),
        build_lsh=lambda physical, output: _write_lsh(calls, physical, output),
    )

    report = pipeline.run(("introspect", "lsh"))

    assert report.status == "succeeded"
    assert calls == ["introspect", "lsh"]
    assert [stage.name for stage in report.stages] == ["introspect", "lsh"]
    active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip()
    published = tmp_path / ".tht-dwh" / "generations" / active
    assert (published / "physical.yaml").read_text() == "catalog"
    assert sorted(path.name for path in published.iterdir()) == [
        "demo_lsh.pkl", "demo_meta.json", "demo_minhashes.pkl",
        "generation-manifest.json", "physical.yaml",
    ]


def test_shared_root_rejects_other_workspace_before_builder_or_read(tmp_path):
    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),
    )
    published = owner.run()
    contender = DwhPreprocessPipeline(
        workspace_id="other", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: calls.append("introspect"),
        build_lsh=lambda physical, output: calls.append("lsh"),
    )

    import pytest
    with pytest.raises(Exception, match="different workspace configuration"):
        contender.run()
    with pytest.raises(Exception, match="different workspace configuration"):
        resolve_dwh_snapshot(snapshot_config(tmp_path, "other"))

    assert calls == []
    assert resolve_dwh_snapshot(snapshot_config(tmp_path)).generation == published.run_id


def test_shared_root_mismatch_fails_without_deadlock_while_owner_reader_is_active(tmp_path):
    import threading

    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()
    contender = DwhPreprocessPipeline(
        workspace_id="other", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: (_ for _ in ()).throw(AssertionError("builder called")),
        build_lsh=lambda physical, output: None,
    )
    finished = threading.Event()
    errors = []
    with lease_dwh_snapshot(snapshot_config(tmp_path)):
        thread = threading.Thread(
            target=lambda: (errors.append(_capture_error(contender.run)), finished.set())
        )
        thread.start()
        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:
        return operation()
    except Exception as error:
        return error


def _write_lsh(calls, physical: Path, output: Path):
    calls.append("lsh")
    assert physical.read_text() == "catalog"
    for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"):
        (output / name).write_text(name)


def test_preprocess_dwh_json_is_pristine(monkeypatch, tmp_path):
    import tht.cli.preprocess_cmd as command

    class Report:
        status = "succeeded"

        def model_dump(self, mode=None):
            return {"status": "succeeded", "run_id": "a" * 32, "stages": []}

    seen = {}

    def run(config, *, steps, resume):
        seen.update(config=config, steps=steps, resume=resume)
        return Report()

    monkeypatch.setattr(command, "run_dwh_from_config", run)
    response = CliRunner().invoke(
        app,
        [
            "preprocess", "dwh", "--steps", "introspect,lsh", "--json",
            "-c", str(tmp_path / "workspace.yaml"),
        ],
    )

    assert response.exit_code == 0, response.output
    assert json.loads(response.output)["run_id"] == "a" * 32
    assert seen["steps"] == ("introspect", "lsh")


def test_preprocess_dwh_rejects_unknown_or_duplicate_steps(monkeypatch, tmp_path):
    import tht.cli.preprocess_cmd as command

    called = False

    def forbidden(*args, **kwargs):
        nonlocal called
        called = True

    monkeypatch.setattr(command, "run_dwh_from_config", forbidden)
    runner = CliRunner()
    for value in ("introspect,unknown", "lsh,lsh", ""):
        response = runner.invoke(
            app,
            ["preprocess", "dwh", "--steps", value, "--json", "-c", str(tmp_path / "w.yaml")],
        )
        assert response.exit_code == 2
        assert json.loads(response.output)["status"] == "failed"
    assert called is False


def test_failed_multi_file_build_never_replaces_active_generation(tmp_path):
    def catalog(output):
        output.write_text("old-catalog")

    first = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=catalog,
        build_lsh=lambda physical, output: [
            (output / name).write_text(name)
            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
        ],
    ).run(("introspect", "lsh"))
    assert first.status == "succeeded"
    old_active = (tmp_path / ".tht-dwh" / "ACTIVE").read_text()

    def partial_lsh(physical, output):
        (output / "demo_lsh.pkl").write_text("new-but-partial")
        raise RuntimeError("crash between LSH files")

    failed = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("new-catalog"),
        build_lsh=partial_lsh,
    ).run(("introspect", "lsh"))

    assert failed.status == "failed"
    assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text() == old_active

    resumed = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: (_ for _ in ()).throw(
            AssertionError("completed introspection must not repeat")
        ),
        build_lsh=lambda physical, output: [
            (output / name).write_text("recovered")
            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
        ],
    ).run(("introspect", "lsh"), resume_run_id=failed.run_id)
    assert resumed.status == "succeeded"


def test_unsafe_lsh_filename_is_rejected(tmp_path):
    import pytest

    with pytest.raises(ValueError, match="flat safe"):
        DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output: None, build_lsh=lambda physical, output: None,
            lsh_filenames=("../escape.pkl", "ok.pkl", "meta.json"),
        )


def test_active_fsync_failure_restores_previous_pointer(monkeypatch, tmp_path):
    import os
    import tht.jobs.dwh_pipeline as module
    def build(physical, output):
        for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json"):
            (output / name).write_text(name)

    first_pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("old"), build_lsh=build,
    )
    first = first_pipeline.run()
    root = tmp_path / ".tht-dwh"
    root_identity = (root.stat().st_dev, root.stat().st_ino)
    original_fsync = module.os.fsync
    failed_once = False

    def fail_active_once(fd):
        nonlocal failed_once
        info = os.fstat(fd)
        if (
            (info.st_dev, info.st_ino) == root_identity
            and "ACTIVE" in os.listdir(fd)
            and not failed_once
        ):
            failed_once = True
            raise OSError("injected directory fsync failure")
        original_fsync(fd)

    second = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("new"), build_lsh=build,
    )
    monkeypatch.setattr(module.os, "fsync", fail_active_once)
    failed = second.run()
    assert failed.status == "failed"
    assert (tmp_path / ".tht-dwh" / "ACTIVE").read_text().strip() == first.run_id


def test_snapshot_root_swap_after_lease_never_reads_replacement(monkeypatch, tmp_path):
    import tht.jobs.dwh_pipeline as module

    pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("trusted"),
        build_lsh=lambda physical, output: [
            (output / name).write_text("trusted")
            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
        ],
    )
    first = pipeline.run()
    assert first.status == "succeeded"
    root = tmp_path / ".tht-dwh"
    moved = tmp_path / "moved-read-root"
    replacement = root
    real_read = module._read_owned_at
    swapped = False

    def swapping_read(directory_fd, name, *, readonly):
        nonlocal swapped
        if name == "ACTIVE" and not swapped:
            swapped = True
            replacement.rename(moved)
            replacement.mkdir(mode=0o700)
            (replacement / "sentinel").write_text("replacement-secret")
        return real_read(directory_fd, name, readonly=readonly)

    monkeypatch.setattr(module, "_read_owned_at", swapping_read)
    try:
        with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot:
            assert snapshot.physical.read_text() == "trusted"
    except Exception as error:
        assert "ACTIVE" in str(error) or "root" in str(error)
    assert swapped
    assert (replacement / "sentinel").read_text() == "replacement-secret"


def test_snapshot_copies_each_validated_artifact_once_without_reopen(monkeypatch, tmp_path):
    import tht.jobs.dwh_pipeline as module

    pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("trusted"),
        build_lsh=lambda physical, output: [
            (output / name).write_text("trusted")
            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
        ],
    )
    assert pipeline.run().status == "succeeded"
    real_read = module._read_owned_at
    reads = {}

    def mutate_on_reopen(directory_fd, name, *, readonly):
        reads[name] = reads.get(name, 0) + 1
        if name.endswith(".pkl") and reads[name] > 1:
            return b"MALICIOUS_PICKLE"
        return real_read(directory_fd, name, readonly=readonly)

    monkeypatch.setattr(module, "_read_owned_at", mutate_on_reopen)
    with lease_dwh_snapshot(snapshot_config(tmp_path)) as snapshot:
        assert (snapshot.lsh_dir / "demo_lsh.pkl").read_text() == "trusted"
        assert "MALICIOUS" not in (snapshot.lsh_dir / "demo_lsh.pkl").read_text()
    assert all(count == 1 for count in reads.values())


def test_reconcile_mismatch_closes_active_generation_fd(monkeypatch, tmp_path):
    import os
    from types import SimpleNamespace
    import pytest
    import tht.jobs.dwh_pipeline as module

    pipeline = 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),
    )
    report = pipeline.run()
    run_dir = tmp_path / ".tht-jobs" / "dwh" / "runs" / report.run_id
    real_active = module._active_generation_fd

    def mismatched_active(root_fd, binding):
        generation, generation_fd = real_active(root_fd, binding)
        return "f" * 32, generation_fd

    monkeypatch.setattr(module, "_active_generation_fd", mismatched_active)
    source = SimpleNamespace(
        run_id=report.run_id,
        stages=(SimpleNamespace(
            status="running", effect_state="intent", name="lsh",
            artifact_files=("physical.yaml", "demo_lsh.pkl", "demo_minhashes.pkl",
                            "demo_meta.json"),
        ),),
    )
    before = len(os.listdir("/dev/fd"))
    with pytest.raises(Exception, match="not ACTIVE"):
        pipeline._reconcile_effects(source, run_dir)
    assert len(os.listdir("/dev/fd")) == before


def test_pipeline_releases_materialized_snapshot_after_every_run(tmp_path):
    import tht.jobs.dwh_pipeline as module

    pipeline = 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),
    )
    baseline = set(module._SNAPSHOT_DIRS)
    for _ in range(3):
        assert pipeline.run().status == "succeeded"
        assert set(module._SNAPSHOT_DIRS) == baseline
        assert pipeline._snapshot_holder is None
    pipeline.introspect = lambda output: (_ for _ in ()).throw(RuntimeError("injected"))
    assert pipeline.run().status == "failed"
    assert set(module._SNAPSHOT_DIRS) == baseline
    assert pipeline._snapshot_holder is None


def test_corrupt_resume_checkpoint_releases_materialized_snapshot(tmp_path):
    import tht.jobs.dwh_pipeline as module
    import pytest

    pipeline = 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),
    )
    assert pipeline.run().status == "succeeded"
    baseline = set(module._SNAPSHOT_DIRS)
    run_id = "e" * 32
    run_dir = tmp_path / ".tht-jobs" / "dwh" / "runs" / run_id
    run_dir.mkdir(parents=True)
    (run_dir / "checkpoint.json").write_text("not-json")

    with pytest.raises(Exception, match="checkpoint is invalid"):
        pipeline.run(resume_run_id=run_id)
    assert pipeline._snapshot_holder is None
    assert set(module._SNAPSHOT_DIRS) == baseline


def test_job_spec_construction_failure_releases_materialized_snapshot(monkeypatch, tmp_path):
    import tht.jobs.dwh_pipeline as module
    import pytest

    pipeline = 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),
    )
    assert pipeline.run().status == "succeeded"
    baseline = set(module._SNAPSHOT_DIRS)
    captured = []
    real_materialize = module._materialize_generation_fd

    def capture(*args, **kwargs):
        holder, root = real_materialize(*args, **kwargs)
        captured.append(holder)
        return holder, root

    monkeypatch.setattr(module, "_materialize_generation_fd", capture)
    monkeypatch.setattr(
        module, "JobSpec",
        lambda **kwargs: (_ for _ in ()).throw(RuntimeError("job spec injected")),
    )
    with pytest.raises(RuntimeError, match="job spec injected"):
        pipeline.run()
    assert pipeline._snapshot_holder is None
    assert set(module._SNAPSHOT_DIRS) == baseline
    assert captured and all(not path.exists() for path in captured)


def test_publish_root_swap_after_lease_never_writes_replacement(monkeypatch, tmp_path):
    import tht.jobs.dwh_pipeline as module

    def make(content):
        return DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output: output.write_text(content),
            build_lsh=lambda physical, output: [
                (output / name).write_text(content)
                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
            ],
        )

    first = make("old").run()
    assert first.status == "succeeded", first
    root = tmp_path / ".tht-dwh"
    moved = tmp_path / "moved-publish-root"
    real_replace = module.os.replace
    swapped = False

    def swapping_replace(source, destination, *args, **kwargs):
        nonlocal swapped
        if destination == "ACTIVE" and kwargs.get("dst_dir_fd") is not None and not swapped:
            swapped = True
            root.rename(moved)
            root.mkdir(mode=0o700)
            (root / "sentinel").write_text("replacement-safe")
        return real_replace(source, destination, *args, **kwargs)

    monkeypatch.setattr(module.os, "replace", swapping_replace)
    result = make("new").run()
    assert result.status in {"succeeded", "failed"}
    assert swapped
    assert (root / "sentinel").read_text() == "replacement-safe"
    moved_active = (moved / "ACTIVE").read_text().strip()
    assert len(moved_active) == 32
    assert (moved / "generations" / moved_active).is_dir()


def test_cleanup_root_swap_after_lease_never_deletes_replacement(monkeypatch, tmp_path):
    import tht.jobs.dwh_pipeline as module

    pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("trusted"),
        build_lsh=lambda physical, output: [
            (output / name).write_text("trusted")
            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
        ],
        retain_generations=1,
    )
    pipeline.run()
    root = tmp_path / ".tht-dwh"
    moved = tmp_path / "moved-cleanup-root"
    real_open = module.os.open
    swapped = False

    def swapping_open(path, flags, *args, **kwargs):
        nonlocal swapped
        if path == "generations" and kwargs.get("dir_fd") is not None and not swapped:
            swapped = True
            root.rename(moved)
            root.mkdir(mode=0o700)
            (root / "sentinel").write_text("replacement-safe")
        return real_open(path, flags, *args, **kwargs)

    monkeypatch.setattr(module.os, "open", swapping_open)
    pipeline._cleanup_generations()
    assert swapped
    assert (root / "sentinel").read_text() == "replacement-safe"


def test_snapshot_stays_on_one_generation_across_publish(tmp_path):
    def pipeline(content):
        return DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output: output.write_text(content),
            build_lsh=lambda physical, output: [
                (output / name).write_text(content)
                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
            ],
        )

    first = pipeline("old").run()
    cfg = snapshot_config(tmp_path)
    snapshot = resolve_dwh_snapshot(cfg)
    pipeline("new").run()
    assert snapshot.generation == first.run_id
    assert snapshot.physical.read_text() == "old"
    assert (snapshot.lsh_dir / "demo_meta.json").read_text() == "old"


def test_generation_retention_keeps_active_and_one_rollback(tmp_path):
    run_ids = []
    for index in range(5):
        report = DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output, i=index: output.write_text(str(i)),
            build_lsh=lambda physical, output, i=index: [
                (output / name).write_text(str(i))
                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
            ],
            retain_generations=2,
        ).run()
        run_ids.append(report.run_id)
    remaining = {path.name for path in (tmp_path / ".tht-dwh" / "generations").iterdir()}
    assert remaining == set(run_ids[-2:])


def test_corrupt_newer_directory_does_not_consume_rollback_slot(tmp_path):
    run_ids = []
    pipeline = None
    for index in range(3):
        pipeline = DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output, i=index: output.write_text(str(i)),
            build_lsh=lambda physical, output, i=index: [
                (output / name).write_text(str(i))
                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
            ],
            retain_generations=3,
        )
        run_ids.append(pipeline.run().run_id)
    generations = tmp_path / ".tht-dwh" / "generations"
    corrupt = generations / ("f" * 32)
    corrupt.mkdir(mode=0o700)
    (corrupt / "junk").write_text("not a published generation")

    pipeline.retain_generations = 2
    pipeline._cleanup_generations()

    assert (generations / run_ids[-1]).is_dir()
    assert (generations / run_ids[-2]).is_dir()
    assert not (generations / run_ids[0]).exists()
    assert corrupt.is_dir()


def test_retention_n_counts_active_plus_n_minus_one_rollbacks_even_if_active_is_old(tmp_path):
    import os

    run_ids = []
    pipeline = None
    for index in range(3):
        pipeline = DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output, i=index: output.write_text(str(i)),
            build_lsh=lambda physical, output, i=index: [
                (output / name).write_text(str(i))
                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
            ], retain_generations=3,
        )
        run_ids.append(pipeline.run().run_id)
    generations = tmp_path / ".tht-dwh" / "generations"
    os.utime(generations / run_ids[-1], ns=(1, 1))

    pipeline.retain_generations = 2
    pipeline._cleanup_generations()

    remaining = {path.name for path in generations.iterdir() if path.is_dir()}
    assert remaining == {run_ids[-1], run_ids[-2]}


def test_retention_candidate_swap_to_symlink_is_never_followed(monkeypatch, tmp_path):
    import tht.jobs.dwh_pipeline as module

    pipeline = DwhPreprocessPipeline(
        workspace_id="demo", workspace_root=tmp_path,
        config_fingerprint=FP, input_fingerprint=FP,
        introspect=lambda output: output.write_text("active"),
        build_lsh=lambda physical, output: [
            (output / name).write_text("active")
            for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
        ], retain_generations=1,
    )
    pipeline.run()
    generations = tmp_path / ".tht-dwh" / "generations"
    candidate_name = "e" * 32
    candidate = generations / candidate_name
    candidate.mkdir(mode=0o700)
    external = tmp_path / "external-crafted"
    external.mkdir()
    sentinel = external / "sentinel"
    sentinel.write_text("must-not-read-or-mutate")
    real_open = module.os.open
    swapped = False

    def swapping_open(path, flags, *args, **kwargs):
        nonlocal swapped
        if path == candidate_name and kwargs.get("dir_fd") is not None and not swapped:
            swapped = True
            candidate.rmdir()
            candidate.symlink_to(external, target_is_directory=True)
        return real_open(path, flags, *args, **kwargs)

    monkeypatch.setattr(module.os, "open", swapping_open)
    pipeline._cleanup_generations()
    assert swapped
    assert sentinel.read_text() == "must-not-read-or-mutate"
    assert candidate.is_symlink()


def test_reader_lease_blocks_retain_one_publisher_until_file_reads_finish(tmp_path):
    import threading
    import time
    def make(content, retain=1):
        return DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output: output.write_text(content),
            build_lsh=lambda physical, output: [
                (output / name).write_text(content)
                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
            ], retain_generations=retain,
        )

    first = make("old").run()
    cfg = snapshot_config(tmp_path)
    completed = threading.Event()
    with lease_dwh_snapshot(cfg) as snapshot:
        thread = threading.Thread(target=lambda: (make("new").run(), completed.set()))
        thread.start()
        time.sleep(0.05)
        assert not completed.is_set()
        assert snapshot.physical.read_text() == "old"
        assert snapshot.generation == first.run_id
    thread.join(timeout=2)
    assert completed.is_set()
    assert not (tmp_path / ".tht-dwh" / "generations" / first.run_id).exists()


def test_cleanup_never_follows_top_level_or_child_symlinks(tmp_path):
    external = tmp_path / "external"
    external.mkdir()
    victim = external / "victim"
    victim.write_text("safe")

    def make(content, retain=1):
        return DwhPreprocessPipeline(
            workspace_id="demo", workspace_root=tmp_path,
            config_fingerprint=FP, input_fingerprint=FP,
            introspect=lambda output: output.write_text(content),
            build_lsh=lambda physical, output: [
                (output / name).write_text(content)
                for name in ("demo_lsh.pkl", "demo_minhashes.pkl", "demo_meta.json")
            ], retain_generations=retain,
        )

    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)
    (old / "hostile-link").symlink_to(victim)
    make("three").run()
    assert victim.read_text() == "safe"
    assert victim.stat().st_mode & 0o200