From da59c89e1c866b2792d9cf45cb0067e4a2b9d320 Mon Sep 17 00:00:00 2001 From: Your Name Date: Sun, 19 Jul 2026 12:08:49 +0800 Subject: [PATCH] fix(sre): require Sentry recurrence closure --- apps/api/src/api/v1/webhooks.py | 5 + .../services/awooop_ansible_audit_service.py | 2 +- .../awooop_ansible_check_mode_service.py | 324 +++++++++++++++++- .../services/awooop_ansible_post_verifier.py | 29 ++ .../tests/test_ansible_verified_closure.py | 278 +++++++++++++++ ...3s_controlled_automation_work_items_api.py | 7 +- .../api/tests/test_sre_typed_domain_router.py | 84 ++++- ...ent-conversation-commitments.snapshot.json | 8 +- 8 files changed, 725 insertions(+), 12 deletions(-) diff --git a/apps/api/src/api/v1/webhooks.py b/apps/api/src/api/v1/webhooks.py index ff492ed91..6158f555a 100644 --- a/apps/api/src/api/v1/webhooks.py +++ b/apps/api/src/api/v1/webhooks.py @@ -27,6 +27,7 @@ import hashlib import hmac import time import uuid +from datetime import UTC, datetime from typing import Any, Literal from fastapi import APIRouter, BackgroundTasks, Header, HTTPException, Request, status @@ -578,6 +579,10 @@ async def _try_auto_repair_background( ), "source_recurrence_source": "current_alert_webhook", "source_recurrence_anchor": "webhook_payload", + "source_received_at": datetime.now(UTC).isoformat(), + "source_canonical_asset_id": typed_target_route.get( + "canonical_asset_id" + ), }, project_id=str(getattr(incident, "project_id", None) or "awoooi"), ) diff --git a/apps/api/src/services/awooop_ansible_audit_service.py b/apps/api/src/services/awooop_ansible_audit_service.py index 6b3c43a15..beefa60bc 100644 --- a/apps/api/src/services/awooop_ansible_audit_service.py +++ b/apps/api/src/services/awooop_ansible_audit_service.py @@ -878,7 +878,7 @@ _CATALOG: tuple[dict[str, Any], ...] = ( "approval_required": False, "risk_level": "medium", "canonical_asset_id": "service:sentry", - "catalog_revision": "2026-07-15-sentry-exact-container-v1", + "catalog_revision": "2026-07-19-sentry-exact-container-v2", }, { "catalog_id": "ansible:110-devops-full-convergence", diff --git a/apps/api/src/services/awooop_ansible_check_mode_service.py b/apps/api/src/services/awooop_ansible_check_mode_service.py index 8251ed67a..fa932aadf 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -8,6 +8,7 @@ catalog rows performs controlled apply after the dry-run passes. from __future__ import annotations import asyncio +import hashlib import json import os import re @@ -199,6 +200,16 @@ _WAZUH_ALLOWED_RECURRENCE_SOURCES = frozenset( ) _WAZUH_SCHEDULED_RECURRENCE_SOURCE = "iwooos_asset_sensor_receipt" _WAZUH_RECURRENCE_MAX_AGE_SECONDS = 15 * 60 +_SENTRY_PROFILING_RECOVERY_CATALOG_ID = ( + "ansible:110-sentry-profiling-consumer-recovery" +) +_SENTRY_CANONICAL_ASSET_ID = "service:sentry" +_SENTRY_PROFILING_CONTAINER = ( + "sentry-self-hosted-snuba-profiling-functions-consumer-1" +) +_SENTRY_SOURCE_RECURRENCE_FENCE_SCHEMA_VERSION = ( + "sentry_source_recurrence_fence_receipt_v1" +) FORCED_COMMAND_BLOCKER = "ansible_repair_ssh_forced_command_denies_ansible_bootstrap" REPAIR_FORCED_COMMAND_KEY_PATH = Path("/etc/repair-ssh/id_ed25519") REPAIR_FORCED_COMMAND_KNOWN_HOSTS_PATH = Path("/etc/repair-known-hosts/known_hosts") @@ -1827,6 +1838,297 @@ def _canonical_risk_severity(risk_level: str) -> str: }.get(str(risk_level or "").lower(), "P2") +def _sentry_source_recurrence_identity( + claim: AnsibleCheckModeClaim, +) -> dict[str, str] | None: + recurrence = claim.input_payload.get("source_recurrence") + typed_target_route = claim.input_payload.get("typed_target_route") + if not isinstance(recurrence, Mapping) or not isinstance( + typed_target_route, + Mapping, + ): + return None + fingerprint = str(recurrence.get("fingerprint") or "").strip().lower() + occurrence_id = str(recurrence.get("occurrence_id") or "").strip() + observed_at_raw = str(recurrence.get("observed_at") or "").strip() + try: + observed_at = datetime.fromisoformat( + observed_at_raw.replace("Z", "+00:00") + ) + except ValueError: + return None + if ( + recurrence.get("schema_version") + != "alert_source_recurrence_receipt_v1" + or recurrence.get("verified") is not True + or len(fingerprint) != 32 + or any(character not in "0123456789abcdef" for character in fingerprint) + or not occurrence_id.startswith("alert-") + or observed_at.tzinfo is None + or observed_at.astimezone(UTC) > datetime.now(UTC) + timedelta(seconds=60) + or recurrence.get("identity_anchor") != "webhook_payload" + or recurrence.get("source") != "current_alert_webhook" + or recurrence.get("canonical_asset_id") != _SENTRY_CANONICAL_ASSET_ID + or typed_target_route.get("canonical_asset_id") + != _SENTRY_CANONICAL_ASSET_ID + or claim.inventory_hosts != ("host_110",) + ): + return None + return { + "fingerprint": fingerprint, + "occurrence_id": occurrence_id, + "observed_at": observed_at.astimezone(UTC).isoformat(), + } + + +async def _verify_sentry_source_recurrence_fence( + claim: AnsibleCheckModeClaim, + *, + apply_op_id: str, + host_postconditions_passed: bool, +) -> dict[str, Any]: + """Prove the exact source occurrence did not recur after controlled apply.""" + + identity = _sentry_source_recurrence_identity(claim) + base: dict[str, Any] = { + "schema_version": _SENTRY_SOURCE_RECURRENCE_FENCE_SCHEMA_VERSION, + "condition_id": "awoooi_sentry_source_recurrence_after_apply", + "condition_type": "source_recurrence", + "independent_source": "awooop_conversation_event_readback", + "catalog_id": claim.catalog_id, + "apply_op_id": apply_op_id, + "passed": False, + "verified": False, + "writes_on_verify": False, + "raw_output_stored": False, + "direct_webhook_recurrence_covered": True, + "source_fingerprint_sha256": ( + hashlib.sha256(identity["fingerprint"].encode()).hexdigest() + if identity is not None + else None + ), + "source_occurrence_id_sha256": ( + hashlib.sha256(identity["occurrence_id"].encode()).hexdigest() + if identity is not None + else None + ), + } + if host_postconditions_passed is not True: + return { + **base, + "active_blockers": [ + "host_postconditions_not_verified_before_source_fence" + ], + } + if identity is None: + return { + **base, + "active_blockers": ["source_recurrence_identity_invalid"], + } + + project_id = str(claim.input_payload.get("project_id") or "").strip() + if not project_id: + return { + **base, + "active_blockers": ["source_recurrence_project_missing"], + } + try: + async with get_db_context(project_id) as db: + await db.execute(text("SET LOCAL statement_timeout = '5000ms'")) + result = await db.execute( + text(""" + WITH apply_boundary AS MATERIALIZED ( + SELECT created_at + FROM automation_operation_log + WHERE op_id = CAST(:apply_op_id AS uuid) + AND operation_type = 'ansible_apply_executed' + LIMIT 1 + ), + source_anchor AS MATERIALIZED ( + SELECT 1 + FROM awooop_conversation_event event_row + WHERE event_row.project_id = :project_id + AND event_row.channel_type = 'internal' + AND event_row.source_envelope ->> 'provider' + = 'alertmanager' + AND event_row.source_envelope ->> 'stage' = 'received' + AND event_row.source_envelope #>> + '{log_correlation,alertname}' + = 'DockerContainerUnhealthy' + AND event_row.source_envelope #>> + '{log_correlation,target_resource}' + = :target_resource + AND coalesce( + event_row.source_envelope #> + '{source_refs,event_ids}', + '[]'::jsonb + ) ? :occurrence_id + AND coalesce( + event_row.source_envelope #> + '{source_refs,fingerprints}', + '[]'::jsonb + ) ? :fingerprint + LIMIT 1 + ), + newer_recurrence AS MATERIALIZED ( + SELECT 1 + FROM awooop_conversation_event event_row + CROSS JOIN apply_boundary + WHERE event_row.project_id = :project_id + AND event_row.channel_type = 'internal' + AND event_row.received_at > apply_boundary.created_at + AND event_row.source_envelope ->> 'provider' + = 'alertmanager' + AND event_row.source_envelope ->> 'stage' = 'received' + AND event_row.source_envelope #>> + '{log_correlation,alertname}' + = 'DockerContainerUnhealthy' + AND event_row.source_envelope #>> + '{log_correlation,target_resource}' + = :target_resource + AND coalesce( + event_row.source_envelope #> + '{source_refs,fingerprints}', + '[]'::jsonb + ) ? :fingerprint + AND NOT (coalesce( + event_row.source_envelope #> + '{source_refs,event_ids}', + '[]'::jsonb + ) ? :occurrence_id) + LIMIT 1 + ) + SELECT + EXISTS (SELECT 1 FROM apply_boundary) + AS apply_anchor_found, + EXISTS (SELECT 1 FROM source_anchor) + AS source_anchor_found, + EXISTS (SELECT 1 FROM newer_recurrence) + AS newer_recurrence_found + """), + { + "apply_op_id": apply_op_id, + "project_id": project_id, + "target_resource": _SENTRY_PROFILING_CONTAINER, + "occurrence_id": identity["occurrence_id"], + "fingerprint": identity["fingerprint"], + }, + ) + row = result.mappings().one_or_none() + except Exception as exc: + return { + **base, + "error_type": type(exc).__name__, + "active_blockers": ["source_recurrence_readback_failed"], + } + + blockers: list[str] = [] + if not isinstance(row, Mapping) or row.get("apply_anchor_found") is not True: + blockers.append("source_recurrence_apply_anchor_missing") + if not isinstance(row, Mapping) or row.get("source_anchor_found") is not True: + blockers.append("source_recurrence_event_anchor_missing") + if not isinstance(row, Mapping) or row.get("newer_recurrence_found") is not False: + blockers.append( + "source_recurrence_observed_after_apply" + if isinstance(row, Mapping) + and row.get("newer_recurrence_found") is True + else "source_recurrence_readback_invalid" + ) + verified = not blockers + return { + **base, + "passed": verified, + "verified": verified, + "durable_readback_verified": verified, + "active_blockers": blockers, + } + + +def _attach_sentry_source_recurrence_fence( + verifier_receipt: Mapping[str, Any], + source_fence: Mapping[str, Any], +) -> dict[str, Any]: + postconditions = [ + dict(item) + for item in verifier_receipt.get("postconditions") or [] + if isinstance(item, Mapping) + ] + postconditions.append({ + "condition_id": source_fence.get("condition_id"), + "condition_type": source_fence.get("condition_type"), + "asset_id": _SENTRY_CANONICAL_ASSET_ID, + "passed": source_fence.get("passed") is True, + "returncode": None, + "timed_out": False, + "duration_ms": 0, + "output_sha256": None, + "error_type": source_fence.get("error_type"), + "raw_output_stored": False, + }) + blockers = [ + str(item) + for item in verifier_receipt.get("active_blockers") or [] + if str(item) + ] + blockers.extend( + str(item) + for item in source_fence.get("active_blockers") or [] + if str(item) and str(item) not in blockers + ) + verified = bool( + verifier_receipt.get("all_postconditions_passed") is True + and source_fence.get("verified") is True + and not blockers + ) + return { + **dict(verifier_receipt), + "verification_result": "success" if verified else "failed", + "all_postconditions_passed": verified, + "technical_postconditions_passed": verified, + "source_recurrence_fence_required": True, + "source_recurrence_fence_verified": ( + source_fence.get("verified") is True + ), + "source_recurrence_fence": dict(source_fence), + "required_postcondition_count": len(postconditions), + "passed_postcondition_count": sum( + item.get("passed") is True for item in postconditions + ), + "postconditions": postconditions, + "active_blockers": blockers, + } + + +def _durable_sentry_source_fence_matches_claim( + receipt: Mapping[str, Any], + claim: AnsibleCheckModeClaim, +) -> bool: + identity = _sentry_source_recurrence_identity(claim) + fence = receipt.get("source_recurrence_fence") + if identity is None or not isinstance(fence, Mapping): + return False + expected_fingerprint_sha = hashlib.sha256( + identity["fingerprint"].encode() + ).hexdigest() + expected_occurrence_sha = hashlib.sha256( + identity["occurrence_id"].encode() + ).hexdigest() + return bool( + receipt.get("source_recurrence_fence_required") is True + and receipt.get("source_recurrence_fence_verified") is True + and fence.get("schema_version") + == _SENTRY_SOURCE_RECURRENCE_FENCE_SCHEMA_VERSION + and fence.get("verified") is True + and fence.get("durable_readback_verified") is True + and fence.get("source_fingerprint_sha256") + == expected_fingerprint_sha + and fence.get("source_occurrence_id_sha256") + == expected_occurrence_sha + and fence.get("writes_on_verify") is False + and not fence.get("active_blockers") + ) + + def _durable_post_verifier_receipt_for_backfill( row: Mapping[str, Any], claim: AnsibleCheckModeClaim, @@ -1859,6 +2161,8 @@ def _durable_post_verifier_receipt_for_backfill( not active_blockers, receipt.get("writes_on_verify") is False, receipt.get("executor_returncode") == 0, + claim.catalog_id != _SENTRY_PROFILING_RECOVERY_CATALOG_ID + or _durable_sentry_source_fence_matches_claim(receipt, claim), ) return receipt if all(checks) else None @@ -6284,7 +6588,13 @@ async def _resolve_ansible_post_verifier_receipt( durable_verifier_receipt: Mapping[str, Any] | None = None, ) -> tuple[dict[str, Any], bool]: automation_run_id = _automation_run_id_for_claim(claim) - if durable_verifier_receipt is not None: + if durable_verifier_receipt is not None and ( + claim.catalog_id != _SENTRY_PROFILING_RECOVERY_CATALOG_ID + or _durable_sentry_source_fence_matches_claim( + durable_verifier_receipt, + claim, + ) + ): return dict(durable_verifier_receipt), True try: receipt = await run_ansible_asset_post_verifier( @@ -6295,6 +6605,18 @@ async def _resolve_ansible_post_verifier_receipt( executor_returncode=result.returncode, **_claim_correlation_kwargs(claim), ) + if claim.catalog_id == _SENTRY_PROFILING_RECOVERY_CATALOG_ID: + source_fence = await _verify_sentry_source_recurrence_fence( + claim, + apply_op_id=apply_op_id, + host_postconditions_passed=( + receipt.get("all_postconditions_passed") is True + ), + ) + receipt = _attach_sentry_source_recurrence_fence( + receipt, + source_fence, + ) return receipt, False except Exception as exc: logger.warning( diff --git a/apps/api/src/services/awooop_ansible_post_verifier.py b/apps/api/src/services/awooop_ansible_post_verifier.py index 659d56986..6b3d15202 100644 --- a/apps/api/src/services/awooop_ansible_post_verifier.py +++ b/apps/api/src/services/awooop_ansible_post_verifier.py @@ -505,6 +505,35 @@ _POSTCONDITION_REGISTRY: dict[str, tuple[AssetPostcondition, ...]] = { "sentry-self-hosted-snuba-profiling-functions-consumer-1 " "2>/dev/null)\" = healthy", ), + AssetPostcondition( + "host_110_sentry_profiling_alert_resolved_after_window", + "metric", + "host_110", + "clean_samples=0; for attempt in $(seq 1 43); do " + "curl -fsS --connect-timeout 3 --max-time 5 --get " + "--data-urlencode " + "'query=ALERTS{alertname=\"DockerContainerUnhealthy\"," + "alertstate=\"firing\",job=\"docker-health-monitor\"," + "host=\"110\",container=\"" + "sentry-self-hosted-snuba-profiling-functions-consumer-1\"} or " + "ALERTS{alertname=\"DockerContainerUnhealthy\"," + "alertstate=\"firing\",job=\"docker-health-monitor\"," + "host=\"110\",name=\"" + "sentry-self-hosted-snuba-profiling-functions-consumer-1\"} or " + "ALERTS{alertname=\"DockerContainerUnhealthy\"," + "alertstate=\"firing\",job=\"docker-health-monitor\"," + "host=\"110\",container_name=\"" + "sentry-self-hosted-snuba-profiling-functions-consumer-1\"}' " + "http://127.0.0.1:9090/api/v1/query | python3 -c '" + "import json,sys; p=json.load(sys.stdin); " + "raise SystemExit(0 if p.get(\"status\")==\"success\" " + "and p.get(\"data\",{}).get(\"result\")==[] else 1)' " + "&& clean_samples=$((clean_samples + 1)) " + "|| clean_samples=0; " + "test \"$clean_samples\" -ge 37 && exit 0; " + "sleep 10; done; exit 1", + 660, + ), ), "ansible:110-devops-full-convergence": ( AssetPostcondition( diff --git a/apps/api/tests/test_ansible_verified_closure.py b/apps/api/tests/test_ansible_verified_closure.py index 99103d7bf..56e8ca1fc 100644 --- a/apps/api/tests/test_ansible_verified_closure.py +++ b/apps/api/tests/test_ansible_verified_closure.py @@ -1,5 +1,6 @@ from __future__ import annotations +import hashlib import inspect import json from contextlib import asynccontextmanager @@ -126,6 +127,283 @@ def _claim() -> service.AnsibleCheckModeClaim: ) +def _sentry_claim() -> service.AnsibleCheckModeClaim: + claim = _claim() + return replace( + claim, + catalog_id="ansible:110-sentry-profiling-consumer-recovery", + playbook_path=( + "infra/ansible/playbooks/" + "110-sentry-profiling-consumer-recovery.yml" + ), + apply_playbook_path=( + "infra/ansible/playbooks/" + "110-sentry-profiling-consumer-recovery.yml" + ), + inventory_hosts=("host_110",), + input_payload={ + **claim.input_payload, + "project_id": "awoooi", + "typed_target_route": { + "canonical_asset_id": "service:sentry", + }, + "source_recurrence": { + "schema_version": "alert_source_recurrence_receipt_v1", + "verified": True, + "fingerprint": "a" * 32, + "occurrence_id": "alert-20260719010101", + "observed_at": datetime.now(UTC).isoformat(), + "identity_anchor": "webhook_payload", + "canonical_asset_id": "service:sentry", + "source": "current_alert_webhook", + "durable_readback_verified": False, + }, + }, + ) + + +@pytest.mark.asyncio +async def test_sentry_source_recurrence_fence_passes_only_with_both_anchors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + db = _SequenceDB( + _MappingResult(), + _MappingResult({ + "apply_anchor_found": True, + "source_anchor_found": True, + "newer_recurrence_found": False, + }), + ) + + @asynccontextmanager + async def db_context(project_id: str): + assert project_id == "awoooi" + yield db + + monkeypatch.setattr(service, "get_db_context", db_context) + + receipt = await service._verify_sentry_source_recurrence_fence( + _sentry_claim(), + apply_op_id="00000000-0000-0000-0000-000000000901", + host_postconditions_passed=True, + ) + + assert receipt["verified"] is True + assert receipt["durable_readback_verified"] is True + assert receipt["direct_webhook_recurrence_covered"] is True + assert receipt["active_blockers"] == [] + sql = db.statements[1] + assert "WITH apply_boundary AS MATERIALIZED" in sql + assert "source_anchor AS MATERIALIZED" in sql + assert "newer_recurrence AS MATERIALIZED" in sql + assert "event_row.received_at > apply_boundary.created_at" in sql + assert "'{source_refs,event_ids}'" in sql + assert "'{source_refs,fingerprints}'" in sql + assert db.parameters[1]["occurrence_id"] == "alert-20260719010101" + assert db.parameters[1]["fingerprint"] == "a" * 32 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("row", "expected_blocker"), + [ + ( + { + "apply_anchor_found": False, + "source_anchor_found": True, + "newer_recurrence_found": False, + }, + "source_recurrence_apply_anchor_missing", + ), + ( + { + "apply_anchor_found": True, + "source_anchor_found": False, + "newer_recurrence_found": False, + }, + "source_recurrence_event_anchor_missing", + ), + ( + { + "apply_anchor_found": True, + "source_anchor_found": True, + "newer_recurrence_found": True, + }, + "source_recurrence_observed_after_apply", + ), + ( + { + "apply_anchor_found": True, + "source_anchor_found": True, + "newer_recurrence_found": None, + }, + "source_recurrence_readback_invalid", + ), + ], +) +async def test_sentry_source_recurrence_fence_fails_closed( + monkeypatch: pytest.MonkeyPatch, + row: dict, + expected_blocker: str, +) -> None: + db = _SequenceDB(_MappingResult(), _MappingResult(row)) + + @asynccontextmanager + async def db_context(_project_id: str): + yield db + + monkeypatch.setattr(service, "get_db_context", db_context) + + receipt = await service._verify_sentry_source_recurrence_fence( + _sentry_claim(), + apply_op_id="00000000-0000-0000-0000-000000000902", + host_postconditions_passed=True, + ) + + assert receipt["verified"] is False + assert expected_blocker in receipt["active_blockers"] + + +@pytest.mark.asyncio +async def test_sentry_post_verifier_wires_durable_recurrence_fence( + monkeypatch: pytest.MonkeyPatch, +) -> None: + host_receipt = { + "schema_version": service.ANSIBLE_POST_VERIFIER_SCHEMA_VERSION, + "verifier": "asset_specific_read_only_host_postconditions", + "independent_source": "broker_ssh_host_runtime_readback", + "automation_run_id": _sentry_claim().input_payload["automation_run_id"], + "apply_op_id": "00000000-0000-0000-0000-000000000903", + "catalog_id": "ansible:110-sentry-profiling-consumer-recovery", + "verification_result": "success", + "all_postconditions_passed": True, + "technical_postconditions_passed": True, + "required_postcondition_count": 3, + "passed_postcondition_count": 3, + "postconditions": [ + {"condition_id": f"host-{index}", "passed": True} + for index in range(3) + ], + "active_blockers": [], + "writes_on_verify": False, + "executor_returncode": 0, + } + host_verifier = AsyncMock(return_value=host_receipt) + monkeypatch.setattr( + service, + "run_ansible_asset_post_verifier", + host_verifier, + ) + fence = { + "schema_version": "sentry_source_recurrence_fence_receipt_v1", + "condition_id": "awoooi_sentry_source_recurrence_after_apply", + "condition_type": "source_recurrence", + "verified": True, + "passed": True, + "durable_readback_verified": True, + "writes_on_verify": False, + "raw_output_stored": False, + "source_fingerprint_sha256": hashlib.sha256(("a" * 32).encode()).hexdigest(), + "source_occurrence_id_sha256": hashlib.sha256( + b"alert-20260719010101" + ).hexdigest(), + "active_blockers": [], + } + monkeypatch.setattr( + service, + "_verify_sentry_source_recurrence_fence", + AsyncMock(return_value=fence), + ) + + receipt, reused = await service._resolve_ansible_post_verifier_receipt( + _sentry_claim(), + service.AnsibleRunResult(0, "", "", 1), + apply_op_id="00000000-0000-0000-0000-000000000903", + ) + + assert reused is False + assert receipt["verification_result"] == "success" + assert receipt["source_recurrence_fence_verified"] is True + assert receipt["required_postcondition_count"] == 4 + assert receipt["passed_postcondition_count"] == 4 + assert receipt["postconditions"][-1]["condition_type"] == ( + "source_recurrence" + ) + + host_verifier.reset_mock() + replayed_receipt, replayed_reused = ( + await service._resolve_ansible_post_verifier_receipt( + _sentry_claim(), + service.AnsibleRunResult(0, "", "", 1), + apply_op_id="00000000-0000-0000-0000-000000000903", + durable_verifier_receipt=host_receipt, + ) + ) + assert replayed_reused is False + assert replayed_receipt["source_recurrence_fence_verified"] is True + host_verifier.assert_awaited_once() + + +def test_sentry_durable_verifier_cannot_bypass_source_recurrence_fence() -> None: + claim = _sentry_claim() + apply_op_id = "00000000-0000-0000-0000-000000000904" + host_receipt = { + "schema_version": service.ANSIBLE_POST_VERIFIER_SCHEMA_VERSION, + "automation_run_id": claim.input_payload["automation_run_id"], + "apply_op_id": apply_op_id, + "catalog_id": claim.catalog_id, + "verification_result": "success", + "all_postconditions_passed": True, + "required_postcondition_count": 3, + "passed_postcondition_count": 3, + "postconditions": [ + {"condition_id": f"host-{index}", "passed": True} + for index in range(3) + ], + "active_blockers": [], + "writes_on_verify": False, + "executor_returncode": 0, + } + row = {"durable_verifier_receipt": host_receipt} + + assert service._durable_post_verifier_receipt_for_backfill( + row, + claim, + apply_op_id=apply_op_id, + ) is None + + fence = { + "schema_version": "sentry_source_recurrence_fence_receipt_v1", + "condition_id": "awoooi_sentry_source_recurrence_after_apply", + "condition_type": "source_recurrence", + "verified": True, + "passed": True, + "durable_readback_verified": True, + "writes_on_verify": False, + "source_fingerprint_sha256": hashlib.sha256(("a" * 32).encode()).hexdigest(), + "source_occurrence_id_sha256": hashlib.sha256( + b"alert-20260719010101" + ).hexdigest(), + "active_blockers": [], + } + fenced = service._attach_sentry_source_recurrence_fence( + host_receipt, + fence, + ) + assert service._durable_post_verifier_receipt_for_backfill( + {"durable_verifier_receipt": fenced}, + claim, + apply_op_id=apply_op_id, + ) == fenced + + fenced["source_recurrence_fence"]["source_fingerprint_sha256"] = "b" * 64 + assert service._durable_post_verifier_receipt_for_backfill( + {"durable_verifier_receipt": fenced}, + claim, + apply_op_id=apply_op_id, + ) is None + + def _verified_result() -> service.AnsibleRunResult: return service.AnsibleRunResult( returncode=0, diff --git a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py index d6902a1b7..d4f23fcab 100644 --- a/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py +++ b/apps/api/tests/test_sre_k3s_controlled_automation_work_items_api.py @@ -155,9 +155,9 @@ def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() - "active_or_completed_commitments": 68, "by_status": { "analysis_or_governance_complete": 6, - "source_implemented_runtime_pending": 24, + "source_implemented_runtime_pending": 25, "in_progress": 32, - "not_started_or_no_current_evidence": 6, + "not_started_or_no_current_evidence": 5, "superseded": 2, }, "product_runtime_closed_commitments": 0, @@ -168,6 +168,9 @@ def test_loader_returns_fixed_architecture_provider_order_and_agent99_bridge() - assert commitments["AIA-CONV-012"]["title"].startswith("Claude API") assert commitments["AIA-CONV-013"]["title"].startswith("Gemini API") assert "Host112" in commitments["AIA-CONV-049"]["title"] + assert commitments["AIA-CONV-036"]["status"] == ( + "source_implemented_runtime_pending" + ) assert commitments["AIA-CONV-050"]["status"] == ( "source_implemented_runtime_pending" ) diff --git a/apps/api/tests/test_sre_typed_domain_router.py b/apps/api/tests/test_sre_typed_domain_router.py index a2d57cd17..37d85e547 100644 --- a/apps/api/tests/test_sre_typed_domain_router.py +++ b/apps/api/tests/test_sre_typed_domain_router.py @@ -5,6 +5,7 @@ from __future__ import annotations import hashlib import json import os +import subprocess from contextlib import asynccontextmanager from pathlib import Path from types import SimpleNamespace @@ -60,6 +61,7 @@ PLAYBOOK = ( ) SENTRY_CONTAINER = "sentry-self-hosted-snuba-profiling-functions-consumer-1" SENTRY_CATALOG = "ansible:110-sentry-profiling-consumer-recovery" +SENTRY_FINGERPRINT = "a" * 32 WAZUH_CATALOG = "ansible:wazuh-manager-posture-readback" @@ -169,7 +171,7 @@ def _sentry_runtime_incident() -> SimpleNamespace: "container_name": SENTRY_CONTAINER, "namespace": "default", "host": "110", - "fingerprint": "sentry-profiling-fingerprint", + "fingerprint": SENTRY_FINGERPRINT, }, annotations={}, ) @@ -182,11 +184,13 @@ def _sentry_runtime_proposal(occurrence_id: str) -> dict: "source": "alert_webhook_controlled_router", "risk_level": "medium", "action": SENTRY_CONTAINER, - "source_fingerprint": "sentry-profiling-fingerprint", + "source_fingerprint": SENTRY_FINGERPRINT, "source_occurrence_id": occurrence_id, "source_recurrence_verified": True, "source_recurrence_source": "current_alert_webhook", "source_recurrence_anchor": "webhook_payload", + "source_received_at": "2026-07-19T00:00:00+00:00", + "source_canonical_asset_id": "service:sentry", } @@ -1388,9 +1392,12 @@ def test_webhook_recurrence_receipt_requires_explicit_occurrence_identity() -> N assert payload is not None recurrence = payload["input"]["source_recurrence"] - assert recurrence["fingerprint"] == "sentry-profiling-fingerprint" + assert recurrence["fingerprint"] == SENTRY_FINGERPRINT assert recurrence["occurrence_id"] is None assert recurrence["verified"] is False + assert recurrence["observed_at"] == "2026-07-19T00:00:00+00:00" + assert recurrence["canonical_asset_id"] == "service:sentry" + assert recurrence["source"] == "current_alert_webhook" def test_sentry_playbook_and_post_verifier_are_exact_container_bounded() -> None: @@ -1405,8 +1412,26 @@ def test_sentry_playbook_and_post_verifier_are_exact_container_bounded() -> None assert catalog["inventory_hosts"] == ["host_110"] assert catalog["auto_apply_enabled"] is True assert catalog["risk_level"] == "medium" - assert len(postconditions) == 2 + assert catalog["catalog_revision"] == ( + "2026-07-19-sentry-exact-container-v2" + ) + assert len(postconditions) == 3 assert all(row.inventory_host == "host_110" for row in postconditions) + recurrence = next( + row + for row in postconditions + if row.condition_id + == "host_110_sentry_profiling_alert_resolved_after_window" + ) + assert recurrence.condition_type == "metric" + assert recurrence.timeout_seconds == 660 + assert 'alertname=\"DockerContainerUnhealthy\"' in recurrence.probe + assert 'job=\"docker-health-monitor\"' in recurrence.probe + assert 'host=\"110\"' in recurrence.probe + assert SENTRY_CONTAINER in recurrence.probe + assert "result\")==[]" in recurrence.probe + assert "clean_samples=$((clean_samples + 1))" in recurrence.probe + assert 'test "$clean_samples" -ge 37' in recurrence.probe assert source.count(SENTRY_CONTAINER) == 1 for forbidden in ( "docker compose", @@ -1418,6 +1443,55 @@ def test_sentry_playbook_and_post_verifier_are_exact_container_bounded() -> None assert forbidden not in source.lower() +def test_sentry_alert_probe_requires_consecutive_clean_samples( + tmp_path: Path, +) -> None: + recurrence = next( + row + for row in postconditions_for_catalog(SENTRY_CATALOG) + if row.condition_id + == "host_110_sentry_profiling_alert_resolved_after_window" + ) + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + count_file = tmp_path / "curl-count" + fake_curl = fake_bin / "curl" + fake_curl.write_text( + """#!/usr/bin/env bash +count=0 +test ! -f "$SENTRY_PROBE_COUNT_FILE" || count=$(cat "$SENTRY_PROBE_COUNT_FILE") +count=$((count + 1)) +printf '%s' "$count" > "$SENTRY_PROBE_COUNT_FILE" +if test "$count" -eq 2; then + printf '%s\n' '{"status":"success","data":{"result":[{"metric":{"alertname":"DockerContainerUnhealthy"}}]}}' +else + printf '%s\n' '{"status":"success","data":{"result":[]}}' +fi +""", + encoding="utf-8", + ) + fake_sleep = fake_bin / "sleep" + fake_sleep.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8") + fake_curl.chmod(0o755) + fake_sleep.chmod(0o755) + + result = subprocess.run( + ["bash", "-c", recurrence.probe], + check=False, + capture_output=True, + text=True, + timeout=10, + env={ + **os.environ, + "PATH": f"{fake_bin}:{os.environ['PATH']}", + "SENTRY_PROBE_COUNT_FILE": str(count_file), + }, + ) + + assert result.returncode == 0, result.stderr + assert count_file.read_text(encoding="utf-8") == "39" + + def test_converged_webhook_recurrence_reenters_generation_preflight() -> None: source = (ROOT / "apps/api/src/api/v1/webhooks.py").read_text( encoding="utf-8" @@ -1426,6 +1500,8 @@ def test_converged_webhook_recurrence_reenters_generation_preflight() -> None: assert '"source_fingerprint": resolved_source_fingerprint' in source assert '"source_occurrence_id": source_alert_id' in source assert '"source_recurrence_verified": bool(' in source + assert '"source_received_at": datetime.now(UTC).isoformat()' in source + assert '"source_canonical_asset_id": typed_target_route.get(' in source assert source.count( 'getattr(updated_approval, "incident_id", None)\n' " and _controlled_ai_policy_allows" diff --git a/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json b/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json index 82c79c7a3..6e3e2f89e 100644 --- a/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json +++ b/docs/operations/sre-ai-agent-conversation-commitments.snapshot.json @@ -1,7 +1,7 @@ { "schema_version": "sre_ai_agent_conversation_commitments_v1", "program_id": "AIA-SRE-P0-20260715", - "generated_at": "2026-07-19T11:40:41+08:00", + "generated_at": "2026-07-19T11:49:00+08:00", "source_scope": "This Codex AI Agent task only; cross-task delegation envelopes are excluded.", "authoritative_for_execution_order": false, "raw_conversation_embedded": false, @@ -53,7 +53,7 @@ {"id":"AIA-CONV-033","category":"named_incident","status":"in_progress","title":"INC-20260711-D037E5 修復 stale candidate 阻擋、自動排隊與 durable closure","linked_work_items":["AIA-SRE-007","AIA-SRE-015"],"terminal_condition":"新候選可 claim，apply 不重複，verifier、Telegram 與 learning receipts 同 run 完成。"}, {"id":"AIA-CONV-034","category":"named_incident","status":"in_progress","title":"AWOOOI CPU 高負載與 P99 上升完成 RCA、bounded repair 與 verifier","linked_work_items":["AIA-SRE-004","AIA-SRE-006","AIA-SRE-010"],"terminal_condition":"相關 metrics/logs/changes 對齊 canonical asset，修復後 latency 與資源 verifier 關閉。"}, {"id":"AIA-CONV-035","category":"named_incident","status":"in_progress","title":"Backup/restore escrow 補 freshness、escrow metadata、restore drill 與 DR scorecard","linked_work_items":["AIA-SRE-011","AIA-SRE-016"],"terminal_condition":"read-only evidence 齊全；缺 escrow/restore receipt 時保持 blocked 而非假綠。"}, - {"id":"AIA-CONV-036","category":"named_incident","status":"not_started_or_no_current_evidence","title":"Sentry Snuba DockerContainerUnhealthy 反覆告警自動修復","linked_work_items":["AIA-SRE-005"],"terminal_condition":"exact-container replay 完成 bounded recovery、健康 verifier 與 recurrence fence。"}, + {"id":"AIA-CONV-036","category":"named_incident","status":"source_implemented_runtime_pending","title":"Sentry Snuba DockerContainerUnhealthy 反覆告警自動修復","linked_work_items":["AIA-SRE-005"],"terminal_condition":"exact-container replay 完成 bounded recovery、健康 verifier 與 recurrence fence。"}, {"id":"AIA-CONV-037","category":"named_incident","status":"source_implemented_runtime_pending","title":"AlertChainBroken_Alertmanager webhook error 大於 10% 自動旁路與修復","linked_work_items":["AIA-SRE-014","AIA-SRE-017"],"terminal_condition":"integration-scoped counters、旁路 delivery、receiver contract 與恢復 receipt 全部成立。"}, {"id":"AIA-CONV-038","category":"named_incident","status":"source_implemented_runtime_pending","title":"provider_freshness_signal 改成 readback、repair 與 verify","linked_work_items":["AIA-SRE-013","AIA-SRE-017"],"terminal_condition":"provider window、last seen、direct reference 與 no-false-green verifier 決定 terminal。"}, {"id":"AIA-CONV-039","category":"named_incident","status":"source_implemented_runtime_pending","title":"Backup.awoooi-frequent、Langfuse、Signoz、AI 等 backup family 分類與閉環","linked_work_items":["AIA-SRE-011","AIA-SRE-017"],"terminal_condition":"每類 backup signal 有正確 route、read-only receipt、去重與 closure。"}, @@ -98,9 +98,9 @@ "active_or_completed_commitments": 68, "by_status": { "analysis_or_governance_complete": 6, - "source_implemented_runtime_pending": 24, + "source_implemented_runtime_pending": 25, "in_progress": 32, - "not_started_or_no_current_evidence": 6, + "not_started_or_no_current_evidence": 5, "superseded": 2 }, "product_runtime_closed_commitments": 0