Release: decode state.db content sentinel in the message projection (#7492) (#7590)

Co-authored-by: n <a@n>

CHANGELOG.md

@@ -3,8 +3,36 @@
 
 ## [Unreleased]
 
+### Changed
+
+- **The chat composer grows natively instead of being resized by JavaScript on every keystroke.** `autoResize()` measured `scrollHeight` and wrote `style.height` on each input event — a forced synchronous reflow on the most-typed-in surface in the app. Browsers that support CSS `field-sizing: content` (Chromium today; also Firefox 152 and Safari 26.2) now own the geometry directly, gated on `CSS.supports()`, and the existing JavaScript path is untouched for every other engine. Measured behaviour is identical across both paths: 44px resting height, no jump when the first character is typed or the last deleted, growth to the 200px ceiling, then internal scrolling. Because `field-sizing` deliberately includes placeholder text in content sizing, `:placeholder-shown` pins fixed sizing while the composer is empty so a long placeholder can't inflate it. Thanks @starship-s. (#6760, #5514)
+
 ### Fixed
 
+- **Structured (multimodal) message content stored in `state.db` renders as its real content instead of a placeholder, without dropping or misattributing turns.** When the session projection decodes the `state.db` content sentinel, list-valued content now gets an out-of-band identity — a `("structured_content", canonical)` tuple that no scalar string can compare equal to — shared across the merge, dedup, content, and visible reconciliation keys, so an in-band string can't collide with a rich row and an image-only row can no longer vanish from the projection. The expensive canonical serialisation is memoised per content object for the duration of a single `merge_session_messages_append_only()` call (scoped through a `ContextVar`, keyed by object identity with the object held alive so an id can't be reused for a different list mid-call), restoring one serialisation per list per call. Empty/falsy content keys exactly as before. Thanks @totalitarian. (#7492)
+
+- **An empty composer no longer inflates to fit its own placeholder on the JavaScript sizing path.** The fallback resizer measured `scrollHeight` for an empty textarea, which reflects the *placeholder* — so a long busy or compression hint (which wraps to two or three lines) grew the empty composer to roughly 71px, and only after a resize that happened while empty, making the resting height depend on history rather than content. The inline height is now cleared when the value is empty so CSS `min-height` defines the resting size, matching the native path. (#6760)
+
+- **Delegated subagent sessions nest under the conversation that spawned them instead of landing at the top of the sidebar.** The sidebar forces a *cross-surface* child row to top level when its parent row belongs to another surface (a Telegram/messaging parent), so an independent continuation doesn't get stacked under a foreign thread. A delegated subagent is also technically cross-source relative to its WebUI parent, so it was swept up by that rule and surfaced as a bare top-level row with no indication of what spawned it. Delegated subagents are now exempt from that branch and attach to their visible parent, while independent cross-surface continuations still stay top level. The delegated role is resolved in strict precedence order from the first non-blank of `raw_source` → `source_tag` → `source` and additionally requires the backend's child-session flag, so a stale lower-priority field can't promote an independent row into a delegated one. Thanks @lowlandsheperd. (#7263)
+
+- **Remote (SSH/Docker) workspaces keep their target-side POSIX paths across session restore, streaming, uploads and project context.** When the terminal backend runs off-host, a workspace path such as `/srv/remote-alice` belongs to the *target* machine — but several resolvers normalized it against the WebUI host's filesystem, so a host-side expansion (macOS firmlinks turning it into `/System/Volumes/Data/srv/remote-alice`, or a local `~` expansion) could be persisted back onto the session and then used as the per-turn runtime CWD. The workspace, streaming, upload and project-context resolvers now receive the owning **session's profile** explicitly instead of resolving against whatever profile happens to be ambient, so remote paths stay target-side and a session owned by one profile can never resolve through another's. Local (non-remote) workspaces normalize exactly as before, and symlink/`..`/outside-root rejection is unchanged. Thanks @alfred-rootson. (#7168, closes #7131)
+
+- **Signal gateway conversations are classified as messaging sessions instead of falling into the default/other bucket.** The messaging-source allowlists never learned about Signal, so a session started on the Signal gateway was normalized to `other`, failed the client-side external-session check the session list uses to render and refresh gateway sessions, and never showed up in the sidebar. `signal` (labeled "Signal") is now in the messaging-source sets on both sides — the Python normalizer (`MESSAGING_SOURCES` / `SOURCE_LABELS`), which also feeds the server-side messaging allowlist, and the client-side classifier (`_MESSAGING_RAW_SOURCES` / `_MESSAGING_SOURCE_LABELS`) — so Signal sessions group, label, and list exactly like Telegram, Discord, Slack, WeCom, and Matrix. No other source's classification changes. Thanks @webtecnica. (#7571)
+
+- **Opening a session no longer stalls just because Codex refreshed its model cache in the background.** The WebUI keys its 24-hour `/api/models` cache partly on Codex's `~/.codex/models_cache.json`, but that file was fingerprinted by `mtime` + size — and Codex rewrites it on its own refresh timer, bumping those stat fields even when the model catalog is byte-identical (only a volatile `fetched_at` timestamp changed). Every Codex refresh therefore invalidated the cache, and the next session open paid a full live rebuild whose serial provider probes stalled the open (#7540). The Codex cache is now fingerprinted by its *content* with the refresh-timestamp fields (`fetched_at`, `updated_at`) stripped, so a timestamp-only rewrite keeps the cache while any genuine catalog change (models, `etag`, `client_version`) still invalidates it. A malformed or pathologically deep cache file degrades safely to the old stat fingerprint rather than erroring. Thanks @webtecnica. (#7556, closes #7540)
+
+- **A timed-out command-approval card can be dismissed again instead of getting stuck on screen forever.** When the agent raised a local approval with no gateway notifier registered, the raw pending entry was stored without an `approval_id`; the frontend needs that id to own the card, so a card that timed out (or otherwise went unanswered) could be neither approved nor dismissed and the poll re-served it indefinitely. Reconciliation now mints a stable `gwlocal-mirrorless:` id on the raw local entry itself — the same tokenization contract already used for gateway producers — so the id survives repeated polls and the client's dismiss marker sticks. Gateway/run-bearing entries are untouched, and the synthetic prefix can't be mistaken for gateway authority (routing keys on exact ids, `run_id`, and mirror tokens, never the prefix). Thanks @CharlesMcquade. (#7510)
+
+- **Manual "Regenerate title" now works for conversations that open with several queued user messages.** For a session whose transcript begins with consecutive user turns (no assistant reply before the second user message), the first-exchange walker stopped at the second user row with no assistant text, so "Regenerate title" skipped the LLM call, silently persisted the local fallback (a truncated first message), and returned `200` with that same wrong title on every retry — leaving the session permanently stuck. Manual regeneration now scans past the opening user rows to the first complete user+assistant pair so it reaches the aux model. The automatic in-stream title path is deliberately unchanged. Thanks @Raylands. (#7545, closes #7543)
+
+- **Custom providers that Hermes auto-discovered now show their full live `/v1/models` catalog again instead of just the saved models.** When a provider entry carries `models_discovered: true` alongside a `models:` mapping of per-model metadata (vision, context length), that mapping is discovery metadata — not a hand-curated allowlist — but the WebUI treated it as one and suppressed the live catalog, collapsing the model picker to only the already-saved IDs. Such a provider now defers to its live `/v1/models` catalog (matching the Hermes Agent's own `_models_config_is_allowlist` semantics), while an explicit `discover_models: false` opt-out (including the Agent-compatible string forms `false`/`no`/`0`) still pins the configured list, and a transient empty live probe falls back to the configured models rather than dropping the provider from the picker. Thanks @evgenyponomarev. (#7409, closes #7404)
+
+- **Ctrl-C and `ctl.sh`-launched daemons now shut down gracefully instead of leaving the server running.** The WebUI installed a SIGTERM handler that drains in-flight work through `serve_forever()`'s `finally`, but SIGINT fell through to Python's default, so a daemon started via `./ctl.sh start` could only be stopped with `kill <pid>` and the Settings "Stop server" button did not stop it. SIGINT now shares the same idempotent graceful-shutdown handler as SIGTERM. Thanks @aniruddhaadak80. (#7315, closes #7078)
+
+- **Static assets are served with correct MIME types instead of a `text/plain` catch-all.** `_serve_static()` defaulted every extension outside its small built-in allowlist to `text/plain; charset=utf-8`, mislabeling PDFs, APKs, WASM and other binaries. Unknown extensions now resolve through Python's standard `mimetypes` database, with an explicit deterministic entry for `.apk` (whose type varies across host MIME databases), and fail closed to `application/octet-stream` for unknown or pre-encoded suffixes (`.svgz`/`.tgz`/`.js.gz`) so still-compressed bytes are never advertised with their decoded media type. Path containment, gzip/ETag/cache handling, and the text charset path are unchanged. Thanks @Oidaladn0. (#7372)
+
+- **Chat-attachment uploads are created atomically so a raced duplicate or symlink can't silently overwrite or redirect the write.** `handle_upload` deduped filenames with an `exists()` check and then wrote with a plain `write_bytes`, so two concurrent uploads of the same name both passed the check and the last writer silently won — leaving one client a `200` naming bytes no longer on disk. The final create now uses the existing descriptor-anchored `open_anchored_create_fd` (`O_CREAT|O_EXCL|O_NOFOLLOW`, mirroring the #3398 workspace-upload hardening): a raced duplicate returns `409` instead of overwriting, and a symlink raced into the attachment path cannot redirect the write outside the session inbox. The normal success path, size/MIME handling, and `-1` dedup suffixing are unchanged. Thanks @zicochaos. (#7337)
+
 - **Cron jobs can be delivered to your messaging platforms again.** `GET /api/crons/delivery-options` had been returning only `local` and `origin`, so Telegram, Discord, Slack, Feishu and every other platform was silently missing from the cron delivery picker — a scheduled job could not be pointed anywhere. The Agent moved its delivery-platform allowlist to a new module, and the endpoint still imported the old path inside a bare `except` that fell back to an empty set, so the import error was swallowed and the feature degraded with nothing logged. The allowlist is now resolved across both module paths so the picker survives future relocations, with a regression test that fails loudly instead of emptying out. (#7525)
 
 - **Auxiliary-model settings no longer silently discard a configured provider that is missing from the model catalog.** The provider dropdown was built only from providers the catalog returned, so when a task's configured provider was absent (for example its group exposes no models), nothing matched, the select fell back to its first entry (`auto`), and the next Apply persisted `auto` — quietly throwing away the user's choice on a row they never touched. The configured provider is now kept selectable so it round-trips through Apply unchanged. Thanks @webtecnica. (#7519, closes #7486)

api/agent_sessions.py

@@ -59,6 +59,7 @@ MESSAGING_SOURCES = {
     'telegram',
     'weixin',
     'matrix',
+    'signal',
 }
 
 CLI_MIN_UNTITLED_MESSAGE_COUNT = 6
@@ -82,6 +83,7 @@ SOURCE_LABELS = {
     'webui': 'WebUI',
     'weixin': 'Weixin',
     'matrix': 'Matrix',
+    'signal': 'Signal',
 }
 
 

api/config.py

@@ -713,10 +713,14 @@ def get_config_for_profile_home(profile_home: "Path | str | None") -> dict:
     bypassing the thread-local resolver entirely. When ``profile_home`` matches
     the path the ambient resolver would pick (the common single-profile case),
     we return the cached ``get_config()`` to preserve in-memory overrides used
-    by tests and runtime callers. Only when the session's profile home diverges
-    from the ambient path do we read the session profile's file directly — a
-    pure read with no global cache mutation, so it is race-free across
-    concurrent sessions on different profiles.
+    by tests and runtime callers, and to honour an authoritative
+    ``HERMES_CONFIG_PATH`` override. Only when the session's profile home
+    diverges from the ambient path do we read the session profile's file
+    directly — a pure read with no global cache mutation, so it is race-free
+    across concurrent sessions on different profiles. Divergent profiles stay
+    isolated: a nonexistent home returns ``{}`` and an existing home without a
+    ``config.yaml`` yields defaults — neither ever falls back to the ambient
+    config (profiles-are-islands).
     """
     if not profile_home:
         return get_config()
@@ -724,10 +728,18 @@ def get_config_for_profile_home(profile_home: "Path | str | None") -> dict:
         target = Path(profile_home).expanduser()
     except Exception:
         return get_config()
+
+    from api.workspace import _safe_resolve as _cfg_safe_resolve
+
+    # Canonicalize BOTH sides before every identity comparison (#7168 re-gate
+    # round 5): when HERMES_HOME (or the config parent) is a symlink alias,
+    # lexical equality fails and an authoritative HERMES_CONFIG_PATH inside
+    # the aliased home would be bypassed in favor of a direct — wrong — read.
+    target = _cfg_safe_resolve(target)
     try:
         from api.profiles import get_active_hermes_home
 
-        if Path(get_active_hermes_home()).expanduser() == target:
+        if _cfg_safe_resolve(Path(get_active_hermes_home()).expanduser()) == target:
             return get_config()
     except Exception:
         pass
@@ -737,7 +749,7 @@ def get_config_for_profile_home(profile_home: "Path | str | None") -> dict:
     # whose directory doesn't physically exist yet (fresh install, monkeypatched
     # cfg) must still resolve through get_config(), not return {} (#4516 gate).
     try:
-        if _get_config_path().parent == target:
+        if _cfg_safe_resolve(_get_config_path().parent) == target:
             return get_config()
     except Exception:
         pass
@@ -1399,6 +1411,37 @@ def _configured_model_ids(raw_models: object) -> list[str]:
     return model_ids
 
 
+def _provider_discover_allowed(provider_cfg: object) -> bool:
+    """Mirror the Hermes Agent ``discover_models`` opt-out (``model_switch_providers._discover_flag``).
+
+    ``discover_models`` defaults to True; the string forms ``"false"``/``"no"``/``"0"``
+    (case-insensitive) mean False. A provider that pins its catalog with
+    ``discover_models: false`` keeps its configured ``models:`` even when the entry is
+    also marked ``models_discovered: true`` — the explicit opt-out wins.
+    """
+    if not isinstance(provider_cfg, dict):
+        return True
+    discover = provider_cfg.get("discover_models", True)
+    if isinstance(discover, str):
+        return discover.strip().lower() not in {"false", "no", "0"}
+    return bool(discover)
+
+
+def _provider_models_are_discovered_catalog(provider_cfg: object) -> bool:
+    """True when ``models:`` is an auto-discovered catalog that should defer to the live probe.
+
+    A provider entry marked ``models_discovered: true`` carries a per-model *metadata*
+    mapping written by Hermes discovery, not a hand-curated allowlist — so the live
+    ``/v1/models`` catalog is authoritative. But an explicit ``discover_models: false``
+    re-pins the configured mapping as the source of truth, so honor that opt-out.
+    """
+    return (
+        isinstance(provider_cfg, dict)
+        and provider_cfg.get("models_discovered") is True
+        and _provider_discover_allowed(provider_cfg)
+    )
+
+
 def _configured_model_options(raw_models: object) -> list[dict[str, str]]:
     """Return picker option rows from supported config allowlist shapes."""
     labels: dict[str, str] = {}
@@ -1418,6 +1461,29 @@ def _configured_model_options(raw_models: object) -> list[dict[str, str]]:
     ]
 
 
+def _merge_model_option_rows(*row_lists: object) -> list[dict[str, str]]:
+    """Merge picker option rows from multiple sources, first-seen order, deduped by id.
+
+    Used to preserve a discovered provider's configured model IDs (ordered first) as a
+    fallback when a live ``/v1/models`` probe transiently returns nothing, merged with
+    any static built-in catalog without producing duplicate ids.
+    """
+    merged: list[dict[str, str]] = []
+    seen: set[str] = set()
+    for rows in row_lists:
+        if not isinstance(rows, (list, tuple)):
+            continue
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            model_id = str(row.get("id") or "").strip()
+            if not model_id or model_id in seen:
+                continue
+            seen.add(model_id)
+            merged.append(row)
+    return merged
+
+
 def _named_custom_provider_slugs(config_obj: dict | None = None) -> set[str]:
     return {
         slug
@@ -5741,8 +5807,12 @@ def _static_models_catalog_without_live_probes() -> dict:
             raw_key = canonical_to_raw_provider_key.get(pid, pid)
             provider_cfg = _get_provider_cfg(raw_key)
             raw_models = []
-            if isinstance(provider_cfg, dict) and "models" in provider_cfg:
-                raw_models = _configured_model_options(provider_cfg["models"])
+            if (
+                isinstance(provider_cfg, dict)
+                and "models" in provider_cfg
+                and not _provider_models_are_discovered_catalog(provider_cfg)
+            ):
+                raw_models = _configured_model_options(provider_cfg.get("models"))
             if not raw_models:
                 raw_models = copy.deepcopy(_PROVIDER_MODELS.get(pid, []))
             # Plugin-only providers (e.g. 9router) are not in _PROVIDER_MODELS
@@ -6124,6 +6194,97 @@ def _models_cache_file_fingerprint(path: Path) -> dict:
     return fingerprint
 
 
+# Codex's ~/.codex/models_cache.json is rewritten on Codex's own refresh timer.
+# Each rewrite bumps mtime_ns + size, but the payload usually only refreshes
+# the volatile timestamp fields (`fetched_at`, plus `updated_at` when present)
+# while the model catalog (client_version, etag, models[]) stays identical.
+# Fingerprinting that file by stat (#2443's _models_cache_file_fingerprint)
+# therefore invalidated the 24h /api/models cache on every Codex refresh, and
+# the next session visit paid a full live rebuild whose serial provider probes
+# starved the session-open path (#7540).
+#
+# This is a DENY-list, not an allow-list, on purpose — same safety direction as
+# _AUTH_FINGERPRINT_VOLATILE_KEYS: every other field (client_version, etag,
+# models, and any future model-affecting key) stays IN the fingerprint, so a
+# genuine catalog change still invalidates the cache.
+_CODEX_CACHE_FINGERPRINT_VOLATILE_KEYS = frozenset({
+    # Whole-file refresh timestamp, rewritten by every Codex models refresh.
+    "fetched_at",
+    # Same-family save timestamp (mirrors the auth.json deny-list).
+    "updated_at",
+})
+
+
+def _strip_volatile_codex_cache_fields(obj):
+    """Recursively drop refresh-timestamp-only keys from a Codex cache tree.
+
+    Pure structural transform; never mutates the input. Any key NOT in the
+    deny-list is preserved verbatim so real catalog changes still show through
+    in the fingerprint.
+    """
+    if isinstance(obj, dict):
+        return {
+            k: _strip_volatile_codex_cache_fields(v)
+            for k, v in obj.items()
+            if k not in _CODEX_CACHE_FINGERPRINT_VOLATILE_KEYS
+        }
+    if isinstance(obj, list):
+        return [_strip_volatile_codex_cache_fields(v) for v in obj]
+    return obj
+
+
+def _codex_models_cache_fingerprint(path: Path) -> dict:
+    """Return a content fingerprint of Codex's models_cache.json.
+
+    Unlike _models_cache_file_fingerprint() (mtime_ns + size), this hashes the
+    JSON content with the refresh-timestamp fields stripped, so a Codex
+    refresh that only bumps `fetched_at` does NOT invalidate the 24h
+    /api/models cache and therefore does not force the live rebuild that
+    stalled session opens (#7540). A change to anything that actually feeds the
+    Codex models we surface (client_version, etag, models[], an unknown future
+    field) still changes the hash and correctly busts the cache.
+
+    Failure modes are deliberately conservative — a missing file is recorded,
+    and an unreadable/undecodable file falls back to the stat-based fingerprint
+    so behaviour is never *less* safe than the stat-only version.
+    """
+    p = Path(path).expanduser()
+    fp: dict = {"path": str(p)}
+    try:
+        st = p.stat()
+    except OSError:
+        fp["missing"] = True
+        return fp
+    try:
+        raw = json.loads(p.read_text(encoding="utf-8"))
+    except Exception:
+        # Unreadable / corrupt / mid-write: keep the stat-based fingerprint.
+        # Strictly no less safe than the pre-fix behaviour (every write still
+        # invalidates) for this rare path only.
+        fp["mtime_ns"] = st.st_mtime_ns
+        fp["size"] = st.st_size
+        fp["semantic"] = "unparsed-fallback"
+        return fp
+    try:
+        # The recursive strip can raise (e.g. RecursionError on a pathologically
+        # deep JSON tree) — keep it inside the fallback try so any transform
+        # failure degrades to the stat fingerprint rather than 500ing /api/models.
+        stripped = _strip_volatile_codex_cache_fields(raw)
+        encoded = json.dumps(
+            stripped,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=True,
+            default=str,
+        ).encode("utf-8")
+        fp["semantic_sha256"] = hashlib.sha256(encoded).hexdigest()
+    except Exception:
+        fp["mtime_ns"] = st.st_mtime_ns
+        fp["size"] = st.st_size
+        fp["semantic"] = "encode-fallback"
+    return fp
+
+
 def _models_cache_catalog_fingerprint() -> dict:
     """Return non-secret model-catalog identity metadata for cache invalidation.
 
@@ -6133,6 +6294,13 @@ def _models_cache_catalog_fingerprint() -> dict:
     deterministic so a server restart after catalog changes does not keep
     serving an otherwise-valid persisted models_cache.json until the 24h TTL
     expires (#2443).
+
+    The Codex axis uses a *content* fingerprint that excludes the refresh
+    timestamp fields (see _codex_models_cache_fingerprint): Codex rewrites
+    ~/.codex/models_cache.json on its own timer, bumping mtime_ns + size while
+    the model payload stays identical, so a stat-based fingerprint invalidated
+    the 24h cache on every Codex refresh and the next session visit paid a live
+    rebuild (#7540).
     """
     catalog_payload = {
         "provider_models": _PROVIDER_MODELS,
@@ -6153,7 +6321,7 @@ def _models_cache_catalog_fingerprint() -> dict:
     codex_home = Path(os.getenv("CODEX_HOME", "").strip() or (HOME / ".codex")).expanduser()
     return {
         "provider_catalog_sha256": provider_catalog_sha,
-        "codex_models_cache": _models_cache_file_fingerprint(codex_home / "models_cache.json"),
+        "codex_models_cache": _codex_models_cache_fingerprint(codex_home / "models_cache.json"),
     }
 
 
@@ -8143,13 +8311,16 @@ def get_available_models(*, prefer_cache: bool = False, force_refresh: bool = Fa
                     # whichever model had local settings. Only Copilot skips the
                     # config-models allowlist branch and asks Hermes CLI for the
                     # live catalog first (static _PROVIDER_MODELS is fallback only).
-                    _uses_models_as_settings_map = pid == "copilot"
+                    _uses_models_as_settings_map = (
+                        pid == "copilot"
+                        or _provider_models_are_discovered_catalog(provider_cfg)
+                    )
                     if (
                         not _uses_models_as_settings_map
                         and isinstance(provider_cfg, dict)
                         and "models" in provider_cfg
                     ):
-                        raw_models = _configured_model_options(provider_cfg["models"])
+                        raw_models = _configured_model_options(provider_cfg.get("models"))
 
                     if not raw_models:
                         if pid == "moa":
@@ -8165,6 +8336,19 @@ def get_available_models(*, prefer_cache: bool = False, force_refresh: bool = Fa
                                 pid,
                                 _read_live_provider_model_ids(pid),
                             )
+                            if (
+                                not raw_models
+                                and _provider_models_are_discovered_catalog(provider_cfg)
+                            ):
+                                # A transient live-catalog failure must not drop a
+                                # provider's persisted discovered models (the empty
+                                # result would then be cached for up to 24h). Fall
+                                # back to the configured discovered IDs, ordered
+                                # first, merged with any static fallback (deduped).
+                                raw_models = _merge_model_option_rows(
+                                    _configured_model_options(provider_cfg.get("models")),
+                                    copy.deepcopy(_PROVIDER_MODELS.get(pid, [])),
+                                )
 
                     if not raw_models:
                         raw_models = copy.deepcopy(_PROVIDER_MODELS.get(pid, []))

api/gateway_chat.py

@@ -584,7 +584,7 @@ def _run_gateway_runs_api_streaming(
             try:
                 from api.streaming import _build_native_multimodal_message
 
-                message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=active_provider, active_model=(model or ""), requested_provider=active_provider)
+                message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=active_provider, active_model=(model or ""), requested_provider=active_provider, profile=getattr(session, "profile", None))
             except Exception:
                 logger.debug("Failed to build runs-API multimodal attachment payload", exc_info=True)
                 message_content = str(msg_text or "")
@@ -1120,7 +1120,7 @@ def _run_gateway_chat_streaming(
                 try:
                     from api.streaming import _build_native_multimodal_message
 
-                    message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=(model_provider or ""), active_model=(model or ""), requested_provider=(model_provider or ""))
+                    message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg, active_provider=(model_provider or ""), active_model=(model or ""), requested_provider=(model_provider or ""), profile=getattr(s, "profile", None))
                 except Exception:
                     logger.debug("Failed to build gateway multimodal attachment payload", exc_info=True)
                     message_content = str(msg_text or "")

api/models.py

@@ -1,5 +1,6 @@
 """Hermes Web UI -- Session model and in-memory session store."""
 import collections
+import contextvars
 import copy
 import datetime
 import hashlib
@@ -34,7 +35,7 @@ from api.config import (
     LOCK, STREAMS, STREAMS_LOCK, DEFAULT_WORKSPACE, DEFAULT_MODEL, PROJECTS_FILE, HOME,
     get_effective_default_model, _get_session_agent_lock,
 )
-from api.workspace import get_last_workspace
+from api.workspace import get_last_workspace, _resolve_path
 from api.usage import prompt_cache_hit_percent
 from api.agent_sessions import (
     _is_continuation_session,
@@ -1264,7 +1265,8 @@ class Session:
                  **kwargs):
         self.session_id = session_id or uuid.uuid4().hex[:12]
         self.title = title
-        self.workspace = str(Path(workspace).expanduser().resolve())
+        self.profile = profile
+        self.workspace = str(_resolve_path(workspace, profile=profile))
         # #6672: immutable snapshot of the workspace at session creation time.
         # s.workspace is updated on every turn when the user switches workspaces
         # mid-session via the WebUI header dropdown; interpolating the live
@@ -1276,7 +1278,7 @@ class Session:
         # without a persisted created_workspace fall back to the workspace
         # recorded on disk, which is the best available approximation.
         self.created_workspace = (
-            str(Path(created_workspace).expanduser().resolve())
+            str(_resolve_path(created_workspace, profile=profile))
             if created_workspace
             else self.workspace
         )
@@ -5098,7 +5100,7 @@ def new_session(workspace=None, model=None, profile=None, model_provider=None, p
     wt = worktree_info if isinstance(worktree_info, dict) else None
     workspace_path = (wt.get('path') if wt and wt.get('path') else workspace) if wt else workspace
     s = Session(
-        workspace=workspace_path or get_last_workspace(),
+        workspace=workspace_path or get_last_workspace(profile=profile),
         model=effective_model,
         model_provider=effective_model_provider,
         profile=profile,
@@ -5763,6 +5765,7 @@ def get_session_for_file_ops(sid: str):
         workspace, recovered = resolve_implicit_workspace_with_recovery(
             stored_workspace,
             get_last_workspace,
+            profile=session_profile or None,
         )
     except ValueError:
         # Preserve the existing file-handler behavior for non-missing trust or
@@ -6797,7 +6800,7 @@ def import_cli_session(
     s = Session(
         session_id=session_id,
         title=title,
-        workspace=get_last_workspace(),
+        workspace=get_last_workspace(profile=profile),
         model=model,
         messages=messages,
         profile=profile,
@@ -7720,7 +7723,10 @@ def _load_cli_sessions_uncached(
     _cli_workspace_cache: list = [None]  # list-as-cell; None = not yet resolved
     def _cli_workspace():
         if _cli_workspace_cache[0] is None:
-            _cli_workspace_cache[0] = str(get_last_workspace())
+            try:
+                _cli_workspace_cache[0] = str(get_last_workspace(profile=_cli_profile))
+            except TypeError:
+                _cli_workspace_cache[0] = str(get_last_workspace())
         return _cli_workspace_cache[0]
 
     _webhook_pid_cache: list[str | None] = [None]
@@ -7934,7 +7940,7 @@ def _load_cli_sessions_uncached(
                 cli_sessions.append({
                     'session_id': sid,
                     'title': _display_title,
-                    'workspace': str(get_last_workspace()),
+                    'workspace': str(get_last_workspace(profile=_cli_profile)),
                     'model': row['model'] or None,
                     'message_count': row['message_count'] or row['actual_message_count'] or 0,
                     'created_at': row['started_at'],
@@ -8222,18 +8228,138 @@ def _state_db_active_rows_digest(rows) -> str:
     return digest.hexdigest() if stamped else ''
 
 
+# hermes_state stores list/dict message content (multimodal parts) as a
+# sentinel-prefixed JSON string because sqlite3 binds only scalars; see
+# hermes_state._CONTENT_JSON_PREFIX and _decode_content(). This module reads
+# that table with its own SQL, so it must apply the same decode. Without it an
+# image part's base64 data URI reaches the transcript as literal text -- a
+# single unbreakable ~65k-character run -- and WebKit computes min-content
+# width by scanning every line-break position, pinning a core for minutes.
+#
+# The decode deliberately does NOT widen `content` beyond the one shape the
+# rest of the WebUI already accepts:
+#   * list roots only. A dict root would reach _getCachedRender() unchanged and
+#     _renderCacheKey() would call text.slice() on an object, blanking the turn.
+#   * no non-finite numbers. Python emits NaN/Infinity, which the browser's
+#     JSON.parse() rejects, breaking the whole /api/session response.
+# Anything else is returned as the original string, exactly as before.
+_STATE_DB_CONTENT_JSON_PREFIX = "\x00json:"
+
+
+def _is_valid_image_part_payload(part) -> bool:
+    """Explicit per-type payload validation for an image content part."""
+    part_type = part.get("type")
+    if part_type in ("image_url", "input_image"):
+        ref = part.get("image_url")
+        if isinstance(ref, dict):
+            ref = ref.get("url")
+        if isinstance(ref, str) and ref.strip():
+            return True
+        file_id = part.get("file_id")
+        return part_type == "input_image" and isinstance(file_id, str) and bool(file_id.strip())
+    if part_type == "image":
+        source = part.get("source")
+        if not isinstance(source, dict):
+            return False
+        if source.get("type") == "base64":
+            data = source.get("data")
+            return (
+                isinstance(data, str)
+                and bool(data)
+                and isinstance(source.get("media_type"), str)
+            )
+        if source.get("type") == "url":
+            url = source.get("url")
+            return isinstance(url, str) and bool(url.strip())
+        return False
+    return False
+
+
+def _is_supported_content_part_list(parts) -> bool:
+    """True only for lists the current WebUI will actually render.
+
+    The shared JS readers keep text parts and discard every image part:
+    ``msgContent()`` joins the text parts and trims, and ``_messageIsRenderable()``
+    hides the row when that is empty. This projection supplies no attachments,
+    so image parts do not render from it either. A list is therefore decoded only
+    when it carries non-whitespace text to show. Image-only lists, and lists with
+    a malformed image part, stay undecoded so the row cannot silently vanish.
+    """
+    if not parts:
+        return False
+    has_text = False
+    for part in parts:
+        if not isinstance(part, dict):
+            return False
+        part_type = part.get("type")
+        if not isinstance(part_type, str):
+            return False
+        if part_type == "text":
+            text = part.get("text")
+            if not isinstance(text, str):
+                return False
+            if text.strip():
+                has_text = True
+        elif part_type in _SESSION_MESSAGE_IMAGE_PART_TYPES:
+            if not _is_valid_image_part_payload(part):
+                return False
+        else:
+            return False
+    return has_text
+
+
+def _reject_non_finite_state_db_json_constant(value):
+    raise ValueError(f"unsupported JSON constant: {value}")
+
+
+def _parse_finite_state_db_json_float(value):
+    parsed = float(value)
+    if not math.isfinite(parsed):
+        raise ValueError(f"unsupported JSON float: {value}")
+    return parsed
+
+
+def _decode_state_db_content(value):
+    """Decode the Agent's structured-content storage form without widening it.
+
+    Returns the original value unchanged for anything that is not a
+    sentinel-prefixed, finite, list-rooted payload.
+    """
+    if isinstance(value, bytes):
+        value = value.decode("utf-8", errors="replace")
+    if not isinstance(value, str) or not value.startswith(_STATE_DB_CONTENT_JSON_PREFIX):
+        return value
+    try:
+        decoded = json.loads(
+            value[len(_STATE_DB_CONTENT_JSON_PREFIX):],
+            parse_constant=_reject_non_finite_state_db_json_constant,
+            parse_float=_parse_finite_state_db_json_float,
+        )
+    except Exception:
+        return value
+    if not isinstance(decoded, list) or not _is_supported_content_part_list(decoded):
+        return value
+    try:
+        # Prove the decoded value survives the trip to the browser before
+        # handing it on: re-serialisable, finite, UTF-8 encodable.
+        json.dumps(decoded, ensure_ascii=False, allow_nan=False).encode("utf-8")
+    except Exception:
+        return value
+    return decoded
+
+
 def _project_state_db_message(row, available, id_col, optional):
     """Authoritative state.db row → WebUI message projection (#6826 r4).
 
     Shared by ``get_state_db_session_messages`` and the regeneration
     single-snapshot helper so the bounded tail can never drift from the
-    canonical reader: JSON-decode tool_calls/reasoning payloads, omit
+    canonical reader: JSON-decode content/tool_calls/reasoning payloads, omit
     empty fields, keep durable row id private (``_state_db_row_id`` only for
     real Agent api_content replays), and apply ``tool_name → name``.
     """
     msg = {
         'role': row['role'],
-        'content': row['content'],
+        'content': _decode_state_db_content(row['content']),
         'timestamp': row['timestamp'],
     }
     for col in optional:
@@ -8636,7 +8762,11 @@ def get_state_db_session_message_keys_before_timestamp(
                 _session_message_visible_key(
                     {
                         "role": row["role"],
-                        "content": row["content"],
+                        # Same guarded decode as the projected tail: prefix and
+                        # tail keys must share one representation or the
+                        # prefix/tail collision proof can miss a genuine
+                        # repeated recovered turn.
+                        "content": _decode_state_db_content(row["content"]),
                         "tool_calls": _json_loads_if_string(row["tool_calls"]),
                         "api_content": row["api_content"] if "api_content" in available else None,
                     },
@@ -8744,7 +8874,7 @@ def get_state_db_regeneration_tail_snapshot(
             prefix_keys = [
                 _session_message_visible_key({
                     "role": r["role"],
-                    "content": r["content"],
+                    "content": _decode_state_db_content(r["content"]),
                     "tool_calls": _json_loads_if_string(r["tool_calls"]) if "tool_calls" in r.keys() and r["tool_calls"] is not None else None,
                     "api_content": r["api_content"] if "api_content" in r.keys() else None,
                 }, normalize_workspace_prefix=True)
@@ -8931,14 +9061,27 @@ def _session_message_multimodal_mirror_key(
     msg: dict,
     *,
     require_image_parts: bool = False,
+    require_scalar_mirror: bool = False,
 ):
-    """Return exact cross-store identity for a native multimodal mirror."""
+    """Return exact cross-store identity for a native multimodal mirror.
+
+    The bridge exists to pair ONE rich image-bearing row with ONE scalar row
+    holding the Agent's stored projection of it. It must never pair two rich
+    rows: distinct image turns can project to the same "[screenshot] <text>"
+    string, so rich-to-rich matching collapses different images into one turn.
+    Callers pass ``require_image_parts`` on the rich side and
+    ``require_scalar_mirror`` on the scalar side to keep the pairing asymmetric.
+    """
     if not isinstance(msg, dict):
         return None
+    if require_image_parts and require_scalar_mirror:
+        return None
     role = str(msg.get("role") or "").strip().lower()
     if role != "user":
         return None
     raw_content = msg.get("content")
+    if require_scalar_mirror and not isinstance(raw_content, str):
+        return None
     content = _agent_durable_multimodal_content(msg)
     if require_image_parts and content is None:
         return None
@@ -8964,6 +9107,53 @@ def _session_message_multimodal_mirror_key(
     )
 
 
+# Per-call memo of structured-content identities. Reconciliation derives merge,
+# dedup, content and visible keys for every message, several per source, so a
+# large multimodal payload would otherwise be serialised once per key. The memo
+# is scoped to a single merge call (set/reset in
+# merge_session_messages_append_only) and keyed by object identity with the
+# object held alive, so a later call always recomputes after mutation and an id
+# can never be reused for a different list within one call.
+_STRUCTURED_IDENTITY_MEMO = contextvars.ContextVar(
+    "_STRUCTURED_IDENTITY_MEMO", default=None
+)
+
+
+def _canonical_structured_content(content) -> str:
+    """Canonical serialisation of structured content -- the expensive step."""
+    try:
+        return json.dumps(content, sort_keys=True, ensure_ascii=False, default=str)
+    except Exception:
+        return repr(content)
+
+
+def _content_identity_for_key(content):
+    """Identity component for message content in every reconciliation key.
+
+    Non-list values key exactly as they always have: ``str(content or "")``.
+
+    Non-empty list content -- which reaches these paths once the state.db
+    sentinel is decoded -- gets an OUT-OF-BAND identity: a tuple, not a string.
+    No scalar can ever compare equal to it, so there is no in-band marker for a
+    message body to imitate. (An earlier revision tagged lists with a string
+    prefix; a scalar containing that prefix collided with the rich row.)
+
+    Merge, dedup, content and visible keys all derive structured content
+    through this one function so the discriminator cannot drift between them.
+    """
+    if not (isinstance(content, list) and content):
+        return str(content or "")
+    memo = _STRUCTURED_IDENTITY_MEMO.get()
+    if memo is not None:
+        hit = memo.get(id(content))
+        if hit is not None and hit[0] is content:
+            return hit[1]
+    identity = ("structured_content", _canonical_structured_content(content))
+    if memo is not None:
+        memo[id(content)] = (content, identity)
+    return identity
+
+
 def _session_message_merge_key(msg: dict):
     if not isinstance(msg, dict):
         return ("non_dict", repr(msg))
@@ -8983,7 +9173,7 @@ def _session_message_merge_key(msg: dict):
     return _session_message_key_with_sidecar((
         "legacy",
         str(msg.get("role") or ""),
-        str(msg.get("content") or ""),
+        _content_identity_for_key(msg.get("content")),
         _normalized_message_timestamp_for_key(msg.get("timestamp")),
         str(msg.get("tool_call_id") or ""),
         str(msg.get("tool_name") or msg.get("name") or ""),
@@ -9234,7 +9424,9 @@ def _copy_api_content_sidecar(target: dict | None, source: dict | None) -> bool:
         )
         if (
             target_mirror_key is None
-            or target_mirror_key != _session_message_multimodal_mirror_key(source)
+            or target_mirror_key != _session_message_multimodal_mirror_key(
+                source, require_scalar_mirror=True
+            )
         ):
             return False
     if target.get("api_content") not in (None, ""):
@@ -9674,7 +9866,7 @@ def _session_message_dedup_key(msg: dict):
     return _session_message_key_with_sidecar((
         "legacy",
         str(msg.get("role") or ""),
-        str(msg.get("content") or ""),
+        _content_identity_for_key(msg.get("content")),
         str(msg.get("timestamp") or ""),
         str(msg.get("tool_call_id") or ""),
         str(msg.get("tool_name") or msg.get("name") or ""),
@@ -9682,10 +9874,19 @@ def _session_message_dedup_key(msg: dict):
     ), msg)
 
 
-def _normalized_session_message_content(msg: dict) -> str:
+def _normalized_session_message_content(msg: dict):
+    """Visible identity for a message's content.
+
+    Scalars normalise whitespace as before. Structured content returns the same
+    out-of-band tuple as the merge/dedup keys, so content and visible keys can
+    never place a list and a string in the same identity space.
+    """
     if not isinstance(msg, dict):
         return repr(msg)
-    return " ".join(str(msg.get("content") or "").split())
+    content = msg.get("content")
+    if isinstance(content, list) and content:
+        return _content_identity_for_key(content)
+    return " ".join(str(content or "").split())
 
 
 def _loose_session_message_content(value: str) -> str:
@@ -9701,7 +9902,7 @@ def _session_message_content_key(
         return ("non_dict", repr(msg))
     role = str(msg.get("role") or "")
     content = _normalized_session_message_content(msg)
-    if role == "user" and normalize_workspace_prefix:
+    if role == "user" and normalize_workspace_prefix and isinstance(content, str):
         # WebUI sends the model a workspace-prefixed user_message
         # ("[Workspace::v1: /path]\n<text>") while the visible/optimistic
         # bubble and the WebUI sidecar row carry only the bare "<text>". The
@@ -9744,7 +9945,7 @@ def _session_message_visible_key(
     _tc_key = json.dumps(_tc, sort_keys=True, default=str) if _tc else ""
     role = str(msg.get("role") or "")
     content = _normalized_session_message_content(msg)
-    if role == "user" and normalize_workspace_prefix:
+    if role == "user" and normalize_workspace_prefix and isinstance(content, str):
         # state.db stores the model-facing workspace-prefixed prompt while the
         # WebUI sidecar owns the bare visible text. Fold that protocol wrapper
         # into the exact key so large-session reconciliation does not depend on
@@ -9769,7 +9970,9 @@ def _build_visible_duplicate_lookup(visible_keys: set[tuple]) -> dict:
             content = key[1]
         except (TypeError, IndexError):
             continue
-        if not content:
+        # Only text identities take part in fuzzy matching; structured content
+        # is exact-identity only and must never fuzzy-match a scalar.
+        if not content or not isinstance(content, str):
             continue
         by_role.setdefault(role, []).append(key)
     # Keep loose_by_key lazy.  Some transcripts contain multi-megabyte tool
@@ -9789,6 +9992,11 @@ def _matching_visible_duplicate(visible_key: tuple, visible_keys: set[tuple], lo
     sidecar = visible_key[3] if len(visible_key) > 3 else None
     if not content:
         return None
+    # Structured content is matched by exact identity only (checked above).
+    # Substring/token matching would otherwise compare a canonical list
+    # serialisation against arbitrary text and pair a rich row with a scalar.
+    if not isinstance(content, str):
+        return None
     # Exact identity above remains authoritative at every size. The fallback
     # below scans the existing keys for every candidate, so it becomes
     # quadratic on long transcripts even when each individual message is small.
@@ -9803,7 +10011,12 @@ def _matching_visible_duplicate(visible_key: tuple, visible_keys: set[tuple], lo
         existing_role = existing_key[0]
         existing_content = existing_key[1] if len(existing_key) > 1 else ""
         existing_sidecar = existing_key[3] if len(existing_key) > 3 else None
-        if role != existing_role or sidecar != existing_sidecar or not existing_content:
+        if (
+            role != existing_role
+            or sidecar != existing_sidecar
+            or not existing_content
+            or not isinstance(existing_content, str)
+        ):
             continue
         # Exact visible-key equality was checked above. For very large payloads
         # (tool logs / request dumps), Python-in substring and fuzzy-token
@@ -10138,6 +10351,30 @@ def merge_session_messages_append_only(
 ) -> list:
     """Merge sidecar/context and state.db messages without deleting local rows.
 
+    Thin wrapper that scopes the structured-content identity memo to this one
+    call; see :func:`_merge_session_messages_append_only_impl` for the merge.
+    """
+    token = _STRUCTURED_IDENTITY_MEMO.set({})
+    try:
+        return _merge_session_messages_append_only_impl(
+            sidecar_messages,
+            state_messages,
+            truncation_watermark=truncation_watermark,
+            truncation_boundary=truncation_boundary,
+        )
+    finally:
+        _STRUCTURED_IDENTITY_MEMO.reset(token)
+
+
+def _merge_session_messages_append_only_impl(
+    sidecar_messages: list,
+    state_messages: list,
+    *,
+    truncation_watermark=None,
+    truncation_boundary=None,
+) -> list:
+    """Merge sidecar/context and state.db messages without deleting local rows.
+
     ``truncation_boundary``: the original truncate cutoff — the
     timestamp of the last message kept by the truncate operation.  When the
     watermark is later advanced (new turn committed), this boundary is preserved
@@ -10228,7 +10465,16 @@ def merge_session_messages_append_only(
                 value = helper(msg)
                 # If this is a legacy message key, keep the already-stringified
                 # content payload for downstream helper calls.
-                if isinstance(value, tuple) and value and value[0] == "legacy":
+                # Structured content is never substituted: its key component is
+                # an identity token, not content, and writing it back would let
+                # a scalar equal to that token impersonate the rich row.
+                if (
+                    isinstance(value, tuple)
+                    and value
+                    and value[0] == "legacy"
+                    and isinstance(value[2], str)
+                    and not isinstance(msg.get("content"), list)
+                ):
                     prepared_msg = dict(msg)
                     prepared_msg["content"] = value[2]
                     _cached_msg_prepared[msg_cache_key] = prepared_msg
@@ -10243,13 +10489,20 @@ def merge_session_messages_append_only(
             prepared_msg = _cached_msg_prepared.get(msg_cache_key)
             if prepared_msg is None:
                 prepared_msg = dict(msg)
-                prepared_msg["content"] = (
-                    merge_key[2]
-                    if isinstance(merge_key, tuple)
+                raw_content = msg.get("content")
+                if isinstance(raw_content, list):
+                    # Keep the real structure; downstream keys derive their own
+                    # out-of-band identity from it.
+                    prepared_msg["content"] = raw_content
+                elif (
+                    isinstance(merge_key, tuple)
                     and len(merge_key) > 2
                     and merge_key[0] == "legacy"
-                    else str(msg.get("content") or "")
-                )
+                    and isinstance(merge_key[2], str)
+                ):
+                    prepared_msg["content"] = merge_key[2]
+                else:
+                    prepared_msg["content"] = str(raw_content or "")
                 _cached_msg_prepared[msg_cache_key] = prepared_msg
 
         value = helper(prepared_msg)
@@ -10392,7 +10645,9 @@ def merge_session_messages_append_only(
     if sidecar_multimodal_mirrors:
         state_multimodal_mirror_identities = {}
         for state_message in state_messages:
-            mirror_key = _session_message_multimodal_mirror_key(state_message)
+            mirror_key = _session_message_multimodal_mirror_key(
+                state_message, require_scalar_mirror=True
+            )
             state_multimodal_mirror_keys[id(state_message)] = mirror_key
             if (
                 mirror_key is not None

api/profiles.py

@@ -1711,19 +1711,22 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict:
     default_workspace = None
     try:
         from api.config import DEFAULT_WORKSPACE as _DW
+        from api.workspace import _resolve_path, _remote_terminal_workspace_candidate
         lw_file = home / 'webui_state' / 'last_workspace.txt'
         if lw_file.exists():
             _p = lw_file.read_text(encoding='utf-8').strip()
             if _p:
-                _pp = Path(_p).expanduser()
-                if _pp.is_dir():
-                    default_workspace = str(_pp.resolve())
+                _pp = _resolve_path(_p, profile=name)
+                remote_cand = _remote_terminal_workspace_candidate(_p, profile=name)
+                if remote_cand is not None or _pp.is_dir():
+                    default_workspace = str(_pp)
         if default_workspace is None:
             for _key in ('workspace', 'default_workspace'):
                 _v = cfg.get(_key)
                 if _v:
-                    _pp = Path(str(_v)).expanduser().resolve()
-                    if _pp.is_dir():
+                    _pp = _resolve_path(str(_v), profile=name)
+                    remote_cand = _remote_terminal_workspace_candidate(str(_v), profile=name)
+                    if remote_cand is not None or _pp.is_dir():
                         default_workspace = str(_pp)
                         break
         if default_workspace is None:
@@ -1731,8 +1734,9 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict:
             if isinstance(_tc, dict):
                 _cwd = _tc.get('cwd', '')
                 if _cwd and str(_cwd) not in ('.', ''):
-                    _pp = Path(str(_cwd)).expanduser().resolve()
-                    if _pp.is_dir():
+                    _pp = _resolve_path(str(_cwd), profile=name)
+                    remote_cand = _remote_terminal_workspace_candidate(str(_cwd), profile=name)
+                    if remote_cand is not None or _pp.is_dir():
                         default_workspace = str(_pp)
         if default_workspace is None:
             default_workspace = str(_DW)

api/route_approvals.py

@@ -265,6 +265,23 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
             live_local_tokens.add(live_entry_token)
             if not str(live_data.get("approval_id") or "").strip():
                 live_data["approval_id"] = f"gwlocal:{live_entry_token}"
+
+    # A raw id-less LOCAL entry in `_pending` (agent-side _pending_result drops
+    # `{command, pattern_key, pattern_keys, description}` verbatim when no
+    # gateway notifier is registered) is unactionable by contract: the frontend
+    # owner-capture requires an approval_id, so its card can neither be
+    # approved nor dismissed — the poll re-serves it forever. Mint a stable id
+    # on the entry itself (the same contract the producer-tokenization loop
+    # above applies to _gateway_queues producers). Minting on the stored dict
+    # keeps the id stable across polls, so frontend dismiss markers persist.
+    for entry in queue_list:
+        if not isinstance(entry, dict) or _is_gateway_mirror_entry(entry):
+            continue
+        if str(entry.get("approval_id") or "").strip():
+            continue
+        if str(entry.get("run_id") or "").strip():
+            continue
+        entry["approval_id"] = f"gwlocal-mirrorless:{uuid.uuid4().hex}"
     live_token = (
         _gateway_mirror_entry_token(live_head_entry)
         if live_head_entry and live_head_data

api/routes.py

@@ -13,6 +13,7 @@ import gzip
 import json
 from api.sse_chunked import end_sse_headers
 import logging
+import mimetypes
 import os
 import queue
 import re
@@ -10543,6 +10544,7 @@ from api.workspace import (
     safe_resolve_ws,
     raw_authorized_escape_target,
     resolve_trusted_workspace,
+    _resolve_path,
     resolve_implicit_workspace_with_recovery,
     open_anchored_fd,
     open_anchored_create_fd,
@@ -14159,22 +14161,39 @@ def handle_get(handler, parsed) -> bool:
         return _handle_session_export(handler, parsed)
 
     if parsed.path == "/api/workspaces":
+        from api.profiles import get_active_profile_name
+        active_profile = get_active_profile_name()
+        try:
+            wss = load_workspaces(profile=active_profile)
+        except TypeError:
+            wss = load_workspaces()
+        try:
+            lw = get_last_workspace(profile=active_profile)
+        except TypeError:
+            lw = get_last_workspace()
         return j(
             handler,
             {
-                "workspaces": load_workspaces(),
-                "last": get_last_workspace(),
+                "workspaces": wss,
+                "last": lw,
                 "terminal_remote_backend": _terminal_remote_backend_enabled(),
             },
         )
 
     if parsed.path == "/api/workspaces/suggest":
+        from api.profiles import get_active_profile_name
+
         qs = parse_qs(parsed.query)
         prefix = qs.get("prefix", [""])[0]
+        active_profile = get_active_profile_name()
+        try:
+            suggestions = list_workspace_suggestions(prefix, profile=active_profile)
+        except TypeError:
+            suggestions = list_workspace_suggestions(prefix)
         return j(
             handler,
             {
-                "suggestions": list_workspace_suggestions(prefix),
+                "suggestions": suggestions,
                 "prefix": prefix,
             },
         )
@@ -14625,7 +14644,10 @@ def handle_get(handler, parsed) -> bool:
         # profile-scoped via the per-request hermes_profile cookie set in server.py.
         # Fail open: a resolution error must never 500 this boot-critical endpoint.
         try:
-            _profile_default_workspace = get_profile_default_workspace()
+            try:
+                _profile_default_workspace = get_profile_default_workspace(profile=active_profile_name)
+            except TypeError:
+                _profile_default_workspace = get_profile_default_workspace()
         except Exception:
             logger.debug("Failed to resolve profile default workspace for /api/profile/active", exc_info=True)
             _profile_default_workspace = None
@@ -14848,26 +14870,41 @@ def _validate_session_toolsets_shape(toolsets):
     return toolsets
 
 
-def _resolve_new_session_workspace(body, visible_prev_session_id):
+def _resolve_new_session_workspace(body, visible_prev_session_id, profile=None):
     """Resolve a new-session workspace, recovering only verified inheritance."""
     candidate = body.get("workspace")
     if not candidate:
         return None
+
+    def _rtw(value):
+        # Legacy test doubles may predate the profile kwarg.
+        try:
+            return resolve_trusted_workspace(value, profile=profile)
+        except TypeError:
+            return resolve_trusted_workspace(value)
+
     if (
         body.get("workspace_inherited_from_prev_session") is not True
         or not visible_prev_session_id
     ):
-        return str(resolve_trusted_workspace(candidate))
+        return str(_rtw(candidate))
     try:
         previous_session = get_session(visible_prev_session_id, metadata_only=True)
     except KeyError:
-        return str(resolve_trusted_workspace(candidate))
+        return str(_rtw(candidate))
     if str(getattr(previous_session, "workspace", None) or "") != str(candidate):
-        return str(resolve_trusted_workspace(candidate))
-    workspace, _recovered = resolve_implicit_workspace_with_recovery(
-        candidate,
-        get_last_workspace,
-    )
+        return str(_rtw(candidate))
+    try:
+        workspace, _recovered = resolve_implicit_workspace_with_recovery(
+            candidate,
+            get_last_workspace,
+            profile=profile,
+        )
+    except TypeError:
+        workspace, _recovered = resolve_implicit_workspace_with_recovery(
+            candidate,
+            get_last_workspace,
+        )
     return str(workspace)
 
 def handle_post(handler, parsed) -> bool:
@@ -15209,7 +15246,9 @@ def handle_post(handler, parsed) -> bool:
         ):
             workspace_prev_session_id = None
         try:
-            workspace = _resolve_new_session_workspace(body, workspace_prev_session_id)
+            workspace = _resolve_new_session_workspace(
+                body, workspace_prev_session_id, profile=body.get("profile") or None
+            )
         except (TypeError, ValueError) as e:
             return bad(handler, str(e))
         worktree_info = None
@@ -15238,7 +15277,17 @@ def handle_post(handler, parsed) -> bool:
                 from api.worktrees import create_worktree_for_workspace
                 base_workspace = workspace
                 if not base_workspace:
-                    base_workspace = str(resolve_trusted_workspace(get_last_workspace()))
+                    _new_profile = body.get("profile") or None
+                    try:
+                        _lw = get_last_workspace(profile=_new_profile)
+                    except TypeError:
+                        _lw = get_last_workspace()
+                    try:
+                        base_workspace = str(
+                            resolve_trusted_workspace(_lw, profile=_new_profile)
+                        )
+                    except TypeError:
+                        base_workspace = str(resolve_trusted_workspace(_lw))
                 worktree_info = create_worktree_for_workspace(base_workspace)
                 workspace = worktree_info["path"]
             except (TypeError, ValueError) as e:
@@ -15803,7 +15852,7 @@ def handle_post(handler, parsed) -> bool:
         old_model = getattr(s, "model", None)
         old_provider = getattr(s, "model_provider", None)
         try:
-            new_ws = str(resolve_trusted_workspace(body.get("workspace", s.workspace)))
+            new_ws = str(resolve_trusted_workspace(body.get("workspace", s.workspace), profile=getattr(s, "profile", None)))
         except ValueError as e:
             return bad(handler, str(e))
         with _get_session_agent_lock(body["session_id"]):
@@ -15837,7 +15886,7 @@ def handle_post(handler, parsed) -> bool:
                 close_terminal(body["session_id"])
             except Exception:
                 logger.debug("Failed to close workspace terminal after workspace update")
-        set_last_workspace(new_ws)
+        set_last_workspace(new_ws, profile=getattr(s, "profile", None))
         return j(
             handler,
             {"session": public_session_projection(s.compact() | {"messages": s.messages})},
@@ -17025,14 +17074,16 @@ def handle_post(handler, parsed) -> bool:
             if _arch_source_tag == "subagent" or _is_subagent_child_session_id(sid):
                 return bad(handler, "Subagent sessions cannot be archived from WebUI", 400)
             if _is_messaging_session_record(cli_meta):
+                _arch_profile = cli_meta.get("profile") or None
                 s = Session(
                     session_id=sid,
                     title=cli_meta.get("title") or title_from(get_cli_session_messages(sid), "CLI Session"),
-                    workspace=get_last_workspace(),
+                    workspace=get_last_workspace(profile=_arch_profile),
                     messages=[],
                     model=cli_meta.get("model") or "unknown",
                     created_at=cli_meta.get("created_at"),
                     updated_at=cli_meta.get("updated_at"),
+                    profile=_arch_profile,
                 )
                 s.is_cli_session = is_cli_session_row(cli_meta)
                 s.source_tag = cli_meta.get("source_tag")
@@ -17710,6 +17761,9 @@ _STATIC_MIME = {
     "webp": "image/webp",
     "woff": "font/woff",
     "woff2": "font/woff2",
+    # Python's built-in MIME table does not include APK, and platform MIME
+    # databases are not consistent across Linux, macOS, and Windows.
+    "apk": "application/vnd.android.package-archive",
 }
 # MIME types that are text-based and should carry charset=utf-8
 _TEXT_MIME_TYPES = {"text/css", "application/javascript", "text/html", "image/svg+xml", "text/plain"}
@@ -17741,7 +17795,13 @@ def _serve_static(handler, parsed):
     if not static_file.exists() or not static_file.is_file():
         return j(handler, {"error": "not found"}, status=404)
     ext = static_file.suffix.lower()
-    ct = _STATIC_MIME.get(ext.lstrip("."), "text/plain")
+    ct = _STATIC_MIME.get(ext.lstrip("."))
+    if ct is None:
+        guessed_type, content_encoding = mimetypes.guess_type(static_file.name)
+        # Encoded suffixes (for example .svgz/.tgz) need Content-Encoding
+        # semantics this route does not implement. Fail closed instead of
+        # advertising the decoded media type for still-compressed bytes.
+        ct = guessed_type if guessed_type and not content_encoding else "application/octet-stream"
     ct_header = f"{ct}; charset=utf-8" if ct in _TEXT_MIME_TYPES else ct
 
     # Look up or populate the per-file cache (raw, optional gzip, ETag).
@@ -18009,15 +18069,26 @@ def _handle_list_dir(handler, parsed):
         except Exception:
             return bad(handler, "Session not found", 404)
     try:
+        _list_profile = getattr(webui_session, "profile", None)
         if webui_session is None:
-            workspace = resolve_trusted_workspace(workspace)
+            try:
+                workspace = resolve_trusted_workspace(workspace, profile=_list_profile)
+            except TypeError:
+                workspace = resolve_trusted_workspace(workspace)
             recovered = False
         else:
             stored_workspace = workspace
-            workspace, recovered = resolve_implicit_workspace_with_recovery(
-                stored_workspace,
-                get_last_workspace,
-            )
+            try:
+                workspace, recovered = resolve_implicit_workspace_with_recovery(
+                    stored_workspace,
+                    get_last_workspace,
+                    profile=_list_profile,
+                )
+            except TypeError:
+                workspace, recovered = resolve_implicit_workspace_with_recovery(
+                    stored_workspace,
+                    get_last_workspace,
+                )
             if recovered:
                 persisted = persist_recovered_workspace_binding(
                     webui_session,
@@ -19059,7 +19130,7 @@ def _handle_terminal_start(handler, body):
                 },
                 status=400,
             )
-        workspace = resolve_trusted_workspace(getattr(session, "workspace", "") or "")
+        workspace = resolve_trusted_workspace(getattr(session, "workspace", "") or "", profile=getattr(session, "profile", None))
         from api.terminal import start_terminal
         term = start_terminal(
             sid,
@@ -22043,10 +22114,11 @@ def _memory_project_context_workspace(parsed) -> Path | None:
             # fall through to Path("").resolve(), which returns the server's own
             # CWD and would surface the install's AGENTS.md/HERMES.md as if it
             # were the user's project context.
-            ws = (get_session(sid).workspace or "").strip()
+            session = get_session(sid)
+            ws = (session.workspace or "").strip()
             if not ws:
                 return None
-            return Path(ws).expanduser().resolve()
+            return _resolve_path(ws, profile=getattr(session, "profile", None))
         except Exception:
             return None
 
@@ -22054,7 +22126,7 @@ def _memory_project_context_workspace(parsed) -> Path | None:
     if not raw_workspace:
         return None
     try:
-        return Path(resolve_trusted_workspace(raw_workspace)).expanduser().resolve()
+        return resolve_trusted_workspace(raw_workspace)
     except Exception:
         logger.debug("Skipping project context for untrusted workspace %s", raw_workspace, exc_info=True)
         return None
@@ -22824,7 +22896,7 @@ def _start_regeneration_stream_locked(
         save_attempted = True
         s.save()
         accepted = True
-        set_last_workspace(workspace)
+        set_last_workspace(workspace, profile=getattr(s, "profile", None))
         release_worker.set()
     except Exception as exc:
         abort_worker.set()
@@ -23163,7 +23235,7 @@ def _start_chat_stream_for_session(
     except Exception:
         logger.warning("Failed to append submitted turn journal event", exc_info=True)
     diag.stage("set_last_workspace") if diag else None
-    set_last_workspace(workspace)
+    set_last_workspace(workspace, profile=getattr(s, "profile", None))
     diag.stage("stream_registration") if diag else None
     stream = create_stream_channel()
     register_stream_owner(stream_id, s.session_id)
@@ -23876,7 +23948,7 @@ def _handle_goal_command(handler, body):
     previous_goal_state = None
     if will_kickoff:
         try:
-            workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace))
+            workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace, profile=getattr(s, "profile", None)))
         except ValueError as e:
             return bad(handler, str(e))
         requested_model = body.get("model") or s.model
@@ -23945,7 +24017,7 @@ def _handle_goal_command(handler, body):
     if kickoff_prompt:
         if workspace is None:
             try:
-                workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace))
+                workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace, profile=getattr(s, "profile", None)))
             except ValueError as e:
                 return bad(handler, str(e))
         if model is None:
@@ -24326,14 +24398,25 @@ def _handle_chat_start(handler, body, diag=None):
 
 def _resolve_chat_workspace_with_recovery(s, requested_workspace) -> str:
     """Recover stale implicit session workspaces without hiding explicit errors."""
+    _session_profile = getattr(s, "profile", None)
     explicit = requested_workspace not in (None, "")
     if explicit:
-        return str(resolve_trusted_workspace(requested_workspace))
+        try:
+            return str(resolve_trusted_workspace(requested_workspace, profile=_session_profile))
+        except TypeError:
+            return str(resolve_trusted_workspace(requested_workspace))
     stored_workspace = getattr(s, "workspace", None)
-    workspace, recovered = resolve_implicit_workspace_with_recovery(
-        stored_workspace,
-        get_last_workspace,
-    )
+    try:
+        workspace, recovered = resolve_implicit_workspace_with_recovery(
+            stored_workspace,
+            get_last_workspace,
+            profile=_session_profile,
+        )
+    except TypeError:
+        workspace, recovered = resolve_implicit_workspace_with_recovery(
+            stored_workspace,
+            get_last_workspace,
+        )
     if not recovered:
         return str(workspace)
     persisted = persist_recovered_workspace_binding(
@@ -24346,12 +24429,23 @@ def _resolve_chat_workspace_with_recovery(s, requested_workspace) -> str:
 
 def _resolve_chat_workspace_for_regeneration(s, requested_workspace) -> str:
     """Resolve regeneration's workspace without persisting before start acceptance."""
+    _session_profile = getattr(s, "profile", None) or None
     if requested_workspace not in (None, ""):
-        return str(resolve_trusted_workspace(requested_workspace))
-    workspace, _recovered = resolve_implicit_workspace_with_recovery(
-        getattr(s, "workspace", None),
-        get_last_workspace,
-    )
+        try:
+            return str(resolve_trusted_workspace(requested_workspace, profile=_session_profile))
+        except TypeError:
+            return str(resolve_trusted_workspace(requested_workspace))
+    try:
+        workspace, _recovered = resolve_implicit_workspace_with_recovery(
+            getattr(s, "workspace", None),
+            get_last_workspace,
+            profile=_session_profile,
+        )
+    except TypeError:
+        workspace, _recovered = resolve_implicit_workspace_with_recovery(
+            getattr(s, "workspace", None),
+            get_last_workspace,
+        )
     return str(workspace)
 
 
@@ -24397,7 +24491,10 @@ def _handle_chat_sync(handler, body):
     if not msg:
         return j(handler, {"error": "empty message"}, status=400)
     try:
-        workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace))
+        try:
+            workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace, profile=getattr(s, "profile", None)))
+        except TypeError:
+            workspace = str(resolve_trusted_workspace(body.get("workspace") or s.workspace))
     except ValueError as e:
         return bad(handler, str(e))
     with _get_session_agent_lock(s.session_id):
@@ -25780,38 +25877,49 @@ def _handle_workspace_add(handler, body):
     # macOS) so pytest's tmp_path_factory paths and other legit user-tmp dirs
     # still register cleanly.
     try:
-        candidate = Path(path_str).expanduser().resolve()
+        from api.workspace import _remote_terminal_workspace_candidate, _resolve_path
+        from api.profiles import get_active_profile_name
+        active_profile = get_active_profile_name()
+        remote_candidate = _remote_terminal_workspace_candidate(path_str, profile=active_profile)
+        candidate = _resolve_path(path_str, profile=active_profile)
     except (ValueError, OSError, RuntimeError) as e:
         # Invalid path (e.g. embedded null byte) — fail closed with a clean 400
         # instead of letting .resolve() raise an uncaught 500.
         return bad(handler, f"Invalid path: {_sanitize_error(e)}")
-    if _is_blocked_system_path(candidate):
-        # Home-directory carve-out, mirroring the validators
-        # (resolve_trusted_workspace / validate_workspace_to_add): a workspace
-        # at or under the active user's home must stay allowed even when that
-        # home lives under an otherwise-blocked root (e.g. systemd-homed
-        # /var/home/<user>/...). Without this the route rejects valid
-        # /var/home workspaces before validate_workspace_to_add()'s carve-out
-        # can run.
-        _home = _home_path()
-        if not (_home != Path("/") and (candidate == _home or _is_within(candidate, _home))):
-            return bad(handler, f"Path points to a system directory: {candidate}")
-    # Now safe to create the directory if requested
-    if auto_create:
-        try:
-            candidate.mkdir(parents=True, exist_ok=True)
-        except (OSError, PermissionError) as e:
-            return bad(handler, f"Could not create directory: {_sanitize_error(e)}")
+    if remote_candidate is None:
+        if _is_blocked_system_path(candidate):
+            # Home-directory carve-out, mirroring the validators
+            # (resolve_trusted_workspace / validate_workspace_to_add): a workspace
+            # at or under the active user's home must stay allowed even when that
+            # home lives under an otherwise-blocked root (e.g. systemd-homed
+            # /var/home/<user>/...). Without this the route rejects valid
+            # /var/home workspaces before validate_workspace_to_add()'s carve-out
+            # can run.
+            _home = _home_path()
+            if not (_home != Path("/") and (candidate == _home or _is_within(candidate, _home))):
+                return bad(handler, f"Path points to a system directory: {candidate}")
+        # Now safe to create the directory if requested
+        if auto_create:
+            try:
+                candidate.mkdir(parents=True, exist_ok=True)
+            except (OSError, PermissionError) as e:
+                return bad(handler, f"Could not create directory: {_sanitize_error(e)}")
     # Full validation (exists, is_dir) — should pass now that dir exists
     try:
-        p = validate_workspace_to_add(path_str)
+        p = validate_workspace_to_add(path_str, profile=active_profile)
     except ValueError as e:
         return bad(handler, str(e))
-    wss = load_workspaces()
+    try:
+        wss = load_workspaces(profile=active_profile)
+    except TypeError:
+        wss = load_workspaces()
     if any(w["path"] == str(p) for w in wss):
         return bad(handler, "Workspace already in list")
     wss.append({"path": str(p), "name": name or p.name})
-    save_workspaces(wss)
+    try:
+        save_workspaces(wss, profile=active_profile)
+    except TypeError:
+        save_workspaces(wss)
     return j(handler, {"ok": True, "workspaces": wss})
 
 
@@ -25819,9 +25927,17 @@ def _handle_workspace_remove(handler, body):
     path_str = body.get("path", "").strip()
     if not path_str:
         return bad(handler, "path is required")
-    wss = load_workspaces()
+    from api.profiles import get_active_profile_name
+    active_profile = get_active_profile_name()
+    try:
+        wss = load_workspaces(profile=active_profile)
+    except TypeError:
+        wss = load_workspaces()
     wss = [w for w in wss if w["path"] != path_str]
-    save_workspaces(wss)
+    try:
+        save_workspaces(wss, profile=active_profile)
+    except TypeError:
+        save_workspaces(wss)
     return j(handler, {"ok": True, "workspaces": wss})
 
 
@@ -25830,14 +25946,22 @@ def _handle_workspace_rename(handler, body):
     name = body.get("name", "").strip()
     if not path_str or not name:
         return bad(handler, "path and name are required")
-    wss = load_workspaces()
+    from api.profiles import get_active_profile_name
+    active_profile = get_active_profile_name()
+    try:
+        wss = load_workspaces(profile=active_profile)
+    except TypeError:
+        wss = load_workspaces()
     for w in wss:
         if w["path"] == path_str:
             w["name"] = name
             break
     else:
         return bad(handler, "Workspace not found", 404)
-    save_workspaces(wss)
+    try:
+        save_workspaces(wss, profile=active_profile)
+    except TypeError:
+        save_workspaces(wss)
     return j(handler, {"ok": True, "workspaces": wss})
 
 
@@ -25851,7 +25975,12 @@ def _handle_workspace_reorder(handler, body):
     paths = body.get("paths", [])
     if not paths or not isinstance(paths, list):
         return bad(handler, "paths is required and must be a list")
-    wss = load_workspaces()
+    from api.profiles import get_active_profile_name
+    active_profile = get_active_profile_name()
+    try:
+        wss = load_workspaces(profile=active_profile)
+    except TypeError:
+        wss = load_workspaces()
     by_path = {w["path"]: w for w in wss}
     # Build reordered list: given order first, then any omitted entries
     reordered = []
@@ -25865,7 +25994,11 @@ def _handle_workspace_reorder(handler, body):
     for w in wss:
         if w["path"] not in seen:
             reordered.append(w)
-    save_workspaces(reordered)
+    try:
+        save_workspaces(reordered, profile=active_profile)
+    except TypeError:
+        # Legacy signature (test doubles with single-arg lambdas, older forks).
+        save_workspaces(reordered)
     return j(handler, {"ok": True, "workspaces": reordered})
 
 
@@ -28381,7 +28514,7 @@ def _handle_session_import_cli(handler, body):
         session_payload = {
             "session_id": sid,
             "title": title,
-            "workspace": str(get_last_workspace()),
+            "workspace": str(get_last_workspace(profile=profile)),
             "model": model,
             "message_count": len(msgs),
             "created_at": created_at,

api/streaming.py

@@ -2556,7 +2556,7 @@ def _aiagent_import_error_detail() -> str:
     lines.append('  Full troubleshooting: docs/troubleshooting.md ("AIAgent not available")')
     return "\n".join(lines)
 from api.models import get_session, title_from
-from api.workspace import set_last_workspace
+from api.workspace import _resolve_path
 
 # Fields that are safe to send to LLM provider APIs.
 # Everything else (attachments, timestamp, _ts, etc.) is display-only
@@ -3385,7 +3385,7 @@ def _resolve_image_input_mode(cfg: dict, active_provider: str = "", active_model
     return "native"
 
 
-def _build_native_multimodal_message(workspace_ctx: str, msg_text: str, attachments, workspace: str, *, cfg: dict = None, active_provider: str = "", active_model: str = "", requested_provider: str = ""):
+def _build_native_multimodal_message(workspace_ctx: str, msg_text: str, attachments, workspace: str, *, cfg: dict = None, active_provider: str = "", active_model: str = "", requested_provider: str = "", profile: str | Path | None = None):
     """Build native multimodal content parts for current-turn image uploads.
 
     WebUI uploads files into the active workspace. For image files, pass the
@@ -3405,7 +3405,7 @@ def _build_native_multimodal_message(workspace_ctx: str, msg_text: str, attachme
         return workspace_ctx + msg_text
 
     parts = [{'type': 'text', 'text': workspace_ctx + msg_text}]
-    workspace_root = Path(workspace).expanduser().resolve()
+    workspace_root = _resolve_path(workspace, profile=profile)
     # Stage-361 maintainer fix (Opus SHOULD-FIX): chat uploads from #2319 now
     # land in ~/.hermes/webui/attachments/<sid>/ (outside workspace_root by
     # design). The pre-existing `path.relative_to(workspace_root)` guard would
@@ -4011,11 +4011,21 @@ def _looks_like_current_user_turn(msg, msg_text) -> bool:
     return any(" ".join(str(candidate or '').split()) == needle for candidate in candidates)
 
 
-def _first_exchange_snippets(messages):
+def _first_exchange_snippets(messages, *, scan_past_consecutive_users: bool = False):
     """Return (first_user_text, first_assistant_text) snippets for title generation.
 
     Prefer the first substantive assistant answer in the opening exchange,
     skipping empty placeholders and assistant tool-call preambles.
+
+    ``scan_past_consecutive_users`` (opt-in) keeps scanning past consecutive
+    opening user rows (queued first turns) until the first COMPLETE user+assistant
+    pair — needed by the manual "Regenerate title" path (#7543), which otherwise
+    aborted at the second user row with an empty assistant snippet and silently
+    persisted the local fallback. It defaults False so the automatic in-stream
+    background-title path keeps its exact prior behavior (a transcript with no
+    assistant text before the second user row yields an empty assistant snippet,
+    which _background_title_generation_inputs treats as "not yet eligible" — the
+    stream teardown then emits stream_end on its synchronous path unchanged).
     """
     user_text = ''
     asst_text = ''
@@ -4025,11 +4035,14 @@ def _first_exchange_snippets(messages):
         role = m.get('role')
         if role == 'user':
             candidate = _strip_thinking_markup(_title_exchange_input_text(m.get('content')))
-            if not user_text and candidate:
+            if candidate and not user_text:
                 user_text = candidate
-                continue
-            if user_text and candidate:
+            elif user_text and candidate and not scan_past_consecutive_users:
+                # Legacy/default behavior: a second populated user row before any
+                # assistant text ends the opening exchange (asst_text stays empty).
                 break
+            # When scan_past_consecutive_users is set, keep going past consecutive
+            # user rows (#7543) until the first user+assistant pair.
         elif role == 'assistant' and user_text:
             candidate = _message_text(m.get('content'))
             # Skip tool-call preambles *only* when content is empty or looks
@@ -5098,7 +5111,12 @@ def generate_session_title_for_session(session, *, prefer_latest: bool = False,
     if prefer_latest:
         user_text, assistant_text = _latest_exchange_snippets(messages)
     else:
-        user_text, assistant_text = _first_exchange_snippets(messages)
+        # Manual "Regenerate title" (#7543): scan past consecutive opening user
+        # rows to the first complete user+assistant pair, so a transcript that
+        # opens with queued user turns still reaches the aux LLM instead of
+        # silently persisting the local fallback. The automatic in-stream path
+        # keeps the default (no scan-past) so its stream teardown is unchanged.
+        user_text, assistant_text = _first_exchange_snippets(messages, scan_past_consecutive_users=True)
     if not user_text:
         return None, 'empty_user_message', ''
     from api import profiles as profiles_api
@@ -9448,19 +9466,19 @@ def _run_agent_streaming(
         # Resolved the same way as `s.workspace` below so the bound cwd and the
         # session's own record cannot disagree. Guarded: a turn must not die
         # here because a workspace path is malformed.
+        s = get_session(session_id)
         try:
-            _turn_workspace_cwd = str(Path(workspace).expanduser().resolve())
+            _turn_workspace_cwd = str(_resolve_path(workspace, profile=getattr(s, 'profile', None)))
         except Exception:
             _turn_workspace_cwd = ""
             logger.debug("per-turn workspace cwd resolve failed", exc_info=True)
         _turn_session_identity_tokens = _set_turn_session_identity(
             session_id, workspace=_turn_workspace_cwd
         )
-        s = get_session(session_id)
         _turn_pending_source = getattr(s, 'pending_user_source', None) or 'webui'
         _active_turn_identity = _active_turn_authority(s, stream_id, msg_text)
         update_active_run(stream_id, phase="running", session_id=session_id)
-        s.workspace = str(Path(workspace).expanduser().resolve())
+        s.workspace = _turn_workspace_cwd
         _last_persisted_model = None
         _last_persisted_provider = None
         _turn_owns_persisted_model = False
@@ -10988,7 +11006,7 @@ def _run_agent_streaming(
             _agent_msg_text = msg_text
             if _process_notifications:
                 _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 ""))
+            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)
             _run_conversation_kwargs = _build_run_conversation_kwargs(
                 agent.run_conversation,
@@ -11041,6 +11059,10 @@ def _run_agent_streaming(
                     active_provider=(resolved_provider or ""),
                     active_model=(resolved_model or ""),
                     requested_provider=(_session_requested_provider or ""),
+                    # Legacy fallback wraps the home STRING in Path: string
+                    # profiles are logical ids only (round-5 grammar gate),
+                    # explicit homes must arrive as Path values.
+                    profile=(getattr(s, "profile", None) or Path(_profile_home)),
                 )
                 _run_conversation_kwargs["user_message"] = user_message
             _result_partial_pre_call_context = list(_previous_context_messages)

api/upload.py

@@ -122,8 +122,8 @@ def _attachment_root() -> Path:
     return (STATE_DIR / 'attachments').resolve()
 
 
-def _upload_destination(session_id: str, safe_name: str) -> Path:
-    dest_dir = _session_attachment_dir(session_id)
+def _upload_destination(session_id: str, safe_name: str, dest_dir: Path | None = None) -> Path:
+    dest_dir = dest_dir if dest_dir is not None else _session_attachment_dir(session_id)
     dest_dir.mkdir(parents=True, exist_ok=True)
     dest = (dest_dir / safe_name).resolve()
     if not dest.is_relative_to(dest_dir):
@@ -224,8 +224,20 @@ def handle_upload(handler):
         if _reject_invisible_session(handler, s):
             return True
         safe_name = _sanitize_upload_name(filename)
-        dest = _upload_destination(session_id, safe_name)
-        dest.write_bytes(file_bytes)
+        dest_dir = _session_attachment_dir(session_id)
+        dest = _upload_destination(session_id, safe_name, dest_dir)
+        # #3398-style TOCTOU hardening, mirrored from the workspace upload
+        # path: O_CREAT|O_EXCL|O_NOFOLLOW anchored open so a raced duplicate
+        # cannot be silently overwritten and a raced symlink subpath cannot
+        # redirect the write outside the attachment dir.
+        try:
+            _wfd = open_anchored_create_fd(dest_dir, dest)
+        except FileExistsError:
+            return j(handler, {'error': f'Upload destination already exists: {safe_name}'}, status=409)
+        except (ValueError, OSError):
+            return j(handler, {'error': 'Upload destination rejected'}, status=403)
+        with os.fdopen(_wfd, 'wb', closefd=True) as _wfh:
+            _wfh.write(file_bytes)
         mime = mimetypes.guess_type(safe_name)[0] or 'application/octet-stream'
         return j(handler, {
             'filename': dest.name,
@@ -623,8 +635,13 @@ def handle_workspace_upload(handler):
         if _reject_invisible_session(handler, session):
             return True
 
-        # Resolve workspace root from session
-        workspace = resolve_trusted_workspace(session.workspace)
+        # Resolve workspace root using the session profile, not the ambient request profile.
+        try:
+            workspace = resolve_trusted_workspace(
+                session.workspace, profile=getattr(session, "profile", None)
+            )
+        except TypeError:
+            workspace = resolve_trusted_workspace(session.workspace)
 
         # Resolve target subdirectory within workspace
         target_dir = safe_resolve_ws(workspace, subpath) if subpath else workspace

api/workspace.py

@@ -12,6 +12,7 @@ import json
 import logging
 import os
 import posixpath
+import re
 import secrets
 import shutil
 import stat
@@ -40,8 +41,14 @@ from api.subprocess_utils import windows_hide_flags
 
 # ── Profile-aware path resolution ───────────────────────────────────────────
 
-def _profile_state_dir() -> Path:
-    """Return the webui_state directory for the active profile.
+# Logical profile-name grammar — mirrors api.profiles._PROFILE_ID_RE. Kept as
+# a local copy so workspace-layer validation does not import profiles at module
+# load time (profiles may not be importable in every embedding context).
+_PROFILE_NAME_RE = re.compile(r'^[a-z0-9][a-z0-9_-]{0,63}$')
+
+
+def _profile_state_dir(profile: str | Path | None = None) -> Path:
+    """Return the webui_state directory for the active or given profile.
 
     For the default profile, returns the global STATE_DIR (respects
     HERMES_WEBUI_STATE_DIR env var for test isolation).
@@ -49,6 +56,28 @@ def _profile_state_dir() -> Path:
     """
     try:
         from api.profiles import get_active_profile_name, get_active_hermes_home
+        if profile is not None:
+            # Literal-"default" STATE routing (#7168 re-gate round 7): the
+            # default profile's workspace state always lives in the global
+            # state files, even when isolated mode pins the default home at
+            # <base>/profiles/default. The round-6 resolver change made
+            # _resolve_profile_home_param("default") return that pinned home,
+            # so the canonical-home check below sent explicit
+            # profile="default" state reads/writes to {pinned}/webui_state/
+            # while ambient calls kept using the global dir — splitting saved
+            # workspaces between two authorities. Config and workspace PATH
+            # resolution still use the pinned home via
+            # _resolve_profile_home_param; only this state-file tier stays
+            # global for the logical string "default".
+            if isinstance(profile, str) and profile.strip() == 'default':
+                return _GLOBAL_WS_FILE.parent
+            profile_home = _resolve_profile_home_param(profile)
+            if not _is_default_profile_home(profile_home):
+                d = profile_home / 'webui_state'
+                d.mkdir(parents=True, exist_ok=True)
+                return d
+            return _GLOBAL_WS_FILE.parent
+
         name = get_active_profile_name()
         if name and name != 'default':
             d = get_active_hermes_home() / 'webui_state'
@@ -59,14 +88,41 @@ def _profile_state_dir() -> Path:
     return _GLOBAL_WS_FILE.parent
 
 
-def _workspaces_file() -> Path:
-    """Return the workspaces.json path for the active profile."""
-    return _profile_state_dir() / 'workspaces.json'
+def _workspaces_file(profile: str | Path | None = None) -> Path:
+    """Return the workspaces.json path for the active or given profile."""
+    return _profile_state_dir(profile=profile) / 'workspaces.json'
 
 
-def _last_workspace_file() -> Path:
-    """Return the last_workspace.txt path for the active profile."""
-    return _profile_state_dir() / 'last_workspace.txt'
+def _last_workspace_file(profile: str | Path | None = None) -> Path:
+    """Return the last_workspace.txt path for the active or given profile."""
+    return _profile_state_dir(profile=profile) / 'last_workspace.txt'
+
+
+def _workspaces_file_for_profile(profile: str | Path | None = None) -> Path | None:
+    """Profile-scoped workspaces.json path, or None for an INVALID profile.
+
+    ``None`` is the fail-closed contract (#7168 re-gate round 4): a malformed
+    profile name must not be clamped onto the default/global state files, so
+    callers treat it as "no readable/writable profile-local state".
+    """
+    try:
+        return _workspaces_file(profile=profile) if profile is not None else _workspaces_file()
+    except TypeError:
+        return _workspaces_file()
+    except ValueError:
+        logger.debug("Ignoring invalid profile name %r for workspaces file", profile)
+        return None
+
+
+def _last_workspace_file_for_profile(profile: str | Path | None = None) -> Path | None:
+    """Profile-scoped last_workspace.txt path, or None for an INVALID profile."""
+    try:
+        return _last_workspace_file(profile=profile) if profile is not None else _last_workspace_file()
+    except TypeError:
+        return _last_workspace_file()
+    except ValueError:
+        logger.debug("Ignoring invalid profile name %r for last-workspace file", profile)
+        return None
 
 
 def _expanduser_path(path: str | Path) -> Path:
@@ -101,8 +157,11 @@ def _expanduser_path(path: str | Path) -> Path:
     return Path(raw)
 
 
-def _resolve_path(path: str | Path) -> Path:
-    """Resolve *path* after env-aware home expansion, without raising."""
+def _resolve_path(path: str | Path, profile: str | Path | None = None) -> Path:
+    """Resolve *path* after env-aware home expansion, preserving remote POSIX paths without raising."""
+    remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile)
+    if remote_candidate is not None:
+        return remote_candidate
     return _safe_resolve(_expanduser_path(path))
 
 
@@ -148,12 +207,75 @@ def _is_remote_terminal_backend(terminal_cfg: dict | None) -> bool:
     return backend not in ('', 'local')
 
 
-def _remote_terminal_cwd() -> str | None:
-    """Return target-side terminal cwd for remote profiles, without local stat()."""
+def _resolve_profile_home_param(profile: str | Path | None) -> Path:
+    """Resolve a profile parameter (name string, directory path string, or Path) to a profile home Path.
+
+    Logical profile names are validated strictly (#7168 re-gate round 4): a
+    name that fails the profile-id grammar raises ValueError instead of being
+    silently clamped onto the default home — the old clamp let a malformed
+    name such as ``"bad name"`` read/write the DEFAULT profile's state.
+    Round 5 tightens the grammar gate: a STRING profile value is strictly a
+    logical profile id and is NEVER treated as a path-shaped home — the old
+    ``"/" in raw`` branch resolved any slash-bearing string directly, so a
+    malformed value like ``"../evil"`` bypassed validation entirely and could
+    read/overwrite an arbitrary ``webui_state/last_workspace.txt``
+    (#7168 re-gate round 5, path traversal on the profile-isolation boundary).
+    Round 6 closes the last isolation hole in this resolver: the logical
+    string ``"default"`` used to short-circuit to ``_DEFAULT_HERMES_HOME``
+    before reaching ``get_hermes_home_for_profile()``, bypassing the
+    isolated-mode clamp in ``api.profiles._resolve_profile_home_for_name``
+    (#7168 re-gate round 6). In an isolated deployment pinned at
+    ``<base>/profiles/default``, a session created with ``profile="default"``
+    therefore resolved workspace/config from the BASE root home instead of
+    the pinned one. The name now flows through the same delegated path as
+    every other logical id; literal-default routing to the global state
+    files is retained in ``_profile_state_dir``/``get_last_workspace`` via
+    canonical ``_is_default_profile_home`` identity.
+    An explicit home directory is expressed as a ``Path`` object (callers such
+    as streaming's legacy ``_profile_home`` fallback wrap their home strings
+    in ``Path``); Path values are honored and canonicalized so identity
+    comparisons never depend on lexical spelling (e.g. a symlink alias of the
+    default home must compare equal to it).
+    """
+    if profile is None or str(profile).strip() == "":
+        from api.profiles import get_active_hermes_home
+        return get_active_hermes_home()
+
+    raw = str(profile).strip()
+
+    if isinstance(profile, Path):
+        return _safe_resolve(profile.expanduser())
+
+    # Strings are LOGICAL PROFILE IDS ONLY — no path-shaped strings, ever.
+    if not _PROFILE_NAME_RE.fullmatch(raw):
+        raise ValueError(f"invalid profile name: {raw!r}")
+
+    from api.profiles import get_hermes_home_for_profile
+    return _safe_resolve(get_hermes_home_for_profile(raw))
+
+
+def _is_default_profile_home(profile_home: Path) -> bool:
+    """Canonical identity check against the root/default Hermes home.
+
+    Compares resolved paths so a symlink alias of _DEFAULT_HERMES_HOME is
+    recognized as the default profile rather than treated as a foreign,
+    lexically-different directory (#7168 re-gate round 4).
+    """
     try:
-        from api.config import get_config
+        from api.profiles import _DEFAULT_HERMES_HOME
+        return _safe_resolve(profile_home) == _safe_resolve(_DEFAULT_HERMES_HOME)
+    except Exception:
+        return False
+
+
+def _remote_terminal_cwd(profile: str | Path | None = None) -> str | None:
+    """Return target-side terminal cwd for a remote profile, without local stat()."""
+    try:
+        from api.config import get_config_for_profile_home
+
+        profile_home = _resolve_profile_home_param(profile)
+        terminal_cfg = get_config_for_profile_home(profile_home).get('terminal', {})
 
-        terminal_cfg = get_config().get('terminal', {})
         if not _is_remote_terminal_backend(terminal_cfg):
             return None
         cwd = str(terminal_cfg.get('cwd') or '').strip()
@@ -165,8 +287,8 @@ def _remote_terminal_cwd() -> str | None:
         return None
 
 
-def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None:
-    """Return a non-stat'ed target-side Path when it is under terminal.cwd.
+def _remote_terminal_workspace_candidate(path: str | Path, profile: str | Path | None = None) -> Path | None:
+    """Return a non-stat'ed target-side Path when it is under terminal.cwd for the given profile.
 
     Remote workspace paths live on the target host (e.g., remote SSH/Docker
     backend). For valid target-side POSIX paths under ``terminal.cwd``, the
@@ -174,7 +296,10 @@ def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None:
     local host-filesystem resolution (avoiding host-specific firmlink rewriting
     such as macOS synthetic ``/home`` -> ``/System/Volumes/Data/home``).
     """
-    cwd = _remote_terminal_cwd()
+    try:
+        cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd()
+    except TypeError:
+        cwd = _remote_terminal_cwd()
     if not cwd:
         return None
     raw = _strip_surrounding_quotes(str(path)).strip()
@@ -192,8 +317,8 @@ def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None:
         if posix_candidate == posix_base or _posix_is_within(posix_candidate, posix_base):
             return Path(normalized_raw)
         return None
-    candidate = _resolve_path(raw)
-    base = _resolve_path(cwd)
+    candidate = _safe_resolve(_expanduser_path(raw))
+    base = _safe_resolve(_expanduser_path(cwd))
     if _is_blocked_workspace_path(candidate, raw) or _is_blocked_workspace_path(base, cwd):
         return None
     if candidate == base or _is_within(candidate, base):
@@ -201,7 +326,7 @@ def _remote_terminal_workspace_candidate(path: str | Path) -> Path | None:
     return None
 
 
-def _profile_default_workspace() -> str:
+def _profile_default_workspace(profile: str | Path | None = None) -> str:
     """Read the profile's default workspace from its config.yaml.
 
     Checks keys in priority order:
@@ -217,8 +342,9 @@ def _profile_default_workspace() -> str:
     Falls back to the live DEFAULT_WORKSPACE from api.config.
     """
     try:
-        from api.config import get_config
-        cfg = get_config()
+        from api.config import get_config_for_profile_home
+        profile_home = _resolve_profile_home_param(profile)
+        cfg = get_config_for_profile_home(profile_home)
         terminal_cfg = cfg.get('terminal', {})
         remote_terminal = _is_remote_terminal_backend(terminal_cfg)
         # Explicit webui workspace keys first
@@ -227,7 +353,7 @@ def _profile_default_workspace() -> str:
             if ws:
                 if remote_terminal:
                     return str(ws).strip()
-                p = _resolve_path(str(ws))
+                p = _resolve_path(str(ws), profile=profile)
                 if remote_terminal or p.is_dir():
                     return str(p)
         # Fall through to terminal.cwd — the agent's configured working directory
@@ -236,7 +362,7 @@ def _profile_default_workspace() -> str:
             if cwd and str(cwd) not in ('.', ''):
                 if remote_terminal:
                     return str(cwd).strip()
-                p = _resolve_path(str(cwd))
+                p = _resolve_path(str(cwd), profile=profile)
                 if remote_terminal or p.is_dir():
                     return str(p)
     except (ImportError, Exception):
@@ -244,14 +370,14 @@ def _profile_default_workspace() -> str:
     try:
         from api.config import DEFAULT_WORKSPACE as _LIVE_DEFAULT_WORKSPACE
 
-        return str(_resolve_path(_LIVE_DEFAULT_WORKSPACE))
+        return str(_resolve_path(_LIVE_DEFAULT_WORKSPACE, profile=profile))
     except Exception:
-        return str(_resolve_path(_BOOT_DEFAULT_WORKSPACE))
+        return str(_resolve_path(_BOOT_DEFAULT_WORKSPACE, profile=profile))
 
 
 # ── Public API ──────────────────────────────────────────────────────────────
 
-def _clean_workspace_list(workspaces: list) -> list:
+def _clean_workspace_list(workspaces: list, profile: str | Path | None = None) -> list:
     """Sanitize a workspace list:
     - Preserve target-side remote terminal workspace paths (SSH/Docker) without
       resolving them against the local WebUI host filesystem.
@@ -270,7 +396,7 @@ def _clean_workspace_list(workspaces: list) -> list:
         name = w.get('name', '')
         if not path:
             continue
-        remote_cand = _remote_terminal_workspace_candidate(path)
+        remote_cand = _remote_terminal_workspace_candidate(path, profile=profile)
         if remote_cand is not None:
             p = remote_cand
         else:
@@ -283,7 +409,14 @@ def _clean_workspace_list(workspaces: list) -> list:
             # p is under ~/.hermes/profiles/ — only skip if it's under a DIFFERENT profile
             try:
                 from api.profiles import get_active_hermes_home
-                own_profile_dir = get_active_hermes_home().resolve()
+                if profile is not None:
+                    # Explicit profile wins: the list belongs to that profile,
+                    # so "own" is defined by the profile parameter, never by the
+                    # ambient home (loading profile A's list under ambient B must
+                    # not silently drop A's own workspaces).
+                    own_profile_dir = _resolve_profile_home_param(profile).resolve()
+                else:
+                    own_profile_dir = get_active_hermes_home().resolve()
                 p.relative_to(own_profile_dir)
                 # p is under our own profile dir — keep it
             except (ValueError, Exception):
@@ -350,12 +483,12 @@ def _migrate_global_workspaces() -> list:
         return []
 
 
-def load_workspaces() -> list:
-    ws_file = _workspaces_file()
-    if ws_file.exists():
+def load_workspaces(profile: str | Path | None = None) -> list:
+    ws_file = _workspaces_file_for_profile(profile)
+    if ws_file is not None and ws_file.exists():
         try:
             raw = json.loads(ws_file.read_text(encoding='utf-8'))
-            cleaned = _clean_workspace_list(raw)
+            cleaned = _clean_workspace_list(raw, profile=profile)
             if len(cleaned) != len(raw):
                 # Persist the cleaned version so stale entries don't keep reappearing
                 try:
@@ -364,7 +497,7 @@ def load_workspaces() -> list:
                     )
                 except Exception:
                     logger.debug("Failed to persist cleaned workspace list")
-            return cleaned or [{'path': _profile_default_workspace(), 'name': 'Home'}]
+            return cleaned or [{'path': _profile_default_workspace(profile=profile), 'name': 'Home'}]
         except Exception:
             logger.debug("Failed to load workspaces from %s", ws_file)
     # No profile-local file yet.
@@ -372,7 +505,11 @@ def load_workspaces() -> list:
     # For NAMED profiles: always start clean with just their own workspace.
     try:
         from api.profiles import get_active_profile_name
-        is_default = get_active_profile_name() in ('default', None)
+        if profile is not None:
+            profile_home = _resolve_profile_home_param(profile)
+            is_default = _is_default_profile_home(profile_home)
+        else:
+            is_default = get_active_profile_name() in ('default', None)
     except ImportError:
         is_default = True
     if is_default:
@@ -380,16 +517,20 @@ def load_workspaces() -> list:
         if migrated:
             return migrated
     # Fresh start: single entry from the profile's configured workspace, labeled "Home"
-    return [{'path': _profile_default_workspace(), 'name': 'Home'}]
+    return [{'path': _profile_default_workspace(profile=profile), 'name': 'Home'}]
 
 
-def save_workspaces(workspaces: list) -> None:
-    ws_file = _workspaces_file()
+def save_workspaces(workspaces: list, profile: str | Path | None = None) -> None:
+    ws_file = _workspaces_file_for_profile(profile)
+    if ws_file is None:
+        # Fail-closed: an invalid profile name must not write any state file
+        # (it would land in the default profile's directory via the old clamp).
+        raise ValueError(f"cannot save workspaces for invalid profile {profile!r}")
     ws_file.parent.mkdir(parents=True, exist_ok=True)
     ws_file.write_text(json.dumps(workspaces, ensure_ascii=False, indent=2), encoding='utf-8')
 
 
-def get_profile_default_workspace() -> str:
+def get_profile_default_workspace(profile: str | Path | None = None) -> str:
     """Resolve the ACTIVE PROFILE's default workspace, never the global file.
 
     Like get_last_workspace() but WITHOUT the global ``_GLOBAL_LW_FILE``
@@ -404,32 +545,38 @@ def get_profile_default_workspace() -> str:
     Priority: profile-scoped ``last_workspace.txt`` -> profile ``config.yaml``
     ``workspace``/``default_workspace`` -> ``terminal.cwd`` -> process default.
     """
-    remote_cwd = _remote_terminal_cwd()
+    try:
+        remote_cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd()
+    except TypeError:
+        remote_cwd = _remote_terminal_cwd()
 
     def _valid(raw: str) -> str | None:
         if not raw:
             return None
         if remote_cwd:
-            if _remote_terminal_workspace_candidate(raw) is not None:
+            if _remote_terminal_workspace_candidate(raw, profile=profile) is not None:
                 return raw
             return None
         if Path(raw).is_dir():
             return raw
         return None
 
-    lw_file = _last_workspace_file()
-    if lw_file.exists():
+    lw_file = _last_workspace_file_for_profile(profile)
+    if lw_file is not None and lw_file.exists():
         try:
             p = _valid(lw_file.read_text(encoding='utf-8').strip())
             if p:
                 return p
         except Exception:
             logger.debug("Failed to read profile last workspace from %s", lw_file)
-    return _profile_default_workspace()
+    return _profile_default_workspace(profile=profile)
 
 
-def get_last_workspace() -> str:
-    remote_cwd = _remote_terminal_cwd()
+def get_last_workspace(profile: str | Path | None = None) -> str:
+    try:
+        remote_cwd = _remote_terminal_cwd(profile=profile) if profile is not None else _remote_terminal_cwd()
+    except TypeError:
+        remote_cwd = _remote_terminal_cwd()
 
     def valid_last_workspace(raw: str) -> str | None:
         if not raw:
@@ -438,35 +585,58 @@ def get_last_workspace() -> str:
             # For remote/SSH profiles, last_workspace is target-side state. Do
             # not accept stale server-local paths merely because they exist on
             # the WebUI host; require the value to stay under terminal.cwd.
-            if _remote_terminal_workspace_candidate(raw) is not None:
+            if _remote_terminal_workspace_candidate(raw, profile=profile) is not None:
                 return raw
             return None
         if Path(raw).is_dir():
             return raw
         return None
 
-    lw_file = _last_workspace_file()
-    if lw_file.exists():
+    lw_file = _last_workspace_file_for_profile(profile)
+    if lw_file is not None and lw_file.exists():
         try:
             p = valid_last_workspace(lw_file.read_text(encoding='utf-8').strip())
             if p:
                 return p
         except Exception:
             logger.debug("Failed to read last workspace from %s", lw_file)
-    # Fallback: try global file
-    if _GLOBAL_LW_FILE.exists():
+    # Fallback: try global file — but ONLY for the root/default profile. A named
+    # profile must never inherit another profile's last-workspace binding through
+    # the legacy global state (#7168 re-gate round 3). Identity is canonical:
+    # a symlink alias of the default home still counts as default, while a
+    # malformed profile name fails validation and is denied the fallback
+    # rather than being clamped onto the global file (#7168 re-gate round 4).
+    _global_fallback_allowed = True  # ambient / no explicit profile: historical behavior
+    if profile is not None:
+        if str(profile).strip() == 'default':
+            _global_fallback_allowed = True
+        else:
+            try:
+                _global_fallback_allowed = _is_default_profile_home(
+                    _resolve_profile_home_param(profile)
+                )
+            except Exception:
+                # Conservative default: an unresolvable explicit profile is NOT the
+                # root profile — deny the global fallback rather than leak.
+                _global_fallback_allowed = False
+    if _global_fallback_allowed and _GLOBAL_LW_FILE.exists():
         try:
             p = valid_last_workspace(_GLOBAL_LW_FILE.read_text(encoding='utf-8').strip())
             if p:
                 return p
         except Exception:
             logger.debug("Failed to read global last workspace")
-    return _profile_default_workspace()
+    return _profile_default_workspace(profile=profile)
 
 
-def set_last_workspace(path: str) -> None:
+def set_last_workspace(path: str, profile: str | Path | None = None) -> None:
     try:
-        lw_file = _last_workspace_file()
+        lw_file = _last_workspace_file_for_profile(profile)
+        if lw_file is None:
+            # Fail-closed: an invalid profile name must not write the default
+            # profile's last_workspace.txt (#7168 re-gate round 4).
+            logger.debug("Refusing to set last workspace for invalid profile %r", profile)
+            return
         lw_file.parent.mkdir(parents=True, exist_ok=True)
         lw_file.write_text(str(path), encoding='utf-8')
     except Exception:
@@ -683,7 +853,15 @@ def _is_within(path: Path, root: Path) -> bool:
         return False
 
 
-def _trusted_workspace_roots() -> list[Path]:
+def _trusted_workspace_roots(profile: str | Path | None = None) -> list[Path]:
+    """Return the host directories workspace suggestions may traverse.
+
+    Saved-workspace roots follow the same trust rule as
+    :func:`resolve_trusted_workspace`: with an explicit *profile*, only
+    workspaces saved under THAT profile widen the boundary (plus the ambient
+    home / boot-default carve-outs); ``None`` keeps the historical ambient /
+    global saved-list behaviour.
+    """
     roots: list[Path] = []
 
     def add(candidate: str | Path | None) -> None:
@@ -702,24 +880,26 @@ def _trusted_workspace_roots() -> list[Path]:
 
     add(_home_path())
     add(_BOOT_DEFAULT_WORKSPACE)
-    for w in load_workspaces():
+    for w in load_workspaces(profile=profile):
         add(w.get("path"))
     roots.sort(key=lambda p: len(str(p)))
     return roots
 
 
-def list_workspace_suggestions(prefix: str = "", limit: int = 12) -> list[str]:
+def list_workspace_suggestions(
+    prefix: str = "", limit: int = 12, profile: str | Path | None = None
+) -> list[str]:
     """Return workspace path suggestions under trusted roots only.
 
     Suggestions are limited to directories under one of:
       - Path.home()
       - the boot default workspace
-      - already-saved workspace roots
+      - already-saved workspace roots (scoped to *profile* when given)
 
     Arbitrary system prefixes return an empty list rather than an error so the
     UI can safely autocomplete while the user types.
     """
-    roots = _trusted_workspace_roots()
+    roots = _trusted_workspace_roots(profile=profile)
     if not roots:
         return []
 
@@ -815,7 +995,7 @@ def list_workspace_suggestions(prefix: str = "", limit: int = 12) -> list[str]:
     return suggestions[:limit]
 
 
-def resolve_trusted_workspace(path: str | Path | None = None) -> Path:
+def resolve_trusted_workspace(path: str | Path | None = None, profile: str | Path | None = None) -> Path:
     """Resolve and validate a workspace path.
 
     A path is trusted if it satisfies at least one of:
@@ -837,12 +1017,12 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path:
     trusted (it was validated at server startup).
     """
     if path in (None, ""):
-        return _resolve_path(_BOOT_DEFAULT_WORKSPACE)
+        return _resolve_path(_BOOT_DEFAULT_WORKSPACE, profile) if profile is not None else _resolve_path(_BOOT_DEFAULT_WORKSPACE)
 
-    candidate = _resolve_path(path)
+    candidate = _resolve_path(path, profile) if profile is not None else _resolve_path(path)
 
     access_error = _workspace_access_error(candidate)
-    remote_candidate = _remote_terminal_workspace_candidate(path)
+    remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile)
     if access_error:
         # For remote terminal profiles, workspace paths belong to the target
         # machine. Allow paths under terminal.cwd so session switching can
@@ -869,6 +1049,11 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path:
 
     # (B) Trusted if already in the saved workspace list — covers non-home installs
     try:
+        saved = load_workspaces(profile=profile)
+        saved_paths = {_resolve_path(w["path"], profile) for w in saved if w.get("path")}
+        if candidate in saved_paths:
+            return candidate
+    except TypeError:
         saved = load_workspaces()
         saved_paths = {_resolve_path(w["path"]) for w in saved if w.get("path")}
         if candidate in saved_paths:
@@ -898,15 +1083,22 @@ def resolve_trusted_workspace(path: str | Path | None = None) -> Path:
 def resolve_implicit_workspace_with_recovery(
     candidate: str | Path | None,
     fallback: str | Path | None | Callable[[], str | Path | None],
+    profile: str | Path | None = None,
 ) -> tuple[Path, bool]:
     """Resolve an implicit workspace, recovering only a genuinely missing path.
 
     The fallback still passes through :func:`resolve_trusted_workspace`. Existing
     but untrusted, inaccessible, or non-directory candidates are not recovery
     cases: their original validation error is preserved so fallback cannot widen
-    the workspace trust boundary.
+    the workspace trust boundary. When *profile* is given, both the trust
+    resolution and the recovery fallback are scoped to that profile.
     """
     try:
+        if profile is not None:
+            try:
+                return resolve_trusted_workspace(candidate, profile=profile), False
+            except TypeError:
+                pass
         return resolve_trusted_workspace(candidate), False
     except ValueError as original_error:
         if candidate in (None, ""):
@@ -917,19 +1109,49 @@ def resolve_implicit_workspace_with_recovery(
         # never prove target-side deletion. Config-read uncertainty also fails
         # closed by preserving the original validation error.
         try:
-            from api.config import get_config
+            from api.config import get_config, get_config_for_profile_home
 
-            terminal_cfg = get_config().get("terminal", {})
+            if profile is not None:
+                # Classify the backend from THIS profile's own config, never the
+                # ambient one: a remote profile without terminal.cwd must still be
+                # recognized as remote when loaded under a different ambient home.
+                terminal_cfg = get_config_for_profile_home(
+                    _resolve_profile_home_param(profile)
+                ).get("terminal", {})
+            else:
+                terminal_cfg = get_config().get("terminal", {})
         except Exception:
             logger.debug("Failed to classify terminal backend for workspace recovery", exc_info=True)
             raise original_error from None
         if _is_remote_terminal_backend(terminal_cfg):
             raise original_error from None
         try:
-            local_candidate = _resolve_path(candidate)
+            local_candidate = (
+                _resolve_path(candidate, profile=profile)
+                if profile is not None
+                else _resolve_path(candidate)
+            )
             local_candidate.stat()
         except FileNotFoundError:
-            fallback_value = fallback() if callable(fallback) else fallback
+            def _profile_bound_fallback():
+                if profile is not None and callable(fallback):
+                    # Profile-bound getter FIRST (#7168 re-gate round 3): the
+                    # production call sites pass profile-aware getters such as
+                    # get_last_workspace, so binding the explicit profile must
+                    # take precedence over any zero-argument compatibility call,
+                    # which would read ambient/global state.
+                    try:
+                        return fallback(profile)
+                    except TypeError:
+                        pass
+                return fallback() if callable(fallback) else fallback
+
+            fallback_value = _profile_bound_fallback()
+            if profile is not None:
+                try:
+                    return resolve_trusted_workspace(fallback_value, profile=profile), True
+                except TypeError:
+                    pass
             return resolve_trusted_workspace(fallback_value), True
         except (OSError, RuntimeError, ValueError):
             raise original_error from None
@@ -957,7 +1179,7 @@ def _strip_surrounding_quotes(path: str) -> str:
     return s
 
 
-def validate_workspace_to_add(path: str) -> Path:
+def validate_workspace_to_add(path: str, profile: str | Path | None = None) -> Path:
     """Validate a path for *adding* to the workspace list (less restrictive than resolve_trusted_workspace).
 
     When a user explicitly adds a new workspace path, we trust their intent — they
@@ -972,10 +1194,10 @@ def validate_workspace_to_add(path: str) -> Path:
     and users routinely paste those into the Add Space input.
     """
     path = _strip_surrounding_quotes(path)
-    candidate = _resolve_path(path)
+    candidate = _resolve_path(path, profile) if profile is not None else _resolve_path(path)
 
     access_error = _workspace_access_error(candidate)
-    remote_candidate = _remote_terminal_workspace_candidate(path)
+    remote_candidate = _remote_terminal_workspace_candidate(path, profile=profile)
     if access_error:
         # Remote terminal profiles validate workspace existence on the target
         # machine, not on the WebUI server. Permit target-side paths under

docs/CONTRACTS.md

@@ -23,6 +23,10 @@ contributor guidance; it does not change runtime behavior or CI gates.
 
 ## Runtime, durability, and state contracts
 
+- [`docs/remote-workspaces.md`](remote-workspaces.md):
+  architecture contract for remote terminal workspaces (SSH/Docker), target-side
+  POSIX path preservation against macOS synthetic firmlink expansion, and
+  per-profile isolation boundaries.
 - [`docs/rfcs/webui-run-state-consistency-contract.md`](rfcs/webui-run-state-consistency-contract.md):
   proposed consistency rules for current WebUI streaming, recovery, replay,
   model-context reconstruction, compression, UI scene/cache, and sidebar metadata

docs/architecture/agent-api-contract.md

@@ -41,10 +41,109 @@ helpers are genuinely shared code.
 | `docker_agent_source_volume` | Compose files and Docker docs expose `hermes-agent-src` and `/opt/hermes` to make the agent checkout visible to WebUI. | Remove the WebUI source mount only after startup install and runtime imports have migrated. This needs Docker/compose follow-up work, not a runtime behavior change in this audit PR. |
 | `startup_dependency_install` | `api/startup.py` discovers `HERMES_WEBUI_AGENT_DIR` or `$HERMES_HOME/hermes-agent`; `server.py` calls `auto_install_agent_deps()` after import verification fails; `docker_init.bash` installs from the staged agent source. | Replace source-tree pip installs with a packaged hermes-agent WebUI client plus an agent health/version capability contract. Keep `HERMES_WEBUI_AGENT_DIR` during migration as an override/debug path, but it should stop being required in normal multi-container startup. |
 | `runtime_auxiliary_model_metadata` | `api/streaming.py`, `api/routes.py`, `api/config.py`, and `api/providers.py` import `agent.auxiliary_client`, `agent.model_metadata`, `agent.models_dev`, `hermes_cli.models`, and `agent.account_usage`. | Existing provider/model WebUI endpoints can keep serving UI data where they already wrap agent helpers. Missing surfaces need hermes-agent endpoints or a client package for auxiliary task config, text auxiliary calls, context length, token estimate, provider catalog, and account usage. |
-| `runtime_session_state` | `api/streaming.py`, `api/goals.py`, and `api/state_sync.py` import `hermes_state.SessionDB` directly. `api/models.py` also opens the active profile's canonical `state.db` for scoped session deletion because the current canonical helper does not preserve branch/compression evidence ahead of inherited delegate metadata or expose retryable artifact-cleanup semantics. | Move cross-container state reads and writes, including destructive session deletion, behind hermes-agent session/state endpoints once the agent API provides equivalent lineage precedence, transaction, and retry-manifest guarantees. WebUI-only presentation state can remain local, but agent session storage should not be opened from the WebUI container. |
+| `runtime_session_state` | `api/streaming.py`, `api/goals.py`, and `api/state_sync.py` import `hermes_state.SessionDB` directly. `api/models.py` also reads the `messages` table directly (see [state.db message content encoding](#statedb-message-content-encoding) for the storage-format coupling that creates) and opens the active profile's canonical `state.db` for scoped session deletion because the current canonical helper does not preserve branch/compression evidence ahead of inherited delegate metadata or expose retryable artifact-cleanup semantics. | Move cross-container state reads and writes, including destructive session deletion, behind hermes-agent session/state endpoints once the agent API provides equivalent lineage precedence, transaction, and retry-manifest guarantees. WebUI-only presentation state can remain local, but agent session storage should not be opened from the WebUI container. |
 | `runtime_gateway_provider` | `api/streaming.py` and `api/routes.py` import `hermes_cli.runtime_provider`; adapter helpers such as `agent.anthropic_adapter` are also imported for gateway normalization. | Provider resolution, runtime routing, and gateway invocation should be hermes-agent API calls. WebUI can keep request validation and display formatting, but it should not import runtime provider internals from the agent checkout. |
 | `webui_local_or_client_package` | WebUI imports `hermes_cli.auth`, `hermes_cli.config`, `hermes_cli.plugins`, `hermes_cli.profiles`, `hermes_cli.goals`, `agent.skill_utils`, `agent.credential_pool`, and `hermes_constants`. | Pure schemas, constants, and parsing helpers can move into a small versioned client/shared package. Privileged data such as credential pools, auth status, profile mutation, plugin discovery, and goal persistence need hermes-agent endpoints. UI-only formatting can remain in WebUI. |
 
+## state.db message content encoding
+
+`api/models.py` reads the agent's `messages` table with its own SQL, so it also
+depends on how hermes-agent *encodes* that table, not only on its schema. This
+is a storage-format coupling and belongs with the `runtime_session_state`
+dependency class above.
+
+`hermes_state` stores list/dict message content (multimodal parts) as a
+sentinel-prefixed JSON string, because sqlite3 binds only scalars:
+
+```
+_CONTENT_JSON_PREFIX = "\x00json:"        # hermes_state.py
+```
+
+It provides `_decode_content()` to reverse this. Any WebUI read path that
+projects that column must apply an equivalent decode; a raw read hands the
+frontend an encoded string that no reader recognises, and an image part's
+base64 data URI then renders as literal transcript text.
+
+### WebUI decoding contract
+
+`_decode_state_db_content()` in `api/models.py` is the single decode point. It
+is deliberately narrower than the agent's own decoder, because the WebUI can
+only accept shapes the rest of its pipeline already renders:
+
+| Input | Result | Why |
+| --- | --- | --- |
+| Sentinel + list with non-whitespace text and only valid image parts | decoded `list` | `msgContent()` joins the text parts, so the row renders its text |
+| Sentinel + image-only list, or text that is empty/whitespace | unchanged string | `msgContent()` discards image parts, so `_messageIsRenderable()` would hide the row with no error |
+| Sentinel + list containing a malformed image part | unchanged string | a part must carry a valid per-type payload, not just a matching `type` |
+| Sentinel + dict or scalar root | unchanged string | a dict reaches `_getCachedRender()`, and `_renderCacheKey()` calls `text.slice()` on it, blanking the turn |
+| Sentinel + `NaN`/`Infinity`/overflowed float | unchanged string | Python emits them, browser `JSON.parse()` rejects the whole `/api/session` payload |
+| Sentinel + unsupported part shapes | unchanged string | `input_text`, `output_text`, scalar and unknown parts are dropped by the JS readers, so decoding them would silently lose content that is visible today |
+| Anything without the sentinel | unchanged | non-sentinel content is not this contract's concern |
+
+Supported parts are `{"type": "text", "text": <str>}` plus image parts whose
+payload validates for their type: `image_url` with a non-empty URL (string or
+`{"url": ...}`), `input_image` with a URL or `file_id`, and `image` with a
+`base64` source carrying `data` and `media_type` or a `url` source. At least one
+text part must contain non-whitespace text.
+
+**Image parts do not render from this projection.** The shared JS readers drop
+them, and the state.db projection supplies no `attachments`. Decoding a
+text-and-image row shows its text and keeps the base64 payload out of the DOM;
+it does not display the image. Rendering images from state.db rows would need a
+shared inline-image projection first, at which point image-only lists could be
+accepted too.
+
+Widening the accepted schema requires teaching every shared content reader
+through one extractor first; until then unsupported shapes must keep falling
+back to the raw string.
+
+### Consequences for identity and bounded reads
+
+Decoding changes the runtime type of `content`, so every consumer that derives
+an identity from it must agree on one representation:
+
+- Every key -- merge, dedup, content, visible and the fuzzy fallback -- derives
+  content identity through `_content_identity_for_key()`. Non-list values key
+  exactly as on master, `str(content or "")`. Non-empty lists get an
+  **out-of-band** tuple identity, so no message body can compare equal to one:
+  an in-band string marker would be forgeable by a scalar that contains it.
+  Two rich turns sharing visible text and timestamp stay distinct when their
+  images differ.
+- Fuzzy duplicate matching is text-only. Structured identities match by exact
+  identity or not at all, so a rich row can never fuzzy-match a scalar.
+- The merge key cache never writes a key component back into message content.
+  To avoid re-serialising large payloads for every key it instead memoises the
+  canonical serialisation per content object, scoped to one
+  `merge_session_messages_append_only()` call.
+- The multimodal mirror bridge pairs one rich image-bearing row with one scalar
+  mirror only. `require_image_parts` and `require_scalar_mirror` are mutually
+  exclusive so rich-to-rich pairing cannot occur.
+- Every read path that projects the `content` column applies the decoder, so
+  keys derived on one path cannot disagree with keys derived on another. There
+  are exactly three such call sites:
+
+  | Call site | Role |
+  | --- | --- |
+  | `_project_state_db_message()` | canonical row projection, shared by the transcript read and the regeneration tail |
+  | `get_state_db_session_message_keys_before_timestamp()` | bounded prefix keys |
+  | `get_state_db_regeneration_tail_snapshot()` | regeneration prefix keys |
+
+  If prefix keys stayed encoded while the projected tail was decoded, the
+  prefix/tail collision proof could miss a genuine repeated recovered turn and
+  `_bounded_tail_snapshot_if_safe` would reject the bounded path, reading the
+  entire transcript during regeneration.
+
+  Two nearby paths deliberately need no decode.
+  `get_state_db_session_message_prefix_summary()` projects only timestamp
+  counts and never selects `content`. State-db sidecar reconstruction
+  (`_sync_sidecar_from_state_db_if_newer()`) sources its rows through
+  `get_state_db_session_messages()`, so it inherits the canonical decoded
+  projection rather than reading the column itself.
+
+When session state moves behind hermes-agent endpoints, this decode should move
+with it: the agent should return structured content over the API and the WebUI
+should stop depending on the sentinel format at all.
+
 ## Replacement contract
 
 ### Existing endpoint candidates

docs/remote-workspaces.md

@@ -0,0 +1,29 @@
+# Remote Terminal Workspaces
+
+Architecture contract and path-resolution semantics for remote terminal profiles (SSH, Docker) in Hermes WebUI.
+
+---
+
+## 1. Overview
+
+When a Hermes profile is configured with a remote terminal backend (e.g. `terminal.backend: "ssh"` or `"docker"`), its working directory (`terminal.cwd`) lives on the remote target host rather than the local WebUI host filesystem.
+
+On hosts such as macOS, local path resolution via `Path.resolve()` or `os.path.realpath()` expands synthetic firmlinks (e.g. rewriting `/home/<user>` to `/System/Volumes/Data/home/<user>`). Because the target-side path does not exist on the local macOS server, unconstrained local resolution causes runtime validation failures and session corruption.
+
+---
+
+## 2. Path Resolution Contract
+
+1. **Target-side POSIX Path Preservation:**
+   - Any valid POSIX path at or beneath a remote profile's configured `terminal.cwd` is preserved verbatim as a `Path` object without calling host-local `Path.resolve()`.
+   - Traversal escapes (`..`), null bytes (`\0`), and blocked system roots (`/etc`, `/usr`, `/var`, `/sys`, `/proc`, `/dev`) continue to be strictly rejected.
+
+2. **Profile Isolation & Boundary Enforcement:**
+   - Remote path recognition is **strictly scoped to the target/active profile**.
+   - If an active profile has a local backend, target-side remote paths belonging to inactive named profiles are **not** treated as remote for the active local profile.
+   - For local profiles, workspace addition (`/api/workspaces/add`) enforces host-local directory existence and permissions.
+   - For remote profiles, workspace addition validates against the profile's remote `terminal.cwd` and skips host-local directory creation (`mkdir`).
+
+3. **Session & Streaming Lifecycle:**
+   - `Session.__init__` and `Session.load` normalize session workspace paths through `_resolve_path(..., profile=profile)`, preserving target-side paths for remote profiles and preventing corruption of `session.workspace` or `session.created_workspace`.
+   - Streaming execution (`_run_agent_streaming`) and multimodal asset root resolution preserve the remote session workspace when updating runtime run state.

server.py

@@ -716,9 +716,9 @@ def main() -> None:
 
     try:
         signal.signal(signal.SIGTERM, _request_shutdown)
+        signal.signal(signal.SIGINT, _request_shutdown)  # Ctrl-C / ctl.sh daemons (#7078)
     except (ValueError, OSError):
-        # Not on the main thread (e.g. embedded/test harness); skip handler.
-        logger.debug("Could not install SIGTERM handler", exc_info=True)
+        logger.debug("Could not install shutdown signal handlers", exc_info=True)
 
     try:
         httpd.serve_forever()

static/messages.js

@@ -7236,7 +7236,28 @@ function autoResize(){
   }
   const el=$('msg');
   const _nextValue=String(el.value||'');
+  if(typeof CSS!=='undefined'&&typeof CSS.supports==='function'&&CSS.supports('field-sizing','content')){
+    if(el.style.height) el.style.height='';
+    _composerLastResizeValue=_nextValue;
+    updateSendBtn();
+    return;
+  }
   const _isAppendOnly=_nextValue.length>_composerLastResizeValue.length&&_nextValue.startsWith(_composerLastResizeValue);
+  // An EMPTY composer has no content to measure, so clear any inline height and
+  // let the CSS `min-height` define the resting size. Measuring instead would
+  // read the PLACEHOLDER's scrollHeight — a long busy/compression hint wraps to
+  // two or three lines and would grow the empty composer (71px for the English
+  // busy hint, 97px for the French compression one) purely because of hint text.
+  // That made the empty height history-dependent on this path: 44px on a fresh
+  // send, but grown after any later resize while empty. The native
+  // `field-sizing` path above always holds the resting height, so clearing here
+  // keeps both paths on the same contract.
+  if(!_nextValue){
+    if(el.style.height) el.style.height='';
+    _composerLastResizeValue=_nextValue;
+    updateSendBtn();
+    return;
+  }
   const _fitsCurrentHeight=el.scrollHeight<=el.offsetHeight;
   // Only a direct append at the natural one-row height can skip the height
   // round trip. Replacements and an already-tall composer must remeasure so the

static/sessions.js

@@ -2424,7 +2424,7 @@ const _HANDOFF_THRESHOLD = 10;  // conversation rounds
 const _HANDOFF_STORAGE_PREFIX = 'handoff:';
 const _HANDOFF_SUFFIX_DISMISSED_AT = 'dismissed_at';
 const _HANDOFF_SUFFIX_SUMMARY_HANDLED_AT = 'summary_handled_at';
-const _MESSAGING_RAW_SOURCES = new Set(['weixin', 'telegram', 'discord', 'slack', 'email', 'wecom', 'wecom_callback', 'matrix']);
+const _MESSAGING_RAW_SOURCES = new Set(['weixin', 'telegram', 'discord', 'slack', 'email', 'wecom', 'wecom_callback', 'matrix', 'signal']);
 const _MESSAGING_SOURCE_LABELS = {
   weixin: 'WeChat',
   telegram: 'Telegram',
@@ -2434,6 +2434,7 @@ const _MESSAGING_SOURCE_LABELS = {
   wecom: 'WeCom',
   wecom_callback: 'WeCom Callback',
   matrix: 'Matrix',
+  signal: 'Signal',
 };
 
 function _isMessagingSession(session) {
@@ -7133,6 +7134,12 @@ function _attachChildSessionsToSidebarRows(collapsedRows, rawSessions, rawRefere
     const childRenderable=!!(child&&child.session_id&&renderableChildIds.has(child.session_id));
     if(child&&child.session_id&&visibleBySid.has(child.session_id)) continue;
     const isForkChild=_isForkWithResolvableParent(child, sessionIdsInList)&&!(child&&child.pinned);
+    const childRawRole=[
+      child&&child.raw_source,
+      child&&child.source_tag,
+      child&&child.source,
+    ].map(source=>String(source||'').trim().toLowerCase()).find(Boolean)||'';
+    const childIsDelegatedSubagent=_isChildSession(child)&&childRawRole==='subagent';
     const childLineageKey=child&&(child._lineage_root_id||child.lineage_root_id||child.parent_session_id);
     const isHiddenLineageReferenceChild=!!(child&&child.archived&&child.parent_session_id&&childLineageKey&&!child.pinned&&!childRenderable);
     if(!_isChildSession(child)&&!isForkChild&&!isHiddenLineageReferenceChild) continue;
@@ -7157,11 +7164,10 @@ function _attachChildSessionsToSidebarRows(collapsedRows, rawSessions, rawRefere
       hiddenArchivedChildTree.add(child.session_id);
       continue;
     }
-    // Cross-surface rows (for example a WebUI continuation from a Telegram
-    // conversation) should remain top-level when there is no WebUI-owned parent
-    // row to stack under.  But if the parent is visible in this same sidebar
-    // render, attach normally — delegated subagent rows are also cross-source
-    // relative to their WebUI parent and should not be forced into orphans.
+    // Independent cross-surface rows (for example a WebUI continuation from a
+    // Telegram conversation) remain top-level instead of nesting under an
+    // external parent. Delegated subagents are also cross-source, but they are
+    // parent-owned work and should still attach to the visible parent row.
     const parentSourceMarker=String(parentRow&&(
       parentRow.session_source||parentRow.raw_source||parentRow.source_tag||parentRow.source
     )||'').toLowerCase();
@@ -7172,7 +7178,7 @@ function _attachChildSessionsToSidebarRows(collapsedRows, rawSessions, rawRefere
       parentRow.session_source==='messaging'||
       (parentSourceMarker&&parentSourceMarker!=='webui'&&parentSourceMarker!=='subagent'&&parentSourceMarker!=='other'&&parentSourceMarker!=='fork')
     );
-    if(parentRow&&child._cross_surface_child_session&&parentIsExternal){
+    if(parentRow&&child._cross_surface_child_session&&parentIsExternal&&!childIsDelegatedSubagent){
       if(childRenderable) orphans.push({...child,_orphan_child_session:true});
       continue;
     }

static/style.css

@@ -2594,7 +2594,8 @@
   @media (hover: hover) {
     .attach-thumb:hover{filter:brightness(1.05);transform:scale(1.04);}
   }
-  textarea#msg{width:100%;background:transparent;border:none;outline:none;color:var(--text);font-size:16px;line-height:1.65;padding:12px 16px 6px;resize:none;min-height:44px;max-height:200px;font-family:inherit;}
+  textarea#msg{width:100%;background:transparent;border:none;outline:none;color:var(--text);font-size:16px;line-height:1.65;padding:12px 16px 6px;resize:none;min-height:44px;max-height:200px;overflow-y:auto;field-sizing:content;font-family:inherit;}
+  textarea#msg:placeholder-shown{field-sizing:fixed;}
   textarea#msg::placeholder{color:var(--muted);}
   .composer-footer{display:flex;align-items:center;justify-content:space-between;gap:10px;padding:6px 10px 10px;position:relative;container-type:inline-size;container-name:composer-footer;}
   .composer-left{display:flex;align-items:center;gap:4px;min-width:0;flex:1;overflow-x:auto;overflow-y:hidden;scrollbar-width:none;}

tests/conftest.py

@@ -231,6 +231,44 @@ def _reset_password_hash_cache():
         _invalidate_password_hash_cache()
 
 
+def _strip_leaked_webui_password_env() -> None:
+    """Remove a leaked HERMES_WEBUI_PASSWORD between tests (#7168 review).
+
+    bootstrap.py runs _load_repo_dotenv() at import time, which copies values
+    from the developer's real repo .env straight into os.environ. When any
+    test imports bootstrap mid-session (e.g. tests/test_bootstrap_foreground.py
+    via its import_bootstrap fixture), a local .env containing
+    HERMES_WEBUI_PASSWORD leaks into the process environment OUTSIDE
+    monkeypatch's undo scope. Every later test then sees is_auth_enabled()
+    True and no-handler cookie helpers raise spurious
+    "build_profile_cookie requires a request handler" errors — exactly the
+    #5588 failure shape, but sourced from the repo .env instead of the hash
+    cache. Tests that legitimately enable auth set the var themselves AFTER
+    this strip; an intentionally-empty value ("") is preserved so
+    ctl.sh-style override semantics keep working.
+
+    HERMES_COMMAND gets the same treatment (#7168 re-gate round 7): a local
+    .env carrying HERMES_COMMAND leaks past bootstrap imports and redirects
+    gateway_restart._resolve_hermes_command() away from its mocked
+    shutil.which result, failing every later active-profile-restart test
+    with a machine-specific CLI path. Upstream code has no
+    HERMES_COMMAND override, so stripping a leaked value restores exact
+    upstream semantics.
+    """
+    if os.environ.get("HERMES_WEBUI_PASSWORD") == "":
+        pass  # intentional empty override preserved for the password var
+    else:
+        os.environ.pop("HERMES_WEBUI_PASSWORD", None)
+    os.environ.pop("HERMES_COMMAND", None)
+
+
+@pytest.fixture(autouse=True)
+def _strip_leaked_webui_password():
+    _strip_leaked_webui_password_env()
+    yield
+    _strip_leaked_webui_password_env()
+
+
 @pytest.fixture(autouse=True)
 def _invalidate_providers_cache():
     """Clear the /api/providers TTL cache around every test (#6010).

tests/test_agent_runtime_revision_guard.py

@@ -854,7 +854,7 @@ def test_stream_admission_uses_one_gateway_ownership_snapshot(monkeypatch, gatew
     monkeypatch.setattr(routes, "_active_run_stream_for_session", lambda _sid: None)
     monkeypatch.setattr(routes, "_is_hidden_empty_session", lambda _session: False)
     monkeypatch.setattr(routes, "_prepare_chat_start_session_for_stream", prepare)
-    monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda _workspace, **_kw: None)
     monkeypatch.setattr(routes.threading, "Thread", FakeThread)
     monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *_args, **_kwargs: {})
 

tests/test_attachment_upload_toctou.py

@@ -0,0 +1,160 @@
+"""TOCTOU hardening for chat-attachment uploads.
+
+handle_upload deduped filenames with an exists() check and then wrote with
+plain write_bytes — two concurrent uploads of the same name could both pass
+the check and the last writer silently won. These tests pin the #3398-style
+anchored O_CREAT|O_EXCL|O_NOFOLLOW creation semantics (mirrored from the
+workspace upload path): a raced duplicate must 409 instead of overwriting,
+and the non-raced happy path / dedup behavior must stay unchanged.
+
+Test-pattern cribbed from tests/test_raw_audio_upload.py (real multipart body
+through parse_multipart + fake handler) and
+tests/test_session_active_profile_authorization.py (monkeypatched
+get_session / _get_active_profile_name).
+"""
+import io
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+import api.upload as upload
+from api.upload import handle_upload
+
+
+def _multipart_body(fields=None, files=None, boundary=b"testboundary"):
+    fields = fields or {}
+    files = files or {}
+    body = b""
+    for name, value in fields.items():
+        body += b"--" + boundary + b"\r\n"
+        body += f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()
+        body += str(value).encode() + b"\r\n"
+    for name, (filename, data, content_type) in files.items():
+        body += b"--" + boundary + b"\r\n"
+        body += (
+            f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'
+            f"Content-Type: {content_type}\r\n\r\n"
+        ).encode()
+        body += data + b"\r\n"
+    body += b"--" + boundary + b"--\r\n"
+    return body, f"multipart/form-data; boundary={boundary.decode()}"
+
+
+class _FakeHandler:
+    def __init__(self, body: bytes, content_type: str):
+        self.rfile = io.BytesIO(body)
+        self.wfile = io.BytesIO()
+        self.headers = {
+            "Content-Type": content_type,
+            "Content-Length": str(len(body)),
+        }
+        self.status = None
+        self.sent_headers = {}
+
+    def send_response(self, status):
+        self.status = status
+
+    def send_header(self, key, value):
+        self.sent_headers[key] = value
+
+    def end_headers(self):
+        pass
+
+    def payload(self):
+        return json.loads(self.wfile.getvalue().decode("utf-8"))
+
+
+@pytest.fixture()
+def attachment_env(tmp_path, monkeypatch):
+    """Isolate the attachment inbox and stub the session lookup."""
+    root = tmp_path / "attachments"
+    monkeypatch.setenv("HERMES_WEBUI_ATTACHMENT_DIR", str(root))
+    monkeypatch.setattr(
+        upload,
+        "get_session",
+        lambda sid: SimpleNamespace(session_id=sid, profile=None),
+    )
+    monkeypatch.setattr(upload, "_get_active_profile_name", lambda: "default")
+    return root
+
+
+def test_raced_duplicate_returns_409_and_preserves_existing_bytes(attachment_env, monkeypatch):
+    """An attacker winning the exists()-check race must not get its bytes written.
+
+    Simulates the race deterministically: the destination file already exists
+    with known content, and _upload_destination returns that exact path as if
+    the dedup check had just passed. The anchored O_EXCL create must fail with
+    FileExistsError -> 409, and the pre-existing bytes must be untouched.
+    """
+    session_id = "race-sess"
+    dest_dir = upload._session_attachment_dir(session_id)
+    dest_dir.mkdir(parents=True, exist_ok=True)
+    target = dest_dir / "report.txt"
+    target.write_bytes(b"ORIGINAL-CONTENT")
+
+    monkeypatch.setattr(
+        upload,
+        "_upload_destination",
+        lambda session_id, safe_name, dest_dir=None: target,
+    )
+
+    body, content_type = _multipart_body(
+        fields={"session_id": session_id},
+        files={"file": ("report.txt", b"ATTACKER-BYTES", "text/plain")},
+    )
+    handler = _FakeHandler(body, content_type)
+    handle_upload(handler)
+
+    assert handler.status == 409
+    assert handler.payload() == {
+        "error": "Upload destination already exists: report.txt"
+    }
+    assert target.read_bytes() == b"ORIGINAL-CONTENT"
+
+
+def test_new_upload_happy_path_unchanged(attachment_env):
+    """A normal upload of a fresh name keeps the exact response shape."""
+    body, content_type = _multipart_body(
+        fields={"session_id": "happy-sess"},
+        files={"file": ("notes.txt", b"hello world", "text/plain")},
+    )
+    handler = _FakeHandler(body, content_type)
+    handle_upload(handler)
+
+    assert handler.status == 200
+    payload = handler.payload()
+    assert set(payload) == {"filename", "path", "size", "mime", "is_image"}
+    assert payload["filename"] == "notes.txt"
+    assert payload["mime"].startswith("text/")
+    assert payload["is_image"] is False
+    assert payload["size"] == len(b"hello world")
+
+    dest = attachment_env / "happy-sess" / "notes.txt"
+    assert Path(payload["path"]) == dest.resolve()
+    assert dest.read_bytes() == b"hello world"
+
+
+def test_non_raced_dedup_still_suffixed(attachment_env):
+    """Uploading an existing name (no race) still picks the -1 suffixed name."""
+    body1, ctype1 = _multipart_body(
+        fields={"session_id": "dedup-sess"},
+        files={"file": ("dup.txt", b"first", "text/plain")},
+    )
+    h1 = _FakeHandler(body1, ctype1)
+    handle_upload(h1)
+    assert h1.status == 200
+
+    body2, ctype2 = _multipart_body(
+        fields={"session_id": "dedup-sess"},
+        files={"file": ("dup.txt", b"second", "text/plain")},
+    )
+    h2 = _FakeHandler(body2, ctype2)
+    handle_upload(h2)
+    assert h2.status == 200
+    assert h2.payload()["filename"] == "dup-1.txt"
+
+    dest_dir = upload._session_attachment_dir("dedup-sess")
+    assert (dest_dir / "dup.txt").read_bytes() == b"first"
+    assert (dest_dir / "dup-1.txt").read_bytes() == b"second"

tests/test_bootstrap_dotenv.py

@@ -197,3 +197,87 @@ class TestBootstrapStructure:
         assert "_load_repo_dotenv" in bootstrap_src, (
             "bootstrap.py must load .env so direct invocation matches start.sh behaviour"
         )
+
+
+class TestLeakedWebuiPasswordIsolation:
+    """#7168 review: a local repo .env containing HERMES_WEBUI_PASSWORD leaks
+    into os.environ when any test imports bootstrap (import-time
+    _load_repo_dotenv() runs OUTSIDE monkeypatch's undo scope). The conftest
+    autouse guard strips the leaked var around every test so later tests
+    don't see is_auth_enabled()==True (the #5588 failure shape)."""
+
+    def _load_guard(self):
+        import importlib.util
+
+        cpath = Path(__file__).resolve().parent / "conftest.py"
+        spec = importlib.util.spec_from_file_location("_test_conftest", cpath)
+        mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+        return mod._strip_leaked_webui_password_env
+
+    def test_conftest_guard_strips_leaked_password(self):
+        strip = self._load_guard()
+
+        os.environ["HERMES_WEBUI_PASSWORD"] = "leaked-from-repo-dotenv"
+        try:
+            strip()
+            assert "HERMES_WEBUI_PASSWORD" not in os.environ
+        finally:
+            os.environ.pop("HERMES_WEBUI_PASSWORD", None)
+
+    def test_conftest_guard_preserves_intentional_empty_override(self):
+        """ctl.sh-style override semantics: an explicitly empty value means
+        'keep auth off' and must survive the strip."""
+        strip = self._load_guard()
+
+        sentinel = object()
+        os.environ["HERMES_WEBUI_PASSWORD"] = ""
+        try:
+            strip()
+            assert os.environ.get("HERMES_WEBUI_PASSWORD", sentinel) == ""
+        finally:
+            os.environ.pop("HERMES_WEBUI_PASSWORD", None)
+
+
+class TestLeakedHermesCommandIsolation:
+    """#7168 re-gate round 7: a local repo .env carrying HERMES_COMMAND leaks
+    into os.environ via bootstrap import-time _load_repo_dotenv() and
+    redirects gateway_restart._resolve_hermes_command() away from its mocked
+    shutil.which result — every later active-profile-restart test then fails
+    on a machine-specific CLI path. The autouse conftest guard strips the
+    leaked var around every test; upstream code never reads HERMES_COMMAND,
+    so stripping restores exact upstream semantics."""
+
+    def _load_guard(self):
+        import importlib.util
+
+        cpath = Path(__file__).resolve().parent / "conftest.py"
+        spec = importlib.util.spec_from_file_location("_test_conftest_hc", cpath)
+        mod = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(mod)
+        return mod._strip_leaked_webui_password_env
+
+    def test_conftest_guard_strips_leaked_hermes_command(self):
+        strip = self._load_guard()
+
+        os.environ["HERMES_COMMAND"] = "/machine/local/hermes-gateway-wrapper"
+        try:
+            strip()
+            assert "HERMES_COMMAND" not in os.environ
+        finally:
+            os.environ.pop("HERMES_COMMAND", None)
+
+    def test_conftest_guard_still_strips_password_when_command_leaks(self):
+        """Both leaked vars are stripped in one pass (password branch must
+        not be short-circuited by the HERMES_COMMAND handling)."""
+        strip = self._load_guard()
+
+        os.environ["HERMES_WEBUI_PASSWORD"] = "leaked-from-repo-dotenv"
+        os.environ["HERMES_COMMAND"] = "/machine/local/hermes-gateway-wrapper"
+        try:
+            strip()
+            assert "HERMES_WEBUI_PASSWORD" not in os.environ
+            assert "HERMES_COMMAND" not in os.environ
+        finally:
+            os.environ.pop("HERMES_WEBUI_PASSWORD", None)
+            os.environ.pop("HERMES_COMMAND", None)

tests/test_claude_code_session_import.py

@@ -200,7 +200,7 @@ def test_session_import_cli_returns_read_only_claude_code_payload(monkeypatch, t
     monkeypatch.setattr(routes, "j", lambda _handler, payload, status=200, extra_headers=None: payload)
     monkeypatch.setattr(routes, "get_cli_session_messages", lambda _sid, profile=None: messages if _sid == sid else [])
     monkeypatch.setattr(routes, "get_cli_sessions", lambda source_filter=None, all_profiles=False: [meta])
-    monkeypatch.setattr(routes, "get_last_workspace", lambda: tmp_path / "workspace")
+    monkeypatch.setattr(routes, "get_last_workspace", lambda profile=None: tmp_path / "workspace")
     monkeypatch.setattr(routes, "import_cli_session", lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("read-only import must not persist")))
 
     response = routes._handle_session_import_cli(object(), {"session_id": sid})

tests/test_early_session_title.py

@@ -93,7 +93,7 @@ def test_start_chat_stream_response_includes_provisional_title(tmp_path, monkeyp
     import api.routes as routes
 
     monkeypatch.setattr(Session, "save", lambda self, *a, **k: None)
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: object())
     monkeypatch.setattr(routes, "_run_agent_streaming", lambda *a, **k: None)
 

tests/test_gateway_sync.py

@@ -1020,6 +1020,7 @@ def test_agent_session_source_normalization_contract():
         'discord': ('messaging', 'Discord'),
         'slack': ('messaging', 'Slack'),
         'matrix': ('messaging', 'Matrix'),
+        'signal': ('messaging', 'Signal'),
         'cron': ('cron', 'Cron'),
         'webhook': ('webhook', 'Webhook'),
         'tool': ('tool', 'Tool'),
@@ -1045,13 +1046,14 @@ def test_sessions_js_treats_email_as_messaging_source():
     raw_section = src[src.find("_MESSAGING_RAW_SOURCES"):src.find("function _isMessagingSession")]
     label_section = src[src.find("_MESSAGING_SOURCE_LABELS"):src.find("function _isMessagingSession")]
 
-    for raw_source in ("email", "wecom", "wecom_callback", "matrix"):
+    for raw_source in ("email", "wecom", "wecom_callback", "matrix", "signal"):
         assert f"'{raw_source}'" in raw_section, f"Missing raw source {raw_source!r} in _MESSAGING_RAW_SOURCES"
 
     assert "email: 'Email'" in label_section
     assert "wecom: 'WeCom'" in label_section
     assert "wecom_callback: 'WeCom Callback'" in label_section
     assert "matrix: 'Matrix'" in label_section
+    assert "signal: 'Signal'" in label_section
 
 
 def test_sessions_js_treats_messaging_sidecars_behaviorally():
@@ -1067,6 +1069,8 @@ const cases = [
   {{ session_source: 'other', raw_source: 'wecom_callback' }},
   {{ session_source: 'other', source_tag: 'wecom' }},
   {{ session_source: 'other', source: 'matrix' }},
+  {{ session_source: 'other', source: 'signal' }},
+  {{ session_source: 'other', raw_source: 'signal' }},
   {{ session_source: 'messaging', source: 'anything' }},
 ];
 for (const c of cases) {{
@@ -1079,6 +1083,47 @@ if (_isMessagingSession({{ session_source: 'other', source: 'cli' }})) {{
     subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True)
 
 
+def test_sessions_js_treats_signal_sidecar_as_external_session():
+    """A Signal gateway session must be exposed as an external session.
+
+    The sidebar only lists sessions the client classifier reports as external,
+    and _isExternalSession routes messaging sources through _isMessagingSession.
+    A Signal sidecar whose persisted session_source is the legacy 'other' value
+    therefore has to be recognized from its raw source, exactly like the Matrix
+    precedent (#6816). This exercises the real sessions.js classifiers in node
+    rather than asserting on the shape of the allowlists.
+    """
+    src = (REPO_ROOT / "static" / "sessions.js").read_text(encoding="utf-8")
+    start = src.index("const _MESSAGING_RAW_SOURCES")
+    end = src.index("\n}", src.index("function _isExternalSession")) + len("\n}")
+    block = src[start:end]
+    script = f"""
+{block}
+if (!_isMessagingSession({{ session_source: 'other', source: 'signal' }})) {{
+  throw new Error('signal session was not classified as messaging');
+}}
+if (_MESSAGING_SOURCE_LABELS['signal'] !== 'Signal') {{
+  throw new Error('unexpected signal source label: ' + _MESSAGING_SOURCE_LABELS['signal']);
+}}
+for (const c of [
+  {{ session_source: 'other', source: 'signal' }},
+  {{ session_source: 'other', raw_source: 'signal' }},
+  {{ session_source: 'other', source_tag: 'signal' }},
+]) {{
+  if (!_isExternalSession(c)) {{
+    throw new Error('signal sidecar is not exposed as an external session: ' + JSON.stringify(c));
+  }}
+}}
+if (_isExternalSession({{ session_source: 'webui', source: 'signal' }})) {{
+  throw new Error('a WebUI-origin session must not be treated as an external session');
+}}
+if (_isExternalSession({{ session_source: 'other', source: 'some_future_platform' }})) {{
+  throw new Error('an unrecognized source must not be treated as an external session');
+}}
+"""
+    subprocess.run(["node", "-e", script], check=True, capture_output=True, text=True)
+
+
 def test_empty_active_gateway_session_does_not_hide_messaging_history(monkeypatch):
     """A zero-message active Gateway row must not hide older Discord history."""
     import api.routes as routes

tests/test_goal_command_webui.py

@@ -397,7 +397,7 @@ def test_goal_endpoint_sets_goal_and_starts_kickoff_stream(
 
     monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager)
     monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession())
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path)
     monkeypatch.setattr(
         routes,
         "webui_gateway_chat_enabled",
@@ -519,7 +519,7 @@ def test_goal_endpoint_adapter_keeps_full_set_text_and_legacy_payload_status(mon
     monkeypatch.setenv("HERMES_WEBUI_RUNTIME_ADAPTER", "legacy-journal")
     monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager)
     monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession())
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path)
     monkeypatch.setattr(
         routes,
         "_resolve_compatible_session_model_state",
@@ -732,7 +732,7 @@ def test_goal_kickoff_forwards_explicit_model_pick_to_resolver(monkeypatch, tmp_
 
     monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager)
     monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession())
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path)
     monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False)
     monkeypatch.setattr(routes, "get_config", lambda: {})
     monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve)
@@ -809,7 +809,7 @@ def test_goal_kickoff_defaults_explicit_model_pick_false(monkeypatch, tmp_path):
 
     monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager)
     monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession())
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path)
     monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False)
     monkeypatch.setattr(routes, "get_config", lambda: {})
     monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve)
@@ -901,7 +901,7 @@ def test_goal_kickoff_stamps_explicit_pick_signature(monkeypatch, tmp_path):
 
     monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager)
     monkeypatch.setattr(routes, "get_session", lambda sid: FakeSession())
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path)
     monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False)
     monkeypatch.setattr(routes, "get_config", lambda: {})
     monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve)
@@ -973,7 +973,7 @@ def test_goal_kickoff_does_not_stamp_signature_without_explicit_pick(monkeypatch
         return model, provider, False
 
     monkeypatch.setattr(webui_goals, "GoalManager", FakeGoalManager)
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda workspace, **_kw: tmp_path)
     monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda _cfg: False)
     monkeypatch.setattr(routes, "get_config", lambda: {})
     monkeypatch.setattr(routes, "_resolve_compatible_session_model_state", fake_resolve)

tests/test_issue1217_transcript_compaction.py

@@ -748,7 +748,7 @@ def test_handle_chat_sync_writeback_dedupes_full_context_replay(tmp_path, monkey
     monkeypatch.setattr(routes, "title_from", models.title_from)
     monkeypatch.setattr(config, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
     monkeypatch.setattr(routes, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path)
     monkeypatch.setattr(routes, "load_settings", lambda: {})
     monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: [])
 

tests/test_issue3930_source_filter_pushdown.py

@@ -241,7 +241,7 @@ def test_load_cli_sessions_uncached_pushes_specific_source_into_state_db_scan(mo
 
     monkeypatch.setattr(models, "get_claude_code_sessions", lambda: claude_calls.append(True) or [])
     monkeypatch.setattr(models, "read_importable_agent_session_rows", fake_read_rows)
-    monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path)
+    monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path)
     monkeypatch.setattr(models, "_profile_has_user_projects", lambda: False)
     monkeypatch.setattr(models, "ensure_cron_project", lambda **_: "cron-project-id")
     monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None)
@@ -283,7 +283,7 @@ def test_cron_source_filter_uses_cron_rescue_limit(monkeypatch, tmp_path):
         ]
 
     monkeypatch.setattr(models, "read_importable_agent_session_rows", fake_read_rows)
-    monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path)
+    monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path)
     monkeypatch.setattr(models, "_profile_has_user_projects", lambda: False)
     monkeypatch.setattr(models, "ensure_cron_project", lambda **_: "cron-project-id")
     monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None)

