diff --git a/agent/memory_manager.py b/agent/memory_manager.py
index 474d9b3ac9..b50b22a103 100644
--- a/agent/memory_manager.py
+++ b/agent/memory_manager.py
@@ -46,28 +46,21 @@ def _has_var_kwargs(params) -> bool:
 
 
 def _accepts_require_checkpoint(fn: Callable[..., Any]) -> bool:
-    """True if ``fn`` can receive the ``require_checkpoint`` keyword.
+    """True if ``fn`` can receive the ``require_checkpoint`` keyword (unreadable signatures -> False).
 
-    v2 providers written against the docs example use the bare ``on_pre_compress(self,
-    messages)`` shape; passing the keyword would raise TypeError, which the host would
-    re-raise as a checkpoint failure even though the durable write succeeded.
-    Unreadable signatures conservatively report False.
+    Bare-shape v2 providers (``on_pre_compress(self, messages)``) would raise TypeError on the
+    keyword, which the host would re-raise as a checkpoint failure despite a successful write.
     """
     params = _signature_params(fn)
     if params is None:
         return False
-    param = params.get("require_checkpoint")
-    return _has_var_kwargs(params) or (
-        param is not None and param.kind in (inspect.Parameter.KEYWORD_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
-    )
+    kind = getattr(params.get("require_checkpoint"), "kind", None)
+    return _has_var_kwargs(params) or kind in (inspect.Parameter.KEYWORD_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
 
 
 def _ctx_bound(fn: Callable[[], Any]) -> Callable[[], Any]:
-    """Bind ``fn`` to the CALLER's contextvars for execution on another thread.
-
-    Profile isolation is a ContextVar-scoped HERMES_HOME override; an unbound provider
-    on a worker thread would silently resolve paths/secrets against the default profile.
-    """
+    """Bind ``fn`` to the CALLER's contextvars for another thread: profile isolation is a
+    ContextVar-scoped HERMES_HOME override, and an unbound worker would silently use the default profile."""
     return partial(contextvars.copy_context().run, fn)
 
 
@@ -88,10 +81,8 @@ def normalize_tool_schema(schema: Any) -> Optional[Dict[str, Any]]:
     return schema if name and isinstance(name, str) else None
 
 
-def memory_provider_tools_enabled(
-    enabled_toolsets: Optional[List[str]], disabled_toolsets: Optional[List[str]] = None, *,
-    memory_tool_present: bool = False,
-) -> bool:
+def memory_provider_tools_enabled(enabled_toolsets: Optional[List[str]], disabled_toolsets: Optional[List[str]] = None,
+                                  *, memory_tool_present: bool = False) -> bool:
     """Return whether external memory-provider tools should be exposed."""
     if disabled_toolsets and "memory" in disabled_toolsets:
         return False
@@ -121,11 +112,9 @@ def memory_provider_tools_exposed(agent: Any) -> bool:
     never advertises tools absent from the tool surface.
     """
     tools = getattr(agent, "tools", None)
-    return memory_provider_tools_enabled(
-        getattr(agent, "enabled_toolsets", None),
-        getattr(agent, "disabled_toolsets", None),
-        memory_tool_present=isinstance(tools, (list, tuple)) and any(_tool_name(t) == "memory" for t in tools),
-    )
+    present = isinstance(tools, (list, tuple)) and any(_tool_name(t) == "memory" for t in tools)
+    enabled, disabled = getattr(agent, "enabled_toolsets", None), getattr(agent, "disabled_toolsets", None)
+    return memory_provider_tools_enabled(enabled, disabled, memory_tool_present=present)
 
 
 def inject_memory_provider_tools(agent: Any) -> int:
@@ -138,7 +127,8 @@ def inject_memory_provider_tools(agent: Any) -> int:
     if not memory_provider_tools_exposed(agent):
         # Say so once: a silent 0 leaves the provider looking "half on" with no clue which
         # config key (platform_toolsets / disabled_toolsets) gated it.
-        _providers = [p for p in (getattr(memory_manager, "providers", None) or []) if getattr(p, "name", "") != "builtin"]
+        _providers = [p for p in getattr(memory_manager, "providers", None) or []
+                      if getattr(p, "name", "") != "builtin"]
         if _providers:
             logger.info(
                 "Memory provider(s) %s configured but the 'memory' toolset is "
@@ -192,10 +182,9 @@ def sanitize_context(text: str) -> str:
 class StreamingContextScrubber:
     """Stateful scrubber for streaming text whose memory-context spans may straddle deltas.
 
-    ``sanitize_context`` needs both tags in one string, so a span split across deltas
-    would leak to the UI. This holds back partial-tag tails between ``feed()`` calls and
-    drops everything inside a span. One scrubber (or ``reset()``) per top-level
-    response; call ``flush()`` at end of stream.
+    ``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>"
@@ -218,13 +207,16 @@ class StreamingContextScrubber:
         out: list[str] = []
         while buf:
             if self._in_span:
-                idx = buf.lower().find(self._CLOSE_TAG)
-                held = self._max_partial_suffix(buf, self._CLOSE_TAG)  # potential partial close tag
                 tag = self._CLOSE_TAG
+                idx = buf.lower().find(tag)
+                held = self._max_partial_suffix(buf, tag)  # potential partial close tag
             else:
-                idx = self._find_boundary_open_tag(buf)
-                held = self._max_pending_open_suffix(buf) or self._max_partial_suffix(buf, self._OPEN_TAG)
                 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:
@@ -248,41 +240,26 @@ class StreamingContextScrubber:
     @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 = tag.lower()
-        buf_lower = buf.lower()
-        for i in range(min(len(buf_lower), len(tag_lower) - 1), 0, -1):
-            if tag_lower.startswith(buf_lower[-i:]):
-                return i
-        return 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 = buf.lower()
-        search_start = 0
-        while True:
-            idx = buf_lower.find(self._OPEN_TAG, search_start)
-            if idx == -1:
-                return -1
-            after_idx = idx + len(self._OPEN_TAG)
-            if self._is_block_boundary(buf, idx) and after_idx < len(buf) and buf[after_idx] in "\r\n":
+        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
-            search_start = idx + 1
-
-    def _max_pending_open_suffix(self, buf: str) -> int:
-        """Hold a complete boundary tag at the buffer end until the following char confirms it."""
-        if buf.lower().endswith(self._OPEN_TAG) and self._is_block_boundary(buf, len(buf) - len(self._OPEN_TAG)):
-            return len(self._OPEN_TAG)
-        return 0
+            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)."""
-        last_newline = text.rfind("\n")
-        if last_newline == -1:
-            return self._at_block_boundary and text.strip() == ""
-        return text[last_newline + 1:].strip() == ""
-
-    def _is_block_boundary(self, buf: str, idx: int) -> bool:
-        return self._at_block_boundary if idx == 0 else self._ends_at_block_boundary(buf[:idx])
+        """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:
@@ -307,11 +284,6 @@ def build_memory_context_block(raw_context: str) -> str:
     )
 
 
-def _nonblank(text: Any) -> Any:
-    """Return ``text`` when it has non-whitespace content, else None."""
-    return text if text and text.strip() else None
-
-
 class MemoryManager:
     """Builtin provider (always first) plus at most one external provider.
 
@@ -323,11 +295,11 @@ class MemoryManager:
         self._providers: List[MemoryProvider] = []
         self._tool_to_provider: Dict[str, MemoryProvider] = {}
         self._has_external: bool = False
-        self._external_prefetch_timeout = (
-            _EXTERNAL_PREFETCH_TIMEOUT_S if external_prefetch_timeout is None else float(external_prefetch_timeout)
-        )
-        if self._external_prefetch_timeout <= 0:
+        timeout = external_prefetch_timeout
+        timeout = _EXTERNAL_PREFETCH_TIMEOUT_S if timeout is None else float(timeout)
+        if timeout <= 0:
             raise ValueError("external_prefetch_timeout must be positive")
+        self._external_prefetch_timeout = timeout
         self._external_prefetch_threads: Dict[str, threading.Thread] = {}
         self._external_prefetch_lock = threading.Lock()
         # Single-worker background executor for end-of-turn sync/prefetch, created lazily so
@@ -342,15 +314,10 @@ class MemoryManager:
             "status": "not_started", "abandoned_writes": 0, "abandoned_prefetches": 0, "active_tasks": 0,
         }
 
-    def _each_provider(
-        self, label: str, call: Callable[[MemoryProvider], Any], *,
-        level: int = logging.DEBUG, providers: Optional[List[MemoryProvider]] = None, exc_info: bool = False,
-    ) -> List[Any]:
-        """Call ``call(provider)`` for each provider, logging and swallowing failures.
-
-        ``label`` completes the log line ``Memory provider '<name>' <label>: <exc>``.
-        Returns the successful results in provider order.
-        """
+    def _each_provider(self, label: str, call: Callable[[MemoryProvider], Any], *, level: int = logging.DEBUG,
+                       providers: Optional[List[MemoryProvider]] = None, exc_info: bool = False) -> List[Any]:
+        """Call ``call(provider)`` per provider, logging+swallowing failures; returns successes in order.
+        ``label`` completes the log line ``Memory provider '<name>' <label>: <exc>``."""
         results: List[Any] = []
         for provider in self._providers if providers is None else providers:
             try:
@@ -409,10 +376,9 @@ class MemoryManager:
 
     def build_system_prompt(self) -> str:
         """Join every provider's non-empty ``system_prompt_block()`` with blank lines."""
-        blocks = self._each_provider(
-            "system_prompt_block() failed", lambda p: _nonblank(p.system_prompt_block()), level=logging.WARNING,
-        )
-        return "\n\n".join(b for b in blocks if b)
+        blocks = self._each_provider("system_prompt_block() failed", lambda p: p.system_prompt_block(),
+                                      level=logging.WARNING)
+        return "\n\n".join(b for b in blocks if b and b.strip())
 
     # A /skill or /bundle turn embeds the whole skill body in the model-facing message;
     # providers get just the user's instruction (None for a bare invocation).
