diff --git a/hermes_cli/middleware.py b/hermes_cli/middleware.py
index 27e15e54a0..7d5888004b 100644
--- a/hermes_cli/middleware.py
+++ b/hermes_cli/middleware.py
@@ -7,7 +7,7 @@ wrapping the actual execution callback. Agent-loop call sites and plugins share
 from __future__ import annotations
 
 import logging
-from copy import copy, deepcopy
+from copy import deepcopy
 from dataclasses import dataclass, field
 from typing import Any, Callable, Dict, List
 
@@ -48,107 +48,16 @@ def middleware_payload(**kwargs: Any) -> Dict[str, Any]:
 
 
 def _safe_copy(payload: Any) -> Any:
-    """Copy a request graph, tolerating non-deepcopyable members.
+    """Deep-copy a request payload, tolerating non-deepcopyable members.
 
     An LLM request can carry clients/callbacks/file handles; a hard ``deepcopy`` failure would
-    otherwise abort the whole request-middleware pass. The fallback isolates surrounding
-    containers and preserves their aliases and cycles; opaque leaves retain their identity.
+    otherwise abort the whole request-middleware pass.
     """
     try:
         return deepcopy(payload)
-    except Exception as exc:
-        logger.debug("deepcopy failed for request payload (%s); retaining opaque leaves", exc)
-
-    # Walk containers without Python recursion: an opaque leaf or a deeply nested
-    # request must not share surrounding containers or bypass the middleware pass. Shallow
-    # copies retain container subclasses (including defaultdict's factory).
-    memo: Dict[int, Any] = {}
-    pending: Dict[int, List[Callable[[Any], None]]] = {}
-    tasks: List[Callable[[], None]] = []
-    root: List[Any] = []
-
-    def copy_state(value: Any, result: Any) -> None:
-        state = getattr(value, "__dict__", None)
-        if state is not None and hasattr(result, "__dict__"):
-            tasks.append(lambda: copy_value(state, lambda copied: setattr(result, "__dict__", copied)))
-
-    def copy_value(value: Any, assign: Callable[[Any], None]) -> None:
-        if type(value) in (str, bytes, int, float, bool, complex, type(None)):
-            assign(value)
-            return
-        identity = id(value)
-        if identity in memo:
-            assign(memo[identity])
-            return
-        if identity in pending:
-            pending[identity].append(assign)
-            return
-        result: Any
-        if isinstance(value, (dict, list, set)):
-            factory = dict if isinstance(value, dict) else list if isinstance(value, list) else set
-            try:
-                result = copy(value)
-                if result is value or not isinstance(result, factory):
-                    result = factory()
-                result.clear()
-            except Exception:
-                result = factory()
-            memo[identity] = result
-            assign(result)
-            copy_state(value, result)
-            if isinstance(result, dict):
-                for key, item in reversed(list(value.items())):
-                    pair: List[Any] = [None, None]
-                    tasks.append(lambda pair=pair, target=result: target.__setitem__(pair[0], pair[1]))
-                    tasks.append(lambda item=item, pair=pair: copy_value(
-                        item, lambda copied: pair.__setitem__(1, copied)))
-                    tasks.append(lambda key=key, pair=pair: copy_value(
-                        key, lambda copied: pair.__setitem__(0, copied)))
-            elif isinstance(result, list):
-                list.extend(result, [None] * len(value))
-                for index, item in reversed(list(enumerate(value))):
-                    tasks.append(lambda index=index, item=item, target=result: copy_value(
-                        item, lambda copied: list.__setitem__(target, index, copied)))
-            else:
-                for item in value:
-                    tasks.append(lambda item=item, target=result: copy_value(
-                        item, lambda copied: set.add(target, copied)))
-            return
-        if isinstance(value, (tuple, frozenset)):
-            pending[identity] = [assign]
-            items: List[Any] = [None] * len(value)
-
-            def finish() -> None:
-                factory = tuple if isinstance(value, tuple) else frozenset
-                result = factory.__new__(type(value), items)
-                memo[identity] = result
-                for setter in pending.pop(identity):
-                    setter(result)
-                copy_state(value, result)
-
-            # Mutable children are allocated before their contents, so tuple/list
-            # cycles can be completed after the tuple itself becomes available.
-            tasks.append(finish)
-            for index, item in reversed(list(enumerate(value))):
-                tasks.append(lambda index=index, item=item: copy_value(
-                    item, lambda copied: items.__setitem__(index, copied)))
-            return
-        # A failing __deepcopy__ may leave partial containers in its memo.
-        # Publish that memo only on success, never after an opaque leaf fails.
-        leaf_memo = memo.copy()
-        try:
-            result = deepcopy(value, leaf_memo)
-        except Exception:
-            result = value
-        else:
-            memo.update(leaf_memo)
-        memo[identity] = result
-        assign(result)
-
-    tasks.append(lambda: copy_value(payload, root.append))
-    while tasks:
-        tasks.pop()()
-    return root[0]
+    except Exception as exc:  # pragma: no cover - exercised via fallback test
+        logger.debug("deepcopy failed for request payload (%s); using shallow copy", exc)
+        return dict(payload) if isinstance(payload, dict) else payload
 
 
 def _apply_request_chain(
@@ -258,11 +167,6 @@ def _run_execution_chain(kind: str, terminal_call: Callable[[Any], Any], **kwarg
     if not callbacks:
         return terminal_call(kwargs[payload_key])
 
-    # Execution-only plugins may rewrite the caller's payload in place, so freeze
-    # its original before entering the first frame, then isolate each frame's view.
-    original_key = "original_" + payload_key
-    kwargs[original_key] = _safe_copy(kwargs[original_key])
-
     def call_at(index: int, payload: Any) -> Any:
         if index >= len(callbacks):
             return terminal_call(payload)
@@ -292,7 +196,6 @@ def _run_execution_chain(kind: str, terminal_call: Callable[[Any], Any], **kwarg
 
         call_kwargs = middleware_payload(**kwargs)
         call_kwargs[payload_key] = payload
-        call_kwargs[original_key] = _safe_copy(kwargs[original_key])
         call_kwargs["next_call"] = next_call
         try:
             return callback(**call_kwargs)
diff --git a/hermes_cli/plugins_dispatch.py b/hermes_cli/plugins_dispatch.py
index 84634a59cb..8015975355 100644
--- a/hermes_cli/plugins_dispatch.py
+++ b/hermes_cli/plugins_dispatch.py
@@ -591,31 +591,28 @@ class PluginDispatchMixin:
     def invoke_middleware(
         self, kind: str, *, _payload_key: Optional[str] = None, **kwargs: Any
     ) -> List[Any]:
-        """Dispatch middleware; request rewrites compose only after a valid return.
+        """Call middleware callbacks for *kind* (each isolated); return non-``None`` results.
 
-        Request callbacks receive a copy of the last accepted payload, so a
-        failing/observer callback cannot mutate an earlier accepted rewrite.
-        Each callback also receives its own pre-middleware snapshot.
+        Request middleware passes ``_payload_key``: a dict returned under it becomes the payload the
+        next callback sees, so rewrites compose, and each callback gets its own copy of the payload and
+        of ``original_<key>`` so an in-place edit cannot leak.
         """
         from hermes_cli.middleware import _safe_copy
 
