fix(gateway): unify MEDIA extraction onto one curated extension set

gateway/platforms/base.py

@@ -857,6 +857,91 @@ def cache_video_from_bytes(data: bytes, ext: str = ".mp4") -> str:
     _HERMES_HOME / "cache" / "screenshots",
 )
 
+# Single source of truth for "what counts as a deliverable file" across every
+# MEDIA: extraction site. Curated as the union of what extract_media and
+# extract_local_files already delivered, so source/script/config extensions
+# (.py .sh .ts .toml .ini .cfg .log) — never in either list — stay excluded:
+# delivering them outbound would be new behavior, and scripts/config often
+# carry secrets.
+_DELIVERY_MEDIA_EXTS = frozenset({
+    # Images (embed inline where supported)
+    ".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".tiff", ".svg",
+    # Video (embed inline where supported)
+    ".mp4", ".mov", ".avi", ".mkv", ".webm",
+    # Audio (voice/audio attachment)
+    ".mp3", ".wav", ".ogg", ".opus", ".m4a", ".flac",
+    # Documents
+    ".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",
+    # GIS / geospatial (deliverable artifacts; route to send_document) — #24032
+    ".kml", ".kmz", ".geojson", ".gpx",
+    # Mobile packages
+    ".apk", ".ipa",
+    # Ebook
+    ".epub",
+})
+
+
+# Longest-first so 'html' beats 'htm' and 'jpeg' beats 'jpg' regardless of the
+# trailing boundary anchor; each extension regex-escaped.
+_MEDIA_EXT_ALT = "|".join(
+    re.escape(ext.lstrip(".")) for ext in sorted(_DELIVERY_MEDIA_EXTS, key=len, reverse=True)
+)
+
+# Explicit MEDIA: tags in model/tool RESPONSE TEXT. Shared by extract_media()
+# and stream_consumer._clean_for_display(). `MEDIA:` is case-SENSITIVE (the
+# documented tag), but the extension match is case-insensitive via (?i:...)
+# so `.PNG`/`.png` both match. Optional **bold** wrappers and surrounding
+# quotes/backticks are consumed so they never leak into cleaned user text.
+_MEDIA_TAG_RE = re.compile(
+    r'''(?P<wrap>\*{0,2})[`"']?MEDIA:\s*'''
+    r'''(?P<path>`[^`\n]+`|"[^"\n]+"|'[^'\n]+'|'''
+    r'''(?:~/|/|[A-Za-z]:[\\/]|\\\\)\S*?(?:[^\S\n]+\S+?)*?\.(?i:''' + _MEDIA_EXT_ALT + r''')'''
+    r'''(?=[\s`"',;:)\]}*]|$))[`"']?(?P=wrap)'''
+)
+
+# Bare local paths (no MEDIA: prefix) in response text. IGNORECASE is safe
+# here — there is no case-sensitive literal to protect.
+_LOCAL_PATH_RE = re.compile(
+    r'(?<![/:\w.])(?:~/|/)(?:[\w.\-]+/)*[\w.\-]+\.(?:' + _MEDIA_EXT_ALT + r')\b',
+    re.IGNORECASE,
+)
+
+# MEDIA: tags inside raw tool/function JSON content (run.py history-dedup +
+# salvage re-injection). Narrower grammar than _MEDIA_TAG_RE (no quoted
+# branches, no spaced paths — JSON paths are single tokens). Defined here so
+# it is built from the same frozenset and lives as one module-level object.
+# Windows drive-letter and UNC paths are recognized for parity with
+# _MEDIA_TAG_RE (so they're stripped from text; delivery still fail-closes).
+_TOOL_MEDIA_RE = re.compile(
+    r'MEDIA:((?:/|~/|[A-Za-z]:[\\/]|\\\\)\S+\.(?:' + _MEDIA_EXT_ALT + r'))\b',
+    re.IGNORECASE,
+)
+
+
+def tool_media_paths(content: str) -> List[str]:
+    """Salvage MEDIA: paths from raw tool/function JSON content, in order.
+
+    Single source for run.py's history-dedup and salvage sites: applies
+    ``_TOOL_MEDIA_RE`` and the same JSON-token cleanup (strip surrounding
+    whitespace + trailing ``",}``) both used, so the two sites can't drift.
+    """
+    if "MEDIA:" not in content:
+        return []
+    paths: List[str] = []
+    for match in _TOOL_MEDIA_RE.finditer(content):
+        path = match.group(1).strip().rstrip('",}')
+        if path:
+            paths.append(path)
+    return paths
+
 # Default recency window for trusting freshly-produced files (seconds).
 # The agent's actual work generally completes well inside 10 minutes; legitimate
 # build artifacts (PDFs from pandoc, plots from matplotlib, etc.) almost always
