From c4a3319a6cb3d698d57f9b5df4b1eeb9d840d8d0 Mon Sep 17 00:00:00 2001 From: olegbrok Date: Wed, 6 May 2026 22:13:17 -0700 Subject: [PATCH 1/3] feat(auth-alerts): operator alert + auth_status when Claude credentials break (#400) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(auth-alerts): notify operator + surface auth_status when Claude auth breaks When the Claude SDK returns error='authentication_failed' the streaming session has always suppressed the response (so raw API errors never reach end users). The downside: the operator who can re-auth has no signal until someone complains. This adds two channels for that signal. 1) AuthFailureTracker (auth_alerts.py): - Per-host sliding window of auth failures across all agents. - Fires an alert at threshold (3 fails in 5 min, OR 3 distinct agents failing at once = host-wide outage). - 30 min cooldown between alerts to prevent spam. - record_success() clears state on the next clean turn. 2) Streaming-session hooks (streaming_session.py): - New auth_alert_callback / auth_success_callback ctor params, called from the existing assistant-error and ResultMessage paths. - _is_auth_error() pattern-matches authentication_failed, invalid_api_key, unauthorized, auth_error, permission_error so future SDK renames keep alerting. 3) Operator delivery (api.py): - _on_auth_failure resolves operator chat (system_settings.operator_chat_id first; otherwise the chat_id appearing across the most approved_users rows) and DMs via the affected agent's own bot through _broker_send. - Codex sessions skip the hook — different auth lifecycle. 4) Health surface: - /admin/watchdog gains auth_status with status=ok|degraded|broken, window/threshold/cooldown config, outage age, alerts sent, and a per-agent failures list. External monitors can alert on broken. 20 new tests covering thresholds, cooldown, window eviction, host-wide detection, success-clears-state, operator resolution fallbacks, and a TestClient integration check on /admin/watchdog. Full suite: 1620 passed. Why now: hit a real outage on the Pi today — all 5 agents auth_failed for 12+ hours and nobody knew because the symptom was "agents stop responding." This makes the next one a 30-second telegram ping instead of a manual log grep. Co-Authored-By: Claude Opus 4.7 * fix(auth-alerts): preserve cooldown on broker delivery failure + plain-text alert PR #400 review (Pushok) caught the regression this PR was meant to prevent: record_failure() advanced _last_alert_at *before* awaiting _broker_send, so a network blip on the very first send-attempt silenced the alert system for 30 minutes — exactly the Sasha-down-9-12h shape we're designing against. Two-phase alerting: - record_failure() now decides only; never mutates cooldown. - New commit_alert() is invoked by _on_auth_failure ONLY after _broker_send succeeds. If delivery raises, cooldown is preserved so the next failure retries instead of being lost. Also fix the markdown parse_mode mismatch Pushok flagged: format_alert_message used `*agent*` and backticks, but _broker_send defaults to no parse_mode so operators would see literal `*sasha*`. Switched to plain text — safer than escaping for MarkdownV2 with arbitrary error strings. Tracker docstring: clarified concurrency contract (single event loop, no internal Lock needed) per Pushok's note that "not async-safe" overstated it. New tests: - test_failed_delivery_preserves_cooldown_and_retries_next_failure (regression) - test_commit_alert_advances_cooldown_and_counter - test_format_alert_uses_no_markdown_so_plain_text_renders_clean - two existing assertions extended to verify the decide/commit split Full suite: 1770 passed, 1 skipped. Ruff clean. Co-Authored-By: Claude Opus 4.7 --------- Co-authored-by: Oleg Co-authored-by: Claude Opus 4.7 --- src/pinky_daemon/api.py | 113 ++++++++- src/pinky_daemon/auth_alerts.py | 316 +++++++++++++++++++++++++ src/pinky_daemon/streaming_session.py | 51 ++++ tests/test_auth_alerts.py | 328 ++++++++++++++++++++++++++ 4 files changed, 806 insertions(+), 2 deletions(-) create mode 100644 src/pinky_daemon/auth_alerts.py create mode 100644 tests/test_auth_alerts.py diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index 1f91c1ce..880be6e7 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -594,6 +594,17 @@ async def security_headers_middleware(request: Request, call_next): _comms_db = db_path.replace(".db", "_agent_comms.db") comms = AgentComms(db_path=_comms_db) + # Auth-failure tracker — counts SDK auth errors across all streaming + # sessions on this host and decides when to alert the operator. The + # callback wiring is created in _build_auth_alert_callbacks() below + # once _broker_send is available; the tracker itself is host-global. + from pinky_daemon.auth_alerts import ( + AuthFailureTracker, + format_alert_message, + resolve_operator_chat, + ) + auth_tracker = AuthFailureTracker() + # Message broker — routes platform messages through approval checks to agent sessions _platform_adapters: dict[tuple[str, str], object] = {} @@ -1571,6 +1582,83 @@ async def _on_stream_event(event: dict): return _on_stream_event + async def _on_auth_failure(agent_name: str, error: str) -> None: + """Streaming-session hook: record an auth failure and alert the operator. + + Called from StreamingSession's reader loop whenever the SDK reports + an authentication-class error (see auth_alerts._is_auth_error). The + AuthFailureTracker decides whether this failure crosses the alert + threshold and whether the cooldown has elapsed; if both, we DM the + operator via _broker_send using the affected agent's own bot. + """ + try: + decision = auth_tracker.record_failure(agent_name, error) + except Exception as e: + _log(f"auth_alerts: tracker.record_failure raised: {e}") + return + + if not decision.get("should_alert"): + return + + try: + chat_id, platform = resolve_operator_chat( + get_setting=agents.get_setting, + list_all_approved_users=agents.list_all_approved_users, + ) + except Exception as e: + _log(f"auth_alerts: resolve_operator_chat raised: {e}") + return + + if not chat_id: + _log( + "auth_alerts: auth failure detected but no operator_chat_id " + "configured and no approved_users to fall back on — " + "alert suppressed (set system_settings.operator_chat_id)" + ) + return + + try: + host_label = agents.get_setting("host_label", "") or "" + except Exception: + host_label = "" + + body = format_alert_message( + agent_name=agent_name, + decision=decision, + error=error, + host_label=host_label, + ) + + try: + await _broker_send(agent_name, platform, chat_id, body) + except Exception as e: + # Delivery failed — do NOT commit the alert. The next failure + # crossing threshold will retry instead of being silenced for + # the cooldown window. + _log( + f"auth_alerts: failed to send operator alert via " + f"{agent_name} bot to {platform}:{chat_id}: {e} " + f"(cooldown not advanced — will retry on next failure)" + ) + return + + try: + auth_tracker.commit_alert() + except Exception as e: + _log(f"auth_alerts: tracker.commit_alert raised: {e}") + _log( + f"auth_alerts: operator alert sent to {platform}:{chat_id} " + f"for {agent_name} (reason={decision.get('reason')}, " + f"agents_failing={decision.get('agents_failing')})" + ) + + def _on_auth_success(agent_name: str) -> None: + """Clear an agent's auth-failure record after a successful turn.""" + try: + auth_tracker.record_success(agent_name) + except Exception as e: + _log(f"auth_alerts: tracker.record_success raised: {e}") + async def _make_streaming_session_id_callback(agent_name: str, label: str): """Persist a streaming session ID when captured from the SDK. @@ -1828,6 +1916,13 @@ def runtime_from_legacy_provider(agent_config) -> str: "analytics_store": analytics, "registry": agents, } + # Auth-failure detection is currently only wired for the SDK-based + # StreamingSession. Codex sessions don't surface an equivalent + # "authentication_failed" signal — they rely on a separate provider + # token lifecycle — so we skip the hooks there. + if not is_codex: + init_kwargs["auth_alert_callback"] = _on_auth_failure + init_kwargs["auth_success_callback"] = _on_auth_success if is_codex: init_kwargs["stream_event_callback"] = await _make_streaming_event_callback(agent_name, label) @@ -6461,8 +6556,22 @@ async def on_shutdown(): @app.get("/admin/watchdog") async def get_watchdog_status(): - """Return session watchdog status and tracked agents.""" - return watchdog.status() + """Return session watchdog status, tracked agents, and Claude auth health. + + Three top-level keys for external monitors: + running / check_interval / agents — session-stuck watchdog state. + auth_status — host-wide Claude auth health (see auth_alerts). + + ``auth_status.status`` is one of ``ok | degraded | broken``; tying a + Prometheus alert to ``broken`` catches credential outages quickly. + """ + out = watchdog.status() + try: + out["auth_status"] = auth_tracker.status() + except Exception as e: + _log(f"admin/watchdog: auth_tracker.status raised: {e}") + out["auth_status"] = {"status": "unknown", "error": str(e)} + return out # ── Admin: Shared MCP Status ───────────────────────── diff --git a/src/pinky_daemon/auth_alerts.py b/src/pinky_daemon/auth_alerts.py new file mode 100644 index 00000000..10e27638 --- /dev/null +++ b/src/pinky_daemon/auth_alerts.py @@ -0,0 +1,316 @@ +"""Auth-failure tracking and operator alerts. + +When the Claude SDK returns ``error='authentication_failed'`` for an agent's +streaming session, the user-facing chat sees nothing — the daemon suppresses +the response so raw error JSON never reaches end users. The downside is the +operator (the person who can re-auth) has no idea Claude credentials are +broken until someone complains. + +This module provides: + +- ``AuthFailureTracker`` — in-memory counter that records auth failures across + all agents on a host, decides when an alert should fire, and enforces a + cooldown so we don't spam the operator. +- ``resolve_operator_chat()`` — best-effort lookup of the operator's + chat_id/platform. Reads ``operator_chat_id`` / ``operator_platform`` from + system_settings if set; otherwise picks the chat_id appearing across the + most ``approved_users`` rows (heuristic: the person with broadest access + is the operator). + +The streaming session calls into the tracker on each auth failure; when the +tracker says "alert", the daemon's alert wiring sends a Telegram DM via the +existing broker. Both are documented inline so the next Brad wedge debug is +faster. +""" +from __future__ import annotations + +import logging +import time +from collections import Counter, defaultdict +from dataclasses import dataclass, field +from typing import Iterable + +_log = logging.getLogger("pinky.auth_alerts").info +_warn = logging.getLogger("pinky.auth_alerts").warning + + +# ── Defaults ──────────────────────────────────────────────────────── + +DEFAULT_FAIL_WINDOW_SECONDS = 300 # 5 min +DEFAULT_FAIL_THRESHOLD = 3 # within window → alert +DEFAULT_ALERT_COOLDOWN_SECONDS = 1800 # 30 min between alerts + + +@dataclass +class _AgentFailures: + """Sliding-window record of recent auth failures for one agent.""" + + timestamps: list[float] = field(default_factory=list) + last_error: str = "" + + def record(self, now: float, error: str, window: float) -> int: + """Append a failure timestamp, prune older ones, return current count.""" + self.timestamps.append(now) + self.last_error = error or "" + cutoff = now - window + self.timestamps = [t for t in self.timestamps if t >= cutoff] + return len(self.timestamps) + + def reset(self) -> None: + self.timestamps = [] + self.last_error = "" + + +class AuthFailureTracker: + """Per-host auth failure detector with cooldown. + + Concurrency: callers must share a single event loop (the streaming-session + reader loop). No internal locking is needed — every method is synchronous + and mutation happens before return. Methods are O(N) over recent + timestamps which is fine — N is bounded by the threshold in practice. + + Two-phase alerting: ``record_failure()`` decides whether an alert *should* + fire but never advances the cooldown by itself. The caller must invoke + ``commit_alert()`` only after the operator notification was actually + delivered. If delivery raises (broker down, bot token revoked, etc.), the + cooldown is preserved so the next failure can retry — that's the whole + reason this PR exists, so a single failed-send doesn't silence the alert + system for 30 minutes. + """ + + def __init__( + self, + *, + fail_window_seconds: int = DEFAULT_FAIL_WINDOW_SECONDS, + fail_threshold: int = DEFAULT_FAIL_THRESHOLD, + alert_cooldown_seconds: int = DEFAULT_ALERT_COOLDOWN_SECONDS, + clock=time.time, + ) -> None: + self._window = fail_window_seconds + self._threshold = fail_threshold + self._cooldown = alert_cooldown_seconds + self._clock = clock + self._agents: dict[str, _AgentFailures] = defaultdict(_AgentFailures) + self._last_alert_at: float = 0.0 + self._first_failure_at: float = 0.0 # When current outage started + self._alert_count: int = 0 # Alerts fired since last clear + + def record_failure(self, agent_name: str, error: str = "") -> dict: + """Record one auth failure. Returns a decision dict. + + Returned keys: + should_alert (bool): True if this failure crossed the threshold + AND cooldown has elapsed. + reason (str): Short message explaining the decision. + count (int): Current failures-in-window for this agent. + agents_failing (int): Count of distinct agents currently failing. + """ + now = self._clock() + if self._first_failure_at == 0.0: + self._first_failure_at = now + + record = self._agents[agent_name] + count = record.record(now, error, self._window) + agents_failing = sum(1 for r in self._agents.values() if r.timestamps) + + # Multi-agent host outage: if ≥ threshold agents failing simultaneously, + # alert immediately on the very first qualifying failure (don't wait for + # any single agent to hit `threshold` failures alone). + threshold_hit = ( + count >= self._threshold or agents_failing >= self._threshold + ) + + if not threshold_hit: + return { + "should_alert": False, + "reason": "below_threshold", + "count": count, + "agents_failing": agents_failing, + } + + if self._last_alert_at and now - self._last_alert_at < self._cooldown: + return { + "should_alert": False, + "reason": "cooldown", + "count": count, + "agents_failing": agents_failing, + "cooldown_remaining": int( + self._cooldown - (now - self._last_alert_at) + ), + } + + # NOTE: do NOT advance _last_alert_at / _alert_count here. Caller must + # invoke commit_alert() after successful delivery; that way a broker + # failure preserves the cooldown so the alert can retry on the next + # failure instead of being silenced for the cooldown window. + return { + "should_alert": True, + "reason": ( + "host_wide" if agents_failing >= self._threshold else "agent_repeated" + ), + "count": count, + "agents_failing": agents_failing, + } + + def commit_alert(self) -> None: + """Mark an alert as delivered: advance cooldown and bump the counter. + + Call this only after the operator notification was actually sent. If + delivery fails, do NOT call this — the next record_failure() that + crosses threshold will return should_alert=True so the alert retries + instead of being lost for the cooldown window. + """ + self._last_alert_at = self._clock() + self._alert_count += 1 + + def record_success(self, agent_name: str) -> None: + """Clear failure tracking for an agent that successfully called Claude. + + If all agents are clear, reset the outage state so the next outage + emits a fresh "outage started" alert. + """ + if agent_name in self._agents: + self._agents[agent_name].reset() + if not any(r.timestamps for r in self._agents.values()): + self._first_failure_at = 0.0 + # Keep _last_alert_at in place — cooldown still respected even + # across a brief recovery → re-failure to avoid alert flapping. + + def status(self) -> dict: + """Snapshot for /admin/watchdog and diagnostics. + + Status semantics: + "ok" — no agent has failures in window. + "degraded" — some failures but below threshold. + "broken" — at or above threshold (alert is or will fire). + """ + now = self._clock() + agents_failing: list[dict] = [] + for name, rec in self._agents.items(): + # Prune stale timestamps lazily here too so /admin/watchdog + # doesn't show ghosts after the window has elapsed. + cutoff = now - self._window + rec.timestamps = [t for t in rec.timestamps if t >= cutoff] + if not rec.timestamps: + continue + agents_failing.append( + { + "agent": name, + "failures_in_window": len(rec.timestamps), + "last_failure_age_s": int(now - rec.timestamps[-1]), + "last_error": rec.last_error[:120], + } + ) + + if not agents_failing: + status = "ok" + elif any( + a["failures_in_window"] >= self._threshold for a in agents_failing + ) or len(agents_failing) >= self._threshold: + status = "broken" + else: + status = "degraded" + + return { + "status": status, + "fail_window_seconds": self._window, + "fail_threshold": self._threshold, + "alert_cooldown_seconds": self._cooldown, + "outage_started_at": self._first_failure_at or None, + "outage_age_seconds": ( + int(now - self._first_failure_at) + if self._first_failure_at + else None + ), + "alerts_sent": self._alert_count, + "last_alert_age_seconds": ( + int(now - self._last_alert_at) if self._last_alert_at else None + ), + "agents_failing": agents_failing, + } + + +# ── Operator chat resolution ──────────────────────────────────────── + + +def resolve_operator_chat( + *, + get_setting, + list_all_approved_users, +) -> tuple[str, str]: + """Pick the (chat_id, platform) that should receive operator alerts. + + Resolution order: + 1. ``operator_chat_id`` + ``operator_platform`` in system_settings. + 2. Fallback: the chat_id appearing across the most approved_users rows + (heuristic: the person with broadest access is the operator). + 3. Empty strings if nothing matches — caller must handle "no operator". + + Returns ("", "") when unresolved. Never raises. + """ + try: + chat_id = (get_setting("operator_chat_id", "") or "").strip() + platform = (get_setting("operator_platform", "") or "").strip() or "telegram" + if chat_id: + return chat_id, platform + except Exception as e: + _warn("auth_alerts: failed to read operator_chat_id setting: %s", e) + + try: + users: Iterable[dict] = list_all_approved_users() or [] + except Exception as e: + _warn("auth_alerts: failed to list approved users: %s", e) + return "", "" + + counter: Counter[str] = Counter() + for u in users: + cid = (u.get("chat_id") or "").strip() if isinstance(u, dict) else "" + if cid: + counter[cid] += 1 + if not counter: + return "", "" + + chat_id, _ = counter.most_common(1)[0] + return chat_id, "telegram" + + +# ── Alert formatting ─────────────────────────────────────────────── + + +def format_alert_message( + *, + agent_name: str, + decision: dict, + error: str, + host_label: str = "", +) -> str: + """Render the operator-alert message body. + + Plain text — _broker_send defaults to no parse_mode, so anything wrapped + in ``*`` or `` ` `` would render literally on Telegram. Keep it readable + without markdown affordances. + + Kept human-readable and short; includes enough detail for the operator + to know which host to log into and re-auth. + """ + reason = decision.get("reason", "") + agents_failing = decision.get("agents_failing", 0) + count = decision.get("count", 0) + + if reason == "host_wide": + headline = ( + f"🚨 Claude auth broken on {host_label or 'this host'}: " + f"{agents_failing} agent(s) failing" + ) + else: + headline = ( + f"🚨 Claude auth broken for {agent_name}: " + f"{count} failures in window" + ) + + detail = error.strip()[:200] or "no error detail" + instructions = ( + "Re-auth needed: SSH to the host, run 'claude' to log back in, " + "then 'sudo systemctl restart pinkybot' (or equivalent)." + ) + return f"{headline}\n\nLast error: {detail}\n\n{instructions}" diff --git a/src/pinky_daemon/streaming_session.py b/src/pinky_daemon/streaming_session.py index fc6bfabd..4c8ec484 100644 --- a/src/pinky_daemon/streaming_session.py +++ b/src/pinky_daemon/streaming_session.py @@ -112,6 +112,27 @@ def _is_outreach_tool(tool_name: str) -> bool: return _tool_basename(tool_name) in _OUTREACH_TOOL_NAMES +# Auth-failure heuristics. The Claude Agent SDK normalises most credential +# problems to error='authentication_failed'; older SDK versions used +# 'invalid_api_key' / 'permission_error'. Pattern-match instead of a strict +# equality check so a future SDK rename doesn't silently break the alerting. +_AUTH_ERROR_TOKENS = ( + "authentication_failed", + "invalid_api_key", + "unauthorized", + "auth_error", + "permission_error", +) + + +def _is_auth_error(err) -> bool: + """True if the SDK error string looks like a credential failure.""" + if not err: + return False + s = str(err).lower() + return any(token in s for token in _AUTH_ERROR_TOKENS) + + def _describe_tool_use(tool_name: str, tool_input: dict) -> str: """Build a human-readable description of a tool invocation.""" name = _tool_basename(tool_name) @@ -174,6 +195,8 @@ def __init__( cost_callback=None, # fn(agent_name, cost_usd, input_tokens, output_tokens, session_id) analytics_store=None, registry=None, # AgentRegistry — for server-side presence stamping + auth_alert_callback=None, # async fn(agent_name, error_str) — fires on auth_failed + auth_success_callback=None, # fn(agent_name) — fires on a successful turn (clears auth fail state) ) -> None: self._config = config self._response_callback = response_callback @@ -181,6 +204,11 @@ def __init__( self._conversation_store = conversation_store self._analytics_store = analytics_store self._registry = registry + # Auth-failure detection plumbing — called from the reader loop when + # the SDK reports an authentication error so the daemon can alert + # the operator. See pinky_daemon/auth_alerts.py for the tracker. + self._auth_alert_callback = auth_alert_callback + self._auth_success_callback = auth_success_callback self._client = None self._reader_task: asyncio.Task | None = None self._connected = False @@ -439,6 +467,20 @@ async def _reader_loop(self) -> None: f"streaming[{self.agent_name}]: assistant error={msg.error!r}" f" stop_reason={msg.stop_reason!r} — suppressing content" ) + # Detect auth failures and notify the operator. The SDK + # surfaces these as msg.error == 'authentication_failed' + # (and historically 'invalid_api_key' on some versions); + # match defensively so any future variants still alert. + if _is_auth_error(msg.error) and self._auth_alert_callback: + try: + await self._auth_alert_callback( + self.agent_name, str(msg.error) + ) + except Exception as exc: + _log( + f"streaming[{self.agent_name}]: " + f"auth_alert_callback raised: {exc}" + ) # Don't touch _last_response; fall through to usage/session_id capture. else: # Extract text and tool uses from content blocks @@ -596,6 +638,15 @@ async def _reader_loop(self) -> None: except Exception as e: _log(f"streaming[{self.agent_name}]: callback error: {e}") + # A successful (non-errored) turn proves Claude auth is + # working again — clear any auth-fail tracking for this + # agent so the next outage emits a fresh alert. + if self._auth_success_callback: + try: + self._auth_success_callback(self.agent_name) + except Exception: + pass + # Track usage from result if msg.total_cost_usd: self.usage.total_cost_usd += msg.total_cost_usd diff --git a/tests/test_auth_alerts.py b/tests/test_auth_alerts.py new file mode 100644 index 00000000..36b9695d --- /dev/null +++ b/tests/test_auth_alerts.py @@ -0,0 +1,328 @@ +"""Tests for the auth-failure tracker and operator-alert wiring.""" +from __future__ import annotations + +from pinky_daemon.auth_alerts import ( + AuthFailureTracker, + format_alert_message, + resolve_operator_chat, +) +from pinky_daemon.streaming_session import _is_auth_error + +# ── _is_auth_error ──────────────────────────────────────────────── + + +def test_is_auth_error_matches_known_tokens(): + assert _is_auth_error("authentication_failed") is True + assert _is_auth_error("invalid_api_key") is True + assert _is_auth_error("Unauthorized") is True # case-insensitive + assert _is_auth_error("auth_error: token expired") is True + assert _is_auth_error("permission_error") is True + + +def test_is_auth_error_skips_other_failures(): + assert _is_auth_error("rate_limit_exceeded") is False + assert _is_auth_error("content_filter") is False + assert _is_auth_error("") is False + assert _is_auth_error(None) is False + + +# ── AuthFailureTracker ──────────────────────────────────────────── + + +class _FakeClock: + def __init__(self, start: float = 1_000_000.0) -> None: + self.now = start + + def __call__(self) -> float: + return self.now + + def tick(self, seconds: float) -> None: + self.now += seconds + + +def _make_tracker(**kw): + clock = _FakeClock() + tracker = AuthFailureTracker( + fail_window_seconds=kw.get("window", 300), + fail_threshold=kw.get("threshold", 3), + alert_cooldown_seconds=kw.get("cooldown", 1800), + clock=clock, + ) + return tracker, clock + + +def test_below_threshold_does_not_alert(): + tracker, _ = _make_tracker() + d = tracker.record_failure("sasha", "authentication_failed") + assert d["should_alert"] is False + assert d["count"] == 1 + assert d["reason"] == "below_threshold" + + +def test_threshold_for_single_agent_fires_alert(): + tracker, clock = _make_tracker(threshold=3) + for _ in range(2): + clock.tick(10) + d = tracker.record_failure("sasha", "authentication_failed") + assert d["should_alert"] is False + clock.tick(10) + d = tracker.record_failure("sasha", "authentication_failed") + assert d["should_alert"] is True + assert d["reason"] == "agent_repeated" + assert d["count"] == 3 + # Decision alone must not commit cooldown — caller hasn't delivered yet. + assert tracker._last_alert_at == 0.0 + assert tracker._alert_count == 0 + + +def test_host_wide_outage_fires_immediately_at_threshold_agents(): + """If 3 different agents fail, alert on the 3rd — don't wait for any + single agent to hit the per-agent threshold.""" + tracker, _ = _make_tracker(threshold=3) + assert tracker.record_failure("sasha", "auth")["should_alert"] is False + assert tracker.record_failure("ivan", "auth")["should_alert"] is False + d = tracker.record_failure("kuzya", "auth") + assert d["should_alert"] is True + assert d["reason"] == "host_wide" + assert d["agents_failing"] == 3 + + +def test_cooldown_blocks_repeat_alerts(): + tracker, clock = _make_tracker(threshold=3, cooldown=600) + # Drive past threshold once and simulate successful delivery. + for _ in range(3): + tracker.record_failure("sasha", "auth") + tracker.commit_alert() + # Next failure within cooldown — must not realert. + clock.tick(60) + d = tracker.record_failure("sasha", "auth") + assert d["should_alert"] is False + assert d["reason"] == "cooldown" + assert d["cooldown_remaining"] > 0 + + +def test_cooldown_elapses_then_alert_fires_again(): + """During a continuous outage we want a re-alert each cooldown period. + + Use a window long enough that failures stay live across the cooldown gap + — that's the realistic outage shape (agent retries every minute or so). + """ + tracker, clock = _make_tracker(threshold=3, window=3600, cooldown=600) + for _ in range(3): + tracker.record_failure("sasha", "auth") # first alert decision here + tracker.commit_alert() # simulate first delivery succeeded + clock.tick(601) # past cooldown, still inside window + d = tracker.record_failure("sasha", "auth") + assert d["should_alert"] is True + assert d["count"] >= 3 + + +def test_failed_delivery_preserves_cooldown_and_retries_next_failure(): + """Regression: a broker_send raise must NOT advance the cooldown. + + Pushok review on PR #400: the original code mutated _last_alert_at + *before* the await, so a network blip on the very first send-attempt + silenced the alert system for 30 minutes — the exact regression this + feature exists to prevent (Sasha-down-9-12h). + """ + tracker, clock = _make_tracker(threshold=3, cooldown=600) + # First three failures cross threshold; caller "tries to deliver" but + # the broker raises, so commit_alert() is NEVER called. + for _ in range(3): + d = tracker.record_failure("sasha", "auth") + assert d["should_alert"] is True + # No commit_alert() — simulating broker_send failure. + assert tracker._last_alert_at == 0.0 # cooldown untouched + + # Next failure (1 second later) must still be should_alert: True so the + # caller can retry delivery. + clock.tick(1) + d = tracker.record_failure("sasha", "auth") + assert d["should_alert"] is True + assert d["reason"] == "agent_repeated" + + +def test_commit_alert_advances_cooldown_and_counter(): + tracker, clock = _make_tracker(threshold=3, cooldown=600) + for _ in range(3): + tracker.record_failure("sasha", "auth") + assert tracker._alert_count == 0 + tracker.commit_alert() + assert tracker._alert_count == 1 + assert tracker._last_alert_at == clock.now + + +def test_window_eviction_drops_old_failures(): + tracker, clock = _make_tracker(window=60, threshold=3) + tracker.record_failure("sasha", "auth") + clock.tick(61) # first entry now stale + tracker.record_failure("sasha", "auth") + clock.tick(1) + d = tracker.record_failure("sasha", "auth") + # Only 2 fresh entries within window — below threshold. + assert d["should_alert"] is False + assert d["count"] == 2 + + +def test_record_success_clears_agent_failures(): + tracker, _ = _make_tracker(threshold=3) + tracker.record_failure("sasha", "auth") + tracker.record_failure("sasha", "auth") + tracker.record_success("sasha") + d = tracker.record_failure("sasha", "auth") + assert d["count"] == 1 + assert d["should_alert"] is False + + +def test_status_reports_ok_when_no_failures(): + tracker, _ = _make_tracker() + s = tracker.status() + assert s["status"] == "ok" + assert s["agents_failing"] == [] + assert s["alerts_sent"] == 0 + + +def test_status_reports_broken_at_or_above_threshold(): + tracker, _ = _make_tracker(threshold=3) + for _ in range(3): + tracker.record_failure("sasha", "auth") + tracker.commit_alert() # simulate delivery succeeded + s = tracker.status() + assert s["status"] == "broken" + assert s["alerts_sent"] == 1 + assert any(a["agent"] == "sasha" for a in s["agents_failing"]) + + +def test_status_reports_degraded_below_threshold(): + tracker, _ = _make_tracker(threshold=3) + tracker.record_failure("sasha", "auth") + s = tracker.status() + assert s["status"] == "degraded" + + +# ── resolve_operator_chat ───────────────────────────────────────── + + +def test_resolve_operator_chat_prefers_setting(): + settings = {"operator_chat_id": "12345", "operator_platform": "telegram"} + chat_id, platform = resolve_operator_chat( + get_setting=lambda k, default="": settings.get(k, default), + list_all_approved_users=lambda: [], + ) + assert chat_id == "12345" + assert platform == "telegram" + + +def test_resolve_operator_chat_falls_back_to_most_common_approved_user(): + users = [ + {"chat_id": "111", "agent_name": "a"}, + {"chat_id": "222", "agent_name": "a"}, + {"chat_id": "111", "agent_name": "b"}, + {"chat_id": "111", "agent_name": "c"}, + ] + chat_id, platform = resolve_operator_chat( + get_setting=lambda k, default="": "", + list_all_approved_users=lambda: users, + ) + assert chat_id == "111" # appears 3× → operator + assert platform == "telegram" + + +def test_resolve_operator_chat_returns_empty_when_nothing_known(): + chat_id, platform = resolve_operator_chat( + get_setting=lambda k, default="": "", + list_all_approved_users=lambda: [], + ) + assert chat_id == "" + assert platform == "" + + +def test_resolve_operator_chat_tolerates_setting_errors(): + def bad_setting(*_a, **_kw): + raise RuntimeError("db down") + + chat_id, platform = resolve_operator_chat( + get_setting=bad_setting, + list_all_approved_users=lambda: [{"chat_id": "999"}], + ) + assert chat_id == "999" # fell through to fallback + assert platform == "telegram" + + +# ── format_alert_message ────────────────────────────────────────── + + +def test_format_alert_includes_agent_when_repeated(): + msg = format_alert_message( + agent_name="sasha", + decision={"reason": "agent_repeated", "count": 3, "agents_failing": 1}, + error="authentication_failed", + host_label="rpi5", + ) + assert "sasha" in msg + assert "authentication_failed" in msg + assert "Re-auth" in msg + + +def test_format_alert_says_host_wide_when_multiple_agents_fail(): + msg = format_alert_message( + agent_name="sasha", + decision={"reason": "host_wide", "count": 1, "agents_failing": 3}, + error="invalid_api_key", + host_label="rpi5", + ) + assert "rpi5" in msg + assert "3 agent" in msg + + +def test_format_alert_truncates_long_error_strings(): + long = "x" * 5000 + msg = format_alert_message( + agent_name="sasha", + decision={"reason": "agent_repeated", "count": 3, "agents_failing": 1}, + error=long, + host_label="", + ) + # Last error block is capped at 200 chars. + assert "x" * 5000 not in msg + + +def test_format_alert_uses_no_markdown_so_plain_text_renders_clean(): + """Pushok review: _broker_send defaults to no parse_mode, so any markdown + in the message renders literally on Telegram (operator sees `*sasha*`). + Keep the message plain text. + """ + msg = format_alert_message( + agent_name="sasha", + decision={"reason": "agent_repeated", "count": 3, "agents_failing": 1}, + error="authentication_failed", + host_label="rpi5", + ) + assert "*" not in msg + assert "`" not in msg + # Sanity: still actually informative. + assert "sasha" in msg + assert "authentication_failed" in msg + + +# ── /admin/watchdog integration ─────────────────────────────────── + + +def test_admin_watchdog_exposes_auth_status(tmp_path): + """End-to-end: /admin/watchdog includes auth_status with the right shape.""" + from fastapi.testclient import TestClient + + from pinky_daemon.api import create_api + + api = create_api(db_path=str(tmp_path / "memory.db")) + with TestClient(api) as client: + resp = client.get("/admin/watchdog") + assert resp.status_code == 200 + body = resp.json() + assert "auth_status" in body + auth = body["auth_status"] + assert auth["status"] == "ok" # no failures yet + assert auth["fail_threshold"] >= 1 + assert auth["agents_failing"] == [] + assert "fail_window_seconds" in auth + assert "alert_cooldown_seconds" in auth From 8596b9216fcde3b94cc742702f83c5ca86827c03 Mon Sep 17 00:00:00 2001 From: olegbrok Date: Fri, 8 May 2026 08:36:34 -0700 Subject: [PATCH 2/3] =?UTF-8?q?chore(deps):=20bump=20claude-agent-sdk=20fl?= =?UTF-8?q?oor=200.1.73=20=E2=86=92=200.1.76=20(#402)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Picks up SDK 0.1.74-0.1.76 fixes on the floor pin: hook event streaming + deferred decisions, strict MCP config validation, eager session-store flushing, permission-suggestions deserialization fix, and bundled CC 2.1.132 (MCP reconnect-flood fix, subprocess-cleanup-on-parent-exit, CLAUDE_CODE_SESSION_ID env in subprocesses). Closes the floor-bump action item from #401. The native API-error-status investigation, subprocess-cleanup verification, and PR #400 auth-alert re-validation remain as separate follow-ups in #401. Verification: - pip install -e . resolves to claude-agent-sdk 0.1.76 (bundled CC 2.1.132) - pytest: 1770 passed, 1 skipped (parity with pre-bump) Co-authored-by: Oleg Co-authored-by: Claude Opus 4.7 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 48b7b6e5..ccba046d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "httpx>=0.28", "fastapi>=0.135", "uvicorn>=0.44", - "claude-agent-sdk>=0.1.73", + "claude-agent-sdk>=0.1.76", "Pillow>=10.0", # Federation crypto (sealed_box_v1): X25519, Ed25519, HKDF, XChaCha20-Poly1305. "cryptography>=41.0", From 95c89e7cee98ae597e2bba2a7baf6fe5fb11a2fe Mon Sep 17 00:00:00 2001 From: olegbrok Date: Fri, 8 May 2026 09:04:19 -0700 Subject: [PATCH 3/3] refactor(auth-detect): structured AssistantMessage.error + ResultMessage.api_error_status (#403) (#404) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(auth-detect): structured AssistantMessage.error + ResultMessage.api_error_status (#403) Replaces the legacy `_AUTH_ERROR_TOKENS` substring tuple in streaming_session.py with structured detection on both message paths, and bumps `claude-agent-sdk` floor 0.1.76 → 0.1.77. ## What changed **streaming_session.py** - Drop `_AUTH_ERROR_TOKENS` substring tuple and `_is_auth_error(str)`. - Add `_is_auth_error_assistant(msg)`: exact match against the `AssistantMessageError` Literal (`"authentication_failed"`). Other Literal values (billing_error, rate_limit, invalid_request, server_error, unknown) are NOT credential failures and must NOT trip the operator alert. - Add `_is_auth_error_result(result)`: detects `api_error_status` in {401, 403} on `ResultMessage` (added in SDK 0.1.76, emitted by CLI >= 2.1.110). Closes the gap where errors that only land at turn completion (not mid-turn on AssistantMessage) were silently swallowed by the existing error-result path. - Update both reader-loop callsites: AssistantMessage path now uses the assistant variant; ResultMessage `is_error` path now also fires the auth alert callback when status is 401/403. **api.py** - Update the `_on_auth_failure` docstring: the comment used to point at `auth_alerts._is_auth_error` (which never existed there); now references both the assistant and result detectors in `streaming_session`. **tests/test_auth_alerts.py** - Replace `_is_auth_error(str)` tests with structured tests for both detectors using `SimpleNamespace` stand-ins for the SDK message types. Coverage: - assistant: only `"authentication_failed"` is true; all 5 other Literal values are false; `None` and missing-attr are false. - result: 401/403 are true; 429 + 5xx are false; `None`/200/missing are false. **pyproject.toml** - `claude-agent-sdk` floor 0.1.76 → 0.1.77. Required for the `api_error_status` field on `ResultMessage` and gets the bundled Claude CLI 2.1.133 (subagent skill discovery fix, parallel-session refresh-token race fix, hook effort context). ## Why this matters Pre-refactor, the substring-tuple matcher carried entries (`invalid_api_key`, `unauthorized`, `auth_error`, `permission_error`) that the SDK Literal cannot produce — pure cargo from before the type became strict. Worse, the ResultMessage error path had no auth detection at all, so credential failures that landed only at turn completion were silently swallowed. After this refactor, both paths detect credential failures with exact-match semantics over typed values, the operator-alert wiring fires from either path, and we can no longer be silently broken by an AssistantMessageError variant rename — a mismatch becomes a test failure instead of a silent regression. ## Test plan - [x] `pytest tests/test_auth_alerts.py -x` — 27 passed - [x] Full suite: `pytest tests/ -x` — 1774 passed, 1 skipped - [x] `ruff check` on modified files — clean Refs: #401, #402, #403 🤖 Opened by Barsik * fix(auth-detect): per-turn dedupe + SDK Literal invariant guards Address Murzik's PR #404 review (P1-ish double-count) and Pushok's forward-looking concern about silent regression on SDK rename. ## Per-turn auth dedupe (Murzik) A single failed turn can surface auth errors on BOTH paths the reader_loop watches: - AssistantMessage with error="authentication_failed" (mid-turn) - ResultMessage with api_error_status=401/403 (turn completion) Without dedupe, AuthFailureTracker.record_failure() increments twice for one real failure — tripping the operator-alert threshold early (after ~2 actual failed turns instead of 3) and skewing the host-wide multi-agent baseline. Fix: a per-turn `auth_reported_this_turn` flag in the reader_loop. The AssistantMessage auth path sets it before invoking the callback (so a callback exception still dedupes). The ResultMessage auth path gates its callback on `not auth_reported_this_turn`. The flag resets at the end of each ResultMessage handling — turn boundary. ## SDK Literal invariant (Pushok) `_is_auth_error_assistant` does exact-match against `AssistantMessageError`. If a future SDK release renames the Literal value, exact-match silently stops detecting credential failures — re-creating the exact regression #400 was built to catch. Two-layer guard: - Import-time assertion in reader_loop checks `_AUTH_ASSISTANT_ERROR in AssistantMessageError.__args__`. Fails loud at session start on any local-dev SDK upgrade that drops the literal. - Unit test in test_auth_alerts.py asserts the same invariant. Fails loud at CI on any PR that bumps the SDK across a rename. ## Other changes - Enrich the ResultMessage auth-callback string with `msg.errors` when present — operators see actionable triage context in their DM, not just the raw status code (SDK 0.1.77 started returning meaningful error messages here). - Bump uv.lock claude-agent-sdk 0.1.73 → 0.1.77 to match pyproject.toml's floor (lock file was stale from PR #402). ## Tests - 4 new reader-loop regression tests in test_streaming_session.py: - both paths emit for one turn → callback fires once - result-only path still fires (dedupe doesn't break the gap-fill) - dedupe resets between turns (two real failures = two callbacks) - non-auth AssistantMessage error doesn't poison next-turn dedupe - 1 new SDK-Literal-invariant unit test in test_auth_alerts.py. - test_api.py fake_types stub gains AssistantMessageError so the import-time invariant check passes under that test's monkey-patch. Full suite: 1779 passed / 1 skipped, ruff clean. Co-Authored-By: Claude Opus 4.7 --------- Co-authored-by: Oleg Co-authored-by: Claude Opus 4.7 --- pyproject.toml | 2 +- src/pinky_daemon/api.py | 11 +- src/pinky_daemon/streaming_session.py | 128 +++++++++++++++---- tests/test_api.py | 11 ++ tests/test_auth_alerts.py | 104 +++++++++++++-- tests/test_streaming_session.py | 174 ++++++++++++++++++++++++++ uv.lock | 16 +-- 7 files changed, 397 insertions(+), 49 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ccba046d..84a7a5bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,7 +32,7 @@ dependencies = [ "httpx>=0.28", "fastapi>=0.135", "uvicorn>=0.44", - "claude-agent-sdk>=0.1.76", + "claude-agent-sdk>=0.1.77", "Pillow>=10.0", # Federation crypto (sealed_box_v1): X25519, Ed25519, HKDF, XChaCha20-Poly1305. "cryptography>=41.0", diff --git a/src/pinky_daemon/api.py b/src/pinky_daemon/api.py index 880be6e7..157c391c 100644 --- a/src/pinky_daemon/api.py +++ b/src/pinky_daemon/api.py @@ -1586,10 +1586,13 @@ async def _on_auth_failure(agent_name: str, error: str) -> None: """Streaming-session hook: record an auth failure and alert the operator. Called from StreamingSession's reader loop whenever the SDK reports - an authentication-class error (see auth_alerts._is_auth_error). The - AuthFailureTracker decides whether this failure crosses the alert - threshold and whether the cooldown has elapsed; if both, we DM the - operator via _broker_send using the affected agent's own bot. + an authentication-class error — either AssistantMessage.error == + "authentication_failed" (see streaming_session._is_auth_error_assistant) + or ResultMessage.api_error_status in {401, 403} (see + streaming_session._is_auth_error_result). The AuthFailureTracker + decides whether this failure crosses the alert threshold and whether + the cooldown has elapsed; if both, we DM the operator via + _broker_send using the affected agent's own bot. """ try: decision = auth_tracker.record_failure(agent_name, error) diff --git a/src/pinky_daemon/streaming_session.py b/src/pinky_daemon/streaming_session.py index 4c8ec484..e0d238ba 100644 --- a/src/pinky_daemon/streaming_session.py +++ b/src/pinky_daemon/streaming_session.py @@ -112,25 +112,38 @@ def _is_outreach_tool(tool_name: str) -> bool: return _tool_basename(tool_name) in _OUTREACH_TOOL_NAMES -# Auth-failure heuristics. The Claude Agent SDK normalises most credential -# problems to error='authentication_failed'; older SDK versions used -# 'invalid_api_key' / 'permission_error'. Pattern-match instead of a strict -# equality check so a future SDK rename doesn't silently break the alerting. -_AUTH_ERROR_TOKENS = ( - "authentication_failed", - "invalid_api_key", - "unauthorized", - "auth_error", - "permission_error", -) - - -def _is_auth_error(err) -> bool: - """True if the SDK error string looks like a credential failure.""" - if not err: - return False - s = str(err).lower() - return any(token in s for token in _AUTH_ERROR_TOKENS) +# Auth-failure detection — uses native SDK types (claude-agent-sdk >= 0.1.76). +# +# Two paths surface credential failures, and we check both: +# +# 1. ``AssistantMessage.error`` is an ``AssistantMessageError`` Literal whose +# only credential-failure value is "authentication_failed". The other +# Literal members (billing_error, rate_limit, invalid_request, +# server_error, unknown) are NOT auth issues and must NOT trip the +# operator alert — billing errors and rate limits in particular would +# spam the operator on perfectly authenticated sessions. +# +# 2. ``ResultMessage.api_error_status`` is the raw HTTP status of a failing +# API call (added in SDK 0.1.76, emitted by CLI >= 2.1.110). 401 and 403 +# are credential failures; 429 and 5xx are transient/operational and +# handled separately. +# +# This replaces an older substring-tuple match against ``msg.error`` that +# pre-dated the Literal type and over-matched (e.g. "permission_error" is +# never a value the SDK can produce). +_AUTH_ASSISTANT_ERROR = "authentication_failed" +_AUTH_HTTP_STATUSES = frozenset({401, 403}) + + +def _is_auth_error_assistant(msg) -> bool: + """True if an ``AssistantMessage`` indicates a credential failure.""" + return getattr(msg, "error", None) == _AUTH_ASSISTANT_ERROR + + +def _is_auth_error_result(result) -> bool: + """True if a ``ResultMessage``'s HTTP status indicates a credential failure.""" + status = getattr(result, "api_error_status", None) + return status in _AUTH_HTTP_STATUSES def _describe_tool_use(tool_name: str, tool_input: dict) -> str: @@ -440,6 +453,7 @@ async def _reader_loop(self) -> None: """Background loop that reads responses and fires callbacks.""" from claude_agent_sdk.types import ( AssistantMessage, + AssistantMessageError, ResultMessage, TextBlock, ThinkingBlock, @@ -447,9 +461,30 @@ async def _reader_loop(self) -> None: ToolUseBlock, ) + # Defensive invariant: ``_is_auth_error_assistant`` does an exact + # match against ``AssistantMessageError`` for "authentication_failed". + # If a future SDK release renames that Literal value, exact-match + # silently stops detecting credential failures — re-creating the + # exact regression mode #400 was built to catch. Fail loud at session + # start instead. (CI also asserts this in test_auth_alerts.py, so + # bumps caught in PR; this guard catches local-dev SDK upgrades.) + assert _AUTH_ASSISTANT_ERROR in AssistantMessageError.__args__, ( + f"claude-agent-sdk renamed AssistantMessageError Literal — " + f"_AUTH_ASSISTANT_ERROR={_AUTH_ASSISTANT_ERROR!r} no longer in " + f"{AssistantMessageError.__args__}. Update the constant and tests." + ) + _log(f"streaming[{self.agent_name}]: reader loop running") turn_tool_uses = [] # Track tool uses per turn turn_thinking: list[str] = [] # Track thinking blocks per turn + # Per-turn dedupe for auth-failure callbacks. A single failed turn can + # surface auth errors on BOTH paths: an AssistantMessage with + # error="authentication_failed", followed by the terminal ResultMessage + # with api_error_status=401. Without dedupe, AuthFailureTracker would + # increment twice for one real failure — tripping the operator-alert + # threshold early and skewing the multi-agent baseline. Reset at the + # end of ResultMessage handling (turn boundary). + auth_reported_this_turn = False try: async for msg in self._client.receive_messages(): @@ -468,10 +503,17 @@ async def _reader_loop(self) -> None: f" stop_reason={msg.stop_reason!r} — suppressing content" ) # Detect auth failures and notify the operator. The SDK - # surfaces these as msg.error == 'authentication_failed' - # (and historically 'invalid_api_key' on some versions); - # match defensively so any future variants still alert. - if _is_auth_error(msg.error) and self._auth_alert_callback: + # types ``msg.error`` as the AssistantMessageError + # Literal; only "authentication_failed" is a credential + # issue. Other Literal values (billing_error, rate_limit, + # invalid_request, server_error, unknown) are NOT auth + # failures and must not trip the operator alert. + if _is_auth_error_assistant(msg) and self._auth_alert_callback: + # Set the dedupe flag BEFORE invoking the callback + # so a callback exception can't cause the + # ResultMessage path to double-fire for the same + # turn. + auth_reported_this_turn = True try: await self._auth_alert_callback( self.agent_name, str(msg.error) @@ -576,8 +618,42 @@ async def _reader_loop(self) -> None: _log( f"streaming[{self.agent_name}]: error result" f" stop_reason={msg.stop_reason!r}" + f" api_error_status={getattr(msg, 'api_error_status', None)!r}" f" errors={msg.errors!r} — suppressing forwarded response" ) + # Detect credential failures on the result path too. The + # AssistantMessage path catches errors the SDK surfaces + # mid-turn; this path catches errors that only land at + # turn completion (api_error_status added in 0.1.76: + # 401/403 = bad creds; 429/5xx = transient, handled + # below as a generic error result). + # + # Skip if the AssistantMessage path already reported + # auth for this turn — both paths firing for one real + # failure double-counts in AuthFailureTracker. + if ( + _is_auth_error_result(msg) + and self._auth_alert_callback + and not auth_reported_this_turn + ): + auth_reported_this_turn = True + # Surface msg.errors into the callback string when + # present — operators get richer triage context + # than the raw status code alone (the SDK started + # returning actionable messages in 0.1.77). + err_detail = f"api_error_status={msg.api_error_status}" + if msg.errors: + err_detail += f" errors={msg.errors!r}" + try: + await self._auth_alert_callback( + self.agent_name, + err_detail, + ) + except Exception as exc: + _log( + f"streaming[{self.agent_name}]: " + f"auth_alert_callback (result) raised: {exc}" + ) # Analytics: still record errored turns _u = msg.usage or {} self._analytics_log_turn_usage( @@ -612,6 +688,8 @@ async def _reader_loop(self) -> None: turn_thinking = [] self._stats["turns"] += 1 self.last_active = time.time() + # Reset per-turn auth dedupe — turn boundary + auth_reported_this_turn = False continue # Turn complete — fire response callback @@ -763,6 +841,10 @@ async def _reader_loop(self) -> None: turn_thinking = [] # Reset for next turn self._stats["turns"] += 1 self.last_active = time.time() + # Reset per-turn auth dedupe — turn boundary. Successful + # turns rarely set this flag (no auth error fired), but + # reset unconditionally to keep the invariant simple. + auth_reported_this_turn = False _log(f"streaming[{self.agent_name}]: turn complete (total: {self._stats['turns']})") diff --git a/tests/test_api.py b/tests/test_api.py index 0661a4b2..31463990 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -233,6 +233,17 @@ def __init__(self, thinking=""): fake_types.ToolResultBlock = ToolResultBlock fake_types.AssistantMessage = AssistantMessage fake_types.ResultMessage = ResultMessage + # Stub for the SDK Literal — reader_loop's import-time invariant + # check (PR #404) reads __args__ to defend against SDK rename. + from typing import Literal as _Literal + fake_types.AssistantMessageError = _Literal[ + "authentication_failed", + "billing_error", + "rate_limit", + "invalid_request", + "server_error", + "unknown", + ] old_sdk_types = sys.modules.get("claude_agent_sdk.types") sys.modules["claude_agent_sdk.types"] = fake_types diff --git a/tests/test_auth_alerts.py b/tests/test_auth_alerts.py index 36b9695d..4f999cc7 100644 --- a/tests/test_auth_alerts.py +++ b/tests/test_auth_alerts.py @@ -1,29 +1,107 @@ """Tests for the auth-failure tracker and operator-alert wiring.""" from __future__ import annotations +from types import SimpleNamespace + from pinky_daemon.auth_alerts import ( AuthFailureTracker, format_alert_message, resolve_operator_chat, ) -from pinky_daemon.streaming_session import _is_auth_error +from pinky_daemon.streaming_session import ( + _AUTH_ASSISTANT_ERROR, + _is_auth_error_assistant, + _is_auth_error_result, +) + +# ── SDK Literal invariant ──────────────────────────────────────── +# +# ``_is_auth_error_assistant`` does an exact string match against +# AssistantMessageError. If a future SDK rename of the Literal value +# happens unnoticed, exact-match silently stops detecting credential +# failures — re-creating the exact regression mode #400 was built to +# catch. This test fails loud at CI time on any SDK bump that drops +# or renames "authentication_failed". + + +def test_auth_assistant_error_literal_present_in_sdk_type(): + """``_AUTH_ASSISTANT_ERROR`` must remain a member of + ``AssistantMessageError.__args__``. If the SDK renames the Literal, + update both the constant and ``_is_auth_error_assistant``.""" + from claude_agent_sdk.types import AssistantMessageError + + assert _AUTH_ASSISTANT_ERROR in AssistantMessageError.__args__, ( + f"claude-agent-sdk renamed the AssistantMessageError Literal — " + f"{_AUTH_ASSISTANT_ERROR!r} no longer in {AssistantMessageError.__args__}. " + f"Update _AUTH_ASSISTANT_ERROR in streaming_session.py." + ) + +# ── _is_auth_error_assistant ────────────────────────────────────── +# +# AssistantMessage.error is the AssistantMessageError Literal: +# authentication_failed, billing_error, rate_limit, invalid_request, +# server_error, unknown +# Only "authentication_failed" should trip the operator alert. + + +def _msg(error=None): + """Lightweight AssistantMessage stand-in (only the .error attr matters).""" + return SimpleNamespace(error=error) + + +def test_is_auth_error_assistant_true_only_for_authentication_failed(): + assert _is_auth_error_assistant(_msg("authentication_failed")) is True + + +def test_is_auth_error_assistant_false_for_other_literal_values(): + # Other AssistantMessageError values are real errors but NOT credential + # failures. Alerting the operator on these would be noise (or actively + # wrong, in the case of billing/rate_limit issues that have nothing to + # do with credentials). + for not_auth in ( + "billing_error", + "rate_limit", + "invalid_request", + "server_error", + "unknown", + ): + assert _is_auth_error_assistant(_msg(not_auth)) is False, not_auth + + +def test_is_auth_error_assistant_false_for_none_and_missing(): + assert _is_auth_error_assistant(_msg(None)) is False + # Missing attribute (e.g. unrelated message type) must not raise. + assert _is_auth_error_assistant(SimpleNamespace()) is False + + +# ── _is_auth_error_result ───────────────────────────────────────── +# +# ResultMessage.api_error_status is int|None — raw HTTP status from a +# failing API call. 401/403 are credential failures; 429 and 5xx are +# transient/operational and must not trip the auth alert. + + +def _result(api_error_status=None): + """Lightweight ResultMessage stand-in.""" + return SimpleNamespace(api_error_status=api_error_status) + -# ── _is_auth_error ──────────────────────────────────────────────── +def test_is_auth_error_result_true_for_401_and_403(): + assert _is_auth_error_result(_result(401)) is True + assert _is_auth_error_result(_result(403)) is True -def test_is_auth_error_matches_known_tokens(): - assert _is_auth_error("authentication_failed") is True - assert _is_auth_error("invalid_api_key") is True - assert _is_auth_error("Unauthorized") is True # case-insensitive - assert _is_auth_error("auth_error: token expired") is True - assert _is_auth_error("permission_error") is True +def test_is_auth_error_result_false_for_transient_statuses(): + # Rate limit (429) and any 5xx must NOT alert as a credential failure. + for not_auth_status in (429, 500, 502, 503, 504, 529): + assert _is_auth_error_result(_result(not_auth_status)) is False, not_auth_status -def test_is_auth_error_skips_other_failures(): - assert _is_auth_error("rate_limit_exceeded") is False - assert _is_auth_error("content_filter") is False - assert _is_auth_error("") is False - assert _is_auth_error(None) is False +def test_is_auth_error_result_false_for_none_and_success(): + assert _is_auth_error_result(_result(None)) is False + assert _is_auth_error_result(_result(200)) is False + # Missing attribute (older SDK / non-result objects) must not raise. + assert _is_auth_error_result(SimpleNamespace()) is False # ── AuthFailureTracker ──────────────────────────────────────────── diff --git a/tests/test_streaming_session.py b/tests/test_streaming_session.py index a1f187a6..def9bed7 100644 --- a/tests/test_streaming_session.py +++ b/tests/test_streaming_session.py @@ -202,3 +202,177 @@ async def test_idle_sleep_returns_false_when_already_disconnected() -> None: assert result is False assert ss.is_idle_sleeping is False + + +# -- Auth-failure dedupe across AssistantMessage + ResultMessage paths -------- +# +# A single failed turn can surface auth errors on BOTH paths the reader_loop +# watches: an AssistantMessage with error="authentication_failed", followed +# by the terminal ResultMessage with api_error_status=401. Without dedupe, +# AuthFailureTracker.record_failure() would increment twice for one real +# failure — tripping the operator-alert threshold early and skewing the +# host-wide multi-agent baseline. Reader_loop must collapse the two-path +# emission into a single auth-alert-callback invocation per turn. +# +# Per Murzik's PR #404 review: "I verified locally with a synthetic reader +# stream: callback fired as [('agent', 'authentication_failed'), ('agent', +# 'api_error_status=401')]." This test pins down the fix. + + +def _make_assistant_message(*, error: str | None = None): + """Build an AssistantMessage with the minimum fields the reader_loop reads.""" + from claude_agent_sdk.types import AssistantMessage + + return AssistantMessage( + content=[], + model="claude-opus-4-7", + error=error, + usage={"input_tokens": 0, "output_tokens": 0}, + stop_reason="error" if error else "end_turn", + ) + + +def _make_result_message( + *, + is_error: bool = False, + api_error_status: int | None = None, + errors: list[str] | None = None, +): + """Build a ResultMessage with the minimum fields the reader_loop reads.""" + from claude_agent_sdk.types import ResultMessage + + return ResultMessage( + subtype="success" if not is_error else "error", + duration_ms=10, + duration_api_ms=5, + is_error=is_error, + num_turns=1, + session_id="test-session", + stop_reason="end_turn" if not is_error else "error", + api_error_status=api_error_status, + errors=errors, + ) + + +async def _run_reader_against_stream(ss: StreamingSession, messages: list) -> None: + """Wire up a fake client whose receive_messages() yields the given list, + then run the reader_loop until the iterator drains.""" + + async def _receive_messages(): + for msg in messages: + yield msg + + ss._client.receive_messages = _receive_messages + await ss._reader_loop() + + +@pytest.mark.asyncio +async def test_auth_callback_fires_once_when_both_paths_signal_for_same_turn() -> None: + """Regression for PR #404 / Murzik's review. + + SDK emits AssistantMessage(error="authentication_failed") followed by + ResultMessage(is_error=True, api_error_status=401) for a single failed + turn. The auth-alert callback must fire exactly once — not once per path.""" + ss = _make_session() + callback = AsyncMock() + ss._auth_alert_callback = callback + + await _run_reader_against_stream( + ss, + [ + _make_assistant_message(error="authentication_failed"), + _make_result_message( + is_error=True, + api_error_status=401, + errors=["Invalid API key"], + ), + ], + ) + + assert callback.await_count == 1, ( + f"Expected one callback per failed turn (dedupe across " + f"AssistantMessage + ResultMessage paths); got {callback.await_count}: " + f"{callback.await_args_list}" + ) + # The AssistantMessage path should win (it fires first in the stream). + args, _ = callback.await_args + assert args == (ss.agent_name, "authentication_failed") + + +@pytest.mark.asyncio +async def test_auth_callback_fires_on_result_path_when_assistant_path_silent() -> None: + """If the failed turn surfaces only at the ResultMessage (no + AssistantMessage error mid-turn), the result-path detection must still + fire the alert. Dedupe must not break this case.""" + ss = _make_session() + callback = AsyncMock() + ss._auth_alert_callback = callback + + await _run_reader_against_stream( + ss, + [ + _make_result_message( + is_error=True, + api_error_status=403, + errors=["Forbidden"], + ), + ], + ) + + assert callback.await_count == 1 + args, _ = callback.await_args + assert args[0] == ss.agent_name + assert "api_error_status=403" in args[1] + # Enriched detail string should include msg.errors for triage context. + assert "Forbidden" in args[1] + + +@pytest.mark.asyncio +async def test_auth_dedupe_resets_between_turns() -> None: + """Two separate failed turns in one session must produce two callback + invocations — dedupe is per-turn, not per-session.""" + ss = _make_session() + callback = AsyncMock() + ss._auth_alert_callback = callback + + await _run_reader_against_stream( + ss, + [ + # Turn 1: AssistantMessage auth + ResultMessage 401 — one alert + _make_assistant_message(error="authentication_failed"), + _make_result_message(is_error=True, api_error_status=401), + # Turn 2: ResultMessage 401 only — one more alert + _make_result_message(is_error=True, api_error_status=401), + ], + ) + + assert callback.await_count == 2, ( + f"Expected two callbacks (one per failed turn); got {callback.await_count}" + ) + + +@pytest.mark.asyncio +async def test_billing_error_assistant_does_not_block_subsequent_auth_alert() -> None: + """Edge case: AssistantMessage with error="billing_error" (NOT auth) on + one turn must not set the dedupe flag, so a real auth failure on the + NEXT turn still alerts.""" + ss = _make_session() + callback = AsyncMock() + ss._auth_alert_callback = callback + + await _run_reader_against_stream( + ss, + [ + # Turn 1: billing error — must not fire auth alert, must not + # set the dedupe flag (it's not an auth error). + _make_assistant_message(error="billing_error"), + _make_result_message(is_error=True, api_error_status=402), + # Turn 2: real auth failure — must alert. + _make_assistant_message(error="authentication_failed"), + _make_result_message(is_error=True, api_error_status=401), + ], + ) + + assert callback.await_count == 1 + args, _ = callback.await_args + assert args == (ss.agent_name, "authentication_failed") diff --git a/uv.lock b/uv.lock index fd90df95..990ce522 100644 --- a/uv.lock +++ b/uv.lock @@ -508,20 +508,20 @@ wheels = [ [[package]] name = "claude-agent-sdk" -version = "0.1.73" +version = "0.1.77" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "anyio" }, { name = "mcp" }, { name = "sniffio" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/92/d7/5207b9428813918532d52fbcc8342aa105e04cb50afb94118228a322af39/claude_agent_sdk-0.1.73.tar.gz", hash = "sha256:7dbb1e885abac558e39374da632c52e87a931861e20c67e1f36c86e7e3be7f19", size = 242906, upload-time = "2026-05-04T23:14:12.366Z" } +sdist = { url = "https://files.pythonhosted.org/packages/60/d4/7e3ca29d1b6f76639b4bd4becb796ef68a492be3a561c6cd19abe5fe903a/claude_agent_sdk-0.1.77.tar.gz", hash = "sha256:cb292268ecab294047f02365298ecbf8cc17146c4c86fda54b068a1d38e1ebbb", size = 250297, upload-time = "2026-05-08T00:03:03.199Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/b4/9d/37bac90d86f50642ba1e9b70b558cdd643da8bb1109c584ce32ad0b1fc6f/claude_agent_sdk-0.1.73-py3-none-macosx_11_0_arm64.whl", hash = "sha256:62171e16840a8d00681207f6ea133736082b5df701a87548a84ec5ab00cdf5ca", size = 63885588, upload-time = "2026-05-04T23:14:16.457Z" }, - { url = "https://files.pythonhosted.org/packages/3e/e2/81212e886a9bc34a14b8d3588aafd3460bb37328edc4793b10e7df205e05/claude_agent_sdk-0.1.73-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:a702aecfdd1d9ad7dd9fcf19aa9ebde41ffc426e28a4125b3ab2e88b49a89c47", size = 65744104, upload-time = "2026-05-04T23:14:20.345Z" }, - { url = "https://files.pythonhosted.org/packages/80/81/2ecea14eb376d2eaf7f121fdc6ee4b00071b9b1b859ff7251d84fe0e4686/claude_agent_sdk-0.1.73-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:6cb2c38fbfd3d6be7edfc41ce342635569403a8b41f77ded07c328f20b6138cf", size = 77049373, upload-time = "2026-05-04T23:14:24.595Z" }, - { url = "https://files.pythonhosted.org/packages/89/16/a02de4e05bd522beddf2aa02351fc670222929e9e25fff2e23add7937df6/claude_agent_sdk-0.1.73-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:938b2f1adf10e92d4e14c0b16d61f8e616171e98cdc7c3cbcd197dd857fdbc22", size = 77225319, upload-time = "2026-05-04T23:14:28.843Z" }, - { url = "https://files.pythonhosted.org/packages/d0/6a/cb0b8b9a9110ca46e5fb2de7c4e4224e0b18ca8fd50430ca0ea4118608cd/claude_agent_sdk-0.1.73-py3-none-win_amd64.whl", hash = "sha256:85454108785058bab796e9de2d654ce5570c2d9f8de9a9889e02a751d4a43158", size = 78636611, upload-time = "2026-05-04T23:14:33.165Z" }, + { url = "https://files.pythonhosted.org/packages/80/fc/dbb90aff01fe7bdd7708077aaff4b36d5be48a38318fd4ca2216aa64ed87/claude_agent_sdk-0.1.77-py3-none-macosx_11_0_arm64.whl", hash = "sha256:a5697c838cb3a0c78f73698933cd2eea98e8baea99be0740e314fc9e1f69fd4c", size = 60659903, upload-time = "2026-05-08T00:03:06.461Z" }, + { url = "https://files.pythonhosted.org/packages/37/6e/c984d60f4b3cbbf7d84157283338493d11a2b557a2f6389118b5a7f8adc2/claude_agent_sdk-0.1.77-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:a85e3a8cf5d521116d964d28939f25e367d18e5bdb2232cae260b8e8bd9d946f", size = 62721493, upload-time = "2026-05-08T00:03:09.808Z" }, + { url = "https://files.pythonhosted.org/packages/00/83/575e798ce53b58e14d775db104cdb2558706e7f8d22dc647752208fa853f/claude_agent_sdk-0.1.77-py3-none-manylinux_2_17_aarch64.whl", hash = "sha256:16864b91bf369e99590e3ffba82e2d711807caaa4329b11462212779224cd355", size = 70393564, upload-time = "2026-05-08T00:03:13.383Z" }, + { url = "https://files.pythonhosted.org/packages/00/67/50aed27534a9183e204cca33fbef36d0fe5f5145ebf4061c01f2edeaf6db/claude_agent_sdk-0.1.77-py3-none-manylinux_2_17_x86_64.whl", hash = "sha256:7db011c9c8ca4168249f1ce90d4372e13ebab2b122498167e6fb2a5c9c1bd427", size = 70582992, upload-time = "2026-05-08T00:03:16.997Z" }, + { url = "https://files.pythonhosted.org/packages/83/3f/a9d396c46e40d025d4209cf0f6f01a62c70c33b70efceb2d2e991e8ed08c/claude_agent_sdk-0.1.77-py3-none-win_amd64.whl", hash = "sha256:fe7dd006a15a85c1d9a2698d6fffd58e0ce689db1f0040f5b5e4c06d51ce2505", size = 71224295, upload-time = "2026-05-08T00:03:20.695Z" }, ] [[package]] @@ -2258,7 +2258,7 @@ requires-dist = [ { name = "beautifulsoup4", marker = "extra == 'web'", specifier = ">=4.12" }, { name = "caldav", marker = "extra == 'calendar'", specifier = ">=1.3" }, { name = "camoufox", marker = "extra == 'web'", specifier = ">=0.4" }, - { name = "claude-agent-sdk", specifier = ">=0.1.73" }, + { name = "claude-agent-sdk", specifier = ">=0.1.77" }, { name = "cryptography", specifier = ">=41.0" }, { name = "cryptography", marker = "extra == 'calendar'", specifier = ">=41.0" }, { name = "discord-py", marker = "extra == 'discord'", specifier = ">=2.3" },