1  """Tests for the ``transform_tool_result`` plugin hook wired into
  2  ``model_tools.handle_function_call``.
  3  
  4  Mirrors the ``transform_terminal_output`` hook tests from Phase 1 but
  5  targets the generic tool-result seam that runs for every tool dispatch.
  6  """
  7  
  8  import json
  9  import os
 10  from pathlib import Path
 11  from unittest.mock import MagicMock
 12  
 13  import hermes_cli.plugins as plugins_mod
 14  import model_tools
 15  
 16  
 17  _UNSET = object()
 18  
 19  
 20  def _run_handle_function_call(
 21      monkeypatch,
 22      *,
 23      tool_name="dummy_tool",
 24      tool_args=None,
 25      dispatch_result='{"output": "original"}',
 26      invoke_hook=_UNSET,
 27  ):
 28      """Drive ``handle_function_call`` with a mocked registry dispatch."""
 29      from tools.registry import registry
 30  
 31      monkeypatch.setattr(
 32          registry, "dispatch",
 33          lambda name, args, **kw: dispatch_result,
 34      )
 35      # Skip unrelated side effects (read-loop tracker).
 36      monkeypatch.setattr(model_tools, "_READ_SEARCH_TOOLS", frozenset())
 37  
 38      if invoke_hook is not _UNSET:
 39          # Patch the symbol actually imported inside handle_function_call.
 40          monkeypatch.setattr("hermes_cli.plugins.invoke_hook", invoke_hook)
 41  
 42      return model_tools.handle_function_call(
 43          tool_name,
 44          tool_args or {},
 45          task_id="t1",
 46          session_id="s1",
 47          tool_call_id="tc1",
 48          skip_pre_tool_call_hook=True,
 49      )
 50  
 51  
 52  def test_result_unchanged_when_no_hook_registered(monkeypatch):
 53      # Real invoke_hook with no plugins loaded returns [].
 54      monkeypatch.setenv("HERMES_HOME", "/tmp/hermes_no_plugins")
 55      # Force a fresh plugin manager so no stale plugins pollute state.
 56      plugins_mod._plugin_manager = plugins_mod.PluginManager()
 57  
 58      out = _run_handle_function_call(monkeypatch)
 59      assert out == '{"output": "original"}'
 60  
 61  
 62  def test_result_unchanged_for_none_hook_return(monkeypatch):
 63      out = _run_handle_function_call(
 64          monkeypatch,
 65          invoke_hook=lambda hook_name, **kw: [None],
 66      )
 67      assert out == '{"output": "original"}'
 68  
 69  
 70  def test_result_ignores_non_string_hook_returns(monkeypatch):
 71      out = _run_handle_function_call(
 72          monkeypatch,
 73          invoke_hook=lambda hook_name, **kw: [{"bad": True}, 123, ["nope"]],
 74      )
 75      assert out == '{"output": "original"}'
 76  
 77  
 78  def test_first_valid_string_return_replaces_result(monkeypatch):
 79      out = _run_handle_function_call(
 80          monkeypatch,
 81          invoke_hook=lambda hook_name, **kw: [None, {"x": 1}, "first", "second"],
 82      )
 83      assert out == "first"
 84  
 85  
 86  def test_hook_receives_expected_kwargs(monkeypatch):
 87      captured = {}
 88  
 89      def _hook(hook_name, **kwargs):
 90          if hook_name == "transform_tool_result":
 91              captured.update(kwargs)
 92          return []
 93  
 94      out = _run_handle_function_call(
 95          monkeypatch,
 96          tool_name="my_tool",
 97          tool_args={"a": 1, "b": "x"},
 98          dispatch_result='{"ok": true}',
 99          invoke_hook=_hook,
100      )
101      assert out == '{"ok": true}'
102      assert captured["tool_name"] == "my_tool"
103      assert captured["args"] == {"a": 1, "b": "x"}
104      assert captured["result"] == '{"ok": true}'
105      assert captured["task_id"] == "t1"
106      assert captured["session_id"] == "s1"
107      assert captured["tool_call_id"] == "tc1"
108  
109  
110  def test_hook_exception_falls_back_to_original(monkeypatch):
111      def _raise(*_a, **_kw):
112          raise RuntimeError("boom")
113  
114      out = _run_handle_function_call(
115          monkeypatch,
116          invoke_hook=_raise,
117      )
118      assert out == '{"output": "original"}'
119  
120  
121  def test_post_tool_call_remains_observational(monkeypatch):
122      """post_tool_call return values must NOT replace the result."""
123      def _hook(hook_name, **kw):
124          if hook_name == "post_tool_call":
125              # Observers returning a string must be ignored.
126              return ["observer return should be ignored"]
127          return []
128  
129      out = _run_handle_function_call(
130          monkeypatch,
131          invoke_hook=_hook,
132      )
133      assert out == '{"output": "original"}'
134  
135  
136  def test_transform_tool_result_runs_after_post_tool_call(monkeypatch):
137      """post_tool_call sees ORIGINAL result; transform_tool_result sees same and may replace."""
138      observed = []
139  
140      def _hook(hook_name, **kw):
141          if hook_name == "post_tool_call":
142              observed.append(("post_tool_call", kw["result"]))
143              return []
144          if hook_name == "transform_tool_result":
145              observed.append(("transform_tool_result", kw["result"]))
146              return ["rewritten"]
147          return []
148  
149      out = _run_handle_function_call(
150          monkeypatch,
151          dispatch_result='{"raw": "value"}',
152          invoke_hook=_hook,
153      )
154      assert out == "rewritten"
155      # Both hooks saw the ORIGINAL (untransformed) result.
156      assert observed == [
157          ("post_tool_call", '{"raw": "value"}'),
158          ("transform_tool_result", '{"raw": "value"}'),
159      ]
160  
161  
162  def test_transform_tool_result_integration_with_real_plugin(monkeypatch, tmp_path):
163      """End-to-end: load a real plugin from HERMES_HOME and verify it rewrites results."""
164      import yaml
165  
166      hermes_home = Path(os.environ["HERMES_HOME"])
167      plugins_dir = hermes_home / "plugins"
168      plugin_dir = plugins_dir / "transform_result_canon"
169      plugin_dir.mkdir(parents=True)
170      (plugin_dir / "plugin.yaml").write_text("name: transform_result_canon\n", encoding="utf-8")
171      (plugin_dir / "__init__.py").write_text(
172          "def register(ctx):\n"
173          '    ctx.register_hook("transform_tool_result", '
174          'lambda **kw: f\'CANON[{kw["tool_name"]}]\' + kw["result"])\n',
175          encoding="utf-8",
176      )
177      # Plugins are opt-in — must be listed in plugins.enabled to load.
178      cfg_path = hermes_home / "config.yaml"
179      cfg_path.write_text(
180          yaml.safe_dump({"plugins": {"enabled": ["transform_result_canon"]}}),
181          encoding="utf-8",
182      )
183  
184      # Force a fresh plugin manager so the new config is picked up.
185      plugins_mod._plugin_manager = plugins_mod.PluginManager()
186      plugins_mod.discover_plugins()
187  
188      out = _run_handle_function_call(
189          monkeypatch,
190          tool_name="some_tool",
191          dispatch_result='{"payload": 42}',
192      )
193      assert out == 'CANON[some_tool]{"payload": 42}'