fix: clone plugin context engines via clone_for_agent(), not blind deepcopy

A general-plugin context engine is one shared instance; agent init copied it per agent
with copy.deepcopy() only. Engines that hold a SQLite connection or lock (hermes-lcm)
already expose clone_for_agent() for exactly this, but it was never called, so every
init logged "could not be safely copied … falling back to built-in compressor" and the
engine was unusable through the plugin system.

ContextEngine grows clone_for_agent() (default: deepcopy, the previous behaviour) and
_select_context_engine calls it; the failure message now names the hook to override.
Docs: context-engine-plugin.md documents the per-agent clone contract.

Test change (existing on main): tests/agent/test_context_engine.py::
test_agent_init_source_deepcopies_singleton_not_aliases was a source-reading pin on the
literal `copy.deepcopy(_candidate)` line, which this fix intentionally replaces. It is
superseded by tests/agent/test_plugin_context_engine_clone.py, which drives the real
_select_context_engine seam and asserts the invariant it guarded (child update_model()
never mutates the shared singleton) plus the new clone_for_agent() path.

Fixes #99640
credit: @stephenschoettler #62374
credit: @686f6c61 #99677

agent/agent_init.py

@@ -1870,22 +1870,20 @@ def _select_context_engine(_agent_cfg):
         except Exception:
             _candidate = None
         if _candidate is not None and _candidate.name == _engine_name:
-            # Deep-copy the shared singleton so a child's update_model() can't mutate the
-            # parent's. Uncopyable state (locks, DB conns) → built-in with an ACCURATE message.
-            import copy
+            # The plugin system holds ONE shared instance; each agent gets its own so a child's
+            # update_model() can't mutate the parent's (#42449). clone_for_agent() defaults to
+            # deepcopy; engines with uncopyable state (locks, DB conns) override it. A failure
+            # falls back to the built-in compressor with an ACCURATE message, not "not found".
             try:
-                # Copy can fail for engines holding uncopyable state (locks, DB connections, clients); in
-                # that case fall back to the built-in compressor with an ACCURATE message rather than
-                # silently mislabelling it "not found". See #42449.
-                _selected_engine = copy.deepcopy(_candidate)
+                _selected_engine = _candidate.clone_for_agent()
             except Exception as _copy_err:
                 _copy_failed = True
                 _ra().logger.warning(
                     "Context engine '%s' could not be safely copied for this "
                     "agent (%s) — falling back to built-in compressor. Plugin "
                     "engines that hold uncopyable state (locks, DB connections) "
-                    "should implement __deepcopy__ to copy only mutable budget "
-                    "state.",
+                    "should override clone_for_agent() (or __deepcopy__) to copy "
+                    "only mutable budget state.",
                     _engine_name, _copy_err,
                 )
 

agent/context_engine.py

@@ -8,6 +8,7 @@ should_compress() / compress() -> on_session_end() at real session boundaries on
 (CLI exit, /reset, gateway expiry), never per-turn.
 """
 
+import copy
 import json
 from abc import ABC, abstractmethod
 from typing import Any, Dict, List, Optional
@@ -222,6 +223,13 @@ class ContextEngine(ABC):
             "compression_count": self.compression_count,
         }
 
+    def clone_for_agent(self) -> "ContextEngine":
+        """Per-agent instance of a plugin-registered engine (the plugin system holds ONE shared
+        instance; every AIAgent gets its own so a child's update_model() cannot mutate the parent's).
+        Override when the engine holds uncopyable state (locks, DB connections): return a fresh
+        engine sharing the durable backend and copying only mutable budget state."""
+        return copy.deepcopy(self)
+
     def update_model(
         self, model: str, context_length: int, base_url: str = "", api_key: str = "",
         provider: str = "", api_mode: str = "",

tests/agent/test_context_engine.py

@@ -237,13 +237,9 @@ class TestInitAgentDoesNotMutatePluginSingleton:
     """Regression coverage for #42449: a child agent's init must not mutate the
     shared plugin context-engine singleton via update_model().
 