tests/test_issue492_workspace_reorder.py

@@ -22,7 +22,8 @@ class TestWorkspaceReorderEndpoint:
             {"path": "/home/user/b", "name": "Beta"},
             {"path": "/home/user/c", "name": "Gamma"},
         ]
-        mock_save.side_effect = lambda wss: wss
+        # Accept the profile-scoped save signature (save_workspaces(wss, profile=...)).
+        mock_save.side_effect = lambda wss, **kwargs: wss
         handler = _make_handler()
         _handle_workspace_reorder(handler, {
             "paths": ["/home/user/c", "/home/user/a", "/home/user/b"]

tests/test_issue5514_composer_grow_scroll_pin.py

@@ -39,6 +39,7 @@ ROOT = Path(__file__).parents[1]
 UI_JS = (ROOT / "static" / "ui.js").read_text(encoding="utf-8")
 MESSAGES_JS = (ROOT / "static" / "messages.js").read_text(encoding="utf-8")
 BOOT_JS = (ROOT / "static" / "boot.js").read_text(encoding="utf-8")
+STYLE_CSS = (ROOT / "static" / "style.css").read_text(encoding="utf-8")
 
 
 def _function_source(source: str, name: str) -> str:
@@ -582,6 +583,91 @@ def test_steady_state_keystroke_preserves_near_bottom_unpinned_reader():
     assert out["repinCalls"] == 0, "must not re-pin an unpinned reader"
 
 
+# ---------------------------------------------------------------------------
+# Behavioral (node vm) — native field-sizing avoids per-keystroke layout reads
+# ---------------------------------------------------------------------------
+
+def test_composer_declares_native_field_sizing_and_capped_overflow():
+    composer_rules = [part.split("}", 1)[0] for part in STYLE_CSS.split("textarea#msg{")[1:]]
+    required = ("field-sizing:content", "overflow-y:auto", "min-height:44px", "max-height:200px")
+    assert any(all(declaration in rule for declaration in required) for rule in composer_rules)
+    base_at = STYLE_CSS.index("  textarea#msg{")
+    state_rule = "  textarea#msg:placeholder-shown{field-sizing:fixed;}"
+    assert state_rule in STYLE_CSS
+    state_at = STYLE_CSS.index(state_rule)
+    assert state_at > base_at
+    assert "field-sizing:content" in STYLE_CSS[base_at : STYLE_CSS.index("}", base_at)]
+
+
+def test_native_field_sizing_skips_geometry_and_height_writes():
+    node = shutil.which("node")
+    if not node:  # pragma: no cover
+        pytest.skip("node not available")
+    body = _autoresize_body()
+    harness = textwrap.dedent(
+        """
+        let geometryReads = 0, heightWrites = 0, sendUpdates = 0;
+        let _composerAutoResizeRaf = 0;
+        let _composerLastResizeValue = 'first line';
+        const msg = {
+          value: 'first line',
+          style: {
+            _height: '',
+            set height(value) { heightWrites += 1; this._height = value; },
+            get height() { return this._height; },
+          },
+          get scrollHeight() { geometryReads += 1; return 200; },
+          get offsetHeight() { geometryReads += 1; return 44; },
+        };
+        const messages = { scrollTop: 0 };
+        const $ = (id) => id === 'msg' ? msg : id === 'messages' ? messages : null;
+        let supportsNative = true;
+        const supportCalls = [];
+        const CSS = {
+          supports: (property, value) => {
+            supportCalls.push([property, value]);
+            return supportsNative && property === 'field-sizing' && value === 'content';
+          },
+        };
+        function getComputedStyle() { geometryReads += 1; return { minHeight: '44px' }; }
+        function updateSendBtn() { sendUpdates += 1; }
+        function _repinMessagesAfterComposerResize() { throw new Error('native path must not repin'); }
+        %(autoresize)s
+
+        msg.value += '\\nsecond line';
+        autoResize();
+        const first = { geometryReads, heightWrites, lastValue: _composerLastResizeValue, sendUpdates };
+
+        msg.style._height = '120px';
+        msg.value += '\\nthird line';
+        autoResize();
+        const native = { first, geometryReads, heightWrites, height: msg.style.height, lastValue: _composerLastResizeValue, sendUpdates };
+
+        supportsNative = false;
+        geometryReads = 0;
+        heightWrites = 0;
+        msg.style._height = '';
+        msg.value += '\\nfourth line';
+        autoResize();
+        console.log(JSON.stringify({ native, fallback: { geometryReads, heightWrites }, supportCalls }));
+        """
+    ) % {"autoresize": body}
+    proc = subprocess.run([node, "-e", harness], capture_output=True, text=True, timeout=30)
+    assert proc.returncode == 0, proc.stderr
+    result = json.loads(proc.stdout)
+    assert result["native"] == {
+        "first": {"geometryReads": 0, "heightWrites": 0, "lastValue": "first line\nsecond line", "sendUpdates": 1},
+        "geometryReads": 0,
+        "heightWrites": 1,
+        "height": "",
+        "lastValue": "first line\nsecond line\nthird line",
+        "sendUpdates": 2,
+    }
+    assert result["fallback"]["geometryReads"] > 0
+    assert result["fallback"]["heightWrites"] == 2
+    assert result["supportCalls"] == [["field-sizing", "content"]] * 3
+
+
 # ---------------------------------------------------------------------------
 # Strict min-height parse (#6349 Codex re-gate): the single-row fast path must
 # only fire when getComputedStyle(el).minHeight is a genuine "<number>px". A

tests/test_issue6611_regeneration_transaction.py

@@ -193,7 +193,7 @@ def test_locked_postacceptance_workspace_exception_does_not_restore_turn(monkeyp
     monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
     monkeypatch.setattr(turn_journal, "append_turn_journal_event", lambda *_args, **_kwargs: {"turn_id": "turn-6611"})
     monkeypatch.setattr(Session, "save", lambda *_args, **_kwargs: None)
-    monkeypatch.setattr(routes, "set_last_workspace", lambda *_args: (_ for _ in ()).throw(RuntimeError("workspace failed")))
+    monkeypatch.setattr(routes, "set_last_workspace", lambda *_args, **_kw: (_ for _ in ()).throw(RuntimeError("workspace failed")))
 
     class FakeThread:
         def __init__(self, *args, **kwargs):

tests/test_issue6751_api_content_agent_replay.py

@@ -435,7 +435,7 @@ def test_issue6751_sync_chat_agent_receives_original_api_content_bytes(monkeypat
     monkeypatch.setattr(routes, "title_from", models.title_from)
     monkeypatch.setattr(config, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
     monkeypatch.setattr(routes, "get_config", lambda: {"model": "test-model", "provider": "test-provider"})
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path)
     monkeypatch.setattr(routes, "load_settings", lambda: {})
     monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: [])
 
@@ -526,7 +526,7 @@ def test_issue6751_json_import_strips_internal_aliases_before_persistence(monkey
     monkeypatch.setattr(models, "SESSION_INDEX_FILE", state_dir / "session_index.json")
     monkeypatch.setattr(models, "SESSIONS", sessions)
     monkeypatch.setattr(routes, "SESSIONS", sessions)
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path)
     monkeypatch.setattr(routes, "get_active_profile_name", lambda: "default")
     monkeypatch.setattr(routes, "publish_session_list_changed", lambda *args, **kwargs: None)
     monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
@@ -625,7 +625,7 @@ def test_issue6751_json_import_nested_tool_calls_are_removed_at_agent_boundary(
     monkeypatch.setattr(models, "SESSIONS", sessions)
     monkeypatch.setattr(routes, "SESSION_INDEX_FILE", state_dir / "session_index.json")
     monkeypatch.setattr(routes, "SESSIONS", sessions)
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path)
     monkeypatch.setattr(routes, "get_active_profile_name", lambda: "default")
     monkeypatch.setattr(routes, "publish_session_list_changed", lambda *args, **kwargs: None)
     monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
@@ -784,7 +784,7 @@ def test_issue6751_json_import_rejects_non_list_session_tool_calls(monkeypatch,
     monkeypatch.setattr(models, "SESSIONS", sessions)
     monkeypatch.setattr(routes, "SESSION_INDEX_FILE", state_dir / "session_index.json")
     monkeypatch.setattr(routes, "SESSIONS", sessions)
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path)
     monkeypatch.setattr(config, "load_settings", lambda: {"api_redact_enabled": False})
     captured = {}
 

tests/test_issue7168_round6_default_isolated_clamp.py

@@ -0,0 +1,294 @@
+"""Re-gate round 6 (#7168): the logical "default" name must honor isolated-mode clamping.
+
+Maintainer round-6 re-gate CORE (one remaining instance of the isolation
+class): ``_resolve_profile_home_param("default")`` short-circuited to
+``_DEFAULT_HERMES_HOME`` BEFORE delegating to
+``api.profiles.get_hermes_home_for_profile()``, so it never reached the
+isolated-mode clamp in ``_resolve_profile_home_for_name`` (which pins every
+lookup to ``_INITIAL_HERMES_HOME`` when ``HERMES_WEBUI_ISOLATED_PROFILE`` is
+enabled). In an isolated deployment pinned at ``<base>/profiles/default``, a
+session created with ``profile="default"`` therefore used the BASE root
+home's workspace + config instead of the pinned one.
+
+Fix under test: the literal shortcut is gone — the logical string ``"default"``
+flows through ``get_hermes_home_for_profile()`` like every other id, so the
+clamp applies. Literal-default routing to the global state files is RETAINED
+(canonical ``_is_default_profile_home`` identity), asserted by the fallback
+tests below.
+"""
+
+import pytest
+
+import api.workspace as workspace
+from api import profiles
+
+
+@pytest.fixture
+def isolated_default_pinned(tmp_path, monkeypatch):
+    """Isolated-mode deployment pinned at <base>/profiles/default.
+
+    Mirrors the gate reproduction: HERMES_WEBUI_ISOLATED_PROFILE=1 with a
+    profile-shaped _INITIAL_HERMES_HOME whose directory NAME is literally
+    'default' (the exact shape where base and pin diverge). Base home gets a
+    DISTINCT config/workspace marker so any leak of the root home is visible.
+    """
+    base = tmp_path / ".hermes"
+    pinned = base / "profiles" / "default"
+    pinned.mkdir(parents=True)
+    (pinned / "webui_state").mkdir()
+
+    # Distinct base-vs-pinned configs: only the PINNED one carries the marker.
+    (pinned / "config.yaml").write_text(
+        "workspace: /srv/pinned-workspace\n", encoding="utf-8"
+    )
+
+    monkeypatch.setenv("HERMES_WEBUI_ISOLATED_PROFILE", "1")
+    monkeypatch.setattr(profiles, "_INITIAL_ISOLATED_PROFILE_OPT_IN", "1")
+    monkeypatch.setattr(profiles, "_INITIAL_HERMES_HOME", str(pinned))
+    monkeypatch.setattr(profiles, "_DEFAULT_HERMES_HOME", base)
+    monkeypatch.setattr(profiles, "_LIST_PROFILES_CACHE", None)
+    # Hermetic w.r.t. the runner's real state dir / remote terminal config.
+    monkeypatch.setattr(workspace, "_remote_terminal_cwd", lambda profile=None: None)
+    return {"base": base, "pinned": pinned, "tmp": tmp_path}
+
+
+class TestRound6DefaultHonorsIsolationClamp:
+    """The logical "default" name resolves through the clamped delegated path."""
+
+    def test_resolver_pins_default_in_isolated_mode(self, isolated_default_pinned):
+        env = isolated_default_pinned
+        assert profiles._is_isolated_profile_mode() is True
+        assert (
+            profiles.get_hermes_home_for_profile("default") == env["pinned"]
+        )  # pre-existing correct behavior (the clamp)
+        got = workspace._resolve_profile_home_param("default")
+        assert got == env["pinned"].resolve(), (
+            f"_resolve_profile_home_param('default') must reach the isolated-mode "
+            f"clamp and return the PINNED home {env['pinned']}, got {got}"
+        )
+        assert got != env["base"], "must not resolve to the base/root home"
+
+    def test_get_last_workspace_uses_global_state_not_base(self, isolated_default_pinned, monkeypatch):
+        """Round 7 contract: 'default' STATE lives in the GLOBAL files only.
+
+        Round-6 follow-up: once the resolver pins config/path resolution at
+        the isolated home, the canonical-home check in _profile_state_dir
+        split explicit profile="default" state I/O onto {pinned}/webui_state/
+        while ambient calls kept the global dir. Round 7 restores ONE state
+        authority: literal "default" reads/writes the global files, so this
+        test now asserts the global binding WINS and the BASE home's
+        webui_state is never consulted.
+        """
+        env = isolated_default_pinned
+        # The authoritative GLOBAL binding...
+        global_ws = env["tmp"] / "srv" / "global-ws"
+        global_ws.mkdir(parents=True)
+        global_lw = env["tmp"] / "global-state"
+        global_lw.mkdir(parents=True)
+        # Both global state files must move TOGETHER: in production
+        # _GLOBAL_WS_FILE and _GLOBAL_LW_FILE share one state dir.
+        monkeypatch.setattr(workspace, "_GLOBAL_WS_FILE", global_lw / "workspaces.json")
+        monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt")
+        (global_lw / "last_workspace.txt").write_text(str(global_ws), encoding="utf-8")
+
+        # ...and a POISONED base binding that must NOT win. Point the base
+        # home's own webui_state at a foreign path to prove no base read.
+        poisoned_ws = env["tmp"] / "srv" / "poisoned-base-ws"
+        poisoned_ws.mkdir(parents=True)
+        base_state = env["base"] / "webui_state"
+        base_state.mkdir(parents=True)
+        (base_state / "last_workspace.txt").write_text(str(poisoned_ws), encoding="utf-8")
+
+        got = workspace.get_last_workspace(profile="default")
+        assert str(got) == str(global_ws.resolve()), (
+            f"profile='default' in isolated mode must read the GLOBAL state "
+            f"authority ({global_lw / 'last_workspace.txt'}), got {got!r}"
+        )
+
+    def test_get_last_workspace_global_fallback_only_when_canonically_default(
+        self, isolated_default_pinned, monkeypatch
+    ):
+        """No profile-local state + non-pinned layout → legacy global fallback.
+
+        Literal-default routing to _GLOBAL_LW_FILE is retained OUTSIDE isolated
+        mode: when the resolved default home IS canonically the default home,
+        get_last_workspace(profile="default") still reads the global file
+        (historical behavior, re-gate round 3 contract).
+        """
+        env = isolated_default_pinned
+        # Disable isolation so 'default' resolves canonically to the base home.
+        monkeypatch.delenv("HERMES_WEBUI_ISOLATED_PROFILE", raising=False)
+        monkeypatch.setattr(profiles, "_INITIAL_ISOLATED_PROFILE_OPT_IN", "")
+        monkeypatch.setattr(profiles, "_INITIAL_HERMES_HOME", str(env["base"]))
+        global_lw = env["tmp"] / "global-state"
+        global_lw.mkdir(parents=True)
+        monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt")
+
+        shared_ws = env["tmp"] / "srv" / "shared-ws"
+        shared_ws.mkdir(parents=True)
+        (global_lw / "last_workspace.txt").write_text(str(shared_ws), encoding="utf-8")
+
+        # The conftest autouse cleanup fixture writes TEST_STATE_DIR /
+        # last_workspace.txt at every teardown; for a canonically-default
+        # profile that IS the profile-local tier and would shadow the global
+        # file under test. Pin the profile-local tier to an absent path so the
+        # GLOBAL fallback is what actually gets exercised.
+        monkeypatch.setattr(
+            workspace,
+            "_last_workspace_file_for_profile",
+            lambda p=None: env["tmp"] / "absent-webui_state" / "last_workspace.txt",
+        )
+
+        got = workspace.get_last_workspace(profile="default")
+        assert str(got) == str(shared_ws), (
+            "outside isolated mode, profile='default' keeps the legacy global "
+            "fallback (canonical identity retained after the round-6 fix)"
+        )
+
+    def test_new_session_binds_global_state_workspace(self, isolated_default_pinned, monkeypatch):
+        """Session creation with profile='default' honors the global state
+        authority and never leaks onto the base home (round-7 contract)."""
+        import api.models as models
+
+        env = isolated_default_pinned
+        bound_ws = env["tmp"] / "srv" / "bound-session-ws"
+        bound_ws.mkdir(parents=True)
+        global_lw = env["tmp"] / "global-state"
+        global_lw.mkdir(parents=True)
+        # Both global state files must move TOGETHER: in production
+        # _GLOBAL_WS_FILE and _GLOBAL_LW_FILE share one state dir.
+        monkeypatch.setattr(workspace, "_GLOBAL_WS_FILE", global_lw / "workspaces.json")
+        monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", global_lw / "last_workspace.txt")
+        (global_lw / "last_workspace.txt").write_text(str(bound_ws), encoding="utf-8")
+
+        # POISONED base-home binding must not win.
+        poisoned_ws = env["tmp"] / "srv" / "poisoned-base-ws"
+        poisoned_ws.mkdir(parents=True)
+        base_state = env["base"] / "webui_state"
+        base_state.mkdir(parents=True)
+        (base_state / "last_workspace.txt").write_text(str(poisoned_ws), encoding="utf-8")
+
+        s = models.new_session(profile="default")
+        assert s.profile == "default"
+        assert str(s.workspace) == str(bound_ws.resolve()), (
+            f"new_session(profile='default') in isolated mode must bind the "
+            f"GLOBAL state authority's workspace, got {s.workspace!r}"
+        )
+        assert str(s.workspace) != str(poisoned_ws.resolve()), (
+            "must never leak onto the BASE home's webui_state binding"
+        )
+
+    def test_session_config_reads_pinned_config(
+        self, isolated_default_pinned, monkeypatch
+    ):
+        """Config for the resolved 'default' home comes from the PINNED home.
+
+        In isolated mode the pinned home IS the active home, so
+        get_config_for_profile_home takes the ambient-match branch — which is
+        exactly the authority we must verify: the ambient resolver (and its
+        HERMES_CONFIG_PATH) must be anchored at the PINNED home, never at the
+        base/root one. Point the authoritative config INSIDE the pinned home;
+        had the resolver leaked to the base home, this config would not win.
+        """
+        from api import config as cfg_mod
+        from api.config import get_config_for_profile_home
+
+        env = isolated_default_pinned
+        monkeypatch.setenv("HERMES_CONFIG_PATH", str(env["pinned"] / "config.yaml"))
+        cfg_mod.reload_config()
+
+        home = workspace._resolve_profile_home_param("default")
+        assert home == env["pinned"].resolve()
+        got = get_config_for_profile_home(home)
+        assert got.get("workspace") == "/srv/pinned-workspace", (
+            "config for the resolved 'default' home must be anchored at the "
+            "PINNED profile's home/config, never the base root's"
+        )
+        # The BASE root has no config.yaml — a leak onto the base home would
+        # have produced the defaults-only dict (no 'workspace' key).
+
+
+class TestRound7DefaultStateRouting:
+    """Explicit and ambient "default" STATE I/O share ONE authority (round 7).
+
+    Round-6 CORE follow-up: routing the literal "default" name through the
+    delegated resolver made the canonical-home check in
+    ``_profile_state_dir`` send EXPLICIT ``profile="default"`` state
+    reads/writes to ``{pinned}/webui_state/`` while AMBIENT calls kept using
+    the global state dir — splitting saved workspaces between two
+    authorities and hiding the pre-upgrade list. Fix under test:
+    literal-"default" STATE routing stays on the global files, while config
+    and workspace PATH resolution keep the pinned home.
+    """
+
+    def _seed_global_state(self, env, monkeypatch):
+        """Point both global state files at an isolated tmp dir."""
+        g = env["tmp"] / "global-state"
+        g.mkdir(parents=True, exist_ok=True)
+        monkeypatch.setattr(workspace, "_GLOBAL_WS_FILE", g / "workspaces.json")
+        monkeypatch.setattr(workspace, "_GLOBAL_LW_FILE", g / "last_workspace.txt")
+        return g
+
+    def test_explicit_default_state_dir_is_global(self, isolated_default_pinned, monkeypatch):
+        env = isolated_default_pinned
+        g = self._seed_global_state(env, monkeypatch)
+        assert workspace._profile_state_dir(profile="default") == g, (
+            "explicit profile='default' must route STATE to the global dir "
+            "(same authority as ambient), not {pinned}/webui_state/"
+        )
+        assert workspace._workspaces_file("default") == g / "workspaces.json"
+        assert workspace._last_workspace_file("default") == g / "last_workspace.txt"
+
+    def test_explicit_and_ambient_share_workspaces_authority(
+        self, isolated_default_pinned, monkeypatch
+    ):
+        import json
+
+        env = isolated_default_pinned
+        g = self._seed_global_state(env, monkeypatch)
+
+        # Pre-upgrade saved list lives ONLY in the global file.
+        pre_dir = env["tmp"] / "srv" / "pre-upgrade"
+        pre_dir.mkdir(parents=True)
+        pre = [{"path": str(pre_dir), "name": "Pre"}]
+        (g / "workspaces.json").write_text(
+            json.dumps(pre, ensure_ascii=False), encoding="utf-8"
+        )
+
+        # Explicit profile="default" reads the pre-upgrade list...
+        assert workspace.load_workspaces(profile="default") == pre
+
+        # ...and explicit writes land in the SAME global file.
+        added = env["tmp"] / "srv" / "added"
+        added.mkdir(parents=True)
+        updated = pre + [{"path": str(added), "name": "Added"}]
+        workspace.save_workspaces(updated, profile="default")
+        assert json.loads((g / "workspaces.json").read_text(encoding="utf-8")) == updated
+
+        # Ambient calls see the identical authority — no split.
+        assert workspace._profile_state_dir(None) == g
+        assert workspace.load_workspaces() == updated
+
+        # No divergent webui_state authority may appear under the pinned home.
+        assert not (env["pinned"] / "webui_state" / "workspaces.json").exists()
+        assert not (env["pinned"] / "webui_state" / "last_workspace.txt").exists()
+
+    def test_last_workspace_round_trip_shared_explicit_ambient(
+        self, isolated_default_pinned, monkeypatch
+    ):
+        env = isolated_default_pinned
+        g = self._seed_global_state(env, monkeypatch)
+
+        ws = env["tmp"] / "srv" / "bound-ws"
+        ws.mkdir(parents=True)
+
+        workspace.set_last_workspace(str(ws), profile="default")
+        assert (g / "last_workspace.txt").exists(), (
+            "set_last_workspace('default') must write the GLOBAL file"
+        )
+        assert not (env["pinned"] / "webui_state" / "last_workspace.txt").exists()
+
+        assert workspace.get_last_workspace(profile="default") == str(ws)
+        # Ambient (isolated mode's active profile is literally "default")
+        # resolves to the same binding.
+        assert workspace.get_last_workspace() == str(ws)

tests/test_issue7404_models_discovered_not_allowlist.py

@@ -0,0 +1,178 @@
+"""Regression coverage for #7404: models_discovered suppresses live catalog.
+
+When a provider sets ``models_discovered: true`` in its config alongside a
+``models:`` dict of per-model metadata, WebUI must not treat the dict as a
+strict allowlist.  It should fall through to the live ``/v1/models`` probe,
+matching the upstream Hermes Agent behaviour.
+"""
+
+
+def _provider_group(payload: dict, provider_id: str) -> dict:
+    for group in payload.get("groups", []):
+        if group.get("provider_id") == provider_id:
+            return group
+    raise AssertionError(f"provider group {provider_id!r} not found: {payload.get('groups')!r}")
+
+
+def test_models_discovered_skips_config_allowlist_and_uses_live_catalog(monkeypatch, tmp_path):
+    import api.config as config
+
+    cfg = {
+        "model": {"default": "model-a", "provider": "custom-llm"},
+        "providers": {
+            "custom-llm": {
+                "name": "Custom LLM",
+                "api": "https://llm.example.com",
+                "transport": "chat_completions",
+                "models_discovered": True,
+                "models": {
+                    "model-a": {"supports_vision": True},
+                    "model-b": {"context_length": 128000},
+                },
+            }
+        },
+    }
+
+    live_ids = ["model-a", "model-b", "model-c", "model-d", "model-e"]
+
+    monkeypatch.setattr(config, "cfg", cfg, raising=False)
+    monkeypatch.setattr(config, "_get_config_path", lambda: tmp_path / "config.yaml")
+    monkeypatch.setattr(config, "_get_auth_store_path", lambda: tmp_path / "auth.json")
+    monkeypatch.setattr(config, "_get_models_cache_path", lambda: tmp_path / "models_cache.json")
+    monkeypatch.setattr(config, "_models_cache_source_fingerprint", lambda: {"test": "fingerprint"})
+    monkeypatch.setattr(config, "reload_config_if_stale", lambda: None)
+    monkeypatch.setattr(config, "reload_config", lambda: None)
+    monkeypatch.setattr(config, "_cfg_mtime", 0.0, raising=False)
+    monkeypatch.setattr(config, "_LIVE_REBUILD_BUDGET_SECONDS", 0.0, raising=False)
+    monkeypatch.setattr(
+        config, "_read_live_provider_model_ids",
+        lambda pid: live_ids if pid == "custom-llm" else [],
+    )
+
+    config.invalidate_models_cache()
+    payload = config.get_available_models(force_refresh=True)
+    group = _provider_group(payload, "custom-llm")
+    ids = [m["id"] for m in group["models"]]
+
+    assert ids == live_ids
+
+
+def _setup_provider(monkeypatch, tmp_path, provider_cfg, live_ids):
+    """Wire config with a single custom-llm provider and a controllable live catalog."""
+    import api.config as config
+
+    cfg = {
+        "model": {"default": "model-a", "provider": "custom-llm"},
+        "providers": {"custom-llm": provider_cfg},
+    }
+    monkeypatch.setattr(config, "cfg", cfg, raising=False)
+    monkeypatch.setattr(config, "_get_config_path", lambda: tmp_path / "config.yaml")
+    monkeypatch.setattr(config, "_get_auth_store_path", lambda: tmp_path / "auth.json")
+    monkeypatch.setattr(config, "_get_models_cache_path", lambda: tmp_path / "models_cache.json")
+    monkeypatch.setattr(config, "_models_cache_source_fingerprint", lambda: {"test": "fingerprint"})
+    monkeypatch.setattr(config, "reload_config_if_stale", lambda: None)
+    monkeypatch.setattr(config, "reload_config", lambda: None)
+    monkeypatch.setattr(config, "_cfg_mtime", 0.0, raising=False)
+    monkeypatch.setattr(config, "_LIVE_REBUILD_BUDGET_SECONDS", 0.0, raising=False)
+    monkeypatch.setattr(
+        config, "_read_live_provider_model_ids",
+        lambda pid: live_ids if pid == "custom-llm" else [],
+    )
+    config.invalidate_models_cache()
+    return config
+
+
+def test_discover_models_false_pins_configured_allowlist_despite_discovered_flag(monkeypatch, tmp_path):
+    """`discover_models: false` overrides `models_discovered: true` (explicit opt-out wins).
+
+    A provider that persisted a discovered catalog but then pinned it with
+    ``discover_models: false`` must keep exactly its configured ``models:`` and NOT
+    probe/expose the broader live catalog.
+    """
+    provider_cfg = {
+        "name": "Custom LLM",
+        "api": "https://llm.example.com",
+        "transport": "chat_completions",
+        "models_discovered": True,
+        "discover_models": False,
+        "models": {
+            "model-a": {"supports_vision": True},
+            "model-b": {"context_length": 128000},
+        },
+    }
+    live_ids = ["model-a", "model-b", "model-c", "model-d", "model-e"]
+    config = _setup_provider(monkeypatch, tmp_path, provider_cfg, live_ids)
+
+    payload = config.get_available_models(force_refresh=True)
+    group = _provider_group(payload, "custom-llm")
+    ids = [m["id"] for m in group["models"]]
+
+    assert ids == ["model-a", "model-b"]
+
+
+def test_discover_models_false_string_form_also_pins(monkeypatch, tmp_path):
+    """Agent-compatible string opt-out (`discover_models: "false"`) also pins the allowlist."""
+    provider_cfg = {
+        "name": "Custom LLM",
+        "api": "https://llm.example.com",
+        "transport": "chat_completions",
+        "models_discovered": True,
+        "discover_models": "false",
+        "models": {"model-a": {}, "model-b": {}},
+    }
+    live_ids = ["model-a", "model-b", "model-c"]
+    config = _setup_provider(monkeypatch, tmp_path, provider_cfg, live_ids)
+
+    payload = config.get_available_models(force_refresh=True)
+    group = _provider_group(payload, "custom-llm")
+    ids = [m["id"] for m in group["models"]]
+
+    assert ids == ["model-a", "model-b"]
+
+
+def test_discovered_catalog_survives_transient_live_probe_failure(monkeypatch, tmp_path):
+    """A transient empty live probe must not drop a discovered provider's configured models.
+
+    With ``models_discovered: true`` (and discovery allowed), the live catalog is
+    authoritative — but when the probe transiently returns nothing, the provider group
+    must fall back to the configured discovered IDs rather than vanishing (which would be
+    cached empty for up to 24h). A provider absent from the static built-in catalog is
+    the exact case #7404's fix must not regress.
+    """
+    provider_cfg = {
+        "name": "Custom LLM",
+        "api": "https://llm.example.com",
+        "transport": "chat_completions",
+        "models_discovered": True,
+        "models": {"model-a": {"supports_vision": True}, "model-b": {"context_length": 128000}},
+    }
+    # Live probe returns NOTHING (transient failure), provider not in _PROVIDER_MODELS.
+    config = _setup_provider(monkeypatch, tmp_path, provider_cfg, [])
+
+    payload = config.get_available_models(force_refresh=True)
+    group = _provider_group(payload, "custom-llm")
+    ids = [m["id"] for m in group["models"]]
+
+    assert ids == ["model-a", "model-b"]
+
+
+def test_discovered_flag_without_models_key_and_empty_probe_does_not_500(monkeypatch, tmp_path):
+    """models_discovered:true with NO models: key + empty live probe must not raise (KeyError->500).
+
+    _provider_models_are_discovered_catalog is True on models_discovered alone, so the
+    probe-failure fallback must read models via .get() (absent -> no configured rows),
+    degrading to the static catalog rather than crashing /api/models.
+    """
+    provider_cfg = {
+        "name": "Custom LLM",
+        "api": "https://llm.example.com",
+        "transport": "chat_completions",
+        "models_discovered": True,
+        # no "models" key at all
+    }
+    config = _setup_provider(monkeypatch, tmp_path, provider_cfg, [])
+
+    # Must not raise; provider absent from the static catalog degrades to an empty
+    # group (filtered out) rather than a 500.
+    payload = config.get_available_models(force_refresh=True)
+    assert isinstance(payload.get("groups"), list)

tests/test_issue7540_codex_catalog_fingerprint.py

@@ -0,0 +1,386 @@
+"""Regression tests for issue #7540 — Codex models_cache.json churn must not
+bust a valid /api/models catalog cache and force a live rebuild on session open.
+
+Bug shape
+---------
+``_models_cache_source_fingerprint()`` covers the small local catalogs the
+catalog depends on, including Codex's ``~/.codex/models_cache.json``, and
+fingerprinted that file by ``mtime_ns`` + ``size``
+(``_models_cache_file_fingerprint``, #2443).  Codex rewrites that file on its
+own refresh timer, and each rewrite bumps both stat fields even when the model
+payload is byte-identical — only ``fetched_at`` / ``updated_at`` move.  So:
+
+    Codex refresh -> fingerprint differs
+    -> ``_get_fresh_memory_models_cache()`` rejects the warm 24h snapshot
+    -> the session visit (#4756) then finds ``_is_loadable_disk_cache()``
+       rejecting the disk snapshot for the same reason
+    -> ``get_available_models(force_refresh=True)`` -> serial live provider
+       probes on the session-open path (the multi-second hang in the report).
+
+Fix (maintainer recommendation on #7540: semantic fingerprint)
+--------------------------------------------------------------
+``_codex_models_cache_fingerprint()`` hashes the file's *content* with the
+refresh-timestamp keys deny-listed, mirroring the auth.json fix
+(``_auth_store_semantic_fingerprint``, RCA t_16551f61).  Metadata churn is
+invisible to the fingerprint, while anything that actually feeds the Codex
+model list we merge still changes it.
+
+These tests are INVARIANTS, not change-detectors:
+
+  * churn-immune           — a ``fetched_at``-only rewrite keeps the
+                             fingerprint identical, keeps a previously-valid
+                             disk cache loadable, and keeps a session visit
+                             served from memory (no live rebuild).
+  * real-change-invalidates — client_version / etag / models[] / an unknown
+                             future field each flip the fingerprint AND reject
+                             the previously-valid disk cache, so the fix cannot
+                             over-stabilise into serving a stale catalog.
+"""
+
+import copy
+import json
+import os
+import time
+
+import pytest
+
+import api.config as config
+
+
+_TS_1 = "2026-09-13T12:00:00+00:00"
+_TS_2 = "2026-09-13T15:45:11+00:00"
+
+
+class _LiveRebuildReached(BaseException):
+    """Sentinel raised if a session visit escalates to a live provider rebuild.
+
+    Deliberately NOT an ``Exception``: the session-visit path catches broad
+    ``Exception`` around its ``force_refresh`` call and falls back to a stale
+    on-disk snapshot, which would mask the regression.
+    """
+
+
+def _codex_cache(fetched_at=_TS_1, *, client_version="0.51.0",
+                 etag='"codex-etag-1"', models=None, extra_top=None):
+    """Build a realistic ~/.codex/models_cache.json payload."""
+    if models is None:
+        models = [
+            {"slug": "gpt-5.3-codex", "visibility": "list", "priority": 1},
+            {"slug": "gpt-5.1-codex-mini", "visibility": "list", "priority": 2},
+        ]
+    payload = {
+        "fetched_at": fetched_at,
+        "client_version": client_version,
+        "etag": etag,
+        "models": models,
+    }
+    if extra_top:
+        payload.update(extra_top)
+    return payload
+
+
+def _write_codex(tmp_path, monkeypatch, payload) -> "os.PathLike":
+    """Write the Codex cache into an isolated CODEX_HOME and point config at it."""
+    codex_home = tmp_path / "codex_home"
+    codex_home.mkdir(parents=True, exist_ok=True)
+    cache_path = codex_home / "models_cache.json"
+    cache_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+    monkeypatch.setenv("CODEX_HOME", str(codex_home))
+    return cache_path
+
+
+def _rewrite(path, payload):
+    """Rewrite the Codex cache and deterministically bump its mtime.
+
+    ``os.utime`` is used instead of ``time.sleep`` so the mtime_ns change is
+    guaranteed on every filesystem/timestamp-resolution combination — the test
+    must prove the *old* stat fingerprint churned, not rely on clock luck.
+    """
+    path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+    bumped = path.stat().st_mtime + 7.0
+    os.utime(path, (bumped, bumped))
+    return path
+
+
+def _catalog_payload():
+    return {
+        "active_provider": "codex",
+        "default_model": "gpt-5.3-codex",
+        "configured_model_badges": {"gpt-5.3-codex": "Codex"},
+        "groups": [{"name": "Codex", "models": ["gpt-5.3-codex"]}],
+    }
+
+
+# ── churn-immunity invariant ────────────────────────────────────────────────
+
+
+def test_codex_fetched_at_only_rewrite_keeps_fingerprint_identical(tmp_path, monkeypatch):
+    """Codex's own refresh moves ``fetched_at`` only — same models, same size."""
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache(fetched_at=_TS_1))
+    fp_before = config._codex_models_cache_fingerprint(p)
+    st_before = p.stat()
+
+    _rewrite(p, _codex_cache(fetched_at=_TS_2))
+    fp_after = config._codex_models_cache_fingerprint(p)
+    st_after = p.stat()
+
+    # Pre-condition: this really is the mtime/size churn from the report — the
+    # old stat-based fingerprint WOULD have changed.
+    assert (st_before.st_mtime_ns, st_before.st_size) != (
+        st_after.st_mtime_ns, st_after.st_size), "test setup: rewrite must churn the stat fields"
+    # Invariant: the semantic fingerprint does not.
+    assert fp_before == fp_after, (
+        "A Codex refresh that only moves fetched_at must NOT change the catalog "
+        "fingerprint (#7540) — otherwise the 24h cache is rejected and the next "
+        "session visit pays a live rebuild"
+    )
+    assert "semantic_sha256" in fp_before
+
+
+def test_unsorted_codex_payload_key_order_keeps_fingerprint_identical(tmp_path, monkeypatch):
+    """Key order is not semantic: a reordered but otherwise identical file must
+    not bust the cache."""
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache())
+    fp_before = config._codex_models_cache_fingerprint(p)
+
+    reordered = {
+        "models": _codex_cache()["models"],
+        "etag": _codex_cache()["etag"],
+        "client_version": _codex_cache()["client_version"],
+        "fetched_at": _codex_cache()["fetched_at"],
+    }
+    _rewrite(p, reordered)
+
+    assert config._codex_models_cache_fingerprint(p) == fp_before
+
+
+def test_codex_metadata_churn_does_not_reject_valid_disk_models_cache(tmp_path, monkeypatch):
+    """End-to-end: the disk snapshot a warm server wrote must still load after
+    Codex refreshed its own cache."""
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache(fetched_at=_TS_1))
+    monkeypatch.setattr(config, "_models_cache_path", tmp_path / "models_cache.json")
+    config._save_models_cache_to_disk(_catalog_payload())
+    assert config._load_models_cache_from_disk() is not None
+
+    _rewrite(p, _codex_cache(fetched_at=_TS_2))
+
+    assert config._load_models_cache_from_disk() is not None, (
+        "Codex metadata churn must NOT reject a valid on-disk models cache "
+        "(#7540) — that rejection is what triggers the live rebuild"
+    )
+
+
+def test_session_visit_after_codex_refresh_needs_no_live_rebuild(tmp_path, monkeypatch):
+    """The user-visible symptom: open a conversation after Codex refreshed.
+
+    The warm in-memory snapshot must still be served; a rebuild here is the
+    multi-second session-open hang in the report.
+    """
+    _write_codex(tmp_path, monkeypatch, _codex_cache(fetched_at=_TS_1))
+    monkeypatch.setattr(config, "_models_cache_path", tmp_path / "models_cache.json")
+    payload = _catalog_payload()
+    config._save_models_cache_to_disk(payload)
+
+    # Warm the memory cache the way a normal /api/models call leaves it.
+    monkeypatch.setattr(config, "_available_models_cache", copy.deepcopy(payload))
+    monkeypatch.setattr(config, "_available_models_cache_ts", time.monotonic())
+    monkeypatch.setattr(
+        config,
+        "_available_models_cache_source_fingerprint",
+        config._models_cache_source_fingerprint(),
+    )
+
+    # Codex's refresh timer fires between two session opens.
+    _rewrite(tmp_path / "codex_home" / "models_cache.json",
+             _codex_cache(fetched_at=_TS_2))
+
+    def _boom(**_kwargs):
+        # BaseException, not Exception: get_available_models_for_session_visit
+        # swallows exceptions from its force_refresh path and falls back to the
+        # on-disk stale snapshot, so an AssertionError sentinel would be eaten
+        # and the test would pass while quietly taking the slow path.
+        raise _LiveRebuildReached(
+            "session visit fell through to the live rebuild: Codex fetched_at "
+            "churn invalidated a still-valid catalog cache (#7540)"
+        )
+
+    monkeypatch.setattr(config, "get_available_models", _boom)
+
+    result = config.get_available_models_for_session_visit()
+
+    assert result is not None
+    assert result["default_model"] == "gpt-5.3-codex"
+
+
+def test_warm_memory_cache_survives_codex_churn(tmp_path, monkeypatch):
+    """Isolates the invalidation point itself (the memory-cache rejection that
+    precedes the disk load and the force_refresh)."""
+    _write_codex(tmp_path, monkeypatch, _codex_cache(fetched_at=_TS_1))
+    monkeypatch.setattr(config, "_models_cache_path", tmp_path / "models_cache.json")
+    payload = _catalog_payload()
+    config._save_models_cache_to_disk(payload)
+    monkeypatch.setattr(config, "_available_models_cache", copy.deepcopy(payload))
+    monkeypatch.setattr(config, "_available_models_cache_ts", time.monotonic())
+    monkeypatch.setattr(
+        config,
+        "_available_models_cache_source_fingerprint",
+        config._models_cache_source_fingerprint(),
+    )
+
+    _rewrite(tmp_path / "codex_home" / "models_cache.json",
+             _codex_cache(fetched_at=_TS_2))
+
+    assert config._get_fresh_memory_models_cache(time.monotonic()) is not None, (
+        "The warm in-memory catalog cache must survive a Codex fetched_at-only "
+        "refresh (#7540)"
+    )
+
+
+# ── real-change-invalidates invariant ───────────────────────────────────────
+
+
+def test_new_codex_model_changes_fingerprint(tmp_path, monkeypatch):
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache())
+    fp_before = config._codex_models_cache_fingerprint(p)
+
+    models = _codex_cache()["models"] + [
+        {"slug": "gpt-5.4-codex", "visibility": "list", "priority": 3},
+    ]
+    _rewrite(p, _codex_cache(models=models))
+
+    assert config._codex_models_cache_fingerprint(p) != fp_before, (
+        "A new Codex model changes the merged catalog and MUST bust the cache"
+    )
+
+
+def test_model_visibility_change_changes_fingerprint(tmp_path, monkeypatch):
+    """``visibility`` gates whether the slug is surfaced — a hide/show flip is a
+    real catalog change at unchanged file size."""
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache())
+    fp_before = config._codex_models_cache_fingerprint(p)
+
+    models = copy.deepcopy(_codex_cache()["models"])
+    models[0]["visibility"] = "hidden"
+    _rewrite(p, _codex_cache(models=models))
+
+    assert config._codex_models_cache_fingerprint(p) != fp_before
+
+
+@pytest.mark.parametrize(
+    "changed",
+    [
+        {"client_version": "0.52.0"},
+        {"etag": '"codex-etag-2"'},
+        {"models": [{"slug": "gpt-5.3-codex", "priority": 9}]},
+    ],
+)
+def test_codex_catalog_fields_stay_in_fingerprint(tmp_path, monkeypatch, changed):
+    """Everything that actually feeds the Codex model list stays covered."""
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache())
+    fp_before = config._codex_models_cache_fingerprint(p)
+
+    _rewrite(p, _codex_cache(**changed))
+
+    assert config._codex_models_cache_fingerprint(p) != fp_before
+
+
+def test_unknown_codex_field_stays_in_fingerprint(tmp_path, monkeypatch):
+    """Deny-list (not allow-list) safety: an unknown field is NOT stripped, so a
+    hypothetical future model-gating field still busts the cache."""
+    p = _write_codex(tmp_path, monkeypatch,
+                     _codex_cache(extra_top={"some_future_model_gate": "v1"}))
+    fp_before = config._codex_models_cache_fingerprint(p)
+
+    _rewrite(p, _codex_cache(extra_top={"some_future_model_gate": "v2"}))
+
+    assert config._codex_models_cache_fingerprint(p) != fp_before, (
+        "An unknown (non-deny-listed) field must remain in the fingerprint — the "
+        "deny-list must fail safe toward over-invalidation"
+    )
+
+
+def test_real_codex_change_still_rejects_previously_valid_disk_cache(tmp_path, monkeypatch):
+    """Anti-over-stabilisation mirror of the churn test: a disk snapshot that WAS
+    valid must be rejected once a genuine Codex model change lands."""
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache())
+    monkeypatch.setattr(config, "_models_cache_path", tmp_path / "models_cache.json")
+    config._save_models_cache_to_disk(_catalog_payload())
+    assert config._load_models_cache_from_disk() is not None
+
+    _rewrite(p, _codex_cache(client_version="0.52.0"))
+
+    assert config._load_models_cache_from_disk() is None, (
+        "A real Codex catalog change MUST reject the stale disk cache — the fix "
+        "must not over-stabilise into serving wrong data"
+    )
+
+
+# ── helper unit guards ──────────────────────────────────────────────────────
+
+
+def test_strip_volatile_codex_fields_is_pure_and_recursive():
+    src = {
+        "fetched_at": "ts",
+        "client_version": "0.51.0",
+        "models": [{"slug": "gpt-5.3-codex", "updated_at": "ts", "priority": 1}],
+    }
+    snapshot = copy.deepcopy(src)
+    out = config._strip_volatile_codex_cache_fields(src)
+
+    assert src == snapshot  # input untouched (pure)
+    assert out == {
+        "client_version": "0.51.0",
+        "models": [{"slug": "gpt-5.3-codex", "priority": 1}],
+    }
+
+
+def test_unparseable_codex_cache_falls_back_to_stat_fingerprint(tmp_path, monkeypatch):
+    """Corrupt / mid-write file: fall back to the old stat fingerprint instead of
+    silently pinning a hash of garbage — behaviour is never less safe than the
+    stat-based version, and a later good file still invalidates."""
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache())
+    p.write_text('{"models": [', encoding="utf-8")
+
+    fp = config._codex_models_cache_fingerprint(p)
+    assert fp.get("semantic") == "unparsed-fallback"
+    assert "mtime_ns" in fp and "size" in fp
+
+    _rewrite(p, _codex_cache(fetched_at=_TS_2))
+    assert config._codex_models_cache_fingerprint(p) != fp
+
+
+def test_missing_codex_cache_fingerprint_is_stable_and_marked(tmp_path, monkeypatch):
+    monkeypatch.setenv("CODEX_HOME", str(tmp_path / "absent"))
+    fp = config._codex_models_cache_fingerprint(tmp_path / "absent" / "models_cache.json")
+
+    assert fp.get("missing") is True
+    assert config._codex_models_cache_fingerprint(
+        tmp_path / "absent" / "models_cache.json") == fp
+
+
+def test_deeply_nested_codex_cache_degrades_to_stat_fallback_without_crashing(tmp_path, monkeypatch):
+    """A strip failure (e.g. RecursionError on a deep tree) must not 500 /api/models.
+
+    The recursive volatile-field strip runs inside the encode try/except, so if
+    it raises (RecursionError on a pathologically deep JSON tree, or any other
+    error) the fingerprint degrades to the stat-based fallback instead of the
+    exception escaping and turning /api/models into an HTTP 500. (#7540 gate
+    finding.) We force the strip to raise directly so the guard is exercised
+    deterministically regardless of the ambient recursion depth.
+    """
+    p = _write_codex(tmp_path, monkeypatch, _codex_cache(fetched_at=_TS_1))
+
+    def _boom(_obj):
+        raise RecursionError("simulated deep-tree strip overflow")
+
+    monkeypatch.setattr(config, "_strip_volatile_codex_cache_fields", _boom)
+
+    # Must not raise; must degrade to the stat-based fingerprint.
+    fp = config._codex_models_cache_fingerprint(p)
+    fp2 = config._codex_models_cache_fingerprint(p)
+
+    assert isinstance(fp, dict)
+    assert fp.get("semantic") == "encode-fallback"
+    assert "mtime_ns" in fp and "size" in fp
+    assert "semantic_sha256" not in fp  # the content hash was NOT produced
+    assert fp2 == fp  # stable across repeated calls on the unchanged file

