From 06b0fec5fc73e087788ffbaa8d0d6f44c24daa53 Mon Sep 17 00:00:00 2001 From: CodexManvik Date: Wed, 23 Sep 2026 02:01:05 +0530 Subject: [PATCH] feat: suppression handles reliably-fixed incidents quietly instead of dropping them (ADR-034) A suppressed Situation used to be dropped before RCA and action, so the fault the system had learned to fix was the one it would no longer fix. - Correlation marks, it does not drop: a signature over the suppression threshold (and min samples) is emitted with Situation.handling="quiet" and still queued on situations.suppressed, with a log line either way. SUPPRESSION_MODE=drop restores the old behaviour. Its reliability map now refreshes every RELIABILITY_REFRESH_SECONDS instead of only at boot, and a baseline reset keeps it. - Action re-checks before waiving anything: the approval is skipped only if THIS playbook's real track record on THIS signature clears the same bar (common/evidence.py - dry runs and pre-`mode` records excluded). Otherwise it falls back to a normal approval. QUIET_SKIP_APPROVAL=false keeps quiet as a label only. - Only the HITL wait is waived. The disabled/reversible/RBAC/denylist gates still apply, and a failed pre-flight rehearsal becomes a hard stop. - Never out of sight: log lines in correlation and action (searchable in Grafana), a `quiet-handling` audit record carrying the evidence ("3/3 real runs worked") plus `quiet-approved` on the execute path, RemediationOutcome.handling="quiet", a console badge, and a quietlyHandled KPI. - Console: the reason explainer described the new unhealthy:no-rollback / unhealthy:rollback-failed outcomes as "nothing executed"; it now says the fix ran and a human must check. Co-Authored-By: Claude Opus 5.5 --- architectural.md | 55 ++++ common/config.py | 10 +- common/contracts.py | 8 + common/evidence.py | 71 +++++ deploy/k8s/platform/templates/configmap.yaml | 4 + deploy/k8s/platform/values.yaml | 5 + flow.md | 2 +- frontend/src/data/mock.ts | 8 +- frontend/src/data/types.ts | 8 + frontend/src/views/Incidents.tsx | 23 ++ services/action/app.py | 6 + services/action/consumer.py | 67 +++++ services/action/remediate.py | 14 +- services/action/tests/test_quiet_handling.py | 248 ++++++++++++++++++ services/correlation/app.py | 33 ++- services/correlation/consumer.py | 13 + services/correlation/engine.py | 28 +- .../correlation/tests/test_engine_suppress.py | 30 +++ services/correlation/tests/test_retrain.py | 40 +++ services/read/projection.py | 7 + services/read/tests/test_metrics.py | 1 + services/read/tests/test_occurrences.py | 10 + tests/test_evidence.py | 49 ++++ 23 files changed, 730 insertions(+), 10 deletions(-) create mode 100644 common/evidence.py create mode 100644 services/action/tests/test_quiet_handling.py create mode 100644 tests/test_evidence.py diff --git a/architectural.md b/architectural.md index 18bd73d..0af6f4d 100644 --- a/architectural.md +++ b/architectural.md @@ -1749,6 +1749,61 @@ surfaced `interrupted:unknown` rather than a silent double execution. --- +### ADR-034 — Suppression means "handle quietly", not "drop" + +**Context.** The closed loop's promise was: a signature the system has reliably fixed stops +paging people. The implementation did something stronger. A suppressed Situation was never +emitted, so it never reached RCA or action, and **the fix never ran**. The fault the system +had learned to fix was the one fault it would no longer fix. This stayed latent for a long +time, because suppression rarely fired. Signatures depended on how many samples landed in a +window, and correlation's reliability map loaded only at boot. Once signatures became stable +(the set of series, not the multiset of samples), suppression started to fire as coded, and +the gap became real. + +**Decision.** Suppression now means **handle quietly**: remediate without paging a human, and +record that it happened. + +- **Correlation marks, it does not drop.** A signature that clears + `reliability_suppress_threshold` over at least `reliability_suppress_min_samples` labelled + outcomes is emitted with `Situation.handling="quiet"` (an additive field), and is still + queued on `situations.suppressed` for the counter. `suppression_mode=drop` keeps the old + behaviour. The reliability map now refreshes every `reliability_refresh_seconds` instead of + only at boot. Suppression no longer loses fixes, so it no longer has to be starved. +- **Quiet is a request; action decides.** Correlation's map is per signature, but the risk is + per playbook. So action re-checks **that playbook's real track record on that signature** + (`common/evidence.py`: at least the same minimum sample count and threshold, with dry runs + and records predating `TrainingRecord.mode` excluded). Only if it clears does the HITL + approval get waived (`quiet_skip_approval`, default on). Otherwise the situation goes + through the normal approval flow. +- **The waiver is the only thing that changes.** The disabled, reversible, RBAC and + destructive-shape gates all still apply. A failed pre-flight rehearsal becomes a hard stop, + as it already was for AUTO, because no human is there to weigh it. +- **Never out of sight.** There are three records of every quiet decision: + - a structured log line in correlation when the situation is marked, and in action when the + waiver is granted or refused (both searchable in Grafana); + - an audit record `quiet-handling` carrying the evidence (e.g. `3/3 real runs worked`), + plus `quiet-approved` on the execute path; + - `RemediationOutcome.handling="quiet"`, which shows up as a badge on the console card and + in the `quietlyHandled` KPI. + +**Consequences.** (+) The loop's promise holds as stated: less paging, the same fixes. (+) The +waiver is self-correcting: a quiet fix that fails is labelled like any other, lowers the +track record, and the next occurrence goes back to a human. (+) Only real runs count, the same +rule graduation uses, so compose's dry-run posture never waives an approval. (−) This is a +second path to unattended execution beside graduation. Graduation is per playbook across every +signature; this is per (signature, playbook), and so narrower. Both paths use the same evidence +rule and the same gates. (−) On a real cluster a fault's first `min_samples` occurrences still +need a human. That is deliberate. + +**Alternatives rejected.** *Keep dropping.* It contradicts the feature's own description. +*Quiet only as a label* (still asking a human). A "quiet" incident that pages someone is not +quiet, and without anyone watching it would time out and hold the serial action consumer for +the whole HITL window. That remains available as `quiet_skip_approval=false`. *Trust +correlation's per-signature figure.* It isn't specific to the playbook RCA picks, and it +included dry runs. + +--- + ## 4. Cross-cutting concerns **Traceability.** A `correlation_id` is threaded through every `AuditRecord`, so one diff --git a/common/config.py b/common/config.py index 9329287..d667ce9 100644 --- a/common/config.py +++ b/common/config.py @@ -22,8 +22,16 @@ class Settings(BaseSettings): reliability_suppress_threshold: float = 0.8 # Labelled outcomes a signature needs before it can be suppressed at all. reliability_suppress_min_samples: int = 3 - # How often RCA re-reads the training store for its runbook track records. + # How often RCA and correlation re-read the training store for track records. reliability_refresh_seconds: float = 60.0 + # What suppression does to a signature the system has reliably fixed: + # "quiet" (default) still diagnoses and remediates it, without paging a human + # when the playbook's own real track record clears the bar below; "drop" is + # the historical behaviour - never emitted, so never fixed. + suppression_mode: str = "quiet" # "quiet" | "drop" + # Whether a quiet situation whose playbook qualifies may skip the HITL + # approval. False keeps quiet handling to logging/labelling only. + quiet_skip_approval: bool = True graduation_min_successes: int = 3 # Whether a dry-run SUCCESS counts toward graduating a playbook to AUTO. Off: # a simulation always "succeeds", so it is no evidence a fix is safe unattended. diff --git a/common/contracts.py b/common/contracts.py index c0f0bee..11fcbd0 100644 --- a/common/contracts.py +++ b/common/contracts.py @@ -77,6 +77,11 @@ class Situation(BaseModel): signature: str peak_score: float | None = None # correlator max z-score for the window baseline: dict | None = None # per-metric {name: {mean, std}} at emit time + # "quiet": correlation has seen this signature fixed reliably and asks for it + # to be handled without paging a human. A REQUEST, not a permission - action + # re-checks the evidence for the specific playbook before skipping approval, + # and every other safety gate still applies. Additive; defaults to "normal". + handling: str = "normal" # "normal" | "quiet" class RootCauseHypothesis(BaseModel): @@ -216,6 +221,9 @@ class RemediationOutcome(BaseModel): steps: list[str] = Field(default_factory=list) mode: str = "dry_run" # "dry_run" | "k8s" | "none" (escalated: no executor ran) preflight: PreflightResult | None = None + # "quiet" when the fix ran without a human approval on the strength of its + # track record (see Situation.handling). Additive; defaults to "normal". + handling: str = "normal" class AuditRecord(BaseModel): diff --git a/common/evidence.py b/common/evidence.py new file mode 100644 index 0000000..38fc56c --- /dev/null +++ b/common/evidence.py @@ -0,0 +1,71 @@ +"""What counts as evidence that a fix works - shared by correlation and action. + +Two decisions rest on a remediation track record: + +- correlation marks a situation `handling="quiet"` when its signature has been + fixed reliably, and +- action lets a quiet situation's playbook run WITHOUT a human approval, but + only after re-checking that track record for that exact playbook. + +Both must judge evidence the same way, or correlation would keep asking for +quiet handling that action keeps refusing. Hence one module. + +Only real runs count. DryRunRemediator + AlwaysHealthyChecker report success +unconditionally, and records written before TrainingRecord carried `mode` (None) +cannot be told apart from dry runs - so neither is evidence that a change to a +real cluster is safe to make unattended. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +_NOT_REAL = frozenset({None, "dry_run", "none"}) + + +def _field(record, name): + return record.get(name) if isinstance(record, dict) else getattr(record, name, None) + + +def is_real(record) -> bool: + return _field(record, "mode") not in _NOT_REAL + + +def real_records(records) -> list: + return [r for r in records if is_real(r)] + + +@dataclass(frozen=True) +class TrackRecord: + worked: int + total: int + + @property + def ratio(self) -> float: + return self.worked / self.total if self.total else 0.0 + + def __str__(self) -> str: + return f"{self.worked}/{self.total} real runs worked" + + +def track_record(records, signature: str, playbook_id: str | None = None) -> TrackRecord: + """Real outcomes for `signature`, optionally narrowed to one playbook.""" + worked = total = 0 + for r in records: + if not is_real(r) or _field(r, "signature") != signature: + continue + if playbook_id is not None and _field(r, "playbook_id") != playbook_id: + continue + total += 1 + if _field(r, "worked"): + worked += 1 + return TrackRecord(worked, total) + + +def quiet_eligible( + records, signature: str, playbook_id: str, threshold: float, min_samples: int +) -> tuple[bool, TrackRecord]: + """May this playbook run on this signature without a human approval?""" + record = track_record(records, signature, playbook_id) + ok = record.total >= max(1, min_samples) and record.ratio >= threshold + return ok, record diff --git a/deploy/k8s/platform/templates/configmap.yaml b/deploy/k8s/platform/templates/configmap.yaml index 29524f8..da471a9 100644 --- a/deploy/k8s/platform/templates/configmap.yaml +++ b/deploy/k8s/platform/templates/configmap.yaml @@ -45,6 +45,10 @@ data: INTELLIOPS_CORRELATION_ROBUST_WINDOW: {{ .Values.env.CORRELATION_ROBUST_WINDOW | default "128" | quote }} INTELLIOPS_CORRELATION_ROBUST_WARMUP: {{ .Values.env.CORRELATION_ROBUST_WARMUP | default "30" | quote }} INTELLIOPS_CORRELATION_ROBUST_KEY_BY: {{ .Values.env.CORRELATION_ROBUST_KEY_BY | default "metric" | quote }} + # ADR-034: what suppression does to a reliably-fixed signature, and whether a + # qualifying playbook may then run without an approval request. + INTELLIOPS_SUPPRESSION_MODE: {{ .Values.env.SUPPRESSION_MODE | default "quiet" | quote }} + INTELLIOPS_QUIET_SKIP_APPROVAL: {{ .Values.env.QUIET_SKIP_APPROVAL | default "true" | quote }} INTELLIOPS_DETECTION_POLICY: {{ .Values.env.DETECTION_POLICY | quote }} INTELLIOPS_RUNBOOK_SELECTOR_MODE: {{ .Values.env.RUNBOOK_SELECTOR_MODE | quote }} INTELLIOPS_REMEDIATOR_MODE: {{ .Values.env.REMEDIATOR_MODE | quote }} diff --git a/deploy/k8s/platform/values.yaml b/deploy/k8s/platform/values.yaml index 221d2e0..2e855c6 100644 --- a/deploy/k8s/platform/values.yaml +++ b/deploy/k8s/platform/values.yaml @@ -79,6 +79,11 @@ env: # every service into one per metric. Re-tune warm-up before flipping it: a # per-series window fills ~5x slower (one sample per poll, not five). CORRELATION_ROBUST_KEY_BY: "metric" + # ADR-034. "quiet": a reliably-fixed signature is still remediated, without an + # approval request when its playbook's REAL track record qualifies (audited + + # logged). "drop": the old behaviour - never emitted, never fixed. + SUPPRESSION_MODE: "quiet" + QUIET_SKIP_APPROVAL: "true" GOVERNANCE_MODE: "in_process" # "http" in values-live — action + governance are GOVERNANCE_URL: "http://governance:8000" # separate pods, so approvals must go over HTTP HITL_POLL_TIMEOUT_SECONDS: "30" # values-live raises this so a human has time to click Approve diff --git a/flow.md b/flow.md index 17303ee..0cd15b4 100644 --- a/flow.md +++ b/flow.md @@ -107,7 +107,7 @@ instead of waiting out a poll interval ([ADR-018](architectural.md#adr-018--real | `telemetry.raw` | ingestion | correlation | `TelemetryEvent` | a normalized signal | | `situations.detected` | correlation | rca, **read** | `Situation` (detected) | an alert storm collapsed into one incident | | `situations.diagnosed` | rca | action, **read** | `DiagnosedSituation` (situation + hypotheses) | incident with likely cause + suggested fix | -| `situations.suppressed` | correlation | **read** | `Situation` (suppressed) | a signature that reliably self-heals — detected, then *not* emitted as an incident (closed-loop suppression, made visible for metrics) | +| `situations.suppressed` | correlation | **read** | `Situation` (suppressed) | a signature the system has reliably fixed. Under the default `suppression_mode=quiet` it is ALSO emitted on `situations.detected` with `handling="quiet"` and remediated without paging a human when the playbook's real track record qualifies; this topic is the record of it ([ADR-034](architectural.md#adr-034--suppression-means-handle-quietly-not-drop)) | | `remediation.outcomes` | action | feedback, **read** | `RemediationOutcome` | did the fix work / roll back | | `audit.events` | all services | governance | `AuditRecord` | append-only trail (fire-and-forget) | diff --git a/frontend/src/data/mock.ts b/frontend/src/data/mock.ts index 330e451..d4668c5 100644 --- a/frontend/src/data/mock.ts +++ b/frontend/src/data/mock.ts @@ -162,16 +162,20 @@ export const situations: Situation[] = [ { description: "Latency/queueing under load — capacity contention", confidence: 0.55, suggested_runbook_id: "scale-service", confidence_source: "rule" }, ], suggested_runbook_id: "restart-pod", - hitl_mode: "auto", + hitl_mode: "hitl", reversible: true, reliability: 0.86, suppressed: false, + // A signature fixed reliably before: remediated without an approval request + // on restart-pod's real track record (still audited) - quiet handling. + handling: "quiet", outcome: { result: "success", health_after: "healthy", - mode: "dry_run", + mode: "k8s", steps: ["restart"], preflight: { passed: true, detail: "sandbox: clone healthy in 7s", mode: "k8s" }, + handling: "quiet", }, }, { diff --git a/frontend/src/data/types.ts b/frontend/src/data/types.ts index 90549de..43df71a 100644 --- a/frontend/src/data/types.ts +++ b/frontend/src/data/types.ts @@ -64,8 +64,13 @@ export interface SituationOutcome { mode: "off" | "k8s"; sandbox_namespace?: string | null; } | null; + /** "quiet": the fix ran without anyone being asked, on the playbook's real track record. */ + handling?: Handling; } +/** "quiet" = a reliably-fixed signature, handled without paging a human (and audited). */ +export type Handling = "normal" | "quiet"; + export interface Situation { id: string; // "sit-" + signature signature: string; @@ -81,6 +86,7 @@ export interface Situation { reversible: boolean; reliability: number; // per-signature reliability (0..1) suppressed: boolean; + handling?: Handling; // "quiet" = correlation asked for this to be handled without paging outcome?: SituationOutcome; // present once remediation has produced a result peak_score?: number | null; baseline?: Record | null; @@ -97,6 +103,7 @@ export interface OutcomeRow { reason: OutcomeReason; ts: number; service: string; + handling?: Handling; } export interface AuditRow { @@ -144,6 +151,7 @@ export interface Metrics { approvalsPending: number; successRate: number; // 0..1 needsAttention: number; // escalations awaiting a human — excluded from successRate + quietlyHandled?: number; // fixes that ran without anyone being asked (proven track record) } export interface MemberEvent { diff --git a/frontend/src/views/Incidents.tsx b/frontend/src/views/Incidents.tsx index 3c4e5de..aa83f12 100644 --- a/frontend/src/views/Incidents.tsx +++ b/frontend/src/views/Incidents.tsx @@ -40,6 +40,10 @@ function failureNote(reason: string | undefined): string { if (!reason) return "no outcome detail recorded"; if (reason === "unhealthy:rolled-back") return "the fix ran, health did not recover, and it was rolled back"; + if (reason === "unhealthy:no-rollback") + return "the fix ran and health did not recover; the playbook has no rollback, so a human must check"; + if (reason === "unhealthy:rollback-failed") + return "the fix ran, health did not recover, and the rollback itself FAILED — a human must check"; if (reason.startsWith("interrupted:")) return "the attempt was interrupted mid-flight — whether the cluster changed is UNKNOWN, so a human must check"; if (reason === "aborted:timeout") @@ -52,6 +56,18 @@ function failureNote(reason: string | undefined): string { return "gate failed closed — nothing executed"; } +/** A reliably-fixed signature: remediated without paging a human, still audited. */ +function QuietChip() { + return ( + + quiet + + ); +} + const METRIC_DOCS: Record = { noise: { title: "Noise reduction", @@ -343,6 +359,7 @@ export function Incidents({
+ {s.handling === "quiet" && }
{timeAgo(s.first_seen)} @@ -379,6 +396,7 @@ export function Incidents({
+ {shown.handling === "quiet" && }

{shown.title}

@@ -523,6 +541,11 @@ export function Incidents({
Resolved · {shown.outcome?.health_after ?? "resolved"} {shown.outcome?.mode === "dry_run" && dry-run} + {shown.outcome?.handling === "quiet" && ( + + handled quietly · no approval needed (proven fix, audited) + + )}
{shown.outcome?.steps && shown.outcome.steps.length > 0 && (
steps: {shown.outcome.steps.join(" → ")}
diff --git a/services/action/app.py b/services/action/app.py index 0ecb48a..bb2dda3 100644 --- a/services/action/app.py +++ b/services/action/app.py @@ -138,6 +138,12 @@ async def lifespan(app: FastAPI): stop_event, make_guard(settings, app.state.bus), ), + { + "training_store": stores.training_store, + "quiet_threshold": settings.reliability_suppress_threshold, + "quiet_min_samples": settings.reliability_suppress_min_samples, + "quiet_skip_approval": settings.quiet_skip_approval, + }, ) app.state.consumer_stop = stop_event app.state.consumer_thread = thread diff --git a/services/action/consumer.py b/services/action/consumer.py index 2e5e876..9610cea 100644 --- a/services/action/consumer.py +++ b/services/action/consumer.py @@ -21,6 +21,7 @@ RemediationResult, ) from common.envelope import iter_models, publish_model +from common.evidence import quiet_eligible from common.idempotency import NullGuard from services.action.remediate import _ACTOR, execute_remediation from services.action.select import select_playbook @@ -47,6 +48,53 @@ def _audit_best_effort(gate, situation_id: str, action: str, decision: str) -> N logger.warning("%s audit write failed for %s: %s", action, situation_id, exc) +def _quiet_decision( + gate, + situation, + playbook, + training_store, + threshold: float, + min_samples: int, + allowed: bool, +) -> bool: + """Should this quiet situation's playbook run without a human approval? + + Correlation marks a signature quiet from its own reliability map; that is a + request, and it is not specific to the playbook RCA then picked. So the + decision is re-made here against THIS playbook's real track record on THIS + signature (common/evidence.py). Whatever the answer, it is audited and + logged: a quiet fix must never be invisible. + + An AUTO playbook never asks a human anyway, so there is nothing to waive. + """ + if playbook.hitl_mode != HitlMode.HITL: + logger.info("quiet situation %s: playbook %s is already auto", situation.id, playbook.id) + return False + record = None + eligible = False + if allowed and training_store is not None: + try: + eligible, record = quiet_eligible( + training_store.read_all(), situation.signature, playbook.id, threshold, min_samples + ) + except Exception as exc: # noqa: BLE001 - no evidence means ask a human + logger.warning("quiet situation %s: track record unreadable (%s)", situation.id, exc) + evidence = str(record) if record is not None else "no track record available" + if not allowed: + evidence = "quiet approval disabled (QUIET_SKIP_APPROVAL=false)" + decision = "quiet:skip-approval" if eligible else "quiet:needs-approval" + logger.info( + "quiet situation %s, playbook %s: %s (%s)", + situation.id, + playbook.id, + "running without approval" if eligible else "asking a human after all", + evidence, + extra={"situation_id": situation.id, "playbook_id": playbook.id, "quiet": eligible}, + ) + _audit_best_effort(gate, situation.id, "quiet-handling", f"{decision} ({evidence})") + return eligible + + def run_consumer( bus, store, @@ -58,6 +106,10 @@ def run_consumer( poll_interval_seconds: float, stop_event: threading.Event, guard=None, + training_store=None, + quiet_threshold: float = 0.8, + quiet_min_samples: int = 3, + quiet_skip_approval: bool = True, ) -> None: guard = guard if guard is not None else NullGuard() for diagnosed in iter_models( @@ -121,6 +173,17 @@ def run_consumer( # The only outcome-producing path here that had no audit trail. _audit_best_effort(gate, situation.id, "escalate", "escalated") else: + skip = False + if situation.handling == "quiet": + skip = _quiet_decision( + gate, + situation, + playbook, + training_store, + quiet_threshold, + quiet_min_samples, + quiet_skip_approval, + ) outcome = execute_remediation( situation, playbook, @@ -130,6 +193,10 @@ def run_consumer( sandbox, timeout_seconds, poll_interval_seconds, + skip_approval=skip, ) + if situation.handling == "quiet" and (skip or playbook.hitl_mode == HitlMode.AUTO): + # Nobody was asked: this is what "handled quietly" means. + outcome = outcome.model_copy(update={"handling": "quiet"}) publish_model(bus, "remediation.outcomes", outcome) guard.set_state(claim_key, "done") diff --git a/services/action/remediate.py b/services/action/remediate.py index 2f39797..c8ca287 100644 --- a/services/action/remediate.py +++ b/services/action/remediate.py @@ -191,7 +191,12 @@ def execute_remediation( sandbox, timeout_seconds: float, poll_interval_seconds: float, + skip_approval: bool = False, ) -> RemediationOutcome: + """`skip_approval` waives ONLY gate 3 (the human approval), and only for a + quiet situation whose playbook the caller has already checked against its + real track record (services/action/consumer.py). Gates 0-2.5 and the + pre-flight rehearsal apply exactly as they do to everything else.""" # Gate 0: disabled playbooks never run. if playbook.hitl_mode == HitlMode.DISABLED: _audit(gate, situation, playbook, "skipped") @@ -227,8 +232,9 @@ def execute_remediation( # approves (and before an auto playbook executes). Fail-safe — the sandbox # never raises; a failure is a PreflightResult(passed=False). preflight = sandbox.rehearse(situation, plan) - if not preflight.passed and playbook.hitl_mode == HitlMode.AUTO: - # Auto has no human to advise — block. + if not preflight.passed and (playbook.hitl_mode == HitlMode.AUTO or skip_approval): + # Auto has no human to advise — block. Neither does a quiet run whose + # approval was waived: a failed rehearsal is a hard stop for it too. _audit(gate, situation, playbook, "preflight-failed") return _outcome( situation, @@ -242,7 +248,9 @@ def execute_remediation( # Gate 3: HITL — wait for an explicit human approval (ADR-008). The human # sees the pre-flight verdict on the request. - if playbook.hitl_mode == HitlMode.HITL: + if playbook.hitl_mode == HitlMode.HITL and skip_approval: + _audit(gate, situation, playbook, "quiet-approved") + elif playbook.hitl_mode == HitlMode.HITL: request = gate.request_approval( ApprovalRequest( id=f"appr-{situation.id}", diff --git a/services/action/tests/test_quiet_handling.py b/services/action/tests/test_quiet_handling.py new file mode 100644 index 0000000..fee19e9 --- /dev/null +++ b/services/action/tests/test_quiet_handling.py @@ -0,0 +1,248 @@ +"""Quiet handling: a reliably-fixed signature is remediated without paging a human, +but only on the playbook's own real track record, and never out of sight. +""" + +from __future__ import annotations + +import threading +from datetime import UTC, datetime + +from common.contracts import ( + ApprovalRequest, + DiagnosedSituation, + HitlMode, + Playbook, + PreflightResult, + RemediationOutcome, + RemediationResult, + RemediationStep, + Situation, + SituationStatus, + TelemetryEvent, + TelemetryKind, + TrainingRecord, +) +from common.envelope import decode_model +from services.action.adapters.health import AlwaysHealthyChecker +from services.action.adapters.remediator import RecordingRemediator +from services.action.adapters.sandbox import NullSandbox +from services.action.consumer import run_consumer +from services.feedback.adapters.training_store import InMemoryTrainingStore +from services.governance.adapters.playbook_store import InMemoryPlaybookStore + +NOW = datetime(2026, 9, 23, tzinfo=UTC) + + +class Gate: + """Records whether anyone was asked; approves if asked.""" + + def __init__(self): + self.audits = [] + self.approval_requests = [] + + def check_rbac(self, actor, action, resource): + return True + + def request_approval(self, request): + self.approval_requests.append(request) + return request + + def await_decision(self, approval_id, timeout_seconds): + return ApprovalRequest( + id=approval_id, + situation_id="s1", + playbook_id="restart-pod", + requested_by="action-service", + status="approved", + decided_by="oncall-alice", + ) + + def write_audit(self, record): + self.audits.append(record) + + +class Bus: + def __init__(self, script): + self._script = script + self.published = [] + + def publish(self, topic, message): + self.published.append((topic, message)) + + def consume(self, topic, group): + yield from self._script + + def send_to_dlq(self, *a, **k): + return False + + +class FailingSandbox: + def rehearse(self, situation, plan): + return PreflightResult(passed=False, mode="k8s", detail="clone never became ready") + + +def _msg(handling="quiet"): + sit = Situation( + id="s1", + status=SituationStatus.DIAGNOSED, + member_events=[ + TelemetryEvent( + source="p", + kind=TelemetryKind.METRIC, + name="memory_usage_mb", + value=900.0, + labels={"service": "meridian-gateway"}, + ts=NOW, + fingerprint="f", + ) + ], + severity="high", + first_seen=NOW, + last_seen=NOW, + signature="sig", + handling=handling, + ) + d = DiagnosedSituation(situation=sit, hypotheses=[], suggested_runbook_id="restart-pod") + return {"data": d.model_dump_json()} + + +def _playbooks(mode=HitlMode.HITL): + s = InMemoryPlaybookStore() + s.register( + Playbook( + id="restart-pod", + name="Restart", + match_rule="x", + steps=[RemediationStep(action="restart")], + hitl_mode=mode, + reversible=True, + rollback_steps=[RemediationStep(action="restart")], + ) + ) + return s + + +def _history(n, mode="k8s", worked=True, pid="restart-pod"): + store = InMemoryTrainingStore() + for _ in range(n): + store.append( + TrainingRecord( + situation_id="sit-sig", + signature="sig", + playbook_id=pid, + result=RemediationResult.SUCCESS if worked else RemediationResult.ROLLED_BACK, + worked=worked, + ts=NOW, + mode=mode, + ) + ) + return store + + +def _run(msg, history, playbooks=None, sandbox=None, skip_allowed=True): + bus, gate, rem = Bus([msg]), Gate(), RecordingRemediator() + run_consumer( + bus, + playbooks or _playbooks(), + gate, + rem, + AlwaysHealthyChecker(), + sandbox or NullSandbox(), + 1.0, + 0.01, + threading.Event(), + training_store=history, + quiet_threshold=0.8, + quiet_min_samples=3, + quiet_skip_approval=skip_allowed, + ) + outcome = decode_model(bus.published[-1][1], RemediationOutcome) + return outcome, gate, rem + + +def _audit_decisions(gate): + return [(a.action, a.decision) for a in gate.audits] + + +def test_proven_playbook_runs_without_asking_and_is_audited(): + outcome, gate, rem = _run(_msg(), _history(3)) + assert gate.approval_requests == [] # nobody was paged + assert rem.executed_plan is not None # but the fix ran + assert outcome.result == RemediationResult.SUCCESS + assert outcome.handling == "quiet" + decisions = _audit_decisions(gate) + assert ("quiet-handling", "quiet:skip-approval (3/3 real runs worked)") in decisions + assert ("execute", "quiet-approved") in decisions + + +def test_too_little_evidence_falls_back_to_a_human(): + outcome, gate, _rem = _run(_msg(), _history(2)) + assert len(gate.approval_requests) == 1 + assert outcome.handling == "normal" + assert ("quiet-handling", "quiet:needs-approval (2/2 real runs worked)") in _audit_decisions( + gate + ) + + +def test_dry_run_history_is_not_evidence(): + outcome, gate, _ = _run(_msg(), _history(10, mode="dry_run")) + assert len(gate.approval_requests) == 1 + assert outcome.handling == "normal" + + +def test_another_playbooks_record_does_not_count(): + _, gate, _ = _run(_msg(), _history(5, pid="scale-service")) + assert len(gate.approval_requests) == 1 + + +def test_a_poor_record_falls_back(): + store = _history(3) + for r in _history(2, worked=False).read_all(): + store.append(r) # 3/5 = 0.6 < 0.8 + _, gate, _ = _run(_msg(), store) + assert len(gate.approval_requests) == 1 + + +def test_skipping_approval_can_be_switched_off(): + outcome, gate, _ = _run(_msg(), _history(3), skip_allowed=False) + assert len(gate.approval_requests) == 1 + assert outcome.handling == "normal" + + +def test_a_failed_rehearsal_blocks_a_quiet_run(): + # With no human to read the preflight verdict, a failed rehearsal is a hard stop. + outcome, gate, rem = _run(_msg(), _history(3), sandbox=FailingSandbox()) + assert rem.executed_plan is None + assert outcome.health_after == "preflight-failed" + assert gate.approval_requests == [] + + +def test_normal_situations_are_untouched(): + outcome, gate, _ = _run(_msg(handling="normal"), _history(10)) + assert len(gate.approval_requests) == 1 + assert outcome.handling == "normal" + assert not any(a.action == "quiet-handling" for a in gate.audits) + + +def test_an_auto_playbook_is_marked_quiet_without_a_waiver(): + outcome, gate, _ = _run(_msg(), _history(0), playbooks=_playbooks(HitlMode.AUTO)) + assert gate.approval_requests == [] + assert outcome.handling == "quiet" + + +def test_the_other_gates_still_apply_to_a_quiet_run(): + # A waived approval is the ONLY thing quiet handling changes. + s = InMemoryPlaybookStore() + s.register( + Playbook( + id="restart-pod", + name="Restart", + match_rule="x", + steps=[RemediationStep(action="restart")], + hitl_mode=HitlMode.HITL, + reversible=False, + ) + ) + outcome, _, rem = _run(_msg(), _history(3), playbooks=s) + assert outcome.health_after == "refused:not-reversible" + assert rem.executed_plan is None diff --git a/services/correlation/app.py b/services/correlation/app.py index 654eed4..b96bc61 100644 --- a/services/correlation/app.py +++ b/services/correlation/app.py @@ -12,6 +12,7 @@ from common.config import get_settings from common.envelope import publish_model +from common.evidence import real_records from common.idempotency import make_guard from common.stores import make_stores from common.supervise import start_supervised @@ -34,6 +35,8 @@ def run_flusher( stop_event: threading.Event, baseline_store=None, snapshot_period: float = 30.0, + training_store=None, + reliability_refresh_seconds: float = 60.0, ) -> None: """Periodically collapse the buffered window into a Situation. @@ -54,8 +57,12 @@ def run_flusher( its own elapsed-time schedule tracked with time.monotonic() rather than once per wake. The snapshot is best-effort (_snapshot_baseline_once never raises), so a persistence hiccup can never crash this flusher. + + It also refreshes the reliability map suppression reads, on its own + schedule. That map used to be loaded once at boot, so a fix that started + working was only recognised after a restart. """ - last_snapshot = time.monotonic() + last_snapshot = last_refresh = time.monotonic() while not stop_event.wait(period_seconds): for emitted in engine.flush_all(): publish_model(bus, "situations.detected", emitted) @@ -64,6 +71,23 @@ def run_flusher( if now - last_snapshot >= snapshot_period: _snapshot_baseline_once(engine, baseline_store) last_snapshot = now + if training_store is not None and now - last_refresh >= reliability_refresh_seconds: + _refresh_reliability(engine, training_store) + last_refresh = now + + +def _training_rows(training_store) -> list[dict]: + """Real outcomes only (common/evidence.py): a dry run always "works", so it + is no evidence a signature is reliably fixed.""" + return [r.model_dump() for r in real_records(training_store.read_all())] + + +def _refresh_reliability(engine, training_store) -> None: + """Best-effort: a failed read keeps the current map.""" + try: + engine.retrain(_training_rows(training_store)) + except Exception as exc: # noqa: BLE001 - refresh again next period + logger.warning("reliability refresh failed (keeping the current map): %s", exc) def _reload_baseline(engine, baseline_store, training_records: list[dict]) -> None: @@ -110,6 +134,7 @@ async def lifespan(app: FastAPI): # config and the engine always used its 0.8 default. suppress_threshold=settings.reliability_suppress_threshold, suppress_min_samples=settings.reliability_suppress_min_samples, + suppress_mode=settings.suppression_mode, ) app.state.engine = engine # Reload-on-boot: restore the durable baseline + reliability BEFORE the @@ -123,13 +148,15 @@ async def lifespan(app: FastAPI): # it must be inside the guard too. baseline_store = None model_store = None + training_store = None training_records: list[dict] = [] try: stores = make_stores(settings) app.state.db_engine = stores.engine baseline_store = stores.baseline_store model_store = getattr(stores, "model_store", None) - training_records = [r.model_dump() for r in stores.training_store.read_all()] + training_store = stores.training_store + training_records = _training_rows(training_store) except Exception as exc: # noqa: BLE001 — a failed boot-load just means a cold start logger.warning("store reload failed, starting cold: %s", exc) # The durable `correlation_baseline` table only understands the river z-score's @@ -161,6 +188,8 @@ async def lifespan(app: FastAPI): stop_event, baseline_store, settings.baseline_snapshot_seconds, + training_store, + settings.reliability_refresh_seconds, ), ) app.state.consumer_stop = stop_event diff --git a/services/correlation/consumer.py b/services/correlation/consumer.py index 1799699..dd6d3f4 100644 --- a/services/correlation/consumer.py +++ b/services/correlation/consumer.py @@ -30,6 +30,19 @@ def _snapshot_baseline_once(engine, baseline_store) -> None: def _drain_suppressed(bus, engine: CorrelationEngine) -> None: while (s := engine.pop_suppressed()) is not None: + # The record of every suppression, whatever it did to the Situation. In + # quiet mode the Situation was ALSO emitted for remediation; in drop mode + # this line and the situations.suppressed event are all that remain. + logger.info( + "suppressed situation %s (signature %s, service %s): %s", + s.id, + s.signature, + next( + (e.labels.get("service") for e in s.member_events if e.labels.get("service")), "?" + ), + "handling quietly" if s.handling == "quiet" else "dropped", + extra={"situation_id": s.id, "signature": s.signature, "handling": s.handling}, + ) publish_model(bus, "situations.suppressed", s) diff --git a/services/correlation/engine.py b/services/correlation/engine.py index 56fdf75..b3abb63 100644 --- a/services/correlation/engine.py +++ b/services/correlation/engine.py @@ -39,6 +39,7 @@ def __init__( group_by: str = "window", min_events: int = 1, suppress_min_samples: int = 1, + suppress_mode: str = "drop", ) -> None: self._correlator = correlator self._correlator_factory = ( @@ -53,6 +54,15 @@ def __init__( self._window = window_seconds self._suppress_threshold = suppress_threshold self._suppress_min_samples = max(1, int(suppress_min_samples)) + # What suppression DOES to a reliably-fixed signature: + # "drop" (historical default here): the Situation is never emitted, so + # it never reaches RCA or action - the fault is left unfixed. + # "quiet": the Situation IS emitted, marked handling="quiet", so it is + # still diagnosed and remediated, just without paging a human + # when action can confirm the playbook's track record. + # Either way it is also queued for situations.suppressed (the counter + # and the log of what was suppressed). + self._suppress_mode = suppress_mode self._group_by = group_by # How many anomalous events a window must hold before it is an incident. # @@ -162,14 +172,23 @@ def _correlate_buffer(self, key: str = _ALL) -> Situation | None: sit = sit.model_copy(update={"peak_score": peak, "baseline": baseline}) self._buffers.pop(key, None) self._max_scores.pop(key, None) - # Closed loop: suppress a Situation whose signature reliably self-heals. + # Closed loop: a signature the system has reliably fixed before. if self._correlator.should_suppress( sit.signature, self._suppress_threshold, self._suppress_min_samples ): + if self._suppress_mode == "quiet": + sit = sit.model_copy(update={"handling": "quiet"}) + self._suppressed.append(sit) + return sit self._suppressed.append(sit) return None return sit + def retrain(self, training_data: list[dict]) -> None: + """Replace the reliability map (what suppression reads) under the lock.""" + with self._lock: + self._correlator.retrain(training_data) + def snapshot(self) -> list[dict]: with self._lock: return self._correlator.snapshot() @@ -185,7 +204,14 @@ def pop_suppressed(self) -> Situation | None: def reset(self) -> None: with self._lock: + # A baseline reset forgets what "normal" looks like, not which fixes + # have worked: the reliability map comes from labelled outcomes and is + # re-derived from them anyway, so carry it across. + old = self._correlator self._correlator = self._correlator_factory() + for attr in ("_reliability", "_samples"): + if hasattr(old, attr): + setattr(self._correlator, attr, getattr(old, attr)) self._buffers = {} self._max_scores = {} self._suppressed = [] diff --git a/services/correlation/tests/test_engine_suppress.py b/services/correlation/tests/test_engine_suppress.py index eddcab5..1656275 100644 --- a/services/correlation/tests/test_engine_suppress.py +++ b/services/correlation/tests/test_engine_suppress.py @@ -94,3 +94,33 @@ def test_every_bucket_suppressed_in_one_flush_is_reported(): while (s := engine.pop_suppressed()) is not None: suppressed.append(s) assert len(suppressed) == 2 + + +def _suppressible_engine(mode): + correlator = RiverCorrelator(z_threshold=3.0) + engine = CorrelationEngine(correlator, window_seconds=30, suppress_mode=mode) + ev = _event(1.0, "fq") + engine._buffers = {engine._ALL: [ev]} + engine._max_scores = {engine._ALL: 9.0} + correlator.retrain([{"signature": correlator._signature([ev]), "worked": True}]) + return engine + + +def test_quiet_mode_still_emits_the_situation_for_remediation(): + engine = _suppressible_engine("quiet") + emitted = engine.flush_all() + assert len(emitted) == 1 and emitted[0].handling == "quiet" + logged = engine.pop_suppressed() # still recorded as a suppression + assert logged is not None and logged.handling == "quiet" + + +def test_drop_mode_keeps_the_historical_behaviour(): + engine = _suppressible_engine("drop") + assert engine.flush_all() == [] + assert engine.pop_suppressed() is not None + + +def test_a_baseline_reset_keeps_what_was_learned_about_fixes(): + engine = _suppressible_engine("quiet") + engine.reset() + assert engine._correlator.reliability(next(iter(engine._correlator._reliability))) == 1.0 diff --git a/services/correlation/tests/test_retrain.py b/services/correlation/tests/test_retrain.py index c0cff70..47c3acb 100644 --- a/services/correlation/tests/test_retrain.py +++ b/services/correlation/tests/test_retrain.py @@ -54,3 +54,43 @@ def test_one_success_is_not_a_track_record(): assert c.should_suppress("s", 0.8, min_samples=3) is False c.retrain([{"signature": "s", "worked": True}] * 3) assert c.should_suppress("s", 0.8, min_samples=3) is True + + +def test_runtime_refresh_uses_only_real_outcomes(): + from datetime import UTC, datetime + + from common.contracts import RemediationResult, TrainingRecord + from services.correlation.app import _refresh_reliability + from services.correlation.engine import CorrelationEngine + + def rec(mode): + return TrainingRecord( + situation_id="sit-s", + signature="s", + playbook_id="restart-pod", + result=RemediationResult.SUCCESS, + worked=True, + ts=datetime(2026, 9, 23, tzinfo=UTC), + mode=mode, + ) + + class Store: + records = [rec("dry_run")] * 5 + + def read_all(self): + return list(self.records) + + engine = CorrelationEngine(RiverCorrelator()) + store = Store() + _refresh_reliability(engine, store) + assert engine._correlator.reliability("s") == 0.0 # simulations prove nothing + store.records = [rec("k8s")] * 3 + _refresh_reliability(engine, store) # picked up without a restart + assert engine._correlator.reliability("s") == 1.0 + + class Broken: + def read_all(self): + raise RuntimeError("db down") + + _refresh_reliability(engine, Broken()) # never raises, keeps the map + assert engine._correlator.reliability("s") == 1.0 diff --git a/services/read/projection.py b/services/read/projection.py index b3648dd..0d9ba52 100644 --- a/services/read/projection.py +++ b/services/read/projection.py @@ -172,6 +172,9 @@ def _apply_detected(self, s: Situation) -> bool: "reversible": existing.get("reversible", True), "reliability": existing.get("reliability", 0.0), "suppressed": False, + # "quiet" = correlation asked for this to be handled without paging a + # human; the outcome says whether that is what actually happened. + "handling": getattr(s, "handling", "normal"), "last_activity": existing.get("last_activity", _epoch_ms(s.first_seen)), "stages": existing.get("stages", {}), } @@ -248,6 +251,7 @@ def _apply_outcome(self, o: RemediationOutcome) -> None: "preflight": ( p.model_dump() if (p := getattr(o, "preflight", None)) is not None else None ), + "handling": getattr(o, "handling", "normal"), } result = o.result.value if isinstance(o.result, RemediationResult) else str(o.result) sit = self._sits.get(o.situation_id, {}) @@ -280,6 +284,7 @@ def _apply_outcome(self, o: RemediationOutcome) -> None: # restarted a pod" and "we simulated it" were indistinguishable # downstream. It is the whole point of the k8s posture. "mode": getattr(o, "mode", "dry_run"), + "handling": getattr(o, "handling", "normal"), "mttr_ms": mttr_ms, }, ) @@ -384,6 +389,8 @@ def _metrics(self) -> dict: "approvalsPending": len(pending), "successRate": round(successes / n_att, 3) if n_att else 0.0, "needsAttention": sum(1 for s in sits if s["status"] == "needs_attention"), + # Fixes that ran without a human being asked, on a proven track record. + "quietlyHandled": sum(1 for o in outs if o.get("handling") == "quiet"), } @staticmethod diff --git a/services/read/tests/test_metrics.py b/services/read/tests/test_metrics.py index 919546f..3b9e7b7 100644 --- a/services/read/tests/test_metrics.py +++ b/services/read/tests/test_metrics.py @@ -55,6 +55,7 @@ def test_empty_metrics_all_zero(): "approvalsPending", "successRate", "needsAttention", + "quietlyHandled", } diff --git a/services/read/tests/test_occurrences.py b/services/read/tests/test_occurrences.py index 6ad6f6e..a7282a0 100644 --- a/services/read/tests/test_occurrences.py +++ b/services/read/tests/test_occurrences.py @@ -100,3 +100,13 @@ def test_an_event_from_an_earlier_occurrence_is_ignored(): card = rm.situation("sit-a") assert card["first_seen"] == int(T2.timestamp() * 1000) assert card["hypotheses"] == [] + + +def test_quiet_handling_is_visible_on_the_card_and_counted(): + rm = ReadModel() + rm.apply_detected(_sit(T1).model_copy(update={"handling": "quiet"})) + assert rm.situation("sit-a")["handling"] == "quiet" + rm.apply_outcome(_outcome(T1 + timedelta(minutes=1)).model_copy(update={"handling": "quiet"})) + assert rm.situation("sit-a")["outcome"]["handling"] == "quiet" + assert rm.outcomes()[0]["handling"] == "quiet" + assert rm.metrics()["quietlyHandled"] == 1 diff --git a/tests/test_evidence.py b/tests/test_evidence.py new file mode 100644 index 0000000..df79bca --- /dev/null +++ b/tests/test_evidence.py @@ -0,0 +1,49 @@ +from datetime import UTC, datetime + +from common.contracts import RemediationResult, TrainingRecord +from common.evidence import is_real, quiet_eligible, real_records, track_record + +NOW = datetime(2026, 9, 23, tzinfo=UTC) + + +def _rec(worked=True, mode="k8s", sig="s", pid="restart-pod"): + return TrainingRecord( + situation_id=f"sit-{sig}", + signature=sig, + playbook_id=pid, + result=RemediationResult.SUCCESS if worked else RemediationResult.ROLLED_BACK, + worked=worked, + ts=NOW, + mode=mode, + ) + + +def test_only_real_runs_are_evidence(): + assert is_real(_rec(mode="k8s")) + assert not is_real(_rec(mode="dry_run")) + assert not is_real(_rec(mode=None)) # predates the mode field: unknowable + assert not is_real({"mode": "none"}) # escalation: nothing ran + assert len(real_records([_rec(), _rec(mode="dry_run"), _rec(mode=None)])) == 1 + + +def test_track_record_narrows_to_signature_and_playbook(): + recs = [_rec(), _rec(worked=False), _rec(pid="scale-service"), _rec(sig="other")] + assert (track_record(recs, "s").worked, track_record(recs, "s").total) == (2, 3) + rp = track_record(recs, "s", "restart-pod") + assert (rp.worked, rp.total) == (1, 2) + + +def test_quiet_needs_enough_real_runs_of_this_playbook(): + three = [_rec(), _rec(), _rec()] + assert quiet_eligible(three, "s", "restart-pod", 0.8, 3)[0] is True + assert quiet_eligible(three[:2], "s", "restart-pod", 0.8, 3)[0] is False # too few + assert quiet_eligible(three, "s", "scale-service", 0.8, 3)[0] is False # other playbook + dry = [_rec(mode="dry_run")] * 5 + assert quiet_eligible(dry, "s", "restart-pod", 0.8, 3)[0] is False # simulations + + +def test_quiet_needs_a_high_enough_success_rate(): + recs = [_rec(), _rec(), _rec(), _rec(worked=False)] # 3/4 = 0.75 + ok, record = quiet_eligible(recs, "s", "restart-pod", 0.8, 3) + assert ok is False + assert str(record) == "3/4 real runs worked"