@@ -424,17 +390,13 @@ class MemoryManager:
         if not clean_query:
             return ""
         parts = self._each_provider(
-            "prefetch failed (non-fatal)",
-            lambda p: _nonblank(self._prefetch_provider(p, clean_query, session_id=session_id)),
+            "prefetch failed (non-fatal)", lambda p: self._prefetch_provider(p, clean_query, session_id=session_id),
         )
-        return "\n\n".join(p for p in parts if p)
+        return "\n\n".join(p for p in parts if p and p.strip())
 
     def _prefetch_provider(self, provider: MemoryProvider, query: str, *, session_id: str = "") -> str:
-        """Run one provider's prefetch; external providers are bounded by a timeout.
-
-        A stuck external call keeps running on its daemon thread and the provider is
-        skipped on later turns until it returns.
-        """
+        """Run one provider's prefetch; external providers are bounded by a timeout. A stuck external
+        call keeps running on its daemon thread and the provider is skipped on later turns until it returns."""
         if provider.name == "builtin":
             return provider.prefetch(query, session_id=session_id)
 
@@ -449,11 +411,9 @@ class MemoryManager:
         thread = threading.Thread(target=_ctx_bound(_run), daemon=True, name=f"memory-prefetch-{provider.name}")
         with self._external_prefetch_lock:
             existing = self._external_prefetch_threads.get(provider.name)
-            if existing is not None:
-                if existing.is_alive():
-                    logger.debug("Memory provider '%s' prefetch is still running; skipping this turn", provider.name)
-                    return ""
-                self._external_prefetch_threads.pop(provider.name, None)
+            if existing is not None and existing.is_alive():
+                logger.debug("Memory provider '%s' prefetch is still running; skipping this turn", provider.name)
+                return ""
             self._external_prefetch_threads[provider.name] = thread
             thread.start()
 
@@ -473,21 +433,15 @@ class MemoryManager:
         return result_box.get("value", "")
 
     def describe_recall(self) -> str:
-        """Deterministic recall indicator line (e.g. ``"🧠 Provider — recalled 3 memories"``).
-
-        Call right after :meth:`prefetch_all` so the user SEES memory was used regardless
-        of whether the model mentions it; ``""`` when no provider injected memory.
-        """
+        """Deterministic recall indicator line (e.g. ``"🧠 Provider — recalled 3 memories"``); ``""`` if none.
+        Call right after :meth:`prefetch_all` so the user SEES memory was used even if the model is silent."""
         segments: List[str] = []
         for status in self._each_provider("recall_status failed (non-fatal)", lambda p: p.recall_status()):
             if status is None:
                 continue
-            if status.count == 1:
-                detail = "recalled 1 memory"
-            elif status.count > 1:
-                detail = f"recalled {status.count} memories"
-            else:  # content injected but no discrete count (reflect)
-                detail = "recalled relevant memory"
+            # count <= 0: content injected but no discrete count (reflect)
+            detail = ("recalled 1 memory" if status.count == 1 else f"recalled {status.count} memories"
+                      if status.count > 1 else "recalled relevant memory")
             segments.append(f"{status.glyph} {status.provider_label} — {detail}")
         return "  ".join(segments)
 
@@ -497,14 +451,10 @@ class MemoryManager:
         clean_query = self._strip_skill_scaffolding(query) if providers else None
         if not clean_query:
             return
-        self._submit_background(
-            lambda: self._each_provider(
-                "queue_prefetch failed (non-fatal)",
-                lambda p: p.queue_prefetch(clean_query, session_id=session_id),
-                providers=providers,
-            ),
-            kind="prefetch",
-        )
+        self._submit_background(lambda: self._each_provider(
+            "queue_prefetch failed (non-fatal)", lambda p: p.queue_prefetch(clean_query, session_id=session_id),
+            providers=providers,
+        ), kind="prefetch")
 
     @staticmethod
     def _provider_sync_accepts_messages(provider: MemoryProvider) -> bool:
@@ -512,15 +462,12 @@ class MemoryManager:
         params = _signature_params(provider.sync_turn)
         return params is None or _has_var_kwargs(params) or "messages" in params
 
-    def sync_all(
-        self, user_content: str, assistant_content: str, *,
-        session_id: str = "", messages: Optional[List[Dict[str, Any]]] = None,
-    ) -> None:
+    def sync_all(self, user_content: str, assistant_content: str, *, session_id: str = "",
+                 messages: Optional[List[Dict[str, Any]]] = None) -> None:
         """Sync a completed turn to all providers on the background worker.
 
-        Never inline: a provider's ``sync_turn`` may block for minutes, which kept
-        ``run_conversation`` open after the user saw the response. The single worker
-        also serializes writes so turn N lands before turn N+1.
+        Never inline: a provider's ``sync_turn`` may block for minutes, which kept ``run_conversation``
+        open after the user saw the response. The single worker also serializes writes (turn N before N+1).
         """
         providers = list(self._providers)
         clean_user_content = self._strip_skill_scaffolding(user_content) if providers else None
@@ -538,13 +485,21 @@ class MemoryManager:
         )
 
     def _submit_background(self, fn, *, kind: str = "write") -> None:
-        """Queue ``fn`` on the serialized worker and track its durability class.
-
-        Runs under the caller's contextvars (``_ctx_bound``). If the executor is
-        unavailable outside shutdown, run inline — the historical fail-safe.
-        """
+        """Queue ``fn`` on the serialized worker (created lazily; None once shutting down) and track its
+        durability class. Runs under the caller's contextvars (``_ctx_bound``). If the executor is
+        unavailable outside shutdown, run inline — the historical fail-safe."""
         fn = _ctx_bound(fn)
-        executor = self._get_sync_executor()
+        executor = None if self._shutting_down else self._sync_executor
+        if executor is None and not self._shutting_down:
+            with self._sync_executor_lock:
+                if self._sync_executor is None and not self._shutting_down:
+                    try:
+                        # Daemon workers: a wedged provider must never block interpreter exit.
+                        from tools.daemon_pool import DaemonThreadPoolExecutor
+                        self._sync_executor = DaemonThreadPoolExecutor(max_workers=1, thread_name_prefix="mem-sync")
+                    except Exception as e:  # pragma: no cover - resource exhaustion
+                        logger.warning("Failed to create memory sync executor: %s", e)
+                executor = self._sync_executor
         future = None
         try:
             # Submit+track atomically with the shutdown snapshot. The callback is attached
@@ -562,53 +517,31 @@ class MemoryManager:
                 return
         if future is not None:
             future.add_done_callback(self._forget_background_future)
-        else:
-            try:
-                fn()
-            except Exception as e:  # pragma: no cover - fn guards internally
-                logger.debug("Inline memory background task failed: %s", e)
+            return
+        try:
+            fn()
+        except Exception as e:  # pragma: no cover - fn guards internally
+            logger.debug("Inline memory background task failed: %s", e)
 
     def _forget_background_future(self, future: Future) -> None:
         with self._sync_executor_lock:
             self._background_futures.pop(future, None)
 
-    def _get_sync_executor(self) -> Optional[ThreadPoolExecutor]:
-        """Lazily create the single-worker background executor (None once shutting down)."""
-        if self._shutting_down:
-            return None
-        if self._sync_executor is not None:
-            return self._sync_executor
-        with self._sync_executor_lock:
-            if self._sync_executor is None and not self._shutting_down:
-                try:
-                    # Daemon workers: a wedged provider must never block interpreter exit.
-                    from tools.daemon_pool import DaemonThreadPoolExecutor
-                    self._sync_executor = DaemonThreadPoolExecutor(max_workers=1, thread_name_prefix="mem-sync")
-                except Exception as e:  # pragma: no cover - resource exhaustion
-                    logger.warning("Failed to create memory sync executor: %s", e)
-            return self._sync_executor
-
     def flush_pending(self, timeout: Optional[float] = None) -> bool:
         """Block until queued sync/prefetch work has drained (False on timeout).
-
-        With a single worker, a sentinel task completing proves every earlier task ran.
-        """
+        With a single worker, a sentinel task completing proves every earlier task ran."""
         executor = self._sync_executor
         if executor is None:
             return True
         try:
             executor.submit(lambda: None).result(timeout=timeout)
-        except RuntimeError:
-            return True  # executor already shut down — nothing pending
-        except Exception:
-            return False
+        except Exception as e:
+            return isinstance(e, RuntimeError)  # executor already shut down — nothing pending
         return True
 
     def get_all_tool_schemas(self) -> List[Dict[str, Any]]:
-        """Collect deduplicated tool schemas from all providers.
-
-        Reserved core tool names are skipped: :meth:`add_provider` refuses to route them.
-        """
+        """Collect deduplicated tool schemas from all providers; reserved core tool names are
+        skipped because :meth:`add_provider` refuses to route them."""
         from toolsets import _HERMES_CORE_TOOLS
 
         schemas: List[Dict[str, Any]] = []
@@ -630,7 +563,7 @@ class MemoryManager:
         return schemas
 
     def get_all_tool_names(self) -> set:
-        return set(self._tool_to_provider.keys())
+        return set(self._tool_to_provider)
 
     def has_tool(self, tool_name: str) -> bool:
         return tool_name in self._tool_to_provider
@@ -650,19 +583,15 @@ class MemoryManager:
         self._each_provider("on_turn_start failed", lambda p: p.on_turn_start(turn_number, message, **kwargs))
 
     def on_session_end(self, messages: List[Dict[str, Any]]) -> None:
-        self._each_provider(
-            "on_session_end failed", lambda p: p.on_session_end(messages), level=logging.WARNING, exc_info=True,
-        )
+        self._each_provider("on_session_end failed", lambda p: p.on_session_end(messages), level=logging.WARNING,
+                            exc_info=True)
 
-    def commit_session_boundary_async(
-        self, messages: List[Dict[str, Any]], *,
-        new_session_id: str, parent_session_id: str = "", reason: str = "new_session",
-    ) -> None:
+    def commit_session_boundary_async(self, messages: List[Dict[str, Any]], *, new_session_id: str,
+                                      parent_session_id: str = "", reason: str = "new_session") -> None:
         """Queue old-session extraction + provider rebinding as ONE serialized task.
 
-        ``on_session_end`` (LLM-bound, seconds) must run strictly BEFORE ``on_session_switch``
-        rebinds provider session state; an ad-hoc thread raced the inline switch and
-        misattributed transcripts. One FIFO task gives immediate return plus ordering.
+        ``on_session_end`` (LLM-bound, seconds) must run strictly BEFORE ``on_session_switch`` rebinds
+        provider state; an ad-hoc thread raced the inline switch and misattributed transcripts.
         """
         if not self._providers:
             return
@@ -680,10 +609,8 @@ class MemoryManager:
 
         self._submit_background(_run)
 
-    def on_session_switch(
-        self, new_session_id: str, *,
-        parent_session_id: str = "", reset: bool = False, rewound: bool = False, **kwargs,
-    ) -> None:
+    def on_session_switch(self, new_session_id: str, *, parent_session_id: str = "", reset: bool = False,
+                          rewound: bool = False, **kwargs) -> None:
         """Notify providers that ``AIAgent.session_id`` rotated without teardown
         (``/resume``, ``/branch``, ``/reset``, ``/new``, compression). ``rewound=True``
         (``/undo``): same id, truncated transcript."""
@@ -704,57 +631,45 @@ class MemoryManager:
         except (TypeError, ValueError):
             return None
 
-    @classmethod
-    def _is_checkpoint_provider(cls, provider: MemoryProvider, api_version: int) -> bool:
-        version = cls._checkpoint_api_version(provider)
-        return (_LEGACY_PRE_COMPRESS_API_VERSION if version is None else version) >= api_version
-
     def supports_pre_compress_checkpoint(self, api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION) -> bool:
         """Return whether an active provider guarantees checkpoint API support."""
-        return any(
-            (version := self._checkpoint_api_version(p)) is not None and version >= api_version
-            for p in self._providers
-        )
+        versions = (self._checkpoint_api_version(p) for p in self._providers)
+        return any(v is not None and v >= api_version for v in versions)
 
-    def on_pre_compress(
-        self, messages: List[Dict[str, Any]], *,
-        evidence_messages: Optional[List[Dict[str, Any]]] = None,
-        require_checkpoint: bool = False,
-        checkpoint_api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION,
-    ) -> str:
+    def on_pre_compress(self, messages: List[Dict[str, Any]], *,
+                        evidence_messages: Optional[List[Dict[str, Any]]] = None, require_checkpoint: bool = False,
+                        checkpoint_api_version: int = PRE_COMPRESS_CHECKPOINT_API_VERSION) -> str:
         """Notify providers before compression; return their combined summary-prompt text.
 
-        ``messages`` is the raw transcript (the v1 contract). ``evidence_messages`` is the
-        host-normalized list handed only to checkpoint (v2+) providers. With
-        ``require_checkpoint``, at least one checkpoint provider must succeed — its
-        exception propagates so the caller keeps the uncompressed transcript.
+        ``messages`` is the raw v1 transcript; ``evidence_messages`` is the host-normalized list handed
+        only to checkpoint (v2+) providers. With ``require_checkpoint`` at least one checkpoint provider
+        must succeed — its exception propagates so the caller keeps the uncompressed transcript.
         """
         parts = []
         checkpoint_succeeded = False
         for provider in self._providers:
-            is_checkpoint_provider = self._is_checkpoint_provider(provider, checkpoint_api_version)
-            provider_messages = messages
+            version = self._checkpoint_api_version(provider)
+            if version is None:
+                version = _LEGACY_PRE_COMPRESS_API_VERSION
+            is_checkpoint_provider = version >= checkpoint_api_version
+            use_evidence = is_checkpoint_provider and evidence_messages is not None
+            provider_messages = evidence_messages if use_evidence else messages
             kwargs: Dict[str, Any] = {}
-            if is_checkpoint_provider:
-                if evidence_messages is not None:
-                    provider_messages = evidence_messages
-                # v1 providers and bare-shape v2 providers never see the signal.
-                if _accepts_require_checkpoint(provider.on_pre_compress):
-                    kwargs["require_checkpoint"] = require_checkpoint
+            # v1 providers and bare-shape v2 providers never see the signal.
+            if is_checkpoint_provider and _accepts_require_checkpoint(provider.on_pre_compress):
+                kwargs["require_checkpoint"] = require_checkpoint
             try:
                 result = provider.on_pre_compress(provider_messages, **kwargs)
                 if result and result.strip():
                     parts.append(result)
+                checkpoint_succeeded = checkpoint_succeeded or is_checkpoint_provider
             except Exception as e:
                 logger.debug("Memory provider '%s' on_pre_compress failed: %s", provider.name, e)
                 if require_checkpoint and is_checkpoint_provider:
                     raise
-                continue
-            checkpoint_succeeded = checkpoint_succeeded or is_checkpoint_provider
         if require_checkpoint and not checkpoint_succeeded:
             raise RuntimeError(
-                "No active memory provider completed pre-compress checkpoint "
-                f"API v{checkpoint_api_version}"
+                f"No active memory provider completed pre-compress checkpoint API v{checkpoint_api_version}"
             )
         return "\n\n".join(parts)
 
@@ -767,9 +682,8 @@ class MemoryManager:
         accepted = sum(p.kind is not inspect.Parameter.VAR_POSITIONAL for p in params.values())
         return "positional" if accepted >= 4 else "legacy"
 
-    def on_memory_write(
-        self, action: str, target: str, content: str, metadata: Optional[Dict[str, Any]] = None,
-    ) -> None:
+    def on_memory_write(self, action: str, target: str, content: str,
+                        metadata: Optional[Dict[str, Any]] = None) -> None:
         """Notify external providers when the built-in memory tool writes (skips builtin, the source)."""
 
         def _notify(provider: MemoryProvider) -> None:
@@ -781,9 +695,8 @@ class MemoryManager:
             else:
                 provider.on_memory_write(action, target, content, metadata=dict(metadata or {}))
 
-        self._each_provider(
-            "on_memory_write failed", _notify, providers=[p for p in self._providers if p.name != "builtin"],
-        )
+        external = [p for p in self._providers if p.name != "builtin"]
+        self._each_provider("on_memory_write failed", _notify, providers=external)
 
     # Actions mirrored to external providers; non-mutating results (errors, staged) are
     # filtered by ``notify_memory_tool_write`` first.
@@ -791,11 +704,8 @@ class MemoryManager:
 
     @staticmethod
     def _memory_tool_result_succeeded(result: Any) -> bool:
-        """True only when the built-in memory tool actually committed a write.
-
-        Fails closed (non-JSON, non-dict, missing ``success``, staged for approval) so
-        providers never mirror a write that did not land.
-        """
+        """True only when the built-in memory tool actually committed a write. Fails closed (non-JSON,
+        non-dict, missing ``success``, staged for approval) so providers never mirror a write that did not land."""
         if isinstance(result, str):
             try:
                 result = json.loads(result)
@@ -803,23 +713,19 @@ class MemoryManager:
                 return False
         return isinstance(result, dict) and result.get("success") is True and result.get("staged") is not True
 
-    def notify_memory_tool_write(
-        self, tool_result: Any, tool_args: Dict[str, Any], *,
-        build_metadata: Optional[Callable[[], Dict[str, Any]]] = None,
-    ) -> None:
+    def notify_memory_tool_write(self, tool_result: Any, tool_args: Dict[str, Any], *,
+                                 build_metadata: Optional[Callable[[], Dict[str, Any]]] = None) -> None:
         """Mirror a built-in memory tool call to external providers.
 
