from __future__ import annotations

import hashlib
import io
import json
import os
import sys
from contextlib import contextmanager, nullcontext
from email.message import Message
from types import SimpleNamespace

import pytest
from test_deploy_broker import BROKER_ROOT, _load_module
from test_projectbrain_delivery_revalidation import _git

REPOSITORY = "jens/itworx_mcp_hub"
REVISION = "0253e7acef01ad952266ff180ed434e22b986ddc"
PREFIX = "/v1/appops/hub-delivery/"
KEY = "hub-delivery-request-2026-10-08"


@pytest.fixture
def controller(monkeypatch):
    module = _load_module("hub_delivery_revalidation_test", BROKER_ROOT / "status_server.py")
    monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True)
    monkeypatch.setattr(module, "_self_upgrade_secret", lambda: b"hub-delivery-test-key-" + b"x" * 40)
    monkeypatch.setattr(module, "_require_controller_release", lambda: {"artifactSha256": "f" * 64})
    return module


def invoke(module, stage: str, payload: dict[str, object], *, trusted: bool = True):
    handler = module.Handler.__new__(module.Handler)
    body = json.dumps(payload, separators=(",", ":")).encode()
    handler.path = PREFIX + stage
    handler.headers = Message()
    handler.headers["Content-Length"] = str(len(body))
    handler.headers["Content-Type"] = "application/json"
    handler.rfile = io.BytesIO(body)
    handler._trusted_client = lambda: trusted
    responses = []
    handler._send_json = lambda status, data: responses.append((status, json.loads(data)))
    handler._single_header = lambda name: (
        "0" * 32 if name == "X-ITWorx-Deployment-Nonce"
        else "0" if name in {"X-ITWorx-Deployment-Timestamp", "X-ITWorx-Deployment-Signature"}
        else handler.headers.get(name)
    )
    handler._handle_hub_delivery()
    return responses[0]


def _snapshot(module, *, preflight_sha: str = "a" * 64):
    return {
        "status": "ready",
        "repository": REPOSITORY,
        "revision": REVISION,
        "originalDelivery": "c" * 64,
        "validationRunId": 22495,
        "validationRunNumber": 22495,
        "validationRunAttempt": 1,
        "validationJobId": 39309,
        "validationJobCount": 2,
        "validationStepCount": 12,
        "validationStatusId": 4,
        "preflightSha256": preflight_sha,
        "_semanticDigest": "c" * 64,
        "_originalFileSha256": "b" * 64,
        "_project": {"slug": "itworx-mcp-hub", "branch": "main"},
        "_validationPolicy": {"contexts": [module.HUB_DELIVERY_FULL_CONTEXT]},
        "_stateSha256": "1" * 64,
        "_migrationIntentSha256": "2" * 64,
        "_baselineSha256": "5" * 64,
        "_profileSha256": "3" * 64,
        "_controllerArtifactSha256": "f" * 64,
        "_validationEvidenceSha256": "4" * 64,
    }


def test_preflight_is_fixed_read_only_and_returns_bounded_evidence(controller, monkeypatch):
    module = controller
    calls = []
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: calls.append(revision) or _snapshot(module))
    monkeypatch.setattr(
        module,
        "_enqueue_hub_terminal_revalidation",
        lambda *_a, **_k: pytest.fail("preflight must never enqueue"),
    )

    code, result = invoke(module, "preflight", {"revision": REVISION})

    assert code == 200
    assert calls == [REVISION]
    assert result == {
        "status": "ready",
        "repository": REPOSITORY,
        "revision": REVISION,
        "originalDelivery": "c" * 64,
        "validationRunId": 22495,
        "validationRunNumber": 22495,
        "validationRunAttempt": 1,
        "validationJobId": 39309,
        "validationJobCount": 2,
        "validationStepCount": 12,
        "validationStatusId": 4,
        "preflightSha256": "a" * 64,
        "noMutation": True,
    }


def test_preflight_auth_and_exact_contract_precede_all_evidence_reads(controller, monkeypatch):
    module = controller
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda *_a: pytest.fail("unauthorized or malformed input read state"))
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", lambda: pytest.fail("lock acquired for malformed input"))
    monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: False)
    assert invoke(module, "preflight", {"revision": REVISION})[0] == 403

    monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True)
    assert invoke(module, "preflight", {"revision": REVISION, "repository": REPOSITORY})[0] == 400
    assert invoke(module, "preflight", {"revision": "A" * 40})[0] == 400


def test_revalidate_recomputes_cas_before_one_enqueue(controller, monkeypatch):
    module = controller
    snap = _snapshot(module)
    queued = []
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: snap)
    monkeypatch.setattr(module, "_hub_delivery_signed_binding", lambda snapshot, key_sha: {"key": key_sha})
    monkeypatch.setattr(
        module,
        "_enqueue_hub_terminal_revalidation",
        lambda revision, project, binding, original_sha: queued.append(
            (revision, project, binding, original_sha)
        ) or ("awaiting_validation", "c" * 64, True),
    )

    code, result = invoke(module, "revalidate", {
        "revision": REVISION,
        "expectedPreflightSha256": "a" * 64,
        "idempotencyKey": KEY,
    })

    assert code == 202
    assert result["status"] == "awaiting_validation"
    assert result["noMutation"] is False
    assert queued == [(REVISION, snap["_project"], {"key": hashlib.sha256(KEY.encode()).hexdigest()}, "b" * 64)]


