diff --git a/kora_cli/alerts/wake_consumer.py b/kora_cli/alerts/wake_consumer.py new file mode 100644 index 000000000000..649fc85f49bf --- /dev/null +++ b/kora_cli/alerts/wake_consumer.py @@ -0,0 +1,790 @@ +"""Alert wake-event CONSUMER — KR-ALERT-INVESTIGATION-WAKE-CONSUMER. + +Activates the ``alert_investigation`` cost-telemetry route literal +end-to-end. Parallels :mod:`kora_cli.probes.wake_consumer` (#166) +for alert events: tails the ``notification.dispatched`` audit seam +(#149 KR-ALERT-NOTIFY) → invokes reasoning with alert context → +DMs operator with the investigation result → emits the new +``alert.investigation_completed`` audit seam. + +# Wake trigger + +The alert notifier already emits ``notification.dispatched`` rows +per alert dispatched (#149); per-alert rows carry ``alert_id`` / +``severity`` / ``category`` / ``channel`` / ``status`` (success/ +fail). Burst-summary + digest rows use synthetic alert_ids +(``burst:N`` / ``digest:N``) — those are skipped (they're +aggregates, not single alerts worth investigating). + +# 4-stream join precedent (matches probe wake consumer) + + 1. ``notification.dispatched`` — alert emitted (existing) + 2. ``alert.investigation_completed`` — investigation done (THIS PR) + 3. ``slack_dm_log.jsonl`` entry — DM with investigation summary + 4. (future) ``tool.alert_autoresolve_attempted`` — when alert + envelope auto-actions get built; reserved seam, not emitted v1 + +CC#2 follow-on (KR-FE-ALERT-INVESTIGATIONS-VIEWER) joins the +4 streams via ``caller_session_id = "alert:{category}:{severity}"``. + +# Debounce policy + +Inline: per (category, severity) → datetime of last dispatched +investigation. Default 10 min window via +``KORA_ALERT_WAKE_DEBOUNCE_SECONDS``. Critical alerts can +optionally bypass via ``KORA_ALERT_WAKE_DEBOUNCE_BYPASS_CRITICAL`` +(default false — fail-closed; even critical wakes debounce +unless operator opts in). + +# Fail-soft + +Every external dependency is fail-soft: + * Engine None / engine raises → fallback DM with verbatim alert + reason + * Slack client None / channel_id unset → log warning + outbound-log entry + * Telemetry / audit record failures → log + continue (DM still sent) + +The listener that drives this consumer must NOT crash on any +single event — failures are recorded + cycle continues. +""" + +from __future__ import annotations + +import logging +import os +import threading +import time +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Any, Callable, Dict, Optional, Tuple + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Env vars + defaults +# --------------------------------------------------------------------------- + + +DEBOUNCE_SECONDS_ENV = "KORA_ALERT_WAKE_DEBOUNCE_SECONDS" +DEFAULT_DEBOUNCE_SECONDS = 600 # 10 min; matches probe wake consumer default + +BYPASS_CRITICAL_ENV = "KORA_ALERT_WAKE_DEBOUNCE_BYPASS_CRITICAL" +JOSHUA_SLACK_USER_ID_ENV = "KORA_SLACK_JOSHUA_USER_ID" + + +# Per-channel filter for the wake trigger. Per-alert rows ride +# either "slack" or "email"; the burst-summary + digest emits use +# synthetic "burst:N" / "digest:N" alert_ids on the same seam and +# are skipped at the consumer level. +_PER_ALERT_CHANNEL_VALUES = frozenset({"slack", "email"}) + + +def _read_debounce_seconds() -> int: + raw = os.environ.get(DEBOUNCE_SECONDS_ENV, "").strip() + if not raw: + return DEFAULT_DEBOUNCE_SECONDS + try: + value = int(raw) + except ValueError: + logger.warning( + "[kora.alert_wake_consumer] %s=%r is not numeric; using " + "default %ds", + DEBOUNCE_SECONDS_ENV, + raw, + DEFAULT_DEBOUNCE_SECONDS, + ) + return DEFAULT_DEBOUNCE_SECONDS + if value < 0: + return DEFAULT_DEBOUNCE_SECONDS + return value + + +def _read_bypass_critical() -> bool: + raw = os.environ.get(BYPASS_CRITICAL_ENV, "").strip().lower() + return raw in {"true", "1", "yes", "on"} + + +# --------------------------------------------------------------------------- +# Outcome (telemetry + test surface) +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True, slots=True) +class AlertWakeOutcome: + """One per-event outcome. Bundled for tests + listener telemetry.""" + + alert_id: str + category: str + severity: str + dispatched: bool + reasoning_invoked: bool + dm_sent: bool + debounce_skipped: bool = False + # ``filtered_skipped`` distinguishes "not a per-alert row" + # (burst / digest) from "debounced or dispatched". Lets the + # listener telemetry break out aggregate-vs-actual. + filtered_skipped: bool = False + error: Optional[str] = None + + +# --------------------------------------------------------------------------- +# AlertWakeConsumer +# --------------------------------------------------------------------------- + + +SlackClientFactory = Callable[[], Optional[Any]] +ReasoningEngineFactory = Callable[[], Optional[Any]] + + +class AlertWakeConsumer: + """Per-event handler for ``notification.dispatched`` rows. + + Stateful only via the debounce map (per (category, severity); + NOT per alert_id because most categories include the source-rule + id in the alert_id and operator wants one investigation per + distinct category-x-severity combination, not per identical + re-fire). + """ + + def __init__( + self, + *, + reasoning_engine_factory: ReasoningEngineFactory, + slack_client_factory: SlackClientFactory, + operator_channel_id_resolver: Callable[[], str] = ( + lambda: os.environ.get(JOSHUA_SLACK_USER_ID_ENV, "").strip() + ), + ) -> None: + self._reasoning_engine_factory = reasoning_engine_factory + self._slack_client_factory = slack_client_factory + self._operator_channel_id_resolver = operator_channel_id_resolver + self._debounce_lock = threading.RLock() + self._last_dispatched: Dict[Tuple[str, str], datetime] = {} + + @property + def debounce_map_size(self) -> int: + return len(self._last_dispatched) + + def reset_debounce_state(self) -> None: + """Clear the in-memory debounce map. Listener shutdown calls + this so subsequent listener start sees a clean slate.""" + with self._debounce_lock: + self._last_dispatched = {} + + # ------------------------------------------------------------------ + # Debounce + # ------------------------------------------------------------------ + + def _is_debounced(self, category: str, severity: str) -> bool: + if severity == "critical" and _read_bypass_critical(): + return False + window = _read_debounce_seconds() + if window <= 0: + return False + with self._debounce_lock: + last = self._last_dispatched.get((category, severity)) + if last is None: + return False + elapsed = (datetime.now(timezone.utc) - last).total_seconds() + return elapsed < window + + def _mark_dispatched(self, category: str, severity: str) -> None: + with self._debounce_lock: + self._last_dispatched[(category, severity)] = datetime.now( + timezone.utc + ) + + # ------------------------------------------------------------------ + # Public entry point + # ------------------------------------------------------------------ + + async def consume_alert_event( + self, event_details: Dict[str, Any] + ) -> AlertWakeOutcome: + """Process one ``notification.dispatched`` event. + + ``event_details`` is the ``details`` dict from the audit row + per AlertNotifier's shape: ``channel`` / ``alert_id`` / + ``severity`` / ``category`` / ``status`` (+ ``error`` on + failed status). + + Fail-soft contract: every path either dispatches OR returns + a structured outcome explaining why it didn't. Never raises. + """ + alert_id = str(event_details.get("alert_id") or "") + category = str(event_details.get("category") or "unknown") + severity = str(event_details.get("severity") or "warning") + channel = str(event_details.get("channel") or "") + status = str(event_details.get("status") or "") + + # Filter aggregate emits (burst_summary / digest_email) + + # rows for failed dispatches. Only investigate alerts that + # actually reached the operator's surface. + if channel not in _PER_ALERT_CHANNEL_VALUES: + return AlertWakeOutcome( + alert_id=alert_id, + category=category, + severity=severity, + dispatched=False, + reasoning_invoked=False, + dm_sent=False, + filtered_skipped=True, + ) + if status != "ok": + return AlertWakeOutcome( + alert_id=alert_id, + category=category, + severity=severity, + dispatched=False, + reasoning_invoked=False, + dm_sent=False, + filtered_skipped=True, + ) + + if self._is_debounced(category, severity): + logger.debug( + "[kora.alert_wake_consumer] debounced category=%s " + "severity=%s", + category, + severity, + ) + return AlertWakeOutcome( + alert_id=alert_id, + category=category, + severity=severity, + dispatched=False, + reasoning_invoked=False, + dm_sent=False, + debounce_skipped=True, + ) + + investigation_started_at = datetime.now(timezone.utc) + investigation_started_monotonic = time.monotonic() + caller_session_id = f"alert:{category}:{severity}" + + reasoning_text: str + reasoning_invoked = False + reasoning_error: Optional[str] = None + reasoning_result: Optional[Any] = None + engine = self._reasoning_engine_factory() + if engine is None: + reasoning_text = format_fallback_text( + event_details, reason="engine_unavailable" + ) + reasoning_error = "engine_unavailable" + logger.warning( + "[kora.alert_wake_consumer] reasoning engine unavailable; " + "sending fallback DM category=%s", + category, + ) + else: + try: + ( + invocation_text, + invocation_error, + invocation_result, + ) = await self._invoke_reasoning( + engine=engine, event_details=event_details + ) + except Exception as exc: + reasoning_text = format_fallback_text( + event_details, + reason=f"engine_exception:{type(exc).__name__}", + ) + reasoning_error = f"engine_exception:{type(exc).__name__}" + logger.warning( + "[kora.alert_wake_consumer] engine.respond raised " + "%r category=%s — sending fallback DM", + exc, + category, + ) + else: + reasoning_result = invocation_result + if invocation_error is None: + reasoning_text = invocation_text + reasoning_invoked = True + else: + reasoning_text = format_fallback_text( + event_details, reason=invocation_error + ) + reasoning_error = invocation_error + logger.warning( + "[kora.alert_wake_consumer] engine returned " + "error=%s category=%s — sending fallback DM", + invocation_error, + category, + ) + + # Stamp dispatched BEFORE attempting DM so a flapping Slack + # client can't trigger duplicate investigations in the next + # cycle (probe wake consumer precedent). + self._mark_dispatched(category, severity) + + dm_outcome = await self._send_operator_dm_routed( + category=category, + severity=severity, + text=reasoning_text, + caller_session_id=caller_session_id, + reasoning_result=reasoning_result, + reasoning_error=reasoning_error, + investigation_started_monotonic=investigation_started_monotonic, + ) + dm_sent = dm_outcome["dm_sent"] + dm_status = dm_outcome["dm_status"] + + self._emit_investigation_completed( + alert_id=alert_id, + category=category, + severity=severity, + caller_session_id=caller_session_id, + reasoning_result=reasoning_result, + reasoning_error=reasoning_error, + reasoning_text_for_dm=reasoning_text, + dm_status=dm_status, + investigation_started_monotonic=investigation_started_monotonic, + ) + + return AlertWakeOutcome( + alert_id=alert_id, + category=category, + severity=severity, + dispatched=True, + reasoning_invoked=reasoning_invoked, + dm_sent=dm_sent, + debounce_skipped=False, + error=reasoning_error, + ) + + # ------------------------------------------------------------------ + # Reasoning invocation + # ------------------------------------------------------------------ + + async def _invoke_reasoning( + self, *, engine: Any, event_details: Dict[str, Any] + ) -> Tuple[str, Optional[str], Optional[Any]]: + """Build the IncomingMessage with ``source="alert_investigation"``, + call engine.respond. Telemetry route attribution fires inside + the engine — :func:`_record_call_to_telemetry` maps the + source to ``ROUTE_ALERT_INVESTIGATION`` (per #190's wire). + """ + from kora_cli.reasoning.engine import ( + ConversationContext, + IncomingMessage, + ) + + message = IncomingMessage( + text=format_investigation_prompt(event_details), + source="alert_investigation", + received_at=datetime.now(timezone.utc), + metadata={ + "alert_id": event_details.get("alert_id") or "unknown", + "category": event_details.get("category") or "unknown", + "severity": event_details.get("severity") or "warning", + "channel": event_details.get("channel") or "unknown", + }, + ) + context = ConversationContext( + recent_messages=[], + current_operational_state="unknown", + current_cost_ladder_rung="unknown", + ) + result = await engine.respond(message, context) + engine_error = getattr(result, "error", None) + if engine_error is not None: + return ("", str(engine_error), result) + text = getattr(result, "text", "") or "" + if not text.strip(): + return ("", "empty_response_text", result) + return (text, None, result) + + # ------------------------------------------------------------------ + # Outbound DM + # ------------------------------------------------------------------ + + async def _send_operator_dm_routed( + self, + *, + category: str, + severity: str, + text: str, + caller_session_id: str, + reasoning_result: Optional[Any], + reasoning_error: Optional[str], + investigation_started_monotonic: float, + ) -> Dict[str, Any]: + """Mirror of probe wake consumer's _send_operator_dm_routed. + + Returns ``{"dm_sent": bool, "dm_status": str}`` with the + same vocabulary as probe wake consumer so CC#2's viewers + can share the dm_status enum. + """ + client = self._slack_client_factory() + channel_id = self._operator_channel_id_resolver() + is_fallback = reasoning_error is not None + + if client is None or not channel_id: + if client is None: + logger.warning( + "[kora.alert_wake_consumer] slack_client_unavailable; " + "DM not sent category=%s", + category, + ) + else: + logger.warning( + "[kora.alert_wake_consumer] %s unset; DM not sent " + "category=%s", + JOSHUA_SLACK_USER_ID_ENV, + category, + ) + self._append_outbound_log( + channel_id=channel_id or "", + text=text, + slack_message_ts=None, + send_status="failed", + failure_reason=( + "slack_client_unavailable" + if client is None + else "channel_id_unset" + ), + caller_session_id=caller_session_id, + reasoning_result=reasoning_result, + reasoning_error=reasoning_error, + investigation_started_monotonic=investigation_started_monotonic, + ) + return { + "dm_sent": False, + "dm_status": ( + "engine_unavailable_failed_send" + if is_fallback + else "failed_send" + ), + } + + dm_text = format_operator_dm( + category=category, severity=severity, reasoning_text=text + ) + post_response: Optional[Dict[str, Any]] = None + send_exception: Optional[BaseException] = None + try: + post_response = await client.post_dm( + channel_id=channel_id, text=dm_text + ) + except Exception as exc: + send_exception = exc + logger.warning( + "[kora.alert_wake_consumer] post_dm raised %r " + "category=%s", + exc, + category, + ) + + slack_message_ts: Optional[str] = None + if isinstance(post_response, dict): + ts_raw = post_response.get("ts") + if isinstance(ts_raw, str): + slack_message_ts = ts_raw + + if send_exception is not None: + self._append_outbound_log( + channel_id=channel_id, + text=dm_text, + slack_message_ts=None, + send_status="failed", + failure_reason=f"post_dm_raised:{type(send_exception).__name__}", + caller_session_id=caller_session_id, + reasoning_result=reasoning_result, + reasoning_error=reasoning_error, + investigation_started_monotonic=investigation_started_monotonic, + ) + return { + "dm_sent": False, + "dm_status": ( + "engine_unavailable_failed_send" + if is_fallback + else "failed_send" + ), + } + + self._append_outbound_log( + channel_id=channel_id, + text=dm_text, + slack_message_ts=slack_message_ts, + send_status="ok", + failure_reason=None, + caller_session_id=caller_session_id, + reasoning_result=reasoning_result, + reasoning_error=reasoning_error, + investigation_started_monotonic=investigation_started_monotonic, + ) + return { + "dm_sent": True, + "dm_status": ( + "engine_unavailable_fallback" if is_fallback else "sent" + ), + } + + def _append_outbound_log( + self, + *, + channel_id: str, + text: str, + slack_message_ts: Optional[str], + send_status: str, + failure_reason: Optional[str], + caller_session_id: str, + reasoning_result: Optional[Any], + reasoning_error: Optional[str], + investigation_started_monotonic: float, + ) -> None: + try: + from kora_cli.handlers.slack_dm_handler import ( + append_outbound_log_entry, + resolve_slack_dm_log_path, + ) + except Exception as exc: + logger.warning( + "[kora.alert_wake_consumer] outbound log import failed: " + "%r — slack_dm_log entry skipped", + exc, + ) + return + + duration_ms = int( + (time.monotonic() - investigation_started_monotonic) * 1000 + ) + meta = _reasoning_meta_from_result(reasoning_result) + try: + append_outbound_log_entry( + log_path=resolve_slack_dm_log_path(), + channel_id=channel_id, + thread_ts=None, + text=text, + slack_message_ts=slack_message_ts, + send_status=send_status, + failure_reason=failure_reason, + model_used=meta.get("model_used"), + input_tokens=meta.get("input_tokens"), + output_tokens=meta.get("output_tokens"), + reasoning_duration_ms=duration_ms, + reasoning_error=reasoning_error, + cache_creation_input_tokens=meta.get( + "cache_creation_input_tokens" + ), + cache_read_input_tokens=meta.get("cache_read_input_tokens"), + caller_session_id=caller_session_id, + ) + except Exception as exc: + logger.warning( + "[kora.alert_wake_consumer] outbound log write raised " + "%r — investigation continues", + exc, + ) + + def _emit_investigation_completed( + self, + *, + alert_id: str, + category: str, + severity: str, + caller_session_id: str, + reasoning_result: Optional[Any], + reasoning_error: Optional[str], + reasoning_text_for_dm: str, + dm_status: str, + investigation_started_monotonic: float, + ) -> None: + try: + from kora_cli.audit.jsonl_sink import emit_audit + except Exception as exc: + logger.warning( + "[kora.alert_wake_consumer] audit import failed: %r " + "— investigation_completed row skipped", + exc, + ) + return + + meta = _reasoning_meta_from_result(reasoning_result) + cost_usd = _compute_total_cost_usd(meta) + duration_ms = int( + (time.monotonic() - investigation_started_monotonic) * 1000 + ) + + details: Dict[str, Any] = { + "alert_id": alert_id, + "category": category, + "severity": severity, + "model_used": meta.get("model_used"), + "input_tokens": meta.get("input_tokens"), + "output_tokens": meta.get("output_tokens"), + "cache_creation_input_tokens": meta.get( + "cache_creation_input_tokens" + ), + "cache_read_input_tokens": meta.get("cache_read_input_tokens"), + "total_cost_usd": cost_usd, + "investigation_duration_ms": duration_ms, + "investigation_summary_text": reasoning_text_for_dm, + "dm_status": dm_status, + # Reserved for the future alert-envelope autoaction + # parallel to probe autofix. v1 always false; the seam + # field is in the payload now so consumers don't have + # to branch on presence later. + "autoaction_attempted": False, + } + if reasoning_error is not None: + details["reasoning_error"] = reasoning_error + + try: + emit_audit( + "alert.investigation_completed", + details, + caller_session_id=caller_session_id, + source="reasoning", + ) + except Exception as exc: + logger.warning( + "[kora.alert_wake_consumer] emit_audit raised %r — " + "alert.investigation_completed row skipped", + exc, + ) + + +# --------------------------------------------------------------------------- +# Formatters (pure functions — easy to test in isolation) +# --------------------------------------------------------------------------- + + +_SEVERITY_EMOJI = { + "critical": "🚨", + "warning": "⚠️", + "info": "ℹ️", +} + + +def format_investigation_prompt(event_details: Dict[str, Any]) -> str: + """Build the prompt the reasoning engine receives. + + Mirrors the probe wake consumer's structure: alert identity + + severity + category, then an instruction to keep the response + operator-friendly + Slack-DM-sized. Alert events DON'T carry + title/detail in the audit row (the AlertNotifier audit shape + is alert_id/severity/category/channel/status only) so the + prompt asks the engine to reason from the categorical signal + alone, augmented by whatever the engine pulls via tools. + """ + alert_id = event_details.get("alert_id") or "unknown" + category = event_details.get("category") or "unknown" + severity = event_details.get("severity") or "warning" + channel = event_details.get("channel") or "unknown" + + lines = [ + f"Alert dispatched: {category} (severity: {severity})", + f"Alert id: {alert_id}", + f"Dispatched via: {channel}", + "", + "Investigate what triggered this alert and propose the next", + "action(s). The response is sent verbatim to the operator as", + "a Slack DM — keep it concise (2-4 sentences for diagnosis +", + "1 line for recommended next step). Use plain text; no", + "markdown headers.", + ] + return "\n".join(lines) + + +def format_operator_dm( + *, category: str, severity: str, reasoning_text: str +) -> str: + """Slack DM body the operator receives. Header carries the + severity emoji + category; body is reasoning text verbatim.""" + emoji = _SEVERITY_EMOJI.get(severity, "🔔") + return f"{emoji} Alert · {category}\n{reasoning_text}" + + +def format_fallback_text( + event_details: Dict[str, Any], *, reason: str +) -> str: + """When reasoning fails, send the alert details verbatim + the + failure reason. Operator still gets actionable signal.""" + category = event_details.get("category") or "unknown" + severity = event_details.get("severity") or "warning" + alert_id = event_details.get("alert_id") or "unknown" + return ( + f"{category} ({severity}): alert id {alert_id}\n" + f"\n" + f"I was unable to investigate — engine returned: {reason}" + ) + + +# --------------------------------------------------------------------------- +# Per-investigation helpers — shape mirrors probe wake consumer's +# _reasoning_meta_from_result + _compute_total_cost_usd so a future +# refactor can extract them into a shared module without divergence. +# --------------------------------------------------------------------------- + + +def _reasoning_meta_from_result(result: Optional[Any]) -> Dict[str, Any]: + """Project a ResponseResult into the 5-key meta dict shared + between outbound log + investigation_completed audit.""" + if result is None: + return { + "model_used": None, + "input_tokens": None, + "output_tokens": None, + "cache_creation_input_tokens": None, + "cache_read_input_tokens": None, + } + return { + "model_used": getattr(result, "model_used", None) or None, + "input_tokens": getattr(result, "input_tokens", None), + "output_tokens": getattr(result, "output_tokens", None), + "cache_creation_input_tokens": getattr( + result, "cache_creation_input_tokens", None + ), + "cache_read_input_tokens": getattr( + result, "cache_read_input_tokens", None + ), + } + + +def _compute_total_cost_usd(meta: Dict[str, Any]) -> Optional[float]: + """Compute the investigation's total cost via the canonical + pricing helper. Returns ``None`` on unknown-model / pricing- + miss paths — consumers render "—" in those cells.""" + model = meta.get("model_used") + if not model: + return None + try: + from agent.usage_pricing import CanonicalUsage, estimate_usage_cost + except Exception as exc: # pragma: no cover — defensive + logger.warning( + "[kora.alert_wake_consumer] usage_pricing import failed: %r", + exc, + ) + return None + + def _int_or_zero(key: str) -> int: + value = meta.get(key) + if isinstance(value, int): + return value + return 0 + + usage = CanonicalUsage( + input_tokens=_int_or_zero("input_tokens"), + output_tokens=_int_or_zero("output_tokens"), + cache_read_tokens=_int_or_zero("cache_read_input_tokens"), + cache_write_tokens=_int_or_zero("cache_creation_input_tokens"), + ) + try: + result = estimate_usage_cost(str(model), usage) + except Exception as exc: + logger.warning( + "[kora.alert_wake_consumer] estimate_usage_cost raised %r " + "model=%s — total_cost_usd recorded as None", + exc, + model, + ) + return None + if result.status == "unknown": + return None + if result.amount_usd is None: + return None + return float(result.amount_usd) diff --git a/kora_cli/audit/jsonl_sink.py b/kora_cli/audit/jsonl_sink.py index 497328bbe091..6d1cd136817f 100644 --- a/kora_cli/audit/jsonl_sink.py +++ b/kora_cli/audit/jsonl_sink.py @@ -238,6 +238,48 @@ # recurring_recommendation_text / blast_radius_summary / # created_at / status. Source is ``reasoning``. "promotion.probe_envelope_action_proposed", + # KR-ALERT-INVESTIGATION-WAKE-CONSUMER — per-investigation summary + # for the alert wake consumer (parallels probe.investigation_completed). + # Emitted by ``kora_cli/alerts/wake_consumer.py`` after reasoning + + # DM dispatch complete. Payload: alert_id / category / severity / + # model_used / input_tokens / output_tokens / + # cache_creation_input_tokens / cache_read_input_tokens / + # total_cost_usd / investigation_duration_ms / + # investigation_summary_text / dm_status / autoaction_attempted + # (v1 always false; reserved for future alert-envelope autoaction + # parallel to probe autofix). Source is ``reasoning`` since the + # emit happens inside the wake consumer's reasoning flow. CC#2 + # follow-on KR-FE-ALERT-INVESTIGATIONS-VIEWER reads this seam + # alongside the existing notification.dispatched + slack_dm_log.jsonl + # to render the 3-stream join (4 streams if/when autoaction lands). + "alert.investigation_completed", + # KR-PROMOTE-EMAIL-INTENT — 6th promotion loop. Reads + # ``intent.email_to_sea_ticket`` rows with ``action="logged_only"`` + # (Joshua-authored emails that no existing intent pattern matched) + # + clusters by subject text similarity. Proposes new regex + # patterns to extend the email-intent registry. Payload: + # proposal_id / cluster_size / sample_subjects (up to 3) / + # proposed_pattern / proposed_action_kind ("save_note" | + # "log_only" | "save_with_reply" — operator picks at approve) / + # confidence / created_at / status / action ("proposed" or + # "auto_applied"). Source is ``email``. Auto-apply OFF by + # default per promotion-loop discipline. + "promotion.email_intent_pattern_proposed", + # KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW — captures operator-level + # "Haiku should have escalated here but didn't" signals. Emitted + # by the engine pre-call when ``select_model_pre_call`` returned + # ``reason in {opus_prefix, force_opus_env}`` on iteration 1 — + # i.e. operator manually forced Opus on a call that Haiku-router + # would have left on Haiku absent the override. Payload: + # original_message_text (truncated to 240 chars) / + # pre_call_decision_reason (verbatim from the router; v1 + # observed reasons are ``opus_prefix`` / ``force_opus_env``) / + # override_source (``operator_prefix`` / ``force_env``) / route + # (the message source). The router-tuning observer (#193) + # consumes this seam to activate its dormant loosen-path + # proposer; the loosen proposal flags routes where operator + # overrode N+ times in the window. + "opus_override.applied", ] SourceName = Literal[ diff --git a/kora_cli/listeners/__init__.py b/kora_cli/listeners/__init__.py index 66f872b4d1b4..9764eec987eb 100644 --- a/kora_cli/listeners/__init__.py +++ b/kora_cli/listeners/__init__.py @@ -101,3 +101,15 @@ from kora_cli.listeners import promote_router_tuning_listener # noqa: F401 from kora_cli.listeners import promote_tool_trimming_listener # noqa: F401 from kora_cli.listeners import promote_probe_fix_envelopes_listener # noqa: F401 +# KR-PROMOTE-EMAIL-INTENT — 6th promotion loop. Observes +# ``intent.email_to_sea_ticket`` action="logged_only" rows + proposes +# regex patterns to extend the email-intent registry. Same propose- +# only discipline as probe-fix-envelopes (manual scaffolding into +# kora_cli/intent/email_to_sea_ticket.py at approve-time). +from kora_cli.listeners import promote_email_intent_listener # noqa: F401 +# KR-ALERT-INVESTIGATION-WAKE-CONSUMER — alerts side of the unified- +# operator-interface (parallels probe wake consumer #166). Tails +# notification.dispatched audit + invokes reasoning + DMs operator. +# Imported AFTER reasoning + slack client listeners so the lazy +# factories resolve to live singletons by the first cycle tick. +from kora_cli.listeners import alert_wake_listener # noqa: F401 diff --git a/kora_cli/listeners/alert_wake_listener.py b/kora_cli/listeners/alert_wake_listener.py new file mode 100644 index 000000000000..93d654a110e0 --- /dev/null +++ b/kora_cli/listeners/alert_wake_listener.py @@ -0,0 +1,276 @@ +"""Alert wake-event listener — KR-ALERT-INVESTIGATION-WAKE-CONSUMER. + +Periodic task that tails ``${KORA_HOME}/kora_audit_log.jsonl`` for +fresh ``notification.dispatched`` rows (emitted by AlertNotifier +#149) and feeds each one to :class:`AlertWakeConsumer`. + +Tail strategy mirrors :mod:`kora_cli.listeners.probe_wake_listener` +(#166): cron-driven, NOT inotify; first-tick-after-startup stamps +``_last_seen_at`` to NOW so historical dispatches don't replay. + +# Tail filter + +The consumer itself filters aggregate rows (burst_summary / +digest_email) + non-ok status rows — this listener pulls every +``notification.dispatched`` entry and the consumer's +``consume_alert_event`` short-circuits the ones it shouldn't act +on. Keeps the tail logic simple + lets the consumer-side filter +remain a single source of truth for "what counts as an +investigatable alert." + +# Cadence + +``KORA_ALERT_WAKE_POLL_SEC`` (default 30s; same as probe wake +listener for operator-grep parity in the boot logs). + +# Fail-soft + +Wraps the consumer call in try/except; single-event failures +inside ``consume_alert_event`` are caught by the consumer itself +per its own contract, and the listener wrapper guards against +anything that escapes (e.g. an unexpected import-time failure +inside a lazily-loaded helper). +""" + +from __future__ import annotations + +import logging +import os +from datetime import datetime, timezone +from typing import Any, Optional + +from kora_cli.alerts.wake_consumer import AlertWakeConsumer +from kora_cli.daemon import DEFAULT_SHUTDOWN_TIMEOUT, register_daemon_listener +from kora_cli.listeners.heartbeat import register_periodic_task + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Cadence config +# --------------------------------------------------------------------------- + + +DEFAULT_POLL_SEC: float = 30.0 +POLL_SEC_ENV: str = "KORA_ALERT_WAKE_POLL_SEC" + + +def _read_poll_sec() -> float: + raw = os.environ.get(POLL_SEC_ENV, "").strip() + if not raw: + return DEFAULT_POLL_SEC + try: + value = float(raw) + except ValueError: + logger.warning( + "[kora.alert_wake_listener] %s=%r is not numeric; using " + "default %ss", + POLL_SEC_ENV, + raw, + DEFAULT_POLL_SEC, + ) + return DEFAULT_POLL_SEC + if value <= 0: + return DEFAULT_POLL_SEC + return value + + +# --------------------------------------------------------------------------- +# Singleton + tail-position state (mirrors probe_wake_listener) +# --------------------------------------------------------------------------- + + +_consumer_singleton: Optional[AlertWakeConsumer] = None +_last_seen_at: Optional[datetime] = None + + +def current_alert_wake_consumer() -> Optional[AlertWakeConsumer]: + """Read-only accessor for tests + introspection.""" + return _consumer_singleton + + +def _set_consumer(consumer: AlertWakeConsumer) -> None: + global _consumer_singleton + _consumer_singleton = consumer + + +def _clear_consumer() -> None: + global _consumer_singleton, _last_seen_at + _consumer_singleton = None + _last_seen_at = None + + +# --------------------------------------------------------------------------- +# Lazy factories for reasoning engine + Slack client +# --------------------------------------------------------------------------- + + +def _reasoning_engine_factory() -> Optional[Any]: + try: + from kora_cli.listeners.reasoning_engine_listener import ( + current_reasoning_engine, + ) + except Exception: + return None + return current_reasoning_engine() + + +def _slack_client_factory() -> Optional[Any]: + try: + from kora_cli.listeners.slack_client_listener import ( + current_slack_client, + ) + except Exception: + return None + return current_slack_client() + + +# --------------------------------------------------------------------------- +# Periodic-task entry +# --------------------------------------------------------------------------- + + +async def run_tail_cycle() -> None: + """One tail tick: pick up fresh notification.dispatched rows + + hand them to the consumer. + + Fail-soft: any exception inside (audit read, consumer raise, + etc.) is caught + logged so the heartbeat scheduler keeps + ticking. + """ + global _last_seen_at + consumer = current_alert_wake_consumer() + if consumer is None: + logger.debug( + "[kora.alert_wake_listener] tick skipped: no active consumer" + ) + return + + now = datetime.now(timezone.utc) + if _last_seen_at is None: + _last_seen_at = now + logger.info( + "[kora.alert_wake_listener] tail-position stamped at boot " + "(no replay of prior notification dispatches)" + ) + return + + try: + from kora_cli.audit.jsonl_reader import read_audit_entries + except Exception as exc: + logger.warning( + "[kora.alert_wake_listener] audit reader import failed: %r", + exc, + ) + return + + try: + rows = read_audit_entries( + seam="notification.dispatched", + since=_last_seen_at, + ) + except Exception as exc: + logger.warning( + "[kora.alert_wake_listener] read_audit_entries raised %r", + exc, + ) + return + + if not rows: + return + + # Reader returns newest-first; reverse so consume + last_seen_at + # advance monotonically (probe wake listener pattern). + rows_chronological = list(reversed(rows)) + + new_max_ts = _last_seen_at + for row in rows_chronological: + try: + await consumer.consume_alert_event( + getattr(row, "details", {}) or {} + ) + except Exception as exc: + logger.warning( + "[kora.alert_wake_listener] consume_alert_event raised " + "%r — continuing past this event", + exc, + ) + emitted_at = getattr(row, "emitted_at", None) + if isinstance(emitted_at, datetime): + if emitted_at.tzinfo is None: + emitted_at = emitted_at.replace(tzinfo=timezone.utc) + if emitted_at > new_max_ts: + new_max_ts = emitted_at + + _last_seen_at = new_max_ts + + +# --------------------------------------------------------------------------- +# Listener lifecycle +# --------------------------------------------------------------------------- + + +class AlertWakeListener: + """Holds the live :class:`AlertWakeConsumer` for the daemon + lifetime. Periodic task picks up fresh notification rows on + each tick + hands them to the consumer. + """ + + async def startup(self) -> None: + try: + consumer = AlertWakeConsumer( + reasoning_engine_factory=_reasoning_engine_factory, + slack_client_factory=_slack_client_factory, + ) + except Exception as exc: + logger.warning( + "[kora.alert_wake_listener] startup raised %r — " + "consumer disabled this run", + exc, + ) + _clear_consumer() + return + + _set_consumer(consumer) + logger.info( + "[kora.alert_wake_listener] AlertWakeConsumer constructed; " + "poll cadence=%ss", + _read_poll_sec(), + ) + + async def shutdown(self) -> None: + consumer = _consumer_singleton + if consumer is not None: + consumer.reset_debounce_state() + _clear_consumer() + logger.info( + "[kora.alert_wake_listener] AlertWakeConsumer cleared" + ) + + +# --------------------------------------------------------------------------- +# Factory + registration (import-time side effect) +# --------------------------------------------------------------------------- + + +def _factory(): + listener = AlertWakeListener() + return (listener.startup, listener.shutdown, DEFAULT_SHUTDOWN_TIMEOUT) + + +register_daemon_listener("alert_wake", _factory) + + +register_periodic_task( + "alert_wake.tail", + interval_seconds=_read_poll_sec(), + callable=run_tail_cycle, +) + + +def _reset_tail_position_for_tests() -> None: + """Test-only: clear the tail-position state. Production code + MUST NOT call this — it would cause replay of all historical + notifications on the next tick.""" + global _last_seen_at + _last_seen_at = None diff --git a/kora_cli/listeners/promote_email_intent_listener.py b/kora_cli/listeners/promote_email_intent_listener.py new file mode 100644 index 000000000000..845718c92ce8 --- /dev/null +++ b/kora_cli/listeners/promote_email_intent_listener.py @@ -0,0 +1,45 @@ +"""Heartbeat-scheduled email-intent promotion cycle — KR-PROMOTE-EMAIL-INTENT. + +Same listener shape as the other promotion-loop listeners. Cadence +operator-tunable via ``KORA_PROMOTE_EMAIL_INTENT_INTERVAL_SEC`` +(default 86400s = 24h). Master kill-switch +``KORA_PROMOTE_EMAIL_INTENT_ENABLED=false`` checked inside the +cycle. +""" + +from __future__ import annotations + +import logging + +from kora_cli.listeners.heartbeat import register_periodic_task +from kora_cli.promote.email_intent.plugin import ( + get_interval_seconds, + run_email_intent_cycle, +) + +logger = logging.getLogger(__name__) + + +async def _periodic_task() -> None: + try: + summary = await run_email_intent_cycle() + logger.debug( + "[kora.promote.email_intent.listener] tick complete: " + "proposals_persisted=%d expired_count=%d duration_ms=%d", + summary.get("proposals_persisted", 0), + summary.get("expired_count", 0), + summary.get("duration_ms", 0), + ) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent.listener] tick raised %r — " + "next scheduled run will retry", + exc, + ) + + +register_periodic_task( + "promote_email_intent_cycle", + interval_seconds=float(get_interval_seconds()), + callable=_periodic_task, +) diff --git a/kora_cli/promote/email_intent/__init__.py b/kora_cli/promote/email_intent/__init__.py new file mode 100644 index 000000000000..2ca9f8d3a8ce --- /dev/null +++ b/kora_cli/promote/email_intent/__init__.py @@ -0,0 +1,47 @@ +"""Email-intent promotion loop — KR-PROMOTE-EMAIL-INTENT. + +Sixth promotion loop (after phrasebook / snapshot-expand / +router-tuning / tool-trimming / probe-fix-envelopes). Observes +``intent.email_to_sea_ticket`` audit rows with +``action="logged_only"`` — Joshua-authored emails that no +existing intent regex matched — and proposes new regex patterns +to extend the email-intent registry. + +# Loop shape + + 1. :mod:`.observer` — read ``intent.email_to_sea_ticket`` rows + where ``action="logged_only"``; extract the subject text + + pattern_matched hint + reason. + 2. :mod:`.proposer` — embed subjects via the shared lexical + embedder + cluster by similarity; for clusters meeting the + min-size threshold, derive a candidate regex from common + tokens + propose a default ``proposed_action_kind`` of + ``"save_note"`` (operator picks at approve-time). + 3. :mod:`.plugin` — orchestrator + listener wiring + audit emit + via ``promotion.email_intent_pattern_proposed``. + +# Why subject-only clustering + +The ``intent.email_to_sea_ticket`` audit payload intentionally +carries the subject text but NOT the body — PII discipline per +the existing module's security posture. v1 clusters on subjects +which is sufficient signal for "operator routinely sends emails +with this subject shape" — the proposer's pattern is a regex +the operator REVIEWS, so coarse signal is fine. + +# Cost discipline + +$0 LLM. The proposer uses the shared lexical embedder + greedy +clustering + token-frequency-based pattern derivation. Per +cycle: $0. Combined with the other 5 loops: still ≤$0.005/day +(phrasebook's Haiku synthesis remains the only LLM cost). + +# Auto-apply + +DEFAULT FALSE. Adding regex patterns to the intent registry +changes how Kora interprets operator-from emails (could +inadvertently auto-Sea_Ticket emails the operator didn't mean to +trigger). v1 is propose-only — operator manually scaffolds +approved patterns into ``kora_cli/intent/email_to_sea_ticket.py`` +(follows the probe-fix-envelopes precedent from #193). +""" diff --git a/kora_cli/promote/email_intent/observer.py b/kora_cli/promote/email_intent/observer.py new file mode 100644 index 000000000000..9d231e048c5c --- /dev/null +++ b/kora_cli/promote/email_intent/observer.py @@ -0,0 +1,114 @@ +"""Email-intent observation collector — KR-PROMOTE-EMAIL-INTENT. + +Reads ``intent.email_to_sea_ticket`` audit rows where +``action="logged_only"`` and projects them into the shape the +proposer clusters on. + +# Filtering + + * ``action != "logged_only"`` rows are skipped — only the + unmatched/below-floor emails carry the "Kora should learn + this" signal. + * Rows where ``subject`` is missing / empty are skipped — + nothing to cluster on. + * ``since`` defaults to 14 days back; long enough to surface + recurring patterns without dragging in stale subjects. + +# PII discipline + +The body text is NOT logged in the source audit (see +``kora_cli/intent/email_to_sea_ticket.py`` security posture). +This observer projects only the subject + the audit row's +``pattern_matched`` / ``reason`` / ``confidence`` for proposer +context. Future bucket may extend audit to carry a hashed body +fingerprint if subject-only clustering proves too noisy. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import List, Optional + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class EmailIntentObservation: + """One logged-only email projected for clustering.""" + + subject: str + pattern_matched: Optional[str] # None when no pattern matched + confidence: str # "unrecognized" / "low" / "medium" / "high" + reason: str # "no_pattern_matched" / "below_floor_" + caller_session_id: str + timestamp: datetime + + +async def collect_recent_logged_only( + *, since: Optional[datetime] = None +) -> List[EmailIntentObservation]: + """Read recent ``intent.email_to_sea_ticket`` audit rows + filter + to ``action="logged_only"``. + + Args: + since: Lower bound (aware datetime). Defaults to 14 days + before now. + + Returns observations sorted by ``timestamp`` ascending. Empty + on reader failure (fail-soft per the other promotion-loop + observers). + """ + if since is None: + since = datetime.now(timezone.utc) - timedelta(days=14) + if since.tzinfo is None: + since = since.replace(tzinfo=timezone.utc) + + try: + from kora_cli.audit.jsonl_reader import read_audit_entries + except Exception as exc: + logger.warning( + "[kora.promote.email_intent.observer] audit reader import " + "failed: %r — no observations", + exc, + ) + return [] + + try: + entries = read_audit_entries( + seam="intent.email_to_sea_ticket", since=since + ) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent.observer] read_audit_entries " + "raised %r — no observations", + exc, + ) + return [] + + out: List[EmailIntentObservation] = [] + for entry in entries: + details = entry.details or {} + if details.get("action") != "logged_only": + continue + subject = details.get("subject") + if not isinstance(subject, str) or not subject.strip(): + continue + pattern_raw = details.get("pattern_matched") + pattern_matched = ( + pattern_raw if isinstance(pattern_raw, str) and pattern_raw else None + ) + out.append( + EmailIntentObservation( + subject=subject.strip(), + pattern_matched=pattern_matched, + confidence=str(details.get("confidence") or "unrecognized"), + reason=str(details.get("reason") or ""), + caller_session_id=str(entry.caller_session_id or ""), + timestamp=entry.emitted_at, + ) + ) + + out.sort(key=lambda o: o.timestamp) + return out diff --git a/kora_cli/promote/email_intent/plugin.py b/kora_cli/promote/email_intent/plugin.py new file mode 100644 index 000000000000..e50aa33db5e4 --- /dev/null +++ b/kora_cli/promote/email_intent/plugin.py @@ -0,0 +1,206 @@ +"""Email-intent cycle orchestrator — KR-PROMOTE-EMAIL-INTENT. + +Called by the periodic-task heartbeat (registered by +:mod:`kora_cli.listeners.promote_email_intent_listener`). + +# Env + + * ``KORA_PROMOTE_EMAIL_INTENT_ENABLED`` (default ``true``) + * ``KORA_PROMOTE_EMAIL_INTENT_INTERVAL_SEC`` (default 86400 = 24h) + * ``KORA_PROMOTE_EMAIL_INTENT_EXPIRY_DAYS`` (default 14) + * ``KORA_PROMOTE_EMAIL_INTENT_WINDOW_DAYS`` (default 14) + * Proposer-side: ``KORA_PROMOTE_EMAIL_INTENT_MIN_CLUSTER``, + ``KORA_PROMOTE_EMAIL_INTENT_COHESION`` + +# Auto-apply + +Default OFF per the email-intent risk profile (a bad regex could +silently auto-Sea_Ticket emails the operator didn't intend). +Operator scaffolds approved patterns manually into +``kora_cli/intent/email_to_sea_ticket.py``. +""" + +from __future__ import annotations + +import logging +import os +import time +from datetime import datetime, timedelta, timezone +from typing import Any, Dict, Optional + +from kora_cli.promote._shared.proposal_store import ( + expire_older_than, + save_pending, +) + +from .observer import collect_recent_logged_only +from .proposer import ( + EmailIntentProposal, + generate_proposals, + proposal_to_dict, +) + +logger = logging.getLogger(__name__) + + +LOOP_NAME = "email_intent" + +ENABLED_ENV = "KORA_PROMOTE_EMAIL_INTENT_ENABLED" +INTERVAL_SEC_ENV = "KORA_PROMOTE_EMAIL_INTENT_INTERVAL_SEC" +EXPIRY_DAYS_ENV = "KORA_PROMOTE_EMAIL_INTENT_EXPIRY_DAYS" +OBSERVATION_WINDOW_DAYS_ENV = "KORA_PROMOTE_EMAIL_INTENT_WINDOW_DAYS" + +DEFAULT_INTERVAL_SEC = 86400 # once daily +DEFAULT_EXPIRY_DAYS = 14 +DEFAULT_OBSERVATION_WINDOW_DAYS = 14 + + +def _is_enabled() -> bool: + raw = os.environ.get(ENABLED_ENV, "true").strip().lower() + return raw in {"true", "1", "yes", "on", ""} + + +def _int_env(name: str, default: int, *, minimum: int = 1) -> int: + raw = os.environ.get(name, "").strip() + if not raw: + return default + try: + value = int(raw) + except ValueError: + return default + if value < minimum: + return default + return value + + +def get_interval_seconds() -> int: + return _int_env(INTERVAL_SEC_ENV, DEFAULT_INTERVAL_SEC, minimum=60) + + +def _emit_audit(proposal: EmailIntentProposal) -> None: + try: + from kora_cli.audit.jsonl_sink import emit_audit + except Exception as exc: + logger.warning( + "[kora.promote.email_intent] audit import failed: %r — " + "promotion.email_intent_pattern_proposed skipped", + exc, + ) + return + payload = proposal_to_dict(proposal) + payload["action"] = "proposed" # v1 — auto-apply OFF + try: + emit_audit( + "promotion.email_intent_pattern_proposed", + payload, + caller_session_id=( + f"promotion:email_intent:{proposal.proposal_id}" + ), + source="email", + ) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent] emit_audit raised %r — " + "proposal persisted; audit row missing", + exc, + ) + + +async def run_email_intent_cycle( + *, now: Optional[datetime] = None +) -> Dict[str, Any]: + """One cycle of the email-intent promotion loop.""" + started_dt = now or datetime.now(timezone.utc) + started_monotonic = time.monotonic() + + summary: Dict[str, Any] = { + "enabled": True, + "observations_read": 0, + "proposals_generated": 0, + "proposals_persisted": 0, + "expired_count": 0, + "auto_apply_mode": False, # v1 hardcoded; future env can flip + "started_at": started_dt.strftime("%Y-%m-%dT%H:%M:%SZ"), + "duration_ms": 0, + } + + if not _is_enabled(): + summary["enabled"] = False + logger.info( + "[kora.promote.email_intent] disabled (%s=false) — skipping", + ENABLED_ENV, + ) + summary["duration_ms"] = int( + (time.monotonic() - started_monotonic) * 1000 + ) + return summary + + window_days = _int_env( + OBSERVATION_WINDOW_DAYS_ENV, + DEFAULT_OBSERVATION_WINDOW_DAYS, + minimum=1, + ) + + try: + observations = await collect_recent_logged_only( + since=started_dt - timedelta(days=window_days), + ) + summary["observations_read"] = len(observations) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent] observer failed: %r — " + "no proposals generated", + exc, + ) + summary["duration_ms"] = int( + (time.monotonic() - started_monotonic) * 1000 + ) + return summary + + try: + proposals = await generate_proposals(observations, now=started_dt) + summary["proposals_generated"] = len(proposals) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent] proposer failed: %r", exc + ) + summary["duration_ms"] = int( + (time.monotonic() - started_monotonic) * 1000 + ) + return summary + + for proposal in proposals: + try: + save_pending( + loop_name=LOOP_NAME, + proposal_id=proposal.proposal_id, + payload=proposal_to_dict(proposal), + ) + summary["proposals_persisted"] += 1 + except Exception as exc: + logger.warning( + "[kora.promote.email_intent] persist failed for " + "%s: %r — proposal lost (audit row still emitted)", + proposal.proposal_id, + exc, + ) + _emit_audit(proposal) + + try: + summary["expired_count"] = expire_older_than( + loop_name=LOOP_NAME, + days=_int_env(EXPIRY_DAYS_ENV, DEFAULT_EXPIRY_DAYS, minimum=1), + ) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent] expire_older_than raised %r", + exc, + ) + + summary["duration_ms"] = int( + (time.monotonic() - started_monotonic) * 1000 + ) + logger.info( + "[kora.promote.email_intent] cycle complete: %s", summary + ) + return summary diff --git a/kora_cli/promote/email_intent/proposer.py b/kora_cli/promote/email_intent/proposer.py new file mode 100644 index 000000000000..b73238bc1437 --- /dev/null +++ b/kora_cli/promote/email_intent/proposer.py @@ -0,0 +1,278 @@ +"""Email-intent proposal generator — KR-PROMOTE-EMAIL-INTENT. + +Input: :class:`EmailIntentObservation` from the observer. +Output: :class:`EmailIntentProposal` records — one per cluster +≥ min_cluster_size. + +# Pipeline + + 1. Embed each observation's subject via + :func:`kora_cli.clustering.text_similarity.embed_texts`. + 2. Cluster via greedy similarity at + :data:`DEFAULT_COHESION_THRESHOLD`. + 3. For each cluster ≥ min_cluster_size: + a. Derive a candidate regex from the cluster's common tokens + (escape all regex metacharacters; emit case-insensitive + alternation of top-3 tokens). + b. Default ``proposed_action_kind = "save_note"`` (the most + common existing action; operator can flip to + ``"log_only"`` or ``"save_with_reply"`` at approve). + c. Confidence derived from cluster size; capped at 1.0. + +# Pattern derivation safety + +Same escape + alternation pattern as the phrasebook proposer +(#186). All regex metacharacters in the cluster's tokens are +escaped before joining; pathological subjects (with literal +``(``/``*``/etc.) can't bomb the regex engine at compile-time. +Operator can refine on approval. +""" + +from __future__ import annotations + +import logging +import os +import re +import uuid +from collections import Counter +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from typing import Any, Dict, List, Literal + +from kora_cli.clustering.text_similarity import ( + cluster_by_similarity, + embed_texts, +) + +from .observer import EmailIntentObservation + +logger = logging.getLogger(__name__) + + +MIN_CLUSTER_SIZE_ENV = "KORA_PROMOTE_EMAIL_INTENT_MIN_CLUSTER" +COHESION_THRESHOLD_ENV = "KORA_PROMOTE_EMAIL_INTENT_COHESION" + +DEFAULT_MIN_CLUSTER_SIZE = 3 +DEFAULT_COHESION_THRESHOLD = 0.65 +# Lower than the phrasebook threshold (0.85) because subjects are +# inherently shorter / sparser than full DM replies; tighter +# clustering would over-fragment. + +SAMPLE_SUBJECTS_CAP = 3 + +ProposalStatus = Literal["pending", "approved", "rejected", "expired"] +ActionKind = Literal["save_note", "log_only", "save_with_reply"] + + +@dataclass(frozen=True, slots=True) +class EmailIntentProposal: + """Wire-stable proposal shape.""" + + proposal_id: str + cluster_size: int + sample_subjects: List[str] + proposed_pattern: str + proposed_action_kind: ActionKind + confidence: float + created_at: datetime + status: ProposalStatus = "pending" + review_notes: str = "" + sample_caller_session_ids: List[str] = field(default_factory=list) + + +def _format_iso(dt: datetime) -> str: + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + + +def proposal_to_dict(p: EmailIntentProposal) -> Dict[str, Any]: + out = asdict(p) + out["created_at"] = _format_iso(p.created_at) + out["sample_subjects"] = list(p.sample_subjects) + out["sample_caller_session_ids"] = list(p.sample_caller_session_ids) + return out + + +def _int_env(name: str, default: int, *, minimum: int = 1) -> int: + raw = os.environ.get(name, "").strip() + if not raw: + return default + try: + value = int(raw) + except ValueError: + return default + if value < minimum: + return default + return value + + +def _float_env(name: str, default: float, *, minimum: float = 0.0) -> float: + raw = os.environ.get(name, "").strip() + if not raw: + return default + try: + value = float(raw) + except ValueError: + return default + if value < minimum: + return default + return value + + +# Stopwords specific to email subjects — common prefix noise like +# "Re:" / "Fwd:" gets stripped so the proposer doesn't cluster +# on transport-level header chatter. +_STOPWORDS = frozenset( + { + "re", + "fwd", + "fw", + "a", + "an", + "and", + "the", + "to", + "of", + "for", + "in", + "on", + "at", + "is", + "are", + "was", + "be", + } +) + + +_TOKEN_RE = re.compile(r"[A-Za-z0-9]+") + + +def _meaningful_tokens(text: str) -> List[str]: + return [ + t + for t in _TOKEN_RE.findall(text.lower()) + if t not in _STOPWORDS and len(t) >= 2 + ] + + +def _derive_pattern(subjects: List[str]) -> str: + """Generate a conservative regex from the cluster's common + subject tokens. Pattern: case-insensitive alternation of the + top-3 cross-subject tokens. All metacharacters escaped. + + Fallback (no shared tokens): a permissive ``(?i).*`` placeholder + that the operator MUST refine before approving — defaults to + "match everything" so the proposal is obviously a placeholder + rather than a silently-bad regex. + """ + df: Counter = Counter() + for subj in subjects: + for tok in set(_meaningful_tokens(subj)): + df[tok] += 1 + if not df: + return "(?i).*" + top = [tok for tok, _ in df.most_common(3)] + escaped = [re.escape(t) for t in top] + return "(?i)(" + "|".join(escaped) + ")" + + +async def generate_proposals( + observations: List[EmailIntentObservation], + *, + now: datetime, +) -> List[EmailIntentProposal]: + """Cluster + propose. Returns proposals sorted by confidence + descending. Fail-soft on embedder error (returns empty list).""" + if not observations: + return [] + min_cluster_size = _int_env( + MIN_CLUSTER_SIZE_ENV, DEFAULT_MIN_CLUSTER_SIZE, minimum=2 + ) + cohesion = _float_env( + COHESION_THRESHOLD_ENV, + DEFAULT_COHESION_THRESHOLD, + minimum=0.0, + ) + if cohesion > 1.0: + cohesion = 1.0 + + subjects = [o.subject for o in observations] + try: + embeddings = await embed_texts(subjects) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent.proposer] embed_texts raised " + "%r — no proposals", + exc, + ) + return [] + + try: + clusters = cluster_by_similarity(embeddings, threshold=cohesion) + except Exception as exc: + logger.warning( + "[kora.promote.email_intent.proposer] cluster_by_similarity " + "raised %r — no proposals", + exc, + ) + return [] + + # Map embeddings back to observations by index — embed_texts + # preserves order, cluster_by_similarity preserves order, so + # we can find each embedding's source observation via + # ``subjects[i]`` equality (stable since same input list). + text_to_obs: Dict[str, List[EmailIntentObservation]] = {} + for o in observations: + text_to_obs.setdefault(o.subject, []).append(o) + + out: List[EmailIntentProposal] = [] + for cluster in clusters: + if len(cluster) < min_cluster_size: + continue + cluster_subjects = [emb.text for emb in cluster] + # Dedup + cap sample subjects so the audit payload stays + # bounded; operator wants representative examples, not the + # full cluster. + seen_subjects: List[str] = [] + for s in cluster_subjects: + if s not in seen_subjects: + seen_subjects.append(s) + if len(seen_subjects) >= SAMPLE_SUBJECTS_CAP: + break + + # Sample caller_session_ids drawn from the matching + # observations. + sample_ids: List[str] = [] + seen_ids = set() + for emb in cluster: + for obs in text_to_obs.get(emb.text, []): + if not obs.caller_session_id or obs.caller_session_id in seen_ids: + continue + seen_ids.add(obs.caller_session_id) + sample_ids.append(obs.caller_session_id) + if len(sample_ids) >= SAMPLE_SUBJECTS_CAP: + break + if len(sample_ids) >= SAMPLE_SUBJECTS_CAP: + break + + confidence = min(1.0, len(cluster) / (2 * min_cluster_size)) + out.append( + EmailIntentProposal( + proposal_id=str(uuid.uuid4()), + cluster_size=len(cluster), + sample_subjects=seen_subjects, + proposed_pattern=_derive_pattern(cluster_subjects), + # v1 default — operator can flip to log_only / + # save_with_reply at approve-time. save_note matches + # the most common existing pattern shape (subject_note_prefix + # / subject_idea_prefix / subject_todo_prefix from + # the current registry). + proposed_action_kind="save_note", + confidence=round(confidence, 4), + created_at=now, + status="pending", + sample_caller_session_ids=sample_ids, + ) + ) + out.sort(key=lambda p: (-p.confidence, -p.cluster_size)) + return out diff --git a/kora_cli/promote/router_tuning/observer.py b/kora_cli/promote/router_tuning/observer.py index 971a607296f6..0ab8a19f8021 100644 --- a/kora_cli/promote/router_tuning/observer.py +++ b/kora_cli/promote/router_tuning/observer.py @@ -2,31 +2,36 @@ Reads the live :func:`kora_cli.telemetry.get_telemetry` singleton's :meth:`snapshot` and projects per-route rollups for the proposer. - -# v1 scope (STOP-ASK §4) - -The bucket STOP-ASK §4 anticipated: the observer wants per-call -"was the Opus reply materially better than Haiku's would be?" data. -That data isn't exposed today — collecting it would need either: - - 1. A second Haiku call per Opus call to do post-hoc quality - scoring (doubles spend on every escalation), OR - 2. A new audit row whenever the operator manually issues - ``/opus`` to fix a Haiku miss (not yet emitted). - -v1 ships with the data we DO have: ``calls_count`` + -``escalation_count`` per route. That's enough to surface "this -route escalates 60% of the time — please review the trigger" for -operator-attention; the actual tuning decision stays operator- -gated regardless. A future bucket can wire option 2 (cheap; one -audit row per operator override) to refine the rationale. +Also reads the ``opus_override.applied`` audit seam (added in +KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW) to surface routes whose +operator-overrides exceed the loosen-path threshold. + +# Two data sources, two proposal kinds + + 1. ``cost_telemetry.snapshot()`` per-route counters → + ``tighten_review`` proposals (route escalates often; review + whether the trigger is paying off). + 2. ``opus_override.applied`` audit rows (this PR) → + ``loosen_review`` proposals (operator manually overrode + Haiku to Opus N+ times on a route; trigger pattern should + probably auto-escalate that case). + +# Why this design earns the loosen path + +#193 left the loosen path dormant because no override audit row +existed. With the seam added, the observer reads it directly + +the proposer wires the second proposal kind. Per the bucket +spec §2 deliverable C: "the loosen-path code that #193 left +dormant becomes active." """ from __future__ import annotations import logging -from dataclasses import dataclass -from typing import Dict, List +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional logger = logging.getLogger(__name__) @@ -42,6 +47,21 @@ class RouteEscalationRollup: cost_estimate_usd_total: float +@dataclass(frozen=True, slots=True) +class RouteOverrideRollup: + """One route's projection from ``opus_override.applied`` audit + rows over the observation window. Feeds the proposer's + loosen_review path (KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW).""" + + route: str + override_count: int + sample_message_texts: List[str] = field(default_factory=list) + # Per-source breakdown ("operator_prefix" / "force_env") so the + # proposer can distinguish per-call /opus moves from a global + # KORA_FORCE_OPUS env flip (which is much weaker signal). + by_source: Dict[str, int] = field(default_factory=dict) + + def _safe_rate(escalations: int, total: int) -> float: if total <= 0: return 0.0 @@ -104,3 +124,92 @@ def collect_route_rollups() -> List[RouteEscalationRollup]: ) out.sort(key=lambda r: r.route) return out + + +# --------------------------------------------------------------------------- +# Loosen-path observer (KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW) +# --------------------------------------------------------------------------- + + +# Cap on sample message texts surfaced per route — keeps the +# proposer payload bounded even when a route accumulates many +# override events. +_SAMPLE_TEXT_CAP = 3 + + +def collect_route_overrides( + *, since: Optional[datetime] = None +) -> List[RouteOverrideRollup]: + """Read ``opus_override.applied`` audit rows + group by route. + + Args: + since: Lower bound (aware datetime). Defaults to 24h before + now — same window as the tighten-path's rolling_24h + cost-telemetry source so the two proposal kinds emit on + comparable observation windows. + + Returns rollups sorted alphabetically by route (stable test + ordering; the proposer reorders by score before emitting). + Fail-soft per the other observer methods: audit reader failure + returns ``[]``. + """ + if since is None: + since = datetime.now(timezone.utc) - timedelta(days=1) + if since.tzinfo is None: + since = since.replace(tzinfo=timezone.utc) + + try: + from kora_cli.audit.jsonl_reader import read_audit_entries + except Exception as exc: + logger.debug( + "[kora.promote.router_tuning.observer] audit reader import " + "failed: %r — no override rollups", + exc, + ) + return [] + + try: + entries = read_audit_entries( + seam="opus_override.applied", since=since + ) + except Exception as exc: + logger.warning( + "[kora.promote.router_tuning.observer] read_audit_entries " + "raised %r — no override rollups", + exc, + ) + return [] + + counts: Dict[str, int] = defaultdict(int) + samples: Dict[str, List[str]] = defaultdict(list) + by_source: Dict[str, Dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) + for entry in entries: + details = entry.details or {} + route = details.get("route") + if not isinstance(route, str) or not route: + route = "unknown" + counts[route] += 1 + source = details.get("override_source") + if isinstance(source, str) and source: + by_source[route][source] += 1 + text = details.get("original_message_text") + if ( + isinstance(text, str) + and text.strip() + and len(samples[route]) < _SAMPLE_TEXT_CAP + ): + samples[route].append(text.strip()) + + out: List[RouteOverrideRollup] = [] + for route in sorted(counts.keys()): + out.append( + RouteOverrideRollup( + route=route, + override_count=counts[route], + sample_message_texts=list(samples[route]), + by_source=dict(by_source[route]), + ) + ) + return out diff --git a/kora_cli/promote/router_tuning/plugin.py b/kora_cli/promote/router_tuning/plugin.py index 3b2059639d0b..c9ed047e39f4 100644 --- a/kora_cli/promote/router_tuning/plugin.py +++ b/kora_cli/promote/router_tuning/plugin.py @@ -31,16 +31,23 @@ import logging import os import time -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Any, Dict, Optional +_ONE_DAY = timedelta(days=1) + from kora_cli.promote._shared.proposal_store import ( expire_older_than, save_pending, ) -from .observer import collect_route_rollups -from .proposer import RouterTuningProposal, generate_proposals, proposal_to_dict +from .observer import collect_route_overrides, collect_route_rollups +from .proposer import ( + RouterTuningProposal, + generate_loosen_proposals, + generate_proposals, + proposal_to_dict, +) logger = logging.getLogger(__name__) @@ -117,6 +124,9 @@ async def run_router_tuning_cycle( summary: Dict[str, Any] = { "enabled": True, "rollups_observed": 0, + # KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW — separate counter so + # operator can see tighten vs loosen volume in one cycle log. + "overrides_observed": 0, "proposals_generated": 0, "proposals_persisted": 0, "expired_count": 0, @@ -151,15 +161,41 @@ async def run_router_tuning_cycle( try: proposals = generate_proposals(rollups, now=started_dt) - summary["proposals_generated"] = len(proposals) except Exception as exc: logger.warning( - "[kora.promote.router_tuning] proposer failed: %r", exc + "[kora.promote.router_tuning] tighten proposer failed: %r", + exc, ) - summary["duration_ms"] = int( - (time.monotonic() - started_monotonic) * 1000 + proposals = [] + + # KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW — loosen-path arm. Reads + # opus_override.applied audit since the cycle's started_at - 24h + # (mirrors the tighten-path's rolling_24h cost-telemetry window + # so the two arms see comparable observation windows). + try: + override_rollups = collect_route_overrides( + since=started_dt - _ONE_DAY, ) - return summary + summary["overrides_observed"] = len(override_rollups) + except Exception as exc: + logger.warning( + "[kora.promote.router_tuning] loosen observer failed: %r", + exc, + ) + override_rollups = [] + + try: + loosen_proposals = generate_loosen_proposals( + override_rollups, now=started_dt + ) + proposals = proposals + loosen_proposals + except Exception as exc: + logger.warning( + "[kora.promote.router_tuning] loosen proposer failed: %r", + exc, + ) + + summary["proposals_generated"] = len(proposals) for proposal in proposals: try: diff --git a/kora_cli/promote/router_tuning/proposer.py b/kora_cli/promote/router_tuning/proposer.py index a57147ff79cb..558d27ac0936 100644 --- a/kora_cli/promote/router_tuning/proposer.py +++ b/kora_cli/promote/router_tuning/proposer.py @@ -1,29 +1,34 @@ """Router-tuning proposal generator — KR-PROMOTE-ROUTER-TUNING. -Input: per-route :class:`RouteEscalationRollup` from the observer. +Inputs: + * Per-route :class:`RouteEscalationRollup` from cost_telemetry + (tighten-path). + * Per-route :class:`RouteOverrideRollup` from + ``opus_override.applied`` audit (loosen-path; activated by + KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW). + Output: zero or more :class:`RouterTuningProposal` records — one -per route whose escalation pattern crosses an operator-attention +per route whose pattern crosses the relevant operator-attention threshold. -# Thresholds (operator-tunable via env) - - * ``KORA_PROMOTE_ROUTER_TUNING_MIN_CALLS`` (default 20) — - minimum calls in window before a route gets considered. - Below this, sample size is too noisy. - * ``KORA_PROMOTE_ROUTER_TUNING_TIGHTEN_THRESHOLD`` (default 0.40) - — escalation_rate ≥ this on an eligible route → tighten_review - proposal. (Default 40% — well above the natural escalation - baseline of <15% from healthy decision-language patterns.) - -# Why no loosen_review in v1 - -The signal for ``loosen_review`` is "operator overrode Haiku to -Opus via /opus N times" — that observation doesn't have its own -audit row yet (it lives in the routing decision logs, not the -JSONL audit). Future bucket can emit a ``router.operator_override`` -seam; the proposer here would then surface routes with high -override-rate as loosen candidates. Documented in -``__init__.py`` v1 scope. +# Tighten path + +``KORA_PROMOTE_ROUTER_TUNING_TIGHTEN_THRESHOLD`` (default 0.40) — +escalation_rate ≥ this on an eligible route → ``tighten_review``. + +# Loosen path + +``KORA_PROMOTE_ROUTER_TUNING_LOOSEN_OVERRIDE_THRESHOLD`` (default +3) — operator forced Opus ≥ this many times in the window on a +single route → ``loosen_review``. The proposer's rationale points +the operator at the sample message texts so they can identify the +trigger pattern that should auto-escalate. + +# Sample size minimum + +``KORA_PROMOTE_ROUTER_TUNING_MIN_CALLS`` (default 20) — applies +to the tighten path only; the loosen path has its own threshold +since override events are intrinsically rarer than total calls. """ from __future__ import annotations @@ -31,20 +36,24 @@ import logging import os import uuid -from dataclasses import asdict, dataclass +from dataclasses import asdict, dataclass, field from datetime import datetime, timezone from typing import Any, Dict, List, Literal, Tuple -from .observer import RouteEscalationRollup +from .observer import RouteEscalationRollup, RouteOverrideRollup logger = logging.getLogger(__name__) MIN_CALLS_ENV = "KORA_PROMOTE_ROUTER_TUNING_MIN_CALLS" TIGHTEN_THRESHOLD_ENV = "KORA_PROMOTE_ROUTER_TUNING_TIGHTEN_THRESHOLD" +LOOSEN_OVERRIDE_THRESHOLD_ENV = ( + "KORA_PROMOTE_ROUTER_TUNING_LOOSEN_OVERRIDE_THRESHOLD" +) DEFAULT_MIN_CALLS = 20 DEFAULT_TIGHTEN_THRESHOLD = 0.40 +DEFAULT_LOOSEN_OVERRIDE_THRESHOLD = 3 ProposalStatus = Literal["pending", "approved", "rejected", "expired"] @@ -55,7 +64,21 @@ class RouterTuningProposal: """Wire-stable proposal shape. Mirrors the snapshot_expand / phrasebook proposal shape conventions (proposal_id / - cluster_size / confidence / created_at / status).""" + cluster_size / confidence / created_at / status). + + Both ``tighten_review`` and ``loosen_review`` proposals use this + single shape; field semantics vary by ``recommendation_kind``: + + * tighten_review: ``calls_count`` / ``escalation_count`` / + ``escalation_rate`` / ``cost_estimate_usd_total`` are the + tighten-path numbers; ``override_count`` / ``sample_message_texts`` + are 0 / empty. + * loosen_review: ``override_count`` + ``sample_message_texts`` + + ``by_override_source`` are the loosen-path numbers; + ``escalation_count`` / ``escalation_rate`` / + ``cost_estimate_usd_total`` may be 0 (the override path + doesn't depend on those). + """ proposal_id: str route: str @@ -69,6 +92,11 @@ class RouterTuningProposal: created_at: datetime status: ProposalStatus = "pending" review_notes: str = "" + # KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW additions. Default to + # 0 / empty so the existing tighten-path callers keep working. + override_count: int = 0 + sample_message_texts: List[str] = field(default_factory=list) + by_override_source: Dict[str, int] = field(default_factory=dict) def _format_iso(dt: datetime) -> str: @@ -78,6 +106,10 @@ def _format_iso(dt: datetime) -> str: def proposal_to_dict(p: RouterTuningProposal) -> Dict[str, Any]: out = asdict(p) out["created_at"] = _format_iso(p.created_at) + # asdict mutates list/dict default_factory fields into new + # containers — defensive copy keeps caller mutation safe. + out["sample_message_texts"] = list(p.sample_message_texts) + out["by_override_source"] = dict(p.by_override_source) return out @@ -177,3 +209,87 @@ def generate_proposals( ) out.sort(key=lambda p: (-p.confidence, -p.escalation_rate)) return out + + +# --------------------------------------------------------------------------- +# Loosen-path generator (KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW) +# --------------------------------------------------------------------------- + + +def _loosen_confidence( + override_count: int, *, threshold: int +) -> float: + """0..1 confidence from override count + threshold. Hits 1.0 + at 3× threshold so 9 overrides on default config is full + confidence.""" + if override_count <= 0: + return 0.0 + return min(1.0, override_count / (3 * threshold)) + + +def generate_loosen_proposals( + override_rollups: List[RouteOverrideRollup], + *, + now: datetime, +) -> List[RouterTuningProposal]: + """For each route whose override_count crosses the loosen + threshold, emit one ``loosen_review`` proposal. Returns + proposals sorted by confidence descending.""" + threshold = _int_env( + LOOSEN_OVERRIDE_THRESHOLD_ENV, + DEFAULT_LOOSEN_OVERRIDE_THRESHOLD, + minimum=1, + ) + out: List[RouterTuningProposal] = [] + for rollup in override_rollups: + if rollup.override_count < threshold: + continue + # Compose a rationale that surfaces the sample texts the + # operator typed when they manually escalated. That's the + # operator-decision-relevant context: "this is what I + # wanted Opus for; the trigger pattern should cover it." + per_source_str = ( + ", ".join( + f"{src}={count}" + for src, count in sorted(rollup.by_source.items()) + ) + if rollup.by_source + else "(no per-source data)" + ) + sample_block = ( + "; ".join(f'"{t[:80]}"' for t in rollup.sample_message_texts) + if rollup.sample_message_texts + else "(no sample texts captured)" + ) + rationale = ( + f"Route {rollup.route!r} saw {rollup.override_count} " + f"operator-driven Opus override(s) in the rolling 24h " + f"window ({per_source_str}). The Haiku-router would " + f"otherwise have left these on Haiku — the trigger " + f"pattern likely needs loosening to auto-escalate " + f"similar messages. Sample message text(s) operator " + f"escalated: {sample_block}." + ) + confidence = _loosen_confidence( + rollup.override_count, threshold=threshold + ) + out.append( + RouterTuningProposal( + proposal_id=str(uuid.uuid4()), + route=rollup.route, + calls_count=0, + escalation_count=0, + escalation_rate=0.0, + cost_estimate_usd_total=0.0, + recommendation_kind="loosen_review", + rationale=rationale, + confidence=round(confidence, 4), + created_at=now, + status="pending", + override_count=rollup.override_count, + sample_message_texts=list(rollup.sample_message_texts), + by_override_source=dict(rollup.by_source), + ) + ) + out.sort(key=lambda p: (-p.confidence, -p.override_count)) + return out diff --git a/kora_cli/reasoning/anthropic_engine.py b/kora_cli/reasoning/anthropic_engine.py index 617045f9fcc2..2ecdcf74a2e6 100644 --- a/kora_cli/reasoning/anthropic_engine.py +++ b/kora_cli/reasoning/anthropic_engine.py @@ -465,6 +465,28 @@ async def _tool_use_loop( iteration=iteration, cost_rung=cost_rung, ) + # KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW — capture the + # operator-loosen signal. When the operator manually + # forced Opus on iteration 1 (via /opus prefix or + # KORA_FORCE_OPUS env), record an audit row the + # router-tuning promotion loop's loosen-path consumes. + # The signal means "Haiku-router would have left this + # on Haiku, but the operator wanted Opus" — i.e. a + # trigger pattern that could be tightened toward Opus. + # Skip on iteration ≥2 (that's the loop's own + # iteration-earning signal, not an override) and on + # decision_language / cost_clamp / etc. (those aren't + # overrides). + if iteration == 1 and decision.reason in ( + "opus_prefix", + "force_opus_env", + ): + self._emit_opus_override_audit( + decision_reason=decision.reason, + message_text=message_text, + source=source, + caller_session_id=caller_session_id, + ) if decision.model is None: # cost_rung == hard_stop_100 — caller (respond) already # short-circuits on this rung BEFORE entering the loop, @@ -830,6 +852,74 @@ def _record_call_to_telemetry( exc, ) + # KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW — single-character map from + # router decision reason to the audit payload's override_source + # field. Kept as a module-private constant rather than inline so + # the test surface can assert against the exact emitted strings. + _OPUS_OVERRIDE_SOURCE_BY_REASON = { + "opus_prefix": "operator_prefix", + "force_opus_env": "force_env", + } + + # Cap on the original message text recorded in the audit row — + # operator-decision-relevant per #182 precedent, but not so long + # that the audit JSONL grows unbounded on a long-message override. + _OPUS_OVERRIDE_MESSAGE_TEXT_CAP = 240 + + def _emit_opus_override_audit( + self, + *, + decision_reason: str, + message_text: str, + source: str, + caller_session_id: str, + ) -> None: + """Best-effort emit of the ``opus_override.applied`` audit row. + + Captures the per-call signal that the router-tuning loop's + loosen-path consumer needs (#193 follow-on). Fail-soft — + any failure here logs at DEBUG + is swallowed; the + reasoning call is unaffected. + + ``message_text`` is the RAW text (pre /opus-prefix stripping) + so the router-tuning observer can recover the trigger text + the operator wanted Opus to see; truncated to + :data:`_OPUS_OVERRIDE_MESSAGE_TEXT_CAP` chars. + """ + try: + from kora_cli.audit.jsonl_sink import emit_audit + except Exception as exc: + logger.debug( + "[kora.reasoning.opus_override] audit import failed: %r " + "— audit row skipped", + exc, + ) + return + override_source = self._OPUS_OVERRIDE_SOURCE_BY_REASON.get( + decision_reason, decision_reason + ) + truncated_text = (message_text or "")[ + : self._OPUS_OVERRIDE_MESSAGE_TEXT_CAP + ] + try: + emit_audit( + "opus_override.applied", + { + "original_message_text": truncated_text, + "pre_call_decision_reason": decision_reason, + "override_source": override_source, + "route": source, + }, + caller_session_id=caller_session_id or None, + source="reasoning", + ) + except Exception as exc: + logger.debug( + "[kora.reasoning.opus_override] emit_audit raised %r " + "— audit row skipped", + exc, + ) + def _extract_tool_use_blocks(self, response: Any) -> list: """Return the list of ``tool_use`` content blocks from an Anthropic SDK response. Each block has ``.id``, ``.name``, diff --git a/kora_cli/web_server.py b/kora_cli/web_server.py index 16912dd06c92..00610e3ade05 100644 --- a/kora_cli/web_server.py +++ b/kora_cli/web_server.py @@ -7065,6 +7065,56 @@ async def reject_probe_envelope_proposal( ) +# --- Email-intent (KR-PROMOTE-EMAIL-INTENT — 6th loop) --------------------- + + +@app.get("/api/promotions/email-intent/pending") +async def list_pending_email_intent_proposals() -> Dict[str, Any]: + """Return pending email-intent proposals. + + Payload per ``kora_cli.promote.email_intent.proposer``: + proposal_id / cluster_size / sample_subjects / + proposed_pattern / proposed_action_kind / confidence / + created_at / status / sample_caller_session_ids. + + Operator scaffolds approved patterns manually into + ``kora_cli/intent/email_to_sea_ticket.py`` (probe-fix-envelope + precedent — see #193). + """ + return _promotion_loop_pending("email_intent") + + +@app.post("/api/promotions/email-intent/{proposal_id}/approve") +async def approve_email_intent_proposal( + proposal_id: str, payload: Optional[Dict[str, Any]] = None +) -> Any: + """Approve an email-intent proposal. v1 transitions status + + emits audit only — the regex registry in + ``kora_cli/intent/email_to_sea_ticket.py`` MUST be edited by + hand. Approved/ proposal file is the audit trail for when the + manual scaffold lands.""" + return _promotion_loop_transition( + loop_name="email_intent", + proposal_id=proposal_id, + new_status="approved", + audit_seam="promotion.approved", + payload=payload, + ) + + +@app.post("/api/promotions/email-intent/{proposal_id}/reject") +async def reject_email_intent_proposal( + proposal_id: str, payload: Optional[Dict[str, Any]] = None +) -> Any: + return _promotion_loop_transition( + loop_name="email_intent", + proposal_id=proposal_id, + new_status="rejected", + audit_seam="promotion.rejected", + payload=payload, + ) + + # --------------------------------------------------------------------------- # Email-intent audit lens (KR-FE-EMAIL-INTENT-LOG-PANEL) # --------------------------------------------------------------------------- diff --git a/tests/kora_cli/alerts/test_wake_consumer.py b/tests/kora_cli/alerts/test_wake_consumer.py new file mode 100644 index 000000000000..73fe8547f1a9 --- /dev/null +++ b/tests/kora_cli/alerts/test_wake_consumer.py @@ -0,0 +1,337 @@ +"""Tests for kora_cli.alerts.wake_consumer — KR-ALERT-INVESTIGATION-WAKE-CONSUMER. + +Covers: + * filter: aggregate / non-ok rows skipped (filtered_skipped=True) + * debounce: second event within window skipped (debounce_skipped=True) + * engine_unavailable → fallback DM sent + * engine raise → fallback DM sent + * happy path: reasoning called → DM sent + alert.investigation_completed + audit emitted with correct shape + * caller_session_id wired ``alert:{category}:{severity}`` everywhere + * source attribution on the engine call → route="alert_investigation" + * slack_dm_log entry written with caller_session_id (the 4-stream join key) +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from kora_cli.alerts.wake_consumer import ( + BYPASS_CRITICAL_ENV, + DEBOUNCE_SECONDS_ENV, + JOSHUA_SLACK_USER_ID_ENV, + AlertWakeConsumer, + format_fallback_text, + format_investigation_prompt, + format_operator_dm, +) + + +_JOSHUA_USER_ID = "U01JOSHUA" + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setenv( + "KORA_AUDIT_LOG_PATH", str(tmp_path / "kora_audit_log.jsonl") + ) + monkeypatch.setenv( + "KORA_SLACK_DM_LOG_PATH", str(tmp_path / "slack_dm_log.jsonl") + ) + monkeypatch.setenv(JOSHUA_SLACK_USER_ID_ENV, _JOSHUA_USER_ID) + monkeypatch.delenv(DEBOUNCE_SECONDS_ENV, raising=False) + monkeypatch.delenv(BYPASS_CRITICAL_ENV, raising=False) + return tmp_path + + +def _make_event( + *, + alert_id: str = "cost_warn_75", + category: str = "cost_ladder", + severity: str = "warning", + channel: str = "slack", + status: str = "ok", +) -> dict: + return { + "alert_id": alert_id, + "category": category, + "severity": severity, + "channel": channel, + "status": status, + } + + +def _make_engine(*, text: str = "Investigation: ...", error=None): + engine = MagicMock() + result = MagicMock() + result.text = text + result.error = error + result.model_used = "claude-haiku-4-5-20251001" + result.input_tokens = 120 + result.output_tokens = 40 + result.cache_creation_input_tokens = 0 + result.cache_read_input_tokens = 0 + engine.respond = AsyncMock(return_value=result) + return engine + + +def _make_slack(): + client = MagicMock() + client.post_dm = AsyncMock(return_value={"ts": "1.0"}) + return client + + +def _make_consumer(*, engine=None, slack=None): + return AlertWakeConsumer( + reasoning_engine_factory=lambda: engine, + slack_client_factory=lambda: slack, + ) + + +def _read_audit(tmp_path) -> list: + path = tmp_path / "kora_audit_log.jsonl" + if not path.is_file(): + return [] + return [ + json.loads(line) + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +def _read_slack_dm_log(tmp_path) -> list: + path = tmp_path / "slack_dm_log.jsonl" + if not path.is_file(): + return [] + return [ + json.loads(line) + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +# =========================================================================== +# Formatters +# =========================================================================== + + +def test_format_investigation_prompt_includes_all_fields(): + text = format_investigation_prompt(_make_event()) + assert "cost_ladder" in text + assert "warning" in text + assert "cost_warn_75" in text + assert "slack" in text + + +def test_format_operator_dm_severity_emoji(): + text = format_operator_dm( + category="cost_ladder", severity="critical", reasoning_text="bar" + ) + assert text.startswith("🚨") + assert "cost_ladder" in text + assert "bar" in text + + +def test_format_fallback_text_includes_reason(): + text = format_fallback_text(_make_event(), reason="cost_ladder_halted") + assert "cost_ladder_halted" in text + + +# =========================================================================== +# Filter (aggregate / non-ok rows) +# =========================================================================== + + +@pytest.mark.asyncio +async def test_burst_summary_channel_filtered_skipped(): + """notification.dispatched with channel="slack-burst-summary" or + a synthetic channel value isn't a per-alert row — skip.""" + consumer = _make_consumer(engine=_make_engine(), slack=_make_slack()) + event = _make_event(channel="burst_summary") # not in allowlist + outcome = await consumer.consume_alert_event(event) + assert outcome.filtered_skipped is True + assert outcome.dispatched is False + + +@pytest.mark.asyncio +async def test_failed_dispatch_filtered_skipped(): + consumer = _make_consumer(engine=_make_engine(), slack=_make_slack()) + event = _make_event(status="failed") + outcome = await consumer.consume_alert_event(event) + assert outcome.filtered_skipped is True + + +# =========================================================================== +# Debounce +# =========================================================================== + + +@pytest.mark.asyncio +async def test_second_dispatch_same_category_severity_debounced(): + consumer = _make_consumer(engine=_make_engine(), slack=_make_slack()) + o1 = await consumer.consume_alert_event(_make_event()) + o2 = await consumer.consume_alert_event(_make_event()) + assert o1.dispatched is True + assert o2.dispatched is False + assert o2.debounce_skipped is True + + +@pytest.mark.asyncio +async def test_different_category_independent(): + consumer = _make_consumer(engine=_make_engine(), slack=_make_slack()) + await consumer.consume_alert_event( + _make_event(category="cost_ladder") + ) + out = await consumer.consume_alert_event( + _make_event(category="service_unhealthy") + ) + assert out.dispatched is True + + +@pytest.mark.asyncio +async def test_debounce_zero_disables(monkeypatch): + monkeypatch.setenv(DEBOUNCE_SECONDS_ENV, "0") + consumer = _make_consumer(engine=_make_engine(), slack=_make_slack()) + await consumer.consume_alert_event(_make_event()) + out = await consumer.consume_alert_event(_make_event()) + # Without the window, both dispatch. + assert out.dispatched is True + + +@pytest.mark.asyncio +async def test_critical_bypass_truthy_skips_debounce(monkeypatch): + monkeypatch.setenv(BYPASS_CRITICAL_ENV, "true") + consumer = _make_consumer(engine=_make_engine(), slack=_make_slack()) + await consumer.consume_alert_event( + _make_event(severity="critical") + ) + out = await consumer.consume_alert_event( + _make_event(severity="critical") + ) + assert out.dispatched is True + + +# =========================================================================== +# Engine paths +# =========================================================================== + + +@pytest.mark.asyncio +async def test_engine_unavailable_sends_fallback_dm(tmp_path): + """engine=None → fallback DM (still sent), no reasoning_invoked.""" + consumer = _make_consumer(engine=None, slack=_make_slack()) + out = await consumer.consume_alert_event(_make_event()) + assert out.dispatched is True + assert out.reasoning_invoked is False + assert out.dm_sent is True + + +@pytest.mark.asyncio +async def test_engine_raise_sends_fallback_dm(tmp_path): + engine = MagicMock() + engine.respond = AsyncMock(side_effect=RuntimeError("boom")) + consumer = _make_consumer(engine=engine, slack=_make_slack()) + out = await consumer.consume_alert_event(_make_event()) + assert out.dispatched is True + assert out.reasoning_invoked is False + assert out.dm_sent is True + assert out.error is not None and out.error.startswith("engine_exception:") + + +@pytest.mark.asyncio +async def test_engine_returns_error_sends_fallback_dm(): + engine = _make_engine(error="cost_ladder_halted", text="") + consumer = _make_consumer(engine=engine, slack=_make_slack()) + out = await consumer.consume_alert_event(_make_event()) + assert out.reasoning_invoked is False + assert out.error == "cost_ladder_halted" + assert out.dm_sent is True + + +@pytest.mark.asyncio +async def test_happy_path_dispatches_and_audits(tmp_path): + """End-to-end happy path: reasoning called → DM sent → both + alert.investigation_completed AND slack_dm_log entry written + with the same caller_session_id.""" + engine = _make_engine(text="Cost is at 76% — review burn rate") + slack = _make_slack() + consumer = _make_consumer(engine=engine, slack=slack) + outcome = await consumer.consume_alert_event(_make_event()) + + assert outcome.dispatched is True + assert outcome.reasoning_invoked is True + assert outcome.dm_sent is True + slack.post_dm.assert_awaited_once() + + # Stream 2 — alert.investigation_completed audit. + audit = _read_audit(tmp_path) + completed = [ + e for e in audit if e["seam"] == "alert.investigation_completed" + ] + assert len(completed) == 1 + details = completed[0]["details"] + assert details["alert_id"] == "cost_warn_75" + assert details["category"] == "cost_ladder" + assert details["severity"] == "warning" + assert details["dm_status"] == "sent" + assert details["model_used"] == "claude-haiku-4-5-20251001" + assert details["investigation_summary_text"].startswith("Cost is at 76%") + assert details["autoaction_attempted"] is False # v1 hardcoded + + # caller_session_id pattern (alert:{category}:{severity}) + assert ( + completed[0]["caller_session_id"] + == "alert:cost_ladder:warning" + ) + + # Stream 3 — slack_dm_log entry with the same caller_session_id. + dm_log = _read_slack_dm_log(tmp_path) + assert len(dm_log) == 1 + assert dm_log[0]["caller_session_id"] == "alert:cost_ladder:warning" + assert dm_log[0]["model_used"] == "claude-haiku-4-5-20251001" + assert dm_log[0]["send_status"] == "ok" + + +@pytest.mark.asyncio +async def test_incoming_message_source_is_alert_investigation(): + """Engine receives IncomingMessage with source='alert_investigation' + so the engine's bypass-path telemetry mapping attributes the call + to ROUTE_ALERT_INVESTIGATION (per #190's wire).""" + engine = _make_engine() + consumer = _make_consumer(engine=engine, slack=_make_slack()) + await consumer.consume_alert_event(_make_event()) + engine.respond.assert_awaited_once() + (message, _context), _ = engine.respond.await_args + assert message.source == "alert_investigation" + assert message.metadata["category"] == "cost_ladder" + assert message.metadata["alert_id"] == "cost_warn_75" + + +@pytest.mark.asyncio +async def test_slack_client_none_skips_dm_but_still_audits(tmp_path): + """slack=None → DM not sent, but the alert.investigation_completed + audit still fires with dm_status reflecting the failure.""" + consumer = _make_consumer(engine=_make_engine(), slack=None) + out = await consumer.consume_alert_event(_make_event()) + assert out.dm_sent is False + audit = _read_audit(tmp_path) + completed = [ + e for e in audit if e["seam"] == "alert.investigation_completed" + ] + assert len(completed) == 1 + assert completed[0]["details"]["dm_status"] == "failed_send" + + +@pytest.mark.asyncio +async def test_reset_debounce_state_clears_map(): + consumer = _make_consumer(engine=_make_engine(), slack=_make_slack()) + await consumer.consume_alert_event(_make_event()) + assert consumer.debounce_map_size == 1 + consumer.reset_debounce_state() + assert consumer.debounce_map_size == 0 diff --git a/tests/kora_cli/promote/email_intent/__init__.py b/tests/kora_cli/promote/email_intent/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/kora_cli/promote/email_intent/test_email_intent.py b/tests/kora_cli/promote/email_intent/test_email_intent.py new file mode 100644 index 000000000000..cdd0fc524bb7 --- /dev/null +++ b/tests/kora_cli/promote/email_intent/test_email_intent.py @@ -0,0 +1,282 @@ +"""Tests for kora_cli.promote.email_intent — KR-PROMOTE-EMAIL-INTENT.""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from kora_cli.audit.jsonl_sink import BATCH_SIZE_ENV, _reset_batching_for_tests +from kora_cli.promote.email_intent.observer import ( + EmailIntentObservation, + collect_recent_logged_only, +) +from kora_cli.promote.email_intent.plugin import ( + ENABLED_ENV, + run_email_intent_cycle, +) +from kora_cli.promote.email_intent.proposer import ( + MIN_CLUSTER_SIZE_ENV, + generate_proposals, +) + + +@pytest.fixture(autouse=True) +def _isolate(tmp_path, monkeypatch): + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setenv("KORA_PROMOTIONS_DIR", str(tmp_path / "promotions")) + monkeypatch.setenv( + "KORA_AUDIT_LOG_PATH", str(tmp_path / "kora_audit_log.jsonl") + ) + monkeypatch.setenv(BATCH_SIZE_ENV, "0") + _reset_batching_for_tests() + yield + _reset_batching_for_tests() + + +def _write_audit(tmp_path: Path, entries: list) -> None: + path = tmp_path / "kora_audit_log.jsonl" + path.write_text( + "\n".join(json.dumps(e) for e in entries) + "\n", encoding="utf-8" + ) + + +def _entry( + *, + subject: str = "Quick idea about probes", + action: str = "logged_only", + pattern: str | None = None, + reason: str = "no_pattern_matched", + confidence: str = "unrecognized", + caller_session_id: str = "email:msg-1", + emitted_at: datetime | None = None, +) -> dict: + if emitted_at is None: + emitted_at = datetime.now(timezone.utc) - timedelta(hours=1) + return { + "emitted_at": emitted_at.isoformat(), + "seam": "intent.email_to_sea_ticket", + "details": { + "action": action, + "subject": subject, + "pattern_matched": pattern, + "confidence": confidence, + "reason": reason, + }, + "caller_session_id": caller_session_id, + "source": "email", + } + + +# =========================================================================== +# Observer +# =========================================================================== + + +@pytest.mark.asyncio +async def test_observer_returns_logged_only_rows(tmp_path): + _write_audit( + tmp_path, + [ + _entry(subject="Idea: ship the v2 panel"), + _entry(subject="Burn warning"), + ], + ) + out = await collect_recent_logged_only( + since=datetime.now(timezone.utc) - timedelta(days=1) + ) + assert len(out) == 2 + assert all(o.confidence == "unrecognized" for o in out) + + +@pytest.mark.asyncio +async def test_observer_skips_other_actions(tmp_path): + _write_audit( + tmp_path, + [ + _entry(subject="Logged only", action="logged_only"), + _entry(subject="Got created", action="created"), + _entry(subject="Dry run", action="dry_run"), + ], + ) + out = await collect_recent_logged_only( + since=datetime.now(timezone.utc) - timedelta(days=1) + ) + assert [o.subject for o in out] == ["Logged only"] + + +@pytest.mark.asyncio +async def test_observer_skips_empty_subjects(tmp_path): + _write_audit( + tmp_path, + [ + _entry(subject=""), + _entry(subject=" "), + _entry(subject="real text"), + ], + ) + out = await collect_recent_logged_only( + since=datetime.now(timezone.utc) - timedelta(days=1) + ) + assert [o.subject for o in out] == ["real text"] + + +@pytest.mark.asyncio +async def test_observer_respects_since(tmp_path): + old = datetime.now(timezone.utc) - timedelta(days=30) + fresh = datetime.now(timezone.utc) - timedelta(hours=1) + _write_audit( + tmp_path, + [ + _entry(subject="too old", emitted_at=old), + _entry(subject="fresh enough", emitted_at=fresh), + ], + ) + out = await collect_recent_logged_only( + since=datetime.now(timezone.utc) - timedelta(days=7) + ) + assert [o.subject for o in out] == ["fresh enough"] + + +# =========================================================================== +# Proposer +# =========================================================================== + + +def _obs(subject: str, csid: str = "email:c1") -> EmailIntentObservation: + return EmailIntentObservation( + subject=subject, + pattern_matched=None, + confidence="unrecognized", + reason="no_pattern_matched", + caller_session_id=csid, + timestamp=datetime.now(timezone.utc), + ) + + +@pytest.mark.asyncio +async def test_proposer_clusters_similar_subjects(monkeypatch): + monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2") + obs = [ + _obs("Question: burn rate for this week", csid=f"q{i}") + for i in range(3) + ] + [ + _obs("Unrelated thing entirely about birds", csid="other") + ] + out = await generate_proposals(obs, now=datetime.now(timezone.utc)) + # The 3 burn-rate subjects should land in one cluster (≥2 min); + # the bird subject is alone and below threshold. + assert len(out) == 1 + assert out[0].cluster_size == 3 + # Pattern is alternation of top-3 cross-subject tokens — exact + # token choice depends on Counter tie-break (the burn-rate + # subjects have multiple tokens tied at frequency 3). Assert + # the pattern contains at least one of the cluster's meaningful + # tokens rather than pinning a specific one. + pat_lower = out[0].proposed_pattern.lower() + assert any( + tok in pat_lower + for tok in ("question", "burn", "rate", "week") + ) + + +@pytest.mark.asyncio +async def test_proposer_skips_under_min_cluster(monkeypatch): + monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "3") + obs = [_obs(f"Idea about X {i}", csid=f"c{i}") for i in range(2)] + out = await generate_proposals(obs, now=datetime.now(timezone.utc)) + assert out == [] + + +@pytest.mark.asyncio +async def test_proposer_default_action_kind_is_save_note(monkeypatch): + monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2") + obs = [_obs(f"Idea: cool feature ABC {i}", csid=f"c{i}") for i in range(3)] + out = await generate_proposals(obs, now=datetime.now(timezone.utc)) + assert len(out) == 1 + assert out[0].proposed_action_kind == "save_note" + + +@pytest.mark.asyncio +async def test_proposer_pattern_is_safe_regex(monkeypatch): + """Subjects with regex metacharacters don't bomb the engine — + derive_pattern escapes them.""" + import re as _re + + monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2") + obs = [ + _obs(f"Note (urgent) about $foo {i}", csid=f"c{i}") for i in range(3) + ] + out = await generate_proposals(obs, now=datetime.now(timezone.utc)) + assert len(out) == 1 + # The pattern must compile without error. + _re.compile(out[0].proposed_pattern) + + +@pytest.mark.asyncio +async def test_proposer_sample_subjects_deduped_and_capped(monkeypatch): + monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2") + obs = [_obs("Same subject", csid=f"c{i}") for i in range(8)] + out = await generate_proposals(obs, now=datetime.now(timezone.utc)) + assert len(out) == 1 + # Same subject dedups to 1 in the sample list. + assert out[0].sample_subjects == ["Same subject"] + # Sample callers list cap. + assert len(out[0].sample_caller_session_ids) <= 3 + + +# =========================================================================== +# Cycle +# =========================================================================== + + +def _read_audit(tmp_path) -> list: + path = tmp_path / "kora_audit_log.jsonl" + if not path.is_file(): + return [] + return [ + json.loads(line) + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +@pytest.mark.asyncio +async def test_cycle_emits_correct_seam(tmp_path, monkeypatch): + monkeypatch.setenv(MIN_CLUSTER_SIZE_ENV, "2") + _write_audit( + tmp_path, + [ + _entry( + subject="Idea: new dashboard panel for cost burn", + caller_session_id=f"email:msg-{i}", + ) + for i in range(3) + ], + ) + summary = await run_email_intent_cycle() + assert summary["enabled"] is True + assert summary["observations_read"] == 3 + assert summary["proposals_generated"] == 1 + assert summary["proposals_persisted"] == 1 + assert summary["auto_apply_mode"] is False + + rows = _read_audit(tmp_path) + promo = [ + r + for r in rows + if r["seam"] == "promotion.email_intent_pattern_proposed" + ] + assert len(promo) == 1 + assert promo[0]["details"]["action"] == "proposed" + assert promo[0]["source"] == "email" + + +@pytest.mark.asyncio +async def test_cycle_disabled_short_circuits(monkeypatch): + monkeypatch.setenv(ENABLED_ENV, "false") + summary = await run_email_intent_cycle() + assert summary["enabled"] is False + assert summary["proposals_generated"] == 0 diff --git a/tests/kora_cli/promote/router_tuning/test_router_tuning.py b/tests/kora_cli/promote/router_tuning/test_router_tuning.py index 1419ae141516..34ab04405617 100644 --- a/tests/kora_cli/promote/router_tuning/test_router_tuning.py +++ b/tests/kora_cli/promote/router_tuning/test_router_tuning.py @@ -219,3 +219,178 @@ async def test_cycle_generates_persists_and_audits(tmp_path, monkeypatch): assert pending_dir.is_dir() files = list(pending_dir.iterdir()) assert len(files) == 1 + + +# =========================================================================== +# KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW — loosen-path activation +# =========================================================================== + + +from datetime import timedelta + +from kora_cli.promote.router_tuning.observer import ( + RouteOverrideRollup, + collect_route_overrides, +) +from kora_cli.promote.router_tuning.proposer import ( + DEFAULT_LOOSEN_OVERRIDE_THRESHOLD, + LOOSEN_OVERRIDE_THRESHOLD_ENV, + generate_loosen_proposals, +) + + +def _write_audit_jsonl(tmp_path, entries): + path = tmp_path / "kora_audit_log.jsonl" + path.write_text( + "\n".join(json.dumps(e) for e in entries) + "\n", encoding="utf-8" + ) + + +def _override_entry( + *, + route: str = "slack_dm", + message_text: str = "please tell me the right call here", + source: str = "operator_prefix", + reason: str = "opus_prefix", + emitted_at=None, +) -> dict: + if emitted_at is None: + emitted_at = datetime.now(timezone.utc) - timedelta(hours=1) + return { + "emitted_at": emitted_at.isoformat(), + "seam": "opus_override.applied", + "details": { + "original_message_text": message_text, + "pre_call_decision_reason": reason, + "override_source": source, + "route": route, + }, + "caller_session_id": "slack_dm:D1:1.001", + "source": "reasoning", + } + + +def test_collect_route_overrides_groups_by_route(tmp_path): + _write_audit_jsonl( + tmp_path, + [ + _override_entry(route="slack_dm"), + _override_entry(route="slack_dm"), + _override_entry(route="email_inbound"), + ], + ) + out = collect_route_overrides( + since=datetime.now(timezone.utc) - timedelta(days=1) + ) + by_route = {r.route: r for r in out} + assert by_route["slack_dm"].override_count == 2 + assert by_route["email_inbound"].override_count == 1 + + +def test_collect_route_overrides_per_source_breakdown(tmp_path): + _write_audit_jsonl( + tmp_path, + [ + _override_entry(source="operator_prefix"), + _override_entry(source="operator_prefix"), + _override_entry(source="force_env"), + ], + ) + out = collect_route_overrides( + since=datetime.now(timezone.utc) - timedelta(days=1) + ) + assert len(out) == 1 + assert out[0].by_source == {"operator_prefix": 2, "force_env": 1} + + +def test_collect_route_overrides_captures_sample_texts(tmp_path): + _write_audit_jsonl( + tmp_path, + [ + _override_entry(message_text="should I ship the migration"), + _override_entry(message_text="what's the right call here"), + _override_entry(message_text="explain the tradeoff"), + _override_entry(message_text="another override"), + ], + ) + out = collect_route_overrides( + since=datetime.now(timezone.utc) - timedelta(days=1) + ) + # _SAMPLE_TEXT_CAP = 3 — only first 3 captured. + assert len(out[0].sample_message_texts) == 3 + + +def _override_rollup( + route: str = "slack_dm", + override_count: int = 5, +) -> RouteOverrideRollup: + return RouteOverrideRollup( + route=route, + override_count=override_count, + sample_message_texts=["should I ship migration X"], + by_source={"operator_prefix": override_count}, + ) + + +def test_loosen_proposer_skips_under_threshold(monkeypatch): + monkeypatch.delenv(LOOSEN_OVERRIDE_THRESHOLD_ENV, raising=False) + rollups = [_override_rollup(override_count=2)] + out = generate_loosen_proposals(rollups, now=datetime.now(timezone.utc)) + assert out == [] + + +def test_loosen_proposer_emits_when_threshold_crossed(): + rollups = [_override_rollup(override_count=5)] + out = generate_loosen_proposals(rollups, now=datetime.now(timezone.utc)) + assert len(out) == 1 + assert out[0].recommendation_kind == "loosen_review" + assert out[0].override_count == 5 + assert "should I ship migration X" in out[0].rationale + + +def test_loosen_proposer_env_override_threshold(monkeypatch): + monkeypatch.setenv(LOOSEN_OVERRIDE_THRESHOLD_ENV, "1") + rollups = [_override_rollup(override_count=1)] + out = generate_loosen_proposals(rollups, now=datetime.now(timezone.utc)) + assert len(out) == 1 + + +@pytest.mark.asyncio +async def test_cycle_loosen_path_end_to_end(tmp_path, monkeypatch): + """Synthetic opus_override.applied audit → cycle reads + overrides → loosen proposals emitted alongside the tighten + path.""" + monkeypatch.setenv(LOOSEN_OVERRIDE_THRESHOLD_ENV, "2") + _write_audit_jsonl( + tmp_path, + [ + _override_entry(route="slack_dm", message_text=f"override {i}") + for i in range(3) + ], + ) + # No telemetry escalations — tighten path produces 0 proposals. + from unittest.mock import MagicMock as _MM + + fake = _MM() + fake.snapshot.return_value = {"rolling_24h": {}, "monthly": {}} + monkeypatch.setattr( + "kora_cli.telemetry.cost_telemetry.get_telemetry", lambda: fake + ) + monkeypatch.setattr( + "kora_cli.telemetry.get_telemetry", lambda: fake + ) + + summary = await run_router_tuning_cycle() + assert summary["overrides_observed"] == 1 + assert summary["proposals_generated"] == 1 + assert summary["proposals_persisted"] == 1 + + audit = _read_audit(tmp_path) + promo = [ + r + for r in audit + if r["seam"] == "promotion.router_trigger_proposed" + ] + assert len(promo) == 1 + assert promo[0]["details"]["recommendation_kind"] == "loosen_review" + assert promo[0]["details"]["override_count"] == 3 diff --git a/tests/kora_cli/reasoning/test_anthropic_engine_router.py b/tests/kora_cli/reasoning/test_anthropic_engine_router.py index 45876a4c4584..3ba1e0d56122 100644 --- a/tests/kora_cli/reasoning/test_anthropic_engine_router.py +++ b/tests/kora_cli/reasoning/test_anthropic_engine_router.py @@ -556,3 +556,136 @@ async def test_telemetry_route_mapping_covers_all_sources( await engine.respond(message, _ctx()) assert len(telemetry_spy) >= 1 assert telemetry_spy[0]["route"] == expected_route + + +# --------------------------------------------------------------------------- +# KR-PROMOTE-ROUTER-LOOSEN-AUDIT-ROW — opus_override.applied emission +# --------------------------------------------------------------------------- + + +@pytest.fixture +def _audit_redirect(tmp_path, monkeypatch): + """Force per-emit sync audit writes for these tests so the + audit JSONL is readable immediately.""" + import json as _json + + from kora_cli.audit.jsonl_sink import ( + BATCH_SIZE_ENV, + _reset_batching_for_tests, + ) + + monkeypatch.setenv("KORA_HOME", str(tmp_path)) + monkeypatch.setenv( + "KORA_AUDIT_LOG_PATH", str(tmp_path / "kora_audit_log.jsonl") + ) + monkeypatch.setenv(BATCH_SIZE_ENV, "0") + _reset_batching_for_tests() + yield tmp_path / "kora_audit_log.jsonl" + _reset_batching_for_tests() + + +def _read_audit_rows(path): + import json as _json + + if not path.is_file(): + return [] + return [ + _json.loads(line) + for line in path.read_text(encoding="utf-8").splitlines() + if line.strip() + ] + + +@pytest.mark.asyncio +async def test_opus_prefix_emits_opus_override_audit( + monkeypatch, system_prompt_path, telemetry_spy, _audit_redirect +): + """/opus prefix on iteration 1 → opus_override.applied audit + row with override_source=operator_prefix + truncated message + text + route from source.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + client = _make_client([_response([_text_block("ok")])]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + await engine.respond( + _msg("/opus tell me about the migration"), _ctx() + ) + rows = _read_audit_rows(_audit_redirect) + override_rows = [ + r for r in rows if r["seam"] == "opus_override.applied" + ] + assert len(override_rows) == 1 + details = override_rows[0]["details"] + assert details["override_source"] == "operator_prefix" + assert details["pre_call_decision_reason"] == "opus_prefix" + assert details["route"] == "slack_dm" + assert "tell me about the migration" in details["original_message_text"] + + +@pytest.mark.asyncio +async def test_force_opus_env_emits_opus_override_audit( + monkeypatch, system_prompt_path, telemetry_spy, _audit_redirect +): + """KORA_FORCE_OPUS=true on iteration 1 → opus_override.applied + with override_source=force_env.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + monkeypatch.setenv("KORA_FORCE_OPUS", "true") + client = _make_client([_response([_text_block("ok")])]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + await engine.respond(_msg("routine question"), _ctx()) + rows = _read_audit_rows(_audit_redirect) + override_rows = [ + r for r in rows if r["seam"] == "opus_override.applied" + ] + assert len(override_rows) == 1 + assert override_rows[0]["details"]["override_source"] == "force_env" + + +@pytest.mark.asyncio +async def test_default_haiku_does_not_emit_opus_override( + monkeypatch, system_prompt_path, telemetry_spy, _audit_redirect +): + """A routine call that the router leaves on Haiku must NOT emit + the override seam — only operator-driven overrides count.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + client = _make_client([_response([_text_block("hi")])]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + await engine.respond(_msg("hello there"), _ctx()) + rows = _read_audit_rows(_audit_redirect) + override_rows = [ + r for r in rows if r["seam"] == "opus_override.applied" + ] + assert override_rows == [] + + +@pytest.mark.asyncio +async def test_decision_language_does_not_emit_opus_override( + monkeypatch, system_prompt_path, telemetry_spy, _audit_redirect +): + """Decision-language pattern → Opus, but that's a router-side + heuristic, NOT an operator override. Don't emit the seam.""" + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + client = _make_client([_response([_text_block("ok")])]) + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + # "should i ship" is a decision-language pattern per the router. + await engine.respond(_msg("should i ship this?"), _ctx()) + rows = _read_audit_rows(_audit_redirect) + override_rows = [ + r for r in rows if r["seam"] == "opus_override.applied" + ] + assert override_rows == []