All checks were successful
CD Pipeline / workflow-shape (push) Successful in 0s
CD Pipeline / cancel-stale-cd (push) Has been skipped
CD Pipeline / tests (push) Successful in 2m32s
CD Pipeline / build-and-deploy (push) Successful in 8m2s
CD Pipeline / post-deploy-checks (push) Successful in 2m22s
Restore D037 durable incident readback without weakening database failure semantics. Fence Agent99 dispatch and terminal learning to canonical identities, durable leases, verifier receipts, and reconciler-owned checkpoints. Drain backup and restore legacy receipts in bounded cohorts with strict transport, claim, projection, and no-false-green controls. Keep SSH service refusal visible while separating it from broker network-policy reachability.
1032 lines
35 KiB
Python
1032 lines
35 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from src.services.agent99_controlled_dispatch_ledger import (
|
|
build_agent99_dispatch_identity,
|
|
build_agent99_dispatch_receipt_envelope,
|
|
)
|
|
from src.services.agent99_sre_bridge import (
|
|
acquire_agent99_sre_single_flight,
|
|
agent99_sre_single_flight_key,
|
|
bridge_alertmanager_to_agent99,
|
|
build_agent99_sre_alert,
|
|
dispatch_agent99_sre_alert_with_receipt,
|
|
post_agent99_sre_alert,
|
|
release_agent99_sre_single_flight,
|
|
resolve_agent99_alert_kind,
|
|
write_agent99_sre_alert,
|
|
)
|
|
|
|
|
|
def test_provider_freshness_alert_builds_agent99_contract() -> None:
|
|
payload = build_agent99_sre_alert(
|
|
alert_id="alert-20260709233000",
|
|
alertname="provider_freshness_signal",
|
|
severity="warning",
|
|
namespace="awoooi-prod",
|
|
target_resource="provider_freshness_signal",
|
|
message=(
|
|
"raw_signal_normalized controlled_runtime_candidate freshness "
|
|
"last seen missing direct ref verifier readback"
|
|
),
|
|
labels={
|
|
"service": "awoooi",
|
|
"token": "must-not-leak",
|
|
},
|
|
annotations={
|
|
"summary": "Provider freshness evidence is missing",
|
|
},
|
|
fingerprint="fp-provider-freshness",
|
|
alert_category="source_provider_freshness_triage",
|
|
notification_type="raw_signal_normalized",
|
|
source_url="https://awoooi.wooo.work/api/v1/webhooks/alertmanager",
|
|
)
|
|
|
|
assert payload["source"] == "awoooi-api-alertmanager"
|
|
assert payload["kind"] == "provider_freshness_signal"
|
|
assert payload["suggestedMode"] == "ProviderFreshness"
|
|
assert payload["controlledApply"] is False
|
|
assert payload["force"] is True
|
|
assert "provider-freshness" in payload["labels"]
|
|
assert "no-false-green" in payload["labels"]
|
|
assert "token=[redacted]" in payload["labels"]
|
|
assert "must-not-leak" not in json.dumps(payload)
|
|
assert "do not switch provider" in payload["instruction"]
|
|
assert payload["routing"]["routeSource"] == "structured_kind"
|
|
assert payload["routing"]["correlationKey"] == payload["routing"]["groupKey"]
|
|
|
|
|
|
def test_explicit_backup_identity_wins_over_generic_freshness_words() -> None:
|
|
payload = build_agent99_sre_alert(
|
|
alert_id="backup-escrow-missing",
|
|
alertname="BackupCredentialEscrowEvidenceMissing",
|
|
severity="critical",
|
|
namespace="awoooi-prod",
|
|
target_resource="backup_restore",
|
|
message=(
|
|
"backup freshness missing; escrow evidence missing; "
|
|
"readback verifier required"
|
|
),
|
|
labels={
|
|
"event_type": "backup_restore_escrow_signal",
|
|
"lane": "backup_restore_escrow_triage",
|
|
},
|
|
fingerprint="fp-backup-freshness-collision",
|
|
)
|
|
|
|
assert payload["kind"] == "backup_health"
|
|
assert payload["suggestedMode"] == "BackupCheck"
|
|
assert payload["controlledApply"] is False
|
|
assert "backup-readback" in payload["labels"]
|
|
assert "no-false-green" in payload["labels"]
|
|
assert "read-only" in payload["labels"]
|
|
assert "do not run backup/restore" in payload["instruction"]
|
|
assert "remote delete" in payload["instruction"]
|
|
assert payload["resolution"]["policy"] == "mode_verifier_can_resolve"
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("alertname", "target", "message", "labels", "expected_kind", "expected_mode"),
|
|
[
|
|
(
|
|
"HostBackupFailed",
|
|
"backup",
|
|
"snapshot stale",
|
|
{},
|
|
"backup_health",
|
|
"BackupCheck",
|
|
),
|
|
(
|
|
"provider_freshness_signal",
|
|
"provider_freshness_signal",
|
|
"last_seen missing",
|
|
{},
|
|
"provider_freshness_signal",
|
|
"ProviderFreshness",
|
|
),
|
|
(
|
|
"RebootAutoRecoveryActiveBlocker",
|
|
"reboot-auto-recovery-slo",
|
|
"all_required_hosts_not_in_10_minute_reboot_window cpu pressure",
|
|
{"service": "awoooi"},
|
|
"host_recovery",
|
|
"Recover",
|
|
),
|
|
(
|
|
"PostgresqlDown",
|
|
"postgresql",
|
|
"database unavailable deployment awoooi",
|
|
{"service": "awoooi"},
|
|
"database_health",
|
|
"Status",
|
|
),
|
|
(
|
|
"AlertChainBroken",
|
|
"alertmanager",
|
|
"monitoring pipeline down awoooi",
|
|
{"service": "awoooi"},
|
|
"alert_chain_health",
|
|
"Status",
|
|
),
|
|
(
|
|
"HostDown",
|
|
"192.168.0.120",
|
|
"host_down vm_not_running",
|
|
{"host": "192.168.0.120"},
|
|
"host_recovery",
|
|
"Recover",
|
|
),
|
|
(
|
|
"Host112GuestNotReady",
|
|
"host112-guest-recovery",
|
|
"Host 112 console LightDM VMware Tools and Wazuh are not ready",
|
|
{"host": "112", "service": "host112-guest-recovery"},
|
|
"host112_guest_recovery",
|
|
"Recover",
|
|
),
|
|
(
|
|
"KubePodCrashLooping",
|
|
"awoooi-api",
|
|
"k3s pod crashloopbackoff",
|
|
{"service": "awoooi"},
|
|
"awoooi_k3s",
|
|
"AwoooRepair",
|
|
),
|
|
(
|
|
"ColdStartGateBlocked",
|
|
"cold-start-gate",
|
|
"full-stack cold-start recovery gate requires remediation",
|
|
{"service": "awoooi"},
|
|
"host_recovery",
|
|
"Recover",
|
|
),
|
|
(
|
|
"AWOOOICICDFailed",
|
|
"AWOOOI deployment",
|
|
"build-and-deploy deployment failed",
|
|
{"service": "awoooi"},
|
|
"cicd_failure",
|
|
"Status",
|
|
),
|
|
],
|
|
)
|
|
def test_agent99_routing_replay_corpus(
|
|
alertname: str,
|
|
target: str,
|
|
message: str,
|
|
labels: dict[str, str],
|
|
expected_kind: str,
|
|
expected_mode: str,
|
|
) -> None:
|
|
payload = build_agent99_sre_alert(
|
|
alert_id=f"replay-{alertname}",
|
|
alertname=alertname,
|
|
severity="critical",
|
|
namespace="awoooi-prod",
|
|
target_resource=target,
|
|
message=message,
|
|
labels=labels,
|
|
fingerprint=f"fp-{alertname}",
|
|
)
|
|
|
|
assert payload["kind"] == expected_kind
|
|
assert payload["suggestedMode"] == expected_mode
|
|
|
|
|
|
def test_agent99_resolution_policy_uses_mode_verifier_only_for_stateful_alerts() -> None:
|
|
cold_start = build_agent99_sre_alert(
|
|
alert_id="cold-start",
|
|
alertname="ColdStartGateBlocked",
|
|
severity="critical",
|
|
namespace="awoooi-prod",
|
|
target_resource="cold-start-gate",
|
|
message="cold-start gate blocked",
|
|
)
|
|
cicd = build_agent99_sre_alert(
|
|
alert_id="cicd-failure",
|
|
alertname="AWOOOICICDFailed",
|
|
severity="critical",
|
|
namespace="awoooi-prod",
|
|
target_resource="AWOOOI deployment",
|
|
message="build-and-deploy failed",
|
|
)
|
|
|
|
assert cold_start["resolution"]["policy"] == "mode_verifier_can_resolve"
|
|
assert cicd["resolution"]["policy"] == "external_source_receipt_required"
|
|
|
|
|
|
def test_host112_guest_recovery_precedes_wazuh_security_classification() -> None:
|
|
payload = build_agent99_sre_alert(
|
|
alert_id="host112-guest-not-ready",
|
|
alertname="Host112GuestNotReady",
|
|
severity="critical",
|
|
namespace="host112",
|
|
target_resource="host112-guest-recovery",
|
|
message="Host 112 Console VMware Tools Wazuh services are not ready",
|
|
labels={"host": "112", "service": "host112-guest-recovery"},
|
|
)
|
|
|
|
assert payload["kind"] == "host112_guest_recovery"
|
|
assert payload["suggestedMode"] == "Recover"
|
|
assert payload["controlledApply"] is True
|
|
assert payload["resolution"]["policy"] == "mode_verifier_can_resolve"
|
|
assert "do not reboot" in payload["instruction"]
|
|
|
|
|
|
def test_alert_kind_mapping_for_route_and_performance() -> None:
|
|
assert (
|
|
resolve_agent99_alert_kind(
|
|
alertname="PublicRoute502",
|
|
severity="critical",
|
|
namespace="awoooi-prod",
|
|
target_resource="awoooi-web",
|
|
message="HTTP 502",
|
|
)
|
|
== "public_route_502"
|
|
)
|
|
assert (
|
|
resolve_agent99_alert_kind(
|
|
alertname="HostHighCpuLoad",
|
|
severity="warning",
|
|
namespace="node",
|
|
target_resource="192.168.0.110",
|
|
message="load per core is high",
|
|
)
|
|
== "performance_pressure"
|
|
)
|
|
|
|
|
|
def test_write_agent99_sre_alert_to_configured_inbox(tmp_path) -> None:
|
|
payload = build_agent99_sre_alert(
|
|
alert_id="alert-write-smoke",
|
|
alertname="HostHighCpuLoad",
|
|
severity="warning",
|
|
namespace="node",
|
|
target_resource="192.168.0.110",
|
|
message="CPU load per core is high",
|
|
labels={"host": "192.168.0.110"},
|
|
)
|
|
|
|
written = write_agent99_sre_alert(payload, inbox_path=str(tmp_path))
|
|
|
|
assert written is not None
|
|
data = json.loads(written.read_text(encoding="utf-8"))
|
|
assert data["id"] == "awoooi-alertmanager-alert-write-smoke"
|
|
assert data["kind"] == "performance_pressure"
|
|
assert data["suggestedMode"] == "Perf"
|
|
|
|
|
|
def test_post_agent99_sre_alert_to_relay(monkeypatch) -> None:
|
|
payload = build_agent99_sre_alert(
|
|
alert_id="relay-smoke",
|
|
alertname="provider_freshness_signal",
|
|
severity="warning",
|
|
namespace="awoooi-prod",
|
|
target_resource="provider_freshness_signal",
|
|
message="freshness last_seen missing",
|
|
labels={"service": "awoooi", "authorization": "must-not-leak"},
|
|
)
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeResponse:
|
|
def getcode(self) -> int:
|
|
return 202
|
|
|
|
def read(self, _limit: int) -> bytes:
|
|
return b'{"ok":true}'
|
|
|
|
def __enter__(self) -> FakeResponse:
|
|
return self
|
|
|
|
def __exit__(self, *_args: object) -> None:
|
|
return None
|
|
|
|
def fake_urlopen(request, timeout): # type: ignore[no-untyped-def]
|
|
seen["url"] = request.full_url
|
|
seen["timeout"] = timeout
|
|
seen["headers"] = dict(request.header_items())
|
|
seen["payload"] = json.loads(request.data.decode("utf-8"))
|
|
return FakeResponse()
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.urllib.request.urlopen",
|
|
fake_urlopen,
|
|
)
|
|
|
|
receipt = post_agent99_sre_alert(
|
|
payload,
|
|
relay_url="http://192.168.0.99:8787/agent99/sre-alert/",
|
|
relay_token="relay-token",
|
|
timeout_seconds=1.5,
|
|
)
|
|
|
|
assert receipt == {"status": 202, "body": '{"ok":true}'}
|
|
assert seen["url"] == "http://192.168.0.99:8787/agent99/sre-alert/"
|
|
assert seen["timeout"] == 1.5
|
|
assert seen["payload"]["kind"] == "provider_freshness_signal"
|
|
assert seen["payload"]["suggestedMode"] == "ProviderFreshness"
|
|
assert "authorization=[redacted]" in seen["payload"]["labels"]
|
|
assert "must-not-leak" not in json.dumps(seen["payload"])
|
|
assert seen["headers"]["X-agent99-relay-token"] == "relay-token"
|
|
|
|
|
|
def test_dispatch_receipt_keeps_relay_acceptance_without_raw_body(monkeypatch) -> None:
|
|
payload = build_agent99_sre_alert(
|
|
alert_id="receipt-smoke",
|
|
alertname="ColdStartGateBlocked",
|
|
severity="critical",
|
|
namespace="awoooi-prod",
|
|
target_resource="cold-start-gate",
|
|
message="cold-start gate blocked",
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.settings.AGENT99_SRE_ALERT_RELAY_URL",
|
|
"http://192.168.0.99:8787/agent99/sre-alert/",
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.post_agent99_sre_alert",
|
|
lambda _payload: {
|
|
"status": 202,
|
|
"body": '{"ok":true,"inboxTriggered":true,"incomingFile":"private-path"}',
|
|
},
|
|
)
|
|
|
|
receipt = dispatch_agent99_sre_alert_with_receipt(payload)
|
|
|
|
assert receipt == {
|
|
"schema_version": "agent99_sre_dispatch_receipt_v1",
|
|
"status": "accepted_inbox_triggered",
|
|
"transport": "relay",
|
|
"alert_id": "awoooi-alertmanager-receipt-smoke",
|
|
"kind": "host_recovery",
|
|
"http_status": 202,
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
"delivery_certainty": "delivered",
|
|
"stores_raw_response": False,
|
|
}
|
|
assert "private-path" not in str(receipt)
|
|
|
|
|
|
def test_agent99_single_flight_key_does_not_expose_fingerprint() -> None:
|
|
key = agent99_sre_single_flight_key("private/source/fingerprint")
|
|
|
|
assert key.startswith("agent99:sre_dispatch:single_flight:")
|
|
assert "private" not in key
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent99_bridge_suppresses_duplicate_single_flight(monkeypatch) -> None:
|
|
async def duplicate(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": False, "reason": "single_flight_duplicate"}
|
|
|
|
dispatched: list[dict[str, object]] = []
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
duplicate,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda payload: dispatched.append(payload)
|
|
or {
|
|
"status": "accepted_inbox_triggered",
|
|
"transport": "relay",
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
},
|
|
)
|
|
|
|
result = await bridge_alertmanager_to_agent99(
|
|
alert_id="duplicate-alert",
|
|
alertname="UnknownObservation",
|
|
severity="critical",
|
|
namespace="node",
|
|
target_resource="192.168.0.120",
|
|
message="observation only",
|
|
fingerprint="same-fingerprint",
|
|
)
|
|
|
|
assert result["status"] == "suppressed"
|
|
assert result["reason"] == "single_flight_duplicate"
|
|
assert dispatched == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent99_bridge_dispatches_single_flight_winner(monkeypatch) -> None:
|
|
async def winner(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": True, "reason": "acquired"}
|
|
|
|
dispatched: list[dict[str, object]] = []
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
winner,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda payload: dispatched.append(payload)
|
|
or {
|
|
"status": "accepted_inbox_triggered",
|
|
"transport": "relay",
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
},
|
|
)
|
|
|
|
result = await bridge_alertmanager_to_agent99(
|
|
alert_id="winner-alert",
|
|
alertname="UnknownObservation",
|
|
severity="critical",
|
|
namespace="node",
|
|
target_resource="192.168.0.120",
|
|
message="observation only",
|
|
fingerprint="winner-fingerprint",
|
|
)
|
|
|
|
assert result["status"] == "dispatched"
|
|
assert result["dispatch"] == "relay"
|
|
assert result["dispatchReceipt"]["accepted"] is True
|
|
assert result["dispatchReceipt"]["inbox_triggered"] is True
|
|
assert result["suggestedMode"] == "Status"
|
|
assert len(dispatched) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent99_bridge_releases_lock_when_receipt_is_not_accepted(
|
|
monkeypatch,
|
|
) -> None:
|
|
async def winner(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": True, "reason": "acquired"}
|
|
|
|
released: list[tuple[str, str]] = []
|
|
|
|
async def release(fingerprint: str, alert_id: str) -> None:
|
|
released.append((fingerprint, alert_id))
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
winner,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.release_agent99_sre_single_flight",
|
|
release,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda _payload: {
|
|
"status": "relay_response_not_accepted",
|
|
"transport": "relay",
|
|
"accepted": False,
|
|
"inbox_triggered": False,
|
|
"delivery_certainty": "not_delivered",
|
|
},
|
|
)
|
|
|
|
result = await bridge_alertmanager_to_agent99(
|
|
alert_id="receipt-rejected",
|
|
alertname="UnknownObservation",
|
|
severity="critical",
|
|
namespace="node",
|
|
target_resource="192.168.0.120",
|
|
message="observation only",
|
|
fingerprint="receipt-rejected-fingerprint",
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert result["reason"] == "relay_response_not_accepted"
|
|
assert result["dispatchReceipt"]["accepted"] is False
|
|
assert released == [("receipt-rejected-fingerprint", "receipt-rejected")]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent99_bridge_releases_lock_when_transport_fails(monkeypatch) -> None:
|
|
async def winner(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": True, "reason": "acquired"}
|
|
|
|
released: list[tuple[str, str]] = []
|
|
|
|
async def release(fingerprint: str, alert_id: str) -> None:
|
|
released.append((fingerprint, alert_id))
|
|
|
|
def fail_transport(_payload: dict[str, object]) -> dict[str, object]:
|
|
raise RuntimeError("relay unavailable")
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
winner,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.release_agent99_sre_single_flight",
|
|
release,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
fail_transport,
|
|
)
|
|
|
|
result = await bridge_alertmanager_to_agent99(
|
|
alert_id="transport-failure",
|
|
alertname="UnknownObservation",
|
|
severity="critical",
|
|
namespace="node",
|
|
target_resource="192.168.0.120",
|
|
message="observation only",
|
|
fingerprint="transport-fingerprint",
|
|
)
|
|
|
|
assert result["status"] == "failed"
|
|
assert result["reason"] == "RuntimeError"
|
|
assert result["dispatchPerformed"] is False
|
|
assert result["runtimeClosureVerified"] is False
|
|
assert released == [("transport-fingerprint", "transport-failure")]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent99_single_flight_uses_redis_nx(monkeypatch) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeRedis:
|
|
async def set(self, key, value, *, ex, nx): # type: ignore[no-untyped-def]
|
|
seen.update({"key": key, "value": value, "ex": ex, "nx": nx})
|
|
return True
|
|
|
|
monkeypatch.setattr("src.core.redis_client.get_redis", lambda: FakeRedis())
|
|
|
|
from src.services.agent99_sre_bridge import try_acquire_agent99_sre_single_flight
|
|
|
|
acquired = await try_acquire_agent99_sre_single_flight(
|
|
"source-fingerprint", "alert-1"
|
|
)
|
|
|
|
assert acquired is True
|
|
assert str(seen["key"]).startswith("agent99:sre_dispatch:single_flight:")
|
|
assert seen["value"] == "alert-1"
|
|
assert seen["ex"] == 600
|
|
assert seen["nx"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_agent99_single_flight_release_is_owner_checked(monkeypatch) -> None:
|
|
seen: dict[str, object] = {}
|
|
|
|
class FakeRedis:
|
|
async def eval(self, script, key_count, key, owner): # type: ignore[no-untyped-def]
|
|
seen.update(
|
|
{"script": script, "key_count": key_count, "key": key, "owner": owner}
|
|
)
|
|
return 1
|
|
|
|
monkeypatch.setattr("src.core.redis_client.get_redis", lambda: FakeRedis())
|
|
|
|
await release_agent99_sre_single_flight("source-fingerprint", "alert-1")
|
|
|
|
assert "redis.call('get'" in str(seen["script"])
|
|
assert seen["key_count"] == 1
|
|
assert seen["owner"] == "alert-1"
|
|
|
|
|
|
class _FakeControlledDispatchLedger:
|
|
def __init__(self) -> None:
|
|
self.reserved = False
|
|
self.receipt: dict[str, object] | None = None
|
|
self.complete_calls = 0
|
|
self.defer_calls = 0
|
|
|
|
async def reserve(self, *, identity, payload): # type: ignore[no-untyped-def]
|
|
if self.reserved:
|
|
return {
|
|
"status": "idempotent_completed",
|
|
"claimed": False,
|
|
"identity": identity.public_dict(),
|
|
"receipt": self.receipt,
|
|
}
|
|
self.reserved = True
|
|
return {
|
|
"status": "reserved",
|
|
"claimed": True,
|
|
"identity": identity.public_dict(),
|
|
"receipt": None,
|
|
"claim_token": "claim-attempt-1",
|
|
}
|
|
|
|
async def complete( # type: ignore[no-untyped-def]
|
|
self, *, identity, claim_token, dispatch_receipt, **kwargs
|
|
) -> dict[str, object]:
|
|
self.complete_calls += 1
|
|
self.receipt = {
|
|
**build_agent99_dispatch_receipt_envelope(
|
|
identity=identity,
|
|
dispatch_receipt=dispatch_receipt,
|
|
),
|
|
"receipt_persisted": True,
|
|
}
|
|
return self.receipt
|
|
|
|
async def mark_delivery_attempt_started( # type: ignore[no-untyped-def]
|
|
self, *, identity, claim_token
|
|
) -> dict[str, object]:
|
|
self.receipt = {
|
|
"status": "dispatch_delivery_unknown_reconcile_only",
|
|
"identity": identity.public_dict(),
|
|
"retry_policy": {"reconcile_only": True, "safe_to_retry": False},
|
|
"receipt_persisted": True,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return self.receipt
|
|
|
|
async def defer_claim_retryable( # type: ignore[no-untyped-def]
|
|
self, *, identity, claim_token, reason, retry_after_seconds
|
|
) -> dict[str, object]:
|
|
self.defer_calls += 1
|
|
self.receipt = {
|
|
"schema_version": "agent99_controlled_dispatch_receipt_v1",
|
|
"status": "dispatch_lock_retryable",
|
|
"identity": identity.public_dict(),
|
|
"dispatch_accepted": False,
|
|
"inbox_triggered": False,
|
|
"dispatch_promoted": False,
|
|
"controlled_apply_authorized": False,
|
|
"retry_policy": {
|
|
"safe_to_retry": True,
|
|
"retry_same_generation": True,
|
|
"requires_generation_advance": False,
|
|
"retry_after_seconds": retry_after_seconds,
|
|
"reason": reason,
|
|
},
|
|
"receipt_persisted": True,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return self.receipt
|
|
|
|
|
|
def test_agent99_dispatch_identity_is_stable_across_projections() -> None:
|
|
alertmanager = build_agent99_dispatch_identity(
|
|
project_id="awoooi",
|
|
incident_id="INC-20260711-11C751",
|
|
source_fingerprint="source-fingerprint",
|
|
route_id="agent99_cold_start_recovery",
|
|
execution_generation="1",
|
|
)
|
|
telegram = build_agent99_dispatch_identity(
|
|
project_id="awoooi",
|
|
incident_id="INC-20260711-11C751",
|
|
source_fingerprint="source-fingerprint",
|
|
route_id="agent99_cold_start_recovery",
|
|
execution_generation="1",
|
|
)
|
|
|
|
assert telegram.run_id == alertmanager.run_id
|
|
assert telegram.trace_id == alertmanager.trace_id
|
|
assert telegram.idempotency_key == alertmanager.idempotency_key
|
|
assert telegram.work_item_id == alertmanager.work_item_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutating_dispatch_is_deduplicated_by_postgres_identity(
|
|
monkeypatch,
|
|
) -> None:
|
|
ledger = _FakeControlledDispatchLedger()
|
|
dispatched: list[dict[str, object]] = []
|
|
|
|
async def acquired(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": True, "reason": "acquired"}
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.get_agent99_dispatch_ledger",
|
|
lambda: ledger,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
acquired,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda payload: dispatched.append(payload)
|
|
or {
|
|
"status": "accepted_inbox_triggered",
|
|
"transport": "relay",
|
|
"alert_id": str(payload["id"]),
|
|
"kind": str(payload["kind"]),
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
},
|
|
)
|
|
common = {
|
|
"alertname": "ColdStartGateBlocked",
|
|
"severity": "critical",
|
|
"namespace": "host_recovery",
|
|
"target_resource": "cold-start-gate",
|
|
"message": "cold-start gate blocked",
|
|
"fingerprint": "source-fingerprint",
|
|
"project_id": "awoooi",
|
|
"incident_id": "INC-20260711-11C751",
|
|
"route_id": "agent99_cold_start_recovery",
|
|
}
|
|
|
|
first = await bridge_alertmanager_to_agent99(alert_id="alertmanager-1", **common)
|
|
second = await bridge_alertmanager_to_agent99(alert_id="telegram-card-2", **common)
|
|
|
|
assert first["status"] == "dispatched"
|
|
assert second["status"] == "deduplicated"
|
|
assert len(dispatched) == 1
|
|
assert dispatched[0]["force"] is False
|
|
assert first["identity"] == second["identity"]
|
|
assert first["correlatedReceipt"]["runtime_execution_authorized"] is False
|
|
assert first["correlatedReceipt"]["runtime_closure_verified"] is False
|
|
assert first["correlatedReceipt"]["verifier"]["status"] == "pending"
|
|
assert first["correlatedReceipt"]["learning_writeback"]["status"] == (
|
|
"pending_verifier"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutating_dispatch_redis_failure_performs_zero_transport(
|
|
monkeypatch,
|
|
) -> None:
|
|
ledger = _FakeControlledDispatchLedger()
|
|
dispatched: list[dict[str, object]] = []
|
|
|
|
async def redis_unavailable(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": False, "reason": "redis_unavailable_fail_closed"}
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.get_agent99_dispatch_ledger",
|
|
lambda: ledger,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
redis_unavailable,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda payload: dispatched.append(payload),
|
|
)
|
|
|
|
result = await bridge_alertmanager_to_agent99(
|
|
alert_id="cold-start-redis-failure",
|
|
alertname="ColdStartGateBlocked",
|
|
severity="critical",
|
|
namespace="host_recovery",
|
|
target_resource="cold-start-gate",
|
|
message="cold-start gate blocked",
|
|
fingerprint="redis-failure-fingerprint",
|
|
incident_id="INC-20260711-11C751",
|
|
route_id="agent99_cold_start_recovery",
|
|
)
|
|
|
|
assert result["status"] == "retryable"
|
|
assert result["reason"] == "redis_unavailable_fail_closed"
|
|
assert result["dispatchPerformed"] is False
|
|
assert dispatched == []
|
|
assert ledger.complete_calls == 0
|
|
assert ledger.defer_calls == 1
|
|
assert result["correlatedReceipt"]["receipt_persisted"] is True
|
|
assert result["correlatedReceipt"]["runtime_closure_verified"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mutating_single_flight_redis_exception_is_fail_closed(
|
|
monkeypatch,
|
|
) -> None:
|
|
monkeypatch.setattr(
|
|
"src.core.redis_client.get_redis",
|
|
lambda: (_ for _ in ()).throw(RuntimeError("redis down")),
|
|
)
|
|
|
|
decision = await acquire_agent99_sre_single_flight(
|
|
"same-controlled-idempotency-key",
|
|
"same-run-owner",
|
|
fail_closed=True,
|
|
)
|
|
|
|
assert decision == {
|
|
"acquired": False,
|
|
"reason": "redis_unavailable_fail_closed",
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backupcheck_read_only_dispatch_has_durable_same_run_receipt(
|
|
monkeypatch,
|
|
) -> None:
|
|
ledger = _FakeControlledDispatchLedger()
|
|
dispatched: list[dict[str, object]] = []
|
|
|
|
async def acquired(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": True, "reason": "acquired"}
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.get_agent99_dispatch_ledger",
|
|
lambda: ledger,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
acquired,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda payload: dispatched.append(payload)
|
|
or {
|
|
"schema_version": "agent99_sre_dispatch_receipt_v1",
|
|
"status": "accepted_inbox_triggered",
|
|
"transport": "relay",
|
|
"alert_id": str(payload["id"]),
|
|
"kind": str(payload["kind"]),
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
"delivery_certainty": "delivered",
|
|
},
|
|
)
|
|
|
|
result = await bridge_alertmanager_to_agent99(
|
|
alert_id="backup-readback",
|
|
alertname="BackupCredentialEscrowEvidenceMissing",
|
|
severity="warning",
|
|
namespace="awoooi-prod",
|
|
target_resource="backup_restore",
|
|
message="freshness escrow restore evidence required",
|
|
labels={"event_type": "backup_restore_escrow_signal"},
|
|
fingerprint="backup-fingerprint",
|
|
incident_id="INC-20260711-BACKUP",
|
|
route_id="agent99:backup_health:BackupCheck",
|
|
)
|
|
|
|
assert result["status"] == "dispatched"
|
|
assert result["identity"]["run_id"]
|
|
assert result["identity"]["trace_id"]
|
|
assert result["identity"]["work_item_id"]
|
|
assert result["correlatedReceipt"]["receipt_persisted"] is True
|
|
assert dispatched[0]["suggestedMode"] == "BackupCheck"
|
|
assert dispatched[0]["controlledApply"] is False
|
|
assert result["correlatedReceipt"]["controlled_apply_authorized"] is False
|
|
|
|
|
|
class _DeliveryUnknownLedger:
|
|
def __init__(self) -> None:
|
|
self.reserved = False
|
|
self.receipt: dict[str, object] | None = None
|
|
|
|
async def reserve(self, *, identity, payload): # type: ignore[no-untyped-def]
|
|
if self.reserved:
|
|
return {
|
|
"status": "idempotent_running",
|
|
"claimed": False,
|
|
"receipt": self.receipt,
|
|
}
|
|
self.reserved = True
|
|
self.receipt = {
|
|
"schema_version": "agent99_controlled_dispatch_receipt_v1",
|
|
"status": "dispatch_reserved",
|
|
"identity": identity.public_dict(),
|
|
"outbox": {"status": "pending_delivery"},
|
|
}
|
|
return {
|
|
"status": "reserved",
|
|
"claimed": True,
|
|
"claim_token": "unknown-delivery-claim",
|
|
"receipt": self.receipt,
|
|
}
|
|
|
|
async def complete(self, **_kwargs): # type: ignore[no-untyped-def]
|
|
return {
|
|
**(self.receipt or {}),
|
|
"status": "receipt_persistence_failed_no_runtime_closure",
|
|
"receipt_persisted": False,
|
|
}
|
|
|
|
async def mark_delivery_attempt_started( # type: ignore[no-untyped-def]
|
|
self, *, identity, claim_token
|
|
) -> dict[str, object]:
|
|
self.receipt = {
|
|
"schema_version": "agent99_controlled_dispatch_receipt_v1",
|
|
"status": "dispatch_delivery_unknown_reconcile_only",
|
|
"identity": identity.public_dict(),
|
|
"retry_policy": {"safe_to_retry": False, "reconcile_only": True},
|
|
"receipt_persisted": True,
|
|
"runtime_closure_verified": False,
|
|
}
|
|
return self.receipt
|
|
|
|
async def defer_claim_retryable(self, **_kwargs): # type: ignore[no-untyped-def]
|
|
raise AssertionError("lock is acquired in this test")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_accepted_transport_with_receipt_write_failure_is_not_replayed(
|
|
monkeypatch,
|
|
) -> None:
|
|
ledger = _DeliveryUnknownLedger()
|
|
dispatched: list[dict[str, object]] = []
|
|
|
|
async def acquired(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": True, "reason": "acquired"}
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.get_agent99_dispatch_ledger",
|
|
lambda: ledger,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
acquired,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda payload: dispatched.append(payload)
|
|
or {
|
|
"status": "accepted_inbox_triggered",
|
|
"transport": "relay",
|
|
"alert_id": str(payload["id"]),
|
|
"kind": str(payload["kind"]),
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
"delivery_certainty": "delivered",
|
|
},
|
|
)
|
|
common = {
|
|
"alertname": "ColdStartGateBlocked",
|
|
"severity": "critical",
|
|
"namespace": "host_recovery",
|
|
"target_resource": "cold-start-gate",
|
|
"message": "cold-start gate blocked",
|
|
"fingerprint": "unknown-delivery-fingerprint",
|
|
"incident_id": "INC-20260711-11C751",
|
|
"route_id": "agent99:host_recovery:Recover",
|
|
}
|
|
|
|
first = await bridge_alertmanager_to_agent99(alert_id="first", **common)
|
|
second = await bridge_alertmanager_to_agent99(alert_id="second", **common)
|
|
|
|
assert first["reason"] == "dispatch_receipt_persistence_failed"
|
|
assert second["dispatchPerformed"] is False
|
|
assert len(dispatched) == 1
|
|
|
|
|
|
class _GenerationAdvanceLedger(_FakeControlledDispatchLedger):
|
|
async def reserve(self, *, identity, payload): # type: ignore[no-untyped-def]
|
|
if identity.execution_generation == "1":
|
|
return {
|
|
"status": "idempotent_failed",
|
|
"claimed": False,
|
|
"receipt": {
|
|
"status": "dispatch_not_delivered_retry_safe",
|
|
"identity": identity.public_dict(),
|
|
"retry_policy": {
|
|
"safe_to_retry": True,
|
|
"requires_generation_advance": True,
|
|
"max_generation": 3,
|
|
},
|
|
},
|
|
}
|
|
return await super().reserve(identity=identity, payload=payload)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_safe_not_delivered_failure_advances_generation_before_dispatch(
|
|
monkeypatch,
|
|
) -> None:
|
|
ledger = _GenerationAdvanceLedger()
|
|
dispatched: list[dict[str, object]] = []
|
|
|
|
async def acquired(*_args, **_kwargs) -> dict[str, object]:
|
|
return {"acquired": True, "reason": "acquired"}
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.get_agent99_dispatch_ledger",
|
|
lambda: ledger,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.acquire_agent99_sre_single_flight",
|
|
acquired,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.agent99_sre_bridge.dispatch_agent99_sre_alert_with_receipt",
|
|
lambda payload: dispatched.append(payload)
|
|
or {
|
|
"status": "accepted_inbox_triggered",
|
|
"transport": "relay",
|
|
"alert_id": str(payload["id"]),
|
|
"kind": str(payload["kind"]),
|
|
"accepted": True,
|
|
"inbox_triggered": True,
|
|
"delivery_certainty": "delivered",
|
|
},
|
|
)
|
|
|
|
result = await bridge_alertmanager_to_agent99(
|
|
alert_id="retry-generation",
|
|
alertname="ColdStartGateBlocked",
|
|
severity="critical",
|
|
namespace="host_recovery",
|
|
target_resource="cold-start-gate",
|
|
message="cold-start gate blocked",
|
|
fingerprint="retry-generation-fingerprint",
|
|
incident_id="INC-20260711-11C751",
|
|
route_id="agent99:host_recovery:Recover",
|
|
)
|
|
|
|
assert result["status"] == "dispatched"
|
|
assert result["identity"]["execution_generation"] == "2"
|
|
assert len(dispatched) == 1
|