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

fix(aiops): verify provider freshness per source

Browse Source
This commit is contained in:
ogt
2026-07-16 23:17:13 +08:00
parent 6188aabd44
commit d9487cd55a
10 changed files with 547 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
apps/api/src/services/backup_restore_signal_automation.py
View File

@@ -40,7 +40,7 @@ BACKUP_RESTORE_POLICY_SOURCE_HASH = (
)
_TOP_EVIDENCE_RE = re.compile(
r"<b>\s*Top evidence\s*</b>(?P<body>.*?)(?:<b>\s*建議下一步\s*</b>|$)",
r"<b>\s*Top evidence\s*</b>(?P<body>.*?)(?:<b>\s*(?:建議下一步|AI 自動處置佇列|禁止事項)\s*</b>|$)",
re.IGNORECASE | re.DOTALL,
)
_TAG_RE = re.compile(r"<[^>]+>")

44
apps/api/src/services/controlled_alert_target_router.py
View File

@@ -63,6 +63,10 @@ _NODE_EXPORTER_ALERTS = {
"nodeexporterunhealthy",
}
_ALERTMANAGER_DELIVERY_ALERTS = {"alertchainbrokenalertmanager"}
_PROVIDER_FRESHNESS_ALERTS = {
"providerfreshnesssignal",
"sourceprovideringestionstale",
}
_OLLAMA_LOCAL_RUNTIME_ALERTS = {
"ollamalocalunavailable",
"ollama111unavailable",
@@ -429,6 +433,46 @@ def resolve_typed_alert_target(
execution_role="single_writer_alert_chain_container_executor",
)
if (
compact_alert in _PROVIDER_FRESHNESS_ALERTS
or "source_provider_freshness_triage" in text
or "alertchain_provider_freshness" in text
):
provider = _text_token(
str(
labels.get("source")
or labels.get("provider")
or labels.get("service")
or ""
)
)
if provider not in {"sentry", "signoz"}:
provider = "sentry-signoz"
route = _typed_route(
resolution_status="resolved",
target_kind="observability_source_ingestion",
target_resource=target_resource or "source-provider-ingestion",
namespace=namespace,
canonical_asset_id=f"observability-provider:{provider}",
host=None,
executor="provider_freshness_readonly_executor",
verifier="provider_freshness_independent_verifier",
risk_class="low",
controlled_apply_allowed=True,
execution_role=(
"single_writer_observability_source_ingestion_readonly_executor"
),
)
route["provider_freshness_contract"] = {
"schema_version": "provider_freshness_typed_route_v1",
"execution_mode": "read_only",
"required_providers": ["sentry", "signoz"],
"allowed_actions": ["provider_specific_coverage_readback"],
"mutating_actions_allowed": False,
"generic_recurrence_can_resolve": False,
}
return route
if compact_alert in _OLLAMA_LOCAL_RUNTIME_ALERTS:
return _typed_route(
resolution_status="resolved",

110
apps/api/src/services/telegram_gateway.py
View File

@@ -469,13 +469,13 @@ _AI_OPS_SIGNAL_SPECS: tuple[AiOpsSignalSpec, ...] = (
),
gate_tokens=("stale", "freshness", "window", "provider", "ingestion", "timeout"),
impact="觀測來源可能過期或 ingestion 中斷；不能把 route 200 或 dashboard up 當健康",
next_step="建立 provider freshness candidate，要求 provider window、last seen、direct ref 與 verifier readback",
next_step="同一 run 讀回 Sentry / SignOz provider coverage、計算 freshness age，接 independent verifier / writeback",
recommendation_lines=(
"├ 對齊 Sentry / SigNoz / provider freshness window 與 direct reference",
"├ 標出缺資料、過期、timeout 與 no-false-green 風險",
"└ 回寫報表資料信任度與 AgentOps readiness，不直接切 provider",
"├ Evidence：分別驗證 Sentry / SignOz provider window、last seen 與 direct reference",
"├ Verify：兩者都 fresh 才能 resolved；任一 stale / timeout 保持 degraded 並自動重試",
"└ Writeback：回寫 incident、KM/RAG/MCP/PlayBook receipt；不直接切 provider",
),
forbidden_summary="不切 provider、不呼叫付費模型、不改 env、不 reload；先補 freshness evidence。",
forbidden_summary="不切 provider、不呼叫付費模型、不改 env、不 reload；generic recurrence 不得作為 resolved 證據。",
),
AiOpsSignalSpec(
event_type="supply_chain_drift",
@@ -596,8 +596,16 @@ def _aiops_signal_evidence_lines(text: str) -> list[str]:
return lines
def format_aiops_signal_alert_card(text: str) -> str:
"""把資安 / 維運 raw signal 轉成 AI 自動化事件包。"""
def format_aiops_signal_alert_card(
text: str,
*,
lifecycle: Mapping[str, object] | None = None,
) -> str:
"""把 raw signal 轉成 fail-closed AI lifecycle 首卡。
沒有 durable run / executor / verifier receipt 時只能顯示 partial；首卡不是
terminal，也不得把自動化缺口包裝成人工「建議下一步」。
"""
spec = _classify_aiops_signal_alert(text)
if spec is None:
return text
@@ -605,6 +613,82 @@ def format_aiops_signal_alert_card(text: str) -> str:
target = _aiops_signal_target(text, spec)
severity = _aiops_signal_severity(text, spec)
evidence_lines = _aiops_signal_evidence_lines(text)
queued_action_lines = tuple(
f"├{line[1:]}" if line.startswith("└") else line
for line in spec.recommendation_lines
)
lifecycle_data = lifecycle if isinstance(lifecycle, Mapping) else {}
provided_run_id = str(lifecycle_data.get("run_id") or "").strip()
provided_receipt_status = str(
lifecycle_data.get("receipt_status") or ""
).strip()
provided_verifier_status = str(
lifecycle_data.get("verifier_status") or ""
).strip()
provided_lifecycle_status = str(
lifecycle_data.get("status") or ""
).strip()
requested_terminal = bool(
lifecycle_data.get("terminal") is True
or provided_lifecycle_status.lower()
in {"completed", "closed", "resolved", "verified_resolved"}
)
terminal = bool(
requested_terminal
and provided_run_id
and provided_receipt_status
and "pending" not in provided_receipt_status.lower()
and provided_verifier_status
and "pending" not in provided_verifier_status.lower()
)
lifecycle_status = str(
(
provided_lifecycle_status
if not requested_terminal or terminal
else "partial"
)
or "partial"
).strip()
receipt_status = str(
(
provided_receipt_status
if not requested_terminal or terminal
else "receipt_pending"
)
or "receipt_pending"
).strip()
run_id = provided_run_id or "receipt_pending"
check_status = str(
lifecycle_data.get("check_status")
or "same_run_check_running_receipt_pending"
).strip()
provided_execution_summary = str(
lifecycle_data.get("execution_summary") or ""
).strip()
execution_summary = str(
(
provided_execution_summary
if not requested_terminal or terminal
else ""
)
or (
"已收到 terminal receipt"
if terminal
else "已建立 Run／AI check執行中"
)
).strip()
verifier_status = str(
(
provided_verifier_status
if not requested_terminal or terminal
else ""
)
or (
"same_run_terminal_receipt_received"
if terminal
else "same_run_verifier_receipt_pending"
)
).strip()
gate_line = (
"├ Gate：<code>controlled_playbook_queue=readback_only</code> / "
"<code>runtime_write_gate=0_until_verifier</code>"
@@ -622,18 +706,22 @@ def format_aiops_signal_alert_card(text: str) -> str:
f"├ 事件類型：<code>{html.escape(spec.event_type)}</code>",
f"├ Target：<code>{html.escape(target)}</code>",
f"├ 影響：<b>{html.escape(spec.impact)}</b>",
"└ 產品判讀：<code>raw_signal_normalized</code> / <code>controlled_runtime_candidate</code>",
f"└ 自動化狀態：<code>{html.escape(lifecycle_status)}</code> / <code>{html.escape(receipt_status)}</code>",
"",
"<b>AI 自動化判讀</b>",
f"├ Lane：<code>{html.escape(spec.lane)}</code>",
gate_line,
f"└ 下一步：{html.escape(spec.next_step)}",
f"├ Run：<code>{html.escape(run_id)}</code>",
f"└ 執行：<b>{html.escape(execution_summary)}</b>",
"",
"<b>Top evidence</b>",
*evidence_lines,
"",
"<b>建議下一步</b>",
*spec.recommendation_lines,
"<b>AI 自動處置佇列</b>",
f"├ Check：<code>{html.escape(check_status)}</code>",
f"├ 自動後續：{html.escape(spec.next_step)}",
*queued_action_lines,
f"└ Verify / Writeback：<code>{html.escape(verifier_status)}</code>；未收到獨立 verifier receipt 前保持 <code>partial</code>",
"",
"<b>禁止事項</b>",
f"└ {html.escape(spec.forbidden_summary)}",

37
apps/api/tests/test_agent99_control_plane_outcome_contract.py
View File

@@ -67,6 +67,43 @@ def test_agent99_outcome_contract_has_runtime_self_test() -> None:
assert '$securityCandidate.state -eq "candidate_ready"' in source
def test_agent99_provider_freshness_is_provider_specific_and_fail_closed() -> None:
source = _CONTROL_PLANE.read_text(encoding="utf-8")
converter = source[
source.index("function Convert-AgentProviderFreshnessCoverageReadback") :
source.index("function Invoke-AgentProviderFreshnessDirectReadback")
]
direct_readback = source[
source.index("function Invoke-AgentProviderFreshnessDirectReadback") :
source.index("function Invoke-AgentProviderFreshnessTriage")
]
synthetic = (
_REPO_ROOT / "agent99-synthetic-tests.ps1"
).read_text(encoding="utf-8")
assert '$requiredProviders = @("sentry", "signoz")' in converter
assert 'schemaVersion = "agent99_provider_freshness_readback_v2"' in converter
assert 'readbackScope = "provider_specific_coverage"' in converter
assert "genericRecurrenceAccepted = $false" in converter
assert "ageMinutes -le $LookbackMinutes" in converter
assert "allProvidersFresh = $allProvidersFresh" in converter
assert "/dossier/coverage?" in direct_readback
assert "provider=$provider" in direct_readback
assert "/dossier/recurrence" not in direct_readback
assert '$providerReadback.allProvidersFresh -eq $true' in source
assert 'provider_freshness_evidence_ref' in source
for case in (
"provider_freshness:mixed_fresh_stale_degraded",
"provider_freshness:both_fresh_resolved",
"provider_freshness:generic_recurrence_not_resolved",
):
assert case in synthetic
assert '$providerMixedDegraded.state -eq "degraded"' in source
assert '$providerFreshResolved.state -eq "resolved"' in source
assert '$providerGenericDegraded.state -eq "degraded"' in source
def test_agent99_posts_fenced_outcome_to_production_ingestion() -> None:
source = _CONTROL_PLANE.read_text(encoding="utf-8")
inbox = _SRE_INBOX.read_text(encoding="utf-8")

25
apps/api/tests/test_sre_typed_domain_router.py
View File

@@ -124,6 +124,31 @@ def test_other_alert_chain_domains_do_not_fall_into_alertmanager_container() ->
assert route["cross_domain_fallback_allowed"] is False
def test_provider_freshness_routes_to_read_only_observability_lane() -> None:
route = resolve_typed_alert_target(
alertname="SourceProviderIngestionStale",
target_resource="source-ingestion",
namespace="monitoring",
labels={"component": "source-ingestion", "source": "signoz"},
alert_category="alertchain_provider_freshness",
)
assert route["resolution_status"] == "resolved"
assert route["target_kind"] == "observability_source_ingestion"
assert route["canonical_asset_id"] == "observability-provider:signoz"
assert route["executor"] == "provider_freshness_readonly_executor"
assert route["verifier"] == "provider_freshness_independent_verifier"
assert route["risk_class"] == "low"
assert route["controlled_apply_allowed"] is True
assert route["cross_domain_fallback_allowed"] is False
assert route["agent99_bridge"]["dispatch_allowed"] is False
contract = route["provider_freshness_contract"]
assert contract["execution_mode"] == "read_only"
assert contract["required_providers"] == ["sentry", "signoz"]
assert contract["mutating_actions_allowed"] is False
assert contract["generic_recurrence_can_resolve"] is False
def _sentry_runtime_incident() -> SimpleNamespace:
return SimpleNamespace(
incident_id="INC-20260714-471307",

10
apps/api/tests/test_telegram_gateway_error_sanitizer.py
View File

@@ -266,6 +266,16 @@ Authorization: Bearer abcdefghijklmnopqrstuvwxyz
"""
card = format_aiops_signal_alert_card(raw_alert)
assert "已建立 Run／AI check執行中" in card
assert "partial" in card
assert "receipt_pending" in card
assert "same_run_check_running_receipt_pending" in card
assert "same_run_verifier_receipt_pending" in card
assert "AI 自動處置佇列" in card
assert "兩者都 fresh 才能 resolved" in card
assert "generic recurrence 不得作為 resolved 證據" in card
assert "建議下一步" not in card
payload = _agent99_alert_from_ai_automation_card(
card,
source="unit-test",

24
apps/api/tests/test_telegram_message_templates.py
View File

@@ -299,6 +299,13 @@ def test_aiops_signal_formatter_covers_non_host_alert_lanes(
assert "ai_automation_alert_card_v1" in result
assert event_type in result
assert lane in result
assert "已建立 Run／AI check執行中" in result
assert "partial" in result
assert "receipt_pending" in result
assert "same_run_check_running_receipt_pending" in result
assert "same_run_verifier_receipt_pending" in result
assert "AI 自動處置佇列" in result
assert "建議下一步" not in result
assert "controlled_playbook_queue" in result
if event_type == "backup_restore_escrow_signal":
assert "controlled_playbook_queue=readback_only" in result
@@ -310,6 +317,23 @@ def test_aiops_signal_formatter_covers_non_host_alert_lanes(
assert "禁止事項" in result
def test_aiops_signal_terminal_claim_requires_complete_same_run_receipts() -> None:
result = format_aiops_signal_alert_card(
'alertname="SourceProviderIngestionStale" source="signoz" stale',
lifecycle={
"status": "resolved",
"terminal": True,
"execution_summary": "resolved without receipts",
},
)
assert "已建立 Run／AI check執行中" in result
assert "resolved without receipts" not in result
assert "<code>partial</code> / <code>receipt_pending</code>" in result
assert "same_run_verifier_receipt_pending" in result
assert "建議下一步" not in result
@pytest.mark.asyncio
async def test_send_alert_notification_normalizes_host_resource_raw_dump(monkeypatch) -> None:
"""send_alert_notification 是最後出口，必須自動套用 AI 自動化事件包。"""