fix(awooop): persist signal metadata and auto-repair prestate
Some checks failed
Code Review / ai-code-review (push) Successful in 11s
CD Pipeline / tests (push) Successful in 1m14s
CD Pipeline / build-and-deploy (push) Failing after 3m33s
CD Pipeline / post-deploy-checks (push) Has been skipped

This commit is contained in:
Your Name
2026-05-18 10:59:54 +08:00
parent 5c240744eb
commit 1a2b04f5cf
8 changed files with 302 additions and 37 deletions

View File

@@ -297,6 +297,45 @@ class EvidenceSnapshot:
)
raise
async def update_pre_execution(
self,
pre_state: dict[str, Any],
) -> None:
"""
補填執行前狀態。
Auto-repair 背景路徑會在 playbook 執行前呼叫
PostExecutionVerifier.capture_pre_execution_state(),同一份 evidence row
後續再寫入 post_execution_state / verification_result讓 truth-chain 能看出
「執行前 → 執行後」是否真的改善。
"""
self.pre_execution_state = pre_state
try:
async with get_db_context() as db:
stmt_result = await db.execute(
update(IncidentEvidence)
.where(IncidentEvidence.id == self.snapshot_id)
.values(pre_execution_state=pre_state)
)
if stmt_result.rowcount < 1:
logger.warning(
"evidence_snapshot_pre_update_no_rows",
snapshot_id=self.snapshot_id,
)
else:
logger.info(
"evidence_snapshot_pre_execution_updated",
snapshot_id=self.snapshot_id,
)
except Exception:
logger.exception(
"evidence_snapshot_pre_update_error",
snapshot_id=self.snapshot_id,
)
raise
async def update_self_healing(
self,
score: float,

View File

@@ -25,11 +25,106 @@ from src.core.redis_client import get_redis
from src.db.base import get_db_context
from src.db.models import IncidentRecord
from src.models.incident import Incident
from src.utils.incident_converter import brain_to_local
logger = structlog.get_logger(__name__)
def _signal_to_dict(signal: Any) -> dict[str, Any]:
"""Normalize brain/local Signal objects and raw dicts into one shape."""
if isinstance(signal, dict):
return signal
if hasattr(signal, "model_dump"):
return signal.model_dump(mode="json")
return {
"alert_name": getattr(signal, "alert_name", None),
"severity": getattr(signal, "severity", None),
"source": getattr(signal, "source", None),
"labels": getattr(signal, "labels", None) or {},
"annotations": getattr(signal, "annotations", None) or {},
"fingerprint": getattr(signal, "fingerprint", None),
}
def _derive_incident_alert_metadata(incident: Incident) -> dict[str, Any]:
"""Derive alert metadata for incidents saved through the lewooogo bridge."""
first_signal = incident.signals[0] if incident.signals else None
signal = _signal_to_dict(first_signal) if first_signal else {}
labels = signal.get("labels") or {}
annotations = signal.get("annotations") or {}
alertname = (
labels.get("alertname")
or signal.get("alert_name")
or signal.get("alertname")
or ""
)
severity = (
signal.get("severity")
or getattr(incident.severity, "value", incident.severity)
or labels.get("severity")
or "warning"
)
severity = getattr(severity, "value", severity)
alert_category = None
notification_type = None
if alertname:
from src.services.incident_service import classify_alert_early
alert_category, notification_type = classify_alert_early(
str(alertname),
str(severity),
labels,
)
description = (
annotations.get("message")
or annotations.get("description")
or annotations.get("summary")
or ""
)
return {
"alertname": str(alertname) if alertname else None,
"severity": str(severity) if severity else None,
"alert_category": alert_category,
"notification_type": notification_type,
"description": str(description) if description else None,
"actor": signal.get("source") or labels.get("source") or "signal_worker",
}
async def _add_signal_timeline_event(
incident: Incident,
metadata: dict[str, Any],
) -> None:
"""Best-effort timeline seed for incidents created outside Alertmanager."""
alertname = metadata.get("alertname")
if not alertname:
return
try:
from src.services.approval_db import get_timeline_service
await get_timeline_service().add_event(
event_type="webhook",
status="success",
title=f"Signal received: {alertname}",
description=metadata.get("description"),
actor=metadata.get("actor"),
actor_role="signal_worker",
risk_level=getattr(incident.severity, "value", incident.severity),
incident_id=incident.incident_id,
)
except Exception as exc:
logger.warning(
"incident_signal_timeline_seed_failed",
incident_id=incident.incident_id,
alertname=alertname,
error=str(exc),
)
# =============================================================================
# Phase 16: IncidentDbAdapter (DI 注入實現)
# =============================================================================
@@ -63,6 +158,8 @@ class IncidentDbAdapter:
async def save(self, incident: Incident) -> bool:
"""儲存 Incident 到 PostgreSQL (upsert)"""
metadata = _derive_incident_alert_metadata(incident)
created = False
try:
async with get_db_context() as db:
from sqlalchemy import select
@@ -85,6 +182,12 @@ class IncidentDbAdapter:
existing.resolved_at = incident.resolved_at
if incident.closed_at:
existing.closed_at = incident.closed_at
if metadata.get("alertname") and not existing.alertname:
existing.alertname = metadata["alertname"]
if metadata.get("notification_type") and not existing.notification_type:
existing.notification_type = metadata["notification_type"]
if metadata.get("alert_category") and not existing.alert_category:
existing.alert_category = metadata["alert_category"]
else:
record = IncidentRecord(
incident_id=incident.incident_id,
@@ -111,9 +214,15 @@ class IncidentDbAdapter:
closed_at=incident.closed_at,
ttl_days=getattr(incident, 'ttl_days', 30),
vectorized=getattr(incident, 'vectorized', False),
alertname=metadata.get("alertname"),
notification_type=metadata.get("notification_type"),
alert_category=metadata.get("alert_category"),
)
db.add(record)
created = True
if created:
await _add_signal_timeline_event(incident, metadata)
logger.debug("db_adapter_save_success", incident_id=incident.incident_id)
return True

View File

@@ -208,10 +208,28 @@ class PostExecutionVerifier:
timeout=TOOL_TIMEOUT_SEC,
)
snapshot.pre_execution_state = state
try:
await snapshot.update_pre_execution(state)
except Exception as exc:
logger.warning(
"verifier_pre_state_persist_failed",
incident_id=incident_id,
snapshot_id=snapshot.snapshot_id,
error=str(exc),
)
logger.debug("verifier_pre_state_captured", incident_id=incident_id)
except Exception:
logger.warning("verifier_pre_state_failed", incident_id=incident_id)
snapshot.pre_execution_state = {}
try:
await snapshot.update_pre_execution({})
except Exception as exc:
logger.warning(
"verifier_empty_pre_state_persist_failed",
incident_id=incident_id,
snapshot_id=snapshot.snapshot_id,
error=str(exc),
)
async def _collect_post_state(self, incident: "Incident") -> dict[str, Any]:
"""