def test_revalidate_stale_cas_has_explicit_no_mutation_and_does_not_enqueue(controller, monkeypatch):
    module = controller
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: _snapshot(module))
    monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: pytest.fail("stale CAS enqueued"))

    code, result = invoke(module, "revalidate", {
        "revision": REVISION,
        "expectedPreflightSha256": "9" * 64,
        "idempotencyKey": KEY,
    })

    assert code == 409
    assert result == {"status": "blocked", "reason": "hub_delivery_preflight_changed", "noMutation": True}


def test_revalidate_same_key_replays_progressed_sibling_without_new_snapshot(controller, monkeypatch):
    module = controller
    expected = {"status": "deployed", "delivery": "c" * 64}
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *args: expected)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda *_a: pytest.fail("replay must not require a fresh snapshot"))
    monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: pytest.fail("replay created another sibling"))

    code, result = invoke(module, "revalidate", {
        "revision": REVISION,
        "expectedPreflightSha256": "a" * 64,
        "idempotencyKey": KEY,
    })

    assert code == 200
    assert result == {**expected, "repository": REPOSITORY, "revision": REVISION, "noMutation": True}


def test_revalidate_write_ambiguity_is_unknown_not_definite_failure(controller, monkeypatch):
    module = controller
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: _snapshot(module))
    monkeypatch.setattr(module, "_hub_delivery_signed_binding", lambda *_a: {"binding": "signed"})
    monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: (_ for _ in ()).throw(OSError("fsync failed")))

    code, result = invoke(module, "revalidate", {
        "revision": REVISION,
        "expectedPreflightSha256": "a" * 64,
        "idempotencyKey": KEY,
    })

    assert code == 503
    assert result == {"status": "unknown", "reason": "hub_delivery_enqueue_outcome_unknown", "noMutation": False}


def test_revalidate_wakeup_failure_after_durable_enqueue_is_reported_unknown(controller, monkeypatch):
    module = controller
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_existing_revalidation", lambda *_a: None)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda revision: _snapshot(module))
    monkeypatch.setattr(module, "_hub_delivery_signed_binding", lambda *_a: {"binding": "signed"})
    monkeypatch.setattr(module, "_enqueue_hub_terminal_revalidation", lambda *_a: ("awaiting_validation", "c" * 64, True))
    monkeypatch.setattr(module, "_webhook_wakeup", SimpleNamespace(set=lambda: (_ for _ in ()).throw(OSError("wake failed"))))

    code, result = invoke(module, "revalidate", {
        "revision": REVISION,
        "expectedPreflightSha256": "a" * 64,
        "idempotencyKey": KEY,
    })

    assert code == 503
    assert result == {"status": "unknown", "reason": "hub_delivery_enqueue_outcome_unknown", "noMutation": False}


def test_hub_binding_is_hmac_bound_and_rejects_tampering(controller, monkeypatch):
    module = controller
    snapshot = _snapshot(module)
    binding = module._hub_delivery_signed_binding(snapshot, hashlib.sha256(KEY.encode()).hexdigest())
    assert module._hub_delivery_binding_is_valid(binding, REPOSITORY, REVISION)
    assert binding["schema"] == module.HUB_DELIVERY_BINDING_SCHEMA
    assert not module._hub_delivery_binding_is_valid(
        {**binding, "stateSha256": "9" * 64}, REPOSITORY, REVISION,
    )
    assert not module._hub_delivery_binding_is_valid(
        {**binding, "repository": "jens/other"}, REPOSITORY, REVISION,
    )


def test_worker_rejects_unbound_hub_revalidation_before_candidate_pipeline(controller):
    module = controller
    assert not module._hub_delivery_worker_admission(
        {"schema": module._WEBHOOK_REVALIDATION_SCHEMA}, REPOSITORY, REVISION,
    )


def test_worker_rechecks_hmac_bound_snapshot_before_candidate_import(controller, monkeypatch):
    module = controller
    snapshot = _snapshot(module)
    key_sha = hashlib.sha256(KEY.encode()).hexdigest()
    binding = module._hub_delivery_signed_binding(snapshot, key_sha)
    record = {
        "schema": module._WEBHOOK_REVALIDATION_SCHEMA,
        "semantic_sha256": binding["semanticSha256"],
        "revalidation_of": binding["semanticSha256"],
        "hub_delivery_revalidation": binding,
    }
    changed = {**snapshot, "preflightSha256": "9" * 64}
    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda *_a: changed)

    assert not module._hub_delivery_worker_admission(record, REPOSITORY, REVISION)


def test_final_hub_admission_persists_deployment_start_under_one_lock(controller, monkeypatch, tmp_path):
    module = controller
    semantic = module._webhook_semantic_digest(7, "refs/heads/main", REVISION)
    snapshot = {**_snapshot(module), "_semanticDigest": semantic}
    binding = module._hub_delivery_signed_binding(
        snapshot, hashlib.sha256(KEY.encode()).hexdigest(),
    )
    record = {
        "schema": module._WEBHOOK_REVALIDATION_SCHEMA,
        "semantic_sha256": semantic,
        "revalidation_of": semantic,
        "hub_delivery_revalidation": binding,
        "status": "validation_passed",
        "audit": [],
    }
    marker = tmp_path / f"{semantic}.revalidation.json"
    lock_depth = []
    persisted = []

    @contextmanager
    def mutation_lock():
        lock_depth.append(True)
        try:
            yield
        finally:
            lock_depth.pop()

    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", mutation_lock)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda _revision: snapshot)
    monkeypatch.setattr(
        module, "_atomic_inbox_update",
        lambda _marker, updated: persisted.append((bool(lock_depth), json.loads(json.dumps(updated)))),
    )

    assert module._hub_delivery_finalize_worker_admission(
        marker, record, REPOSITORY, REVISION,
    )
    assert len(persisted) == 1
    assert persisted[0][0]
    assert persisted[0][1]["status"] == "deploying"
    assert persisted[0][1]["audit"][-1]["stage"] == "deployment_started"


