Gateway MEDIA extraction false-positives and unsafe path handling

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #34833 fix(gateway): scope tool-result MEDIA scan to current turn (#34608)

  merged to main

  files: gateway/run.py, tests/gateway/test_media_extraction.py

  Show the diff

  • gateway/run.py

    @@ -17478,13 +17478,33 @@ def _approval_notify_sync(approval_data: dict) -> None:
                 # append any that aren't already present in the final response, so the
                 # adapter's extract_media() can find and deliver the files exactly once.
                 #
    -            # Uses path-based deduplication against _history_media_paths (collected
    -            # before run_conversation) instead of index slicing. This is safe even
    -            # when context compression shrinks the message list. (Fixes #160)
    +            # Scope the scan to THIS turn's tool results only. ``agent_history``
    +            # was passed into run_conversation as ``conversation_history``, so the
    +            # agent's returned ``messages`` list is ``agent_history`` followed by
    +            # the messages produced this turn. Slicing at ``len(agent_history)``
    +            # isolates the current turn precisely, so a stale MEDIA: path emitted
    +            # by a tool several turns earlier (still present in the full message
    +            # list) can never leak onto a later text-only reply. (Fixes #34608)
    +            #
    +            # Path-based deduplication against _history_media_paths (collected
    +            # before run_conversation) is retained as a secondary guard. It is
    +            # also the sole guard on the fallback branch taken when mid-run
    +            # context compression shrinks the message list below the original
    +            # history length, preserving the compression-safe behaviour of #160.
                 if "MEDIA:" not in final_response:
                     media_tags = []
                     has_voice_directive = False
    -                for msg in result.get("messages", []):
    +                _all_msgs = result.get("messages", [])
    +                _history_len = len(agent_history)
    +                # Only trust the slice boundary when the message list still
    +                # contains the full history prefix. Mid-run compression can
    +                # rewrite/shrink the list; in that case fall back to scanning
    +                # everything and rely on _history_media_paths for dedup.
    +                if _history_len and len(_all_msgs) >= _history_len:
    +                    _scan_msgs = _all_msgs[_history_len:]
    +                else:
    +                    _scan_msgs = _all_msgs
    +                for msg in _scan_msgs:
                         if msg.get("role") in {"tool", "function"}:
                             content = msg.get("content", "")
                             if "MEDIA:" in content:

  • tests/gateway/test_media_extraction.py

    @@ -5,6 +5,10 @@
     messages in the CURRENT turn, not from the full conversation history.
     This prevents voice messages from accumulating and being sent multiple
     times per reply. (Regression test for #160)
    +
    +Also covers #34608: a stale MEDIA: path emitted by an execute_code /
    +make_image tool several turns earlier must not leak onto a later
    +text-only reply, even when the path-based dedup set fails to capture it.
     """
     
     import pytest
    @@ -43,6 +47,37 @@ def extract_media_tags_fixed(result_messages, history_len):
         return media_tags, has_voice_directive
     
     
    +def extract_media_tags_production(result_messages, history_len, history_media_paths):
    +    """Mirror of the production scan in gateway/run.py after the #34608 fix.
    +
    +    Primary guard: scope the scan to the current turn via ``history_len``
    +    slicing (matching how ``agent_history`` is passed as
    +    ``conversation_history`` into ``run_conversation``). Secondary guard:
    +    path-based dedup against ``history_media_paths`` (the #160 compression-safe
    +    fallback, also used when compression shrinks the list below history_len).
    +    """
    +    media_tags = []
    +    has_voice_directive = False
    +
    +    if len(result_messages) >= history_len and history_len:
    +        scan_msgs = result_messages[history_len:]
    +    else:
    +        scan_msgs = result_messages
    +
    +    for msg in scan_msgs:
    +        if msg.get("role") == "tool" or msg.get("role") == "function":
    +            content = msg.get("content", "")
    +            if "MEDIA:" in content:
    +                for match in re.finditer(r'MEDIA:(\S+)', content):
    +                    path = match.group(1).strip().rstrip('",}')
    +                    if path and path not in history_media_paths:
    +                        media_tags.append(f"MEDIA:{path}")
    +                if "[[audio_as_voice]]" in content:
    +                    has_voice_directive = True
    +
    +    return media_tags, has_voice_directive
    +
    +
     def extract_media_tags_broken(result_messages):
         """
         The BROKEN behavior: extract MEDIA tags from ALL messages including history.
    @@ -180,5 +215,104 @@ def test_deduplication_within_current_turn(self):
             assert len(unique) == 2  # After dedup: same.ogg and different.ogg
     
     
    +class TestStaleToolMediaLeak:
    +    """Regression tests for #34608.
    +
    +    A MEDIA: path emitted by an execute_code / make_image tool several turns
    +    earlier remains in the full conversation message list. A later text-only
    +    reply (zero MEDIA directives) must NOT attach that stale image.
    +
    +    The production code previously relied solely on path-based dedup against
    +    paths reconstructed from the replayable transcript. When that
    +    reconstruction does not byte-match the in-memory tool content (timestamp
    +    stripping, observed-context withholding, compression rewrites), the stale
    +    path is absent from the dedup set and leaks. Turn-scoped slicing closes
    +    this class of bug deterministically.
    +    """
    +
    +    def test_stale_execute_code_media_not_attached_to_text_only_reply(self):
    +        """The exact #34608 scenario: make_image cover from an earlier turn."""
    +        # Prior turn generated an image via execute_code stdout.
    +        history = [
    +            {"role": "user", "content": "Make a cover image"},
    +            {"role": "assistant", "content": None,
    +             "tool_calls": [{"id": "1", "function": {"name": "execute_code"}}]},
    +            {"role": "tool", "tool_call_id": "1",
    +             "content": "Generating cover...\nMEDIA:/tmp/seosmi_cover.png\nDone."},
    +            {"role": "assistant", "content": "Here is your cover."},
    +        ]
    +        # Current turn: plain text status update, zero MEDIA directives.
    +        new_messages = [
    +            {"role": "user", "content": "What skill version am I on?"},
    +            {"role": "assistant", "content": "You're on v0.15.1."},
    +        ]
    +        all_messages = history + new_messages
    +        history_len = len(history)
    +
    +        # Simulate the dedup set FAILING to capture the stale path (the real
    +        # #34608 condition: replayable-history reconstruction diverged from
    +        # the in-memory tool content, so the path is not in the set).
    +        history_media_paths = set()
    +
    +        tags, voice = extract_media_tags_production(
    +            all_messages, history_len, history_media_paths
    +        )
    +        assert tags == [], (
    +            "Stale tool MEDIA from a prior turn must not leak onto a "
    +            f"later text-only reply, got {tags}"
    +        )
    +        assert voice is False
    +
    +        # The pre-fix production behaviour (scan everything, dedup only) would
    +        # have leaked the stale path when the dedup set missed it.
    +        broken_tags, _ = extract_media_tags_broken(all_messages)
    +        assert any("seosmi_cover.png" in t for t in broken_tags), (
    +            "Sanity: the unscoped scan does surface the stale path"
    +        )
    +
    +    def test_current_turn_media_still_attached_when_dedup_set_empty(self):
    +        """Turn-scoping must not suppress genuinely new media."""
    +        history = [
    +            {"role": "user", "content": "hi"},
    +            {"role": "assistant", "content": "hello"},
    +        ]
    +        new_messages = [
    +            {"role": "user", "content": "Make me a cover image"},
    +            {"role": "assistant", "content": None,
    +             "tool_calls": [{"id": "9", "function": {"name": "execute_code"}}]},
    +            {"role": "tool", "tool_call_id": "9",
    +             "content": "MEDIA:/tmp/fresh_cover.png"},
    +            {"role": "assistant", "content": "Here it is."},
    +        ]
    +        all_messages = history + new_messages
    +        tags, _ = extract_media_tags_production(
    +            all_messages, len(history), set()
    +        )
    +        assert len(tags) == 1 and "fresh_cover.png" in tags[0]
    +
    +    def test_compression_shrink_falls_back_to_path_dedup(self):
    +        """When the list is shorter than history_len (mid-run compression),
    +        fall back to scanning everything with path-based dedup so the #160
    +        compression-safe guarantee is preserved."""
    +        # Post-compression list is shorter than the original history length.
    +        compressed_messages = [
    +            {"role": "user", "content": "summary so far..."},
    +            {"role": "tool", "tool_call_id": "7",
    +             "content": "MEDIA:/tmp/old_from_history.png"},
    +            {"role": "assistant", "content": "ok"},
    +        ]
    +        original_history_len = 12  # larger than the compressed list
    +        # The old path IS captured in the dedup set here (history scan ran
    +        # before compression), so it must still be excluded.
    +        history_media_paths = {"/tmp/old_from_history.png"}
    +        tags, _ = extract_media_tags_production(
    +            compressed_messages, original_history_len, history_media_paths
    +        )
    +        assert tags == [], (
    +            "On the compression fallback path, path-dedup must still exclude "
    +            f"known-old media, got {tags}"
    +        )
    +
    +
     if __name__ == "__main__":
         pytest.main([__file__, "-v"])

• #34844 fix(gateway): unify MEDIA: extraction extension set + close the unknown-ext black hole (#34517)

  merged to main

  files: gateway/platforms/base.py, gateway/run.py, gateway/stream_consumer.py, scripts/release.py, tests/gateway/test_platform_base.py

  Show the diff

  • gateway/platforms/base.py

    @@ -1131,6 +1131,75 @@ def _log_safe_path(path: str) -> str:
     }
     
     
    +# ---------------------------------------------------------------------------
    +# Media-delivery extension allowlist — SINGLE SOURCE OF TRUTH
    +#
    +# Both extractors that turn response text into native attachments derive their
    +# extension set from this tuple:
    +#   * ``extract_media()``       — explicit ``MEDIA:<path>`` tags
    +#   * ``extract_local_files()`` — bare absolute/home paths the agent mentions
    +#
    +# Historically these two carried independently-maintained extension lists.
    +# ``extract_media`` had a narrow list (no .md/.json/.yaml/.xml/.html/...) while
    +# ``extract_local_files`` had a broad one. Combined with the unconditional
    +# ``MEDIA:\\s*\\S+`` cleanup at the dispatch sites, that mismatch created a
    +# silent black hole: a ``MEDIA:/report.md`` tag failed the narrow extract_media
    +# match, got stripped from the body by the loose cleanup regex, and was then
    +# invisible to extract_local_files — the file was never delivered (issue
    +# #34517). Keeping one list eliminates the drift; building the cleanup regexes
    +# from the same set means a tag is only stripped when its extension is one we
    +# can actually deliver, so an unknown-extension path survives in the body
    +# instead of vanishing.
    +#
    +# Covers images (inline), video (inline where supported), audio (voice/audio),
    +# documents/spreadsheets/presentations (send_document), archives, and rendered
    +# web output. The dispatch partition (image vs video vs document) lives in
    +# ``gateway/run.py``.
    +# ---------------------------------------------------------------------------
    +
    +MEDIA_DELIVERY_EXTS: Tuple[str, ...] = (
    +    # Images (embed inline)
    +    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".svg",
    +    # Video (embed inline where supported)
    +    ".mp4", ".mov", ".avi", ".mkv", ".webm",
    +    # Audio (delivered as voice/audio where supported)
    +    ".mp3", ".wav", ".ogg", ".opus", ".m4a", ".flac",
    +    # Documents (uploaded as file attachments)
    +    ".pdf", ".docx", ".doc", ".odt", ".rtf", ".txt", ".md", ".epub",
    +    # Spreadsheets / data
    +    ".xlsx", ".xls", ".ods", ".csv", ".tsv", ".json", ".xml", ".yaml", ".yml",
    +    # Presentations
    +    ".pptx", ".ppt", ".odp", ".key",
    +    # Archives
    +    ".zip", ".tar", ".gz", ".tgz", ".bz2", ".xz", ".7z", ".rar", ".apk", ".ipa",
    +    # Web / rendered output
    +    ".html", ".htm",
    +)
    +
    +# Regex alternation fragment of bare extensions (no leading dot), e.g.
    +# ``png|jpe?g|...``. ``jpe?g`` collapses jpg/jpeg into one branch. Sorted
    +# longest-first so the alternation never matches a shorter ext as a prefix of
    +# a longer one (e.g. ``.tar`` before ``.tar.gz`` components).
    +_MEDIA_EXT_ALTERNATION = "|".join(
    +    sorted((e.lstrip(".") for e in MEDIA_DELIVERY_EXTS), key=len, reverse=True)
    +)
    +
    +# Anchored ``MEDIA:<path>`` cleanup pattern. Unlike the old loose
    +# ``MEDIA:\\s*\\S+``, this only strips a tag whose path ends in a known
    +# deliverable extension (optionally quoted/backticked). A ``MEDIA:`` tag with
    +# an unknown extension is left in the text so it can still be picked up by the
    +# bare-path detector (extract_local_files) downstream rather than silently
    +# deleted. Shared by the non-streaming dispatch path and the streaming
    +# consumer so both behave identically.
    +MEDIA_TAG_CLEANUP_RE = re.compile(
    +    r'''[`"']?MEDIA:\s*'''
    +    r'''(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|'''
    +    r'''(?:~/|/)\S+(?:[^\S\n]+\S+)*?\.(?:''' + _MEDIA_EXT_ALTERNATION + r'''))'''
    +    r'''(?=[\s`"',;:)\]}]|$)[`"']?''',
    +    re.IGNORECASE,
    +)
    +
    +
     def get_document_cache_dir() -> Path:
         """Return the document cache directory, creating it if it doesn't exist."""
         DOCUMENT_CACHE_DIR.mkdir(parents=True, exist_ok=True)
    @@ -2542,10 +2611,10 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
             cleaned = cleaned.replace("[[as_document]]", "")
             
             # Extract MEDIA:<path> tags, allowing optional whitespace after the colon
    -        # and quoted/backticked paths for LLM-formatted outputs.
    -        media_pattern = re.compile(
    -            r'''[`"']?MEDIA:\s*(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|(?:~/|/)\S+(?:[^\S\n]+\S+)*?\.(?:png|jpe?g|gif|webp|mp4|mov|avi|mkv|webm|ogg|opus|mp3|wav|m4a|flac|epub|pdf|zip|rar|7z|docx?|xlsx?|pptx?|txt|csv|apk|ipa)(?=[\s`"',;:)\]}]|$))[`"']?'''
    -        )
    +        # and quoted/backticked paths for LLM-formatted outputs. The extension
    +        # set is the shared MEDIA_DELIVERY_EXTS source of truth (built once into
    +        # MEDIA_TAG_CLEANUP_RE) so it can never drift from extract_local_files.
    +        media_pattern = MEDIA_TAG_CLEANUP_RE
             for match in media_pattern.finditer(content):
                 path = match.group("path").strip()
                 if len(path) >= 2 and path[0] == path[-1] and path[0] in "`\"'":
    @@ -2591,24 +2660,7 @@ def extract_local_files(content: str) -> Tuple[List[str], str]:
                 Tuple of (list of expanded file paths, cleaned text with the
                 raw path strings removed).
             """
    -        _LOCAL_MEDIA_EXTS = (
    -            # Images (embed inline)
    -            '.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp', '.tiff', '.svg',
    -            # Video (embed inline where supported)
    -            '.mp4', '.mov', '.avi', '.mkv', '.webm',
    -            # Audio (delivered as voice/audio where supported)
    -            '.mp3', '.wav', '.ogg', '.m4a', '.flac',
    -            # Documents (uploaded as file attachments)
    -            '.pdf', '.docx', '.doc', '.odt', '.rtf', '.txt', '.md',
    -            # Spreadsheets / data
    -            '.xlsx', '.xls', '.ods', '.csv', '.tsv', '.json', '.xml', '.yaml', '.yml',
    -            # Presentations
    -            '.pptx', '.ppt', '.odp', '.key',
    -            # Archives
    -            '.zip', '.tar', '.gz', '.tgz', '.bz2', '.xz', '.7z', '.rar',
    -            # Web / rendered output
    -            '.html', '.htm',
    -        )
    +        _LOCAL_MEDIA_EXTS = MEDIA_DELIVERY_EXTS
             ext_part = '|'.join(e.lstrip('.') for e in _LOCAL_MEDIA_EXTS)
     
             # (?<![/:\w.]) prevents matching inside URLs (e.g. https://…/img.png)
    @@ -3729,7 +3781,12 @@ async def _stop_typing_task() -> None:
                     # Strip any remaining internal directives from message body (fixes #1561)
                     text_content = text_content.replace("[[audio_as_voice]]", "").strip()
                     text_content = text_content.replace("[[as_document]]", "").strip()
    -                text_content = re.sub(r"MEDIA:\s*\S+", "", text_content).strip()
    +                # Strip only MEDIA: tags whose path has a deliverable extension
    +                # (shared MEDIA_TAG_CLEANUP_RE). A MEDIA: tag with an unknown
    +                # extension is intentionally left in the body so extract_local_files
    +                # below can still pick up the bare path — otherwise the file would
    +                # be silently dropped (issue #34517).
    +                text_content = MEDIA_TAG_CLEANUP_RE.sub("", text_content).strip()
                     if images:
                         logger.info("[%s] extract_images found %d image(s) in response (%d chars)", self.name, len(images), len(response))
     

  • gateway/run.py

    @@ -11743,9 +11743,16 @@ async def _deliver_media_from_response(
     
                 from gateway.platforms.base import BasePlatformAdapter, should_send_media_as_audio
     
    -            media_files, _ = adapter.extract_media(response)
    +            media_files, cleaned = adapter.extract_media(response)
                 media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files)
    -            _, cleaned = adapter.extract_images(response)
    +            # Chain the cleaned text through each extractor (extract_media →
    +            # extract_images → extract_local_files) so MEDIA: tags and image URLs
    +            # are removed before the bare-path auto-detect runs. Previously the
    +            # cleaned text from extract_media was dropped (``_``) and
    +            # extract_local_files scanned text that still contained MEDIA: tags,
    +            # producing false-positive bare-path matches with the MEDIA: prefix
    +            # glued on. This matches the chain order in gateway/platforms/base.py.
    +            _, cleaned = adapter.extract_images(cleaned)
                 local_files, _ = adapter.extract_local_files(cleaned)
                 local_files = BasePlatformAdapter.filter_local_delivery_paths(local_files)
     

  • gateway/stream_consumer.py

    @@ -26,6 +26,7 @@
     
     from gateway.platforms.base import BasePlatformAdapter as _BasePlatformAdapter
     from gateway.platforms.base import _custom_unit_to_cp
    +from gateway.platforms.base import MEDIA_TAG_CLEANUP_RE
     from gateway.config import (
         DEFAULT_STREAMING_EDIT_INTERVAL as _DEFAULT_STREAMING_EDIT_INTERVAL,
         DEFAULT_STREAMING_BUFFER_THRESHOLD as _DEFAULT_STREAMING_BUFFER_THRESHOLD,
    @@ -645,10 +646,13 @@ async def run(self) -> None:
             except Exception as e:
                 logger.error("Stream consumer error: %s", e)
     
    -    # Pattern to strip MEDIA:<path> tags (including optional surrounding quotes).
    -    # Matches the simple cleanup regex used by the non-streaming path in
    -    # gateway/platforms/base.py for post-processing.
    -    _MEDIA_RE = re.compile(r'''[`"']?MEDIA:\s*\S+[`"']?''')
    +    # Strip MEDIA:<path> tags before display. Uses the shared anchored
    +    # MEDIA_TAG_CLEANUP_RE from gateway/platforms/base.py — only tags whose
    +    # path ends in a deliverable extension are removed, so an unknown-extension
    +    # path stays visible instead of being silently dropped (issue #34517).
    +    # Streaming and non-streaming paths share the same regex, so a tag is
    +    # treated identically whichever path delivered the text.
    +    _MEDIA_RE = MEDIA_TAG_CLEANUP_RE
     
         @staticmethod
         def _clean_for_display(text: str) -> str:

  • scripts/release.py

    @@ -523,6 +523,8 @@
         "[redacted:email]": "JezzaHehn",
         "[redacted:email]": "dodo-reach",
         "[redacted:email]": "Bartok9",
    +    "[redacted:email]": "banditburai",
    +    "[redacted:email]": "Kyzcreig",
         "[redacted:email]": "crayfish-ai",
         "[redacted:email]": "MestreY0d4-Uninter",
         "[redacted:email]": "Roy-oss1",

  • tests/gateway/test_platform_base.py

    @@ -362,6 +362,54 @@ def test_both_directives_can_coexist(self):
             assert "[[as_document]]" not in cleaned
     
     
    +class TestMediaExtensionAllowlistParity:
    +    """Regression coverage for issue #34517 — the MEDIA: extension black hole.
    +
    +    extract_media used to carry a narrow extension allowlist that omitted
    +    .md/.json/.yaml/.xml/.html etc., while extract_local_files had a broad one.
    +    Combined with an unconditional ``MEDIA:\\s*\\S+`` strip at the dispatch
    +    sites, an unmatched MEDIA: tag for one of those extensions was deleted from
    +    the body before extract_local_files could pick up the bare path — the file
    +    was silently dropped. Both extractors now derive from the single
    +    MEDIA_DELIVERY_EXTS source of truth, and the strip is anchored to that set.
    +    """
    +
    +    DROPPED_BEFORE = ["md", "json", "yaml", "yml", "xml", "html", "htm",
    +                      "tsv", "svg"]
    +
    +    def test_previously_dropped_extensions_now_extract(self):
    +        for ext in self.DROPPED_BEFORE:
    +            path = f"/tmp/report.{ext}"
    +            media, _ = BasePlatformAdapter.extract_media(f"Here: MEDIA:{path}")
    +            assert media == [(path, False)], f".{ext} should extract via MEDIA:"
    +
    +    def test_extract_media_and_local_files_share_one_extension_set(self):
    +        from gateway.platforms.base import MEDIA_DELIVERY_EXTS
    +        # Both functions reference MEDIA_DELIVERY_EXTS; assert the documents
    +        # that motivated the bug are present in the shared set.
    +        for ext in (".md", ".json", ".yaml", ".yml", ".xml", ".html", ".htm"):
    +            assert ext in MEDIA_DELIVERY_EXTS
    +
    +    def test_unknown_extension_not_black_holed_by_cleanup(self):
    +        """A MEDIA: tag with an unknown extension is NOT stripped from the
    +        body — it survives so extract_local_files can still see the bare path,
    +        rather than vanishing entirely (the core of issue #34517)."""
    +        from gateway.platforms.base import MEDIA_TAG_CLEANUP_RE
    +        text = "Saved to MEDIA:/tmp/data.weirdext done"
    +        media, _ = BasePlatformAdapter.extract_media(text)
    +        assert media == []  # unknown extension is not a deliverable MEDIA tag
    +        stripped = MEDIA_TAG_CLEANUP_RE.sub("", text)
    +        assert "/tmp/data.weirdext" in stripped  # path preserved, not dropped
    +
    +    def test_known_extension_tag_is_stripped_from_body(self):
    +        from gateway.platforms.base import MEDIA_TAG_CLEANUP_RE
    +        text = "Here is your report: MEDIA:/tmp/report.md"
    +        stripped = MEDIA_TAG_CLEANUP_RE.sub("", text).strip()
    +        assert "MEDIA:" not in stripped
    +        assert "/tmp/report.md" not in stripped
    +        assert "Here is your report:" in stripped
    +
    +
     class TestMediaDeliveryPathValidation:
         def _patch_roots(self, monkeypatch, *roots):
             monkeypatch.setattr(

• #36275 fix(gateway): salvage MEDIA: false-positive guards (#16721 + #35699)

  merged to main

  files: gateway/platforms/base.py, gateway/run.py, scripts/release.py, tests/gateway/test_media_extraction.py, tests/gateway/test_platform_base.py

  Show the diff

  • gateway/platforms/base.py

    @@ -2579,6 +2579,48 @@ def filter_local_delivery_paths(file_paths) -> List[str]:
                     logger.warning("Skipping unsafe local file path: %s", _log_safe_path(raw))
             return safe_paths
     
    +
    +    @staticmethod
    +    def _mask_protected_spans(content: str) -> str:
    +        """Replace content inside fenced code blocks, inline code spans,
    +        and blockquotes with spaces to prevent MEDIA: false positives.
    +
    +        Preserves character count so regex match offsets stay valid.
    +        Skips masking backtick-quoted paths in MEDIA: tags (e.g.
    +        ``MEDIA:`/path/to/file.png` ``) to avoid breaking path extraction.
    +        """
    +        chars = list(content)
    +        n = len(chars)
    +
    +        # Build list of (start, end) spans to mask
    +        spans: list = []
    +
    +        # Fenced code blocks: ```...```
    +        for m in re.finditer(r'```[^\n]*\n.*?```', content, re.DOTALL):
    +            spans.append((m.start(), m.end()))
    +
    +        # Inline code: `...` but NOT backtick-quoted paths in MEDIA: tags
    +        for m in re.finditer(r'`[^`\n]+`', content):
    +            start = m.start()
    +            # Check if this is a backtick-quoted path after MEDIA:
    +            prefix = content[max(0, start - 20):start]
    +            if re.search(r'MEDIA:\s*$', prefix):
    +                continue  # This is a MEDIA path quote, not inline code
    +            spans.append((start, m.end()))
    +
    +        # Blockquote lines: > at line start
    +        for m in re.finditer(r'^>.*$', content, re.MULTILINE):
    +            spans.append((m.start(), m.end()))
    +
    +        # Apply masking
    +        for start, end in spans:
    +            for i in range(start, end):
    +                if chars[i] != '\n':
    +                    chars[i] = ' '
    +
    +        return ''.join(chars)
    +
    +
         @staticmethod
         def _mask_json_string_media(content: str) -> str:
             """Blank out ``MEDIA:<bare-path>`` occurrences that sit inside a JSON
    @@ -2661,9 +2703,14 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
             # set is the shared MEDIA_DELIVERY_EXTS source of truth (built once into
             # MEDIA_TAG_CLEANUP_RE) so it can never drift from extract_local_files.
             media_pattern = MEDIA_TAG_CLEANUP_RE
    -        # Mask MEDIA: embedded inside serialized JSON string values so stale
    -        # paths from stored tool results are never re-delivered (#34375).
    -        scan_content = BasePlatformAdapter._mask_json_string_media(content)
    +        # Mask example/stored MEDIA: paths before scanning so they are never
    +        # delivered as real attachments:
    +        #  - code blocks / inline code / blockquotes hold prose examples (#35695)
    +        #  - serialized JSON string values hold stored tool-result text (#34375)
    +        # Both maskers are offset-preserving (chars -> spaces) so match offsets
    +        # stay valid; chaining them masks the union of both protected regions.
    +        scan_content = BasePlatformAdapter._mask_protected_spans(content)
    +        scan_content = BasePlatformAdapter._mask_json_string_media(scan_content)
             for match in media_pattern.finditer(scan_content):
                 path = match.group("path").strip()
                 if len(path) >= 2 and path[0] == path[-1] and path[0] in "`\"'":
    @@ -2677,15 +2724,17 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
                         # and dropping every other attachment in the response.
                         continue
     
    -        # Remove the delivered MEDIA tags from the user-visible text. Mask
    -        # ``cleaned`` (same length, so offsets line up with it), find the real
    -        # tag spans there, and delete those spans from the *unmasked* ``cleaned``.
    -        # This strips real tags while leaving JSON-embedded MEDIA: text intact —
    -        # it is stored data, not a directive, and must read back verbatim
    -        # (#34375). Masking ``cleaned`` (not ``content``) keeps offsets valid
    -        # after the [[audio_as_voice]] / [[as_document]] directives are removed.
    +        # Remove the delivered MEDIA tags from the user-visible text. Mask a
    +        # length-equal copy of ``cleaned`` (same union of protected regions) to
    +        # *locate* the real tag spans, then delete exactly those spans from the
    +        # *unmasked* ``cleaned``. Masking is only a locator — protected spans
    +        # (code blocks, quotes, JSON-embedded MEDIA: text) must survive verbatim
    +        # in the delivered text, not be blanked to whitespace. Masking
    +        # ``cleaned`` (not ``content``) keeps offsets valid after the
    +        # [[audio_as_voice]] / [[as_document]] directives are removed.
             if media:
    -            masked_cleaned = BasePlatformAdapter._mask_json_string_media(cleaned)
    +            masked_cleaned = BasePlatformAdapter._mask_protected_spans(cleaned)
    +            masked_cleaned = BasePlatformAdapter._mask_json_string_media(masked_cleaned)
                 spans = [m.span() for m in media_pattern.finditer(masked_cleaned)]
                 if spans:
                     chars = list(cleaned)

  • gateway/run.py

    @@ -669,6 +669,88 @@ def _last_transcript_timestamp(history: Optional[List[Dict[str, Any]]]) -> Any:
         return None
     
     
    +# Tool results can contain literal MEDIA: examples in docs, logs, or other
    +# ordinary outputs. Only tools that intentionally create deliverable media
    +# artifacts should be eligible for automatic append when the model omits them
    +# from the final gateway reply.
    +_AUTO_APPEND_MEDIA_TOOL_NAMES = {"text_to_speech", "text_to_speech_tool"}
    +
    +
    +# Extension-anchored MEDIA: matcher for tool results. Mirrors the dispatch-site
    +# pattern so a bare ``MEDIA:`` token in prose (no deliverable extension) is never
    +# auto-appended. Kept local to the auto-append path; the producer-tool allowlist
    +# below is the primary guard, this is the secondary precision guard.
    +_TOOL_MEDIA_RE = re.compile(
    +    r'MEDIA:((?:[A-Za-z]:[/\\]|/|~\/)\S+\.(?:png|jpe?g|gif|webp|'
    +    r'mp4|mov|avi|mkv|webm|ogg|opus|mp3|wav|m4a|'
    +    r'flac|epub|pdf|zip|rar|7z|docx?|xlsx?|pptx?|'
    +    r'txt|csv|apk|ipa))',
    +    re.IGNORECASE,
    +)
    +
    +
    +def _collect_auto_append_media_tags(
    +    messages: List[Dict[str, Any]],
    +    history_offset: int = 0,
    +    history_media_paths: Optional[set] = None,
    +) -> tuple[List[str], bool]:
    +    """Collect real media tags from current-turn producer-tool results only.
    +
    +    Two layered guards keep stale/example MEDIA: strings out of the reply:
    +
    +    1. Producer-tool allowlist: only tools that intentionally emit deliverable
    +       artifacts (TTS) are eligible. Documentation, logs, and search results can
    +       contain example strings such as MEDIA:/absolute/path/to/file, which must
    +       never be delivered as attachments. (Fixes the original report behind #16721.)
    +    2. Current-turn isolation: only messages produced this turn are scanned, so a
    +       tool result from an earlier turn (still present in the full message list)
    +       cannot leak onto a later text-only reply (#34608).
    +
    +    Mid-run context compression can rewrite/shrink the message list below the
    +    original history length. When that happens the slice boundary is no longer
    +    trustworthy, so fall back to scanning every message and rely on
    +    ``history_media_paths`` for dedup, preserving the compression-safe behaviour
    +    of #160. The producer-tool allowlist still applies on the fallback path.
    +    """
    +    history_media_paths = history_media_paths or set()
    +    # Only trust the slice boundary when the message list still contains the
    +    # full history prefix. Otherwise scan everything (compression-safe fallback).
    +    if history_offset and len(messages) >= history_offset:
    +        new_messages = messages[history_offset:]
    +    else:
    +        new_messages = messages
    +
    +    tool_name_by_call_id: Dict[str, str] = {}
    +    for msg in new_messages:
    +        if msg.get("role") != "assistant":
    +            continue
    +        for call in msg.get("tool_calls") or []:
    +            call_id = call.get("id") or call.get("call_id")
    +            fn = call.get("function") or {}
    +            name = str(fn.get("name") or call.get("name") or "")
    +            if call_id and name:
    +                tool_name_by_call_id[str(call_id)] = name
    +
    +    media_tags: List[str] = []
    +    has_voice_directive = False
    +    for msg in new_messages:
    +        if msg.get("role") not in ("tool", "function"):
    +            continue
    +        call_id = str(msg.get("tool_call_id") or msg.get("call_id") or "")
    +        if tool_name_by_call_id.get(call_id) not in _AUTO_APPEND_MEDIA_TOOL_NAMES:
    +            continue
    +        content = str(msg.get("content") or "")
    +        if "MEDIA:" not in content:
    +            continue
    +        for match in _TOOL_MEDIA_RE.finditer(content):
    +            path = match.group(1).strip().rstrip('\",}')
    +            if path and path not in history_media_paths:
    +                media_tags.append(f"MEDIA:{path}")
    +        if "[[audio_as_voice]]" in content:
    +            has_voice_directive = True
    +
    +    return media_tags, has_voice_directive
    +
     # ---------------------------------------------------------------------------
     # SSL certificate auto-detection for NixOS and other non-standard systems.
     # Must run BEFORE any HTTP library (discord, aiohttp, etc.) is imported.
    @@ -17931,36 +18013,12 @@ def _approval_notify_sync(approval_data: dict) -> None:
                 # context compression shrinks the message list below the original
                 # history length, preserving the compression-safe behaviour of #160.
                 if "MEDIA:" not in final_response:
    -                media_tags = []
    -                has_voice_directive = False
    -                _all_msgs = result.get("messages", [])
    -                _history_len = len(agent_history)
    -                # Only trust the slice boundary when the message list still
    -                # contains the full history prefix. Mid-run compression can
    -                # rewrite/shrink the list; in that case fall back to scanning
    -                # everything and rely on _history_media_paths for dedup.
    -                if _history_len and len(_all_msgs) >= _history_len:
    -                    _scan_msgs = _all_msgs[_history_len:]
    -                else:
    -                    _scan_msgs = _all_msgs
    -                for msg in _scan_msgs:
    -                    if msg.get("role") in {"tool", "function"}:
    -                        content = msg.get("content", "")
    -                        if "MEDIA:" in content:
    -                            _TOOL_MEDIA_RE = re.compile(
    -                                r'MEDIA:((?:[A-Za-z]:[/\\]|/|~\/)\S+\.(?:png|jpe?g|gif|webp|'
    -                                r'mp4|mov|avi|mkv|webm|ogg|opus|mp3|wav|m4a|'
    -                                r'flac|epub|pdf|zip|rar|7z|docx?|xlsx?|pptx?|'
    -                                r'txt|csv|apk|ipa))',
    -                                re.IGNORECASE
    -                            )
    -                            for match in _TOOL_MEDIA_RE.finditer(content):
    -                                path = match.group(1).strip().rstrip('",}')
    -                                if path and path not in _history_media_paths:
    -                                    media_tags.append(f"MEDIA:{path}")
    -                            if "[[audio_as_voice]]" in content:
    -                                has_voice_directive = True
    -                
    +                media_tags, has_voice_directive = _collect_auto_append_media_tags(
    +                    result.get("messages", []),
    +                    history_offset=len(agent_history),
    +                    history_media_paths=_history_media_paths,
    +                )
    +
                     if media_tags:
                         seen = set()
                         unique_tags = []

  • scripts/release.py

    @@ -1032,6 +1032,7 @@
         "[redacted:email]": "holynn-q",
         "[redacted:email]": "jacdevos",
         "[redacted:email]": "liuhao1024",
    +    "[redacted:email]": "VinciZhu",
         "[redacted:email]": "rodrigoeqnit",
         "[redacted:email]": "sgtworkman",
         "[redacted:email]": "keepcalmqqf",

  • tests/gateway/test_media_extraction.py

    @@ -102,6 +102,63 @@ def extract_media_tags_broken(result_messages):
     
     class TestMediaExtraction:
         """Tests for MEDIA tag extraction from tool results."""
    +
    +    def test_gateway_auto_append_ignores_media_examples_in_skill_docs(self):
    +        """Skill/documentation examples must not be appended as real attachments."""
    +        from gateway.run import _collect_auto_append_media_tags
    +
    +        messages = [
    +            {"role": "user", "content": "How should I format gateway media?"},
    +            {
    +                "role": "assistant",
    +                "tool_calls": [
    +                    {"id": "call_skill", "function": {"name": "skill_view"}}
    +                ],
    +            },
    +            {
    +                "role": "tool",
    +                "tool_call_id": "call_skill",
    +                "content": """
    +Recommended pattern:
    +```text
    +MEDIA:/absolute/path/to/image.png
    +```
    +Second message:
    +```text
    +caption
    +```
    +""",
    +            },
    +            {"role": "assistant", "content": "Use a standalone media message."},
    +        ]
    +
    +        tags, voice = _collect_auto_append_media_tags(messages, history_offset=0)
    +        assert tags == []
    +        assert voice is False
    +
    +    def test_gateway_auto_append_keeps_real_tts_media_tag(self):
    +        """TTS tool media tags are still auto-appended when the model omits them."""
    +        from gateway.run import _collect_auto_append_media_tags
    +
    +        messages = [
    +            {"role": "user", "content": "Say this as audio"},
    +            {
    +                "role": "assistant",
    +                "tool_calls": [
    +                    {"id": "call_tts", "function": {"name": "text_to_speech"}}
    +                ],
    +            },
    +            {
    +                "role": "tool",
    +                "tool_call_id": "call_tts",
    +                "content": '{"success": true, "media_tag": "[[audio_as_voice]]\\nMEDIA:/tmp/voice.ogg"}',
    +            },
    +            {"role": "assistant", "content": "Done."},
    +        ]
    +
    +        tags, voice = _collect_auto_append_media_tags(messages, history_offset=0)
    +        assert tags == ["MEDIA:/tmp/voice.ogg"]
    +        assert voice is True
         
         def test_media_tags_not_extracted_from_history(self):
             """MEDIA tags from previous turns should NOT be extracted again."""

  • tests/gateway/test_platform_base.py

    @@ -400,6 +400,64 @@ def test_relative_path_still_ignored(self):
             )
             assert media == []
     
    +    # --- Code block / inline code / blockquote false-positive guards (#35695) ---
    +
    +    def test_media_in_fenced_code_block_ignored(self):
    +        """MEDIA: inside ``` fenced code blocks must not be extracted."""
    +        content = "Here is an example:\n```text\nMEDIA:/path/to/example.png\n```\nDone."
    +        media, cleaned = BasePlatformAdapter.extract_media(content)
    +        assert media == []
    +        assert "example" in cleaned.lower()
    +
    +    def test_media_in_inline_code_ignored(self):
    +        """MEDIA: inside backtick inline code must not be extracted."""
    +        content = "Use `MEDIA:/path/to/file.png` in your response."
    +        media, cleaned = BasePlatformAdapter.extract_media(content)
    +        assert media == []
    +        assert "MEDIA:" in cleaned  # preserved as text
    +
    +    def test_media_in_blockquote_ignored(self):
    +        """MEDIA: inside a > blockquote must not be extracted."""
    +        content = "> To send an image, include MEDIA:/path/to/image.jpg\nEnd."
    +        media, cleaned = BasePlatformAdapter.extract_media(content)
    +        assert media == []
    +        assert "End." in cleaned
    +
    +    def test_media_outside_code_blocks_still_extracted(self):
    +        """Real MEDIA: tags outside protected regions must still work."""
    +        content = "MEDIA:/real/file.png\n```code\nMEDIA:/fake/file.png\n```"
    +        media, _ = BasePlatformAdapter.extract_media(content)
    +        assert len(media) == 1
    +        assert media[0][0] == "/real/file.png"
    +
    +    def test_media_mixed_code_and_prose(self):
    +        """Real MEDIA: in prose + example in code block: only prose extracted,
    +        and the code block survives verbatim in the delivered text."""
    +        content = (
    +            "Here is your file:\n"
    +            "MEDIA:/output/report.pdf\n"
    +            "Example usage:\n"
    +            "```text\nMEDIA:/example/path.pdf\n```\n"
    +            "Done."
    +        )
    +        media, cleaned = BasePlatformAdapter.extract_media(content)
    +        assert len(media) == 1
    +        assert media[0][0] == "/output/report.pdf"
    +        assert "Done." in cleaned
    +        # The real tag is stripped from the delivered text...
    +        assert "MEDIA:/output/report.pdf" not in cleaned
    +        # ...but the fenced code block (incl. its example MEDIA: line) must
    +        # survive verbatim — masking is a locator, not a text rewrite.
    +        assert "```text\nMEDIA:/example/path.pdf\n```" in cleaned
    +
    +    def test_inline_code_survives_when_real_media_present(self):
    +        """When a real MEDIA: tag is delivered, an inline-code example in the
    +        same reply must not be blanked to whitespace."""
    +        content = "See MEDIA:/r/a.png and `MEDIA:/ex/b.png` inline"
    +        media, cleaned = BasePlatformAdapter.extract_media(content)
    +        assert [p for p, _ in media] == ["/r/a.png"]
    +        assert "`MEDIA:/ex/b.png`" in cleaned
    +
     
     class TestMediaInsideSerializedJson:
         """Regression coverage for #34375 — MEDIA: embedded in serialized JSON

• #36283 fix(gateway): skip MEDIA: inside serialized JSON string values (#34388)

  merged to main

  files: gateway/platforms/base.py, tests/gateway/test_platform_base.py

  Show the diff

  • gateway/platforms/base.py

    @@ -2579,6 +2579,46 @@ def filter_local_delivery_paths(file_paths) -> List[str]:
                     logger.warning("Skipping unsafe local file path: %s", _log_safe_path(raw))
             return safe_paths
     
    +    @staticmethod
    +    def _mask_json_string_media(content: str) -> str:
    +        """Blank out ``MEDIA:<bare-path>`` occurrences that sit inside a JSON
    +        string *value* so they are never delivered as real attachments.
    +
    +        Serialized tool results frequently embed a previous reply's text, e.g.::
    +
    +            {"result": "MEDIA:/Users/x/.hermes/media/generated/stale.png"}
    +
    +        Here the ``MEDIA:`` is part of stored text, not an outbound directive,
    +        but the bare-path branch of ``MEDIA_TAG_CLEANUP_RE`` would still match it
    +        and re-deliver a stale file. (Regression report #34375.)
    +
    +        The discriminator is precise so legitimate tags are untouched:
    +
    +        * Only spans opened by a JSON value-context quote (``:``, ``,``, ``{`` or
    +          ``[`` immediately before the ``"``) are considered.
    +        * Within such a span, only a ``MEDIA:`` followed by a **bare** path
    +          (``/``, ``~/`` or ``X:\\``) is masked. A ``MEDIA:"..."`` quoted-path
    +          tag — a real LLM output format the extractor supports — is not bare and
    +          is left alone.
    +        * Tags at line start, after prose whitespace, or indented are outside any
    +          JSON value span and are never affected.
    +
    +        Offsets are preserved (matched chars replaced with spaces, newlines kept)
    +        so downstream match positions stay valid.
    +        """
    +        if '"' not in content or "MEDIA:" not in content:
    +            return content
    +        chars = list(content)
    +        # JSON value-context string: a quote preceded by : , { or [ (optional ws),
    +        # capturing the (escape-aware) string body up to the closing quote.
    +        for m in re.finditer(r'(?<=[:,{\[])\s*"((?:[^"\\\n]|\\.)*)"', content):
    +            seg = m.group(1)
    +            if re.search(r'MEDIA:\s*(?:~/|/|[A-Za-z]:[/\\])', seg):
    +                for i in range(m.start(1), m.end(1)):
    +                    if chars[i] != '\n':
    +                        chars[i] = ' '
    +        return ''.join(chars)
    +
         @staticmethod
         def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
             """
    @@ -2621,7 +2661,10 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
             # set is the shared MEDIA_DELIVERY_EXTS source of truth (built once into
             # MEDIA_TAG_CLEANUP_RE) so it can never drift from extract_local_files.
             media_pattern = MEDIA_TAG_CLEANUP_RE
    -        for match in media_pattern.finditer(content):
    +        # Mask MEDIA: embedded inside serialized JSON string values so stale
    +        # paths from stored tool results are never re-delivered (#34375).
    +        scan_content = BasePlatformAdapter._mask_json_string_media(content)
    +        for match in media_pattern.finditer(scan_content):
                 path = match.group("path").strip()
                 if len(path) >= 2 and path[0] == path[-1] and path[0] in "`\"'":
                     path = path[1:-1].strip()
    @@ -2634,10 +2677,22 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
                         # and dropping every other attachment in the response.
                         continue
     
    -        # Remove MEDIA tags from content (including surrounding quote/backtick wrappers)
    +        # Remove the delivered MEDIA tags from the user-visible text. Mask
    +        # ``cleaned`` (same length, so offsets line up with it), find the real
    +        # tag spans there, and delete those spans from the *unmasked* ``cleaned``.
    +        # This strips real tags while leaving JSON-embedded MEDIA: text intact —
    +        # it is stored data, not a directive, and must read back verbatim
    +        # (#34375). Masking ``cleaned`` (not ``content``) keeps offsets valid
    +        # after the [[audio_as_voice]] / [[as_document]] directives are removed.
             if media:
    -            cleaned = media_pattern.sub('', cleaned)
    -            cleaned = re.sub(r'\n{3,}', '\n\n', cleaned).strip()
    +            masked_cleaned = BasePlatformAdapter._mask_json_string_media(cleaned)
    +            spans = [m.span() for m in media_pattern.finditer(masked_cleaned)]
    +            if spans:
    +                chars = list(cleaned)
    +                for start, end in sorted(spans, reverse=True):
    +                    del chars[start:end]
    +                cleaned = "".join(chars)
    +                cleaned = re.sub(r'\n{3,}', '\n\n', cleaned).strip()
             
             return media, cleaned
     

  • tests/gateway/test_platform_base.py

    @@ -401,6 +401,93 @@ def test_relative_path_still_ignored(self):
             assert media == []
     
     
    +class TestMediaInsideSerializedJson:
    +    """Regression coverage for #34375 — MEDIA: embedded in serialized JSON
    +    string values (e.g. a stored previous reply inside a tool result) must not
    +    be re-delivered as a real attachment, while legitimate MEDIA: tags in prose,
    +    at line start, indented, or as quoted-path tags keep working.
    +    """
    +
    +    def test_media_in_json_value_not_extracted(self):
    +        content = '{"result": "MEDIA:/tmp/stale.png"}'
    +        media, _ = BasePlatformAdapter.extract_media(content)
    +        assert media == [], f"JSON value MEDIA: leaked: {media}"
    +
    +    def test_media_in_pretty_json_value_not_extracted(self):
    +        content = '{\n  "tool_result": "MEDIA:/var/old.jpg"\n}'
    +        media, _ = BasePlatformAdapter.extract_media(content)
    +        assert media == [], f"pretty JSON MEDIA: leaked: {media}"
    +
    +    def test_media_in_json_array_not_extracted(self):
    +        content = '["MEDIA:/a/b.png", "other"]'
    +        media, _ = BasePlatformAdapter.extract_media(content)
    +        assert media == [], f"JSON array MEDIA: leaked: {media}"
    +
    +    def test_media_in_nested_json_value_not_extracted(self):
    +        content = '{"a":{"b":"see MEDIA:/x/y.pdf here"}}'
    +        media, _ = BasePlatformAdapter.extract_media(content)
    +        assert media == [], f"nested JSON MEDIA: leaked: {media}"
    +
    +    def test_media_in_embedded_serialized_reply_not_extracted(self):
    +        """A serialized tool result that embeds a prior reply's MEDIA: tag."""
    +        content = (
    +            '{"content":"previous reply MEDIA:/Users/ex/.hermes/media/'
    +            'generated/stale.png and more text"}'
    +        )
    +        media, _ = BasePlatformAdapter.extract_media(content)
    +        assert media == [], f"embedded serialized reply leaked: {media}"
    +
    +    # --- Legitimate tags must still extract (no regression vs line-start anchor) ---
    +
    +    def test_media_at_line_start_still_extracted(self):
    +        media, _ = BasePlatformAdapter.extract_media("MEDIA:/real/file.png")
    +        assert len(media) == 1 and media[0][0] == "/real/file.png"
    +
    +    def test_media_after_prose_same_line_still_extracted(self):
    +        media, _ = BasePlatformAdapter.extract_media(
    +            "Here is your file: MEDIA:/out/report.pdf"
    +        )
    +        assert len(media) == 1 and media[0][0] == "/out/report.pdf"
    +
    +    def test_media_indented_still_extracted(self):
    +        media, _ = BasePlatformAdapter.extract_media("  MEDIA:/tmp/x.png")
    +        assert len(media) == 1 and media[0][0] == "/tmp/x.png"
    +
    +    def test_quoted_path_media_still_extracted(self):
    +        """MEDIA:"..." quoted-path form (a real LLM output) is not JSON-masked."""
    +        media, _ = BasePlatformAdapter.extract_media(
    +            'MEDIA:"/path/with space/file.png"'
    +        )
    +        assert len(media) == 1 and media[0][0] == "/path/with space/file.png"
    +
    +    def test_tts_two_line_still_extracted(self):
    +        media, _ = BasePlatformAdapter.extract_media(
    +            "[[audio_as_voice]]\nMEDIA:/tmp/v.ogg"
    +        )
    +        assert len(media) == 1 and media[0][0] == "/tmp/v.ogg"
    +        assert media[0][1] is True  # voice flag
    +
    +    # --- cleaned-text invariants: real tags stripped, JSON data kept verbatim ---
    +
    +    def test_json_embedded_media_kept_verbatim_in_cleaned_text(self):
    +        """A real tag is delivered+stripped; a JSON-embedded MEDIA: stays as
    +        literal text (stored data must read back unchanged)."""
    +        content = 'MEDIA:/real/r.png\nlog: {"old":"MEDIA:/stale/s.png"}'
    +        media, cleaned = BasePlatformAdapter.extract_media(content)
    +        assert [p for p, _ in media] == ["/real/r.png"]
    +        # The JSON-embedded path must survive verbatim — not blanked to spaces.
    +        assert '{"old":"MEDIA:/stale/s.png"}' in cleaned
    +
    +    def test_cleaned_text_after_directive_not_truncated(self):
    +        """Stripping a tag preceded by a [[as_document]] directive must not
    +        shift offsets and chop the path or trailing text."""
    +        content = "See [[as_document]] MEDIA:/d/report.pdf now"
    +        media, cleaned = BasePlatformAdapter.extract_media(content)
    +        assert [p for p, _ in media] == ["/d/report.pdf"]
    +        assert "MEDIA:" not in cleaned          # real tag removed
    +        assert cleaned.endswith("now")          # trailing text intact (not chopped)
    +
    +
     class TestMediaExtensionAllowlistParity:
         """Regression coverage for issue #34517 — the MEDIA: extension black hole.
     

• #72193 fix(gateway): cross-turn media dedupe, spaced/GIS MEDIA paths, code-block-safe display strip

  merged to main

  files: gateway/platforms/base.py, gateway/run.py, tests/gateway/test_media_spaced_paths_and_history_dedupe.py, tests/gateway/test_stream_consumer.py