refactor(memory): move memory-context fencing into agent/memory_context.py

agent/memory_manager.py mixes provider fan-out, background work and the
text formatting that fences recalled memory for the model. The formatting
half (sanitize_context, StreamingContextScrubber, build_memory_context_block
and their regexes) is pure text work with no provider calls or profile
state, so it moves unchanged to its own module. Callers import it from
there; nothing in plugins/ imported it, so no re-export is kept.

The only visible difference is the logger name on the "pre-wrapped
context" warning (agent.memory_context). This leaves the manager room for
the profile-ownership checks the next commit adds without crossing its
line cap.

Refs #134889 (PLG-11)
Refs #140208

agent/agent_init.py

@@ -27,7 +27,7 @@ from agent.agent_init_fallback import _fallback_entries, _init_fallback_chain, r
 from agent.agent_init_params import _CALLBACK_PARAMS, _GATEWAY_IDENTITY_PARAMS, _PASSTHROUGH_PARAMS
 from agent.agent_runtime_helpers import _ra
 from agent.iteration_budget import IterationBudget, normalize_budget_warning_ratio
-from agent.memory_manager import StreamingContextScrubber
+from agent.memory_context import StreamingContextScrubber
 from agent.memory_provider import is_core_memory_provider
 from agent.session_activity import ActivityProvenance
 from agent.model_metadata import (

agent/memory_context.py

@@ -0,0 +1,185 @@
+"""Memory-context fencing: how recalled memory is wrapped for the model and kept out of output.
+
+``build_memory_context_block`` fences prefetched recall for the prompt; ``sanitize_context`` and
+``StreamingContextScrubber`` strip those fences (and anything a provider pre-wrapped) from text that
+leaves the agent. Pure text formatting: no provider calls, no profile state.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+
+logger = logging.getLogger(__name__)
+
+_FENCE_TAG_RE = re.compile(r'</?\s*memory-context\s*>', re.IGNORECASE)
+_INTERNAL_CONTEXT_RE = re.compile(r'<\s*memory-context\s*>[\s\S]*?</\s*memory-context\s*>', re.IGNORECASE)
+_INTERNAL_NOTE_RE = re.compile(
+    r'\[System note:\s*The following is recalled memory context,\s*NOT new user input\.\s*Treat as (?:informational background data|authoritative reference data[^\]]*)\.\]\s*',
+    re.IGNORECASE,
+)
+
+
+def sanitize_context(text: str) -> str:
+    """Strip fence tags, injected context blocks, and system notes from provider output."""
+    for pattern in (_INTERNAL_CONTEXT_RE, _INTERNAL_NOTE_RE, _FENCE_TAG_RE):
+        text = pattern.sub('', text)
+    return text
+
+
+class StreamingContextScrubber:
+    """Stateful scrubber for streaming text whose memory-context spans may straddle deltas.
+
+    ``sanitize_context`` needs both tags in one string, so a split span would leak to the UI;
+    this holds back partial-tag tails between ``feed()`` calls and drops span interiors.
+    One scrubber (or ``reset()``) per top-level response; call ``flush()`` at end of stream.
+    """
+
+    _OPEN_TAG = "<memory-context>"
+    _CLOSE_TAG = "</memory-context>"
+
+    def __init__(self) -> None:
+        self.reset()
+
+    def reset(self) -> None:
+        self._in_span: bool = False
+        self._buf: str = ""
+        self._at_block_boundary: bool = True
+
+    def feed(self, text: str) -> str:
+        """Return the visible portion of ``text``; a possible partial tag tail is held for the next call."""
+        if not text:
+            return ""
+        buf = self._buf + text
+        self._buf = ""
+        out: list[str] = []
+        while buf:
+            if self._in_span:
+                tag = self._CLOSE_TAG
+                idx = buf.lower().find(tag)
+                held = self._max_partial_suffix(buf, tag)  # potential partial close tag
+            else:
+                tag = self._OPEN_TAG
+                idx = self._find_boundary_open_tag(buf)
+                # A complete boundary tag at the buffer end is held until the next char confirms it.
+                n = len(tag)
+                pending = n if buf.lower().endswith(tag) and self._ends_at_block_boundary(buf[:-n]) else 0
+                held = pending or self._max_partial_suffix(buf, tag)
+            if idx == -1:
+                # Hold back the possible partial tag; inside a span the rest is dropped.
+                if not self._in_span:
+                    self._append_visible(out, buf[:-held] if held else buf)
+                self._buf = buf[-held:] if held else ""
+                break
+            if not self._in_span:
+                self._append_visible(out, buf[:idx])
+            buf = buf[idx + len(tag):]
+            self._in_span = not self._in_span
+        return "".join(out)
+
+    def flush(self) -> str:
+        """Emit the held-back tail at end-of-stream; inside an unterminated span it is discarded
+        (leaking partial memory context is worse than a truncated answer)."""
+        tail = "" if self._in_span else self._buf
+        self._buf = ""
+        self._in_span = False
+        return tail
+
+    @staticmethod
+    def _max_partial_suffix(buf: str, tag: str) -> int:
+        """Length of the longest buf-suffix that is a (case-insensitive) prefix of ``tag``, else 0."""
+        tag_lower, buf_lower = tag.lower(), buf.lower()
+        span = range(min(len(buf_lower), len(tag_lower) - 1), 0, -1)
+        return next((i for i in span if tag_lower.startswith(buf_lower[-i:])), 0)
+
+    def _find_boundary_open_tag(self, buf: str) -> int:
+        """Find an opening fence only when it starts a block-like span (own line, newline after)."""
+        buf_lower, tag_len = buf.lower(), len(self._OPEN_TAG)
+        idx = buf_lower.find(self._OPEN_TAG)
+        while idx != -1:
+            after_idx = idx + tag_len
+            if self._ends_at_block_boundary(buf[:idx]) and after_idx < len(buf) and buf[after_idx] in "\r\n":
+                return idx
+            idx = buf_lower.find(self._OPEN_TAG, idx + 1)
+        return -1
+
+    def _ends_at_block_boundary(self, text: str) -> bool:
+        """Whether emitting ``text`` leaves the stream at a line start (blank tail after the last newline;
+        no newline at all -> only whitespace and already at a boundary)."""
+        _head, sep, tail = text.rpartition("\n")
+        return tail.strip() == "" and (bool(sep) or self._at_block_boundary)
+
+    def _append_visible(self, out: list[str], text: str) -> None:
+        if text:
+            out.append(text)
+            self._at_block_boundary = self._ends_at_block_boundary(text)
+
+
+# A markdown bullet: a marker, whitespace, then content. The whitespace matters — it is what keeps
+# ``**Preferences**`` (a bold heading) and ``*emphasis*`` out of the rule.
+_RECALL_BULLET_RE = re.compile(r"[-*+]\s+\S")
+
+
+def _drop_repeated_recall_lines(text: str) -> str:
+    """Drop a recalled bullet that an EARLIER line of this same block already states.
+
+    Providers merge several stores (and this merges several providers), so one prefetch routinely
+    surfaces the same fact two or three times. A byte-identical repeat inside one block tells the
+    model nothing the block has not already said, and it is not free: the composed block is stamped
+    into the user row's ``api_content`` sidecar and replayed verbatim on every later request for as
+    long as that row is in context, so each duplicate is paid once per turn, forever.
+
+    The ``seen`` set is scoped per section — every non-bullet line at column 0 (a heading of any
+    style, a ``---`` rule, prose) starts a new one — so a repeat is only dropped when the SAME
+    section already states it.
+
+    Only a SELF-CONTAINED bullet is considered — a marker, whitespace, content, and no continuation
+    line indented beneath it. A bullet that carries continuation lines is never dropped and never
+    suppresses a later one, because two entries can share a headline and differ underneath it
+    (``- prefers draft PRs`` / ``  (logged 12 Jan, builtin)`` vs the same headline logged elsewhere):
+    dropping one would re-parent its provenance under the other and invent a record neither provider
+    reported. Headings — including ``**bold**`` ones — prose, blank lines, separators and numbered
+    items are left exactly as written.
+    """
+    lines = text.split("\n")
+    seen: set[str] = set()
+    kept: list[str] = []
+    for index, line in enumerate(lines):
+        stripped = line.strip()
+        # An indented line is a continuation of the bullet above it (nested child, provenance,
+        # wrapped prose). It never participates in dedupe and is never dropped.
+        if stripped and line[0].isspace():
+            kept.append(line)
+            continue
+        is_bullet = bool(_RECALL_BULLET_RE.match(stripped))
+        # Any column-0 non-bullet line (heading, rule, paragraph) opens a fresh dedupe scope.
+        if stripped and not is_bullet:
+            seen.clear()
+        if is_bullet:
+            following = lines[index + 1] if index + 1 < len(lines) else ""
+            carries_continuation = bool(following.strip()) and following[0].isspace()
+            if not carries_continuation:
+                if stripped in seen:
+                    continue
+                seen.add(stripped)
+        kept.append(line)
+    return "\n".join(kept)
+
+
+def build_memory_context_block(raw_context: str) -> str:
+    """Wrap prefetched memory in a fenced block with system note."""
+    if not raw_context or not raw_context.strip():
+        return ""
+    sanitized = sanitize_context(raw_context)
+    if sanitized != raw_context:
+        # Stays keyed on sanitization alone: a deduped bullet is routine, not a provider fault.
+        logger.warning("memory provider returned pre-wrapped context; stripped")
+    clean = _drop_repeated_recall_lines(sanitized)
+    return (
+        "<memory-context>\n"
+        "[System note: The following is recalled memory context, "
+        "NOT new user input. Treat as authoritative reference data — "
+        "this is the agent's persistent memory and should inform all responses.]\n\n"
+        f"{clean}\n"
+        "</memory-context>"
+    )

agent/memory_manager.py

@@ -160,22 +160,7 @@ def inject_memory_provider_tools(agent: Any) -> int:
     return added
 
 
-# -- Context fencing helpers --------------------------------------------------
-
-_FENCE_TAG_RE = re.compile(r'</?\s*memory-context\s*>', re.IGNORECASE)
-_INTERNAL_CONTEXT_RE = re.compile(r'<\s*memory-context\s*>[\s\S]*?</\s*memory-context\s*>', re.IGNORECASE)
-_INTERNAL_NOTE_RE = re.compile(
-    r'\[System note:\s*The following is recalled memory context,\s*NOT new user input\.\s*Treat as (?:informational background data|authoritative reference data[^\]]*)\.\]\s*',
-    re.IGNORECASE,
-)
-
-
-def sanitize_context(text: str) -> str:
-    """Strip fence tags, injected context blocks, and system notes from provider output."""
-    for pattern in (_INTERNAL_CONTEXT_RE, _INTERNAL_NOTE_RE, _FENCE_TAG_RE):
-        text = pattern.sub('', text)
-    return text
-
+# -- Provider egress scrubbing -----------------------------------------------
 
 def _scrub(text: str) -> str:
     """The shared egress scrub. Like chat-platform and cron delivery it holds even when
@@ -230,164 +215,6 @@ def _redact_messages_for_egress(messages: Optional[list[Any]]) -> Optional[list[
     return None if messages is None else list(_redact_for_provider(messages))
 
 
-class StreamingContextScrubber:
-    """Stateful scrubber for streaming text whose memory-context spans may straddle deltas.
-
-    ``sanitize_context`` needs both tags in one string, so a split span would leak to the UI;
-    this holds back partial-tag tails between ``feed()`` calls and drops span interiors.
-    One scrubber (or ``reset()``) per top-level response; call ``flush()`` at end of stream.
-    """
-
-    _OPEN_TAG = "<memory-context>"
-    _CLOSE_TAG = "</memory-context>"
-
-    def __init__(self) -> None:
-        self.reset()
-
-    def reset(self) -> None:
-        self._in_span: bool = False
-        self._buf: str = ""
-        self._at_block_boundary: bool = True
-
-    def feed(self, text: str) -> str:
-        """Return the visible portion of ``text``; a possible partial tag tail is held for the next call."""
-        if not text:
-            return ""
-        buf = self._buf + text
-        self._buf = ""
-        out: list[str] = []
-        while buf:
-            if self._in_span:
-                tag = self._CLOSE_TAG
-                idx = buf.lower().find(tag)
-                held = self._max_partial_suffix(buf, tag)  # potential partial close tag
-            else:
-                tag = self._OPEN_TAG
-                idx = self._find_boundary_open_tag(buf)
-                # A complete boundary tag at the buffer end is held until the next char confirms it.
-                n = len(tag)
-                pending = n if buf.lower().endswith(tag) and self._ends_at_block_boundary(buf[:-n]) else 0
-                held = pending or self._max_partial_suffix(buf, tag)
-            if idx == -1:
-                # Hold back the possible partial tag; inside a span the rest is dropped.
-                if not self._in_span:
-                    self._append_visible(out, buf[:-held] if held else buf)
-                self._buf = buf[-held:] if held else ""
-                break
-            if not self._in_span:
-                self._append_visible(out, buf[:idx])
-            buf = buf[idx + len(tag):]
-            self._in_span = not self._in_span
-        return "".join(out)
-
-    def flush(self) -> str:
-        """Emit the held-back tail at end-of-stream; inside an unterminated span it is discarded
-        (leaking partial memory context is worse than a truncated answer)."""
-        tail = "" if self._in_span else self._buf
-        self._buf = ""
-        self._in_span = False
-        return tail
-
-    @staticmethod
-    def _max_partial_suffix(buf: str, tag: str) -> int:
-        """Length of the longest buf-suffix that is a (case-insensitive) prefix of ``tag``, else 0."""
-        tag_lower, buf_lower = tag.lower(), buf.lower()
-        span = range(min(len(buf_lower), len(tag_lower) - 1), 0, -1)
-        return next((i for i in span if tag_lower.startswith(buf_lower[-i:])), 0)
-
-    def _find_boundary_open_tag(self, buf: str) -> int:
-        """Find an opening fence only when it starts a block-like span (own line, newline after)."""
-        buf_lower, tag_len = buf.lower(), len(self._OPEN_TAG)
-        idx = buf_lower.find(self._OPEN_TAG)
-        while idx != -1:
-            after_idx = idx + tag_len
-            if self._ends_at_block_boundary(buf[:idx]) and after_idx < len(buf) and buf[after_idx] in "\r\n":
-                return idx
-            idx = buf_lower.find(self._OPEN_TAG, idx + 1)
-        return -1
-
-    def _ends_at_block_boundary(self, text: str) -> bool:
-        """Whether emitting ``text`` leaves the stream at a line start (blank tail after the last newline;
-        no newline at all -> only whitespace and already at a boundary)."""
-        _head, sep, tail = text.rpartition("\n")
-        return tail.strip() == "" and (bool(sep) or self._at_block_boundary)
-
-    def _append_visible(self, out: list[str], text: str) -> None:
-        if text:
-            out.append(text)
-            self._at_block_boundary = self._ends_at_block_boundary(text)
-
-
-# A markdown bullet: a marker, whitespace, then content. The whitespace matters — it is what keeps
-# ``**Preferences**`` (a bold heading) and ``*emphasis*`` out of the rule.
-_RECALL_BULLET_RE = re.compile(r"[-*+]\s+\S")
-
-
-def _drop_repeated_recall_lines(text: str) -> str:
-    """Drop a recalled bullet that an EARLIER line of this same block already states.
-
-    Providers merge several stores (and this merges several providers), so one prefetch routinely
-    surfaces the same fact two or three times. A byte-identical repeat inside one block tells the
-    model nothing the block has not already said, and it is not free: the composed block is stamped
-    into the user row's ``api_content`` sidecar and replayed verbatim on every later request for as
-    long as that row is in context, so each duplicate is paid once per turn, forever.
-
-    The ``seen`` set is scoped per section — every non-bullet line at column 0 (a heading of any
-    style, a ``---`` rule, prose) starts a new one — so a repeat is only dropped when the SAME
-    section already states it.
-
-    Only a SELF-CONTAINED bullet is considered — a marker, whitespace, content, and no continuation
-    line indented beneath it. A bullet that carries continuation lines is never dropped and never
-    suppresses a later one, because two entries can share a headline and differ underneath it
-    (``- prefers draft PRs`` / ``  (logged 12 Jan, builtin)`` vs the same headline logged elsewhere):
-    dropping one would re-parent its provenance under the other and invent a record neither provider
-    reported. Headings — including ``**bold**`` ones — prose, blank lines, separators and numbered
-    items are left exactly as written.
-    """
-    lines = text.split("\n")
-    seen: set[str] = set()
-    kept: list[str] = []
-    for index, line in enumerate(lines):
-        stripped = line.strip()
-        # An indented line is a continuation of the bullet above it (nested child, provenance,
-        # wrapped prose). It never participates in dedupe and is never dropped.
-        if stripped and line[0].isspace():
-            kept.append(line)
-            continue
-        is_bullet = bool(_RECALL_BULLET_RE.match(stripped))
-        # Any column-0 non-bullet line (heading, rule, paragraph) opens a fresh dedupe scope.
-        if stripped and not is_bullet:
-            seen.clear()
-        if is_bullet:
-            following = lines[index + 1] if index + 1 < len(lines) else ""
-            carries_continuation = bool(following.strip()) and following[0].isspace()
-            if not carries_continuation:
-                if stripped in seen:
-                    continue
-                seen.add(stripped)
-        kept.append(line)
-    return "\n".join(kept)
-
-
-def build_memory_context_block(raw_context: str) -> str:
-    """Wrap prefetched memory in a fenced block with system note."""
-    if not raw_context or not raw_context.strip():
-        return ""
-    sanitized = sanitize_context(raw_context)
-    if sanitized != raw_context:
-        # Stays keyed on sanitization alone: a deduped bullet is routine, not a provider fault.
-        logger.warning("memory provider returned pre-wrapped context; stripped")
-    clean = _drop_repeated_recall_lines(sanitized)
-    return (
-        "<memory-context>\n"
-        "[System note: The following is recalled memory context, "
-        "NOT new user input. Treat as authoritative reference data — "
-        "this is the agent's persistent memory and should inform all responses.]\n\n"
-        f"{clean}\n"
-        "</memory-context>"
-    )
-
-
 class MemoryManager:
     """Builtin provider (always first) plus at most one external provider.
 

agent/session_persistence.py

@@ -19,7 +19,7 @@ from agent.context_compressor import (
     user_originated_turn_view,
 )
 from agent.lazy_forward import forward as _forward, forward_static as _forward_static
-from agent.memory_manager import sanitize_context
+from agent.memory_context import sanitize_context
 
 from agent.tool_dispatch_helpers import _is_multimodal_tool_result, _multimodal_text_summary
 from agent.trajectory import save_trajectory as _save_trajectory_to_file

agent/stream_delivery.py

@@ -8,7 +8,7 @@ import re
 import threading
 from typing import Any, Dict, List
 
-from agent.memory_manager import sanitize_context
+from agent.memory_context import sanitize_context
 from agent.message_content import flatten_message_text
 from agent.history_commentary import visible_commentary
 

agent/turn_context.py

@@ -19,7 +19,7 @@ from typing import Any, Dict, List, Mapping, Optional, Tuple
 
 from agent.conversation_compression import recover_rotated_compression_session
 from agent.iteration_budget import IterationBudget
-from agent.memory_manager import build_memory_context_block
+from agent.memory_context import build_memory_context_block
 from agent.memory_provider import is_trivial_prompt
 from agent.message_content import flatten_message_text
 from agent.lsp.deferred import forward_to_agent as forward_lsp_verdicts

gateway/run_inbound.py

@@ -1776,7 +1776,7 @@ class GatewayInboundMixin(GatewayInboundHooksMixin, GatewayPluginInjectionMixin)
         Description *and* local cache path are injected so the model understands the image without
         a tool call and can re-examine it with vision_analyze."""
         from tools.vision_tools import vision_analyze_tool
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
 
         analysis_prompt = (
             "Concisely describe this image in 2-4 sentences "

hermes_cli/cli_session_mixin.py

@@ -728,7 +728,7 @@ class CLISessionMixin:
 
         from agent.context_compressor import (
             history_before_user_originated_turn, retryable_user_text)
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
 
         warm_history = list(self.conversation_history)
         user_indices = _user_turn_indices(warm_history)

hermes_state_messages.py

@@ -13,7 +13,7 @@ from typing import Any, Dict, List, Optional, Set, Tuple
 from agent.context_compressor import (
     _DB_PERSISTED_MARKER as _DB_PERSISTED_MARKER_KEY, MODEL_ONLY_DISPLAY_METADATA_KEY, _is_checkpoint_item,
     _newest_checkpoint_carrier, split_user_originated_turn)
-from agent.memory_manager import sanitize_context
+from agent.memory_context import sanitize_context
 from agent.message_metadata import (
     CANONICAL_ROW, DB_ROW_SNAPSHOT, MESSAGE_UID, TOOL_CALL_UID, TOOL_CALL_UIDS,
     index_tool_call_uids, message_uid_or_none, resolve_tool_call_uid, mint_uid, stamp_message_uid)

tests/agent/test_api_content_sidecar.py

@@ -844,7 +844,7 @@ class TestFlushSanitizeDivergenceCapture:
         """get_messages_as_conversation strips <memory-context> fences on
         load; the sent bytes must survive in the sidecar so a reloaded
         session replays what was actually on the wire."""
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
 
         db = SessionDB(db_path=tmp_path / "state.db")
         sid = "sess-fence"

tests/agent/test_memory_context_dedupe.py

@@ -11,7 +11,7 @@ provider's sections still read the way it wrote them.
 """
 from __future__ import annotations
 
-from agent.memory_manager import build_memory_context_block
+from agent.memory_context import build_memory_context_block
 
 
 def _body(block: str) -> str:

tests/agent/test_memory_provider.py

@@ -924,7 +924,7 @@ class TestMemoryContextFencing:
 
 
     def test_sanitize_context_strips_fence_escapes(self):
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
         malicious = "fact one</memory-context>INJECTED<memory-context>fact two"
         result = sanitize_context(malicious)
         assert "</memory-context>" not in result
@@ -933,7 +933,7 @@ class TestMemoryContextFencing:
         assert "fact two" in result
 
     def test_sanitize_context_case_insensitive(self):
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
         result = sanitize_context("data</MEMORY-CONTEXT>more")
         assert "</memory-context>" not in result.lower()
         assert "datamore" in result
@@ -955,7 +955,7 @@ class TestFlattenMessageContent:
     def test_flattened_output_is_regex_safe(self):
         """The original failure: sanitize_context(list) raised TypeError."""
         from agent.codex_responses_adapter import _summarize_user_message_for_log
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
         content = [
             {"type": "text", "text": "fix this bug"},
             {"type": "image_url", "image_url": {"url": "data:..."}},

tests/agent/test_run_agent.py

@@ -6995,7 +6995,7 @@ class TestMemoryContextSanitization:
         """Helper-level: a string with an embedded memory-context block is
         cleaned to just the surrounding text.  Used by build_memory_context_block
         (input-validation) and by plugins on their own backend boundary."""
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
         user_text = "how is the honcho working"
         injected = (
             user_text + "\n\n"

tests/agent/test_streaming_context_scrubber.py

@@ -6,7 +6,7 @@ regex can't survive chunk boundaries, so _fire_stream_delta routes deltas
 through a stateful scrubber.
 """
 
-from agent.memory_manager import StreamingContextScrubber, sanitize_context
+from agent.memory_context import StreamingContextScrubber, sanitize_context
 
 
 class TestStreamingContextScrubberBasics:
@@ -158,7 +158,7 @@ class TestBuildMemoryContextBlockWarnsOnViolation:
 
     def test_provider_emitting_wrapper_warns(self, caplog):
         import logging
-        from agent.memory_manager import build_memory_context_block
+        from agent.memory_context import build_memory_context_block
 
         prewrapped = (
             "<memory-context>\n"
@@ -166,7 +166,7 @@ class TestBuildMemoryContextBlockWarnsOnViolation:
             "real fact\n"
             "</memory-context>"
         )
-        with caplog.at_level(logging.WARNING, logger="agent.memory_manager"):
+        with caplog.at_level(logging.WARNING, logger="agent.memory_context"):
             out = build_memory_context_block(prewrapped)
 
         assert any("pre-wrapped" in rec.message for rec in caplog.records)
@@ -175,9 +175,9 @@ class TestBuildMemoryContextBlockWarnsOnViolation:
 
     def test_clean_provider_output_does_not_warn(self, caplog):
         import logging
-        from agent.memory_manager import build_memory_context_block
+        from agent.memory_context import build_memory_context_block
 
-        with caplog.at_level(logging.WARNING, logger="agent.memory_manager"):
+        with caplog.at_level(logging.WARNING, logger="agent.memory_context"):
             out = build_memory_context_block("plain fact about user")
 
         assert not any("pre-wrapped" in rec.message for rec in caplog.records)

tests/hermes_cli/test_cli_retry.py

@@ -151,7 +151,7 @@ def test_retry_fails_closed_when_transcript_changes_after_snapshot(
 def test_rewind_matches_warm_raw_carrier_to_durable_sanitized_sidecar(
     tmp_path, command
 ):
-    from agent.memory_manager import sanitize_context
+    from agent.memory_context import sanitize_context
 
     cli = _make_cli()
     cli._session_db.close()

tui_gateway/methods_prompt.py

@@ -43,7 +43,7 @@ def _mem_db_pair_agrees(mem, db_msg) -> bool:
         return False
     if mem.get("role") == "user":
         from agent.context_compressor import user_originated_turn_view
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
         mem_view = user_originated_turn_view(mem)
         db_view = user_originated_turn_view(db_msg)
         if (mem_view is None) != (db_view is None):
@@ -174,7 +174,7 @@ def _merge_carrier_holds_target_text(carrier: dict, db_msg: dict) -> bool:
     content or appears in it as a whole ``"\\n\\n"``-delimited part. Both views are
     user-originated by construction (the caller only visits user turns)."""
     from agent.context_compressor import user_originated_turn_view
-    from agent.memory_manager import sanitize_context
+    from agent.memory_context import sanitize_context
     mem_view = user_originated_turn_view(carrier)
     db_view = user_originated_turn_view(db_msg)
     if mem_view is None or db_view is None:

tui_gateway/session_history.py

@@ -192,7 +192,7 @@ def _user_image_display_text(content: Any) -> str | None:
     if len(image_parts) != len(paths) or len(content) not in (1 + len(paths), 2 + len(paths)):
         return None
     if len(content) > 1 + len(paths):
-        from agent.memory_manager import sanitize_context
+        from agent.memory_context import sanitize_context
         memory = content[-1]
         if not isinstance(memory, dict) or memory.get("type") != "text" or not isinstance(memory.get("text"), str) or sanitize_context(memory["text"]).strip():
             return None