@pytest.mark.parametrize(
    "drift_field",
    [
        "_stateSha256",
        "_profileSha256",
        "_controllerArtifactSha256",
        "_validationEvidenceSha256",
        "_originalFileSha256",
    ],
)
def test_worker_rechecks_hub_binding_after_candidate_prep_before_deployment(
    controller, monkeypatch, tmp_path, drift_field,
):
    module = controller
    semantic = module._webhook_semantic_digest(7, "refs/heads/main", REVISION)
    snapshot = {**_snapshot(module), "_semanticDigest": semantic}
    binding = module._hub_delivery_signed_binding(
        snapshot, hashlib.sha256(KEY.encode()).hexdigest(),
    )
    changed = {
        **snapshot,
        drift_field: "9" * 64,
        "preflightSha256": "e" * 64,
    }
    snapshot_reads = 0
    preparation_calls = []
    deployment_calls = []
    persisted = []
    project = {"slug": "itworx-mcp-hub"}
    record = {
        "schema": module._WEBHOOK_REVALIDATION_SCHEMA,
        "semantic_sha256": semantic,
        "revalidation_of": semantic,
        "body_sha256": "6" * 64,
        "gitea_repository_id": 7,
        "ref": "refs/heads/main",
        "repository": REPOSITORY,
        "revision": REVISION,
        "status": "validation_passed",
        "validation": {},
        "audit": [],
        "hub_delivery_revalidation": binding,
    }
    marker = tmp_path / f"{semantic}.revalidation.json"

    def read_snapshot(_revision):
        nonlocal snapshot_reads
        snapshot_reads += 1
        return snapshot if snapshot_reads == 1 else changed

    monkeypatch.setattr(module, "_hub_delivery_mutation_lock", nullcontext)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", read_snapshot)
    monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project})
    monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (7, "main"))
    monkeypatch.setattr(
        module, "_validation_policy",
        lambda *_a: {"contexts": [module.HUB_DELIVERY_FULL_CONTEXT]},
    )
    monkeypatch.setattr(module, "_record_policy_for_processing", lambda *_a: True)
    monkeypatch.setattr(module, "_is_current_branch_head", lambda *_a: True)
    monkeypatch.setattr(module, "_supersede_older_deliveries", lambda *_a: None)
    monkeypatch.setattr(module, "_read_inbox_record", lambda _marker: record)
    monkeypatch.setattr(
        module, "_atomic_inbox_update",
        lambda _marker, updated: persisted.append(json.loads(json.dumps(updated))),
    )
    monkeypatch.setattr(
        module, "_prepare_hub_candidate_delivery",
        lambda *_a: preparation_calls.append(True) or True,
    )
    monkeypatch.setattr(
        module, "_run_registered_deploy",
        lambda *_a, **_kw: deployment_calls.append(True) or SimpleNamespace(returncode=0),
    )

    module._process_webhook_record(marker)

    assert snapshot_reads == 2
    assert preparation_calls == [True]
    assert deployment_calls == []
    assert persisted[-1]["status"] == "rejected"
    assert persisted[-1]["reason"] == "hub_delivery_revalidation_preflight_changed"
    assert persisted[-1]["audit"][-1]["stage"] == "rejected"


