diff --git a/config.sample.yaml b/config.sample.yaml index 7ea6603..ac6d9dc 100644 --- a/config.sample.yaml +++ b/config.sample.yaml @@ -10,3 +10,9 @@ trusted_contacts: number: "+15559876543" - name: "Bob" number: "+15551112222" + +# Working directory for Goose sessions (absolute path recommended) +working_directory: "/home/user/projects" + +# Forward Goose's chain-of-thought reasoning to Signal (can be noisy) +forward_thoughts: false diff --git a/docs/plan.md b/docs/plan.md new file mode 100644 index 0000000..2386600 --- /dev/null +++ b/docs/plan.md @@ -0,0 +1,233 @@ +# Honker ACP Bridge — Implementation Plan + +## Current State + +What's implemented today: +- ✅ Initialize ACP connection to Goose (`goose acp` over stdio) +- ✅ Create new sessions (one per Signal contact) +- ✅ Send text prompts and collect text responses +- ✅ Interactive permission requests over Signal (numbered options) +- ✅ Signal-cli lifecycle management and socket communication + +--- + +## Phase 1: Essential — Make the Bridge Robust + +### 1.1 Terminal Execution (Client ← Agent) +**Priority: HIGH — Goose cannot function without this** + +Goose's primary tool is running shell commands. The ACP protocol has the agent +ask the *client* to execute commands via `create_terminal`, then polls output +with `terminal_output` and `wait_for_terminal_exit`. + +Currently we return `None` for all terminal methods, which means Goose can't +run any commands. We need to: + +- `create_terminal(command, args, cwd, env)` → spawn a real subprocess, return + a `terminal_id` +- `terminal_output(terminal_id)` → return accumulated stdout/stderr +- `wait_for_terminal_exit(terminal_id)` → await process completion, return exit code +- `release_terminal(terminal_id)` → clean up +- `kill_terminal(terminal_id)` → kill the subprocess + +Implementation: a `TerminalManager` class that tracks running subprocesses +keyed by terminal_id. + +### 1.2 File System Operations (Client ← Agent) +**Priority: HIGH — Goose cannot edit code without this** + +The agent asks the client to read/write files: + +- `read_text_file(path, line, limit)` → read from the local filesystem, return content +- `write_text_file(path, content)` → write to the local filesystem + +Currently we return `None`, so Goose can't read or write files. +These should be straightforward to implement against the local filesystem. + +We should also advertise these capabilities in `ClientCapabilities` during +`initialize`: +```python +ClientCapabilities( + fs=FileSystemCapabilities(read_text_file=True, write_text_file=True), + terminal=True, +) +``` + +### 1.3 Cancellation +**Priority: MEDIUM** + +Allow Signal users to cancel a running prompt. The user could send a special +command (e.g. `/cancel`) while a prompt is in progress. We'd call +`conn.cancel(session_id)` to tell Goose to stop. + +The `PromptResponse.stop_reason` already distinguishes `cancelled` from +`end_turn`, `max_tokens`, `refusal`, and `max_turn_requests` — we should +handle each appropriately in the reply. + +### 1.4 Stop Reason Handling +**Priority: MEDIUM** + +Currently we only look at the response text. We should handle all stop reasons: +- `end_turn` — normal completion (current behavior) +- `max_tokens` — context limit hit; notify user +- `max_turn_requests` — agent hit its turn limit; notify user +- `refusal` — agent refused the request; notify user +- `cancelled` — user cancelled; acknowledge + +--- + +## Phase 2: Enriched Feedback + +### 2.1 Tool Call Notifications +**Priority: MEDIUM** + +Forward tool activity as status messages so the Signal user knows what Goose +is doing while working. On `ToolCallStart`, send a brief message like: +"🔧 Running: `ls -la /src`" + +On `ToolCallProgress` with status updates, optionally send progress. We need +to be careful not to spam — batch or throttle these. + +### 2.2 Agent Thoughts +**Priority: LOW-MEDIUM** + +`AgentThoughtChunk` contains the agent's chain-of-thought reasoning. Could +forward these optionally (config flag) as italicized or prefixed messages. +Useful for debugging but noisy for everyday use. + +### 2.3 Plan Updates +**Priority: LOW-MEDIUM** + +`AgentPlanUpdate` sends a structured plan with entries (content, priority, +status). Could render as a checklist: +``` +📋 Plan: + ✅ Read the project structure + 🔄 Implement the new feature + ⬚ Write tests + ⬚ Update documentation +``` + +### 2.4 Usage/Cost Updates +**Priority: LOW** + +`UsageUpdate` reports token usage (size, used) and optional `Cost` (amount, +currency). Could send periodic summaries or on-demand via a `/usage` command. + +--- + +## Phase 3: Session Management + +### 3.1 Session Persistence (Load/Resume) +**Priority: MEDIUM** + +Goose supports `load_session` and `resume_session`. We could persist the +session_id→contact mapping in SQLite (as the README mentions) and restore +sessions across honker restarts. + +- `load_session(cwd, session_id)` — restore conversation history +- `resume_session(cwd, session_id)` — resume an interrupted session +- `list_sessions()` — list available sessions +- `close_session(session_id)` — explicitly end a session + +User commands: `/sessions`, `/load `, `/new` (force new session) + +### 3.2 Session Commands +**Priority: MEDIUM** + +Signal user sends special commands (prefix with `/`): +- `/new` — start a fresh session +- `/sessions` — list previous sessions +- `/load ` — resume a previous session +- `/cancel` — cancel current prompt +- `/mode ` — change session mode (via `set_session_mode`) +- `/model ` — change model (via `set_session_model`) +- `/usage` — show token usage / cost + +### 3.3 Session Modes +**Priority: LOW** + +Goose supports modes (e.g., "auto" vs "manual" approval). `set_session_mode` +and `CurrentModeUpdate` notifications. Could expose via `/mode` command. + +### 3.4 Session Configuration +**Priority: LOW** + +`set_config_option` lets the client change boolean/select config options. +`ConfigOptionUpdate` notifications report changes. Could expose via a +`/config` command. + +--- + +## Phase 4: Rich Content + +### 4.1 Image Support (Input) +**Priority: MEDIUM** + +Signal users can send images. signal-cli provides attachment data. ACP +supports `ImageContentBlock` in prompts (with base64 `data` and `mime_type`). +We could forward Signal image attachments to Goose as image content blocks. + +### 4.2 Image Support (Output) +**Priority: LOW** + +If Goose responds with `ImageContentBlock` in agent messages, we could send +them as Signal attachments. Less common but possible. + +### 4.3 File/Resource Content +**Priority: LOW** + +`ResourceContentBlock` and `EmbeddedResourceContentBlock` in agent messages. +Could send file contents as Signal attachments or inline code blocks. + +### 4.4 Audio Support +**Priority: LOW** + +ACP has `AudioContentBlock`. Signal supports voice messages. Theoretically +bridgeable but Goose doesn't currently use audio. + +--- + +## Phase 5: Advanced / Optional + +### 5.1 Multi-User Concurrency +**Priority: MEDIUM** + +Currently each contact gets their own session, and prompts run as async tasks. +But there's no protection against a user sending a second message while the +first is still processing. Options: +- Queue messages per-user (process sequentially) +- Allow concurrent prompts (may confuse Goose session state) +- Tell the user "please wait" if a prompt is in progress + +### 5.2 Group Messaging +**Priority: LOW** + +Signal groups could map to shared Goose sessions. Would need the account +number (already in config) for group message handling. signal-cli has +different envelope structure for group messages. + +### 5.3 Extension Methods +**Priority: LOW** + +`ext_method` and `ext_notification` are catch-alls for custom protocol +extensions. Log them for now; implement specific ones if Goose uses them. + +### 5.4 MCP Server Configuration +**Priority: LOW** + +`new_session` accepts `mcp_servers` to connect external tool servers. Could +be configured per-contact or globally in honker config. + +--- + +## Decisions Made + +1. **Terminal execution**: Permission gate is enough. No additional sandboxing. +2. **Tool call notifications**: Light — send "🔧 Running: ..." on tool start. + Goose already keeps extensive logs at `~/.local/state/goose/logs/`. +3. **Agent thoughts**: Opt-in via config flag (`forward_thoughts: true`). +4. **Cancellation UX**: `/cancel` plus general `/command` prefix convention. +5. **Multi-user concurrency**: "Please wait" — reject concurrent prompts per user. +6. **Image attachments**: Yes, wire up in Phase 1. +7. **Working directory**: Global config option (`working_directory`). diff --git a/src/honker/bridge.py b/src/honker/bridge.py index c1128f7..c5e2f80 100644 --- a/src/honker/bridge.py +++ b/src/honker/bridge.py @@ -1,7 +1,9 @@ """Bridge — ties Signal and Goose together.""" import asyncio +import base64 import logging +from pathlib import Path from honker.config import Config from honker.goose import GooseConnection @@ -29,7 +31,6 @@ def _split_message(text: str, max_len: int = MAX_SIGNAL_MESSAGE_LENGTH) -> list[ # Try to break at a newline split_at = text.rfind("\n", 0, max_len) if split_at == -1: - # No newline found, break at max_len split_at = max_len chunks.append(text[:split_at]) text = text[split_at:].lstrip("\n") @@ -47,16 +48,28 @@ class Bridge: # Pending interactive questions: source phone -> Future[str] self._pending_replies: dict[str, asyncio.Future[str]] = {} - # Wire up the ask_user callback so GooseClient can ask permission questions + # Track which users have a prompt in progress + self._active_prompts: set[str] = set() + + # Wire up callbacks so GooseClient can communicate with Signal users self.goose.client.set_ask_user(self._ask_user) + self.goose.client.set_notify_user(self._notify_user) + + async def _send(self, source: str, text: str) -> None: + """Send a message to a Signal user, splitting if needed.""" + for chunk in _split_message(text): + await self.signal.send_message(source, chunk) + + async def _notify_user(self, session_id: str, message: str) -> None: + """Send a status notification to the Signal user for a session.""" + source = self.goose.session_key_for_id(session_id) + if source: + await self._send(source, message) async def _ask_user( self, session_id: str, question: str, option_labels: list[str] ) -> str: - """Send a question to the Signal user and wait for their reply. - - Called by GooseClient.request_permission during a prompt() call. - """ + """Send a question to the Signal user and wait for their reply.""" source = self.goose.session_key_for_id(session_id) if source is None: raise RuntimeError(f"No contact found for session {session_id}") @@ -65,10 +78,7 @@ class Bridge: contact_name = contact.name if contact else source log.info("Asking %s for permission [%s]", contact_name, session_id[:8]) - - # Send the question - for chunk in _split_message(question): - await self.signal.send_message(source, chunk) + await self._send(source, question) # Create a future and wait for the user's reply future: asyncio.Future[str] = asyncio.get_event_loop().create_future() @@ -80,9 +90,7 @@ class Bridge: return reply except asyncio.TimeoutError: log.warning("Permission request timed out for %s [%s]", contact_name, session_id[:8]) - await self.signal.send_message( - source, "⏰ Permission request timed out — denying by default." - ) + await self._send(source, "⏰ Permission request timed out — denying by default.") raise finally: self._pending_replies.pop(source, None) @@ -93,7 +101,8 @@ class Bridge: async for incoming in self.signal.receive_messages(): source = incoming["source"] - message = incoming["message"] + message = incoming.get("message", "") + attachments = incoming.get("attachments", []) # Check trust if not self.config.is_trusted(source): @@ -114,37 +123,121 @@ class Bridge: self._pending_replies[source].set_result(message) continue + # Check for slash commands + if message.startswith("/"): + await self._handle_command(source, contact_name, message) + continue + + # Reject if a prompt is already in progress + if source in self._active_prompts: + log.info("Rejecting message from %s — prompt in progress", contact_name) + await self._send(source, "⏳ Please wait — I'm still working on your previous message.") + continue + log.info("Message from %s (%s): %s", contact_name, source, message[:80]) - # Process the message as a new prompt in the background so we can - # continue reading Signal messages (needed for permission replies) + # Process the prompt in the background asyncio.create_task( - self._handle_prompt(source, contact_name, message), + self._handle_prompt(source, contact_name, message, attachments), name=f"prompt-{source}", ) - async def _handle_prompt(self, source: str, contact_name: str, message: str) -> None: + async def _handle_command(self, source: str, contact_name: str, message: str) -> None: + """Handle a slash command from a Signal user.""" + parts = message.strip().split(maxsplit=1) + command = parts[0].lower() + args = parts[1] if len(parts) > 1 else "" + + log.info("Command from %s: %s", contact_name, command) + + if command == "/cancel": + if source in self._active_prompts: + cancelled = await self.goose.cancel(source) + if cancelled: + await self._send(source, "🛑 Cancelling current request…") + else: + await self._send(source, "No active request to cancel.") + else: + await self._send(source, "No active request to cancel.") + + elif command == "/new": + # Close existing session and start fresh + await self.goose.close_session(source) + await self._send(source, "🆕 Starting a new session.") + + elif command == "/help": + help_text = ( + "Available commands:\n" + "/cancel — Cancel the current request\n" + "/new — Start a new session\n" + "/help — Show this help message" + ) + await self._send(source, help_text) + + else: + await self._send(source, f"Unknown command: {command}\nType /help for available commands.") + + async def _handle_prompt( + self, + source: str, + contact_name: str, + message: str, + attachments: list[dict], + ) -> None: """Handle a single prompt from a Signal user.""" + self._active_prompts.add(source) try: - # Get or create a Goose session for this contact session_id = await self.goose.get_or_create_session(source) - # Send the message to Goose - response_text = await self.goose.prompt(session_id, message) + # Check for image attachments + image_data, mime_type = await self._extract_image(attachments) + + if image_data: + response_text = await self.goose.prompt_with_image( + session_id, message or None, image_data, mime_type + ) + else: + if not message: + return # No text and no image, nothing to do + response_text = await self.goose.prompt(session_id, message) if not response_text.strip(): response_text = "(Goose returned an empty response)" - # Send the reply back via Signal - for chunk in _split_message(response_text): - await self.signal.send_message(source, chunk) - log.info("Sent reply to %s (%d chars)", contact_name, len(chunk)) + await self._send(source, response_text) + log.info("Sent reply to %s (%d chars)", contact_name, len(response_text)) except Exception: log.exception("Error processing message from %s", contact_name) try: - await self.signal.send_message( + await self._send( source, "⚠️ Sorry, something went wrong processing your message." ) except Exception: log.exception("Failed to send error message to %s", contact_name) + finally: + self._active_prompts.discard(source) + + async def _extract_image(self, attachments: list[dict]) -> tuple[str | None, str | None]: + """Extract the first image attachment as base64 data. + + Returns (base64_data, mime_type) or (None, None). + """ + for att in attachments: + content_type = att.get("contentType", "") + if not content_type.startswith("image/"): + continue + + # signal-cli stores attachments locally; the path is in the "file" field + file_path = att.get("file") + if file_path and Path(file_path).is_file(): + log.info("Reading image attachment: %s (%s)", file_path, content_type) + data = Path(file_path).read_bytes() + return base64.b64encode(data).decode("ascii"), content_type + + # Some versions use "id" - try fetching via signal-cli's attachment path + att_id = att.get("id") + if att_id: + log.info("Attachment has id %s but no file path; skipping", att_id) + + return None, None diff --git a/src/honker/config.py b/src/honker/config.py index 4767b5a..dee0459 100644 --- a/src/honker/config.py +++ b/src/honker/config.py @@ -17,6 +17,8 @@ class Contact: class Config: account: str trusted_contacts: list[Contact] = field(default_factory=list) + working_directory: str = "." + forward_thoughts: bool = False def is_trusted(self, number: str) -> bool: """Check whether a phone number belongs to a trusted contact.""" @@ -52,4 +54,12 @@ def load_config(path: Path | None = None) -> Config: for c in raw.get("trusted_contacts", []) ] - return Config(account=raw["account"], trusted_contacts=contacts) + working_directory = raw.get("working_directory", ".") + forward_thoughts = raw.get("forward_thoughts", False) + + return Config( + account=raw["account"], + trusted_contacts=contacts, + working_directory=working_directory, + forward_thoughts=forward_thoughts, + ) diff --git a/src/honker/goose.py b/src/honker/goose.py index 0a69b9a..65eddf6 100644 --- a/src/honker/goose.py +++ b/src/honker/goose.py @@ -15,6 +15,7 @@ from acp import ( ) from acp.schema import ( AgentMessageChunk, + AgentThoughtChunk, AllowedOutcome, ClientCapabilities, DeniedOutcome, @@ -32,6 +33,9 @@ log = logging.getLogger(__name__) # Type for the callback the bridge provides: (session_id, question, options) -> answer AskUserFunc = Callable[[str, str, list[str]], Awaitable[str]] +# Type for sending a notification to the user: (session_id, message) -> None +NotifyUserFunc = Callable[[str, str], Awaitable[None]] + def _format_tool_call(tool_call: ToolCallUpdate) -> str: """Format a tool call into a human-readable description.""" @@ -67,32 +71,59 @@ def _format_permission_message( class GooseClient(Client): - """ACP Client implementation that collects agent responses.""" + """ACP Client implementation that handles agent requests and notifications. - def __init__(self) -> None: + Goose runs locally and uses its own built-in tools for file I/O and shell + execution, so we don't advertise client-side fs or terminal capabilities. + The client-side methods exist only as no-op stubs required by the protocol. + """ + + def __init__(self, forward_thoughts: bool = False) -> None: self._response_chunks: dict[str, list[str]] = {} # session_id -> chunks + self._forward_thoughts = forward_thoughts self._ask_user: AskUserFunc | None = None + self._notify_user: NotifyUserFunc | None = None def set_ask_user(self, func: AskUserFunc) -> None: """Set the callback used to ask the Signal user a question.""" self._ask_user = func + def set_notify_user(self, func: NotifyUserFunc) -> None: + """Set the callback used to send status notifications to the Signal user.""" + self._notify_user = func + def take_response(self, session_id: str) -> str: """Take the accumulated response text for a session, clearing the buffer.""" chunks = self._response_chunks.pop(session_id, []) return "".join(chunks) + # -- Session update notifications -- + async def session_update(self, session_id: str, update, **kwargs: Any) -> None: if isinstance(update, AgentMessageChunk): content = update.content if isinstance(content, TextContentBlock): self._response_chunks.setdefault(session_id, []).append(content.text) log.debug("Agent chunk [%s]: %s", session_id[:8], content.text[:80]) + + elif isinstance(update, AgentThoughtChunk): + content = update.content + if isinstance(content, TextContentBlock): + log.debug("Agent thought [%s]: %s", session_id[:8], content.text[:80]) + if self._forward_thoughts and self._notify_user: + await self._notify_user(session_id, f"💭 {content.text}") + elif isinstance(update, ToolCallStart): - log.info("Tool call started [%s]: %s", session_id[:8], update.title) + title = update.title or "(unknown)" + log.info("Tool call started [%s]: %s", session_id[:8], title) + if self._notify_user: + await self._notify_user(session_id, f"🔧 {title}") + elif isinstance(update, ToolCallProgress): log.debug("Tool call progress [%s]", session_id[:8]) + # -- Permission requests -- + async def request_permission( self, options: list[PermissionOption], @@ -132,16 +163,18 @@ class GooseClient(Client): outcome=DeniedOutcome(outcome="cancelled") ) - async def write_text_file(self, content, path, session_id, **kwargs): - log.info("Agent wants to write file [%s]: %s", session_id[:8], path) - return None + # -- Client-side stubs (Goose uses its own built-in tools instead) -- async def read_text_file(self, path, session_id, **kwargs): - log.info("Agent wants to read file [%s]: %s", session_id[:8], path) + log.debug("read_text_file called (unused) [%s]: %s", session_id[:8], path) + return None + + async def write_text_file(self, content, path, session_id, **kwargs): + log.debug("write_text_file called (unused) [%s]: %s", session_id[:8], path) return None async def create_terminal(self, command, session_id, **kwargs): - log.info("Agent wants to create terminal [%s]: %s", session_id[:8], command) + log.debug("create_terminal called (unused) [%s]: %s", session_id[:8], command) return None async def terminal_output(self, session_id, terminal_id, **kwargs): @@ -157,10 +190,11 @@ class GooseClient(Client): return None async def ext_method(self, method, params): + log.debug("ext_method: %s", method) return {} async def ext_notification(self, method, params): - pass + log.debug("ext_notification: %s", method) def on_connect(self, conn: Agent) -> None: pass @@ -197,9 +231,9 @@ def _parse_permission_reply( class GooseConnection: """Manages the ACP connection to a Goose agent process.""" - def __init__(self, cwd: str = "."): + def __init__(self, cwd: str = ".", forward_thoughts: bool = False): self.cwd = cwd - self.client = GooseClient() + self.client = GooseClient(forward_thoughts=forward_thoughts) self._conn = None self._process = None self._ctx = None @@ -217,7 +251,8 @@ class GooseConnection: self._ctx = spawn_agent_process(self.client, "goose", "acp", env=os.environ.copy()) self._conn, self._process = await self._ctx.__aenter__() - # Initialize the ACP protocol + # Initialize — don't advertise fs or terminal capabilities since + # Goose runs locally and uses its own built-in tools resp = await self._conn.initialize( protocol_version=PROTOCOL_VERSION, client_info=Implementation(name="honker", version="0.1.0"), @@ -247,9 +282,32 @@ class GooseConnection: log.info("Created new Goose session for %s: %s", key, session_id[:8]) return session_id + async def close_session(self, key: str) -> bool: + """Close and remove the session for the given key. Returns True if found.""" + session_id = self._sessions.pop(key, None) + if session_id is None: + return False + try: + await self._conn.close_session(session_id=session_id) + except Exception: + log.exception("Error closing session %s", session_id[:8]) + log.info("Closed session for %s: %s", key, session_id[:8]) + return True + + async def cancel(self, key: str) -> bool: + """Cancel the current prompt for the given key. Returns True if found.""" + session_id = self._sessions.get(key) + if session_id is None: + return False + try: + await self._conn.cancel(session_id=session_id) + log.info("Cancelled prompt for %s [%s]", key, session_id[:8]) + except Exception: + log.exception("Error cancelling session %s", session_id[:8]) + return True + async def prompt(self, session_id: str, text: str) -> str: """Send a prompt and wait for the complete response.""" - # Clear any prior state self.client._response_chunks.pop(session_id, None) prompt_content = [TextContentBlock(type="text", text=text)] @@ -257,4 +315,33 @@ class GooseConnection: log.info("Prompt complete [%s]: stop_reason=%s", session_id[:8], response.stop_reason) + text_response = self.client.take_response(session_id) + + # Append stop reason context if not a normal end + stop_reason = response.stop_reason + if stop_reason == "max_tokens": + text_response += "\n\n⚠️ Response was cut short — context limit reached." + elif stop_reason == "max_turn_requests": + text_response += "\n\n⚠️ Response was cut short — turn limit reached." + elif stop_reason == "refusal": + text_response += "\n\n🚫 The agent refused this request." + elif stop_reason == "cancelled": + text_response += "\n\n🛑 Request was cancelled." + + return text_response + + async def prompt_with_image(self, session_id: str, text: str | None, image_data: str, mime_type: str) -> str: + """Send a prompt with an image and wait for the complete response.""" + from acp.schema import ImageContentBlock + + self.client._response_chunks.pop(session_id, None) + + prompt_content = [] + if text: + prompt_content.append(TextContentBlock(type="text", text=text)) + prompt_content.append(ImageContentBlock(type="image", data=image_data, mime_type=mime_type)) + + response = await self._conn.prompt(prompt=prompt_content, session_id=session_id) + log.info("Prompt (with image) complete [%s]: stop_reason=%s", session_id[:8], response.stop_reason) + return self.client.take_response(session_id) diff --git a/src/honker/main.py b/src/honker/main.py index dc18785..a5e7757 100644 --- a/src/honker/main.py +++ b/src/honker/main.py @@ -51,9 +51,12 @@ async def run(args: argparse.Namespace) -> None: await signal_client.connect() # Start Goose via ACP - goose = GooseConnection() + goose = GooseConnection( + cwd=config.working_directory, + forward_thoughts=config.forward_thoughts, + ) await goose.start() - log.info("Goose ACP connection established") + log.info("Goose ACP connection established (cwd=%s)", config.working_directory) # Run the bridge; cancel on SIGINT/SIGTERM bridge = Bridge(config, signal_client, goose) diff --git a/src/honker/signal.py b/src/honker/signal.py index 7d26e07..f036ecf 100644 --- a/src/honker/signal.py +++ b/src/honker/signal.py @@ -82,12 +82,17 @@ class SignalClient: source = envelope.get("source") data_message = envelope.get("dataMessage") - if source and data_message and data_message.get("message"): - await self._incoming_queue.put({ - "source": source, - "message": data_message["message"], - "timestamp": data_message.get("timestamp"), - }) + if source and data_message: + message_text = data_message.get("message", "") + attachments = data_message.get("attachments", []) + # Only enqueue if there's text or attachments + if message_text or attachments: + await self._incoming_queue.put({ + "source": source, + "message": message_text, + "timestamp": data_message.get("timestamp"), + "attachments": attachments, + }) except asyncio.CancelledError: pass except Exception: