refactor: address Phase-2 review findings on /new boundary handoff

- Return the boundary snapshot from
  _launch_session_boundary_memory_flush as a local value instead of
  staging it on self._session_boundary_snapshot. The instance-attr
  handoff could leak (no memory manager configured) or mis-fire a
  stale snapshot on a later /new if an exception hit between staging
  and consumption. A local variable eliminates the class; the helper
  also returns None when no memory manager is configured so
  new_session takes the inline-switch path.
- Drop the now-dead session_id kwarg from commit_memory_session:
  after the redesign no production caller passes it (gateway, TUI,
  compression all use the default), and speculative params are
  rejected per AGENTS.md. The explicit-old-session need is served by
  cli.py's direct engine call + commit_session_boundary_async.
- Drop the dead providers snapshot in commit_session_boundary_async
  (only the emptiness check used it).
- Tests updated accordingly (dead-kwarg test removed, snapshot
  assertion now covered by return-value contract).

Phase-2 gates: 2a tests/cli 1048 passed + 6 memory files 137 passed;
2b programmatic live smoke 0.38ms non-blocking caller, end→switch→sync
ordering verified; 2c structured 4-angle review — no Criticals, these
warnings fixed.

agent/memory_manager.py

@@ -811,8 +811,7 @@ class MemoryManager:
         ``_submit_background`` degrades to inline execution — the pre-#16454
         synchronous behavior, slow but correct.
         """
-        providers = list(self._providers)
-        if not providers:
+        if not self._providers:
             return
         snapshot = list(messages or [])
 

cli.py

@@ -4144,9 +4144,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         # Background task tracking: {task_id: threading.Thread}
         self._background_tasks: Dict[str, threading.Thread] = {}
         self._background_task_counter = 0
-        # History snapshot staged by _launch_session_boundary_memory_flush for
-        # the /new end→switch boundary task (see new_session()).
-        self._session_boundary_snapshot: Optional[list] = None
 
     def _claim_active_session(self, surface: str = "cli", *, stderr: bool = False) -> bool:
         """Claim a global active-session slot for this CLI process."""
@@ -6977,7 +6974,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         history_snapshot: list,
         *,
         session_id: Optional[str] = None,
-    ) -> None:
+    ) -> Optional[list]:
         """Stage old-session memory extraction so /new stays responsive.
 
         The context-engine ``on_session_end`` boundary is delivered
@@ -6986,18 +6983,20 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         rebinds the engine to the new session.
 
         The memory-provider half (LLM-bound extraction, seconds) is NOT run
-        here. It is queued later in ``new_session()`` via
+        here. The returned snapshot is handed by ``new_session()`` to
         ``MemoryManager.commit_session_boundary_async`` as a single
         end→switch task on the manager's serialized background worker, so
         extraction can never race the provider rebinding (providers key off
         internal ``_session_id`` state — a late ``on_session_end`` after
         ``on_session_switch`` would misattribute the old transcript to the
-        new session). This method just snapshots the history for that call.
+        new session).
+
+        Returns the history snapshot to queue, or ``None`` when there is
+        nothing to extract (no agent / empty history / no memory manager).
         """
-        self._session_boundary_snapshot = None
         agent = getattr(self, "agent", None)
         if not agent or not history_snapshot:
-            return
+            return None
 
         engine = getattr(agent, "context_compressor", None)
         if engine is not None and hasattr(engine, "on_session_end"):
@@ -7009,17 +7008,22 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
                     exc_info=True,
                 )
 
-        self._session_boundary_snapshot = list(history_snapshot)
+        # No provider extraction to queue when no memory manager is
+        # configured — new_session() falls back to the inline switch path.
+        if getattr(agent, "_memory_manager", None) is None:
+            return None
+        return history_snapshot
 
     def new_session(self, silent=False, title=None):
         """Start a fresh session with a new session ID and cleared agent state."""
         old_session_id = self.session_id
+        _boundary_snapshot = None
         if self.agent and self.conversation_history:
-            # Trigger memory extraction on the old session without blocking
-            # session rotation. Snapshot the history and old session id first:
-            # the background thread may run after ``self.agent.session_id`` is
-            # updated to the new session below.
-            self._launch_session_boundary_memory_flush(
+            # Deliver the context-engine boundary synchronously and get back
+            # the history snapshot for the deferred provider extraction —
+            # queued below (after rotation) so /new never blocks on the
+            # LLM-bound extraction call.
+            _boundary_snapshot = self._launch_session_boundary_memory_flush(
                 list(self.conversation_history),
                 session_id=old_session_id,
             )
@@ -7124,11 +7128,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             try:
                 _mm = getattr(self.agent, "_memory_manager", None)
                 if _mm is not None:
-                    _boundary_snapshot = getattr(
-                        self, "_session_boundary_snapshot", None
-                    )
                     if _boundary_snapshot:
-                        self._session_boundary_snapshot = None
                         _mm.commit_session_boundary_async(
                             _boundary_snapshot,
                             new_session_id=self.session_id,

run_agent.py

@@ -3300,18 +3300,11 @@ class AIAgent:
             except Exception:
                 pass
 
-    def commit_memory_session(
-        self,
-        messages: list = None,
-        session_id: Optional[str] = None,
-    ) -> None:
+    def commit_memory_session(self, messages: list = None) -> None:
         """Trigger end-of-session extraction without tearing providers down.
         Called when session_id rotates (e.g. /new, context compression);
         providers keep their state and continue running under the old
         session_id — they just flush pending extraction now."""
-        ended_session_id = (
-            session_id if session_id is not None else (self.session_id or "")
-        )
         if self._memory_manager:
             try:
                 self._memory_manager.on_session_end(messages or [])
@@ -3326,7 +3319,7 @@ class AIAgent:
         if hasattr(self, "context_compressor") and self.context_compressor:
             try:
                 self.context_compressor.on_session_end(
-                    ended_session_id,
+                    self.session_id or "",
                     messages or [],
                 )
             except Exception:

tests/cli/test_cli_new_session.py

@@ -194,8 +194,6 @@ def test_new_session_queues_boundary_commit_with_snapshot(tmp_path):
     assert kwargs["reason"] == "new_session"
     # The queued path replaces the inline switch — not both.
     mm.on_session_switch.assert_not_called()
-    # Snapshot is consumed (a later /new without history must not re-send it).
-    assert cli._session_boundary_snapshot is None
 
 
 def test_new_session_without_history_switches_inline(tmp_path):

tests/run_agent/test_commit_memory_session_context_engine.py

@@ -56,19 +56,6 @@ def test_commit_memory_session_with_no_messages_passes_empty_list():
     ctx.on_session_end.assert_called_once_with("sess-7", [])
 
 
-def test_commit_memory_session_can_finalize_explicit_old_session_id():
-    """Async callers may run after agent.session_id already changed."""
-    mm = MagicMock()
-    ctx = MagicMock()
-    agent = _make_minimal_agent(mm, ctx, session_id="new-session")
-
-    msgs = [{"role": "user", "content": "old turn"}]
-    agent.commit_memory_session(msgs, session_id="old-session")
-
-    mm.on_session_end.assert_called_once_with(msgs)
-    ctx.on_session_end.assert_called_once_with("old-session", msgs)
-
-
 def test_commit_memory_session_no_memory_manager_still_notifies_context_engine():
     """If only the context engine is configured, it still gets the hook."""
     ctx = MagicMock()