awoooi/apps/api/tests/test_agent99_same_run_reconcile.py

from __future__ import annotations

import json
from copy import deepcopy
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock

import pytest

from src.jobs import agent99_controlled_dispatch_reconciler_job as job
from src.services import (
    agent99_controlled_dispatch_ledger as ledger_module,
)
from src.services import agent99_same_run_reconcile as reconcile
from src.services import agent99_sre_bridge as bridge
from src.services.agent99_controlled_dispatch_ledger import (
    PostgresAgent99DispatchLedger,
    build_agent99_dispatch_identity,
    build_agent99_dispatch_receipt_envelope,
)

IDENTITY = build_agent99_dispatch_identity(
    project_id="awoooi",
    incident_id="INC-20260711-11C751",
    source_fingerprint="legacy-cold-start:" + "a" * 64,
    route_id="agent99:host_recovery:Recover",
    approval_id="00000000-0000-0000-0000-000000000751",
    work_item_id=(
        "agent99-dispatch:awoooi:INC-20260711-11C751:legacy-cold-start"
    ),
)


@pytest.fixture(autouse=True)
def _bind_test_identity_to_bounded_run(monkeypatch) -> None:
    monkeypatch.setattr(
        reconcile,
        "AGENT99_COLD_START_RUN_ID",
        str(IDENTITY.run_id),
    )


def _source_receipt() -> dict[str, object]:
    return {
        "schema_version": "agent99_cold_start_source_resolution_v1",
        "status": "source_resolved",
        "resolved": True,
        "source_result": "degraded",
        "degraded_provenance": True,
        "receipt_ref": (
            "prometheus:cold-start:1784020000:pass-95:"
            "warn-1:blocked-0:result-degraded:blocking-alerts-0"
        ),
        "observed_at_epoch": 1784020030,
        "age_seconds": 30.0,
        "values": {
            "monitor_up": 1,
            "pass_gates": 95,
            "warn_gates": 1,
            "blocked_gates": 0,
            "green_result": 0,
            "degraded_result": 1,
            "blocked_result": 0,
            "check_failed_result": 0,
            "firing_blocking_alerts": 0,
        },
        "blockers": [],
        "stores_raw_evidence": False,
    }


def _valid_outcome() -> dict[str, object]:
    identity = IDENTITY.public_dict()
    return {
        "schemaVersion": "agent99_outcome_readback_v1",
        "sameRunContractFieldsComplete": True,
        "automationRunId": str(IDENTITY.run_id),
        "traceId": IDENTITY.trace_id,
        "workItemId": IDENTITY.work_item_id,
        "projectId": IDENTITY.project_id,
        "incidentId": IDENTITY.incident_id,
        "routeId": IDENTITY.route_id,
        "executionGeneration": IDENTITY.execution_generation,
        "approvalId": IDENTITY.approval_id,
        "evidenceReceiptId": reconcile.agent99_same_run_evidence_id(IDENTITY),
        "identity": identity,
        "mode": "Status",
        "controlledApply": False,
        "reconcileOnly": True,
        "runtimeWritePerformed": False,
        "outcome": {
            "schemaVersion": "agent99_outcome_contract_v1",
            "state": "resolved",
            "resolved": True,
            "transportOk": True,
            "verifierName": reconcile.AGENT99_SAME_RUN_STATUS_VERIFIER,
            "verifierPassed": True,
            "sourceEventResolved": True,
            "sourceEventEvidence": _source_receipt()["receipt_ref"],
            "identity": identity,
            "verifiedAt": "2026-07-14T16:30:00+08:00",
        },
    }


def _evidence_refs() -> dict[str, str]:
    return {
        "agent99_outcome_receipt_id": (
            f"agent99-relay:outcome:{IDENTITY.run_id}"
        ),
        "post_verifier_evidence_ref": (
            "agent99:Status:no-write:"
            f"{reconcile.agent99_same_run_evidence_id(IDENTITY)}"
        ),
        "source_event_evidence_ref": str(_source_receipt()["receipt_ref"]),
    }


def test_cold_start_source_queries_include_host112_scope() -> None:
    expected_scope = 'scope="110_112_120_121_188"'

    assert reconcile.AGENT99_COLD_START_METRIC_SCOPE == "110_112_120_121_188"
    for name, query in reconcile._COLD_START_QUERIES.items():
        if name == "firing_blocking_alerts":
            continue
        assert expected_scope in query
        assert 'scope="110_120_121_188"' not in query


def test_live_cold_start_identity_requires_canonical_dispatch_scope() -> None:
    identity = build_agent99_dispatch_identity(
        project_id="awoooi",
        incident_id="INC-20260715-A1B2C3",
        source_fingerprint="a" * 32,
        route_id="agent99:host_recovery:Recover",
        work_item_id=(
            "agent99-dispatch:awoooi:INC-20260715-A1B2C3:"
            "agent99:host_recovery:Recover"
        ),
    )
    scope = {
        "dispatch_scope": {
            "schema_version": "agent99_dispatch_scope_v1",
            "kind": "host_recovery",
            "suggested_mode": "Recover",
            "target_resource": "cold-start-gate",
            "controlled_apply_requested": True,
        }
    }

    assert reconcile.is_cold_start_same_run_identity(identity, scope) is True
    wrong_target = deepcopy(scope)
    wrong_target["dispatch_scope"]["target_resource"] = "other-host"
    assert reconcile.is_cold_start_same_run_identity(identity, wrong_target) is False
    wrong_work_item = build_agent99_dispatch_identity(
        project_id="awoooi",
        incident_id="INC-20260715-A1B2C3",
        source_fingerprint="a" * 32,
        route_id="agent99:host_recovery:Recover",
        work_item_id="agent99-dispatch:awoooi:INC-20260715-A1B2C3:other",
    )
    assert (
        reconcile.is_cold_start_same_run_identity(wrong_work_item, scope) is False
    )
    assert reconcile.is_cold_start_same_run_identity(identity, {}) is False


