from __future__ import annotations

import gzip
import hashlib
import json
from pathlib import Path
from threading import Barrier, Thread

import pytest


SAMPLES = "wgs-samples:" + "a" * 64
REFERENCE = "wgs-reference:" + "b" * 64
COMBINE_ANALYSIS = "wgs-analysis:" + "c" * 64
INDEX_ANALYSIS = "wgs-analysis:" + "d" * 64


def _record(workspace: Path, path: Path, role: str) -> dict:
    return {
        "artifact": role,
        "role": role,
        "path": str(path.relative_to(workspace)),
        "full_path": str(path.resolve()),
        "size": path.stat().st_size,
        "size_bytes": path.stat().st_size,
        "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
    }


def _write_vcf(path: Path, samples: list[str], *, gvcf: bool = True) -> None:
    with gzip.open(path, "wt", encoding="utf-8") as handle:
        handle.write("##fileformat=VCFv4.2\n")
        if gvcf:
            handle.write('##ALT=<ID=NON_REF,Description="NON_REF">\n')
        handle.write(
            "#CHROM\tPOS\tID\tREF\tALT\tQUAL\tFILTER\tINFO\tFORMAT\t"
            + "\t".join(samples)
            + "\n"
        )
        alt = "<NON_REF>" if gvcf else "C"
        handle.write(
            f"chr1\t1\t.\tA\t{alt}\t50\tPASS\t.\tGT\t"
            + "\t".join("0/0" for _ in samples)
            + "\n"
        )


def _authorities(workspace: Path) -> tuple[Path, Path]:
    final = workspace / "final"
    combine_bundle = (
        final / ".wgs_gvcf_combine_runs" / "Cohort" / ("7" * 32)
    )
    combine_bundle.mkdir(parents=True)
    combined = (
        combine_bundle / "Cohort.wgs_gvcf_combine.cohort.combined.g.vcf.gz"
    )
    combined_index = Path(f"{combined}.tbi")
    combine_log = combine_bundle / "Cohort.wgs_gvcf_combine.cohort.combine.log"
    _write_vcf(combined, ["S1", "S2", "S3"])
    combined_index.write_bytes(b"TBI:combined")
    combine_log.write_text("CombineGVCFs exit=0\n", encoding="utf-8")
    combine = final / "Cohort.wgs_gvcf_combine.manifest.json"
    combine.write_text(
        json.dumps(
            {
                "manifest_version": "1.0",
                "stage": "wgs_gvcf_combine",
                "status": "success",
                "prefix": "Cohort",
                "job_id": "job-1",
                "run_id": "7" * 32,
                "workspace": str(workspace.resolve()),
                "wgs_sample_set_id": SAMPLES,
                "wgs_reference_id": REFERENCE,
                "wgs_analysis_id": COMBINE_ANALYSIS,
                "upstream_wgs_analysis_ids": {
                    "wgs_haplotype_calling": "wgs-analysis:" + "e" * 64,
                    "wgs_index": INDEX_ANALYSIS,
                },
                "metrics": {
                    "input_sample_ids": ["S1", "S2", "S3"],
                    "output_sample_ids": ["S1", "S2", "S3"],
                },
                "artifacts": {
                    "combined_gvcf": _record(
                        workspace, combined, "combined_gvcf"
                    ),
                    "combined_gvcf_index": _record(
                        workspace, combined_index, "combined_gvcf_index"
                    ),
                    "combine_log": _record(
                        workspace, combine_log, "combine_log"
                    ),
                },
            }
        ),
        encoding="utf-8",
    )

    index_bundle = final / ".wgs_index_runs" / "Cohort" / ("2" * 32)
    index_bundle.mkdir(parents=True)
    reference = index_bundle / "Cohort.wgs_index.reference.fasta"
    fai = Path(f"{reference}.fai")
    dictionary = reference.with_suffix(".dict")
    reference.write_text(">chr1\nAAAAAAAAAA\n", encoding="utf-8")
    fai.write_text("chr1\t10\t6\t10\t11\n", encoding="utf-8")
    dictionary.write_text("@HD\tVN:1.6\n@SQ\tSN:chr1\tLN:10\n", encoding="utf-8")
    index = final / "Cohort.wgs_index.manifest.json"
    index.write_text(
        json.dumps(
            {
                "manifest_version": "1.0",
                "stage": "wgs_index",
                "status": "success",
                "prefix": "Cohort",
                "job_id": "job-1",
                "run_id": "2" * 32,
                "workspace": str(workspace.resolve()),
                "wgs_reference_id": REFERENCE,
                "wgs_analysis_id": INDEX_ANALYSIS,
                "artifacts": {
                    "reference_fasta": _record(
                        workspace, reference, "reference_fasta"
                    ),
                    "reference_fai": _record(
                        workspace, fai, "reference_fai"
                    ),
                    "reference_dict": _record(
                        workspace, dictionary, "reference_dict"
                    ),
                },
            }
        ),
        encoding="utf-8",
    )
    return combine, index


