Merge pull request #6981 from nesquena/release/stage-6979-1786619874

Release: per-message media preview snapshots (#6979)

CHANGELOG.md

@@ -5,6 +5,8 @@
 
 ### Fixed
 
+- **Historical chat media previews no longer change when a file is overwritten in place.** After the recent switch to revalidating `/api/media` responses, overwriting a file with the same name (e.g. regenerating `chart.png`) silently rewrote every earlier chat preview that referenced it, so you lost the ability to compare the old and new versions. Each settled message now freezes the bytes of its local-file previews into a content-addressed snapshot store and serves those frozen bytes for that message, while new turns still show the current file. The snapshot store is internal (never reachable as a normal media path), bounded (50 MB per file, 2 GB total, oldest-evicted-first), and each stored digest is bound to the exact source path it was captured from, so a snapshot can only ever be served for its own message. Thanks @happy5318. (#6979)
+
 - **Multi-container Docker deployments can reach the agent API again.** In the two- and three-container Compose topologies, the agent's gateway API bound to loopback only, so the WebUI (and dashboard) containers — separate hosts on the Docker network — couldn't reach `http://hermes-agent:8642` even with `HERMES_API_URL` and `API_SERVER_KEY` set correctly, breaking cron/Tasks and gateway-backed features. The agent service now sets `API_SERVER_HOST=0.0.0.0` so it's reachable across the Compose network; the host port publish stays `127.0.0.1`-only, so this does not expose the API beyond the Docker network and host loopback. Thanks @bragov4ik. (#6968)
 
 - **Two WebUI sessions with the same auto-generated title no longer leave the second one blank.** Follow-up to the `state.db` title sync: hermes-agent enforces a uniqueness rule on session titles, so when two sessions independently generated a byte-identical auto-title, the second sync raised (and silently swallowed) a collision error and left that session untitled in `hermes sessions list`. The sync now catches the collision, derives a de-duplicated variant via the existing lineage helper (`"My Session"` → `"My Session #2"`), and retries — and because the retry goes through the same provenance guard, a title you set manually is still never overwritten. Thanks @webtecnica. (#6964)

api/media_snapshots.py

@@ -0,0 +1,500 @@
+"""Per-message media snapshots: freeze file bytes at message-settle time.
+
+Problem
+-------
+``/api/media`` serves a file's CURRENT bytes.  Since the ETag-revalidation
+work (#6922) made browsers revalidate on every use, an in-place overwrite of
+``report.html`` also rewrites every historical chat preview that referenced
+it — the user loses the old/new comparison they had when the agent emitted
+the file twice under the same name.
+
+This module is the storage half of the fix.  When a turn settles
+(``api/streaming.py``), every local-file ``MEDIA:`` reference in the new
+assistant messages is snapshotted into a content-addressed store:
+
+    <STATE_DIR>/media_snapshots/<sha256>.snap
+
+Content addressing gives free dedup (an unchanged file re-settles to the
+same digest and the copy is skipped) and makes every stored object
+IMMUTABLE — the serving side can therefore cache snapshots aggressively.
+Messages carry a ``_media_snapshots`` annotation mapping absolute path to
+digest (sidecar JSON preserves extra message fields); the frontend appends
+``&snap=<digest>`` to ``/api/media`` URLs and ``api/routes.py`` serves the
+stored bytes instead of the live file.
+
+Security model
+--------------
+* Digests are validated by whole-string ``fullmatch`` against ``[0-9a-f]{64}``
+  before any path is built — a crafted ``snap=`` value cannot traverse out of
+  the store.
+* Capture uses the SAME deny predicate as the ``/api/media`` serve path
+  (``routes._media_deny_reason``): anything the endpoint refuses to serve is
+  never captured in the first place.
+* Every captured digest carries a server-owned source-path binding; a digest
+  is only ever served back for the exact canonical path it was captured from,
+  so it can never act as a bearer capability through a different allowed path.
+* The store lives under STATE_DIR; capture is restricted to files that are
+  regular files within caller-approved roots (the caller reuses the same
+  allow-list reasoning as ``/api/media`` — this module never decides what a
+  user may see, only stores bytes it is handed).
+* Writes are tmp-file + fsync + rename: a crash never leaves a torn
+  snapshot that would serve corrupt bytes forever (content addressing means
+  a torn file could never be overwritten with good bytes later).
+
+Caps
+----
+* Per-file cap (default 50 MB): larger files are not snapshotted; previews
+  for them gracefully degrade to the live file.
+* Total store cap (default 2 GB): when exceeded after a capture, oldest
+  snapshots (by mtime) are evicted until the store fits.  Eviction only
+  degrades OLD previews to live-file behaviour; it never corrupts anything.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import re
+import threading
+from pathlib import Path
+
+logger = logging.getLogger("hermes.webui")
+
+# Strict whole-string shape.  ``\\Z`` (not ``$``) so a terminal newline cannot
+# sneak past the gate; ``fullmatch`` is used at call sites.
+_DIGEST_RE = re.compile(r"[0-9a-f]{64}\Z")
+
+# Default caps.  Overridable via env var for operators with unusual disks.
+DEFAULT_MAX_FILE_BYTES = 50 * 1024 * 1024          # 50 MB per snapshot
+DEFAULT_TOTAL_CAP_BYTES = 2 * 1024 * 1024 * 1024   # 2 GB total store
+
+_SNAPSHOT_DIR_ENV = "HERMES_WEBUI_MEDIA_SNAPSHOT_DIR"
+
+# Capture and eviction run on the streaming worker thread; serialize them so
+# two concurrent settles cannot race the same tmp file or the quota scan.
+_LOCK = threading.Lock()
+
+def media_capture_allowed(path: Path) -> bool:
+    """Allow-list predicate for snapshot capture (same gate as serve).
+
+    True only when ``path`` is a regular file inside an allowed root and NOT
+    inside a denied Hermes-internal state location.  The deny half is the
+    SAME predicate the ``/api/media`` serve path uses (``routes._media_deny_reason``,
+    the #3234 state/profile deny set): anything the endpoint would refuse to
+    serve is never captured in the first place — capture and serve can never
+    diverge on what is denied.  Any failure mode returns False —
+    snapshotting is best-effort durability, never a reason to widen file
+    access.
+    """
+    import stat as stat_mod
+
+    try:
+        resolved = path.resolve()
+        st = resolved.stat()
+    except OSError:
+        return False
+    if not stat_mod.S_ISREG(st.st_mode):
+        return False
+
+    within_any_root = any(
+        _path_within(resolved, root) for root in _allowed_roots_for_capture()
+    )
+    if not within_any_root:
+        return False
+    # Authoritative deny parity with the serve path (#6979 Round 2 MUST-FIX 1):
+    # STATE_DIR subdirs, webui_state subdirs, named-profile roots, secret
+    # basenames and the snapshot store itself are all denied exactly as the
+    # route denies them.
+    try:
+        from api.routes import _media_deny_reason
+
+        if _media_deny_reason(resolved):
+            return False
+    except Exception:
+        return False  # fail closed: never snapshot when the gate is unclear
+    return True
+
+
+def _allowed_roots_for_capture() -> list[Path]:
+    """Roots capture is permitted in — same shape as ``/api/media``'s list."""
+    roots: list[Path] = []
+    home = Path(os.path.expanduser("~"))
+    hermes_home = Path(os.getenv("HERMES_HOME", str(home / ".hermes"))).expanduser()
+    for candidate in (hermes_home, Path("/tmp"), home / ".hermes"):
+        try:
+            resolved = candidate.resolve()
+        except OSError:
+            continue
+        if resolved not in roots:
+            roots.append(resolved)
+    try:
+        from api.workspace import get_last_workspace
+
+        ws = Path(get_last_workspace()).resolve()
+        if ws.is_dir() and ws not in roots:
+            roots.append(ws)
+    except Exception:
+        pass
+    extra = os.environ.get("MEDIA_ALLOWED_ROOTS", "").strip()
+    if extra:
+        for root in extra.split(os.pathsep):
+            root = root.strip()
+            if not root:
+                continue
+            try:
+                rp = Path(root).resolve()
+            except OSError:
+                continue
+            if rp.is_dir() and rp not in roots:
+                roots.append(rp)
+    return roots
+
+
+def _path_within(child: Path, root: Path) -> bool:
+    try:
+        child.resolve().relative_to(root)
+        return True
+    except (ValueError, OSError):
+        return False
+
+
+def resolve_media_ref(raw_ref: str) -> Path | None:
+    """Map a raw ``MEDIA:`` token to an absolute local path, or None.
+
+    Only bare local paths and ``file://`` URLs resolve; http(s)/data/other
+    schemes return None (nothing to snapshot — they are not server files).
+    ``~`` is expanded.  No existence check: callers decide whether absence
+    matters (capture skips missing files).
+    """
+    ref = str(raw_ref or "").strip()
+    if not ref:
+        return None
+    if ref.startswith("data:") or re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*://", ref):
+        if ref.lower().startswith("file://"):
+            from urllib.parse import unquote, urlparse
+
+            try:
+                parsed = urlparse(ref)
+                ref = unquote(parsed.path or "")
+            except Exception:
+                ref = ref[len("file://"):]
+        else:
+            return None
+    if not ref or ref.startswith(("http:", "https:")):
+        return None
+    try:
+        return Path(ref).expanduser().resolve()
+    except (OSError, RuntimeError):
+        return None
+
+
+def _default_snapshot_dir() -> Path:
+    from api.config import STATE_DIR
+
+    return Path(STATE_DIR) / "media_snapshots"
+
+
+def get_snapshot_dir() -> Path:
+    """Snapshot store root (created lazily by capture)."""
+    override = os.getenv(_SNAPSHOT_DIR_ENV, "").strip()
+    if override:
+        return Path(override).expanduser()
+    return _default_snapshot_dir()
+
+
+def is_valid_digest(digest: str) -> bool:
+    """Strict digest shape check — the ONLY gate before path construction.
+
+    ``fullmatch`` with a ``\\Z``-anchored pattern: ``$`` would also match
+    before a terminal newline, letting ``<64 hex>\\n`` slip through.
+    """
+    try:
+        return _DIGEST_RE.fullmatch(str(digest or "")) is not None
+    except (TypeError, ValueError):
+        return False
+
+
+def snapshot_path_for_digest(digest: str) -> Path | None:
+    """Return the on-disk path for a digest, or None if absent/invalid.
+
+    Never creates anything; serving uses this to decide snapshot vs live-file
+    fallback.
+    """
+    if not is_valid_digest(digest):
+        return None
+    candidate = get_snapshot_dir() / f"{digest}.snap"
+    return candidate if candidate.is_file() else None
+
+
+def _binding_path_for_digest(digest: str) -> Path:
+    """Sidecar holding the source-path set for a digest (``<digest>.src.json``)."""
+    return get_snapshot_dir() / f"{digest}.src.json"
+
+
+def _record_source_binding(digest: str, source: Path) -> None:
+    """Persist the server-owned canonical source-path ↔ digest association.
+
+    A digest is only ever served back for the EXACT path it was captured from
+    (see :func:`snapshot_servable_for_path`): a digest must never become a
+    bearer capability readable through any other allowed path.  The sidecar is
+    a tiny path list, rewritten tmp+rename atomically.  Caller holds
+    ``_LOCK`` (capture is serialized), so concurrent settles cannot race it.
+    """
+    try:
+        binding_file = _binding_path_for_digest(digest)
+        sources: set[str] = set()
+        try:
+            data = json.loads(binding_file.read_text(encoding="utf-8"))
+            sources = set(data.get("sources", []))
+        except (OSError, ValueError, TypeError):
+            sources = set()
+        sources.add(str(Path(source).resolve()))
+        tmp = binding_file.with_name(binding_file.name + ".tmp")
+        tmp.write_text(
+            json.dumps({"digest": digest, "sources": sorted(sources)}, sort_keys=True),
+            encoding="utf-8",
+        )
+        os.replace(tmp, binding_file)
+    except OSError as exc:
+        logger.debug("media snapshot source binding failed for %s: %s", source, exc)
+
+
+def snapshot_servable_for_path(digest: str, target: Path) -> bool:
+    """True only when ``digest`` was captured from canonical path ``target``.
+
+    Serve-side half of the source-path binding (#6979 Round 2 MUST-FIX 1):
+    the snapshot branch serves a digest only for the exact authorized path it
+    was captured from, so replaying a digest through a different (allowed)
+    path can never read stored bytes the normal ``path=`` gate would refuse.
+    A missing/invalid sidecar returns False — the caller falls back to the
+    live file.
+    """
+    if not is_valid_digest(digest):
+        return False
+    try:
+        want = str(target.resolve())
+    except OSError:
+        return False
+    try:
+        data = json.loads(_binding_path_for_digest(digest).read_text(encoding="utf-8"))
+    except (OSError, ValueError, TypeError):
+        return False
+    return want in set(data.get("sources", []))
+
+
+def _total_cap_bytes() -> int:
+    try:
+        return max(0, int(os.getenv("HERMES_WEBUI_MEDIA_SNAPSHOT_CAP_BYTES", "")))
+    except ValueError:
+        return DEFAULT_TOTAL_CAP_BYTES
+
+
+def _max_file_bytes() -> int:
+    try:
+        return max(0, int(os.getenv("HERMES_WEBUI_MEDIA_SNAPSHOT_MAX_FILE_BYTES", "")))
+    except ValueError:
+        return DEFAULT_MAX_FILE_BYTES
+
+
+def _store_size_and_entries(directory: Path) -> tuple[int, list[tuple[float, int, Path]]]:
+    """Scan the store: (total bytes, [(mtime, size, path), ...])."""
+    total = 0
+    entries: list[tuple[float, int, Path]] = []
+    try:
+        for child in directory.iterdir():
+            if child.suffix != ".snap":
+                continue
+            try:
+                st = child.stat()
+            except OSError:
+                continue
+            total += st.st_size
+            entries.append((st.st_mtime, st.st_size, child))
+    except OSError:
+        pass
+    return total, entries
+
+
+def _enforce_quota_locked(directory: Path) -> None:
+    """Evict oldest snapshots until the store is under the total cap.
+
+    Caller holds ``_LOCK``.  Eviction is best-effort: an unlink failure is
+    logged and skipped, never raised into the settle path.
+    """
+    cap = _total_cap_bytes()
+    total, entries = _store_size_and_entries(directory)
+    if total <= cap:
+        return
+    entries.sort()  # oldest mtime first
+    for _mtime, size, path in entries:
+        if total <= cap:
+            break
+        try:
+            path.unlink()
+            total -= size
+            logger.info("media snapshot quota: evicted %s (%d bytes)", path.name, size)
+            # Drop the digest's source-binding sidecar with its blob.
+            try:
+                _binding_path_for_digest(path.stem).unlink()
+            except OSError:
+                pass
+        except OSError as exc:
+            logger.debug("media snapshot eviction failed for %s: %s", path, exc)
+
+
+def capture_snapshot(source: Path, *, max_file_bytes: int | None = None) -> str | None:
+    """Copy ``source`` into the content-addressed store; return its digest.
+
+    Returns None (caller falls back to live-file previews) when:
+    * the file is missing / not a regular file / unreadable,
+    * the file exceeds the per-file cap,
+    * any I/O error occurs mid-copy (the torn tmp is removed).
+
+    Never raises — snapshotting is a durability enhancement and must not be
+    able to break the settle path.
+    """
+    cap = _max_file_bytes() if max_file_bytes is None else max_file_bytes
+    try:
+        st = source.stat()
+    except OSError:
+        return None
+    import stat as stat_mod
+
+    if not stat_mod.S_ISREG(st.st_mode):
+        return None
+    if cap and st.st_size > cap:
+        return None
+
+    with _LOCK:
+        directory = get_snapshot_dir()
+        try:
+            directory.mkdir(parents=True, exist_ok=True)
+        except OSError:
+            return None
+
+        digest = hashlib.sha256()
+        tmp_path: Path | None = None
+        try:
+            tmp_path = directory / f".tmp.{os.getpid()}.{threading.get_ident()}"
+            with open(source, "rb") as src, open(tmp_path, "wb") as dst:
+                while True:
+                    chunk = src.read(1024 * 1024)
+                    if not chunk:
+                        break
+                    digest.update(chunk)
+                    dst.write(chunk)
+                dst.flush()
+                os.fsync(dst.fileno())
+            hex_digest = digest.hexdigest()
+            final_path = directory / f"{hex_digest}.snap"
+            if final_path.exists():
+                # Content already stored (dedup) — drop the duplicate copy.
+                try:
+                    tmp_path.unlink()
+                except OSError:
+                    pass
+            else:
+                os.replace(tmp_path, final_path)
+            tmp_path = None
+            # Server-owned source-path binding: this digest may only be served
+            # back for THIS canonical path (see snapshot_servable_for_path).
+            _record_source_binding(hex_digest, source)
+            _enforce_quota_locked(directory)
+            return hex_digest
+        except OSError as exc:
+            logger.debug("media snapshot capture failed for %s: %s", source, exc)
+            if tmp_path is not None:
+                try:
+                    tmp_path.unlink()
+                except OSError:
+                    pass
+            return None
+
+
+def annotate_media_snapshots(
+    messages: list,
+    *,
+    resolve_ref=None,
+    allowed_predicate=None,
+) -> int:
+    """Scan settled messages and snapshot every local-file MEDIA: reference.
+
+    Writes a ``_media_snapshots`` dict ({absolute path: digest}) onto each
+    assistant message that carries at least one local-file ``MEDIA:`` ref.
+    Messages whose refs are already fully annotated are skipped (idempotent
+    across repeated settles).  A recorded digest is FINAL: even if its blob is
+    later evicted by quota, a re-settle must not re-capture the CURRENT live
+    bytes and silently rebind the historical message — evicted blobs degrade
+    to live-file serving instead (#6979 Round 2 SHOULD-FIX).
+
+    ``resolve_ref(raw_ref) -> Path | None`` maps a raw MEDIA token to an
+    absolute file path (defaults to :func:`resolve_media_ref`); refs it cannot
+    resolve (remote URLs, data: URIs) are ignored.  ``allowed_predicate(path)
+    -> bool`` applies the same allow/deny reasoning as ``/api/media``
+    (defaults to :func:`media_capture_allowed`) so the store never receives
+    bytes the endpoint would not serve.
+
+    Returns the number of new snapshots captured (0 on a repeat settle).
+    """
+    import re as _re
+
+    if resolve_ref is None:
+        resolve_ref = resolve_media_ref
+    if allowed_predicate is None:
+        allowed_predicate = media_capture_allowed
+    media_re = _re.compile(r"MEDIA:([^\s\)\]]+)")
+    captured = 0
+    for msg in messages or []:
+        if not isinstance(msg, dict) or msg.get("role") != "assistant":
+            continue
+        content = msg.get("content")
+        if not isinstance(content, str) or "MEDIA:" not in content:
+            continue
+        refs = media_re.findall(content)
+        if not refs:
+            continue
+        existing = msg.get("_media_snapshots")
+        snaps = dict(existing) if isinstance(existing, dict) else {}
+        changed = False
+        for raw_ref in refs:
+            if resolve_ref is not None:
+                try:
+                    path = resolve_ref(raw_ref)
+                except Exception:
+                    path = None
+            else:
+                path = None
+            if path is None:
+                continue
+            # Index by BOTH the resolved absolute path and the raw token as the
+            # frontend embeds it (file:// unwrapped, ~/ kept verbatim). One of
+            # the two always matches the path= query param in the rendered URL.
+            keys = [str(path)]
+            if raw_ref not in keys:
+                keys.append(raw_ref)
+            # A recorded digest is FINAL once stamped (blob presence is NOT
+            # re-checked): quota eviction must not cause a re-settle to
+            # re-capture the current live bytes and rebind the historical
+            # message — that would defeat per-message immutability and thrash
+            # the store with re-capture I/O. Evicted blobs simply fall back to
+            # live-file serving on the serve side.
+            pending = [k for k in keys if not (snaps.get(k) and is_valid_digest(snaps[k]))]
+            if not pending:
+                continue  # already stored under every key — zero-I/O fast path
+            if allowed_predicate is not None:
+                try:
+                    if not allowed_predicate(path):
+                        continue
+                except Exception:
+                    continue
+            digest = capture_snapshot(path)
+            if digest:
+                for k in keys:
+                    snaps[k] = digest
+                changed = True
+                captured += 1
+        if changed:
+            msg["_media_snapshots"] = snaps
+    return captured

api/models.py

@@ -8414,6 +8414,12 @@ _SESSION_MESSAGE_DISPLAY_METADATA_KEYS = (
     "_statusCard",
     "_anchor_stream_id",
     "_anchor_activity_scene",
+    # Map of absolute media path -> content-addressed snapshot digest, stamped
+    # at turn-settle time (api/media_snapshots.py) so historical previews keep
+    # showing the file bytes the turn emitted even after the file is
+    # overwritten in place. Display-only metadata: must survive the
+    # sidecar/state.db merge exactly like the other keys above.
+    "_media_snapshots",
 )
 
 

api/routes.py

@@ -18499,12 +18499,15 @@ def _etag_and_snapshot(fd, *, file_size: int) -> tuple[str | None, bytes | None,
     return _bytes_etag(data), data, actual_size
 
 
-def _serve_file_bytes(handler, target: Path, mime: str, disposition: str, cache_control: str, *, csp: str | None = None, anchor_root: Path | None = None):
+def _serve_file_bytes(handler, target: Path, mime: str, disposition: str, cache_control: str, *, csp: str | None = None, anchor_root: Path | None = None, download_name: str | None = None):
     """Serve a file with correct MIME/disposition and optional byte-range support.
 
     Supports conditional GET via If-None-Match (ETag) — when the ETag matches,
     the request is short-circuited with 304 so revalidating clients (e.g.
     `no-cache` responses) do not re-download unchanged files.
+
+    ``download_name`` overrides the Content-Disposition filename (used when
+    serving an immutable media snapshot whose on-disk name is a content digest).
     """
     fd = None
     try:
@@ -18590,7 +18593,7 @@ def _serve_file_bytes(handler, target: Path, mime: str, disposition: str, cache_
         if byte_range:
             handler.send_header("Content-Range", f"bytes {start}-{end}/{file_size}")
         handler.send_header("Cache-Control", cache_control)
-        handler.send_header("Content-Disposition", _content_disposition_value(disposition, target.name))
+        handler.send_header("Content-Disposition", _content_disposition_value(disposition, download_name or target.name))
         if csp:
             # Sandboxed inline HTML must remain frameable for workspace previews;
             # X-Frame-Options: DENY would block the iframe before CSP sandbox applies.
@@ -19336,6 +19339,201 @@ def _path_is_within_root(child: Path, root: Path) -> bool:
         return False
 
 
+def _media_deny_reason(target: Path) -> str | None:
+    """Return a reason string when ``target`` must be hard-denied, else None.
+
+    The ``/api/media`` #3234 state/profile deny model, extracted so the
+    snapshot CAPTURE side (api/media_snapshots.media_capture_allowed) shares
+    the EXACT same predicate — anything the serve path refuses is never
+    captured in the first place (#6979 Round 2 MUST-FIX 1 deny parity).
+
+    Model: the ACTIVE WORKSPACE is a legitimate-media carve-out — the user is
+    entitled to their own workspace files (that is also how the workspace file
+    browser reaches them), even when a workspace happens to live under a
+    Hermes root. The deny rules target Hermes's OWN internal state, which lives
+    OUTSIDE any workspace. So: if the target is inside the active workspace, it
+    is never denied here; otherwise we deny known secret/config basenames and
+    the internal state subdirectories across every Hermes root the allowlist
+    accepts (active-profile HERMES_HOME, base ~/.hermes, the api.profiles
+    default home, and STATE_DIR — which also defends sibling profiles).
+    """
+    import os as _os
+
+    _HOME = Path(_os.path.expanduser("~"))
+    _HERMES_HOME = Path(_os.getenv("HERMES_HOME", str(_HOME / ".hermes"))).expanduser()
+
+    _DENY_FILENAMES = {
+        "settings.json", "state.db", "state.db-wal", "state.db-shm",
+        "auth.json", "auth.lock", "config.yaml", "config.yml", ".env",
+        ".signing_key", ".pbkdf2_key", ".sessions.json",
+        "google_token.json", "google_client_secret.json",
+        "gateway_state.json", "channel_directory.json", "jobs.json",
+        "passkeys.json", ".passkey_challenges.json", ".login_attempts.json",
+    }
+    # Internal state subdirs that are sensitive in their entirety. NOTE:
+    # `profiles` is intentionally NOT here — it is a container of profile roots,
+    # each of which has its own legitimate workspace/. We instead enumerate each
+    # named-profile root below and deny ITS state subdirs, so a sibling profile's
+    # secrets are blocked without 403-ing a named-profile workspace. (#3234.)
+    _DENY_SUBDIRS = (
+        "sessions", "memories", "cron", "logs",
+        "checkpoints", "backups",
+        # Content-addressed media snapshots (api/media_snapshots.py) are an
+        # internal store: digest bytes are only reachable through the validated
+        # `snap=` parameter, never as a bare `path=` request. (#media-snapshots)
+        "media_snapshots",
+    )
+    _state_dir = None
+    try:
+        from api.config import STATE_DIR as _STATE_DIR
+        _state_dir = Path(_STATE_DIR).resolve()
+    except Exception:
+        _state_dir = None
+    _base_hermes_home = None
+    try:
+        from api.profiles import _DEFAULT_HERMES_HOME as _BASE_HH
+        _base_hermes_home = Path(_BASE_HH).resolve()
+    except Exception:
+        _base_hermes_home = None
+    _hermes_roots = []
+    for _r in (
+        _HERMES_HOME.resolve(),
+        (_HOME / ".hermes").resolve(),
+        _base_hermes_home,
+        _state_dir,
+    ):
+        if _r is not None and _r not in _hermes_roots:
+            _hermes_roots.append(_r)
+    # Enumerate named-profile roots (<root>/profiles/<name>) and treat each as a
+    # Hermes root in its own right, so a sibling/other profile's sensitive subdirs
+    # + secret files are denied — WITHOUT denying the whole `profiles` container
+    # (which would block a legit named-profile workspace at
+    # <root>/profiles/<name>/workspace/). (Codex review #3234.)
+    _profile_roots = []
+    for _root in list(_hermes_roots):
+        _profiles_dir = (_root / "profiles")
+        try:
+            if _profiles_dir.is_dir():
+                for _pchild in _profiles_dir.iterdir():
+                    if _pchild.is_dir():
+                        _pr = _pchild.resolve()
+                        if _pr not in _hermes_roots and _pr not in _profile_roots:
+                            _profile_roots.append(_pr)
+        except OSError:
+            pass
+    _hermes_roots.extend(_profile_roots)
+
+    # Case-insensitive path helpers so STATE.DB / Sessions/ casing variants
+    # cannot bypass the deny on macOS/Windows filesystems (Codex review #3234).
+    def _norm(p):
+        return os.path.normcase(str(Path(p).resolve())).casefold()
+    def _within_ci(child, root):
+        try:
+            c, r = _norm(child), _norm(root)
+            return os.path.commonpath([c, r]) == r
+        except (ValueError, OSError):
+            return False
+    def _equal_ci(a, b):
+        try:
+            return _norm(a) == _norm(b)
+        except (ValueError, OSError):
+            return False
+
+    # State-subdir deny set: each DENY_SUBDIR directly under any Hermes root
+    # (which includes STATE_DIR — so STATE_DIR/sessions, STATE_DIR/memories,
+    # etc. are covered). These ALWAYS apply — even to a file under the active
+    # workspace — so a workspace pointed at (or overlapping) a state dir cannot
+    # expose sessions/memories/profiles/etc. We do NOT deny STATE_DIR itself
+    # wholesale: the default workspace lives at STATE_DIR/workspace, and that is
+    # legitimate user media — direct sensitive files there are still caught by
+    # the filename denies below. (Codex review #3234.)
+    _deny_dirs = []
+    for _root in _hermes_roots:
+        for _sub in _DENY_SUBDIRS:
+            _deny_dirs.append((_root / _sub).resolve())
+        # Per-profile WebUI state lives at <root>/webui_state (api/workspace.py),
+        # so its state subdirs (<root>/webui_state/sessions, etc.) must be denied
+        # too — they are NOT direct children of <root>. (Codex review #3234.)
+        _ws_state = (_root / "webui_state")
+        for _sub in _DENY_SUBDIRS:
+            _deny_dirs.append((_ws_state / _sub).resolve())
+    # The configured media-snapshot store root itself: blobs are internal and
+    # only reachable through the validated `snap=` parameter on an authorized
+    # path, so a bare `path=` request at or below the store is rejected
+    # REGARDLESS of the store's configured name/location
+    # (HERMES_WEBUI_MEDIA_SNAPSHOT_DIR may point anywhere, e.g. /tmp/custom-name;
+    # the literal "media_snapshots" entry above only covers the default layout).
+    # (#6979 Round 2 MUST-FIX 2.)
+    try:
+        from api.media_snapshots import get_snapshot_dir
+        _snap_store = get_snapshot_dir().resolve()
+    except Exception:
+        _snap_store = None
+    if _snap_store is not None and _within_ci(target, _snap_store):
+        return "media snapshot store is internal"
+    _deny_names_ci = {n.casefold() for n in _DENY_FILENAMES}
+
+    # Active-workspace carve-out: a file inside a genuine PROJECT workspace is
+    # the user's own content, so the secret/config FILENAME denies are relaxed
+    # for it. The carve-out is DISABLED when the workspace is a broad/internal
+    # location ($HOME, a Hermes root itself, an ANCESTOR of a Hermes root, a
+    # */profiles dir, a named-profile root, or a state subdir) — honoring those
+    # would re-open the disclosure. A workspace that is a proper DESCENDANT of a
+    # Hermes root (e.g. STATE_DIR/workspace) is still a legit project workspace
+    # and keeps the carve-out. The dir-based denies above are NOT relaxed.
+    _active_workspace = None
+    try:
+        from api.workspace import get_last_workspace
+        _aw = Path(get_last_workspace()).resolve()
+        if _aw.is_dir():
+            _active_workspace = _aw
+    except Exception:
+        _active_workspace = None
+
+    def _workspace_is_safe_carveout(ws):
+        if ws is None:
+            return False
+        if _equal_ci(ws, _HOME):
+            return False
+        for _root in _hermes_roots:
+            # ws IS a root, or ws is an ANCESTOR of a root → unsafe. (A proper
+            # descendant of a root is fine — that's a normal project workspace.)
+            if _equal_ci(ws, _root) or _within_ci(_root, ws):
+                return False
+        if ws.name == "profiles" or ws.parent.name == "profiles":
+            return False
+        if ws.name in _DENY_SUBDIRS:
+            return False
+        return True
+
+    _in_active_workspace = (
+        _active_workspace is not None
+        and _workspace_is_safe_carveout(_active_workspace)
+        and _within_ci(target, _active_workspace)
+    )
+
+    # Dir-based denies always fire (even inside the active workspace).
+    if any(_within_ci(target, d) for d in _deny_dirs):
+        return "denied state subdir"
+    # Filename-based denies fire for files under a Hermes root, UNLESS the file
+    # is inside a genuine project workspace (carve-out).
+    if not _in_active_workspace:
+        _under_hermes_root = any(_within_ci(target, _root) for _root in _hermes_roots)
+        _name_cf = target.name.casefold()
+        # Exact secret/state basenames, plus atomic-write temp files for those
+        # (api/auth.py and api/passkeys.py write via a `tmp*.<name>.tmp` / `tmp*.tmp`
+        # sidecar then rename) — deny those suffixes too so a momentary temp file
+        # cannot be fetched. (Codex review #3234.)
+        _deny_tmp_suffixes = (".sessions.tmp", ".login_attempts.tmp",
+                              ".passkeys.tmp", ".passkey_challenges.tmp")
+        if _under_hermes_root and (
+            _name_cf in _deny_names_ci
+            or _name_cf.endswith(_deny_tmp_suffixes)
+        ):
+            return "denied state filename"
+    return None
+
+
 def _handle_media(handler, parsed):
     """Serve a local file by absolute path for inline display in the chat.
 
@@ -19437,175 +19635,21 @@ def _handle_media(handler, parsed):
     # covers every entry path (bare file:// URLs, markdown anchors, MEDIA:
     # tokens, and session-token grants).
     #
-    # Model: the ACTIVE WORKSPACE is a legitimate-media carve-out — the user is
-    # entitled to their own workspace files (that is also how the workspace file
-    # browser reaches them), even when a workspace happens to live under a
-    # Hermes root. The deny rules target Hermes's OWN internal state, which lives
-    # OUTSIDE any workspace. So: if the target is inside the active workspace, it
-    # is never denied here; otherwise we deny known secret/config basenames and
-    # the internal state subdirectories across every Hermes root the allowlist
-    # accepts (active-profile HERMES_HOME, base ~/.hermes, the api.profiles
-    # default home, and STATE_DIR — which also defends sibling profiles).
-    _DENY_FILENAMES = {
-        "settings.json", "state.db", "state.db-wal", "state.db-shm",
-        "auth.json", "auth.lock", "config.yaml", "config.yml", ".env",
-        ".signing_key", ".pbkdf2_key", ".sessions.json",
-        "google_token.json", "google_client_secret.json",
-        "gateway_state.json", "channel_directory.json", "jobs.json",
-        "passkeys.json", ".passkey_challenges.json", ".login_attempts.json",
-    }
-    # Internal state subdirs that are sensitive in their entirety. NOTE:
-    # `profiles` is intentionally NOT here — it is a container of profile roots,
-    # each of which has its own legitimate workspace/. We instead enumerate each
-    # named-profile root below and deny ITS state subdirs, so a sibling profile's
-    # secrets are blocked without 403-ing a named-profile workspace. (#3234.)
-    _DENY_SUBDIRS = (
-        "sessions", "memories", "cron", "logs",
-        "checkpoints", "backups",
-    )
-    _state_dir = None
-    try:
-        from api.config import STATE_DIR as _STATE_DIR
-        _state_dir = Path(_STATE_DIR).resolve()
-    except Exception:
-        _state_dir = None
-    _base_hermes_home = None
-    try:
-        from api.profiles import _DEFAULT_HERMES_HOME as _BASE_HH
-        _base_hermes_home = Path(_BASE_HH).resolve()
-    except Exception:
-        _base_hermes_home = None
-    _hermes_roots = []
-    for _r in (
-        _HERMES_HOME.resolve(),
-        (_HOME / ".hermes").resolve(),
-        _base_hermes_home,
-        _state_dir,
-    ):
-        if _r is not None and _r not in _hermes_roots:
-            _hermes_roots.append(_r)
-    # Enumerate named-profile roots (<root>/profiles/<name>) and treat each as a
-    # Hermes root in its own right, so a sibling/other profile's sensitive subdirs
-    # + secret files are denied — WITHOUT denying the whole `profiles` container
-    # (which would block a legit named-profile workspace at
-    # <root>/profiles/<name>/workspace/). (Codex review #3234.)
-    _profile_roots = []
-    for _root in list(_hermes_roots):
-        _profiles_dir = (_root / "profiles")
-        try:
-            if _profiles_dir.is_dir():
-                for _pchild in _profiles_dir.iterdir():
-                    if _pchild.is_dir():
-                        _pr = _pchild.resolve()
-                        if _pr not in _hermes_roots and _pr not in _profile_roots:
-                            _profile_roots.append(_pr)
-        except OSError:
-            pass
-    _hermes_roots.extend(_profile_roots)
-
-    # Case-insensitive path helpers so STATE.DB / Sessions/ casing variants
-    # cannot bypass the deny on macOS/Windows filesystems (Codex review #3234).
-    def _norm(p):
-        return os.path.normcase(str(Path(p).resolve())).casefold()
-    def _within_ci(child, root):
-        try:
-            c, r = _norm(child), _norm(root)
-            return os.path.commonpath([c, r]) == r
-        except (ValueError, OSError):
-            return False
-    def _equal_ci(a, b):
-        try:
-            return _norm(a) == _norm(b)
-        except (ValueError, OSError):
-            return False
-
-    # State-subdir deny set: each DENY_SUBDIR directly under any Hermes root
-    # (which includes STATE_DIR — so STATE_DIR/sessions, STATE_DIR/memories,
-    # etc. are covered). These ALWAYS apply — even to a file under the active
-    # workspace — so a workspace pointed at (or overlapping) a state dir cannot
-    # expose sessions/memories/profiles/etc. We do NOT deny STATE_DIR itself
-    # wholesale: the default workspace lives at STATE_DIR/workspace, and that is
-    # legitimate user media — direct sensitive files there are still caught by
-    # the filename denies below. (Codex review #3234.)
-    _deny_dirs = []
-    for _root in _hermes_roots:
-        for _sub in _DENY_SUBDIRS:
-            _deny_dirs.append((_root / _sub).resolve())
-        # Per-profile WebUI state lives at <root>/webui_state (api/workspace.py),
-        # so its state subdirs (<root>/webui_state/sessions, etc.) must be denied
-        # too — they are NOT direct children of <root>. (Codex review #3234.)
-        _ws_state = (_root / "webui_state")
-        for _sub in _DENY_SUBDIRS:
-            _deny_dirs.append((_ws_state / _sub).resolve())
-    _deny_names_ci = {n.casefold() for n in _DENY_FILENAMES}
-
-    # Active-workspace carve-out: a file inside a genuine PROJECT workspace is
-    # the user's own content, so the secret/config FILENAME denies are relaxed
-    # for it. The carve-out is DISABLED when the workspace is a broad/internal
-    # location ($HOME, a Hermes root itself, an ANCESTOR of a Hermes root, a
-    # */profiles dir, a named-profile root, or a state subdir) — honoring those
-    # would re-open the disclosure. A workspace that is a proper DESCENDANT of a
-    # Hermes root (e.g. STATE_DIR/workspace) is still a legit project workspace
-    # and keeps the carve-out. The dir-based denies above are NOT relaxed.
-    _active_workspace = None
-    try:
-        from api.workspace import get_last_workspace
-        _aw = Path(get_last_workspace()).resolve()
-        if _aw.is_dir():
-            _active_workspace = _aw
-    except Exception:
-        _active_workspace = None
-
-    def _workspace_is_safe_carveout(ws):
-        if ws is None:
-            return False
-        if _equal_ci(ws, _HOME):
-            return False
-        for _root in _hermes_roots:
-            # ws IS a root, or ws is an ANCESTOR of a root → unsafe. (A proper
-            # descendant of a root is fine — that's a normal project workspace.)
-            if _equal_ci(ws, _root) or _within_ci(_root, ws):
-                return False
-        if ws.name == "profiles" or ws.parent.name == "profiles":
-            return False
-        if ws.name in _DENY_SUBDIRS:
-            return False
-        return True
-
-    _in_active_workspace = (
-        _active_workspace is not None
-        and _workspace_is_safe_carveout(_active_workspace)
-        and _within_ci(target, _active_workspace)
-    )
-
-    # Dir-based denies always fire (even inside the active workspace).
-    if any(_within_ci(target, d) for d in _deny_dirs):
-        return bad(handler, "Path not in allowed location", 403)
-    # Filename-based denies fire for files under a Hermes root, UNLESS the file
-    # is inside a genuine project workspace (carve-out).
-    if not _in_active_workspace:
-        _under_hermes_root = any(_within_ci(target, _root) for _root in _hermes_roots)
-        _name_cf = target.name.casefold()
-        # Exact secret/state basenames, plus atomic-write temp files for those
-        # (api/auth.py and api/passkeys.py write via a `tmp*.<name>.tmp` / `tmp*.tmp`
-        # sidecar then rename) — deny those suffixes too so a momentary temp file
-        # cannot be fetched. (Codex review #3234.)
-        _deny_tmp_suffixes = (".sessions.tmp", ".login_attempts.tmp",
-                              ".passkeys.tmp", ".passkey_challenges.tmp")
-        if _under_hermes_root and (
-            _name_cf in _deny_names_ci
-            or _name_cf.endswith(_deny_tmp_suffixes)
-        ):
+    # The predicate is SHARED with snapshot capture
+    # (api/media_snapshots.media_capture_allowed) so capture and serve can
+    # never diverge on what is denied — anything denied here is never
+    # snapshotted in the first place. (#6979 Round 2 MUST-FIX 1.)
+    deny_reason = _media_deny_reason(target)
+    if deny_reason:
         return bad(handler, "Path not in allowed location", 403)
     # ── end #3234 deny ───────────────────────────────────────────────────────
 
     if not within_allowed and not session_media_allowed:
         return bad(handler, "Path not in allowed location", 403)
 
-    if not target.exists() or not target.is_file():
-        return j(handler, {"error": "not found"}, status=404)
-
-    # Determine MIME type
+    # Determine MIME type from the requested path's extension. Computed BEFORE
+    # the existence check because the requested file may have been overwritten
+    # or deleted while its message-level snapshot still exists below.
     ext = target.suffix.lower()
     mime = MIME_MAP.get(ext, "application/octet-stream")
 
@@ -19624,6 +19668,58 @@ def _handle_media(handler, parsed):
     ) else "attachment"
     # _serve_file_bytes sends Content-Security-Policy when csp is set.
     csp = "sandbox allow-scripts" if html_inline_ok else None
+
+    # ── Message-level snapshot serving (?snap=<sha256>) ─────────────────────
+    # Historical chat previews carry a content-addressed snapshot digest of the
+    # file as it existed when the message settled (see api/media_snapshots.py).
+    # Serving the frozen bytes instead of the live file means an in-place
+    # overwrite (same filename) no longer rewrites old previews — the user can
+    # still compare old vs new. The allow/deny checks above still gate the
+    # request: `snap` only selects WHICH bytes to serve for an already-
+    # authorized path; it never grants access to a path that would be denied
+    # without it. A missing/evicted snapshot falls back to the live file.
+    snap_digest = qs.get("snap", [""])[0].strip().lower()
+    snapshot_file = None
+    snap_dir = None
+    if snap_digest:
+        from api.media_snapshots import (
+            get_snapshot_dir,
+            is_valid_digest,
+            snapshot_path_for_digest,
+            snapshot_servable_for_path,
+        )
+
+        snap_dir = get_snapshot_dir().resolve()
+        if is_valid_digest(snap_digest):
+            snapshot_file = snapshot_path_for_digest(snap_digest)
+            # Server-owned source-path binding (#6979 Round 2 MUST-FIX 1): a
+            # digest may only be served back for the EXACT canonical path it
+            # was captured from. Replaying a digest through a different
+            # (allowed) path must not leak the stored bytes — treat it as an
+            # invalid snapshot and fall back to the live file (or 404 when the
+            # live file is absent).
+            if snapshot_file is not None and not snapshot_servable_for_path(snap_digest, target):
+                snapshot_file = None
+    if snapshot_file is not None:
+        # Content-addressed and immutable: the digest IS the SHA-256 of the
+        # exact bytes, so the browser may cache forever and never revalidate.
+        # The blob is opened ANCHORED inside the store root (no-follow), and
+        # the store root itself is deny-listed from bare path= fetches above
+        # (#6979 Round 2 MUST-FIX 2).
+        return _serve_file_bytes(
+            handler,
+            snapshot_file,
+            mime,
+            disposition,
+            "private, max-age=31536000, immutable",
+            csp=csp,
+            download_name=target.name,
+            anchor_root=snap_dir,
+        )
+
+    if not target.exists() or not target.is_file():
+        return j(handler, {"error": "not found"}, status=404)
+
     # HTML inline previews change frequently (agent edits + re-renders).
     # Use no-store so the browser always fetches fresh content, avoiding stale
     # previews that require a manual full-page refresh to update.

api/streaming.py

@@ -1746,6 +1746,24 @@ def _prepare_marker_clean_writeback(
     return [], list(previous_context_messages or []), provenance
 
 
+def _annotate_media_snapshots_for_settled_messages(messages) -> None:
+    """Freeze local-file MEDIA: bytes at settle time so historical previews
+    keep showing the version the turn emitted, even after the file is
+    overwritten in place (#6922 follow-up).
+
+    Runs AFTER the display merge so the annotation rides the exact messages the
+    frontend will render. The store is content-addressed and the annotator is
+    idempotent (already-stored digests short-circuit), so re-settling the same
+    transcript is cheap. Never raises — snapshotting is best-effort durability.
+    """
+    try:
+        from api.media_snapshots import annotate_media_snapshots
+
+        annotate_media_snapshots(messages)
+    except Exception:
+        logger.debug("Media snapshot annotation failed during settle", exc_info=True)
+
+
 def _settle_result_messages(
     session,
     previous_messages,
@@ -1808,6 +1826,7 @@ def _settle_result_messages(
         source=source,
         verification_nudge_provenance=verification_nudge_provenance,
     )
+    _annotate_media_snapshots_for_settled_messages(session.messages)
     _compact_session_image_parts_for_persistence(session)
     _advance_truncation_watermark_after_commit(session)  # #3831
     return result_messages

static/ui.js

@@ -1514,6 +1514,53 @@ function _getCachedRender(text, isUser){
   _renderCache.set(key, rendered);
   return rendered;
 }
+// ── Message-level media snapshot stamping ─────────────────────────────────
+// /api/media serves a file's CURRENT bytes. Since ETag revalidation (#6922),
+// an in-place overwrite (same filename) also rewrites every historical chat
+// preview that referenced it — the old/new comparison is lost. At settle time
+// the backend freezes the bytes of each local-file MEDIA: reference into a
+// content-addressed store and stamps the message with
+// `_media_snapshots: {path: digest}`. This helper rewrites the rendered HTML
+// of ONE message to append `&snap=<digest>` to the matching /api/media URLs
+// (and `data-snap` on lazy-preview placeholders), so old previews keep
+// showing the file as it was when the message was emitted.
+// Runs AFTER the text-keyed render cache: the cache stays pure-text, and each
+// message stamps its own digests — two messages with identical text but
+// different snapshots (exactly the old/new case) resolve independently.
+function _stampMediaSnapshots(html, snaps){
+  if(!html || !snaps || typeof snaps !== 'object') return html;
+  let out = String(html);
+  // Direct media URLs: rewrite each COMPLETE `path=` query value atomically.
+  // Value-level parsing (decode the whole value, exact map lookup) instead of
+  // substring split/join: when one path is a PREFIX of another
+  // (/tmp/a.png vs /tmp/a.png.backup) the naive rewrite corrupts the longer
+  // URL and drops its own digest. Matching is boundary-aware — the value runs
+  // to the next `&` separator or an attribute quote/whitespace — so nothing
+  // outside the value ever moves.
+  out = out.replace(/api\/media[?&]path=([^&"'\s<>]+)/g, (match, encodedPath)=>{
+    let decoded;
+    try{ decoded = decodeURIComponent(encodedPath); }
+    catch(e){ return match; }
+    const digest = snaps[decoded];
+    if(typeof digest === 'string' && /^[0-9a-f]{64}$/.test(digest)){
+      return match + '&snap=' + digest;
+    }
+    return match;
+  });
+  // Lazy-preview placeholders: data-path="<html-escaped raw path>" — match the
+  // complete attribute value, unescape HTML entities, then exact lookup (same
+  // prefix-safety: a longer path's attribute can never be partially matched).
+  const _unescapeHtml=(s)=>String(s||'').replace(/&quot;/g,'"').replace(/&#39;/g,"'").replace(/&lt;/g,'<').replace(/&gt;/g,'>').replace(/&amp;/g,'&');
+  out = out.replace(/data-path="([^"]*)"/g, (match, rawValue)=>{
+    const decoded = _unescapeHtml(rawValue);
+    const digest = snaps[decoded];
+    if(typeof digest === 'string' && /^[0-9a-f]{64}$/.test(digest)){
+      return match + ' data-snap="' + digest + '"';
+    }
+    return match;
+  });
+  return out;
+}
 function _currentMessageRenderWindowSize(){
   return Math.max(
     MESSAGE_RENDER_WINDOW_DEFAULT,
@@ -16564,6 +16611,13 @@ function renderMessages(options){
       }).join('')}</div>`;
     }
     let bodyHtml = _getCachedRender(displayContent, isUser);
+    // Message-level media snapshots: settled assistant messages carry a
+    // path→digest map (written at settle time) freezing the file bytes the
+    // turn emitted. Stamp it AFTER the text-keyed render cache so identical
+    // text with different snapshots (old/new comparison) never collides.
+    if(!isUser && m && m._media_snapshots && typeof m._media_snapshots==='object'){
+      bodyHtml = _stampMediaSnapshots(bodyHtml, m._media_snapshots);
+    }
     if(!isUser&&m.provider_details){
       const summary=m.provider_details_label||'Provider details';
       bodyHtml += `<details class="provider-error-details"><summary>${esc(String(summary))}</summary><pre><code>${esc(String(m.provider_details))}</code></pre></details>`;
@@ -16784,10 +16838,15 @@ function renderMessages(options){
         if(!firstSeg&&thinkingText&&window._showThinking!==false&&!((isCompactWorklogMode()||isTransparentStream())&&_assistantThinkingBelongsInWorklog(m, rawIdx, toolCallAssistantIdxs))) orderedSeg.insertAdjacentHTML('beforeend', _thinkingCardHtml(thinkingText));
         const isLastTextPart=partIdx===lastTextPartIdx;
         const partBodyHtml=_getCachedRender(partDisplayText,false);
+        // Message-level media snapshots: transparent ordered segments carry the
+        // same per-message path→digest map as the main transcript; stamp it so
+        // historical previews freeze (&snap=) instead of following overwrites.
+        // Inlined in the template (no intermediate variable) so test-harness
+        // block extraction of the ordered-segment slice stays self-contained.
         if(isLastTextPart&&statusHtml){
           orderedSeg.insertAdjacentHTML('beforeend', statusHtml);
         }
-        orderedSeg.insertAdjacentHTML('beforeend', `${isLastTextPart?filesHtml:''}<div class="msg-body">${partBodyHtml}</div>${isLastTextPart?footHtml:''}`);
+        orderedSeg.insertAdjacentHTML('beforeend', `${isLastTextPart?filesHtml:''}<div class="msg-body">${(typeof m!=='undefined'&&m&&m._media_snapshots&&typeof m._media_snapshots==='object')?_stampMediaSnapshots(partBodyHtml,m._media_snapshots):partBodyHtml}</div>${isLastTextPart?footHtml:''}`);
         blocks.appendChild(orderedSeg);
         if(!firstSeg) firstSeg=orderedSeg;
       });
@@ -19165,7 +19224,8 @@ function loadDiffInline(container){
   root.querySelectorAll('.diff-inline-load:not([data-loaded])').forEach(el=>{
     el.setAttribute('data-loaded','1');
     const path=el.dataset.path;
-    fetch('api/media?path='+encodeURIComponent(path))
+    const snapQuery=_mediaSnapQuery(el);
+    fetch('api/media?path='+encodeURIComponent(path)+snapQuery)
       .then(r=>{if(!r.ok) throw new Error(r.status);return r.text();})
       .then(text=>{
         if(text.length>DIFF_MAX_SIZE){
@@ -19194,8 +19254,18 @@ function _mediaSessionQuery(){
   return mediaSessionId?'&session_id='+encodeURIComponent(mediaSessionId):'';
 }
 
+// Message-level media snapshots: lazy preview loaders (pdf/html/csv/diff/
+// excalidraw) build their fetch URL from data-path at load time. The stamping
+// pass in _stampMediaSnapshots carries the content digest in data-snap;
+// append it here so the preview shows the file as the message emitted it.
+function _mediaSnapQuery(el){
+  const snap=el&&el.dataset?el.dataset.snap:'';
+  return (snap&&/^[0-9a-f]{64}$/.test(snap))?('&snap='+snap):'';
+}
+
 function _csvMediaUrl(path, opts={}){
   let url='api/media?path='+encodeURIComponent(path)+_mediaSessionQuery();
+  if(opts.snap) url+='&snap='+encodeURIComponent(opts.snap);
   if(opts.download) url+='&download=1';
   return url;
 }
@@ -19235,8 +19305,9 @@ function loadCsvInline(container){
   root.querySelectorAll('.csv-inline-load:not([data-loaded])').forEach(el=>{
     el.setAttribute('data-loaded','1');
     const path=el.dataset.path;
-    const mediaUrl=_csvMediaUrl(path);
-    const downloadUrl=_csvMediaUrl(path,{download:true});
+    const snap=_mediaSnapQuery(el).replace(/^&snap=/,'');
+    const mediaUrl=_csvMediaUrl(path,{snap:snap||undefined});
+    const downloadUrl=_csvMediaUrl(path,{download:true,snap:snap||undefined});
     fetch(mediaUrl)
       .then(r=>{if(!r.ok) throw new Error(r.status);return r.text();})
       .then(text=>{
@@ -19255,7 +19326,8 @@ function loadExcalidrawInline(container){
   root.querySelectorAll('.excalidraw-inline-load:not([data-loaded])').forEach(el=>{
     el.setAttribute('data-loaded','1');
     const path=el.dataset.path;
-    fetch('api/media?path='+encodeURIComponent(path))
+    const snapQuery=_mediaSnapQuery(el);
+    fetch('api/media?path='+encodeURIComponent(path)+snapQuery)
       .then(r=>{if(!r.ok) throw new Error(r.status);return r.text();})
       .then(text=>{
         if(text.length>EXCALIDRAW_MAX_SIZE){
@@ -19384,14 +19456,15 @@ function loadPdfInline(container){
     const path=el.dataset.path;
     const fname=path.split('/').pop()||path;
     const mediaSessionId=(typeof S!=='undefined'&&S&&S.session&&S.session.session_id)?String(S.session.session_id):'';
+    const snapQuery=_mediaSnapQuery(el);
     const publicMediaUrl='api/media?path='+encodeURIComponent(path);
-    const mediaUrl=publicMediaUrl+(mediaSessionId?'&session_id='+encodeURIComponent(mediaSessionId):'');
+    const mediaUrl=publicMediaUrl+(mediaSessionId?'&session_id='+encodeURIComponent(mediaSessionId):'')+snapQuery;
     const loadPdf=(pdfjsLib)=>{
       fetch(mediaUrl)
         .then(r=>{if(!r.ok) throw new Error(r.status); return r.arrayBuffer();})
         .then(buf=>{
           if(buf.byteLength>PDF_MAX_SIZE){
-            const dlUrl=publicMediaUrl+'&download=1';
+            const dlUrl=publicMediaUrl+'&download=1'+snapQuery;
             el.outerHTML=`<div class="pdf-preview-fallback"><a class="msg-media-link" href="${dlUrl}" download="${esc(fname)}">📎 ${esc(fname)}</a><br><span style="color:var(--muted);font-size:12px">${t('pdf_too_large')}</span></div>`;
             return;
           }
@@ -19399,7 +19472,7 @@ function loadPdfInline(container){
         })
         .then(pdf=>{
           if(!pdf) return;
-          const dlUrl=publicMediaUrl+'&download=1';
+          const dlUrl=publicMediaUrl+'&download=1'+snapQuery;
           const total=pdf.numPages;
           const pagesLabel=total>1?` · ${total} pages`:'';
           const wrap=document.createElement('div');
@@ -19438,7 +19511,7 @@ function loadPdfInline(container){
           renderPage(1);
         })
         .catch(()=>{
-          const dlUrl=publicMediaUrl+'&download=1';
+          const dlUrl=publicMediaUrl+'&download=1'+snapQuery;
           el.outerHTML=`<div class="pdf-preview-fallback"><a class="msg-media-link" href="${dlUrl}" download="${esc(fname)}">📎 ${esc(fname)}</a><br><span style="color:var(--muted);font-size:12px">${t('pdf_error')}</span></div>`;
         });
     };
@@ -19458,7 +19531,7 @@ function loadPdfInline(container){
       window.addEventListener('pdfjs-ready',()=>{ _pdfjsReady=true; loadPdf(window._pdfjsLib); },{once:true});
       setTimeout(()=>{
         if(!_pdfjsReady){
-          const dlUrl=publicMediaUrl+'&download=1';
+          const dlUrl=publicMediaUrl+'&download=1'+snapQuery;
           if(el.parentNode){
             el.outerHTML=`<div class="pdf-preview-fallback"><a class="msg-media-link" href="${dlUrl}" download="${esc(fname)}">📎 ${esc(fname)}</a><br><span style="color:var(--muted);font-size:12px">${t('pdf_error')}</span></div>`;
           }
@@ -19479,22 +19552,23 @@ function loadHtmlInline(container){
     const path=el.dataset.path;
     const fname=path.split('/').pop()||path;
     const mediaSessionId=(typeof S!=='undefined'&&S&&S.session&&S.session.session_id)?String(S.session.session_id):'';
+    const snapQuery=_mediaSnapQuery(el);
     const publicMediaUrl='api/media?path='+encodeURIComponent(path);
-    const mediaUrl=publicMediaUrl+(mediaSessionId?'&session_id='+encodeURIComponent(mediaSessionId):'');
+    const mediaUrl=publicMediaUrl+(mediaSessionId?'&session_id='+encodeURIComponent(mediaSessionId):'')+snapQuery;
     fetch(mediaUrl, {cache:'no-store'})
       .then(r=>{if(!r.ok) throw new Error(r.status); return r.text();})
       .then(html=>{
         if(html.length>HTML_MAX_SIZE){
-          const openUrl=publicMediaUrl+'&inline=1';
+          const openUrl=publicMediaUrl+'&inline=1'+snapQuery;
           el.outerHTML=`<div class="html-preview-fallback"><a class="msg-media-link" href="${openUrl}" target="_blank" rel="noopener">📎 ${esc(fname)}</a><br><span style="color:var(--muted);font-size:12px">${t('html_too_large')}</span></div>`;
           return;
         }
-        const openUrl=publicMediaUrl+'&inline=1';
+        const openUrl=publicMediaUrl+'&inline=1'+snapQuery;
         const safeHtml=html.replace(/&/g,'&amp;').replace(/"/g,'&quot;').replace(/</g,'&lt;').replace(/>/g,'&gt;');
         el.outerHTML=`<div class="html-preview-wrap"><div class="html-preview-header"><span>${t('html_sandbox_label')}</span><a href="${openUrl}" target="_blank" rel="noopener" class="html-open-link">${t('html_open_full')} ↗</a></div><iframe srcdoc="${safeHtml}" sandbox="allow-scripts" class="html-preview-iframe" loading="lazy"></iframe></div>`;
       })
       .catch(()=>{
-        const dlUrl=publicMediaUrl+'&download=1';
+        const dlUrl=publicMediaUrl+'&download=1'+snapQuery;
         el.outerHTML=`<div class="html-preview-fallback"><a class="msg-media-link" href="${dlUrl}" download="${esc(fname)}">📎 ${esc(fname)}</a><br><span style="color:var(--muted);font-size:12px">${t('html_error')}</span></div>`;
       });
   });

tests/test_media_message_snapshots.py

@@ -0,0 +1,697 @@
+"""Message-level media snapshots: freeze file bytes at settle time.
+
+PR #6922 made /api/media revalidate on every use (no-cache + ETag), so an
+in-place overwrite of a file (same filename) also rewrites every historical
+chat preview that referenced it — the old/new comparison is lost. This suite
+covers the fix: at settle time the WebUI snapshots each local-file MEDIA:
+reference into a content-addressed store and stamps the message with
+``_media_snapshots``; the frontend appends ``&snap=<digest>`` to historical
+preview URLs and /api/media serves the frozen bytes instead of the live file.
+
+Key property under test: a snapshot survives the original file being
+overwritten AND deleted, while a request WITHOUT a snap keeps serving the live
+(possibly new) bytes.
+"""
+
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class _FakeHandler:
+    def __init__(self, headers=None):
+        self.status = None
+        self.sent_headers: list[tuple[str, str]] = []
+        self.body = bytearray()
+        self.wfile = self
+        self.headers = dict(headers or {})
+
+    def send_response(self, code):
+        self.status = code
+
+    def send_header(self, key, value):
+        self.sent_headers.append((key, value))
+
+    def end_headers(self):
+        pass
+
+    def write(self, data):
+        self.body.extend(data)
+
+    def header(self, key):
+        return next((v for k, v in self.sent_headers if k == key), "") or ""
+
+
+@pytest.fixture
+def routes():
+    from api import routes
+
+    return routes
+
+
+@pytest.fixture
+def snap_dir(tmp_path, monkeypatch):
+    """Isolate the snapshot store per test and point it at tmp_path."""
+    store = tmp_path / "media_snapshots"
+    monkeypatch.setenv("HERMES_WEBUI_MEDIA_SNAPSHOT_DIR", str(store))
+    return store
+
+
+def _media_get(routes, monkeypatch, target, headers=None, query_extra=""):
+    monkeypatch.setattr("api.auth.is_auth_enabled", lambda: False)
+    handler = _FakeHandler(headers)
+    parsed = SimpleNamespace(path="/api/media", query=f"path={target}{query_extra}")
+    routes._handle_media(handler, parsed)
+    return handler
+
+
+# ── capture_snapshot ───────────────────────────────────────────────────────
+
+
+def test_capture_snapshot_stores_content_addressed_bytes(snap_dir, tmp_path):
+    from api.media_snapshots import capture_snapshot, snapshot_path_for_digest
+
+    source = tmp_path / "report.html"
+    source.write_bytes(b"<html>v1</html>")
+
+    digest = capture_snapshot(source)
+    assert digest and len(digest) == 64
+
+    stored = snapshot_path_for_digest(digest)
+    assert stored is not None
+    assert stored.read_bytes() == b"<html>v1</html>"
+
+
+def test_capture_snapshot_dedupes_identical_content(snap_dir, tmp_path):
+    from api.media_snapshots import capture_snapshot
+
+    a = tmp_path / "a.png"
+    b = tmp_path / "b.png"
+    a.write_bytes(b"same-bytes")
+    b.write_bytes(b"same-bytes")
+
+    d1 = capture_snapshot(a)
+    d2 = capture_snapshot(b)
+    assert d1 == d2
+    # One .snap blob only (the dedup contract); the source-binding sidecar
+    # (.src.json) is a separate small file and does not count as a blob.
+    assert len(list(snap_dir.glob("*.snap"))) == 1
+
+
+def test_capture_snapshot_skips_missing_and_over_cap(snap_dir, tmp_path):
+    from api.media_snapshots import capture_snapshot
+
+    assert capture_snapshot(tmp_path / "missing.png") is None
+
+    big = tmp_path / "big.mp4"
+    big.write_bytes(b"x" * 1024)
+    assert capture_snapshot(big, max_file_bytes=512) is None
+
+
+def test_capture_snapshot_skips_directories(snap_dir, tmp_path):
+    from api.media_snapshots import capture_snapshot
+
+    assert capture_snapshot(tmp_path) is None
+
+
+# ── resolve_media_ref / media_capture_allowed ──────────────────────────────
+
+
+def test_resolve_media_ref_handles_file_url_and_expands_home(tmp_path, monkeypatch):
+    from api.media_snapshots import resolve_media_ref
+
+    target = tmp_path / "x.html"
+    target.write_text("hi")
+
+    assert resolve_media_ref(str(target)) == target.resolve()
+    assert resolve_media_ref("file://" + str(target)) == target.resolve()
+    assert resolve_media_ref("https://example.com/a.png") is None
+    assert resolve_media_ref("data:image/png;base64,AAAA") is None
+    assert resolve_media_ref("") is None
+
+
+def test_media_capture_allowed_denies_hermes_state(tmp_path, monkeypatch):
+    from api.media_snapshots import media_capture_allowed
+
+    # Files under an allowed root (tmp) are fine...
+    allowed = tmp_path / "ok.html"
+    allowed.write_text("x")
+    assert media_capture_allowed(allowed) is True
+
+    # ...but a deny-listed filename under a Hermes root is never snapshotted.
+    # Point HOME at the fake tree so <fake-home>/.hermes counts as a root.
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    secret = hermes_home / "settings.json"
+    secret.write_text("{}")
+    monkeypatch.setenv("HOME", str(tmp_path))
+    assert media_capture_allowed(secret) is False
+
+
+# ── annotate_media_snapshots ───────────────────────────────────────────────
+
+
+def test_annotate_stamps_assistant_messages_with_snapshots(snap_dir, tmp_path):
+    from api.media_snapshots import annotate_media_snapshots
+
+    target = tmp_path / "report.html"
+    target.write_text("<html>v1</html>")
+
+    messages = [
+        {"role": "user", "content": "please build it"},
+        {"role": "assistant", "content": f"done: MEDIA:{target}"},
+        {"role": "assistant", "content": "no media here"},
+    ]
+    captured = annotate_media_snapshots(messages)
+    assert captured == 1
+
+    stamped = messages[1]["_media_snapshots"]
+    assert str(target) in stamped
+    assert len(stamped[str(target)]) == 64
+
+
+def test_annotate_is_idempotent_across_settles(snap_dir, tmp_path):
+    from api.media_snapshots import annotate_media_snapshots
+
+    target = tmp_path / "report.html"
+    target.write_text("<html>v1</html>")
+    messages = [{"role": "assistant", "content": f"MEDIA:{target}"}]
+
+    assert annotate_media_snapshots(messages) == 1
+    assert annotate_media_snapshots(messages) == 0  # fast-path skip
+    assert len(list(snap_dir.glob("*.snap"))) == 1
+
+
+def test_annotate_skips_remote_and_data_refs(snap_dir, tmp_path):
+    from api.media_snapshots import annotate_media_snapshots
+
+    messages = [
+        {"role": "assistant", "content": "MEDIA:https://example.com/a.png"},
+        {"role": "assistant", "content": "MEDIA:data:image/png;base64,AAAA"},
+    ]
+    assert annotate_media_snapshots(messages) == 0
+    assert "_media_snapshots" not in messages[0]
+    assert "_media_snapshots" not in messages[1]
+
+
+# ── /api/media?snap= serving ───────────────────────────────────────────────
+
+
+def test_handle_media_serves_snapshot_after_inplace_overwrite(routes, monkeypatch, snap_dir, tmp_path):
+    """THE regression test: same filename overwritten must not rewrite old
+    previews when the message pins a snapshot digest."""
+    from api.media_snapshots import capture_snapshot
+
+    target = tmp_path / "report.html"
+    target.write_text("<html>v1</html>")
+    digest = capture_snapshot(target)
+
+    # Overwrite the file in place (the scenario that used to break history).
+    time.sleep(0.01)
+    target.write_text("<html>v2 - completely different</html>")
+
+    # Without snap: live file (new bytes).
+    live = _media_get(routes, monkeypatch, target)
+    assert live.status == 200
+    assert bytes(live.body) == b"<html>v2 - completely different</html>"
+
+    # With snap: frozen v1 bytes, immutable caching.
+    pinned = _media_get(routes, monkeypatch, target, query_extra=f"&snap={digest}")
+    assert pinned.status == 200
+    assert bytes(pinned.body) == b"<html>v1</html>"
+    assert pinned.header("Cache-Control") == "private, max-age=31536000, immutable"
+
+
+def test_handle_media_snapshot_survives_file_deletion(routes, monkeypatch, snap_dir, tmp_path):
+    from api.media_snapshots import capture_snapshot
+
+    target = tmp_path / "pic.png"
+    target.write_bytes(b"png-bytes-v1")
+    digest = capture_snapshot(target)
+
+    target.unlink()
+
+    pinned = _media_get(routes, monkeypatch, target, query_extra=f"&snap={digest}")
+    assert pinned.status == 200
+    assert bytes(pinned.body) == b"png-bytes-v1"
+
+
+def test_handle_media_invalid_snap_falls_back_to_live(routes, monkeypatch, snap_dir, tmp_path):
+    target = tmp_path / "pic.png"
+    target.write_bytes(b"live-bytes")
+    replayed = _media_get(routes, monkeypatch, target, query_extra="&snap=not-a-digest")
+    assert replayed.status == 200
+    assert bytes(replayed.body) == b"live-bytes"
+
+
+def test_handle_media_missing_snap_falls_back_to_live(routes, monkeypatch, snap_dir, tmp_path):
+    target = tmp_path / "pic.png"
+    target.write_bytes(b"live-bytes")
+    missing = _media_get(
+        routes, monkeypatch, target, query_extra="&snap=" + "0" * 64
+    )
+    assert missing.status == 200
+    assert bytes(missing.body) == b"live-bytes"
+
+
+def test_handle_media_snap_does_not_bypass_deny(routes, monkeypatch, snap_dir, tmp_path):
+    """snap= must never widen the path allow-list: a denied path stays denied
+    even when a valid snapshot digest is supplied."""
+    from api.media_snapshots import capture_snapshot
+
+    hermes_home = tmp_path / ".hermes"
+    hermes_home.mkdir()
+    secret = hermes_home / "settings.json"
+    secret.write_text("{}")
+    monkeypatch.setenv("HOME", str(tmp_path))
+    digest = capture_snapshot(secret)  # capture itself may be blocked; if not...
+
+    denied = _media_get(routes, monkeypatch, secret, query_extra=f"&snap={digest or '0'*64}")
+    # settings.json under a hermes root is denied by the #3234 deny list.
+    assert denied.status == 403
+
+
+def test_handle_media_denies_direct_store_path(routes, monkeypatch, tmp_path):
+    """The snapshot STORE directory itself is not a servable media path.
+
+    The store lives under STATE_DIR (a Hermes root) in production; the #3234
+    deny list must reject a bare path= fetch of a snapshot blob there, so the
+    store is only reachable through the validated snap= parameter.
+    """
+    from api.media_snapshots import capture_snapshot
+
+    # Simulate the production layout: the store lives under a Hermes root, with
+    # HOME pointing at tmp_path so that directory counts as a Hermes root. The
+    # #3234 deny list denies <hermes_root>/media_snapshots.
+    hermes_home = tmp_path / ".hermes"
+    store = hermes_home / "media_snapshots"
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("HERMES_WEBUI_MEDIA_SNAPSHOT_DIR", str(store))
+
+    target = tmp_path / "a.png"
+    target.write_bytes(b"bytes")
+    digest = capture_snapshot(target)
+    store_file = store / f"{digest}.snap"
+    assert store_file.exists()
+
+    denied = _media_get(routes, monkeypatch, store_file)
+    assert denied.status == 403
+
+
+def test_handle_media_snapshot_range_request(routes, monkeypatch, snap_dir, tmp_path):
+    from api.media_snapshots import capture_snapshot
+
+    target = tmp_path / "clip.mp4"
+    target.write_bytes(b"0123456789")
+    digest = capture_snapshot(target)
+
+    handler = _media_get(
+        routes,
+        monkeypatch,
+        target,
+        headers={"Range": "bytes=2-4"},
+        query_extra=f"&snap={digest}",
+    )
+    assert handler.status == 206
+    assert bytes(handler.body) == b"234"
+    assert handler.header("Content-Range") == "bytes 2-4/10"
+
+
+def test_handle_media_snapshot_download_name_uses_original(routes, monkeypatch, snap_dir, tmp_path):
+    from api.media_snapshots import capture_snapshot
+
+    target = tmp_path / "report.html"
+    target.write_bytes(b"<html>v1</html>")
+    digest = capture_snapshot(target)
+    target.unlink()
+
+    handler = _media_get(routes, monkeypatch, target, query_extra=f"&snap={digest}")
+    disposition = handler.header("Content-Disposition")
+    assert "report.html" in disposition
+    assert f"{digest}.snap" not in disposition
+
+
+# ── Round 2: capture/serve deny parity + source-path binding (#6979) ────────
+
+
+def test_media_capture_allowed_denies_default_webui_state_layout(tmp_path, monkeypatch):
+    """MUST-FIX 1 repro: default-layout <HERMES_HOME>/webui/sessions/victim.json
+    (== STATE_DIR/sessions) must NEVER be captured.
+
+    Round 1 capture omitted STATE_DIR from its deny roots; the serve path
+    denies it. Capture now shares the serve predicate, so this returns False.
+    """
+    from api.media_snapshots import media_capture_allowed
+
+    hermes_home = tmp_path / ".hermes"
+    state_dir = hermes_home / "webui"
+    (state_dir / "sessions").mkdir(parents=True)
+    victim = state_dir / "sessions" / "victim.json"
+    victim.write_text("{}")
+
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.setattr("api.config.STATE_DIR", str(state_dir))
+
+    assert media_capture_allowed(victim) is False
+
+
+def test_annotate_never_captures_denied_state_file(snap_dir, tmp_path, monkeypatch):
+    """MUST-FIX 1 end-to-end: annotating a message whose MEDIA: ref points at a
+    denied state file must capture NOTHING (Round 1 captured it and the digest
+    then acted as a bearer capability)."""
+    from api.media_snapshots import annotate_media_snapshots
+
+    hermes_home = tmp_path / ".hermes"
+    state_dir = hermes_home / "webui"
+    (state_dir / "sessions").mkdir(parents=True)
+    victim = state_dir / "sessions" / "victim.json"
+    victim.write_text('{"secret": 1}')
+
+    monkeypatch.setenv("HOME", str(tmp_path))
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.setattr("api.config.STATE_DIR", str(state_dir))
+
+    messages = [{"role": "assistant", "content": f"here it is: MEDIA:{victim}"}]
+    captured = annotate_media_snapshots(messages)
+    assert captured == 0
+    assert "_media_snapshots" not in messages[0]
+    assert len(list(snap_dir.glob("*.snap"))) == 0
+
+
+def test_handle_media_snap_requires_source_path_binding(routes, monkeypatch, snap_dir, tmp_path):
+    """MUST-FIX 1 serve-side repro: a digest captured from path A must not be
+    servable through path B, even when B is an allowed path.
+
+    Round 1 served stored bytes for ANY allowed path carrying the digest
+    (/tmp/whatever.png?snap=<digest> -> 200 with A's bytes). With the binding:
+    B falls back to its live file (404 when absent; live bytes when present).
+    """
+    from api.media_snapshots import capture_snapshot
+
+    source = tmp_path / "a.png"
+    source.write_bytes(b"frozen-bytes-from-a")
+    digest = capture_snapshot(source)
+
+    # Bound path: snapshot bytes served.
+    bound = _media_get(routes, monkeypatch, source, query_extra=f"&snap={digest}")
+    assert bound.status == 200
+    assert bytes(bound.body) == b"frozen-bytes-from-a"
+
+    # Unbound + absent path: NOT the snapshot; live fallback -> 404.
+    other_missing = tmp_path / "b.png"
+    unbound = _media_get(routes, monkeypatch, other_missing, query_extra=f"&snap={digest}")
+    assert unbound.status == 404
+
+    # Unbound + present path: live bytes, never the frozen snapshot.
+    other_live = tmp_path / "b.png"
+    other_live.write_bytes(b"live-bytes-of-b")
+    unbound_live = _media_get(routes, monkeypatch, other_live, query_extra=f"&snap={digest}")
+    assert unbound_live.status == 200
+    assert bytes(unbound_live.body) == b"live-bytes-of-b"
+
+
+def test_handle_media_snap_dedup_binds_every_source_path(routes, monkeypatch, snap_dir, tmp_path):
+    """Dedup must not break source binding: identical bytes captured from two
+    different paths share one digest, and BOTH paths may serve it (each was a
+    legitimate capture source); a third path may not."""
+    from api.media_snapshots import capture_snapshot
+
+    a = tmp_path / "a.png"
+    b = tmp_path / "b.png"
+    a.write_bytes(b"same-bytes")
+    b.write_bytes(b"same-bytes")
+    d1 = capture_snapshot(a)
+    d2 = capture_snapshot(b)
+    assert d1 == d2
+
+    sa = _media_get(routes, monkeypatch, a, query_extra=f"&snap={d1}")
+    sb = _media_get(routes, monkeypatch, b, query_extra=f"&snap={d2}")
+    assert sa.status == 200 and bytes(sa.body) == b"same-bytes"
+    assert sb.status == 200 and bytes(sb.body) == b"same-bytes"
+
+    c = tmp_path / "c.png"
+    c.write_bytes(b"live-different-bytes")
+    sc = _media_get(routes, monkeypatch, c, query_extra=f"&snap={d1}")
+    assert sc.status == 200
+    # Live file served (distinguishable from the frozen bytes), NOT the snapshot.
+    assert bytes(sc.body) == b"live-different-bytes"
+
+
+def test_handle_media_denies_custom_named_store_via_bare_path(routes, monkeypatch, tmp_path):
+    """MUST-FIX 2 repro: a custom-named snapshot store (HERMES_WEBUI_MEDIA_SNAPSHOT_DIR
+    pointing anywhere, e.g. /tmp/custom-store-name) must not be readable through
+    a bare path= request. Round 1 only deny-listed the literal name
+    'media_snapshots', so the blobs leaked with no snap= needed."""
+    from api.media_snapshots import capture_snapshot
+
+    store = tmp_path / "custom-store-name"
+    monkeypatch.setenv("HERMES_WEBUI_MEDIA_SNAPSHOT_DIR", str(store))
+
+    target = tmp_path / "a.png"
+    target.write_bytes(b"stored-bytes")
+    digest = capture_snapshot(target)
+    store_file = store / f"{digest}.snap"
+    assert store_file.exists()
+
+    denied = _media_get(routes, monkeypatch, store_file)
+    assert denied.status == 403
+
+
+def test_is_valid_digest_rejects_trailing_newline():
+    """SHOULD-FIX: `$` matched before a terminal newline; fullmatch must not."""
+    from api.media_snapshots import is_valid_digest
+
+    good = "a" * 64
+    assert is_valid_digest(good) is True
+    assert is_valid_digest(good + "\n") is False
+    assert is_valid_digest("a" * 63) is False
+    assert is_valid_digest("A" * 64) is False  # lowercase hex only
+    assert is_valid_digest("") is False
+    assert is_valid_digest(None) is False  # type: ignore[arg-type]
+
+
+def test_annotate_evicted_snapshot_is_final(snap_dir, tmp_path):
+    """SHOULD-FIX: a recorded digest is FINAL — after its blob is evicted, a
+    re-settle must not re-capture the CURRENT live bytes and rebind the
+    historical message (Round 1 did, silently following overwrites again)."""
+    from api.media_snapshots import annotate_media_snapshots
+
+    target = tmp_path / "report.html"
+    target.write_text("<html>v1</html>")
+    messages: list = [{"role": "assistant", "content": f"MEDIA:{target}"}]
+
+    assert annotate_media_snapshots(messages) == 1
+    snaps0: dict = messages[0].get("_media_snapshots") or {}
+    digest = snaps0[str(target)]
+
+    # Simulate quota eviction: blob gone.
+    blob = snap_dir / f"{digest}.snap"
+    assert blob.exists()
+    blob.unlink()
+
+    # Overwrite the live file, then re-settle the SAME message.
+    target.write_text("<html>v2 - overwritten</html>")
+    assert annotate_media_snapshots(messages) == 0
+    snaps1: dict = messages[0].get("_media_snapshots") or {}
+    assert snaps1[str(target)] == digest
+    assert not blob.exists()  # no re-capture
+
+
+def test_quota_eviction_drops_source_binding_sidecar(snap_dir, tmp_path, monkeypatch):
+    """Evicting a snapshot blob must also drop its source-binding sidecar."""
+    from api.media_snapshots import _binding_path_for_digest, capture_snapshot
+
+    monkeypatch.setenv("HERMES_WEBUI_MEDIA_SNAPSHOT_CAP_BYTES", "64")
+    old = tmp_path / "old.png"
+    new = tmp_path / "new.png"
+    old.write_bytes(b"x" * 64)
+    new.write_bytes(b"y" * 64)
+    d_old = capture_snapshot(old)
+    d_new = capture_snapshot(new)
+    assert d_old and d_new
+
+    if not _binding_path_for_digest(d_old).exists():
+        # Oldest was evicted (its binding sidecar must be gone too).
+        assert not (snap_dir / f"{d_old}.snap").exists()
+    else:
+        assert not _binding_path_for_digest(d_new).exists()
+        assert not (snap_dir / f"{d_new}.snap").exists()
+
+
+# ── display-metadata persistence (state.db ↔ sidecar merge) ────────────────
+
+
+def test_media_snapshots_registered_in_display_metadata_keys():
+    from api.models import _SESSION_MESSAGE_DISPLAY_METADATA_KEYS
+
+    assert "_media_snapshots" in _SESSION_MESSAGE_DISPLAY_METADATA_KEYS
+
+
+def test_merge_session_display_metadata_preserves_snapshots():
+    from api.models import _merge_session_display_metadata
+
+    target = {"role": "assistant", "content": "x"}
+    source = {"role": "assistant", "content": "x", "_media_snapshots": {"/a": "a" * 64}}
+    _merge_session_display_metadata(target, source)
+    assert target["_media_snapshots"] == {"/a": "a" * 64}
+
+
+# ── frontend stamping helper (behavioral, node-executed) ───────────────────
+
+
+def _extract_stamp_helper():
+    """Extract esc() and _stampMediaSnapshots() verbatim from static/ui.js."""
+    src = open(ROOT / "static" / "ui.js", encoding="utf-8").read()
+
+    def extract_function(name):
+        start = src.find(f"function {name}(")
+        if start < 0:
+            raise AssertionError(f"{name} not found in ui.js")
+        i = src.find("{", start)
+        depth = 1
+        i += 1
+        while i < len(src) and depth:
+            if src[i] == "{":
+                depth += 1
+            elif src[i] == "}":
+                depth -= 1
+            i += 1
+        return src[start:i]
+
+    esc_line = next(line for line in src.split("\n") if line.startswith("const esc="))
+    return esc_line, extract_function("_stampMediaSnapshots")
+
+
+def _run_stamp(html, snaps):
+    """Run the real _stampMediaSnapshots under node with mocked collaborators."""
+    import json
+    import subprocess
+    import tempfile
+
+    esc_def, fn_def = _extract_stamp_helper()
+    js_code = esc_def + "\n" + fn_def + "\n"
+    js_code += "var html=process.argv[2];\n"
+    js_code += "var snaps=JSON.parse(process.argv[3]);\n"
+    js_code += "process.stdout.write(_stampMediaSnapshots(html,snaps));\n"
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".js", delete=False, encoding="utf-8") as tf:
+        tf.write(js_code)
+        tfname = tf.name
+    try:
+        result = subprocess.run(
+            ["node", tfname, html, json.dumps(snaps)],
+            capture_output=True, text=True, timeout=30,
+        )
+        if result.returncode != 0:
+            raise RuntimeError(f"node error: {result.stderr}")
+        return result.stdout
+    finally:
+        os.unlink(tfname)
+
+
+def test_frontend_stamp_appends_snap_to_media_urls():
+    html = '<img class="msg-media-img" src="api/media?path=%2Fhome%2Fx%2Freport.html">'
+    snaps = {"/home/x/report.html": "a" * 64}
+    out = _run_stamp(html, snaps)
+    assert f"&snap={'a' * 64}" in out
+    assert "api/media?path=%2Fhome%2Fx%2Freport.html&snap=" in out
+
+
+def test_frontend_stamp_ignores_unrelated_paths():
+    html = '<img src="api/media?path=%2Fother%2Ffile.png">'
+    snaps = {"/home/x/report.html": "a" * 64}
+    out = _run_stamp(html, snaps)
+    assert "&snap=" not in out
+
+
+def test_frontend_stamp_ignores_invalid_digests():
+    html = '<img src="api/media?path=%2Fhome%2Fx%2Freport.html">'
+    snaps = {"/home/x/report.html": "not-a-digest"}
+    out = _run_stamp(html, snaps)
+    assert "&snap=" not in out
+
+
+def test_frontend_stamp_tags_lazy_preview_placeholders():
+    html = '<div class="html-preview-load" data-path="/home/x/report.html">…</div>'
+    snaps = {"/home/x/report.html": "b" * 64}
+    out = _run_stamp(html, snaps)
+    assert f'data-snap="{"b" * 64}"' in out
+
+
+def test_frontend_stamp_prefix_paths_do_not_corrupt():
+    """MUST-FIX 3 repro: /tmp/report.png is a PREFIX of /tmp/report.png.backup.
+    Round 1's substring split/join rewrote the longer URL as the shorter path +
+    the FIRST digest + '.backup' and dropped its own digest — the .backup
+    preview then pointed at the wrong bytes. Each path= value must be rewritten
+    atomically with its own digest."""
+    html = ('<img src="api/media?path=%2Ftmp%2Freport.png">'
+            '<img src="api/media?path=%2Ftmp%2Freport.png.backup">')
+    snaps = {"/tmp/report.png": "a" * 64, "/tmp/report.png.backup": "b" * 64}
+    out = _run_stamp(html, snaps)
+    assert f"path=%2Ftmp%2Freport.png&snap={'a' * 64}" in out
+    assert f"path=%2Ftmp%2Freport.png.backup&snap={'b' * 64}" in out
+    # The shorter URL must not have swallowed the longer one.
+    assert f"path=%2Ftmp%2Freport.png&snap={'a' * 64}.backup" not in out
+
+
+def test_frontend_stamp_prefix_data_paths_do_not_corrupt():
+    """Same prefix class for lazy-preview placeholders (data-path attributes)."""
+    html = ('<div class="diff-inline-load" data-path="/tmp/a.png">x</div>'
+            '<div class="diff-inline-load" data-path="/tmp/a.png.backup">y</div>')
+    snaps = {"/tmp/a.png": "c" * 64, "/tmp/a.png.backup": "d" * 64}
+    out = _run_stamp(html, snaps)
+    assert f'data-path="/tmp/a.png" data-snap="{"c" * 64}"' in out
+    assert f'data-path="/tmp/a.png.backup" data-snap="{"d" * 64}"' in out
+
+
+def test_frontend_stamp_handles_html_escaped_data_paths():
+    """data-path values are HTML-escaped (esc()): a path containing & must still
+    resolve exactly after unescaping, and must not get a bogus stamp."""
+    html = '<div class="html-preview-load" data-path="/tmp/a&amp;b.html">…</div>'
+    snaps = {"/tmp/a&b.html": "e" * 64}
+    out = _run_stamp(html, snaps)
+    assert f'data-snap="{"e" * 64}"' in out
+    # Non-matching escaped path is untouched.
+    html2 = '<div class="html-preview-load" data-path="/tmp/other&amp;x.html">…</div>'
+    out2 = _run_stamp(html2, snaps)
+    assert "data-snap=" not in out2
+
+
+def test_frontend_stamp_handles_file_url_forms():
+    # The backend indexes under BOTH the resolved path and the raw token; the
+    # helper must stamp either key form.
+    html = '<img src="api/media?path=%2Fhome%2Fx%2Freport.html">'
+    snaps = {"file:///home/x/report.html": "c" * 64}
+    out = _run_stamp(html, snaps)
+    assert "&snap=" not in out  # resolved form is the URL param; raw key alone
+    # must not wrongly stamp, but the resolved key MUST:
+    snaps2 = {"/home/x/report.html": "c" * 64}
+    out2 = _run_stamp(html, snaps2)
+    assert f"&snap={'c' * 64}" in out2
+
+
+def test_frontend_stamp_source_invariants():
+    """Non-vacuous source checks: the helper must be called at BOTH settled
+    render sites with the message's _media_snapshots, and lazy loaders must
+    consume data-snap."""
+    src = open(ROOT / "static" / "ui.js", encoding="utf-8").read()
+    # Main transcript path and transparent ordered segments must stamp.
+    assert "_stampMediaSnapshots(bodyHtml, m._media_snapshots)" in src
+    # Transparent segments: original _getCachedRender line is preserved (source
+    # window contract), stamping applied on the next line via *_Stamped.
+    assert "_getCachedRender(partDisplayText,false);" in src
+    assert "_stampMediaSnapshots(partBodyHtml,m._media_snapshots)" in src
+    # Worklog scene prose path is intentionally NOT stamped (folded view; the
+    # scene render chain is exercised by harness-extracted tests that would
+    # break on new helper references — snapshot support covers the main
+    # transcript + transparent segments, which are the comparison surface).
+    # Lazy loaders must read the stamped digest.
+    assert "_mediaSnapQuery(el)" in src
+    assert "el.dataset.snap" in src

tests/test_pdf_html_preview.py

@@ -173,7 +173,7 @@ class TestLoadHtmlInlineFunction:
     def test_fallback_on_error(self):
         ui = _read_js('ui.js')
         idx = ui.find('function loadHtmlInline')
-        body = ui[idx:idx + 2000]
+        body = ui[idx:idx + 2600]  # snapshot-stamped derived links (+snapQuery) grew the function
         assert 'html_error' in body, 'Must show error fallback on failure'
 
     def test_uses_srcdoc_attribute(self):
@@ -202,7 +202,11 @@ class TestLoadHtmlInlineFunction:
             "browser cache for stale HTML preview (got body without no-store)"
         )
         assert "const publicMediaUrl='api/media?path='+encodeURIComponent(path);" in body
-        assert "const openUrl=publicMediaUrl+'&inline=1';" in body
+        assert "const openUrl=publicMediaUrl+'&inline=1'+snapQuery;" in body, (
+            "HTML 'open full page' / fallback links must carry the message's "
+            "snapshot digest (snapQuery) so they open the frozen bytes, not the "
+            "current live file"
+        )
 
     def test_pdf_fetch_url_includes_session_id_for_session_media_artifacts(self):
         ui = _read_js('ui.js')
@@ -212,7 +216,11 @@ class TestLoadHtmlInlineFunction:
         assert "'&session_id='+encodeURIComponent(mediaSessionId)" in body
         assert 'fetch(mediaUrl)' in body
         assert "const publicMediaUrl='api/media?path='+encodeURIComponent(path);" in body
-        assert "const dlUrl=publicMediaUrl+'&download=1';" in body
+        assert "const dlUrl=publicMediaUrl+'&download=1'+snapQuery;" in body, (
+            "PDF download/fallback links must carry the message's snapshot "
+            "digest (snapQuery) so they download the frozen bytes, not the "
+            "current live file"
+        )
 
 
 # ── requestAnimationFrame integration ──────────────────────────────────────