wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

fix(ai-loop): validate durable Wazuh closure receipts

Browse Source
This commit is contained in:
Your Name
2026-07-22 20:36:25 +08:00
parent 6b87cf6d7f
commit bb9be8827b
2 changed files with 512 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
apps/api/src/services/ai_agent_autonomous_runtime_control.py
View File

@@ -3720,9 +3720,7 @@ def _autonomous_execution_loop_ledger(
for row in telegram_rows
if _durable_controlled_apply_telegram_receipt(
row,
require_provider_receipt=(
catalog_id == _WAZUH_POSTURE_CATALOG_ID
),
catalog_id=catalog_id,
)
and (
not incident_id
@@ -3943,29 +3941,122 @@ def _autonomous_execution_loop_ledger(
def _durable_controlled_apply_telegram_receipt(
row: Mapping[str, Any],
*,
require_provider_receipt: bool = False,
catalog_id: str = "",
retry_terminal_receipt: bool = False,
) -> bool:
"""Accept provider sends or policy-complete no-send suppression receipts."""
if str(row.get("action") or "") != "controlled_apply_result":
return False
wazuh_posture = catalog_id == _WAZUH_POSTURE_CATALOG_ID
send_status = str(row.get("send_status") or "")
if send_status == "sent":
return (
bool(row.get("provider_message_id"))
if require_provider_receipt
else True
if not wazuh_posture:
return True
if not row.get("provider_message_id"):
return False
policy_version = str(row.get("notification_policy_version") or "")
provider_delivery = str(
row.get("notification_provider_delivery") or ""
)
if require_provider_receipt:
lifecycle_state = str(row.get("notification_lifecycle_state") or "")
if retry_terminal_receipt:
execution_kind = str(
row.get("notification_execution_kind") or ""
)
return bool(
str(row.get("notification_disposition") or "") == "sent"
and lifecycle_state == "failed"
and (
(
execution_kind == "no_write_posture_readback"
and (
(
policy_version
== "telegram_posture_lifecycle_transition_v1"
and provider_delivery == "state_transition"
)
or (
policy_version
== "telegram_posture_state_digest_v1"
and provider_delivery == "digest"
)
)
)
or (
execution_kind == "no_write_replay"
and policy_version == "telegram_failure_digest_v2"
and provider_delivery == "digest"
)
)
)
return bool(
policy_version == "telegram_posture_lifecycle_transition_v1"
and str(row.get("notification_disposition") or "") == "sent"
and provider_delivery == "state_transition"
and lifecycle_state == "recovered"
)
if send_status != "shadow" or row.get("provider_message_id"):
return False
policy_version = str(row.get("notification_policy_version") or "")
provider_delivery = str(row.get("notification_provider_delivery") or "")
suppression_reason = str(row.get("notification_suppression_reason") or "")
generic_shadow = bool(
policy_version
and str(row.get("notification_disposition") or "") == "suppressed"
and provider_delivery in {"shadow_only", "state_transition", "digest"}
and row.get("notification_provider_send_suppressed") is True
)
if not wazuh_posture:
return generic_shadow
lifecycle_state = str(row.get("notification_lifecycle_state") or "")
if retry_terminal_receipt:
if not generic_shadow or lifecycle_state != "failed":
return False
execution_kind = str(row.get("notification_execution_kind") or "")
return bool(
(
execution_kind == "no_write_posture_readback"
and policy_version
== "telegram_posture_lifecycle_transition_v1"
and provider_delivery == "state_transition"
and suppression_reason == "duplicate_lifecycle_state"
)
or (
execution_kind == "no_write_posture_readback"
and policy_version == "telegram_posture_state_digest_v1"
and provider_delivery == "shadow_only"
and suppression_reason == "historical_projection_backfill"
)
or (
execution_kind == "no_write_replay"
and policy_version == "telegram_failure_digest_v2"
and (
(
provider_delivery == "digest"
and suppression_reason
== "duplicate_failure_fingerprint_window"
)
or (
provider_delivery == "shadow_only"
and suppression_reason == "historical_projection_backfill"
)
)
)
)
if not generic_shadow or lifecycle_state != "recovered":
return False
return bool(
send_status == "shadow"
and not row.get("provider_message_id")
and str(row.get("notification_policy_version") or "")
and str(row.get("notification_disposition") or "") == "suppressed"
and str(row.get("notification_provider_delivery") or "")
in {"shadow_only", "state_transition", "digest"}
and row.get("notification_provider_send_suppressed") is True
(
policy_version == "telegram_posture_lifecycle_transition_v1"
and provider_delivery == "state_transition"
and suppression_reason == "duplicate_lifecycle_state"
)
or (
policy_version == "telegram_posture_state_digest_v1"
and provider_delivery == "shadow_only"
and suppression_reason == "historical_projection_backfill"
)
)
@@ -4038,9 +4129,7 @@ def _latest_flow_closure(
for row in telegram_rows
if _durable_controlled_apply_telegram_receipt(
row,
require_provider_receipt=(
catalog_id == _WAZUH_POSTURE_CATALOG_ID
),
catalog_id=catalog_id,
)
and (
not incident_id
@@ -5041,6 +5130,15 @@ def _build_autonomous_retry_rollback_terminal_readback(
"notification_provider_delivery": row.get(
"telegram_notification_provider_delivery"
),
"notification_execution_kind": row.get(
"telegram_notification_execution_kind"
),
"notification_lifecycle_state": row.get(
"telegram_notification_lifecycle_state"
),
"notification_suppression_reason": row.get(
"telegram_notification_suppression_reason"
),
"notification_provider_send_suppressed": row.get(
"telegram_provider_send_suppressed"
),
@@ -5109,7 +5207,11 @@ def _build_autonomous_retry_rollback_terminal_readback(
),
"telegram_receipt_same_run": bool(
row.get("telegram_message_id")
and _durable_controlled_apply_telegram_receipt(telegram_receipt)
and _durable_controlled_apply_telegram_receipt(
telegram_receipt,
catalog_id=str(row.get("catalog_id") or ""),
retry_terminal_receipt=True,
)
and row.get("telegram_automation_run_id") == automation_run_id
),
"public_safe_receipt": bool(
@@ -8146,6 +8248,12 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """
AS telegram_notification_disposition,
telegram.notification_provider_delivery
AS telegram_notification_provider_delivery,
telegram.notification_execution_kind
AS telegram_notification_execution_kind,
telegram.notification_lifecycle_state
AS telegram_notification_lifecycle_state,
telegram.notification_suppression_reason
AS telegram_notification_suppression_reason,
telegram.provider_send_suppressed
AS telegram_provider_send_suppressed,
replay.created_at AS retry_created_at
@@ -8263,6 +8371,17 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """
outbound.source_envelope #>>
'{notification_policy,provider_delivery}'
AS notification_provider_delivery,
coalesce(
outbound.source_envelope #>>
'{callback_reply,execution_kind}',
outbound.source_envelope ->> 'execution_kind'
) AS notification_execution_kind,
outbound.source_envelope #>>
'{notification_policy,lifecycle_state}'
AS notification_lifecycle_state,
outbound.source_envelope #>>
'{notification_policy,suppression_reason}'
AS notification_suppression_reason,
outbound.source_envelope #>>
'{notification_policy,provider_send_performed}'
= 'false' AS provider_send_suppressed
@@ -8270,7 +8389,10 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """
WHERE outbound.project_id = :project_id
AND outbound.channel_type = 'telegram'
AND (
outbound.send_status = 'sent'
(
outbound.send_status = 'sent'
AND outbound.provider_message_id IS NOT NULL
)
OR (
outbound.send_status = 'shadow'
AND outbound.provider_message_id IS NULL
@@ -8302,6 +8424,17 @@ _RUNTIME_RETRY_TERMINAL_LATEST_SQL = """
replay.input ->> 'automation_run_id',
apply.input ->> 'automation_run_id'
)
AND coalesce(
outbound.source_envelope #>>
'{callback_reply,incident_id}',
outbound.source_envelope #>>
'{source_refs,incident_ids,0}'
) = coalesce(
apply.incident_id::text,
apply.input ->> 'incident_id'
)
AND outbound.source_envelope #>>
'{callback_reply,apply_op_id}' = apply.op_id::text
ORDER BY outbound.sent_at DESC NULLS LAST, outbound.queued_at DESC
LIMIT 1
) telegram ON TRUE
@@ -8789,6 +8922,10 @@ _RUNTIME_TELEGRAM_LATEST_SQL = """
AS notification_disposition,
source_envelope #>> '{notification_policy,provider_delivery}'
AS notification_provider_delivery,
source_envelope #>> '{notification_policy,lifecycle_state}'
AS notification_lifecycle_state,
source_envelope #>> '{notification_policy,suppression_reason}'
AS notification_suppression_reason,
source_envelope #>> '{notification_policy,provider_send_performed}'
= 'false' AS notification_provider_send_suppressed,
queued_at,
@@ -8844,6 +8981,10 @@ _RUNTIME_TELEGRAM_LATEST_DIRECT_SQL = """
AS notification_disposition,
source_envelope #>> '{notification_policy,provider_delivery}'
AS notification_provider_delivery,
source_envelope #>> '{notification_policy,lifecycle_state}'
AS notification_lifecycle_state,
source_envelope #>> '{notification_policy,suppression_reason}'
AS notification_suppression_reason,
source_envelope #>> '{notification_policy,provider_send_performed}'
= 'false' AS notification_provider_send_suppressed,
queued_at,