def test_stage08_rejects_producer_from_different_index_generation_at_all_gates(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main
    from apps.web import pipeline_backend
    from gfa_core.mas.tools.stage_status import get_stage_status

    workspace_root = tmp_path / "workspaces"
    workspace = workspace_root / "job-1"
    combine, index = _authorities(workspace)
    gatk = _gatk(tmp_path)
    assert main(_argv(workspace, combine, index, gatk)) == 0
    output = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    output_analysis = json.loads(output.read_text())["wgs_analysis_id"]

    producer = json.loads(combine.read_text())
    producer["upstream_wgs_analysis_ids"]["wgs_index"] = (
        "wgs-analysis:" + "0" * 64
    )
    combine.write_text(json.dumps(producer))
    assert main(_argv(workspace, combine, index, gatk)) == 1

    called = {"value": False}
    def forbidden_run(*_args, **_kwargs):
        called["value"] = True
        return 0
    monkeypatch.setattr(pipeline_backend, "run_cmd_logged", forbidden_run)
    message, _ = pipeline_backend.run_stage(
        stage="wgs_joint_genotyping",
        job=pipeline_backend.JobInfo(
            "job-1", "", str(workspace), str(workspace / "logs"), "now"
        ),
        prefix="Cohort",
        wgs_combine_manifest=str(combine),
        wgs_index_manifest=str(index),
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_combine_analysis_id=COMBINE_ANALYSIS,
        wgs_index_analysis_id=INDEX_ANALYSIS,
        wgs_gatk=str(gatk),
    )
    assert "lineage" in message.lower()
    assert not called["value"]
    assert not get_stage_status(
        "job-1", "Cohort", "wgs_joint_genotyping", workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id=output_analysis,
    ).completed


def _gatk(tmp_path: Path) -> Path:
    tool = tmp_path / "gatk"
    tool.write_text(
        """#!/usr/bin/env python3
import gzip, json, os, pathlib, sys
args = sys.argv[1:]
if args == ['--version']:
    print('The Genome Analysis Toolkit (GATK) v4.6.0.0')
    raise SystemExit(0)
command = args[0]
def value(flag):
    return args[args.index(flag) + 1]
calls = os.environ.get('GATK_CALLS')
if calls:
    with open(calls, 'a', encoding='utf-8') as handle:
        handle.write(json.dumps(args) + '\\n')
if os.environ.get('FAIL_COMMAND') == command:
    raise SystemExit(23)
if command == 'ValidateVariants':
    path = pathlib.Path(value('-V'))
    index = pathlib.Path(str(path) + '.tbi')
    if not path.is_file() or not index.is_file():
        raise SystemExit(2)
    if index.read_bytes().startswith(b'CORRUPT'):
        raise SystemExit(24)
    raise SystemExit(0)
if command == 'GenotypeGVCFs':
    source = pathlib.Path(value('-V'))
    with gzip.open(source, 'rt') as handle:
        header = next(line for line in handle if line.startswith('#CHROM'))
    samples = header.rstrip().split('\\t')[9:]
    if os.environ.get('BAD_OUTPUT_HEADER'):
        samples[-1] = 'WRONG'
    output = pathlib.Path(value('-O'))
    with gzip.open(output, 'wt') as handle:
        handle.write('##fileformat=VCFv4.2\\n')
        handle.write('#CHROM\\tPOS\\tID\\tREF\\tALT\\tQUAL\\tFILTER\\tINFO'
                     '\\tFORMAT\\t' + '\\t'.join(samples) + '\\n')
        handle.write('chr1\\t1\\t.\\tA\\tC\\t50\\tPASS\\t.\\tGT\\t'
                     + '\\t'.join('0/0' for _ in samples) + '\\n')
    raise SystemExit(0)
if command == 'IndexFeatureFile':
    path = pathlib.Path(value('-I'))
    content = b'CORRUPT:output' if os.environ.get('CORRUPT_OUTPUT_INDEX') else b'TBI:joint'
    pathlib.Path(str(path) + '.tbi').write_bytes(content)
    raise SystemExit(0)
if command == 'VariantsToTable':
    pathlib.Path(value('-O')).write_text(
        'CHROM\\tPOS\\tREF\\tALT\\tQUAL\\nchr1\\t1\\tA\\tC\\t50\\n',
        encoding='utf-8',
    )
    raise SystemExit(0)
raise SystemExit(2)
""",
        encoding="utf-8",
    )
    tool.chmod(0o755)
    return tool


def _argv(
    workspace: Path, combine: Path, index: Path, gatk: Path
) -> list[str]:
    return [
        "--workspace",
        str(workspace),
        "--job-id",
        "job-1",
        "--prefix",
        "Cohort",
        "--combine-manifest",
        str(combine),
        "--index-manifest",
        str(index),
        "--gatk",
        str(gatk),
    ]


def test_wgs_joint_genotyping_exposes_import_safe_cli() -> None:
    from apps.cli.run_wgs_joint_genotyping import build_parser, main

    assert callable(build_parser)
    assert callable(main)


def test_parser_locks_scientific_defaults() -> None:
    from apps.cli.run_wgs_joint_genotyping import build_parser

    parsed = build_parser().parse_args(
        [
            "--workspace", "w", "--job-id", "j", "--prefix", "P",
            "--combine-manifest", "c", "--index-manifest", "i",
        ]
    )
    assert parsed.max_alternate_alleles == 6
    assert parsed.calling_threshold == 10
    assert parsed.include_non_variant_sites is False


def test_three_samples_produce_only_canonical_outputs_and_parameter_snapshot(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    calls = tmp_path / "gatk.calls"
    monkeypatch.setenv("GATK_CALLS", str(calls))
    assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 0
    authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    manifest = json.loads(authority.read_text())
    assert manifest["metrics"]["input_sample_ids"] == ["S1", "S2", "S3"]
    assert manifest["metrics"]["output_sample_ids"] == ["S1", "S2", "S3"]
    assert manifest["upstream_wgs_analysis_ids"] == {
        "wgs_gvcf_combine": COMBINE_ANALYSIS,
        "wgs_index": INDEX_ANALYSIS,
    }
    assert manifest["parameters"] == {
        "gatk": str(_gatk(tmp_path)),
        "operation": "GenotypeGVCFs",
        "max_alternate_alleles": 6,
        "calling_threshold": 10,
        "include_non_variant_sites": False,
    }
    assert set(manifest["artifacts"]) == {
        "joint_vcf", "joint_vcf_index", "joint_summary", "joint_log"
    }
    paths = {
        item["role"]: Path(item["full_path"])
        for item in manifest["artifacts"].values()
    }
    assert paths["joint_vcf"].name == (
        "Cohort.wgs_joint_genotyping.cohort.joint.vcf.gz"
    )
    assert paths["joint_vcf_index"] == Path(f"{paths['joint_vcf']}.tbi")
    assert paths["joint_summary"].read_text().splitlines() == [
        "CHROM\tPOS\tREF\tALT\tQUAL", "chr1\t1\tA\tC\t50"
    ]
    with gzip.open(paths["joint_vcf"], "rt") as handle:
        header = next(line for line in handle if line.startswith("#CHROM"))
    assert header.rstrip().split("\t")[9:] == ["S1", "S2", "S3"]
    invocations = [json.loads(line) for line in calls.read_text().splitlines()]
    genotype = next(call for call in invocations if call[0] == "GenotypeGVCFs")
    assert genotype[genotype.index("--max-alternate-alleles") + 1] == "6"
    assert genotype[genotype.index("--standard-min-confidence-threshold-for-calling") + 1] == "10"
    assert genotype[genotype.index("--include-non-variant-sites") + 1] == "false"
    assert any(call[0] == "ValidateVariants" for call in invocations)


def test_corrupt_nonempty_input_tbi_is_rejected_by_indexed_query(
    tmp_path: Path
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    payload = json.loads(combine.read_text())
    record = payload["artifacts"]["combined_gvcf_index"]
    tbi = Path(record["full_path"])
    tbi.write_bytes(b"CORRUPT:input")
    record["size"] = record["size_bytes"] = tbi.stat().st_size
    record["sha256"] = hashlib.sha256(tbi.read_bytes()).hexdigest()
    combine.write_text(json.dumps(payload))
    assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1


@pytest.mark.parametrize(
    ("flag", "value"),
    (
        ("--max-alternate-alleles", "7"),
        ("--calling-threshold", "11"),
        ("--include-non-variant-sites", "true"),
    ),
)
def test_scientific_parameters_are_locked(
    tmp_path: Path, flag: str, value: str
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    argv = _argv(workspace, combine, index, _gatk(tmp_path))
    argv.extend([flag, value])
    assert main(argv) == 1


def test_declared_index_must_be_the_vcf_sidecar(tmp_path: Path) -> None:
    from apps.cli.run_wgs_joint_genotyping import main

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    payload = json.loads(combine.read_text())
    record = payload["artifacts"]["combined_gvcf_index"]
    unrelated = Path(record["full_path"]).with_name("unrelated.tbi")
    unrelated.write_bytes(b"TBI:unrelated")
    record.update(
        {
            "path": str(unrelated.relative_to(workspace)),
            "full_path": str(unrelated),
            "size": unrelated.stat().st_size,
            "size_bytes": unrelated.stat().st_size,
            "sha256": hashlib.sha256(unrelated.read_bytes()).hexdigest(),
        }
    )
    combine.write_text(json.dumps(payload))
    assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1


@pytest.mark.parametrize(
    "failure", ("GenotypeGVCFs", "IndexFeatureFile", "VariantsToTable")
)
def test_each_tool_return_code_fails_and_preserves_old_authority(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    authority.write_text(
        json.dumps({"stage": "wgs_joint_genotyping", "status": "success", "sentinel": True})
    )
    before = hashlib.sha256(authority.read_bytes()).hexdigest()
    monkeypatch.setenv("FAIL_COMMAND", failure)
    assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1
    assert hashlib.sha256(authority.read_bytes()).hexdigest() == before
    sidecars = list(
        (workspace / "final").glob(
            "Cohort.wgs_joint_genotyping.failed.*.manifest.json"
        )
    )
    assert len(sidecars) == 1
    assert json.loads(sidecars[0].read_text())["artifacts"] == {}


@pytest.mark.parametrize("environment", ("BAD_OUTPUT_HEADER", "CORRUPT_OUTPUT_INDEX"))
def test_output_content_or_index_failure_is_not_published(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, environment: str
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    monkeypatch.setenv(environment, "1")
    assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1
    failure = json.loads(
        (workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json").read_text()
    )
    assert failure["status"] == "failed"
    assert failure["artifacts"] == {}


@pytest.mark.parametrize(
    "mutation", ("combine_analysis", "reference", "path", "symlink", "roles")
)
def test_two_upstream_authorities_are_strictly_bound(
    tmp_path: Path, mutation: str
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    if mutation == "combine_analysis":
        payload = json.loads(combine.read_text())
        payload.pop("wgs_analysis_id")
        combine.write_text(json.dumps(payload))
    elif mutation == "reference":
        payload = json.loads(index.read_text())
        payload["wgs_reference_id"] = "wgs-reference:" + "f" * 64
        index.write_text(json.dumps(payload))
    elif mutation == "path":
        payload = json.loads(combine.read_text())
        payload["artifacts"]["combined_gvcf"]["path"] = "../escape.vcf.gz"
        combine.write_text(json.dumps(payload))
    elif mutation == "symlink":
        payload = json.loads(combine.read_text())
        record = payload["artifacts"]["combined_gvcf"]
        original = Path(record["full_path"])
        link = original.with_name("linked.g.vcf.gz")
        link.symlink_to(original)
        record["path"] = str(link.relative_to(workspace))
        record["full_path"] = str(link)
        combine.write_text(json.dumps(payload))
    else:
        payload = json.loads(combine.read_text())
        payload["artifacts"].pop("combined_gvcf_index")
        combine.write_text(json.dumps(payload))
    assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 1


def test_status_and_results_require_exact_identity_and_singleton_roles(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    import asyncio

    from apps.cli.run_wgs_joint_genotyping import main
    from apps.web.vue3 import main as web_main
    from gfa_core.mas.tools.stage_status import get_stage_status

    workspace_root = tmp_path / "workspaces"
    workspace = workspace_root / "job-1"
    combine, index = _authorities(workspace)
    assert main(_argv(workspace, combine, index, _gatk(tmp_path))) == 0
    manifest_path = (
        workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    )
    manifest = json.loads(manifest_path.read_text())
    analysis_id = manifest["wgs_analysis_id"]
    status = get_stage_status(
        "job-1",
        "Cohort",
        "wgs_joint_genotyping",
        workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id=analysis_id,
    )
    assert status.completed
    monkeypatch.setattr(web_main, "WORKSPACE_ROOT", workspace_root)
    results = asyncio.run(
        web_main.get_results(
            job_id="job-1",
            stage="wgs_joint_genotyping",
            wgs_sample_set_id=SAMPLES,
            wgs_reference_id=REFERENCE,
            wgs_analysis_id=analysis_id,
        )
    )
    serialized = json.dumps(results)
    assert "joint_vcf" in serialized
    assert "joint_summary" in serialized
    manifest["artifacts"].pop("joint_vcf_index")
    manifest_path.write_text(json.dumps(manifest))
    assert not get_stage_status(
        "job-1",
        "Cohort",
        "wgs_joint_genotyping",
        workspace_root,
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_analysis_id=analysis_id,
    ).completed


def test_backend_builds_stage08_command_and_accepts_only_fresh_manifest(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    from apps.cli.run_wgs_joint_genotyping import main
    from apps.web import pipeline_backend

    workspace = tmp_path / "workspace"
    combine, index = _authorities(workspace)
    gatk = _gatk(tmp_path)
    calls: list[list[str]] = []

    def fake_run(command, *_args, **_kwargs):
        calls.append(command)
        return main(command[3:])

    monkeypatch.setattr(pipeline_backend, "run_cmd_logged", fake_run)
    message, output = pipeline_backend.run_stage(
        stage="wgs_joint_genotyping",
        job=pipeline_backend.JobInfo(
            "job-1", "", str(workspace), str(workspace / "logs"), "now"
        ),
        prefix="Cohort",
        wgs_combine_manifest=str(combine),
        wgs_index_manifest=str(index),
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_combine_analysis_id=COMBINE_ANALYSIS,
        wgs_index_analysis_id=INDEX_ANALYSIS,
        wgs_gatk=str(gatk),
        wgs_max_alternate_alleles=6,
        wgs_calling_threshold=10,
        wgs_include_non_variant_sites=False,
    )
    assert "completed" in message.lower()
    assert output == (
        workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    )
    command = calls[0]
    assert command[:3] == [
        pipeline_backend.sys.executable,
        "-m",
        "apps.cli.run_wgs_joint_genotyping",
    ]
    assert command[command.index("--combine-manifest") + 1] == str(combine)
    assert command[command.index("--max-alternate-alleles") + 1] == "6"
    assert command[command.index("--calling-threshold") + 1] == "10"
    assert command[command.index("--include-non-variant-sites") + 1] == "false"
    before = output.read_bytes()
    monkeypatch.setattr(
        pipeline_backend, "run_cmd_logged", lambda *_args, **_kwargs: 0
    )
    stale_message, stale_output = pipeline_backend.run_stage(
        stage="wgs_joint_genotyping",
        job=pipeline_backend.JobInfo(
            "job-1", "", str(workspace), str(workspace / "logs"), "now"
        ),
        prefix="Cohort",
        wgs_combine_manifest=str(combine),
        wgs_index_manifest=str(index),
        wgs_sample_set_id=SAMPLES,
        wgs_reference_id=REFERENCE,
        wgs_combine_analysis_id=COMBINE_ANALYSIS,
        wgs_index_analysis_id=INDEX_ANALYSIS,
        wgs_gatk=str(gatk),
    )
    assert "failed" in stale_message.lower()
    assert stale_output == output
    assert output.read_bytes() == before


def test_publish_lock_prevents_older_attempt_replacing_newer(
    tmp_path: Path
) -> None:
    from apps.cli.run_wgs_joint_genotyping import _publish

    workspace = tmp_path / "workspace"
    authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    authority.parent.mkdir(parents=True)
    authority.write_text(
        json.dumps({"status": "success", "start_time": "2026-07-23T12:00:00Z"})
    )
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    (bundle / "artifact").write_text("older")
    manifest_work = tmp_path / "manifest.json"
    manifest_work.write_text("{}")
    assert not _publish(
        bundle,
        workspace / "final" / ".wgs_joint_genotyping_runs" / "Cohort" / ("9" * 32),
        manifest_work,
        authority,
        "2026-07-23T11:00:00Z",
    )


def test_publish_lock_rejects_equal_timestamp_attempt(tmp_path: Path) -> None:
    from apps.cli.run_wgs_joint_genotyping import _publish

    workspace = tmp_path / "workspace"
    authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    authority.parent.mkdir(parents=True)
    timestamp = "2026-07-23T12:00:00Z"
    authority.write_text(
        json.dumps({"status": "success", "start_time": timestamp})
    )
    bundle = tmp_path / "bundle"
    bundle.mkdir()
    (bundle / "artifact").write_text("equal")
    manifest_work = tmp_path / "manifest.json"
    manifest_work.write_text("{}")
    assert not _publish(
        bundle,
        workspace / "final" / ".wgs_joint_genotyping_runs" / "Cohort" / ("8" * 32),
        manifest_work,
        authority,
        timestamp,
    )


def test_concurrent_publish_has_one_complete_authority(tmp_path: Path) -> None:
    from apps.cli.run_wgs_joint_genotyping import _publish

    workspace = tmp_path / "workspace"
    authority = workspace / "final" / "Cohort.wgs_joint_genotyping.manifest.json"
    authority.parent.mkdir(parents=True)
    barrier = Barrier(2)

    def worker(name: str, started_at: str) -> None:
        bundle = tmp_path / f"bundle-{name}"
        bundle.mkdir()
        (bundle / "artifact").write_text(name)
        manifest = tmp_path / f"manifest-{name}.json"
        manifest.write_text(
            json.dumps({"status": "success", "start_time": started_at, "winner": name})
        )
        barrier.wait()
        _publish(
            bundle,
            workspace / "final" / ".wgs_joint_genotyping_runs" / "Cohort" / name,
            manifest,
            authority,
            started_at,
        )

    threads = [
        Thread(target=worker, args=("a" * 32, "2026-07-23T11:00:00Z")),
        Thread(target=worker, args=("b" * 32, "2026-07-23T12:00:00Z")),
    ]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    assert json.loads(authority.read_text())["winner"] == "b" * 32