diff --git a/plugins/memory/hindsight/__init__.py b/plugins/memory/hindsight/__init__.py
index 20236b0a4e69f..1f10d6a62a608 100644
--- a/plugins/memory/hindsight/__init__.py
+++ b/plugins/memory/hindsight/__init__.py
@@ -1022,19 +1022,25 @@ def _retain_batch(self, item: dict, *, bank_id: str, document_id: str | None = N
 
     def _make_turn_retain_job(self, turns: list[str], *, document_id: str, update_mode: str | None,
                               label: str, track_ops: bool = True) -> Callable[[], None]:
-        """Writer job shipping *turns* as one document. Inputs are snapshotted NOW: the
-        writer runs after later sync_turn() calls mutate _session_turns/_turn_index/_session_id."""
-        content = "[" + ",".join(turns) + "]"
-        metadata = self._build_metadata(message_count=len(turns) * 2, turn_index=self._turn_index)
-        lineage = (("session", self._session_id), ("parent", self._parent_session_id))
-        tags = [f"{kind}:{sid}" for kind, sid in lineage if sid] or None
-        bank_id, retain_async, retain_context = self._bank_id, self._retain_async, self._retain_context
+        """Writer job shipping *turns* as one document. Everything is snapshotted NOW —
+        the item included — because the writer runs after later sync_turn() calls or an
+        on_session_switch() mutate the provider's mutable item-config attributes
+        (_retain_tags, _observation_scopes, _retain_source). Building the item at run
+        time would stamp an OLD-session retain with NEW-session tags/scopes."""
+        item = self._build_retain_kwargs(
+            "[" + ",".join(turns) + "]",
+            context=self._retain_context,
+            metadata=self._build_metadata(message_count=len(turns) * 2, turn_index=self._turn_index),
+            tags=[f"{kind}:{sid}" for kind, sid in
+                  (("session", self._session_id), ("parent", self._parent_session_id)) if sid] or None,
+            update_mode=update_mode,
+        )
+        bank_id, retain_async = self._bank_id, self._retain_async
 
         def _job() -> None:
-            item = self._build_retain_kwargs(content, context=retain_context, metadata=metadata,
-                                             tags=tags, update_mode=update_mode)
             logger.debug("Hindsight %s: bank=%s, doc=%s, mode=%s, async=%s, content_len=%d, num_turns=%d",
-                         label, bank_id, document_id, update_mode, retain_async, len(content), len(turns))
+                         label, bank_id, document_id, update_mode, retain_async,
+                         len(item["content"]), len(turns))
             resp = self._retain_batch(item, bank_id=bank_id, document_id=document_id, retain_async=retain_async)
             # Async retains are only *accepted* here; track the op id(s) so the
             # next-turn prefetch can wait for true server-side completion.
diff --git a/tests/plugins/memory/test_hindsight_provider.py b/tests/plugins/memory/test_hindsight_provider.py
index 93a429464e55a..a17a77168b0dc 100644
--- a/tests/plugins/memory/test_hindsight_provider.py
+++ b/tests/plugins/memory/test_hindsight_provider.py
@@ -1777,3 +1777,61 @@ def test_append_mode_trims_retained_turns_without_dropping_any(provider, monkeyp
     assert len(provider._session_turns) == 1  # only the un-retained tail (turn 7)
     assert provider._last_retained_turn_count == 0
     assert len(shipped) == 6 and len(set(shipped)) == 6
+
+
+# ---------------------------------------------------------------------------
+# Retain identity isolation — a queued retain must ship the identity and item
+# config that were authoritative at ENQUEUE time, not whatever the provider
+# holds when the writer thread eventually drains the queue.
+# ---------------------------------------------------------------------------
+
+
+class TestRetainIdentityIsolation:
+    """Regression for the enqueue/run-time identity gap.
+
+    ``_make_turn_retain_job`` snapshots turns/metadata/lineage/bank_id at enqueue
+    time, but the writer job still calls ``_build_retain_kwargs`` at run time,
+    which re-reads mutable per-session attributes (``_retain_tags``,
+    ``_observation_scopes``, ``_retain_source``). A session switch in between
+    stamps a queued OLD-session retain with the NEW session's item config.
+    """
+
+    def _gated_provider(self, provider_with_config):
+        p = provider_with_config(
+            retain_tags=["old-tag"],
+            retain_source="old-source",
+            observation_scopes="per_tag",
+        )
+        p._client = _make_mock_client()
+
+        # Park the writer exactly between dequeue and job execution so the main
+        # thread can mutate provider state while the job is still pending.
+        gate = threading.Event()
+        released = threading.Event()
+
+        def _gate():
+            gate.set()
+            released.wait(timeout=5.0)
+
+        p._retain_queue.put(_gate)
+        return p, gate, released
+
+    def test_queued_retain_keeps_enqueue_time_item_config(self, provider_with_config):
+        p, gate, released = self._gated_provider(provider_with_config)
+        p.sync_turn("old-user", "old-assistant")
+        gate.wait(timeout=5.0)
+
+        # Mutate the item-config attributes the writer must NOT observe.
+        p._retain_tags = ["new-tag"]
+        p._observation_scopes = "all_combinations"
+        p._retain_source = "new-source"
+
+        released.set()
+        p._retain_queue.join()
+
+        p._client.aretain_batch.assert_called_once()
+        item = p._client.aretain_batch.call_args.kwargs["items"][0]
+        assert item["tags"] == ["old-tag", "session:test-session"]
+        assert item["observation_scopes"] == "per_tag"
+        assert item["metadata"]["source"] == "old-source"
+