def test_cold_start_source_resolution_allows_warning_only_but_requires_fresh_zero(
    monkeypatch,
) -> None:
    values = {
        "monitor_up": 1.0,
        "pass_gates": 95.0,
        "warn_gates": 1.0,
        "blocked_gates": 0.0,
        "last_run_timestamp": 1784020000.0,
        "green_result": 0.0,
        "degraded_result": 1.0,
        "blocked_result": 0.0,
        "check_failed_result": 0.0,
        "firing_blocking_alerts": 0.0,
    }

    def fake_query(_url, query, *, timeout_seconds):  # type: ignore[no-untyped-def]
        del timeout_seconds
        for name, expected_query in reconcile._COLD_START_QUERIES.items():
            if query == expected_query:
                return values[name]
        raise AssertionError(query)

    monkeypatch.setattr(reconcile, "_query_prometheus_scalar", fake_query)

    receipt = reconcile.read_cold_start_source_resolution(
        prometheus_url="http://prometheus.invalid",
        now_epoch=1784020030,
    )

    assert receipt["resolved"] is True
    assert receipt["status"] == "source_resolved"
    assert receipt["values"]["warn_gates"] == 1
    assert receipt["values"]["blocked_gates"] == 0
    assert receipt["receipt_ref"] == (
        "prometheus:cold-start:1784020000:pass-95:"
        "warn-1:blocked-0:result-degraded:blocking-alerts-0"
    )
    assert receipt["stores_raw_evidence"] is False


@pytest.mark.parametrize(
    ("field", "value", "blocker"),
    [
        ("blocked_gates", 1.0, "cold_start_blocked_gates_present"),
        (
            "firing_blocking_alerts",
            1.0,
            "cold_start_blocking_alert_still_firing",
        ),
        ("monitor_up", 0.0, "cold_start_monitor_not_up"),
    ],
)
def test_cold_start_source_resolution_fails_closed(
    monkeypatch,
    field: str,
    value: float,
    blocker: str,
) -> None:
    values = {
        "monitor_up": 1.0,
        "pass_gates": 95.0,
        "warn_gates": 0.0,
        "blocked_gates": 0.0,
        "last_run_timestamp": 1784020000.0,
        "green_result": 1.0,
        "degraded_result": 0.0,
        "blocked_result": 0.0,
        "check_failed_result": 0.0,
        "firing_blocking_alerts": 0.0,
    }
    values[field] = value
    query_to_name = {
        query: name for name, query in reconcile._COLD_START_QUERIES.items()
    }
    monkeypatch.setattr(
        reconcile,
        "_query_prometheus_scalar",
        lambda _url, query, **_kwargs: values[query_to_name[query]],
    )

    receipt = reconcile.read_cold_start_source_resolution(
        prometheus_url="http://prometheus.invalid",
        now_epoch=1784020030,
    )

    assert receipt["resolved"] is False
    assert blocker in receipt["blockers"]
    assert receipt["receipt_ref"] is None


def test_same_run_status_request_preserves_identity_and_no_write_fences(
    monkeypatch,
) -> None:
    seen: dict[str, object] = {}

    class FakeResponse:
        def getcode(self) -> int:
            return 202

        def read(self, _limit: int) -> bytes:
            return json.dumps({
                "ok": True,
                "status": "same_run_status_reconcile_queued",
                "automationRunId": str(IDENTITY.run_id),
                "reconcileOnly": True,
                "controlledApply": False,
                "evidenceReceiptId": reconcile.agent99_same_run_evidence_id(
                    IDENTITY
                ),
            }).encode()

        def __enter__(self):  # type: ignore[no-untyped-def]
            return self

        def __exit__(self, *_args):  # type: ignore[no-untyped-def]
            return None

    def fake_urlopen(request, timeout):  # type: ignore[no-untyped-def]
        seen["url"] = request.full_url
        seen["timeout"] = timeout
        seen["payload"] = json.loads(request.data.decode())
        return FakeResponse()

    monkeypatch.setattr(reconcile.urllib.request, "urlopen", fake_urlopen)

    result = reconcile.request_agent99_same_run_status_reconcile(
        IDENTITY,
        _source_receipt(),
        relay_url="http://agent99.invalid/agent99/sre-alert/",
        relay_token="configured-not-returned",
        timeout_seconds=2,
    )

    assert result["accepted"] is True
    assert result["runtime_write_authorized"] is False
    payload = seen["payload"]
    assert payload["schemaVersion"] == "agent99_same_run_status_reconcile_v1"
    assert payload["mode"] == "Status"
    assert payload["controlledApply"] is False
    assert payload["reconcileOnly"] is True
    assert payload["suppressTelegram"] is True
    assert payload["evidenceReceiptId"] == reconcile.agent99_same_run_evidence_id(
        IDENTITY
    )
    assert payload["identity"] == IDENTITY.public_dict()
    assert payload["identity"]["execution_generation"] == "1"
    assert "configured-not-returned" not in json.dumps(result)


