diff --git a/kora_cli/alerts/__init__.py b/kora_cli/alerts/__init__.py new file mode 100644 index 000000000000..0555ce67143e --- /dev/null +++ b/kora_cli/alerts/__init__.py @@ -0,0 +1,16 @@ +"""Alert aggregation — KR-ALERTS-PANEL-FLIP. + +Aggregates operator-attention signals from the 5 existing data +sources (OperationalStateHolder + cost-ladder holder + HealthRollup ++ audit JSONL + heartbeat probe snapshots) into the FE's +``Alert`` shape from ``web/src/lib/api.ts``. + +Public surface: + * :class:`Alert` — wire-shape dataclass mirroring the TS interface + * :func:`compute_active_alerts` — single-call aggregator the + endpoint consumes +""" + +from kora_cli.alerts.aggregator import Alert, compute_active_alerts + +__all__ = ["Alert", "compute_active_alerts"] diff --git a/kora_cli/alerts/aggregator.py b/kora_cli/alerts/aggregator.py new file mode 100644 index 000000000000..af412cb4a46f --- /dev/null +++ b/kora_cli/alerts/aggregator.py @@ -0,0 +1,526 @@ +"""Active-alert aggregator — KR-ALERTS-PANEL-FLIP. + +:func:`compute_active_alerts` calls 5 per-source helpers, each +isolated by try/except so a single source failure (uninitialized +holder, unreadable JSONL, transient probe glitch) NEVER bubbles a +500 to the operator. Degraded data is better than no panel. + +# Rule taxonomy (10 rules) + +| Rule | Source | Severity | Trigger | +|---|---|---|---| +| ``cost_ladder_warned`` | cost holder | warning | active_rung == WARN_75 | +| ``cost_ladder_downshifted`` | cost holder | warning | active_rung == DOWNSHIFT_90 | +| ``cost_ladder_halted`` | cost holder | critical | active_rung == HARD_STOP_100 | +| ``operator_paused`` | operational state | critical | primary_state == PAUSED | +| ``operator_stopped`` | operational state | critical | primary_state == STOPPED | +| ``webhook_dead_letters_24h`` | audit JSONL | warning | count > 5 | +| ``capability_denied_24h`` | audit JSONL | info | count > 10 | +| ``reasoning_errors_24h`` | audit JSONL | warning | count > 5 | +| ``service_unhealthy`` | probe snapshots | warning | per service in {degraded, unhealthy} | +| ``slack_dm_reply_failed_24h`` | audit JSONL | warning | count > 3 | + +Thresholds are PROPOSED (per bucket §2(b)) and tunable per +operator feedback; they're module-level constants below for ease +of edits. + +# K-DG K-DG (yes, twice — paranoid by design) + +Per §1 of the bucket spec + the live grep at HEAD ``054f4086``: + + * Cost holder accessor: ``agent.cost_state_holder.get_cost_holder()``. + Rung resolution: ``.active_rung()`` — METHOD, not property + (PM-locked #126 catch). + * Operational state accessor: ``agent.operational_state_holder.get_holder()`` + (NOT ``get_operational_state_holder()`` — bucket spec drift + versus actual symbol). ``holder.current`` is @PROPERTY + (#112 catch); ``.primary_state`` is a bare enum field on the + inner ``OperationalState`` dataclass. + * Health rollup accessor: + ``agent.health_rollup_holder.get_health_rollup_holder()`` + (NOT ``get_health_holder()`` — bucket spec drift). ``.current()`` + is METHOD; returns a frozen dataclass with BARE field names + ``overall`` / ``control_plane`` / ``worker`` (no @property + wrapper per #112 catch). + * Audit reader: ``kora_cli.audit.jsonl_reader.read_audit_entries`` + accepts ``seam=`` and ``since=`` kwargs (jsonl_reader.py:58). + * Heartbeat snapshots: ``kora_cli.heartbeat_probes.runner.current_service_snapshots()`` + returns ``dict[str, ServiceHealthSnapshot]``; status enum + drawn from ``ServiceStatus`` Literal in + ``heartbeat_probes/types.py:28``: {healthy, degraded, + unhealthy, unknown}. + +# Forward-compat note: capability_denied + +The ``capability_denied_24h`` rule is forward-looking: today the +``mcp.tool_called`` audit emit at +``kora_cli/listeners/mcp_tools.py:714`` is reached AFTER the +capability gate (``listeners/mcp.py:181``), so denial responses +are NOT currently audit-logged. This rule's matcher uses +``details.result == "capability_denied"`` so when a follow-on +bucket adds audit-on-denial the rule activates automatically; in +the meantime it emits zero alerts (no false negatives — the data +genuinely isn't there). +""" + +from __future__ import annotations + +import logging +from dataclasses import asdict, dataclass +from datetime import datetime, timedelta, timezone +from typing import List, Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Thresholds — proposed per bucket §2(b); tune from operator feedback +# --------------------------------------------------------------------------- + + +WEBHOOK_DEAD_LETTER_24H_THRESHOLD = 5 +CAPABILITY_DENIED_24H_THRESHOLD = 10 +REASONING_ERRORS_24H_THRESHOLD = 5 +SLACK_DM_REPLY_FAILED_24H_THRESHOLD = 3 + + +# Severity rank for ordering (lower index = higher priority). +_SEVERITY_RANK = {"critical": 0, "warning": 1, "info": 2} + + +@dataclass(frozen=True, slots=True) +class Alert: + """One operator-attention alert. + + Wire-stable shape — keys match the FE's ``Alert`` interface in + ``web/src/lib/api.ts`` verbatim. ``to_dict()`` returns a plain + dict the endpoint serializes through FastAPI's JSON encoder. + """ + + id: str + severity: str # "critical" | "warning" | "info" + category: str # e.g. "cost_ladder" / "operational_state" / "service_unhealthy" + title: str + detail: str + source_panel: str + source_panel_route: str + first_seen_at: str # ISO-8601 UTC with Z suffix + + def to_dict(self) -> dict: + return asdict(self) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +# --------------------------------------------------------------------------- +# Per-source rule helpers +# --------------------------------------------------------------------------- + + +def _rules_from_cost_holder() -> List[Alert]: + """Cost-ladder rules (3 — one per non-NORMAL rung).""" + try: + from agent.cost_state_holder import CostRung, get_cost_holder + except Exception as exc: + logger.warning( + "[kora.alerts] cost_state_holder import failed: %r — skipping " + "cost-ladder rules", + exc, + ) + return [] + + holder = get_cost_holder() + if holder is None: + return [] + + try: + rung = holder.active_rung() + except Exception as exc: + logger.warning( + "[kora.alerts] cost holder.active_rung() raised %r — skipping", + exc, + ) + return [] + + try: + pct = holder.current_pct_used() + except Exception: + pct = None + + now = _now_iso() + pct_str = f"{int(pct * 100)}%" if isinstance(pct, (int, float)) else "?%" + + if rung is CostRung.WARN_75: + return [ + Alert( + id="cost_ladder_warned", + severity="warning", + category="cost_ladder", + title=f"Budget at {pct_str} of monthly cap", + detail=( + "Cost ladder at warn_75 — reasoning still runs on the " + "selected model; reply-failure rates may climb if the " + "rung escalates." + ), + source_panel="cost", + source_panel_route="/cost-state", + first_seen_at=now, + ) + ] + if rung is CostRung.DOWNSHIFT_90: + return [ + Alert( + id="cost_ladder_downshifted", + severity="warning", + category="cost_ladder", + title=f"Reasoning downshifted at {pct_str}", + detail=( + "Cost ladder at downshift_90 — calls that requested " + "opus are being routed to sonnet (and sonnet → haiku) " + "per agent/cost_downshift.py." + ), + source_panel="cost", + source_panel_route="/cost-state", + first_seen_at=now, + ) + ] + if rung is CostRung.HARD_STOP_100: + return [ + Alert( + id="cost_ladder_halted", + severity="critical", + category="cost_ladder", + title=f"Reasoning halted at {pct_str} of budget", + detail=( + "Cost ladder at hard_stop_100 — non-critical reasoning " + "calls refuse with cost_ladder_halted; AUTO_REPLY paths " + "fall back to canned text." + ), + source_panel="cost", + source_panel_route="/cost-state", + first_seen_at=now, + ) + ] + # CostRung.NORMAL → no alert + return [] + + +def _rules_from_operational_state() -> List[Alert]: + """Operational state rules (PAUSED + STOPPED).""" + try: + from agent.operational_state import PrimaryState + from agent.operational_state_holder import get_holder + except Exception as exc: + logger.warning( + "[kora.alerts] operational_state_holder import failed: %r — " + "skipping", + exc, + ) + return [] + + holder = get_holder() + if holder is None: + return [] + + try: + state = holder.current # @property — value snapshot + ps = state.primary_state + except Exception as exc: + logger.warning( + "[kora.alerts] operational holder.current raised %r — skipping", + exc, + ) + return [] + + now = _now_iso() + + if ps is PrimaryState.PAUSED: + return [ + Alert( + id="operator_paused", + severity="critical", + category="operational_state", + title="Kora paused", + detail=( + "Slack DM + email handlers drop inbound traffic at " + "the state gate; reasoning engine refuses calls. " + "Resume from the operational-state panel." + ), + source_panel="ops", + source_panel_route="/operational-state", + first_seen_at=now, + ) + ] + if ps is PrimaryState.STOPPED: + return [ + Alert( + id="operator_stopped", + severity="critical", + category="operational_state", + title="Kora STOPPED", + detail=( + "Terminal state — all surfaces refuse traffic. " + "STOPPED requires manual operator action to clear " + "(typically a daemon restart)." + ), + source_panel="ops", + source_panel_route="/operational-state", + first_seen_at=now, + ) + ] + return [] + + +def _count_audit_entries_in_window( + seam: str, + *, + since: datetime, + detail_match: Optional[dict] = None, +) -> int: + """Helper: read audit entries for ``seam`` since ``since`` + return + the count, optionally filtered to entries whose ``details`` dict + contains every key/value pair in ``detail_match``. + + Returns 0 on any read failure — caller's fail-soft posture. + """ + try: + from kora_cli.audit.jsonl_reader import read_audit_entries + except Exception as exc: + logger.warning( + "[kora.alerts] audit jsonl_reader import failed: %r", exc + ) + return 0 + try: + entries = read_audit_entries(seam=seam, since=since) + except Exception as exc: + logger.warning( + "[kora.alerts] read_audit_entries(seam=%s) raised %r", + seam, + exc, + ) + return 0 + if detail_match is None: + return len(entries) + matched = 0 + for entry in entries: + details = entry.details or {} + if all(details.get(k) == v for k, v in detail_match.items()): + matched += 1 + return matched + + +def _rules_from_webhook_audit() -> List[Alert]: + since = datetime.now(timezone.utc) - timedelta(hours=24) + count = _count_audit_entries_in_window( + seam="webhook.dead_letter", since=since + ) + if count <= WEBHOOK_DEAD_LETTER_24H_THRESHOLD: + return [] + return [ + Alert( + id="webhook_dead_letters_24h", + severity="warning", + category="webhook_dead_letter", + title=f"{count} webhook dead-letters in last 24h", + detail=( + f"Threshold {WEBHOOK_DEAD_LETTER_24H_THRESHOLD} exceeded. " + "Common causes: signing-secret mismatch, malformed payloads " + "from upstream, or webhook listener panic. Check the " + "webhook-events panel for per-event diagnostic." + ), + source_panel="webhook_events", + source_panel_route="/webhook-events", + first_seen_at=_now_iso(), + ) + ] + + +def _rules_from_capability_denied_audit() -> List[Alert]: + """Forward-looking — see module docstring's forward-compat note. + + Today the audit log at ``mcp.tool_called`` only records successful + invocations (capability gate at ``listeners/mcp.py:181`` returns + BEFORE the audit emit at ``listeners/mcp_tools.py:714``). This + rule's matcher is forward-compatible: when a follow-on bucket + adds audit-on-denial it'll start firing without an aggregator + edit. + """ + since = datetime.now(timezone.utc) - timedelta(hours=24) + count = _count_audit_entries_in_window( + seam="mcp.tool_called", + since=since, + detail_match={"result": "capability_denied"}, + ) + if count <= CAPABILITY_DENIED_24H_THRESHOLD: + return [] + return [ + Alert( + id="capability_denied_24h", + severity="info", + category="agent_capability_denied", + title=f"{count} capability_denied responses in 24h", + detail=( + f"Threshold {CAPABILITY_DENIED_24H_THRESHOLD} exceeded. " + "Unconfigured caller actor_kinds may indicate misconfigured " + "third-party agents. Review mcp_callers.yaml + the " + "agent-activity panel for the denied caller distribution." + ), + source_panel="agent_activity", + source_panel_route="/agent-activity", + first_seen_at=_now_iso(), + ) + ] + + +def _rules_from_reasoning_audit() -> List[Alert]: + since = datetime.now(timezone.utc) - timedelta(hours=24) + count = _count_audit_entries_in_window( + seam="reasoning.tool_called", + since=since, + detail_match={"tool_status": "execution_error"}, + ) + if count <= REASONING_ERRORS_24H_THRESHOLD: + return [] + return [ + Alert( + id="reasoning_errors_24h", + severity="warning", + category="reasoning_halted", + title=f"{count} reasoning failures in 24h", + detail=( + f"Threshold {REASONING_ERRORS_24H_THRESHOLD} exceeded " + "(tool_status=execution_error). Common causes: 3P tool " + "transport failure, malformed Pydantic args from the " + "model, or transient SDK 5xx. Check the reasoning panel " + "for per-call diagnostic." + ), + source_panel="reasoning", + source_panel_route="/reasoning", + first_seen_at=_now_iso(), + ) + ] + + +def _rules_from_slack_dm_audit() -> List[Alert]: + since = datetime.now(timezone.utc) - timedelta(hours=24) + count = _count_audit_entries_in_window( + seam="slack_dm.reply_failed", since=since + ) + if count <= SLACK_DM_REPLY_FAILED_24H_THRESHOLD: + return [] + return [ + Alert( + id="slack_dm_reply_failed_24h", + severity="warning", + category="reasoning_halted", + title=f"{count} Slack DM reply failures in 24h", + detail=( + f"Threshold {SLACK_DM_REPLY_FAILED_24H_THRESHOLD} " + "exceeded. Causes typically split between Slack-API " + "transport failures and reasoning-engine errors. Check " + "the slack-dm panel for the failure-reason taxonomy." + ), + source_panel="slack_dm", + source_panel_route="/slack-dm", + first_seen_at=_now_iso(), + ) + ] + + +def _rules_from_probe_snapshots() -> List[Alert]: + """One alert per service in ``degraded`` or ``unhealthy``. + + Per bucket §2(b): emits N alerts (not 1) so the operator sees + each affected service's name in the alerts list — clicking the + alert routes to the heartbeat panel where the per-service + diagnostic lives. + """ + try: + from kora_cli.heartbeat_probes.runner import current_service_snapshots + except Exception as exc: + logger.warning( + "[kora.alerts] heartbeat snapshots import failed: %r", exc + ) + return [] + + try: + snapshots = current_service_snapshots() or {} + except Exception as exc: + logger.warning( + "[kora.alerts] current_service_snapshots raised %r", exc + ) + return [] + + out: List[Alert] = [] + for service_name, snapshot in snapshots.items(): + try: + status = snapshot.status + except Exception: + continue + if status not in {"degraded", "unhealthy"}: + continue + severity = "critical" if status == "unhealthy" else "warning" + out.append( + Alert( + id=f"service_unhealthy:{service_name}", + severity=severity, + category="service_unhealthy", + title=f"{service_name} probe: {status}", + detail=( + f"Heartbeat probe for {service_name} returned " + f"status={status}. See the heartbeat panel for the " + "last_check_at + sanitized error diagnostic." + ), + source_panel="heartbeat", + source_panel_route="/heartbeat", + first_seen_at=_now_iso(), + ) + ) + return out + + +# --------------------------------------------------------------------------- +# Public entry point +# --------------------------------------------------------------------------- + + +_RULE_HELPERS = ( + _rules_from_cost_holder, + _rules_from_operational_state, + _rules_from_webhook_audit, + _rules_from_capability_denied_audit, + _rules_from_reasoning_audit, + _rules_from_slack_dm_audit, + _rules_from_probe_snapshots, +) + + +def compute_active_alerts() -> List[Alert]: + """Run every rule helper + return the merged alert list. + + Sort: severity rank (critical → warning → info), then by ``id`` + for stable ordering within a severity tier so the FE list + doesn't reshuffle between polls. + + Fail-soft: each helper already catches per-source exceptions + + returns ``[]`` on any failure; an unexpected error past that + layer is caught here too (defense in depth) so the endpoint + NEVER 500s. + """ + alerts: List[Alert] = [] + for helper in _RULE_HELPERS: + try: + alerts.extend(helper()) + except Exception as exc: + # Reachable only if a helper bypassed its own try/except — + # which shouldn't happen but the contract is the endpoint + # never crashes the panel. + logger.warning( + "[kora.alerts] helper %s raised past inner catch: %r", + helper.__name__, + exc, + ) + alerts.sort(key=lambda a: (_SEVERITY_RANK.get(a.severity, 99), a.id)) + return alerts diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py index 174120a72a03..38df566a719f 100644 --- a/kora_cli/web_server.py +++ b/kora_cli/web_server.py @@ -5916,84 +5916,44 @@ async def list_recent_reasoning(limit: int = 50): async def list_current_alerts(): """Return currently-active operator-attention alerts. - v1 stub — pinned shape so the deferred alert-collector backend - can swap the body without touching the FE. - - Per-alert fields: - id — opaque id - severity — "critical" | "warning" | "info" - category — alert category (drives the icon mapping); - cost_ladder | operational_state | - webhook_dead_letter | agent_capability_denied - | reasoning_halted | service_unhealthy | - boot_gate_failure - title — short headline (single line, bold) - detail — secondary explanation (rendered as text) - source_panel — short id of the originating panel - source_panel_route — FE route to navigate to; uses the flat - ``/`` convention established by - every prior panel in this branch - (not the bucket-spec's ``/admin/``) - first_seen_at — ISO-8601 when this alert first fired + KR-ALERTS-PANEL-FLIP swaps the v1 stub (PR #134) for a real + aggregator that pulls from 5 sources: OperationalStateHolder + + cost-ladder holder + HealthRollup + audit JSONL + + heartbeat-probe snapshots. See + ``kora_cli/alerts/aggregator.py`` for the rule taxonomy + + fail-soft contract. + + Per-alert fields (matches FE ``Alert`` in ``web/src/lib/api.ts``): + id, severity, category, title, detail, source_panel, + source_panel_route, first_seen_at. + + Behaviour: + * Any per-source failure (holder uninitialized, JSONL + unreadable, probe import error, etc.) is caught inside + the aggregator + that source's rules drop silently; other + rules still emit. Operator NEVER sees a 500. + * stub: false always — the endpoint reads live state even + when no alerts are active (empty list, stub:false). + * Sort: severity rank (critical → warning → info), then by + alert id for stable intra-tier ordering. """ + from kora_cli.alerts import compute_active_alerts + + alerts = compute_active_alerts() + by_severity: Dict[str, int] = {"critical": 0, "warning": 0, "info": 0} + for a in alerts: + by_severity[a.severity] = by_severity.get(a.severity, 0) + 1 + + from datetime import datetime, timezone + return { - "alerts": [ - { - "id": "stub-1", - "severity": "warning", - "category": "cost_ladder", - "title": "Budget at 78% of monthly cap", - "detail": ( - "Reasoning model downshifted opus → sonnet " - "at warn_75 rung" - ), - "source_panel": "cost", - "source_panel_route": "/cost-state", - "first_seen_at": "2026-05-22T17:48:00Z", - }, - { - "id": "stub-2", - "severity": "critical", - "category": "operational_state", - "title": "Operator paused Kora 12 min ago", - "detail": ( - "Slack DM handler dropping messages; reasoning " - "engine refusing calls" - ), - "source_panel": "ops", - "source_panel_route": "/operational-state", - "first_seen_at": "2026-05-22T17:48:00Z", - }, - { - "id": "stub-3", - "severity": "warning", - "category": "webhook_dead_letter", - "title": "8 webhook dead-letters in last 24h", - "detail": ( - "Threshold 5 exceeded; check signing-secret match" - ), - "source_panel": "webhook_events", - "source_panel_route": "/webhook-events", - "first_seen_at": "2026-05-22T13:00:00Z", - }, - { - "id": "stub-4", - "severity": "info", - "category": "agent_capability_denied", - "title": "12 capability_denied responses in 24h", - "detail": ( - "Unconfigured caller actor_kinds — review " - "mcp_callers.yaml" - ), - "source_panel": "agent_activity", - "source_panel_route": "/agent-activity", - "first_seen_at": "2026-05-22T08:00:00Z", - }, - ], - "stub": True, - "generated_at": "2026-05-22T18:00:00Z", - "total_active": 4, - "by_severity": {"critical": 1, "warning": 2, "info": 1}, + "alerts": [a.to_dict() for a in alerts], + "stub": False, + "generated_at": datetime.now(timezone.utc).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ), + "total_active": len(alerts), + "by_severity": by_severity, } diff --git a/tests/kora_cli/alerts/__init__.py b/tests/kora_cli/alerts/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/kora_cli/alerts/test_aggregator.py b/tests/kora_cli/alerts/test_aggregator.py new file mode 100644 index 000000000000..a6cf181bf7d0 --- /dev/null +++ b/tests/kora_cli/alerts/test_aggregator.py @@ -0,0 +1,733 @@ +"""Tests for the KR-ALERTS-PANEL-FLIP aggregator. + +Bucket §2(f) scenarios: + + Per-rule coverage (10 rules): + 1. cost_ladder_warned fires when rung == WARN_75 + 2. cost_ladder_downshifted fires when rung == DOWNSHIFT_90 + 3. cost_ladder_halted fires when rung == HARD_STOP_100 + 4. cost ladder NORMAL → no alert + 5. operator_paused fires when state == PAUSED + 6. operator_stopped fires when state == STOPPED + 7. operational state ACTIVE → no alert + 8. webhook_dead_letters_24h fires when count > 5 + 9. capability_denied_24h fires when count > 10 + 10. reasoning_errors_24h fires when execution_error count > 5 + 11. service_unhealthy fires per service in {degraded, unhealthy} + 12. service_unhealthy unhealthy → severity=critical + 13. service_unhealthy degraded → severity=warning + 14. slack_dm_reply_failed_24h fires when count > 3 + + Sort + shape: + 15. Severity sort: critical → warning → info + 16. Empty state (no triggers) → empty list + 17. Alert.to_dict shape matches FE Alert TS interface + + Fail-soft: + 18. Cost holder raises → other rules still emit + 19. Operational holder None → no cost alerts but webhook alerts emit + 20. JSONL reader raises → other rules still emit + 21. Probe snapshots accessor raises → other rules still emit + 22. compute_active_alerts NEVER raises + + Endpoint integration: + 23. GET /api/alerts/current returns expected top-level shape + 24. stub: false always + 25. by_severity sums to total_active + 26. SECURITY: walk-payload sweep — no PII / token shapes +""" + +from __future__ import annotations + +import json +import re +from datetime import datetime, timezone +from typing import Any, Dict, List, Optional +from unittest.mock import MagicMock, patch + +import pytest + +from kora_cli.alerts.aggregator import ( + CAPABILITY_DENIED_24H_THRESHOLD, + REASONING_ERRORS_24H_THRESHOLD, + SLACK_DM_REPLY_FAILED_24H_THRESHOLD, + WEBHOOK_DEAD_LETTER_24H_THRESHOLD, + Alert, + compute_active_alerts, +) + + +_EMAIL_ADDRESS = re.compile( + r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b" +) +_ANTHROPIC_KEY = re.compile(r"\bsk-ant-[A-Za-z0-9_-]{8,}\b") +_HEX_SECRET_SHAPE = re.compile(r"\b[0-9a-fA-F]{32,}\b") +_BEARER_TOKEN_SHAPE = re.compile( + r"\b(?:Bearer|Authorization)\s*[: ]\s*[A-Za-z0-9+/_.-]{8,}", + re.IGNORECASE, +) + + +# --------------------------------------------------------------------------- +# Helpers — synthesize source-state for mocks +# --------------------------------------------------------------------------- + + +def _make_cost_holder(rung_name: str, pct: float = 0.5): + from agent.cost_state_holder import CostRung + + rung = getattr(CostRung, rung_name) + holder = MagicMock() + holder.active_rung.return_value = rung + holder.current_pct_used.return_value = pct + return holder + + +def _make_operational_holder(primary_state_name: str): + from agent.operational_state import PrimaryState + + state = MagicMock() + state.primary_state = getattr(PrimaryState, primary_state_name) + holder = MagicMock() + holder.current = state # @property — set attribute on the mock + return holder + + +def _make_audit_entry(seam: str, details: Optional[dict] = None) -> Any: + """Build a mock that quacks like AuditEntry: ``.seam``, + ``.emitted_at``, ``.details``.""" + entry = MagicMock() + entry.seam = seam + entry.emitted_at = datetime.now(timezone.utc) + entry.details = details or {} + return entry + + +def _make_snapshot(name: str, status: str): + snap = MagicMock() + snap.name = name + snap.status = status + return snap + + +# Reusable patches: each helper module imports its dependency lazily, +# so we patch where the helper looks for it (NOT where it's defined). + + +@pytest.fixture +def patch_sources(): + """Yield a context manager that patches all 5 sources to baseline + (no-alert) shapes; individual tests override per-source.""" + from contextlib import ExitStack + + stack = ExitStack() + + cost_holder = _make_cost_holder("NORMAL") + op_holder = _make_operational_holder("ACTIVE") + + stack.enter_context( + patch( + "agent.cost_state_holder.get_cost_holder", + return_value=cost_holder, + ) + ) + stack.enter_context( + patch( + "agent.operational_state_holder.get_holder", + return_value=op_holder, + ) + ) + stack.enter_context( + patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + return_value=[], + ) + ) + stack.enter_context( + patch( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + return_value={}, + ) + ) + + sources = { + "cost_holder": cost_holder, + "op_holder": op_holder, + } + try: + yield sources, stack + finally: + stack.close() + + +# =========================================================================== +# Cost-ladder rules +# =========================================================================== + + +def test_cost_ladder_warned_fires(patch_sources): + sources, _ = patch_sources + sources["cost_holder"].active_rung.return_value = _enum("CostRung", "WARN_75") + sources["cost_holder"].current_pct_used.return_value = 0.80 + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "cost_ladder_warned" in ids + matching = next(a for a in alerts if a.id == "cost_ladder_warned") + assert matching.severity == "warning" + assert "80%" in matching.title + + +def test_cost_ladder_downshifted_fires(patch_sources): + sources, _ = patch_sources + sources["cost_holder"].active_rung.return_value = _enum( + "CostRung", "DOWNSHIFT_90" + ) + sources["cost_holder"].current_pct_used.return_value = 0.92 + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "cost_ladder_downshifted"] + assert len(matching) == 1 + assert matching[0].severity == "warning" + assert "92%" in matching[0].title + + +def test_cost_ladder_halted_fires(patch_sources): + sources, _ = patch_sources + sources["cost_holder"].active_rung.return_value = _enum( + "CostRung", "HARD_STOP_100" + ) + sources["cost_holder"].current_pct_used.return_value = 1.05 + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "cost_ladder_halted"] + assert len(matching) == 1 + assert matching[0].severity == "critical" + + +def test_cost_ladder_normal_no_alert(patch_sources): + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "cost_ladder_warned" not in ids + assert "cost_ladder_downshifted" not in ids + assert "cost_ladder_halted" not in ids + + +def _enum(enum_class_name: str, member_name: str): + if enum_class_name == "CostRung": + from agent.cost_state_holder import CostRung + + return getattr(CostRung, member_name) + if enum_class_name == "PrimaryState": + from agent.operational_state import PrimaryState + + return getattr(PrimaryState, member_name) + raise ValueError(enum_class_name) + + +# =========================================================================== +# Operational-state rules +# =========================================================================== + + +def test_operator_paused_fires(patch_sources): + sources, _ = patch_sources + sources["op_holder"].current.primary_state = _enum( + "PrimaryState", "PAUSED" + ) + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "operator_paused"] + assert len(matching) == 1 + assert matching[0].severity == "critical" + + +def test_operator_stopped_fires(patch_sources): + sources, _ = patch_sources + sources["op_holder"].current.primary_state = _enum( + "PrimaryState", "STOPPED" + ) + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "operator_stopped"] + assert len(matching) == 1 + assert matching[0].severity == "critical" + + +def test_operational_active_no_alert(patch_sources): + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "operator_paused" not in ids + assert "operator_stopped" not in ids + + +# =========================================================================== +# Audit JSONL rules +# =========================================================================== + + +def test_webhook_dead_letters_fires_when_over_threshold(patch_sources): + sources, _ = patch_sources + entries = [ + _make_audit_entry("webhook.dead_letter") + for _ in range(WEBHOOK_DEAD_LETTER_24H_THRESHOLD + 2) + ] + + def fake_read(seam=None, since=None): + return entries if seam == "webhook.dead_letter" else [] + + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=fake_read, + ): + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "webhook_dead_letters_24h"] + assert len(matching) == 1 + assert str(WEBHOOK_DEAD_LETTER_24H_THRESHOLD + 2) in matching[0].title + + +def test_webhook_dead_letters_below_threshold_no_alert(patch_sources): + sources, _ = patch_sources + entries = [ + _make_audit_entry("webhook.dead_letter") + for _ in range(WEBHOOK_DEAD_LETTER_24H_THRESHOLD) + ] + + def fake_read(seam=None, since=None): + return entries if seam == "webhook.dead_letter" else [] + + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=fake_read, + ): + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "webhook_dead_letters_24h" not in ids + + +def test_capability_denied_fires_when_over_threshold(patch_sources): + sources, _ = patch_sources + entries = [ + _make_audit_entry( + "mcp.tool_called", details={"result": "capability_denied"} + ) + for _ in range(CAPABILITY_DENIED_24H_THRESHOLD + 1) + ] + [ + # Mixed-in OK entries — should NOT count. + _make_audit_entry( + "mcp.tool_called", details={"result": "ok"} + ) + for _ in range(5) + ] + + def fake_read(seam=None, since=None): + return entries if seam == "mcp.tool_called" else [] + + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=fake_read, + ): + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "capability_denied_24h"] + assert len(matching) == 1 + assert matching[0].severity == "info" + # Title shows the matched-only count, not the total. + assert str(CAPABILITY_DENIED_24H_THRESHOLD + 1) in matching[0].title + + +def test_capability_denied_today_no_alert_since_audit_doesnt_emit_denials( + patch_sources, +): + """Forward-compat: today the audit emit at mcp_tools.py:714 runs + AFTER the cap-gate, so denial responses aren't logged. Rule + emits zero alerts in the current state — documented in the + aggregator module docstring.""" + sources, _ = patch_sources + # Only "ok" entries — no capability_denied results. + entries = [ + _make_audit_entry("mcp.tool_called", details={"result": "ok"}) + for _ in range(CAPABILITY_DENIED_24H_THRESHOLD + 50) + ] + + def fake_read(seam=None, since=None): + return entries if seam == "mcp.tool_called" else [] + + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=fake_read, + ): + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "capability_denied_24h" not in ids + + +def test_reasoning_errors_fires_when_over_threshold(patch_sources): + sources, _ = patch_sources + entries = [ + _make_audit_entry( + "reasoning.tool_called", + details={"tool_status": "execution_error"}, + ) + for _ in range(REASONING_ERRORS_24H_THRESHOLD + 1) + ] + [ + # Successful ones should not count. + _make_audit_entry( + "reasoning.tool_called", details={"tool_status": "ok"} + ) + for _ in range(3) + ] + + def fake_read(seam=None, since=None): + return entries if seam == "reasoning.tool_called" else [] + + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=fake_read, + ): + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "reasoning_errors_24h"] + assert len(matching) == 1 + assert matching[0].severity == "warning" + + +def test_slack_dm_reply_failed_fires_when_over_threshold(patch_sources): + sources, _ = patch_sources + entries = [ + _make_audit_entry("slack_dm.reply_failed") + for _ in range(SLACK_DM_REPLY_FAILED_24H_THRESHOLD + 1) + ] + + def fake_read(seam=None, since=None): + return entries if seam == "slack_dm.reply_failed" else [] + + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=fake_read, + ): + alerts = compute_active_alerts() + matching = [a for a in alerts if a.id == "slack_dm_reply_failed_24h"] + assert len(matching) == 1 + + +# =========================================================================== +# Service-snapshot rules +# =========================================================================== + + +def test_service_unhealthy_one_alert_per_affected_service(patch_sources): + sources, _ = patch_sources + snapshots = { + "vercel": _make_snapshot("vercel", "unhealthy"), + "sentry": _make_snapshot("sentry", "degraded"), + "doppler": _make_snapshot("doppler", "healthy"), + "supabase": _make_snapshot("supabase", "unknown"), + "fly": _make_snapshot("fly", "unhealthy"), + } + with patch( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + return_value=snapshots, + ): + alerts = compute_active_alerts() + service_alerts = [a for a in alerts if a.category == "service_unhealthy"] + # vercel (unhealthy) + sentry (degraded) + fly (unhealthy) = 3 + # doppler (healthy) + supabase (unknown) excluded + assert len(service_alerts) == 3 + ids = {a.id for a in service_alerts} + assert "service_unhealthy:vercel" in ids + assert "service_unhealthy:sentry" in ids + assert "service_unhealthy:fly" in ids + assert "service_unhealthy:doppler" not in ids + assert "service_unhealthy:supabase" not in ids + + +def test_service_unhealthy_severity_mapping(patch_sources): + sources, _ = patch_sources + snapshots = { + "vercel": _make_snapshot("vercel", "unhealthy"), + "sentry": _make_snapshot("sentry", "degraded"), + } + with patch( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + return_value=snapshots, + ): + alerts = compute_active_alerts() + vercel_alert = next(a for a in alerts if a.id == "service_unhealthy:vercel") + sentry_alert = next(a for a in alerts if a.id == "service_unhealthy:sentry") + assert vercel_alert.severity == "critical" + assert sentry_alert.severity == "warning" + + +def test_service_snapshots_empty_no_alerts(patch_sources): + alerts = compute_active_alerts() + service_alerts = [a for a in alerts if a.category == "service_unhealthy"] + assert service_alerts == [] + + +# =========================================================================== +# Sort + shape +# =========================================================================== + + +def test_severity_sort_critical_first(patch_sources): + sources, _ = patch_sources + sources["cost_holder"].active_rung.return_value = _enum( + "CostRung", "HARD_STOP_100" + ) + sources["cost_holder"].current_pct_used.return_value = 1.0 + sources["op_holder"].current.primary_state = _enum( + "PrimaryState", "ACTIVE" + ) + snapshots = { + "sentry": _make_snapshot("sentry", "degraded"), + } + with patch( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + return_value=snapshots, + ): + alerts = compute_active_alerts() + # critical (cost_ladder_halted) must come before warning (sentry). + severities = [a.severity for a in alerts] + assert severities[0] == "critical" + assert "warning" in severities + + +def test_empty_state_returns_empty_list(patch_sources): + alerts = compute_active_alerts() + assert alerts == [] + + +def test_alert_to_dict_matches_fe_shape(patch_sources): + """Alert.to_dict keys must match the FE EmailMessage TS interface + exactly: id, severity, category, title, detail, source_panel, + source_panel_route, first_seen_at.""" + sources, _ = patch_sources + sources["op_holder"].current.primary_state = _enum( + "PrimaryState", "PAUSED" + ) + alerts = compute_active_alerts() + assert len(alerts) >= 1 + d = alerts[0].to_dict() + assert set(d.keys()) == { + "id", + "severity", + "category", + "title", + "detail", + "source_panel", + "source_panel_route", + "first_seen_at", + } + + +# =========================================================================== +# Fail-soft +# =========================================================================== + + +def test_cost_holder_raises_other_rules_still_emit(patch_sources): + sources, _ = patch_sources + sources["cost_holder"].active_rung.side_effect = RuntimeError( + "kaboom" + ) + sources["op_holder"].current.primary_state = _enum( + "PrimaryState", "PAUSED" + ) + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + # Cost rule shouldn't emit but operator_paused must still fire. + assert "operator_paused" in ids + # No cost alerts at all. + assert not any(a.id.startswith("cost_ladder") for a in alerts) + + +def test_operational_holder_none_other_rules_still_emit(patch_sources): + sources, _ = patch_sources + with patch( + "agent.operational_state_holder.get_holder", return_value=None + ): + sources["cost_holder"].active_rung.return_value = _enum( + "CostRung", "WARN_75" + ) + sources["cost_holder"].current_pct_used.return_value = 0.80 + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "cost_ladder_warned" in ids + assert "operator_paused" not in ids + assert "operator_stopped" not in ids + + +def test_audit_reader_raises_other_rules_still_emit(patch_sources): + sources, _ = patch_sources + sources["op_holder"].current.primary_state = _enum( + "PrimaryState", "STOPPED" + ) + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=OSError("disk crash"), + ): + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "operator_stopped" in ids + + +def test_probe_snapshots_raises_other_rules_still_emit(patch_sources): + sources, _ = patch_sources + sources["op_holder"].current.primary_state = _enum( + "PrimaryState", "PAUSED" + ) + with patch( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + side_effect=RuntimeError("probe runner crashed"), + ): + alerts = compute_active_alerts() + ids = [a.id for a in alerts] + assert "operator_paused" in ids + + +def test_compute_active_alerts_never_raises(monkeypatch): + """Even if EVERY source raises, the aggregator returns a list + (possibly empty) rather than propagating.""" + monkeypatch.setattr( + "agent.cost_state_holder.get_cost_holder", + MagicMock(side_effect=RuntimeError("cost dead")), + ) + monkeypatch.setattr( + "agent.operational_state_holder.get_holder", + MagicMock(side_effect=RuntimeError("op dead")), + ) + monkeypatch.setattr( + "kora_cli.audit.jsonl_reader.read_audit_entries", + MagicMock(side_effect=RuntimeError("audit dead")), + ) + monkeypatch.setattr( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + MagicMock(side_effect=RuntimeError("probe dead")), + ) + alerts = compute_active_alerts() + assert isinstance(alerts, list) + + +# =========================================================================== +# Endpoint integration +# =========================================================================== + + +@pytest.fixture +def _isolate_endpoint(tmp_path, monkeypatch): + """Apply CC#2 #137 fixture-isolation discipline + reset all + accessor sources to baseline no-alert state.""" + monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setattr("kora_constants.get_kora_home", lambda: tmp_path) + monkeypatch.setattr("kora_cli.config.get_kora_home", lambda: tmp_path) + monkeypatch.setattr( + "kora_cli.web_server.get_kora_home", lambda: tmp_path + ) + cost_holder = _make_cost_holder("NORMAL") + op_holder = _make_operational_holder("ACTIVE") + monkeypatch.setattr( + "agent.cost_state_holder.get_cost_holder", + lambda: cost_holder, + ) + monkeypatch.setattr( + "agent.operational_state_holder.get_holder", + lambda: op_holder, + ) + monkeypatch.setattr( + "kora_cli.audit.jsonl_reader.read_audit_entries", + lambda **kwargs: [], + ) + monkeypatch.setattr( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + lambda: {}, + ) + return {"cost_holder": cost_holder, "op_holder": op_holder} + + +@pytest.mark.asyncio +async def test_endpoint_returns_expected_shape(_isolate_endpoint): + from kora_cli import web_server + + result = await web_server.list_current_alerts() + assert set(result.keys()) == { + "alerts", + "stub", + "generated_at", + "total_active", + "by_severity", + } + assert result["stub"] is False + assert isinstance(result["alerts"], list) + assert isinstance(result["by_severity"], dict) + + +@pytest.mark.asyncio +async def test_endpoint_no_alerts_returns_empty_list(_isolate_endpoint): + from kora_cli import web_server + + result = await web_server.list_current_alerts() + assert result["alerts"] == [] + assert result["total_active"] == 0 + assert result["by_severity"] == {"critical": 0, "warning": 0, "info": 0} + + +@pytest.mark.asyncio +async def test_endpoint_by_severity_reconciles_total(_isolate_endpoint): + _isolate_endpoint["cost_holder"].active_rung.return_value = _enum( + "CostRung", "HARD_STOP_100" + ) + _isolate_endpoint["cost_holder"].current_pct_used.return_value = 1.0 + _isolate_endpoint["op_holder"].current.primary_state = _enum( + "PrimaryState", "PAUSED" + ) + from kora_cli import web_server + + result = await web_server.list_current_alerts() + by_sev_sum = sum(result["by_severity"].values()) + assert by_sev_sum == result["total_active"] + assert result["total_active"] >= 2 + assert result["by_severity"]["critical"] >= 2 + + +# =========================================================================== +# SECURITY — walk-payload sweep +# =========================================================================== + + +@pytest.mark.asyncio +async def test_no_pii_or_secret_shapes_in_payload(_isolate_endpoint): + """Trigger many rules to exercise diverse alert text; sweep the + whole serialized response for PII / secret shapes.""" + _isolate_endpoint["cost_holder"].active_rung.return_value = _enum( + "CostRung", "HARD_STOP_100" + ) + _isolate_endpoint["cost_holder"].current_pct_used.return_value = 1.0 + _isolate_endpoint["op_holder"].current.primary_state = _enum( + "PrimaryState", "PAUSED" + ) + + def lots_of_entries(seam=None, since=None): + if seam == "webhook.dead_letter": + return [_make_audit_entry(seam) for _ in range(20)] + if seam == "reasoning.tool_called": + return [ + _make_audit_entry( + seam, details={"tool_status": "execution_error"} + ) + for _ in range(10) + ] + if seam == "slack_dm.reply_failed": + return [_make_audit_entry(seam) for _ in range(15)] + return [] + + with patch( + "kora_cli.audit.jsonl_reader.read_audit_entries", + side_effect=lots_of_entries, + ), patch( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + return_value={"vercel": _make_snapshot("vercel", "unhealthy")}, + ): + from kora_cli import web_server + + result = await web_server.list_current_alerts() + + blob = json.dumps(result) + assert _EMAIL_ADDRESS.findall(blob) == [] + assert _ANTHROPIC_KEY.findall(blob) == [] + assert _HEX_SECRET_SHAPE.findall(blob) == [] + assert _BEARER_TOKEN_SHAPE.findall(blob) == [] diff --git a/tests/kora_cli/test_web_server_alerts.py b/tests/kora_cli/test_web_server_alerts.py index bd470487abb9..9623c55fcea1 100644 --- a/tests/kora_cli/test_web_server_alerts.py +++ b/tests/kora_cli/test_web_server_alerts.py @@ -1,27 +1,22 @@ -"""Tests for the KR-ALERTS-PANEL stub endpoint + banner. - -Bucket §2 scenarios: - 1. GET /api/alerts/current returns 200 - 2. Top-level shape (alerts + stub:true + generated_at + - total_active + by_severity) - 3. 4 representative stub alerts present - 4. Stub spans all 3 severity tiers (critical / warning / info) - so the operator's first look exercises the severity sort + - banner border-tone mapping - 5. Per-entry shape + valid severity enum + source_panel_route - uses the flat ``/`` FE convention (not /admin/) - 6. SECURITY: walk-payload sweeps for token shapes (Anthropic - sk-ant-, Slack xox*-, HMAC hex), email PII, raw Slack U-IDs - 7. SECURITY: companion FE pin — AlertsPanel.tsx never uses - dangerouslySetInnerHTML for title/detail - 8. SECURITY: companion FE pin — AlertsBanner.tsx uses - sessionStorage (per-tab dismissal), NOT localStorage - (which would persist across sessions and silence alerts - wrongly) - 9. Empty state: AlertsPanel renders positive-reinforcement - CheckCircle2 + "Daemon healthy" when alerts.length === 0 - 10. by_severity sum reconciles to total_active - 11. Cron-regression sanity +"""Tests for the KR-ALERTS-PANEL endpoint (post KR-ALERTS-PANEL-FLIP). + +After the flip the endpoint reads from the live aggregator +(``kora_cli/alerts/aggregator.py``) instead of the v1 stub. This +module keeps the original PR #134 walk-payload security guards + +FE source pins that apply to BOTH the old stub and the new live +endpoint: + + * Top-level response shape (now with ``stub: false`` always) + * Walk-payload security guards (no Anthropic/Slack token shapes, + no email/Slack-ID PII, no long-hex secret shapes) + * FE source pins (no dangerouslySetInnerHTML, title/detail + rendered as JSX children, sessionStorage vs localStorage + dismissal, empty-state daemon-healthy view) + * by_severity / total_active reconciliation + * Cron-regression sanity + +Rule-trigger tests + per-rule severity verification live in +``tests/kora_cli/alerts/test_aggregator.py``. """ import re @@ -56,14 +51,42 @@ def _strip_ts_comments(src: str) -> str: @pytest.fixture(autouse=True) def _isolate_config(tmp_path, monkeypatch): + """CC#2 #137 fixture-isolation discipline + reset all aggregator + sources to baseline (no-alert) state. The aggregator pulls from + 5 holders/accessors; without all-5 monkeypatching, parallel + pytest-xdist workers can see each other's state.""" monkeypatch.setenv("HERMES_HOME", str(tmp_path)) + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setattr("kora_constants.get_kora_home", lambda: tmp_path) monkeypatch.setattr("kora_cli.config.get_kora_home", lambda: tmp_path) + monkeypatch.setattr( + "kora_cli.web_server.get_kora_home", lambda: tmp_path + ) monkeypatch.setattr( "kora_cli.config.get_config_path", lambda: tmp_path / "config.yaml" ) monkeypatch.setattr( "kora_cli.config.get_env_path", lambda: tmp_path / ".env" ) + # Reset aggregator sources to no-alert baseline so this file's + # walk-payload + shape tests don't accidentally trigger rules + # (which would change the payload they're sweeping). + monkeypatch.setattr( + "agent.cost_state_holder.get_cost_holder", + lambda: None, + ) + monkeypatch.setattr( + "agent.operational_state_holder.get_holder", + lambda: None, + ) + monkeypatch.setattr( + "kora_cli.audit.jsonl_reader.read_audit_entries", + lambda **kwargs: [], + ) + monkeypatch.setattr( + "kora_cli.heartbeat_probes.runner.current_service_snapshots", + lambda: {}, + ) return tmp_path @@ -83,6 +106,8 @@ async def test_endpoint_returns_200(_isolate_config): @pytest.mark.asyncio async def test_response_shape_has_required_keys(_isolate_config): + """Top-level shape stays the same post-flip; ``stub`` is now + always ``False`` (live aggregator, even when no alerts active).""" from kora_cli import web_server result = await web_server.list_current_alerts() @@ -97,40 +122,19 @@ async def test_response_shape_has_required_keys(_isolate_config): assert isinstance(result["generated_at"], str) assert isinstance(result["total_active"], int) assert isinstance(result["by_severity"], dict) - assert result["stub"] is True - - -# ---- 3. Expected stub alerts ---------------------------------------- - - -@pytest.mark.asyncio -async def test_stub_returns_four_representative_alerts(_isolate_config): - """Pin the bucket §1(a) canonical 4-alert stub list. The deferred - real-data collector will swap the body but shape must stay - stable so the FE banner + panel render correctly during - cut-over.""" - from kora_cli import web_server - - result = await web_server.list_current_alerts() - assert len(result["alerts"]) == 4 - ids = {a["id"] for a in result["alerts"]} - assert ids == {"stub-1", "stub-2", "stub-3", "stub-4"} + assert result["stub"] is False @pytest.mark.asyncio -async def test_stub_spans_all_three_severity_tiers(_isolate_config): - """The 4 stub alerts deliberately span critical + warning + info - so the operator's first look exercises: - * severity sort order (critical → warning → info) - * banner border-tone mapping (red / yellow / blue) - * category icon variety - Pin so a future stub edit can't homogenize to one severity tier - that would mask the visual differentiation.""" +async def test_no_alerts_baseline_returns_empty_list(_isolate_config): + """With all aggregator sources patched to no-alert baseline + (per the fixture), the response is an empty list + stub:false.""" from kora_cli import web_server result = await web_server.list_current_alerts() - severities = {a["severity"] for a in result["alerts"]} - assert severities == {"critical", "warning", "info"} + assert result["alerts"] == [] + assert result["total_active"] == 0 + assert result["by_severity"] == {"critical": 0, "warning": 0, "info": 0} # ---- 4. Per-entry shape + enums + route convention -----------------