tests/test_issue7543_title_first_exchange.py

@@ -0,0 +1,165 @@
+"""Regression coverage for nesquena/hermes-webui#7543.
+
+Bug: manual "Regenerate title" failed with missing_exchange for sessions
+whose transcript opens with consecutive user rows (no assistant text before
+the second user turn) — _first_exchange_snippets() aborted at the second
+user message, the aux call was skipped, and the deterministic local fallback
+was persisted (200 + identical wrong title on every retry).
+"""
+
+from unittest.mock import MagicMock
+
+import api.profiles as profiles_api
+import api.streaming as streaming
+
+
+class _ProfileEnv:
+    def __enter__(self):
+        return None
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+
+def _capturing_llm_result(captured, title="LLM Title", status="llm_aux"):
+    # Guard-faithful stand-in for the aux route: mirrors the documented
+    # generate_title_raw_via_aux contract — empty assistant_text is rejected
+    # with missing_exchange (the provider edge), never the selection logic.
+    def _fake(user_text, assistant_text, **kwargs):
+        captured["user_text"] = user_text
+        captured["assistant_text"] = assistant_text
+        if not user_text or not assistant_text:
+            return None, "missing_exchange", ""
+        return title, status, ""
+
+    return _fake
+
+
+def _run_generation(monkeypatch, messages, captured, prefer_latest=False):
+    monkeypatch.setattr(profiles_api, "profile_env_for_background_worker", lambda *a, **k: _ProfileEnv())
+    monkeypatch.setattr(streaming, "_aux_title_generation_enabled", lambda: True)
+    monkeypatch.setattr(streaming, "_generate_llm_session_title_via_aux", _capturing_llm_result(captured))
+    session = MagicMock()
+    session.messages = messages
+    session.session_id = "issue7543"
+    return streaming.generate_session_title_for_session(session, prefer_latest=prefer_latest)
+
+
+# --- Fix A: _first_exchange_snippets scans past consecutive user rows (opt-in) ---
+
+
+def test_first_exchange_snippets_scan_past_consecutive_user_rows():
+    # Channel-backed import/projection shape: opening run of user rows,
+    # first assistant answer only much later (#7543 repro shape).
+    # scan_past_consecutive_users=True is the manual-regen behavior.
+    messages = [
+        {"role": "user", "content": "Opening question about certificates"},
+        {"role": "user", "content": "Follow-up that used to trigger the break"},
+        {"role": "user", "content": "Another queued user turn"},
+        {"role": "assistant", "content": "## Real first answer with substance"},
+    ]
+    user_text, asst_text = streaming._first_exchange_snippets(
+        messages, scan_past_consecutive_users=True
+    )
+    assert user_text == "Opening question about certificates"
+    assert asst_text == "## Real first answer with substance"
+
+
+def test_first_exchange_snippets_default_stops_at_second_user_row():
+    # The DEFAULT (automatic in-stream background-title path) must keep master's
+    # exact behavior: a second populated user row before any assistant text ends
+    # the opening exchange with an empty assistant snippet. This is load-bearing —
+    # _background_title_generation_inputs treats an empty assistant snippet as
+    # "not yet eligible", which keeps the stream teardown on its synchronous
+    # stream_end path (regression guard for test_issue3929 emits_done).
+    messages = [
+        {"role": "user", "content": "Opening question about certificates"},
+        {"role": "user", "content": "Follow-up that used to trigger the break"},
+        {"role": "user", "content": "Another queued user turn"},
+        {"role": "assistant", "content": "## Real first answer with substance"},
+    ]
+    user_text, asst_text = streaming._first_exchange_snippets(messages)
+    assert user_text == "Opening question about certificates"
+    assert asst_text == ""
+
+
+def test_first_exchange_snippets_normal_pair_unchanged():
+    # Classic [user, assistant] opening must keep its exact behavior in BOTH modes.
+    messages = [
+        {"role": "user", "content": "Please fix the stale sidebar title controls"},
+        {"role": "assistant", "content": "I will add a regenerate-title action."},
+        {"role": "user", "content": "Second question"},
+    ]
+    for scan in (False, True):
+        user_text, asst_text = streaming._first_exchange_snippets(
+            messages, scan_past_consecutive_users=scan
+        )
+        assert user_text == "Please fix the stale sidebar title controls"
+        assert asst_text == "I will add a regenerate-title action."
+
+
+def test_first_exchange_snippets_without_any_assistant_text_still_empty():
+    # No assistant text anywhere -> still unusable for the LLM path in BOTH modes;
+    # the missing_exchange rejection in generate_title_raw_via_aux stays intact.
+    messages = [
+        {"role": "user", "content": "Question one"},
+        {"role": "user", "content": "Question two"},
+    ]
+    for scan in (False, True):
+        user_text, asst_text = streaming._first_exchange_snippets(
+            messages, scan_past_consecutive_users=scan
+        )
+        assert user_text == "Question one"
+        assert asst_text == ""
+
+
+def test_issue7543_real_transcript_shape_reaches_llm_path(monkeypatch):
+    captured = {}
+    messages = [
+        {"role": "user", "content": "Wie kann ich auf einem windows server 2025 ein zertifikat erstellen"} ,
+        {"role": "user", "content": "Wächst das Log so nicht in eine unendliche Schleife"},
+        {"role": "assistant", "content": "## Zertifikat für Windows Admin Center mit AD-CS"},
+    ]
+    title, status, _raw = _run_generation(monkeypatch, messages, captured)
+    assert status == "llm_aux"
+    assert title == "LLM Title"
+    assert captured["user_text"].startswith("Wie kann ich auf einem windows server 2025")
+    assert captured["assistant_text"].startswith("## Zertifikat")
+
+
+def test_regenerate_helper_errors_with_real_walkers_when_no_user_text_exists(monkeypatch):
+    """Observable error path through the real walkers: no user text anywhere
+    (orphan assistant rows only) -> empty_user_message, aux never called."""
+    captured = {}
+    messages = [{"role": "assistant", "content": "orphan answer without any user turn"}]
+    title, status, _raw = _run_generation(monkeypatch, messages, captured)
+    assert title is None
+    assert status == "empty_user_message"
+    assert "user_text" not in captured  # aux path never reached
+
+
+def test_regenerate_helper_prefer_latest_path_unchanged(monkeypatch):
+    captured = {}
+    messages = [
+        {"role": "user", "content": "First question"},
+        {"role": "assistant", "content": "First answer"},
+        {"role": "user", "content": "Latest question"},
+        {"role": "assistant", "content": "Latest answer"},
+    ]
+    title, status, _raw = _run_generation(monkeypatch, messages, captured, prefer_latest=True)
+    assert status == "llm_aux"
+    assert captured["user_text"] == "Latest question"
+    assert captured["assistant_text"] == "Latest answer"
+
+
+def test_regenerate_helper_prefer_latest_with_empty_last_user_message_errors(monkeypatch):
+    captured = {}
+    messages = [
+        {"role": "user", "content": "First question"},
+        {"role": "assistant", "content": "First answer"},
+        {"role": "user", "content": "   "},
+    ]
+    title, status, _raw = _run_generation(monkeypatch, messages, captured, prefer_latest=True)
+    assert title is None
+    assert status == "empty_user_message"
+    assert "user_text" not in captured

