diff --git a/kora_cli/plugins.py b/kora_cli/plugins.py index 99b9bfd6ab30..b687f2f3a0a7 100644 --- a/kora_cli/plugins.py +++ b/kora_cli/plugins.py @@ -151,6 +151,19 @@ def _install_plugin_debug_handler(force: bool = False) -> None: # override wins. Enables route-specific tool manifests # without mutating ``agent.tools`` (which is process-wide). "pre_tool_list_finalized", + # KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B — fires inside + # ``model_tools.handle_function_call`` AFTER the + # ``pre_tool_call`` block-check and BEFORE Hermes's default + # ``registry.dispatch``. Plugins return ``{"result": + # ""}`` to short-circuit Hermes dispatch + # with a plugin-provided result. First non-None ``result`` + # wins. Returning ``None`` (or non-dict / missing + # ``result`` key) falls through to other plugins, then + # Hermes default. Fail-safe: plugin exception → log + fall + # through. Enables fork-specific tool registries (e.g. + # Kora's reasoning tools) to dispatch via plugin code + # without registering them as Hermes-native tools. + "pre_tool_call_can_provide_result", # Transform LLM output before it's returned to the user. # Plugins return a string to replace the response text, or None/empty to leave unchanged. # First non-None string wins. Useful for vocabulary/personality transformation. diff --git a/kora_cli/reasoning/anthropic_engine.py b/kora_cli/reasoning/anthropic_engine.py index 631dce70e9fc..bc8b76e72b82 100644 --- a/kora_cli/reasoning/anthropic_engine.py +++ b/kora_cli/reasoning/anthropic_engine.py @@ -1161,14 +1161,33 @@ async def _respond_via_gateway( max_tokens=self._max_output_tokens, quiet_mode=True, # daemon path; no print() to stdout ) - # Override agent.tools = [] — Kora's reasoning tools are - # NOT bridged into Hermes's toolset model in this ST. - # The bridge is the explicit ST2B follow-on bucket. With - # the toggle OFF in production, the bypass path retains - # full tool capability; toggling ON loses tool-use until - # ST2B lands. - agent.tools = [] - agent.valid_tool_names = set() + # KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B — populate + # agent.tools from Kora's reasoning registry (replaces + # ST2's ``agent.tools = []`` toolless-only posture). The + # kora_hermes plugin's ``pre_tool_call_can_provide_result`` + # hook intercepts Hermes's dispatch for these tool names + # and routes them to Kora's existing reasoning dispatch + # (``execute_reasoning_tool``). Empty list — registry + # unavailable / failed import — falls back to toolless + # route-through same as ST2. + try: + from plugins.kora_hermes import get_kora_tools_for_agent + + kora_tools = get_kora_tools_for_agent() + agent.tools = kora_tools + agent.valid_tool_names = { + t["function"]["name"] + for t in kora_tools + if isinstance(t, dict) and "function" in t + } + except Exception as exc: + logger.warning( + "[kora.reasoning.gateway] tool-bridge tool population " + "failed: %r — falling back to toolless route-through", + exc, + ) + agent.tools = [] + agent.valid_tool_names = set() # Route field threading — kora_hermes plugin's hooks gate # on this. Setting it to "" (when source isn't mapped) diff --git a/kora_docs/14_research/hermes_local_extensions_2026-05-23.md b/kora_docs/14_research/hermes_local_extensions_2026-05-23.md index a78253fa7be6..838f5379e52a 100644 --- a/kora_docs/14_research/hermes_local_extensions_2026-05-23.md +++ b/kora_docs/14_research/hermes_local_extensions_2026-05-23.md @@ -161,6 +161,34 @@ ctx.register_background_daemon( **Thread safety**: registry methods are RLock-wrapped. Plugin discovery happens at import time on the main thread; consumers may iterate from any thread. +### Extension 5: `pre_tool_call_can_provide_result` hook (added in KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B) + +**File**: `model_tools.py` (insertion in `handle_function_call` after the existing `pre_tool_call` block-check and before `registry.dispatch`). +**VALID_HOOKS entry**: added between `pre_tool_list_finalized` and the `transform_llm_output` block in `kora_cli/plugins.py:155-167`. + +**Signature**: +```python +invoke_hook( + "pre_tool_call_can_provide_result", + tool_name=..., + args=..., + task_id=..., + session_id=..., + tool_call_id=..., +) +``` + +**Return contract**: +- `None` / non-dict / missing `"result"` key → no-op, fall through to other plugins, then Hermes default `registry.dispatch` +- `{"result": ""}` → short-circuits Hermes dispatch; the plugin-provided string becomes the tool result +- First non-None `result` wins (matches existing override-shape semantics for `pre_api_request_mutable` and `pre_tool_list_finalized`) + +**Failure handling**: any hook exception is caught + DEBUG-logged + fall-through to Hermes default. Fail-safe. + +**Backward compat**: no existing plugin registers it; non-Kora-route plugins gate themselves on tool-name or route checks (the kora_hermes plugin returns None when the tool isn't a Kora reasoning tool — confirms safety for Hermes-fork users loading the plugin). + +**Backward-compat for the dispatch site**: Hermes's `pre_tool_call` block-check + `post_tool_call` audit hook + `transform_tool_result` hook ALL still fire on the same code path; the new hook slots between the block-check and Hermes's default dispatch without altering observer ordering. + ### Extension DEFERRED: post-LLM re-issue hook **Why deferred**: re-issuing `messages.create` inside `conversation_loop` requires re-running portions of the loop (streaming consumption, tool-result processing, conversation-history append). Each of those has substantial state machinery. Doing this safely needs a coordinated control-flow refactor — beyond the scope of this bucket per §4. @@ -227,6 +255,16 @@ Each extension is designed to package cleanly into a future Hermes-upstream PR. - **Upstream framing**: "Add `route: str` context field to all conversation-loop hooks. Threading purely from `getattr(agent, 'route', '')` so legacy callers (no route set) see no change. Use case: per-route observability (cost telemetry, latency tracking, debug logging that groups by use-case)." - **Upstream prep work needed**: extract Hermes-friendly `KNOWN_ROUTES` taxonomy + RFC the vocabulary before the PR. +### Extension 5: `pre_tool_call_can_provide_result` (KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B) — **PR-ready after route-through battle-tests it** + +- [x] New hook added cleanly between existing block-check and Hermes default dispatch +- [x] First-non-None-result wins; non-dict / missing-key returns no-op +- [x] Fail-safe: hook exception → caught + logged → Hermes default dispatch +- [x] Backward compat: existing pre_tool_call (block-check) + post_tool_call + transform_tool_result fire unchanged on the same code path +- [x] Tests in `tests/plugins/test_kora_hermes_plugin_st2b.py` (Hermes-side wiring + Kora plugin consumer + bridge handler + Kora-tool-via-bridge sample trace + non-Kora-tool-fall-through-to-Hermes sample trace) +- [ ] **Battle-test gap**: ST3 (default-flip) + a 24-48h burn-in is the natural integration test. After ST3 lands, this hook is upstreamable. +- **Upstream framing**: "Allow plugins to short-circuit Hermes's tool dispatch with a plugin-computed result. Use case: a fork that maintains its own tool registry (parallel to Hermes's) wants those tools dispatched via fork code without registering them as Hermes tools. Fork-specific tool dispatch keeps the fork's code organization clean + enables behaviors (async tool dispatch, custom error envelopes, plugin-mediated security checks) that don't fit the Hermes tool registry's signature." + ### Extension 4: `register_background_daemon` + registry — **PR-ready in isolation; consumer wiring is a separate PR** - [x] New module is import-side-effect-free diff --git a/model_tools.py b/model_tools.py index 137e1e53e449..8fbd5755cc99 100644 --- a/model_tools.py +++ b/model_tools.py @@ -829,21 +829,56 @@ def handle_function_call( # to wrap every tool manually. We use monotonic() so the value is # unaffected by wall-clock adjustments during the call. _dispatch_start = time.monotonic() - if function_name == "execute_code": - # Prefer the caller-provided list so subagents can't overwrite - # the parent's tool set via the process-global. - sandbox_enabled = enabled_tools if enabled_tools is not None else _last_resolved_tool_names - result = registry.dispatch( - function_name, function_args, - task_id=task_id, - enabled_tools=sandbox_enabled, + + # KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B — give plugins a + # chance to provide the result themselves (e.g. Kora's + # reasoning tools are dispatched via the kora_hermes + # plugin, NOT via Hermes's tool registry). First non-None + # ``{"result": }`` return wins; on no plugin result, + # fall through to Hermes's default ``registry.dispatch``. + # Fail-safe: any hook exception → caught + logged → fall + # through. Non-Kora-route plugins return None on their own + # gate; this is purely additive on the Hermes flow. + result: Optional[str] = None + try: + from kora_cli.plugins import invoke_hook as _invoke_hook + _provide_results = _invoke_hook( + "pre_tool_call_can_provide_result", + tool_name=function_name, + args=function_args, + task_id=task_id or "", + session_id=session_id or "", + tool_call_id=tool_call_id or "", ) - else: - result = registry.dispatch( - function_name, function_args, - task_id=task_id, - user_task=user_task, + for _provide in _provide_results: + if isinstance(_provide, dict): + _r = _provide.get("result") + if isinstance(_r, str): + result = _r + break + except Exception as _hook_err: + logger.debug( + "pre_tool_call_can_provide_result hook error: %s", + _hook_err, ) + result = None + + if result is None: + if function_name == "execute_code": + # Prefer the caller-provided list so subagents can't overwrite + # the parent's tool set via the process-global. + sandbox_enabled = enabled_tools if enabled_tools is not None else _last_resolved_tool_names + result = registry.dispatch( + function_name, function_args, + task_id=task_id, + enabled_tools=sandbox_enabled, + ) + else: + result = registry.dispatch( + function_name, function_args, + task_id=task_id, + user_task=user_task, + ) duration_ms = int((time.monotonic() - _dispatch_start) * 1000) try: diff --git a/plugins/kora_hermes/__init__.py b/plugins/kora_hermes/__init__.py index 0c0d3c5f8ef2..c07dfa676f30 100644 --- a/plugins/kora_hermes/__init__.py +++ b/plugins/kora_hermes/__init__.py @@ -285,6 +285,164 @@ def _post_llm_call( ) +# --------------------------------------------------------------------------- +# 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 # --------------------------------------------------------------------------- @@ -292,15 +450,19 @@ def _post_llm_call( 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.""" + 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: 6 hooks against KORA_ROUTES=%s", + "[kora_hermes] plugin registered: 7 hooks against KORA_ROUTES=%s", sorted(KORA_ROUTES), ) diff --git a/tests/plugins/test_kora_hermes_plugin.py b/tests/plugins/test_kora_hermes_plugin.py index 0954bae58c77..a997377df674 100644 --- a/tests/plugins/test_kora_hermes_plugin.py +++ b/tests/plugins/test_kora_hermes_plugin.py @@ -74,11 +74,11 @@ def test_plugin_is_discovered_but_opt_in(): assert "not enabled in config" in (loaded.error or "") -def test_register_function_wires_six_hooks(): - """The plugin's register(ctx) function registers exactly 6 - hooks. Test directly with a mock context — bypasses Hermes's - opt-in plugins.enabled gate (which is operator-policy - territory, not the plugin's responsibility).""" +def test_register_function_wires_seven_hooks(): + """The plugin's register(ctx) function registers exactly 7 + hooks (ST2B added pre_tool_call_can_provide_result to ST1's + 6). Test directly with a mock context — bypasses Hermes's + opt-in plugins.enabled gate (operator-policy territory).""" from plugins.kora_hermes import register registered = [] @@ -96,6 +96,7 @@ def register_hook(self, name, callback): "pre_tool_call", "post_tool_call", "post_llm_call", + "pre_tool_call_can_provide_result", # ST2B added ]) # Each registered callback is callable. for name, callback in registered: diff --git a/tests/plugins/test_kora_hermes_plugin_st2.py b/tests/plugins/test_kora_hermes_plugin_st2.py index 90b823397942..e10b03bcb8eb 100644 --- a/tests/plugins/test_kora_hermes_plugin_st2.py +++ b/tests/plugins/test_kora_hermes_plugin_st2.py @@ -351,7 +351,15 @@ async def test_respond_via_gateway_end_to_end( assert ctor_kwargs["quiet_mode"] is True # Post-construction tool override + route set. - assert fake_agent.tools == [] + # KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B — agent.tools is + # now populated from Kora's reasoning registry (no longer + # the toolless ``[]``). Verify shape: list of Hermes-shaped + # tool dicts ``{"type": "function", "function": {...}}``. + assert isinstance(fake_agent.tools, list) + assert len(fake_agent.tools) >= 1 + for t in fake_agent.tools: + assert t["type"] == "function" + assert "name" in t["function"] assert fake_agent.route == "slack_dm" # Result projection. diff --git a/tests/plugins/test_kora_hermes_plugin_st2b.py b/tests/plugins/test_kora_hermes_plugin_st2b.py new file mode 100644 index 000000000000..435991e0b3b1 --- /dev/null +++ b/tests/plugins/test_kora_hermes_plugin_st2b.py @@ -0,0 +1,581 @@ +"""ST2B tool-bridge tests for KR-REASONING-ROUTE-THROUGH-GATEWAY. + +Covers the new ``pre_tool_call_can_provide_result`` Hermes hook ++ the Kora plugin's bridge handler + ``get_kora_tools_for_agent`` +shape conversion + the ``_respond_via_gateway`` tool-population +integration. + +ST2B scope: + - Hook surface (added to VALID_HOOKS; fires inside + ``handle_function_call``; first-non-None-result wins; + fail-safe on plugin exceptions) + - Bridge handler short-circuits Hermes dispatch for Kora's + reasoning tools; returns None for non-Kora tools + (Hermes-fork safety — plugin loaded on non-Kora deploys + doesn't break their tool dispatch) + - get_kora_tools_for_agent converts Anthropic → Hermes shape + - _respond_via_gateway populates agent.tools from the bridge + + agent.valid_tool_names tracks the names +""" + +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + + +# --------------------------------------------------------------------------- +# Hook surface — pre_tool_call_can_provide_result +# --------------------------------------------------------------------------- + + +def test_new_hook_in_valid_hooks(): + from kora_cli.plugins import VALID_HOOKS + + assert "pre_tool_call_can_provide_result" in VALID_HOOKS + + +def test_register_hook_accepts_new_hook_without_warning(caplog): + """Registering the new hook MUST NOT trigger the unknown-hook + warning — confirms it's properly in VALID_HOOKS.""" + import logging + + from kora_cli.plugins import PluginContext, PluginManifest + + caplog.set_level(logging.WARNING) + manifest = MagicMock(spec=PluginManifest) + manifest.name = "test_plugin" + manager = MagicMock() + ctx = PluginContext.__new__(PluginContext) + ctx.manifest = manifest + ctx._manager = manager + + ctx.register_hook( + "pre_tool_call_can_provide_result", lambda **kw: None + ) + warns = [r.getMessage() for r in caplog.records if r.levelname == "WARNING"] + assert not any("pre_tool_call_can_provide_result" in w for w in warns) + + +# --------------------------------------------------------------------------- +# get_kora_tools_for_agent — Anthropic → Hermes shape +# --------------------------------------------------------------------------- + + +def test_get_kora_tools_for_agent_returns_hermes_shape(): + """Each entry is ``{"type": "function", "function": {...}}``.""" + from plugins.kora_hermes import get_kora_tools_for_agent + + tools = get_kora_tools_for_agent() + assert len(tools) >= 1 # registry ships >= 1 tool + for t in tools: + assert t["type"] == "function" + assert set(t["function"].keys()) >= { + "name", + "description", + "parameters", + } + + +def test_get_kora_tools_for_agent_all_names_in_allowlist(): + """Every tool returned matches the reasoning-tool allowlist — + drift-guarded so a registry-side rename can't leak a tool + that the bridge handler won't recognize.""" + from kora_cli.reasoning.tool_registry import REASONING_TOOL_ALLOWLIST + from plugins.kora_hermes import get_kora_tools_for_agent + + names = {t["function"]["name"] for t in get_kora_tools_for_agent()} + assert names <= set(REASONING_TOOL_ALLOWLIST), ( + f"agent tools contain names outside the reasoning allowlist: " + f"{names - set(REASONING_TOOL_ALLOWLIST)}" + ) + + +def test_get_kora_tools_for_agent_empty_on_registry_failure(monkeypatch): + """If the registry import / call fails, function returns [] — + engine falls back to toolless route-through (ST2 posture).""" + from plugins import kora_hermes as kh + + def _explode(): + raise RuntimeError("registry down") + + monkeypatch.setattr( + "kora_cli.reasoning.tool_registry.get_reasoning_available_tools", + _explode, + ) + assert kh.get_kora_tools_for_agent() == [] + + +# --------------------------------------------------------------------------- +# _tool_bridge_provide_result — bridge handler +# --------------------------------------------------------------------------- + + +def test_bridge_returns_none_for_non_kora_tool(): + """Hermes-fork safety: non-Kora tools → None → Hermes default + dispatch runs unchanged.""" + from plugins.kora_hermes import _tool_bridge_provide_result + + assert _tool_bridge_provide_result(tool_name="write_file", args={}) is None + assert _tool_bridge_provide_result(tool_name="bash", args={"cmd": "ls"}) is None + assert _tool_bridge_provide_result(tool_name="", args={}) is None + + +def test_bridge_dispatches_kora_tool_to_reasoning_registry(monkeypatch): + """Kora-allowlisted tool → bridge calls execute_reasoning_tool + + returns ``{"result": }``.""" + from plugins.kora_hermes import _tool_bridge_provide_result + + # Stub the registry's executor to return a Pydantic-like model. + class _StubResult: + def model_dump_json(self): + return '{"value": "stubbed"}' + + async def _fake_execute(name, tool_input): + assert name == "kora__get_operational_state" + return _StubResult() + + monkeypatch.setattr( + "kora_cli.reasoning.tool_registry.execute_reasoning_tool", + _fake_execute, + ) + + out = _tool_bridge_provide_result( + tool_name="kora__get_operational_state", args={} + ) + assert isinstance(out, dict) + assert "result" in out + payload = json.loads(out["result"]) + assert payload == {"value": "stubbed"} + + +def test_bridge_handles_dispatch_exception_with_is_error_result(monkeypatch): + """Tool dispatch raises → bridge returns + ``{"result": '{"error": ...}'}`` so the reasoning loop sees + a tool_result with an error (vs Hermes crashing on + unregistered tool).""" + from plugins.kora_hermes import _tool_bridge_provide_result + + async def _exploding_execute(name, tool_input): + raise RuntimeError("substrate down") + + monkeypatch.setattr( + "kora_cli.reasoning.tool_registry.execute_reasoning_tool", + _exploding_execute, + ) + + out = _tool_bridge_provide_result( + tool_name="kora__get_operational_state", args={} + ) + assert isinstance(out, dict) + assert "result" in out + payload = json.loads(out["result"]) + assert "error" in payload + assert "kora_tool_dispatch_error" in payload["error"] + assert "RuntimeError" in payload["error"] + + +def test_bridge_handles_non_pydantic_result(monkeypatch): + """If execute_reasoning_tool returns something without + ``model_dump_json``, fall back to ``json.dumps(default=str)``.""" + from plugins.kora_hermes import _tool_bridge_provide_result + + async def _fake_execute(name, tool_input): + return {"raw": "dict"} + + monkeypatch.setattr( + "kora_cli.reasoning.tool_registry.execute_reasoning_tool", + _fake_execute, + ) + + out = _tool_bridge_provide_result( + tool_name="kora__get_operational_state", args={} + ) + assert json.loads(out["result"]) == {"raw": "dict"} + + +# --------------------------------------------------------------------------- +# Hook fires in model_tools.handle_function_call +# --------------------------------------------------------------------------- + + +def test_handle_function_call_fires_provide_result_hook(monkeypatch): + """When the hook returns ``{"result": ...}``, Hermes's + registry.dispatch MUST be short-circuited.""" + from kora_cli.plugins import PluginManager + + captured: list = [] + + class _FakeManager: + def invoke_hook(self, name, **kw): + captured.append((name, kw)) + if name == "pre_tool_call_can_provide_result": + return [{"result": "stubbed-bridge-result"}] + return [] + + fake = _FakeManager() + monkeypatch.setattr( + "kora_cli.plugins.get_plugin_manager", lambda: fake + ) + + # Spy on registry.dispatch — it must NOT be called when the + # hook short-circuited. + import model_tools + + dispatch_calls: list = [] + + def _spy_dispatch(name, args, **kw): + dispatch_calls.append(name) + return "should-not-reach" + + monkeypatch.setattr(model_tools.registry, "dispatch", _spy_dispatch) + + result = model_tools.handle_function_call("any_tool", {}) + assert result == "stubbed-bridge-result" + assert dispatch_calls == [], ( + "registry.dispatch must NOT be called when the provide_result " + "hook short-circuits" + ) + # The hook fired before dispatch. + hook_calls = [n for n, _ in captured if n == "pre_tool_call_can_provide_result"] + assert len(hook_calls) == 1 + + +def test_handle_function_call_falls_through_on_no_provide(monkeypatch): + """When no plugin returns ``{"result": ...}``, Hermes's default + dispatch must run.""" + + class _FakeManager: + def invoke_hook(self, name, **kw): + if name == "pre_tool_call_can_provide_result": + return [None, {"other": "ignored"}, {}] # all no-op + return [] + + monkeypatch.setattr( + "kora_cli.plugins.get_plugin_manager", lambda: _FakeManager() + ) + + import model_tools + + dispatch_calls: list = [] + + def _spy_dispatch(name, args, **kw): + dispatch_calls.append(name) + return "hermes-default-result" + + monkeypatch.setattr(model_tools.registry, "dispatch", _spy_dispatch) + + result = model_tools.handle_function_call("any_tool", {}) + assert result == "hermes-default-result" + assert dispatch_calls == ["any_tool"] + + +def test_handle_function_call_hook_exception_falls_through( + monkeypatch, caplog +): + """Plugin raises in the hook → caught + debug-logged → Hermes + default dispatch runs (fail-safe).""" + import logging + + class _FakeManager: + def invoke_hook(self, name, **kw): + if name == "pre_tool_call_can_provide_result": + raise RuntimeError("plugin imploded") + return [] + + monkeypatch.setattr( + "kora_cli.plugins.get_plugin_manager", lambda: _FakeManager() + ) + + import model_tools + + def _spy_dispatch(name, args, **kw): + return "hermes-default-after-hook-failure" + + monkeypatch.setattr(model_tools.registry, "dispatch", _spy_dispatch) + caplog.set_level(logging.DEBUG) + result = model_tools.handle_function_call("any_tool", {}) + assert result == "hermes-default-after-hook-failure" + + +# --------------------------------------------------------------------------- +# End-to-end: _respond_via_gateway populates tools + dispatch fires +# --------------------------------------------------------------------------- + + +def _make_incoming(text: str = "what's my burn?", source: str = "slack_dm"): + from kora_cli.reasoning.engine import IncomingMessage + + return IncomingMessage( + text=text, + source=source, + received_at=datetime.now(timezone.utc), + metadata={}, + ) + + +def _make_context(rung: str = "normal", state: str = "ready"): + from kora_cli.reasoning.engine import ConversationContext + + return ConversationContext( + recent_messages=[], + current_operational_state=state, + current_cost_ladder_rung=rung, + ) + + +@pytest.fixture +def system_prompt_path(tmp_path): + p = tmp_path / "kora_system_prompt.md" + p.write_text("You are Kora.\n", encoding="utf-8") + return p + + +@pytest.fixture(autouse=True) +def _oauth_env(monkeypatch): + monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "sk-ant-oat-test") + + +def _make_engine(system_prompt_path): + from kora_cli.reasoning.anthropic_engine import AnthropicReasoningEngine + + return AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=MagicMock() + ) + + +@pytest.mark.asyncio +async def test_respond_via_gateway_populates_agent_tools( + monkeypatch, system_prompt_path +): + """When the toggle is ON, _respond_via_gateway populates + agent.tools with Kora's 5 reasoning tools (replaces ST2's + toolless ``= []``).""" + fake_agent = MagicMock() + fake_agent.model = "claude-haiku-4-5-20251001" + fake_agent.run_conversation = lambda u, s=None: { + "final_response": "ok", + "model": "claude-haiku-4-5-20251001", + "input_tokens": 1, + "output_tokens": 1, + "cache_read_tokens": 0, + "cache_write_tokens": 0, + "completed": True, + "interrupted": False, + } + fake_class = MagicMock(return_value=fake_agent) + monkeypatch.setattr("run_agent.AIAgent", fake_class) + monkeypatch.setenv("KORA_REASONING_USE_GATEWAY", "true") + + engine = _make_engine(system_prompt_path) + await engine.respond(_make_incoming(), _make_context()) + + # agent.tools populated from Kora's registry — Hermes-shape. + assert isinstance(fake_agent.tools, list) + assert len(fake_agent.tools) >= 1 + names = { + t["function"]["name"] + for t in fake_agent.tools + if isinstance(t, dict) and "function" in t + } + assert "kora__get_operational_state" in names + assert fake_agent.valid_tool_names == names + + +@pytest.mark.asyncio +async def test_respond_via_gateway_falls_back_when_tools_unavailable( + monkeypatch, system_prompt_path +): + """get_kora_tools_for_agent raises → engine falls back to + agent.tools = [] (toolless route-through; ST2 posture).""" + fake_agent = MagicMock() + fake_agent.run_conversation = lambda u, s=None: { + "final_response": "ok", + "model": "h", + "input_tokens": 0, + "output_tokens": 0, + "cache_read_tokens": 0, + "cache_write_tokens": 0, + "completed": True, + "interrupted": False, + } + fake_class = MagicMock(return_value=fake_agent) + monkeypatch.setattr("run_agent.AIAgent", fake_class) + monkeypatch.setenv("KORA_REASONING_USE_GATEWAY", "true") + + monkeypatch.setattr( + "plugins.kora_hermes.get_kora_tools_for_agent", + lambda: (_ for _ in ()).throw(RuntimeError("registry down")), + ) + + engine = _make_engine(system_prompt_path) + await engine.respond(_make_incoming(), _make_context()) + assert fake_agent.tools == [] + assert fake_agent.valid_tool_names == set() + + +# --------------------------------------------------------------------------- +# Sample tool-use trace: end-to-end through handle_function_call +# --------------------------------------------------------------------------- + + +@pytest.fixture +def _register_bridge_hook_in_global_manager(): + """Append the bridge handler to the process-global + PluginManager's hook list for a sample-trace test, then + remove it on teardown. Uses try/finally semantic via yield + so a test failure inside the body still cleans up — + without this, a partial test run pollutes downstream tests + that rely on un-registered hook state (the global manager + is a singleton across tests).""" + from kora_cli.plugins import get_plugin_manager + from plugins.kora_hermes import _tool_bridge_provide_result + + mgr = get_plugin_manager() + mgr._hooks.setdefault( + "pre_tool_call_can_provide_result", [] + ).append(_tool_bridge_provide_result) + try: + yield _tool_bridge_provide_result + finally: + callbacks = mgr._hooks.get( + "pre_tool_call_can_provide_result", [] + ) + if _tool_bridge_provide_result in callbacks: + callbacks.remove(_tool_bridge_provide_result) + + +def test_sample_tool_use_trace_kora_tool_via_bridge( + monkeypatch, _register_bridge_hook_in_global_manager +): + """Sample trace: Hermes asks to dispatch ``kora__get_ + operational_state`` → bridge intercepts → Kora's + execute_reasoning_tool runs → result string is returned to + Hermes loop. (No Hermes registry.dispatch call.)""" + import model_tools + + # 1. Stub execute_reasoning_tool to return a controlled + # result (avoids substrate-dep flakiness). + class _Stub: + def model_dump_json(self): + return '{"primary_state": "ready", "claim_permission": "normal"}' + + async def _fake_execute(name, tool_input): + return _Stub() + + monkeypatch.setattr( + "kora_cli.reasoning.tool_registry.execute_reasoning_tool", + _fake_execute, + ) + + # 2. Spy on Hermes default dispatch — MUST NOT fire. + dispatch_calls: list = [] + monkeypatch.setattr( + model_tools.registry, + "dispatch", + lambda name, args, **kw: dispatch_calls.append(name) or "WRONG", + ) + + # 3. Trace: invoke handle_function_call as Hermes would. + result = model_tools.handle_function_call( + function_name="kora__get_operational_state", + function_args={}, + task_id="trace_test", + ) + + # 4. Verify the trace: + # - Hermes dispatch NEVER fired + assert dispatch_calls == [] + # - Bridge result is the Kora result JSON + parsed = json.loads(result) + assert parsed["primary_state"] == "ready" + assert parsed["claim_permission"] == "normal" + + +def test_sample_tool_use_trace_non_kora_tool_falls_through( + monkeypatch, _register_bridge_hook_in_global_manager +): + """Sample trace: Hermes asks to dispatch ``write_file`` (a + Hermes-native tool, NOT in Kora's allowlist) → bridge returns + None → Hermes default dispatch runs unchanged. Confirms + Hermes-fork users with the plugin loaded see no regression + on their own tools.""" + import model_tools + + dispatch_calls: list = [] + monkeypatch.setattr( + model_tools.registry, + "dispatch", + lambda name, args, **kw: ( + dispatch_calls.append(name) or "hermes-default" + ), + ) + + result = model_tools.handle_function_call( + function_name="write_file", + function_args={"path": "/tmp/x", "content": "hi"}, + ) + + # Hermes dispatch DID fire for non-Kora tool — fork users safe. + assert dispatch_calls == ["write_file"] + assert result == "hermes-default" + + +# --------------------------------------------------------------------------- +# Bypass-path regression — toggle OFF unchanged +# --------------------------------------------------------------------------- + + +def _fake_anthropic_response(text: str = "bypass works"): + usage = MagicMock() + usage.input_tokens = 10 + usage.output_tokens = 5 + usage.cache_creation_input_tokens = 0 + usage.cache_read_input_tokens = 0 + block = MagicMock() + block.type = "text" + block.text = text + r = MagicMock() + r.content = [block] + r.stop_reason = "end_turn" + r.model = "claude-haiku-4-5-20251001" + r.usage = usage + return r + + +@pytest.mark.asyncio +async def test_toggle_off_bypass_unchanged_post_st2b( + monkeypatch, system_prompt_path +): + """ST2B's tool-bridge changes don't leak into the toggle-OFF + bypass path. Default behavior: existing bypass runs cleanly.""" + monkeypatch.delenv("KORA_REASONING_USE_GATEWAY", raising=False) + from kora_cli.listeners import mcp_tools + + monkeypatch.setattr(mcp_tools, "_get_active_provider", lambda: None) + + fake_aiagent_class = MagicMock() + monkeypatch.setattr("run_agent.AIAgent", fake_aiagent_class) + + from kora_cli.reasoning.anthropic_engine import ( + AnthropicReasoningEngine, + ) + + client = MagicMock() + client.messages = MagicMock() + client.messages.create = AsyncMock( + return_value=_fake_anthropic_response("bypass works") + ) + client.close = AsyncMock() + engine = AnthropicReasoningEngine( + system_prompt_path=system_prompt_path, client=client + ) + + result = await engine.respond(_make_incoming(), _make_context()) + assert result.text == "bypass works" + assert result.error is None + fake_aiagent_class.assert_not_called() diff --git a/tests/test_model_tools.py b/tests/test_model_tools.py index 978712302bdb..347e88dd2c1e 100644 --- a/tests/test_model_tools.py +++ b/tests/test_model_tools.py @@ -54,6 +54,10 @@ def test_tool_hooks_receive_session_and_tool_call_ids(self): ) assert result == '{"ok":true}' + # KR-REASONING-ROUTE-THROUGH-GATEWAY-ST2B added the new + # ``pre_tool_call_can_provide_result`` hook between + # pre_tool_call's block-check and Hermes's + # registry.dispatch. Updated to reflect the new sequence. assert mock_invoke_hook.call_args_list == [ call( "pre_tool_call", @@ -63,6 +67,14 @@ def test_tool_hooks_receive_session_and_tool_call_ids(self): session_id="session-1", tool_call_id="call-1", ), + call( + "pre_tool_call_can_provide_result", + tool_name="web_search", + args={"q": "test"}, + task_id="task-1", + session_id="session-1", + tool_call_id="call-1", + ), call( "post_tool_call", tool_name="web_search",