-        Gates on a committed write, expands single-op and batched ``operations`` shapes,
-        keeps only mutating actions, and forwards ``old_text`` plus per-op provenance from
-        ``build_metadata`` (the loop knows session/task/tool-call identity; we do not).
+        Gates on a committed write, expands single-op and batched ``operations`` shapes, keeps only
+        mutating actions, and forwards ``old_text`` plus provenance from ``build_metadata`` (the loop
+        knows session/task/tool-call identity; we do not).
         """
         if not self._memory_tool_result_succeeded(tool_result):
             return
         target = str(tool_args.get("target") or "memory")
         operations = tool_args.get("operations")
-        if not (isinstance(operations, list) and operations):
-            operations = [tool_args]
-        for op in operations:
+        for op in operations if isinstance(operations, list) and operations else [tool_args]:
             action = str(op.get("action") or "") if isinstance(op, dict) else ""
             if action not in self._MIRRORED_MEMORY_ACTIONS:
                 continue
@@ -841,9 +747,8 @@ class MemoryManager:
     def shutdown_all(self) -> None:
         """Drain the background executor (bounded), then shut providers down in reverse order."""
         self._drain_sync_executor()
-        self._each_provider(
-            "shutdown failed", lambda p: p.shutdown(), level=logging.WARNING, providers=list(reversed(self._providers)),
-        )
+        self._each_provider("shutdown failed", lambda p: p.shutdown(), level=logging.WARNING,
+                            providers=self._providers[::-1])
 
     @property
     def shutdown_drain_state(self) -> Dict[str, Any]:
@@ -860,8 +765,7 @@ class MemoryManager:
             tracked = dict(self._background_futures)
             self._shutdown_drain_state = {
                 "status": "draining" if executor is not None else "drained",
-                "abandoned_writes": 0,
-                "abandoned_prefetches": 0,
+                "abandoned_writes": 0, "abandoned_prefetches": 0,
                 "active_tasks": sum(not future.done() for future in tracked),
             }
         if executor is None:
@@ -871,21 +775,17 @@ class MemoryManager:
         # tracked futures lets the worker run every queued task in order up to the deadline.
         executor.shutdown(wait=False, cancel_futures=False)
         _, pending = wait(tuple(tracked), timeout=_SYNC_DRAIN_TIMEOUT_S)
-        if not pending:
-            with self._sync_executor_lock:
-                self._shutdown_drain_state.update(status="drained", active_tasks=0)
-            return
-
         cancelled = [tracked[future] for future in pending if future.cancel()]
         active_tasks = len(pending) - len(cancelled)
         abandoned_prefetches = cancelled.count("prefetch")
         abandoned_writes = len(cancelled) - abandoned_prefetches
-
         with self._sync_executor_lock:
             self._shutdown_drain_state.update(
-                status="timed_out", abandoned_writes=abandoned_writes,
+                status="timed_out" if pending else "drained", abandoned_writes=abandoned_writes,
                 abandoned_prefetches=abandoned_prefetches, active_tasks=active_tasks,
             )
+        if not pending:
+            return
         logger.warning(
             "Memory shutdown drain timed out after %.2fs; abandoning %d queued "
             "memory write(s) and %d queued prefetch(es); %d active task(s) remain detached",
@@ -897,6 +797,5 @@ class MemoryManager:
         if "hermes_home" not in kwargs:
             from hermes_constants import get_hermes_home
             kwargs["hermes_home"] = str(get_hermes_home())
-        self._each_provider(
-            "initialize failed", lambda p: p.initialize(session_id=session_id, **kwargs), level=logging.WARNING,
-        )
+        self._each_provider("initialize failed", lambda p: p.initialize(session_id=session_id, **kwargs),
+                            level=logging.WARNING)
diff --git a/agent/message_sanitization.py b/agent/message_sanitization.py
index 12a9e7f3d1..54f8f91e24 100644
--- a/agent/message_sanitization.py
+++ b/agent/message_sanitization.py
@@ -26,9 +26,7 @@ _MESSAGE_CORE_KEYS = frozenset({"content", "name", "tool_calls", "role"})
 
 def _sanitize_surrogates(text: str) -> str:
     """Replace lone surrogate code points with U+FFFD; no-op when none present."""
-    if _SURROGATE_RE.search(text):
-        return _SURROGATE_RE.sub('\ufffd', text)
-    return text
+    return _SURROGATE_RE.sub('\ufffd', text)
 
 
 def _strip_non_ascii(text: str) -> str:
@@ -39,12 +37,11 @@ def _strip_non_ascii(text: str) -> str:
 def _fix_str_field(container: Any, key: Any, fix: Callable[[str], str]) -> bool:
     """Apply ``fix`` to ``container[key]`` if it is a str; True if it changed."""
     value = container.get(key) if isinstance(container, dict) else container[key]
-    if isinstance(value, str):
-        fixed = fix(value)
-        if fixed != value:
-            container[key] = fixed
-            return True
-    return False
+    fixed = fix(value) if isinstance(value, str) else value
+    if fixed == value:
+        return False
+    container[key] = fixed
+    return True
 
 
 def _sanitize_structure(payload: Any, fix: Callable[[str], str]) -> bool:
@@ -63,36 +60,25 @@ def _sanitize_structure(payload: Any, fix: Callable[[str], str]) -> bool:
 
 
 def _sanitize_messages(messages: list, fix: Callable[[str], str], *, deep: bool) -> bool:
-    """Apply ``fix`` to the string fields of every message dict in-place.
-
-    Covers content / content-part text, name, tool_call arguments, and every non-core
-    top-level str field. ``deep=True`` additionally covers tool_call ids, function names,
-    and NESTED non-core fields (``reasoning_details`` arrays from byte-level reasoning models).
-    """
+    """Apply ``fix`` to the string fields of every message dict in-place (content / part text,
+    name, tool_call arguments, non-core top-level str fields). ``deep=True`` adds tool_call ids,
+    function names, and NESTED non-core fields (``reasoning_details`` from byte-level models)."""
     found = False
     for msg in messages:
         if not isinstance(msg, dict):
             continue
         content = msg.get("content")
-        if isinstance(content, list):
-            for part in content:
-                if isinstance(part, dict):
-                    found |= _fix_str_field(part, "text", fix)
-        else:
-            found |= _fix_str_field(msg, "content", fix)
-        found |= _fix_str_field(msg, "name", fix)
+        parts = [(p, "text") for p in content if isinstance(p, dict)] if isinstance(content, list) else None
+        fields = parts if parts is not None else [(msg, "content")]
+        fields.append((msg, "name"))
         tool_calls = msg.get("tool_calls")
         for tc in tool_calls if isinstance(tool_calls, list) else ():
-            if not isinstance(tc, dict):
-                continue
-            fn = tc.get("function")
-            fn_fields = [(fn, "name"), (fn, "arguments")] if isinstance(fn, dict) else []
-            for container, key in [(tc, "id")] + fn_fields:
-                if deep or key == "arguments":
-                    found |= _fix_str_field(container, key, fix)
-        for key, value in list(msg.items()):
-            if key in _MESSAGE_CORE_KEYS:
-                continue
+            fn = tc.get("function") if isinstance(tc, dict) else None
+            fields += [(tc, "id")] if deep and isinstance(tc, dict) else []
+            fields += ([(fn, "name")] if deep else []) + [(fn, "arguments")] if isinstance(fn, dict) else []
+        for container, key in fields:
+            found |= _fix_str_field(container, key, fix)
+        for key, value in [kv for kv in msg.items() if kv[0] not in _MESSAGE_CORE_KEYS]:
             if isinstance(value, str):
                 found |= _fix_str_field(msg, key, fix)
             elif deep and isinstance(value, (dict, list)):
@@ -169,14 +155,12 @@ def _repair_tool_call_arguments(raw_args: str, tool_name: str = "?") -> str:
     fixed += '}' * max(0, fixed.count('{') - fixed.count('}'))
     fixed += ']' * max(0, fixed.count('[') - fixed.count(']'))
     for _ in range(50):
-        if _loads_ok(fixed):
-            break
-        if (fixed.endswith('}') and fixed.count('}') > fixed.count('{')) or (
-            fixed.endswith(']') and fixed.count(']') > fixed.count('[')
+        if _loads_ok(fixed) or not (
+            (fixed.endswith('}') and fixed.count('}') > fixed.count('{'))
+            or (fixed.endswith(']') and fixed.count(']') > fixed.count('['))
         ):
-            fixed = fixed[:-1]
-        else:
             break
+        fixed = fixed[:-1]
 
     if _loads_ok(fixed):
         logger.warning("Repaired malformed tool_call arguments for %s: %s → %s", tool_name, raw_stripped[:80], fixed[:80])
@@ -199,12 +183,10 @@ def _repair_tool_call_arguments(raw_args: str, tool_name: str = "?") -> str:
 
 
 def close_interrupted_tool_sequence(messages: list, final_response: Any = None) -> bool:
-    """Append a synthetic assistant turn when an interrupted tail is a tool result.
-
-    A transcript ending on a raw ``tool`` message makes the next user message land as
-    ``tool → user`` — an alternation violation strict providers (Gemini, Claude) answer by
-    hallucinating a continuation. Mutates in place; True if a closing turn was appended.
-    """
+    """Append a synthetic assistant turn when an interrupted tail is a tool result: a transcript
+    ending on a raw ``tool`` message makes the next user message land as ``tool → user``, an
+    alternation violation strict providers (Gemini, Claude) answer by hallucinating a
+    continuation. Mutates in place; True if a closing turn was appended."""
     last = messages[-1] if messages else None
     if not isinstance(last, dict) or last.get("role") != "tool":
         return False
@@ -216,12 +198,9 @@ def close_interrupted_tool_sequence(messages: list, final_response: Any = None)
 
 
 def serialized_messages_bytes(messages: list) -> int:
-    """Exact serialized byte size of the ``messages`` payload (HTTP 413 recovery).
-
-    A 413 is a BYTE-size error, but the token estimator prices images at a flat cost, so
-    it cannot score recovery from an image-dominated 413. Non-serializable values fall
-    back to ``str()`` so a malformed message can never crash recovery.
-    """
+    """Exact serialized byte size of ``messages`` (HTTP 413 is a BYTE-size error the token
+    estimator, pricing images flat, cannot score). Non-serializable values fall back to
+    ``str()`` so a malformed message can never crash recovery."""
     if not isinstance(messages, list) or not messages:
         return 0
     try:
@@ -236,19 +215,16 @@ _IMAGE_PART_TYPES = {"image_url", "image", "input_image"}
 def _strip_images_from_messages(messages: list) -> bool:
     """Remove image content parts from all messages in-place (server rejected images).
 
-    ``tool`` messages and assistant messages carrying ``tool_calls`` whose content was
-    entirely images get a placeholder, NOT deleted (deleting orphans the paired
-    ``tool_call_id`` → HTTP 400); other now-empty messages are dropped. Rewritten messages
-    lose their ``api_content`` sidecar (it carries the images being removed).
+    ``tool`` / ``tool_calls`` messages left empty get a placeholder, NOT deleted (deleting
+    orphans the paired ``tool_call_id`` → HTTP 400); other now-empty messages are dropped.
+    Rewritten messages lose their ``api_content`` sidecar (it carries the removed images).
     """
     from agent.turn_context import drop_stale_api_content
 
     found = False
     to_delete = []
     for i, msg in enumerate(messages):
-        if not isinstance(msg, dict):
-            continue
-        content = msg.get("content")
+        content = msg.get("content") if isinstance(msg, dict) else None
         if not isinstance(content, list):
             continue
         new_parts = [p for p in content if not (isinstance(p, dict) and p.get("type") in _IMAGE_PART_TYPES)]
@@ -326,10 +302,7 @@ def _tc_field(tc: Any, key: str) -> Any:
 
 
 def _tc_set(tc: Any, key: str, value: Any) -> None:
-    if isinstance(tc, dict):
-        tc[key] = value
-    else:
-        setattr(tc, key, value)
+    tc.__setitem__(key, value) if isinstance(tc, dict) else setattr(tc, key, value)
 
 
 def deterministic_call_id(fn_name: str, arguments: str, index: int = 0) -> str:
@@ -344,10 +317,8 @@ def _expand_tool_id_variants(values: tuple[Any, ...]) -> frozenset[str]:
     variants: set[str] = set()
     for raw in values:
         value = raw.strip() if isinstance(raw, str) else ""
-        if not value:
-            continue
-        variants.add(value)
-        if "|" in value:
+        if value:
+            variants.add(value)
             variants.update(p for p in (part.strip() for part in value.split("|")) if p)
     return frozenset(variants)
 
@@ -363,11 +334,9 @@ def tool_result_id_variants(tool_call_id: Any) -> frozenset[str]:
 
 
 def coalesce_tool_call_id(tc: Any) -> str:
-    """Effective call id of a tool_call entry (dict or object); ``""`` when none.
-
-    Codex Responses calls carry ``call_id`` (authoritative pairing key), Chat Completions
-    carry ``id`` only, and bridge ids may be ``call_id|response_item_id``.
-    """
+    """Effective call id of a tool_call entry (dict or object); ``""`` when none. Codex Responses
+    carry ``call_id`` (authoritative pairing key), Chat Completions ``id`` only, and bridge ids
+    may be ``call_id|response_item_id``."""
     for raw in (_tc_field(tc, "call_id"), _tc_field(tc, "id")):
         value = raw.strip() if isinstance(raw, str) else ""
         if value:
@@ -379,10 +348,10 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
     """Ensure every tool call in one assistant turn has a distinct id.
 
     Some providers reuse one id across a batch; the pre-API sanitizer then keeps only the