tests/test_issue_raw_pending_approval_id.py

@@ -0,0 +1,248 @@
+"""Raw id-less local approval entries in ``_pending`` must be actionable.
+
+Regression for the Sep 2026 live incident. When a guarded tool needs approval
+and no gateway notifier is registered for the session (``unregister_gateway_
+notify`` fires at turn end; a queued wakeup/continuation then hits a guard),
+the agent-side fallback ``tools/approval.py::_pending_result`` writes a raw
+dict — ``{command, description, pattern_key, pattern_keys}`` and NOTHING else,
+no ``approval_id``, no ``request_id`` — directly into the shared
+``tools.approval._pending[session_key]`` dict that the WebUI imports.
+
+Before the fix:
+- ``GET /api/approval/pending`` served that raw dict verbatim (reconcile passes
+  non-mirror entries through untouched) with NO ``approval_id``, so the
+  frontend owner-capture (``_captureApprovalResponseOwner``) bailed and every
+  button on the approval card was a no-op;
+- the frontend dismiss (X) only hid the card locally and the 1.5s poll
+  re-rendered it forever — the card could neither be approved nor dismissed;
+- there is no waiter thread behind this shape (``_pending_result`` returns the
+  STOP message immediately), so draining the entry is purely UI hygiene — but
+  leaving it forever means the session shows a permanent phantom card.
+
+After the fix (mint at the reconcile chokepoint):
+- id-less, run-less, non-mirror ``_pending`` entries get a stable minted
+  ``approval_id`` (persisted in the stored entry dict, so it is stable across
+  polls and dismissals persist);
+- an exact-id response pops the entry and reports ``ok``;
+- a stale/different explicit id still fails closed (#527 guard preserved);
+- Skip-all (yolo release) drains the entry instead of dead-ending.
+
+The sibling shape — an orphaned ``_gateway_queues`` producer entry — is already
+covered upstream (minting ``gwlocal:<token>`` in the producer-tokenization loop,
+commit 5826d76d); the tests in ``test_gateway_approval_legacy_path.py`` and this
+file's history document it. This file covers the raw-``_pending`` shape, which
+no existing test exercised.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from unittest.mock import patch
+
+import pytest
+
+from api import models
+from api import routes
+
+try:
+    import tools.approval as ta
+    APPROVAL_AVAILABLE = True
+except ImportError:
+    ta = None
+    APPROVAL_AVAILABLE = False
+
+pytestmark = pytest.mark.skipif(
+    not APPROVAL_AVAILABLE,
+    reason="tools.approval not available in this environment",
+)
+
+
+class _FakeHandler:
+    def __init__(self):
+        self.status = None
+        self._body = b""
+        self.client_address = ("127.0.0.1", 0)
+        self.headers = {}
+
+        class _W:
+            def __init__(self, outer):
+                self.outer = outer
+
+            def write(self, b):
+                self.outer._body += b
+
+        self.wfile = _W(self)
+
+    def send_response(self, code):
+        self.status = code
+
+    def send_header(self, k, v):
+        pass
+
+    def end_headers(self):
+        pass
+
+    def json(self):
+        return json.loads(self._body.decode("utf-8"))
+
+
+def _register_session(sid: str, active_stream_id: str | None = None):
+    s = models.Session(session_id=sid, title="approval-raw-pending-orphan")
+    s.active_stream_id = active_stream_id
+    with models.LOCK:
+        models.SESSIONS[sid] = s
+    return s
+
+
+def _seed_raw_pending(sid: str, command: str = "rm -rf /tmp/qa-probe-home",
+                      key: str = "recursive delete") -> dict:
+    """Seed ``_pending[sid]`` with the exact ``_pending_result`` shape.
+
+    Mirrors agent-side tools/approval.py::_pending_result when no gateway
+    notifier is registered: a raw dict with no approval_id, no request_id,
+    no run_id, written directly into the shared _pending dict.
+    """
+    pending = {
+        "command": command,
+        "pattern_key": key,
+        "pattern_keys": [key],
+        "description": "recursive delete",
+    }
+    with ta._lock:
+        ta._pending[sid] = pending
+        ta._gateway_queues.pop(sid, None)
+    return pending
+
+
+def _cleanup(sid: str):
+    with ta._lock:
+        ta._pending.pop(sid, None)
+        ta._gateway_queues.pop(sid, None)
+    with models.LOCK:
+        models.SESSIONS.pop(sid, None)
+    try:
+        ta.disable_session_yolo(sid)
+    except Exception:
+        pass
+
+
+def _poll_pending(sid: str) -> dict:
+    h = _FakeHandler()
+    routes._handle_approval_pending(h, type("P", (), {"query": f"session_id={sid}"})())
+    assert h.status == 200, f"pending poll failed: {h.status} {h.json()!r}"
+    return h.json()
+
+
+def _respond(sid: str, body: dict):
+    h = _FakeHandler()
+    with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=True):
+        routes._handle_approval_respond(h, body)
+    return h
+
+
+MINTED_ID_RE = re.compile(r"^gwlocal-mirrorless:[0-9a-f]{32}$")
+
+
+def test_raw_pending_entry_surfaces_with_stable_actionable_id():
+    """The id-less raw entry must be served with a minted id that is stable
+    across polls (dismiss persistence keys on sid+approval_id)."""
+    sid = f"raw-id-{uuid.uuid4().hex[:8]}"
+    entry = _seed_raw_pending(sid)
+    try:
+        data = _poll_pending(sid)
+        assert data["pending_count"] >= 1
+        pending = data["pending"]
+        assert pending is not None
+        approval_id = pending.get("approval_id")
+        assert approval_id, (
+            "pending payload served a raw id-less local entry with no "
+            "approval_id; the frontend cannot act on it (buttons no-op, "
+            "dismiss cannot stick)"
+        )
+        assert MINTED_ID_RE.match(approval_id), approval_id
+
+        # Stability: the second poll must return the SAME id — the mint must
+        # persist in the stored entry, not re-mint per poll.
+        data2 = _poll_pending(sid)
+        assert data2["pending"]["approval_id"] == approval_id
+        assert entry["approval_id"] == approval_id
+    finally:
+        _cleanup(sid)
+
+
+def test_raw_pending_entry_respond_with_minted_id_pops_entry():
+    """An exact-id response must pop the raw entry and report ok (no waiter
+    exists behind this shape — draining is the correct resolution)."""
+    sid = f"raw-resp-{uuid.uuid4().hex[:8]}"
+    _seed_raw_pending(sid)
+    # The incident session had a LIVE run pointer; the stale stream-pointer
+    # guard must not 409 the click before the legacy resolver pops the entry.
+    _register_session(sid, active_stream_id="stream-raw-resp-live")
+    from api.gateway_chat import _STREAM_RUN_IDS
+
+    _STREAM_RUN_IDS["stream-raw-resp-live"] = "run-live-1"
+    try:
+        approval_id = _poll_pending(sid)["pending"]["approval_id"]
+        assert approval_id
+
+        h = _respond(sid, {"session_id": sid, "choice": "deny",
+                           "approval_id": approval_id})
+        body = h.json()
+        assert h.status == 200, f"respond failed: {h.status} {body!r}"
+        assert body.get("ok") is True, f"respond not accepted: {body!r}"
+
+        # The entry must actually be gone — no more phantom card.
+        data = _poll_pending(sid)
+        assert data["pending"] is None
+        assert data["pending_count"] == 0
+    finally:
+        _STREAM_RUN_IDS.pop("stream-raw-resp-live", None)
+        _cleanup(sid)
+
+
+def test_raw_pending_entry_stale_id_fails_closed():
+    """#527 guard preserved: a different explicit id must not pop the live
+    entry and must keep it surfaced with its stable id."""
+    sid = f"raw-stale-{uuid.uuid4().hex[:8]}"
+    entry = _seed_raw_pending(sid)
+    _register_session(sid)
+    try:
+        approval_id = _poll_pending(sid)["pending"]["approval_id"]
+        assert approval_id
+
+        h = _respond(sid, {"session_id": sid, "choice": "once",
+                           "approval_id": "stale-different-id"})
+        body = h.json()
+        assert body.get("ok") is not True, f"stale id must fail closed: {body!r}"
+
+        # The live entry remains pending, unchanged, with its stable id.
+        data = _poll_pending(sid)
+        assert data["pending"] is not None
+        assert data["pending"]["approval_id"] == approval_id
+        assert entry["approval_id"] == approval_id
+    finally:
+        _cleanup(sid)
+
+
+def test_raw_pending_entry_yolo_skip_all_drains():
+    """Skip-all (yolo release) must drain the raw entry instead of leaving the
+    phantom card, and report the yolo state truthfully."""
+    sid = f"raw-yolo-{uuid.uuid4().hex[:8]}"
+    _seed_raw_pending(sid)
+    _register_session(sid)
+    try:
+        approval_id = _poll_pending(sid)["pending"]["approval_id"]
+        assert approval_id
+
+        h = _respond(sid, {"session_id": sid, "choice": "once",
+                           "approval_id": approval_id, "yolo": True})
+        body = h.json()
+        assert h.status == 200, f"yolo respond failed: {h.status} {body!r}"
+        assert body.get("ok") is True, body
+        assert body.get("yolo_enabled") is True
+        assert _poll_pending(sid)["pending"] is None
+    finally:
+        # Restore: the yolo release enables session YOLO as a side effect.
+        _cleanup(sid)

tests/test_merge_append_only_normalization_cache.py

@@ -23,6 +23,33 @@ class _ExplodingList(list):
         raise RuntimeError("content __str__ must not run")
 
 
+def _count_content_normalizations(monkeypatch):
+    """Count how often structured content is actually normalized.
+
+    These tests originally counted ``__str__`` on the content list, because the
+    key builders stringified content with ``str()``. Structured content is now
+    serialized canonically (type-namespaced) instead, so ``str()`` never runs on
+    a list and the old counter reads zero.
+
+    The property under test is unchanged -- expensive normalization of the same
+    content must happen a bounded number of times, not once per key -- so the
+    counter now observes the normalization helper itself and ignores the cheap
+    passthrough calls where content has already been reduced to a string.
+    """
+    counts = {"count": 0}
+    original = models._canonical_structured_content
+
+    def _wrapped(content):
+        counts["count"] += 1
+        return original(content)
+
+    # Observe the serialisation itself, not the identity lookups: every key
+    # derivation asks for the identity, but a list must be serialised only once
+    # per merge call.
+    monkeypatch.setattr(models, "_canonical_structured_content", _wrapped)
+    return counts
+
+
 def _install_key_wrappers(monkeypatch):
     call_counts: dict[str, dict[int, int]] = {
         "merge": defaultdict(int),
@@ -66,6 +93,7 @@ def test_merge_append_only_caches_canonical_keys_and_preserves_identity(monkeypa
     sidecar_messages = [repeated_user, repeated_user, repeated_user, repeated_assistant]
     state_messages = [repeated_assistant, repeated_user]
 
+    normalizations = _count_content_normalizations(monkeypatch)
     call_counts = _install_key_wrappers(monkeypatch)
     merged = models.merge_session_messages_append_only(sidecar_messages, state_messages)
 
@@ -77,7 +105,8 @@ def test_merge_append_only_caches_canonical_keys_and_preserves_identity(monkeypa
     ]
     assert merged == [repeated_user, repeated_user, repeated_user, repeated_assistant]
 
-    assert str_calls["count"] == 2
+    assert normalizations["count"] == 2
+    assert str_calls["count"] == 0
     assert dict(call_counts["merge"]) == {
         id(repeated_user): 1,
         id(repeated_assistant): 1,
@@ -203,6 +232,7 @@ def test_merge_append_only_recomputes_after_mutation(monkeypatch):
         "timestamp": 3000,
     }
 
+    normalizations = _count_content_normalizations(monkeypatch)
     call_counts = _install_key_wrappers(monkeypatch)
     first = models.merge_session_messages_append_only([], [state_message])
     state_message["content"].append("second")
@@ -211,7 +241,8 @@ def test_merge_append_only_recomputes_after_mutation(monkeypatch):
     assert first == [state_message]
     assert second == [state_message]
     assert first is not second
-    assert str_calls["count"] == 2
+    assert normalizations["count"] == 2
+    assert str_calls["count"] == 0
     assert dict(call_counts["dedup"]) == {
         id(state_message): 2,
     }

tests/test_profile_switch_1200.py

@@ -696,7 +696,7 @@ def test_chat_start_retags_empty_session_to_request_profile(monkeypatch, tmp_pat
         "_resolve_compatible_session_model_state",
         lambda model, provider, **_: (model, provider, False),
     )
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: object())
 
     started_threads = []
@@ -768,7 +768,7 @@ def test_chat_start_does_not_retag_non_empty_session(monkeypatch, tmp_path):
         "_resolve_compatible_session_model_state",
         lambda model, provider, **_: (model, provider, False),
     )
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: object())
 
     class FakeThread:

tests/test_provider_mismatch.py

@@ -1059,9 +1059,9 @@ def test_issue1734_chat_start_persists_repaired_codex_provider(monkeypatch):
         "_resolve_chat_workspace_with_recovery",
         lambda current, _requested: current.workspace,
     )
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: value)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: value)
     monkeypatch.setattr(routes, "_get_session_agent_lock", lambda sid: contextlib.nullcontext())
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: object())
     monkeypatch.setattr(routes.threading, "Thread", FakeThread)
 

tests/test_recovered_anchor_settle_duplication.py

@@ -537,7 +537,7 @@ def test_handle_chat_sync_passes_result_turn_authority_to_settlement(tmp_path, m
     monkeypatch.setattr(routes, "get_session", models.get_session)
     monkeypatch.setattr(routes, "title_from", models.title_from)
     monkeypatch.setattr(routes, "get_config", lambda: {"model": "m", "provider": "p"})
-    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value: tmp_path)
+    monkeypatch.setattr(routes, "resolve_trusted_workspace", lambda value, **_kw: tmp_path)
     monkeypatch.setattr(routes, "load_settings", lambda: {})
     monkeypatch.setattr(routes, "_resolve_cli_toolsets", lambda: [])
     monkeypatch.setattr(routes, "_agent_runtime_barrier_response", lambda **_k: None)

tests/test_session_lineage_collapse.py

@@ -499,6 +499,181 @@ console.log(JSON.stringify(rows));
     assert "_orphan_child_session" not in rows[0]["_child_sessions"][0]
 
 
+def test_cross_surface_subagents_stack_under_visible_messaging_parent():
+    """Delegated subagents stay nested even when their visible parent is external."""
+    js = SESSIONS_JS_PATH.read_text(encoding="utf-8")
+    source = f"""
+const src = {js!r};
+function extractFunc(name) {{
+  const re = new RegExp('function\\\\s+' + name + '\\\\s*\\\\(');
+  const start = src.search(re);
+  if (start < 0) throw new Error(name + ' not found');
+  let i = src.indexOf('{{', start);
+  let depth = 1; i++;
+  while (depth > 0 && i < src.length) {{
+    if (src[i] === '{{') depth++;
+    else if (src[i] === '}}') depth--;
+    i++;
+  }}
+  return src.slice(start, i);
+}}
+function _isExternalSession(s) {{ return !!(s && (s.is_cli_session || s.session_source === 'messaging')); }}
+eval(extractFunc('_isChildSession'));
+eval(extractFunc('_isForkWithResolvableParent'));
+eval(extractFunc('_sidebarLineageKeyForRow'));
+eval(extractFunc('_attachChildSessionsToSidebarRows'));
+const collapsed = [{{
+  session_id:'messaging_parent',
+  title:'Messaging parent',
+  raw_source:'weixin',
+  source_tag:'weixin',
+  session_source:'messaging',
+  message_count:3,
+}}];
+const raw = [
+  collapsed[0],
+  ...['delegate_a', 'delegate_b', 'delegate_c'].map(session_id => ({{
+    session_id,
+    title:'Subagent Session',
+    parent_session_id:'messaging_parent',
+    relationship_type:'child_session',
+    raw_source:'subagent',
+    source_tag:'subagent',
+    session_source:'other',
+    source_label:'Subagent',
+    _parent_lineage_root_id:'messaging_parent',
+    _cross_surface_child_session:true,
+  }})),
+];
+const rows = _attachChildSessionsToSidebarRows(collapsed, raw);
+console.log(JSON.stringify(rows));
+"""
+    rows = json.loads(_run_node(source))
+    assert [row["session_id"] for row in rows] == ["messaging_parent"]
+    assert rows[0]["_child_session_count"] == 3
+    assert [child["session_id"] for child in rows[0]["_child_sessions"]] == [
+        "delegate_a",
+        "delegate_b",
+        "delegate_c",
+    ]
+    assert all("_orphan_child_session" not in child for child in rows[0]["_child_sessions"])
+
+
+
+
+def test_delegated_subagent_source_precedence_ignores_stale_metadata():
+    """Only the first nonblank raw-role marker may assert delegation."""
+    js = SESSIONS_JS_PATH.read_text(encoding="utf-8")
+    source = f"""
+const src = {js!r};
+function extractFunc(name) {{
+  const re = new RegExp('function\\\\s+' + name + '\\\\s*\\\\(');
+  const start = src.search(re);
+  if (start < 0) throw new Error(name + ' not found');
+  let i = src.indexOf('{{', start);
+  let depth = 1; i++;
+  while (depth > 0 && i < src.length) {{
+    if (src[i] === '{{') depth++;
+    else if (src[i] === '}}') depth--;
+    i++;
+  }}
+  return src.slice(start, i);
+}}
+eval(extractFunc('_isChildSession'));
+eval(extractFunc('_isForkWithResolvableParent'));
+eval(extractFunc('_sidebarLineageKeyForRow'));
+eval(extractFunc('_attachChildSessionsToSidebarRows'));
+const parent = {{
+  session_id:'messaging_parent',
+  title:'Messaging parent',
+  raw_source:'weixin',
+  source_tag:'weixin',
+  session_source:'messaging',
+}};
+function runCase(name, overrides, includeChildInCollapsed=false) {{
+  const child = {{
+    session_id:name,
+    title:name,
+    parent_session_id:'messaging_parent',
+    relationship_type:'child_session',
+    raw_source:'api_server',
+    source_tag:'api_server',
+    source:'api_server',
+    session_source:'api',
+    _parent_lineage_root_id:'messaging_parent',
+    _cross_surface_child_session:true,
+    ...overrides,
+  }};
+  const collapsed = includeChildInCollapsed ? [{{...parent}}, child] : [{{...parent}}];
+  const rows = _attachChildSessionsToSidebarRows(collapsed, [collapsed[0], child]);
+  const parentRow = rows.find(row => row.session_id === 'messaging_parent');
+  return {{
+    name,
+    topLevel:rows.map(row => row.session_id),
+    nested:(parentRow._child_sessions || []).map(row => row.session_id),
+  }};
+}}
+const results = [
+  runCase('stale_source_tag', {{
+    raw_source:'api_server',
+    source_tag:'subagent',
+  }}),
+  runCase('stale_session_source', {{
+    raw_source:'api_server',
+    source_tag:'api_server',
+    source:'api_server',
+    session_source:'subagent',
+  }}),
+  runCase('source_tag_fallback', {{
+    raw_source:'   ',
+    source_tag:' SubAgent ',
+    source:'api_server',
+    session_source:'other',
+  }}),
+  runCase('authentic_raw_source', {{
+    raw_source:'subagent',
+    source_tag:'api_server',
+    source:'api_server',
+    session_source:'other',
+  }}),
+  runCase('not_child_session', {{
+    relationship_type:null,
+    raw_source:'subagent',
+    source_tag:'subagent',
+    source:'subagent',
+    session_source:'other',
+  }}, true),
+];
+console.log(JSON.stringify(results));
+"""
+    assert json.loads(_run_node(source)) == [
+        {
+            "name": "stale_source_tag",
+            "topLevel": ["messaging_parent", "stale_source_tag"],
+            "nested": [],
+        },
+        {
+            "name": "stale_session_source",
+            "topLevel": ["messaging_parent", "stale_session_source"],
+            "nested": [],
+        },
+        {
+            "name": "source_tag_fallback",
+            "topLevel": ["messaging_parent"],
+            "nested": ["source_tag_fallback"],
+        },
+        {
+            "name": "authentic_raw_source",
+            "topLevel": ["messaging_parent"],
+            "nested": ["authentic_raw_source"],
+        },
+        {
+            "name": "not_child_session",
+            "topLevel": ["messaging_parent", "not_child_session"],
+            "nested": [],
+        },
+    ]
+
 
 def test_cross_surface_subagent_child_stacks_under_visible_fork_parent():
     """Forked WebUI conversations can still own subagent child rows."""

tests/test_sprint5.py

@@ -117,7 +117,7 @@ def test_workspace_suggest_preserves_tilde_prefix(monkeypatch, tmp_path):
     child = home / "Projects"
     child.mkdir(parents=True)
     monkeypatch.setenv("HOME", str(home))
-    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [home.resolve()])
+    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [home.resolve()])
 
     suggestions = workspace.list_workspace_suggestions("~/")
 
@@ -132,7 +132,7 @@ def test_workspace_suggest_preserves_tilde_prefix_for_partial_child(monkeypatch,
     child = home / "Projects"
     child.mkdir(parents=True)
     monkeypatch.setenv("HOME", str(home))
-    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [home.resolve()])
+    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [home.resolve()])
 
     suggestions = workspace.list_workspace_suggestions("~/Pro")
 
@@ -152,7 +152,7 @@ def test_workspace_suggest_expands_tilde_when_home_is_symlink(monkeypatch, tmp_p
         pytest.skip(f"symlinks are not available in this environment: {exc}")
 
     monkeypatch.setenv("HOME", str(link_home))
-    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [actual_home.resolve()])
+    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [actual_home.resolve()])
 
     suggestions = workspace.list_workspace_suggestions("~/Doc")
 
@@ -166,7 +166,7 @@ def test_workspace_suggest_keeps_absolute_prefix_absolute(monkeypatch, tmp_path)
     child = home / "Projects"
     child.mkdir(parents=True)
     monkeypatch.setenv("HOME", str(home))
-    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda: [home.resolve()])
+    monkeypatch.setattr(workspace, "_trusted_workspace_roots", lambda *a, **kw: [home.resolve()])
 
     suggestions = workspace.list_workspace_suggestions(str(home) + "/Pro")
 

tests/test_stale_stream_cleanup.py

@@ -134,7 +134,7 @@ def test_chat_start_rechecks_active_stream_under_session_lock(monkeypatch, tmp_p
 
     monkeypatch.setattr(routes, "_get_session_agent_lock", lambda sid: MutatingSessionLock())
     monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})())
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
     monkeypatch.setattr(routes.threading, "Thread", NoopThread)
 
@@ -199,7 +199,7 @@ def test_chat_start_blocks_same_session_active_run_after_cancel_clears_stream_id
             return None
 
     monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})())
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
     monkeypatch.setattr(routes.threading, "Thread", NoopThread)
 
@@ -257,7 +257,7 @@ def test_chat_start_allows_same_session_after_active_run_unregisters(monkeypatch
             return None
 
     monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})())
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
     monkeypatch.setattr(routes.threading, "Thread", NoopThread)
 
@@ -332,7 +332,7 @@ def test_chat_start_not_permanently_blocked_by_stale_active_run(monkeypatch, tmp
             return None
 
     monkeypatch.setattr(routes.uuid, "uuid4", lambda: type("FakeUuid", (), {"hex": "new-stream"})())
-    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace: None)
+    monkeypatch.setattr(routes, "set_last_workspace", lambda workspace, **_kw: None)
     monkeypatch.setattr(routes, "create_stream_channel", lambda: queue.Queue())
     monkeypatch.setattr(routes.threading, "Thread", NoopThread)
 

tests/test_state_db_content_sentinel_decode.py

@@ -0,0 +1,416 @@
+"""Regression tests for the state.db structured-content sentinel.
+
+hermes_state stores list/dict message content as a NUL-sentinel JSON string.
+While the WebUI projector left it undecoded, an uploaded image's base64 data
+URI reached the transcript as literal text -- one unbreakable ~65k-character
+run -- and the browser spent minutes computing its min-content width.
+
+The decode is deliberately narrow: it must not widen ``content`` into any
+shape the rest of the WebUI pipeline cannot already render.
+"""
+import json
+
+from api.models import (
+    _content_identity_for_key,
+    _decode_state_db_content,
+    _project_state_db_message,
+    _session_message_dedup_key,
+    _session_message_merge_key,
+    _session_message_multimodal_mirror_key,
+    _session_message_visible_key,
+)
+
+PREFIX = "\x00json:"
+TEXT_AND_IMAGE = [
+    {"type": "text", "text": "here is a screenshot"},
+    {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}},
+]
+
+
+def _sentinel(payload_json: str) -> str:
+    return PREFIX + payload_json
+
+
+# --- the fix itself -------------------------------------------------------
+
+def test_supported_list_root_is_decoded():
+    assert _decode_state_db_content(_sentinel(json.dumps(TEXT_AND_IMAGE))) == TEXT_AND_IMAGE
+
+
+def test_projector_decodes_content():
+    row = {"role": "user", "content": _sentinel(json.dumps(TEXT_AND_IMAGE)),
+           "timestamp": 1.0, "id": 1}
+    msg = _project_state_db_message(row, available=set(), id_col=False, optional=())
+    assert msg["content"] == TEXT_AND_IMAGE
+
+
+# --- finding #1: dict roots must NOT be widened ---------------------------
+
+def test_dict_root_falls_back_to_raw_string():
+    """A dict root reaches _renderCacheKey(), which calls text.slice() on it
+    and throws, blanking the turn. It must stay a string."""
+    raw = _sentinel(json.dumps({"type": "text", "text": "hi"}))
+    assert _decode_state_db_content(raw) == raw
+
+
+def test_scalar_roots_fall_back_to_raw_string():
+    for payload in ("42", '"just a string"', "true", "null"):
+        raw = _sentinel(payload)
+        assert _decode_state_db_content(raw) == raw
+
+
+# --- finding #2: non-finite numbers break browser JSON.parse --------------
+
+def test_non_finite_constants_fall_back_to_raw_string():
+    for literal in ("NaN", "Infinity", "-Infinity"):
+        raw = _sentinel('[{"type": "text", "text": 1}, %s]' % literal)
+        assert _decode_state_db_content(raw) == raw
+
+
+def test_overflowed_float_falls_back_to_raw_string():
+    raw = _sentinel('[{"type": "text", "text": "x", "score": 1e400}]')
+    assert _decode_state_db_content(raw) == raw
+
+
+def test_decoded_payload_is_always_browser_parseable():
+    decoded = _decode_state_db_content(_sentinel(json.dumps(TEXT_AND_IMAGE)))
+    json.loads(json.dumps(decoded, allow_nan=False))
+
+
+# --- finding #3: only the schema the UI actually renders ------------------
+
+def test_unsupported_part_shapes_fall_back_to_raw_string():
+    unsupported = [
+        [{"type": "input_text", "text": "dropped by the JS readers"}],
+        [{"type": "output_text", "text": "also dropped"}],
+        [{"type": "tool_use", "id": "t1"}],
+        ["a bare scalar part"],
+        [{"text": "no type key"}],
+        [{"type": "text", "text": {"not": "a string"}}],
+        [],
+    ]
+    for payload in unsupported:
+        raw = _sentinel(json.dumps(payload))
+        assert _decode_state_db_content(raw) == raw, payload
+
+
+def test_image_only_list_stays_raw_because_it_would_not_render():
+    """msgContent() discards image parts and this projection supplies no
+    attachments, so an image-only row would decode to nothing visible and
+    _messageIsRenderable() would hide it. It must stay a raw string instead."""
+    payload = [{"type": "image_url", "image_url": {"url": "data:image/png;base64,AA"}}]
+    raw = _sentinel(json.dumps(payload))
+    assert _decode_state_db_content(raw) == raw
+
+
+# --- passthrough / malformed ---------------------------------------------
+
+def test_plain_values_pass_through_unchanged():
+    for value in ("hello", "", None, 42, ["already", "a", "list"], "see json: below"):
+        assert _decode_state_db_content(value) == value
+
+
+def test_malformed_sentinel_payload_returns_raw_string():
+    raw = _sentinel("{not valid json")
+    assert _decode_state_db_content(raw) == raw
+
+
+# --- finding #4: identities must be type-namespaced ----------------------
+
+def test_structured_content_cannot_collide_with_its_own_repr():
+    structured = {"role": "user", "content": TEXT_AND_IMAGE, "timestamp": 1.0}
+    scalar = {"role": "user", "content": str(TEXT_AND_IMAGE), "timestamp": 1.0}
+    assert _session_message_merge_key(structured) != _session_message_merge_key(scalar)
+    assert _session_message_dedup_key(structured) != _session_message_dedup_key(scalar)
+
+
+def test_rich_turns_with_different_images_keep_distinct_identities():
+    def turn(url):
+        return {"role": "user", "timestamp": 1.0, "content": [
+            {"type": "text", "text": "same visible text"},
+            {"type": "image_url", "image_url": {"url": url}},
+        ]}
+    a, b = turn("data:image/png;base64,AAAA"), turn("data:image/png;base64,BBBB")
+    assert _session_message_merge_key(a) != _session_message_merge_key(b)
+    assert _session_message_dedup_key(a) != _session_message_dedup_key(b)
+
+
+def test_scalar_content_identity_is_unchanged():
+    """Existing scalar behaviour must not shift."""
+    assert _content_identity_for_key("plain text") == "plain text"
+    assert _content_identity_for_key(None) == ""
+    assert _content_identity_for_key("") == ""
+
+
+def test_mirror_bridge_never_pairs_rich_to_rich():
+    rich = {"role": "user", "timestamp": 1.0, "content": TEXT_AND_IMAGE}
+    assert _session_message_multimodal_mirror_key(rich, require_image_parts=True) is not None
+    # the scalar side of the bridge must refuse a rich row
+    assert _session_message_multimodal_mirror_key(rich, require_scalar_mirror=True) is None
+    # and the two flags are mutually exclusive by construction
+    assert _session_message_multimodal_mirror_key(
+        rich, require_image_parts=True, require_scalar_mirror=True
+    ) is None
+
+
+def test_mirror_bridge_still_accepts_a_scalar_mirror():
+    scalar = {"role": "user", "timestamp": 1.0, "content": "[screenshot] here is a screenshot"}
+    assert _session_message_multimodal_mirror_key(scalar, require_scalar_mirror=True) is not None
+
+
+# --- finding #5: prefix and tail keys share one representation -----------
+
+def test_prefix_and_tail_keys_agree_for_a_sentinel_row():
+    raw = _sentinel(json.dumps(TEXT_AND_IMAGE))
+    row = {"role": "user", "content": raw, "timestamp": 5.0, "id": 7}
+    tail_msg = _project_state_db_message(row, available=set(), id_col=False, optional=())
+    prefix_msg = {
+        "role": row["role"],
+        "content": _decode_state_db_content(row["content"]),
+        "tool_calls": None,
+        "api_content": None,
+    }
+    tail_key = _session_message_visible_key(
+        {"role": tail_msg.get("role"), "content": tail_msg.get("content"),
+         "tool_calls": None, "api_content": None},
+        normalize_workspace_prefix=True,
+    )
+    prefix_key = _session_message_visible_key(prefix_msg, normalize_workspace_prefix=True)
+    assert prefix_key == tail_key
+
+
+# --- every read path that projects content must decode (behavioural) ------
+#
+# Keys derived on one read path are compared against keys derived on another,
+# so a read path that projected the column raw while another decoded it would
+# silently reintroduce the prefix/tail mismatch. These exercise each path
+# against a real sentinel-encoded row rather than inspecting source.
+
+def _make_state_db(path, sid, rows):
+    import sqlite3
+    conn = sqlite3.connect(path)
+    conn.execute(
+        "CREATE TABLE sessions (id TEXT PRIMARY KEY, source TEXT, title TEXT, "
+        "model TEXT, started_at REAL, message_count INTEGER)"
+    )
+    conn.execute(
+        "CREATE TABLE messages (id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, "
+        "role TEXT, content TEXT, timestamp REAL, tool_call_id TEXT, tool_calls TEXT, "
+        "tool_name TEXT, active INTEGER DEFAULT 1, api_content TEXT)"
+    )
+    conn.execute(
+        "INSERT INTO sessions (id, source, title, model, started_at, message_count) "
+        "VALUES (?, ?, ?, ?, ?, ?)",
+        (sid, "webui", "Sentinel", "test-model", 1000.0, len(rows)),
+    )
+    for row in rows:
+        conn.execute(
+            "INSERT INTO messages (session_id, role, content, timestamp, active) "
+            "VALUES (?, ?, ?, ?, 1)",
+            (sid, row["role"], row["content"], row["timestamp"]),
+        )
+    conn.commit()
+    conn.close()
+
+
+def _sentinel_session(tmp_path, monkeypatch):
+    """A session whose first row carries sentinel-encoded multimodal content."""
+    from api import models
+
+    sid = "sentineltest"
+    db = tmp_path / "state.db"
+    _make_state_db(db, sid, [
+        {"role": "user", "content": _sentinel(json.dumps(TEXT_AND_IMAGE)), "timestamp": 1000.0},
+        {"role": "assistant", "content": "plain reply", "timestamp": 2000.0},
+    ])
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db, raising=False)
+    return models, sid
+
+
+def test_transcript_read_decodes_sentinel_rows(tmp_path, monkeypatch):
+    """The canonical projection hands the frontend structured content."""
+    models, sid = _sentinel_session(tmp_path, monkeypatch)
+    messages = models.get_state_db_session_messages(sid)
+    assert messages, "expected the fixture session to load"
+    assert messages[0]["content"] == TEXT_AND_IMAGE
+    # the base64 payload must not survive as literal transcript text
+    assert not isinstance(messages[0]["content"], str)
+
+
+def test_regeneration_prefix_and_tail_keys_agree_for_a_sentinel_row(tmp_path, monkeypatch):
+    """The same row keyed as prefix and as tail must produce one identity.
+
+    Decoding the projected tail while leaving prefix keys encoded would make
+    `_bounded_tail_snapshot_if_safe` reject the bounded path and re-read the
+    whole transcript, and could hide a genuine repeated recovered turn.
+    """
+    models, sid = _sentinel_session(tmp_path, monkeypatch)
+
+    # floor above the sentinel row -> it is part of the prefix proof
+    as_prefix = models.get_state_db_regeneration_tail_snapshot(sid, 1500.0)
+    # floor below it -> the same row is part of the bounded tail
+    as_tail = models.get_state_db_regeneration_tail_snapshot(sid, 500.0)
+    assert as_prefix is not None and as_tail is not None
+
+    assert as_prefix["prefix_keys"], "sentinel row should sit in the prefix"
+    assert as_tail["tail_keys"], "sentinel row should sit in the tail"
+    assert as_prefix["prefix_keys"][0] == as_tail["tail_keys"][0]
+
+
+def test_bounded_prefix_reader_agrees_with_the_projected_tail(tmp_path, monkeypatch):
+    """The standalone prefix-key reader shares the tail's representation."""
+    models, sid = _sentinel_session(tmp_path, monkeypatch)
+
+    prefix_keys = models.get_state_db_session_message_keys_before_timestamp(sid, 1500.0)
+    tail = models.get_state_db_regeneration_tail_snapshot(sid, 500.0)
+    assert prefix_keys, "expected a prefix key for the sentinel row"
+    assert tail is not None and tail["tail_keys"]
+    assert prefix_keys[0] == tail["tail_keys"][0]
+
+
+def test_unsupported_sentinel_shape_survives_the_read_path_as_text(tmp_path, monkeypatch):
+    """A shape the UI cannot render stays a string end-to-end, not silently dropped."""
+    from api import models
+
+    sid = "unsupported"
+    db = tmp_path / "state.db"
+    raw = _sentinel(json.dumps({"type": "text", "text": "dict root"}))
+    _make_state_db(db, sid, [{"role": "user", "content": raw, "timestamp": 1000.0}])
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db, raising=False)
+
+    messages = models.get_state_db_session_messages(sid)
+    assert messages
+    assert messages[0]["content"] == raw
+
+
+# --- re-gate finding 2: only decode what will actually render --------------
+
+def test_whitespace_only_text_with_an_image_stays_raw():
+    payload = [
+        {"type": "text", "text": "   \n\t "},
+        {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA"}},
+    ]
+    raw = _sentinel(json.dumps(payload))
+    assert _decode_state_db_content(raw) == raw
+
+
+def test_malformed_image_parts_stay_raw():
+    text = {"type": "text", "text": "caption"}
+    malformed = [
+        {"type": "image_url", "payload": "invalid"},
+        {"type": "image_url", "image_url": {"url": ""}},
+        {"type": "image_url", "image_url": {"href": "x"}},
+        {"type": "input_image"},
+        {"type": "image", "source": {"type": "base64", "data": "AA"}},  # no media_type
+        {"type": "image", "source": {"type": "url"}},
+        {"type": "image", "source": "not-a-dict"},
+    ]
+    for bad in malformed:
+        raw = _sentinel(json.dumps([text, bad]))
+        assert _decode_state_db_content(raw) == raw, bad
+
+
+def test_text_with_each_valid_image_payload_shape_decodes():
+    text = {"type": "text", "text": "caption"}
+    valid = [
+        {"type": "image_url", "image_url": {"url": "data:image/png;base64,AA"}},
+        {"type": "image_url", "image_url": "https://example.test/a.png"},
+        {"type": "input_image", "image_url": "data:image/png;base64,AA"},
+        {"type": "input_image", "file_id": "file-123"},
+        {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "AA"}},
+        {"type": "image", "source": {"type": "url", "url": "https://example.test/a.png"}},
+    ]
+    for good in valid:
+        payload = [text, good]
+        assert _decode_state_db_content(_sentinel(json.dumps(payload))) == payload, good
+
+
+def test_image_only_row_stays_visible_through_the_read_path(tmp_path, monkeypatch):
+    from api import models
+
+    sid = "imageonly"
+    db = tmp_path / "state.db"
+    raw = _sentinel(json.dumps([{"type": "image_url", "image_url": {"url": "data:image/png;base64,AA"}}]))
+    _make_state_db(db, sid, [{"role": "user", "content": raw, "timestamp": 1000.0}])
+    monkeypatch.setattr(models, "_active_state_db_path", lambda: db, raising=False)
+    messages = models.get_state_db_session_messages(sid)
+    assert messages and messages[0]["content"] == raw
+
+
+# --- re-gate finding 1: out-of-band identity, consistent across every key ---
+
+from api.models import (  # noqa: E402
+    _matching_visible_duplicate,
+    _session_message_content_key,
+    merge_session_messages_append_only,
+)
+
+# Exactly what the previous in-band tag produced for TEXT_AND_IMAGE.
+_OLD_INBAND_TOKEN = "\x00list:" + json.dumps(TEXT_AND_IMAGE, sort_keys=True, default=str)
+
+
+def test_structured_identity_is_not_a_string():
+    """Out of band: no message body can equal it, whatever it contains."""
+    assert not isinstance(_content_identity_for_key(TEXT_AND_IMAGE), str)
+
+
+def test_scalar_imitating_a_structured_identity_collides_with_no_key():
+    rich = {"role": "user", "content": TEXT_AND_IMAGE, "timestamp": 1.0}
+    forged = {"role": "user", "content": _OLD_INBAND_TOKEN, "timestamp": 1.0}
+    assert _session_message_merge_key(rich) != _session_message_merge_key(forged)
+    assert _session_message_dedup_key(rich) != _session_message_dedup_key(forged)
+    assert _session_message_content_key(rich) != _session_message_content_key(forged)
+    assert _session_message_visible_key(rich) != _session_message_visible_key(forged)
+
+
+def test_merge_keeps_the_rich_row_when_a_scalar_imitates_its_identity():
+    rich = {"role": "user", "content": TEXT_AND_IMAGE, "timestamp": 1000.0}
+    forged = {"role": "user", "content": _OLD_INBAND_TOKEN, "timestamp": 1000.0}
+    merged = merge_session_messages_append_only([forged], [rich])
+    assert any(m.get("content") == TEXT_AND_IMAGE for m in merged), merged
+
+
+def test_non_list_scalars_key_exactly_as_on_master():
+    """No silent dedup change for ordinary non-string content."""
+    assert _content_identity_for_key(42) == "42"
+    assert _content_identity_for_key(3.5) == "3.5"
+    assert _content_identity_for_key(True) == "True"
+    assert _content_identity_for_key({"a": 1}) == str({"a": 1})
+    assert _content_identity_for_key([]) == ""
+    msg = {"role": "user", "content": 42, "timestamp": 1.0}
+    assert "42" in _session_message_merge_key(msg)
+
+
+def test_structured_visible_key_never_fuzzy_matches_a_scalar():
+    rich_key = _session_message_visible_key({"role": "user", "content": TEXT_AND_IMAGE})
+    canonical = _content_identity_for_key(TEXT_AND_IMAGE)[1]
+    for text in (canonical, "prefix " + canonical + " suffix", str(TEXT_AND_IMAGE)):
+        scalar_key = _session_message_visible_key({"role": "user", "content": text})
+        assert _matching_visible_duplicate(rich_key, {scalar_key}) is None
+        assert _matching_visible_duplicate(scalar_key, {rich_key}) is None
+
+
+def test_merge_never_writes_an_identity_into_message_content():
+    rich = {"role": "user", "content": TEXT_AND_IMAGE, "timestamp": 1000.0}
+    reply = {"role": "assistant", "content": "ok", "timestamp": 1001.0}
+    merged = merge_session_messages_append_only([rich], [rich, reply])
+    for message in merged:
+        assert isinstance(message.get("content"), (str, list))
+        assert not isinstance(message.get("content"), tuple)
+    assert any(m.get("content") == TEXT_AND_IMAGE for m in merged)
+
+
+def test_structured_content_is_serialised_once_per_merge_call(monkeypatch):
+    from api import models
+
+    calls = {"n": 0}
+    real = models._canonical_structured_content
+
+    def counting(content):
+        calls["n"] += 1
+        return real(content)
+
+    monkeypatch.setattr(models, "_canonical_structured_content", counting)
+    rich = {"role": "user", "content": list(TEXT_AND_IMAGE), "timestamp": 1000.0}
+    merge_session_messages_append_only([rich, rich], [rich])
+    assert calls["n"] == 1

