feat(middleware): add adaptive execution intercepts

Signed-off-by: Bryan Bednarski <bbednarski@nvidia.com>

tests/hermes_cli/test_plugins.py

@@ -18,8 +18,15 @@ from hermes_cli.plugins import (
     get_plugin_command_handler,
     get_plugin_commands,
     get_pre_tool_call_block_message,
+    has_middleware,
     resolve_plugin_command_result,
 )
+from hermes_cli.middleware import (
+    VALID_MIDDLEWARE,
+    apply_llm_request_middleware,
+    apply_tool_request_middleware,
+    run_tool_execution_middleware,
+)
 
 
 # ── Helpers ────────────────────────────────────────────────────────────────
@@ -96,6 +103,110 @@ class TestPluginDiscovery:
         assert "hello_plugin" in mgr._plugins
         assert mgr._plugins["hello_plugin"].enabled
 
+    def test_plugin_can_register_and_invoke_middleware(self, tmp_path, monkeypatch):
+        plugins_dir = tmp_path / "hermes_test" / "plugins"
+        _make_plugin_dir(
+            plugins_dir,
+            "mw_plugin",
+            register_body=(
+                "ctx.register_middleware('llm_request', "
+                "lambda **kw: {'request': {**kw['request'], 'mw': True}})\n"
+                "    ctx.register_middleware('tool_request', "
+                "lambda **kw: {'args': {**kw['args'], 'mw': True}})"
+            ),
+        )
+        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
+
+        mgr = PluginManager()
+        mgr.discover_and_load()
+
+        assert "llm_request" in VALID_MIDDLEWARE
+        assert "tool_request" in VALID_MIDDLEWARE
+        assert set(mgr._plugins["mw_plugin"].middleware_registered) == {"llm_request", "tool_request"}
+        assert mgr.invoke_middleware("llm_request", request={"messages": []}) == [
+            {"request": {"messages": [], "mw": True}}
+        ]
+        assert mgr.invoke_middleware("tool_request", args={"path": "README.md"}) == [
+            {"args": {"path": "README.md", "mw": True}}
+        ]
+        assert mgr.has_middleware("llm_request") is True
+
+    def test_execution_middleware_does_not_retry_downstream_failure(self, monkeypatch):
+        calls = []
+
+        def middleware(**kwargs):
+            return kwargs["next_call"](kwargs["args"])
+
+        manager = types.SimpleNamespace(_middleware={"tool_execution": [middleware]})
+        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)
+
+        def terminal(args):
+            calls.append(args)
+            raise RuntimeError("tool failed")
+
+        with pytest.raises(RuntimeError, match="tool failed"):
+            run_tool_execution_middleware("terminal", {"command": "false"}, terminal)
+
+        assert calls == [{"command": "false"}]
+
+    def test_middleware_helpers_skip_no_listener_work(self, monkeypatch):
+        manager = types.SimpleNamespace(_middleware={})
+        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)
+
+        request = {"messages": []}
+        args = {"path": "README.md"}
+
+        llm_result = apply_llm_request_middleware(request)
+        tool_result = apply_tool_request_middleware("read_file", args)
+
+        assert llm_result.payload is request
+        assert llm_result.original_payload is request
+        assert llm_result.changed is False
+        assert llm_result.trace == []
+        assert tool_result.payload is args
+        assert tool_result.original_payload is args
+        assert tool_result.changed is False
+        assert tool_result.trace == []
+        assert run_tool_execution_middleware("terminal", args, lambda payload: payload) is args
+        assert has_middleware("tool_request") is False
+
+    def test_request_middleware_changed_tracks_trace_not_deep_equality(self, monkeypatch):
+        def same_payload_middleware(**kwargs):
+            return {"args": kwargs["args"], "source": "same-payload"}
+
+        manager = types.SimpleNamespace(
+            _middleware={"tool_request": [same_payload_middleware]},
+            invoke_middleware=lambda kind, **kwargs: [same_payload_middleware(**kwargs)],
+        )
+        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)
+
+        args = {"path": "README.md"}
+        result = apply_tool_request_middleware("read_file", args)
+
+        assert result.payload == args
+        assert result.original_payload == args
+        assert result.changed is True
+        assert result.trace == [{"source": "same-payload"}]
+
+    def test_execution_middleware_post_next_call_error_does_not_retry(self, monkeypatch):
+        calls = []
+
+        def middleware(**kwargs):
+            result = kwargs["next_call"](kwargs["args"])
+            raise RuntimeError(f"post-processing failed after {result}")
+
+        manager = types.SimpleNamespace(_middleware={"tool_execution": [middleware]})
+        monkeypatch.setattr("hermes_cli.plugins.get_plugin_manager", lambda: manager)
+
+        def terminal(args):
+            calls.append(args)
+            return "terminal-result"
+
+        result = run_tool_execution_middleware("terminal", {"command": "printf ok"}, terminal)
+
+        assert result == "terminal-result"
+        assert calls == [{"command": "printf ok"}]
+
     def test_discover_project_plugins(self, tmp_path, monkeypatch):
         """Plugins in ./.hermes/plugins/ are discovered."""
         project_dir = tmp_path / "project"