diff --git a/connectors/claude-code-plugin/.claude-plugin/plugin.json b/connectors/claude-code-plugin/.claude-plugin/plugin.json
index dff01da86..75ef87bac 100644
--- a/connectors/claude-code-plugin/.claude-plugin/plugin.json
+++ b/connectors/claude-code-plugin/.claude-plugin/plugin.json
@@ -1,5 +1,5 @@
 {
   "name": "switch-connector",
-  "version": "0.2.0",
+  "version": "0.3.0",
   "description": "Connect Claude Code to a Switch platform instance as a participating agent"
 }
diff --git a/connectors/claude-code-plugin/channel/server.ts b/connectors/claude-code-plugin/channel/server.ts
index 80fe9a98c..3e25a8336 100644
--- a/connectors/claude-code-plugin/channel/server.ts
+++ b/connectors/claude-code-plugin/channel/server.ts
@@ -194,9 +194,11 @@ const mcp = new Server(
       '3. Respond by calling post_message (or send_targeted_message if addressing a specific agent).',
       '',
       'If a message event has an image_path attribute, the sender attached one or more images. Each path is a local file already downloaded for you (comma-separated if several) — Read it to see the image before responding.',
+      'If it has a file_path attribute, the sender attached one or more non-image files (.md, .csv, .pdf, logs, code — comma-separated if several), already downloaded for you. Read them before responding.',
+      'A failed_attachments attribute lists files the sender attached that could NOT be retrieved. Do not pretend you saw them — say so.',
       '',
-      'To view an image that appears in read_context history but did NOT arrive with an image_path (e.g. an unaddressed image posted earlier), call the download_attachment tool with the attachment\'s mxc (from the read_context attachments field). It writes the file locally and returns the path — then Read that path.',
-      'To send an image (or other file) into the room, call the send_attachment tool with the local file path and an optional caption/thread_id. It posts as a native room attachment and bridged platforms (Slack, Mattermost) receive it as a real file upload.',
+      'To view a file that appears in read_context history but did NOT arrive with an image_path/file_path (e.g. an unaddressed file posted earlier), call the download_attachment tool with the attachment\'s mxc (from the read_context attachments field). It writes the file locally and returns the path — then Read that path.',
+      'To send files into the room, call the send_attachment tool with `path` (one file) or `paths` (several, delivered as ONE message) plus an optional caption/thread_id. Any file type works. They post as native room attachments and bridged platforms (Slack, Mattermost) receive them as real file uploads.',
       '',
       'When you receive a task_delegate event (only delivered if your integration profile has can_accept=true):',
       '1. Call accept_task with the task_id to move it to ongoing.',
@@ -222,8 +224,9 @@ const DOWNLOAD_ATTACHMENT_TOOL = {
   description:
     "Download a room attachment (by its mxc:// URI, as returned in an " +
     "attachment's `mxc` field from read_context) to a local file and return " +
-    "the path. Use this to view an image from history that did not arrive " +
-    'with an image_path. Operates on the currently connected room unless ' +
+    "the path. Works for any file type. Use this to view a file from history " +
+    'that did not arrive with an image_path/file_path. Operates on the ' +
+    'currently connected room unless ' +
     'room_id is given.',
   inputSchema: {
     type: 'object',
@@ -254,10 +257,14 @@ const DOWNLOAD_ATTACHMENT_TOOL = {
 const SEND_ATTACHMENT_TOOL = {
   name: 'send_attachment',
   description:
-    'Send a local file (e.g. an image) into the connected Switch room as an ' +
-    'attachment. It enters the room as a native image/file event and bridged ' +
-    'platforms (Slack, Mattermost) receive it as a real file upload. ' +
-    'Operates on the currently connected room unless room_id is given.',
+    'Send one or more local files of ANY type (image, .md, .csv, .pdf, log, ' +
+    'code) into the connected Switch room as attachments. They enter the room ' +
+    'as native image/file events and bridged platforms (Slack, Mattermost) ' +
+    'receive them as real file uploads. Several files sent in one call arrive ' +
+    'as ONE message carrying all of them. Pass `path` for a single file or ' +
+    '`paths` for several. Oversize or unreadable files fail the whole call — ' +
+    'nothing is sent silently. Operates on the currently connected room unless ' +
+    'room_id is given.',
   inputSchema: {
     type: 'object',
     properties: {
@@ -265,6 +272,13 @@ const SEND_ATTACHMENT_TOOL = {
         type: 'string',
         description: 'Absolute path of the local file to send.',
       },
+      paths: {
+        type: 'array',
+        items: { type: 'string' },
+        description:
+          'Absolute paths of several local files to send as one message. ' +
+          'Use instead of `path` for a multi-attachment message.',
+      },
       caption: {
         type: 'string',
         description: 'Optional text to accompany the attachment.',
@@ -281,7 +295,6 @@ const SEND_ATTACHMENT_TOOL = {
           'Optional Switch room id. Defaults to the currently polling room.',
       },
     },
-    required: ['path'],
   },
 }
 
@@ -334,8 +347,9 @@ async function handleDownloadAttachment(rawArgs: Record<string, unknown>) {
   }
 }
 
-// Minimal extension → mimetype map for common attachment types; anything else
-// goes up as application/octet-stream and enters the room as an m.file.
+// Extension → mimetype map. Anything unlisted goes up as
+// application/octet-stream, which still relays fine — the mapping exists to
+// preserve type fidelity so platforms render/preview the file properly.
 const MIME_BY_EXT: Record<string, string> = {
   '.png': 'image/png',
   '.jpg': 'image/jpeg',
@@ -346,18 +360,54 @@ const MIME_BY_EXT: Record<string, string> = {
   '.bmp': 'image/bmp',
   '.pdf': 'application/pdf',
   '.txt': 'text/plain',
+  '.md': 'text/markdown',
+  '.csv': 'text/csv',
+  '.tsv': 'text/tab-separated-values',
+  '.log': 'text/plain',
+  '.json': 'application/json',
+  '.yaml': 'application/yaml',
+  '.yml': 'application/yaml',
+  '.toml': 'application/toml',
+  '.xml': 'application/xml',
+  '.html': 'text/html',
+  '.css': 'text/css',
+  '.js': 'text/javascript',
+  '.mjs': 'text/javascript',
+  '.ts': 'text/x-typescript',
+  '.tsx': 'text/x-typescript',
+  '.jsx': 'text/javascript',
+  '.py': 'text/x-python',
+  '.rs': 'text/x-rust',
+  '.go': 'text/x-go',
+  '.java': 'text/x-java',
+  '.c': 'text/x-c',
+  '.h': 'text/x-c',
+  '.cpp': 'text/x-c++',
+  '.sh': 'application/x-sh',
+  '.sql': 'application/sql',
+  '.zip': 'application/zip',
+  '.gz': 'application/gzip',
+  '.tar': 'application/x-tar',
 }
 
 async function handleSendAttachment(rawArgs: Record<string, unknown>) {
   const args = rawArgs as {
     path?: string
+    paths?: unknown
     caption?: string
     thread_id?: string
     room_id?: string
   }
-  const filePath = typeof args.path === 'string' ? args.path : ''
-  if (!filePath) {
-    return { isError: true, content: [{ type: 'text', text: 'path is required' }] }
+  const filePaths: string[] = []
+  if (typeof args.path === 'string' && args.path) filePaths.push(args.path)
+  if (Array.isArray(args.paths)) {
+    for (const p of args.paths) if (typeof p === 'string' && p) filePaths.push(p)
+  }
+  if (filePaths.length === 0) {
+    return {
+      isError: true,
+      content: [{ type: 'text', text: 'path (or paths) is required' }],
+    }
   }
   const roomId = args.room_id ?? pollingRoomId
   if (!roomId) {
@@ -372,20 +422,31 @@ async function handleSendAttachment(rawArgs: Record<string, unknown>) {
     }
   }
 
-  let bytes: Buffer
-  try {
-    bytes = fs.readFileSync(filePath)
-  } catch (err) {
-    return {
-      isError: true,
-      content: [{ type: 'text', text: `Cannot read ${filePath}: ${err}` }],
+  // Read every file up front: one unreadable path fails the whole call rather
+  // than posting a partial message.
+  const files: { name: string; bytes: Buffer; mimetype: string }[] = []
+  for (const filePath of filePaths) {
+    let bytes: Buffer
+    try {
+      bytes = fs.readFileSync(filePath)
+    } catch (err) {
+      return {
+        isError: true,
+        content: [{ type: 'text', text: `Cannot read ${filePath}: ${err}` }],
+      }
     }
+    files.push({
+      name: path.basename(filePath),
+      bytes,
+      mimetype:
+        MIME_BY_EXT[path.extname(filePath).toLowerCase()] ?? 'application/octet-stream',
+    })
   }
 
-  const filename = path.basename(filePath)
-  const mimetype = MIME_BY_EXT[path.extname(filePath).toLowerCase()] ?? 'application/octet-stream'
   const form = new FormData()
-  form.append('file', new Blob([bytes], { type: mimetype }), filename)
+  for (const file of files) {
+    form.append('files', new Blob([file.bytes], { type: file.mimetype }), file.name)
+  }
   if (typeof args.caption === 'string' && args.caption) form.append('caption', args.caption)
   if (typeof args.thread_id === 'string' && args.thread_id) form.append('thread_id', args.thread_id)
 
@@ -399,11 +460,14 @@ async function handleSendAttachment(rawArgs: Record<string, unknown>) {
       throw new Error(`HTTP ${resp.status}: ${await resp.text()}`)
     }
     const data = (await resp.json()) as { event_id?: string }
+    const names = files.map((f) => f.name).join(', ')
     return {
       content: [
         {
           type: 'text',
-          text: `Sent ${filename} to the room (event_id: ${data.event_id ?? 'unknown'}).`,
+          text:
+            `Sent ${files.length === 1 ? names : `${files.length} files (${names})`} ` +
+            `to the room (event_id: ${data.event_id ?? 'unknown'}).`,
         },
       ],
     }
@@ -690,15 +754,23 @@ async function handleEvent(event: AgentEvent) {
     // where the agent finishes without replying.
     void setTyping(room_id, true)
 
-    // Materialise image attachments to local files so Claude can Read them.
-    // image_path is the attribute the channel instructions tell Claude to read.
+    // Materialise every attachment to a local file so Claude can Read it,
+    // whatever the type. Images are surfaced as image_path (Claude renders
+    // them); everything else as file_path. A download that fails is reported
+    // as failed_attachment rather than being dropped quietly.
     const imagePaths: string[] = []
+    const filePaths: string[] = []
+    const failedAttachments: string[] = []
     const attachments = msg.attachments ?? []
     for (let i = 0; i < attachments.length; i++) {
       const att = attachments[i]
-      if (!att.mimetype.startsWith('image/')) continue
       const localPath = await downloadAttachment(room_id, att, msg.message_id, i)
-      if (localPath) imagePaths.push(localPath)
+      if (!localPath) {
+        failedAttachments.push(att.filename)
+        continue
+      }
+      if (att.mimetype.startsWith('image/')) imagePaths.push(localPath)
+      else filePaths.push(localPath)
     }
 
     const ts = new Date(msg.timestamp).toISOString()
@@ -714,6 +786,10 @@ async function handleEvent(event: AgentEvent) {
         // Lets an addressed agent reply back into the same thread.
         ...(msg.thread_id ? { thread_id: msg.thread_id } : {}),
         ...(imagePaths.length ? { image_path: imagePaths.join(',') } : {}),
+        ...(filePaths.length ? { file_path: filePaths.join(',') } : {}),
+        ...(failedAttachments.length
+          ? { failed_attachments: failedAttachments.join(',') }
+          : {}),
       },
     )
     return
diff --git a/connectors/claude-code-plugin/skills/switch/SKILL.md b/connectors/claude-code-plugin/skills/switch/SKILL.md
index 055114cb8..461249f65 100644
--- a/connectors/claude-code-plugin/skills/switch/SKILL.md
+++ b/connectors/claude-code-plugin/skills/switch/SKILL.md
@@ -86,30 +86,43 @@ Both `post_message` and `send_targeted_message` accept an optional
 
 ## Sending and receiving attachments
 
-Messages can carry file attachments (images most commonly). Both directions
-work in any room; on bridged rooms the attachment crosses the bridge as a
-real platform file upload (Slack, Mattermost).
-
-- **Receiving:** an addressed message with images arrives with an
-  `image_path` on the notification (already downloaded — Read the path). For
-  an attachment seen in `read_context` history (its `attachments` field),
-  pass its `mxc` to the channel's `download_attachment` tool, then Read the
-  returned path.
-- **Sending:** call the channel's **`send_attachment`** tool with the local
-  file `path`, an optional `caption`, and an optional `thread_id` (same
-  threading semantics as `post_message`). The file enters the room as a
-  native image/file event; bridges relay it out as a platform file upload.
-  Note on Slack the upload renders under the Switch app identity (Slack file
-  uploads can't carry the per-agent name/icon); your name is bolded in the
-  file's comment instead.
+Messages can carry file attachments of **any type** — images, `.md`, `.csv`,
+`.pdf`, logs, code — and a single message can carry **several**. Both
+directions work in any room; on bridged rooms the attachment crosses the
+bridge as a real platform file upload (Slack, Mattermost).
+
+- **Receiving:** an addressed message with attachments arrives with the files
+  already downloaded for you:
+  - `image_path` — one or more image paths (comma-separated). Read them to
+    see the image.
+  - `file_path` — one or more non-image paths (comma-separated: `.md`,
+    `.csv`, `.pdf`, logs, code). Read them too.
+  - `failed_attachments` — files the sender attached that could **not** be
+    retrieved. Say so rather than pretending you saw them.
+
+  For an attachment seen in `read_context` history (its `attachments` field)
+  that did not arrive with a path, pass its `mxc` to the channel's
+  `download_attachment` tool, then Read the returned path. It works for any
+  file type.
+- **Sending:** call the channel's **`send_attachment`** tool with either
+  `path` (one file) or `paths` (several — they arrive as **one** message
+  carrying all of them), plus an optional `caption` and `thread_id` (same
+  threading semantics as `post_message`). Any file type works. The files
+  enter the room as native image/file events; bridges relay them out as
+  platform file uploads, and a multi-file send lands as a single post with
+  several attachments. Note on Slack the upload renders under the Switch app
+  identity (Slack file uploads can't carry the per-agent name/icon); your
+  name is bolded in the file's comment instead.
 - **No channel tool available?** (e.g. a switchdash-managed session where the
   channel process is not running): upload directly to the bridge API —
   `curl -X POST "$SWITCH_API_ENDPOINT/agents/$SWITCH_AGENT_ID/rooms/<room_id>/media"
-  -H "Authorization: Bearer $SWITCH_API_TOKEN" -F "file=@/path/to/image.png"
-  -F "caption=..."` (optional `-F "thread_id=..."`). Returns the posted
-  `event_id`.
+  -H "Authorization: Bearer $SWITCH_API_TOKEN" -F "files=@/path/to/report.md"
+  -F "caption=..."` (optional `-F "thread_id=..."`; repeat `-F "files=@..."`
+  for several files in one message). Returns the posted `event_id`.
 - Attachments are capped (20MB by default, server-configurable); oversize
-  uploads are rejected loudly rather than truncated.
+  uploads are rejected loudly rather than truncated. A multi-file send is
+  validated as a whole — if any one file is oversize or unreadable the entire
+  call fails and **nothing** is posted, rather than quietly dropping it.
 
 **Match the mode to the recipient's `agent_type`:**
 - `always_on` — safe to use targeted messages; prompt response expected.
diff --git a/core/switch_core/attachments.py b/core/switch_core/attachments.py
new file mode 100644
index 000000000..52a32c46c
--- /dev/null
+++ b/core/switch_core/attachments.py
@@ -0,0 +1,41 @@
+"""Shared constants for multi-attachment messages.
+
+Matrix has no native way to put several files on one message: an
+`m.room.message` media event carries exactly one `url`. The proposals that
+would change that (MSC4274 inline media galleries, MSC2881 message
+attachments) are unmerged, so Switch groups at the edges instead — a message
+with n files is sent as n media events that share a group id, and the
+receiving side coalesces them back into a single logical message.
+
+An event without the group key is a group of one, so this is fully backward
+compatible with media events already in a room.
+"""
+
+# Event-content key carrying {"id": str, "index": int, "total": int}.
+ATTACHMENT_GROUP_KEY = "com.switch.attachment_group"
+
+
+def parse_attachment_group(
+    content: dict[str, object],
+) -> tuple[str, int, int] | None:
+    """Extract (group_id, index, total) from event content, or None when the
+    event is not part of a multi-attachment group (or the marker is malformed —
+    a bad marker degrades to an ungrouped attachment rather than raising, so a
+    single odd event can never stall a receiver's buffer)."""
+    raw = content.get(ATTACHMENT_GROUP_KEY)
+    if not isinstance(raw, dict):
+        return None
+    group_id = raw.get("id")
+    index = raw.get("index")
+    total = raw.get("total")
+    if not isinstance(group_id, str) or not group_id:
+        return None
+    # bool is a subclass of int; a bool here means a malformed marker, not an
+    # index of 0/1.
+    if isinstance(index, bool) or isinstance(total, bool):
+        return None
+    if not isinstance(index, int) or not isinstance(total, int):
+        return None
+    if total < 1 or index < 0 or index >= total:
+        return None
+    return group_id, index, total
diff --git a/core/switch_core/bridges/agent/api/handlers.py b/core/switch_core/bridges/agent/api/handlers.py
index edeb6bafb..bd3b89893 100644
--- a/core/switch_core/bridges/agent/api/handlers.py
+++ b/core/switch_core/bridges/agent/api/handlers.py
@@ -428,27 +428,45 @@ async def download_media(
 async def upload_media(
     agent_id: str,
     room_id: str,
-    file: UploadFile,
     agent: Annotated[Agent, Depends(get_agent_from_scope)],
     protocol: Annotated[ProtocolService, Depends(get_protocol)],
+    file: UploadFile | None = None,
+    files: list[UploadFile] | None = None,
     caption: Annotated[str | None, Form()] = None,
     thread_id: Annotated[str | None, Form()] = None,
 ) -> dict[str, object]:
-    """Post an attachment to a room as the agent (multipart upload).
+    """Post one or more attachments to a room as the agent (multipart upload).
 
     The inverse of the GET media endpoint: the local channel (or any connector
-    holding the bridge API token) sends the file's bytes here; they are
-    uploaded to the Matrix media repo and posted to the room as an
-    m.image / m.file event, with optional caption and threading.
+    holding the bridge API token) sends the files' bytes here; they are
+    uploaded to the Matrix media repo and posted to the room as
+    m.image / m.file events, with optional caption and threading.
+
+    Accepts either a single `file` part or repeated `files` parts. Several
+    files become one logical message (they share an attachment-group marker).
+    Validation is all-or-nothing: if any file is empty or oversize the whole
+    request fails with 400 and nothing is posted.
     """
-    data = await file.read()
+    uploads = list(files or [])
+    if file is not None:
+        uploads.insert(0, file)
+    if not uploads:
+        raise HTTPException(
+            status_code=400, detail="no file provided (expected 'file' or 'files')"
+        )
+    payload = [
+        (
+            await upload.read(),
+            upload.filename or "attachment",
+            upload.content_type or "application/octet-stream",
+        )
+        for upload in uploads
+    ]
     try:
         result = await protocol.send_media(
             agent.id,
             room_id,
-            data,
-            filename=file.filename or "attachment",
-            mimetype=file.content_type or "application/octet-stream",
+            payload,
             caption=caption,
             thread_id=thread_id,
         )
diff --git a/core/switch_core/bridges/agent/mcp/server.py b/core/switch_core/bridges/agent/mcp/server.py
index 5a91bd129..77ca37709 100644
--- a/core/switch_core/bridges/agent/mcp/server.py
+++ b/core/switch_core/bridges/agent/mcp/server.py
@@ -489,9 +489,10 @@ async def read_context(
     reply into that thread.
 
     `attachments` is a (usually empty) list of files on the message, each
-    {"filename", "mimetype", "size", "mxc", "msgtype"}. To actually view an
-    image attachment, pass its `mxc` to the channel's `download_attachment`
-    tool, which fetches it to a local file you can read.
+    {"filename", "mimetype", "size", "mxc", "msgtype"}. Any file type can
+    appear, and a message may carry several. To actually view one, pass its
+    `mxc` to the channel's `download_attachment` tool, which fetches it to a
+    local file you can read.
 
     Args:
         limit: Maximum number of messages to scan (default 50), grouped into
diff --git a/core/switch_core/bridges/agent/protocol/service.py b/core/switch_core/bridges/agent/protocol/service.py
index 42df19ba9..e59b08d40 100644
--- a/core/switch_core/bridges/agent/protocol/service.py
+++ b/core/switch_core/bridges/agent/protocol/service.py
@@ -4,6 +4,7 @@
 import logging
 import re
 import secrets
+import uuid
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
@@ -759,30 +760,40 @@ async def send_media(
         self,
         agent_id: str,
         room_id: str,
-        data: bytes,
-        filename: str,
-        mimetype: str,
+        files: list[tuple[bytes, str, str]],
         caption: str | None = None,
         thread_id: str | None = None,
-    ) -> dict[str, str]:
-        """Upload bytes to the Matrix media repository and post them to a room
-        as an m.image / m.file event. Returns {"event_id": ..., "mxc": ...}.
-
-        Membership in `room_id` is required, mirroring download_media. The
-        payload is capped by config.agent_media_max_bytes — an oversize upload
-        raises rather than being truncated. When `caption` is set it becomes
-        the event body (the filename rides in the `filename` field, per the
-        caption convention the bridges already use inbound). When `thread_id`
-        is set the event is posted into that thread (normalised to its root).
+    ) -> dict[str, object]:
+        """Upload one or more files to the Matrix media repository and post them
+        to a room as m.image / m.file events. `files` is a list of
+        (data, filename, mimetype). Returns
+        {"event_id": <first>, "mxc": <first>, "attachments": [{event_id, mxc,
+        filename}, ...]}.
+
+        Membership in `room_id` is required, mirroring download_media. Every
+        file is validated (non-empty, within config.agent_media_max_bytes)
+        BEFORE anything is sent, so a bad file in the batch fails the whole call
+        rather than leaving a half-posted message in the room. When `caption` is
+        set it becomes the body of the first event (the filename rides in the
+        `filename` field, per the caption convention the bridges already use
+        inbound). When `thread_id` is set the events are posted into that thread
+        (normalised to its root).
+
+        With more than one file the events share an attachment-group marker so
+        receivers can coalesce them into one logical message — Matrix itself has
+        no multi-attachment event.
         """
-        if not data:
-            raise ValueError("attachment is empty")
+        if not files:
+            raise ValueError("no attachments provided")
         max_bytes = self.config.agent_media_max_bytes
-        if len(data) > max_bytes:
-            raise ValueError(
-                f"attachment '{filename}' is {len(data)} bytes, over the "
-                f"{max_bytes}-byte limit (AGENT_MEDIA_MAX_BYTES)"
-            )
+        for data, filename, _mimetype in files:
+            if not data:
+                raise ValueError(f"attachment '{filename}' is empty")
+            if len(data) > max_bytes:
+                raise ValueError(
+                    f"attachment '{filename}' is {len(data)} bytes, over the "
+                    f"{max_bytes}-byte limit (AGENT_MEDIA_MAX_BYTES)"
+                )
         room = await self.require_room_member(agent_id, room_id)
         client = self.client_lifecycle.get_by_agent_id(agent_id)
         if client is None:
@@ -792,27 +803,44 @@ async def send_media(
             thread_root_id = await self._resolve_thread_root(
                 client, room.matrix_room_id, thread_id
             )
-        mxc = await client.upload_media(data, mimetype, filename)
-        msgtype = "m.image" if mimetype.startswith("image/") else "m.file"
-        event_id = await client.send_media(
-            room.matrix_room_id,
-            mxc,
-            filename,
-            mimetype,
-            len(data),
-            msgtype=msgtype,
-            caption=caption if caption and caption.strip() else None,
-            thread_root_id=thread_root_id,
-        )
-        if event_id is None:
-            raise ValueError("Failed to send media message")
+
+        total = len(files)
+        group_id = str(uuid.uuid4()) if total > 1 else None
+        posted: list[dict[str, str]] = []
+        for index, (data, filename, mimetype) in enumerate(files):
+            mxc = await client.upload_media(data, mimetype, filename)
+            msgtype = "m.image" if mimetype.startswith("image/") else "m.file"
+            event_id = await client.send_media(
+                room.matrix_room_id,
+                mxc,
+                filename,
+                mimetype,
+                len(data),
+                msgtype=msgtype,
+                caption=(
+                    caption if index == 0 and caption and caption.strip() else None
+                ),
+                thread_root_id=thread_root_id,
+                group=(
+                    {"id": group_id, "index": index, "total": total}
+                    if group_id is not None
+                    else None
+                ),
+            )
+            if event_id is None:
+                raise ValueError(f"Failed to send media message for '{filename}'")
+            posted.append({"event_id": event_id, "mxc": mxc, "filename": filename})
         try:
             await self.set_typing(agent_id, room_id, False)
         except Exception:
             logger.warning(
                 "Failed to clear typing indicator for room %s", room_id, exc_info=True
             )
-        return {"event_id": event_id, "mxc": mxc}
+        return {
+            "event_id": posted[0]["event_id"],
+            "mxc": posted[0]["mxc"],
+            "attachments": posted,
+        }
 
     async def send_targeted_message(
         self,
diff --git a/core/switch_core/bridges/collaboration/adapter.py b/core/switch_core/bridges/collaboration/adapter.py
index ed6188aeb..a922f6873 100644
--- a/core/switch_core/bridges/collaboration/adapter.py
+++ b/core/switch_core/bridges/collaboration/adapter.py
@@ -10,6 +10,7 @@
     InboundCommand,
     InboundMessage,
     InboundUserJoin,
+    OutboundAttachment,
 )
 
 
@@ -22,6 +23,14 @@ def __init__(self) -> None:
         )
         self._on_user_joined: Callable[[InboundUserJoin], Awaitable[None]] | None = None
         self._on_app_joined: Callable[[InboundAppJoin], Awaitable[None]] | None = None
+        # Inbound attachment size ceiling, set by the lifecycle service from
+        # config.agent_media_max_bytes. Adapters check a platform-reported file
+        # size against this before downloading so an oversize file is rejected
+        # loudly instead of being pulled down and discarded.
+        self._max_attachment_bytes = 20 * 1024 * 1024
+
+    def set_max_attachment_bytes(self, max_bytes: int) -> None:
+        self._max_attachment_bytes = max_bytes
 
     @abstractmethod
     async def start(
@@ -91,10 +100,10 @@ async def send_attachment(
         """Relay a file attachment to the external channel as `sender_name`,
         returning the platform message ref (like send_message).
 
-        Platforms with a file API override this to upload the bytes natively.
-        This default is the disclosed degradation for adapters without native
-        support: a text message that names the attachment instead of silently
-        dropping it.
+        Platforms with a file API override this to upload the bytes natively,
+        for any mimetype. This default is the disclosed degradation for adapters
+        without native support: a text message that names the attachment instead
+        of silently dropping it.
         """
         note = f"_sent an attachment that couldn't be relayed: {filename}_"
         body = f"{caption}\n{note}" if caption else note
@@ -102,6 +111,35 @@ async def send_attachment(
             channel_id, sender_name, self.translate_outbound(body), thread_root_id
         )
 
+    async def send_attachments(
+        self,
+        channel_id: str,
+        sender_name: str,
+        files: list[OutboundAttachment],
+        caption: str | None = None,
+        thread_root_id: str | None = None,
+    ) -> str | None:
+        """Relay several files as a SINGLE post on the external platform.
+
+        Adapters whose platform can attach multiple files to one message
+        override this. The default posts them one at a time — correct, but the
+        files land as separate messages rather than one.
+        """
+        message_ref: str | None = None
+        for index, file in enumerate(files):
+            ref = await self.send_attachment(
+                channel_id,
+                sender_name,
+                file.filename,
+                file.mimetype,
+                file.data,
+                caption=caption if index == 0 else None,
+                thread_root_id=thread_root_id,
+            )
+            if index == 0:
+                message_ref = ref
+        return message_ref
+
     @abstractmethod
     async def update_message(
         self, channel_id: str, message_ref: str, new_content: str
diff --git a/core/switch_core/bridges/collaboration/bridge_core.py b/core/switch_core/bridges/collaboration/bridge_core.py
index 397f35b3b..9273a4e25 100644
--- a/core/switch_core/bridges/collaboration/bridge_core.py
+++ b/core/switch_core/bridges/collaboration/bridge_core.py
@@ -3,6 +3,8 @@
 import asyncio
 import logging
 import re
+import uuid
+from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any
 
 from nio import (
@@ -16,6 +18,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from switch_core.aliases import AliasError, validate_alias_format
+from switch_core.attachments import parse_attachment_group
 from switch_core.bridges.collaboration.adapter import CollaborationAdapter
 from switch_core.bridges.collaboration.models import (
     ChannelType,
@@ -24,6 +27,7 @@
     InboundCommand,
     InboundMessage,
     InboundUserJoin,
+    OutboundAttachment,
 )
 from switch_core.clients.admin_messages import ADMIN_MARKER, AdminMessageType
 from switch_core.clients.client_base import ClientBase, ClientConfig
@@ -44,6 +48,21 @@
 
 logger = logging.getLogger(__name__)
 
+# How long to hold an incomplete outbound attachment group before relaying the
+# parts that arrived, flagged as incomplete (see _schedule_outbound_group_flush).
+OUTBOUND_GROUP_TIMEOUT_SECONDS = 5.0
+
+
+@dataclass
+class _PendingOutboundGroup:
+    """Files of a multi-attachment message seen so far, keyed by group index."""
+
+    total: int
+    parts: dict[int, OutboundAttachment] = field(default_factory=dict)
+    caption: str | None = None
+    first_event_id: str | None = None
+
+
 _LOBBY_DEPRECATION_NOTICE = (
     "👋 This isn't where you talk to agents — direct messages to the Switch "
     "app aren't routed to anyone. Head to a channel and @-mention an agent "
@@ -112,6 +131,11 @@ def __init__(
         # spawn a duplicate room. Handlers skip adoption while a channel is in
         # this set. See begin_provisioning / end_provisioning.
         self._provisioning_channels: set[str] = set()
+        # Outbound multi-attachment messages still assembling, with their
+        # safety-net timers. Cleared on completion or timeout so a group that
+        # never completes cannot leak.
+        self._outbound_groups: dict[str, _PendingOutboundGroup] = {}
+        self._outbound_group_timers: dict[str, asyncio.TimerHandle] = {}
 
     @property
     def adapter(self) -> CollaborationAdapter:
@@ -382,6 +406,16 @@ async def _handle_inbound_message(self, msg: InboundMessage) -> None:
                     matrix_room_id,
                 )
 
+        # An attachment the platform offered but we could not relay must be
+        # visible in the room, not swallowed. Append it to the message body so
+        # both the agent and the humans see that a file went missing.
+        if msg.attachment_failures:
+            notes = "\n".join(
+                f"_attachment not relayed: {failure.filename} — {failure.reason}_"
+                for failure in msg.attachment_failures
+            )
+            content = f"{content}\n{notes}" if content.strip() else notes
+
         if not msg.attachments:
             event_id = await puppet.send_message(
                 matrix_room_id,
@@ -404,6 +438,11 @@ async def _handle_inbound_message(self, msg: InboundMessage) -> None:
         # event stands in for the post for threading / correlation purposes.
         caption = content if content.strip() else None
         first_event_id: str | None = None
+        total = len(msg.attachments)
+        # A platform post hands us all its files at once, so the group is known
+        # up front — no waiting on the receiving side to learn how many to
+        # expect. Matrix carries them as `total` events sharing this id.
+        group_id = str(uuid.uuid4()) if total > 1 else None
         for index, attachment in enumerate(msg.attachments):
             mxc = await puppet.upload_media(
                 attachment.data, attachment.mimetype, attachment.filename
@@ -420,6 +459,11 @@ async def _handle_inbound_message(self, msg: InboundMessage) -> None:
                 msgtype=msgtype,
                 caption=caption if index == 0 else None,
                 thread_root_id=thread_root_id,
+                group=(
+                    {"id": group_id, "index": index, "total": total}
+                    if group_id is not None
+                    else None
+                ),
             )
             if index == 0:
                 first_event_id = event_id
@@ -977,10 +1021,14 @@ async def handle_outbound_media(
         Mirrors handle_outbound_message: puppet media is skipped (it originated
         on the platform), the caption convention is unpacked (a `filename` key
         means the body is a caption), and the relayed post is recorded in the
-        message map so replies thread both ways. Images relay natively via the
-        adapter's send_attachment; other file types get a disclosed text notice
-        (matching the images-only inbound path) rather than a silent drop.
-        `client` is the bridge's Matrix client, used to fetch the media bytes.
+        message map so replies thread both ways. Any file type relays natively
+        via the adapter; a file whose bytes can't be fetched or that exceeds the
+        relay cap gets a disclosed text notice rather than a silent drop.
+
+        A message carrying several files arrives as several Matrix events
+        sharing a group marker; they are buffered here and relayed as ONE
+        platform post. `client` is the bridge's Matrix client, used to fetch the
+        media bytes.
         """
         logger.debug(
             "[BRIDGE-OUT] matrix media event from=%s room=%s body=%s",
@@ -1026,8 +1074,17 @@ async def handle_outbound_media(
         )
 
         message_ref: str | None
-        if event_content.get("msgtype") != "m.image":
-            note = f"_sent a file that isn't relayed yet: {filename}_"
+        data = await self._download_matrix_media(client, event.url, filename)
+        if data is None or len(data) > self._max_attachment_bytes:
+            if data is not None:
+                logger.warning(
+                    "[BRIDGE-OUT] attachment %s is %d bytes, over the "
+                    "%d-byte relay cap",
+                    filename,
+                    len(data),
+                    self._max_attachment_bytes,
+                )
+            note = f"_sent an attachment that couldn't be relayed: {filename}_"
             body = f"{caption}\n{note}" if caption else note
             message_ref = await self._adapter.send_message(
                 channel_id,
@@ -1036,34 +1093,41 @@ async def handle_outbound_media(
                 thread_root_id=thread_root_ref,
             )
         else:
-            data = await self._download_matrix_media(client, event.url, filename)
-            if data is None or len(data) > self._max_attachment_bytes:
-                if data is not None:
-                    logger.warning(
-                        "[BRIDGE-OUT] attachment %s is %d bytes, over the "
-                        "%d-byte relay cap",
-                        filename,
-                        len(data),
-                        self._max_attachment_bytes,
-                    )
-                note = f"_sent an attachment that couldn't be relayed: {filename}_"
-                body = f"{caption}\n{note}" if caption else note
-                message_ref = await self._adapter.send_message(
-                    channel_id,
-                    sender_name,
-                    self._adapter.translate_outbound(body),
-                    thread_root_id=thread_root_ref,
+            # Part of a multi-file message? Hold it until the whole group has
+            # arrived, then relay all of it as one platform post.
+            group = parse_attachment_group(event_content)
+            if group is not None:
+                group_id, index, total = group
+                pending = self._outbound_groups.setdefault(
+                    group_id, _PendingOutboundGroup(total=total)
                 )
-            else:
-                message_ref = await self._adapter.send_attachment(
-                    channel_id,
-                    sender_name,
-                    filename,
-                    mimetype,
-                    data,
-                    caption=caption,
-                    thread_root_id=thread_root_ref,
+                pending.parts[index] = OutboundAttachment(
+                    filename=filename, mimetype=mimetype, data=data
+                )
+                if index == 0:
+                    pending.caption = caption
+                    pending.first_event_id = event.event_id
+                if len(pending.parts) < total:
+                    self._schedule_outbound_group_flush(
+                        group_id, channel_id, sender_name, thread_root_ref
+                    )
+                    return
+                self._cancel_outbound_group_flush(group_id)
+                self._outbound_groups.pop(group_id, None)
+                await self._relay_outbound_group(
+                    pending, channel_id, sender_name, thread_root_ref
                 )
+                return
+
+            message_ref = await self._adapter.send_attachment(
+                channel_id,
+                sender_name,
+                filename,
+                mimetype,
+                data,
+                caption=caption,
+                thread_root_id=thread_root_ref,
+            )
 
         if message_ref is not None:
             await self._record_message_map(
@@ -1072,6 +1136,88 @@ async def handle_outbound_media(
                 external_post_id=message_ref,
             )
 
+    def _schedule_outbound_group_flush(
+        self,
+        group_id: str,
+        channel_id: str,
+        sender_name: str,
+        thread_root_ref: str | None,
+    ) -> None:
+        """Arm the safety net for an incomplete outbound attachment group.
+
+        A group normally completes immediately — the sender posts its events
+        back-to-back. This timer guarantees that a batch that never completes
+        still reaches the platform, flagged, instead of being held forever.
+
+        Armed once per group, NOT re-armed per part, so the deadline bounds the
+        whole group rather than the gap between parts.
+        """
+        if group_id in self._outbound_group_timers:
+            return
+        self._outbound_group_timers[group_id] = asyncio.get_running_loop().call_later(
+            OUTBOUND_GROUP_TIMEOUT_SECONDS,
+            lambda: asyncio.create_task(
+                self._flush_incomplete_outbound_group(
+                    group_id, channel_id, sender_name, thread_root_ref
+                )
+            ),
+        )
+
+    def _cancel_outbound_group_flush(self, group_id: str) -> None:
+        timer = self._outbound_group_timers.pop(group_id, None)
+        if timer is not None:
+            timer.cancel()
+
+    async def _flush_incomplete_outbound_group(
+        self,
+        group_id: str,
+        channel_id: str,
+        sender_name: str,
+        thread_root_ref: str | None,
+    ) -> None:
+        self._outbound_group_timers.pop(group_id, None)
+        pending = self._outbound_groups.pop(group_id, None)
+        if pending is None:
+            return
+        received = len(pending.parts)
+        logger.error(
+            "[BRIDGE-OUT] attachment group %s incomplete: %d of %d parts arrived "
+            "within %ss; relaying what arrived",
+            group_id,
+            received,
+            pending.total,
+            OUTBOUND_GROUP_TIMEOUT_SECONDS,
+        )
+        notice = (
+            f"_incomplete attachment group: relaying {received} of "
+            f"{pending.total} files_"
+        )
+        pending.caption = f"{pending.caption}\n{notice}" if pending.caption else notice
+        await self._relay_outbound_group(
+            pending, channel_id, sender_name, thread_root_ref
+        )
+
+    async def _relay_outbound_group(
+        self,
+        pending: _PendingOutboundGroup,
+        channel_id: str,
+        sender_name: str,
+        thread_root_ref: str | None,
+    ) -> None:
+        message_ref = await self._adapter.send_attachments(
+            channel_id,
+            sender_name,
+            [pending.parts[i] for i in sorted(pending.parts)],
+            caption=pending.caption,
+            thread_root_id=thread_root_ref,
+        )
+        if message_ref is not None and pending.first_event_id is not None:
+            await self._record_message_map(
+                external_channel_id=channel_id,
+                matrix_event_id=pending.first_event_id,
+                external_post_id=message_ref,
+            )
+
     async def _download_matrix_media(
         self, client: ClientBase[Any], mxc: str | None, filename: str
     ) -> bytes | None:
diff --git a/core/switch_core/bridges/collaboration/discord/adapter.py b/core/switch_core/bridges/collaboration/discord/adapter.py
index 29525bb15..4189cf63e 100644
--- a/core/switch_core/bridges/collaboration/discord/adapter.py
+++ b/core/switch_core/bridges/collaboration/discord/adapter.py
@@ -13,6 +13,7 @@
 from switch_core.bridges.collaboration.adapter import CollaborationAdapter
 from switch_core.bridges.collaboration.models import (
     Attachment,
+    AttachmentFailure,
     BridgeConnectionConfig,
     ChannelType,
     InboundAgentJoin,
@@ -721,7 +722,7 @@ async def _handle_message(self, message: Any) -> None:
         if self._on_message is None:
             return
 
-        attachments = await self._fetch_image_attachments(
+        attachments, attachment_failures = await self._fetch_attachments(
             getattr(message, "attachments", []) or []
         )
         self_mention = (
@@ -739,39 +740,67 @@ async def _handle_message(self, message: Any) -> None:
                 root_id=root_id,
                 channel_name=channel_name,
                 attachments=attachments,
+                attachment_failures=attachment_failures,
                 self_mention_token=str(self._bot_user_id) if self_mention else None,
             )
         )
 
     # ── Attachments ──────────────────────────────────────────────────────────
 
-    async def _fetch_image_attachments(self, files: list[Any]) -> list[Attachment]:
-        """Download image attachments from a Discord message.
+    async def _fetch_attachments(
+        self, files: list[Any]
+    ) -> tuple[list[Attachment], list[AttachmentFailure]]:
+        """Download every attachment from a Discord message, whatever the type.
 
-        Non-image files are skipped (images-only for now, mirroring Slack and
-        Mattermost). A single file that fails to download is logged and
-        skipped rather than dropping the whole message.
+        Returns the downloaded attachments and, separately, the ones that could
+        not be relayed (oversize, download failure) so the bridge can disclose
+        them in the room rather than dropping them silently.
         """
         attachments: list[Attachment] = []
+        failures: list[AttachmentFailure] = []
         for file in files:
-            mimetype = str(getattr(file, "content_type", "") or "")
-            if not mimetype.startswith("image/"):
-                logger.debug(
-                    "Skipping non-image Discord attachment %s (%s)",
-                    getattr(file, "id", "?"),
-                    mimetype or "unknown",
+            mimetype = (
+                str(getattr(file, "content_type", "") or "")
+                or "application/octet-stream"
+            )
+            filename = str(getattr(file, "filename", "") or "file")
+            size = getattr(file, "size", None)
+            if isinstance(size, int) and size > self._max_attachment_bytes:
+                logger.warning(
+                    "Discord attachment %s is %d bytes, over the %d cap",
+                    filename,
+                    size,
+                    self._max_attachment_bytes,
+                )
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename,
+                        reason=f"{size} bytes exceeds the {self._max_attachment_bytes} byte limit",
+                    )
                 )
                 continue
-            filename = str(getattr(file, "filename", "") or "image")
             try:
                 data = await file.read()
-            except Exception:
+            except Exception as exc:
                 logger.exception("Failed to download Discord attachment %s", filename)
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename, reason=f"download failed: {exc}"
+                    )
+                )
+                continue
+            if len(data) > self._max_attachment_bytes:
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename,
+                        reason=f"{len(data)} bytes exceeds the {self._max_attachment_bytes} byte limit",
+                    )
+                )
                 continue
             attachments.append(
                 Attachment(filename=filename, mimetype=mimetype, data=data)
             )
-        return attachments
+        return attachments, failures
 
     # ── Webhooks & channels ──────────────────────────────────────────────────
 
diff --git a/core/switch_core/bridges/collaboration/lifecycle_service.py b/core/switch_core/bridges/collaboration/lifecycle_service.py
index c1f0d7fe3..745b2edb0 100644
--- a/core/switch_core/bridges/collaboration/lifecycle_service.py
+++ b/core/switch_core/bridges/collaboration/lifecycle_service.py
@@ -156,6 +156,7 @@ async def start(self, bridge_id: str) -> None:
         adapter.set_service_url_persister(
             lambda service_url: self._persist_service_url(bridge_id, service_url)
         )
+        adapter.set_max_attachment_bytes(self._config.agent_media_max_bytes)
 
         async with self._session_factory() as session:
             bridge_client_record = await self._client_store.get(
diff --git a/core/switch_core/bridges/collaboration/mattermost/adapter.py b/core/switch_core/bridges/collaboration/mattermost/adapter.py
index 06df5d550..c09ab128d 100644
--- a/core/switch_core/bridges/collaboration/mattermost/adapter.py
+++ b/core/switch_core/bridges/collaboration/mattermost/adapter.py
@@ -20,6 +20,7 @@
 from switch_core.bridges.collaboration.adapter import CollaborationAdapter
 from switch_core.bridges.collaboration.models import (
     Attachment,
+    AttachmentFailure,
     BridgeConnectionConfig,
     ChannelType,
     InboundAgentJoin,
@@ -27,6 +28,7 @@
     InboundCommand,
     InboundMessage,
     InboundUserJoin,
+    OutboundAttachment,
 )
 
 logger = logging.getLogger(__name__)
@@ -272,6 +274,87 @@ def _upload() -> list[str]:
             )
             return None
 
+    async def send_attachments(
+        self,
+        channel_id: str,
+        sender_name: str,
+        files: list[OutboundAttachment],
+        caption: str | None = None,
+        thread_root_id: str | None = None,
+    ) -> str | None:
+        """Upload several files and attach them all to ONE post — Mattermost
+        posts natively carry a list of file ids."""
+        if not files:
+            return None
+        if len(files) == 1:
+            file = files[0]
+            return await self.send_attachment(
+                channel_id,
+                sender_name,
+                file.filename,
+                file.mimetype,
+                file.data,
+                caption,
+                thread_root_id,
+            )
+        loop = self._main_loop
+        if loop is None:
+            logger.error("Cannot send attachments: event loop not initialized")
+            return None
+        driver = self._bot_drivers.get(sender_name)
+        if not driver:
+            logger.error("No Mattermost driver found for sender '%s'", sender_name)
+            if not self._admin_driver:
+                return None
+            driver = self._admin_driver
+
+        def _upload_all() -> list[str]:
+            ids: list[str] = []
+            for file in files:
+                result = driver.files.upload_file(
+                    channel_id,
+                    files={
+                        "files": (file.filename, io.BytesIO(file.data), file.mimetype)
+                    },
+                )
+                ids.extend(info["id"] for info in result.get("file_infos", []))
+            return ids
+
+        try:
+            file_ids = await loop.run_in_executor(None, _upload_all)
+        except Exception as e:
+            logger.error(
+                "Failed to upload %d attachments to Mattermost channel %s: %s",
+                len(files),
+                channel_id,
+                e,
+            )
+            file_ids = []
+        if not file_ids:
+            return await super().send_attachments(
+                channel_id, sender_name, files, caption, thread_root_id
+            )
+
+        post: dict[str, object] = {
+            "channel_id": channel_id,
+            "message": self.translate_outbound(caption) if caption else "",
+            "file_ids": file_ids,
+        }
+        if thread_root_id is not None:
+            post["root_id"] = thread_root_id
+        try:
+            result = await loop.run_in_executor(None, driver.posts.create_post, post)
+            post_id: str = result.get("id", "")
+            return post_id or None
+        except Exception as e:
+            logger.error(
+                "Failed to post %d attachments to Mattermost channel %s: %s",
+                len(files),
+                channel_id,
+                e,
+            )
+            return None
+
     async def _create_post(
         self,
         driver: Driver,
@@ -1015,7 +1098,7 @@ async def _ws_handler(self, event_data: str, agent_name: str) -> None:
             return
 
         if self._on_message:
-            attachments = await self._fetch_image_attachments(
+            attachments, attachment_failures = await self._fetch_attachments(
                 post.get("file_ids", []), ws_loop
             )
             inbound = InboundMessage(
@@ -1029,50 +1112,74 @@ async def _ws_handler(self, event_data: str, agent_name: str) -> None:
                 agent_name=agent_name,
                 channel_name=channel_name,
                 attachments=attachments,
+                attachment_failures=attachment_failures,
             )
             coro = self._on_message(inbound)
             self._dispatch(coro, loop)  # type: ignore[arg-type]
 
-    async def _fetch_image_attachments(
+    async def _fetch_attachments(
         self, file_ids: list[str], loop: asyncio.AbstractEventLoop
-    ) -> list[Attachment]:
-        """Download image attachments for a post's file ids.
+    ) -> tuple[list[Attachment], list[AttachmentFailure]]:
+        """Download every attachment for a post's file ids, whatever the type.
 
-        Non-image files are skipped (images-only for now). Metadata and bytes
-        are fetched via the admin driver off the websocket loop. A single file
-        that fails to download is logged and skipped rather than dropping the
-        whole message.
+        Metadata and bytes are fetched via the admin driver off the websocket
+        loop. Returns the downloaded attachments and, separately, the ones that
+        could not be relayed (oversize, download failure) so the bridge can
+        disclose them in the room rather than dropping them silently.
         """
         if not file_ids or self._admin_driver is None:
-            return []
+            return [], []
 
         driver = self._admin_driver
         attachments: list[Attachment] = []
+        failures: list[AttachmentFailure] = []
         for file_id in file_ids:
+            filename = file_id
             try:
                 meta = await loop.run_in_executor(
                     None, driver.files.get_file_metadata, file_id
                 )
-                mimetype = str(meta.get("mime_type", ""))
-                if not mimetype.startswith("image/"):
-                    logger.debug(
-                        "[MM-INBOUND] skipping non-image attachment %s (%s)",
-                        file_id,
-                        mimetype or "unknown",
+                mimetype = str(meta.get("mime_type", "")) or "application/octet-stream"
+                filename = str(meta.get("name", file_id))
+                size = meta.get("size")
+                if isinstance(size, int) and size > self._max_attachment_bytes:
+                    logger.warning(
+                        "[MM-INBOUND] attachment %s is %d bytes, over the %d cap",
+                        filename,
+                        size,
+                        self._max_attachment_bytes,
+                    )
+                    failures.append(
+                        AttachmentFailure(
+                            filename=filename,
+                            reason=f"{size} bytes exceeds the {self._max_attachment_bytes} byte limit",
+                        )
                     )
                     continue
-                filename = str(meta.get("name", file_id))
                 resp = await loop.run_in_executor(None, driver.files.get_file, file_id)
                 data: bytes = resp.content
-            except Exception:
+            except Exception as exc:
                 logger.exception(
                     "[MM-INBOUND] failed to download attachment %s", file_id
                 )
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename, reason=f"download failed: {exc}"
+                    )
+                )
+                continue
+            if len(data) > self._max_attachment_bytes:
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename,
+                        reason=f"{len(data)} bytes exceeds the {self._max_attachment_bytes} byte limit",
+                    )
+                )
                 continue
             attachments.append(
                 Attachment(filename=filename, mimetype=mimetype, data=data)
             )
-        return attachments
+        return attachments, failures
 
     async def _handle_user_added(self, event: dict[str, Any]) -> None:
         data: dict[str, Any] = event.get("data", {})
diff --git a/core/switch_core/bridges/collaboration/models.py b/core/switch_core/bridges/collaboration/models.py
index e235e9dcb..70b47e81f 100644
--- a/core/switch_core/bridges/collaboration/models.py
+++ b/core/switch_core/bridges/collaboration/models.py
@@ -6,7 +6,7 @@
 
 
 class Attachment(BaseModel):
-    """An inbound file attachment (image only, for now) with its raw bytes.
+    """An inbound file attachment of any type, with its raw bytes.
 
     `data` holds the downloaded file content; the bridge uploads it to the
     Matrix media repository and discards the bytes afterwards.
@@ -17,6 +17,25 @@ class Attachment(BaseModel):
     data: bytes
 
 
+class OutboundAttachment(BaseModel):
+    """A file on its way out to an external platform, with its raw bytes."""
+
+    filename: str
+    mimetype: str
+    data: bytes
+
+
+class AttachmentFailure(BaseModel):
+    """An inbound attachment that could not be relayed, and why.
+
+    Carried alongside the successful attachments so the bridge can disclose the
+    failure in the room. An attachment that fails must never vanish silently.
+    """
+
+    filename: str
+    reason: str
+
+
 class InboundMessage(BaseModel):
     channel_id: str
     channel_type: ChannelType
@@ -30,6 +49,9 @@ class InboundMessage(BaseModel):
     agent_name: str | None = None
     channel_name: str | None = None
     attachments: list[Attachment] = []
+    # Attachments the platform offered but the bridge could not relay (oversize,
+    # download failure). Disclosed in the room rather than dropped.
+    attachment_failures: list[AttachmentFailure] = []
     # The bridge's own bot handle when this message @-mentions the bridge bot
     # itself (e.g. Slack's "Agent Switch" app). None when the bot was not
     # tagged. Lets the bridge guide users who tag the app instead of an agent.
diff --git a/core/switch_core/bridges/collaboration/slack/adapter.py b/core/switch_core/bridges/collaboration/slack/adapter.py
index bdd9b8336..4f8ee5fda 100644
--- a/core/switch_core/bridges/collaboration/slack/adapter.py
+++ b/core/switch_core/bridges/collaboration/slack/adapter.py
@@ -17,6 +17,7 @@
 from switch_core.bridges.collaboration.adapter import CollaborationAdapter
 from switch_core.bridges.collaboration.models import (
     Attachment,
+    AttachmentFailure,
     BridgeConnectionConfig,
     ChannelType,
     InboundAgentJoin,
@@ -24,6 +25,7 @@
     InboundCommand,
     InboundMessage,
     InboundUserJoin,
+    OutboundAttachment,
 )
 
 logger = logging.getLogger(__name__)
@@ -282,6 +284,71 @@ async def send_attachment(
         ts = self._extract_share_ts(result.get("files") or [], channel_id)
         return f"{channel_id}:{ts}" if ts else None
 
+    async def send_attachments(
+        self,
+        channel_id: str,
+        sender_name: str,
+        files: list[OutboundAttachment],
+        caption: str | None = None,
+        thread_root_id: str | None = None,
+    ) -> str | None:
+        """Upload several files as ONE Slack post via files_upload_v2's
+        `file_uploads` list, so N files share a single message and comment."""
+        if not files:
+            return None
+        if len(files) == 1:
+            file = files[0]
+            return await self.send_attachment(
+                channel_id,
+                sender_name,
+                file.filename,
+                file.mimetype,
+                file.data,
+                caption,
+                thread_root_id,
+            )
+        if not self._web_client:
+            logger.error("Cannot send attachments: Slack client not connected")
+            return None
+
+        thread_ts: str | None = None
+        if thread_root_id:
+            thread_ts = (
+                self._parse_message_ref(thread_root_id)[1]
+                if ":" in thread_root_id
+                else thread_root_id
+            )
+
+        names = ", ".join(f"`{file.filename}`" for file in files)
+        comment = (
+            f"*{sender_name}*: {self.translate_outbound(caption)}"
+            if caption
+            else f"*{sender_name}* sent {names}"
+        )
+
+        try:
+            result = await self._web_client.files_upload_v2(
+                channel=channel_id,
+                file_uploads=[
+                    {"file": file.data, "filename": file.filename} for file in files
+                ],
+                initial_comment=comment,
+                thread_ts=thread_ts,
+            )
+        except SlackApiError as e:
+            logger.error(
+                "Failed to upload %d attachments to Slack channel %s: %s",
+                len(files),
+                channel_id,
+                e,
+            )
+            return await super().send_attachments(
+                channel_id, sender_name, files, caption, thread_root_id
+            )
+
+        ts = self._extract_share_ts(result.get("files") or [], channel_id)
+        return f"{channel_id}:{ts}" if ts else None
+
     @staticmethod
     def _extract_share_ts(
         files: list[dict[str, object]], channel_id: str
@@ -775,7 +842,7 @@ async def _handle_message_event(self, event: dict[str, object]) -> None:
             return
 
         if self._on_message:
-            attachments = await self._fetch_image_attachments(
+            attachments, attachment_failures = await self._fetch_attachments(
                 event.get("files", []) or []  # type: ignore[arg-type]
             )
             self_mention = bool(self._bot_user_id) and f"<@{self._bot_user_id}>" in text
@@ -790,6 +857,7 @@ async def _handle_message_event(self, event: dict[str, object]) -> None:
                     root_id=root_id,
                     channel_name=channel_name,
                     attachments=attachments,
+                    attachment_failures=attachment_failures,
                     self_mention_token=self._bot_user_id if self_mention else None,
                 )
             )
@@ -873,41 +941,70 @@ def _to_channel_type(slack_type: str) -> ChannelType:
 
     # ── Attachments ──────────────────────────────────────────────────────────
 
-    async def _fetch_image_attachments(
+    async def _fetch_attachments(
         self, files: list[dict[str, object]]
-    ) -> list[Attachment]:
-        """Download image attachments from a Slack message's `files`.
-
-        Non-image files are skipped (images-only for now, mirroring Mattermost).
-        A single file that fails to download is logged and skipped rather than
-        dropping the whole message.
+    ) -> tuple[list[Attachment], list[AttachmentFailure]]:
+        """Download every attachment from a Slack message's `files`, whatever
+        the type.
+
+        Returns the successfully downloaded attachments and, separately, the
+        ones that could not be relayed. A file that is oversize or fails to
+        download is reported as a failure so the bridge can disclose it in the
+        room — never dropped silently.
         """
         attachments: list[Attachment] = []
+        failures: list[AttachmentFailure] = []
         for file in files:
-            mimetype = str(file.get("mimetype", ""))
-            if not mimetype.startswith("image/"):
-                logger.debug(
-                    "Skipping non-image Slack attachment %s (%s)",
-                    file.get("id", "?"),
-                    mimetype or "unknown",
-                )
-                continue
+            mimetype = str(file.get("mimetype", "")) or "application/octet-stream"
+            filename = str(file.get("name") or file.get("id") or "file")
             url = str(file.get("url_private_download") or file.get("url_private") or "")
             if not url:
                 logger.warning(
-                    "Slack image attachment %s has no download url", file.get("id", "?")
+                    "Slack attachment %s has no download url", file.get("id", "?")
+                )
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename, reason="no download url from Slack"
+                    )
+                )
+                continue
+            size = file.get("size")
+            if isinstance(size, int) and size > self._max_attachment_bytes:
+                logger.warning(
+                    "Slack attachment %s is %d bytes, over the %d cap",
+                    filename,
+                    size,
+                    self._max_attachment_bytes,
+                )
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename,
+                        reason=f"{size} bytes exceeds the {self._max_attachment_bytes} byte limit",
+                    )
                 )
                 continue
-            filename = str(file.get("name") or file.get("id") or "image")
             try:
                 data = await self._download_file(url)
-            except Exception:
+            except Exception as exc:
                 logger.exception("Failed to download Slack attachment %s", filename)
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename, reason=f"download failed: {exc}"
+                    )
+                )
+                continue
+            if len(data) > self._max_attachment_bytes:
+                failures.append(
+                    AttachmentFailure(
+                        filename=filename,
+                        reason=f"{len(data)} bytes exceeds the {self._max_attachment_bytes} byte limit",
+                    )
+                )
                 continue
             attachments.append(
                 Attachment(filename=filename, mimetype=mimetype, data=data)
             )
-        return attachments
+        return attachments, failures
 
     async def _download_file(self, url: str) -> bytes:
         """Fetch a Slack private file URL with the bot token, returning bytes."""
diff --git a/core/switch_core/clients/agent_client.py b/core/switch_core/clients/agent_client.py
index a64ccc9c2..72c48151a 100644
--- a/core/switch_core/clients/agent_client.py
+++ b/core/switch_core/clients/agent_client.py
@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 import random
 import re
+from dataclasses import dataclass, field
 from typing import Literal
 
 from nio import (
@@ -15,6 +17,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from switch_core.addressing import SenderKind, can_address, parse_policy
+from switch_core.attachments import parse_attachment_group
 from switch_core.bridges.agent.commands import (
     AGENT_GREETINGS,
     COMMANDS_BY_NAME,
@@ -85,6 +88,26 @@
 # m.room.message keeps the reply rendering normally for humans.
 AUTO_REPLY_FLAG = "com.switch.auto_reply"
 
+# How long to hold an incomplete multi-attachment group before delivering the
+# parts that did arrive, flagged as incomplete. Groups normally complete in
+# milliseconds (one sender, back-to-back events); this is a safety net so a
+# broken batch surfaces rather than being buffered indefinitely.
+ATTACHMENT_GROUP_TIMEOUT_SECONDS = 5.0
+
+
+@dataclass
+class _PendingAttachmentGroup:
+    """Parts of a multi-attachment message seen so far, keyed by group index."""
+
+    total: int
+    parts: dict[int, AttachmentRef] = field(default_factory=dict)
+    body: str = ""
+    # The index-0 event, kept so a group that has to be flushed incomplete is
+    # still anchored on its canonical first part (message_id / timestamp /
+    # sender) rather than on whichever part happened to arrive last.
+    first_event: RoomMessageMedia | None = None
+
+
 _UNAVAILABLE_MESSAGES = {
     "always_on": (
         "I'm currently offline — my connector isn't reporting in. "
@@ -186,6 +209,11 @@ def __init__(
         )
         self._agent: Agent | None = None
         self._room_meta: dict[str, RoomMeta | None] = {}
+        # In-flight multi-attachment groups, by group id, with their safety-net
+        # timers. Both are cleared when a group completes or times out, so a
+        # never-completed group cannot leak.
+        self._attachment_groups: dict[str, _PendingAttachmentGroup] = {}
+        self._attachment_group_timers: dict[str, asyncio.TimerHandle] = {}
 
     @property
     def agent(self) -> Agent:
@@ -381,6 +409,143 @@ async def on_media(self, room: MatrixRoom, event: RoomMessageMedia) -> None:
         if relates.get("rel_type") == "m.thread":
             thread_id = relates.get("event_id")
 
+        # Several files posted as one message arrive as separate Matrix events
+        # sharing a group marker (Matrix has no multi-attachment event). Hold
+        # them until the group is complete, then emit ONE payload carrying all
+        # of them, so the agent sees one message with N attachments.
+        group = parse_attachment_group(content)
+        if group is None:
+            await self._emit_media(
+                room,
+                event,
+                meta,
+                is_addressed,
+                sender_name,
+                thread_id,
+                [attachment],
+                event.body,
+            )
+            return
+
+        group_id, index, total = group
+        pending = self._attachment_groups.setdefault(
+            group_id, _PendingAttachmentGroup(total=total)
+        )
+        pending.parts[index] = attachment
+        if index == 0:
+            pending.body = event.body
+            pending.first_event = event
+        if len(pending.parts) < total:
+            self._schedule_attachment_group_flush(
+                group_id, room, event, meta, is_addressed, sender_name, thread_id
+            )
+            return
+
+        self._cancel_attachment_group_flush(group_id)
+        self._attachment_groups.pop(group_id, None)
+        # Anchor the coalesced message on part 0, not on whichever part
+        # happened to complete the group, so message_id / timestamp are stable.
+        await self._emit_media(
+            room,
+            pending.first_event or event,
+            meta,
+            is_addressed,
+            sender_name,
+            thread_id,
+            [pending.parts[i] for i in sorted(pending.parts)],
+            pending.body or event.body,
+        )
+
+    def _schedule_attachment_group_flush(
+        self,
+        group_id: str,
+        room: MatrixRoom,
+        event: RoomMessageMedia,
+        meta: RoomMeta,
+        is_addressed: bool,
+        sender_name: str,
+        thread_id: str | None,
+    ) -> None:
+        """Arm the safety-net timer for an incomplete attachment group.
+
+        The group should normally complete within milliseconds — every event is
+        sent back-to-back by one sender. The timer exists so a group that never
+        completes (a failed send mid-batch, a dropped event) surfaces what did
+        arrive, clearly flagged, instead of being buffered forever.
+
+        Armed once per group, NOT re-armed per part: the deadline bounds the
+        whole group, so a batch dribbling in just under the timeout can't hold
+        the buffer open indefinitely.
+        """
+        if group_id in self._attachment_group_timers:
+            return
+        self._attachment_group_timers[group_id] = asyncio.get_running_loop().call_later(
+            ATTACHMENT_GROUP_TIMEOUT_SECONDS,
+            lambda: asyncio.create_task(
+                self._flush_incomplete_attachment_group(
+                    group_id, room, event, meta, is_addressed, sender_name, thread_id
+                )
+            ),
+        )
+
+    def _cancel_attachment_group_flush(self, group_id: str) -> None:
+        timer = self._attachment_group_timers.pop(group_id, None)
+        if timer is not None:
+            timer.cancel()
+
+    async def _flush_incomplete_attachment_group(
+        self,
+        group_id: str,
+        room: MatrixRoom,
+        event: RoomMessageMedia,
+        meta: RoomMeta,
+        is_addressed: bool,
+        sender_name: str,
+        thread_id: str | None,
+    ) -> None:
+        self._attachment_group_timers.pop(group_id, None)
+        pending = self._attachment_groups.pop(group_id, None)
+        if pending is None:
+            return
+        received = len(pending.parts)
+        logger.error(
+            "Attachment group %s incomplete: %d of %d parts arrived within %ss; "
+            "delivering what arrived",
+            group_id,
+            received,
+            pending.total,
+            ATTACHMENT_GROUP_TIMEOUT_SECONDS,
+        )
+        body = pending.body or event.body
+        notice = (
+            f"[incomplete attachment group: {received} of {pending.total} "
+            f"files arrived]"
+        )
+        # Anchor on part 0 when we have it, so the payload's message_id matches
+        # the completed-group case and replies thread off the canonical event.
+        anchor = pending.first_event or event
+        await self._emit_media(
+            room,
+            anchor,
+            meta,
+            is_addressed,
+            sender_name,
+            thread_id,
+            [pending.parts[i] for i in sorted(pending.parts)],
+            f"{body}\n{notice}" if body else notice,
+        )
+
+    async def _emit_media(
+        self,
+        room: MatrixRoom,
+        event: RoomMessageMedia,
+        meta: RoomMeta,
+        is_addressed: bool,
+        sender_name: str,
+        thread_id: str | None,
+        attachments: list[AttachmentRef],
+        body: str,
+    ) -> None:
         reply_thread_root = thread_id if thread_id is not None else event.event_id
         is_addressed = await self._gate_addressed(
             room, event, meta, reply_thread_root, is_addressed
@@ -396,10 +561,10 @@ async def on_media(self, room: MatrixRoom, event: RoomMessageMedia) -> None:
                 sender=event.sender,
                 sender_name=sender_name,
                 message_id=event.event_id,
-                body=event.body,
+                body=body,
                 timestamp=event.server_timestamp,
                 thread_id=thread_id,
-                attachments=[attachment],
+                attachments=attachments,
             ),
         )
 
diff --git a/core/switch_core/clients/client_base.py b/core/switch_core/clients/client_base.py
index 67bc54ac9..cb1a8e3b3 100644
--- a/core/switch_core/clients/client_base.py
+++ b/core/switch_core/clients/client_base.py
@@ -26,6 +26,7 @@
 from pydantic import BaseModel
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
+from switch_core.attachments import ATTACHMENT_GROUP_KEY
 from switch_core.bridges.resource.events import (
     ResourceLoadRequest,
     ResourceLoadResponse,
@@ -619,6 +620,7 @@ async def send_media(
         msgtype: str,
         caption: str | None = None,
         thread_root_id: str | None = None,
+        group: dict[str, object] | None = None,
     ) -> str | None:
         """Send an m.image / m.file event pointing at an uploaded mxc URI.
 
@@ -626,6 +628,13 @@ async def send_media(
         filename carried separately in `filename`, per the rich-media-caption
         convention); otherwise `body` is the filename. When `thread_root_id` is
         set the event is related into that thread (mirrors send_message).
+
+        `group` marks this event as one part of a multi-attachment message —
+        `{"id": ..., "index": i, "total": n}`. Matrix has no native
+        multi-attachment event (MSC4274 / MSC2881 are unmerged), so a message
+        carrying several files is sent as n events sharing a group id, which
+        receivers coalesce back into one logical message. Absent the field, an
+        event is simply a group of one.
         """
         content: dict[str, object] = {
             "msgtype": msgtype,
@@ -636,6 +645,8 @@ async def send_media(
         }
         if caption:
             content["filename"] = filename
+        if group is not None:
+            content[ATTACHMENT_GROUP_KEY] = group
         if thread_root_id is not None:
             content["m.relates_to"] = {
                 "rel_type": "m.thread",
diff --git a/core/tests/switch_core/bridges/agent/protocol/test_send_media.py b/core/tests/switch_core/bridges/agent/protocol/test_send_media.py
index 2833c9a07..3ac1d42de 100644
--- a/core/tests/switch_core/bridges/agent/protocol/test_send_media.py
+++ b/core/tests/switch_core/bridges/agent/protocol/test_send_media.py
@@ -28,6 +28,7 @@ async def send_media(
         msgtype: str,
         caption: str | None = None,
         thread_root_id: str | None = None,
+        group: dict[str, object] | None = None,
     ) -> str | None:
         self.sends.append(
             {
@@ -39,9 +40,10 @@ async def send_media(
                 "msgtype": msgtype,
                 "caption": caption,
                 "thread_root_id": thread_root_id,
+                "group": group,
             }
         )
-        return "$media-event"
+        return f"$media-event-{len(self.sends)}"
 
 
 def _build_service(client: _FakeClient, *, max_bytes: int = 100) -> ProtocolService:
@@ -70,16 +72,26 @@ async def test_send_media_uploads_and_posts_image() -> None:
     svc = _build_service(client)
 
     result = await svc.send_media(
-        "agent-1", "room-1", b"png-bytes", filename="cat.png", mimetype="image/png"
+        "agent-1", "room-1", [(b"png-bytes", "cat.png", "image/png")]
     )
 
-    assert result == {"event_id": "$media-event", "mxc": "mxc://s/uploaded"}
+    assert result["event_id"] == "$media-event-1"
+    assert result["mxc"] == "mxc://s/uploaded"
+    assert result["attachments"] == [
+        {
+            "event_id": "$media-event-1",
+            "mxc": "mxc://s/uploaded",
+            "filename": "cat.png",
+        }
+    ]
     assert client.uploads == [(b"png-bytes", "image/png", "cat.png")]
     sent = client.sends[0]
     assert sent["room_id"] == "!room"
     assert sent["msgtype"] == "m.image"
     assert sent["caption"] is None
     assert sent["thread_root_id"] is None
+    # A lone attachment carries no group marker.
+    assert sent["group"] is None
 
 
 async def test_send_media_caption_and_thread() -> None:
@@ -89,9 +101,7 @@ async def test_send_media_caption_and_thread() -> None:
     await svc.send_media(
         "agent-1",
         "room-1",
-        b"x",
-        filename="cat.png",
-        mimetype="image/png",
+        [(b"x", "cat.png", "image/png")],
         caption="look",
         thread_id="$mid-thread",
     )
@@ -105,9 +115,7 @@ async def test_send_media_non_image_is_file_msgtype() -> None:
     client = _FakeClient()
     svc = _build_service(client)
 
-    await svc.send_media(
-        "agent-1", "room-1", b"%PDF", filename="doc.pdf", mimetype="application/pdf"
-    )
+    await svc.send_media("agent-1", "room-1", [(b"%PDF", "doc.pdf", "application/pdf")])
 
     assert client.sends[0]["msgtype"] == "m.file"
 
@@ -117,9 +125,7 @@ async def test_send_media_oversize_raises() -> None:
     svc = _build_service(client, max_bytes=3)
 
     with pytest.raises(ValueError, match="over the 3-byte limit"):
-        await svc.send_media(
-            "agent-1", "room-1", b"toolarge", filename="c.png", mimetype="image/png"
-        )
+        await svc.send_media("agent-1", "room-1", [(b"toolarge", "c.png", "image/png")])
     assert client.uploads == []
 
 
@@ -128,6 +134,69 @@ async def test_send_media_empty_raises() -> None:
     svc = _build_service(client)
 
     with pytest.raises(ValueError, match="empty"):
+        await svc.send_media("agent-1", "room-1", [(b"", "c.png", "image/png")])
+
+
+async def test_send_media_multiple_files_share_a_group_marker() -> None:
+    client = _FakeClient()
+    svc = _build_service(client)
+
+    result = await svc.send_media(
+        "agent-1",
+        "room-1",
+        [
+            (b"png-bytes", "cat.png", "image/png"),
+            (b"# notes", "notes.md", "text/markdown"),
+            (b"a,b", "data.csv", "text/csv"),
+        ],
+        caption="three files",
+    )
+
+    assert len(client.sends) == 3
+    groups = [sent["group"] for sent in client.sends]
+    assert {g["id"] for g in groups} == {groups[0]["id"]}, "all parts share one id"
+    assert [g["index"] for g in groups] == [0, 1, 2]
+    assert all(g["total"] == 3 for g in groups)
+    # Caption rides on the first part only, mirroring the inbound convention.
+    assert client.sends[0]["caption"] == "three files"
+    assert client.sends[1]["caption"] is None
+    assert client.sends[2]["caption"] is None
+    # Non-image files keep their own msgtype within the group.
+    assert [sent["msgtype"] for sent in client.sends] == [
+        "m.image",
+        "m.file",
+        "m.file",
+    ]
+    assert [a["filename"] for a in result["attachments"]] == [
+        "cat.png",
+        "notes.md",
+        "data.csv",
+    ]
+
+
+async def test_send_media_rejects_whole_batch_when_one_file_is_oversize() -> None:
+    """A bad file in the batch must abort everything — never a half-posted
+    message with the good files and a silently missing one."""
+    client = _FakeClient()
+    svc = _build_service(client, max_bytes=5)
+
+    with pytest.raises(ValueError, match="over the 5-byte limit"):
         await svc.send_media(
-            "agent-1", "room-1", b"", filename="c.png", mimetype="image/png"
+            "agent-1",
+            "room-1",
+            [
+                (b"ok", "small.md", "text/markdown"),
+                (b"way-too-large", "big.bin", "application/octet-stream"),
+            ],
         )
+
+    assert client.uploads == []
+    assert client.sends == []
+
+
+async def test_send_media_rejects_empty_batch() -> None:
+    client = _FakeClient()
+    svc = _build_service(client)
+
+    with pytest.raises(ValueError, match="no attachments"):
+        await svc.send_media("agent-1", "room-1", [])
diff --git a/core/tests/switch_core/bridges/collaboration/test_bridge_inbound_media.py b/core/tests/switch_core/bridges/collaboration/test_bridge_inbound_media.py
new file mode 100644
index 000000000..3bc8b9915
--- /dev/null
+++ b/core/tests/switch_core/bridges/collaboration/test_bridge_inbound_media.py
@@ -0,0 +1,224 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any
+
+from switch_core.bridges.collaboration.bridge_core import BridgeCore
+from switch_core.bridges.collaboration.models import (
+    Attachment,
+    AttachmentFailure,
+    InboundMessage,
+)
+
+
+class _FakePuppet:
+    matrix_user_id = "@puppet:s"
+
+    def __init__(self) -> None:
+        self.uploads: list[dict[str, Any]] = []
+        self.media: list[dict[str, Any]] = []
+        self.messages: list[dict[str, Any]] = []
+
+    async def upload_media(self, data: bytes, mimetype: str, filename: str) -> str:
+        self.uploads.append({"data": data, "mimetype": mimetype, "filename": filename})
+        return f"mxc://s/{filename}"
+
+    async def send_media(
+        self,
+        matrix_room_id,
+        mxc,
+        filename,
+        mimetype,
+        size,
+        msgtype,
+        caption=None,
+        thread_root_id=None,
+        group=None,
+    ):  # noqa: ANN001, ANN201
+        self.media.append(
+            {
+                "matrix_room_id": matrix_room_id,
+                "mxc": mxc,
+                "filename": filename,
+                "mimetype": mimetype,
+                "size": size,
+                "msgtype": msgtype,
+                "caption": caption,
+                "thread_root_id": thread_root_id,
+                "group": group,
+            }
+        )
+        return f"$evt-{len(self.media) - 1}"
+
+    async def send_message(
+        self, matrix_room_id, content, format=None, thread_root_id=None
+    ):  # noqa: ANN001, ANN201, A002
+        self.messages.append({"content": content, "thread_root_id": thread_root_id})
+        return "$evt-text"
+
+
+class _FakeAdapter:
+    def translate_inbound(self, content: str) -> str:
+        return content
+
+
+def _fake_bridge() -> SimpleNamespace:
+    puppet = _FakePuppet()
+    recorded: list[dict[str, str]] = []
+
+    async def _is_registered_agent(_name: str) -> bool:
+        return False
+
+    async def _ensure_user_in_matrix_room(**_kwargs: Any) -> _FakePuppet:
+        return puppet
+
+    async def _record_message_map(**kwargs: str) -> None:
+        recorded.append(kwargs)
+
+    return SimpleNamespace(
+        _adapter=_FakeAdapter(),
+        _channel_to_room={"chan-1": ("room-1", "!room:s")},
+        _is_registered_agent=_is_registered_agent,
+        _ensure_user_in_matrix_room=_ensure_user_in_matrix_room,
+        _record_message_map=_record_message_map,
+        puppet=puppet,
+        recorded=recorded,
+    )
+
+
+def _msg(
+    *,
+    content: str = "here you go",
+    attachments: list[Attachment] | None = None,
+    attachment_failures: list[AttachmentFailure] | None = None,
+) -> InboundMessage:
+    return InboundMessage(
+        channel_id="chan-1",
+        channel_type="channel_public",
+        sender_id="U1",
+        sender_name="alice",
+        content=content,
+        message_ref="post-1",
+        attachments=attachments or [],
+        attachment_failures=attachment_failures or [],
+    )
+
+
+def _attachment(filename: str, mimetype: str) -> Attachment:
+    return Attachment(filename=filename, mimetype=mimetype, data=b"bytes")
+
+
+async def test_three_attachments_are_stamped_as_one_group() -> None:
+    bridge = _fake_bridge()
+
+    await BridgeCore._handle_inbound_message(
+        bridge,
+        _msg(
+            content="three files",
+            attachments=[
+                _attachment("cat.png", "image/png"),
+                _attachment("notes.md", "text/markdown"),
+                _attachment("data.csv", "text/csv"),
+            ],
+        ),
+    )
+
+    media = bridge.puppet.media
+    assert len(media) == 3
+
+    group_ids = {m["group"]["id"] for m in media}
+    assert len(group_ids) == 1
+    assert [m["group"]["index"] for m in media] == [0, 1, 2]
+    assert {m["group"]["total"] for m in media} == {3}
+
+    # Caption convention: text rides on the first attachment only.
+    assert [m["caption"] for m in media] == ["three files", None, None]
+    assert [m["msgtype"] for m in media] == ["m.image", "m.file", "m.file"]
+    assert [m["filename"] for m in media] == ["cat.png", "notes.md", "data.csv"]
+
+    # The first media event stands in for the post, so replies thread back.
+    assert bridge.recorded == [
+        {
+            "external_channel_id": "chan-1",
+            "matrix_event_id": "$evt-0",
+            "external_post_id": "post-1",
+        }
+    ]
+    assert bridge.puppet.messages == []
+
+
+async def test_single_attachment_carries_no_group_marker() -> None:
+    # A group of one needs no marker — the receiver treats an unmarked event as
+    # a complete message and never buffers it.
+    bridge = _fake_bridge()
+
+    await BridgeCore._handle_inbound_message(
+        bridge, _msg(attachments=[_attachment("cat.png", "image/png")])
+    )
+
+    assert len(bridge.puppet.media) == 1
+    assert bridge.puppet.media[0]["group"] is None
+    assert bridge.puppet.media[0]["caption"] == "here you go"
+
+
+async def test_attachment_failures_are_disclosed_alongside_text() -> None:
+    bridge = _fake_bridge()
+
+    await BridgeCore._handle_inbound_message(
+        bridge,
+        _msg(
+            content="see attached",
+            attachment_failures=[
+                AttachmentFailure(filename="huge.zip", reason="too large")
+            ],
+        ),
+    )
+
+    body = bridge.puppet.messages[0]["content"]
+    assert body.startswith("see attached")
+    assert "huge.zip" in body
+    assert "too large" in body
+    assert "attachment not relayed" in body
+
+
+async def test_attachment_failures_are_disclosed_when_text_is_empty() -> None:
+    # An attachment-only post whose single file failed must still say something
+    # in the room — never a silent drop.
+    bridge = _fake_bridge()
+
+    await BridgeCore._handle_inbound_message(
+        bridge,
+        _msg(
+            content="   ",
+            attachment_failures=[
+                AttachmentFailure(filename="huge.zip", reason="too large"),
+                AttachmentFailure(filename="broken.pdf", reason="download failed"),
+            ],
+        ),
+    )
+
+    body = bridge.puppet.messages[0]["content"]
+    assert body.splitlines() == [
+        "_attachment not relayed: huge.zip — too large_",
+        "_attachment not relayed: broken.pdf — download failed_",
+    ]
+
+
+async def test_attachment_failures_ride_on_the_caption_of_relayed_media() -> None:
+    bridge = _fake_bridge()
+
+    await BridgeCore._handle_inbound_message(
+        bridge,
+        _msg(
+            content="two files, one failed",
+            attachments=[_attachment("cat.png", "image/png")],
+            attachment_failures=[
+                AttachmentFailure(filename="huge.zip", reason="too large")
+            ],
+        ),
+    )
+
+    assert bridge.puppet.messages == []
+    caption = bridge.puppet.media[0]["caption"]
+    assert "two files, one failed" in caption
+    assert "huge.zip" in caption
diff --git a/core/tests/switch_core/bridges/collaboration/test_bridge_outbound_media.py b/core/tests/switch_core/bridges/collaboration/test_bridge_outbound_media.py
index ade988fea..bfd970dba 100644
--- a/core/tests/switch_core/bridges/collaboration/test_bridge_outbound_media.py
+++ b/core/tests/switch_core/bridges/collaboration/test_bridge_outbound_media.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 from types import SimpleNamespace
 from typing import Any
 
@@ -17,13 +18,18 @@ def _media_event(
     sender: str = "@agent:s",
     sender_name: str | None = "agent-a",
     thread_root: str | None = None,
+    group: dict[str, Any] | None = None,
+    mimetype: str = "image/png",
+    event_id: str = "$media-event",
 ) -> nio.RoomMessageMedia:
     content: dict[str, Any] = {
         "msgtype": msgtype,
         "body": body,
         "url": "mxc://s/abc",
-        "info": {"mimetype": "image/png", "size": 5},
+        "info": {"mimetype": mimetype, "size": 5},
     }
+    if group is not None:
+        content["com.switch.attachment_group"] = group
     if filename is not None:
         content["filename"] = filename
     if sender_name is not None:
@@ -34,7 +40,7 @@ def _media_event(
     return cls.from_dict(
         {
             "type": "m.room.message",
-            "event_id": "$media-event",
+            "event_id": event_id,
             "sender": sender,
             "origin_server_ts": 1700000000000,
             "content": content,
@@ -45,6 +51,7 @@ def _media_event(
 class _FakeAdapter:
     def __init__(self) -> None:
         self.attachments: list[dict[str, Any]] = []
+        self.batches: list[dict[str, Any]] = []
         self.messages: list[dict[str, Any]] = []
 
     async def send_attachment(
@@ -70,6 +77,20 @@ async def send_attachment(
         )
         return "ext-ref-1"
 
+    async def send_attachments(
+        self, channel_id, sender_name, files, caption=None, thread_root_id=None
+    ):  # noqa: ANN001, ANN201
+        self.batches.append(
+            {
+                "channel_id": channel_id,
+                "sender_name": sender_name,
+                "files": files,
+                "caption": caption,
+                "thread_root_id": thread_root_id,
+            }
+        )
+        return "ext-ref-3"
+
     async def send_message(self, channel_id, sender_name, content, thread_root_id=None):  # noqa: ANN001, ANN201
         self.messages.append(
             {
@@ -109,12 +130,24 @@ async def _record_message_map(**kwargs: str) -> None:
         _external_post_for_matrix_event=_external_post_for_matrix_event,
         _record_message_map=_record_message_map,
         recorded=recorded,
+        _outbound_groups={},
+        _outbound_group_timers={},
     )
     ns._find_channel = lambda room_id=None, matrix_room_id=None: (
         "chan-1" if matrix_room_id == "!room:s" else None
     )
     ns._outbound_thread_root_ref = BridgeCore._outbound_thread_root_ref.__get__(ns)
     ns._download_matrix_media = BridgeCore._download_matrix_media.__get__(ns)
+    ns._schedule_outbound_group_flush = (
+        BridgeCore._schedule_outbound_group_flush.__get__(ns)
+    )
+    ns._cancel_outbound_group_flush = BridgeCore._cancel_outbound_group_flush.__get__(
+        ns
+    )
+    ns._relay_outbound_group = BridgeCore._relay_outbound_group.__get__(ns)
+    ns._flush_incomplete_outbound_group = (
+        BridgeCore._flush_incomplete_outbound_group.__get__(ns)
+    )
 
     async def _nio_download(mxc: str):
         return download
@@ -203,21 +236,25 @@ async def test_media_without_sender_name_is_skipped() -> None:
     assert bridge._adapter.messages == []
 
 
-async def test_non_image_file_posts_disclosed_notice() -> None:
+async def test_non_image_file_relays_natively() -> None:
+    """A .pdf / .md / .csv must upload as a real file, not degrade to a text
+    notice — that was the reported bug."""
     bridge = _fake_bridge()
 
     await BridgeCore.handle_outbound_media(
         bridge,
         _room(),
-        _media_event(msgtype="m.file", body="report.pdf"),
+        _media_event(msgtype="m.file", body="report.pdf", mimetype="application/pdf"),
         bridge.client,
     )
 
-    assert bridge._adapter.attachments == []
-    assert len(bridge._adapter.messages) == 1
-    assert "report.pdf" in bridge._adapter.messages[0]["content"]
-    # The notice still threads/correlates like a real relay.
-    assert bridge.recorded[0]["external_post_id"] == "ext-ref-2"
+    assert bridge._adapter.messages == []
+    assert len(bridge._adapter.attachments) == 1
+    sent = bridge._adapter.attachments[0]
+    assert sent["filename"] == "report.pdf"
+    assert sent["mimetype"] == "application/pdf"
+    assert sent["data"] == b"bytes"
+    assert bridge.recorded[0]["external_post_id"] == "ext-ref-1"
 
 
 async def test_download_failure_posts_disclosed_fallback() -> None:
@@ -241,3 +278,77 @@ async def test_oversize_media_posts_disclosed_fallback() -> None:
 
     assert bridge._adapter.attachments == []
     assert "couldn't be relayed" in bridge._adapter.messages[0]["content"]
+
+
+async def test_grouped_attachments_relay_as_one_platform_post() -> None:
+    """Three files sent as one message must arrive as ONE post carrying all
+    three, not three separate posts."""
+    bridge = _fake_bridge()
+    group_id = "grp-1"
+
+    for index, (name, mimetype) in enumerate(
+        [
+            ("cat.png", "image/png"),
+            ("notes.md", "text/markdown"),
+            ("data.csv", "text/csv"),
+        ]
+    ):
+        await BridgeCore.handle_outbound_media(
+            bridge,
+            _room(),
+            _media_event(
+                msgtype="m.image" if index == 0 else "m.file",
+                body="three files" if index == 0 else name,
+                filename=name if index == 0 else None,
+                mimetype=mimetype,
+                event_id=f"$part-{index}",
+                group={"id": group_id, "index": index, "total": 3},
+            ),
+            bridge.client,
+        )
+
+    # Nothing relayed until the group completed.
+    assert bridge._adapter.attachments == []
+    assert bridge._adapter.messages == []
+    assert len(bridge._adapter.batches) == 1
+
+    batch = bridge._adapter.batches[0]
+    assert [f.filename for f in batch["files"]] == ["cat.png", "notes.md", "data.csv"]
+    assert batch["caption"] == "three files"
+    # The group correlates via its first event, so replies thread back.
+    assert bridge.recorded[0]["matrix_event_id"] == "$part-0"
+    assert bridge.recorded[0]["external_post_id"] == "ext-ref-3"
+    assert bridge._outbound_groups == {}
+
+
+async def test_incomplete_group_is_flushed_with_a_disclosed_notice() -> None:
+    """A group that never completes must still reach the platform, flagged —
+    never silently held forever."""
+    import switch_core.bridges.collaboration.bridge_core as bc
+
+    original = bc.OUTBOUND_GROUP_TIMEOUT_SECONDS
+    bc.OUTBOUND_GROUP_TIMEOUT_SECONDS = 0.01
+    try:
+        bridge = _fake_bridge()
+        await BridgeCore.handle_outbound_media(
+            bridge,
+            _room(),
+            _media_event(
+                msgtype="m.file",
+                body="only.md",
+                mimetype="text/markdown",
+                group={"id": "grp-2", "index": 0, "total": 3},
+            ),
+            bridge.client,
+        )
+        assert bridge._adapter.batches == []
+
+        await asyncio.sleep(0.1)
+    finally:
+        bc.OUTBOUND_GROUP_TIMEOUT_SECONDS = original
+
+    assert len(bridge._adapter.batches) == 1
+    batch = bridge._adapter.batches[0]
+    assert [f.filename for f in batch["files"]] == ["only.md"]
+    assert "1 of 3" in (batch["caption"] or "")
+    assert bridge._outbound_groups == {}
diff --git a/core/tests/switch_core/bridges/collaboration/test_discord_adapter.py b/core/tests/switch_core/bridges/collaboration/test_discord_adapter.py
index b1693585f..105f05805 100644
--- a/core/tests/switch_core/bridges/collaboration/test_discord_adapter.py
+++ b/core/tests/switch_core/bridges/collaboration/test_discord_adapter.py
@@ -416,10 +416,12 @@ def __init__(
         content_type: str | None,
         data: bytes = b"x",
         fail: bool = False,
+        size: int | None = None,
     ) -> None:
         self.id = 1
         self.filename = filename
         self.content_type = content_type
+        self.size = size if size is not None else len(data)
         self._data = data
         self._fail = fail
 
@@ -429,14 +431,14 @@ async def read(self) -> bytes:
         return self._data
 
 
-def test_image_attachments_downloaded_others_skipped() -> None:
+def test_attachments_of_every_type_downloaded() -> None:
     adapter = _adapter()
     captured = _capture_messages(adapter)
     files = [
         _FakeFile("shot.png", "image/png", b"png-bytes"),
-        _FakeFile("notes.pdf", "application/pdf"),
-        _FakeFile("broken.jpg", "image/jpeg", fail=True),
-        _FakeFile("unknown.bin", None),
+        _FakeFile("notes.pdf", "application/pdf", b"%PDF"),
+        _FakeFile("readme.md", "text/markdown", b"# hi"),
+        _FakeFile("unknown.bin", None, b"raw"),
     ]
 
     _run(
@@ -446,10 +448,58 @@ def test_image_attachments_downloaded_others_skipped() -> None:
     )
 
     atts = captured[0].attachments
-    assert len(atts) == 1
-    assert atts[0].filename == "shot.png"
-    assert atts[0].mimetype == "image/png"
+    assert [a.filename for a in atts] == [
+        "shot.png",
+        "notes.pdf",
+        "readme.md",
+        "unknown.bin",
+    ]
     assert atts[0].data == b"png-bytes"
+    assert atts[2].mimetype == "text/markdown"
+    # A file Discord reports no content type for still comes through.
+    assert atts[3].mimetype == "application/octet-stream"
+    assert captured[0].attachment_failures == []
+
+
+def test_failed_download_reported_as_failure_not_dropped() -> None:
+    adapter = _adapter()
+    captured = _capture_messages(adapter)
+    files = [
+        _FakeFile("good.md", "text/markdown", b"# hi"),
+        _FakeFile("broken.jpg", "image/jpeg", fail=True),
+    ]
+
+    _run(
+        adapter._handle_message(
+            _gateway_message(channel=_FakeChannel(), attachments=files)
+        )
+    )
+
+    assert [a.filename for a in captured[0].attachments] == ["good.md"]
+    failures = captured[0].attachment_failures
+    assert [f.filename for f in failures] == ["broken.jpg"]
+    assert failures[0].reason
+
+
+def test_oversize_attachment_reported_as_failure() -> None:
+    adapter = _adapter()
+    adapter.set_max_attachment_bytes(10)
+    captured = _capture_messages(adapter)
+    files = [
+        _FakeFile("small.md", "text/markdown", b"# hi"),
+        _FakeFile("huge.bin", "application/octet-stream", b"x", size=100),
+    ]
+
+    _run(
+        adapter._handle_message(
+            _gateway_message(channel=_FakeChannel(), attachments=files)
+        )
+    )
+
+    assert [a.filename for a in captured[0].attachments] == ["small.md"]
+    failures = captured[0].attachment_failures
+    assert [f.filename for f in failures] == ["huge.bin"]
+    assert "exceeds" in failures[0].reason
 
 
 # ── Outbound messaging ───────────────────────────────────────────────────────
diff --git a/core/tests/switch_core/bridges/collaboration/test_mattermost_attachments.py b/core/tests/switch_core/bridges/collaboration/test_mattermost_attachments.py
index 9c45d9efa..a6b8bc448 100644
--- a/core/tests/switch_core/bridges/collaboration/test_mattermost_attachments.py
+++ b/core/tests/switch_core/bridges/collaboration/test_mattermost_attachments.py
@@ -40,7 +40,7 @@ def _adapter_with_files(files: _FakeFiles) -> MattermostAdapter:
 def _fetch(adapter: MattermostAdapter, file_ids: list[str]):
     loop = asyncio.new_event_loop()
     try:
-        return loop.run_until_complete(adapter._fetch_image_attachments(file_ids, loop))
+        return loop.run_until_complete(adapter._fetch_attachments(file_ids, loop))
     finally:
         loop.close()
 
@@ -52,8 +52,9 @@ def test_downloads_image_attachment() -> None:
     )
     adapter = _adapter_with_files(files)
 
-    attachments = _fetch(adapter, ["f1"])
+    attachments, failures = _fetch(adapter, ["f1"])
 
+    assert failures == []
     assert len(attachments) == 1
     assert attachments[0].filename == "cat.png"
     assert attachments[0].mimetype == "image/png"
@@ -61,21 +62,25 @@ def test_downloads_image_attachment() -> None:
     assert files.get_file_calls == ["f1"]
 
 
-def test_skips_non_image_attachment() -> None:
+def test_downloads_non_image_attachment() -> None:
     files = _FakeFiles(
-        meta={"f1": {"mime_type": "application/pdf", "name": "doc.pdf", "size": 9}},
-        data={"f1": b"%PDF"},
+        meta={"f1": {"mime_type": "text/markdown", "name": "notes.md", "size": 7}},
+        data={"f1": b"# notes"},
     )
     adapter = _adapter_with_files(files)
 
-    attachments = _fetch(adapter, ["f1"])
+    attachments, failures = _fetch(adapter, ["f1"])
 
-    # Non-image is skipped entirely — and we never download its bytes.
-    assert attachments == []
-    assert files.get_file_calls == []
+    # Every file type is relayed now — not just images.
+    assert failures == []
+    assert len(attachments) == 1
+    assert attachments[0].filename == "notes.md"
+    assert attachments[0].mimetype == "text/markdown"
+    assert attachments[0].data == b"# notes"
+    assert files.get_file_calls == ["f1"]
 
 
-def test_failed_download_skips_only_that_file() -> None:
+def test_failed_download_reported_as_failure() -> None:
     class _ExplodingFiles(_FakeFiles):
         def get_file_metadata(self, file_id: str) -> dict[str, object]:
             if file_id == "bad":
@@ -88,14 +93,51 @@ def get_file_metadata(self, file_id: str) -> dict[str, object]:
     )
     adapter = _adapter_with_files(files)
 
-    attachments = _fetch(adapter, ["bad", "good"])
+    attachments, failures = _fetch(adapter, ["bad", "good"])
 
+    # The good file still comes through; the bad one is disclosed, not dropped.
     assert [a.filename for a in attachments] == ["ok.jpg"]
+    assert [f.filename for f in failures] == ["bad"]
+    assert failures[0].reason
+
+
+def test_oversize_attachment_reported_as_failure() -> None:
+    files = _FakeFiles(
+        meta={"f1": {"mime_type": "application/zip", "name": "huge.zip", "size": 100}},
+        data={"f1": b"x" * 100},
+    )
+    adapter = _adapter_with_files(files)
+    adapter.set_max_attachment_bytes(10)
+
+    attachments, failures = _fetch(adapter, ["f1"])
+
+    assert attachments == []
+    assert files.get_file_calls == []
+    assert [f.filename for f in failures] == ["huge.zip"]
+    assert failures[0].reason
+
+
+def test_multiple_files_all_returned() -> None:
+    files = _FakeFiles(
+        meta={
+            "f1": {"mime_type": "image/png", "name": "cat.png", "size": 3},
+            "f2": {"mime_type": "text/markdown", "name": "notes.md", "size": 7},
+            "f3": {"mime_type": "application/pdf", "name": "doc.pdf", "size": 4},
+        },
+        data={"f1": b"abc", "f2": b"# notes", "f3": b"%PDF"},
+    )
+    adapter = _adapter_with_files(files)
+
+    attachments, failures = _fetch(adapter, ["f1", "f2", "f3"])
+
+    assert failures == []
+    assert [a.filename for a in attachments] == ["cat.png", "notes.md", "doc.pdf"]
+    assert [a.data for a in attachments] == [b"abc", b"# notes", b"%PDF"]
 
 
 def test_no_file_ids_returns_empty() -> None:
     adapter = _adapter_with_files(_FakeFiles(meta={}, data={}))
-    assert _fetch(adapter, []) == []
+    assert _fetch(adapter, []) == ([], [])
 
 
 # ── Outbound attachments ─────────────────────────────────────────────────────
diff --git a/core/tests/switch_core/bridges/collaboration/test_slack_adapter.py b/core/tests/switch_core/bridges/collaboration/test_slack_adapter.py
index a283f8676..b81a67464 100644
--- a/core/tests/switch_core/bridges/collaboration/test_slack_adapter.py
+++ b/core/tests/switch_core/bridges/collaboration/test_slack_adapter.py
@@ -90,8 +90,9 @@ async def fake_download(url: str) -> bytes:
             "url_private_download": "https://files.slack.com/cat.png",
         }
     ]
-    attachments = _run(adapter._fetch_image_attachments(files))
+    attachments, failures = _run(adapter._fetch_attachments(files))
 
+    assert failures == []
     assert len(attachments) == 1
     assert attachments[0].filename == "cat.png"
     assert attachments[0].mimetype == "image/png"
@@ -99,32 +100,36 @@ async def fake_download(url: str) -> bytes:
     assert downloaded == ["https://files.slack.com/cat.png"]
 
 
-def test_fetch_skips_non_image_without_downloading() -> None:
+def test_fetch_downloads_non_image_attachment() -> None:
     adapter = _adapter()
     downloaded: list[str] = []
 
     async def fake_download(url: str) -> bytes:
         downloaded.append(url)
-        return b"x"
+        return b"# notes"
 
     adapter._download_file = fake_download  # type: ignore[assignment]
 
     files = [
         {
             "id": "F1",
-            "name": "doc.pdf",
-            "mimetype": "application/pdf",
-            "url_private_download": "https://files.slack.com/doc.pdf",
+            "name": "notes.md",
+            "mimetype": "text/markdown",
+            "url_private_download": "https://files.slack.com/notes.md",
         }
     ]
-    attachments = _run(adapter._fetch_image_attachments(files))
+    attachments, failures = _run(adapter._fetch_attachments(files))
 
-    # Non-image is skipped entirely — and we never download its bytes.
-    assert attachments == []
-    assert downloaded == []
+    # Every file type is relayed now — not just images.
+    assert failures == []
+    assert len(attachments) == 1
+    assert attachments[0].filename == "notes.md"
+    assert attachments[0].mimetype == "text/markdown"
+    assert attachments[0].data == b"# notes"
+    assert downloaded == ["https://files.slack.com/notes.md"]
 
 
-def test_fetch_failed_download_skips_only_that_file() -> None:
+def test_fetch_failed_download_reported_as_failure() -> None:
     adapter = _adapter()
 
     async def fake_download(url: str) -> bytes:
@@ -148,14 +153,81 @@ async def fake_download(url: str) -> bytes:
             "url_private": "https://files.slack.com/good.jpg",
         },
     ]
-    attachments = _run(adapter._fetch_image_attachments(files))
+    attachments, failures = _run(adapter._fetch_attachments(files))
 
+    # The good file still comes through; the bad one is disclosed, not dropped.
     assert [a.filename for a in attachments] == ["good.jpg"]
+    assert [f.filename for f in failures] == ["bad.png"]
+    assert failures[0].reason
+
+
+def test_fetch_oversize_file_reported_without_downloading() -> None:
+    adapter = _adapter()
+    downloaded: list[str] = []
+
+    async def fake_download(url: str) -> bytes:
+        downloaded.append(url)
+        return b"x" * 100
+
+    adapter._download_file = fake_download  # type: ignore[assignment]
+    adapter.set_max_attachment_bytes(10)
+
+    files = [
+        {
+            "id": "F1",
+            "name": "huge.bin",
+            "mimetype": "application/octet-stream",
+            "size": 100,
+            "url_private_download": "https://files.slack.com/huge.bin",
+        }
+    ]
+    attachments, failures = _run(adapter._fetch_attachments(files))
+
+    # Slack reports the size up front, so we never spend the download.
+    assert attachments == []
+    assert downloaded == []
+    assert [f.filename for f in failures] == ["huge.bin"]
+    assert failures[0].reason
+
+
+def test_fetch_multiple_files_all_returned() -> None:
+    adapter = _adapter()
+
+    async def fake_download(url: str) -> bytes:
+        return url.rsplit("/", 1)[-1].encode()
+
+    adapter._download_file = fake_download  # type: ignore[assignment]
+
+    files = [
+        {
+            "id": "F1",
+            "name": "cat.png",
+            "mimetype": "image/png",
+            "url_private": "https://files.slack.com/cat.png",
+        },
+        {
+            "id": "F2",
+            "name": "notes.md",
+            "mimetype": "text/markdown",
+            "url_private": "https://files.slack.com/notes.md",
+        },
+        {
+            "id": "F3",
+            "name": "doc.pdf",
+            "mimetype": "application/pdf",
+            "url_private": "https://files.slack.com/doc.pdf",
+        },
+    ]
+    attachments, failures = _run(adapter._fetch_attachments(files))
+
+    assert failures == []
+    assert [a.filename for a in attachments] == ["cat.png", "notes.md", "doc.pdf"]
+    assert [a.data for a in attachments] == [b"cat.png", b"notes.md", b"doc.pdf"]
 
 
 def test_fetch_no_files_returns_empty() -> None:
     adapter = _adapter()
-    assert _run(adapter._fetch_image_attachments([])) == []
+    assert _run(adapter._fetch_attachments([])) == ([], [])
 
 
 # ── Inbound threading ────────────────────────────────────────────────────────
diff --git a/core/tests/switch_core/clients/test_agent_client_attachment_groups.py b/core/tests/switch_core/clients/test_agent_client_attachment_groups.py
new file mode 100644
index 000000000..3432c3d3f
--- /dev/null
+++ b/core/tests/switch_core/clients/test_agent_client_attachment_groups.py
@@ -0,0 +1,295 @@
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from typing import Any
+
+import nio
+
+import switch_core.clients.agent_client as ac
+from switch_core.clients.agent_client import AgentClient
+from switch_core.clients.room_meta import RoomMeta
+
+
+def _media_event(
+    *,
+    body: str,
+    filename: str | None = None,
+    mimetype: str = "image/png",
+    msgtype: str = "m.image",
+    group: dict[str, Any] | None = None,
+    event_id: str = "$evt",
+) -> nio.RoomMessageMedia:
+    content: dict[str, Any] = {
+        "msgtype": msgtype,
+        "body": body,
+        "url": "mxc://s/abc",
+        "info": {"mimetype": mimetype, "size": 5},
+        "sender_name": "alice",
+    }
+    if filename is not None:
+        content["filename"] = filename
+    if group is not None:
+        content["com.switch.attachment_group"] = group
+    cls = nio.RoomMessageImage if msgtype == "m.image" else nio.RoomMessageFile
+    return cls.from_dict(
+        {
+            "type": "m.room.message",
+            "event_id": event_id,
+            "sender": "@alice:s",
+            "origin_server_ts": 1700000000000,
+            "content": content,
+        }
+    )
+
+
+class _FakeQueue:
+    def __init__(self) -> None:
+        self.events: list[Any] = []
+
+    def enqueue(self, _agent_id: str, _room_id: str, event: Any) -> None:
+        self.events.append(event)
+
+
+def _fake_client() -> SimpleNamespace:
+    queue = _FakeQueue()
+    meta = RoomMeta(room_id="room-1", name="Room", bridge_id="bridge-1")
+
+    async def _resolve_room_meta(_matrix_room_id: str) -> RoomMeta:
+        return meta
+
+    async def _compute_addressed(_event: Any, _meta: RoomMeta) -> bool:
+        return True
+
+    async def _gate_addressed(
+        _room: Any, _event: Any, _meta: Any, _root: Any, is_addressed: bool
+    ) -> bool:
+        return is_addressed
+
+    ns = SimpleNamespace(
+        agent=SimpleNamespace(id="agent-1", name="agent-a"),
+        _event_queue=queue,
+        _attachment_groups={},
+        _attachment_group_timers={},
+        _resolve_room_meta=_resolve_room_meta,
+        _compute_addressed=_compute_addressed,
+        _gate_addressed=_gate_addressed,
+        queue=queue,
+    )
+    ns._emit_media = AgentClient._emit_media.__get__(ns)
+    ns._schedule_attachment_group_flush = (
+        AgentClient._schedule_attachment_group_flush.__get__(ns)
+    )
+    ns._cancel_attachment_group_flush = (
+        AgentClient._cancel_attachment_group_flush.__get__(ns)
+    )
+    ns._flush_incomplete_attachment_group = (
+        AgentClient._flush_incomplete_attachment_group.__get__(ns)
+    )
+    return ns
+
+
+def _room() -> SimpleNamespace:
+    return SimpleNamespace(room_id="!room:s")
+
+
+async def test_grouped_media_coalesces_into_one_event() -> None:
+    client = _fake_client()
+    parts = [
+        ("cat.png", "image/png", "m.image"),
+        ("notes.md", "text/markdown", "m.file"),
+        ("data.csv", "text/csv", "m.file"),
+    ]
+
+    for index, (name, mimetype, msgtype) in enumerate(parts):
+        await AgentClient.on_media(
+            client,
+            _room(),
+            _media_event(
+                body="three files" if index == 0 else name,
+                filename=name if index == 0 else None,
+                mimetype=mimetype,
+                msgtype=msgtype,
+                event_id=f"$part-{index}",
+                group={"id": "grp-1", "index": index, "total": 3},
+            ),
+        )
+        if index < 2:
+            assert client.queue.events == []
+
+    assert len(client.queue.events) == 1
+    payload = client.queue.events[0].payload
+    assert [a.filename for a in payload.attachments] == [
+        "cat.png",
+        "notes.md",
+        "data.csv",
+    ]
+    assert payload.body == "three files"
+    assert client._attachment_groups == {}
+    assert client._attachment_group_timers == {}
+
+
+async def test_grouped_media_coalesces_out_of_order() -> None:
+    client = _fake_client()
+    order = [1, 0, 2]
+    names = {0: "a.png", 1: "b.png", 2: "c.png"}
+
+    for index in order:
+        await AgentClient.on_media(
+            client,
+            _room(),
+            _media_event(
+                body="captioned" if index == 0 else names[index],
+                filename=names[index] if index == 0 else None,
+                event_id=f"$part-{index}",
+                group={"id": "grp-2", "index": index, "total": 3},
+            ),
+        )
+
+    assert len(client.queue.events) == 1
+    payload = client.queue.events[0].payload
+    # Sorted by group index, not arrival order.
+    assert [a.filename for a in payload.attachments] == ["a.png", "b.png", "c.png"]
+    assert payload.body == "captioned"
+    assert client._attachment_groups == {}
+    assert client._attachment_group_timers == {}
+
+
+async def test_ungrouped_media_emits_immediately() -> None:
+    client = _fake_client()
+
+    await AgentClient.on_media(
+        client, _room(), _media_event(body="cat.png", mimetype="image/png")
+    )
+
+    assert len(client.queue.events) == 1
+    payload = client.queue.events[0].payload
+    assert [a.filename for a in payload.attachments] == ["cat.png"]
+    assert payload.body == "cat.png"
+    assert client._attachment_groups == {}
+    assert client._attachment_group_timers == {}
+
+
+async def test_incomplete_group_flushes_with_disclosed_notice() -> None:
+    original = ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS
+    ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS = 0.01
+    try:
+        client = _fake_client()
+        for index, name in [(0, "cat.png"), (1, "notes.md")]:
+            await AgentClient.on_media(
+                client,
+                _room(),
+                _media_event(
+                    body="two of three" if index == 0 else name,
+                    filename=name if index == 0 else None,
+                    event_id=f"$part-{index}",
+                    group={"id": "grp-3", "index": index, "total": 3},
+                ),
+            )
+        assert client.queue.events == []
+
+        await asyncio.sleep(0.15)
+    finally:
+        ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS = original
+
+    assert len(client.queue.events) == 1
+    payload = client.queue.events[0].payload
+    assert [a.filename for a in payload.attachments] == ["cat.png", "notes.md"]
+    assert "two of three" in payload.body
+    assert "incomplete attachment group: 2 of 3" in payload.body
+    # No leak: the buffer and its timer are gone once flushed.
+    assert client._attachment_groups == {}
+    assert client._attachment_group_timers == {}
+
+
+async def test_group_is_anchored_on_part_zero_not_the_completing_part() -> None:
+    """The coalesced message must carry part 0's event id, so a reply threads
+    off the canonical first event rather than whichever part landed last."""
+    client = _fake_client()
+    # Part 0 arrives FIRST, so a later part completes the group — otherwise the
+    # completing event happens to be part 0 and the assertion proves nothing.
+    for index, name in [(0, "a.png"), (1, "b.md"), (2, "c.csv")]:
+        await AgentClient.on_media(
+            client,
+            _room(),
+            _media_event(
+                body="three" if index == 0 else name,
+                filename=name if index == 0 else None,
+                event_id=f"$part-{index}",
+                group={"id": "grp-anchor", "index": index, "total": 3},
+            ),
+        )
+
+    assert len(client.queue.events) == 1
+    # Part 2 completed the group, but part 0 anchors the payload.
+    assert client.queue.events[0].payload.message_id == "$part-0"
+
+
+async def test_incomplete_group_is_anchored_on_part_zero() -> None:
+    original = ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS
+    ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS = 0.01
+    try:
+        client = _fake_client()
+        for index, name in [(0, "a.png"), (2, "c.csv")]:
+            await AgentClient.on_media(
+                client,
+                _room(),
+                _media_event(
+                    body="anchored" if index == 0 else name,
+                    filename=name if index == 0 else None,
+                    event_id=f"$part-{index}",
+                    group={"id": "grp-anchor-2", "index": index, "total": 3},
+                ),
+            )
+        await asyncio.sleep(0.15)
+    finally:
+        ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS = original
+
+    assert len(client.queue.events) == 1
+    assert client.queue.events[0].payload.message_id == "$part-0"
+
+
+async def test_group_timeout_bounds_the_group_not_the_gap_between_parts() -> None:
+    """The safety-net timer is armed once per group. A batch dribbling in just
+    under the timeout must not be able to hold the buffer open indefinitely."""
+    original = ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS
+    ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS = 0.12
+    try:
+        client = _fake_client()
+        await AgentClient.on_media(
+            client,
+            _room(),
+            _media_event(
+                body="slow batch",
+                filename="a.png",
+                event_id="$part-0",
+                group={"id": "grp-slow", "index": 0, "total": 4},
+            ),
+        )
+        first_timer = client._attachment_group_timers["grp-slow"]
+
+        # Parts keep trickling in below the deadline; the timer must NOT be
+        # pushed back by each arrival.
+        for index, name in [(1, "b.md"), (2, "c.csv")]:
+            await asyncio.sleep(0.05)
+            await AgentClient.on_media(
+                client,
+                _room(),
+                _media_event(
+                    body=name,
+                    event_id=f"$part-{index}",
+                    group={"id": "grp-slow", "index": index, "total": 4},
+                ),
+            )
+            assert client._attachment_group_timers["grp-slow"] is first_timer
+
+        await asyncio.sleep(0.15)
+    finally:
+        ac.ATTACHMENT_GROUP_TIMEOUT_SECONDS = original
+
+    # Fired on the group's own deadline rather than being extended forever.
+    assert len(client.queue.events) == 1
+    payload = client.queue.events[0].payload
+    assert "incomplete attachment group: 3 of 4" in payload.body
+    assert client._attachment_groups == {}
+    assert client._attachment_group_timers == {}
diff --git a/core/tests/switch_core/test_attachments.py b/core/tests/switch_core/test_attachments.py
new file mode 100644
index 000000000..fa1e8744c
--- /dev/null
+++ b/core/tests/switch_core/test_attachments.py
@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from switch_core.attachments import ATTACHMENT_GROUP_KEY, parse_attachment_group
+
+
+def _content(marker: Any) -> dict[str, object]:
+    return {"msgtype": "m.image", "body": "cat.png", ATTACHMENT_GROUP_KEY: marker}
+
+
+def test_valid_marker_returns_tuple() -> None:
+    assert parse_attachment_group(_content({"id": "grp", "index": 1, "total": 3})) == (
+        "grp",
+        1,
+        3,
+    )
+
+
+def test_group_of_one_is_valid() -> None:
+    assert parse_attachment_group(_content({"id": "grp", "index": 0, "total": 1})) == (
+        "grp",
+        0,
+        1,
+    )
+
+
+def test_missing_key_returns_none() -> None:
+    assert parse_attachment_group({"msgtype": "m.image", "body": "cat.png"}) is None
+
+
+@pytest.mark.parametrize("marker", ["not-a-dict", 7, None, [], ["id", 0, 1]])
+def test_non_dict_marker_returns_none(marker: Any) -> None:
+    assert parse_attachment_group(_content(marker)) is None
+
+
+@pytest.mark.parametrize(
+    "marker",
+    [
+        {"index": 0, "total": 2},  # missing id
+        {"id": "", "index": 0, "total": 2},  # blank id
+        {"id": 7, "index": 0, "total": 2},  # non-str id
+        {"id": "grp", "total": 2},  # missing index
+        {"id": "grp", "index": 0},  # missing total
+        {"id": "grp", "index": "0", "total": 2},  # non-int index
+        {"id": "grp", "index": 0, "total": "2"},  # non-int total
+        {"id": "grp", "index": 0.0, "total": 2},  # float index
+        {"id": "grp", "index": 0, "total": 0},  # total < 1
+        {"id": "grp", "index": 0, "total": -1},  # negative total
+        {"id": "grp", "index": -1, "total": 2},  # index < 0
+        {"id": "grp", "index": 2, "total": 2},  # index == total
+        {"id": "grp", "index": 5, "total": 2},  # index > total
+    ],
+)
+def test_malformed_marker_degrades_to_ungrouped(marker: dict[str, Any]) -> None:
+    # A bad marker must never raise: it degrades to an ungrouped attachment so
+    # one odd event can't stall a receiver's buffer waiting for parts that
+    # will never arrive.
+    assert parse_attachment_group(_content(marker)) is None
+
+
+def test_bool_index_or_total_is_rejected() -> None:
+    """bool is a subclass of int — a bool marker is malformed, not index 0/1."""
+    assert (
+        parse_attachment_group(
+            {ATTACHMENT_GROUP_KEY: {"id": "g", "index": False, "total": True}}
+        )
+        is None
+    )
+    assert (
+        parse_attachment_group(
+            {ATTACHMENT_GROUP_KEY: {"id": "g", "index": True, "total": 3}}
+        )
+        is None
+    )
+    assert (
+        parse_attachment_group(
+            {ATTACHMENT_GROUP_KEY: {"id": "g", "index": 0, "total": True}}
+        )
+        is None
+    )
diff --git a/core/uv.lock b/core/uv.lock
index 3c66c3702..44a6dddc6 100644
--- a/core/uv.lock
+++ b/core/uv.lock
@@ -2433,7 +2433,7 @@ wheels = [
 
 [[package]]
 name = "switch-core"
-version = "0.8.1"
+version = "0.9.0"
 source = { editable = "." }
 dependencies = [
     { name = "aiohttp" },