def test_same_run_status_request_rejects_unresolved_source_without_transport(
    monkeypatch,
) -> None:
    monkeypatch.setattr(
        reconcile.urllib.request,
        "urlopen",
        lambda *_args, **_kwargs: pytest.fail("transport must not run"),
    )

    result = reconcile.request_agent99_same_run_status_reconcile(
        IDENTITY,
        {"resolved": False, "receipt_ref": None},
        relay_url="http://agent99.invalid/agent99/sre-alert/",
        relay_token="configured",
    )

    assert result == {
        "status": "source_not_resolved_fail_closed",
        "accepted": False,
        "runtime_write_authorized": False,
    }


def test_live_same_run_request_carries_dispatch_scope_into_identity_gate(
    monkeypatch,
) -> None:
    identity = build_agent99_dispatch_identity(
        project_id="awoooi",
        incident_id="INC-20260715-D4E5F6",
        source_fingerprint="b" * 64,
        route_id="agent99:host_recovery:Recover",
    )
    dispatch_receipt = {
        "dispatch_scope": {
            "schema_version": "agent99_dispatch_scope_v1",
            "kind": "host_recovery",
            "suggested_mode": "Recover",
            "target_resource": "cold-start-gate",
            "controlled_apply_requested": True,
        }
    }
    monkeypatch.setattr(
        reconcile.urllib.request,
        "urlopen",
        lambda *_args, **_kwargs: pytest.fail("transport must not run"),
    )

    eligible = reconcile.request_agent99_same_run_status_reconcile(
        identity,
        {"resolved": False, "receipt_ref": None},
        dispatch_receipt,
        relay_url="http://agent99.invalid/agent99/sre-alert/",
        relay_token="configured",
    )
    missing_scope = reconcile.request_agent99_same_run_status_reconcile(
        identity,
        {"resolved": False, "receipt_ref": None},
        relay_url="http://agent99.invalid/agent99/sre-alert/",
        relay_token="configured",
    )

    assert eligible["status"] == "source_not_resolved_fail_closed"
    assert missing_scope["status"] == "identity_not_eligible_fail_closed"


@pytest.mark.parametrize(
    ("relay_complete", "remove_runtime_field", "invalid_generation_type"),
    [(False, False, False), (True, True, False), (True, False, True)],
)
def test_outcome_readback_rejects_incomplete_relay_contract(
    monkeypatch,
    relay_complete: bool,
    remove_runtime_field: bool,
    invalid_generation_type: bool,
) -> None:
    payload = deepcopy(_valid_outcome())
    payload["sameRunContractFieldsComplete"] = relay_complete
    if remove_runtime_field:
        payload.pop("runtimeWritePerformed")
    if invalid_generation_type:
        payload["executionGeneration"] = 1

    class FakeResponse:
        def getcode(self) -> int:
            return 200

        def read(self, _limit: int) -> bytes:
            return json.dumps({"found": True, **payload}).encode()

        def __enter__(self):  # type: ignore[no-untyped-def]
            return self

        def __exit__(self, *_args):  # type: ignore[no-untyped-def]
            return None

    monkeypatch.setattr(
        bridge.urllib.request,
        "urlopen",
        lambda *_args, **_kwargs: FakeResponse(),
    )

    result = bridge.read_agent99_sre_outcome(
        str(IDENTITY.run_id),
        relay_url="http://agent99.invalid/agent99/sre-alert/",
        relay_token="configured-not-returned",
        timeout_seconds=2,
    )

    assert result is not None
    assert result["sameRunContractFieldsComplete"] is False
    assert job._is_bound_no_write_status_outcome(
        result,
        identity=IDENTITY,
        source_receipt_ref=str(_source_receipt()["receipt_ref"]),
    ) is False


class _Ledger:
    def __init__(self) -> None:
        self.verifier_calls: list[dict[str, object]] = []
        self.external_calls: list[dict[str, object]] = []
        self.list_receipt: dict[str, object] = {
            "status": "dispatch_accepted_verifier_pending",
            "post_verifier_passed": False,
        }

    async def list_reconcilable(self, **_kwargs):  # type: ignore[no-untyped-def]
        return [{
            "identity": IDENTITY,
            "receipt": self.list_receipt,
        }]

    async def record_verifier(self, **kwargs):  # type: ignore[no-untyped-def]
        self.verifier_calls.append(kwargs)
        return {
            "post_verifier_passed": True,
            "receipt_persisted": True,
            "verifier": {"evidence_refs": kwargs["evidence_refs"]},
        }

    async def record_external_no_write_reconciliation(
        self,
        **kwargs,
    ):  # type: ignore[no-untyped-def]
        self.external_calls.append(kwargs)
        return {
            "status": "external_recovery_no_write_reconciled_terminal",
            "receipt_persisted": True,
            "runtime_closure_verified": False,
        }


@pytest.mark.asyncio
async def test_reconciler_requests_status_only_after_production_source_resolves(
    monkeypatch,
) -> None:
    ledger = _Ledger()
    requests: list[tuple[object, object, object]] = []
    monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger)
    monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: None)
    monkeypatch.setattr(
        job,
        "read_cold_start_source_resolution",
        lambda: _source_receipt(),
    )
    monkeypatch.setattr(
        job,
        "request_agent99_same_run_status_reconcile",
        lambda identity, source, receipt: requests.append(
            (identity, source, receipt)
        )
        or {"accepted": True},
    )

    result = await job.reconcile_agent99_controlled_dispatches_once()

    assert result["status_reconcile_requested"] == 1
    assert result["verifier_written"] == 0
    assert requests == [(
        IDENTITY,
        _source_receipt(),
        ledger.list_receipt,
    )]
    assert ledger.verifier_calls == []


