DataOps-platform/tests/test_phase3_wp04_edge_gateway.py

from __future__ import annotations

import hashlib
import json
import sqlite3
from concurrent.futures import ThreadPoolExecutor
from dataclasses import FrozenInstanceError
from datetime import UTC, datetime
from threading import Barrier

import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from app.core.edge_gateway import (
    CONTROL_PLANE_ALLOWED,
    EDGE_ONLY,
    RETENTION_DAYS,
    SCHEMA_VERSION,
    EdgeContractError,
    EdgeEgressPolicy,
    EdgeEventContract,
    EdgePolicyError,
    EdgeQueueConflictError,
    EdgeQueueLeaseError,
    EdgeQueueSchemaError,
    EdgeTaskContract,
    SignedTaskEnvelope,
    SqliteEdgeQueue,
    canonical_sha256,
    stable_event_id,
)
from app.core.edge_gateway import queue as edge_queue_module

POLICY_DIGEST = "a" * 64
APPROVED_TASK = {
    "task_id": "task-20260809-001",
    "gateway_id": "gateway-enterprise-001",
    "environment": "production",
    "network_zone": "enterprise-zone-a",
    "purpose": "governed-data-quality",
    "classification": "raw",
    "task_type": "profile",
    "contract_version": 1,
    "deadline_at": "2099-08-09T12:00:00Z",
    "attempt": 1,
    "idempotency_key": "profile:source-1:20260809",
    "policy_digest": POLICY_DIGEST,
}


def _event(
    *,
    classification: str = "statistics",
    payload: dict[str, object] | None = None,
    occurred_at: str = "2026-08-09T08:00:00Z",
) -> dict[str, object]:
    content = payload or {"metric_count": 3, "null_ratio": 0.125}
    event = {
        "task_id": APPROVED_TASK["task_id"],
        "gateway_id": APPROVED_TASK["gateway_id"],
        "environment": APPROVED_TASK["environment"],
        "network_zone": APPROVED_TASK["network_zone"],
        "purpose": APPROVED_TASK["purpose"],
        "classification": classification,
        "contract_version": 1,
        "occurred_at": occurred_at,
        "attempt": 1,
        "idempotency_key": f"event:{APPROVED_TASK['task_id']}:{classification}",
        "policy_digest": POLICY_DIGEST,
        "payload": content,
    }
    return {"event_id": stable_event_id(event), **event}


def _ack(*, received_at: str = "2026-08-09T09:00:00+01:00") -> dict[str, str]:
    return {
        "message_id": "control-message-1",
        "received_at": received_at,
        "status": "accepted",
    }


def _signed_task(
    private_key: Ed25519PrivateKey,
    *,
    task: dict[str, object] | None = None,
    issued_at: str = "2026-08-09T07:59:00Z",
    expires_at: str = "2099-08-09T12:00:00Z",
) -> dict[str, object]:
    payload = {
        "task": task or APPROVED_TASK,
        "authority_key_id": "control-authority-2026-01",
        "signature_algorithm": "Ed25519",
        "contract_digest": canonical_sha256(task or APPROVED_TASK),
        "gateway_id": (task or APPROVED_TASK)["gateway_id"],
        "environment": (task or APPROVED_TASK)["environment"],
        "network_zone": (task or APPROVED_TASK)["network_zone"],
        "policy_digest": (task or APPROVED_TASK)["policy_digest"],
        "purpose": (task or APPROVED_TASK)["purpose"],
        "issued_at": issued_at,
        "expires_at": expires_at,
    }
    unsigned = SignedTaskEnvelope.canonical_unsigned_bytes(payload)
    return {**payload, "signature": private_key.sign(unsigned).hex()}


def test_task_contract_is_immutable_strict_and_allows_only_supported_operations():
    task = EdgeTaskContract.from_mapping(APPROVED_TASK)

    assert task.task_type == "profile"
    assert task.operation == "profile"
    assert task.to_mapping() == APPROVED_TASK
    with pytest.raises(FrozenInstanceError):
        task.task_id = "changed"  # type: ignore[misc]
    with pytest.raises(EdgeContractError, match="unknown properties"):
        EdgeTaskContract.from_mapping({**APPROVED_TASK, "parameters": {}})
    for operation in ("collect", "profile", "quality", "lineage", "controlled_query"):
        assert EdgeTaskContract.from_mapping(
            {**APPROVED_TASK, "task_type": operation}
        ).task_type == operation
    with pytest.raises(EdgeContractError, match="operation"):
        EdgeTaskContract.from_mapping({**APPROVED_TASK, "task_type": "shell"})


def test_direct_task_construction_cannot_bypass_shared_validation():
    fields = {
        **APPROVED_TASK,
        "operation": APPROVED_TASK["task_type"],
    }
    fields.pop("task_type")

    valid = EdgeTaskContract(**fields)
    assert valid.deadline_at == APPROVED_TASK["deadline_at"]
    for changed in (
        {"operation": "shell"},
        {"classification": "unknown"},
        {"contract_version": 2},
        {"attempt": 6},
        {"policy_digest": "bad"},
        {"deadline_at": "not-a-time"},
    ):
        with pytest.raises(EdgeContractError):
            EdgeTaskContract(**{**fields, **changed})


def test_direct_event_construction_and_queue_instance_input_are_revalidated(tmp_path):
    mapping = _event()
    fields = dict(mapping)
    valid = EdgeEventContract(**fields)
    assert valid.payload == mapping["payload"]
    with pytest.raises(EdgeContractError):
        EdgeEventContract(**{**fields, "event_id": "evt_invalid"})
    with pytest.raises(EdgeContractError):
        EdgeEventContract(**{**fields, "attempt": 6})

    task = EdgeTaskContract.from_mapping(APPROVED_TASK)
    object.__setattr__(task, "operation", "shell")
    with pytest.raises(EdgeContractError):
        SqliteEdgeQueue(tmp_path / "edge.db").enqueue(task)


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("classification", "unknown"),
        ("contract_version", 0),
        ("attempt", True),
        ("attempt", 0),
        ("deadline_at", "2099-08-09 12:00:00"),
        ("deadline_at", "2099-08-09T12:00:00"),
        ("policy_digest", "not-a-sha256"),
    ],
)
def test_task_contract_fails_closed_for_malformed_enums_numbers_and_timestamps(
    field, value
):
    with pytest.raises(EdgeContractError):
        EdgeTaskContract.from_mapping({**APPROVED_TASK, field: value})


def test_equivalent_rfc3339_deadlines_canonicalize_to_one_utc_digest():
    offset = EdgeTaskContract.from_mapping(
        {**APPROVED_TASK, "deadline_at": "2026-08-09T09:00:05+01:00"}
    )
    utc = EdgeTaskContract.from_mapping(
        {**APPROVED_TASK, "deadline_at": "2026-08-09T08:00:05Z"}
    )
    fractional = EdgeTaskContract.from_mapping(
        {**APPROVED_TASK, "deadline_at": "2026-08-09T08:00:05.120000+00:00"}
    )

    assert offset.deadline_at == utc.deadline_at == "2026-08-09T08:00:05Z"
    assert offset.digest == utc.digest
    assert fractional.deadline_at == "2026-08-09T08:00:05.12Z"


def test_equivalent_event_timestamps_canonicalize_before_id_and_digest_validation():
    utc_mapping = _event(occurred_at="2026-08-09T08:00:05Z")
    offset_mapping = _event(occurred_at="2026-08-09T09:00:05+01:00")
    offset_mapping["event_id"] = utc_mapping["event_id"]

    utc = EdgeEventContract.from_mapping(utc_mapping)
    offset = EdgeEventContract.from_mapping(offset_mapping)

    assert offset.occurred_at == utc.occurred_at == "2026-08-09T08:00:05Z"
    assert offset.event_id == utc.event_id
    assert offset.digest == utc.digest