tests/test_static_asset_compression_and_cache.py

@@ -214,6 +214,62 @@ def test_image_is_not_gzipped(isolated_static):
     assert h.header("Content-Type") == "image/png"
 
 
+def test_standard_binary_extension_uses_system_mime_type(isolated_static):
+    """Known binary formats outside the hand-written map keep their real MIME."""
+    from api import routes
+
+    payload = b"%PDF-1.7\n" + b"\x00" * 128
+    _make_static_file(isolated_static, "manual.pdf", payload)
+
+    h = _serve(routes, "/static/manual.pdf")
+    assert h.status == 200
+    assert h.header("Content-Type") == "application/pdf"
+    assert bytes(h.body) == payload
+
+
+def test_apk_is_served_as_android_package_when_platform_database_lacks_it(
+    isolated_static, monkeypatch
+):
+    """Android package MIME must not depend on the host's MIME database."""
+    from api import routes
+
+    monkeypatch.setattr(routes.mimetypes, "guess_type", lambda _name: (None, None))
+    payload = b"PK\x03\x04" + b"\x00" * 128
+    _make_static_file(isolated_static, "hermes-webui.apk", payload)
+
+    h = _serve(routes, "/static/hermes-webui.apk")
+    assert h.status == 200
+    assert h.header("Content-Type") == "application/vnd.android.package-archive"
+    assert bytes(h.body) == payload
+
+
+def test_unknown_static_extension_falls_back_to_octet_stream(isolated_static):
+    """Unknown files fail closed as downloads instead of being exposed as text."""
+    from api import routes
+
+    payload = b"\x00\xff\x10binary"
+    _make_static_file(isolated_static, "artifact.hermes-unknown", payload)
+
+    h = _serve(routes, "/static/artifact.hermes-unknown")
+    assert h.status == 200
+    assert h.header("Content-Type") == "application/octet-stream"
+    assert bytes(h.body) == payload
+
+
+def test_encoded_static_suffix_falls_back_to_octet_stream(isolated_static):
+    """Do not advertise decoded media types without Content-Encoding support."""
+    from api import routes
+
+    payload = b"\x1f\x8b" + b"compressed-svg"
+    _make_static_file(isolated_static, "diagram.svgz", payload)
+
+    h = _serve(routes, "/static/diagram.svgz")
+    assert h.status == 200
+    assert h.header("Content-Type") == "application/octet-stream"
+    assert h.header("Content-Encoding") is None
+    assert bytes(h.body) == payload
+
+
 def test_tiny_file_is_not_gzipped(isolated_static):
     """Files under 1 KB skip gzip — framing overhead exceeds savings."""
     from api import routes