@pytest.mark.asyncio
async def test_reconciler_consumes_only_bound_no_write_status_outcome(
    monkeypatch,
) -> None:
    ledger = _Ledger()
    source = _source_receipt()
    outcome = _valid_outcome()
    monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger)
    monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: outcome)
    monkeypatch.setattr(job, "read_cold_start_source_resolution", lambda: source)
    monkeypatch.setattr(
        job,
        "request_agent99_same_run_status_reconcile",
        lambda *_args: pytest.fail("fresh bound outcome must be consumed"),
    )
    finalize = AsyncMock(side_effect=AssertionError("learning must not run"))
    telegram = AsyncMock(side_effect=AssertionError("Telegram must not run"))
    playbook = AsyncMock(side_effect=AssertionError("PlayBook must not run"))
    km = AsyncMock(side_effect=AssertionError("KM must not run"))
    monkeypatch.setattr(job, "_finalize_learning", finalize)
    monkeypatch.setattr(job, "_ensure_telegram_receipt", telegram)
    monkeypatch.setattr(job, "_ensure_playbook_trust", playbook)
    monkeypatch.setattr(job, "_ensure_km_writeback", km)

    result = await job.reconcile_agent99_controlled_dispatches_once()

    assert result["external_no_write_reconciled"] == 1
    assert result["verifier_written"] == 0
    assert len(ledger.external_calls) == 1
    assert (
        ledger.external_calls[0]["evidence_refs"]["source_event_evidence_ref"]
        == source["receipt_ref"]
    )
    assert ledger.verifier_calls == []
    finalize.assert_not_awaited()
    telegram.assert_not_awaited()
    playbook.assert_not_awaited()
    km.assert_not_awaited()


@pytest.mark.asyncio
async def test_reconciler_rejects_status_outcome_with_mismatched_identity(
    monkeypatch,
) -> None:
    ledger = _Ledger()
    source = _source_receipt()
    other = build_agent99_dispatch_identity(
        project_id=IDENTITY.project_id,
        incident_id=IDENTITY.incident_id,
        source_fingerprint=IDENTITY.source_fingerprint,
        route_id=IDENTITY.route_id,
        work_item_id="different-work-item",
    )
    outcome = {
        "identity": other.public_dict(),
        "mode": "Status",
        "controlledApply": False,
        "reconcileOnly": True,
        "runtimeWritePerformed": False,
        "outcome": {
            "schemaVersion": "agent99_outcome_contract_v1",
            "sourceEventResolved": True,
            "sourceEventEvidence": source["receipt_ref"],
            "identity": other.public_dict(),
        },
    }
    requests: list[tuple[object, object, object]] = []
    monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger)
    monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: outcome)
    monkeypatch.setattr(job, "read_cold_start_source_resolution", lambda: source)
    monkeypatch.setattr(
        job,
        "request_agent99_same_run_status_reconcile",
        lambda identity, source, receipt: requests.append(
            (identity, source, receipt)
        )
        or {"accepted": True},
    )

    result = await job.reconcile_agent99_controlled_dispatches_once()

    assert result["status_reconcile_requested"] == 1
    assert result["verifier_written"] == 0
    assert requests == [(IDENTITY, source, ledger.list_receipt)]
    assert ledger.verifier_calls == []


def _mutate_path(payload: dict[str, object], path: str, value: object) -> None:
    target: dict[str, object] = payload
    parts = path.split(".")
    for part in parts[:-1]:
        nested = target[part]
        assert isinstance(nested, dict)
        target = nested
    target[parts[-1]] = value


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("path", "value"),
    [
        ("outcome.state", "degraded"),
        ("outcome.state", "failed"),
        ("outcome.resolved", False),
        ("outcome.transportOk", False),
        ("outcome.verifierPassed", False),
        ("outcome.verifierName", "agent99-status-post-condition-v1"),
        ("reconcileOnly", False),
        ("runtimeWritePerformed", True),
        ("sameRunContractFieldsComplete", False),
        ("traceId", "00-00000000000000000000000000000000-0000000000000000-01"),
        ("workItemId", "different-work-item"),
        ("projectId", "other-project"),
        ("incidentId", "INC-OTHER"),
        ("automationRunId", "00000000-0000-0000-0000-000000000000"),
        ("routeId", "agent99:host_recovery:Status"),
        ("executionGeneration", "2"),
        ("evidenceReceiptId", "same-run-status-wrong"),
        ("outcome.sourceEventEvidence", "prometheus:cold-start:stale"),
    ],
)
async def test_reconciler_rejects_synthetic_bad_same_run_outcomes(
    monkeypatch,
    path: str,
    value: object,
) -> None:
    ledger = _Ledger()
    source = _source_receipt()
    outcome = deepcopy(_valid_outcome())
    _mutate_path(outcome, path, value)
    requests: list[tuple[object, object, object]] = []
    monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger)
    monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: outcome)
    monkeypatch.setattr(job, "read_cold_start_source_resolution", lambda: source)
    monkeypatch.setattr(
        job,
        "request_agent99_same_run_status_reconcile",
        lambda identity, source, receipt: requests.append(
            (identity, source, receipt)
        )
        or {"accepted": True},
    )
    finalize = AsyncMock(side_effect=AssertionError("learning must not run"))
    monkeypatch.setattr(job, "_finalize_learning", finalize)

    result = await job.reconcile_agent99_controlled_dispatches_once()

    assert result["status_reconcile_requested"] == 1
    assert result["external_no_write_reconciled"] == 0
    assert result["verifier_written"] == 0
    assert requests == [(IDENTITY, source, ledger.list_receipt)]
    assert ledger.external_calls == []
    assert ledger.verifier_calls == []
    finalize.assert_not_awaited()