def test_hub_native_full_validation_requires_real_successful_steps(controller, monkeypatch):
    module = controller
    run = {
        "id": 22495, "run_number": 1180, "run_attempt": 1,
        "created_at": "2026-10-08T00:00:00.000Z", "status": "queued",
        "path": "managed-validation.yml@refs/heads/main", "event": "push",
        "head_branch": "main", "head_sha": REVISION,
        "repository": {"full_name": REPOSITORY},
        "head_repository": {"full_name": REPOSITORY}, "pull_requests": [],
    }
    full = {
        "id": 39309, "name": "full", "run_id": 22495, "run_attempt": 1,
        "head_sha": REVISION, "head_branch": "main", "status": "completed",
        "conclusion": "success",
        "steps": [
            {"number": 0, "name": module.HUB_DELIVERY_CHECKOUT_STEP, "status": "completed", "conclusion": "success"},
            {"number": 1, "name": module.HUB_DELIVERY_DOTNET_STEP, "status": "completed", "conclusion": "skipped"},
            {"number": 2, "name": module.HUB_DELIVERY_VALIDATION_STEP, "status": "completed", "conclusion": "success"},
        ],
    }
    release = {
        "id": 39310, "name": "release-images", "run_id": 22495,
        "run_attempt": 1, "head_sha": REVISION, "head_branch": "main",
        "status": "queued",
    }
    jobs = [full, release]
    status = {
        "id": 4,
        "context": module.HUB_DELIVERY_FULL_CONTEXT,
        "status": "success",
        "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22495/jobs/39309",
        "url": f"https://gitea.example/api/v1/repos/Jens/ITWorx_MCP_Hub/statuses/{REVISION}",
        "updated_at": "2026-10-08T00:05:00.000Z",
    }
    statuses = [status]
    fake_candidate = SimpleNamespace(
        WORKFLOW_LIST_PATH="/repos/jens/itworx_mcp_hub/actions/workflows",
        REPOSITORY_API="/repos/jens/itworx_mcp_hub",
        RUN_PATH="managed-validation.yml@refs/heads/main",
        _api_json=lambda _api, path, **_kwargs: (
            {"workflows": [{"id": "managed-validation.yml", "path": ".gitea/workflows/managed-validation.yml", "state": "active"}], "total_count": 1}
            if path.endswith("/actions/workflows") else
            {"protected": True, "commit": {"id": REVISION}}
        ),
        _fixed_workflow=lambda *_a: None,
        _run_matches=lambda row, revision: (
            row.get("head_sha") == revision
            and row.get("head_branch") == "main"
            and row.get("event") == "push"
            and row.get("path") == "managed-validation.yml@refs/heads/main"
            and row.get("repository", {}).get("full_name") == REPOSITORY
            and row.get("head_repository", {}).get("full_name") == REPOSITORY
            and row.get("pull_requests") == []
        ),
        _paged_rows=lambda _api, path, **_kwargs: [run] if path.endswith("/actions/runs") else jobs,
    )
    monkeypatch.setitem(sys.modules, "hub_candidate_import", fake_candidate)
    monkeypatch.setattr(module, "_hub_candidate_api_json", lambda path, **_k: statuses if "/statuses/" in path else {})
    monkeypatch.setattr(module, "_gitea_validation_settings", lambda: ("https://gitea.example/api/v1", b"token"))

    proof = module._hub_delivery_validation_evidence(REVISION)
    assert proof["runId"] == 22495
    assert proof["runNumber"] == 1180
    assert proof["jobId"] == 39309
    assert proof["jobCount"] == 2
    assert proof["releaseJobId"] == 39310
    assert proof["stepCount"] == 3
    assert proof["statusId"] == 4

    run["status"] = "requested"
    release["status"] = "requested"
    assert module._hub_delivery_validation_evidence(REVISION)["runNumber"] == 1180
    run["status"] = "in_progress"
    release["status"] = "queued"

    full["steps"][2]["conclusion"] = "failure"
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_validation_evidence(REVISION)

    full["steps"][2]["conclusion"] = "success"
    full["run_attempt"] = True
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_validation_evidence(REVISION)
    full["run_attempt"] = 1
    del full["steps"][0]
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_validation_evidence(REVISION)

    full["steps"].insert(0, {
        "number": 0, "name": "Run actions/checkout@v4", "status": "completed", "conclusion": "success",
    })
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_validation_evidence(REVISION)

    full["steps"][0]["name"] = module.HUB_DELIVERY_CHECKOUT_STEP
    statuses.append({
        **status,
        "id": 5,
        "status": "failure",
        "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22496/jobs/39309",
        "updated_at": "2026-10-08T00:06:00.000Z",
    })
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_validation_evidence(REVISION)

    statuses[:] = [status]
    jobs.pop()
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_validation_evidence(REVISION)
    jobs.append(release)
    release["id"] = full["id"]
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_validation_evidence(REVISION)


@pytest.mark.parametrize(
    ("success_time", "failure_time", "success_id", "failure_id", "refused"),
    [
        ("2026-10-08T00:05:00Z", "2026-10-08T00:05:00.500Z", 4, 5, True),
        ("2026-10-08T00:05:00.500Z", "2026-10-08T00:05:00Z", 5, 4, False),
        ("2026-10-08T00:05:00Z", "2026-10-08T00:05:00.000Z", 4, 5, True),
        ("2026-10-08T00:05:00.000Z", "2026-10-08T00:05:00Z", 5, 4, False),
        ("2026-10-08T00:05:00Z", "2026-02-30T00:05:00Z", 4, 5, True),
    ],
)
def test_hub_required_status_uses_chronological_utc_order(
    controller, monkeypatch, success_time, failure_time, success_id, failure_id, refused,
):
    module = controller
    source = f"https://gitea.example/api/v1/repos/Jens/ITWorx_MCP_Hub/statuses/{REVISION}"
    statuses = [
        {"id": success_id, "context": module.HUB_DELIVERY_FULL_CONTEXT,
         "status": "success", "url": source,
         "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22495/jobs/39309",
         "updated_at": success_time},
        {"id": failure_id, "context": module.HUB_DELIVERY_FULL_CONTEXT,
         "status": "failure", "url": source,
         "target_url": "/Jens/ITWorx_MCP_Hub/actions/runs/22496/jobs/39311",
         "updated_at": failure_time},
    ]
    monkeypatch.setitem(sys.modules, "hub_candidate_import", SimpleNamespace(
        REPOSITORY_API="/repos/jens/itworx_mcp_hub",
    ))
    monkeypatch.setattr(module, "_gitea_validation_settings", lambda: ("https://gitea.example/api/v1", b"token"))
    monkeypatch.setattr(module, "_hub_candidate_api_json", lambda *_a, **_k: statuses)
    if refused:
        with pytest.raises(module.HubDeliveryRefused):
            module._hub_delivery_full_status_evidence(REVISION, 22495, 39309)
    else:
        proof = module._hub_delivery_full_status_evidence(REVISION, 22495, 39309)
        assert proof["id"] == success_id
        assert proof["updatedAt"] == success_time


def test_hub_status_transport_allows_only_fixed_exact_sha_pages(controller, monkeypatch):
    module = controller
    requests = []
    monkeypatch.setattr(
        module, "_hub_candidate_request",
        lambda target, **_kwargs: requests.append(target) or object(),
    )
    monkeypatch.setattr(module, "_hub_candidate_transport", lambda *_a, **_kwargs: b"[]")

    path = f"/repos/jens/itworx_mcp_hub/statuses/{REVISION}"
    assert module._hub_candidate_api_json(path, query={"page": "1", "limit": "100"}) == []
    assert requests == [path + "?page=1&limit=100"]
    with pytest.raises(module._HubCandidateTransportError):
        module._hub_candidate_api_json(path, query={"page": "1", "limit": "100", "state": "success"})
    with pytest.raises(module._HubCandidateTransportError):
        module._hub_candidate_api_json(f"/repos/jens/other/statuses/{REVISION}", query={"page": "1", "limit": "100"})