-    first call/result pair per id and strict providers reject duplicates. First occurrence
-    keeps its id; later collisions get a deterministic ``<id>_d<n>`` suffix (never uuid4 —
-    cache-prefix stability). Mutates entries in place (SDK models / SimpleNamespace /
-    dicts). Blank ids are left for the deterministic fallback in ``build_assistant_message``.
+    first call/result pair per id and strict providers reject duplicates. Later collisions
+    get a deterministic ``<id>_d<n>`` suffix (never uuid4 — cache-prefix stability). Mutates
+    entries (SDK models / SimpleNamespace / dicts) in place. Blank ids are left for the
+    deterministic fallback in ``build_assistant_message``.
     """
     seen: set = set()
     for tc in tool_calls or []:
@@ -396,18 +365,14 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
         if cid not in seen:
             seen.add(cid)
             continue
-        n = 2
-        while f"{cid}_d{n}" in seen:
-            n += 1
-        new_id = f"{cid}_d{n}"
+        # range is bounded: at most len(seen) suffixes can already be taken.
+        new_id = next(f"{cid}_d{n}" for n in range(2, len(seen) + 3) if f"{cid}_d{n}" not in seen)
         seen.add(new_id)
 
-        def _renamed(value):
-            # Keep a composite id's response-item half so the provider's fc_/item id survives.
-            return f"{new_id}|{value.split('|', 1)[1]}" if isinstance(value, str) and "|" in value else new_id
-
         try:
-            _tc_set(tc, "id", _renamed(_tc_field(tc, "id")))
+            # Keep a composite id's response-item half so the provider's fc_/item id survives.
+            old = _tc_field(tc, "id")
+            _tc_set(tc, "id", f"{new_id}|{old.split('|', 1)[1]}" if isinstance(old, str) and "|" in old else new_id)
             if _tc_field(tc, "call_id"):
                 _tc_set(tc, "call_id", new_id)
         except Exception:
@@ -415,22 +380,17 @@ def uniquify_tool_call_ids(tool_calls: list) -> list:
             continue
         _fn_name = _tc_field(_tc_field(tc, "function"), "name") or "?"
         logger.warning(
-            "Model reused tool call id %s within one turn; renamed the "
-            "duplicate to %s (tool=%s) to keep call/result pairing "
-            "lossless.", cid, new_id, _fn_name,
+            "Model reused tool call id %s within one turn; renamed the duplicate to %s (tool=%s) to keep "
+            "call/result pairing lossless.", cid, new_id, _fn_name,
         )
     return tool_calls
 
 
 # -- reasoning_content policy: single owner of strip-vs-re-pad; adapters keep only SYNTAX --
-#   require-side (echo-back enforced; replays 400 without the field):
-#     kimi     — provider kimi-coding/kimi-coding-cn, or host api.kimi.com / moonshot.ai /
-#                moonshot.cn. Host-driven on purpose: aggregators re-exporting kimi reject it.
-#     deepseek — provider "deepseek", model contains "deepseek", or host api.deepseek.com.
-#                V4 rejects empty-string pads → " " single space.
-#     mimo     — provider "xiaomi", model contains "mimo", or host *.xiaomimimo.com.
-#   strict side (field rejected 400/422 "Extra inputs are not permitted"): everyone else —
-#     Mistral, Cerebras, Groq, SambaNova, … Strip the key entirely, even a one-space pad.
+# Require side (echo-back enforced; replays 400 without the field): the families below. Kimi
+# is host-driven on purpose (aggregators re-exporting kimi reject it); DeepSeek V4 rejects
+# empty-string pads → " ". Strict side (400/422 "Extra inputs are not permitted"): everyone
+# else — Mistral, Cerebras, Groq, SambaNova, … Strip the key entirely, even a one-space pad.
 
 _REASONING_ECHO_RULES: tuple = (
     # (family, exact providers (raw), exact providers (lowered), model substrings (lowered), hosts)
@@ -449,20 +409,16 @@ def matches_reasoning_echo_family(family: str, provider: Any, model: Any, base_u
     _, raw_providers, lowered_providers, model_subs, hosts = _REASONING_ECHO_RULE_BY_FAMILY[family]
     model_lower = (model or "").lower()
     return (
-        provider in raw_providers
-        or (provider or "").lower() in lowered_providers
-        or any(sub in model_lower for sub in model_subs)
-        or any(base_url_host_matches(base_url, host) for host in hosts)
+        provider in raw_providers or (provider or "").lower() in lowered_providers
+        or any(sub in model_lower for sub in model_subs) or any(base_url_host_matches(base_url, host) for host in hosts)
     )
 
 
 def reasoning_echo_family(provider: Any, model: Any, base_url: Any) -> "str | None":
     """``"kimi"`` / ``"deepseek"`` / ``"mimo"`` (first match in table order) when the
     endpoint enforces reasoning_content echo-back, else ``None`` (strip side)."""
-    return next(
-        (rule[0] for rule in _REASONING_ECHO_RULES if matches_reasoning_echo_family(rule[0], provider, model, base_url)),
-        None,
-    )
+    families = (rule[0] for rule in _REASONING_ECHO_RULES)
+    return next((f for f in families if matches_reasoning_echo_family(f, provider, model, base_url)), None)
 
 
 def needs_reasoning_echo(provider: Any, model: Any, base_url: Any) -> bool:
@@ -471,14 +427,12 @@ def needs_reasoning_echo(provider: Any, model: Any, base_url: Any) -> bool:
 
 
 def stale_thinking_reaches_wire(api_mode: Any, provider: Any, model: Any, base_url: Any) -> bool:
-    """True when stale assistant ``reasoning``/``reasoning_content`` text is actually replayed
-    on the wire for the active route.
+    """True when stale assistant reasoning text is actually replayed on the wire for the route.
 
     The single wire-truth predicate the compaction TRIGGER estimator and the tail-budget
     walks must share: if they disagree, a reasoning-heavy session can look over-threshold
     to preflight yet fully tail-protected to the walk — an infinite compaction loop.
-    ``codex_responses`` never reads the text keys (continuity rides the encrypted sidecar);
-    echo-back families replay stored ``reasoning_content`` verbatim; everyone else strips.
+    ``codex_responses`` never reads the text keys (continuity rides the encrypted sidecar).
     """
     return (api_mode or "") != "codex_responses" and needs_reasoning_echo(provider, model, base_url)
 
