diff --git a/agent/context_engine.py b/agent/context_engine.py
index 91c055383b..b772125c0c 100644
--- a/agent/context_engine.py
+++ b/agent/context_engine.py
@@ -286,12 +286,21 @@ class ContextEngine(ABC):
     ) -> None:
         """Observe a finished user turn (post-turn ingestion / observation).
 
-        Called once after the assistant/tool loop completes for a turn, with
-        the finalized in-memory transcript snapshot. This is the complement to
-        ``select_context()``: selection happens *before* the request, while
-        observation happens *after* the turn. It lets an engine ingest, index,
-        summarize, or update routing / topic / session state from what actually
-        happened — so the next ``select_context()`` can act on it.
+        Called from the standard turn-finalization path once the assistant/tool
+        loop completes, with the finalized in-memory transcript snapshot. This
+        is the complement to ``select_context()``: selection happens *before*
+        the request, while observation happens *after* the turn. It lets an
+        engine ingest, index, summarize, or update routing / topic / session
+        state from what actually happened — so the next ``select_context()``
+        can act on it.
+
+        Coverage: this fires from the normal finalization seam. Some abnormal
+        early-return paths in the loop (e.g. a content-policy block or a
+        provider terminal failure) persist and return without routing through
+        finalization, and therefore do not currently emit this hook. Treat it
+        as a best-effort post-turn observation for completed turns, not a
+        guaranteed callback for every possible early exit; unifying all
+        terminal paths behind one finalization seam is a separate follow-up.
 
         Together the two hooks remove the need to abuse ``should_compress()`` /
         ``compress()`` as a generic per-turn callback just to observe history,
@@ -310,8 +319,8 @@ class ContextEngine(ABC):
         ``input_tokens`` / ``output_tokens`` / ``cache_read_tokens`` /
         ``cache_write_tokens`` / ``reasoning_tokens`` buckets) so an engine can
         weigh how large/expensive the selected context actually was when
-        deciding the next ``select_context()``. It is ``None`` on turns that
-        never reached a provider response (early failure / interrupt); engines
+        deciding the next ``select_context()``. It is ``None`` on finalized
+        turns that never reached a provider response (e.g. interrupt); engines
         must treat it as optional.
 
         Default is a no-op.
diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py
index c11694ed49..ef4b48cc12 100644
--- a/agent/conversation_loop.py
+++ b/agent/conversation_loop.py
@@ -740,11 +740,21 @@ def _apply_context_engine_selection(
         return api_messages
 
     session_label = getattr(agent, "session_id", None) or "-"
+    # Pass shallow copies of the reference-only inputs so an engine that
+    # mutates them in place cannot alter persisted transcript state. Only
+    # ``request_messages`` (the per-call request list) is meant to be acted on,
+    # and it may be replaced wholesale via the return value — never mutated in
+    # place either. ``conversation_messages`` / ``incoming_message`` are
+    # read-only context; copying enforces the request-only contract rather than
+    # merely documenting it.
+    _conv_copy = [dict(m) if isinstance(m, dict) else m for m in conversation_messages] \
+        if conversation_messages is not None else None
+    _incoming_copy = dict(incoming_message) if isinstance(incoming_message, dict) else incoming_message
     try:
         selected = engine.select_context(
             api_messages,
-            conversation_messages=conversation_messages,
-            incoming_message=incoming_message,
+            conversation_messages=_conv_copy,
+            incoming_message=_incoming_copy,
             budget_tokens=getattr(engine, "context_length", 0) or 0,
         )
     except Exception:
diff --git a/tests/agent/test_context_engine_select_context.py b/tests/agent/test_context_engine_select_context.py
index 906441870e..350acc360f 100644
--- a/tests/agent/test_context_engine_select_context.py
+++ b/tests/agent/test_context_engine_select_context.py
@@ -196,6 +196,42 @@ def test_empty_list_keeps_original_request():
     assert logger.warning.called
 
 
+def test_engine_mutating_inputs_cannot_corrupt_persisted_state():
+    """An engine that mutates its read-only inputs in place must not affect the
+    persisted conversation history / incoming message.
+
+    ``conversation_messages`` and ``incoming_message`` are reference-only
+    context. The host passes shallow copies, so even a misbehaving engine that
+    appends to / edits them in ``select_context()`` cannot alter the live
+    persisted objects. Enforces the request-only contract (not just documents).
+    """
+    history = [{"role": "user", "content": "hello"}]
+    incoming = history[-1]
+    history_snapshot = [dict(m) for m in history]
+    incoming_snapshot = dict(incoming)
+
+    class _Engine(_MinimalEngine):
+        def select_context(self, request_messages, *, conversation_messages=None,
+                            incoming_message=None, **kwargs):
+            # Misbehaving engine: mutate the read-only inputs in place.
+            if conversation_messages is not None:
+                conversation_messages.append({"role": "user", "content": "INJECTED"})
+                if conversation_messages and isinstance(conversation_messages[0], dict):
+                    conversation_messages[0]["content"] = "TAMPERED"
+            if isinstance(incoming_message, dict):
+                incoming_message["content"] = "TAMPERED"
+            return None
+
+    agent = _agent_with(_Engine())
+    _apply_context_engine_selection(
+        agent, REQUEST, history, incoming, logger=MagicMock()
+    )
+    # Persisted history + incoming message are untouched despite the engine's
+    # in-place mutation of the copies it received.
+    assert history == history_snapshot
+    assert incoming == incoming_snapshot
+
+
 def test_persisted_history_not_mutated():
     """The hook must not mutate the persisted conversation history."""