def test_canonical_sha256_and_event_identifier_are_stable_and_order_independent():
    left = {"task_id": "task-1", "payload": {"b": 2, "a": [1, "中"]}}
    right = {"payload": {"a": [1, "中"], "b": 2}, "task_id": "task-1"}

    assert canonical_sha256(left) == canonical_sha256(right)
    assert len(canonical_sha256(left)) == 64
    assert stable_event_id(left) == stable_event_id(right)
    assert stable_event_id(left).startswith("evt_")


def test_canonical_protocol_has_explicit_numeric_semantics_and_type_boundaries():
    assert canonical_sha256(1) == canonical_sha256(1.0)
    assert canonical_sha256(0) == canonical_sha256(-0.0)
    assert canonical_sha256("1") != canonical_sha256(1)
    assert canonical_sha256(True) != canonical_sha256(1)
    assert canonical_sha256(None) != canonical_sha256("null")
    assert canonical_sha256({"é": 1, "a": 2}) == canonical_sha256(
        {"a": 2.0, "é": 1.0}
    )


def test_stable_event_identity_binds_every_safe_envelope_dimension():
    event = _event()
    identity = {key: value for key, value in event.items() if key != "event_id"}
    baseline = stable_event_id(identity)

    for field, value in {
        "task_id": "task-other",
        "gateway_id": "gateway-other",
        "environment": "staging",
        "network_zone": "zone-other",
        "purpose": "other-purpose",
        "classification": "evidence",
        "contract_version": 2,
        "attempt": 2,
        "idempotency_key": "other-key",
        "policy_digest": "b" * 64,
        "occurred_at": "2026-08-09T08:00:01Z",
        "payload": {"metric_count": 4},
    }.items():
        assert stable_event_id({**identity, field: value}) != baseline


def test_stable_event_id_normalizes_equivalent_rfc3339_identity_timestamps():
    base = {
        "task_id": "task-1",
        "gateway_id": "gateway-1",
        "classification": "statistics",
        "payload": {"count": 1},
    }

    assert stable_event_id(
        {**base, "occurred_at": "2026-08-09T09:00:05+01:00"}
    ) == stable_event_id({**base, "occurred_at": "2026-08-09T08:00:05Z"})


def test_event_contract_requires_stable_identifier_and_approved_classification():
    event = EdgeEventContract.from_mapping(_event())

    assert event.event_id == _event()["event_id"]
    assert event.classification == "statistics"
    with pytest.raises(EdgeContractError, match="stable event identifier"):
        EdgeEventContract.from_mapping({**_event(), "event_id": "evt_changed"})
    with pytest.raises(EdgeContractError, match="classification"):
        EdgeEventContract.from_mapping(_event(classification="raw"))
    with pytest.raises(EdgeContractError, match="timestamp"):
        EdgeEventContract.from_mapping(_event(occurred_at="yesterday"))
    with pytest.raises(EdgeContractError, match="unknown properties"):
        EdgeEventContract.from_mapping({**_event(), "raw_result": []})


def test_contract_and_policy_preflight_reject_cycles_and_extreme_depth():
    cyclic: dict[str, object] = {}
    cyclic["self"] = cyclic
    deep: dict[str, object] = {"leaf": 1}
    for _ in range(1_100):
        deep = {"next": deep}

    with pytest.raises(EdgeContractError, match="cycle"):
        canonical_sha256(cyclic)
    with pytest.raises(EdgeContractError, match="depth"):
        canonical_sha256(deep)
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})
    with pytest.raises(EdgePolicyError):
        policy.approve_event({"classification": "statistics", "payload": cyclic})
    with pytest.raises(EdgePolicyError):
        policy.approve_event({"classification": "statistics", "payload": deep})


def test_preflight_rejects_oversized_strings_and_containers_before_encoding():
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})
    with pytest.raises(EdgePolicyError):
        policy.approve_event(
            {"classification": "statistics", "payload": {"value": "x" * 200_000}}
        )
    with pytest.raises(EdgePolicyError):
        policy.approve_event(
            {
                "classification": "statistics",
                "payload": {"values": list(range(20_000))},
            }
        )


def test_raw_and_recent_detail_never_cross_the_control_plane_boundary():
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})

    assert {"raw", "recent_detail", "restricted"} == EDGE_ONLY
    assert {
        "desensitized_metadata",
        "statistics",
        "lineage",
        "evidence",
        "health_summary",
        "diagnostic_summary",
    } == CONTROL_PLANE_ALLOWED
    for classification in EDGE_ONLY:
        with pytest.raises(EdgePolicyError, match="edge-only"):
            policy.approve_event(
                {"classification": classification, "payload": {"count": 1}}
            )


@pytest.mark.parametrize(
    "payload",
    [
        {"api_key": "abc"},
        {"access_token": "abc"},
        {"nested": {"password": "abc"}},
        {"rows": [{"id": 1}]},
        {"customer_rows": [{"id": 1}]},
        {"raw_records": [{"id": 1}]},
        {"sql": "SELECT * FROM customer"},
        {"query_text": "SELECT * FROM customer"},
        {"diagnostic": "DELETE FROM customer"},
        {"key": "-----BEGIN PRIVATE KEY-----\nabc"},
        {"authorization": "Bearer abcdefghijklmnopqrstuvwxyz"},
    ],
)
def test_policy_recursively_rejects_secret_raw_detail_sql_and_private_keys(payload):
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})

    with pytest.raises(EdgePolicyError, match="sensitive"):
        policy.approve_event({"classification": "statistics", "payload": payload})


@pytest.mark.parametrize(
    "payload",
    [
        {"diagnostic": "connection failed: client_secret = hunter2; retry later"},
        {
            "diagnostic": (
                "planner output follows: SELECT customer_id FROM customers "
                "WHERE active = 1"
            )
        },
        {"result": [{"customer_id": "customer-1", "email": "raw@example.com"}]},
    ],
)
def test_policy_rejects_reviewed_sensitive_values_and_raw_row_containers(payload):
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})

    with pytest.raises(EdgePolicyError, match="sensitive"):
        policy.approve_event({"classification": "diagnostic_summary", "payload": payload})


@pytest.mark.parametrize(
    "payload",
    [
        {"accessToken": "secret-value"},
        {"pass-word": "secret-value"},
        {"rawRows": [{"id": 1}]},
        {"row": {"id": 1}},
        {"item": {"id": 1}},
        {"diagnostic": "prefix SEL/**/ECT id FR/**/OM customer suffix"},
    ],
)
def test_policy_rejects_camelcase_punctuation_single_rows_and_obfuscated_sql(payload):
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})
    with pytest.raises(EdgePolicyError, match="sensitive"):
        policy.approve_event({"classification": "diagnostic_summary", "payload": payload})


@pytest.mark.parametrize(
    "payload",
    [
        {"ｐassword": "secret-value"},
        {"sqlText": "harmless-looking-value"},
        {"diagnostic": "prefix EX/**/EC dbo.rotate_secret suffix"},
        {"diagnostic": "prefix EXECUTE\n dbo.rotate_secret suffix"},
        {"diagnostic": "prefix CALL customer_refresh() suffix"},
    ],
)
def test_policy_nfkc_normalizes_keys_and_rejects_stored_procedure_sql(payload):
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})
    with pytest.raises(EdgePolicyError, match="sensitive"):
        policy.approve_event({"classification": "diagnostic_summary", "payload": payload})


def test_policy_does_not_misclassify_ordinary_execution_text():
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})
    policy.approve_event(
        {
            "classification": "diagnostic_summary",
            "payload": {
                "text": "Execution completed and callback scheduling succeeded."
            },
        }
    )