class _ScalarResult:
    def __init__(self, *, value=None, row=None) -> None:
        self.value = value
        self.row = row

    def scalar_one_or_none(self):  # type: ignore[no-untyped-def]
        return self.value

    def one_or_none(self):  # type: ignore[no-untyped-def]
        return self.row


class _Context:
    def __init__(self, db) -> None:  # type: ignore[no-untyped-def]
        self.db = db

    async def __aenter__(self):  # type: ignore[no-untyped-def]
        return self.db

    async def __aexit__(self, *_args):  # type: ignore[no-untyped-def]
        return False


@pytest.mark.asyncio
async def test_external_no_write_terminal_never_records_recover_success_or_closure(
    monkeypatch,
) -> None:
    envelope = build_agent99_dispatch_receipt_envelope(
        identity=IDENTITY,
        dispatch_receipt={
            "accepted": True,
            "inbox_triggered": True,
            "delivery_certainty": "delivered",
        },
        controlled_apply_authorized=True,
    )
    envelope["runtime_execution_attempted"] = True

    class ReconcileDB:
        def __init__(self) -> None:
            self.call = 0
            self.statements: list[str] = []

        async def execute(self, statement):  # type: ignore[no-untyped-def]
            self.call += 1
            self.statements.append(str(statement))
            if self.call == 1:
                return _ScalarResult(
                    row=SimpleNamespace(
                        state="waiting_tool",
                        error_detail=json.dumps(envelope),
                    )
                )
            if self.call == 2:
                return _ScalarResult(value=IDENTITY.run_id)
            return _ScalarResult()

    db = ReconcileDB()
    monkeypatch.setattr(
        ledger_module,
        "get_db_context",
        lambda _project_id: _Context(db),
    )

    result = await PostgresAgent99DispatchLedger().record_external_no_write_reconciliation(
        identity=IDENTITY,
        outcome_receipt=_valid_outcome(),
        source_receipt=_source_receipt(),
        evidence_refs=_evidence_refs(),
    )

    assert result["receipt_persisted"] is True
    assert result["run_terminal_state"] == "cancelled"
    assert result["automation_execution_success"] is False
    assert result["runtime_execution_attempted"] is True
    assert result["same_run_status_runtime_execution_attempted"] is False
    assert result["post_verifier_passed"] is False
    assert result["external_status_verifier_passed"] is True
    assert result["runtime_closure_verified"] is False
    assert result["closure_complete"] is False
    assert result["incident_resolution_authorized"] is False
    assert result["external_recovery_reconciliation"][
        "recover_execution_success"
    ] is False
    assert result["external_recovery_reconciliation"][
        "telegram_authorized"
    ] is False
    sql = "\n".join(db.statements).lower()
    assert "incident_records" not in sql
    assert "alert_operation_logs" not in sql
    assert "playbook" not in sql
    assert "telegram" not in sql


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("path", "value"),
    [
        ("runtimeWritePerformed", True),
        ("outcome.verifierPassed", False),
        ("outcome.state", "degraded"),
    ],
)
async def test_external_no_write_terminal_rejects_invalid_contract_before_db(
    monkeypatch,
    path: str,
    value: object,
) -> None:
    outcome = deepcopy(_valid_outcome())
    _mutate_path(outcome, path, value)
    monkeypatch.setattr(
        ledger_module,
        "get_db_context",
        lambda _project_id: pytest.fail("invalid contract must not touch DB"),
    )

    result = await PostgresAgent99DispatchLedger().record_external_no_write_reconciliation(
        identity=IDENTITY,
        outcome_receipt=outcome,
        source_receipt=_source_receipt(),
        evidence_refs=_evidence_refs(),
    )

    assert result["status"] == "external_recovery_contract_invalid_fail_closed"
    assert result["receipt_persisted"] is False


@pytest.mark.asyncio
@pytest.mark.parametrize("malformed", ["95", float("nan"), float("inf"), -float("inf")])
async def test_external_no_write_terminal_rejects_malformed_source_counts(
    monkeypatch,
    malformed: object,
) -> None:
    source = deepcopy(_source_receipt())
    values = source["values"]
    assert isinstance(values, dict)
    values["pass_gates"] = malformed
    monkeypatch.setattr(
        ledger_module,
        "get_db_context",
        lambda _project_id: pytest.fail("malformed source must not touch DB"),
    )

    result = await PostgresAgent99DispatchLedger().record_external_no_write_reconciliation(
        identity=IDENTITY,
        outcome_receipt=_valid_outcome(),
        source_receipt=source,
        evidence_refs=_evidence_refs(),
    )

    assert result["status"] == "external_recovery_contract_invalid_fail_closed"
    assert result["receipt_persisted"] is False


def test_dispatch_timeout_elapsed_handles_naive_and_aware_datetimes() -> None:
    past = datetime.now(UTC) - timedelta(minutes=1)
    future = datetime.now(UTC) + timedelta(minutes=1)

    assert job._dispatch_timeout_elapsed(past) is True
    assert job._dispatch_timeout_elapsed(past.replace(tzinfo=None)) is True
    assert job._dispatch_timeout_elapsed(future) is False
    assert job._dispatch_timeout_elapsed(None) is False