def test_original_terminal_record_is_preserved_and_sibling_is_idempotent(controller, monkeypatch, tmp_path):
    module = controller
    repository_id = 314
    semantic = module._webhook_semantic_digest(repository_id, "refs/heads/main", REVISION)
    original = {
        "schema": module._WEBHOOK_SCHEMA,
        "semantic_sha256": semantic,
        "body_sha256": "d" * 64,
        "delivery": "01234567-89ab-4cde-8fab-0123456789ab",
        "gitea_repository_id": repository_id,
        "ref": "refs/heads/main",
        "repository": REPOSITORY,
        "revision": REVISION,
        "source": "gitea_webhook",
        "status": "validation_timeout",
        "reason": "validation_timeout",
        "validation": {"validation_gate_version": 1},
        "audit": [{"at": 1, "stage": "received"}, {"at": 2, "stage": "validation_timeout"}],
        "received_at": 1,
        "updated_at": 2,
    }
    raw = module._hub_delivery_canonical(original) + b"\n"
    original_path = tmp_path / f"{semantic}.json"
    sibling_path = tmp_path / f"{semantic}.revalidation.json"
    original_path.write_bytes(raw)
    original_path.chmod(0o600)
    tmp_path.chmod(0o700)
    monkeypatch.setattr(module, "WEBHOOK_INBOX_ROOT", tmp_path)
    project = {"slug": "itworx-mcp-hub", "branch": "main", "gitea_repository_id": repository_id}
    monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project})
    policy = {"validation_gate_version": 1, "contexts": [module.HUB_DELIVERY_FULL_CONTEXT],
              "timeout_seconds": 60, "poll_interval_seconds": 2, "max_poll_interval_seconds": 30}
    monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (repository_id, "main", "hub-secret"))
    monkeypatch.setattr(module, "_validation_policy", lambda *_a: policy)
    monkeypatch.setattr(module, "_is_current_branch_head", lambda *_a: True)
    monkeypatch.setattr(module, "_validation_snapshot_matches", lambda observed, expected: observed == {"validation_gate_version": 1})
    monkeypatch.setattr(module, "_webhook_wakeup", SimpleNamespace(set=lambda: None))
    binding_snapshot = {
        "revision": REVISION,
        "_semanticDigest": semantic,
        "preflightSha256": "a" * 64,
        "_originalFileSha256": hashlib.sha256(raw).hexdigest(),
        "_stateSha256": "1" * 64,
        "_migrationIntentSha256": "2" * 64,
        "_baselineSha256": "5" * 64,
        "_profileSha256": "3" * 64,
        "_controllerArtifactSha256": "f" * 64,
        "_validationEvidenceSha256": "4" * 64,
    }
    binding = module._hub_delivery_signed_binding(
        binding_snapshot, hashlib.sha256(KEY.encode()).hexdigest(),
    )

    status, queued_semantic, created = module._enqueue_hub_terminal_revalidation(
        REVISION, project, binding, hashlib.sha256(raw).hexdigest(),
    )
    assert (status, queued_semantic, created) == ("awaiting_validation", semantic, True)
    assert original_path.read_bytes() == raw
    sibling = json.loads(sibling_path.read_text())
    assert sibling["hub_delivery_revalidation"] == binding
    assert sibling["revalidation_of"] == semantic

    sibling["status"] = "deployed"
    sibling_path.write_bytes(module._hub_delivery_canonical(sibling) + b"\n")
    status, queued_semantic, created = module._enqueue_hub_terminal_revalidation(
        REVISION, project, binding, hashlib.sha256(raw).hexdigest(),
    )
    assert (status, queued_semantic, created) == ("deployed", semantic, False)
    assert original_path.read_bytes() == raw
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_existing_revalidation(
            REVISION, hashlib.sha256(b"different-idempotency-key").hexdigest(), "a" * 64,
        )


@pytest.mark.skipif(os.name == "nt", reason="POSIX inbox permission contract")
@pytest.mark.parametrize(("root_mode", "file_mode"), [(0o700, 0o644), (0o755, 0o600)])
def test_hub_original_reader_refuses_public_fixture_permissions(
    controller, monkeypatch, tmp_path, root_mode, file_mode,
):
    module = controller
    semantic = "c" * 64
    raw = b'{"fixture":true}\n'
    original = tmp_path / f"{semantic}.json"
    original.write_bytes(raw)
    original.chmod(0o600)
    tmp_path.chmod(0o700)
    monkeypatch.setattr(module, "WEBHOOK_INBOX_ROOT", tmp_path)
    assert module._hub_delivery_read_original(semantic) == (
        {"fixture": True}, hashlib.sha256(raw).hexdigest(),
    )
    original.chmod(file_mode)
    tmp_path.chmod(root_mode)
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_read_original(semantic)
    assert original.read_bytes() == raw


