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

feat(sre): add 24h alert learning corpus

Browse Source
This commit is contained in:
Your Name
2026-07-22 21:54:31 +08:00
parent df50718ac5
commit 3243bc33e0
5 changed files with 467 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

336
apps/api/src/services/automation_slo_scorecard.py
View File

@@ -2,6 +2,8 @@
from __future__ import annotations
import hashlib
import json
import re
from collections.abc import Mapping, Sequence
from datetime import UTC, datetime, timedelta
@@ -25,6 +27,8 @@ _INCIDENT_SAMPLES_SQL = text(
COALESCE(operation.incident_id, operation.id::text) AS sample_id,
operation.event_type::text AS event_type,
operation.created_at,
operation.action_detail,
operation.success,
operation.context,
incident.resolved_at AS incident_resolved_at,
COALESCE(
@@ -43,6 +47,66 @@ _INCIDENT_SAMPLES_SQL = text(
NULLIF(operation.context ->> 'project_id', ''),
'awoooi'
) = :project_id
), execution_evidence AS (
SELECT
*,
(
event_type IN (
'EXECUTION_STARTED', 'EXECUTION_COMPLETED',
'CHANGE_APPLIED'
)
) AS execution_like_event,
(
event_type = 'CHANGE_APPLIED'
OR lower(COALESCE(context ->> 'execution_mode', ''))
IN ('production', 'controlled_apply')
OR lower(COALESCE(context ->> 'execution_kind', ''))
= 'executable'
OR lower(COALESCE(context ->> 'side_effect_performed', ''))
IN ('true', '1', 'yes')
OR lower(COALESCE(context ->> 'repair_executed', ''))
IN ('true', '1', 'yes')
OR lower(COALESCE(
context ->> 'controlled_apply_executed', ''
)) IN ('true', '1', 'yes')
OR lower(COALESCE(context ->> 'runtime_apply_executed', ''))
IN ('true', '1', 'yes')
OR lower(COALESCE(action_detail, '')) LIKE 'controlled_apply_%'
) AS production_execution_evidence,
(
lower(COALESCE(context ->> 'is_shadow', 'false'))
IN ('true', '1', 'yes')
OR lower(COALESCE(context ->> 'execution_mode', '')) IN (
'shadow', 'canary', 'check_mode', 'dry_run',
'read_only', 'no_write', 'no_write_replay',
'controlled_retry_check_mode_replay',
'stdin_boundary_check_mode_replay'
)
OR lower(COALESCE(context ->> 'execution_kind', ''))
IN ('shadow', 'no_action', 'no_write', 'no_write_replay')
OR lower(COALESCE(context ->> 'side_effect_performed', ''))
IN ('false', '0', 'no')
OR lower(COALESCE(action_detail, '')) LIKE ANY (ARRAY[
'%shadow%', '%no_write%', '%check_mode%', '%dry_run%'
])
) AS execution_exclusion_evidence
FROM scoped_events
), classified_events AS (
SELECT
*,
(
execution_like_event
AND production_execution_evidence
AND NOT execution_exclusion_evidence
) AS production_execution_event,
(
execution_like_event
AND NOT (
production_execution_evidence
AND NOT execution_exclusion_evidence
)
) AS excluded_execution_event
FROM execution_evidence
), per_incident AS (
SELECT
sample_id,
@@ -79,17 +143,24 @@ _INCIDENT_SAMPLES_SQL = text(
) AS recurrence,
bool_or(event_type IN ('USER_ACTION', 'MANUAL_FIX_RECORDED'))
AS human_intervention,
bool_or(event_type IN (
'AUTO_REPAIR_TRIGGERED', 'EXECUTION_STARTED',
'EXECUTION_COMPLETED', 'CHANGE_APPLIED'
)) AS execution_observed,
bool_or(event_type = 'AUTO_REPAIR_TRIGGERED') AS repair_observed,
bool_or(
event_type = 'AUTO_REPAIR_TRIGGERED' AND success IS TRUE
) AS repair_accepted,
bool_or(production_execution_event) AS execution_observed,
bool_or(excluded_execution_event)
AS shadow_or_no_write_execution_observed,
bool_or(
context ? 'verifier_passed'
OR context ? 'post_verifier_passed'
OR context ? 'verification_result'
) AS verifier_observed,
bool_or(
lower(COALESCE(context ->> 'verifier_passed', 'false'))
IN ('true', '1', 'yes')
OR lower(COALESCE(
context ->> 'post_verifier_passed', 'false'
)) IN ('true', '1', 'yes')
OR lower(COALESCE(context ->> 'verification_result', ''))
IN ('success', 'passed', 'verified')
) AS verifier_passed,
@@ -109,7 +180,7 @@ _INCIDENT_SAMPLES_SQL = text(
AND COALESCE(context ->> 'asset_identity_status', 'resolved')
<> 'asset_identity_unresolved'
) AS asset_identity_resolved
FROM scoped_events
FROM classified_events
GROUP BY sample_id
)
SELECT *
@@ -196,6 +267,21 @@ def _average(values: Sequence[float]) -> float | None:
return round(sum(values) / len(values), 2) if values else None
def _iso_timestamp(value: object) -> str | None:
parsed = _as_datetime(value)
return parsed.isoformat() if parsed else None
def _sha256_json(value: object) -> str:
canonical = json.dumps(
value,
ensure_ascii=True,
separators=(",", ":"),
sort_keys=True,
)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def _metrics_for(
samples: Sequence[Mapping[str, Any]],
) -> tuple[dict[str, Any], dict[str, int]]:
@@ -213,8 +299,11 @@ def _metrics_for(
total = len(samples)
false_observed = sum(_flag(row.get("false_positive_observed")) for row in samples)
recurrence_observed = sum(_flag(row.get("recurrence_observed")) for row in samples)
execution_observed = sum(_flag(row.get("execution_observed")) for row in samples)
verifier_observed = sum(_flag(row.get("verifier_observed")) for row in samples)
execution_samples = [row for row in samples if _flag(row.get("execution_observed"))]
execution_observed = len(execution_samples)
verifier_observed = sum(
_flag(row.get("verifier_observed")) for row in execution_samples
)
freshness_observed = sum(_flag(row.get("freshness_observed")) for row in samples)
denominators = {
"alert_count": total,
@@ -222,8 +311,17 @@ def _metrics_for(
"mttr_sample_count": len(mttr),
"false_positive_classified_count": false_observed,
"recurrence_classified_count": recurrence_observed,
"repair_observed_count": sum(
_flag(row.get("repair_observed")) for row in samples
),
"repair_accepted_count": sum(
_flag(row.get("repair_accepted")) for row in samples
),
"execution_observed_count": execution_observed,
"verifier_observed_count": verifier_observed,
"shadow_or_no_write_execution_excluded_count": sum(
_flag(row.get("shadow_or_no_write_execution_observed")) for row in samples
),
"freshness_observed_count": freshness_observed,
}
metrics = {
@@ -242,11 +340,15 @@ def _metrics_for(
total,
),
"verifier_pass_rate": _rate(
sum(_flag(row.get("verifier_passed")) for row in samples),
sum(
_flag(row.get("verifier_observed"))
and _flag(row.get("verifier_passed"))
for row in execution_samples
),
verifier_observed,
),
"rollback_rate": _rate(
sum(_flag(row.get("rollback_performed")) for row in samples),
sum(_flag(row.get("rollback_performed")) for row in execution_samples),
execution_observed,
),
"signal_freshness_slo_pass_rate": _rate(
@@ -267,6 +369,198 @@ def _metrics_for(
return metrics, denominators
def _build_learning_corpus(
*,
project_id: str,
incident_samples: Sequence[Mapping[str, Any]],
safe_runs: Mapping[str, int],
provider_usage: Mapping[str, Any],
window_start: datetime,
window_end: datetime,
) -> dict[str, Any]:
records: list[dict[str, Any]] = []
missing_source_identity_count = 0
for sample in incident_samples:
source_identity = str(sample.get("sample_id") or "").strip()
if not source_identity:
missing_source_identity_count += 1
identity_material = f"{project_id}\0{source_identity}".encode()
record_id = (
f"sha256:{hashlib.sha256(identity_material).hexdigest()}"
if source_identity
else ""
)
lane = str(sample.get("lane") or "unclassified")[:64]
execution_observed = _flag(sample.get("execution_observed"))
verifier_observed = execution_observed and _flag(
sample.get("verifier_observed")
)
recurrence_observed = _flag(sample.get("recurrence_observed"))
false_positive_observed = _flag(sample.get("false_positive_observed"))
freshness_observed = _flag(sample.get("freshness_observed"))
asset_identity_resolved = _flag(sample.get("asset_identity_resolved"))
received_at = _iso_timestamp(sample.get("received_at"))
resolved_at = _iso_timestamp(sample.get("resolved_at"))
learning_eligible = bool(
record_id
and received_at
and resolved_at
and lane != "unclassified"
and recurrence_observed
and false_positive_observed
and freshness_observed
and asset_identity_resolved
and (not execution_observed or verifier_observed)
)
records.append(
{
"record_id": record_id,
"lane": lane,
"timestamps": {
"alert_received_at": received_at,
"first_action_at": _iso_timestamp(sample.get("first_action_at")),
"resolved_at": resolved_at,
},
"outcomes": {
"repair_observed": _flag(sample.get("repair_observed")),
"repair_accepted": _flag(sample.get("repair_accepted")),
"production_execution_observed": execution_observed,
"shadow_or_no_write_execution_excluded": _flag(
sample.get("shadow_or_no_write_execution_observed")
),
"verifier_observed": verifier_observed,
"verifier_passed": (
_flag(sample.get("verifier_passed"))
if verifier_observed
else None
),
"recurrence_observed": recurrence_observed,
"recurrence": (
_flag(sample.get("recurrence")) if recurrence_observed else None
),
"false_positive_observed": false_positive_observed,
"false_positive": (
_flag(sample.get("false_positive"))
if false_positive_observed
else None
),
"human_intervention": _flag(sample.get("human_intervention")),
"rollback_performed": (
_flag(sample.get("rollback_performed"))
if execution_observed
else None
),
},
"quality": {
"signal_freshness_observed": freshness_observed,
"signal_freshness_slo_passed": (
_flag(sample.get("freshness_passed"))
if freshness_observed
else None
),
"asset_identity_resolved": asset_identity_resolved,
},
"learning_eligible": learning_eligible,
}
)
records.sort(
key=lambda row: (
str(row["timestamps"].get("alert_received_at") or ""),
str(row.get("record_id") or ""),
)
)
denominators = {
"alert_count": len(records),
"repair_observed_count": sum(
row["outcomes"]["repair_observed"] for row in records
),
"repair_accepted_count": sum(
row["outcomes"]["repair_accepted"] for row in records
),
"production_execution_observed_count": sum(
row["outcomes"]["production_execution_observed"] for row in records
),
"verifier_observed_execution_count": sum(
row["outcomes"]["verifier_observed"] for row in records
),
"recurrence_classified_count": sum(
row["outcomes"]["recurrence_observed"] for row in records
),
"controlled_non_shadow_run_count": int(
safe_runs.get("controlled_runs_with_steps") or 0
),
"provider_call_count": int(provider_usage.get("call_count") or 0),
}
excluded_from_execution = {
"shadow_run_count": max(
int(safe_runs.get("total_runs") or 0)
- int(safe_runs.get("non_shadow_runs") or 0),
0,
),
"non_shadow_no_step_run_count": max(
int(safe_runs.get("non_shadow_runs") or 0)
- int(safe_runs.get("controlled_runs_with_steps") or 0),
0,
),
"alert_count_with_shadow_or_no_write_execution_evidence": sum(
row["outcomes"]["shadow_or_no_write_execution_excluded"] for row in records
),
}
eligible_record_count = sum(row["learning_eligible"] for row in records)
status = (
"no_runtime_sample_fail_closed"
if not records
else "partial_missing_receipt_coverage"
if eligible_record_count < len(records) or missing_source_identity_count
else "learning_corpus_ready_same_run_verification_pending"
)
fingerprint_payload = {
"window": {
"start": window_start.astimezone(UTC).isoformat(),
"end": window_end.astimezone(UTC).isoformat(),
},
"records": records,
"denominators": denominators,
"excluded_from_execution": excluded_from_execution,
"provider_usage": provider_usage,
}
return {
"schema_version": "sre_alert_learning_corpus_v1",
"window": fingerprint_payload["window"],
"status": status,
"runtime_closed": False,
"read_only": True,
"raw_context_included": False,
"record_count": len(records),
"learning_eligible_record_count": eligible_record_count,
"missing_source_identity_count": missing_source_identity_count,
"records": records,
"denominators": denominators,
"provider_usage": dict(provider_usage),
"provider_cost_usd": str(provider_usage.get("cost_usd") or "0.0000"),
"excluded_from_execution": excluded_from_execution,
"recomputation_contract": {
"alert": "count(records)",
"repair": "count(records where outcomes.repair_observed=true)",
"execution": (
"count(records where outcomes.production_execution_observed=true); "
"shadow, canary, check-mode, dry-run and no-write are excluded"
),
"verifier": (
"count(executed records where outcomes.verifier_observed=true)"
),
"recurrence": ("count(records where outcomes.recurrence_observed=true)"),
"cost": (
"sum(provider_usage.rows.call_count) and "
"sum(provider_usage.rows.cost_usd)"
),
},
"corpus_sha256": _sha256_json(fingerprint_payload),
"same_run_receipt_verification": "pending",
}
def build_automation_slo_scorecard(
*,
project_id: str,
@@ -349,6 +643,21 @@ def build_automation_slo_scorecard(
"controlled_failed_runs",
)
}
provider_usage = {
"rows": costs,
"call_count": sum(row["call_count"] for row in costs),
"prompt_tokens": sum(row["prompt_tokens"] for row in costs),
"completion_tokens": sum(row["completion_tokens"] for row in costs),
"cost_usd": f"{total_cost:.4f}",
}
learning_corpus = _build_learning_corpus(
project_id=project_id,
incident_samples=incident_samples,
safe_runs=safe_runs,
provider_usage=provider_usage,
window_start=window_start,
window_end=window_end,
)
return {
"schema_version": SCHEMA_VERSION,
"project_id": project_id,
@@ -363,13 +672,8 @@ def build_automation_slo_scorecard(
"denominators": denominators,
"lane_scorecards": lane_scorecards,
"runs": safe_runs,
"provider_usage": {
"rows": costs,
"call_count": sum(row["call_count"] for row in costs),
"prompt_tokens": sum(row["prompt_tokens"] for row in costs),
"completion_tokens": sum(row["completion_tokens"] for row in costs),
"cost_usd": f"{total_cost:.4f}",
},
"provider_usage": provider_usage,
"learning_corpus": learning_corpus,
"missing_metrics": missing_metrics,
"coverage_blockers": coverage_blockers,
"evidence": {