Merge branch 'master' into feat/gateway-run-reattach-after-restart

api/agent_sessions.py

@@ -366,28 +366,59 @@ def is_cli_session_row_visible(row: dict) -> bool:
     return _count_user_turns(row) >= CLI_MIN_UNTITLED_USER_MESSAGE_COUNT
 
 
-def _branch_markers(row: dict | None) -> tuple[str | None, str | None]:
-    """Return ``(_branched_from, _delegate_from)`` from a row's ``model_config``.
+# Every model_config marker Hermes Agent binds to ``parent_session_id`` in its
+# non-continuation child predicate (``_NON_CONTINUATION_CHILD_FILTER_SQL``):
+# ``_delegate_from`` (delegate_task), ``_branched_from`` (/branch) and
+# ``_reset_from`` (gateway reset children, stamped at creation or durably by
+# ``reopen_session()`` in ``gateway/session_recovery.py``).
+_MODEL_CONFIG_LINEAGE_KEYS = ('_delegate_from', '_branched_from', '_reset_from')
 
-    Hermes Agent stamps explicit branches and delegate/subagent runs in the
-    ``model_config`` JSON column (``_branched_from`` / ``_delegate_from``);
-    ``session_source='fork'`` alone only covers WebUI-created forks. Missing,
-    empty, or unparsable ``model_config`` degrades to ``(None, None)``.
+
+def _branch_markers(row: dict | None) -> tuple[str, dict[str, str]]:
+    """Return ``(state, markers)`` for a row's ``model_config`` lineage identity.
+
+    Hermes Agent stamps explicit branches, delegate/subagent runs and gateway
+    reset children in the ``model_config`` JSON column (see
+    ``_MODEL_CONFIG_LINEAGE_KEYS``); ``session_source='fork'`` alone only
+    covers WebUI-created forks.
+
+    ``state`` is one of:
+
+    - ``'none'``: no ``model_config``, or an object carrying no lineage marker;
+    - ``'markers'``: every non-null marker is a non-empty string, returned by key;
+    - ``'unknown'``: identity evidence exists but cannot be trusted (payload is
+      not JSON, not a JSON object, too deeply nested to decode, or a marker is
+      not a non-empty string). Callers must fail closed on this state and treat
+      the row as a lineage boundary, never as a continuation.
     """
     if not row:
-        return None, None
+        return 'none', {}
     raw = row.get('model_config')
-    if isinstance(raw, dict):
-        config = raw
-    elif isinstance(raw, str) and raw.strip():
+    if raw is None:
+        return 'none', {}
+    if isinstance(raw, (str, bytes, bytearray)):
+        if not raw.strip():
+            return 'none', {}
         try:
-            parsed = json.loads(raw)
-        except (TypeError, ValueError):
-            parsed = None
-        config = parsed if isinstance(parsed, dict) else {}
+            config = json.loads(raw)
+        except (TypeError, ValueError, RecursionError):
+            # RecursionError: a valid but pathologically deep payload exceeds
+            # the decoder's recursion limit; it is untrusted identity evidence
+            # like any other undecodable payload, never an escaping crash.
+            return 'unknown', {}
     else:
-        config = {}
-    return config.get('_branched_from'), config.get('_delegate_from')
+        config = raw
+    if not isinstance(config, dict):
+        return 'unknown', {}
+    markers: dict[str, str] = {}
+    for key in _MODEL_CONFIG_LINEAGE_KEYS:
+        value = config.get(key)
+        if value is None:
+            continue
+        if not isinstance(value, str) or not value.strip():
+            return 'unknown', {}
+        markers[key] = value.strip()
+    return ('markers', markers) if markers else ('none', {})
 
 
 def _is_continuation_session(parent: dict | None, child: dict | None) -> bool:
@@ -405,30 +436,36 @@ def _is_continuation_session(parent: dict | None, child: dict | None) -> bool:
     conversation; otherwise the tip inherits the root's title/source metadata and
     can disappear under messaging/sidebar policies.
 
-    Explicit branches/delegates never continue: a child whose ``model_config``
-    ``_branched_from`` / ``_delegate_from`` points at this parent is a fork of
-    the conversation and must stay visible. Beyond the bounded tolerance window
-    there is no reliable continuation signal — titles are user-controlled and
-    non-unique, so they are never used to widen the window (#7021 re-gate).
+    Explicit branches/delegates/resets never continue: a child whose
+    ``model_config`` ``_branched_from`` / ``_delegate_from`` / ``_reset_from``
+    points at this parent is a separate conversation and must stay visible.
+    Unparsable or non-string marker evidence fails closed as a boundary.
+    Beyond the bounded tolerance window there is no reliable continuation
+    signal — titles are user-controlled and non-unique, so they are never used
+    to widen the window (#7021 re-gate).
     """
     if not parent or not child:
         return False
     if str(child.get('session_source') or '').strip().lower() == 'fork':
         return False
-    # Real Agent branches/delegates are marked in model_config (not
+    # Real Agent branches/delegates/resets are marked in model_config (not
     # session_source, which only WebUI-created forks carry). The marker must
     # reference THIS parent: compression continuations inherit the rotated
     # agent's model_config verbatim, so presence alone (a delegate's
     # continuation still carries the delegate's own ``_delegate_from``) would
-    # misclassify real continuations.
-    child_branched_from, child_delegate_from = _branch_markers(child)
-    parent_id = parent.get('id')
-    if parent_id and (
-        child_branched_from == parent_id or child_delegate_from == parent_id
-    ):
+    # misclassify real continuations. Untrusted identity evidence fails closed.
+    marker_state, child_markers = _branch_markers(child)
+    if marker_state == 'unknown':
+        return False
+    parent_id = str(parent.get('id') or '').strip()
+    if parent_id and parent_id in child_markers.values():
         return False
     parent_source = str(parent.get('source') or '').strip().lower()
     child_source = str(child.get('source') or '').strip().lower()
+    # Agent lineage excludes tool children even when the parent has the same
+    # source (or no source); a timestamp overlap cannot override that boundary.
+    if child_source == 'tool':
+        return False
     if parent_source and child_source and parent_source != child_source:
         return False
     if parent.get('end_reason') not in {'compression', 'cli_close'}:
@@ -576,7 +613,7 @@ def _project_agent_session_rows(rows: list[dict]) -> list[dict]:
         # expects from "Show agent sessions" sorted by activity.
         for key in (
             'id', 'model', 'message_count', 'actual_message_count', 'actual_user_message_count',
-            'ended_at', 'end_reason', 'last_activity',
+            'ended_at', 'end_reason', 'last_activity', 'archived',
         ):
             if key in tip:
                 merged[key] = tip[key]