@pytest.mark.asyncio
async def test_reconciler_terminalizes_expired_run_without_outcome(
    monkeypatch,
) -> None:
    identity = build_agent99_dispatch_identity(
        project_id="awoooi",
        incident_id="INC-BRR-EXPIRED",
        source_fingerprint="backup-restore:" + "b" * 64,
        route_id="agent99:backup_health:Status",
        work_item_id="agent99-dispatch:awoooi:INC-BRR-EXPIRED",
    )

    class TimeoutLedger(_Ledger):
        def __init__(self) -> None:
            super().__init__()
            self.timeout_calls: list[object] = []

        async def list_reconcilable(self, **_kwargs):  # type: ignore[no-untyped-def]
            return [{
                "identity": identity,
                "receipt": self.list_receipt,
                "timeout_at": datetime.now(UTC) - timedelta(minutes=1),
            }]

        async def record_outcome_timeout_no_write_terminal(
            self,
            **kwargs,
        ):  # type: ignore[no-untyped-def]
            self.timeout_calls.append(kwargs["identity"])
            return {
                "status": "outcome_timeout_no_write_terminal",
                "receipt_persisted": True,
                "runtime_closure_verified": False,
            }

    ledger = TimeoutLedger()
    monkeypatch.setattr(job, "get_agent99_dispatch_ledger", lambda: ledger)
    monkeypatch.setattr(job, "read_agent99_sre_outcome", lambda _run: None)

    result = await job.reconcile_agent99_controlled_dispatches_once()

    assert result["outcome_timeout_terminalized"] == 1
    assert result["verifier_written"] == 0
    assert ledger.timeout_calls == [identity]


@pytest.mark.asyncio
async def test_outcome_timeout_terminal_does_not_resolve_incident_or_replay(
    monkeypatch,
) -> None:
    envelope = build_agent99_dispatch_receipt_envelope(
        identity=IDENTITY,
        dispatch_receipt={
            "accepted": True,
            "inbox_triggered": True,
            "queue_accepted": True,
            "dispatch_identity_matched": True,
            "delivery_certainty": "delivered",
        },
        controlled_apply_authorized=True,
    )

    class TimeoutDB:
        def __init__(self) -> None:
            self.call = 0
            self.statements: list[str] = []

        async def execute(self, statement):  # type: ignore[no-untyped-def]
            self.call += 1
            self.statements.append(str(statement))
            if self.call == 1:
                return _ScalarResult(
                    row=SimpleNamespace(
                        state="waiting_tool",
                        error_detail=json.dumps(envelope),
                        timeout_at=(
                            datetime.now(UTC).replace(tzinfo=None)
                            - timedelta(minutes=1)
                        ),
                    )
                )
            if self.call == 2:
                return _ScalarResult(value=IDENTITY.run_id)
            return _ScalarResult()

    db = TimeoutDB()
    monkeypatch.setattr(
        ledger_module,
        "get_db_context",
        lambda _project_id: _Context(db),
    )

    result = (
        await PostgresAgent99DispatchLedger()
        .record_outcome_timeout_no_write_terminal(identity=IDENTITY)
    )

    assert result["receipt_persisted"] is True
    assert result["run_terminal_state"] == "failed"
    assert result["prior_controlled_apply_authorized"] is True
    assert result["controlled_apply_authorized"] is False
    assert result["terminalizer_runtime_write_performed"] is False
    assert result["outcome_runtime_write_status"] == (
        "unknown_without_authenticated_outcome"
    )
    assert result["transport_replayed"] is False
    assert result["automation_execution_success"] is False
    assert result["post_verifier_passed"] is False
    assert result["runtime_closure_verified"] is False
    assert result["incident_resolution_authorized"] is False
    assert result["learning_writeback"]["status"] == (
        "not_applicable_outcome_timeout_no_write"
    )
    sql = "\n".join(db.statements).lower()
    assert "incident_records" not in sql
    assert "alert_operation_logs" not in sql
    assert "telegram" not in sql


@pytest.mark.asyncio
async def test_outcome_timeout_terminal_rejects_unexpired_run(
    monkeypatch,
) -> None:
    envelope = build_agent99_dispatch_receipt_envelope(
        identity=IDENTITY,
        dispatch_receipt={
            "accepted": True,
            "inbox_triggered": True,
            "queue_accepted": True,
            "dispatch_identity_matched": True,
        },
        controlled_apply_authorized=False,
    )

    class FutureTimeoutDB:
        def __init__(self) -> None:
            self.call = 0

        async def execute(self, _statement):  # type: ignore[no-untyped-def]
            self.call += 1
            return _ScalarResult(
                row=SimpleNamespace(
                    state="waiting_tool",
                    error_detail=json.dumps(envelope),
                    timeout_at=(
                        datetime.now(UTC).replace(tzinfo=None)
                        + timedelta(minutes=1)
                    ),
                )
            )

    db = FutureTimeoutDB()
    monkeypatch.setattr(
        ledger_module,
        "get_db_context",
        lambda _project_id: _Context(db),
    )

    result = (
        await PostgresAgent99DispatchLedger()
        .record_outcome_timeout_no_write_terminal(identity=IDENTITY)
    )

    assert result["status"] == "outcome_timeout_not_due_no_write"
    assert result["receipt_persisted"] is False
    assert db.call == 1


