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 ]