@@ -683,6 +720,7 @@ def read_importable_agent_session_rows(
         origin_chat_id_expr = _optional_col('origin_chat_id', session_cols)
         origin_user_id_expr = _optional_col('origin_user_id', session_cols)
         platform_expr = _optional_col('platform', session_cols)
+        archived_expr = _optional_col('archived', session_cols)
         # Older/minimal state.db schemas can have NO ``messages`` table at all,
         # or a ``messages`` table without a ``session_id`` / ``timestamp`` column.
         # The projection SQL below joins ``messages`` and aggregates
@@ -793,6 +831,7 @@ def read_importable_agent_session_rows(
                    {parent_expr},
                    {ended_expr},
                    {end_reason_expr},
+                   {archived_expr},
                    {actual_count_expr} AS actual_message_count,
                    {user_message_count_expr} AS actual_user_message_count,
                    {last_activity_expr} AS last_activity

api/background_process.py

@@ -1883,7 +1883,22 @@ def recover_processes_for_webui(process_registry=None, get_session_fn=None) -> i
             try:
                 proc_session = process_registry.get(process_id)
                 session_key = str(getattr(proc_session, "session_key", "") or "")
-                if not session_key or get_session_fn(session_key, metadata_only=True) is None:
+                if not session_key:
+                    continue
+                # The session resolver (``api.models._resolve_session_once``)
+                # raises ``KeyError(sid)`` for a missing session — that is its
+                # documented contract for the "owner gone" outcome, not a
+                # fault. Treat it identically to a ``None`` return: skip the
+                # process without logging, and let a neighbouring live
+                # process still rebind. The narrow ``except KeyError`` lives
+                # INSIDE the outer try so an unrelated ``KeyError`` from
+                # ``process_registry.get()`` (or any routing-index work) still
+                # surfaces in the existing warning path.
+                try:
+                    resolved = get_session_fn(session_key, metadata_only=True)
+                except KeyError:
+                    continue
+                if resolved is None:
                     continue
             except Exception:
                 logger.warning(

api/commands.py

@@ -257,11 +257,28 @@ def _run_codex_runtime_command(arg_string: str) -> str:
 
 
 def _run_reload_mcp_command() -> str:
-    """Execute the MCP reconnect path and return a short user-facing summary."""
+    """Execute the MCP reconnect path and return a short user-facing summary.
+
+    Scoped to the request profile: only that profile's connections are shut
+    down and rediscovered from that profile's config, so other profiles'
+    same-named servers keep running (agents with profile-scoped MCP; older
+    agents keep their process-wide reload).
+    """
     with _RELOAD_MCP_LOCK:
         try:
+            import importlib
             from tools.mcp_tool import _servers, _lock
+            # Resolved by full name (like the ``from`` import above) so a stubbed
+            # ``tools.mcp_tool`` without a ``tools`` package still works.
+            _core = importlib.import_module("tools.mcp_tool")
             from api.agent_compat import agent_attr
+            from api.mcp_runtime import (
+                accepts_keywords,
+                clear_profile_connect_cooldowns,
+                ledger_key_serves_view,
+                mcp_key_name,
+                mcp_runtime_scope,
+            )
 
             shutdown_mcp_servers = agent_attr(
                 "tools.mcp_tool", "shutdown_mcp_servers", "tools.mcp_tool_lifecycle"
@@ -269,22 +286,74 @@ def _run_reload_mcp_command() -> str:
             discover_mcp_tools = agent_attr(
                 "tools.mcp_tool", "discover_mcp_tools", "tools.mcp_tool_discovery"
             )
+            try:
+                register_mcp_servers = agent_attr(
+                    "tools.mcp_tool", "register_mcp_servers", "tools.mcp_tool_discovery"
+                )
+            except Exception:
+                register_mcp_servers = None  # older agent: no overlay reconciliation
         except Exception as exc:
             logger.warning("Failed to import MCP runtime for /reload-mcp", exc_info=True)
             raise RuntimeError("MCP runtime unavailable") from exc
 
-        try:
-            with _lock:
-                old_servers = set(_servers.keys())
+        with mcp_runtime_scope("/reload-mcp") as view:
+            if not view.trusted:
+                # Resetting under an unconfirmed scope could stop another profile's
+                # connection or rebuild this one under the wrong owner.
+                raise RuntimeError(
+                    "MCP reload unavailable: this profile's MCP runtime scope could not be "
+                    "confirmed (a chat turn on this profile may be running); retry when it finishes"
+                )
+            try:
+                # Profile-scoped teardown needs both the scoped ledger and a
+                # shutdown that takes scope/names; otherwise keep the legacy reload.
+                legacy = view.legacy or not accepts_keywords(shutdown_mcp_servers, "scope", "names")
 
-            shutdown_mcp_servers()
-            new_tools = discover_mcp_tools()
+                def live_names() -> set:
+                    # Same criteria as the status page: the connection serves this
+                    # profile (owned or adopted) and is live. A parked task (failed
+                    # spawn retained for retry) has no session and is not "connected".
+                    return {
+                        mcp_key_name(key) for key, server in _servers.items()
+                        if (legacy or ledger_key_serves_view(_core, key, view))
+                        and getattr(server, "session", True) is not None
+                    }
 
-            with _lock:
-                connected_servers = set(_servers.keys())
-        except Exception as exc:
-            logger.warning("Failed to reload MCP servers", exc_info=True)
-            raise RuntimeError("Failed to reload MCP servers") from exc
+                with _lock:
+                    old_servers = live_names()
+                    # Every ledger entry serving this profile, parked ones included:
+                    # a parked task left in place would make discovery skip the server.
+                    owned_names = {
+                        mcp_key_name(key) for key in _servers
+                        if ledger_key_serves_view(_core, key, view)
+                    }
+
+                if legacy:
+                    shutdown_mcp_servers()
+                else:
+                    # ``names`` is required: ``scope=None`` alone is the process-wide
+                    # wildcard. Only names with a ledger key are passed; shutdown
+                    # ignores anything else, and a key of another owner is never selected.
+                    shutdown_mcp_servers(scope=view.registry_scope, names=owned_names)
+                    # The wildcard also dropped connect backoff; keep that for this
+                    # profile's own failed servers so the reload retries them now.
+                    clear_profile_connect_cooldowns(_core, view)
+                    # A connection this profile ADOPTED (another profile's identical
+                    # route) is not torn down above, and ``discover_mcp_tools()``
+                    # returns before its reconcile step when the profile's config
+                    # has no server left, so the adopted tools would stay callable.
+                    # ``register_mcp_servers({})`` judges every server serving this
+                    # scope against the profile's config on disk and detaches only
+                    # this profile's overlay; the owner's connection keeps running.
+                    if register_mcp_servers is not None and view.registry_scope is not None:
+                        register_mcp_servers({})
+                new_tools = discover_mcp_tools()
+
+                with _lock:
+                    connected_servers = live_names()
+            except Exception as exc:
+                logger.warning("Failed to reload MCP servers", exc_info=True)
+                raise RuntimeError("Failed to reload MCP servers") from exc
 
     added = connected_servers - old_servers
     removed = old_servers - connected_servers

api/config.py

@@ -98,10 +98,19 @@ STATE_DIR = (
     .resolve()
 )
 
+
+def _resolve_settings_file(state_dir: Path) -> Path:
+    """Resolve an optional per-instance settings file without splitting session state."""
+    configured = os.getenv("HERMES_WEBUI_SETTINGS_FILE", "").strip()
+    if configured:
+        return Path(configured).expanduser().resolve()
+    return state_dir / "settings.json"
+
+
 SESSION_DIR = STATE_DIR / "sessions"
 WORKSPACES_FILE = STATE_DIR / "workspaces.json"
 SESSION_INDEX_FILE = SESSION_DIR / "_index.json"
-SETTINGS_FILE = STATE_DIR / "settings.json"
+SETTINGS_FILE = _resolve_settings_file(STATE_DIR)
 LAST_WORKSPACE_FILE = STATE_DIR / "last_workspace.txt"
 PROJECTS_FILE = STATE_DIR / "projects.json"
 

api/gateway_chat.py

@@ -38,6 +38,7 @@ from api.config import (
 from api.helpers import _redact_text, redact_session_data
 from api.models import clear_process_wakeup_pause, get_session, merge_session_messages_append_only
 from api.run_journal import RunJournalWriter, bound_run_journal_snapshot_args
+from api.turn_journal import append_turn_journal_event_for_stream
 
 logger = logging.getLogger(__name__)
 
@@ -1515,7 +1516,11 @@ def _run_gateway_chat_streaming(
             # role/content ordering instead of turn order.
             assistant_ts = now + 0.000001
             pending_source = getattr(s, "pending_user_source", None) or "webui"
-            from api.streaming import _active_turn_authority, _materialize_active_turn_user
+            from api.streaming import (
+                _active_turn_authority,
+                _active_turn_token_matches,
+                _materialize_active_turn_user,
+            )
 
             active_turn_identity = _active_turn_authority(s, stream_id, msg_text)
             user_msg = _materialize_active_turn_user(
@@ -1576,6 +1581,9 @@ def _run_gateway_chat_streaming(
                     s.context_messages,
                     str(msg_text or ""),
                     source=pending_source,
+                    verification_nudge_provenance={
+                        "active_turn_identity": active_turn_identity,
+                    },
                 )
             except Exception:
                 logger.debug("Failed to merge gateway display transcript", exc_info=True)
@@ -1588,6 +1596,14 @@ def _run_gateway_chat_streaming(
                         if latest_text == msg_norm:
                             display = display[:-1]
                 s.messages = display + [user_msg, assistant_msg]
+            if active_turn_identity.get("token"):
+                current_display_rows = [
+                    message
+                    for message in s.messages
+                    if _active_turn_token_matches(message, active_turn_identity)
+                ]
+                if len(current_display_rows) == 1:
+                    current_display_rows[0]["timestamp"] = user_msg["timestamp"]
             s.active_stream_id = None
             s.gateway_run = None
             s.pending_user_message = None
@@ -1623,6 +1639,21 @@ def _run_gateway_chat_streaming(
             if cancel_event.is_set():
                 _restore_cancelled_success_writeback()
                 return
+            # #6366 re-gate: record the durable same-stream completion
+            # event in the crash-safe turn journal. The run journal's
+            # terminal state is only reached on its own ``stream_end``
+            # write path, so a Gateway run whose terminal write is lost
+            # would otherwise leave no completion evidence at all and
+            # stale-cancel recovery would re-append a duplicate
+            # recovered row after the valid final answer.
+            try:
+                append_turn_journal_event_for_stream(
+                    session_id,
+                    stream_id,
+                    {"event": "completed", "created_at": time.time()},
+                )
+            except Exception:
+                logger.debug("Failed to append completed turn journal event", exc_info=True)
             success_writeback_committed = True
         try:
             from api.goals import evaluate_goal_after_turn, has_active_goal

api/gateway_watcher.py

@@ -9,6 +9,7 @@ This enables real-time session list updates in the sidebar without
 requiring any changes to hermes-agent.
 """
 import hashlib
+import json
 import logging
 import os
 import queue
@@ -27,12 +28,27 @@ logger = logging.getLogger(__name__)
 # ── State hash tracking ─────────────────────────────────────────────────────
 
 def _snapshot_hash(sessions: list) -> str:
-    """Create a lightweight hash of session IDs and timestamps for change detection."""
-    key = '|'.join(
-        f"{s['session_id']}:{s.get('updated_at', 0)}:{s.get('message_count', 0)}"
-        for s in sorted(sessions, key=lambda x: x['session_id'])
-    )
-    return hashlib.md5(key.encode(), usedforsecurity=False).hexdigest()
+    """Hash the complete published session payload for change detection.
+
+    Every emitted field participates, not only the id / activity timestamp /
+    message count triple. Projection authority (compression collapse,
+    ``model_config`` lineage markers, title promotion) can change a row's
+    ``title``, ``created_at`` or source metadata while that triple stays fixed;
+    hashing only the triple let the projection rerun, keep ``_last_sessions``
+    stale and emit no ``sessions_changed`` event. Entries are canonicalised
+    (sorted keys, stable separators) and ordered by ``session_id`` so the hash
+    is deterministic and independent of sidebar ordering; cost is bounded by
+    the projection's own row limit.
+    """
+    digest = hashlib.md5(usedforsecurity=False)
+    for session in sorted(sessions, key=lambda x: str(x.get('session_id') or '')):
+        digest.update(
+            json.dumps(
+                session, sort_keys=True, separators=(',', ':'), default=str
+            ).encode('utf-8', 'replace')
+        )
+        digest.update(b'\x1e')
+    return digest.hexdigest()
 
 
 # Sources excluded from the WebUI sidebar projection. Must match the default
@@ -70,10 +86,10 @@ def _cheap_change_fingerprint(db_path: Path) -> str | None:
     # are always present (``source`` is required for the projection to run at
     # all); the rest are optional on older agent schemas and filtered below.
     _PROJECTION_SESSION_COLS = (
-        'id', 'source', 'session_source', 'title', 'model', 'message_count',
-        'started_at', 'ended_at', 'end_reason', 'parent_session_id', 'archived',
-        'user_id', 'chat_id', 'chat_type', 'thread_id', 'session_key',
-        'origin_chat_id', 'origin_user_id', 'platform',
+        'id', 'source', 'session_source', 'model_config', 'title', 'model',
+        'message_count', 'started_at', 'ended_at', 'end_reason',
+        'parent_session_id', 'archived', 'user_id', 'chat_id', 'chat_type',
+        'thread_id', 'session_key', 'origin_chat_id', 'origin_user_id', 'platform',
     )
     try:
         with closing(open_state_db_readonly(db_path)) as conn:

api/helpers.py

@@ -24,6 +24,8 @@ _PUBLIC_MESSAGE_INTERNAL_FIELDS = frozenset({
     "_active_turn_token",
     "_active_turn_user",
     "_fork_child_turn",
+    "_webui_trusted_agent_input_text",
+    "_webui_unmatched_native_image_mirror",
 })
 
 

api/mcp_runtime.py

@@ -0,0 +1,325 @@
+"""Profile-bound view of Hermes Agent's in-process MCP runtime.
+
+The MCP status page, the tool inventory and ``/reload-mcp`` read or reset
+Hermes Agent's MCP ledger from the HTTP request thread. Hermes Agent scopes
+that ledger per profile (connection key ``(profile_home_key, name)``) only
+when the calling task serves a *routed* profile, i.e. its context-local
+Hermes-home override differs from the process home. These helpers bind the
+request profile's home (without touching ``os.environ``) and check that the
+agent's routing decision matches the profile WebUI resolved, so a read never
+shows another profile's connection and a reload never resets one.
+
+Ledger helpers below take ``core`` (``tools.mcp_tool``) explicitly and read
+its maps with defaults, so an agent that lacks a map degrades to "unscoped"
+rather than raising inside a status read.
+"""
+
+from __future__ import annotations
+
+import importlib
+import inspect
+import logging
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Generator, Iterable, Optional
+
+logger = logging.getLogger(__name__)
+
+# Scope label reported by the MCP endpoints.
+SCOPE_PROFILE = "profile"            # runtime view bound to the request profile
+SCOPE_LEGACY = "legacy_process"      # agent predates profile-scoped MCP; process-wide view
+SCOPE_UNAVAILABLE = "unavailable"    # profile scope could not be confirmed; runtime hidden
+
+
+@dataclass(frozen=True)
+class McpRuntimeView:
+    """How the current context sees Hermes Agent's MCP runtime.
+
+    ``trusted``: runtime reads/writes in this context belong to ``profile_home``.
+    ``legacy``: the agent has no profile-scoped MCP ledger (process-wide by name).
+    ``registry_scope``: the tool-registry slot owning this profile's MCP tools
+    (the profile overlay key when routed, ``None`` for the process profile).
+    """
+
+    profile_home: Path
+    trusted: bool
+    legacy: bool
+    registry_scope: Optional[str] = None
+
+    @property
+    def scope_label(self) -> str:
+        if not self.trusted:
+            return SCOPE_UNAVAILABLE
+        return SCOPE_LEGACY if self.legacy else SCOPE_PROFILE
+
+
+def _routing_view(profile_home: Path, override_bound: bool) -> McpRuntimeView:
+    try:
+        from agent.secret_scope import serves_routed_profile
+        from hermes_constants import hermes_home_key
+        from tools.registry import registry
+    except ImportError:
+        return McpRuntimeView(profile_home, trusted=True, legacy=True)
+    from api.profiles import get_process_profile_home
+
+    try:
+        expected_routed = hermes_home_key(profile_home) != hermes_home_key(get_process_profile_home())
+        # Without the override a routed profile cannot be served; with it, the agent must
+        # agree. It disagrees when a streaming turn has mirrored this profile's home into
+        # os.environ['HERMES_HOME'] on an agent without a pinned process home.
+        if expected_routed and not override_bound:
+            return McpRuntimeView(profile_home, trusted=False, legacy=False)
+        if bool(serves_routed_profile()) != expected_routed:
+            return McpRuntimeView(profile_home, trusted=False, legacy=False)
+        scope = registry.current_scope_key() if expected_routed else None
+    except Exception:
+        logger.debug("Failed to resolve MCP runtime scope for %s", profile_home, exc_info=True)
+        return McpRuntimeView(profile_home, trusted=False, legacy=False)
+    return McpRuntimeView(profile_home, trusted=True, legacy=False, registry_scope=scope)
+
+
+@contextmanager
+def mcp_runtime_scope(purpose: str) -> Generator[McpRuntimeView, None, None]:
+    """Bind the active request profile for Hermes Agent MCP runtime calls.
+
+    Installs the context-local Hermes-home override and secret scope for the
+    request profile (including the root profile) and restores them on exit,
+    including on exceptions. Never mutates ``os.environ``, starts, or probes
+    MCP servers.
+    """
+    from api.profiles import get_active_hermes_home, profile_env_for_active_request_readonly
+
+    profile_home = Path(get_active_hermes_home())
+    with profile_env_for_active_request_readonly(
+        purpose, logger_override=logger, include_root=True
+    ) as override_bound:
+        yield _routing_view(profile_home, bool(override_bound))
+
+
+def accepts_keywords(fn: Callable, *names: str) -> bool:
+    """True when *fn*'s signature exposes every keyword in *names*.
+
+    Introspection failures (builtins, C callables, exotic wrappers) count as
+    "not supported" so callers fall back to the bare call instead of dropping
+    the whole read.
+    """
+    try:
+        params = inspect.signature(fn).parameters
+    except (TypeError, ValueError):
+        return False
+    return all(name in params for name in names)
+
+
+# ── Ledger keys ──────────────────────────────────────────────────────────────
+# Hermes Agent's own helpers (``tools.mcp_tool_scope``) are preferred so the key
+# shape has one owner; the lambdas only cover agents that predate that module.
+
+def _key_helper(name: str, fallback: Callable) -> Callable:
+    try:
+        # Full-name import: works with a stubbed module and no ``tools`` package.
+        mcp_tool_scope = importlib.import_module("tools.mcp_tool_scope")
+    except Exception:
+        return fallback
+    helper = getattr(mcp_tool_scope, name, None)
+    return helper if callable(helper) else fallback
+
+
+def mcp_key_name(key: Any) -> str:
+    """Server name of an MCP connection-ledger key (bare name or ``(scope, name)``)."""
+    return str(_key_helper("_key_name", lambda k: k[1] if isinstance(k, tuple) else k)(key))
+
+
+def mcp_key_scope(key: Any) -> Optional[str]:
+    """Owning registry scope encoded in a ledger key (``None`` for a bare key)."""
+    return _key_helper("_key_scope", lambda k: k[0] if isinstance(k, tuple) else None)(key)
+
+
+def ledger_key_owner(core: Any, key: Any) -> Optional[str]:
+    """Scope owning the connection under *key*: recorded at adoption, else the key's own."""
+    owners = getattr(core, "_server_scope_keys", None)
+    if isinstance(owners, dict) and key in owners:
+        return owners[key]
+    return mcp_key_scope(key)
+
+
+def ledger_key_serves_view(core: Any, key: Any, view: McpRuntimeView) -> bool:
+    """True when the connection under *key* serves the profile in *view*.
+
+    Owned by the view's registry scope, or adopted into it (a routed profile
+    sharing an identical connection; ``_server_tool_scopes``). Hermes Agent's
+    own launch-profile view (scope ``None``) is process-wide, so this predicate
+    is what narrows the root profile to its unscoped connections. Legacy agents
+    have one process-wide ledger, so every key counts.
+    """
+    if view.legacy:
+        return True
+    if ledger_key_owner(core, key) == view.registry_scope:
+        return True
+    if view.registry_scope is None:
+        return False
+    adopters = getattr(core, "_server_tool_scopes", None)
+    if not isinstance(adopters, dict):
+        return False
+    return view.registry_scope in (adopters.get(key) or ())
+
+
+@dataclass(frozen=True)
+class ServedRuntime:
+    """Runtime state of one server name as seen from a profile's ledger keys."""
+
+    live_tools: Optional[int] = None   # tool count of a live (session-bearing) connection
+    connecting: bool = False
+    error: Optional[str] = None
+    lazy_tools: Optional[int] = None
+
+    @property
+    def live(self) -> bool:
+        return self.live_tools is not None
+
+
+def _served_keys(core: Any, ledger: str, view: McpRuntimeView) -> list:
+    keys = getattr(core, ledger, None)
+    if not isinstance(keys, (dict, set, frozenset, list, tuple)):
+        return []
+    return [key for key in list(keys) if ledger_key_serves_view(core, key, view)]
+
+
+def view_served_runtime(core: Any, view: McpRuntimeView) -> dict[str, ServedRuntime]:
+    """Per-name runtime state of the ledger entries serving *view*.
+
+    Caller holds ``core._lock`` or tolerates a racy read (status surfaces).
+    """
+    live: dict[str, int] = {}
+    connecting: set[str] = set()
+    errors: dict[str, str] = {}
+    lazy: dict[str, int] = {}
+    servers = getattr(core, "_servers", None)
+    if isinstance(servers, dict):
+        for key in _served_keys(core, "_servers", view):
+            server = servers.get(key)
+            if server is None or getattr(server, "session", None) is None:
+                continue  # parked / mid-connect task: retained, not connected
+            names = getattr(server, "_registered_tool_names", None)
+            if not isinstance(names, (list, tuple, set)):
+                names = getattr(server, "_tools", None) or ()
+            live[mcp_key_name(key)] = len(names)
+    for key in _served_keys(core, "_server_connecting", view):
+        connecting.add(mcp_key_name(key))
+    error_map = getattr(core, "_server_connect_errors", None)
+    if isinstance(error_map, dict):
+        for key in _served_keys(core, "_server_connect_errors", view):
+            errors[mcp_key_name(key)] = str(error_map.get(key, ""))
+    lazy_tools = getattr(core, "_lazy_server_tool_names", None)
+    for key in _served_keys(core, "_lazy_server_configs", view):
+        count = lazy_tools.get(key) if isinstance(lazy_tools, dict) else None
+        lazy[mcp_key_name(key)] = len(count) if isinstance(count, (list, tuple, set)) else 0
+    out: dict[str, ServedRuntime] = {}
+    for name in set(live) | connecting | set(errors) | set(lazy):
+        out[name] = ServedRuntime(
+            live_tools=live.get(name),
+            connecting=name in connecting,
+            error=errors.get(name),
+            lazy_tools=lazy.get(name),
+        )
+    return out
+
+
+_RUNTIME_ROW_FIELDS = ("connected", "tools", "status", "error", "sampling", "tool_schemas")
+
+
+def filter_runtime_status_to_view(statuses: Iterable[Any], view: McpRuntimeView) -> list:
+    """Rebuild the runtime fields of ``get_mcp_status()`` rows from the keys serving *view*.
+
+    Under a routed scope the agent already filters by owner/adopter; under the
+    launch profile (scope ``None``) it reports every profile's connection and
+    indexes them by bare name, so the root profile would show "Active, N tools"
+    for a routed profile's server it never connected (or its own parked server
+    as connected because another profile's same-named one is live). Rows keep
+    their config-derived fields (name, transport, disabled) and get
+    ``connected`` / ``tools`` / ``status`` / ``error`` from this profile's own
+    ledger entries only.
+    """
+    rows = [row for row in statuses if isinstance(row, dict)]
+    if view.legacy or not rows:
+        return rows
+    try:
+        core = importlib.import_module("tools.mcp_tool")
+    except Exception:
+        return rows
+    lock = getattr(core, "_lock", None)
+    try:
+        if lock is not None:
+            with lock:
+                served = view_served_runtime(core, view)
+        else:
+            served = view_served_runtime(core, view)
+    except Exception:
+        logger.debug("Failed to read MCP ledger ownership for %s", view.profile_home, exc_info=True)
+        return []
+    out = []
+    for row in rows:
+        rebuilt = {k: v for k, v in row.items() if k not in _RUNTIME_ROW_FIELDS}
+        state = served.get(str(row.get("name")))
+        disabled = bool(row.get("disabled")) or row.get("status") == "disabled"
+        rebuilt["connected"] = bool(state and state.live)
+        rebuilt["tools"] = state.live_tools if state and state.live else (
+            state.lazy_tools if state and state.lazy_tools is not None and not disabled else 0
+        )
+        if state and state.live:
+            rebuilt["status"] = "connected"
+        elif disabled:
+            rebuilt["status"] = "disabled"
+        elif state and state.connecting:
+            rebuilt["status"] = "connecting"
+        elif state and state.error is not None:
+            rebuilt["status"] = "failed"
+            rebuilt["error"] = state.error
+        elif state and state.lazy_tools is not None:
+            rebuilt["status"] = "lazy"
+        else:
+            rebuilt["status"] = "configured"
+        out.append(rebuilt)
+    return out
+
+
+_COOLDOWN_LEDGERS = ("_server_connect_retry_after", "_server_connect_failures", "_server_connect_errors")
+
+
+def clear_profile_connect_cooldowns(core: Any, view: McpRuntimeView) -> None:
+    """Forget connect backoff/error state of the servers *view*'s profile owns.
+
+    A server that failed to spawn never reaches ``_servers``, so a scoped
+    ``shutdown_mcp_servers(scope=..., names=...)`` (which clears cooldowns for
+    the live keys it tears down) leaves its backoff in place and the following
+    ``discover_mcp_tools()`` skips it. The process-wide wildcard cleared every
+    cooldown; this keeps ``/reload-mcp`` a retry for the profile's own servers
+    without touching another owner's backoff. Owned keys only, never adopted ones.
+    """
+    lock = getattr(core, "_lock", None)
+    if lock is None:
+        return
+    with lock:
+        for ledger in _COOLDOWN_LEDGERS:
+            entries = getattr(core, ledger, None)
+            if not isinstance(entries, dict):
+                continue
+            for key in [k for k in entries if ledger_key_owner(core, k) == view.registry_scope]:
+                entries.pop(key, None)
+
+
+def registry_tool_owned_by_view(registry, tool_name: str, view: McpRuntimeView) -> bool:
+    """True when *tool_name* is registered in this profile's own registry slot.
+
+    The merged registry view overlays a profile's tools on the global slot, so a
+    routed profile would otherwise also list the process profile's MCP tools.
+    """
+    if view.legacy:
+        return True
+    snapshot = getattr(registry, "snapshot_registration", None)
+    if not callable(snapshot):
+        return True
+    try:
+        return snapshot(tool_name, scope=view.registry_scope) is not None
+    except Exception:
+        return False

api/profiles.py

@@ -56,6 +56,11 @@ _loaded_profile_env_keys: set[str] = set()
 # process-global _active_profile.
 _tls = threading.local()
 
+# Home of the profile this process serves as its own (set by init_profile_state).
+# Streaming turns mirror their profile into os.environ['HERMES_HOME'], so the live
+# env var is not a stable identity; see _pin_process_profile_home().
+_PROCESS_PROFILE_HOME: Optional[str] = None
+
 _SKILL_HOME_MODULES = ("tools.skills_tool", "tools.skill_manager_tool")
 _SKILL_HOME_MODULE_PATCH_LOCK = threading.RLock()
 
@@ -1360,6 +1365,8 @@ def profile_env_for_background_worker(
 def profile_env_for_active_request_readonly(
     purpose: str = "provider/model read",
     logger_override: Optional[logging.Logger] = None,
+    *,
+    include_root: bool = False,
 ):
     """Apply the active per-request profile's env to thread-local state only (#3957).
 
@@ -1376,11 +1383,18 @@ def profile_env_for_active_request_readonly(
     process-global ``os.environ``.
 
     No-ops for the default/root profile, which is the common single-profile
-    deployment case.
+    deployment case, unless ``include_root`` is set: callers whose Hermes Agent
+    reads must not follow a streaming turn's mirrored ``HERMES_HOME`` (MCP
+    runtime status/reload) bind the root profile's home explicitly too.
+
+    Yields True when the context-local Hermes-home override is installed for
+    the resolved profile, else False (no-op, older agent, or resolution error).
     """
     profile = (get_active_profile_name() or "").strip()
-    if not profile or _is_root_profile(profile):
-        yield
+    if include_root and not profile:
+        profile = "default"
+    if not profile or (_is_root_profile(profile) and not include_root):
+        yield False
         return
     try:
         from api.config import _clear_thread_env, _set_thread_env, _thread_ctx
@@ -1396,7 +1410,7 @@ def profile_env_for_active_request_readonly(
             purpose,
             exc_info=True,
         )
-        yield
+        yield False
         return
     try:
         from hermes_constants import (
@@ -1440,7 +1454,7 @@ def profile_env_for_active_request_readonly(
                     purpose,
                     exc_info=True,
                 )
-        yield
+        yield home_override_installed
     finally:
         if _has_scope and _secret_scope_mod is not None:
             try:
@@ -1529,8 +1543,14 @@ def profile_scope_for_detached_worker(
 
 
 def _set_hermes_home(home: Path):
-    """Set HERMES_HOME env var and monkey-patch cached module-level paths."""
+    """Set HERMES_HOME env var and monkey-patch cached module-level paths.
+
+    Every process-wide home change (startup, ``switch_profile(process_wide=True)``)
+    goes through here, so the process-profile pin used for MCP routing decisions
+    is updated in the same step and cannot drift from ``HERMES_HOME``.
+    """
     os.environ['HERMES_HOME'] = str(home)
+    _pin_process_profile_home(home)
 
     patch_skill_home_modules(home)
 
@@ -1610,11 +1630,45 @@ def init_profile_state() -> None:
     else:
         _active_profile = _read_active_profile_file()
         home = get_active_hermes_home()
-    _set_hermes_home(home)
+    _set_hermes_home(home)  # also pins the process-profile home (MCP routing anchor)
     install_cron_scheduler_profile_isolation()
     _reload_dotenv(home)
 
 
+def _pin_process_profile_home(home: Path) -> None:
+    """Record the profile home this process serves as its own, for WebUI and Hermes Agent.
+
+    Called from ``_set_hermes_home()`` so startup and process-wide profile switches
+    keep one owner for the value. Hermes Agent keys MCP connections and registry
+    overlays by profile only when a task serves a *routed* profile: the context-local
+    Hermes-home override differs from the process home. Streaming turns mirror their
+    profile into ``os.environ['HERMES_HOME']`` for legacy readers, which makes every
+    turn's own profile look like the process profile, so same-named MCP servers of
+    different profiles share one bare-name connection. Agents exposing
+    ``hermes_constants.pin_process_hermes_home`` take a stable anchor instead; older
+    agents keep following the env var, and ``get_process_profile_home()`` still gives
+    WebUI the anchor to detect that skew.
+    """
+    global _PROCESS_PROFILE_HOME
+    _PROCESS_PROFILE_HOME = str(home)
+    try:
+        import hermes_constants
+        pin = getattr(hermes_constants, 'pin_process_hermes_home', None)
+        if callable(pin):
+            pin(str(home))
+    except Exception:
+        logger.debug("Hermes Agent process-home pin unavailable", exc_info=True)
+
+
+def get_process_profile_home() -> Path:
+    """Return the profile home this WebUI process serves as its own (stable per process)."""
+    if _PROCESS_PROFILE_HOME:
+        return Path(_PROCESS_PROFILE_HOME)
+    if _INITIAL_HERMES_HOME:
+        return Path(_INITIAL_HERMES_HOME).expanduser()
+    return _DEFAULT_HERMES_HOME
+
+
 def switch_profile(name: str, *, process_wide: bool = True) -> dict:
     """Switch the active profile.
 

api/route_session_list_cache.py

@@ -606,6 +606,22 @@ def _session_list_cache_source_stamp(key: tuple) -> tuple[tuple[int, int], tuple
         session_index_path = _session_list_cache_session_dir() / "_index.json"
     except Exception:
         session_index_path = None
+    all_profiles_state_stamp = None
+    if _cache_all_profiles:
+        try:
+            from api.models import _all_profiles_cli_contexts
+            contexts, _ = _all_profiles_cli_contexts()
+            all_profiles_state_stamp = tuple(
+                (
+                    _session_list_cache_path_stamp(ctx_db),
+                    _session_list_cache_path_stamp(ctx_db.with_name(f"{ctx_db.name}-wal") if ctx_db else None),
+                    _session_list_cache_state_db_fingerprint(ctx_db),
+                )
+                for _home, ctx_db, _prof in contexts
+            )
+        except Exception:
+            all_profiles_state_stamp = None
+
     return (
         _session_list_cache_path_stamp(state_db_path),
         _session_list_cache_path_stamp(state_db_wal_path),
@@ -617,6 +633,7 @@ def _session_list_cache_source_stamp(key: tuple) -> tuple[tuple[int, int], tuple
         # frame size), so without this a freshly-committed CLI/gateway session
         # could be served stale for the cache TTL. Mirrors the models-layer fix.
         _session_list_cache_state_db_fingerprint(state_db_path),
+        all_profiles_state_stamp,
         swv,
     )
 

api/routes.py

@@ -4067,6 +4067,7 @@ def _run_journal_live_snapshot(stream_id: str | None, *, handler=None) -> dict |
         "tool_calls": tool_calls,
         "last_assistant_text": assistant_text,
         "last_reasoning_text": reasoning_text,
+        "runtime_model": runtime_model_from_events(session_id, stream_id, events),
         "activity_burst_anchors": activity_burst_anchors,
         "current_activity_burst_id": current_activity_burst_id,
         "current_live_segment_seq": current_live_segment_seq,
@@ -7183,7 +7184,35 @@ def _context_length_lookup_inputs_for_model(
         if not effective_provider:
             effective_provider = _canonical_context_provider(model_cfg.get("provider"))
         if not effective_base_url:
-            effective_base_url = str(model_cfg.get("base_url") or "").strip()
+            # #7535: the global model.base_url may only fill an empty slot when
+            # the session provider IS the configured model.provider owner
+            # (mirror the ownership predicate used for model_cfg's API key in
+            # _context_length_config_api_key_for_provider). A built-in registry
+            # provider (empty base_url by design) must keep the slot empty so
+            # the registry endpoint resolves instead of another provider's URL.
+            #
+            # Two shapes cannot own the slot and therefore cannot conflict, so
+            # they keep master's backfill: a config that declares no provider
+            # at all (the profile-setup path writes model.base_url without one)
+            # and the two spellings of the same built-in id (opencode_go ==
+            # opencode-go), folded through api.config._canonicalise_provider_id
+            # so distinct custom:* slugs stay distinct.
+            _model_cfg_provider = _canonical_context_provider(model_cfg.get("provider"))
+            _owner_provider = _model_cfg_provider
+            _session_provider = effective_provider
+            try:
+                from api.config import _canonicalise_provider_id as _canon_provider_id
+
+                _owner_provider = _canon_provider_id(_owner_provider) or _owner_provider
+                _session_provider = _canon_provider_id(_session_provider) or _session_provider
+            except Exception:
+                pass
+            if (
+                not effective_provider
+                or not _model_cfg_provider
+                or _providers_match_for_context(_owner_provider, _session_provider)
+            ):
+                effective_base_url = str(model_cfg.get("base_url") or "").strip()
 
     custom_providers = cfg.get("custom_providers") if isinstance(cfg, dict) else None
     if not isinstance(custom_providers, list):
@@ -9103,9 +9132,11 @@ def _parse_msg_limit(raw):
     """Parse and clamp the ``?msg_limit=`` query value.
 
     Returns a positive int clamped to ``[1, _MAX_MSG_LIMIT]``, or ``None`` when
-    the value is absent/empty/malformed (the bare no-``msg_limit`` path, which
-    intentionally returns the full transcript for callers that need it).
-    Extracted from the handler so the clamp expression has direct test coverage.
+    the value is absent/empty/malformed.  ``?msg_limit=all`` also returns
+    ``None`` — an explicit escape hatch for the full-transcript paths the
+    frontend genuinely needs; the handler distinguishes it from a bare request
+    via :func:`_resolve_effective_msg_limit`.  Extracted from the handler so
+    the clamp expression has direct test coverage.
     """
     if not raw:
         return None
@@ -9116,6 +9147,26 @@ def _parse_msg_limit(raw):
     return max(1, min(value, _MAX_MSG_LIMIT))
 
 
+def _resolve_effective_msg_limit(raw_limit):
+    """Resolve the effective ``msg_limit`` for ``GET /api/session``.
+
+    Returns ``(effective_limit, explicit_all)``.
+
+    - numeric ``?msg_limit=N`` → clamped int (existing pagination).
+    - ``?msg_limit=all`` → ``(None, True)``: explicit full-transcript escape
+      hatch.  Frontend paths that address rows by absolute transcript index
+      (outline jump, jump-to-start) genuinely need everything; they pass
+      ``all`` instead of relying on the bare no-limit shape.
+    - any other bare shape (no limit) → ``(None, False)``: the historical
+      full-transcript contract is preserved (contract tests pin tool-row
+      preservation and the runtime-journal snapshot on this shape), so the
+      bounded-window fix is enforced at the frontend call sites instead.
+    """
+    explicit_all = str(raw_limit or "").strip().lower() == "all"
+    limit = _parse_msg_limit(raw_limit)
+    return limit, explicit_all
+
+
 # If a sidecar JSON file exceeds this threshold, the display-path tail
 # optimization fires regardless of message count.  Sessions with few messages
 # but large tool outputs (multi-MB JSON) should not force a full-scan merge.
@@ -9298,6 +9349,13 @@ def _limited_webui_messages_for_display_with_sidecar(
     state_db_messages = list(state_db_messages or [])
     if not state_db_messages:
         return sidecar_messages
+    state_db_messages = _suppress_native_image_display_mirrors(
+        session,
+        state_db_messages,
+    )
+    if not state_db_messages:
+        return sidecar_messages
+
     # NOTE: do not short-circuit to the sidecar when state.db has no strictly
     # newer rows. A state.db row whose timestamp is at-or-before the sidecar's
     # newest (recovery / edited-in-place / missing-timestamp cases) is still
@@ -9358,6 +9416,11 @@ def _limited_webui_messages_for_display_with_sidecar(
         truncation_boundary=getattr(session, "truncation_boundary", None),
         incoming_provenance="state_db",
     )
+    merged = _project_native_image_payload_conflicts_for_display(
+        sidecar_messages,
+        state_db_messages,
+        merged,
+    )
     if cache_key is not None:
         _state_key = cache_key[4]
         _streaming_key = (
@@ -10053,7 +10116,35 @@ def _merged_session_messages_for_display(session, cli_messages=None) -> list:
 
 
 
-def _merged_webui_lineage_messages_for_display(session, messages=None) -> list:
+_LINEAGE_PARENT_SESSION_UNSET = object()
+
+
+def _webui_lineage_parent_session_for_display(session):
+    """Load the immediate parent only for display-eligible continuations."""
+    parent_id = str(getattr(session, "parent_session_id", "") or "").strip()
+    if not parent_id:
+        return None
+    if (
+        str(getattr(session, "compression_recovery_source_session_id", "") or "").strip()
+        and str(getattr(session, "compression_recovery_action", "") or "").strip()
+    ):
+        return None
+    source = str(getattr(session, "session_source", "") or "").strip().lower()
+    relationship = str(getattr(session, "relationship_type", "") or "").strip().lower()
+    if source == "fork" or relationship == "child_session":
+        return None
+    try:
+        return get_session(parent_id, metadata_only=False)
+    except Exception:
+        return None
+
+
+def _merged_webui_lineage_messages_for_display(
+    session,
+    messages=None,
+    *,
+    parent_session=_LINEAGE_PARENT_SESSION_UNSET,
+) -> list:
     """Include immediate parent-only rows when a WebUI continuation sidecar is partial.
 
     Compression/continuation sessions should render as one conversation. Most
@@ -10064,23 +10155,9 @@ def _merged_webui_lineage_messages_for_display(session, messages=None) -> list:
     subset of their parent.
     """
     primary_messages = list(messages if messages is not None else (getattr(session, "messages", []) or []))
-    parent_id = str(getattr(session, "parent_session_id", "") or "").strip()
-    if not parent_id:
-        return primary_messages
-    if (
-        str(getattr(session, "compression_recovery_source_session_id", "") or "").strip()
-        and str(getattr(session, "compression_recovery_action", "") or "").strip()
-    ):
-        return primary_messages
-    source = str(getattr(session, "session_source", "") or "").strip().lower()
-    relationship = str(getattr(session, "relationship_type", "") or "").strip().lower()
-    if source == "fork" or relationship == "child_session":
-        return primary_messages
-    try:
-        parent = get_session(parent_id, metadata_only=False)
-    except Exception:
-        return primary_messages
-    parent_messages = list(getattr(parent, "messages", []) or [])
+    if parent_session is _LINEAGE_PARENT_SESSION_UNSET:
+        parent_session = _webui_lineage_parent_session_for_display(session)
+    parent_messages = list(getattr(parent_session, "messages", []) or [])
     if not parent_messages:
         return primary_messages
     if _messages_start_with_visible_prefix(primary_messages, parent_messages):
@@ -10560,7 +10637,6 @@ from api.models import (
     new_session,
     all_sessions,
     title_from,
-    _write_session_index,
     SESSION_INDEX_FILE,
     _active_state_db_path,
     load_projects,
@@ -10574,6 +10650,8 @@ from api.models import (
     get_state_db_session_message_keys_before_timestamp,
     get_state_db_session_summary,
     merge_session_messages_append_only,
+    _project_native_image_payload_conflicts_for_display,
+    _suppress_native_image_display_mirrors,
     _reconcile_api_content_sidecars,
     _enrich_sidebar_lineage_metadata,
     _active_stream_ids,
@@ -10839,6 +10917,7 @@ from api.streaming import (
 )
 from api.gateway_chat import _run_gateway_chat_streaming, webui_gateway_chat_enabled
 from api.run_journal import (
+    runtime_model_from_events,
     _parse_run_journal_event_id as _shared_parse_run_journal_event_id,
     _summary_from_events,
     bound_run_journal_snapshot_args,
@@ -13151,8 +13230,10 @@ def _handle_session_get(handler, parsed) -> bool:
     # payload; the existing _messages_truncated signal covers the clamped
     # case (the client sees there are more rows than returned). Parsing +
     # clamping live in _parse_msg_limit so the expression has direct test
-    # coverage; None means the bare no-msg_limit path (full transcript).
-    msg_limit = _parse_msg_limit(query.get("msg_limit", [None])[0])
+    # coverage.  The frontend recovery paths request a bounded tail
+    # explicitly (msg_limit=30), and the two absolute-index paths opt in to
+    # the full transcript via msg_limit=all (#7310/#7625).
+    _raw_msg_limit = query.get("msg_limit", [None])[0]
     # ?msg_before=N — 0-based index into the full message array.
     # Returns messages before this index (for scroll-to-top lazy loading).
     # Combined with msg_limit for paging.
@@ -13161,6 +13242,9 @@ def _handle_session_get(handler, parsed) -> bool:
         msg_before = int(_msg_before) if _msg_before else None
     except (ValueError, TypeError):
         msg_before = None
+    msg_limit, _msg_limit_explicit_all = _resolve_effective_msg_limit(
+        _raw_msg_limit,
+    )
     # ?expand_renderable=1 is retained for compatibility with older
     # frontends. msg_limit now counts visible transcript rows by default, so
     # the flag no longer changes the server-side pagination semantics.
@@ -13314,13 +13398,33 @@ def _handle_session_get(handler, parsed) -> bool:
                         msg_before=msg_before,
                     )
             else:
+                state_db_messages = _suppress_native_image_display_mirrors(
+                    s,
+                    state_db_messages,
+                )
+                sidecar_messages = _webui_sidecar_lineage_messages_for_display(s)
+                lineage_parent = _webui_lineage_parent_session_for_display(s)
+                projection_sidecar_messages = _merged_webui_lineage_messages_for_display(
+                    s,
+                    sidecar_messages,
+                    parent_session=lineage_parent,
+                )
                 _all_msgs = merge_session_messages_append_only(
-                    _webui_sidecar_lineage_messages_for_display(s),
+                    sidecar_messages,
                     state_db_messages,
                     truncation_watermark=getattr(s, "truncation_watermark", None),
                     truncation_boundary=getattr(s, "truncation_boundary", None),
                 )
-                _all_msgs = _merged_webui_lineage_messages_for_display(s, _all_msgs)
+                _all_msgs = _merged_webui_lineage_messages_for_display(
+                    s,
+                    _all_msgs,
+                    parent_session=lineage_parent,
+                )
+                _all_msgs = _project_native_image_payload_conflicts_for_display(
+                    projection_sidecar_messages,
+                    state_db_messages,
+                    _all_msgs,
+                )
         else:
             if is_messaging_session and cli_messages:
                 _all_msgs = _merged_session_messages_for_display(s, cli_messages)
@@ -13370,9 +13474,13 @@ def _handle_session_get(handler, parsed) -> bool:
             _messages_offset = 0
         # Index of the first returned message in the full message array.
         # Frontend uses this as cursor for scroll-to-top paging.
+        # Session-level tool_calls windowing keys off whether the returned
+        # message array was actually truncated (msg_before paging, any
+        # effective msg_limit) rather than whether a limit parameter was
+        # present — the full-transcript shape returns everything, so the
+        # length comparison alone decides (#7310/#7625).
         _windowed_messages = (
             load_messages
-            and msg_limit is not None
             and (msg_before is not None or len(_truncated_msgs) < len(_all_msgs))
         )
         # Resolve effective context_length with model-metadata fallback so
@@ -13468,7 +13576,7 @@ def _handle_session_get(handler, parsed) -> bool:
             "tool_calls": _session_tool_calls,
             "active_stream_id": getattr(s, "active_stream_id", None),
             "pending_user_message": getattr(s, "pending_user_message", None),
-            "pending_attachments": getattr(s, "pending_attachments", []) if load_messages else [],
+            "pending_attachments": getattr(s, "pending_attachments", []) if (load_messages or getattr(s, "pending_user_message", None)) else [],
             "pending_started_at": getattr(s, "pending_started_at", None),
             "pending_user_source": getattr(s, "pending_user_source", None),
             "context_length": _persisted_cl,
@@ -23106,6 +23214,7 @@ def _prepare_chat_start_session_for_stream(
     s.pending_attachments = attachments
     s.pending_started_at = started_at if started_at is not None else time.time()
     s.pending_user_source = effective_source
+    s._webui_pending_user_timestamp_identity = None
     if retained_user is not None:
         from api.process_event_utils import build_active_turn_token
 
@@ -25157,10 +25266,12 @@ def _handle_chat_sync(handler, body):
                 _active_turn_boundary,
                 _assign_stable_message_ids,
                 _dedupe_replayed_context_messages,
+                _find_active_turn_checkpoint_index,
                 _merge_display_messages_after_agent_result,
                 _resolve_active_turn_authority,
                 _restore_display_reasoning_metadata,
                 _restore_reasoning_metadata_before_boundary,
+                _settle_current_turn_boundary,
                 _sanitize_messages_for_agent,
                 _compact_session_image_parts_for_persistence,
                 _context_messages_for_new_turn,
@@ -25225,6 +25336,33 @@ def _handle_chat_sync(handler, body):
             result=result,
             agent=agent,
         )
+        if (
+            isinstance(_active_turn_identity, dict)
+            and _active_turn_identity.get("agent_turn_boundary_resolved") is True
+            and not _active_turn_identity.get("token")
+        ):
+            _active_image_index = _find_active_turn_checkpoint_index(
+                _result_messages,
+                _previous_context_messages,
+                _active_turn_identity,
+                msg,
+            )
+            _active_image_content = (
+                _result_messages[_active_image_index].get("content")
+                if _active_image_index is not None
+                else None
+            )
+            if isinstance(_active_image_content, list) and any(
+                isinstance(part, dict)
+                and part.get("type") in {"image", "image_url", "input_image"}
+                for part in _active_image_content
+            ):
+                from api.process_event_utils import build_active_turn_token
+
+                _active_turn_identity["token"] = build_active_turn_token(
+                    f"sync:{s.session_id}:{_active_turn_identity['turn_id']}",
+                    time.time(),
+                )
         _turn_boundary = _active_turn_boundary(
             _result_messages, _previous_context_messages, _active_turn_identity, msg,
         )
@@ -25243,6 +25381,14 @@ def _handle_chat_sync(handler, body):
             _next_context_messages,
             msg,
         )
+        if _active_turn_identity.get("token"):
+            _next_context_messages = _settle_current_turn_boundary(
+                _previous_context_messages,
+                _next_context_messages,
+                _active_turn_identity,
+                msg,
+                getattr(s, "pending_user_source", None) or "webui",
+            )
         s.context_messages = _next_context_messages
         s.messages = _merge_display_messages_after_agent_result(
             _previous_messages,
@@ -25252,6 +25398,9 @@ def _handle_chat_sync(handler, body):
             ),
             msg,
             source=getattr(s, "pending_user_source", None) or "webui",
+            verification_nudge_provenance={
+                "active_turn_identity": _active_turn_identity,
+            },
         )
         _compact_session_image_parts_for_persistence(s)
         # Only auto-generate title when still default; preserves user renames
@@ -29285,17 +29434,34 @@ def _parse_mcp_enabled(value) -> bool:
     return True
 
 
-def _mcp_runtime_status_by_name() -> dict[str, dict]:
+def _mcp_runtime_status_by_name(servers=None, view=None) -> dict[str, dict]:
     """Return already-known MCP runtime status without starting servers.
 
     ``tools.mcp_tool.get_mcp_status()`` only reads the existing MCP registry and
     configuration; it does not probe or spawn MCP subprocesses. If Hermes Agent
     is unavailable, fall back to an empty map so the API remains safe.
+
+    Call it inside ``mcp_runtime_scope()`` and pass its ``view``: the agent
+    filters a routed profile's connections itself, but its launch-profile view
+    is process-wide, so rows are narrowed to the connections serving ``view``
+    (``api.mcp_runtime.filter_runtime_status_to_view``). ``servers`` (the
+    profile's ``mcp_servers`` WebUI displays) is passed as ``configured`` when
+    supported so both read the same config.
     """
     try:
         from api.agent_compat import agent_attr
+        from api.mcp_runtime import accepts_keywords, filter_runtime_status_to_view
         get_mcp_status = agent_attr("tools.mcp_tool", "get_mcp_status", "tools.mcp_tool_discovery")
-        statuses = get_mcp_status()
+        if isinstance(servers, dict) and accepts_keywords(get_mcp_status, "configured"):
+            # Invalid entries are summarized as invalid_config by WebUI; keep them
+            # out of the agent call so one bad entry cannot blank every status.
+            statuses = get_mcp_status(configured={
+                str(name): scfg for name, scfg in servers.items() if isinstance(scfg, dict)
+            })
+        else:
+            statuses = get_mcp_status()
+        if view is not None and isinstance(statuses, list):
+            statuses = filter_runtime_status_to_view(statuses, view)
     except Exception:
         return {}
     if not isinstance(statuses, list):
@@ -29478,10 +29644,18 @@ def _mcp_tools_from_runtime_status(runtime_by_name, server_summaries):
     return tools
 
 
-def _mcp_tools_from_registry(server_summaries):
-    """Read already-registered MCP tool schemas without probing MCP servers."""
+def _mcp_tools_from_registry(server_summaries, view=None):
+    """Read already-registered MCP tool schemas without probing MCP servers.
+
+    With a profile ``view`` (see ``api.mcp_runtime``), only tools registered in
+    that profile's own registry slot are listed. The slot is the isolation check;
+    the raw ``mcp_servers`` config is not an allowlist: the agent merges portable
+    plugin servers into the running config at runtime, and their tools are
+    registered in the same slot without a ``config.yaml`` entry.
+    """
     try:
         from tools.registry import registry
+        from api.mcp_runtime import registry_tool_owned_by_view
     except Exception:
         return []
     tools = []
@@ -29497,6 +29671,10 @@ def _mcp_tools_from_registry(server_summaries):
         if not isinstance(toolset, str) or not toolset.startswith("mcp-"):
             continue
         server_name = toolset[len("mcp-"):]
+        if view is not None and not view.legacy and not registry_tool_owned_by_view(
+            registry, tool_name, view
+        ):
+            continue
         schema = registry.get_schema(tool_name) or {}
         server_summary = server_summaries.get(server_name, {
             "name": server_name,
@@ -29508,22 +29686,45 @@ def _mcp_tools_from_registry(server_summaries):
     return tools
 
 
+def _mcp_profile_runtime_inventory(servers, purpose, *, include_tools=True):
+    """Build server summaries and tools from ONE runtime view of the request profile.
+
+    Status, tool count and inventory are all read inside the same
+    ``mcp_runtime_scope()`` so they describe the same profile's connections.
+    When that profile scope cannot be confirmed, runtime data is withheld rather
+    than showing another profile's connection. Passive: never starts or probes
+    MCP servers.
+    """
+    from api.mcp_runtime import mcp_runtime_scope
+
+    runtime = {}
+    tools = []
+    source = "none"
+    with mcp_runtime_scope(purpose) as view:
+        if view.trusted:
+            runtime = _mcp_runtime_status_by_name(servers, view)
+        server_summaries = {
+            str(name): _server_summary(str(name), scfg, runtime.get(str(name)))
+            for name, scfg in servers.items()
+        }
+        if include_tools and view.trusted:
+            tools = _mcp_tools_from_runtime_status(runtime, server_summaries)
+            source = "mcp_runtime_status"
+            if not tools:
+                tools = _mcp_tools_from_registry(server_summaries, view)
+                source = "tool_registry" if tools else "none"
+    return server_summaries, tools, source, view.scope_label
+
+
 def _handle_mcp_tools_list(handler):
     """List known MCP tools from already-available runtime inventory only."""
     cfg = get_config_for_profile_home(get_active_hermes_home())
     servers = cfg.get("mcp_servers", {})
     if not isinstance(servers, dict):
         servers = {}
-    runtime = _mcp_runtime_status_by_name()
-    server_summaries = {
-        str(name): _server_summary(str(name), scfg, runtime.get(str(name)))
-        for name, scfg in servers.items()
-    }
-    tools = _mcp_tools_from_runtime_status(runtime, server_summaries)
-    source = "mcp_runtime_status"
-    if not tools:
-        tools = _mcp_tools_from_registry(server_summaries)
-        source = "tool_registry" if tools else "none"
+    server_summaries, tools, source, runtime_scope = _mcp_profile_runtime_inventory(
+        servers, "/api/mcp/tools"
+    )
     tools.sort(key=lambda row: (row.get("server", ""), row.get("name", "")))
     unavailable_servers = [
         summary["name"] for summary in server_summaries.values()
@@ -29534,6 +29735,7 @@ def _handle_mcp_tools_list(handler):
         "total": len(tools),
         "source": source,
         "inventory_scope": "already_known_runtime_only",
+        "runtime_scope": runtime_scope,
         "unavailable_servers": unavailable_servers,
     })
 
@@ -29723,21 +29925,15 @@ def _handle_notes_sources_list(handler):
     servers = cfg.get("mcp_servers", {})
     if not isinstance(servers, dict):
         servers = {}
-    runtime = _mcp_runtime_status_by_name()
-    server_summaries = {
-        str(name): _server_summary(str(name), scfg, runtime.get(str(name)))
-        for name, scfg in servers.items()
-    }
-    tools = _mcp_tools_from_runtime_status(runtime, server_summaries)
-    source = "mcp_runtime_status"
-    if not tools:
-        tools = _mcp_tools_from_registry(server_summaries)
-        source = "tool_registry" if tools else "none"
+    server_summaries, tools, source, runtime_scope = _mcp_profile_runtime_inventory(
+        servers, "/api/notes/sources"
+    )
     return j(handler, {
         "enabled": True,
         "sources": _notes_sources_from_mcp_inventory(server_summaries, tools),
         "source": source,
         "inventory_scope": "already_known_runtime_only",
+        "runtime_scope": runtime_scope,
         "attach_supported": False,
         "automatic_recall_unchanged": True,
         "recent_ai_notes": _joplin_recent_ai_notes(limit=6),
@@ -30028,15 +30224,14 @@ def _handle_mcp_servers_list(handler):
     servers = cfg.get("mcp_servers", {})
     if not isinstance(servers, dict):
         servers = {}
-    runtime = _mcp_runtime_status_by_name()
-    result = [
-        _server_summary(name, scfg, runtime.get(str(name)))
-        for name, scfg in servers.items()
-    ]
+    server_summaries, _tools, _source, runtime_scope = _mcp_profile_runtime_inventory(
+        servers, "/api/mcp/servers", include_tools=False
+    )
     return j(handler, {
-        "servers": result,
+        "servers": list(server_summaries.values()),
         "toggle_supported": True,
         "reload_required": True,
+        "runtime_scope": runtime_scope,
     })
 
 

api/run_journal.py

@@ -10,6 +10,7 @@ import os
 import re
 import threading
 import time
+from copy import deepcopy
 from collections import OrderedDict
 from pathlib import Path
 from typing import Any, Iterable
@@ -130,7 +131,7 @@ def _get_cached_summary(path: Path) -> dict | None:
             _SUMMARY_CACHE.pop(key, None)
             return None
         _SUMMARY_CACHE.move_to_end(key)
-        return dict(summary)
+        return deepcopy(summary)
 
 
 def _cache_summary(
@@ -147,7 +148,7 @@ def _cache_summary(
         return
     key = str(path)
     with _SUMMARY_CACHE_LOCK:
-        _SUMMARY_CACHE[key] = (signature, dict(summary))
+        _SUMMARY_CACHE[key] = (signature, deepcopy(summary))
         _SUMMARY_CACHE.move_to_end(key)
         while len(_SUMMARY_CACHE) > _SUMMARY_CACHE_MAX_ENTRIES:
             _SUMMARY_CACHE.popitem(last=False)
@@ -534,6 +535,37 @@ def select_authoritative_terminal_event(events: Iterable[dict]) -> dict | None:
     )
 
 
+def runtime_model_from_events(session_id: str, stream_id: str, events: Iterable[dict]) -> dict | None:
+    """Project only observed serving identity for this journal owner.
+
+    A fallback warning or malformed later observation invalidates old evidence;
+    the configured selection and free-text status never supply serving identity.
+    """
+    observed = None
+    for event in events:
+        if not isinstance(event, dict) or event.get("session_id") != session_id or event.get("run_id") != stream_id:
+            continue
+        payload = event.get("payload")
+        if not isinstance(payload, dict):
+            if event.get("event") == "runtime_model":
+                observed = None
+            continue
+        if event.get("event") == "warning" and payload.get("type") == "fallback":
+            observed = None
+        elif event.get("event") == "runtime_model":
+            observed = None
+            if (payload.get("session_id") != session_id or payload.get("stream_id") != stream_id
+                or not isinstance(payload.get("model"), str) or not payload["model"].strip()
+                or not isinstance(payload.get("fallback_active"), bool)
+                or payload.get("phase") not in ("observed_output", "route_observed")):
+                continue
+            observed = {key: payload[key] for key in (
+                "session_id", "stream_id", "model", "fallback_active", "phase")}
+            if isinstance(payload.get("provider"), str) and payload["provider"].strip():
+                observed["provider"] = payload["provider"]
+    return observed
+
+
 def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -> dict:
     ordered = [event for event in events if isinstance(event, dict)]
     last = ordered[-1] if ordered else None
@@ -549,6 +581,7 @@ def _summary_from_events(session_id: str, run_id: str, events: Iterable[dict]) -
         "terminal": bool(terminal),
         "terminal_state": status,
         "last_event": (last or {}).get("event"),
+        "runtime_model": runtime_model_from_events(session_id, run_id, ordered),
     }
 
 

api/session_recovery.py

@@ -601,7 +601,14 @@ def _state_db_row_to_sidecar(row: dict) -> dict:
         'session_id': row.get('id'),
         'title': row.get('title') or 'Recovered WebUI Session',
         'workspace': workspace_value if isinstance(workspace_value, str) else '',
-        'message_count': row.get('message_count') if isinstance(row.get('message_count'), int) else len(messages),
+        # state.db is canonical for the recovered message rows.  Its denormalized
+        # sessions.message_count is absent from older schemas (read above as 0)
+        # and can be stale after recovery, so never copy it into a sidecar whose
+        # `messages` array we just materialized.  Session.save() uses this field
+        # to decide whether an overwrite shrinks the transcript; it must match
+        # the payload exactly or the #1558 backup / empty-snapshot guards can be
+        # bypassed.
+        'message_count': len(messages if isinstance(messages, list) else []),
         'worktree_path': row.get('worktree_path') or None,
         'worktree_branch': row.get('worktree_branch') or None,
         'worktree_repo_root': row.get('worktree_repo_root') or None,

api/streaming.py

@@ -75,7 +75,11 @@ from api.turn_journal import append_turn_journal_event_for_stream
 from api.usage import prompt_cache_hit_percent
 from api.models import (
     StateDBSessionMessagesSnapshot,
+    _WEBUI_TRUSTED_AGENT_INPUT_FIELD,
     _is_empty_partial_activity_message,
+    _message_exact_timestamp_details,
+    _message_private_identity_compatible,
+    _validated_webui_pending_user_timestamp_identity,
     _evict_sessions_over_cap,
     clear_process_wakeup_pause,
     get_state_db_session_messages,
@@ -891,6 +895,9 @@ def _is_fallback_lifecycle_message(kind: str, message: str) -> bool:
             or 'falling back' in m
             or 'fallback activated' in m
             or 'trying fallback' in m
+            or 'model fallback:' in m
+            or 'switched to fallback' in m
+            or 'primary model restored:' in m
         )
     )
 
@@ -2102,6 +2109,58 @@ def _mark_active_turn_checkpoint_in_history(messages, identity, msg_text, *, all
     return messages, True
 
 
+def _active_turn_user_text_matches(message, expected_text):
+    """Validate the Agent's declared user row against the submitted prompt."""
+    if not isinstance(message, dict) or message.get('role') != 'user' or not isinstance(expected_text, str):
+        return False
+    content = message.get('content')
+    if isinstance(content, list):
+        leading_text = []
+        for part in content:
+            if not isinstance(part, dict):
+                return False
+            part_type = str(part.get('type') or '').lower()
+            if part_type in {'image', 'image_url', 'input_image'}:
+                break
+            if part_type not in {'', 'text', 'input_text', 'output_text'}:
+                return False
+            text = part.get('text', part.get('input_text', part.get('output_text', '')))
+            if not isinstance(text, str):
+                return False
+            leading_text.append(text)
+        if not content:
+            return False
+        actual_text = '\n'.join(leading_text)
+    elif isinstance(content, str):
+        actual_text = content
+    else:
+        return False
+    if (
+        isinstance(content, str)
+        and " ".join(actual_text.split()) == " ".join(expected_text.split())
+    ):
+        return True
+    return _submitted_user_text_matches(actual_text, expected_text)
+
+
+def _submitted_user_text_matches(actual_text, expected_text):
+    """Match exact prompt text with at most its generated workspace prefix."""
+    if actual_text == expected_text:
+        return True
+    if not isinstance(actual_text, str) or not isinstance(expected_text, str):
+        return False
+    if not actual_text.endswith(expected_text):
+        return False
+    prefix = actual_text[:-len(expected_text)] if expected_text else actual_text
+    return bool(
+        prefix
+        and (
+            _WORKSPACE_PREFIX_RE.fullmatch(prefix)
+            or _LEGACY_WORKSPACE_PREFIX_RE.fullmatch(prefix)
+        )
+    )
+
+
 def _owner_projection_current_turn_row(messages, identity):
     messages = list(messages or [])
     if not isinstance(identity, dict):
@@ -2138,16 +2197,19 @@ def _find_active_turn_checkpoint_index(result_messages, previous_context, identi
                 return idx
     if not _active_turn_boundary_is_valid(identity):
         return None
-    expected_text = identity.get('text') if identity.get('text') is not None else msg_text
     idx = identity['current_turn_user_idx']
     if idx < 0 or idx >= len(result_messages):
         return None
     message = result_messages[idx]
+    expected_text = identity.get('text') if identity.get('text') is not None else msg_text
+    if _active_turn_user_text_matches(message, expected_text):
+        return idx
+    trusted_agent_input = identity.get('trusted_agent_input_text')
     if (
-        isinstance(message, dict)
-        and message.get('role') == 'user'
-        and _normalize_user_text(_message_text(message.get('content')))
-        == _normalize_user_text(expected_text)
+        identity.get('text') == msg_text
+        and isinstance(trusted_agent_input, str)
+        and trusted_agent_input != expected_text
+        and _active_turn_user_text_matches(message, trusted_agent_input)
     ):
         return idx
     return None
@@ -2236,17 +2298,19 @@ def _settle_current_turn_boundary(previous_context, result_messages, identity, m
     )
     if _checkpoint_idx is not None:
         existing_checkpoint = result_messages[_checkpoint_idx]
-        if isinstance(identity.get('checkpoint'), dict):
-            retained_checkpoint = _materialize_active_turn_user(identity, msg_text, source)
-            if (
-                retained_checkpoint.get('id') is None
-                and isinstance(existing_checkpoint, dict)
-                and existing_checkpoint.get('id') is not None
-            ):
-                retained_checkpoint['id'] = existing_checkpoint['id']
-            result_messages[_checkpoint_idx] = retained_checkpoint
-        else:
-            _mark_active_turn_checkpoint(existing_checkpoint, identity)
+        _mark_active_turn_checkpoint(existing_checkpoint, identity)
+        checkpoint = identity.get('checkpoint')
+        if isinstance(checkpoint, dict):
+            for key in ('id', 'timestamp'):
+                if existing_checkpoint.get(key) is None and checkpoint.get(key) is not None:
+                    existing_checkpoint[key] = copy.deepcopy(checkpoint[key])
+            if checkpoint.get('attachments'):
+                existing_checkpoint['attachments'] = copy.deepcopy(checkpoint['attachments'])
+            stamp_message_source(
+                existing_checkpoint,
+                identity.get('source') or source or 'webui',
+                active_turn_token=identity.get('token'),
+            )
         return result_messages
     previous_context = list(previous_context or [])
     if _messages_have_prefix(result_messages, previous_context):
@@ -2316,6 +2380,14 @@ def _prepare_marker_clean_writeback(
     cleaned, has_verification_nudge = _clean_synthetic_control_messages_with_provenance(
         result_messages
     )
+    # Same internal-control class, second home: a consumed mid-turn /steer is
+    # appended to the turn's last tool result wrapped in
+    # [OUT-OF-BAND USER MESSAGE ...] ... [/OUT-OF-BAND USER MESSAGE]. Strip it
+    # here, on the rows both writebacks are built from, so neither
+    # session.messages (rendered verbatim) nor session.context_messages keeps
+    # the raw wrapper. Stripping the incoming rows too keeps them identity-equal
+    # to the marker-free rows persisted by earlier turns. (#7600)
+    cleaned = _strip_oob_markers_from_messages(cleaned)
     provenance = {
         'verification_nudge_seen': has_verification_nudge,
         'active_turn_identity': copy.deepcopy(active_turn_identity),
@@ -2428,6 +2500,50 @@ def _settle_result_messages(
         source=source,
         verification_nudge_provenance=verification_nudge_provenance,
     )
+    trusted_agent_input = (
+        active_turn_identity.get('trusted_agent_input_text')
+        if isinstance(active_turn_identity, dict)
+        else None
+    )
+    if (
+        isinstance(trusted_agent_input, str)
+        and active_turn_identity.get('text') == msg_text
+        and active_turn_identity.get('token')
+    ):
+        current_display_rows = [
+            message for message in session.messages
+            if _active_turn_token_matches(message, active_turn_identity)
+        ]
+        current_context_rows = [
+            message for message in session.context_messages
+            if _active_turn_token_matches(message, active_turn_identity)
+        ]
+        if (
+            len(current_display_rows) == 1
+            and current_display_rows[0].get('content') == msg_text
+            and len(current_context_rows) == 1
+        ):
+            context_user = current_context_rows[0]
+            context_content = context_user.get('content')
+            if (
+                isinstance(context_content, list)
+                and any(
+                    isinstance(part, dict)
+                    and str(part.get('type') or '').lower()
+                    in {'image', 'image_url', 'input_image'}
+                    for part in context_content
+                )
+                and _active_turn_user_text_matches(
+                    context_user,
+                    trusted_agent_input,
+                )
+            ):
+                context_user[_WEBUI_TRUSTED_AGENT_INPUT_FIELD] = trusted_agent_input
+    # The merge carries earlier display rows across turns verbatim, so a row
+    # settled before this guard existed would keep its raw wrapper forever.
+    # Scrub the persisted display copy too — after the merge, so identity
+    # matching above still saw the rows unchanged. (#7600)
+    session.messages = _strip_oob_markers_from_messages(session.messages)
     _annotate_media_snapshots_for_settled_messages(session.messages)
     _compact_session_image_parts_for_persistence(session)
     _advance_truncation_watermark_after_commit(session)  # #3831
@@ -2455,6 +2571,61 @@ def _current_turn_already_has_visible_assistant_answer(messages, *, active_turn_
     return False
 
 
+def _trusted_native_image_context_display_mirror_tokens(previous_display, previous_context):
+    """Find unique token-paired rich image rows already represented in display."""
+    display_by_token = {}
+    context_by_token = {}
+    for messages, rows_by_token in (
+        (previous_display, display_by_token),
+        (previous_context, context_by_token),
+    ):
+        for message in messages or []:
+            if not isinstance(message, dict) or message.get('role') != 'user':
+                continue
+            token = message.get('_active_turn_token')
+            if not isinstance(token, str) or not token:
+                continue
+            rows_by_token.setdefault(token, []).append(message)
+
+    mirrored_tokens = set()
+    for token, context_rows in context_by_token.items():
+        display_rows = display_by_token.get(token, [])
+        if len(context_rows) != 1 or len(display_rows) != 1:
+            continue
+        context, display = context_rows[0], display_rows[0]
+        content = context.get('content')
+        if (
+            not isinstance(content, list)
+            or not any(
+                isinstance(part, dict)
+                and str(part.get('type') or '').lower() in {'image', 'image_url', 'input_image'}
+                for part in content
+            )
+            or not isinstance(
+                context.get(_WEBUI_TRUSTED_AGENT_INPUT_FIELD),
+                str,
+            )
+            or not _active_turn_user_text_matches(
+                context,
+                context.get(_WEBUI_TRUSTED_AGENT_INPUT_FIELD),
+            )
+            or not _message_private_identity_compatible(display, context)
+        ):
+            continue
+        display_ts, display_ts_valid = _message_exact_timestamp_details(display)
+        context_ts, context_ts_valid = _message_exact_timestamp_details(context)
+        if (
+            not display_ts_valid
+            or not context_ts_valid
+            or display_ts is None
+            or context_ts is None
+            or display_ts != context_ts
+        ):
+            continue
+        mirrored_tokens.add(token)
+    return mirrored_tokens
+
+
 def _agent_result_tool_limit_reached(result) -> bool:
     """Return True when current-turn metadata says the tool iteration cap fired."""
     if not isinstance(result, dict):
@@ -5590,6 +5761,94 @@ def _strip_oob_blocks(content):
     return content
 
 
+_OOB_ANY_OPEN_RE = re.compile(
+    r'\[OUT-OF-BAND\s+USER\s+MESSAGE(?:\s*(?:—|-)\s*.*?)?\]',
+    re.IGNORECASE,
+)
+_OOB_ANY_CLOSE_RE = re.compile(
+    r'\[/OUT-OF-BAND\s+USER\s+MESSAGE\]',
+    re.IGNORECASE,
+)
+
+
+def _unwrap_single_oob_frame(content: str) -> str | None:
+    """Unwrap exactly one fully-validated [OUT-OF-BAND USER MESSAGE] frame.
+
+    Returns the extracted inner user text if and only if ``content`` consists of
+    exactly one valid opening tag and one valid closing tag wrapping the user
+    content. If markers are multiple, nested, incomplete, or ambiguous, returns
+    None so caller preserves the row byte-for-byte.
+    """
+    if not isinstance(content, str):
+        return None
+    stripped = content.strip()
+    if not stripped:
+        return None
+
+    open_matches = list(_OOB_ANY_OPEN_RE.finditer(stripped))
+    close_matches = list(_OOB_ANY_CLOSE_RE.finditer(stripped))
+
+    # Must have exactly one opening marker and one closing marker
+    if len(open_matches) != 1 or len(close_matches) != 1:
+        return None
+
+    open_m = open_matches[0]
+    close_m = close_matches[0]
+
+    # Opening marker must be at the very start of stripped content
+    if open_m.start() != 0:
+        return None
+
+    # Closing marker must be at the very end of stripped content
+    if close_m.end() != len(stripped):
+        return None
+
+    # Opening marker must end before closing marker starts
+    if open_m.end() > close_m.start():
+        return None
+
+    inner = stripped[open_m.end():close_m.start()]
+    # Strip surrounding whitespace/newlines from the extracted user text
+    return inner.strip('\r\n').strip()
+
+
+def _unwrap_steer_row_oob_marker(message: dict) -> None:
+    """Extract inner steer text from a typed steer row in place (#7600).
+
+    The Hermes Agent emits mid-turn steers as standalone typed user rows
+    (`role == 'user'`, `display_kind == 'steer'`). The control wrapper
+    `[OUT-OF-BAND USER MESSAGE ...] ... [/OUT-OF-BAND USER MESSAGE]` is
+    extracted to preserve only the user-authored instruction.
+
+    Mutates caller-row in place to maintain object identity. If the marker
+    frame is malformed, nested, multiple, or legacy, preserves the row
+    byte-for-byte.
+    """
+    if not isinstance(message, dict):
+        return
+    if message.get('role') != 'user' or message.get('display_kind') != 'steer':
+        return
+    content = message.get('content')
+    if isinstance(content, str):
+        unwrapped = _unwrap_single_oob_frame(content)
+        if unwrapped is not None:
+            message['content'] = unwrapped
+    elif isinstance(content, list):
+        if len(content) == 1 and isinstance(content[0], dict):
+            part = content[0]
+            if part.get('type') == 'text' and isinstance(part.get('text'), str):
+                unwrapped = _unwrap_single_oob_frame(part['text'])
+                if unwrapped is not None:
+                    part['text'] = unwrapped
+
+
+def _strip_oob_markers_from_messages(messages):
+    """Unwrap OOB steer markers from typed steer rows in place (#7600)."""
+    for message in messages or []:
+        _unwrap_steer_row_oob_marker(message)
+    return messages
+
+
 def _content_has_reasoning_only_parts(content) -> bool:
     if not isinstance(content, list) or not content:
         return False
@@ -7426,6 +7685,46 @@ def _merge_display_messages_after_agent_result(
     result_messages = _drop_synthetic_control_messages(result_messages)
     if not result_messages:
         return previous_display
+    active_turn_row_index = _find_active_turn_checkpoint_index(
+        result_messages, previous_context, _active_turn_identity, msg_text,
+    )
+    active_turn_row = (
+        result_messages[active_turn_row_index]
+        if active_turn_row_index is not None
+        else None
+    )
+    active_turn_display_text = None
+    active_content = active_turn_row.get('content') if isinstance(active_turn_row, dict) else None
+    if (
+        isinstance(active_turn_row, dict)
+        and active_turn_row.get('role') == 'user'
+        and isinstance(active_content, list)
+        and any(
+            isinstance(part, dict)
+            and isinstance(part.get('type'), str)
+            and part.get('type') in {'image', 'image_url', 'input_image'}
+            for part in active_content
+        )
+    ):
+        active_turn_display_text = (
+            _active_turn_identity.get('text')
+            if _active_turn_identity.get('text') is not None
+            else msg_text
+        )
+        if not isinstance(active_turn_display_text, str):
+            active_turn_display_text = None
+        else:
+            for display_message in previous_display:
+                if _active_turn_token_matches(display_message, _active_turn_identity):
+                    display_message['content'] = active_turn_display_text
+                    display_message.pop('api_content', None)
+                    for key in ('id', 'timestamp', '_row_id'):
+                        if display_message.get(key) is None and active_turn_row.get(key) is not None:
+                            display_message[key] = copy.deepcopy(active_turn_row[key])
+                    if not display_message.get('attachments') and active_turn_row.get('attachments'):
+                        display_message['attachments'] = copy.deepcopy(
+                            active_turn_row['attachments']
+                        )
     previous_user_tail = _stale_user_tail_candidate(_last_user_row(previous_context))
 
     # ── Backfill normal turns from previous_context that are missing from
@@ -7440,16 +7739,34 @@ def _merge_display_messages_after_agent_result(
     # at/after a cursor. Any context messages between the cursor and that
     # match are context-only gaps that get spliced in before the display msg.
     if previous_display and previous_context:
+        _displayed_native_image_context_tokens = (
+            _trusted_native_image_context_display_mirror_tokens(
+                previous_display,
+                previous_context,
+            )
+        )
         _display_id_set = {_message_identity(m) for m in previous_display}
         _context_id_set = {
             _message_identity(m)
             for m in previous_context
+            if not (
+                isinstance(m, dict)
+                and m.get('_active_turn_token') in _displayed_native_image_context_tokens
+            )
             if not _is_context_compression_marker(m)
             and not _is_compressed_context_tool_result_summary_message(m)
         }
         _has_context_only_turns = bool(_context_id_set - _display_id_set)
         if _has_context_only_turns:
-            context_keys = [_message_identity(m) for m in previous_context]
+            context_keys = [
+                None
+                if (
+                    isinstance(m, dict)
+                    and m.get('_active_turn_token') in _displayed_native_image_context_tokens
+                )
+                else _message_identity(m)
+                for m in previous_context
+            ]
             # Precompute display keys once; avoids repeated json.dumps calls inside
             # the inner any() loop (was O(D²·C) — see perf fix below).
             _display_keys = [_message_identity(m) for m in previous_display]
@@ -7649,10 +7966,55 @@ def _merge_display_messages_after_agent_result(
             or _is_compressed_context_tool_result_summary_message(msg)
         ):
             continue
+        is_active_image_row = (
+            active_turn_display_text is not None
+            and isinstance(msg, dict)
+            and msg.get('role') == 'user'
+            and (
+                msg is active_turn_row
+                or _active_turn_token_matches(msg, _active_turn_identity)
+            )
+        )
+        if is_active_image_row:
+            display_row = _materialize_active_turn_user(
+                _active_turn_identity,
+                active_turn_display_text,
+                source,
+            )
+            display_row['content'] = active_turn_display_text
+            display_row.pop('api_content', None)
+            for key in ('id', 'timestamp', '_row_id'):
+                if display_row.get(key) is None and msg.get(key) is not None:
+                    display_row[key] = copy.deepcopy(msg[key])
+            if not display_row.get('attachments') and msg.get('attachments'):
+                display_row['attachments'] = copy.deepcopy(msg['attachments'])
+            _mark_active_turn_checkpoint(display_row, _active_turn_identity)
+            existing_idx = next(
+                (
+                    idx for idx, existing in enumerate(merged)
+                    if _active_turn_token_matches(existing, _active_turn_identity)
+                ),
+                None,
+            )
+            if existing_idx is not None:
+                existing = merged[existing_idx]
+                existing['content'] = active_turn_display_text
+                existing.pop('api_content', None)
+                for key in ('id', 'timestamp', '_row_id'):
+                    if existing.get(key) is None and msg.get(key) is not None:
+                        existing[key] = copy.deepcopy(msg[key])
+                if not existing.get('attachments') and msg.get('attachments'):
+                    existing['attachments'] = copy.deepcopy(msg['attachments'])
+                if not existing.get('attachments') and display_row.get('attachments'):
+                    existing['attachments'] = copy.deepcopy(display_row['attachments'])
+                _mark_active_turn_checkpoint(existing, _active_turn_identity)
+                continue
+            msg = display_row
         key = _message_identity(msg)
         is_current_user_turn = _looks_like_current_user_turn(msg, msg_text)
         if (
-            ((key is not None and key == current_user_key) or is_current_user_turn)
+            not is_active_image_row
+            and ((key is not None and key == current_user_key) or is_current_user_turn)
             and merged
             and (
                 _message_identity(merged[-1]) == current_user_key
@@ -7692,12 +8054,14 @@ def _merge_display_messages_after_agent_result(
             continue
         display_msg = msg
         if (
-            ((key is not None and key == current_user_key) or is_current_user_turn)
+            ((key is not None and key == current_user_key) or is_current_user_turn or is_active_image_row)
             and isinstance(msg, dict)
             and msg.get('role') == 'user'
         ):
             display_msg = copy.deepcopy(msg)
-            display_msg['content'] = msg_text
+            display_msg['content'] = (
+                active_turn_display_text if is_active_image_row else msg_text
+            )
             stamp_message_source(display_msg, source)
         merged.append(copy.deepcopy(display_msg))
         if key is not None:
@@ -8903,6 +9267,39 @@ def _add_supported_run_conversation_kwarg(callable_obj, kwargs, name, value):
     return False
 
 
+def _register_pending_user_timestamp_identity(
+    callable_obj, session, persist_user_timestamp,
+):
+    """Persist the pending-turn proof while the caller owns the session lock."""
+    if session is None:
+        return
+    try:
+        timestamp_parameter = inspect.signature(callable_obj).parameters.get(
+            "persist_user_timestamp"
+        )
+    except (TypeError, ValueError):
+        timestamp_parameter = None
+    has_timestamp_contract = (
+        timestamp_parameter is not None
+        and timestamp_parameter.kind is not inspect.Parameter.POSITIONAL_ONLY
+    )
+    identity = (
+        _validated_webui_pending_user_timestamp_identity(
+            session,
+            (getattr(session, "active_stream_id", None), persist_user_timestamp),
+        )
+        if has_timestamp_contract
+        else None
+    )
+    previous = getattr(session, "_webui_pending_user_timestamp_identity", None)
+    session._webui_pending_user_timestamp_identity = identity
+    session_id = getattr(session, "session_id", None)
+    save = getattr(session, "save", None)
+    if callable(save) and session_id and previous != identity:
+        # Persist before run_conversation can write this turn to state.db.
+        save(touch_updated_at=False, skip_index=True)
+
+
 def _build_run_conversation_kwargs(
     callable_obj,
     *,
@@ -8914,7 +9311,7 @@ def _build_run_conversation_kwargs(
     persist_user_message,
     persist_user_timestamp,
 ):
-    """Build one rolling-compatible Agent invocation contract.
+    """Build one rolling-compatible Agent invocation contract without mutation.
 
     ``persist_user_timestamp`` is signature-gated (#6935): an older
     hermes-agent whose ``run_conversation()`` predates the kwarg must not
@@ -9760,6 +10157,37 @@ def _run_agent_streaming(
         except Exception:
             logger.debug("Failed to put event to queue")
 
+    _last_runtime_model_identity = None
+    _runtime_model_session_id = session_id
+
+    def _observe_runtime_model():
+        """Publish this turn's Agent identity at output, never at an attempted route."""
+        nonlocal _last_runtime_model_identity
+        raw_model = getattr(agent, 'model', None)
+        if not isinstance(raw_model, str) or not raw_model.strip():
+            return
+        raw_provider = getattr(agent, 'provider', None)
+        provider = str(raw_provider).strip().lstrip('@').lower() if isinstance(raw_provider, str) else ''
+        # The Agent may carry a provider-qualified routing hint in its model.
+        from api.config import _parse_provider_qualified_model_id
+        parsed = _parse_provider_qualified_model_id(raw_model.strip())
+        model_id = (parsed[0] if parsed else raw_model).strip()
+        if not model_id:
+            return
+        fallback_active = getattr(agent, '_provider_fallback_active', None) is True
+        identity = (provider, model_id.lower(), fallback_active)
+        if identity == _last_runtime_model_identity:
+            return
+        payload = {
+            'session_id': _runtime_model_session_id, 'stream_id': stream_id,
+            'model': model_id, 'fallback_active': fallback_active,
+            'phase': 'observed_output',
+        }
+        if provider:
+            payload['provider'] = provider
+        put('runtime_model', payload)
+        _last_runtime_model_identity = identity
+
     # #5940: capture a terminal (non-retryable) provider error the Agent emits via
     # its lifecycle status_callback. The Agent aborts a non-retryable API error
     # (e.g. HTTP 400 "invalid model / no credentials") with
@@ -9780,6 +10208,7 @@ def _run_agent_streaming(
         turn-completion classifier can report the real cause instead of the
         generic no_response fallback. All other lifecycle messages are dropped.
         """
+        nonlocal _last_runtime_model_identity
         _message = str(message or '').strip()
         _kind = str(kind or '').strip().lower()
         if not _message:
@@ -9809,6 +10238,7 @@ def _run_agent_streaming(
             return
         _is_fallback_notice = _is_fallback_lifecycle_message(_kind, _message)
         if _is_fallback_notice:
+            _last_runtime_model_identity = None
             put('warning', {'type': 'fallback', 'message': _message})
 
     # xsession wakeup misroute root fix (Option 1): pre-init so the outer
@@ -10134,14 +10564,11 @@ def _run_agent_streaming(
         # loaded the default profile's `mcp_servers`, even when the session
         # was stamped with a non-default profile.  See issue #1968.
         #
-        # NOTE: `_servers` in `tools/mcp_tool.py` is a process-global registry
-        # keyed by server name.  This means once profile A registers a server
-        # named e.g. `postgres`, profile B's discovery sees it as already
-        # connected and skips it — even if B's config points at a different
-        # binary.  Fully fixing multi-profile concurrent use requires keying
-        # `_servers` by `(profile_home, name)` upstream in hermes-agent; that
-        # lives outside this WebUI repo.  This change fixes the headline bug
-        # for users who run a single non-default profile per WebUI process.
+        # NOTE: hermes-agent keys MCP connections by `(profile_home, name)` only
+        # for a routed profile (override != process home).  The HERMES_HOME
+        # mirror above would make this turn's own profile look like the process
+        # profile, falling back to bare, cross-profile names, unless the agent
+        # honours the pin from `api.profiles._pin_process_profile_home()`.
         try:
             from api.agent_compat import agent_attr
             discover_mcp_tools = agent_attr("tools.mcp_tool", "discover_mcp_tools", "tools.mcp_tool_discovery")
@@ -10356,6 +10783,8 @@ def _run_agent_streaming(
                 nonlocal _token_sent
                 if text is None:
                     return  # end-of-stream sentinel
+                if text:
+                    _observe_runtime_model()
                 # #4729: visible output is starting — flush any buffered reasoning tail
                 # first so the live Thinking stream is complete before/at the transition.
                 _flush_reasoning_buffer()
@@ -10399,6 +10828,8 @@ def _run_agent_streaming(
                     # partial window is not lost when the reasoning phase ends.
                     _flush_reasoning_buffer()
                     return
+                if text:
+                    _observe_runtime_model()
                 _tool_boundary_advanced = False
                 reasoning_delta = str(text)
                 # Some runtimes mirror user-visible progress text through the
@@ -11494,6 +11925,11 @@ def _run_agent_streaming(
                 _agent_msg_text = "\n\n".join([*_process_notifications, msg_text]).strip()
             user_message = _build_native_multimodal_message(workspace_ctx, _agent_msg_text, attachments, workspace, cfg=_cfg, active_provider=(resolved_provider or ""), active_model=(resolved_model or ""), requested_provider=(_session_requested_provider or ""), profile=(getattr(s, "profile", None) or Path(_profile_home)))
             _persistent_state_before = _persistent_state_snapshot(_profile_home)
+            with _agent_lock:
+                _persist_user_timestamp = getattr(s, 'pending_started_at', None)
+                _register_pending_user_timestamp_identity(
+                    agent.run_conversation, s, _persist_user_timestamp
+                )
             _run_conversation_kwargs = _build_run_conversation_kwargs(
                 agent.run_conversation,
                 user_message=user_message,
@@ -11509,7 +11945,7 @@ def _run_agent_streaming(
                 conversation_history_revision=_conversation_history_revision,
                 task_id=session_id,
                 persist_user_message=msg_text,
-                persist_user_timestamp=getattr(s, 'pending_started_at', None),
+                persist_user_timestamp=_persist_user_timestamp,
             )
             # Only pass moa_config when a /moa override is actually active, so a
             # normal send never trips a TypeError on an older hermes-agent whose
@@ -11551,6 +11987,11 @@ def _run_agent_streaming(
                     profile=(getattr(s, "profile", None) or Path(_profile_home)),
                 )
                 _run_conversation_kwargs["user_message"] = user_message
+            # This exact WebUI-composed input is identity evidence for the
+            # Agent-indexed current row. Keep identity['text'] as the clean
+            # submitted prompt for display. Assignment follows notification
+            # rejection/rebuild so rejected text cannot claim a returned row.
+            _active_turn_identity['trusted_agent_input_text'] = _agent_msg_text
             _result_partial_pre_call_context = list(_previous_context_messages)
             if not _agent_can_invoke(agent):
                 with _agent_lock:
@@ -11601,6 +12042,11 @@ def _run_agent_streaming(
                     if isinstance(_m, dict) and _m.get('role') == 'assistant':
                         _answer = str(_m.get('content', ''))
                         break
+                if (_answer.strip() and not result.get('error')
+                    and not _agent_result_terminal_failure(result)
+                    and not getattr(agent, '_last_error', None)
+                    and not _captured_terminal_error[0]):
+                    _observe_runtime_model()
                 # /btw is intentionally non-persistent, but its terminal SSE
                 # payload is still public output.  Project the ephemeral
                 # session before enqueueing it so raw Agent ``api_content`` or
@@ -12068,6 +12514,15 @@ def _run_agent_streaming(
                                     _heal_context_messages,
                                     _heal_conversation_history_revision,
                                 ) = _refresh_context_and_revision_from_state_db()
+                                _heal_persist_user_timestamp = getattr(
+                                    s, 'pending_started_at', None
+                                )
+                                # The returned-error path already owns _agent_lock.
+                                _register_pending_user_timestamp_identity(
+                                    agent.run_conversation,
+                                    s,
+                                    _heal_persist_user_timestamp,
+                                )
                                 _heal_kwargs = _build_run_conversation_kwargs(
                                     agent.run_conversation,
                                     user_message=user_message,
@@ -12085,7 +12540,7 @@ def _run_agent_streaming(
                                     ),
                                     task_id=session_id,
                                     persist_user_message=msg_text,
-                                    persist_user_timestamp=getattr(s, 'pending_started_at', None),
+                                    persist_user_timestamp=_heal_persist_user_timestamp,
                                 )
                                 if moa_config is not None:
                                     _heal_kwargs["moa_config"] = moa_config
@@ -12303,6 +12758,8 @@ def _run_agent_streaming(
                         # the catch-all label, hint, and provider details.
                         return  # apperror already closes the stream on the client side
 
+                _observe_runtime_model()
+
                 # ── Handle context compression side effects ──
                 # Also detect compression via the result dict or compressor state
                 if not _compressed:
@@ -12445,13 +12902,9 @@ def _run_agent_streaming(
                 if attachments:
                     display_attachments = [_attachment_name(a) for a in attachments if _attachment_name(a)]
                     for m in reversed(s.messages):
-                        if m.get('role') == 'user':
-                            content = str(m.get('content', ''))
-                            # Match if content is part of the sent message or vice-versa
-                            base_text = msg_text.split('\n\n[Attached files:')[0].strip() if '\n\n[Attached files:' in msg_text else msg_text
-                            if base_text[:60] in content or content[:60] in msg_text:
-                                m['attachments'] = display_attachments
-                                break
+                        if _active_turn_token_matches(m, _active_turn_identity):
+                            m['attachments'] = display_attachments
+                            break
                 # Persist reasoning trace in the session so it survives reload.
                 # Must run BEFORE s.save() — otherwise the mutation lives only in
                 # memory until the next turn's save, and the last-turn thinking card
@@ -12509,7 +12962,9 @@ def _run_agent_streaming(
                 # mutates agent.model when a fallback fires, so the pre-run
                 # resolved_model would mis-attribute exactly the turns where
                 # attribution matters most.
-                _used_model = getattr(agent, 'model', None) or resolved_model or model
+                # The configured selection is not proof that it served this turn.
+                _observed_model = getattr(agent, 'model', None)
+                _used_model = _observed_model.strip() if isinstance(_observed_model, str) else None
                 if _gateway_routing:
                     s.gateway_routing = _gateway_routing
                     _history = list(getattr(s, 'gateway_routing_history', None) or [])
@@ -13395,6 +13850,9 @@ def _run_agent_streaming(
                     if 'credential_pool' in _agent_params:
                         _heal_kwargs['credential_pool'] = _runtime_bundle['credential_pool']
                     _heal_agent = _AIAgent(**_heal_kwargs)
+                    # Delta callbacks close over `agent`; the replacement, not
+                    # the failed original, owns any successful retry output.
+                    agent = _heal_agent
                     _agent_sig = _compute_agent_cache_signature(
                         resolved_model,
                         resolved_api_key,
@@ -13423,6 +13881,15 @@ def _run_agent_streaming(
                             _heal_context_messages,
                             _heal_conversation_history_revision,
                         ) = _refresh_context_and_revision_from_state_db()
+                        with _agent_lock:
+                            _heal_persist_user_timestamp = getattr(
+                                s, 'pending_started_at', None
+                            )
+                            _register_pending_user_timestamp_identity(
+                                _heal_agent.run_conversation,
+                                s,
+                                _heal_persist_user_timestamp,
+                            )
                         _heal_kwargs2 = _build_run_conversation_kwargs(
                             _heal_agent.run_conversation,
                             user_message=user_message,
@@ -13440,7 +13907,7 @@ def _run_agent_streaming(
                             ),
                             task_id=session_id,
                             persist_user_message=msg_text,
-                            persist_user_timestamp=getattr(s, 'pending_started_at', None),
+                            persist_user_timestamp=_heal_persist_user_timestamp,
                         )
                         if moa_config is not None:
                             _heal_kwargs2["moa_config"] = moa_config
@@ -13522,6 +13989,7 @@ def _run_agent_streaming(
                                             s, tool_calls=s.tool_calls
                                         )
                                     )
+                            _observe_runtime_model()
                             if _done_session_payload is not None:
                                 put('done', {
                                     'session': _done_session_payload,