def test_windows_relay_and_control_plane_enforce_same_run_no_write_contract() -> None:
    root = Path(__file__).resolve().parents[3]
    relay = (root / "agent99-sre-alert-relay.ps1").read_text(encoding="utf-8")
    control = (root / "agent99-control-plane.ps1").read_text(encoding="utf-8")
    bootstrap = (root / "agent99-bootstrap.ps1").read_text(encoding="utf-8")
    reconcile_source = (
        root / "apps/api/src/services/agent99_same_run_reconcile.py"
    ).read_text(encoding="utf-8")

    assert (
        'AGENT99_COLD_START_RUN_ID = "bf30ca01-1080-5725-b2d1-3e1534dfc811"'
        in reconcile_source
    )

    assert "agent99_same_run_status_reconcile_v1" in relay
    assert "function Test-AgentSameRunDispatchIdentity" in relay
    assert 'route_id" "") -ne "agent99:host_recovery:Recover"' in relay
    assert 'project_id" "") -ne "awoooi"' in relay
    assert 'trace_id" "") -notmatch "^00-[0-9a-f]{32}-[0-9a-f]{16}-01$"' in relay
    assert '$incidentId -match "^INC-[0-9]{8}-[0-9A-F]{6}$"' in relay
    assert '$executionGeneration -match "^[1-3]$"' in relay
    assert '$sourceFingerprint -match "^[0-9a-f]{32,64}$"' in relay
    assert 'function Test-AgentStoredColdStartRecoverEnvelope' in relay
    assert "function Get-AgentExistingRecoverIdentity" in relay
    assert '$AcceptedIgnoredDir = Join-Path $AlertRoot "ignored"' in relay
    assert '$acceptedIgnoredPattern = "ignored-awoooi-agent99-$RunId-*.json"' in relay
    assert '$files = @($files | Sort-Object LastWriteTime -Descending | Select-Object -First 256)' in relay
    assert 'Get-ChildItem -LiteralPath $AcceptedIgnoredDir -Filter $acceptedIgnoredPattern' in relay
    assert 'Get-AgentField $awoooi "agent99DispatchIdentity"' in relay
    assert '$source -eq "awoooi-api-alertmanager"' in relay
    assert '$kind -eq "host_recovery"' in relay
    assert '$service -eq "cold-start-gate"' in relay
    assert "existing_dispatch_identity_not_found" in relay
    assert "existing_dispatch_identity_mismatch" in relay
    assert "function Test-AgentSameRunOutcomeEligible" in relay
    assert 'sameRunContractFieldsComplete = $sameRunContractFieldsComplete' in relay
    assert 'sameRunContractFieldsComplete" $false' in relay
    assert "existing_same_run_status_outcome_invalid_fail_closed" in relay
    assert "[IO.FileMode]::CreateNew" in relay
    assert "agent99_same_run_single_flight_lock_v1" in relay
    assert "relay_auth_not_configured" in relay
    assert 'mode = "Status"' in relay
    assert 'controlledApply = $false' in relay
    assert 'reconcileOnly = $true' in relay
    assert 'suppressTelegram = $true' in relay
    assert 'sourceEventResolutionPolicy = "external_source_receipt_required"' in relay
    assert 'reason = [string](Get-AgentField $reconcile "reason" "")' in relay
    assert 'httpStatus = [int]$reconcile.httpStatus' in relay
    assert '"-ReconcileQueueOnly", "-ReconcileQueueId", $queueId' in relay
    assert '"-SuppressAlerts", "-SuppressReason"' in relay

    assert "[switch]$ReconcileOnly" in control
    assert '[string]$ReconcileEvidenceId = ""' in control
    assert "[switch]$ReconcileQueueOnly" in control
    assert '[string]$ReconcileQueueId = ""' in control
    assert 'param([string]$OnlyCommandId = "")' in control
    assert 'Get-Item -LiteralPath $onlyPath' in control
    assert 'reason = "same_run_reconcile_exact_queue_contract_required"' in control
    assert "$canonicalDigestMatches" in control
    assert "$sameRunReconcileContractExact" in control
    assert "$automationRunIdValid" in control
    assert "$legacyColdStartIdentity" in control
    assert "$liveColdStartIdentity" in control
    assert "$coldStartIdentityEligible" in control
    assert '$dispatchRouteId -eq "agent99:host_recovery:Recover"' in control
    assert '$lockOwner -eq $automationRunId' in control
    assert '$projectId -eq "awoooi"' in control
    assert '$commandSource -eq "agent99-same-run-reconcile"' in control
    assert '$alertSource -eq "awoooi-production-source-verifier"' in control
    assert '$traceId -match "^00-[0-9a-f]{32}-[0-9a-f]{16}-01$"' in control
    assert '$reconcileEvidenceId -eq $expectedReconcileEvidenceId' in control
    assert "function Get-AgentEvidenceAtPath" in control
    assert 'Get-AgentEvidenceAtPath $exactEvidencePath' in control
    assert 'Get-AgentBootState -Persist:(-not $ReconcileOnly)' in control
    assert '$Mode -eq "Status" -and\n  -not $ReconcileOnly -and' in control
    assert '"reconcile_only_no_runtime_write"' in control
    assert '$data.runtimeWritePerformed -is [bool]' in control
    assert '$data.host112Recovery.runtimeWritePerformed -is [bool]' in control
    assert '$data.reconcileEvidenceId -is [string]' in control
    assert '"-ReconcileOnly", "-ReconcileEvidenceId", $reconcileEvidenceId' in control
    assert 'status = "not_written_same_run_reconcile"' in control
    assert 'status = "suppressed_same_run_reconcile"' in control
    assert (
        'status = "suppressed_same_run_reconcile_source_recheck_required"'
        in control
    )
    assert 'Invoke-AgentControlTick -QueueOnly:$ReconcileQueueOnly' in control
    assert "-ReconcileOnly:`$ReconcileOnly" in bootstrap
    assert "-ReconcileEvidenceId `$ReconcileEvidenceId" in bootstrap
    assert "-ReconcileQueueOnly:`$ReconcileQueueOnly" in bootstrap

    reconcile_job = (
        root
        / "apps/api/src/jobs/agent99_controlled_dispatch_reconciler_job.py"
    ).read_text(encoding="utf-8")
    cold_branch = reconcile_job[
        reconcile_job.index("if is_cold_start_same_run_identity(identity, receipt):") :
        reconcile_job.index(
            'if receipt.get("post_verifier_passed") is not True:'
        )
    ]
    assert "record_external_no_write_reconciliation" in cold_branch
    assert "_finalize_learning" not in cold_branch
    assert "_ensure_telegram_receipt" not in cold_branch
    assert "_ensure_playbook_trust" not in cold_branch
    assert "_ensure_km_writeback" not in cold_branch

    queue_function = control.index("function Invoke-AgentQueuedCommands")
    launch_args = control.index("$args = @(", queue_function)
    queue_branch = control[
        control.index("if ($reconcileOnly) {", launch_args) :
        control.index("$exitCode = if", launch_args)
    ]
    assert '"-ReconcileOnly", "-ReconcileEvidenceId"' in queue_branch
    assert "Get-AgentEvidenceAtPath $exactEvidencePath" in queue_branch
    assert "Send-AgentTelegram" not in queue_branch
    assert "Invoke-AgentCompletionCallback" not in queue_branch