@@ -494,41 +448,37 @@ def apply_reasoning_content_policy(source_msg: dict, api_msg: dict, needs_thinki
         # non-string value (None after compaction): never pass null to the API.
         api_msg.pop("reasoning_content", None)
         return
-    existing = source_msg.get("reasoning_content")
+    existing, reasoning = source_msg.get("reasoning_content"), source_msg.get("reasoning")
     if isinstance(existing, str):
         # Explicit value: preserve verbatim, upgrading legacy "" to " " (DeepSeek V4 400s on "").
         api_msg["reasoning_content"] = existing or " "
-        return
-    reasoning = source_msg.get("reasoning")
-    if isinstance(reasoning, str) and reasoning and not source_msg.get("tool_calls"):
+    elif isinstance(reasoning, str) and reasoning and not source_msg.get("tool_calls"):
         # Healthy session: promote internal 'reasoning' → 'reasoning_content'.
         api_msg["reasoning_content"] = reasoning
-        return
-    # tool_calls + 'reasoning' but no 'reasoning_content' means the reasoning came from
-    # ANOTHER provider (DeepSeek's own build pins reasoning_content for tool-call turns):
-    # pad without leaking foreign CoT. No reasoning at all: every assistant turn still needs
-    # the field; " " (not "") because DeepSeek V4 rejects empty string.
-    api_msg["reasoning_content"] = " "
+    else:
+        # tool_calls + 'reasoning' but no 'reasoning_content' means the reasoning came from
+        # ANOTHER provider (DeepSeek's own build pins reasoning_content for tool-call turns):
+        # pad without leaking foreign CoT. No reasoning at all: every assistant turn still needs
+        # the field; " " (not "") because DeepSeek V4 rejects empty string.
+        api_msg["reasoning_content"] = " "
 
 
 def reapply_reasoning_echo(api_messages: list, needs_thinking_pad: bool) -> int:
     """Re-pad (or strip) assistant turns' reasoning_content for the ACTIVE provider.
 
-    ``api_messages`` is built once under the primary provider; a mid-conversation
-    fallback can switch providers, so baked-in fields must be reconciled: TO a
-    require-side provider re-applies the pad (else 400), TO a strict provider strips it
-    (else 422). Idempotent. Returns the number of assistant turns changed.
+    ``api_messages`` is built once under the primary provider; a mid-conversation fallback
+    can switch providers, so baked-in fields must be reconciled: TO a require-side provider
+    re-applies the pad (else 400), TO a strict one strips it (else 422). Idempotent.
+    Returns the number of assistant turns changed.
     """
     changed = 0
     for api_msg in api_messages:
         if api_msg.get("role") != "assistant":
             continue
         if needs_thinking_pad:
-            if api_msg.get("reasoning_content"):
-                continue
-            apply_reasoning_content_policy(api_msg, api_msg, needs_thinking_pad)
-            if api_msg.get("reasoning_content"):
-                changed += 1
+            if not api_msg.get("reasoning_content"):
+                apply_reasoning_content_policy(api_msg, api_msg, needs_thinking_pad)
+                changed += 1 if api_msg.get("reasoning_content") else 0
         elif "reasoning_content" in api_msg:
             api_msg.pop("reasoning_content", None)
             changed += 1
diff --git a/agent/micro_compaction.py b/agent/micro_compaction.py
index 78abdd0037..6e4e5a01c6 100644
--- a/agent/micro_compaction.py
+++ b/agent/micro_compaction.py
@@ -37,60 +37,49 @@ class MicroCompactionMixin:
     """Rolling micro-compaction; host must be a ``ContextCompressor``."""
 
     def _resolve_compact_cursor(self, messages: List[Dict[str, Any]], head_end: int, tail_start: int) -> int:
-        """Index of the first message not yet absorbed into the rolling summary.
-
-        Uses the in-memory cursor when valid; otherwise scans for the last summary marker.
-        """
+        """Index of the first message not yet absorbed into the rolling summary: the in-memory
+        cursor when valid, else just past the last summary marker."""
         if head_end < self._micro_compact_cursor < tail_start:
             return self._micro_compact_cursor
-        last_summary_idx = max(
-            (idx for idx in range(head_end, tail_start) if self._is_context_summary_message(messages[idx])),
-            default=-1,
-        )
-        cursor = head_end
-        if last_summary_idx >= head_end:
-            cursor = last_summary_idx + 1
+        summaries = (i for i in range(head_end, tail_start) if self._is_context_summary_message(messages[i]))
+        last = max(summaries, default=-1)
+        cursor = last + 1 if last >= head_end else head_end
+        if last >= head_end:
             # Resumed session: rehydrate the rolling summary from the surviving marker so the next
             # pass merges, not replaces.
             recovered = "" if self._micro_compact_rolling_summary.strip() else (
-                self._rolling_summary_from_marker(messages[last_summary_idx].get("content"))
+                self._rolling_summary_from_marker(messages[last].get("content"))
             )
             if recovered:
                 self._micro_compact_rolling_summary = recovered
                 # Rehydration proves containment: this marker (batch or micro) becomes
                 # supersede/defrag-eligible; unabsorbed markers never get the key.
-                messages[last_summary_idx][_cc().MICRO_COMPACT_MARKER_KEY] = True
-                logger.info(
-                    "Micro-compaction: recovered rolling summary from "
-                    "transcript (%d chars)", len(recovered),
-                )
+                messages[last][_cc().MICRO_COMPACT_MARKER_KEY] = True
+                logger.info("Micro-compaction: recovered rolling summary from transcript (%d chars)", len(recovered))
         self._micro_compact_cursor = cursor
         return cursor
 
     def _find_one_exchange(
         self, messages: List[Dict[str, Any]], start: int, tail_start: int,
     ) -> Optional[tuple[int, int]]:
-        """Find the next complete exchange (full agent turn) starting at *start*.
-
-        Returns ``(exchange_start, exchange_end)`` or ``None``. Spans assistant+tool rows up to the
-        next user message; user turns are never absorbed (alternation safety, verbatim user text).
-        """
+        """Find the next complete exchange (full agent turn) starting at *start*; returns
+        ``(exchange_start, exchange_end)`` or ``None``. Spans assistant+tool rows up to the next
+        user message; user turns are never absorbed (alternation safety, verbatim user text)."""
         limit = min(tail_start, len(messages))
 
-        def _real_assistant(msg: Dict[str, Any]) -> bool:
-            return msg.get("role") == "assistant" and not self._is_context_summary_message(msg)
+        def _turn_row(idx: int, roles: tuple) -> bool:
+            return messages[idx].get("role") in roles and not self._is_context_summary_message(messages[idx])
 
         # Skip user messages and (assistant-role) summary markers to reach a real assistant message;
         # otherwise a rehydrated cursor could absorb the marker itself.
         idx = start
-        while idx < limit and not _real_assistant(messages[idx]):
+        while idx < limit and not _turn_row(idx, ("assistant",)):
             idx += 1
         if idx >= limit:
             return None
-
         exchange_start = idx
         idx += 1
-        while idx < limit and messages[idx].get("role") in ("assistant", "tool") and not self._is_context_summary_message(messages[idx]):
+        while idx < limit and _turn_row(idx, ("assistant", "tool")):
             idx += 1
 
         # Boundary must close the turn: a mid-turn stop at tail_start would put the assistant marker
@@ -138,11 +127,8 @@ class MicroCompactionMixin:
             call_kwargs["model"] = self.summary_model
         if self.model:
             call_kwargs.setdefault("main_runtime", {
-                "model": self.model,
-                "provider": self.provider or "",
-                "base_url": self.base_url or "",
-                "api_key": self.api_key or "",
-                "api_mode": getattr(self, "api_mode", "") or "",
+                "model": self.model, "provider": self.provider or "", "base_url": self.base_url or "",
+                "api_key": self.api_key or "", "api_mode": getattr(self, "api_mode", "") or "",
             })
 
         try:
@@ -155,8 +141,7 @@ class MicroCompactionMixin:
         # A length stop means a partial merge; leave the exchange unabsorbed so a later pass retries.
         if _cc()._response_finish_reason(response) == "length":
             logger.warning(
-                "micro-summarization output hit the token cap "
-                "(finish_reason=length) — discarding partial summary",
+                "micro-summarization output hit the token cap (finish_reason=length) — discarding partial summary",
             )
             return None
 
@@ -176,33 +161,29 @@ class MicroCompactionMixin:
 
     def _defrag_rolling_summary(self, messages: List[Dict[str, Any]]) -> bool:
         """Re-summarize the rolling summary text and rewrite the marker in place.
-
-        Transcript-shape-neutral (no splice, no cursor move). Returns True when it rewrote.
-        """
+        Transcript-shape-neutral (no splice, no cursor move). Returns True when it rewrote."""
         old_summary = self._micro_compact_rolling_summary
         if not old_summary.strip():
             return False
         # Empty base turns the merge prompt into a rewrite-compactly instruction.
         self._micro_compact_rolling_summary = ""
         fresh_summary = self._micro_summarize_one(old_summary)
+        self._micro_compact_rolling_summary = fresh_summary or old_summary
         if not fresh_summary:
-            self._micro_compact_rolling_summary = old_summary
             return False
-        self._micro_compact_rolling_summary = fresh_summary
         # Rewrite only the newest MICRO marker (resume rehydrates from it); a batch marker holds
         # history we lack.
-        for entry in reversed(messages):
-            if _is_micro_marker(entry):
-                entry["content"] = self._render_micro_marker_content(fresh_summary)
-                # Content changed: clear the persisted stamp so the DB sync rewrites the row. An
-                # in-place pop on a live dict would be identity-skipped by the bounded flush scan;
-                # flag the finalizer.
-                entry.pop(_cc()._DB_PERSISTED_MARKER, None)
-                self._flush_scan_cursor_invalidated = True
-                break
+        entry = next((e for e in reversed(messages) if _is_micro_marker(e)), None)
+        if entry is not None:
+            entry["content"] = self._render_micro_marker_content(fresh_summary)
+            # Content changed: clear the persisted stamp so the DB sync rewrites the row. An
+            # in-place pop on a live dict would be identity-skipped by the bounded flush scan;
+            # flag the finalizer.
+            entry.pop(_cc()._DB_PERSISTED_MARKER, None)
+            self._flush_scan_cursor_invalidated = True
         logger.info(
-            "Micro-compaction defrag: rolling summary re-summarized "
-            "(%d -> %d chars)", len(old_summary), len(fresh_summary),
+            "Micro-compaction defrag: rolling summary re-summarized (%d -> %d chars)",
+            len(old_summary), len(fresh_summary),
         )
         return True
 
@@ -211,11 +192,9 @@ class MicroCompactionMixin:
         self._micro_compact_last_failure_cursor = -1
 
     def _micro_compact(self, messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """Run one round of micro-compaction; public entry point from ``finalize_turn()``.
-
-        Returns the (possibly modified) list and syncs the session DB via ``archive_and_compact``
-        (the append-only flush alone would double-load on resume).
-        """
+        """Run one round of micro-compaction (entry point from ``finalize_turn()``). Returns the
+        (possibly modified) list and syncs the session DB via ``archive_and_compact`` (the
+        append-only flush alone would double-load on resume)."""
         if not self._micro_compact_enabled:
             return messages
 
@@ -229,9 +208,7 @@ class MicroCompactionMixin:
             self._micro_compact_turns_since_pass = 0
 
         n_messages = len(messages)
-        if n_messages < 4:
-            return messages
-        exchange = self._next_exchange(messages)
+        exchange = self._next_exchange(messages) if n_messages >= 4 else None
         if exchange is None:
             return messages
         exchange_start, exchange_end = exchange
@@ -253,10 +230,8 @@ class MicroCompactionMixin:
             if defragged:
                 self._sync_micro_compact_to_db(messages)
                 self._reset_micro_failure_tracking()
-            _telemetry(
-                "defrag" if defragged else "defrag_failed", messages,
-                tokens_after=estimate_messages_tokens_rough(messages),
-            )
+            outcome = "defrag" if defragged else "defrag_failed"
+            _telemetry(outcome, messages, tokens_after=estimate_messages_tokens_rough(messages))
             return messages
 
         # Cumulative iff it subsumes an earlier marker; captured before summarizing.
@@ -278,24 +253,20 @@ class MicroCompactionMixin:
         self._micro_compact_cursor = self._cursor_after_splice(result, exchange_start + 1)
         self._sync_micro_compact_to_db(result)
         _telemetry(
-            "absorbed", result,
-            tokens_after=estimate_messages_tokens_rough(result), exchange_tokens=_exchange_tokens,
+            "absorbed", result, tokens_after=estimate_messages_tokens_rough(result), exchange_tokens=_exchange_tokens,
         )
         return result
 
     def _record_micro_failure(self, exchange_start: int, exchange_end: int) -> str:
         """Count a summarize failure at this cursor; skip the exchange after too many in a row."""
         # Track consecutive failures at the same cursor to avoid busy-looping every turn.
-        if exchange_start == self._micro_compact_last_failure_cursor:
-            self._micro_compact_consecutive_failures += 1
-        else:
-            self._micro_compact_consecutive_failures = 1
-            self._micro_compact_last_failure_cursor = exchange_start
+        same_cursor = exchange_start == self._micro_compact_last_failure_cursor
+        self._micro_compact_consecutive_failures = self._micro_compact_consecutive_failures + 1 if same_cursor else 1
+        self._micro_compact_last_failure_cursor = exchange_start
         if self._micro_compact_consecutive_failures < _cc()._MICRO_COMPACT_MAX_CONSECUTIVE_FAILURES:
             return "summarize_failed"
         logger.info(
-            "Micro-compaction: skipping exchange at cursor %d "
-            "after %d consecutive failures",
+            "Micro-compaction: skipping exchange at cursor %d after %d consecutive failures",
             exchange_start, self._micro_compact_consecutive_failures,
         )
         # Skip the stuck exchange; it stays in the transcript for batch compression/defrag.
@@ -310,9 +281,7 @@ class MicroCompactionMixin:
         if compress_start >= compress_end:
             return None
         cursor = self._resolve_compact_cursor(messages, compress_start, compress_end)
-        if cursor >= compress_end:
-            return None
-        return self._find_one_exchange(messages, cursor, compress_end)
+        return None if cursor >= compress_end else self._find_one_exchange(messages, cursor, compress_end)
 
     @staticmethod
     def _rolling_summary_from_marker(content: Any) -> str:
@@ -320,70 +289,44 @@ class MicroCompactionMixin:
         cc = _cc()
         if not isinstance(content, str) or not content.strip():
             return ""
-        body = content
         # rfind: SUMMARY_PREFIX itself mentions the heading, so the first hit is in the preamble.
-        idx = body.rfind(cc.HISTORICAL_TASK_HEADING)
-        if idx != -1:
-            body = body[idx + len(cc.HISTORICAL_TASK_HEADING):]
+        idx = content.rfind(cc.HISTORICAL_TASK_HEADING)
+        body = content[idx + len(cc.HISTORICAL_TASK_HEADING):] if idx != -1 else content
         end = body.find(cc._SUMMARY_END_MARKER)
-        if end != -1:
-            body = body[:end]
-        return body.strip()
+        return (body[:end] if end != -1 else body).strip()
 
     def _cursor_after_splice(self, result: List[Dict[str, Any]], fallback: int) -> int:
-        """Cursor position just past the summary marker in *result*.
-
-        Must derive from the SPLICED list: a splice collapses several rows into one marker
-        (and may drop a superseded one), so pre-splice indices land inside a later exchange
-        and silently skip it.
-        """
-        for idx in range(len(result) - 1, -1, -1):
-            if _is_summary_marker(result[idx]):
-                return idx + 1
-        return fallback
+        """Cursor position just past the summary marker in *result*. Must derive from the SPLICED
+        list: a splice collapses several rows into one marker (and may drop a superseded one), so
+        pre-splice indices land inside a later exchange and silently skip it."""
+        return next((idx + 1 for idx in range(len(result) - 1, -1, -1) if _is_summary_marker(result[idx])), fallback)
 
     def _emit_micro_compaction_telemetry(
-        self, *, outcome: str, messages_before: int, messages_after: int,
-        tokens_before: int | None, tokens_after: int | None,
-        exchange_tokens: int | None = None, duration_ms: int | None = None,
+        self, *, outcome: str, messages_before: int, messages_after: int, tokens_before: int | None,
+        tokens_after: int | None, exchange_tokens: int | None = None, duration_ms: int | None = None,
     ) -> None:
         """Emit one content-free JSON log line for a micro-compaction pass.
-
-        ``tokens_delta`` < 0 means the pass shrank the transcript; ``*_total`` fields accumulate.
-        """
+        ``tokens_delta`` < 0 means the pass shrank the transcript; ``*_total`` fields accumulate."""
         _safe_int = _cc()._safe_int
         try:
-            delta = None
-            if tokens_before is not None and tokens_after is not None:
-                delta = tokens_after - tokens_before
-                self._micro_compact_tokens_saved_total -= delta
+            delta = tokens_after - tokens_before if tokens_before is not None and tokens_after is not None else None
+            self._micro_compact_tokens_saved_total -= delta or 0
             self._micro_compact_passes += 1
             # Cached reads only: the lazy properties can fire a synchronous /models probe.
             threshold = self._threshold_tokens
-            occupancy = None
-            if threshold and tokens_after is not None and threshold > 0:
-                occupancy = round(tokens_after / threshold * 100, 1)
+            has_occupancy = threshold and tokens_after is not None and threshold > 0
+            occupancy = round(tokens_after / threshold * 100, 1) if has_occupancy else None
             payload = {
-                "event": "micro_compaction",
-                "session_id": getattr(self, "_session_id", "") or "",
-                "outcome": outcome,
-                "messages_before": messages_before,
-                "messages_after": messages_after,
-                "tokens_before": _safe_int(tokens_before),
-                "tokens_after": _safe_int(tokens_after),
-                "tokens_delta": _safe_int(delta),
-                "exchange_tokens": _safe_int(exchange_tokens),
+                "event": "micro_compaction", "session_id": getattr(self, "_session_id", "") or "", "outcome": outcome,
+                "messages_before": messages_before, "messages_after": messages_after,
+                "tokens_before": _safe_int(tokens_before), "tokens_after": _safe_int(tokens_after),
+                "tokens_delta": _safe_int(delta), "exchange_tokens": _safe_int(exchange_tokens),
                 "rolling_summary_tokens": estimate_tokens_rough(self._micro_compact_rolling_summary),
-                "cursor": _safe_int(self._micro_compact_cursor),
-                "passes_total": self._micro_compact_passes,
-                "tokens_saved_total": self._micro_compact_tokens_saved_total,
-                "duration_ms": _safe_int(duration_ms),
+                "cursor": _safe_int(self._micro_compact_cursor), "passes_total": self._micro_compact_passes,
+                "tokens_saved_total": self._micro_compact_tokens_saved_total, "duration_ms": _safe_int(duration_ms),
                 # Headroom: how full the window is being kept.
-                "threshold_tokens": _safe_int(threshold),
-                "context_limit": _safe_int(self._resolved_context_length),
-                "occupancy_pct": occupancy,
-                "main_model": self.model or "",
-                "aux_model": self.summary_model or "",
+                "threshold_tokens": _safe_int(threshold), "context_limit": _safe_int(self._resolved_context_length),
+                "occupancy_pct": occupancy, "main_model": self.model or "", "aux_model": self.summary_model or "",
             }
             logger.info("micro compaction telemetry: %s", json.dumps(payload, sort_keys=True, separators=(",", ":")))
         except Exception as exc:
@@ -391,10 +334,8 @@ class MicroCompactionMixin:
 
     def _sync_micro_compact_to_db(self, compacted_messages: List[Dict[str, Any]]) -> None:
         """Persist the micro-compacted set to the session DB atomically and stamp rows persisted.
-
-        Without this the old exchange rows stay ``active=1`` and a resume double-loads
-        both the summary and the originals.
-        """
+        Without this the old exchange rows stay ``active=1`` and a resume double-loads both the
+        summary and the originals."""
         session_db, session_id = getattr(self, "_session_db", None), getattr(self, "_session_id", "")
         if not session_db or not session_id:
             return
@@ -413,17 +354,14 @@ class MicroCompactionMixin:
         self, messages: List[Dict[str, Any]], splice_start: int, splice_end: int, supersede: bool = True,
     ) -> List[Dict[str, Any]]:
         """Replace *messages[splice_start:splice_end]* with an assistant-role summary marker.
-
-        Merges user turns left adjacent by a superseded marker so the result is alternation-valid.
-        """
+        Merges user turns left adjacent by a superseded marker so the result is alternation-valid."""
         cc = _cc()
         summary_text = self._micro_compact_rolling_summary
         if not summary_text.strip():
             return messages
 
         summary_msg = {
-            "role": "assistant",
-            "content": self._render_micro_marker_content(summary_text),
+            "role": "assistant", "content": self._render_micro_marker_content(summary_text),
             cc.COMPRESSED_SUMMARY_METADATA_KEY: True,
             # Micro marker: eligible for supersede/defrag; batch markers never carry this key.
             cc.MICRO_COMPACT_MARKER_KEY: True,
@@ -434,10 +372,9 @@ class MicroCompactionMixin:
 
         # Cumulative summary: keep only the newest marker. Drop an older one only if supersede AND
         # it has MICRO_COMPACT_MARKER_KEY (provably absorbed); a batch marker holds MORE history.
-        if supersede:
-            marker_idxs = [i for i, m in enumerate(result) if _is_micro_marker(m)]
-            if len(marker_idxs) > 1:
-                result = self._merge_adjacent_user_turns([m for i, m in enumerate(result) if i not in marker_idxs[:-1]])
+        stale = [i for i, m in enumerate(result) if _is_micro_marker(m)][:-1] if supersede else []
+        if stale:
+            result = self._merge_adjacent_user_turns([m for i, m in enumerate(result) if i not in stale])
 
         # Deliberately no _strip_persistence_markers: micro archives in place under the same session
         # id, so stamps stay accurate and a failed archive keeps the append-only flush idempotent.
@@ -451,17 +388,15 @@ class MicroCompactionMixin:
 
     @staticmethod
     def _merge_adjacent_user_turns(result: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
-        """Merge consecutive plain-text real user turns left by a supersede.
-
-        Same ``\\n\\n`` join as ``repair_message_sequence`` pass 2, done here so the marker
-        and cursor are never collateral damage of the downstream repair. Lists untouched.
-        """
+        """Merge consecutive plain-text real user turns left by a supersede. Same ``\\n\\n`` join as
+        ``repair_message_sequence`` pass 2, done here so the marker and cursor are never collateral
+        damage of the downstream repair. Lists untouched."""
         from agent.turn_context import drop_stale_api_content
 
         def _plain_user(m: Any) -> bool:
             return (
-                isinstance(m, dict) and m.get("role") == "user"
-                and not _is_summary_marker(m) and isinstance(m.get("content"), str)
+                isinstance(m, dict) and m.get("role") == "user" and not _is_summary_marker(m)
+                and isinstance(m.get("content"), str)
             )
 
         merged: List[Dict[str, Any]] = []
diff --git a/agent/moa_loop.py b/agent/moa_loop.py
index c0f155c0fb..e5ac42fb3d 100644
--- a/agent/moa_loop.py
+++ b/agent/moa_loop.py
@@ -202,10 +202,9 @@ def _slot_label(slot: dict[str, Any]) -> str:
 
 def _slot_reasoning_config(slot: dict[str, Any]) -> dict[str, Any] | None:
     """Translate optional per-MoA-slot reasoning_effort into runtime config."""
-    effort = slot.get("reasoning_effort")
     try:
         from hermes_constants import parse_reasoning_effort
-        return parse_reasoning_effort(effort)
+        return parse_reasoning_effort(slot.get("reasoning_effort"))
     except Exception:  # pragma: no cover - bad config must not break MoA
         return None
 
@@ -246,11 +245,10 @@ def _slot_runtime(slot: dict[str, Any]) -> dict[str, Any]:
         from hermes_cli.runtime_provider import resolve_runtime_provider
         rt = resolve_runtime_provider(requested=provider, target_model=model)
         out.update({k: rt[k] for k in ("base_url", "api_key", "api_mode") if rt.get(k)})
-        request_overrides = rt.get("request_overrides")
-        if isinstance(request_overrides, dict):
-            extra_body = request_overrides.get("extra_body")
-            if isinstance(extra_body, dict) and extra_body:
-                out["extra_body"] = dict(extra_body)
+        overrides = rt.get("request_overrides")
+        extra_body = overrides.get("extra_body") if isinstance(overrides, dict) else None
+        if isinstance(extra_body, dict) and extra_body:
+            out["extra_body"] = dict(extra_body)
     except Exception as exc:  # pragma: no cover - defensive
         logger.debug("MoA slot runtime resolution failed for %s: %s", _slot_label(slot), exc)
         return out
@@ -261,27 +259,21 @@ def _slot_runtime(slot: dict[str, Any]) -> dict[str, Any]:
 
 def _merge_slot_extra_body(slot_extra_body: Any, caller_extra_body: Any) -> Any:
     """Merge slot defaults with a caller override (caller wins) for ``call_llm``."""
-    if isinstance(slot_extra_body, dict) and slot_extra_body:
-        if isinstance(caller_extra_body, dict):
-            return {**slot_extra_body, **caller_extra_body}
-        if caller_extra_body:
-            return caller_extra_body
-        return dict(slot_extra_body)
-    return caller_extra_body
+    if not (isinstance(slot_extra_body, dict) and slot_extra_body):
+        return caller_extra_body
+    if isinstance(caller_extra_body, dict):
+        return {**slot_extra_body, **caller_extra_body}
+    return caller_extra_body or dict(slot_extra_body)
 
 
 def _agent_cache_opts(agent: Any) -> tuple[Any, Any]:
     """The live agent's ``(_cache_disabled, _cache_ttl)``; ``(None, None)`` without an agent."""
-    if agent is None:
-        return None, None
-    return getattr(agent, "_cache_disabled", None), getattr(agent, "_cache_ttl", None)
+    return (None, None) if agent is None else (getattr(agent, "_cache_disabled", None), getattr(agent, "_cache_ttl", None))
 
 
 def _with_cache_disabled(runtime: dict[str, Any], cache_disabled: Any) -> dict[str, Any]:
     """Pin the live agent's cache disable onto a runtime snapshot (None is a no-op)."""
-    if cache_disabled is None:
-        return runtime
-    return {**runtime, "_cache_disabled": cache_disabled}
+    return runtime if cache_disabled is None else {**runtime, "_cache_disabled": cache_disabled}
 
 
 def _maybe_apply_moa_cache_control(
@@ -697,9 +689,7 @@ def _preset_temperature(preset: dict[str, Any], key: str) -> float | None:
 
 
 def _hash_messages(msgs: list[dict[str, Any]]) -> str:
-    return hashlib.sha256(
-        "\u0000".join(f"{m.get('role')}:{m.get('content')}" for m in msgs).encode("utf-8", "replace")
-    ).hexdigest()
+    return hashlib.sha256("\u0000".join(f"{m.get('role')}:{m.get('content')}" for m in msgs).encode("utf-8", "replace")).hexdigest()
 
 
 def _is_failed_reference(text: str) -> bool:
@@ -777,10 +767,8 @@ def aggregate_moa_context(
         logger.debug("MoA privacy filter check failed", exc_info=True)
     agg_refs, degraded, all_failed = _guidance_inputs(reference_outputs, privacy_full, degraded_reference_policy)
     joined = _join_reference_outputs(agg_refs, degraded)
-
-    # Every reference failed: skip the aggregator (synthesizing over nothing can block
-    # for the full provider timeout) and return only the sanitized notice.
     if all_failed:
+        # Skip the aggregator: synthesizing over nothing can block for the full provider timeout.
         logger.warning("MoA: all %d reference(s) failed — skipping aggregator synthesis", len(reference_outputs))
         return (
             "[Mixture of Agents context — all reference models failed. "
@@ -806,14 +794,12 @@ def aggregate_moa_context(
         # Same cache_control decoration as the advisor calls; this synthesis call is
         # a third independent MoA call path that otherwise re-bills its full input.
         agg_messages = _maybe_apply_moa_cache_control(
-            [{"role": "user", "content": synth_prompt}],
-            _with_cache_disabled(agg_runtime, cache_disabled), cache_ttl=cache_ttl,
+            [{"role": "user", "content": synth_prompt}], _with_cache_disabled(agg_runtime, cache_disabled), cache_ttl=cache_ttl,
         )
-        response = call_llm(
+        synthesis = _extract_text(call_llm(
             task="moa_aggregator", messages=agg_messages, temperature=aggregator_temperature,
             reasoning_config=_aggregator_reasoning_config(aggregator), **agg_runtime,
-        )
-        synthesis = _extract_text(response)
+        ))
     except Exception as exc:
         logger.warning("MoA aggregator model %s failed: %s", agg_label, exc)
         synthesis = ""
@@ -838,9 +824,7 @@ def _completed_response_as_stream_chunk(response: Any) -> Any:
     if isinstance(raw_tool_calls, (list, tuple)) and raw_tool_calls:
         tool_call_deltas = [
             SimpleNamespace(
-                index=getattr(tc, "index", index),
-                id=getattr(tc, "id", None),
-                type=getattr(tc, "type", None) or "function",
+                index=getattr(tc, "index", index), id=getattr(tc, "id", None), type=getattr(tc, "type", None) or "function",
                 function=SimpleNamespace(
                     name=getattr(getattr(tc, "function", None), "name", None),
                     arguments=getattr(getattr(tc, "function", None), "arguments", None),
@@ -853,12 +837,10 @@ def _completed_response_as_stream_chunk(response: Any) -> Any:
         **{k: getattr(message, k, None) for k in ("reasoning_content", "reasoning", "reasoning_details")},
     )
     choice = SimpleNamespace(
-        index=getattr(first_choice, "index", 0), delta=delta,
-        finish_reason=getattr(first_choice, "finish_reason", None) or "stop",
+        index=getattr(first_choice, "index", 0), delta=delta, finish_reason=getattr(first_choice, "finish_reason", None) or "stop",
     )
     return SimpleNamespace(
-        id=getattr(response, "id", None), model=getattr(response, "model", None), choices=[choice],
-        usage=getattr(response, "usage", None),
+        id=getattr(response, "id", None), model=getattr(response, "model", None), choices=[choice], usage=getattr(response, "usage", None),
     )
 
 
@@ -871,15 +853,13 @@ def _attach_reference_guidance(agg_messages: list[dict[str, Any]], guidance: str
     consecutive user turns would be rejected by strict providers).
     """
     last = agg_messages[-1] if agg_messages else None
-    if last is not None and last.get("role") == "user":
-        last_content = last.get("content")
-        if isinstance(last_content, str):
-            last["content"] = last_content + "\n\n" + guidance
-            return
-        if isinstance(last_content, list):
-            last["content"] = [*last_content, {"type": "text", "text": "\n\n" + guidance}]
-            return
-    agg_messages.append({"role": "user", "content": guidance})
+    last_content = last.get("content") if last is not None and last.get("role") == "user" else None
+    if isinstance(last_content, str):
+        last["content"] = last_content + "\n\n" + guidance
+    elif isinstance(last_content, list):
+        last["content"] = [*last_content, {"type": "text", "text": "\n\n" + guidance}]
+    else:
+        agg_messages.append({"role": "user", "content": guidance})
 
 
 def peel_reference_guidance(messages: list[dict[str, Any]], guidance: Any) -> list[dict[str, Any]]:
@@ -892,26 +872,21 @@ def peel_reference_guidance(messages: list[dict[str, Any]], guidance: Any) -> li
     if not isinstance(last, dict) or last.get("role") != "user":
         return messages
     content = last.get("content")
-    if content == guidance_text:
-        # Attach shape (c): guidance was appended as its own user message.
+    if content == guidance_text:  # shape (c): guidance was its own user message
         return list(messages[:-1])
     suffix = "\n\n" + guidance_text
-    if isinstance(content, str) and content.endswith(suffix):
-        # Attach shape (a): merged into a trailing string user turn.
+    if isinstance(content, str) and content.endswith(suffix):  # shape (a): merged into a string turn
         return [*messages[:-1], {**last, "content": content[: -len(suffix)]}]
     if isinstance(content, list) and content:
         last_part = content[-1]
         if isinstance(last_part, dict) and last_part.get("type", "text") == "text":
             text = last_part.get("text") or ""
             if text in (suffix, guidance_text):
-                # Attach shape (b): guidance rode as its own trailing part. Guidance as
-                # the only content drops the whole message (mirrors shape c).
-                if len(content) == 1:
-                    return list(messages[:-1])
-                return [*messages[:-1], {**last, "content": list(content[:-1])}]
+                # Shape (b): guidance rode as its own trailing part. Guidance as the
+                # only content drops the whole message (mirrors shape c).
+                return list(messages[:-1]) if len(content) == 1 else [*messages[:-1], {**last, "content": list(content[:-1])}]
             if text.endswith(suffix):
-                new_part = {**last_part, "text": text[: -len(suffix)]}
-                return [*messages[:-1], {**last, "content": [*content[:-1], new_part]}]
+                return [*messages[:-1], {**last, "content": [*content[:-1], {**last_part, "text": text[: -len(suffix)]}]}]
     return messages