@@ -1029,6 +1114,18 @@ def validate_media_delivery_path(path: str) -> Optional[str]:
     if not candidate:
         return None
 
+    # Fail-closed for Windows/UNC paths (`C:\`, `C:/`, `\\server\share`):
+    # extraction recognizes them so the MEDIA: tag is stripped from user text
+    # (#28989, #24032), but native-Windows *delivery* needs a Windows-aware
+    # allowlist the POSIX denylist below can't provide — deferred to the L0
+    # path-validation PR (refs #32644). Until then they're dropped (surfaced
+    # via media_dropped), never delivered — which also closes a latent
+    # fail-open on Windows hosts. `\w` (not [A-Za-z]) catches Unicode-homoglyph
+    # drives; no POSIX path starts with word-char + `:` + slash, so it never
+    # over-rejects.
+    if re.match(r'\w:[\\/]|\\\\', candidate):
+        return None
+
     try:
         expanded = Path(os.path.expanduser(candidate))
     except (OSError, RuntimeError, ValueError):
@@ -2479,16 +2576,32 @@ def validate_media_delivery_path(path: str) -> Optional[str]:
         return validate_media_delivery_path(path)
 
     @staticmethod
-    def filter_media_delivery_paths(media_files) -> List[Tuple[str, bool]]:
-        """Drop unsafe MEDIA paths and normalize accepted paths."""
+    def partition_media_delivery_paths(
+        media_files,
+    ) -> Tuple[List[Tuple[str, bool]], List[Tuple[str, bool]]]:
+        """Split MEDIA paths into (safe, dropped) using the delivery gate.
+
+        Same gate as ``validate_media_delivery_path``; the dropped list
+        preserves the caller's original ``(path, is_voice)`` pairs so the
+        caller can surface a warning instead of letting the send silently
+        succeed with text only (issue #32644).
+        """
         safe_media: List[Tuple[str, bool]] = []
+        dropped_media: List[Tuple[str, bool]] = []
         for media_path, is_voice in media_files or []:
             raw = str(media_path)
             safe_path = validate_media_delivery_path(raw)
             if safe_path:
                 safe_media.append((safe_path, bool(is_voice)))
             else:
+                dropped_media.append((raw, bool(is_voice)))
                 logger.warning("Skipping unsafe MEDIA directive path: %s", _log_safe_path(raw))
+        return safe_media, dropped_media
+
+    @staticmethod
+    def filter_media_delivery_paths(media_files) -> List[Tuple[str, bool]]:
+        """Drop unsafe MEDIA paths and normalize accepted paths."""
+        safe_media, _dropped = BasePlatformAdapter.partition_media_delivery_paths(media_files)
         return safe_media
 
     @staticmethod