def test_windows_relay_allows_only_one_audited_failed_no_write_status_retry() -> None:
    root = Path(__file__).resolve().parents[3]
    relay = (root / "agent99-sre-alert-relay.ps1").read_text(encoding="utf-8")

    field_helper = relay[relay.index("function Get-AgentField") :]
    field_helper = field_helper[: field_helper.index("function Get-AgentEnvValue")]
    assert "$Object -is [Collections.IDictionary]" in field_helper
    assert "$Object.Contains($Name)" in field_helper
    assert "return $Object[$Name]" in field_helper

    retry_start = relay.index(
        "function Test-AgentSameRunFailedNoWriteOutcomeRetryEligible"
    )
    retry_end = relay.index("function Start-AgentSameRunStatusReconcile")
    retry_contract = relay[retry_start:retry_end]

    assert '$SameRunRetryAuditDir = Join-Path $SameRunStateDir "retry-audit"' in relay
    assert '"agent99_same_run_status_retry_v1"' in retry_contract
    assert 'retryOrdinal = 1' in retry_contract
    assert 'status = "retry_reserved"' in retry_contract
    assert 'preservesFailedOutcome = $true' in retry_contract
    assert 'runtimeWriteAuthorized = $false' in retry_contract
    assert '[string](Get-AgentField $Result "mode" "") -eq "Status"' in retry_contract
    assert '[string](Get-AgentField $outcome "state" "") -eq "failed"' in retry_contract
    assert '-not [bool](Get-AgentField $Result "controlledApply" $true)' in retry_contract
    assert '[bool](Get-AgentField $Result "reconcileOnly" $false)' in retry_contract
    assert '-not [bool](Get-AgentField $Result "runtimeWritePerformed" $true)' in retry_contract
    assert '[bool](Get-AgentField $Result "suppressTelegram" $false)' in retry_contract
    assert '-not [bool](Get-AgentField $outcome "resolved" $true)' in retry_contract
    assert '-not [bool](Get-AgentField $outcome "transportOk" $true)' in retry_contract
    assert '-not [bool](Get-AgentField $outcome "verifierPassed" $true)' in retry_contract
    assert '[bool](Get-AgentField $outcome "sourceEventResolved" $false)' in retry_contract
    assert "Test-AgentPublicReceiptRef $SourceEvidence" in retry_contract
    assert "Test-AgentPublicReceiptRef $previousSourceEvidence" in retry_contract
    assert '$previousSourceEvidence.StartsWith("prometheus:cold-start:")' in retry_contract
    assert "Test-AgentSameRunIdentityMatchesExisting $canonicalIdentity $Identity" in retry_contract
    assert '"agent99_same_run_single_flight_lock_v1"' in retry_contract
    assert '[IO.FileMode]::CreateNew' in retry_contract
    assert 'failedOutcomeSha256 = $failedOutcomeSha256' in retry_contract
    assert 'previousLockSha256 = $previousLockSha256' in retry_contract
    assert 'Move-Item -LiteralPath $ProcessedPath -Destination $outcomeAuditPath' in retry_contract
    assert 'Move-Item -LiteralPath $LockPath -Destination $lockAuditPath' in retry_contract
    assert 'same_run_status_retry_already_consumed_fail_closed' in retry_contract
    assert 'same_run_status_retry_archive_failed_fail_closed' in retry_contract
    assert 'mode = "Recover"' not in retry_contract
    assert 'controlledApply = $true' not in retry_contract

    start = relay[relay.index("function Start-AgentSameRunStatusReconcile") :]
    marker_branch = start.index(
        "if (-not $singleRetryPrepared -and "
        "(Test-Path -LiteralPath $retryMarkerPath -PathType Leaf))"
    )
    marker_rejected = start.index(
        'reason = "same_run_status_retry_already_consumed_fail_closed"',
        marker_branch,
    )
    marker_pending = start.index(
        'status = "same_run_status_reconcile_pending"',
        marker_rejected,
    )
    general_pending = start.index(
        "if ($queuePending -or $runningPending -or $lockPending)",
        marker_pending,
    )
    assert marker_branch < marker_rejected < marker_pending < general_pending
    assert "(($queuePending -or $runningPending) -and $lockPending)" in start[
        marker_branch:marker_pending
    ]