-    Note: ``test_child_init_does_not_corrupt_parent_singleton`` replicates the
-    init_agent selection-block *pattern* (it cannot cheaply spin up a full
-    init_agent), so it documents/verifies the deepcopy approach but does NOT by
-    itself guard a production revert. The real revert guard is
-    ``test_agent_init_source_deepcopies_singleton_not_aliases`` (source-pin),
-    and ``test_unpicklable_engine_falls_back_gracefully`` covers the
-    copy-failure path.
+    Note: these replicate the init_agent selection-block *pattern*; the production
+    seam (``_select_context_engine`` → ``clone_for_agent()``) is driven directly by
+    ``tests/agent/test_plugin_context_engine_clone.py``.
     """
 
     def test_child_init_does_not_corrupt_parent_singleton(self, monkeypatch):
@@ -317,31 +313,3 @@ class TestInitAgentDoesNotMutatePluginSingleton:
         assert selected is None
         # The original engine is untouched (no partial mutation).
         assert engine.context_length == 1_000_000
-
-    def test_agent_init_source_deepcopies_singleton_not_aliases(self):
-        """Source-pin guarding the production fix in agent/agent_init.py:
-        the plugin-singleton fallback MUST deepcopy the candidate, not alias
-        it (`_selected_engine = _candidate`). Full init_agent is too heavy to
-        drive here, so this pins the exact line so a future revert to direct
-        assignment fails CI. Regression for #42449."""
-        import inspect
-        import re
-        import agent.agent_init as _ai
-
-        src = inspect.getsource(_ai)
-        # The candidate fetched from the plugin singleton must be deep-copied
-        # before becoming _selected_engine (which is later mutated by
-        # update_model). A bare `_selected_engine = _candidate` is the bug.
-        assert re.search(
-            r"_selected_engine\s*=\s*(copy|_copy)\.deepcopy\(\s*_candidate\s*\)",
-            src,
-        ), (
-            "agent_init must deepcopy the plugin context-engine singleton "
-            "(`_selected_engine = copy.deepcopy(_candidate)`) — a bare "
-            "`_selected_engine = _candidate` re-introduces #42449 (child "
-            "update_model corrupts the parent's shared singleton)."
-        )
-        # And the bug-shape alias must NOT be present on that path.
-        assert not re.search(
-            r"_selected_engine\s*=\s*_candidate\b", src
-        ), "found the #42449 bug-shape alias `_selected_engine = _candidate`"

tests/agent/test_plugin_context_engine_clone.py

@@ -0,0 +1,62 @@
+"""A plugin-registered context engine is one shared instance; agent init hands each agent its own
+copy through ``clone_for_agent()`` (default deepcopy), so engines with uncopyable state (locks,
+SQLite connections — hermes-lcm) stay selectable and a child's model never leaks into the parent
+(#99640, #42449)."""
+
+import threading
+from unittest.mock import patch
+
+from agent.agent_init import _select_context_engine
+from agent.context_engine import ContextEngine
+
+
+class _Engine(ContextEngine):
+    engine_name = "lcm"
+
+    @property
+    def name(self):
+        return self.engine_name
+
+    def update_from_response(self, usage):
+        pass
+
+    def should_compress(self, prompt_tokens=None):
+        return False
+
+    def compress(self, messages, current_tokens=None):
+        return messages
+
+
+class _LockedEngine(_Engine):
+    """Holds a lock (deepcopy raises) and hands out per-agent clones like hermes-lcm does."""
+
+    def __init__(self):
+        super().__init__()
+        self._lock = threading.Lock()
+        self.clones = 0
+
+    def clone_for_agent(self):
+        self.clones += 1
+        return _LockedEngine()
+
+
+def _select(engine):
+    with (patch("plugins.context_engine.load_context_engine", return_value=None),
+          patch("hermes_cli.plugins.get_plugin_context_engine", return_value=engine)):
+        return _select_context_engine({"context": {"engine": engine.name}})
+
+
+def test_engine_with_uncopyable_state_is_selected_via_clone_for_agent():
+    singleton = _LockedEngine()
+    selected = _select(singleton)
+    assert isinstance(selected, _LockedEngine) and selected is not singleton
+    assert singleton.clones == 1
+
+
+def test_default_clone_isolates_parent_from_child_update_model():
+    singleton = _Engine()
+    singleton.update_model(model="big", context_length=1_000_000)
+    child = _select(singleton)
+    child.update_model(model="small", context_length=204_800)
+    assert child is not singleton
+    assert (singleton.context_length, child.context_length) == (1_000_000, 204_800)

website/docs/developer-guide/context-engine-plugin.md

@@ -99,6 +99,7 @@ These have sensible defaults in the ABC. Override as needed:
 | `get_status()` | Standard token/threshold dict | You have custom metrics to expose |
 | `select_context(request_messages, *, conversation_messages, incoming_message, budget_tokens)` | Returns `None` (no-op) | You select/route which context enters **this** request (retrieval, topic routing) — see below |
 | `on_turn_complete(messages, usage=None, **kwargs)` | No-op | You ingest/index/observe the finished turn — see below |
+| `clone_for_agent()` | `copy.deepcopy(self)` | Your engine holds uncopyable state (locks, SQLite/DB connections) — see [Via general plugin system](#via-general-plugin-system) |
 
 ## Per-turn context selection and observation
 
@@ -204,6 +205,21 @@ def register(ctx):
 
 Only one engine can be registered. A second plugin attempting to register is rejected with a warning.
 
+The registered instance is shared process-wide, but every `AIAgent` (parent, subagents, gateway
+sessions) needs its own engine so a child's `update_model()` cannot mutate the parent's budget.
+Hermes therefore calls `engine.clone_for_agent()` on the registered instance at each agent init.
+The default is `copy.deepcopy(self)`; override it when the engine holds state that cannot be
+deep-copied (locks, SQLite or HTTP connections) and return a fresh engine sharing the durable
+backend while copying only the mutable budget fields. If the clone raises, the agent falls back to
+the built-in compressor and logs `Context engine 'X' could not be safely copied for this agent`.
+
+```python
+def clone_for_agent(self):
+    clone = LCMEngine(db_path=self.db_path)  # reopens its own connection
+    clone.threshold_percent = self.threshold_percent
+    return clone
+```
+
 ## Lifecycle
 
 ```