@@ -2530,7 +2643,8 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
         Returns:
             Tuple of (list of (path, is_voice) pairs, cleaned content with tags removed).
         """
-        media = []
+        media: List[Tuple[str, bool]] = []
+        seen_paths: set[str] = set()
         cleaned = content
 
         # Check for [[audio_as_voice]] directive
@@ -2540,30 +2654,36 @@ def extract_media(content: str) -> Tuple[List[Tuple[str, bool]], str]:
         # ``content`` for it (so they can still react to it); here we just
         # keep it out of the user-visible cleaned text.
         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`"',;:)\]}]|$))[`"']?'''
-        )
-        for match in media_pattern.finditer(content):
+
+        # Extract MEDIA:<path> tags using the shared, module-level
+        # _MEDIA_TAG_RE built from _DELIVERY_MEDIA_EXTS (single source of
+        # truth — uppercase exts, .md/.html, **bold** wrappers, quoted and
+        # spaced paths all handled identically here and in stream_consumer).
+        for match in _MEDIA_TAG_RE.finditer(content):
             path = match.group("path").strip()
             if len(path) >= 2 and path[0] == path[-1] and path[0] in "`\"'":
                 path = path[1:-1].strip()
-            path = path.lstrip("`\"'").rstrip("`\"',.;:)}]")
-            if path:
-                try:
-                    media.append((os.path.expanduser(path), has_voice_tag))
-                except (OSError, RuntimeError, ValueError):
-                    # Skip a crafted ~\x00 path rather than aborting extraction
-                    # and dropping every other attachment in the response.
-                    continue
-
-        # Remove MEDIA tags from content (including surrounding quote/backtick wrappers)
-        if media:
-            cleaned = media_pattern.sub('', cleaned)
+            path = path.lstrip("`\"'").rstrip("`\"',.;:)}]*")
+            if not path:
+                continue
+            try:
+                expanded = os.path.expanduser(path)
+            except (OSError, RuntimeError, ValueError):
+                # Skip a crafted ~\x00 path rather than aborting extraction
+                # and dropping every other attachment in the response.
+                continue
+            if expanded in seen_paths:
+                continue
+            seen_paths.add(expanded)
+            media.append((expanded, has_voice_tag))
+
+        # Remove MEDIA tags from content (including surrounding quote/backtick
+        # and **bold** wrappers). The .search() guard strips bold/quoted tags
+        # even in the defensive case where dedup left ``media`` empty.
+        if media or _MEDIA_TAG_RE.search(cleaned):
+            cleaned = _MEDIA_TAG_RE.sub('', cleaned)
             cleaned = re.sub(r'\n{3,}', '\n\n', cleaned).strip()
-        
+
         return media, cleaned
 
     @staticmethod
@@ -2591,33 +2711,10 @@ 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',
-        )
-        ext_part = '|'.join(e.lstrip('.') for e in _LOCAL_MEDIA_EXTS)
-
-        # (?<![/:\w.]) prevents matching inside URLs (e.g. https://…/img.png)
-        #             and relative paths (./foo.png)
-        # (?:~/|/)    anchors to absolute or home-relative paths
-        path_re = re.compile(
-            r'(?<![/:\w.])(?:~/|/)(?:[\w.\-]+/)*[\w.\-]+\.(?:' + ext_part + r')\b',
-            re.IGNORECASE,
-        )
+        # Extension set + compiled regex live at module level
+        # (_DELIVERY_MEDIA_EXTS / _LOCAL_PATH_RE) so MEDIA: tags and bare
+        # paths can never drift apart.
+        path_re = _LOCAL_PATH_RE
 
         # Build spans covered by fenced code blocks and inline code
         code_spans: list = []
@@ -3729,7 +3826,13 @@ 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 residual MEDIA: tags via the shared _MEDIA_TAG_RE (same
+                # object as extract_media / stream_consumer) instead of a blind
+                # MEDIA:\S+ — recognized tags were already removed above, so
+                # this is a defensive no-op for them; unknown-extension tags are
+                # left visible rather than silently erased (issue #32644,
+                # consistent with the streaming display path).
+                text_content = _MEDIA_TAG_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

@@ -1043,6 +1043,7 @@ def _reload_runtime_env_preserving_config_authority() -> None:
     MessageType,
     _reply_anchor_for_event,
     merge_pending_message_event,
+    tool_media_paths,
 )
 from gateway.restart import (
     DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT,
@@ -17181,18 +17182,7 @@ def _clarify_callback_sync(question: str, choices) -> str:
             for _hm in agent_history:
                 if _hm.get("role") in {"tool", "function"}:
                     _hc = _hm.get("content", "")
-                    if "MEDIA:" in _hc:
-                        _TOOL_MEDIA_RE = re.compile(
-                            r'MEDIA:((?:/|~\/)\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(_hc):
-                            _p = _match.group(1).strip().rstrip('",}')
-                            if _p:
-                                _history_media_paths.add(_p)
+                    _history_media_paths.update(tool_media_paths(_hc))
 
             # Register per-session gateway approval callback so dangerous
             # command approval blocks the agent thread (mirrors CLI input()).
@@ -17488,16 +17478,8 @@ def _approval_notify_sync(approval_data: dict) -> None:
                     if msg.get("role") in {"tool", "function"}:
                         content = msg.get("content", "")
                         if "MEDIA:" in content:
-                            _TOOL_MEDIA_RE = re.compile(
-                                r'MEDIA:((?:/|~\/)\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:
+                            for path in tool_media_paths(content):
+                                if path not in _history_media_paths:
                                     media_tags.append(f"MEDIA:{path}")
                             if "[[audio_as_voice]]" in content:
                                 has_voice_directive = True

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_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,10 @@ 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+[`"']?''')
+    # Shared with base.extract_media() so streaming display and the
+    # non-streaming final text strip MEDIA: tags identically (one object,
+    # zero drift). Extension-anchored + space-aware; see _MEDIA_TAG_RE.
+    _MEDIA_RE = _MEDIA_TAG_RE
 
     @staticmethod
     def _clean_for_display(text: str) -> str: