Permission prompting

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)

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)

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