-        copy_keys = {"original_request", "original_args"}
-        if _payload_key is not None:
-            copy_keys.add(_payload_key)
         results: List[Any] = []
         for cb in self._middleware.get(kind, []):
+            call_kwargs = kwargs
+            if _payload_key:
+                original_key = "original_" + _payload_key
+                call_kwargs = {**kwargs, _payload_key: _safe_copy(kwargs[_payload_key]),
+                               original_key: _safe_copy(kwargs[original_key])}
             try:
-                callback_kwargs = {
-                    key: _safe_copy(value) if key in copy_keys else value
-                    for key, value in kwargs.items()
-                }
-                ret = cb(**callback_kwargs)
+                ret = cb(**call_kwargs)
                 if ret is not None:
-                    if (_payload_key is not None and isinstance(ret, dict)
-                            and isinstance(ret.get(_payload_key), dict)):
-                        ret = {**ret, _payload_key: _safe_copy(ret[_payload_key])}
-                        kwargs[_payload_key] = ret[_payload_key]
                     results.append(ret)
+                    if _payload_key and isinstance(ret, dict) and isinstance(ret.get(_payload_key), dict):
+                        kwargs[_payload_key] = ret[_payload_key]
             except (Exception, SystemExit) as exc:
+                # Runs once per tool call like a hook, so a mis-declared callback floods identically.
                 self._report_hook_failure(kind, cb, kwargs, exc, surface="Middleware")
         return results