def test_controller_snapshot_binds_full_profile_state_and_intent(controller, monkeypatch):
    module = controller
    revision = REVISION
    project = {
        "slug": "itworx-mcp-hub",
        "branch": "main",
        "gitea_repository_id": 314,
        "auto_apply": True,
        "environment": {"APPOPS_SOURCE_REVISION": "{revision}"},
    }
    policy = {"validation_gate_version": 1, "contexts": [module.HUB_DELIVERY_FULL_CONTEXT],
              "timeout_seconds": 60, "poll_interval_seconds": 2, "max_poll_interval_seconds": 30}
    semantic = module._webhook_semantic_digest(314, "refs/heads/main", revision)
    original = {"schema": module._WEBHOOK_SCHEMA, "semantic_sha256": semantic,
                "body_sha256": "d" * 64, "delivery": "01234567-89ab-4cde-8fab-0123456789ab", "gitea_repository_id": 314,
                "ref": "refs/heads/main", "repository": REPOSITORY, "revision": revision,
                "source": "gitea_webhook", "status": "validation_timeout", "reason": "validation_timeout",
                "validation": {
                    "validation_gate_version": 1,
                    "required_contexts": [module.HUB_DELIVERY_FULL_CONTEXT],
                    "timeout_seconds": 60,
                    "poll_interval_seconds": 2,
                    "max_poll_interval_seconds": 30,
                },
                "audit": [{"stage": "validation_timeout"}]}
    original_bytes = module._hub_delivery_canonical(original) + b"\n"
    original_sha = hashlib.sha256(original_bytes).hexdigest()
    monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project})
    monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (314, "main", "secret"))
    monkeypatch.setattr(module, "_validation_policy", lambda *_a: policy)
    monkeypatch.setattr(module, "_is_current_branch_head", lambda *_a: True)
    monkeypatch.setattr(module, "_hub_delivery_read_original", lambda *_a: (original, original_sha))
    monkeypatch.setattr(module, "_hub_delivery_validation_evidence", lambda *_a: {
        "revision": revision,
        "branch": "main",
        "event": "push",
        "workflowPath": "managed-validation.yml@refs/heads/main",
        "validationContext": module.HUB_DELIVERY_FULL_CONTEXT,
        "runId": 22495,
        "runNumber": 1180,
        "runAttempt": 1,
        "jobId": 39309,
        "releaseJobId": 39310,
        "jobCount": 2,
        "stepCount": 12,
        "stepSummarySha256": "6" * 64,
        "statusId": 4,
        "statusUpdatedAt": "2026-10-08T00:05:00.000Z",
    })
    monkeypatch.setattr(module._mcp_hub_release_store, "read_hub_state_bytes", lambda **_k: b"{\"state\":1}")
    monkeypatch.setitem(sys.modules, "deploy", SimpleNamespace(_mcp_hub_read_migration_intent=lambda: None))
    monkeypatch.setattr(module, "_self_upgrade_secret", lambda: b"hub-test-key-" + b"x" * 40)

    first = module._hub_delivery_snapshot(revision)
    project["environment"]["APPOPS_SOURCE_REVISION"] = "changed"
    second = module._hub_delivery_snapshot(revision)
    assert first["preflightSha256"] != second["preflightSha256"]
    assert first["_originalFileSha256"] == original_sha
    assert first["_stateSha256"] == hashlib.sha256(b"{\"state\":1}").hexdigest()
    project["environment"]["APPOPS_SOURCE_REVISION"] = "{revision}"
    original["audit"].append({"stage": "deployment_started"})
    with pytest.raises(module.HubDeliveryRefused):
        module._hub_delivery_snapshot(revision)


def test_hub_delivery_client_and_mcp_tool_keep_closed_contract(monkeypatch):
    from backend.app import mcp_server
    from backend.app.services.deployment_broker import DeploymentBrokerClient, settings

    monkeypatch.setattr(settings, "mcp_token", "t" * 48)
    client = DeploymentBrokerClient(base_url="https://controller.example", verify_tls=True)
    requests = []
    monkeypatch.setattr(
        client,
        "_request",
        lambda *args, **kwargs: requests.append((args, kwargs)) or {"status": "ready", "noMutation": True},
    )
    preflight_payload = {"revision": REVISION}
    assert client.hub_delivery_operation("preflight", preflight_payload)["status"] == "ready"
    assert requests[-1][0][:2] == ("POST", PREFIX + "preflight")
    assert requests[-1][1]["payload"] == preflight_payload
    revalidate_payload = {
        "revision": REVISION,
        "expectedPreflightSha256": "a" * 64,
        "idempotencyKey": KEY,
    }
    assert client.hub_delivery_operation("revalidate", revalidate_payload)["status"] == "ready"
    assert requests[-1][0][:2] == ("POST", PREFIX + "revalidate")
    assert requests[-1][1]["payload"] == revalidate_payload
    monkeypatch.setattr(
        client, "_request", lambda *_a, **_k: {"status": "ready", "noMutation": False},
    )
    assert client.hub_delivery_operation("preflight", preflight_payload) == {
        "status": "unknown", "reason": "hub_delivery_preflight_mutation_unexpected", "noMutation": False,
    }
    with pytest.raises(ValueError):
        client.hub_delivery_operation("preflight", {**preflight_payload, "repository": REPOSITORY})
    with pytest.raises(ValueError):
        client.hub_delivery_operation("revalidate", {**revalidate_payload, "idempotencyKey": "short"})

    # A transport failure after an operation request is outcome-unknown. The
    # fixed client makes one request and never retries it implicitly.
    calls = []
    monkeypatch.setattr(
        client,
        "_request",
        lambda *args, **kwargs: calls.append((args, kwargs)) or {"status": "unreachable"},
    )
    outcome = client.hub_delivery_operation("revalidate", revalidate_payload)
    assert outcome == {"status": "unknown", "reason": "hub_delivery_outcome_unknown", "noMutation": False}
    assert len(calls) == 1
    monkeypatch.setattr(client, "_request", lambda *_a, **_k: {"status": "server_error"})
    assert client.hub_delivery_operation("revalidate", revalidate_payload) == {
        "status": "unknown", "reason": "hub_delivery_outcome_unknown", "noMutation": False,
    }

    tool_calls = []
    monkeypatch.setattr(
        mcp_server,
        "DeploymentBrokerClient",
        lambda: SimpleNamespace(
            hub_delivery_operation=lambda stage, payload: tool_calls.append((stage, payload))
            or {"status": "ready", "noMutation": True},
        ),
    )
    assert mcp_server.appops_hub_delivery_preflight(REVISION)["status"] == "ready"
    with pytest.raises(ValueError):
        mcp_server.appops_hub_delivery_revalidate(REVISION, "a" * 64, "short")
    assert mcp_server.appops_hub_delivery_revalidate(REVISION, "a" * 64, KEY)["status"] == "ready"
    assert tool_calls == [("preflight", preflight_payload), ("revalidate", revalidate_payload)]


