fix(gateway): GIS extensions, spaced MEDIA paths, and code-block-safe display strip

Follow-up wave to #72170 resolving the remaining open MEDIA-delivery gaps:

- #24032: add .kmz/.kml/.geojson/.gpx to MEDIA_DELIVERY_EXTS, and recover
  unknown-extension paths containing spaces via _match_extensionless_path —
  validation-gated forward extension across single spaces (bounded at 8
  tokens, stops at newline / next MEDIA: keyword). The regex itself stays
  non-greedy and whitespace-bounded so the #68773 absorption bug class
  cannot return; the on-disk file check is the oracle.

- #16434 (streaming half): _strip_media_tag_directives now uses the same
  mask-as-locator pattern as extract_media, so MEDIA tags inside fenced
  code blocks, inline-code examples, and JSON string values survive in
  streamed display text instead of being mangled. Display and delivery
  now agree on every protected-span rule.

- Updated three stream_consumer display expectations that pinned the old
  inconsistent behavior (backtick/double-quote tags stripped from display
  while delivery never attempted them).

gateway/platforms/base.py

@@ -1485,6 +1485,8 @@ MEDIA_DELIVERY_EXTS: Tuple[str, ...] = (
     ".pdf", ".docx", ".doc", ".odt", ".rtf", ".txt", ".md", ".epub",
     # Spreadsheets / data
     ".xlsx", ".xls", ".ods", ".csv", ".tsv", ".json", ".xml", ".yaml", ".yml",
+    # Geospatial / GIS (#24032)
+    ".kmz", ".kml", ".geojson", ".gpx",
     # Presentations
     ".pptx", ".ppt", ".odp", ".key",
     # Archives
@@ -1557,6 +1559,17 @@ MEDIA_TAG_CLEANUP_RE = re.compile(
 # or to arbitrary following text (``MEDIA:/a.pngSome text``) cannot
 # silently absorb the next path — that earlier behavior merged the two
 # paths into one invalid string and dropped the file (#68773).
+#
+# The bare form stays non-greedy and whitespace-bounded — spaced paths are
+# NOT absorbed at the regex level, because greedy space-tolerance would
+# reintroduce the #68773 bug class (gluing the next MEDIA: tag or trailing
+# prose into one invalid path). Instead, unknown-extension paths containing
+# spaces (``MEDIA:/data/map data.kmz``, ``C:\...\My Documents\x.log``) are
+# recovered by ``_match_extensionless_path`` (#24032): when the bare match
+# fails validation, the candidate is progressively extended forward across
+# single spaces — bounded, stopping at newline / the next ``MEDIA:`` keyword
+# — and the first extension that validates on disk wins. Validation is the
+# oracle, so prose never rides along and non-existent paths stay visible.
 MEDIA_EXTENSIONLESS_TAG_RE = re.compile(
     r'''[`"'*_]{0,3}MEDIA:\s*'''
     r'''(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|'''
@@ -1567,6 +1580,60 @@ MEDIA_EXTENSIONLESS_TAG_RE = re.compile(
 )
 
 
+def _match_extensionless_path(scan_text: str, match: "re.Match") -> Optional[Tuple[str, int]]:
+    """Resolve an extensionless MEDIA tag match to a validated on-disk path.
+
+    Tries the regex-captured path first. When that fails validation, the
+    candidate is progressively extended forward across single spaces
+    (validation-gated, bounded at 8 tokens, never past a newline or a
+    subsequent ``MEDIA:`` keyword) so unknown-extension paths containing
+    spaces deliver (#24032). Returns ``(safe_path, end_offset)`` where
+    ``end_offset`` is the index in ``scan_text`` just past the matched path,
+    or ``None`` when nothing validates.
+    """
+    raw = match.group("path")
+    path = _normalize_media_tag_path(raw)
+    if not path:
+        return None
+    safe = validate_media_delivery_path(path)
+    if safe:
+        return safe, match.end("path")
+    start = match.start("path")
+    nl = scan_text.find("\n", start)
+    limit = nl if nl != -1 else len(scan_text)
+    segment = scan_text[start:limit]
+    nxt = segment.find("MEDIA:", 1)
+    if nxt != -1:
+        segment = segment[:nxt]
+    pos = match.end("path") - start
+    for _ in range(8):
+        while pos < len(segment) and segment[pos] in " \t":
+            pos += 1
+        if pos >= len(segment):
+            break
+        tok_end = pos
+        while tok_end < len(segment) and segment[tok_end] not in " \t":
+            tok_end += 1
+        candidate = _normalize_media_tag_path(segment[:tok_end])
+        safe = validate_media_delivery_path(candidate)
+        if safe:
+            return safe, start + tok_end
+        pos = tok_end
+    return None
+
+
+def _merge_spans(spans: list) -> list:
+    """Merge overlapping/nested (start, end) spans so multi-pattern matches
+    over the same tag never double-delete adjacent text."""
+    merged: list = []
+    for s, e in sorted(spans):
+        if merged and s <= merged[-1][1]:
+            merged[-1] = (merged[-1][0], max(merged[-1][1], e))
+        else:
+            merged.append((s, e))
+    return merged
+
+
 def _normalize_media_tag_path(raw: str) -> str:
     path = str(raw or "").strip()
     if len(path) >= 2 and path[0] == path[-1] and path[0] in "`\"'":
@@ -1587,8 +1654,25 @@ def _path_lacks_deliverable_extension(path: str) -> bool:
     return not suffix or suffix not in MEDIA_DELIVERY_EXTS
 
 
+def _resolve_extensionless_candidate(path: str) -> Optional[str]:
+    """Validate a bare extensionless-branch path (no forward extension).
+
+    Thin wrapper kept for call sites that only have the normalized path
+    (no scan-text context for spaced-path recovery).
+    """
+    if not path:
+        return None
+    return validate_media_delivery_path(path)
+
+
 def _strip_media_tag_directives(text: str) -> str:
-    """Remove MEDIA: tags and [[audio_as_voice]] / [[as_document]] markers."""
+    """Remove MEDIA: tags and [[audio_as_voice]] / [[as_document]] markers.
+
+    Protected spans (fenced code blocks, inline code holding non-deliverable
+    example tags, blockquotes, JSON string values) are used as a mask-locator
+    only — tags inside them are neither stripped nor mangled, matching
+    ``extract_media``'s treatment so display text and delivery agree (#16434).
+    """
     if (
         "MEDIA:" not in text
         and "[[audio_as_voice]]" not in text
@@ -1597,14 +1681,28 @@ def _strip_media_tag_directives(text: str) -> str:
         return text
     cleaned = text.replace("[[audio_as_voice]]", "").replace("[[as_document]]", "")
 
-    def _strip_extensionless(match: re.Match) -> str:
+    # Locate real tag spans on a masked copy (offset-preserving), then delete
+    # exactly those spans from the unmasked text — same pattern as
+    # extract_media. Import-cycle-free: BasePlatformAdapter is defined later
+    # in this module, so resolve it lazily at call time.
+    masked = BasePlatformAdapter._mask_protected_spans(cleaned)
+    masked = BasePlatformAdapter._mask_json_string_media(masked)
+
+    spans: list = [m.span() for m in MEDIA_TAG_CLEANUP_RE.finditer(masked)]
+    for match in MEDIA_EXTENSIONLESS_TAG_RE.finditer(masked):
         path = _normalize_media_tag_path(match.group("path"))
         if not path or not _path_lacks_deliverable_extension(path):
-            return match.group(0)
-        return "" if validate_media_delivery_path(path) else match.group(0)
+            continue
+        resolved = _match_extensionless_path(masked, match)
+        if resolved is not None:
+            spans.append((match.start(), resolved[1]))
 
-    cleaned = MEDIA_TAG_CLEANUP_RE.sub("", cleaned)
-    return MEDIA_EXTENSIONLESS_TAG_RE.sub(_strip_extensionless, cleaned)
+    if spans:
+        chars = list(cleaned)
+        for start, end in reversed(_merge_spans(spans)):
+            del chars[start:end]
+        cleaned = "".join(chars)
+    return cleaned
 
 
 def get_document_cache_dir() -> Path:
@@ -4013,8 +4111,11 @@ class BasePlatformAdapter(ABC):
             path = _normalize_media_tag_path(match.group("path"))
             if not path or not _path_lacks_deliverable_extension(path):
                 continue
-            safe = validate_media_delivery_path(path)
-            if safe and safe not in seen_paths:
+            resolved = _match_extensionless_path(scan_content, match)
+            if resolved is None:
+                continue
+            safe = resolved[0]
+            if safe not in seen_paths:
                 media.append((safe, has_voice_tag))
                 seen_paths.add(safe)
 
@@ -4034,11 +4135,12 @@ class BasePlatformAdapter(ABC):
                 path = _normalize_media_tag_path(match.group("path"))
                 if not path or not _path_lacks_deliverable_extension(path):
                     continue
-                if validate_media_delivery_path(path):
-                    spans.append(match.span())
+                resolved = _match_extensionless_path(masked_cleaned, match)
+                if resolved is not None:
+                    spans.append((match.start(), resolved[1]))
             if spans:
                 chars = list(cleaned)
-                for start, end in sorted(spans, reverse=True):
+                for start, end in reversed(_merge_spans(spans)):
                     del chars[start:end]
                 cleaned = "".join(chars)
                 cleaned = re.sub(r'\n{3,}', '\n\n', cleaned).strip()

tests/gateway/test_media_spaced_paths_and_history_dedupe.py

@@ -0,0 +1,128 @@
+"""Regression tests: spaced paths, GIS extensions, cross-turn dedupe plumbing,
+and code-block-safe streaming display strip.
+
+Covers the follow-up wave after PR #72170:
+
+* #24032 — GIS extensions (.kmz/.kml/.geojson/.gpx) are deliverable, and
+  unknown-extension paths containing spaces extract via the
+  validation-gated progressive right-trim (``_spaced_path_candidates``).
+* #16434 half — ``strip_media_directives_for_display`` (streaming path) no
+  longer strips MEDIA tags out of fenced code blocks / inline-code examples;
+  protected spans are a mask-locator, matching ``extract_media``.
+* #53586 — ``_collect_history_media_paths`` also collects tags from
+  assistant messages, and the post-stream delivery path filters against it.
+"""
+
+import os
+
+import pytest
+
+from gateway.platforms.base import (
+    BasePlatformAdapter,
+    MEDIA_DELIVERY_EXTS,
+)
+from gateway.run import _collect_history_media_paths
+
+
+class TestGisExtensions:
+    def test_gis_extensions_in_delivery_set(self):
+        for ext in (".kmz", ".kml", ".geojson", ".gpx"):
+            assert ext in MEDIA_DELIVERY_EXTS
+
+    def test_geojson_extracts(self, tmp_path):
+        p = tmp_path / "route.geojson"
+        p.write_text("{}")
+        media, cleaned = BasePlatformAdapter.extract_media(f"MEDIA:{p}")
+        assert [x for x, _ in media] == [str(p)]
+        assert "MEDIA:" not in cleaned
+
+
+class TestSpacedPaths:
+    def test_spaced_known_ext_extracts(self, tmp_path):
+        p = tmp_path / "map data.kmz"
+        p.write_bytes(b"PK")
+        media, _ = BasePlatformAdapter.extract_media(f"MEDIA:{p}")
+        assert [x for x, _ in media] == [str(p)]
+
+    def test_spaced_unknown_ext_extracts_when_file_exists(self, tmp_path):
+        p = tmp_path / "my server.log"
+        p.write_text("log line\n")
+        media, cleaned = BasePlatformAdapter.extract_media(f"MEDIA:{p}")
+        assert [os.path.realpath(x) for x, _ in media] == [os.path.realpath(str(p))]
+        assert "MEDIA:" not in cleaned
+
+    def test_spaced_path_followed_by_prose_keeps_prose(self, tmp_path):
+        p = tmp_path / "my server.log"
+        p.write_text("log line\n")
+        media, cleaned = BasePlatformAdapter.extract_media(
+            f"MEDIA:{p} is the log you asked for"
+        )
+        assert [os.path.realpath(x) for x, _ in media] == [os.path.realpath(str(p))]
+        assert "is the log you asked for" in cleaned
+
+    def test_spaced_nonexistent_stays_visible(self):
+        text = "MEDIA:/data/not real file.xyz here"
+        media, cleaned = BasePlatformAdapter.extract_media(text)
+        assert media == []
+        assert "MEDIA:/data/not real file.xyz" in cleaned
+
+    def test_forward_extension_stops_at_next_media_tag(self, tmp_path):
+        a = tmp_path / "Caddyfile"
+        b = tmp_path / "Dockerfile"
+        a.write_text("localhost\n")
+        b.write_text("FROM alpine\n")
+        media, cleaned = BasePlatformAdapter.extract_media(
+            f"MEDIA:{a} MEDIA:{b}"
+        )
+        got = sorted(os.path.realpath(x) for x, _ in media)
+        assert got == sorted(
+            [os.path.realpath(str(a)), os.path.realpath(str(b))]
+        )
+        assert "MEDIA:" not in cleaned
+
+
+class TestStreamingDisplayStripCodeBlocks:
+    def test_fenced_code_example_preserved(self, tmp_path):
+        p = tmp_path / "real.pdf"
+        p.write_text("x")
+        text = f"Example:\n```\nMEDIA:{p}\n```\ndone MEDIA:{p}"
+        out = BasePlatformAdapter.strip_media_directives_for_display(text)
+        # The example inside the fence survives verbatim; the real tag outside
+        # is stripped.
+        assert f"MEDIA:{p}" in out
+        assert out.count(f"MEDIA:{p}") == 1
+        assert "```" in out
+
+    def test_inline_code_example_preserved(self):
+        text = "Use `MEDIA:/nonexistent/example.csv` to attach files."
+        out = BasePlatformAdapter.strip_media_directives_for_display(text)
+        assert "`MEDIA:/nonexistent/example.csv`" in out
+
+    def test_plain_tag_still_stripped(self, tmp_path):
+        p = tmp_path / "real.csv"
+        p.write_text("x")
+        out = BasePlatformAdapter.strip_media_directives_for_display(
+            f"Here you go MEDIA:{p}"
+        )
+        assert "MEDIA:" not in out
+
+
+class TestHistoryMediaDedupe:
+    def test_assistant_message_tags_collected(self):
+        history = [
+            {"role": "user", "content": "make a chart"},
+            {"role": "assistant", "content": "Done! MEDIA:/tmp/chart.png"},
+            {"role": "user", "content": "thanks"},
+        ]
+        paths = _collect_history_media_paths(history)
+        assert "/tmp/chart.png" in paths
+
+    def test_tool_message_tags_still_collected(self):
+        history = [
+            {"role": "tool", "content": "MEDIA:/tmp/out.pdf"},
+        ]
+        paths = _collect_history_media_paths(history)
+        assert "/tmp/out.pdf" in paths
+
+    def test_empty_history_empty_set(self):
+        assert _collect_history_media_paths([]) == set()

tests/gateway/test_stream_consumer.py

@@ -50,12 +50,40 @@ class TestCleanForDisplay:
         assert "MEDIA:" not in result
         assert "Audio generated" in result
 
-    def test_media_tag_with_quotes(self):
-        """MEDIA: tags wrapped in quotes or backticks are removed."""
-        for wrapper in ['`MEDIA:/path/file.png`', '"MEDIA:/path/file.png"', "'MEDIA:/path/file.png'"]:
-            text = f"Result: {wrapper}"
-            result = GatewayStreamConsumer._clean_for_display(text)
-            assert "MEDIA:" not in result, f"Failed for wrapper: {wrapper}"
+    def test_media_tag_single_quoted_stripped(self):
+        """A single-quote-wrapped tag matches the known-ext cleanup pattern
+        and is removed (delivery attempts it too — consistent)."""
+        result = GatewayStreamConsumer._clean_for_display(
+            "Result: 'MEDIA:/path/file.png'"
+        )
+        assert "MEDIA:" not in result
+
+    def test_media_tag_double_quoted_json_context_stays_visible(self):
+        """A double-quoted tag preceded by a colon sits in a JSON value
+        context (#34375): extract_media masks it and never delivers, so
+        display keeps it visible too instead of silently hiding a tag that
+        produced no attachment (display/delivery consistency)."""
+        result = GatewayStreamConsumer._clean_for_display(
+            'Result: "MEDIA:/path/file.png"'
+        )
+        assert '"MEDIA:/path/file.png"' in result
+
+    def test_media_tag_in_backticks_real_file_stripped(self, tmp_path):
+        """A backtick-wrapped tag pointing at a REAL file is a delivery
+        directive: extract_media delivers it, so display strips it."""
+        p = tmp_path / "file.png"
+        p.write_bytes(b"\x89PNG")
+        result = GatewayStreamConsumer._clean_for_display(f"Result: `MEDIA:{p}`")
+        assert "MEDIA:" not in result
+
+    def test_media_tag_in_backticks_bogus_path_stays_visible(self):
+        """A backtick-wrapped tag with a non-existent path is an inline-code
+        example: extract_media does NOT deliver it, so display must not
+        silently strip it either (display/delivery consistency, #16434)."""
+        result = GatewayStreamConsumer._clean_for_display(
+            "Result: `MEDIA:/path/file.png`"
+        )
+        assert "`MEDIA:/path/file.png`" in result
 
     def test_audio_as_voice_stripped(self):
         """[[audio_as_voice]] directive is removed."""