diff --git a/tests/hermes_cli/test_middleware_isolation.py b/tests/hermes_cli/test_middleware_isolation.py
deleted file mode 100644
index f7bf082459..0000000000
--- a/tests/hermes_cli/test_middleware_isolation.py
+++ /dev/null
@@ -1,264 +0,0 @@
-"""Request rewrites and original snapshots stay isolated across discovered middleware."""
-
-from collections import OrderedDict, defaultdict
-from typing import Any, NamedTuple, cast
-
-import pytest
-
-from hermes_cli import plugins
-from hermes_cli.config import atomic_config_write
-from hermes_cli.middleware import (
-    apply_llm_request_middleware,
-    apply_tool_request_middleware,
-    run_llm_execution_middleware,
-    run_tool_execution_middleware,
-)
-
-
-class RequestDict(dict):
-    pass
-
-
-class RequestList(list):
-    pass
-
-
-class RequestTuple(NamedTuple):
-    data: Any
-    file: Any
-
-
-class OpaqueMember:
-    def __init__(self, related):
-        self.related = related
-
-    def __deepcopy__(self, memo):
-        memo[id(self.related)] = {"partial": True}
-        raise TypeError("opaque member cannot be copied")
-
-
-def _load_plugin(tmp_path, monkeypatch, source):
-    home = tmp_path / "home"
-    plugin = home / "plugins" / "copy_probe"
-    plugin.mkdir(parents=True)
-    monkeypatch.setenv("HERMES_HOME", str(home))
-    plugin.joinpath("plugin.yaml").write_text(
-        "name: copy_probe\nversion: 0.1.0\ndescription: Middleware isolation probe\n",
-        encoding="utf-8",
-    )
-    plugin.joinpath("__init__.py").write_text(source, encoding="utf-8")
-    atomic_config_write(home / "config.yaml", {"plugins": {"enabled": ["copy_probe"]}})
-    manager = plugins.PluginManager()
-    manager.discover_and_load()
-    monkeypatch.setattr(plugins, "get_plugin_manager", lambda: manager)
-    assert any(manager.has_middleware(kind) for kind in (
-        "llm_request", "tool_request", "llm_execution", "tool_execution",
-    ))
-
-
-def _apply(kind, payload):
-    if kind == "llm":
-        return apply_llm_request_middleware(payload)
-    return apply_tool_request_middleware("probe", payload)
-
-
-@pytest.mark.parametrize("kind", ["llm", "tool"])
-@pytest.mark.parametrize("stage", ["request", "execution", "execution_only"])
-@pytest.mark.parametrize("outcome", ["observe", "raise", "rewrite"])
-@pytest.mark.parametrize("opaque", [False, True])
-def test_original_snapshot_survives_each_middleware_callback(
-    tmp_path, monkeypatch, kind, stage, outcome, opaque,
-):
-    key = "request" if kind == "llm" else "args"
-    request_listener = stage != "execution_only"
-    if not request_listener:
-        stage = "execution"
-    _load_plugin(tmp_path, monkeypatch, f"""
-KIND = {kind!r}
-KEY = {key!r}
-STAGE = {stage!r}
-OUTCOME = {outcome!r}
-REQUEST_LISTENER = {request_listener!r}
-
-def mutate(**kw):
-    original = kw['original_' + KEY]
-    original['input'].append('snapshot-write')
-    original['injected'] = True
-    if not REQUEST_LISTENER:
-        kw[KEY]['input'].append('execution-write')
-    if OUTCOME == 'raise':
-        raise RuntimeError('snapshot probe failure')
-    if OUTCOME == 'rewrite':
-        payload = {{**kw[KEY], 'rewritten': True}}
-        if STAGE == 'request':
-            return {{KEY: payload, 'source': 'mutate'}}
-        return kw['next_call'](payload)
-    if STAGE == 'execution':
-        return kw['next_call']()
-
-def verify(**kw):
-    original = kw['original_' + KEY]
-    stage = 'execution' if 'next_call' in kw else 'request'
-    payload = {{**kw[KEY], stage + '_snapshot': list(original['input'])}}
-    payload[stage + '_injected'] = 'injected' in original
-    if 'next_call' in kw:
-        return kw['next_call'](payload)
-    return {{KEY: payload, 'source': 'verify'}}
-
-def register(ctx):
-    ctx.register_middleware(KIND + '_' + STAGE, mutate)
-    if REQUEST_LISTENER:
-        ctx.register_middleware(KIND + '_request', verify)
-    ctx.register_middleware(KIND + '_execution', verify)
-""")
-    with (tmp_path / "upload.txt").open("w+", encoding="utf-8") as handle:
-        original = {"input": ["original"], **({"file": handle} if opaque else {})}
-        result = _apply(kind, original) if request_listener else None
-        payload = result.payload if result else original
-        context = {"original_" + key: result.original_payload} if result else {}
-        calls = []
-
-        def terminal(payload):
-            calls.append(payload)
-            return payload
-
-        if kind == "llm":
-            executed = run_llm_execution_middleware(
-                payload, terminal, **context,
-            )
-        else:
-            executed = run_tool_execution_middleware(
-                "probe", payload, terminal, **context,
-            )
-
-        expected_input = ["original"] if request_listener else ["original", "execution-write"]
-        assert original["input"] == expected_input
-        assert "injected" not in original
-        if result:
-            assert result.original_payload["input"] == ["original"]
-            assert "injected" not in result.original_payload
-        for phase in ("request", "execution") if request_listener else ("execution",):
-            assert executed[phase + "_snapshot"] == ["original"]
-            assert executed[phase + "_injected"] is False
-        assert executed["input"] == expected_input
-        assert bool(executed.get("rewritten")) == (outcome == "rewrite")
-        assert calls == [executed]
-        expected_sources = ["verify"]
-        if stage == "request" and outcome == "rewrite":
-            expected_sources.insert(0, "mutate")
-        if result:
-            assert [entry["source"] for entry in result.trace] == expected_sources
-        if opaque:
-            assert executed["file"] is original["file"] is handle
-            if result:
-                assert result.original_payload["file"] is handle
-            assert not handle.closed
-
-
-def _leaf(node):
-    while isinstance(node, dict) and "child" in node:
-        node = node["child"]
-    while isinstance(node, (list, tuple)):
-        node = node[0]
-    return node
-
-
-@pytest.mark.parametrize("kind", ["llm", "tool"])
-@pytest.mark.parametrize("container", [
-    dict, OrderedDict, defaultdict, RequestDict, list, RequestList, tuple, RequestTuple,
-])
-@pytest.mark.parametrize("depth,wrapper", [
-    (0, dict), (1500, dict), (6000, dict), (6000, list), (6000, tuple),
-])
-def test_opaque_request_graph_preserves_isolation_aliases_and_cycles(
-    tmp_path, monkeypatch, kind, container, depth, wrapper,
-):
-    key = "request" if kind == "llm" else "args"
-    _load_plugin(tmp_path, monkeypatch, f"""
-KIND = {kind!r}
-KEY = {key!r}
-
-def leaf(node):
-    while isinstance(node, dict) and 'child' in node:
-        node = node['child']
-    while isinstance(node, (list, tuple)):
-        node = node[0]
-    return node
-
-def first(**kw):
-    return {{KEY: {{**kw[KEY], 'accepted': True}}, 'source': 'first'}}
-
-def broken(**kw):
-    leaf(kw[KEY]['node'])['steps'].append('broken')
-    raise RuntimeError('graph probe failure')
-
-def observer(**kw):
-    leaf(kw[KEY]['node'])['steps'].append('observer')
-
-def second(**kw):
-    leaf(kw[KEY]['node'])['steps'].append('second')
-    return {{KEY: kw[KEY], 'source': 'second'}}
-
-def register(ctx):
-    for callback in (first, broken, observer, second):
-        ctx.register_middleware(KIND + '_request', callback)
-""")
-    with (tmp_path / "upload.txt").open("w+", encoding="utf-8") as handle:
-        leaf = {"steps": ["original"], "file": handle}
-        if issubclass(container, dict):
-            node = container(list, leaf) if container is defaultdict else container(leaf)
-        elif container is RequestTuple:
-            node = container(leaf, handle)
-        else:
-            node = container([leaf, handle])
-        if container in (RequestDict, RequestList):
-            node.metadata = ["original"]
-        for _ in range(depth):
-            node = {"child": node} if wrapper is dict else wrapper([node])
-        cycle = []
-        pair = (cycle, handle)
-        cycle.append(pair)
-        opaque = OpaqueMember(node)
-        original: dict[str, Any] = {
-            "opaque": opaque, "node": node, "alias": node, "cycle": pair,
-            "cycle_alias": pair, "tags": {"original"},
-            "frozen": frozenset({RequestTuple("original", handle)}),
-        }
-        try:
-            result = _apply(kind, original)
-        except RecursionError:
-            # Keep pytest from comparing deeply nested frame locals while locating
-            # recursion: that can hang its traceback formatter on this graph.
-            raise AssertionError("Deep requests must still run middleware") from None
-
-        assert result.payload["accepted"] is True
-        assert _leaf(result.payload["node"])["steps"] == ["original", "second"]
-        assert _leaf(result.original_payload["node"])["steps"] == ["original"]
-        assert _leaf(original["node"])["steps"] == ["original"]
-        assert result.payload["node"] is result.payload["alias"]
-        assert result.original_payload["node"] is result.original_payload["alias"]
-        assert result.payload["cycle"][0][0] is result.payload["cycle"]
-        assert result.original_payload["cycle"][0][0] is result.original_payload["cycle"]
-        assert result.payload["cycle_alias"] is result.payload["cycle"]
-        assert result.original_payload["cycle_alias"] is result.original_payload["cycle"]
-        assert result.payload["opaque"] is result.original_payload["opaque"] is opaque
-        assert result.payload["frozen"] == result.original_payload["frozen"] == original["frozen"]
-        assert _leaf(result.payload["node"])["file"] is handle
-        assert not handle.closed
-        assert [entry["source"] for entry in result.trace] == ["first", "second"]
-        copied_node = result.payload["node"]
-        snapshot_node = result.original_payload["node"]
-        source_node = cast(Any, original["node"])
-        for _ in range(depth):
-            index = "child" if wrapper is dict else 0
-            copied_node = copied_node[index]
-            snapshot_node = snapshot_node[index]
-            source_node = source_node[index]
-        assert type(copied_node) is type(snapshot_node) is type(source_node) is container
-        if container is defaultdict:
-            assert copied_node.default_factory is snapshot_node.default_factory is list
-        if container in (RequestDict, RequestList):
-            copied_node.metadata.append("accepted-only")
-            assert source_node.metadata == snapshot_node.metadata == ["original"]
-        result.payload["tags"].add("accepted-only")
-        assert original["tags"] == result.original_payload["tags"] == {"original"}
diff --git a/tests/hermes_cli/test_plugins.py b/tests/hermes_cli/test_plugins.py
index 21da587110..3fe75a3b2a 100644
--- a/tests/hermes_cli/test_plugins.py
+++ b/tests/hermes_cli/test_plugins.py
@@ -2718,76 +2718,3 @@ class TestAsyncHookOnCallerLoop:
             results = asyncio.run(mgr.ainvoke_hook("pre_gateway_dispatch", event="e", gateway="g"))
         assert results == [{"seen": "e"}, {"seen_async": "e"}]
         assert "async plugin blew up" in caplog.text