def test_hub_deploy_gate_requires_the_exact_signed_deploying_sibling(controller, monkeypatch):
    module = controller
    project = {
        "slug": module.HUB_DELIVERY_PROJECT_SLUG,
        "branch": "main",
        "gitea_repository_id": 7,
    }
    semantic = module._webhook_semantic_digest(7, "refs/heads/main", REVISION)
    snapshot = {**_snapshot(module), "_semanticDigest": semantic}
    binding = module._hub_delivery_signed_binding(
        snapshot, hashlib.sha256(KEY.encode()).hexdigest(),
    )
    attempt = {
        "schema": module._WEBHOOK_REVALIDATION_SCHEMA,
        "semantic_sha256": semantic,
        "revalidation_of": semantic,
        "repository": REPOSITORY,
        "revision": REVISION,
        "ref": "refs/heads/main",
        "gitea_repository_id": 7,
        "status": "deploying",
        "audit": [{"stage": "deployment_started"}],
        "hub_delivery_revalidation": binding,
    }
    monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: project})
    monkeypatch.setattr(module, "_webhook_policy", lambda *_a: (7, "main"))
    monkeypatch.setattr(module, "_read_inbox_record", lambda marker: attempt)
    monkeypatch.setattr(module, "_hub_delivery_snapshot", lambda _revision: snapshot)

    assert module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "a" * 64)
    assert not module._hub_delivery_deploy_gate("jens/other", REVISION, "a" * 64)
    assert not module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "short")

    attempt["status"] = "validation_passed"
    assert not module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "a" * 64)
    attempt["status"] = "deploying"
    snapshot["preflightSha256"] = "e" * 64
    assert not module._hub_delivery_deploy_gate(REPOSITORY, REVISION, "a" * 64)


def test_deploy_hub_preflight_binding_is_fixed_and_fails_closed(controller, monkeypatch):
    deploy = _load_module("hub_delivery_locked_deploy_gate_test", BROKER_ROOT / "deploy.py")
    calls = []
    gate = SimpleNamespace(
        _hub_delivery_deploy_gate=lambda *args: calls.append(args) or True,
    )
    monkeypatch.setitem(sys.modules, "status_server", gate)

    deploy._verify_hub_delivery_preflight(REPOSITORY, REVISION, "a" * 64)
    assert calls == [(REPOSITORY, REVISION, "a" * 64)]
    for repository, revision, fingerprint in (
        ("jens/other", REVISION, "a" * 64),
        (REPOSITORY, "invalid", "a" * 64),
        (REPOSITORY, REVISION, "invalid"),
    ):
        with pytest.raises(deploy.DeployError, match="preflight binding is invalid"):
            deploy._verify_hub_delivery_preflight(repository, revision, fingerprint)
    assert len(calls) == 1

    gate._hub_delivery_deploy_gate = lambda *_a: False
    with pytest.raises(deploy.DeployError, match="preflight changed before deployment"):
        deploy._verify_hub_delivery_preflight(REPOSITORY, REVISION, "a" * 64)

    monkeypatch.setenv(deploy.HUB_DELIVERY_PREFLIGHT_ENV, "a" * 64)
    monkeypatch.setattr(sys, "argv", ["deploy.py", "describe", "jens/other", REVISION])
    with pytest.raises(deploy.DeployError, match="preflight binding is invalid"):
        deploy.main()


def test_run_registered_deploy_passes_only_the_fixed_hub_binding(controller, monkeypatch):
    module = controller
    captured = []
    monkeypatch.setattr(module, "_load_projects", lambda: {REPOSITORY: {"auto_apply": True}})
    monkeypatch.setenv("APPOPS_HUB_DELIVERY_PREFLIGHT_SHA256", "c" * 64)
    monkeypatch.setenv("APPOPS_PROJECTBRAIN_DELIVERY_PREFLIGHT_SHA256", "d" * 64)
    monkeypatch.setattr(
        module.subprocess, "run",
        lambda argv, **kwargs: captured.append((argv, kwargs)) or SimpleNamespace(returncode=0),
    )

    module._run_registered_deploy(
        REPOSITORY, REVISION, hub_delivery_preflight_sha256="a" * 64,
    )

    argv, kwargs = captured[0]
    assert argv[-3:] == ["deploy", REPOSITORY, REVISION]
    assert kwargs["env"]["APPOPS_HUB_DELIVERY_PREFLIGHT_SHA256"] == "a" * 64
    assert "APPOPS_PROJECTBRAIN_DELIVERY_PREFLIGHT_SHA256" not in kwargs["env"]
    with pytest.raises(ValueError, match="Hub delivery binding invalid"):
        module._run_registered_deploy("jens/other", REVISION, hub_delivery_preflight_sha256="a" * 64)
    with pytest.raises(ValueError, match="Hub delivery binding invalid"):
        module._run_registered_deploy(REPOSITORY, REVISION, hub_delivery_preflight_sha256="short")


