From eb7ebaebecb5e4b1f64f27b661ad28a444ba106f Mon Sep 17 00:00:00 2001 From: CC#3 Kora Runtime Date: Sat, 23 May 2026 22:14:32 -0700 Subject: [PATCH] =?UTF-8?q?feat(kora):=20KR-PLUGIN-COST-LADDER=20=E2=80=94?= =?UTF-8?q?=20first=20plugin=20extraction=20(cost-ladder)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First plugin extraction per the bucket spec. Moves cost-router code into the canonical Hermes plugin sub-directory shape; sets the template for KR-PLUGIN-AUDIT / KR-PLUGIN-CACHING / etc. follow-ons. Behavior: unchanged. KORA_REASONING_USE_GATEWAY stays default OFF; the bypass path + the route-through path both keep working. # Canonical location kora_cli/reasoning/kora_hermes_plugin/ __init__.py — re-exports KoraHermesPlugin + register plugin.py — orchestrator (6 hooks); delegates to sub-plugins cost_ladder/ __init__.py — re-exports cost-ladder public surface constants.py — DEFAULT_HAIKU_MODEL / regex defaults / rungs selector.py — RoutingDecision + select_model_pre_call + … plugin.py — pre_api_request_mutable hook handler + register # Backward-compat shims (pure re-exports; old import paths keep working) kora_cli/router/cost_router.py — 431 → 75 lines (shim only) plugins/kora_hermes/__init__.py — 527 → 70 lines (discovery shim) # Tests tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/test_selector.py 50 behavioral tests moved 1:1 from tests/kora_cli/router/test_cost_router.py; only the import path changed (now canonical). tests/kora_cli/router/test_cost_router.py — converted to a small 3-test shim-verification suite (asserts every public symbol still re-exports, via identity-against-canonical). # Scoped regression (xdist) tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/ 50 pass tests/kora_cli/router/ 3 pass tests/plugins/test_kora_hermes_plugin*.py 67 pass tests/kora_cli/snapshot/test_state_snapshot.py … pass tests/kora_cli/reasoning/test_anthropic_engine_router.py … pass Total directly-affected surface: 180/180 passing. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../reasoning/kora_hermes_plugin/__init__.py | 29 + .../cost_ladder/__init__.py | 55 ++ .../cost_ladder/constants.py | 69 +++ .../kora_hermes_plugin/cost_ladder/plugin.py | 185 ++++++ .../cost_ladder/selector.py | 346 +++++++++++ .../reasoning/kora_hermes_plugin/plugin.py | 329 +++++++++++ kora_cli/router/cost_router.py | 431 ++------------ plugins/kora_hermes/__init__.py | 527 ++--------------- .../reasoning/kora_hermes_plugin/__init__.py | 0 .../cost_ladder/__init__.py | 0 .../cost_ladder/test_selector.py | 468 +++++++++++++++ tests/kora_cli/router/test_cost_router.py | 547 ++++-------------- 12 files changed, 1714 insertions(+), 1272 deletions(-) create mode 100644 kora_cli/reasoning/kora_hermes_plugin/__init__.py create mode 100644 kora_cli/reasoning/kora_hermes_plugin/cost_ladder/__init__.py create mode 100644 kora_cli/reasoning/kora_hermes_plugin/cost_ladder/constants.py create mode 100644 kora_cli/reasoning/kora_hermes_plugin/cost_ladder/plugin.py create mode 100644 kora_cli/reasoning/kora_hermes_plugin/cost_ladder/selector.py create mode 100644 kora_cli/reasoning/kora_hermes_plugin/plugin.py create mode 100644 tests/kora_cli/reasoning/kora_hermes_plugin/__init__.py create mode 100644 tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/__init__.py create mode 100644 tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/test_selector.py diff --git a/kora_cli/reasoning/kora_hermes_plugin/__init__.py b/kora_cli/reasoning/kora_hermes_plugin/__init__.py new file mode 100644 index 000000000000..5af9dafbc3ff --- /dev/null +++ b/kora_cli/reasoning/kora_hermes_plugin/__init__.py @@ -0,0 +1,29 @@ +"""KoraHermesPlugin package — canonical Kora-side home for the +Hermes-plugin behaviors that drive Kora's reasoning route-through. + +Per KR-PLUGIN-COST-LADDER: this package is the new canonical +location for the Kora plugin code. The Hermes plugin discovery +entry at ``plugins/kora_hermes/`` re-exports from here so +``PluginManager.discover_and_load`` picks the plugin up while +the actual implementation lives in this importable Kora package +(future-proof for ``pip install kora-cost-ladder-plugin``-shape +distribution). + +Sub-plugins: + - ``cost_ladder/`` — first extraction (this bucket). Owns the + ``pre_api_request_mutable`` hook (model selection + cache + markers). + - (future) ``audit/`` — KR-PLUGIN-AUDIT will move audit emit + here (post_tool_call + post_llm_call audit JSONL writes) + - (future) ``caching/`` — KR-PLUGIN-CACHING will split the + caching half from cost_ladder + - (future) ``short_circuit/`` — KR-PLUGIN-SHORT-CIRCUIT + - (future) ``state_holders/`` — KR-PLUGIN-STATE-HOLDERS +""" + +from kora_cli.reasoning.kora_hermes_plugin.plugin import ( + KoraHermesPlugin, + register, +) + +__all__ = ["KoraHermesPlugin", "register"] diff --git a/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/__init__.py b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/__init__.py new file mode 100644 index 000000000000..c2025f335993 --- /dev/null +++ b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/__init__.py @@ -0,0 +1,55 @@ +"""Cost-ladder sub-plugin — default-Haiku with earned Opus escalation. + +See ``selector.py`` for the pure decision functions, +``constants.py`` for env-var names + defaults, ``plugin.py`` +for the ``pre_api_request_mutable`` hook handler + sub-register. + +Public surface (re-exported for both plugin discovery and the +backward-compat shim at ``kora_cli/router/cost_router.py``): +""" + +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.constants import ( + DEFAULT_DECISION_PATTERNS, + DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS, + DEFAULT_HAIKU_MODEL, + DEFAULT_OPUS_MODEL, + ENV_FORCE_OPUS, + ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, + ENV_OPUS_PREFIX, + ENV_OPUS_TRIGGER_PATTERNS, + RUNG_DOWNSHIFT_90, + RUNG_HARD_STOP_100, + RUNG_NORMAL, + RUNG_WARN_75, +) +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.plugin import ( + cost_ladder_and_caching_hook, + register, +) +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.selector import ( + RoutingDecision, + select_model_pre_call, + should_escalate_post_call, + strip_opus_prefix, +) + +__all__ = [ + "DEFAULT_DECISION_PATTERNS", + "DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS", + "DEFAULT_HAIKU_MODEL", + "DEFAULT_OPUS_MODEL", + "ENV_FORCE_OPUS", + "ENV_HAIKU_LOW_CONFIDENCE_PATTERNS", + "ENV_OPUS_PREFIX", + "ENV_OPUS_TRIGGER_PATTERNS", + "RUNG_DOWNSHIFT_90", + "RUNG_HARD_STOP_100", + "RUNG_NORMAL", + "RUNG_WARN_75", + "RoutingDecision", + "cost_ladder_and_caching_hook", + "register", + "select_model_pre_call", + "should_escalate_post_call", + "strip_opus_prefix", +] diff --git a/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/constants.py b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/constants.py new file mode 100644 index 000000000000..3d401ae3c5fd --- /dev/null +++ b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/constants.py @@ -0,0 +1,69 @@ +"""Constants for the cost-ladder plugin. + +Default model identifiers + env-var names + the bundled +decision-language / low-confidence pattern lists. Operators +override the patterns via env (see ``selector.py`` for the +``_compiled_*_patterns`` accessors). + +Moved verbatim from ``kora_cli.router.cost_router`` per +KR-PLUGIN-COST-LADDER; the shim at the old path re-exports +these symbols so existing call sites keep working. +""" + +from __future__ import annotations + +from typing import List + +# Match the long-form ID the engine has used since KR-FEAT-AGENTIC- +# REASONING ST1 (``MODEL_HAIKU = "claude-haiku-4-5-20251001"``). +# Using the long form keeps cache-key stability with already-warm +# caches in production. +DEFAULT_HAIKU_MODEL = "claude-haiku-4-5-20251001" +DEFAULT_OPUS_MODEL = "claude-opus-4-7" + + +# Env names — operator overrides. +ENV_FORCE_OPUS = "KORA_FORCE_OPUS" +ENV_OPUS_TRIGGER_PATTERNS = "KORA_OPUS_TRIGGER_PATTERNS" +ENV_OPUS_PREFIX = "KORA_OPUS_PREFIX" +ENV_HAIKU_LOW_CONFIDENCE_PATTERNS = "KORA_HAIKU_LOW_CONFIDENCE_PATTERNS" + + +# Default decision-language patterns (case-insensitive). Operator +# overrides via ``KORA_OPUS_TRIGGER_PATTERNS`` (comma-separated +# regex list). Tuned for the question shapes operators use when +# they actually want a careful answer rather than a status read. +DEFAULT_DECISION_PATTERNS: List[str] = [ + r"\b(should|do|can|will|would)\s+(i|we|you)\b", + r"\b(decide|decision|approve|approval|reject|go/no-go|ship or not)\b", + r"\bis it (better|worth|safe|right)\b", + r"\bwhat should (i|we)\b", + r"\b(plan|strategy|approach)\b.*\b(for|to)\b", +] + + +# Low-confidence markers in a Haiku response. Operator overrides +# via ``KORA_HAIKU_LOW_CONFIDENCE_PATTERNS``. +DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS: List[str] = [ + r"i'?m not (sure|certain|confident)", + r"i don'?t (have enough|know|understand)", + r"i might be wrong", + r"this is (a guess|just a guess|speculative)", + r"i can'?t (tell|determine|verify|confirm)", + r"unclear (whether|if|how)", +] + + +# Heuristic: a Haiku response that's clearly too short for a +# substantive question. Tunable inline; operator can edit constant +# in this module if they need to shift the threshold (no env for +# this one — it's a structural signal, not a phrase list). +_LOW_CONFIDENCE_SHORT_RESPONSE_THRESHOLD = 50 +_LOW_CONFIDENCE_LONG_INPUT_THRESHOLD = 200 + + +# Cost-rung literals (matching ``agent.cost_state_holder.CostRung``). +RUNG_NORMAL = "normal" +RUNG_WARN_75 = "warn_75" +RUNG_DOWNSHIFT_90 = "downshift_90" +RUNG_HARD_STOP_100 = "hard_stop_100" diff --git a/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/plugin.py b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/plugin.py new file mode 100644 index 000000000000..2b6c1f775c06 --- /dev/null +++ b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/plugin.py @@ -0,0 +1,185 @@ +"""Cost-ladder plugin module — hook handler + sub-register. + +The ``pre_api_request_mutable`` hook handler that drives Kora's +default-Haiku-with-earned-Opus model selection + the prompt- +caching ``cache_control: ephemeral`` markers on system + last +tool. Per KR-PLUGIN-COST-LADDER, this is the first plugin +extraction sub-module; the same pattern applies to follow-on +KR-PLUGIN-* buckets. + +# Bundled responsibility (transitional) + +Today's handler bundles cost-ladder model selection AND prompt- +caching cache_control wrapping. KR-PLUGIN-CACHING will extract +the caching half into its own ``caching/`` sub-plugin file. +Until then, both behaviors live here. The bundling is +intentional for v1: both fire at the same hook site +(``pre_api_request_mutable``), both build the same +``{"override": {...}}`` return shape, and splitting now would +add a second hook fire without behavior benefit. + +# Activation gate + +The handler short-circuits to ``None`` (no override) when the +call isn't a Kora-tagged route — protects Hermes-fork users who +load the plugin from inadvertently running Kora logic on their +sessions. Uses the same ``_is_kora_call`` predicate as the +top-level plugin's other hook handlers. +""" + +from __future__ import annotations + +import logging +from typing import Any, Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _current_cost_rung() -> str: + """Read the active cost-ladder rung. Defaults to ``"normal"`` + on any failure (holder unwired, exception, etc.) — matches + cost_router's expectation.""" + try: + from agent.cost_state_holder import get_cost_holder + + holder = get_cost_holder() + if holder is None: + return "normal" + rung = holder.active_rung() + return getattr(rung, "value", str(rung)) or "normal" + except Exception: + return "normal" + + +def _is_kora_call(route_value: Any) -> bool: + """Mirror of the top-level plugin's KORA_ROUTES gate. Imported + locally to avoid a circular import from + ``plugins.kora_hermes`` (the top-level Hermes-discovery + entry point); the route literal set is small + stable so the + duplication cost is low + the import isolation is worth it.""" + # Lazy import to break a potential circle. + from plugins.kora_hermes import KORA_ROUTES + + if not isinstance(route_value, str) or not route_value: + return False + return route_value in KORA_ROUTES + + +# --------------------------------------------------------------------------- +# Hook handler: pre_api_request_mutable +# --------------------------------------------------------------------------- + + +def cost_ladder_and_caching_hook( + *, + route: str = "", + api_kwargs: Optional[dict] = None, + api_call_count: int = 0, + user_message: str = "", + **kw, +) -> Optional[dict]: + """``pre_api_request_mutable`` handler. + + For Kora-tagged calls: + 1. Call the cost-router (``select_model_pre_call``) to + pick Haiku-default-or-Opus-earned based on iteration + + decision-language + cost rung + force-Opus env signals. + 2. Wrap ``system`` + ``tools`` with + ``cache_control: ephemeral`` markers so Anthropic + caches them (KR-CHEAP-PROMPT-CACHING semantic via the + hook layer rather than the bypass loop's inline wrap). + + Returns ``{"override": {...}}`` with the keys to replace in + api_kwargs. ``None`` (no-op) for non-Kora calls. + """ + if not _is_kora_call(route): + return None + + if not isinstance(api_kwargs, dict): + return None + + override: dict = {} + + # --- Cost-ladder model selection --- + # Use the (now-extracted) selector. Hermes's + # ``api_call_count`` is 1-indexed per-call; matches Kora's + # iteration count from the bypass loop. cost_rung comes from + # the process-global CostStateHolder. + try: + from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.selector import ( + select_model_pre_call, + ) + + cost_rung = _current_cost_rung() + decision = select_model_pre_call( + message_text=user_message or "", + iteration=max(int(api_call_count or 1), 1), + cost_rung=cost_rung, + ) + if decision.model is not None: + override["model"] = decision.model + except Exception as exc: + logger.warning( + "[kora_hermes.cost_ladder] select_model_pre_call raised " + "%r — leaving api_kwargs['model'] unchanged", + exc, + ) + + # --- Caching: wrap system + tools with cache_control markers --- + try: + from kora_cli.reasoning.anthropic_engine import ( + _wrap_system_as_cacheable, + _wrap_tools_as_cacheable, + ) + + # System: may be str (Hermes default) OR already a list + # (e.g. test fixture passed a content-block list). Wrap + # only the str case so we don't double-wrap. + existing_system = api_kwargs.get("system") + if isinstance(existing_system, str) and existing_system: + override["system"] = _wrap_system_as_cacheable(existing_system) + + # Tools: tools_for_api is a list (may be empty in + # toolless v1 route-through). The wrapper handles empty + # list by returning empty list — safe to always call. + existing_tools = api_kwargs.get("tools") or [] + if isinstance(existing_tools, list) and existing_tools: + override["tools"] = _wrap_tools_as_cacheable(existing_tools) + except Exception as exc: + logger.warning( + "[kora_hermes.cost_ladder] caching wrap raised %r — " + "leaving api_kwargs unchanged", + exc, + ) + + if not override: + return None + return {"override": override} + + +# --------------------------------------------------------------------------- +# Sub-register +# --------------------------------------------------------------------------- + + +def register(ctx) -> None: + """Sub-plugin register. Called by the top-level + ``KoraHermesPlugin.register`` to wire the cost-ladder hook. + + The top-level plugin owns plugin-context ownership; this + sub-register is just a thin function that calls + ``ctx.register_hook`` for the one hook this sub-plugin + owns. Keeps the file structure clean for the eventual + upstream PR (each sub-plugin is independently packageable). + """ + ctx.register_hook( + "pre_api_request_mutable", cost_ladder_and_caching_hook + ) + logger.debug( + "[kora_hermes.cost_ladder] sub-plugin registered" + ) diff --git a/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/selector.py b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/selector.py new file mode 100644 index 000000000000..8d87734e1ef7 --- /dev/null +++ b/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/selector.py @@ -0,0 +1,346 @@ +"""Cost-ladder selector — pure functions for model selection. + +Per Council R3 Lock R3-3 (the cost-ladder FLIP): the reasoning +engine no longer defaults to Opus with reactive downshift; it +defaults to Haiku 4.5 and escalates to Opus 4.7 **only when a +signal earns it**. The existing cost-ladder rung downshift stays +as a Layer 2 defensive backstop — when budget pressure crosses +WARN_75/DOWNSHIFT_90 the router clamps every call to Haiku +regardless of any earning signal. + +# Earning signals + +Pre-call (decided BEFORE the first API call): + + 1. **Operator override**: ``/opus`` prefix in the message text. + Case-insensitive. Stripped from the prompt before the SDK + call so the model doesn't see the routing instruction. + 2. **Operator defensive switch**: ``KORA_FORCE_OPUS=true`` env. + 3. **Decision-language regex**: message text matches a tunable + pattern list (``KORA_OPUS_TRIGGER_PATTERNS``). Defaults + cover the common decision-making phrases ("should I", "do I", + "decide", "approve", "go/no-go", etc.). + 4. **Tool-use iteration ≥ 2**: when the engine has already + dispatched tools once, the subsequent iterations get Opus. + This catches Haiku-started reasoning that needs more compute + after the model saw tool results. + +Post-call (decided AFTER the Haiku attempt on iteration 1): + + 5. **Haiku response low-confidence**: either an explicit + uncertainty marker ("I'm not sure", "I don't have enough", + etc.) OR a too-short response for a non-trivial input + (heuristic: ``len(reply) < 50`` when ``len(input) > 200``). + +# Layer 2: cost-ladder backstop + +The cost-rung input to ``select_model_pre_call`` short-circuits: + + - ``hard_stop_100`` → ``model=None`` (caller fail-fasts) + - ``warn_75`` / ``downshift_90`` → forced Haiku, ignoring any + earning signal. The escalation count still tracks what + WOULD have been Opus so the cockpit can show "cost-clamped" + decisions. + - ``normal`` (default) → router runs its full decision tree. + +# Telemetry contract + +Every routing decision is observable via the returned +``RoutingDecision.reason`` string. The engine emits per-call +``cost_telemetry.record_call(..., escalated_to_opus=...)`` so +PR #161's panels can show the Haiku-vs-Opus split + the +escalation rate per route. + +# What this is NOT + +This module is independent of ``agent/cost_downshift.py``. That +module governs **substrate Sea_Ticket queue** decisions +(defer vs run, criticality clamping) and runs on a different +code path than the slack-DM / email-inbound reasoning engine. +Both can co-exist; both name "downshift" in different domain +contexts (substrate-tier downshift vs LLM-tier downshift). + +# Module structure (KR-PLUGIN-COST-LADDER) + +This file holds pure functions only — no hook handlers, no +Hermes plugin integration. The hook handler lives in +``plugin.py``; the constants live in ``constants.py``. The +``register()`` entry on the top-level Hermes plugin +(``kora_hermes_plugin.plugin``) calls the sub-register here. + +The shim at ``kora_cli/router/cost_router.py`` re-exports +everything in this module verbatim so existing call sites +keep working. +""" + +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass +from typing import List, Optional, Tuple + +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.constants import ( + DEFAULT_DECISION_PATTERNS, + DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS, + DEFAULT_HAIKU_MODEL, + DEFAULT_OPUS_MODEL, + ENV_FORCE_OPUS, + ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, + ENV_OPUS_PREFIX, + ENV_OPUS_TRIGGER_PATTERNS, + RUNG_DOWNSHIFT_90, + RUNG_HARD_STOP_100, + RUNG_WARN_75, + _LOW_CONFIDENCE_LONG_INPUT_THRESHOLD, + _LOW_CONFIDENCE_SHORT_RESPONSE_THRESHOLD, +) + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# RoutingDecision +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class RoutingDecision: + """The router's per-iteration verdict. + + Attributes: + model: The model identifier to send. ``None`` when the + cost rung is HARD_STOP_100 — caller must fail-fast. + reason: Stable human-readable + telemetry-readable code. + Reasons: ``default_haiku`` / ``opus_prefix`` / + ``force_opus_env`` / ``decision_language`` / + ``tool_loop_iteration`` / ``cost_clamp:`` / + ``cost_ladder_halted`` / ``escalated_post_haiku:``. + escalated: ``True`` when the model is Opus AND the choice + was an "earning signal" path. Cost-clamps that landed on + Haiku set this False; explicit Opus signals that got + clamped to Haiku ALSO set False (the call IS Haiku). + haiku_context_for_opus: When a post-call escalation produces + a Decision wrapping Opus, this carries Haiku's response so + the caller can include it in the re-issue's messages. + None on every pre-call decision. + """ + + model: Optional[str] + reason: str + escalated: bool + haiku_context_for_opus: Optional[str] = None + + +# --------------------------------------------------------------------------- +# Pre-call decision +# --------------------------------------------------------------------------- + + +def _compiled_decision_patterns() -> List[re.Pattern[str]]: + """Compile decision-language patterns from env or default.""" + raw = os.environ.get(ENV_OPUS_TRIGGER_PATTERNS, "").strip() + sources = ( + [s for s in raw.split(",") if s.strip()] + if raw + else DEFAULT_DECISION_PATTERNS + ) + out: List[re.Pattern[str]] = [] + for src in sources: + try: + out.append(re.compile(src.strip(), re.IGNORECASE)) + except re.error as exc: + logger.warning( + "[kora.router] decision pattern %r invalid: %r — " + "skipping", + src, + exc, + ) + return out + + +def _compiled_low_confidence_patterns() -> List[re.Pattern[str]]: + raw = os.environ.get(ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, "").strip() + sources = ( + [s for s in raw.split(",") if s.strip()] + if raw + else DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS + ) + out: List[re.Pattern[str]] = [] + for src in sources: + try: + out.append(re.compile(src.strip(), re.IGNORECASE)) + except re.error as exc: + logger.warning( + "[kora.router] low-confidence pattern %r invalid: %r — " + "skipping", + src, + exc, + ) + return out + + +def _opus_prefix() -> str: + return os.environ.get(ENV_OPUS_PREFIX, "/opus").strip() or "/opus" + + +def strip_opus_prefix(message_text: str) -> str: + """Return ``message_text`` with the operator's Opus prefix + removed (if present). Strips leading whitespace + the prefix + + one trailing space. Idempotent on text without the prefix. + + The engine should call this on the inbound message text BEFORE + handing it to ``messages.create`` so the routing instruction + doesn't leak into the prompt and confuse the model. + """ + if not isinstance(message_text, str): + return message_text + prefix = _opus_prefix() + stripped = message_text.lstrip() + if stripped.lower().startswith(prefix.lower()): + # Strip prefix + at most one space after it. + remainder = stripped[len(prefix) :] + if remainder.startswith(" "): + remainder = remainder[1:] + return remainder + return message_text + + +def _has_opus_prefix(message_text: str) -> bool: + if not isinstance(message_text, str): + return False + prefix = _opus_prefix() + return message_text.lstrip().lower().startswith(prefix.lower()) + + +def select_model_pre_call( + *, + message_text: str, + iteration: int, + cost_rung: str, + force_opus_env: Optional[bool] = None, +) -> RoutingDecision: + """Pre-call model decision. + + Decision order (first match wins): + + 1. ``cost_rung == hard_stop_100`` → ``model=None``, + reason=``cost_ladder_halted``. + 2. ``cost_rung in {warn_75, downshift_90}`` → forced Haiku, + reason=``cost_clamp:``. The escalation flag is + False even if an earning signal also fired — the call + IS Haiku and that's what telemetry should reflect. + 3. ``force_opus_env`` → Opus, reason=``force_opus_env``. + 4. ``iteration >= 2`` → Opus, reason=``tool_loop_iteration``. + 5. Operator ``/opus`` prefix → Opus, reason=``opus_prefix``. + 6. Decision-language pattern match → Opus, + reason=``decision_language``. + 7. Default → Haiku, reason=``default_haiku``. + + Args: + message_text: The raw inbound text. Used for prefix + + pattern matching. Prefix detection is case-insensitive + and tolerates leading whitespace. + iteration: 1-indexed iteration count inside the tool-use + loop. Iteration 1 is the first API call. + cost_rung: One of the four CostRung literals. + force_opus_env: Override for the ``KORA_FORCE_OPUS`` env + check. ``None`` (default) → read the env. Explicit + bool used by tests to avoid env mutation. + """ + if cost_rung == RUNG_HARD_STOP_100: + return RoutingDecision( + model=None, + reason="cost_ladder_halted", + escalated=False, + ) + + if cost_rung in (RUNG_WARN_75, RUNG_DOWNSHIFT_90): + return RoutingDecision( + model=DEFAULT_HAIKU_MODEL, + reason=f"cost_clamp:{cost_rung}", + escalated=False, + ) + + if force_opus_env is None: + force_opus_env = ( + os.environ.get(ENV_FORCE_OPUS, "").strip().lower() == "true" + ) + if force_opus_env: + return RoutingDecision( + model=DEFAULT_OPUS_MODEL, + reason="force_opus_env", + escalated=True, + ) + + if iteration >= 2: + return RoutingDecision( + model=DEFAULT_OPUS_MODEL, + reason="tool_loop_iteration", + escalated=True, + ) + + if _has_opus_prefix(message_text): + return RoutingDecision( + model=DEFAULT_OPUS_MODEL, + reason="opus_prefix", + escalated=True, + ) + + for pattern in _compiled_decision_patterns(): + if pattern.search(message_text or ""): + return RoutingDecision( + model=DEFAULT_OPUS_MODEL, + reason="decision_language", + escalated=True, + ) + + return RoutingDecision( + model=DEFAULT_HAIKU_MODEL, + reason="default_haiku", + escalated=False, + ) + + +# --------------------------------------------------------------------------- +# Post-call escalation +# --------------------------------------------------------------------------- + + +def should_escalate_post_call( + *, + haiku_response_text: str, + original_message_text: str, +) -> Tuple[bool, str]: + """Return ``(should_escalate, reason)`` based on the Haiku + reply. Caller (engine's iteration 1 post-call hook) re-issues + to Opus with Haiku's response as context when this returns True. + + Reasons: + - ``low_confidence_marker`` — Haiku used an explicit + uncertainty phrase. + - ``short_response_for_long_input`` — Haiku gave a tiny + reply to a substantive question (heuristic). + - ``haiku_confident`` (escalate=False) — no signal fired. + """ + if not isinstance(haiku_response_text, str): + return (False, "haiku_confident") + + text = haiku_response_text.strip() + + # Length-heuristic FIRST so a long uncertainty-marker-containing + # response (which is actually fine — Haiku explained the + # uncertainty thoroughly) doesn't get short-response-tagged. + # Actually run BOTH; pattern match is cheaper + more specific. + for pattern in _compiled_low_confidence_patterns(): + if pattern.search(text): + return (True, "low_confidence_marker") + + if ( + len(original_message_text or "") > _LOW_CONFIDENCE_LONG_INPUT_THRESHOLD + and len(text) < _LOW_CONFIDENCE_SHORT_RESPONSE_THRESHOLD + ): + return (True, "short_response_for_long_input") + + return (False, "haiku_confident") diff --git a/kora_cli/reasoning/kora_hermes_plugin/plugin.py b/kora_cli/reasoning/kora_hermes_plugin/plugin.py new file mode 100644 index 000000000000..245941a57f12 --- /dev/null +++ b/kora_cli/reasoning/kora_hermes_plugin/plugin.py @@ -0,0 +1,329 @@ +"""Top-level KoraHermesPlugin — orchestrates sub-plugin registration. + +Per KR-PLUGIN-COST-LADDER: this is the canonical Kora-side home +for the orchestration logic. Each sub-plugin (cost_ladder/, +future audit/, future caching/, etc.) owns its own hook +handlers + ``register()`` function; the orchestrator just +calls each sub-register so the top-level Hermes plugin entry +(``plugins/kora_hermes/__init__.py``) is a thin shim. + +Sub-plugin extraction order (follow-on buckets): + 1. cost_ladder — **THIS BUCKET** (KR-PLUGIN-COST-LADDER, #182) + 2. audit — KR-PLUGIN-AUDIT (recommended next per CC#3 ST1) + 3. caching — KR-PLUGIN-CACHING (split caching from cost_ladder) + 4. short_circuit — KR-PLUGIN-SHORT-CIRCUIT + 5. state_holders — KR-PLUGIN-STATE-HOLDERS + +Until each sub-plugin extraction lands, the corresponding hook +handler lives in this orchestrator file (the +``_on_session_start`` / ``_pre_tool_list_finalized`` / +``_pre_tool_call`` / ``_post_tool_call`` / ``_post_llm_call`` + +the ST2B tool-bridge ``_tool_bridge_provide_result``). +""" + +from __future__ import annotations + +import logging +from typing import Any, Optional + +logger = logging.getLogger(__name__) + + +# --------------------------------------------------------------------------- +# KORA_ROUTES gate — re-exported from the discovery shim +# --------------------------------------------------------------------------- + + +# Kora route literals — matches ``kora_cli/telemetry/cost_telemetry. +# KNOWN_ROUTES`` (the existing telemetry vocabulary). When +# ``agent.route`` is in this set we treat the call as Kora- +# originated; else we no-op. Sourced as a literal to keep plugin +# discovery import-light (verified-in-sync by a pin test). +KORA_ROUTES = frozenset( + { + "slack_dm", + "email_inbound", + "email_outbound_compose", + "mcp_tool", + "alert_investigation", + "probe_investigation", + "tool_loop_iteration", + "scheduled_task", + } +) + + +def _is_kora_call(route_value: Any) -> bool: + """Return True when the route kwarg signals a Kora-originated call.""" + if not isinstance(route_value, str) or not route_value: + return False + return route_value in KORA_ROUTES + + +# --------------------------------------------------------------------------- +# Hook handlers that haven't been extracted to their own sub-plugin yet +# --------------------------------------------------------------------------- + + +def _on_session_start(*, route: str = "", **kw) -> None: + """Fires once when Hermes starts a conversation_loop session. + For Kora calls, ensures state holders are initialized. Today: + no-op (state holders init at daemon boot via DaemonCoordinator; + this hook may become relevant once KR-PLUGIN-STATE-HOLDERS + wires per-session Kora init).""" + if not _is_kora_call(route): + return + logger.debug( + "[kora_hermes] on_session_start fired for route=%s (no-op)", + route, + ) + + +def _pre_tool_list_finalized( + *, + route: str = "", + tools: Optional[list] = None, + **kw, +) -> Optional[dict]: + """KR-HERMES-LOCAL-EXTENSIONS hook. For Kora calls, filter the + tool list per-route. Today: no-op; future KR-PLUGIN-TOOL- + DESC-TRIM will plumb route-specific tool manifests.""" + if not _is_kora_call(route): + return None + logger.debug( + "[kora_hermes] pre_tool_list_finalized fired for route=%s " + "(no-op; future KR-PLUGIN-TOOL-DESC-TRIM will plumb)", + route, + ) + return None + + +def _pre_tool_call( + *, + tool_name: str = "", + args: Optional[dict] = None, + **kw, +) -> Optional[dict]: + """Constitution pre-screen (today lives in Kora-side + `_execute_single_tool_block` allowlist). Today: no-op; + KR-PLUGIN-CONSTITUTION will plumb KoraConstitution.evaluate_ + tool_call() once that extraction bucket lands.""" + route = kw.get("route", "") or "" + if not _is_kora_call(route): + return None + logger.debug( + "[kora_hermes] pre_tool_call fired for tool=%s route=%s " + "(no-op)", + tool_name, + route, + ) + return None + + +def _post_tool_call( + *, + tool_name: str = "", + result: Any = None, + **kw, +) -> None: + """Audit emit per tool call. Today: no-op (Kora's existing + audit runs inside `_execute_single_tool_block` in the bypass + loop; KR-PLUGIN-AUDIT will wire this to call ``_emit_audit`` + directly).""" + route = kw.get("route", "") or "" + if not _is_kora_call(route): + return + logger.debug( + "[kora_hermes] post_tool_call fired for tool=%s route=%s " + "(no-op)", + tool_name, + route, + ) + + +def _post_llm_call( + *, + route: str = "", + model: str = "", + **kw, +) -> None: + """Structured-log marker for the per-call cost-telemetry + timeline. Per-call ``CanonicalUsage`` accumulation lives at + the handler layer (slack_dm_handler's + ``_record_inference_to_cost_ladder``); KR-PLUGIN-AUDIT will + wire this hook to emit the audit JSONL row from the plugin + instead. + + Note: ``post_llm_call`` fires per CONVERSATION END (not per + API roundtrip). + """ + if not _is_kora_call(route): + return + logger.info( + "[kora.gateway.post_llm_call] route=%s model=%s", + route, + model or "", + ) + + +# --------------------------------------------------------------------------- +# KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B — tool bridge +# --------------------------------------------------------------------------- + + +def _is_kora_reasoning_tool(tool_name: str) -> bool: + """True iff ``tool_name`` is one of Kora's reasoning-allowlist + tools (``kora__*``). Import is lazy so plugin discovery + doesn't fault when the registry isn't importable in CI.""" + try: + from kora_cli.reasoning.tool_registry import REASONING_TOOL_ALLOWLIST + except Exception: + return False + return tool_name in REASONING_TOOL_ALLOWLIST + + +def _tool_bridge_provide_result( + *, + tool_name: str = "", + args: Optional[dict] = None, + **kw, +) -> Optional[dict]: + """Bridge handler for ``pre_tool_call_can_provide_result``. + + Intercepts dispatch for Kora's reasoning tools and returns + the result the kora_cli reasoning code would have produced + in the bypass loop. Hermes's default ``registry.dispatch`` + would otherwise fail (Kora's tools aren't registered as + Hermes tools). + + See ``plugins/kora_hermes/__init__.py`` history (pre- + KR-PLUGIN-COST-LADDER) for the full design + failure-mode + handling rationale. Behavior preserved verbatim by this + refactor. + """ + import asyncio + import json + + if not _is_kora_reasoning_tool(tool_name): + return None + + try: + from kora_cli.reasoning.tool_registry import execute_reasoning_tool + + result_model = asyncio.run( + execute_reasoning_tool(tool_name, args or {}) + ) + + if hasattr(result_model, "model_dump_json"): + result_str = result_model.model_dump_json() + else: + result_str = json.dumps(result_model, default=str) + + logger.debug( + "[kora_hermes.tool_bridge] dispatched %s via Kora registry " + "(result %d chars)", + tool_name, + len(result_str), + ) + return {"result": result_str} + except Exception as exc: + error_msg = ( + f"kora_tool_dispatch_error: {type(exc).__name__}: {exc!s}" + ) + logger.exception( + "[kora_hermes.tool_bridge] dispatch raised for %s", + tool_name, + ) + return {"result": json.dumps({"error": error_msg})} + + +def get_kora_tools_for_agent() -> list: + """Return Kora's reasoning tools in Hermes/OpenAI tool shape + for ``agent.tools`` population. Behavior preserved verbatim + by the KR-PLUGIN-COST-LADDER refactor.""" + try: + from kora_cli.reasoning.tool_registry import ( + get_reasoning_available_tools, + ) + + anthropic_tools = get_reasoning_available_tools() or [] + except Exception as exc: + logger.warning( + "[kora_hermes.tool_bridge] tool registry unavailable: %r " + "— agent.tools stays empty", + exc, + ) + return [] + + hermes_tools: list = [] + for tool in anthropic_tools: + try: + hermes_tools.append( + { + "type": "function", + "function": { + "name": tool["name"], + "description": tool.get("description", ""), + "parameters": tool.get("input_schema", {}), + }, + } + ) + except Exception as exc: + logger.warning( + "[kora_hermes.tool_bridge] tool %r conversion raised " + "%r — skipping", + tool.get("name", ""), + exc, + ) + return hermes_tools + + +# --------------------------------------------------------------------------- +# Top-level plugin class +# --------------------------------------------------------------------------- + + +class KoraHermesPlugin: + """Orchestrator that delegates to sub-plugins. + + Per KR-PLUGIN-COST-LADDER: each sub-plugin owns its hook + callbacks + sub-register. The orchestrator calls each sub- + register, then registers any remaining (not-yet-extracted) + handlers itself. Future extractions move handlers OUT of + this orchestrator INTO their own sub-plugin files. + """ + + def register(self, ctx) -> None: + # --- Sub-plugin registration (each owns its hooks) --- + # KR-PLUGIN-COST-LADDER (first extraction). + from kora_cli.reasoning.kora_hermes_plugin.cost_ladder import ( + register as register_cost_ladder, + ) + + register_cost_ladder(ctx) + + # --- Handlers still living in the orchestrator (await + # their own KR-PLUGIN-* extraction buckets) --- + ctx.register_hook("on_session_start", _on_session_start) + ctx.register_hook( + "pre_tool_list_finalized", _pre_tool_list_finalized + ) + ctx.register_hook("pre_tool_call", _pre_tool_call) + ctx.register_hook("post_tool_call", _post_tool_call) + ctx.register_hook("post_llm_call", _post_llm_call) + ctx.register_hook( + "pre_tool_call_can_provide_result", + _tool_bridge_provide_result, + ) + + logger.info( + "[kora_hermes] plugin registered: cost_ladder sub-plugin + " + "6 orchestrator-resident hooks against KORA_ROUTES=%s", + sorted(KORA_ROUTES), + ) + + +def register(ctx) -> None: + """Module-level register — what the Hermes discovery shim + calls. Instantiates the plugin + delegates to its register.""" + KoraHermesPlugin().register(ctx) diff --git a/kora_cli/router/cost_router.py b/kora_cli/router/cost_router.py index 781ca3b9ebc2..91ed69ac1a6e 100644 --- a/kora_cli/router/cost_router.py +++ b/kora_cli/router/cost_router.py @@ -1,378 +1,71 @@ -"""KR-HAIKU-ROUTER — default-Haiku model routing with earned Opus escalation. +"""Backward-compat shim — canonical location moved. -Per Council R3 Lock R3-3 (the cost-ladder FLIP): the reasoning -engine no longer defaults to Opus with reactive downshift; it -defaults to Haiku 4.5 and escalates to Opus 4.7 **only when a -signal earns it**. The existing cost-ladder rung downshift stays -as a Layer 2 defensive backstop — when budget pressure crosses -WARN_75/DOWNSHIFT_90 the router clamps every call to Haiku -regardless of any earning signal. +Per KR-PLUGIN-COST-LADDER: the cost-ladder code now lives at +``kora_cli/reasoning/kora_hermes_plugin/cost_ladder/`` (split +into ``selector.py`` for pure decision functions, +``constants.py`` for the model + env-var + pattern constants, +and ``plugin.py`` for the Hermes hook handler + sub-register). -# Earning signals +This file re-exports the public surface from the new location +so existing imports keep working: -Pre-call (decided BEFORE the first API call): + - ``from kora_cli.router.cost_router import select_model_pre_call`` + - ``from kora_cli.router.cost_router import DEFAULT_HAIKU_MODEL`` + - ``from kora_cli.router import ...`` (via the package's + ``__init__.py`` which imports from this module) - 1. **Operator override**: ``/opus`` prefix in the message text. - Case-insensitive. Stripped from the prompt before the SDK - call so the model doesn't see the routing instruction. - 2. **Operator defensive switch**: ``KORA_FORCE_OPUS=true`` env. - 3. **Decision-language regex**: message text matches a tunable - pattern list (``KORA_OPUS_TRIGGER_PATTERNS``). Defaults - cover the common decision-making phrases ("should I", "do I", - "decide", "approve", "go/no-go", etc.). - 4. **Tool-use iteration ≥ 2**: when the engine has already - dispatched tools once, the subsequent iterations get Opus. - This catches Haiku-started reasoning that needs more compute - after the model saw tool results. +New code should import from the canonical location directly: -Post-call (decided AFTER the Haiku attempt on iteration 1): + - ``from kora_cli.reasoning.kora_hermes_plugin.cost_ladder import + select_model_pre_call`` - 5. **Haiku response low-confidence**: either an explicit - uncertainty marker ("I'm not sure", "I don't have enough", - etc.) OR a too-short response for a non-trivial input - (heuristic: ``len(reply) < 50`` when ``len(input) > 200``). - -# Layer 2: cost-ladder backstop - -The cost-rung input to ``select_model_pre_call`` short-circuits: - - - ``hard_stop_100`` → ``model=None`` (caller fail-fasts) - - ``warn_75`` / ``downshift_90`` → forced Haiku, ignoring any - earning signal. The escalation count still tracks what - WOULD have been Opus so the cockpit can show "cost-clamped" - decisions. - - ``normal`` (default) → router runs its full decision tree. - -# Telemetry contract - -Every routing decision is observable via the returned -``RoutingDecision.reason`` string. The engine emits per-call -``cost_telemetry.record_call(..., escalated_to_opus=...)`` so -PR #161's panels can show the Haiku-vs-Opus split + the -escalation rate per route. - -# What this is NOT - -This module is independent of ``agent/cost_downshift.py``. That -module governs **substrate Sea_Ticket queue** decisions -(defer vs run, criticality clamping) and runs on a different -code path than the slack-DM / email-inbound reasoning engine. -Both can co-exist; both name "downshift" in different domain -contexts (substrate-tier downshift vs LLM-tier downshift). +The shim is retained indefinitely (other modules in the +kora_cli tree + tests at multiple paths import from here); +deprecation is not on the roadmap. """ -from __future__ import annotations - -import logging -import os -import re -from dataclasses import dataclass -from typing import List, Optional, Tuple - -logger = logging.getLogger(__name__) - - -# --------------------------------------------------------------------------- -# Defaults — overridable via env (see _compiled_*_patterns / DEFAULT_*) -# --------------------------------------------------------------------------- - - -# Match the long-form ID the engine has used since KR-FEAT-AGENTIC- -# REASONING ST1 (``MODEL_HAIKU = "claude-haiku-4-5-20251001"``). -# Using the long form keeps cache-key stability with already-warm -# caches in production. -DEFAULT_HAIKU_MODEL = "claude-haiku-4-5-20251001" -DEFAULT_OPUS_MODEL = "claude-opus-4-7" - - -# Env names — operator overrides. -ENV_FORCE_OPUS = "KORA_FORCE_OPUS" -ENV_OPUS_TRIGGER_PATTERNS = "KORA_OPUS_TRIGGER_PATTERNS" -ENV_OPUS_PREFIX = "KORA_OPUS_PREFIX" -ENV_HAIKU_LOW_CONFIDENCE_PATTERNS = "KORA_HAIKU_LOW_CONFIDENCE_PATTERNS" - - -# Default decision-language patterns (case-insensitive). Operator -# overrides via ``KORA_OPUS_TRIGGER_PATTERNS`` (comma-separated -# regex list). Tuned for the question shapes operators use when -# they actually want a careful answer rather than a status read. -DEFAULT_DECISION_PATTERNS: List[str] = [ - r"\b(should|do|can|will|would)\s+(i|we|you)\b", - r"\b(decide|decision|approve|approval|reject|go/no-go|ship or not)\b", - r"\bis it (better|worth|safe|right)\b", - r"\bwhat should (i|we)\b", - r"\b(plan|strategy|approach)\b.*\b(for|to)\b", -] - - -# Low-confidence markers in a Haiku response. Operator overrides -# via ``KORA_HAIKU_LOW_CONFIDENCE_PATTERNS``. -DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS: List[str] = [ - r"i'?m not (sure|certain|confident)", - r"i don'?t (have enough|know|understand)", - r"i might be wrong", - r"this is (a guess|just a guess|speculative)", - r"i can'?t (tell|determine|verify|confirm)", - r"unclear (whether|if|how)", +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.constants import ( + DEFAULT_DECISION_PATTERNS, + DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS, + DEFAULT_HAIKU_MODEL, + DEFAULT_OPUS_MODEL, + ENV_FORCE_OPUS, + ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, + ENV_OPUS_PREFIX, + ENV_OPUS_TRIGGER_PATTERNS, + RUNG_DOWNSHIFT_90, + RUNG_HARD_STOP_100, + RUNG_NORMAL, + RUNG_WARN_75, + _LOW_CONFIDENCE_LONG_INPUT_THRESHOLD, + _LOW_CONFIDENCE_SHORT_RESPONSE_THRESHOLD, +) +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.selector import ( + RoutingDecision, + _compiled_decision_patterns, + _compiled_low_confidence_patterns, + _has_opus_prefix, + _opus_prefix, + select_model_pre_call, + should_escalate_post_call, + strip_opus_prefix, +) + +__all__ = [ + "DEFAULT_DECISION_PATTERNS", + "DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS", + "DEFAULT_HAIKU_MODEL", + "DEFAULT_OPUS_MODEL", + "ENV_FORCE_OPUS", + "ENV_HAIKU_LOW_CONFIDENCE_PATTERNS", + "ENV_OPUS_PREFIX", + "ENV_OPUS_TRIGGER_PATTERNS", + "RUNG_DOWNSHIFT_90", + "RUNG_HARD_STOP_100", + "RUNG_NORMAL", + "RUNG_WARN_75", + "RoutingDecision", + "select_model_pre_call", + "should_escalate_post_call", + "strip_opus_prefix", ] - - -# Heuristic: a Haiku response that's clearly too short for a -# substantive question. Tunable inline; operator can edit constant -# in this module if they need to shift the threshold (no env for -# this one — it's a structural signal, not a phrase list). -_LOW_CONFIDENCE_SHORT_RESPONSE_THRESHOLD = 50 -_LOW_CONFIDENCE_LONG_INPUT_THRESHOLD = 200 - - -# Cost-rung literals (matching ``agent.cost_state_holder.CostRung``). -RUNG_NORMAL = "normal" -RUNG_WARN_75 = "warn_75" -RUNG_DOWNSHIFT_90 = "downshift_90" -RUNG_HARD_STOP_100 = "hard_stop_100" - - -# --------------------------------------------------------------------------- -# RoutingDecision -# --------------------------------------------------------------------------- - - -@dataclass(frozen=True) -class RoutingDecision: - """The router's per-iteration verdict. - - Attributes: - model: The model identifier to send. ``None`` when the - cost rung is HARD_STOP_100 — caller must fail-fast. - reason: Stable human-readable + telemetry-readable code. - Reasons: ``default_haiku`` / ``opus_prefix`` / - ``force_opus_env`` / ``decision_language`` / - ``tool_loop_iteration`` / ``cost_clamp:`` / - ``cost_ladder_halted`` / ``escalated_post_haiku:``. - escalated: ``True`` when the model is Opus AND the choice - was an "earning signal" path. Cost-clamps that landed on - Haiku set this False; explicit Opus signals that got - clamped to Haiku ALSO set False (the call IS Haiku). - haiku_context_for_opus: When a post-call escalation produces - a Decision wrapping Opus, this carries Haiku's response so - the caller can include it in the re-issue's messages. - None on every pre-call decision. - """ - - model: Optional[str] - reason: str - escalated: bool - haiku_context_for_opus: Optional[str] = None - - -# --------------------------------------------------------------------------- -# Pre-call decision -# --------------------------------------------------------------------------- - - -def _compiled_decision_patterns() -> List[re.Pattern[str]]: - """Compile decision-language patterns from env or default.""" - raw = os.environ.get(ENV_OPUS_TRIGGER_PATTERNS, "").strip() - sources = ( - [s for s in raw.split(",") if s.strip()] - if raw - else DEFAULT_DECISION_PATTERNS - ) - out: List[re.Pattern[str]] = [] - for src in sources: - try: - out.append(re.compile(src.strip(), re.IGNORECASE)) - except re.error as exc: - logger.warning( - "[kora.router] decision pattern %r invalid: %r — " - "skipping", - src, - exc, - ) - return out - - -def _compiled_low_confidence_patterns() -> List[re.Pattern[str]]: - raw = os.environ.get(ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, "").strip() - sources = ( - [s for s in raw.split(",") if s.strip()] - if raw - else DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS - ) - out: List[re.Pattern[str]] = [] - for src in sources: - try: - out.append(re.compile(src.strip(), re.IGNORECASE)) - except re.error as exc: - logger.warning( - "[kora.router] low-confidence pattern %r invalid: %r — " - "skipping", - src, - exc, - ) - return out - - -def _opus_prefix() -> str: - return os.environ.get(ENV_OPUS_PREFIX, "/opus").strip() or "/opus" - - -def strip_opus_prefix(message_text: str) -> str: - """Return ``message_text`` with the operator's Opus prefix - removed (if present). Strips leading whitespace + the prefix - + one trailing space. Idempotent on text without the prefix. - - The engine should call this on the inbound message text BEFORE - handing it to ``messages.create`` so the routing instruction - doesn't leak into the prompt and confuse the model. - """ - if not isinstance(message_text, str): - return message_text - prefix = _opus_prefix() - stripped = message_text.lstrip() - if stripped.lower().startswith(prefix.lower()): - # Strip prefix + at most one space after it. - remainder = stripped[len(prefix) :] - if remainder.startswith(" "): - remainder = remainder[1:] - return remainder - return message_text - - -def _has_opus_prefix(message_text: str) -> bool: - if not isinstance(message_text, str): - return False - prefix = _opus_prefix() - return message_text.lstrip().lower().startswith(prefix.lower()) - - -def select_model_pre_call( - *, - message_text: str, - iteration: int, - cost_rung: str, - force_opus_env: Optional[bool] = None, -) -> RoutingDecision: - """Pre-call model decision. - - Decision order (first match wins): - - 1. ``cost_rung == hard_stop_100`` → ``model=None``, - reason=``cost_ladder_halted``. - 2. ``cost_rung in {warn_75, downshift_90}`` → forced Haiku, - reason=``cost_clamp:``. The escalation flag is - False even if an earning signal also fired — the call - IS Haiku and that's what telemetry should reflect. - 3. ``force_opus_env`` → Opus, reason=``force_opus_env``. - 4. ``iteration >= 2`` → Opus, reason=``tool_loop_iteration``. - 5. Operator ``/opus`` prefix → Opus, reason=``opus_prefix``. - 6. Decision-language pattern match → Opus, - reason=``decision_language``. - 7. Default → Haiku, reason=``default_haiku``. - - Args: - message_text: The raw inbound text. Used for prefix + - pattern matching. Prefix detection is case-insensitive - and tolerates leading whitespace. - iteration: 1-indexed iteration count inside the tool-use - loop. Iteration 1 is the first API call. - cost_rung: One of the four CostRung literals. - force_opus_env: Override for the ``KORA_FORCE_OPUS`` env - check. ``None`` (default) → read the env. Explicit - bool used by tests to avoid env mutation. - """ - if cost_rung == RUNG_HARD_STOP_100: - return RoutingDecision( - model=None, - reason="cost_ladder_halted", - escalated=False, - ) - - if cost_rung in (RUNG_WARN_75, RUNG_DOWNSHIFT_90): - return RoutingDecision( - model=DEFAULT_HAIKU_MODEL, - reason=f"cost_clamp:{cost_rung}", - escalated=False, - ) - - if force_opus_env is None: - force_opus_env = ( - os.environ.get(ENV_FORCE_OPUS, "").strip().lower() == "true" - ) - if force_opus_env: - return RoutingDecision( - model=DEFAULT_OPUS_MODEL, - reason="force_opus_env", - escalated=True, - ) - - if iteration >= 2: - return RoutingDecision( - model=DEFAULT_OPUS_MODEL, - reason="tool_loop_iteration", - escalated=True, - ) - - if _has_opus_prefix(message_text): - return RoutingDecision( - model=DEFAULT_OPUS_MODEL, - reason="opus_prefix", - escalated=True, - ) - - for pattern in _compiled_decision_patterns(): - if pattern.search(message_text or ""): - return RoutingDecision( - model=DEFAULT_OPUS_MODEL, - reason="decision_language", - escalated=True, - ) - - return RoutingDecision( - model=DEFAULT_HAIKU_MODEL, - reason="default_haiku", - escalated=False, - ) - - -# --------------------------------------------------------------------------- -# Post-call escalation -# --------------------------------------------------------------------------- - - -def should_escalate_post_call( - *, - haiku_response_text: str, - original_message_text: str, -) -> Tuple[bool, str]: - """Return ``(should_escalate, reason)`` based on the Haiku - reply. Caller (engine's iteration 1 post-call hook) re-issues - to Opus with Haiku's response as context when this returns True. - - Reasons: - - ``low_confidence_marker`` — Haiku used an explicit - uncertainty phrase. - - ``short_response_for_long_input`` — Haiku gave a tiny - reply to a substantive question (heuristic). - - ``haiku_confident`` (escalate=False) — no signal fired. - """ - if not isinstance(haiku_response_text, str): - return (False, "haiku_confident") - - text = haiku_response_text.strip() - - # Length-heuristic FIRST so a long uncertainty-marker-containing - # response (which is actually fine — Haiku explained the - # uncertainty thoroughly) doesn't get short-response-tagged. - # Actually run BOTH; pattern match is cheaper + more specific. - for pattern in _compiled_low_confidence_patterns(): - if pattern.search(text): - return (True, "low_confidence_marker") - - if ( - len(original_message_text or "") > _LOW_CONFIDENCE_LONG_INPUT_THRESHOLD - and len(text) < _LOW_CONFIDENCE_SHORT_RESPONSE_THRESHOLD - ): - return (True, "short_response_for_long_input") - - return (False, "haiku_confident") diff --git a/plugins/kora_hermes/__init__.py b/plugins/kora_hermes/__init__.py index c07dfa676f30..8ded173c8612 100644 --- a/plugins/kora_hermes/__init__.py +++ b/plugins/kora_hermes/__init__.py @@ -1,468 +1,69 @@ -"""KR-REASONING-ROUTE-THROUGH-GATEWAY-CORE ST1 — Kora behaviors as a Hermes plugin. - -This bundled plugin registers Kora's reasoning behaviors as Hermes -hook callbacks so the (forthcoming) gateway-route-through code path -exercises them via the standard Hermes plugin contract instead of -the bypass loop in ``kora_cli/reasoning/anthropic_engine.py``. - -# Scope (ST1) - -This file is intentionally a **dispatch shim** — the hook handlers -do not move logic out of ``kora_cli/*`` modules. They check whether -the active call is a Kora call (``agent.route`` in KORA_ROUTES) and -delegate to the existing Kora module code when it is, no-op when -it isn't. KR-PLUGIN-COST-LADDER / KR-PLUGIN-AUDIT / etc. follow-on -buckets extract clean plugin bodies; this bucket just wires the -discovery + dispatch surface. - -# Activation gate - -Every hook handler short-circuits to no-op when ``agent.route`` is -empty or not in :data:`KORA_ROUTES`. This protects every Hermes -CLI / Gateway user from inadvertently running Kora logic on their -sessions — the plugin is harmless to ship as a bundled package -even on a non-Kora Hermes deployment. - -# ST2 follow-on - -Hook handler bodies currently delegate to ``kora_hermes_plugin`` -in ``kora_cli/reasoning/``. ST2 (KR-PLUGIN-COST-LADDER first per -the bucket's recommendation) will move logic out of router / -short_circuit / etc. modules into clean plugin files. +"""Hermes-plugin-discovery entry point for the Kora plugin. + +Per KR-PLUGIN-COST-LADDER: the canonical Kora plugin code lives +at ``kora_cli/reasoning/kora_hermes_plugin/``. This file is the +**thin shim** Hermes's ``PluginManager.discover_and_load`` finds +under the bundled ``plugins//`` convention. It re-exports +the public surface from the canonical location so any test or +external consumer that imports ``plugins.kora_hermes`` keeps +working. + +Why the split: + - Hermes plugin discovery requires a ``plugins//`` + directory with a ``plugin.yaml`` manifest + ``__init__.py`` + with a ``register(ctx)`` function. That's the bundled- + plugin convention. + - Future ``pip install kora-cost-ladder-plugin`` distribution + requires the plugin code be a normal importable Python + package — that lives under ``kora_cli/reasoning/ + kora_hermes_plugin/`` so it's clean to extract from this + repo into its own distribution when the time comes. + +This shim re-exports the orchestrator + sub-plugin helpers so +tests at ``tests/plugins/test_kora_hermes_plugin*.py`` (and any +external consumer that learned the public surface at the old +location) keep working unchanged. """ -from __future__ import annotations - -import logging -from typing import Any, Optional - -logger = logging.getLogger(__name__) - - -# Kora route literals — matches ``kora_cli/telemetry/cost_telemetry.KNOWN_ROUTES`` -# (the existing telemetry vocabulary). When ``agent.route`` is in this -# set we treat the call as Kora-originated; else we no-op. -# -# Sourced as a literal list to keep the plugin discovery side import- -# light (avoids pulling in the telemetry module at plugin-load time); -# verified-in-sync via tests/plugins/test_kora_hermes_plugin.py. -KORA_ROUTES = frozenset( - { - "slack_dm", - "email_inbound", - "email_outbound_compose", - "mcp_tool", - "alert_investigation", - "probe_investigation", - "tool_loop_iteration", - "scheduled_task", - } +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.plugin import ( + _current_cost_rung, ) +# Backward-compat alias for the pre-extraction handler name. +# Tests at ``tests/plugins/test_kora_hermes_plugin*.py`` import +# ``_pre_api_request_mutable`` from this module; the renamed +# handler lives in the cost_ladder sub-plugin now. +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder.plugin import ( + cost_ladder_and_caching_hook as _pre_api_request_mutable, +) +from kora_cli.reasoning.kora_hermes_plugin.plugin import ( + KORA_ROUTES, + KoraHermesPlugin, + _is_kora_call, + _is_kora_reasoning_tool, + _on_session_start, + _post_llm_call, + _post_tool_call, + _pre_tool_call, + _pre_tool_list_finalized, + _tool_bridge_provide_result, + get_kora_tools_for_agent, + register, +) -def _is_kora_call(route_value: Any) -> bool: - """Return True when the route kwarg signals a Kora-originated call.""" - if not isinstance(route_value, str) or not route_value: - return False - return route_value in KORA_ROUTES - - -# --------------------------------------------------------------------------- -# Hook handlers — all are dispatch shims to ``kora_hermes_plugin`` module -# --------------------------------------------------------------------------- - - -def _on_session_start(*, route: str = "", **kw) -> None: - """Fires once when Hermes starts a conversation_loop session. - For Kora calls, ensures state holders are initialized. ST1: - no-op (state holders init at daemon boot via DaemonCoordinator; - this hook may become relevant once ST2 wires per-session - Kora init).""" - if not _is_kora_call(route): - return - logger.debug( - "[kora_hermes] on_session_start fired for route=%s (ST1 no-op)", - route, - ) - - -def _pre_api_request_mutable( - *, - route: str = "", - api_kwargs: Optional[dict] = None, - api_call_count: int = 0, - user_message: str = "", - **kw, -) -> Optional[dict]: - """KR-HERMES-LOCAL-EXTENSIONS hook — ST2 real wiring. - - For Kora-tagged calls: - 1. Call the cost-router (``select_model_pre_call``) to - pick Haiku-default-or-Opus-earned based on iteration + - decision-language + cost rung + force-Opus env signals. - 2. Wrap ``system`` + ``tools`` with - ``cache_control: ephemeral`` markers so Anthropic - caches them (KR-CHEAP-PROMPT-CACHING semantic via the - hook layer rather than the bypass loop's inline wrap). - - Returns ``{"override": {...}}`` with the keys to replace in - api_kwargs. None / no-op for non-Kora calls. - """ - if not _is_kora_call(route): - return None - - if not isinstance(api_kwargs, dict): - return None - - override: dict = {} - - # --- Cost-ladder model selection --- - # Use the existing router. iteration semantics: Hermes's - # ``api_call_count`` is 1-indexed per-call; matches Kora's - # iteration count from the bypass loop. cost_rung comes from - # the process-global CostStateHolder. - try: - from kora_cli.router import select_model_pre_call - - cost_rung = _current_cost_rung() - decision = select_model_pre_call( - message_text=user_message or "", - iteration=max(int(api_call_count or 1), 1), - cost_rung=cost_rung, - ) - if decision.model is not None: - override["model"] = decision.model - except Exception as exc: - logger.warning( - "[kora_hermes] cost-ladder select_model_pre_call raised " - "%r — leaving api_kwargs['model'] unchanged", - exc, - ) - - # --- Caching: wrap system + tools with cache_control markers --- - try: - from kora_cli.reasoning.anthropic_engine import ( - _wrap_system_as_cacheable, - _wrap_tools_as_cacheable, - ) - - # System: may be str (Hermes default) OR already a list - # (e.g. test fixture passed a content-block list). Wrap - # only the str case so we don't double-wrap. - existing_system = api_kwargs.get("system") - if isinstance(existing_system, str) and existing_system: - override["system"] = _wrap_system_as_cacheable(existing_system) - - # Tools: tools_for_api is a list (may be empty in - # toolless v1 route-through). The wrapper handles empty - # list by returning empty list — safe to always call. - existing_tools = api_kwargs.get("tools") or [] - if isinstance(existing_tools, list) and existing_tools: - override["tools"] = _wrap_tools_as_cacheable(existing_tools) - except Exception as exc: - logger.warning( - "[kora_hermes] caching wrap raised %r — leaving " - "api_kwargs unchanged", - exc, - ) - - if not override: - return None - return {"override": override} - - -def _current_cost_rung() -> str: - """Read the active cost-ladder rung. Defaults to ``"normal"`` - on any failure (holder unwired, exception, etc.) — matches - cost_router's expectation.""" - try: - from agent.cost_state_holder import get_cost_holder - - holder = get_cost_holder() - if holder is None: - return "normal" - rung = holder.active_rung() - return getattr(rung, "value", str(rung)) or "normal" - except Exception: - return "normal" - - -def _pre_tool_list_finalized( - *, - route: str = "", - tools: Optional[list] = None, - **kw, -) -> Optional[dict]: - """KR-HERMES-LOCAL-EXTENSIONS hook. For Kora calls, filter the - tool list per-route. ST1: no-op.""" - if not _is_kora_call(route): - return None - logger.debug( - "[kora_hermes] pre_tool_list_finalized fired for route=%s " - "(ST1 no-op; future KR-PLUGIN-TOOL-DESC-TRIM will plumb)", - route, - ) - return None - - -def _pre_tool_call( - *, - tool_name: str = "", - args: Optional[dict] = None, - **kw, -) -> Optional[dict]: - """Constitution pre-screen (today lives in Kora-side - `_execute_single_tool_block` allowlist). ST1: no-op; ST2 plumbs - KoraConstitution.evaluate_tool_call() once that extraction - bucket lands.""" - route = kw.get("route", "") or "" - if not _is_kora_call(route): - return None - logger.debug( - "[kora_hermes] pre_tool_call fired for tool=%s route=%s " - "(ST1 no-op)", - tool_name, - route, - ) - return None - - -def _post_tool_call( - *, - tool_name: str = "", - result: Any = None, - **kw, -) -> None: - """Audit emit per tool call. ST1: no-op (Kora's existing audit - runs inside `_execute_single_tool_block` in the bypass loop; - ST2 wires this to call `_emit_audit` directly).""" - route = kw.get("route", "") or "" - if not _is_kora_call(route): - return - logger.debug( - "[kora_hermes] post_tool_call fired for tool=%s route=%s " - "(ST1 no-op)", - tool_name, - route, - ) - - -def _post_llm_call( - *, - route: str = "", - model: str = "", - **kw, -) -> None: - """ST2 real wiring — per-call cost-telemetry record_call. - - Records the call to the telemetry counter so the cockpit's - cost panel shows Kora's route-through spend alongside the - bypass path's spend (same telemetry vocabulary; the route - discriminator is the only diff). - - Note: ``post_llm_call`` fires per CONVERSATION END (not per - API roundtrip). The conversation-loop accumulates tokens - across iterations; the response_text + final usage is what - we see here. For per-iteration accounting (which the bypass - loop does via ``_record_call_to_telemetry`` per call) we'd - need a different hook or the existing ``post_api_request`` - observer — out of scope for ST2 (ST2B follow-on). - """ - if not _is_kora_call(route): - return - - # Hermes's post_llm_call kwargs don't include a usage object - # (the assistant_response is a string). For ST2 the cost- - # ladder write happens at the handler layer when the - # ResponseResult comes back; this hook is a structured-log - # marker for the telemetry timeline. Per-call CanonicalUsage - # accumulation lives in the handler (slack_dm_handler's - # ``_record_inference_to_cost_ladder``). - logger.info( - "[kora.gateway.post_llm_call] route=%s model=%s", - route, - model or "", - ) - - -# --------------------------------------------------------------------------- -# KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B — tool bridge -# --------------------------------------------------------------------------- - - -def _is_kora_reasoning_tool(tool_name: str) -> bool: - """True iff ``tool_name`` is one of Kora's reasoning-allowlist - tools (``kora__*``). Import is lazy so plugin discovery - doesn't fault when the registry isn't importable in CI.""" - try: - from kora_cli.reasoning.tool_registry import REASONING_TOOL_ALLOWLIST - except Exception: - return False - return tool_name in REASONING_TOOL_ALLOWLIST - - -def _tool_bridge_provide_result( - *, - tool_name: str = "", - args: Optional[dict] = None, - **kw, -) -> Optional[dict]: - """Bridge handler for ``pre_tool_call_can_provide_result``. - - Intercepts dispatch for Kora's reasoning tools and returns - the result the kora_cli reasoning code would have produced - in the bypass loop. Hermes's default ``registry.dispatch`` - would otherwise fail (Kora's tools aren't registered as - Hermes tools). - - Returns: - * ``{"result": }`` when the tool IS in Kora's - allowlist + dispatch succeeded → short-circuits Hermes - * ``{"result": }`` when the tool IS - in Kora's allowlist BUT dispatch raised → short-circuits - Hermes with an is_error result so the reasoning loop - sees the error rather than getting Hermes's - "tool not found" envelope. - * ``None`` when the tool isn't a Kora tool OR the call - isn't a Kora-route call → falls through to other plugins - or Hermes default. Non-Kora-route safety: Hermes-fork - users with this plugin loaded see no behavior change on - their own (non-Kora) sessions. - - Implementation note: ``handle_function_call`` is sync; - ``execute_reasoning_tool`` is async. We bridge via - ``asyncio.run`` when no loop is running, OR via - ``asyncio.new_event_loop`` + ``run_until_complete`` when - nested under an existing loop (the daemon path runs - ``handle_function_call`` inside an ``asyncio.to_thread`` - call from ``_respond_via_gateway`` — the thread has no - running loop, so ``asyncio.run`` is the right primitive). - """ - import asyncio - import json - - route = kw.get("route", "") or "" - # ``handle_function_call`` doesn't currently forward - # ``route`` to the hook (the kwargs it passes are tool_name, - # args, task_id, session_id, tool_call_id). For ST2B v1 we - # gate on the tool name's belonging to Kora's allowlist - # alone — this is the cleaner check anyway since the tool - # name uniquely identifies whether Kora can serve it. - # Future ST2C: thread ``route`` through the hook kwargs so - # we can also restrict to Kora routes (defense-in-depth). - - if not _is_kora_reasoning_tool(tool_name): - return None - - # Dispatch via Kora's reasoning tool registry. - try: - from kora_cli.reasoning.tool_registry import execute_reasoning_tool - - result_model = asyncio.run( - execute_reasoning_tool(tool_name, args or {}) - ) - - # Project the Pydantic model into a JSON string. Models - # have ``model_dump_json`` per the existing registry's - # contract; fall back to ``str()`` if not Pydantic. - if hasattr(result_model, "model_dump_json"): - result_str = result_model.model_dump_json() - else: - result_str = json.dumps(result_model, default=str) - - logger.debug( - "[kora_hermes.tool_bridge] dispatched %s via Kora registry " - "(result %d chars)", - tool_name, - len(result_str), - ) - return {"result": result_str} - except Exception as exc: - # Convert dispatch failure to an is_error tool_result so - # the reasoning loop can see the error rather than crash. - # Match Kora's existing bypass-loop error envelope shape - # (the JSON-serializable dict with ``error`` key, matching - # what Hermes's _sanitize_tool_error produces on its own - # dispatch errors). - error_msg = ( - f"kora_tool_dispatch_error: {type(exc).__name__}: {exc!s}" - ) - logger.exception( - "[kora_hermes.tool_bridge] dispatch raised for %s", - tool_name, - ) - return {"result": json.dumps({"error": error_msg})} - - -def get_kora_tools_for_agent() -> list: - """Return Kora's reasoning tools in Hermes/OpenAI tool shape - for ``agent.tools`` population. The kora_cli registry stores - Anthropic-shaped descriptors (``{"name", "description", - "input_schema"}``); Hermes's ``agent.tools`` reads - ``tool["function"]["name"]`` (OpenAI shape) at multiple - sites. We convert here so consumers (e.g. - ``_respond_via_gateway``) get the right shape. - - Returns ``[]`` on any error (registry unavailable, schema - drift) — engine falls back to toolless route-through. - """ - try: - from kora_cli.reasoning.tool_registry import ( - get_reasoning_available_tools, - ) - - anthropic_tools = get_reasoning_available_tools() or [] - except Exception as exc: - logger.warning( - "[kora_hermes.tool_bridge] tool registry unavailable: %r " - "— agent.tools stays empty", - exc, - ) - return [] - - hermes_tools: list = [] - for tool in anthropic_tools: - try: - hermes_tools.append( - { - "type": "function", - "function": { - "name": tool["name"], - "description": tool.get("description", ""), - "parameters": tool.get("input_schema", {}), - }, - } - ) - except Exception as exc: - logger.warning( - "[kora_hermes.tool_bridge] tool %r conversion raised " - "%r — skipping", - tool.get("name", ""), - exc, - ) - return hermes_tools - - -# --------------------------------------------------------------------------- -# Plugin entry point — called once at plugin discovery -# --------------------------------------------------------------------------- - - -def register(ctx) -> None: - """Plugin entry. Called by ``PluginManager.discover_and_load`` - once at process startup. Registers Kora behaviors against the - 7 Hermes hooks Kora's reasoning path needs (post-ST2B).""" - ctx.register_hook("on_session_start", _on_session_start) - ctx.register_hook("pre_api_request_mutable", _pre_api_request_mutable) - ctx.register_hook("pre_tool_list_finalized", _pre_tool_list_finalized) - ctx.register_hook("pre_tool_call", _pre_tool_call) - ctx.register_hook("post_tool_call", _post_tool_call) - ctx.register_hook("post_llm_call", _post_llm_call) - # KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B — tool-bridge hook. - ctx.register_hook( - "pre_tool_call_can_provide_result", _tool_bridge_provide_result - ) - logger.info( - "[kora_hermes] plugin registered: 7 hooks against KORA_ROUTES=%s", - sorted(KORA_ROUTES), - ) +__all__ = [ + "KORA_ROUTES", + "KoraHermesPlugin", + "_current_cost_rung", + "_is_kora_call", + "_is_kora_reasoning_tool", + "_on_session_start", + "_post_llm_call", + "_post_tool_call", + "_pre_api_request_mutable", + "_pre_tool_call", + "_pre_tool_list_finalized", + "_tool_bridge_provide_result", + "get_kora_tools_for_agent", + "register", +] diff --git a/tests/kora_cli/reasoning/kora_hermes_plugin/__init__.py b/tests/kora_cli/reasoning/kora_hermes_plugin/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/__init__.py b/tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/test_selector.py b/tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/test_selector.py new file mode 100644 index 000000000000..99a6dcc14ce5 --- /dev/null +++ b/tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/test_selector.py @@ -0,0 +1,468 @@ +"""Unit tests for the cost-ladder selector module. + +Per KR-PLUGIN-COST-LADDER (PR #182): moved from +``tests/kora_cli/router/test_cost_router.py`` to mirror the +canonical code location at ``kora_cli/reasoning/ +kora_hermes_plugin/cost_ladder/selector.py``. All 50 test +cases preserved verbatim; only the import path updated to use +the canonical location. + +The backward-compat shim at ``kora_cli/router/cost_router.py`` +keeps the old import path working for downstream consumers; +asserted in the now-small ``tests/kora_cli/router/test_cost_ +router.py`` (shim-verification suite). + +Covers KR-HAIKU-ROUTER (PR #165) §2 acceptance: + - Default → Haiku (no earning signal) + - /opus prefix → Opus, prefix stripped from prompt + - KORA_FORCE_OPUS env → Opus regardless of other signals + - Iteration ≥ 2 → Opus (the iteration earning signal) + - Decision-language regex match → Opus + - HARD_STOP_100 → model=None, caller fail-fast + - DOWNSHIFT_90 + Opus signal → STILL Haiku (cost backstop) + - WARN_75 + Opus signal → STILL Haiku + - Post-call low-confidence markers → escalate + - Post-call short response for long input → escalate + - Post-call confident response → no escalate + - strip_opus_prefix is idempotent on non-prefix text + - Operator env override of trigger patterns + - Operator env override of low-confidence patterns + - Operator env override of /opus prefix string +""" + +from __future__ import annotations + +import pytest + +from kora_cli.reasoning.kora_hermes_plugin.cost_ladder import ( + DEFAULT_DECISION_PATTERNS, + DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS, + DEFAULT_HAIKU_MODEL, + DEFAULT_OPUS_MODEL, + ENV_FORCE_OPUS, + ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, + ENV_OPUS_PREFIX, + ENV_OPUS_TRIGGER_PATTERNS, + RUNG_DOWNSHIFT_90, + RUNG_HARD_STOP_100, + RUNG_NORMAL, + RUNG_WARN_75, + RoutingDecision, + select_model_pre_call, + should_escalate_post_call, + strip_opus_prefix, +) + + +@pytest.fixture(autouse=True) +def _clean_env(monkeypatch): + """Strip every router env so each test gets the bundled + defaults (unless the test explicitly sets one).""" + for var in ( + ENV_FORCE_OPUS, + ENV_OPUS_TRIGGER_PATTERNS, + ENV_OPUS_PREFIX, + ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, + ): + monkeypatch.delenv(var, raising=False) + yield + + +# --------------------------------------------------------------------------- +# Pre-call decision — default + earning signals +# --------------------------------------------------------------------------- + + +def test_default_returns_haiku(): + d = select_model_pre_call( + message_text="anything goes", + iteration=1, + cost_rung=RUNG_NORMAL, + ) + assert d.model == DEFAULT_HAIKU_MODEL + assert d.reason == "default_haiku" + assert d.escalated is False + + +def test_opus_prefix_escalates(): + d = select_model_pre_call( + message_text="/opus tell me everything", + iteration=1, + cost_rung=RUNG_NORMAL, + ) + assert d.model == DEFAULT_OPUS_MODEL + assert d.reason == "opus_prefix" + assert d.escalated is True + + +def test_opus_prefix_case_insensitive(): + d = select_model_pre_call( + message_text="/OPUS heavy lift", iteration=1, cost_rung=RUNG_NORMAL + ) + assert d.model == DEFAULT_OPUS_MODEL + assert d.reason == "opus_prefix" + + +def test_opus_prefix_tolerates_leading_whitespace(): + d = select_model_pre_call( + message_text=" /opus heavy", iteration=1, cost_rung=RUNG_NORMAL + ) + assert d.reason == "opus_prefix" + + +def test_force_opus_env_overrides_default(monkeypatch): + monkeypatch.setenv(ENV_FORCE_OPUS, "true") + d = select_model_pre_call( + message_text="anything", + iteration=1, + cost_rung=RUNG_NORMAL, + ) + assert d.model == DEFAULT_OPUS_MODEL + assert d.reason == "force_opus_env" + assert d.escalated is True + + +def test_force_opus_env_explicit_arg_overrides_env_read(): + """Tests pass explicit `force_opus_env=True` even when env isn't + set; this keeps tests env-free.""" + d = select_model_pre_call( + message_text="anything", + iteration=1, + cost_rung=RUNG_NORMAL, + force_opus_env=True, + ) + assert d.model == DEFAULT_OPUS_MODEL + + +def test_iteration_two_escalates(): + d = select_model_pre_call( + message_text="hi", iteration=2, cost_rung=RUNG_NORMAL + ) + assert d.model == DEFAULT_OPUS_MODEL + assert d.reason == "tool_loop_iteration" + assert d.escalated is True + + +def test_iteration_three_escalates(): + """Same signal as iteration 2 — covers the >= 2 inclusive check.""" + d = select_model_pre_call( + message_text="hi", iteration=3, cost_rung=RUNG_NORMAL + ) + assert d.model == DEFAULT_OPUS_MODEL + assert d.reason == "tool_loop_iteration" + + +@pytest.mark.parametrize( + "text", + [ + "should i ship this?", + "do we approve the migration?", + "what should i do about the alert?", + "is it safe to merge?", + "decide whether to scale up", + "go/no-go on the deploy", + "plan strategy for next quarter", + ], +) +def test_decision_language_patterns_escalate(text): + d = select_model_pre_call( + message_text=text, iteration=1, cost_rung=RUNG_NORMAL + ) + assert d.model == DEFAULT_OPUS_MODEL, ( + f"decision-language phrase {text!r} should have escalated" + ) + assert d.reason == "decision_language" + + +@pytest.mark.parametrize( + "text", + [ + "hi", + "thanks", + "what's my burn", + "how are you", + "any alerts?", + "ok cool", + ], +) +def test_non_decision_text_stays_haiku(text): + d = select_model_pre_call( + message_text=text, iteration=1, cost_rung=RUNG_NORMAL + ) + assert d.model == DEFAULT_HAIKU_MODEL, ( + f"non-decision text {text!r} should have stayed Haiku" + ) + + +# --------------------------------------------------------------------------- +# Cost-rung backstop (Layer 2) +# --------------------------------------------------------------------------- + + +def test_hard_stop_returns_none_model(): + d = select_model_pre_call( + message_text="anything", + iteration=1, + cost_rung=RUNG_HARD_STOP_100, + ) + assert d.model is None + assert d.reason == "cost_ladder_halted" + assert d.escalated is False + + +def test_downshift_90_clamps_to_haiku_even_with_opus_signal(): + """The cost backstop wins over an earning signal.""" + d = select_model_pre_call( + message_text="/opus heavy lift", + iteration=1, + cost_rung=RUNG_DOWNSHIFT_90, + ) + assert d.model == DEFAULT_HAIKU_MODEL + assert d.reason == "cost_clamp:downshift_90" + assert d.escalated is False + + +def test_downshift_90_clamps_even_on_iteration_two(): + d = select_model_pre_call( + message_text="hi", + iteration=2, + cost_rung=RUNG_DOWNSHIFT_90, + ) + assert d.model == DEFAULT_HAIKU_MODEL + assert d.reason == "cost_clamp:downshift_90" + + +def test_warn_75_clamps_to_haiku_even_with_decision_language(): + d = select_model_pre_call( + message_text="should i ship this?", + iteration=1, + cost_rung=RUNG_WARN_75, + ) + assert d.model == DEFAULT_HAIKU_MODEL + assert d.reason == "cost_clamp:warn_75" + + +def test_warn_75_clamps_even_with_force_opus_env(monkeypatch): + """Cost backstop wins over even the operator's defensive + KORA_FORCE_OPUS — the rung is the LAST line of defense and + not overridable from app-layer signals.""" + monkeypatch.setenv(ENV_FORCE_OPUS, "true") + d = select_model_pre_call( + message_text="anything", + iteration=1, + cost_rung=RUNG_WARN_75, + ) + assert d.model == DEFAULT_HAIKU_MODEL + + +# --------------------------------------------------------------------------- +# /opus prefix stripping +# --------------------------------------------------------------------------- + + +def test_strip_opus_prefix_removes_prefix(): + assert strip_opus_prefix("/opus what's up") == "what's up" + + +def test_strip_opus_prefix_case_insensitive(): + assert strip_opus_prefix("/OPUS heavy") == "heavy" + + +def test_strip_opus_prefix_idempotent_on_non_prefix_text(): + assert strip_opus_prefix("hi there") == "hi there" + + +def test_strip_opus_prefix_tolerates_leading_whitespace(): + assert strip_opus_prefix(" /opus heavy") == "heavy" + + +def test_strip_opus_prefix_non_string_passthrough(): + assert strip_opus_prefix(None) is None # type: ignore[arg-type] + assert strip_opus_prefix(42) == 42 # type: ignore[arg-type] + + +def test_strip_opus_prefix_respects_env_override(monkeypatch): + monkeypatch.setenv(ENV_OPUS_PREFIX, "!think") + assert strip_opus_prefix("!think hard") == "hard" + assert strip_opus_prefix("/opus hard") == "/opus hard" + + +# --------------------------------------------------------------------------- +# Post-call escalation +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "haiku_text", + [ + "I'm not sure about that.", + "im not sure honestly", + "I don't know what to tell you.", + "I might be wrong here.", + "this is a guess but try X.", + "I can't determine that.", + "Unclear whether this is the right path.", + ], +) +def test_post_call_low_confidence_marker_escalates(haiku_text): + should, reason = should_escalate_post_call( + haiku_response_text=haiku_text, + original_message_text="anything", + ) + assert should is True + assert reason == "low_confidence_marker" + + +def test_post_call_short_response_for_long_input_escalates(): + long_input = "x" * 250 + should, reason = should_escalate_post_call( + haiku_response_text="ok.", + original_message_text=long_input, + ) + assert should is True + assert reason == "short_response_for_long_input" + + +def test_post_call_confident_short_response_for_short_input_does_not_escalate(): + """Short input + short reply is fine — terse Q gets terse A.""" + should, reason = should_escalate_post_call( + haiku_response_text="ok.", + original_message_text="ok?", + ) + assert should is False + assert reason == "haiku_confident" + + +def test_post_call_confident_long_response_does_not_escalate(): + should, reason = should_escalate_post_call( + haiku_response_text=( + "Here's the full plan: step A, step B, step C. " + "Each is independently verifiable. " + "Rollback at step B is the safest cutover point." + ), + original_message_text="walk me through the plan", + ) + assert should is False + assert reason == "haiku_confident" + + +def test_post_call_non_string_does_not_escalate(): + should, reason = should_escalate_post_call( + haiku_response_text=None, # type: ignore[arg-type] + original_message_text="anything", + ) + assert should is False + + +def test_post_call_low_confidence_pattern_env_override(monkeypatch): + """Operator can swap in a custom low-confidence vocabulary.""" + monkeypatch.setenv( + ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, + r"this is suss,unclear", + ) + should, reason = should_escalate_post_call( + haiku_response_text="this is suss", + original_message_text="anything", + ) + assert should is True + # Old default markers no longer trigger. + should2, _ = should_escalate_post_call( + haiku_response_text="I'm not sure about that", + original_message_text="anything", + ) + assert should2 is False + + +# --------------------------------------------------------------------------- +# Decision-language env override +# --------------------------------------------------------------------------- + + +def test_decision_pattern_env_override(monkeypatch): + """Operator can shrink or expand the decision-language list.""" + monkeypatch.setenv(ENV_OPUS_TRIGGER_PATTERNS, r"\bcritical\b") + d = select_model_pre_call( + message_text="this is critical", + iteration=1, + cost_rung=RUNG_NORMAL, + ) + assert d.model == DEFAULT_OPUS_MODEL + assert d.reason == "decision_language" + + # An old default pattern that's no longer active. + d2 = select_model_pre_call( + message_text="should i ship this?", + iteration=1, + cost_rung=RUNG_NORMAL, + ) + assert d2.model == DEFAULT_HAIKU_MODEL + + +def test_decision_pattern_invalid_regex_skipped(monkeypatch, caplog): + """A bad pattern in the env list is logged + skipped; others + still work.""" + import logging + + caplog.set_level(logging.WARNING) + monkeypatch.setenv( + ENV_OPUS_TRIGGER_PATTERNS, r"[unclosed,\bcritical\b" + ) + d = select_model_pre_call( + message_text="this is critical", + iteration=1, + cost_rung=RUNG_NORMAL, + ) + assert d.model == DEFAULT_OPUS_MODEL + warns = [r.getMessage() for r in caplog.records if r.levelname == "WARNING"] + assert any("invalid" in w for w in warns) + + +# --------------------------------------------------------------------------- +# RoutingDecision shape +# --------------------------------------------------------------------------- + + +def test_routing_decision_is_frozen(): + d = RoutingDecision(model="x", reason="r", escalated=False) + with pytest.raises((AttributeError, TypeError)): + d.model = "y" # type: ignore[misc] + + +def test_routing_decision_default_haiku_context_is_none(): + d = RoutingDecision(model="x", reason="r", escalated=False) + assert d.haiku_context_for_opus is None + + +# --------------------------------------------------------------------------- +# Earning-signal precedence — Q4 boundaries +# --------------------------------------------------------------------------- + + +def test_force_opus_env_beats_default_haiku_but_loses_to_cost_clamp( + monkeypatch, +): + """Precedence: hard_stop > cost_clamp > force_opus > iteration + > prefix > decision_language > default. This test pins the + boundary between force_opus and cost_clamp.""" + monkeypatch.setenv(ENV_FORCE_OPUS, "true") + # Normal rung — force wins. + d = select_model_pre_call( + message_text="hi", iteration=1, cost_rung=RUNG_NORMAL + ) + assert d.model == DEFAULT_OPUS_MODEL + # warn_75 — cost clamp wins. + d2 = select_model_pre_call( + message_text="hi", iteration=1, cost_rung=RUNG_WARN_75 + ) + assert d2.model == DEFAULT_HAIKU_MODEL + + +def test_iteration_two_beats_opus_prefix(): + """Both signals would pick Opus; iteration check fires first + in the decision tree (matches the spec's pseudocode order + putting iteration check before prefix check).""" + d = select_model_pre_call( + message_text="/opus heavy", iteration=2, cost_rung=RUNG_NORMAL + ) + assert d.model == DEFAULT_OPUS_MODEL + assert d.reason == "tool_loop_iteration" diff --git a/tests/kora_cli/router/test_cost_router.py b/tests/kora_cli/router/test_cost_router.py index 99f3bd7b5cef..dec927b99939 100644 --- a/tests/kora_cli/router/test_cost_router.py +++ b/tests/kora_cli/router/test_cost_router.py @@ -1,456 +1,123 @@ -"""Unit tests for KR-HAIKU-ROUTER cost_router module. - -Covers spec §2 acceptance: - - Default → Haiku (no earning signal) - - /opus prefix → Opus, prefix stripped from prompt - - KORA_FORCE_OPUS env → Opus regardless of other signals - - Iteration ≥ 2 → Opus (the iteration earning signal) - - Decision-language regex match → Opus (every default pattern) - - HARD_STOP_100 → model=None, caller fail-fast - - DOWNSHIFT_90 + Opus signal → STILL Haiku (cost backstop) - - WARN_75 + Opus signal → STILL Haiku - - Post-call low-confidence markers → escalate - - Post-call short response for long input → escalate - - Post-call confident response → no escalate - - strip_opus_prefix is idempotent on non-prefix text - - Operator env override of trigger patterns - - Operator env override of low-confidence patterns - - Operator env override of /opus prefix string +"""Shim-verification suite for ``kora_cli/router/cost_router.py``. + +The 50 behavioral tests moved to +``tests/kora_cli/reasoning/kora_hermes_plugin/cost_ladder/ +test_selector.py`` in KR-PLUGIN-COST-LADDER (PR #182) — mirrors +the canonical code location at +``kora_cli/reasoning/kora_hermes_plugin/cost_ladder/selector.py``. + +This file remains as a small assertion suite that the +backward-compat shim at ``kora_cli/router/cost_router.py`` still +re-exports the cost-ladder public surface so downstream callers +(any external module that has historically imported from +``kora_cli.router``) keep working without modification. + +If this file ever fails, the shim has drifted from the canonical +location — fix by re-exporting the missing symbol in +``kora_cli/router/cost_router.py``. """ from __future__ import annotations -import pytest - -from kora_cli.router import ( - DEFAULT_DECISION_PATTERNS, - DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS, - DEFAULT_HAIKU_MODEL, - DEFAULT_OPUS_MODEL, - ENV_FORCE_OPUS, - ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, - ENV_OPUS_PREFIX, - ENV_OPUS_TRIGGER_PATTERNS, - RUNG_DOWNSHIFT_90, - RUNG_HARD_STOP_100, - RUNG_NORMAL, - RUNG_WARN_75, - RoutingDecision, - select_model_pre_call, - should_escalate_post_call, - strip_opus_prefix, -) - - -@pytest.fixture(autouse=True) -def _clean_env(monkeypatch): - """Strip every router env so each test gets the bundled - defaults (unless the test explicitly sets one).""" - for var in ( - ENV_FORCE_OPUS, - ENV_OPUS_TRIGGER_PATTERNS, - ENV_OPUS_PREFIX, - ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, - ): - monkeypatch.delenv(var, raising=False) - yield - - -# --------------------------------------------------------------------------- -# Pre-call decision — default + earning signals -# --------------------------------------------------------------------------- - - -def test_default_returns_haiku(): - d = select_model_pre_call( - message_text="anything goes", - iteration=1, - cost_rung=RUNG_NORMAL, - ) - assert d.model == DEFAULT_HAIKU_MODEL - assert d.reason == "default_haiku" - assert d.escalated is False - - -def test_opus_prefix_escalates(): - d = select_model_pre_call( - message_text="/opus tell me everything", - iteration=1, - cost_rung=RUNG_NORMAL, - ) - assert d.model == DEFAULT_OPUS_MODEL - assert d.reason == "opus_prefix" - assert d.escalated is True - - -def test_opus_prefix_case_insensitive(): - d = select_model_pre_call( - message_text="/OPUS heavy lift", iteration=1, cost_rung=RUNG_NORMAL - ) - assert d.model == DEFAULT_OPUS_MODEL - assert d.reason == "opus_prefix" - - -def test_opus_prefix_tolerates_leading_whitespace(): - d = select_model_pre_call( - message_text=" /opus heavy", iteration=1, cost_rung=RUNG_NORMAL - ) - assert d.reason == "opus_prefix" - - -def test_force_opus_env_overrides_default(monkeypatch): - monkeypatch.setenv(ENV_FORCE_OPUS, "true") - d = select_model_pre_call( - message_text="anything", - iteration=1, - cost_rung=RUNG_NORMAL, - ) - assert d.model == DEFAULT_OPUS_MODEL - assert d.reason == "force_opus_env" - assert d.escalated is True - - -def test_force_opus_env_explicit_arg_overrides_env_read(): - """Tests pass explicit `force_opus_env=True` even when env isn't - set; this keeps tests env-free.""" - d = select_model_pre_call( - message_text="anything", - iteration=1, - cost_rung=RUNG_NORMAL, - force_opus_env=True, - ) - assert d.model == DEFAULT_OPUS_MODEL - - -def test_iteration_two_escalates(): - d = select_model_pre_call( - message_text="hi", iteration=2, cost_rung=RUNG_NORMAL - ) - assert d.model == DEFAULT_OPUS_MODEL - assert d.reason == "tool_loop_iteration" - assert d.escalated is True - - -def test_iteration_three_escalates(): - """Same signal as iteration 2 — covers the >= 2 inclusive check.""" - d = select_model_pre_call( - message_text="hi", iteration=3, cost_rung=RUNG_NORMAL - ) - assert d.model == DEFAULT_OPUS_MODEL - assert d.reason == "tool_loop_iteration" - - -@pytest.mark.parametrize( - "text", - [ - "should i ship this?", - "do we approve the migration?", - "what should i do about the alert?", - "is it safe to merge?", - "decide whether to scale up", - "go/no-go on the deploy", - "plan strategy for next quarter", - ], -) -def test_decision_language_patterns_escalate(text): - d = select_model_pre_call( - message_text=text, iteration=1, cost_rung=RUNG_NORMAL - ) - assert d.model == DEFAULT_OPUS_MODEL, ( - f"decision-language phrase {text!r} should have escalated" - ) - assert d.reason == "decision_language" - - -@pytest.mark.parametrize( - "text", - [ - "hi", - "thanks", - "what's my burn", - "how are you", - "any alerts?", - "ok cool", - ], -) -def test_non_decision_text_stays_haiku(text): - d = select_model_pre_call( - message_text=text, iteration=1, cost_rung=RUNG_NORMAL - ) - assert d.model == DEFAULT_HAIKU_MODEL, ( - f"non-decision text {text!r} should have stayed Haiku" - ) - - -# --------------------------------------------------------------------------- -# Cost-rung backstop (Layer 2) -# --------------------------------------------------------------------------- - -def test_hard_stop_returns_none_model(): - d = select_model_pre_call( - message_text="anything", +def test_shim_reexports_full_public_surface(): + """The shim must re-export every symbol the historical + ``kora_cli.router`` (and ``kora_cli.router.cost_router``) + public surface advertised. Asserted by import + identity- + against-canonical: each shim attr is the SAME object as the + canonical module's attr (not a separately-imported copy).""" + from kora_cli.reasoning.kora_hermes_plugin.cost_ladder import ( + constants as canonical_constants, + selector as canonical_selector, + ) + from kora_cli.router import cost_router as shim + + # Constants from constants.py + for name in [ + "DEFAULT_HAIKU_MODEL", + "DEFAULT_OPUS_MODEL", + "ENV_FORCE_OPUS", + "ENV_OPUS_TRIGGER_PATTERNS", + "ENV_OPUS_PREFIX", + "ENV_HAIKU_LOW_CONFIDENCE_PATTERNS", + "DEFAULT_DECISION_PATTERNS", + "DEFAULT_HAIKU_LOW_CONFIDENCE_PATTERNS", + "RUNG_NORMAL", + "RUNG_WARN_75", + "RUNG_DOWNSHIFT_90", + "RUNG_HARD_STOP_100", + "_LOW_CONFIDENCE_SHORT_RESPONSE_THRESHOLD", + "_LOW_CONFIDENCE_LONG_INPUT_THRESHOLD", + ]: + assert getattr(shim, name) is getattr(canonical_constants, name), ( + f"shim drift: {name} not re-exported (or is a copy, " + f"not a reference) — fix kora_cli/router/cost_router.py" + ) + + # Functions + dataclass from selector.py + for name in [ + "RoutingDecision", + "_compiled_decision_patterns", + "_compiled_low_confidence_patterns", + "_opus_prefix", + "_has_opus_prefix", + "strip_opus_prefix", + "select_model_pre_call", + "should_escalate_post_call", + ]: + assert getattr(shim, name) is getattr(canonical_selector, name), ( + f"shim drift: {name} not re-exported — fix " + f"kora_cli/router/cost_router.py" + ) + + +def test_package_level_reexport_still_works(): + """``from kora_cli.router import select_model_pre_call`` (the + package-level import path used by ``plugins/kora_hermes/__init__.py`` + and external callers) must keep resolving.""" + from kora_cli.router import ( + DEFAULT_HAIKU_MODEL, + RoutingDecision, + select_model_pre_call, + should_escalate_post_call, + strip_opus_prefix, + ) + + decision = select_model_pre_call( + message_text="hello", iteration=1, - cost_rung=RUNG_HARD_STOP_100, + cost_rung="normal", + force_opus_env=False, ) - assert d.model is None - assert d.reason == "cost_ladder_halted" - assert d.escalated is False + assert isinstance(decision, RoutingDecision) + assert decision.model == DEFAULT_HAIKU_MODEL + assert decision.reason == "default_haiku" + assert decision.escalated is False - -def test_downshift_90_clamps_to_haiku_even_with_opus_signal(): - """The cost backstop wins over an earning signal.""" - d = select_model_pre_call( - message_text="/opus heavy lift", - iteration=1, - cost_rung=RUNG_DOWNSHIFT_90, - ) - assert d.model == DEFAULT_HAIKU_MODEL - assert d.reason == "cost_clamp:downshift_90" - assert d.escalated is False - - -def test_downshift_90_clamps_even_on_iteration_two(): - d = select_model_pre_call( - message_text="hi", - iteration=2, - cost_rung=RUNG_DOWNSHIFT_90, + # should_escalate_post_call + strip_opus_prefix still resolve. + escalate, reason = should_escalate_post_call( + haiku_response_text="The answer is 42.", + original_message_text="What is the answer?", ) - assert d.model == DEFAULT_HAIKU_MODEL - assert d.reason == "cost_clamp:downshift_90" - - -def test_warn_75_clamps_to_haiku_even_with_decision_language(): - d = select_model_pre_call( - message_text="should i ship this?", - iteration=1, - cost_rung=RUNG_WARN_75, - ) - assert d.model == DEFAULT_HAIKU_MODEL - assert d.reason == "cost_clamp:warn_75" - - -def test_warn_75_clamps_even_with_force_opus_env(monkeypatch): - """Cost backstop wins over even the operator's defensive - KORA_FORCE_OPUS — the rung is the LAST line of defense and - not overridable from app-layer signals.""" - monkeypatch.setenv(ENV_FORCE_OPUS, "true") - d = select_model_pre_call( - message_text="anything", - iteration=1, - cost_rung=RUNG_WARN_75, - ) - assert d.model == DEFAULT_HAIKU_MODEL - - -# --------------------------------------------------------------------------- -# /opus prefix stripping -# --------------------------------------------------------------------------- - - -def test_strip_opus_prefix_removes_prefix(): - assert strip_opus_prefix("/opus what's up") == "what's up" - - -def test_strip_opus_prefix_case_insensitive(): - assert strip_opus_prefix("/OPUS heavy") == "heavy" - - -def test_strip_opus_prefix_idempotent_on_non_prefix_text(): - assert strip_opus_prefix("hi there") == "hi there" - - -def test_strip_opus_prefix_tolerates_leading_whitespace(): - assert strip_opus_prefix(" /opus heavy") == "heavy" - - -def test_strip_opus_prefix_non_string_passthrough(): - assert strip_opus_prefix(None) is None # type: ignore[arg-type] - assert strip_opus_prefix(42) == 42 # type: ignore[arg-type] - - -def test_strip_opus_prefix_respects_env_override(monkeypatch): - monkeypatch.setenv(ENV_OPUS_PREFIX, "!think") - assert strip_opus_prefix("!think hard") == "hard" - assert strip_opus_prefix("/opus hard") == "/opus hard" - - -# --------------------------------------------------------------------------- -# Post-call escalation -# --------------------------------------------------------------------------- - - -@pytest.mark.parametrize( - "haiku_text", - [ - "I'm not sure about that.", - "im not sure honestly", - "I don't know what to tell you.", - "I might be wrong here.", - "this is a guess but try X.", - "I can't determine that.", - "Unclear whether this is the right path.", - ], -) -def test_post_call_low_confidence_marker_escalates(haiku_text): - should, reason = should_escalate_post_call( - haiku_response_text=haiku_text, - original_message_text="anything", - ) - assert should is True - assert reason == "low_confidence_marker" - - -def test_post_call_short_response_for_long_input_escalates(): - long_input = "x" * 250 - should, reason = should_escalate_post_call( - haiku_response_text="ok.", - original_message_text=long_input, - ) - assert should is True - assert reason == "short_response_for_long_input" - - -def test_post_call_confident_short_response_for_short_input_does_not_escalate(): - """Short input + short reply is fine — terse Q gets terse A.""" - should, reason = should_escalate_post_call( - haiku_response_text="ok.", - original_message_text="ok?", - ) - assert should is False + assert escalate is False assert reason == "haiku_confident" + assert strip_opus_prefix("/opus do the thing") == "do the thing" -def test_post_call_confident_long_response_does_not_escalate(): - should, reason = should_escalate_post_call( - haiku_response_text=( - "Here's the full plan: step A, step B, step C. " - "Each is independently verifiable. " - "Rollback at step B is the safest cutover point." - ), - original_message_text="walk me through the plan", +def test_canonical_path_is_the_authoritative_source(): + """Sanity: the canonical path is the place where the public + symbols are DEFINED (``__module__`` attribute). The shim + re-exports; the canonical module owns.""" + from kora_cli.reasoning.kora_hermes_plugin.cost_ladder import ( + RoutingDecision, + select_model_pre_call, ) - assert should is False - assert reason == "haiku_confident" - - -def test_post_call_non_string_does_not_escalate(): - should, reason = should_escalate_post_call( - haiku_response_text=None, # type: ignore[arg-type] - original_message_text="anything", - ) - assert should is False - -def test_post_call_low_confidence_pattern_env_override(monkeypatch): - """Operator can swap in a custom low-confidence vocabulary.""" - monkeypatch.setenv( - ENV_HAIKU_LOW_CONFIDENCE_PATTERNS, - r"this is suss,unclear", - ) - should, reason = should_escalate_post_call( - haiku_response_text="this is suss", - original_message_text="anything", - ) - assert should is True - # Old default markers no longer trigger. - should2, _ = should_escalate_post_call( - haiku_response_text="I'm not sure about that", - original_message_text="anything", + assert ( + RoutingDecision.__module__ + == "kora_cli.reasoning.kora_hermes_plugin.cost_ladder.selector" ) - assert should2 is False - - -# --------------------------------------------------------------------------- -# Decision-language env override -# --------------------------------------------------------------------------- - - -def test_decision_pattern_env_override(monkeypatch): - """Operator can shrink or expand the decision-language list.""" - monkeypatch.setenv(ENV_OPUS_TRIGGER_PATTERNS, r"\bcritical\b") - d = select_model_pre_call( - message_text="this is critical", - iteration=1, - cost_rung=RUNG_NORMAL, - ) - assert d.model == DEFAULT_OPUS_MODEL - assert d.reason == "decision_language" - - # An old default pattern that's no longer active. - d2 = select_model_pre_call( - message_text="should i ship this?", - iteration=1, - cost_rung=RUNG_NORMAL, - ) - assert d2.model == DEFAULT_HAIKU_MODEL - - -def test_decision_pattern_invalid_regex_skipped(monkeypatch, caplog): - """A bad pattern in the env list is logged + skipped; others - still work.""" - import logging - - caplog.set_level(logging.WARNING) - monkeypatch.setenv( - ENV_OPUS_TRIGGER_PATTERNS, r"[unclosed,\bcritical\b" - ) - d = select_model_pre_call( - message_text="this is critical", - iteration=1, - cost_rung=RUNG_NORMAL, - ) - assert d.model == DEFAULT_OPUS_MODEL - warns = [r.getMessage() for r in caplog.records if r.levelname == "WARNING"] - assert any("invalid" in w for w in warns) - - -# --------------------------------------------------------------------------- -# RoutingDecision shape -# --------------------------------------------------------------------------- - - -def test_routing_decision_is_frozen(): - d = RoutingDecision(model="x", reason="r", escalated=False) - with pytest.raises((AttributeError, TypeError)): - d.model = "y" # type: ignore[misc] - - -def test_routing_decision_default_haiku_context_is_none(): - d = RoutingDecision(model="x", reason="r", escalated=False) - assert d.haiku_context_for_opus is None - - -# --------------------------------------------------------------------------- -# Earning-signal precedence — Q4 boundaries -# --------------------------------------------------------------------------- - - -def test_force_opus_env_beats_default_haiku_but_loses_to_cost_clamp( - monkeypatch, -): - """Precedence: hard_stop > cost_clamp > force_opus > iteration - > prefix > decision_language > default. This test pins the - boundary between force_opus and cost_clamp.""" - monkeypatch.setenv(ENV_FORCE_OPUS, "true") - # Normal rung — force wins. - d = select_model_pre_call( - message_text="hi", iteration=1, cost_rung=RUNG_NORMAL - ) - assert d.model == DEFAULT_OPUS_MODEL - # warn_75 — cost clamp wins. - d2 = select_model_pre_call( - message_text="hi", iteration=1, cost_rung=RUNG_WARN_75 - ) - assert d2.model == DEFAULT_HAIKU_MODEL - - -def test_iteration_two_beats_opus_prefix(): - """Both signals would pick Opus; iteration check fires first - in the decision tree (matches the spec's pseudocode order - putting iteration check before prefix check).""" - d = select_model_pre_call( - message_text="/opus heavy", iteration=2, cost_rung=RUNG_NORMAL + assert ( + select_model_pre_call.__module__ + == "kora_cli.reasoning.kora_hermes_plugin.cost_ladder.selector" ) - assert d.model == DEFAULT_OPUS_MODEL - assert d.reason == "tool_loop_iteration"