Compare commits

...

2 Commits

Author SHA1 Message Date
teknium1
1b5eb9df84 feat: Telegram send_document and send_video for native file attachments
Implement send_document() and send_video() overrides in TelegramAdapter
so the agent can deliver files (PDFs, CSVs, docs, etc.) and videos as
native Telegram attachments instead of just printing the file path as
text.

The base adapter already routes MEDIA:<path> tags by extension — audio
goes to send_voice(), images to send_image_file(), and everything else
falls through to send_document(). But TelegramAdapter didn't override
send_document() or send_video(), so those fell back to plain text.

Now when the agent includes MEDIA:/path/to/report.pdf in its response,
users get a proper downloadable file attachment in Telegram.

Features:
- send_document: sends files via bot.send_document with display name,
  caption (truncated to 1024), and reply_to support
- send_video: sends videos via bot.send_video with inline playback
- Both fall back to base class text if the Telegram API call fails
- 10 new tests covering success, custom filename, file-not-found,
  not-connected, caption truncation, API error fallback, and reply_to

Requested by @TigerHixTang on Twitter.
2026-03-09 12:17:35 -07:00
teknium1
c754135965 fix: banner wraps in narrow terminals (Kitty, small windows)
The full HERMES-AGENT ASCII logo needs ~95 columns, and the
side-by-side caduceus + tools panel needs ~80. In narrow terminals
(Kitty default, resized windows) everything wraps into visual garbage.

Fixes:
- show_banner() auto-detects terminal width and falls back to compact
  banner when < 80 columns
- build_welcome_banner() skips the ASCII logo when < 95 columns
- Compact banner now dynamically sized via _build_compact_banner()
  instead of a hardcoded 64-char box that also wrapped in narrow terms
- Same width checks applied to /clear command's banner refresh

The up/down arrow key issue in Kitty terminal for multiline input is
a known Kitty keyboard protocol (CSI u) vs prompt_toolkit compatibility
gap — arrow keys work correctly in standard terminals and tmux. Users
can work around it by running in tmux or setting TERM=xterm-256color.
2026-03-09 05:57:36 -07:00
3 changed files with 295 additions and 7 deletions

43
cli.py
View File

@@ -725,6 +725,7 @@ HERMES_CADUCEUS = """[#CD7F32]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⢀⣀⡀⠀⣀⣀
[#B8860B]⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠈⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀⠀[/]"""
# Compact banner for smaller terminals (fallback)
# Note: built dynamically by _build_compact_banner() to fit terminal width
COMPACT_BANNER = """
[bold #FFD700]╔══════════════════════════════════════════════════════════════╗[/]
[bold #FFD700]║[/] [#FFBF00]⚕ NOUS HERMES[/] [dim #B8860B]- AI Agent Framework[/] [bold #FFD700]║[/]
@@ -733,6 +734,26 @@ COMPACT_BANNER = """
"""
def _build_compact_banner() -> str:
"""Build a compact banner that fits the current terminal width."""
w = min(shutil.get_terminal_size().columns - 2, 64)
if w < 30:
return "\n[#FFBF00]⚕ NOUS HERMES[/] [dim #B8860B]- Nous Research[/]\n"
inner = w - 2 # inside the box border
bar = "" * w
line1 = "⚕ NOUS HERMES - AI Agent Framework"
line2 = "Messenger of the Digital Gods · Nous Research"
# Truncate and pad to fit
line1 = line1[:inner - 2].ljust(inner - 2)
line2 = line2[:inner - 2].ljust(inner - 2)
return (
f"\n[bold #FFD700]╔{bar}╗[/]\n"
f"[bold #FFD700]║[/] [#FFBF00]{line1}[/] [bold #FFD700]║[/]\n"
f"[bold #FFD700]║[/] [dim #B8860B]{line2}[/] [bold #FFD700]║[/]\n"
f"[bold #FFD700]╚{bar}╝[/]\n"
)
def _get_available_skills() -> Dict[str, List[str]]:
"""
Scan ~/.hermes/skills/ and return skills grouped by category.
@@ -930,10 +951,12 @@ def build_welcome_banner(console: Console, model: str, cwd: str, tools: List[dic
padding=(0, 2),
)
# Print the big HERMES-AGENT logo first (no panel wrapper for full width)
console.print()
console.print(HERMES_AGENT_LOGO)
# Print the big HERMES-AGENT logo — skip if terminal is too narrow
console.print()
term_width = shutil.get_terminal_size().columns
if term_width >= 95:
console.print(HERMES_AGENT_LOGO)
console.print()
# Print the panel with caduceus and info
console.print(outer_panel)
@@ -1383,8 +1406,13 @@ class HermesCLI:
"""Display the welcome banner in Claude Code style."""
self.console.clear()
if self.compact:
self.console.print(COMPACT_BANNER)
# Auto-compact for narrow terminals — the full banner with caduceus
# + tool list needs ~80 columns minimum to render without wrapping.
term_width = shutil.get_terminal_size().columns
use_compact = self.compact or term_width < 80
if use_compact:
self.console.print(_build_compact_banner())
self._show_status()
else:
# Get tools for display
@@ -2394,8 +2422,9 @@ class HermesCLI:
# and gets mangled by patch_stdout).
if self._app:
cc = ChatConsole()
if self.compact:
cc.print(COMPACT_BANNER)
term_w = shutil.get_terminal_size().columns
if self.compact or term_w < 80:
cc.print(_build_compact_banner())
else:
tools = get_tool_definitions(enabled_toolsets=self.enabled_toolsets, quiet_mode=True)
cwd = os.getenv("TERMINAL_CWD", os.getcwd())

View File

@@ -346,6 +346,64 @@ class TelegramAdapter(BasePlatformAdapter):
print(f"[{self.name}] Failed to send local image: {e}")
return await super().send_image_file(chat_id, image_path, caption, reply_to)
async def send_document(
self,
chat_id: str,
file_path: str,
caption: Optional[str] = None,
file_name: Optional[str] = None,
reply_to: Optional[str] = None,
) -> SendResult:
"""Send a document/file natively as a Telegram file attachment."""
if not self._bot:
return SendResult(success=False, error="Not connected")
try:
if not os.path.exists(file_path):
return SendResult(success=False, error=f"File not found: {file_path}")
display_name = file_name or os.path.basename(file_path)
with open(file_path, "rb") as f:
msg = await self._bot.send_document(
chat_id=int(chat_id),
document=f,
filename=display_name,
caption=caption[:1024] if caption else None,
reply_to_message_id=int(reply_to) if reply_to else None,
)
return SendResult(success=True, message_id=str(msg.message_id))
except Exception as e:
print(f"[{self.name}] Failed to send document: {e}")
return await super().send_document(chat_id, file_path, caption, file_name, reply_to)
async def send_video(
self,
chat_id: str,
video_path: str,
caption: Optional[str] = None,
reply_to: Optional[str] = None,
) -> SendResult:
"""Send a video natively as a Telegram video message."""
if not self._bot:
return SendResult(success=False, error="Not connected")
try:
if not os.path.exists(video_path):
return SendResult(success=False, error=f"Video file not found: {video_path}")
with open(video_path, "rb") as f:
msg = await self._bot.send_video(
chat_id=int(chat_id),
video=f,
caption=caption[:1024] if caption else None,
reply_to_message_id=int(reply_to) if reply_to else None,
)
return SendResult(success=True, message_id=str(msg.message_id))
except Exception as e:
print(f"[{self.name}] Failed to send video: {e}")
return await super().send_video(chat_id, video_path, caption, reply_to)
async def send_image(
self,
chat_id: str,

View File

@@ -20,6 +20,7 @@ from gateway.config import Platform, PlatformConfig
from gateway.platforms.base import (
MessageEvent,
MessageType,
SendResult,
SUPPORTED_DOCUMENT_TYPES,
)
@@ -336,3 +337,203 @@ class TestDocumentDownloadBlock:
await adapter._handle_media_message(update, MagicMock())
# handle_message should still be called (the handler catches the exception)
adapter.handle_message.assert_called_once()
# ---------------------------------------------------------------------------
# TestSendDocument — outbound file attachment delivery
# ---------------------------------------------------------------------------
class TestSendDocument:
"""Tests for TelegramAdapter.send_document() — sending files to users."""
@pytest.fixture()
def connected_adapter(self, adapter):
"""Adapter with a mock bot attached."""
bot = AsyncMock()
adapter._bot = bot
return adapter
@pytest.mark.asyncio
async def test_send_document_success(self, connected_adapter, tmp_path):
"""A local file is sent via bot.send_document and returns success."""
# Create a real temp file
test_file = tmp_path / "report.pdf"
test_file.write_bytes(b"%PDF-1.4 fake content")
mock_msg = MagicMock()
mock_msg.message_id = 99
connected_adapter._bot.send_document = AsyncMock(return_value=mock_msg)
result = await connected_adapter.send_document(
chat_id="12345",
file_path=str(test_file),
caption="Here's the report",
)
assert result.success is True
assert result.message_id == "99"
connected_adapter._bot.send_document.assert_called_once()
call_kwargs = connected_adapter._bot.send_document.call_args[1]
assert call_kwargs["chat_id"] == 12345
assert call_kwargs["filename"] == "report.pdf"
assert call_kwargs["caption"] == "Here's the report"
@pytest.mark.asyncio
async def test_send_document_custom_filename(self, connected_adapter, tmp_path):
"""The file_name parameter overrides the basename for display."""
test_file = tmp_path / "doc_abc123_ugly.csv"
test_file.write_bytes(b"a,b,c\n1,2,3")
mock_msg = MagicMock()
mock_msg.message_id = 100
connected_adapter._bot.send_document = AsyncMock(return_value=mock_msg)
result = await connected_adapter.send_document(
chat_id="12345",
file_path=str(test_file),
file_name="clean_data.csv",
)
assert result.success is True
call_kwargs = connected_adapter._bot.send_document.call_args[1]
assert call_kwargs["filename"] == "clean_data.csv"
@pytest.mark.asyncio
async def test_send_document_file_not_found(self, connected_adapter):
"""Missing file returns error without calling Telegram API."""
result = await connected_adapter.send_document(
chat_id="12345",
file_path="/nonexistent/file.pdf",
)
assert result.success is False
assert "not found" in result.error.lower()
connected_adapter._bot.send_document.assert_not_called()
@pytest.mark.asyncio
async def test_send_document_not_connected(self, adapter):
"""If bot is None, returns not connected error."""
result = await adapter.send_document(
chat_id="12345",
file_path="/some/file.pdf",
)
assert result.success is False
assert "Not connected" in result.error
@pytest.mark.asyncio
async def test_send_document_caption_truncated(self, connected_adapter, tmp_path):
"""Captions longer than 1024 chars are truncated."""
test_file = tmp_path / "data.json"
test_file.write_bytes(b"{}")
mock_msg = MagicMock()
mock_msg.message_id = 101
connected_adapter._bot.send_document = AsyncMock(return_value=mock_msg)
long_caption = "x" * 2000
await connected_adapter.send_document(
chat_id="12345",
file_path=str(test_file),
caption=long_caption,
)
call_kwargs = connected_adapter._bot.send_document.call_args[1]
assert len(call_kwargs["caption"]) == 1024
@pytest.mark.asyncio
async def test_send_document_api_error_falls_back(self, connected_adapter, tmp_path):
"""If Telegram API raises, falls back to base class text message."""
test_file = tmp_path / "file.pdf"
test_file.write_bytes(b"data")
connected_adapter._bot.send_document = AsyncMock(
side_effect=RuntimeError("Telegram API error")
)
# The base fallback calls self.send() which is also on _bot, so mock it
# to avoid cascading errors.
connected_adapter.send = AsyncMock(
return_value=SendResult(success=True, message_id="fallback")
)
result = await connected_adapter.send_document(
chat_id="12345",
file_path=str(test_file),
)
# Should have fallen back to base class
assert result.success is True
assert result.message_id == "fallback"
@pytest.mark.asyncio
async def test_send_document_reply_to(self, connected_adapter, tmp_path):
"""reply_to parameter is forwarded as reply_to_message_id."""
test_file = tmp_path / "spec.md"
test_file.write_bytes(b"# Spec")
mock_msg = MagicMock()
mock_msg.message_id = 102
connected_adapter._bot.send_document = AsyncMock(return_value=mock_msg)
await connected_adapter.send_document(
chat_id="12345",
file_path=str(test_file),
reply_to="50",
)
call_kwargs = connected_adapter._bot.send_document.call_args[1]
assert call_kwargs["reply_to_message_id"] == 50
# ---------------------------------------------------------------------------
# TestSendVideo — outbound video delivery
# ---------------------------------------------------------------------------
class TestSendVideo:
"""Tests for TelegramAdapter.send_video() — sending videos to users."""
@pytest.fixture()
def connected_adapter(self, adapter):
bot = AsyncMock()
adapter._bot = bot
return adapter
@pytest.mark.asyncio
async def test_send_video_success(self, connected_adapter, tmp_path):
test_file = tmp_path / "clip.mp4"
test_file.write_bytes(b"\x00\x00\x00\x1c" + b"ftyp" + b"\x00" * 100)
mock_msg = MagicMock()
mock_msg.message_id = 200
connected_adapter._bot.send_video = AsyncMock(return_value=mock_msg)
result = await connected_adapter.send_video(
chat_id="12345",
video_path=str(test_file),
caption="Check this out",
)
assert result.success is True
assert result.message_id == "200"
connected_adapter._bot.send_video.assert_called_once()
@pytest.mark.asyncio
async def test_send_video_file_not_found(self, connected_adapter):
result = await connected_adapter.send_video(
chat_id="12345",
video_path="/nonexistent/video.mp4",
)
assert result.success is False
assert "not found" in result.error.lower()
@pytest.mark.asyncio
async def test_send_video_not_connected(self, adapter):
result = await adapter.send_video(
chat_id="12345",
video_path="/some/video.mp4",
)
assert result.success is False
assert "Not connected" in result.error