def test_deploy_hub_preflight_is_rechecked_at_both_locked_boundaries(monkeypatch, tmp_path):
    deploy = _load_module("hub_delivery_two_boundary_deploy_test", BROKER_ROOT / "deploy.py")
    repository_root = tmp_path / "repository"
    repository_root.mkdir()
    _git(repository_root, "init", "-b", "main")
    _git(repository_root, "config", "user.email", "hub-delivery-test@example.invalid")
    _git(repository_root, "config", "user.name", "Hub delivery test")
    source_file = repository_root / "README.md"
    source_file.write_text("prior live source\n", encoding="utf-8")
    _git(repository_root, "add", ".")
    _git(repository_root, "commit", "-m", "prior")
    prior_revision = _git(repository_root, "rev-parse", "HEAD")
    source_file.write_text("candidate source\n", encoding="utf-8")
    _git(repository_root, "add", ".")
    _git(repository_root, "commit", "-m", "candidate")
    target_revision = _git(repository_root, "rev-parse", "HEAD")
    repository_git_dir = repository_root / ".git"
    source_dir = tmp_path / "source"
    deploy.sync_source(repository_git_dir, prior_revision, source_dir, "itworx-mcp-hub")

    project = {
        "slug": "itworx-mcp-hub",
        "branch": "main",
        "source_dir": str(source_dir),
        "containers": ["itworx-mcp-hub"],
        "compose_project": "itworx-mcp-hub",
        "compose_files": ["docker-compose.yml"],
    }
    previous_state = {
        "schema_version": 3,
        "repository": REPOSITORY,
        "live_revision": prior_revision,
        "live_status": "healthy",
        "live_legacy_provenance": False,
        "live_artifacts": {"containers": {"hub": {"image_id": "sha256:" + "a" * 64}}},
        "live_receipt": {"revision": prior_revision, "testOnly": True},
    }
    active_locks = []
    project_lock = deploy.STATE_ROOT / "locks" / "itworx-mcp-hub.lock"

    @contextmanager
    def lock(path, _description):
        active_locks.append(path)
        try:
            yield
        finally:
            active_locks.pop()

    monkeypatch.setattr(deploy, "exclusive_lock", lock)
    monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: lock(
        deploy.GLOBAL_DEPLOY_LOCK, "global",
    ))
    monkeypatch.setattr(deploy, "load_projects", lambda: {REPOSITORY: dict(project)})
    monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, config: config)
    monkeypatch.setattr(deploy, "repository_path", lambda *_a: repository_git_dir)
    monkeypatch.setattr(deploy, "bind_appops_self_upgrade_recovery", lambda *_a: None)
    monkeypatch.setattr(deploy, "bind_registered_legacy_request", lambda *_a: None)
    monkeypatch.setattr(deploy, "mcp_hub_prepare_current_receipt_model", lambda *_a: None)
    monkeypatch.setattr(deploy, "validate_commit", lambda *_a: target_revision)
    monkeypatch.setattr(deploy, "deployment_metadata", lambda *_a: {
        "revision": target_revision,
        "tree": "f" * 40,
    })
    monkeypatch.setattr(deploy, "validate_project_config", lambda *_a: None)
    monkeypatch.setattr(deploy, "read_state", lambda _project: dict(previous_state))
    monkeypatch.setattr(deploy, "verified_current_receipt", lambda *_a: None)
    monkeypatch.setattr(deploy, "validate_forward_revision", lambda *_a: None)
    monkeypatch.setattr(deploy, "_mcp_hub_assert_no_unresolved_migration_intent", lambda *_a: None)

    source_syncs = []
    actual_sync_source = deploy.sync_source

    def record_sync(repo, revision, target, slug, clean_untracked_paths=None, *, exact=False):
        source_syncs.append((revision, tuple(active_locks)))
        return actual_sync_source(
            repo, revision, target, slug, clean_untracked_paths, exact=exact,
        )

    monkeypatch.setattr(deploy, "sync_source", record_sync)
    monkeypatch.setattr(deploy, "copy_environment", lambda *_a, **_k: None)
    gate_calls = []
    gate_results = iter((True, False))

    def gate(repository, revision, fingerprint):
        gate_calls.append((repository, revision, fingerprint, tuple(active_locks)))
        return next(gate_results)

    monkeypatch.setitem(sys.modules, "status_server", SimpleNamespace(
        _hub_delivery_deploy_gate=gate,
    ))
    state_updates = []
    monkeypatch.setattr(
        deploy, "write_state",
        lambda repository, config, revision, status, message, **kwargs: state_updates.append({
            "repository": repository,
            "project": config,
            "revision": revision,
            "status": status,
            "message": message,
            **kwargs,
        }),
    )
    monkeypatch.setattr(
        deploy, "capture_compose_rollback",
        lambda *_a: pytest.fail("production snapshot must not run after stale Hub proof"),
    )
    monkeypatch.setattr(
        deploy, "deploy_project",
        lambda *_a: pytest.fail("production mutation must not run after stale Hub proof"),
    )
    monkeypatch.delenv(deploy.PROJECTBRAIN_DELIVERY_PREFLIGHT_ENV, raising=False)
    monkeypatch.setenv(deploy.HUB_DELIVERY_PREFLIGHT_ENV, "b" * 64)
    monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", REPOSITORY, target_revision])

    with pytest.raises(deploy.DeployError, match="preflight changed before deployment"):
        deploy.main()

    assert gate_calls == [
        (REPOSITORY, target_revision, "b" * 64, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)),
        (REPOSITORY, target_revision, "b" * 64, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)),
    ]
    assert source_syncs == [
        (target_revision, (project_lock,)),
        (prior_revision, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)),
    ]
    assert source_dir.joinpath("README.md").read_text(encoding="utf-8") == "prior live source\n"
    assert len(state_updates) == 1
    assert state_updates[0]["status"] == "failed"
    assert state_updates[0]["live_revision"] == prior_revision
    assert state_updates[0]["live_status"] == "healthy"