diff --git a/kora_cli/handlers/slack_dm_handler.py b/kora_cli/handlers/slack_dm_handler.py index 9722ff7c7bd1..2e77cdb699aa 100644 --- a/kora_cli/handlers/slack_dm_handler.py +++ b/kora_cli/handlers/slack_dm_handler.py @@ -115,21 +115,27 @@ def __init__( self, log_path: Optional[Path] = None, slack_client: Optional[Any] = None, + reasoning_engine: Optional[Any] = None, ) -> None: """Construct the handler. Args: log_path: Override the JSONL log file path. Production callers leave this ``None``; tests inject a tmp_path. - slack_client: ST2 — inject a SlackClient for outbound DM - replies. Production code leaves this ``None``; the - handler lazy-creates a SlackClient on first reply via - ``_get_or_create_slack_client``. Tests can inject a mock - client OR leave it ``None`` to test the lazy-creation - failure modes. + slack_client: KR-FEAT-SLACK-DM ST2 — inject a SlackClient + for outbound DM replies. Production leaves ``None``; + the handler lazy-creates a SlackClient on first reply. + reasoning_engine: KR-FEAT-AI-RESPONSE-LOOP ST2 — inject a + ReasoningEngine for reply-content generation. + Production leaves ``None``; the handler resolves + ``kora_cli.listeners.reasoning_engine_listener.current_reasoning_engine()`` + at reply-time. ``None`` from both injection + accessor + → canned fallback (handler stays alive; Joshua isn't + crickets). """ self._log_path = log_path or _resolve_log_path() self._slack_client: Optional[Any] = slack_client + self._reasoning_engine: Optional[Any] = reasoning_engine async def handle_event(self, payload: Dict[str, Any]) -> Dict[str, Any]: """Process a Slack Events payload. @@ -338,49 +344,103 @@ def _emit_received_event(self, payload: Dict[str, Any]) -> None: ) # ------------------------------------------------------------------ - # ST2 — outbound reply + # Outbound reply — reasoning-engine driven (KR-FEAT-AI-RESPONSE-LOOP ST2) # ------------------------------------------------------------------ - # Echo format LOCKED per PM ruling. The trailing slice keeps the - # reply Slack-renderable even if Joshua pastes a >40k-char message. - # Real AI-driven reply generation lands in the KR-FEAT-SLACK-DM-AI - # follow-on; until then this confirms the round-trip. - _ECHO_TEXT_MAX = 200 + # Canned fallback text per PM ruling. Sent when the reasoning + # engine is unavailable OR returned an error. NOT a re-echo — + # Joshua needs to know reasoning didn't work, but not be hit + # with a dump of his own message. + _CANNED_FALLBACK_TEXT = ( + "Kora is currently unable to respond; operator notified." + ) async def _send_echo_reply(self, payload: Dict[str, Any]) -> None: - """Reply to a verified Joshua DM via SlackClient.post_dm. - - Failure modes (each writes one outbound JSONL entry with - ``send_status: "failed"`` + a ``[kora.slack_dm.reply_failed]`` - structured-log emit — never crashes the inbound handler): - - - SlackClient construction fails (missing - ``KORA_SLACK_BOT_TOKEN``) - - SlackTransportError (transport / retry exhaustion / non- - retryable HTTP error) - - SlackAPIError (Slack returned 2xx + ``ok: false``) + """Reply to a verified Joshua DM. + + Method name retained for diff-minimization with KR-FEAT- + SLACK-DM ST2 (#122); body swapped from echo construction + to reasoning-engine call per KR-FEAT-AI-RESPONSE-LOOP ST2. + + Flow: + + 1. Resolve reasoning engine (injection → daemon singleton + → None). If None: canned fallback + outbound entry + + early return. + 2. Build conversation context from JSONL (last 10 turns + same thread). + 3. Call ``engine.respond(message, context)``. Result + error-set → canned fallback + record reasoning_error + in outbound entry. Error-unset → use result.text + + record reasoning metadata in outbound entry. + 4. Send via SlackClient with retry / dead-letter shape + from KR-FEAT-SLACK-DM ST2. + 5. After successful response (NOT canned), call cost- + ladder ``record_inference()`` to bill the tokens. + + Failure modes (each writes one outbound JSONL entry + + structured log; never crashes the inbound handler): + + - SlackClient unavailable → failed outbound entry, + ``failure_reason="slack_client_not_configured"`` + - SlackTransportError / SlackAPIError → failed outbound + entry, stable failure-reason taxonomy from ST2 + - Reasoning engine unavailable → canned reply sent, + outbound ``reasoning_error="engine_unavailable"`` + - Reasoning engine returned error → canned reply sent, + outbound ``reasoning_error=`` """ + from datetime import datetime, timezone + channel_id = _safe_extract(payload, "event", "channel") or "" original_text = _safe_extract(payload, "event", "text") or "" - # Per the bucket spec: thread under the originating DM via - # event.thread_ts (already in-thread) or event.ts (new thread). + # Per spec: thread under originating DM via event.thread_ts + # (already in-thread) or event.ts (new thread). thread_ts = _safe_extract(payload, "event", "thread_ts") or _safe_extract( payload, "event", "ts" ) - echo_text = f"Kora received: {original_text[: self._ECHO_TEXT_MAX]}" + # ---- Reasoning engine acquisition ---- + engine = self._resolve_reasoning_engine() + reasoning_meta: Dict[str, Any] = { + "model_used": None, + "input_tokens": None, + "output_tokens": None, + "reasoning_duration_ms": None, + "reasoning_error": None, + } + + if engine is None: + # Daemon misconfigured OR running outside-coordinator + # test path. Send canned fallback so Joshua isn't met + # with silence; record the reason for operator triage. + reply_text = self._CANNED_FALLBACK_TEXT + reasoning_meta["reasoning_error"] = "engine_unavailable" + logger.warning( + "[kora.slack_dm.reasoning_skipped] reason=engine_unavailable " + "channel=%s — sending canned fallback", + channel_id, + ) + else: + reply_text, reasoning_meta = await self._call_reasoning_engine( + engine=engine, + payload=payload, + channel_id=channel_id, + thread_ts=thread_ts, + original_text=original_text, + ) + + # ---- Slack outbound ---- client = self._get_or_create_slack_client() if client is None: - # SlackClient construction failed — already logged the - # reason inside _get_or_create. Surface as a failed - # outbound entry. self._append_outbound_log_entry( channel_id=channel_id, thread_ts=thread_ts, - text=echo_text, + text=reply_text, slack_message_ts=None, send_status="failed", failure_reason="slack_client_not_configured", + **reasoning_meta, ) self._emit_reply_failed_event( channel_id=channel_id, @@ -391,21 +451,19 @@ async def _send_echo_reply(self, payload: Dict[str, Any]) -> None: try: response = await client.post_dm( channel_id=channel_id, - text=echo_text, + text=reply_text, thread_ts=thread_ts, ) except Exception as exc: - # Includes SlackAPIError + SlackTransportError. Caught - # broadly so even an unexpected client-side failure - # (e.g. httpx version mismatch) doesn't propagate. reason = self._reply_failure_reason(exc) self._append_outbound_log_entry( channel_id=channel_id, thread_ts=thread_ts, - text=echo_text, + text=reply_text, slack_message_ts=None, send_status="failed", failure_reason=reason, + **reasoning_meta, ) self._emit_reply_failed_event( channel_id=channel_id, reason=reason @@ -419,11 +477,192 @@ async def _send_echo_reply(self, payload: Dict[str, Any]) -> None: self._append_outbound_log_entry( channel_id=channel_id, thread_ts=thread_ts, - text=echo_text, + text=reply_text, slack_message_ts=str(message_ts) if message_ts else None, send_status="ok", + **reasoning_meta, ) + # ---- Cost-ladder write (only on successful, non-canned reply) ---- + # Bill the tokens against the $200/mo Agent SDK pool. Skip + # if the reply was canned (no real inference happened). + if reasoning_meta["reasoning_error"] is None: + self._record_inference_to_cost_ladder(reasoning_meta) + + # ------------------------------------------------------------------ + # Reasoning engine helpers + # ------------------------------------------------------------------ + + def _resolve_reasoning_engine(self) -> Optional[Any]: + """Return injected engine, else daemon-singleton, else None.""" + if self._reasoning_engine is not None: + return self._reasoning_engine + try: + from kora_cli.listeners.reasoning_engine_listener import ( + current_reasoning_engine, + ) + except Exception: + # Listener module didn't import — daemon not active. + return None + return current_reasoning_engine() + + async def _call_reasoning_engine( + self, + *, + engine: Any, + payload: Dict[str, Any], + channel_id: str, + thread_ts: Optional[str], + original_text: str, + ) -> tuple[str, Dict[str, Any]]: + """Call engine.respond + project result into reply_text + + reasoning_meta dict. + + Returns ``(reply_text, reasoning_meta)``. On error the + reply_text is the canned fallback; meta carries the error + code so the outbound JSONL records it. + """ + from datetime import datetime, timezone + + # Lazy import — keeps non-reasoning test paths fast + + # avoids forcing the anthropic SDK import at module-load. + from kora_cli.reasoning.context_loader import ( + load_slack_dm_context, + ) + from kora_cli.reasoning.engine import IncomingMessage + + try: + context = load_slack_dm_context( + channel_id=channel_id, thread_ts=thread_ts + ) + except Exception as exc: + logger.warning( + "[kora.slack_dm.reasoning_skipped] context-load failed: %r " + "channel=%s — using empty context", + exc, + channel_id, + ) + from kora_cli.reasoning.engine import ConversationContext + + context = ConversationContext() + + message = IncomingMessage( + text=original_text, + source="slack_dm", + received_at=datetime.now(timezone.utc), + metadata={ + "channel_id": channel_id, + "thread_ts": thread_ts, + "user_id": _safe_extract(payload, "event", "user"), + "event_ts": _safe_extract(payload, "event", "ts"), + }, + ) + + try: + result = await engine.respond(message, context) + except Exception as exc: + # An engine that itself raises (not just ResponseResult.error) + # is a runtime bug — caught defensively so the handler + # stays alive. + logger.warning( + "[kora.slack_dm.reasoning_skipped] engine.respond raised %r " + "channel=%s — canned fallback", + exc, + channel_id, + ) + return ( + self._CANNED_FALLBACK_TEXT, + { + "model_used": None, + "input_tokens": None, + "output_tokens": None, + "reasoning_duration_ms": None, + "reasoning_error": f"engine_exception:{type(exc).__name__}", + }, + ) + + meta = { + "model_used": result.model_used or None, + "input_tokens": result.input_tokens, + "output_tokens": result.output_tokens, + "reasoning_duration_ms": result.reasoning_duration_ms, + "reasoning_error": result.error, + } + + if result.error is not None: + # Engine refused (paused / cost-halted / SDK failure). + # Send canned text so Joshua sees something; record + # the error code so operator can triage. + logger.warning( + "[kora.slack_dm.reasoning_failed] error=%s channel=%s", + result.error, + channel_id, + ) + return (self._CANNED_FALLBACK_TEXT, meta) + + # Success — engine produced a real response. Defensive + # check: empty text from a successful call shouldn't + # happen but if it does, fall back to canned so Joshua + # doesn't see a blank message. + if not result.text.strip(): + logger.warning( + "[kora.slack_dm.reasoning_failed] empty text on success " + "model=%s channel=%s — canned fallback", + result.model_used, + channel_id, + ) + meta["reasoning_error"] = "empty_response_text" + return (self._CANNED_FALLBACK_TEXT, meta) + + return (result.text, meta) + + @staticmethod + def _record_inference_to_cost_ladder( + reasoning_meta: Dict[str, Any], + ) -> None: + """Bill the reply's tokens to the cost-ladder ($200/mo pool). + + Fail-soft: holder uninitialized → skip (test path / partial + daemon boot). record_inference itself is fail-soft per + ``agent.cost_state_holder``'s docstring (pricing-lookup miss + accumulates 0). + """ + try: + from agent.cost_state_holder import get_cost_holder + from agent.usage_pricing import CanonicalUsage + except Exception as exc: + logger.warning( + "[kora.slack_dm.cost_ladder_skipped] import failed: %r", + exc, + ) + return + + holder = get_cost_holder() + if holder is None: + return + + model_name = reasoning_meta.get("model_used") + input_tokens = reasoning_meta.get("input_tokens") or 0 + output_tokens = reasoning_meta.get("output_tokens") or 0 + if not model_name or (input_tokens == 0 and output_tokens == 0): + return + + try: + holder.record_inference( + CanonicalUsage( + input_tokens=int(input_tokens), + output_tokens=int(output_tokens), + ), + model_name=str(model_name), + provider="anthropic", + ) + except Exception as exc: + logger.warning( + "[kora.slack_dm.cost_ladder_skipped] record_inference " + "raised %r — continuing", + exc, + ) + def _get_or_create_slack_client(self) -> Optional[Any]: """Lazy SlackClient construction. @@ -462,10 +701,32 @@ def _append_outbound_log_entry( slack_message_ts: Optional[str], send_status: str, failure_reason: Optional[str] = None, + # KR-FEAT-AI-RESPONSE-LOOP ST2 — reasoning metadata. All + # optional + backwards-compatible; pre-ST2 outbound entries + # don't have these fields and consumers must handle absence + # (same JSONL file accumulates both shapes). + model_used: Optional[str] = None, + input_tokens: Optional[int] = None, + output_tokens: Optional[int] = None, + reasoning_duration_ms: Optional[int] = None, + reasoning_error: Optional[str] = None, ) -> None: """Outbound-side JSONL entry. Distinct schema from inbound entries (``sent_at`` instead of ``received_at``) so operator - log-analysis can branch on key presence.""" + log-analysis can branch on key presence. + + ST2 extended fields (all optional, all None on canned- + fallback / non-reasoning paths so historical entries stay + readable): + + - ``model_used``: e.g. ``"claude-opus-4-7"`` + - ``input_tokens`` / ``output_tokens``: from SDK usage + - ``reasoning_duration_ms``: engine-side wall-clock + - ``reasoning_error``: stable error code from + ``ResponseResult.error`` (``cost_ladder_halted`` / + ``sdk_5xx`` / ``engine_unavailable`` / etc.) — None on + successful reasoning calls + """ entry: Dict[str, Any] = { "sent_at": _now_iso(), "channel_id": channel_id, @@ -476,6 +737,20 @@ def _append_outbound_log_entry( } if failure_reason: entry["failure_reason"] = failure_reason + # Reasoning meta — write fields when set (None is the + # placeholder for non-reasoning paths; recording None as a + # null in JSONL is fine, but skipping empties keeps the + # entry lean). + if model_used is not None: + entry["model_used"] = model_used + if input_tokens is not None: + entry["input_tokens"] = int(input_tokens) + if output_tokens is not None: + entry["output_tokens"] = int(output_tokens) + if reasoning_duration_ms is not None: + entry["reasoning_duration_ms"] = int(reasoning_duration_ms) + if reasoning_error is not None: + entry["reasoning_error"] = reasoning_error try: self._log_path.parent.mkdir(parents=True, exist_ok=True) diff --git a/kora_cli/listeners/__init__.py b/kora_cli/listeners/__init__.py index f2051b89c35e..e0fc40b274e5 100644 --- a/kora_cli/listeners/__init__.py +++ b/kora_cli/listeners/__init__.py @@ -29,3 +29,9 @@ # construction happens per cycle (stateless across cycles), so # startup is a clean no-op + LOG line. from kora_cli.listeners import heartbeat_probes_listener # noqa: F401 +# KR-FEAT-AI-RESPONSE-LOOP ST2 — reasoning engine listener. +# Constructs the AnthropicReasoningEngine at daemon startup; +# fail-CLOSED on missing creds / missing system prompt (coordinator +# aborts boot). Module-level `current_reasoning_engine()` accessor +# mirrors `current_pool()` so SlackDMHandler reads cross-cuttingly. +from kora_cli.listeners import reasoning_engine_listener # noqa: F401 diff --git a/kora_cli/listeners/reasoning_engine_listener.py b/kora_cli/listeners/reasoning_engine_listener.py new file mode 100644 index 000000000000..aa20468cced8 --- /dev/null +++ b/kora_cli/listeners/reasoning_engine_listener.py @@ -0,0 +1,152 @@ +"""Reasoning engine daemon listener — KR-FEAT-AI-RESPONSE-LOOP ST2. + +Wraps :class:`AnthropicReasoningEngine` in the +:class:`DaemonCoordinator` lifecycle: + + - Startup: construct the engine (loads system prompt from + ``kora_docs/00_canonical_current_state/kora_system_prompt.md``; + resolves credential cascade OAuth-first → API key → fail-CLOSED). + Construction failure → daemon aborts boot (matches the bucket + spec's "engine startup failure → daemon fails-CLOSED" since + a daemon that can't reason is one that can't fulfill its + primary purpose). + - Hold: module-level ``_engine_singleton`` set via + ``_set_singleton``; cleared on shutdown. + - Shutdown: close the engine's underlying HTTP client. Best- + effort; the coordinator's per-listener timeout (default 10s) + caps the wait. + +Mirrors ``kora_cli/listeners/mcp_consumption.py`` shape — singleton +pattern + ``current_reasoning_engine()`` accessor for cross-cutting +read from any code path (notably ``SlackDMHandler`` in ST2). + +# Why startup failure should be FATAL + +The bucket spec is explicit: "Engine startup failure during daemon +boot → daemon fails-CLOSED (no echoes since previous behavior is +replaced by reasoning that can't run)." After ST2 wires the +handler to the engine, the prior echo path is gone — if the engine +can't construct, the daemon has no useful response path. Better +to abort boot loudly than to ship a daemon that drops Joshua's +DMs into a canned-fallback loop. + +The coordinator's :class:`DaemonCoordinator` handles this +naturally: any exception from ``startup()`` aborts the boot + +unwinds already-started listeners (KR-D-DAEMON ST1's lifecycle). +""" + +from __future__ import annotations + +import logging +from typing import Optional + +from kora_cli.daemon import DEFAULT_SHUTDOWN_TIMEOUT, register_daemon_listener +from kora_cli.reasoning.anthropic_engine import ( + AnthropicReasoningEngine, + ReasoningEngineError, +) +from kora_cli.reasoning.engine import ReasoningEngine + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Module-level singleton + accessor (mirrors current_pool pattern) +# --------------------------------------------------------------------------- + + +_engine_singleton: Optional[ReasoningEngine] = None + + +def _set_singleton(engine: ReasoningEngine) -> None: + global _engine_singleton + _engine_singleton = engine + + +def _clear_singleton() -> None: + global _engine_singleton + _engine_singleton = None + + +def current_reasoning_engine() -> Optional[ReasoningEngine]: + """Return the live :class:`ReasoningEngine`, or ``None``. + + ``None`` cases: + - Daemon not running + - Listener not yet started + - Listener stopped (post-shutdown) + - Listener startup failed AND the daemon proceeded anyway + (shouldn't happen — fatal-CLOSED — but defensive) + + Mirrors :func:`kora_cli.listeners.mcp_consumption.current_pool`. + """ + return _engine_singleton + + +# --------------------------------------------------------------------------- +# Listener lifecycle wrapper +# --------------------------------------------------------------------------- + + +class ReasoningEngineListener: + """Owns the engine instance + sets the module-level singleton. + + Tests inject a pre-built engine via the constructor arg; the + factory leaves it ``None`` so production startup creates a + real ``AnthropicReasoningEngine``. + """ + + def __init__( + self, engine: Optional[ReasoningEngine] = None + ) -> None: + self._engine: Optional[ReasoningEngine] = engine + + async def startup(self) -> None: + if self._engine is None: + # Construction can raise ReasoningEngineNotConfigured / + # ReasoningSystemPromptError. We do NOT catch — the + # coordinator's startup-failure path unwinds the daemon, + # which is the spec-mandated fail-CLOSED behavior. + try: + self._engine = AnthropicReasoningEngine() + except ReasoningEngineError as exc: + logger.error( + "[kora.reasoning] engine construction failed: %r " + "— daemon will abort boot (fail-CLOSED). Operator " + "must configure credentials + system prompt before " + "the daemon can reply to DMs.", + exc, + ) + raise + _set_singleton(self._engine) + logger.info("[kora.reasoning] engine listener active") + + async def shutdown(self) -> None: + engine = self._engine + _clear_singleton() + if engine is None: + return + try: + close_method = getattr(engine, "close", None) + if close_method is not None: + result = close_method() + if hasattr(result, "__await__"): + await result + except Exception as exc: + logger.warning( + "[kora.reasoning] engine shutdown raised %r — continuing", + exc, + ) + + +# --------------------------------------------------------------------------- +# Factory + registration (import-time side effect) +# --------------------------------------------------------------------------- + + +def _factory(): + listener = ReasoningEngineListener() + return (listener.startup, listener.shutdown, DEFAULT_SHUTDOWN_TIMEOUT) + + +register_daemon_listener("reasoning_engine", _factory) diff --git a/kora_cli/reasoning/anthropic_engine.py b/kora_cli/reasoning/anthropic_engine.py index 3b264a3ac335..7106484fde81 100644 --- a/kora_cli/reasoning/anthropic_engine.py +++ b/kora_cli/reasoning/anthropic_engine.py @@ -1,24 +1,24 @@ -"""AnthropicReasoningEngine — KR-FEAT-AI-RESPONSE-LOOP ST1. +"""AnthropicReasoningEngine — KR-FEAT-AI-RESPONSE-LOOP ST1+ST2. Implements :class:`kora_cli.reasoning.engine.ReasoningEngine` against -Anthropic's Python SDK (``anthropic==0.86.0``, declared in the -``[web]`` extra alongside fastapi/uvicorn/slowapi so daemon installs -auto-pick it up). +Anthropic's Python SDK (``anthropic==0.86.0``, runtime dep — promoted +from extra in ST2 per PM ruling 2026-05-22 since the reasoning_engine +listener imports it unconditionally at boot). -# Credential cascade +# Credential cascade — OAuth FIRST (PM ruling 2026-05-22 ST2) -Two supported credential sources (K-DG drift surfaced in the ST1 -PR body — bucket spec said ``KORA_ANTHROPIC_API_KEY`` only; the -existing env mapping doc plus the gate-2 anti-secret block + the -"Max plan via Agent SDK billing" framing in §1 imply -``CLAUDE_CODE_OAUTH_TOKEN`` is the canonical credential): +Two supported credential sources. **OAuth-first** because Joshua's +Max 20x plan + the post-May-15 SDK billing split route the $200/mo +Agent SDK pool via the OAuth token path. OAuth = production; +API key = fallback for testing / dev / local-without-Max-setup. - 1. ``KORA_ANTHROPIC_API_KEY`` (if set) → SDK constructed with - ``api_key=...``. Billing: Anthropic Console (operator must - provision an API key separately). - 2. ``CLAUDE_CODE_OAUTH_TOKEN`` (fallback) → SDK constructed with + 1. ``CLAUDE_CODE_OAUTH_TOKEN`` (if set) → SDK constructed with ``auth_token=...``. Billing: Max plan ($200/mo Agent SDK pool). Existing Doppler ``kora-runtime-anthropic`` secret. + **Production path.** + 2. ``KORA_ANTHROPIC_API_KEY`` (fallback) → SDK constructed with + ``api_key=...``. Billing: Anthropic Console (operator must + provision an API key separately). Test / dev escape hatch. Both unset → ``ReasoningEngineNotConfigured`` raised at construction. **Fail-CLOSED** per @@ -153,18 +153,22 @@ def __init__( # stand-in (anything with an async ``messages.create``). client: Optional[Any] = None, ) -> None: - # Credential cascade. Read once at construction (production - # rotation pattern: redeploy, not hot-reload — same as - # SlackClient's bot-token model). - api_key = os.environ.get(API_KEY_ENV, "").strip() or None + # Credential cascade — OAuth FIRST (PM ruling 2026-05-22). + # OAuth = production path (Max plan billing); API key = + # fallback for dev/testing. Read once at construction + # (production rotation pattern: redeploy, not hot-reload — + # same as SlackClient's bot-token model). oauth_token = os.environ.get(OAUTH_TOKEN_ENV, "").strip() or None - if not api_key and not oauth_token: + api_key = os.environ.get(API_KEY_ENV, "").strip() or None + if not oauth_token and not api_key: raise ReasoningEngineNotConfigured( - f"both {API_KEY_ENV} and {OAUTH_TOKEN_ENV} are unset — " - "daemon cannot reason. Set one in Doppler " - "(kora-runtime-anthropic project). See env-mapping doc." + f"both {OAUTH_TOKEN_ENV} and {API_KEY_ENV} are unset — " + "daemon cannot reason. Set CLAUDE_CODE_OAUTH_TOKEN in " + "Doppler (kora-runtime-anthropic project) — the " + "production path via Joshua's Max plan." ) - self._auth_mode: str = "api_key" if api_key else "oauth_token" + # OAuth wins when both are set. + self._auth_mode: str = "oauth_token" if oauth_token else "api_key" # Stored under underscore-prefixed attrs to discourage casual # serialization. NEVER logged. self._api_key = api_key @@ -286,15 +290,15 @@ async def _ensure_client(self) -> Any: # never see the real client created. from anthropic import AsyncAnthropic - if self._api_key: + # OAuth-first per PM ruling: prefer Max-plan billing path. + # Falls back to API key only when OAuth is absent. + if self._oauth_token: self._client = AsyncAnthropic( - api_key=self._api_key, timeout=self._timeout + auth_token=self._oauth_token, timeout=self._timeout ) else: - # OAuth token via the SDK's ``auth_token`` constructor - # arg (Max plan billing path). self._client = AsyncAnthropic( - auth_token=self._oauth_token, timeout=self._timeout + api_key=self._api_key, timeout=self._timeout ) return self._client diff --git a/pyproject.toml b/pyproject.toml index a826277ed3d4..69ed1ed11a19 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,6 +83,12 @@ dependencies = [ # over on import (task #269). PR #125 traced the cascade. Same # placement discipline as aiosmtplib above. "slowapi==0.1.9", + # Anthropic SDK — required by the reasoning engine listener (KR-FEAT- + # AI-RESPONSE-LOOP). Promoted to runtime per the same rule that moved + # aiosmtplib + the slowapi-fix lesson: the daemon's reasoning_engine + # listener imports it unconditionally at boot. Previously under + # [anthropic] + [web] extras; consolidated here. + "anthropic==0.86.0", ] [project.urls] @@ -91,9 +97,6 @@ Repository = "https://github.com/rafe-walker/kora" Upstream = "https://github.com/NousResearch/hermes-agent" [project.optional-dependencies] -# Native Anthropic provider — only needed when provider=anthropic (not via -# OpenRouter or other aggregators). -anthropic = ["anthropic==0.86.0"] # Web search backends — each only loaded when the user picks it as their # search provider (configured via `hermes tools` or config.yaml). exa = ["exa-py==2.10.2"] @@ -198,7 +201,7 @@ youtube = [ "youtube-transcript-api==1.2.4", ] # `hermes dashboard` (localhost SPA + API). Not in core to keep the default install lean. -web = ["fastapi==0.133.1", "uvicorn[standard]==0.41.0", "anthropic==0.86.0"] +web = ["fastapi==0.133.1", "uvicorn[standard]==0.41.0"] all = [ # Policy (2026-05-12): `[all]` includes only extras that genuinely # CAN'T be lazy-installed via `tools/lazy_deps.py` — i.e. things every diff --git a/tests/kora_cli/handlers/test_slack_dm_reply.py b/tests/kora_cli/handlers/test_slack_dm_reply.py index 214a24430177..7171f9a89de3 100644 --- a/tests/kora_cli/handlers/test_slack_dm_reply.py +++ b/tests/kora_cli/handlers/test_slack_dm_reply.py @@ -116,47 +116,119 @@ def __init__(self): # --------------------------------------------------------------------------- -@pytest.mark.asyncio -async def test_joshua_dm_triggers_echo_reply_with_locked_format( - log_path, mock_client +def _make_reasoning_engine( + text: str = "kora's thoughtful reply", + model: str = "claude-opus-4-7", + input_tokens: int = 120, + output_tokens: int = 80, + error: str | None = None, ): - handler = SlackDMHandler(log_path=log_path, slack_client=mock_client) + """Build an AsyncMock-style ReasoningEngine returning a fixed + ResponseResult. ST2 wires this in place of the prior echo path — + tests that previously asserted echo behavior now assert against + the mocked engine's text.""" + from unittest.mock import AsyncMock + + from kora_cli.reasoning.engine import ResponseResult + + class _MockEngine: + def __init__(self): + self.respond = AsyncMock( + return_value=ResponseResult( + text=text, + model_used=model, + input_tokens=input_tokens, + output_tokens=output_tokens, + reasoning_duration_ms=42, + error=error, + ) + ) + self.close = AsyncMock() + + return _MockEngine() + + +@pytest.mark.asyncio +async def test_joshua_dm_triggers_reasoning_reply(log_path, mock_client): + """ST2 swap: handler now calls the reasoning engine instead of + constructing the echo. The engine's `text` is what gets sent.""" + engine = _make_reasoning_engine(text="here is your answer") + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) await handler.handle_event(_make_payload(text="ping")) + engine.respond.assert_awaited_once() mock_client.post_dm.assert_awaited_once() call_kwargs = mock_client.post_dm.await_args.kwargs assert call_kwargs["channel_id"] == "D01CHAN01" - # Echo format LOCKED. - assert call_kwargs["text"] == "Kora received: ping" + # Engine's response text is what Slack receives — NOT an echo. + assert call_kwargs["text"] == "here is your answer" # thread_ts defaults to event.ts when event.thread_ts is absent. assert call_kwargs["thread_ts"] == "1700000000.001" @pytest.mark.asyncio async def test_thread_ts_used_when_present(log_path, mock_client): - handler = SlackDMHandler(log_path=log_path, slack_client=mock_client) + """Reply threads under the original thread, not the latest msg. + Independent of the reply text — the threading logic is the same + pre + post ST2.""" + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=_make_reasoning_engine(), + ) await handler.handle_event( _make_payload(thread_ts="1699999999.000", ts="1700000000.001") ) call_kwargs = mock_client.post_dm.await_args.kwargs - # Reply threads under the original thread, not the latest message. assert call_kwargs["thread_ts"] == "1699999999.000" @pytest.mark.asyncio -async def test_echo_text_truncated_at_200_chars(log_path, mock_client): +async def test_reasoning_engine_receives_full_inbound_text( + log_path, mock_client +): + """ST2 swap removes the prior 200-char echo truncation. The + reasoning engine sees the full inbound text + decides its own + response length per the system prompt.""" long_text = "x" * 5000 - handler = SlackDMHandler(log_path=log_path, slack_client=mock_client) + engine = _make_reasoning_engine(text="short engine reply") + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) await handler.handle_event(_make_payload(text=long_text)) + + # Engine got the full untruncated message. + engine_call = engine.respond.await_args + assert engine_call.args[0].text == long_text # IncomingMessage.text + # Slack received the engine's response, not a truncated echo. sent_text = mock_client.post_dm.await_args.kwargs["text"] - # "Kora received: " is 15 chars, plus up to 200 of original. - assert sent_text.startswith("Kora received: ") - assert len(sent_text) == len("Kora received: ") + 200 + assert sent_text == "short engine reply" @pytest.mark.asyncio -async def test_jsonl_has_inbound_then_outbound_entry(log_path, mock_client): - handler = SlackDMHandler(log_path=log_path, slack_client=mock_client) +async def test_jsonl_has_inbound_then_outbound_entry_with_reasoning_meta( + log_path, mock_client +): + """ST2 schema extension: outbound entries carry the new + reasoning fields (model_used, input_tokens, output_tokens, + reasoning_duration_ms) on successful reasoning calls.""" + engine = _make_reasoning_engine( + text="thoughtful answer", + model="claude-opus-4-7", + input_tokens=150, + output_tokens=60, + ) + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) await handler.handle_event(_make_payload(text="hi")) entries = _read_lines(log_path) @@ -167,12 +239,20 @@ async def test_jsonl_has_inbound_then_outbound_entry(log_path, mock_client): assert "received_at" in entries[0] assert "sent_at" not in entries[0] - # Outbound second (ok). - assert entries[1]["send_status"] == "ok" - assert "sent_at" in entries[1] - assert "received_at" not in entries[1] - assert entries[1]["slack_message_ts"] == "1700000001.999" - assert entries[1]["text"] == "Kora received: hi" + # Outbound second (ok) — with reasoning meta. + out = entries[1] + assert out["send_status"] == "ok" + assert "sent_at" in out + assert "received_at" not in out + assert out["slack_message_ts"] == "1700000001.999" + assert out["text"] == "thoughtful answer" + # New ST2 fields populated on successful reasoning. + assert out["model_used"] == "claude-opus-4-7" + assert out["input_tokens"] == 150 + assert out["output_tokens"] == 60 + assert out["reasoning_duration_ms"] == 42 + # No reasoning_error key on success. + assert "reasoning_error" not in out # --------------------------------------------------------------------------- @@ -386,3 +466,325 @@ async def test_bot_token_never_in_jsonl(log_path, monkeypatch, mock_client): "bot token env value appeared in JSONL — handler must NEVER " "log secret material" ) + + +# =========================================================================== +# KR-FEAT-AI-RESPONSE-LOOP ST2 — reasoning integration paths +# =========================================================================== + + +@pytest.mark.asyncio +async def test_engine_unavailable_sends_canned_fallback( + log_path, mock_client, caplog +): + """No injected engine + listener accessor returns None → + canned fallback text sent + outbound JSONL records + reasoning_error='engine_unavailable'. Handler does NOT crash.""" + caplog.set_level(logging.WARNING) + + # No reasoning_engine injection — handler will call the listener + # accessor, which returns None outside a running daemon. + handler = SlackDMHandler(log_path=log_path, slack_client=mock_client) + result = await handler.handle_event(_make_payload(text="hi")) + assert result == {"ok": True} + + # Canned text sent to Slack. + sent_text = mock_client.post_dm.await_args.kwargs["text"] + assert sent_text == handler._CANNED_FALLBACK_TEXT + + # Outbound JSONL has reasoning_error. + out = _read_lines(log_path)[1] + assert out["send_status"] == "ok" + assert out["text"] == handler._CANNED_FALLBACK_TEXT + assert out["reasoning_error"] == "engine_unavailable" + # No reasoning meta fields (None values not written per the + # entry-builder's "if X is not None" gate). + assert "model_used" not in out + assert "input_tokens" not in out + + # Structured-log line recorded. + assert any( + "reasoning_skipped" in r.getMessage() + and "engine_unavailable" in r.getMessage() + for r in caplog.records + ) + + +@pytest.mark.asyncio +async def test_engine_returns_error_sends_canned_fallback( + log_path, mock_client, caplog +): + """Engine returns ResponseResult(error='cost_ladder_halted') → + canned text sent + outbound records the error code. Handler + does NOT crash.""" + caplog.set_level(logging.WARNING) + engine = _make_reasoning_engine( + text="", # engine signals "no real text" on error + model="", + input_tokens=0, + output_tokens=0, + error="cost_ladder_halted", + ) + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + await handler.handle_event(_make_payload(text="hi")) + + sent_text = mock_client.post_dm.await_args.kwargs["text"] + assert sent_text == handler._CANNED_FALLBACK_TEXT + + out = _read_lines(log_path)[1] + assert out["reasoning_error"] == "cost_ladder_halted" + assert out["text"] == handler._CANNED_FALLBACK_TEXT + # Reasoning meta IS captured even on error (operator triage). + assert out["reasoning_duration_ms"] == 42 + + assert any( + "reasoning_failed" in r.getMessage() + and "cost_ladder_halted" in r.getMessage() + for r in caplog.records + ) + + +@pytest.mark.asyncio +async def test_engine_paused_state_sends_canned_fallback( + log_path, mock_client +): + """Engine error='operational_state_paused' → canned text.""" + engine = _make_reasoning_engine( + text="", + error="operational_state_paused", + ) + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + await handler.handle_event(_make_payload(text="hi")) + out = _read_lines(log_path)[1] + assert out["reasoning_error"] == "operational_state_paused" + assert out["text"] == handler._CANNED_FALLBACK_TEXT + + +@pytest.mark.asyncio +async def test_engine_raises_exception_sends_canned_fallback( + log_path, mock_client, caplog +): + """Engine itself raises (not just sets ResponseResult.error) — + handler catches defensively + sends canned text + records + 'engine_exception:'.""" + from unittest.mock import AsyncMock + + caplog.set_level(logging.WARNING) + + class _CrashingEngine: + def __init__(self): + self.respond = AsyncMock(side_effect=RuntimeError("boom")) + self.close = AsyncMock() + + engine = _CrashingEngine() + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + result = await handler.handle_event(_make_payload(text="hi")) + assert result == {"ok": True} + + sent_text = mock_client.post_dm.await_args.kwargs["text"] + assert sent_text == handler._CANNED_FALLBACK_TEXT + + out = _read_lines(log_path)[1] + assert out["reasoning_error"] == "engine_exception:RuntimeError" + + +@pytest.mark.asyncio +async def test_empty_engine_text_on_success_falls_back( + log_path, mock_client, caplog +): + """Defensive: engine returns ResponseResult with error=None but + empty text → canned fallback (Joshua shouldn't see a blank + message).""" + caplog.set_level(logging.WARNING) + engine = _make_reasoning_engine(text=" ", error=None) + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + await handler.handle_event(_make_payload(text="hi")) + + sent_text = mock_client.post_dm.await_args.kwargs["text"] + assert sent_text == handler._CANNED_FALLBACK_TEXT + out = _read_lines(log_path)[1] + assert out["reasoning_error"] == "empty_response_text" + + +@pytest.mark.asyncio +async def test_reasoning_engine_receives_message_metadata( + log_path, mock_client +): + """Verify the IncomingMessage passed to engine.respond carries + the source + channel_id + thread_ts + user_id metadata the + engine may want to use in its prompt context.""" + engine = _make_reasoning_engine() + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + await handler.handle_event( + _make_payload(text="hi", thread_ts="1699999.000", ts="1700000.001") + ) + + msg = engine.respond.await_args.args[0] + assert msg.text == "hi" + assert msg.source == "slack_dm" + assert msg.metadata["channel_id"] == "D01CHAN01" + assert msg.metadata["thread_ts"] == "1699999.000" + assert msg.metadata["user_id"] == JOSHUA_ID + assert msg.metadata["event_ts"] == "1700000.001" + + +# --------------------------------------------------------------------------- +# Cost-ladder integration +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_cost_ladder_record_inference_called_on_success( + log_path, mock_client, monkeypatch +): + """Successful reasoning call → cost-ladder record_inference() + invoked with CanonicalUsage built from result tokens + + provider='anthropic'.""" + from agent.cost_state_holder import ( + CostStateHolder, + init_cost_holder, + _reset_cost_holder_for_tests, + ) + from datetime import datetime, timezone + + _reset_cost_holder_for_tests() + init_cost_holder( + billing_period_start=datetime.now(timezone.utc), + credit_pool_usd=200.0, + extra_usage_off=True, + ) + + # Spy on holder.record_inference to verify the call. + from agent import cost_state_holder as csh_mod + real_holder = csh_mod._HOLDER + record_calls: list = [] + + original = real_holder.record_inference + + def _spy(canonical_usage, *, model_name, provider=None, base_url=None): + record_calls.append( + { + "input_tokens": canonical_usage.input_tokens, + "output_tokens": canonical_usage.output_tokens, + "model_name": model_name, + "provider": provider, + } + ) + return original( + canonical_usage, + model_name=model_name, + provider=provider, + base_url=base_url, + ) + + monkeypatch.setattr(real_holder, "record_inference", _spy) + + engine = _make_reasoning_engine( + text="answer", + model="claude-opus-4-7", + input_tokens=150, + output_tokens=60, + ) + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + await handler.handle_event(_make_payload(text="hi")) + + assert len(record_calls) == 1 + assert record_calls[0]["input_tokens"] == 150 + assert record_calls[0]["output_tokens"] == 60 + assert record_calls[0]["model_name"] == "claude-opus-4-7" + assert record_calls[0]["provider"] == "anthropic" + + _reset_cost_holder_for_tests() + + +@pytest.mark.asyncio +async def test_cost_ladder_skipped_on_canned_fallback( + log_path, mock_client, monkeypatch +): + """Canned-fallback path must NOT call record_inference — there + was no real inference to bill.""" + from agent.cost_state_holder import ( + init_cost_holder, + _reset_cost_holder_for_tests, + ) + from agent import cost_state_holder as csh_mod + from datetime import datetime, timezone + + _reset_cost_holder_for_tests() + init_cost_holder( + billing_period_start=datetime.now(timezone.utc), + credit_pool_usd=200.0, + extra_usage_off=True, + ) + + record_calls: list = [] + original = csh_mod._HOLDER.record_inference + + def _spy(canonical_usage, *, model_name, provider=None, base_url=None): + record_calls.append(model_name) + + monkeypatch.setattr(csh_mod._HOLDER, "record_inference", _spy) + + # Engine returns an error → canned fallback → NO cost-ladder write. + engine = _make_reasoning_engine( + text="", error="cost_ladder_halted" + ) + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + await handler.handle_event(_make_payload(text="hi")) + + assert record_calls == [] + _reset_cost_holder_for_tests() + + +@pytest.mark.asyncio +async def test_cost_ladder_skipped_when_holder_uninitialized( + log_path, mock_client +): + """No cost holder → handler skips silently; no crash.""" + from agent import cost_state_holder as csh_mod + + # Ensure holder is None. + csh_mod._reset_cost_holder_for_tests() + + engine = _make_reasoning_engine( + text="answer", model="claude-opus-4-7", input_tokens=10, output_tokens=5 + ) + handler = SlackDMHandler( + log_path=log_path, + slack_client=mock_client, + reasoning_engine=engine, + ) + result = await handler.handle_event(_make_payload(text="hi")) + assert result == {"ok": True} + # Outbound still recorded. + out = _read_lines(log_path)[1] + assert out["send_status"] == "ok" + assert out["model_used"] == "claude-opus-4-7" diff --git a/tests/kora_cli/reasoning/test_anthropic_engine.py b/tests/kora_cli/reasoning/test_anthropic_engine.py index 9802a41f20f2..9bce997cfa1d 100644 --- a/tests/kora_cli/reasoning/test_anthropic_engine.py +++ b/tests/kora_cli/reasoning/test_anthropic_engine.py @@ -163,15 +163,19 @@ def test_construction_succeeds_with_api_key(monkeypatch, system_prompt_path): assert engine._auth_mode == "api_key" -def test_api_key_wins_over_oauth_when_both_set( +def test_oauth_wins_over_api_key_when_both_set( monkeypatch, system_prompt_path ): + """PM ruling 2026-05-22 ST2: OAuth is production (Max plan + billing); API key is dev/test fallback. OAuth wins when both + are set so the daemon never accidentally bills to the wrong + surface in deploys where both happen to be present.""" monkeypatch.setenv(API_KEY_ENV, "sk-ant-key") monkeypatch.setenv(OAUTH_TOKEN_ENV, "sk-ant-oat-token") engine = AnthropicReasoningEngine( system_prompt_path=system_prompt_path ) - assert engine._auth_mode == "api_key" + assert engine._auth_mode == "oauth_token" def test_whitespace_only_credential_is_treated_as_unset( diff --git a/tests/kora_cli/test_listeners/test_reasoning_engine_listener.py b/tests/kora_cli/test_listeners/test_reasoning_engine_listener.py new file mode 100644 index 000000000000..489819043a69 --- /dev/null +++ b/tests/kora_cli/test_listeners/test_reasoning_engine_listener.py @@ -0,0 +1,121 @@ +"""Tests for ``kora_cli.listeners.reasoning_engine_listener`` — ST2. + +Covers: + - current_reasoning_engine() returns None pre-startup + - startup with injected engine → singleton set + - shutdown → singleton cleared + - shutdown closes the underlying engine's HTTP client (best-effort) + - Production startup failure (missing creds) re-raises so the + daemon coordinator aborts boot (fail-CLOSED) + - Registered in LISTENER_REGISTRY under "reasoning_engine" +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from kora_cli.listeners import reasoning_engine_listener as rel +from kora_cli.listeners.reasoning_engine_listener import ( + ReasoningEngineListener, + _clear_singleton, + _set_singleton, + current_reasoning_engine, +) + + +@pytest.fixture(autouse=True) +def _reset_singleton(): + _clear_singleton() + yield + _clear_singleton() + + +@pytest.fixture(autouse=True) +def _clear_credential_envs(monkeypatch): + """Ensure neither credential is set unless a test sets it + explicitly — keeps the fail-CLOSED path testable.""" + monkeypatch.delenv("KORA_ANTHROPIC_API_KEY", raising=False) + monkeypatch.delenv("CLAUDE_CODE_OAUTH_TOKEN", raising=False) + + +@pytest.mark.asyncio +async def test_current_engine_is_none_before_startup(): + assert current_reasoning_engine() is None + + +@pytest.mark.asyncio +async def test_startup_with_injected_engine_sets_singleton(): + fake_engine = MagicMock() + fake_engine.close = AsyncMock() + listener = ReasoningEngineListener(engine=fake_engine) + await listener.startup() + assert current_reasoning_engine() is fake_engine + + +@pytest.mark.asyncio +async def test_shutdown_clears_singleton_and_closes_engine(): + fake_engine = MagicMock() + fake_engine.close = AsyncMock() + listener = ReasoningEngineListener(engine=fake_engine) + await listener.startup() + assert current_reasoning_engine() is fake_engine + + await listener.shutdown() + assert current_reasoning_engine() is None + fake_engine.close.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_shutdown_safe_when_close_raises(): + """An engine whose close() throws shouldn't propagate; we still + clear the singleton + return cleanly.""" + fake_engine = MagicMock() + fake_engine.close = AsyncMock(side_effect=RuntimeError("close boom")) + listener = ReasoningEngineListener(engine=fake_engine) + await listener.startup() + await listener.shutdown() # must not raise + assert current_reasoning_engine() is None + + +@pytest.mark.asyncio +async def test_shutdown_safe_without_startup(): + """Shutdown called without prior startup → no-op, no crash.""" + listener = ReasoningEngineListener() + await listener.shutdown() + + +@pytest.mark.asyncio +async def test_production_startup_reraises_on_misconfig(): + """No injected engine + both credential envs unset → production + startup tries to construct AnthropicReasoningEngine, which + raises ReasoningEngineNotConfigured. The listener re-raises so + the daemon coordinator aborts boot (fail-CLOSED per spec). + """ + from kora_cli.reasoning.anthropic_engine import ( + ReasoningEngineNotConfigured, + ) + + listener = ReasoningEngineListener() # no engine injected + with pytest.raises(ReasoningEngineNotConfigured): + await listener.startup() + + +def test_registered_in_listener_registry(): + """Module-import side effect: register_daemon_listener('reasoning_engine', _factory).""" + from kora_cli.daemon import LISTENER_REGISTRY + + names = [name for name, _factory in LISTENER_REGISTRY] + assert "reasoning_engine" in names + + +@pytest.mark.asyncio +async def test_set_and_clear_singleton_helpers(): + """The private helpers are used by the listener — verify they + do what they say.""" + fake = MagicMock() + _set_singleton(fake) + assert current_reasoning_engine() is fake + _clear_singleton() + assert current_reasoning_engine() is None