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 9edb2b681..8439a5716 100644 --- a/apps/api/src/services/awooop_ansible_check_mode_service.py +++ b/apps/api/src/services/awooop_ansible_check_mode_service.py @@ -56,6 +56,20 @@ _CONTROLLED_RETRY_ERROR_BACKOFF_MIN_SECONDS = 15 _CONTROLLED_RETRY_ERROR_BACKOFF_MAX_SECONDS = 30 _STDIN_BOUNDARY_REPLAY_WINDOW_HOURS = 7 * 24 _CATALOG_REPAIR_REPLAY_WINDOW_HOURS = 72 +_STALE_CLAIM_REPLAY_CONTROL_FIELDS = ( + "bounded_execution_scope", + "catalog_drift_replay", + "catalog_repair_of_apply_op_id", + "catalog_replay_revision", + "previous_catalog_revision", + "previous_check_mode_playbook_path", + "replay_of_check_mode_op_id", + "semantic_repair_of_apply_op_id", + "semantic_route_reconciliation", +) +_HISTORICAL_SHADOW_REPLAY_TAGS = frozenset( + {"catalog_drift_replay", "semantic_catalog_reconciliation"} +) _STDIN_BOUNDARY_TERMINAL_PROJECTION_BATCH_LIMIT = 5 _EXECUTION_CAPABILITY_MIN_TTL_SECONDS = 300 _EXECUTION_CAPABILITY_MAX_TTL_SECONDS = 1_800 @@ -1136,6 +1150,24 @@ def _claim_from_stale_check_mode_row( "incident_semantic_route_rechecked": incident_routing_evaluated, } ) + for field_name in _STALE_CLAIM_REPLAY_CONTROL_FIELDS: + if field_name in input_payload: + claim_input[field_name] = input_payload[field_name] + row_tags = { + str(tag) + for tag in (row.get("tags") or []) + if isinstance(tag, str) + } + historical_shadow_replay = bool( + row_tags & _HISTORICAL_SHADOW_REPLAY_TAGS + ) + if ( + historical_shadow_replay + or input_payload.get("historical_projection_backfill") is True + or input_payload.get("telegram_provider_delivery") == "shadow_only" + ): + claim_input["historical_projection_backfill"] = True + claim_input["telegram_provider_delivery"] = "shadow_only" return AnsibleCheckModeClaim( op_id=str(row.get("op_id") or ""), source_candidate_op_id=source_candidate_op_id, @@ -7443,6 +7475,7 @@ async def claim_stale_pending_check_modes( SELECT check_mode.op_id, check_mode.parent_op_id, + check_mode.tags, coalesce( check_mode.incident_id::text, check_mode.input ->> 'incident_id' diff --git a/apps/api/tests/test_host_disk_pressure_ansible_catalog.py b/apps/api/tests/test_host_disk_pressure_ansible_catalog.py index 1314b40c0..52fb479e8 100644 --- a/apps/api/tests/test_host_disk_pressure_ansible_catalog.py +++ b/apps/api/tests/test_host_disk_pressure_ansible_catalog.py @@ -17,6 +17,7 @@ from src.services.awooop_ansible_check_mode_service import ( _send_controlled_apply_telegram_receipt, claim_catalog_drift_failed_check_modes, claim_semantically_misrouted_verified_applies, + claim_stale_pending_check_modes, run_pending_check_modes_once, ) from src.services.awooop_ansible_post_verifier import postconditions_for_catalog @@ -100,6 +101,7 @@ def test_stale_generic_host188_row_is_semantically_remapped() -> None: { "op_id": "00000000-0000-0000-0000-000000000188", "parent_op_id": "00000000-0000-0000-0000-000000000100", + "tags": ["ansible", "check_mode", "catalog_drift_replay"], "incident_id": "INC-DISK-188", "incident_alertname": "HostOutOfDiskSpace", "incident_alert_category": "host_resource", @@ -114,6 +116,15 @@ def test_stale_generic_host188_row_is_semantically_remapped() -> None: "playbook_path": "infra/ansible/playbooks/110-devops.yml", "inventory_hosts": ["host_110"], "risk_level": "medium", + "automation_run_id": ( + "00000000-0000-0000-0000-000000000188" + ), + "catalog_drift_replay": True, + "replay_of_check_mode_op_id": ( + "00000000-0000-0000-0000-000000000099" + ), + "historical_projection_backfill": True, + "telegram_provider_delivery": "shadow_only", }, } ) @@ -123,6 +134,37 @@ def test_stale_generic_host188_row_is_semantically_remapped() -> None: assert claim.inventory_hosts == ("host_188",) assert claim.input_payload["catalog_route_changed"] is True assert claim.input_payload["incident_semantic_route_rechecked"] is True + assert claim.input_payload["catalog_drift_replay"] is True + assert claim.input_payload["historical_projection_backfill"] is True + assert claim.input_payload["telegram_provider_delivery"] == "shadow_only" + + +def test_stale_catalog_replay_tag_recovers_missing_shadow_policy() -> None: + claim = _claim_from_stale_check_mode_row( + { + "op_id": "00000000-0000-0000-0000-000000000198", + "parent_op_id": "00000000-0000-0000-0000-000000000100", + "tags": ["ansible", "check_mode", "catalog_drift_replay"], + "incident_id": "INC-DISK-188-TAG-FALLBACK", + "incident_alertname": "HostOutOfDiskSpace", + "incident_alert_category": "host_resource", + "incident_affected_services": ["node-exporter-188"], + "input": { + "source_candidate_op_id": ( + "00000000-0000-0000-0000-000000000100" + ), + "incident_id": "INC-DISK-188-TAG-FALLBACK", + "catalog_id": "ansible:110-devops", + "playbook_path": "infra/ansible/playbooks/110-devops.yml", + "inventory_hosts": ["host_110"], + "risk_level": "medium", + }, + } + ) + + assert claim is not None + assert claim.input_payload["historical_projection_backfill"] is True + assert claim.input_payload["telegram_provider_delivery"] == "shadow_only" def test_disk_catalogs_have_independent_threshold_verifiers() -> None: @@ -141,12 +183,14 @@ def test_disk_catalogs_have_independent_threshold_verifiers() -> None: def test_historical_catalog_replay_is_per_failed_row_and_provider_silent() -> None: claim_source = inspect.getsource(claim_catalog_drift_failed_check_modes) + stale_source = inspect.getsource(claim_stale_pending_check_modes) send_source = inspect.getsource(_send_controlled_apply_telegram_receipt) assert "replay_of_check_mode_op_id" in claim_source assert "failed_check_mode_op_id" in claim_source assert '"historical_projection_backfill": True' in claim_source assert '"telegram_provider_delivery": "shadow_only"' in claim_source + assert "check_mode.tags" in stale_source assert "telegram_provider_delivery" in send_source