diff --git a/src/honker/bridge.py b/src/honker/bridge.py new file mode 100644 index 0000000..c1128f7 --- /dev/null +++ b/src/honker/bridge.py @@ -0,0 +1,150 @@ +"""Bridge — ties Signal and Goose together.""" + +import asyncio +import logging + +from honker.config import Config +from honker.goose import GooseConnection +from honker.signal import SignalClient + +log = logging.getLogger(__name__) + +# Signal has a message size limit; we split long responses +MAX_SIGNAL_MESSAGE_LENGTH = 4096 + +# How long to wait for a permission reply before timing out +PERMISSION_TIMEOUT_SECONDS = 120 + + +def _split_message(text: str, max_len: int = MAX_SIGNAL_MESSAGE_LENGTH) -> list[str]: + """Split a long message into chunks, breaking at newlines where possible.""" + if len(text) <= max_len: + return [text] + + chunks = [] + while text: + if len(text) <= max_len: + chunks.append(text) + break + # 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") + return chunks + + +class Bridge: + """Orchestrates message flow between Signal and Goose.""" + + def __init__(self, config: Config, signal: SignalClient, goose: GooseConnection): + self.config = config + self.signal = signal + self.goose = goose + + # 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 + self.goose.client.set_ask_user(self._ask_user) + + 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. + """ + source = self.goose.session_key_for_id(session_id) + if source is None: + raise RuntimeError(f"No contact found for session {session_id}") + + contact = self.config.contact_by_number(source) + 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) + + # Create a future and wait for the user's reply + future: asyncio.Future[str] = asyncio.get_event_loop().create_future() + self._pending_replies[source] = future + + try: + reply = await asyncio.wait_for(future, timeout=PERMISSION_TIMEOUT_SECONDS) + log.info("Got permission reply from %s: %s", contact_name, reply[:80]) + 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." + ) + raise + finally: + self._pending_replies.pop(source, None) + + async def run(self) -> None: + """Main loop: read Signal messages, forward to Goose, send replies.""" + log.info("Bridge running — waiting for messages from trusted contacts") + + async for incoming in self.signal.receive_messages(): + source = incoming["source"] + message = incoming["message"] + + # Check trust + if not self.config.is_trusted(source): + log.warning("Ignoring message from untrusted number: %s", source) + continue + + contact = self.config.contact_by_number(source) + contact_name = contact.name if contact else source + + # If there's a pending permission question for this user, + # route the reply there instead of creating a new prompt + if source in self._pending_replies: + log.info( + "Routing reply from %s to pending permission request: %s", + contact_name, + message[:80], + ) + self._pending_replies[source].set_result(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) + asyncio.create_task( + self._handle_prompt(source, contact_name, message), + name=f"prompt-{source}", + ) + + async def _handle_prompt(self, source: str, contact_name: str, message: str) -> None: + """Handle a single prompt from a Signal user.""" + 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) + + 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)) + + except Exception: + log.exception("Error processing message from %s", contact_name) + try: + await self.signal.send_message( + source, "⚠️ Sorry, something went wrong processing your message." + ) + except Exception: + log.exception("Failed to send error message to %s", contact_name) diff --git a/src/honker/goose.py b/src/honker/goose.py new file mode 100644 index 0000000..68b3186 --- /dev/null +++ b/src/honker/goose.py @@ -0,0 +1,255 @@ +"""Goose ACP client — manages the connection and sessions.""" + +import asyncio +import logging +import os +from typing import Any, Awaitable, Callable + +from acp import ( + Agent, + Client, + spawn_agent_process, + PROTOCOL_VERSION, + RequestPermissionRequest, + RequestPermissionResponse, +) +from acp.schema import ( + AgentMessageChunk, + AllowedOutcome, + ClientCapabilities, + DeniedOutcome, + Implementation, + PermissionOption, + PermissionOptionKind, + TextContentBlock, + ToolCallStart, + ToolCallProgress, + ToolCallUpdate, +) + +log = logging.getLogger(__name__) + +# Type for the callback the bridge provides: (session_id, question, options) -> answer +AskUserFunc = Callable[[str, str, list[str]], Awaitable[str]] + + +def _format_tool_call(tool_call: ToolCallUpdate) -> str: + """Format a tool call into a human-readable description.""" + parts = [] + if tool_call.title: + parts.append(tool_call.title) + if tool_call.raw_input: + raw = str(tool_call.raw_input) + if len(raw) > 200: + raw = raw[:200] + "…" + parts.append(f"Input: {raw}") + return "\n".join(parts) if parts else "(unknown tool call)" + + +def _format_permission_message( + tool_call: ToolCallUpdate, options: list[PermissionOption] +) -> tuple[str, list[str]]: + """Build the question text and option labels for a permission request.""" + question_parts = [ + "🔐 *Permission requested*", + "", + _format_tool_call(tool_call), + "", + ] + option_labels = [] + for i, opt in enumerate(options, 1): + label = f"{i}. {opt.name}" + option_labels.append(label) + question_parts.append("\n".join(option_labels)) + question_parts.append("") + question_parts.append("Reply with a number to choose:") + return "\n".join(question_parts), [opt.name for opt in options] + + +class GooseClient(Client): + """ACP Client implementation that collects agent responses.""" + + def __init__(self) -> None: + self._response_chunks: dict[str, list[str]] = {} # session_id -> chunks + self._ask_user: AskUserFunc | 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 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) + + 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, ToolCallStart): + log.info("Tool call started [%s]: %s", session_id[:8], update.title) + elif isinstance(update, ToolCallProgress): + log.debug("Tool call progress [%s]", session_id[:8]) + + async def request_permission( + self, + options: list[PermissionOption], + session_id: str, + tool_call: ToolCallUpdate, + **kwargs: Any, + ) -> RequestPermissionResponse: + """Ask the Signal user for permission via the bridge callback.""" + question, option_labels = _format_permission_message(tool_call, options) + + if self._ask_user is not None: + try: + reply = await self._ask_user(session_id, question, option_labels) + chosen = _parse_permission_reply(reply, options) + if chosen is not None: + log.info( + "User chose [%s]: %s (option: %s)", + session_id[:8], + chosen.kind, + chosen.name, + ) + return RequestPermissionResponse( + outcome=AllowedOutcome( + outcome="selected", option_id=chosen.option_id + ) + ) + else: + log.warning( + "Could not parse permission reply [%s]: %r", session_id[:8], reply + ) + except Exception: + log.exception("Error asking user for permission [%s]", session_id[:8]) + + # Fallback: deny if we couldn't get a valid answer + log.info("Denying permission request [%s] (no valid user response)", session_id[:8]) + return RequestPermissionResponse( + 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 + + async def read_text_file(self, path, session_id, **kwargs): + log.info("Agent wants to read file [%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) + return None + + async def terminal_output(self, session_id, terminal_id, **kwargs): + return None + + async def release_terminal(self, session_id, terminal_id, **kwargs): + return None + + async def wait_for_terminal_exit(self, session_id, terminal_id, **kwargs): + return None + + async def kill_terminal(self, session_id, terminal_id, **kwargs): + return None + + async def ext_method(self, method, params): + return {} + + async def ext_notification(self, method, params): + pass + + def on_connect(self, conn: Agent) -> None: + pass + + +def _parse_permission_reply( + reply: str, options: list[PermissionOption] +) -> PermissionOption | None: + """Parse a user's reply to a permission prompt. + + Accepts: + - A number (1-based index into the options list) + - The option name (case-insensitive, partial match) + """ + reply = reply.strip() + + # Try as a number + try: + idx = int(reply) - 1 + if 0 <= idx < len(options): + return options[idx] + except ValueError: + pass + + # Try as a name match + reply_lower = reply.lower() + for opt in options: + if reply_lower in opt.name.lower(): + return opt + + return None + + +class GooseConnection: + """Manages the ACP connection to a Goose agent process.""" + + def __init__(self, cwd: str = "."): + self.cwd = cwd + self.client = GooseClient() + self._conn = None + self._process = None + self._ctx = None + self._sessions: dict[str, str] = {} # key -> session_id + + def session_key_for_id(self, session_id: str) -> str | None: + """Look up the key (phone number) for a session ID.""" + for key, sid in self._sessions.items(): + if sid == session_id: + return key + return None + + async def start(self) -> None: + """Spawn goose and initialize the ACP connection.""" + 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 + resp = await self._conn.initialize( + protocol_version=PROTOCOL_VERSION, + client_info=Implementation(name="honker", version="0.1.0"), + client_capabilities=ClientCapabilities(), + ) + log.info("ACP initialized: %s", resp) + + async def stop(self) -> None: + """Shut down the ACP connection and goose process.""" + if self._ctx is not None: + await self._ctx.__aexit__(None, None, None) + self._ctx = None + + async def get_or_create_session(self, key: str) -> str: + """Get an existing session or create a new one for the given key.""" + if key in self._sessions: + return self._sessions[key] + + resp = await self._conn.new_session(cwd=self.cwd) + session_id = resp.session_id + self._sessions[key] = session_id + log.info("Created new Goose session for %s: %s", key, session_id[:8]) + return session_id + + 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)] + response = await self._conn.prompt(prompt=prompt_content, session_id=session_id) + + log.info("Prompt complete [%s]: stop_reason=%s", session_id[:8], response.stop_reason) + + return self.client.take_response(session_id) diff --git a/src/honker/signal.py b/src/honker/signal.py new file mode 100644 index 0000000..7d26e07 --- /dev/null +++ b/src/honker/signal.py @@ -0,0 +1,158 @@ +"""Signal-cli JSON-RPC client over UNIX socket.""" + +import asyncio +import json +import logging +from collections.abc import AsyncIterator +from pathlib import Path + +log = logging.getLogger(__name__) + + +class SignalClient: + """Communicates with signal-cli daemon over its UNIX socket JSON-RPC interface. + + Uses a single socket connection with a centralized reader task that + dispatches JSON-RPC responses to waiting request futures and pushes + notifications into a queue for receive_messages() to consume. + """ + + def __init__(self, socket_path: Path, account: str): + self.socket_path = socket_path + self.account = account + self._reader: asyncio.StreamReader | None = None + self._writer: asyncio.StreamWriter | None = None + self._request_id = 0 + self._pending_requests: dict[int, asyncio.Future[dict]] = {} + self._incoming_queue: asyncio.Queue[dict] = asyncio.Queue() + self._reader_task: asyncio.Task | None = None + + async def connect(self) -> None: + """Connect to the signal-cli UNIX socket, retrying until available.""" + for attempt in range(30): + if self.socket_path.exists(): + try: + self._reader, self._writer = await asyncio.open_unix_connection( + str(self.socket_path) + ) + log.info("Connected to signal-cli socket at %s", self.socket_path) + self._reader_task = asyncio.create_task( + self._read_loop(), name="signal-reader" + ) + return + except (ConnectionRefusedError, OSError) as e: + log.debug("Connection attempt %d failed: %s", attempt + 1, e) + else: + log.debug("Socket not yet available (attempt %d)", attempt + 1) + await asyncio.sleep(1) + raise ConnectionError( + f"Could not connect to signal-cli socket at {self.socket_path} after 30 attempts" + ) + + async def _read_loop(self) -> None: + """Centralized reader: dispatch responses and queue notifications.""" + assert self._reader is not None + try: + while True: + raw = await self._reader.readline() + if not raw: + log.warning("signal-cli socket closed") + break + + log.debug("signal-cli <<< %s", raw.decode().strip()) + + try: + msg = json.loads(raw) + except json.JSONDecodeError: + log.warning("Unparseable line from signal-cli: %s", raw) + continue + + # JSON-RPC response — resolve the waiting future + if "id" in msg and msg["id"] in self._pending_requests: + future = self._pending_requests.pop(msg["id"]) + if not future.done(): + future.set_result(msg) + continue + + # JSON-RPC notification — parse and enqueue + params = msg.get("params", {}) + envelope = params.get("envelope", {}) + if not envelope: + continue + + 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"), + }) + except asyncio.CancelledError: + pass + except Exception: + log.exception("signal-cli reader loop crashed") + finally: + # Fail any pending requests + for future in self._pending_requests.values(): + if not future.done(): + future.set_exception( + ConnectionError("signal-cli reader loop stopped") + ) + self._pending_requests.clear() + + async def close(self) -> None: + if self._reader_task is not None: + self._reader_task.cancel() + try: + await self._reader_task + except asyncio.CancelledError: + pass + self._reader_task = None + if self._writer is not None: + self._writer.close() + await self._writer.wait_closed() + self._writer = None + self._reader = None + + def _next_id(self) -> int: + self._request_id += 1 + return self._request_id + + async def _send_request(self, method: str, params: dict) -> dict: + """Send a JSON-RPC request and return the result.""" + assert self._writer is not None + req_id = self._next_id() + request = { + "jsonrpc": "2.0", + "method": method, + "id": req_id, + "params": params, + } + line = json.dumps(request) + "\n" + log.debug("signal-cli >>> %s", line.strip()) + + future: asyncio.Future[dict] = asyncio.get_event_loop().create_future() + self._pending_requests[req_id] = future + + self._writer.write(line.encode()) + await self._writer.drain() + + msg = await future + if "error" in msg: + raise RuntimeError(f"signal-cli error: {msg['error']}") + return msg.get("result", {}) + + async def send_message(self, recipient: str, message: str) -> dict: + """Send a text message to a recipient.""" + return await self._send_request("send", { + "account": self.account, + "recipient": [recipient], + "message": message, + }) + + async def receive_messages(self) -> AsyncIterator[dict]: + """Yield incoming messages from the queue.""" + while True: + msg = await self._incoming_queue.get() + yield msg