-
-
-def _load_request_chain_plugin(tmp_path, monkeypatch, failing=False):
-    from hermes_cli import plugins
-    home = tmp_path / "home"
-    monkeypatch.setenv("HERMES_HOME", str(home))
-    plugin = _make_plugin_dir(home / "plugins", "request_chain")
-    plugin.joinpath("__init__.py").write_text("""
-def register(ctx):
-    for kind, key in [('llm_request', 'request'), ('tool_request', 'args')]:
-        def first(_key=key, **kw):
-            return {_key: {**kw[_key], 'steps': ['first']}, 'source': 'first'}
-        def broken(_key=key, **kw):
-            kw[_key]['steps'].append('broken')
-            kw[_key]['input'].append('broken')
-            raise RuntimeError('fixture failure')
-        def observer(_key=key, **kw):
-            kw[_key]['steps'].append('unaccepted')
-        def second(_key=key, **kw):
-            return {_key: {**kw[_key], 'second': True}, 'source': 'second'}
-        ctx.register_middleware(kind, first)
-        if FAILING:
-            ctx.register_middleware(kind, broken)
-            ctx.register_middleware(kind, observer)
-        ctx.register_middleware(kind, second)
-""".replace("FAILING", str(failing)))
-    manager = PluginManager()
-    manager.discover_and_load()
-    monkeypatch.setattr(plugins, "get_plugin_manager", lambda: manager)
-    assert manager.has_middleware("llm_request")
-    return manager
-
-
-def test_request_middleware_chains_discovered_plugins(tmp_path, monkeypatch):
-    _load_request_chain_plugin(tmp_path, monkeypatch)
-    for apply in (apply_llm_request_middleware,
-                  lambda payload: apply_tool_request_middleware("fixture", payload)):
-        original = {"input": ["original"]}
-        result = apply(original)
-        assert result.payload == {**original, "steps": ["first"], "second": True}
-        assert result.original_payload == original == {"input": ["original"]}
-        assert [entry["source"] for entry in result.trace] == ["first", "second"]
-
-
-def test_failed_or_observer_middleware_cannot_mutate_accepted_rewrites(tmp_path, monkeypatch):
-    _load_request_chain_plugin(tmp_path, monkeypatch, failing=True)
-    # Real provider payloads can contain opaque handles. One such leaf must not
-    # turn every nested request container into a shared, mutable reference.
-    with (tmp_path / "upload.txt").open("w+", encoding="utf-8") as handle:
-        for opaque in ({}, {"file": handle}):
-            for apply in (apply_llm_request_middleware,
-                          lambda payload: apply_tool_request_middleware("fixture", payload)):
-                original = {"input": ["original"], **opaque}
-                shared = {"tags": {"original"}}
-                original["metadata"] = (shared, shared)
-                if opaque:
-                    cycle = []
-                    original["cycle"] = (cycle, handle)
-                    cycle.append(original["cycle"])
-                result = apply(original)
-                assert result.payload["steps"] == ["first"]
-                assert result.payload["second"] is True
-                assert original["input"] == result.original_payload["input"] == ["original"]
-                assert result.payload["input"] == ["original"]
-                assert [entry["source"] for entry in result.trace] == ["first", "second"]
-                assert result.payload["metadata"][0] is result.payload["metadata"][1]
-                result.payload["metadata"][0]["tags"].add("accepted-only")
-                assert shared["tags"] == {"original"}
-                assert result.original_payload["metadata"][0]["tags"] == {"original"}
-                if opaque:
-                    assert result.payload["file"] is handle
-                    assert not handle.closed
-                    assert result.payload["cycle"][0][0] is result.payload["cycle"]
diff --git a/tests/hermes_cli/test_plugins_middleware_chain.py b/tests/hermes_cli/test_plugins_middleware_chain.py
new file mode 100644
index 0000000000..5f32a996aa
--- /dev/null
+++ b/tests/hermes_cli/test_plugins_middleware_chain.py
@@ -0,0 +1,67 @@
+"""Request middleware chaining (#128638): rewrites compose in registration order."""
+
+import pytest
+
+from hermes_cli import plugins
+from hermes_cli.middleware import apply_llm_request_middleware, apply_tool_request_middleware
+from hermes_cli.plugins import PluginManager
+
+_REQUEST_KINDS = [
+    ("llm_request", apply_llm_request_middleware),
+    ("tool_request", lambda payload: apply_tool_request_middleware("fixture", payload)),
+]
+
+# Per request kind: ``first`` rewrites, ``meddler`` mutates its views in place and returns None,
+# ``second`` records what it saw and rewrites.
+_PLUGIN = """
+SEEN = {}
+
+def register(ctx):
+    for kind, key in [('llm_request', 'request'), ('tool_request', 'args')]:
+        def first(_key=key, **kw):
+            return {_key: {**kw[_key], 'steps': ['first']}, 'source': 'first'}
+        def meddler(_key=key, **kw):
+            kw[_key].setdefault('steps', []).append('meddler')
+            kw['original_' + _key]['input'].append('meddler')
+        def second(_key=key, _kind=kind, **kw):
+            SEEN[_kind] = (kw[_key], kw['original_' + _key])
+            return {_key: {**kw[_key], 'second': True}, 'source': 'second'}
+        for callback in (first, meddler, second):
+            ctx.register_middleware(kind, callback)
+"""
+
+
+@pytest.fixture
+def seen(tmp_path, monkeypatch):
+    home = tmp_path / "home"
+    plugin = home / "plugins" / "request_chain"
+    plugin.mkdir(parents=True)
+    (plugin / "plugin.yaml").write_text("name: request_chain\nversion: 0.1.0\n")
+    (plugin / "__init__.py").write_text(_PLUGIN)
+    (home / "config.yaml").write_text("plugins:\n  enabled: [request_chain]\n")
+    monkeypatch.setenv("HERMES_HOME", str(home))
+    manager = PluginManager()
+    manager.discover_and_load()
+    monkeypatch.setattr(plugins, "get_plugin_manager", lambda: manager)
+    return manager._plugins["request_chain"].module.SEEN
+
+
+@pytest.mark.parametrize("kind,apply", _REQUEST_KINDS, ids=[k for k, _ in _REQUEST_KINDS])
+def test_request_middleware_rewrites_compose(seen, kind, apply):
+    result = apply({"input": ["original"]})
+
+    # The second callback sees the first one's rewrite, not the original request.
+    assert seen[kind][0] == {"input": ["original"], "steps": ["first"]}
+    assert result.payload == {"input": ["original"], "steps": ["first"], "second": True}
+    assert [entry["source"] for entry in result.trace] == ["first", "second"]
+
+
+@pytest.mark.parametrize("kind,apply", _REQUEST_KINDS, ids=[k for k, _ in _REQUEST_KINDS])
+def test_request_middleware_in_place_mutation_does_not_leak(seen, kind, apply):
+    original = {"input": ["original"]}
+
+    result = apply(original)
+
+    # Each callback gets its own copy of the payload and of the pre-middleware snapshot.
+    assert "meddler" not in result.payload["steps"]
+    assert seen[kind][1] == result.original_payload == original == {"input": ["original"]}