def test_policy_allows_lineage_graphs_and_aggregate_scalar_containers():
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})

    policy.approve_event(
        {
            "classification": "lineage",
            "payload": {
                "nodes": [{"id": "source-a"}, {"id": "product-b"}],
                "edges": [{"source": "source-a", "target": "product-b"}],
            },
        }
    )
    policy.approve_event(
        {
            "classification": "statistics",
            "payload": {
                "rows": 12,
                "records": 12,
                "result": 12,
                "data": "aggregated",
                "items": [1, 2],
            },
        }
    )


def test_policy_has_independent_byte_count_depth_and_string_limits():
    policy = EdgeEgressPolicy(
        allowed_control_hosts={"control.example.com"},
        byte_limits={"statistics": 96, "evidence": 512},
        count_limits={"statistics": 2, "evidence": 8},
        max_depth=3,
        max_string_bytes=32,
    )

    policy.approve_event(
        {"classification": "evidence", "payload": {"items": [1, 2, 3]}}
    )
    with pytest.raises(EdgePolicyError, match="byte limit"):
        policy.approve_event(
            {"classification": "statistics", "payload": {"value": "x" * 90}}
        )
    with pytest.raises(EdgePolicyError, match="item count"):
        policy.approve_event(
            {"classification": "statistics", "payload": {"items": [1, 2, 3]}}
        )
    with pytest.raises(EdgePolicyError, match="depth"):
        policy.approve_event(
            {"classification": "evidence", "payload": {"a": {"b": {"c": 1}}}}
        )
    with pytest.raises(EdgePolicyError, match="string limit"):
        policy.approve_event(
            {"classification": "evidence", "payload": {"detail": "x" * 33}}
        )


@pytest.mark.parametrize(
    "url",
    [
        "http://control.example.com/events",
        "https://evil.example.com/events",
        "https://control.example.com.evil.test/events",
        "https://user:pass@control.example.com/events",
        "https://control.example.com.:443/events",
    ],
)
def test_destination_requires_https_and_exact_approved_control_host(url):
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})

    with pytest.raises(EdgePolicyError, match="destination"):
        policy.validate_destination(url)


def test_proxy_is_optional_but_if_present_requires_exact_https_allowlist_match():
    policy = EdgeEgressPolicy(
        allowed_control_hosts={"control.example.com"},
        allowed_proxy_hosts={"proxy.enterprise.example"},
        allowed_proxy_origins={"https://proxy.enterprise.example:8443"},
    )

    policy.validate_destination(
        "https://control.example.com/v1/edge/events",
        proxy_url="https://proxy.enterprise.example:8443",
    )
    with pytest.raises(EdgePolicyError, match="proxy"):
        policy.validate_destination(
            "https://control.example.com/v1/edge/events",
            proxy_url="https://proxy.enterprise.example.evil.test",
        )


def test_retention_classes_are_explicit_and_malformed_values_fail_closed():
    policy = EdgeEgressPolicy(allowed_control_hosts={"control.example.com"})

    assert RETENTION_DAYS == {
        "raw": 0,
        "recent_detail": 365,
        "metadata": 1095,
        "evidence": 2190,
    }
    assert policy.retention_days("raw") == 0
    assert policy.retention_days("recent_detail") == 365
    with pytest.raises(EdgePolicyError, match="retention"):
        policy.retention_days("forever")


def test_queue_requires_explicit_on_disk_path_and_enables_wal(tmp_path):
    with pytest.raises(ValueError, match="explicit"):
        SqliteEdgeQueue("")
    with pytest.raises(ValueError, match="memory"):
        SqliteEdgeQueue(":memory:")

    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    with sqlite3.connect(queue.db_path) as connection:
        assert connection.execute("PRAGMA journal_mode").fetchone()[0].lower() == "wal"
        assert connection.execute("PRAGMA busy_timeout").fetchone()[0] > 0
        assert connection.execute("PRAGMA user_version").fetchone()[0] == SCHEMA_VERSION


def test_queue_fails_closed_for_unknown_or_ambiguous_sqlite_schema(tmp_path):
    unknown = tmp_path / "unknown.db"
    with sqlite3.connect(unknown) as connection:
        connection.execute("PRAGMA user_version = 99")
    with pytest.raises(EdgeQueueSchemaError, match="version"):
        SqliteEdgeQueue(unknown)

    ambiguous = tmp_path / "ambiguous.db"
    with sqlite3.connect(ambiguous) as connection:
        connection.execute("CREATE TABLE edge_tasks (task_id TEXT PRIMARY KEY)")
    with pytest.raises(EdgeQueueSchemaError, match="unversioned"):
        SqliteEdgeQueue(ambiguous)

    damaged = tmp_path / "damaged.db"
    with sqlite3.connect(damaged) as connection:
        connection.execute("CREATE TABLE edge_tasks (task_id TEXT PRIMARY KEY)")
        connection.execute("PRAGMA user_version = 1")
    with pytest.raises(EdgeQueueSchemaError, match="schema"):
        SqliteEdgeQueue(damaged)


def test_queue_rejects_versioned_schema_with_wrong_types_constraints_and_indexes(
    tmp_path,
):
    db_path = tmp_path / "forged-v1.db"
    task_columns = (
        "task_id", "digest", "task_json", "status", "attempt_count",
        "available_at", "deadline_at", "deadline_epoch_us", "lease_owner",
        "lease_token", "lease_expires_at", "error_code", "created_at", "updated_at",
    )
    event_columns = (
        "event_id", "digest", "event_json", "status", "attempt_count",
        "available_at", "lease_owner", "lease_token", "lease_expires_at",
        "error_code", "acknowledgement_json", "ack_digest", "lease_token_digest",
        "created_at", "updated_at",
    )
    with sqlite3.connect(db_path) as connection:
        connection.execute(
            "CREATE TABLE edge_tasks ("
            + ",".join(f"{column} TEXT" for column in task_columns)
            + ")"
        )
        connection.execute(
            "CREATE TABLE edge_outbound_events ("
            + ",".join(f"{column} TEXT" for column in event_columns)
            + ")"
        )
        connection.execute(
            "CREATE INDEX edge_tasks_claim_idx ON edge_tasks(task_id)"
        )
        connection.execute(
            "CREATE INDEX edge_events_claim_idx ON edge_outbound_events(event_id)"
        )
        connection.execute("PRAGMA user_version = 1")

    with pytest.raises(EdgeQueueSchemaError, match="schema"):
        SqliteEdgeQueue(db_path)


