fix(gateway): surface dropped MEDIA paths in send_message result

gateway/platforms/base.py

@@ -2809,12 +2809,30 @@ def validate_media_delivery_path(path: str, session_key: str = "") -> Optional[s
         """Return a resolved path if it is safe for native attachment upload."""
         return validate_media_delivery_path(path, session_key=session_key)
 
+    @staticmethod
+    def partition_media_delivery_paths(
+        media_files, session_key: str = "",
+    ) -> Tuple[List[Tuple[str, bool]], List[Tuple[str, bool]]]:
+        """Split MEDIA paths into accepted and rejected delivery paths."""
+        safe_media: List[Tuple[str, bool]] = []
+        dropped_media: List[Tuple[str, bool]] = []
+        for media_path, is_voice in media_files or []:
+            safe_path = _validated_delivery_path(
+                media_path, session_key, "MEDIA directive path"
+            )
+            if safe_path:
+                safe_media.append((safe_path, bool(is_voice)))
+            else:
+                dropped_media.append((str(media_path), bool(is_voice)))
+        return safe_media, dropped_media
+
     @staticmethod
     def filter_media_delivery_paths(media_files, session_key: str = "") -> List[Tuple[str, bool]]:
         """Drop unsafe MEDIA paths and normalize accepted paths."""
-        return [
-            (safe_path, bool(is_voice)) for media_path, is_voice in media_files or []
-            if (safe_path := _validated_delivery_path(media_path, session_key, "MEDIA directive path"))]
+        safe_media, _dropped_media = BasePlatformAdapter.partition_media_delivery_paths(
+            media_files, session_key=session_key
+        )
+        return safe_media
 
     @staticmethod
     def filter_local_delivery_paths(file_paths, session_key: str = "") -> List[str]:

tests/gateway/test_platform_base.py

@@ -33,7 +33,7 @@ def test_media_delivery_denies_encrypted_bitwarden_cache(tmp_path, monkeypatch):
     monkeypatch.setattr(base, "_HERMES_ROOT", hermes_home)
     path = hermes_home / "cache" / "bws_cache.enc.json"
     path.parent.mkdir()
-    path.write_text("encrypted-secret-cache")
+    path.write_text("encrypted-secret-cache", encoding="utf-8")
 
     assert path in base._media_delivery_denied_paths()
     assert base.validate_media_delivery_path(str(path)) is None
@@ -517,6 +517,27 @@ def test_filter_keeps_safe_media_and_drops_unsafe(self, tmp_path, monkeypatch):
 
         assert filtered == [(str(safe.resolve()), True)]
 
+    def test_partition_returns_dropped_paths_so_caller_can_warn(self, tmp_path, monkeypatch):
+        """``partition_media_delivery_paths`` is the silent-drop fix: the
+        caller needs both the safe and the dropped list so a text-only send
+        does not silently report success on a stripped attachment
+        (issue #32644).
+        """
+        root = tmp_path / "media-cache"
+        safe = root / "speech.ogg"
+        unsafe = tmp_path / "outside.ogg"
+        safe.parent.mkdir(parents=True)
+        safe.write_bytes(b"OggS")
+        unsafe.write_bytes(b"OggS")
+        self._patch_roots(monkeypatch, root)
+
+        safe_list, dropped_list = BasePlatformAdapter.partition_media_delivery_paths([
+            (str(unsafe), False),
+            (str(safe), True),
+        ])
+
+        assert safe_list == [(str(safe.resolve()), True)]
+        assert dropped_list == [(str(unsafe), False)]
 
     def test_allows_stale_kanban_attachment_but_not_neighboring_workspace(
         self, tmp_path, monkeypatch,
@@ -596,7 +617,7 @@ def test_accepts_stale_file_outside_allowlist(self, tmp_path, monkeypatch):
         self._patch_roots(monkeypatch)
 
         notes = tmp_path / "notes.md"
-        notes.write_text("# Old notes\n")
+        notes.write_text("# Old notes\n", encoding="utf-8")
         old_mtime = time.time() - 7200  # 2 hours ago — far outside any window
         os.utime(notes, (old_mtime, old_mtime))
 
@@ -622,7 +643,7 @@ def test_denylist_blocks_mcp_oauth_tokens(self, tmp_path, monkeypatch, rel):
         hermes_dir = fake_home / ".hermes"
         (hermes_dir / "mcp-tokens").mkdir(parents=True)
         secret = hermes_dir / rel
-        secret.write_text('{"access_token": "live-bearer-abc123"}')
+        secret.write_text('{"access_token": "live-bearer-abc123"}', encoding="utf-8")
         monkeypatch.setenv("HOME", str(fake_home))
         monkeypatch.setattr(
             "gateway.platforms.base._HERMES_HOME",
@@ -649,7 +670,9 @@ def test_denylist_blocks_google_token_default_mode(self, tmp_path, monkeypatch):
         hermes_dir = fake_home / ".hermes"
         hermes_dir.mkdir(parents=True)
         token = hermes_dir / "google_token.json"
-        token.write_text('{"access_token": "***", "refresh_token": "***"}')
+        token.write_text(
+            '{"access_token": "***", "refresh_token": "***"}', encoding="utf-8"
+        )
         monkeypatch.setenv("HOME", str(fake_home))
         monkeypatch.setattr("gateway.platforms.base._HERMES_HOME", hermes_dir)
         monkeypatch.setattr("gateway.platforms.base._HERMES_ROOT", hermes_dir)
@@ -903,7 +926,7 @@ def test_container_credential_path_never_translates_through_home(self, tmp_path,
         home = sandbox / "docker" / "default" / "home"
         secret = home / ".hermes"
         secret.mkdir(parents=True)
-        (secret / "auth.json").write_text('{"token": "SECRET"}')
+        (secret / "auth.json").write_text('{"token": "SECRET"}', encoding="utf-8")
         hermes_home = tmp_path / ".hermes"
         hermes_home.mkdir()
         monkeypatch.setenv("HERMES_HOME", str(hermes_home))
@@ -1399,7 +1422,7 @@ def test_home_mount_translates_stray_root_writes(self, monkeypatch):
         home = self._sandbox_dir() / "home"
         home.mkdir(parents=True, exist_ok=True)
         produced = home / "note.txt"
-        produced.write_text("hi")
+        produced.write_text("hi", encoding="utf-8")
 
         assert BasePlatformAdapter.validate_media_delivery_path(
             "/root/note.txt", session_key=self.SESSION_KEY
@@ -1413,7 +1436,7 @@ def test_home_credential_surface_still_refused(self, monkeypatch):
         for task in ("default", f"session:{self.SESSION_KEY}"):
             secrets = self._sandbox_dir(task) / "home" / ".hermes"
             secrets.mkdir(parents=True, exist_ok=True)
-            (secrets / "auth.json").write_text("{}")
+            (secrets / "auth.json").write_text("{}", encoding="utf-8")
 
         assert (
             BasePlatformAdapter.validate_media_delivery_path(

tests/tools/test_send_message_tool.py

@@ -345,7 +345,18 @@ def test_media_tag_outside_allowed_roots_is_not_sent(self, tmp_path, monkeypatch
                 )
             )
 
+        # Text sent, but the result must flag that the MEDIA attachment was
+        # dropped so CLI/MCP callers do not read "success" as "media
+        # delivered" (issue #32644).
         assert result["success"] is True
+        assert result["media_dropped"] == [str(secret)]
+        assert result["warnings"] == [
+            "1 MEDIA attachment(s) were dropped because the path is outside the gateway's "
+            "allowed delivery roots; only text was delivered. Move the file under a "
+            "Hermes-managed cache or add its directory to gateway.media_delivery_allow_dirs. "
+            "Alternatively, when gateway.strict is enabled, set gateway.trust_recent_files to true "
+            "to allow recently produced files."
+        ]
         send_mock.assert_awaited_once_with(
             Platform.TELEGRAM,
             telegram_cfg,
@@ -356,6 +367,34 @@ def test_media_tag_outside_allowed_roots_is_not_sent(self, tmp_path, monkeypatch
             force_document=False,
         )
 
+    def test_media_dropped_warning_not_added_when_send_errors(self, tmp_path, monkeypatch):
+        """An upstream send failure must not be masked by the media-dropped
+        warning; ``media_dropped`` is only attached to a successful result.
+        """
+        monkeypatch.setenv("HERMES_MEDIA_DELIVERY_STRICT", "1")
+        monkeypatch.setenv("HERMES_MEDIA_TRUST_RECENT_FILES", "0")
+        config, _telegram_cfg = _make_config()
+        secret = tmp_path / "secret.pdf"
+        secret.write_bytes(b"%PDF secret")
+
+        with patch("gateway.config.load_gateway_config", return_value=config), \
+             patch("tools.interrupt.is_interrupted", return_value=False), \
+             patch("model_tools._run_async", side_effect=_run_async_immediately), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"error": "boom"})):
+            result = json.loads(
+                send_message_tool(
+                    {
+                        "action": "send",
+                        "target": "telegram:12345",
+                        "message": f"hello\nMEDIA:{secret}",
+                    }
+                )
+            )
+
+        assert "error" in result
+        assert "media_dropped" not in result
+        assert "warnings" not in result
+
     def test_top_level_send_failure_redacts_query_token(self):
         config, _telegram_cfg = _make_config()
         leaked = "very-secret-query-token-123456"
@@ -870,7 +909,7 @@ def test_thread_not_found_for_media_retries_without_message_thread_id(self, monk
 
         # Create a test file
         test_file = tmp_path / "doc.txt"
-        test_file.write_text("test content")
+        test_file.write_text("test content", encoding="utf-8")
 
         asyncio.run(
             _send_telegram(

tools/send_message_tool.py

@@ -120,7 +120,7 @@ def _handle_send(args):
     # Capture [[as_document]] before extract_media strips it (images keep original bytes via send_document).
     force_document_attachments = "[[as_document]]" in message
     media_files, cleaned_message = BasePlatformAdapter.extract_media(message)
-    media_files = BasePlatformAdapter.filter_media_delivery_paths(media_files)
+    media_files, dropped_media = BasePlatformAdapter.partition_media_delivery_paths(media_files)
     mirror_text = cleaned_message.strip() or _describe_media_for_mirror(media_files)
     used_home_channel = not chat_id
     if used_home_channel:
@@ -146,6 +146,17 @@ def _handle_send(args):
         if isinstance(result, dict) and result.get("success"):
             if used_home_channel:
                 result["note"] = f"Sent to {platform_name} home channel (chat_id: {chat_id})"
+            if dropped_media:
+                dropped_paths = [path for path, _is_voice in dropped_media]
+                warning = (
+                    f"{len(dropped_paths)} MEDIA attachment(s) were dropped because the path is outside "
+                    "the gateway's allowed delivery roots; only text was delivered. Move the file under a "
+                    "Hermes-managed cache or add its directory to gateway.media_delivery_allow_dirs. "
+                    "Alternatively, when gateway.strict is enabled, set gateway.trust_recent_files to true "
+                    "to allow recently produced files."
+                )
+                result["warnings"] = [*result.get("warnings", []), warning]
+                result["media_dropped"] = dropped_paths
             if mirror_text and _mirror_sent_message(platform_name, chat_id, mirror_text, thread_id):
                 result["mirrored"] = True
         if isinstance(result, dict) and "error" in result: