feat(gateway): deliverable mode — ship artifacts as native uploads from any agent surface (#27813)

The agent can now produce a chart, PDF, spreadsheet, or any other supported
file type and have it land in Slack / Discord / Telegram / WhatsApp / etc.
as a native attachment, just by mentioning the absolute path in its
response. Same primitive works for kanban-worker completions: workers
attach artifacts via kanban_complete(artifacts=[...]) and the gateway
notifier uploads them alongside the completion message.

Changes:

- gateway/platforms/base.py: extract_local_files now covers PDFs, docx,
  spreadsheets (xlsx/csv/json/yaml), presentations (pptx), archives
  (zip/tar/gz), audio (mp3/wav/...), and html — not just images and video.
  Image/video extensions still embed inline; everything else routes to
  send_document via the existing dispatch partition in gateway/run.py.

- tools/kanban_tools.py + hermes_cli/kanban_db.py: kanban_complete gains
  an explicit ``artifacts`` parameter. The handler stashes it in
  metadata.artifacts (for downstream workers) and the kernel promotes
  it onto the completed-event payload so the notifier can find it
  without a second SQL round-trip.

- gateway/run.py: _kanban_notifier_watcher now calls a new helper
  _deliver_kanban_artifacts after sending the completion text. The
  helper reads payload.artifacts (preferred), falls back to scanning
  the payload summary and task.result with extract_local_files, then
  partitions images / videos / documents and uploads each via
  send_multiple_images / send_video / send_document.

- website/docs/user-guide/features/deliverable-mode.md + sidebars.ts:
  user-facing docs page covering the extension list, the kanban
  artifacts pattern, and the MCP-for-connector-breadth recommendation.

Tests:

- tests/gateway/test_extract_local_files.py: 7 new test cases
  (documents, spreadsheets, presentations, audio, archives, html,
  chart-pdf canonical case). 44 passing, 0 regressions.
- tests/tools/test_kanban_tools.py: 4 new cases covering the artifacts
  arg shape (list / string / merge with existing metadata / type
  rejection). 17 passing.
- tests/hermes_cli/test_kanban_notify.py: 2 new cases covering full
  notifier → artifact-upload path and missing-file silent-skip. 12
  passing.
- E2E (real files, real kanban kernel, real BasePlatformAdapter):
  worker calls kanban_complete(artifacts=[png,pdf,csv]) → metadata +
  event payload land → notifier helper partitions correctly →
  send_multiple_images called once with the PNG, send_document called
  twice with PDF + CSV.

What's NOT in this PR (deferred to follow-ups):

- Ad-hoc "research this for two hours, ping the thread when done"
  slash command — covered today by kanban subscriptions; a dedicated
  slash command can ride a follow-up PR if needed.
- Setup-wizard prompt for recommended MCP servers (Notion, GitHub,
  Linear, etc.) — docs page lists them; UI is a separate change.

Plan and rationale captured in ~/.hermes/docs/perplexity-computer-parity.pdf
(local doc, not shipped).

tests/gateway/test_extract_local_files.py

@@ -74,6 +74,58 @@ class TestBasicDetection:
             assert len(paths) == 1, f"Failed for {ext}"
             assert paths[0] == f"/tmp/pic{ext}"
 
+    def test_document_extensions(self):
+        """Documents (PDF, Word, plain text, etc.) ship as file uploads."""
+        for ext in (".pdf", ".docx", ".doc", ".odt", ".rtf", ".txt", ".md"):
+            text = f"Report at /tmp/report{ext} attached"
+            paths, _ = _extract(text)
+            assert len(paths) == 1, f"Failed for {ext}"
+            assert paths[0] == f"/tmp/report{ext}"
+
+    def test_spreadsheet_and_data_extensions(self):
+        """Spreadsheets and structured data ship as file uploads."""
+        for ext in (".xlsx", ".xls", ".csv", ".tsv", ".json", ".xml", ".yaml", ".yml"):
+            text = f"Data at /tmp/data{ext} ready"
+            paths, _ = _extract(text)
+            assert len(paths) == 1, f"Failed for {ext}"
+            assert paths[0] == f"/tmp/data{ext}"
+
+    def test_presentation_extensions(self):
+        """Presentations ship as file uploads."""
+        for ext in (".pptx", ".ppt", ".odp"):
+            text = f"Deck at /tmp/deck{ext} done"
+            paths, _ = _extract(text)
+            assert len(paths) == 1, f"Failed for {ext}"
+            assert paths[0] == f"/tmp/deck{ext}"
+
+    def test_audio_extensions(self):
+        """Audio files are detected and routed by the gateway dispatch."""
+        for ext in (".mp3", ".wav", ".ogg", ".m4a", ".flac"):
+            text = f"Audio at /tmp/sound{ext} ready"
+            paths, _ = _extract(text)
+            assert len(paths) == 1, f"Failed for {ext}"
+            assert paths[0] == f"/tmp/sound{ext}"
+
+    def test_archive_extensions(self):
+        """Archives ship as file uploads."""
+        for ext in (".zip", ".tar", ".gz", ".tgz", ".bz2", ".7z"):
+            text = f"Archive at /tmp/bundle{ext} ready"
+            paths, _ = _extract(text)
+            assert len(paths) == 1, f"Failed for {ext}"
+            assert paths[0] == f"/tmp/bundle{ext}"
+
+    def test_html_extension(self):
+        paths, _ = _extract("Open /tmp/report.html in browser")
+        assert paths == ["/tmp/report.html"]
+
+    def test_chart_pdf_path(self):
+        """Common case: agent renders a chart via matplotlib and references the file."""
+        text = "Here is the comparison chart: /tmp/q3-sales.pdf"
+        paths, cleaned = _extract(text)
+        assert paths == ["/tmp/q3-sales.pdf"]
+        assert "/tmp/q3-sales.pdf" not in cleaned
+        assert "comparison chart" in cleaned
+
     def test_case_insensitive_extension(self):
         paths, _ = _extract("See /tmp/PHOTO.PNG and /tmp/vid.MP4 now")
         assert len(paths) == 2
@@ -269,8 +321,15 @@ class TestEdgeCases:
         assert cleaned == ""
 
     def test_no_media_extensions(self):
-        """Non-media extensions should not be matched."""
-        paths, _ = _extract("See /tmp/data.csv and /tmp/script.py and /tmp/notes.txt")
+        """Extensions outside the supported list should not be matched.
+
+        ``.py`` and ``.log`` are intentionally excluded because (a) most
+        source files are quoted in inline code or fenced blocks anyway,
+        and (b) auto-shipping arbitrary source files would be a
+        surprise.  Documents (.pdf, .docx), data (.csv, .json),
+        archives (.zip), and presentations (.pptx) ARE matched.
+        """
+        paths, _ = _extract("See /tmp/script.py and /tmp/server.log here")
         assert paths == []
 
     def test_path_with_spaces_not_matched(self):

tests/hermes_cli/test_kanban_notify.py

@@ -479,3 +479,162 @@ async def test_gateway_create_autosubscribes_on_explicit_board(kanban_home):
         assert kb.list_notify_subs(conn) == []
     finally:
         conn.close()
+
+
+@pytest.mark.asyncio
+async def test_notifier_uploads_artifacts_on_completion(kanban_home, tmp_path):
+    """When a completed event carries ``artifacts`` in its payload, the
+    notifier uploads each file to the subscribed chat as a native
+    attachment. Images batch through send_multiple_images; documents
+    route through send_document. See the artifacts wiring in
+    gateway/run.py._deliver_kanban_artifacts.
+    """
+    import hermes_cli.kanban_db as kb
+    from gateway.run import GatewayRunner
+    from gateway.config import Platform
+    from tools import kanban_tools as kt
+
+    # Materialize real files so os.path.isfile passes inside the helper.
+    chart_path = tmp_path / "q3-revenue.png"
+    chart_path.write_bytes(b"PNG-fake-bytes")
+    report_path = tmp_path / "report.pdf"
+    report_path.write_bytes(b"%PDF-fake")
+
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="render q3 chart", assignee="worker1")
+        kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1")
+    finally:
+        conn.close()
+
+    # Use the production handler so we exercise the full path: tool args
+    # → metadata.artifacts → event payload promotion.
+    import os
+    os.environ["HERMES_KANBAN_TASK"] = tid
+    try:
+        out = kt._handle_complete({
+            "summary": "rendered the chart",
+            "artifacts": [str(chart_path), str(report_path)],
+        })
+    finally:
+        os.environ.pop("HERMES_KANBAN_TASK", None)
+    import json as _json
+    assert _json.loads(out)["ok"] is True
+
+    runner = object.__new__(GatewayRunner)
+    runner._running = True
+    runner._kanban_sub_fail_counts = {}
+
+    fake_adapter = MagicMock()
+    fake_adapter.name = "telegram"
+
+    sends: list = []
+    images_uploaded: list = []
+    documents_uploaded: list = []
+
+    async def _send(chat_id, msg, metadata=None):
+        sends.append((chat_id, msg))
+        runner._running = False
+
+    async def _send_images(chat_id, images, metadata=None, **_kw):
+        images_uploaded.extend(p for p, _ in images)
+
+    async def _send_document(chat_id, file_path, metadata=None, **_kw):
+        documents_uploaded.append(file_path)
+
+    fake_adapter.send = AsyncMock(side_effect=_send)
+    fake_adapter.send_multiple_images = AsyncMock(side_effect=_send_images)
+    fake_adapter.send_document = AsyncMock(side_effect=_send_document)
+    # extract_local_files is used internally for legacy path fallback;
+    # the real BasePlatformAdapter implementation lives there, so wire it.
+    from gateway.platforms.base import BasePlatformAdapter
+    fake_adapter.extract_local_files = BasePlatformAdapter.extract_local_files
+
+    runner.adapters = {Platform.TELEGRAM: fake_adapter}
+
+    _orig_sleep = asyncio.sleep
+
+    async def _fast_sleep(_):
+        await _orig_sleep(0)
+
+    with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep):
+        await asyncio.wait_for(
+            runner._kanban_notifier_watcher(interval=1),
+            timeout=10.0,
+        )
+
+    # The text completion notification fired.
+    assert len(sends) == 1
+    # The PNG rode the image-batch path.
+    assert any("q3-revenue.png" in p for p in images_uploaded), images_uploaded
+    # The PDF rode the document path.
+    assert any("report.pdf" in p for p in documents_uploaded), documents_uploaded
+
+
+@pytest.mark.asyncio
+async def test_notifier_artifact_delivery_skips_missing_files(kanban_home, tmp_path):
+    """Missing artifact paths are silently skipped — they may have been
+    referenced by name only. The notifier must not crash and must still
+    deliver any artifacts that do exist."""
+    import hermes_cli.kanban_db as kb
+    from gateway.run import GatewayRunner
+    from gateway.config import Platform
+    from tools import kanban_tools as kt
+
+    real_pdf = tmp_path / "real.pdf"
+    real_pdf.write_bytes(b"%PDF-fake")
+
+    conn = kb.connect()
+    try:
+        tid = kb.create_task(conn, title="t", assignee="worker1")
+        kb.add_notify_sub(conn, task_id=tid, platform="telegram", chat_id="chat1")
+    finally:
+        conn.close()
+
+    import os
+    os.environ["HERMES_KANBAN_TASK"] = tid
+    try:
+        kt._handle_complete({
+            "summary": "one real, one ghost",
+            "artifacts": [str(real_pdf), "/tmp/definitely-does-not-exist.pdf"],
+        })
+    finally:
+        os.environ.pop("HERMES_KANBAN_TASK", None)
+
+    runner = object.__new__(GatewayRunner)
+    runner._running = True
+    runner._kanban_sub_fail_counts = {}
+
+    fake_adapter = MagicMock()
+    fake_adapter.name = "telegram"
+
+    documents_uploaded: list = []
+
+    async def _send(chat_id, msg, metadata=None):
+        runner._running = False
+
+    async def _send_document(chat_id, file_path, metadata=None, **_kw):
+        documents_uploaded.append(file_path)
+
+    fake_adapter.send = AsyncMock(side_effect=_send)
+    fake_adapter.send_document = AsyncMock(side_effect=_send_document)
+    fake_adapter.send_multiple_images = AsyncMock()
+    from gateway.platforms.base import BasePlatformAdapter
+    fake_adapter.extract_local_files = BasePlatformAdapter.extract_local_files
+
+    runner.adapters = {Platform.TELEGRAM: fake_adapter}
+
+    _orig_sleep = asyncio.sleep
+
+    async def _fast_sleep(_):
+        await _orig_sleep(0)
+
+    with patch("gateway.run.asyncio.sleep", side_effect=_fast_sleep):
+        await asyncio.wait_for(
+            runner._kanban_notifier_watcher(interval=1),
+            timeout=10.0,
+        )
+
+    # Only the real file was uploaded.
+    assert len(documents_uploaded) == 1
+    assert "real.pdf" in documents_uploaded[0]

tests/tools/test_kanban_tools.py

@@ -318,6 +318,93 @@ def test_complete_with_result_only(worker_env):
     assert d["ok"] is True
 
 
+def test_complete_with_artifacts_lands_in_event_payload(worker_env):
+    """``artifacts=[...]`` rides into the completed event payload so the
+    gateway notifier can upload them as native attachments. See the
+    kanban notifier in gateway/run.py for the consumer side."""
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    out = kt._handle_complete({
+        "summary": "rendered the chart",
+        "artifacts": ["/tmp/q3-revenue.png", "/tmp/q3-report.pdf"],
+    })
+    assert json.loads(out)["ok"] is True
+
+    conn = kb.connect()
+    try:
+        events = kb.list_events(conn, worker_env)
+        # Find the completion event
+        completed = [e for e in events if e.kind == "completed"]
+        assert len(completed) == 1
+        payload = completed[0].payload or {}
+        assert payload.get("artifacts") == [
+            "/tmp/q3-revenue.png",
+            "/tmp/q3-report.pdf",
+        ]
+        # And the artifacts also live on metadata for downstream workers
+        run = kb.latest_run(conn, worker_env)
+        assert run.metadata.get("artifacts") == [
+            "/tmp/q3-revenue.png",
+            "/tmp/q3-report.pdf",
+        ]
+    finally:
+        conn.close()
+
+
+def test_complete_artifacts_accepts_single_string(worker_env):
+    """A bare string is auto-promoted to a single-element list for convenience."""
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    out = kt._handle_complete({
+        "summary": "one chart",
+        "artifacts": "/tmp/chart.png",
+    })
+    assert json.loads(out)["ok"] is True
+
+    conn = kb.connect()
+    try:
+        run = kb.latest_run(conn, worker_env)
+        assert run.metadata.get("artifacts") == ["/tmp/chart.png"]
+    finally:
+        conn.close()
+
+
+def test_complete_artifacts_merges_with_explicit_metadata_field(worker_env):
+    """If the worker passes metadata.artifacts AND the top-level artifacts
+    param, merge the two without duplicates."""
+    from hermes_cli import kanban_db as kb
+    from tools import kanban_tools as kt
+
+    out = kt._handle_complete({
+        "summary": "merged",
+        "metadata": {"artifacts": ["/tmp/a.png"], "other": "fact"},
+        "artifacts": ["/tmp/b.pdf", "/tmp/a.png"],
+    })
+    assert json.loads(out)["ok"] is True
+
+    conn = kb.connect()
+    try:
+        run = kb.latest_run(conn, worker_env)
+        # Order: existing entries first, then new ones, deduplicated.
+        assert run.metadata.get("artifacts") == ["/tmp/a.png", "/tmp/b.pdf"]
+        assert run.metadata.get("other") == "fact"
+    finally:
+        conn.close()
+
+
+def test_complete_rejects_non_list_artifacts(worker_env):
+    """Non-list, non-string artifacts should be rejected with a clear error."""
+    from tools import kanban_tools as kt
+    out = kt._handle_complete({
+        "summary": "bad shape",
+        "artifacts": {"not": "a list"},
+    })
+    err = json.loads(out).get("error", "")
+    assert "artifacts must be a list" in err
+
+
 def test_complete_rejects_no_handoff(worker_env):
     from tools import kanban_tools as kt
     out = kt._handle_complete({})