tests/test_webhook_project_sessions.py

@@ -108,7 +108,7 @@ def test_webhook_rows_get_webhook_project_id(monkeypatch, tmp_path):
     db.write_text("", encoding="utf-8")
 
     monkeypatch.setattr(models, "read_importable_agent_session_rows", lambda *_a, **_kw: [_agent_row()])
-    monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path)
+    monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path)
     monkeypatch.setattr(models, "ensure_cron_project", lambda: "cron-project-id")
     monkeypatch.setattr(models, "ensure_webhook_project", lambda: "webhook-project-id", raising=False)
     monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None)
@@ -137,7 +137,7 @@ def test_webhook_second_pass_keeps_older_project_rows_available(monkeypatch, tmp
         return []
 
     monkeypatch.setattr(models, "read_importable_agent_session_rows", fake_read_rows)
-    monkeypatch.setattr(models, "get_last_workspace", lambda: tmp_path)
+    monkeypatch.setattr(models, "get_last_workspace", lambda profile=None: tmp_path)
     monkeypatch.setattr(models, "ensure_cron_project", lambda: "cron-project-id")
     monkeypatch.setattr(models, "ensure_webhook_project", lambda: "webhook-project-id", raising=False)
     monkeypatch.setattr(models.Session, "load_metadata_only", lambda _sid: None)

tests/test_workspace_upload.py

@@ -136,7 +136,7 @@ def _configure_direct_office_upload(monkeypatch, tmp_path):
 
     monkeypatch.setattr(upload, "get_session", lambda _sid: session)
     monkeypatch.setattr(upload, "_reject_invisible_session", lambda *_args: False)
-    monkeypatch.setattr(upload, "resolve_trusted_workspace", lambda path: path)
+    monkeypatch.setattr(upload, "resolve_trusted_workspace", lambda path, **_kw: path)
     return upload, office_documents, workspace