def test_queue_exact_task_replay_returns_terminal_and_changed_digest_conflicts(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db", lease_seconds=30)
    first = queue.enqueue(APPROVED_TASK)
    claimed = queue.claim("worker-a")
    assert claimed is not None
    queue.complete(first.task_id, lease_token=claimed.lease_token)

    replay = queue.enqueue(APPROVED_TASK)
    assert replay.status == "completed"
    with pytest.raises(EdgeQueueConflictError, match="digest"):
        queue.enqueue({**APPROVED_TASK, "purpose": "different-purpose"})


def test_exact_terminal_replay_wins_after_the_original_task_deadline(tmp_path):
    now = [datetime(2099, 8, 9, 11, 59, tzinfo=UTC)]
    queue = SqliteEdgeQueue(tmp_path / "edge.db", clock=lambda: now[0])
    queue.enqueue(APPROVED_TASK)
    claimed = queue.claim("worker")
    queue.complete(claimed.task_id, lease_token=claimed.lease_token)
    now[0] = datetime(2099, 8, 9, 12, 1, tzinfo=UTC)

    replay = queue.enqueue(APPROVED_TASK)

    assert replay.status == "completed"
    with pytest.raises(EdgeQueueConflictError, match="digest"):
        queue.enqueue({**APPROVED_TASK, "purpose": "changed-after-deadline"})


def test_two_connections_atomically_claim_one_task_only(tmp_path):
    db_path = tmp_path / "edge.db"
    first = SqliteEdgeQueue(db_path)
    second = SqliteEdgeQueue(db_path)
    first.enqueue(APPROVED_TASK)

    claims = [first.claim("worker-a"), second.claim("worker-b")]

    assert sum(claim is not None for claim in claims) == 1
    claimed = next(claim for claim in claims if claim is not None)
    assert claimed.lease_owner in {"worker-a", "worker-b"}
    assert claimed.lease_token
    assert claimed.lease_expires_at is not None


def test_two_threads_racing_independent_task_connections_have_one_winner(tmp_path):
    db_path = tmp_path / "edge.db"
    first = SqliteEdgeQueue(db_path)
    second = SqliteEdgeQueue(db_path)
    first.enqueue(APPROVED_TASK)
    barrier = Barrier(2)

    def claim(queue, owner):
        barrier.wait(timeout=5)
        return queue.claim(owner)

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(
            pool.map(claim, (first, second), ("worker-a", "worker-b"))
        )
    assert sum(result is not None for result in results) == 1


def test_claim_atomically_fails_a_task_whose_deadline_passed_while_queued(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    task = {**APPROVED_TASK, "deadline_at": "2026-08-09T08:00:05Z"}
    queue = SqliteEdgeQueue(tmp_path / "edge.db", clock=lambda: now[0])
    queue.enqueue(task)
    now[0] = datetime(2026, 8, 9, 8, 0, 6, tzinfo=UTC)

    assert queue.claim("worker") is None
    expired = queue.get_task(task["task_id"])
    assert expired.status == "failed"
    assert expired.error_code == "task_deadline_expired"


def test_claim_fails_an_expired_offset_deadline_instead_of_leasing_it(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    task = {
        **APPROVED_TASK,
        "deadline_at": "2026-08-09T09:00:05+01:00",
        "task_id": "task-offset-deadline",
    }
    queue = SqliteEdgeQueue(tmp_path / "edge.db", clock=lambda: now[0])
    queued = queue.enqueue(task)
    assert queued.task["deadline_at"] == "2026-08-09T08:00:05Z"
    now[0] = datetime(2026, 8, 9, 8, 0, 6, tzinfo=UTC)

    assert queue.claim("worker") is None
    expired = queue.get_task(task["task_id"])
    assert expired.status == "failed"
    assert expired.error_code == "task_deadline_expired"


def test_stale_task_lease_is_reclaimed_and_old_owner_is_fenced(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db", clock=lambda: now[0], lease_seconds=10
    )
    queue.enqueue(APPROVED_TASK)
    old = queue.claim("worker-old")
    assert old is not None
    now[0] = datetime(2026, 8, 9, 8, 0, 11, tzinfo=UTC)

    reclaimed = queue.claim("worker-new")

    assert reclaimed is not None
    assert reclaimed.lease_token != old.lease_token
    with pytest.raises(EdgeQueueLeaseError, match="lease"):
        queue.complete(old.task_id, lease_token=old.lease_token)


def test_expired_lease_is_fenced_even_before_another_worker_reclaims_it(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db", clock=lambda: now[0], lease_seconds=10
    )
    queue.enqueue(APPROVED_TASK)
    claimed = queue.claim("worker-old")
    now[0] = datetime(2026, 8, 9, 8, 0, 11, tzinfo=UTC)

    with pytest.raises(EdgeQueueLeaseError, match="expired"):
        queue.complete(claimed.task_id, lease_token=claimed.lease_token)


def test_expired_final_attempt_is_terminal_instead_of_remaining_leased(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db",
        clock=lambda: now[0],
        lease_seconds=10,
        max_attempts=1,
    )
    queue.enqueue(APPROVED_TASK)
    queue.claim("worker")
    now[0] = datetime(2026, 8, 9, 8, 0, 11, tzinfo=UTC)

    assert queue.claim("next-worker") is None
    assert queue.get_task(APPROVED_TASK["task_id"]).status == "failed"


def test_fail_uses_bounded_exponential_retry_then_becomes_terminal(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db",
        clock=lambda: now[0],
        lease_seconds=10,
        max_attempts=3,
        retry_base_seconds=2,
        retry_max_seconds=3,
    )
    queue.enqueue(APPROVED_TASK)

    first = queue.claim("worker")
    retried = queue.fail(
        first.task_id, lease_token=first.lease_token, error_code="source_unavailable"
    )
    assert retried.status == "pending"
    assert retried.available_at == "2026-08-09T08:00:02Z"
    assert queue.claim("too-early") is None

    now[0] = datetime(2026, 8, 9, 8, 0, 2, tzinfo=UTC)
    second = queue.claim("worker")
    queue.fail(
        second.task_id, lease_token=second.lease_token, error_code="source_unavailable"
    )
    now[0] = datetime(2026, 8, 9, 8, 0, 5, tzinfo=UTC)
    third = queue.claim("worker")
    terminal = queue.fail(
        third.task_id, lease_token=third.lease_token, error_code="source_unavailable"
    )
    assert terminal.status == "failed"
    assert terminal.attempt_count == 3
    assert queue.enqueue(APPROVED_TASK).status == "failed"


def test_cancel_is_atomic_and_prevents_lease_completion(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    queue.enqueue(APPROVED_TASK)
    claimed = queue.claim("worker")

    cancelled = queue.cancel(claimed.task_id)

    assert cancelled.status == "cancelled"
    assert queue.cancel(claimed.task_id).status == "cancelled"
    with pytest.raises(EdgeQueueLeaseError):
        queue.complete(claimed.task_id, lease_token=claimed.lease_token)


def test_cancel_and_complete_race_reaches_one_consistent_terminal_state(tmp_path):
    db_path = tmp_path / "edge.db"
    first = SqliteEdgeQueue(db_path)
    second = SqliteEdgeQueue(db_path)
    first.enqueue(APPROVED_TASK)
    claimed = first.claim("worker")
    barrier = Barrier(2)

    def cancel():
        barrier.wait(timeout=5)
        return second.cancel(claimed.task_id).status

    def complete():
        barrier.wait(timeout=5)
        try:
            return first.complete(
                claimed.task_id, lease_token=claimed.lease_token
            ).status
        except EdgeQueueLeaseError:
            return "lease_rejected"

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = [pool.submit(cancel), pool.submit(complete)]
        outcomes = {result.result(timeout=5) for result in results}
    final = first.get_task(claimed.task_id)
    assert final.status in {"cancelled", "completed"}
    assert "completed" not in outcomes or final.status == "completed"


def test_transaction_exception_rolls_back_and_queue_remains_usable(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    queue.enqueue(APPROVED_TASK)

    with (
        pytest.raises(RuntimeError, match="injected"),
        queue._transaction() as connection,
    ):
        connection.execute(
            "UPDATE edge_tasks SET status = 'failed' WHERE task_id = ?",
            (APPROVED_TASK["task_id"],),
        )
        raise RuntimeError("injected lock-path failure")

    assert queue.get_task(APPROVED_TASK["task_id"]).status == "pending"
    assert queue.claim("worker") is not None


def test_outbound_event_is_persisted_before_claim_and_exact_ack_replays(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    event = _event()
    persisted = queue.persist_event(event)
    assert persisted.status == "pending"
    with sqlite3.connect(queue.db_path) as connection:
        stored = connection.execute(
            "SELECT event_json FROM edge_outbound_events WHERE event_id = ?",
            (event["event_id"],),
        ).fetchone()
    assert json.loads(stored[0]) == event

    sending = queue.claim_event("sender-a")
    acknowledged = queue.acknowledge_event(
        sending.event_id,
        lease_token=sending.lease_token,
        acknowledgement=_ack(),
    )
    assert acknowledged.status == "acknowledged"
    replay = queue.persist_event(event)
    assert replay.status == "acknowledged"
    assert replay.acknowledgement == {
        "message_id": "control-message-1",
        "received_at": "2026-08-09T08:00:00Z",
        "status": "accepted",
    }


def test_acknowledgement_exact_replay_binds_ack_and_original_lease_token(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    queue.persist_event(_event())
    sending = queue.claim_event("sender")
    first = queue.acknowledge_event(
        sending.event_id,
        lease_token=sending.lease_token,
        acknowledgement=_ack(),
    )
    replay = queue.acknowledge_event(
        sending.event_id,
        lease_token=sending.lease_token,
        acknowledgement=_ack(received_at="2026-08-09T08:00:00Z"),
    )
    assert replay.acknowledgement == first.acknowledgement

    with pytest.raises(EdgeQueueLeaseError):
        queue.acknowledge_event(
            sending.event_id,
            lease_token="different-token",
            acknowledgement=_ack(received_at="2026-08-09T08:00:00Z"),
        )
    with pytest.raises(EdgeQueueConflictError):
        queue.acknowledge_event(
            sending.event_id,
            lease_token=sending.lease_token,
            acknowledgement={
                **_ack(received_at="2026-08-09T08:00:00Z"),
                "message_id": "different-message",
            },
        )
    with sqlite3.connect(queue.db_path) as connection:
        row = connection.execute(
            """
            SELECT lease_token, ack_digest, lease_token_digest
            FROM edge_outbound_events WHERE event_id = ?
            """,
            (sending.event_id,),
        ).fetchone()
    assert row[0] is None
    assert row[1] == canonical_sha256(dict(first.acknowledgement))
    assert row[2] == canonical_sha256(sending.lease_token)


def test_acknowledgement_requires_exact_fields_timestamp_and_status(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    queue.persist_event(_event())
    sending = queue.claim_event("sender")
    for acknowledgement in (
        {"message_id": "id", "status": "accepted"},
        {**_ack(), "extra": "no"},
        {**_ack(), "received_at": "yesterday"},
        {**_ack(), "status": "unknown"},
    ):
        with pytest.raises(EdgePolicyError):
            queue.acknowledge_event(
                sending.event_id,
                lease_token=sending.lease_token,
                acknowledgement=acknowledgement,
            )


def test_event_persistence_digests_normalized_contract_before_exact_replay(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    utc_event = _event(occurred_at="2026-08-09T08:00:05Z")
    offset_event = _event(occurred_at="2026-08-09T09:00:05+01:00")
    offset_event["event_id"] = utc_event["event_id"]
    contract = EdgeEventContract.from_mapping(offset_event)

    persisted = queue.persist_event(offset_event)
    replay = queue.persist_event(utc_event)

    assert replay.event_id == persisted.event_id
    assert replay.digest == persisted.digest == contract.digest
    assert replay.event["occurred_at"] == "2026-08-09T08:00:05Z"
    with sqlite3.connect(queue.db_path) as connection:
        stored_digest = connection.execute(
            "SELECT digest FROM edge_outbound_events WHERE event_id = ?",
            (contract.event_id,),
        ).fetchone()[0]
    assert stored_digest == contract.digest


def test_outbound_event_changed_digest_conflicts_and_unsafe_payload_never_persists(
    tmp_path,
):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    event = _event()
    queue.persist_event(event)
    changed = {**event, "payload": {"metric_count": 4}}
    with pytest.raises(EdgeQueueConflictError, match="digest"):
        queue.persist_event(changed)

    unsafe = _event(payload={"rows": [{"password": "secret"}]})
    with pytest.raises((EdgeContractError, EdgePolicyError)):
        queue.persist_event(unsafe)
    with sqlite3.connect(queue.db_path) as connection:
        serialized = " ".join(
            str(value)
            for row in connection.execute(
                "SELECT event_json FROM edge_outbound_events"
            ).fetchall()
            for value in row
        ).lower()
    assert "password" not in serialized
    assert "secret" not in serialized


def test_reviewed_sensitive_payloads_are_rejected_before_event_persistence(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    payloads = [
        {"diagnostic": "upstream says api_key: sk-sensitive-value"},
        {"diagnostic": "prefix INSERT INTO customer VALUES (1) suffix"},
        {"data": [{"customer_id": "customer-1", "name": "Raw Name"}]},
    ]

    for payload in payloads:
        with pytest.raises(EdgePolicyError, match="sensitive"):
            queue.persist_event(
                _event(classification="diagnostic_summary", payload=payload)
            )

    with sqlite3.connect(queue.db_path) as connection:
        assert connection.execute(
            "SELECT COUNT(*) FROM edge_outbound_events"
        ).fetchone()[0] == 0


def test_two_connections_atomically_claim_one_outbound_event_only(tmp_path):
    db_path = tmp_path / "edge.db"
    first = SqliteEdgeQueue(db_path)
    second = SqliteEdgeQueue(db_path)
    first.persist_event(_event())

    claims = [first.claim_event("sender-a"), second.claim_event("sender-b")]

    assert sum(claim is not None for claim in claims) == 1
    claimed = next(claim for claim in claims if claim is not None)
    assert claimed.lease_owner in {"sender-a", "sender-b"}


def test_two_threads_racing_independent_event_connections_have_one_winner(tmp_path):
    db_path = tmp_path / "edge.db"
    first = SqliteEdgeQueue(db_path)
    second = SqliteEdgeQueue(db_path)
    first.persist_event(_event())
    barrier = Barrier(2)

    def claim(queue, owner):
        barrier.wait(timeout=5)
        return queue.claim_event(owner)

    with ThreadPoolExecutor(max_workers=2) as pool:
        results = list(
            pool.map(claim, (first, second), ("sender-a", "sender-b"))
        )
    assert sum(result is not None for result in results) == 1


def test_event_retry_is_bounded_and_uses_independent_event_attempts(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db",
        clock=lambda: now[0],
        max_attempts=2,
        retry_base_seconds=1,
        retry_max_seconds=2,
    )
    queue.persist_event(_event())
    first = queue.claim_event("sender")
    retry = queue.fail_event(
        first.event_id, lease_token=first.lease_token, error_code="network_unavailable"
    )
    assert retry.status == "pending"
    now[0] = datetime(2026, 8, 9, 8, 0, 1, tzinfo=UTC)
    second = queue.claim_event("sender")
    failed = queue.fail_event(
        second.event_id,
        lease_token=second.lease_token,
        error_code="network_unavailable",
    )
    assert failed.status == "failed"
    assert failed.attempt_count == 2


def test_signed_task_envelope_verifies_canonical_ed25519_authority_and_bindings():
    private = Ed25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    raw = _signed_task(private)

    envelope = SignedTaskEnvelope.verify_mapping(
        raw,
        authority_keys={"control-authority-2026-01": public},
        now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
    )

    assert envelope.task.to_mapping() == APPROVED_TASK
    assert envelope.contract_digest == canonical_sha256(APPROVED_TASK)
    assert envelope.to_mapping() == raw
    for changed in (
        {**raw, "purpose": "other-purpose"},
        {**raw, "contract_digest": "b" * 64},
        {**raw, "signature": "00" * 64},
        {**raw, "extra": "rejected"},
    ):
        with pytest.raises(EdgeContractError):
            SignedTaskEnvelope.verify_mapping(
                changed,
                authority_keys={"control-authority-2026-01": public},
                now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
            )
    with pytest.raises(EdgeContractError, match="expired"):
        SignedTaskEnvelope.verify_mapping(
            _signed_task(private, expires_at="2026-08-09T07:59:59Z"),
            authority_keys={"control-authority-2026-01": public},
            now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
        )


def test_signed_task_authority_clock_skew_only_tolerates_bounded_future_issue_time():
    private = Ed25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    now = datetime(2026, 8, 9, 8, 0, tzinfo=UTC)
    accepted = SignedTaskEnvelope.verify_mapping(
        _signed_task(private, issued_at="2026-08-09T08:01:00Z"),
        authority_keys={"control-authority-2026-01": public},
        now=now,
        allowed_future_skew_seconds=60,
    )
    assert accepted.issued_at == "2026-08-09T08:01:00Z"
    with pytest.raises(EdgeContractError, match="not yet valid"):
        SignedTaskEnvelope.verify_mapping(
            _signed_task(private, issued_at="2026-08-09T08:01:01Z"),
            authority_keys={"control-authority-2026-01": public},
            now=now,
            allowed_future_skew_seconds=60,
        )
    with pytest.raises(EdgeContractError, match="expired"):
        SignedTaskEnvelope.verify_mapping(
            _signed_task(
                private,
                issued_at="2026-08-09T07:59:00Z",
                expires_at="2026-08-09T08:00:00Z",
            ),
            authority_keys={"control-authority-2026-01": public},
            now=now,
            allowed_future_skew_seconds=60,
        )
    for invalid in (-1, 301, True):
        with pytest.raises(EdgeContractError, match="clock skew"):
            SignedTaskEnvelope.verify_mapping(
                _signed_task(private),
                authority_keys={"control-authority-2026-01": public},
                now=now,
                allowed_future_skew_seconds=invalid,
            )


@pytest.mark.parametrize("task_status", ["pending", "leased"])
def test_v1_queue_refuses_active_unsigned_tasks_without_partial_schema_change(
    tmp_path, task_status
):
    db_path = tmp_path / "edge.db"
    with sqlite3.connect(db_path) as connection:
        connection.execute(edge_queue_module._V1_TASK_TABLE_SQL)
        connection.execute(edge_queue_module._TASK_INDEX_SQL)
        connection.execute(edge_queue_module._V1_EVENT_TABLE_SQL)
        connection.execute(edge_queue_module._EVENT_INDEX_SQL)
        connection.execute(
            """INSERT INTO edge_tasks (
            task_id,digest,task_json,status,attempt_count,available_at,deadline_at,
            deadline_epoch_us,lease_owner,lease_token,lease_expires_at,created_at,updated_at
            ) VALUES (?,?,?,?,0,?,?,?,?,?,?,?,?)""",
            (
                APPROVED_TASK["task_id"], canonical_sha256(APPROVED_TASK),
                json.dumps(APPROVED_TASK, sort_keys=True, separators=(",", ":")),
                task_status,
                "2026-08-09T08:00:00Z", APPROVED_TASK["deadline_at"],
                4090060800000000,
                "worker" if task_status == "leased" else None,
                "local-lease" if task_status == "leased" else None,
                "2026-08-09T08:01:00Z" if task_status == "leased" else None,
                "2026-08-09T08:00:00Z", "2026-08-09T08:00:00Z",
            ),
        )
        connection.execute("PRAGMA user_version = 1")

    with pytest.raises(EdgeQueueSchemaError, match="drain"):
        SqliteEdgeQueue(db_path)
    with sqlite3.connect(db_path) as connection:
        assert connection.execute("PRAGMA user_version").fetchone()[0] == 1
        assert {
            row[1] for row in connection.execute("PRAGMA table_info(edge_tasks)")
        } == {
            row[0] for row in edge_queue_module._V1_EXPECTED_TABLE_INFO["edge_tasks"]
        }
        assert connection.execute(
            "SELECT COUNT(*) FROM sqlite_master WHERE name='edge_local_artifacts'"
        ).fetchone()[0] == 0


@pytest.mark.parametrize("event_status", ["pending", "sending", "failed"])
def test_v1_queue_refuses_unacknowledged_event_without_partial_migration(
    tmp_path, event_status
):
    db_path = tmp_path / "edge.db"
    event = _event(occurred_at="2026-08-09T08:00:00Z")
    with sqlite3.connect(db_path) as connection:
        connection.execute(edge_queue_module._V1_TASK_TABLE_SQL)
        connection.execute(edge_queue_module._TASK_INDEX_SQL)
        connection.execute(edge_queue_module._V1_EVENT_TABLE_SQL)
        connection.execute(edge_queue_module._EVENT_INDEX_SQL)
        connection.execute(
            """INSERT INTO edge_outbound_events (
            event_id,digest,event_json,status,attempt_count,available_at,
            created_at,updated_at
            ) VALUES (?,?,?,?,0,?,?,?)""",
            (
                event["event_id"],
                canonical_sha256(event),
                json.dumps(event, sort_keys=True, separators=(",", ":")),
                event_status,
                "2026-08-09T08:00:00Z",
                "2026-08-09T08:00:00Z",
                "2026-08-09T08:00:00Z",
            ),
        )
        connection.execute("PRAGMA user_version = 1")

    with pytest.raises(EdgeQueueSchemaError, match="drain"):
        SqliteEdgeQueue(db_path)
    with sqlite3.connect(db_path) as connection:
        assert connection.execute("PRAGMA user_version").fetchone()[0] == 1
        assert connection.execute(
            "SELECT COUNT(*) FROM pragma_table_info('edge_outbound_events') "
            "WHERE name='task_digest'"
        ).fetchone()[0] == 0


def test_v1_queue_migrates_terminal_tasks_and_acknowledged_events_then_accepts_signed(
    tmp_path,
):
    db_path = tmp_path / "edge.db"
    event = _event(occurred_at="2026-08-09T08:00:00Z")
    with sqlite3.connect(db_path) as connection:
        connection.execute(edge_queue_module._V1_TASK_TABLE_SQL)
        connection.execute(edge_queue_module._TASK_INDEX_SQL)
        connection.execute(edge_queue_module._V1_EVENT_TABLE_SQL)
        connection.execute(edge_queue_module._EVENT_INDEX_SQL)
        connection.execute(
            """INSERT INTO edge_tasks (
            task_id,digest,task_json,status,attempt_count,available_at,deadline_at,
            deadline_epoch_us,created_at,updated_at
            ) VALUES (?,?,?,'completed',1,?,?,?,?,?)""",
            (
                APPROVED_TASK["task_id"], canonical_sha256(APPROVED_TASK),
                json.dumps(APPROVED_TASK, sort_keys=True, separators=(",", ":")),
                "2026-08-09T08:00:00Z", APPROVED_TASK["deadline_at"],
                4090060800000000, "2026-08-09T08:00:00Z", "2026-08-09T08:00:00Z",
            ),
        )
        connection.execute(
            """INSERT INTO edge_outbound_events (
            event_id,digest,event_json,status,attempt_count,available_at,
            acknowledgement_json,ack_digest,created_at,updated_at
            ) VALUES (?,?,?,'acknowledged',1,?,?,?,?,?)""",
            (
                event["event_id"],
                canonical_sha256(event),
                json.dumps(event, sort_keys=True, separators=(",", ":")),
                "2026-08-09T08:00:00Z",
                '{"status":"accepted"}',
                "b" * 64,
                "2026-08-09T08:00:00Z",
                "2026-08-09T08:00:00Z",
            ),
        )
        connection.execute("PRAGMA user_version = 1")

    migrated = SqliteEdgeQueue(db_path)
    assert SCHEMA_VERSION == 2
    assert migrated.get_task(APPROVED_TASK["task_id"]).status == "completed"
    assert migrated.get_event(event["event_id"]).status == "acknowledged"

    private = Ed25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    task = {**APPROVED_TASK, "task_id": "task-signed-v2"}
    envelope = SignedTaskEnvelope.verify_mapping(
        _signed_task(private, task=task),
        authority_keys={"control-authority-2026-01": public},
        now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
    )
    queued = migrated.enqueue_signed(envelope)
    assert queued.authority_key_id == "control-authority-2026-01"
    with sqlite3.connect(db_path) as connection:
        row = connection.execute(
            "SELECT authority_key_id,authority_digest,authority_json FROM edge_tasks WHERE task_id=?",
            (task["task_id"],),
        ).fetchone()
    assert row[0] == "control-authority-2026-01"
    assert row[1] == envelope.digest
    assert "signature" in row[2]


def test_remote_lease_binding_and_atomic_event_artifact_completion(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(tmp_path / "edge.db", clock=lambda: now[0])
    queue.enqueue(APPROVED_TASK)
    claimed = queue.claim("worker")
    bound = queue.bind_remote_lease(
        claimed.task_id,
        lease_token=claimed.lease_token,
        remote_lease_token="remote-control-lease-1",
        remote_lease_expires_at="2026-08-09T08:05:00Z",
        remote_attempt=claimed.attempt_count,
    )
    assert bound.remote_lease_token_digest == hashlib.sha256(b"remote-control-lease-1").hexdigest()
    assert bound.remote_lease_expires_at == "2026-08-09T08:05:00Z"

    event = _event(occurred_at="2026-08-09T08:00:00Z")
    artifact_ref = str((tmp_path / "artifact.bin").resolve())
    task, persisted = queue.complete_with_event(
        claimed.task_id,
        lease_token=claimed.lease_token,
        remote_lease_token="remote-control-lease-1",
        event=event,
        artifacts=[{
            "artifact_digest": "f" * 64,
            "artifact_ref": artifact_ref,
            "artifact_ref_hash": canonical_sha256(artifact_ref),
            "classification": "raw",
            "retention_until": "2026-08-09T08:00:00Z",
        }],
    )
    assert task.status == "completed"
    assert persisted.status == "pending"
    artifacts = queue.list_local_artifacts(task_id=claimed.task_id)
    assert artifacts[0].artifact_ref == artifact_ref
    assert artifacts[0].cleanup_status == "pending"
    with sqlite3.connect(queue.db_path) as connection:
        columns = {row[1] for row in connection.execute("PRAGMA table_info(edge_local_artifacts)")}
        assert "body" not in columns and "content" not in columns


def test_atomic_completion_rechecks_cancel_deadline_and_remote_lease(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(tmp_path / "edge.db", clock=lambda: now[0], lease_seconds=600)
    queue.enqueue({**APPROVED_TASK, "deadline_at": "2026-08-09T08:00:05Z"})
    claimed = queue.claim("worker")
    queue.bind_remote_lease(
        claimed.task_id, lease_token=claimed.lease_token,
        remote_lease_token="remote", remote_lease_expires_at="2026-08-09T08:00:04Z",
        remote_attempt=1,
    )
    now[0] = datetime(2026, 8, 9, 8, 0, 5, tzinfo=UTC)
    with pytest.raises(EdgeQueueLeaseError):
        queue.complete_with_event(
            claimed.task_id, lease_token=claimed.lease_token,
            remote_lease_token="remote", event=_event(), artifacts=[],
        )
    assert queue.get_event(_event()["event_id"]) is None

    second_task = {**APPROVED_TASK, "task_id": "task-cancelled"}
    queue.enqueue(second_task)
    second = queue.claim("worker")
    queue.bind_remote_lease(
        second.task_id, lease_token=second.lease_token,
        remote_lease_token="remote-2", remote_lease_expires_at="2099-01-01T00:00:00Z",
        remote_attempt=1,
    )
    queue.cancel(second.task_id)
    cancelled_event = _event()
    cancelled_event["task_id"] = second.task_id
    cancelled_event["event_id"] = stable_event_id({k: v for k, v in cancelled_event.items() if k != "event_id"})
    with pytest.raises(EdgeQueueLeaseError):
        queue.complete_with_event(
            second.task_id, lease_token=second.lease_token,
            remote_lease_token="remote-2", event=cancelled_event, artifacts=[],
        )
    assert queue.get_event(cancelled_event["event_id"]) is None


def test_bound_event_ack_requires_event_digest_and_remote_control_leases(tmp_path):
    queue = SqliteEdgeQueue(tmp_path / "edge.db")
    queue.enqueue(APPROVED_TASK)
    claimed = queue.claim("worker")
    queue.bind_remote_lease(
        claimed.task_id, lease_token=claimed.lease_token,
        remote_lease_token="remote", remote_lease_expires_at="2099-01-01T00:00:00Z",
        remote_attempt=1,
    )
    _, persisted = queue.complete_with_event(
        claimed.task_id, lease_token=claimed.lease_token,
        remote_lease_token="remote", event=_event(), artifacts=[],
    )
    sending = queue.claim_event("sender")
    acknowledgement = {
        "event_id": sending.event_id,
        "event_digest": sending.digest,
        "remote_lease_digest": hashlib.sha256(b"remote").hexdigest(),
        "received_at": "2026-08-09T08:00:00Z",
        "status": "accepted",
    }
    acknowledged = queue.acknowledge_event(
        persisted.event_id, lease_token=sending.lease_token,
        acknowledgement=acknowledgement,
    )
    assert acknowledged.status == "acknowledged"
    with pytest.raises(EdgePolicyError):
        queue.acknowledge_event(
            persisted.event_id, lease_token=sending.lease_token,
            acknowledgement={**acknowledgement, "remote_lease_digest": "0" * 64},
        )


def test_retry_jitter_is_injected_and_used_for_task_and_event(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db", clock=lambda: now[0], retry_base_seconds=10,
        retry_max_seconds=100, random_source=lambda: 0.0,
    )
    queue.enqueue(APPROVED_TASK)
    task = queue.claim("worker")
    retried = queue.fail(task.task_id, lease_token=task.lease_token, error_code="temporary")
    assert retried.available_at == "2026-08-09T08:00:05Z"

    queue.persist_event(_event())
    event = queue.claim_event("sender")
    event_retry = queue.fail_event(event.event_id, lease_token=event.lease_token, error_code="temporary")
    assert event_retry.available_at == "2026-08-09T08:00:05Z"


def test_signed_acceptance_encrypts_recoverable_remote_lease_atomically(tmp_path):
    private = Ed25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    envelope = SignedTaskEnvelope.verify_mapping(
        _signed_task(private),
        authority_keys={"control-authority-2026-01": public},
        now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
    )
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db",
        clock=lambda: now[0],
        remote_lease_encoder=lambda token: "cipher:" + token[::-1],
        remote_lease_decoder=lambda value: value.removeprefix("cipher:")[::-1],
    )
    queued = queue.accept_signed_task(
        envelope,
        remote_lease_token="remote-secret",
        remote_lease_expires_at="2026-08-09T08:05:00Z",
        remote_attempt=1,
    )
    assert queued.remote_lease_token_digest == hashlib.sha256(b"remote-secret").hexdigest()
    assert "remote-secret" not in repr(queued)
    assert queue.recover_remote_lease(queued.task_id) == "remote-secret"
    with sqlite3.connect(queue.db_path) as connection:
        stored = connection.execute(
            "SELECT remote_lease_ciphertext FROM edge_tasks WHERE task_id=?",
            (queued.task_id,),
        ).fetchone()[0]
    assert stored != "remote-secret"
    now[0] = datetime(2026, 8, 9, 8, 6, tzinfo=UTC)
    renewed = queue.accept_signed_task(
        envelope,
        remote_lease_token="renewed-secret",
        remote_lease_expires_at="2026-08-09T08:10:00Z",
        remote_attempt=1,
    )
    assert renewed.authority_digest == queued.authority_digest
    assert queue.recover_remote_lease(queued.task_id) == "renewed-secret"


def test_remote_attempt_remains_contract_attempt_across_local_crash_recovery(tmp_path):
    private = Ed25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    envelope = SignedTaskEnvelope.verify_mapping(
        _signed_task(private),
        authority_keys={"control-authority-2026-01": public},
        now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
    )
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db",
        clock=lambda: now[0],
        lease_seconds=5,
        remote_lease_encoder=lambda token: "cipher:" + token[::-1],
        remote_lease_decoder=lambda value: value.removeprefix("cipher:")[::-1],
    )
    queue.accept_signed_task(
        envelope,
        remote_lease_token="remote-a",
        remote_lease_expires_at="2026-08-09T08:00:04Z",
        remote_attempt=1,
    )
    crashed_claim = queue.claim("crashed-agent")
    assert crashed_claim.attempt_count == 1

    now[0] = datetime(2026, 8, 9, 8, 0, 6, tzinfo=UTC)
    queue.accept_signed_task(
        envelope,
        remote_lease_token="remote-b",
        remote_lease_expires_at="2026-08-09T08:05:00Z",
        remote_attempt=1,
    )
    recovered_claim = queue.claim("replacement-agent")
    assert recovered_claim.attempt_count == 2
    assert recovered_claim.remote_attempt == 1
    completed, event = queue.complete_with_event(
        recovered_claim.task_id,
        lease_token=recovered_claim.lease_token,
        remote_lease_token="remote-b",
        event=_event(occurred_at="2026-08-09T08:00:06Z"),
    )
    assert completed.status == "completed"
    assert event.attempt_count == 0
    with sqlite3.connect(queue.db_path) as connection:
        assert connection.execute(
            "SELECT count(*) FROM edge_outbound_events"
        ).fetchone()[0] == 1


def test_expired_remote_lease_rebinds_pending_event_for_exact_ack_replay(tmp_path):
    private = Ed25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    envelope = SignedTaskEnvelope.verify_mapping(
        _signed_task(private),
        authority_keys={"control-authority-2026-01": public},
        now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
    )
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db",
        clock=lambda: now[0],
        remote_lease_encoder=lambda token: "cipher:" + token[::-1],
        remote_lease_decoder=lambda value: value.removeprefix("cipher:")[::-1],
    )
    queue.accept_signed_task(
        envelope,
        remote_lease_token="remote-a",
        remote_lease_expires_at="2026-08-09T08:00:05Z",
        remote_attempt=1,
    )
    task = queue.claim("worker")
    _, pending = queue.complete_with_event(
        task.task_id,
        lease_token=task.lease_token,
        remote_lease_token="remote-a",
        event=_event(),
    )
    first_send = queue.claim_event("sender-a")
    queue.fail_event(
        first_send.event_id,
        lease_token=first_send.lease_token,
        error_code="ack_lost",
    )

    now[0] = datetime(2026, 8, 9, 8, 0, 6, tzinfo=UTC)
    queue.accept_signed_task(
        envelope,
        remote_lease_token="remote-b",
        remote_lease_expires_at="2026-08-09T08:05:00Z",
        remote_attempt=1,
    )
    rebound = queue.get_event(pending.event_id)
    assert rebound.remote_lease_token_digest == hashlib.sha256(b"remote-b").hexdigest()
    second_send = queue.claim_event("sender-b")
    acknowledged = queue.acknowledge_event(
        second_send.event_id,
        lease_token=second_send.lease_token,
        acknowledgement={
            "event_id": second_send.event_id,
            "event_digest": second_send.digest,
            "remote_lease_digest": hashlib.sha256(b"remote-b").hexdigest(),
            "received_at": "2026-08-09T08:00:06Z",
            "status": "accepted",
        },
    )
    assert acknowledged.status == "acknowledged"


def test_remote_reacquire_revives_only_execution_lease_expiry(tmp_path):
    private = Ed25519PrivateKey.generate()
    public = private.public_key().public_bytes(
        serialization.Encoding.Raw, serialization.PublicFormat.Raw
    )
    envelope = SignedTaskEnvelope.verify_mapping(
        _signed_task(private),
        authority_keys={"control-authority-2026-01": public},
        now=datetime(2026, 8, 9, 8, 0, tzinfo=UTC),
    )
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]

    def new_queue(path):
        return SqliteEdgeQueue(
            path,
            clock=lambda: now[0],
            remote_lease_encoder=lambda token: "cipher:" + token[::-1],
            remote_lease_decoder=lambda value: value.removeprefix("cipher:")[::-1],
        )

    recoverable = new_queue(tmp_path / "recoverable.db")
    recoverable.accept_signed_task(
        envelope,
        remote_lease_token="remote-a",
        remote_lease_expires_at="2026-08-09T08:00:05Z",
        remote_attempt=1,
    )
    claimed = recoverable.claim("worker")
    recoverable.fail(
        claimed.task_id,
        lease_token=claimed.lease_token,
        error_code="execution_lease_expired",
        retryable=False,
    )
    now[0] = datetime(2026, 8, 9, 8, 0, 6, tzinfo=UTC)
    revived = recoverable.accept_signed_task(
        envelope,
        remote_lease_token="remote-b",
        remote_lease_expires_at="2026-08-09T08:05:00Z",
        remote_attempt=1,
    )
    assert revived.status == "pending"
    assert revived.error_code is None

    business = new_queue(tmp_path / "business.db")
    business.accept_signed_task(
        envelope,
        remote_lease_token="remote-a",
        remote_lease_expires_at="2026-08-09T08:00:07Z",
        remote_attempt=1,
    )
    business_claim = business.claim("worker")
    business.fail(
        business_claim.task_id,
        lease_token=business_claim.lease_token,
        error_code="runner_execution_failed",
        retryable=False,
    )
    now[0] = datetime(2026, 8, 9, 8, 0, 8, tzinfo=UTC)
    with pytest.raises(EdgeQueueConflictError, match="terminal task"):
        business.accept_signed_task(
            envelope,
            remote_lease_token="remote-b",
            remote_lease_expires_at="2026-08-09T08:05:00Z",
            remote_attempt=1,
        )


def test_retry_jitter_is_capped_after_jitter_for_task_and_event(tmp_path):
    now = [datetime(2026, 8, 9, 8, 0, tzinfo=UTC)]
    queue = SqliteEdgeQueue(
        tmp_path / "edge.db",
        clock=lambda: now[0],
        retry_base_seconds=10,
        retry_max_seconds=10,
        random_source=lambda: 1.0,
    )
    queue.enqueue(APPROVED_TASK)
    task = queue.claim("worker")
    task_retry = queue.fail(
        task.task_id, lease_token=task.lease_token, error_code="temporary"
    )
    assert task_retry.available_at == "2026-08-09T08:00:10Z"

    queue.persist_event(_event())
    event = queue.claim_event("sender")
    event_retry = queue.fail_event(
        event.event_id, lease_token=event.lease_token, error_code="temporary"
    )
    assert event_retry.available_at == "2026-08-09T08:00:10Z"


def test_release_state_is_durable_and_transitions_with_atomic_cas(tmp_path):
    db_path = tmp_path / "edge.db"
    first = SqliteEdgeQueue(db_path)
    state = first.put_release_state(
        release_id="release-1", manifest_digest="d" * 64,
        version="3.1.0", rollback_version="3.0.0",
    )
    assert state.status == "offered"
    accepted = first.compare_and_set_release_state(
        "release-1", expected_status="offered", target_status="accepted",
        previous_version="3.0.0",
    )
    assert accepted.status == "accepted"
    restarted = SqliteEdgeQueue(db_path)
    assert restarted.get_release_state("release-1") == accepted
    with pytest.raises(EdgeQueueConflictError, match="compare-and-set"):
        restarted.compare_and_set_release_state(
            "release-1", expected_status="offered", target_status="accepted"
        )
    candidate = restarted.compare_and_set_release_state(
        "release-1", expected_status="accepted", target_status="candidate"
    )
    installed = restarted.compare_and_set_release_state(
        "release-1", expected_status="candidate", target_status="installed",
        current_version="3.1.0",
    )
    assert candidate.status == "candidate"
    assert installed.current_version == "3.1.0"
    assert restarted.list_release_states() == (installed,)