diff --git a/.env.example b/.env.example index 5c8577c..ce188fd 100644 --- a/.env.example +++ b/.env.example @@ -1,5 +1,6 @@ # Signal Configuration SIGNAL__SERVICE_URL=http://signal-api:8080 +# The bot's own Signal number (E.164). Required for the Pacto→Signal !signal relay. SIGNAL__PHONE_NUMBER=+1234567890 SIGNAL__POLL_INTERVAL=1s @@ -38,6 +39,27 @@ TRANSLATE_ALL__MAX_MESSAGES_PER_MINUTE=30 GROUP_PREFERENCES__PERSIST=true GROUP_PREFERENCES__STORAGE_PATH=/data/group_prefs.enc +# Pacto Messaging (via pacto-bot-api daemon) +# Set to true to enable Pacto integration (requires a running pacto-bot-api +# daemon; see docker/pacto-bot-api.toml.example) +PACTO__ENABLED=false +# Inbound DM agent: when true (and PACTO__ENABLED), Pacto users get the same DM +# experience as Signal users (AI chat + !verify/!clear/!models/!help/!privacy). +# Set false for outbound-only (!pact) deployments. +PACTO__AGENT_ENABLED=true +PACTO__SOCKET_PATH=/var/run/pacto/pacto-bot-api.sock +PACTO__BOT_ID=sigstack +# Optional: recipient used by `!pact ` when no npub is given +# PACTO__DEFAULT_RECIPIENT=npub1... +PACTO__TIMEOUT=15s +# Let Pacto users DM Signal users via `!signal <+number> `. Off by +# default; can relay to arbitrary numbers, so it is allowlist-gated. +# Requires SIGNAL__PHONE_NUMBER (the "from" number). +PACTO__SIGNAL_RELAY_ENABLED=false +# Comma-separated E.164 numbers reachable via !signal. Empty = deny all; +# "*" = allow any (open relay — abuse risk, use deliberately). +# PACTO__SIGNAL_RELAY_ALLOWLIST=+14155551234,+14155555678 + # Payment Configuration (x402) # Set to true to enable credit tracking and payment system PAYMENTS__ENABLED=false diff --git a/.gitignore b/.gitignore index 9da6c0e..863d65f 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,9 @@ docker/.env **/secrets/ +# Per-deployment Pacto daemon config (copy from pacto-bot-api.toml.example) +docker/pacto-bot-api.toml + # IDE .idea/ .vscode/ diff --git a/CLAUDE.md b/CLAUDE.md index 8cce7cb..eb022ed 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -200,6 +200,7 @@ crates/ signal-client/ # Signal CLI REST API client signal-registration-proxy/ # Multi-tenant registration service tools/ # Tool use system (calculator, weather, web search) + pacto-client/ # JSON-RPC client for the pacto-bot-api daemon (!pact) web/ # React frontend (Vite + Tailwind, deployed to Vercel) docker/ # Docker Compose configs for local and Phala deployment ``` @@ -492,6 +493,141 @@ crates/tools/ The tool system uses OpenAI-compatible function calling schema, which NEAR AI supports. +## Pacto Messaging (!pact + two-way DM agent) + +The bot bridges [Pacto](https://github.com/covenant-gov/pacto-app) in **both +directions** via a co-located [pacto-bot-api](https://github.com/covenant-gov/pacto-bot-api) +daemon. Disabled by default. + +- **Outbound** (`!pact`): a Signal user sends an encrypted DM into Pacto. +- **Inbound** (DM agent): a Pacto user DMs the bot and gets the *same* + experience a Signal DM user gets — AI chat (with tools), `!verify`, `!clear`, + `!models`, `!help`, `!privacy`, `!list-langs`, and AI-driven translation. +- **Cross-network relay** (`!signal`, optional): a Pacto user DMs a Signal user + through the bot. Off by default and allowlist-gated (see below). + +### Feature parity ceiling + +Pacto parity covers the full **DM** experience. It does **not** cover Signal's +group or voice features, because the daemon only delivers `dm_received` events — +it exposes neither inbound MLS **group messages** nor **audio attachments** to +handlers (`parse_event_type` accepts only `dm_received` / `mls_welcome_received`). +So group translation and voice transcription for Pacto are blocked upstream in +`pacto-bot-api` (its own docs list MLS group participation as a future phase), +not in this integration. When the daemon gains inbound group/attachment +delivery, the inbound agent can be extended to feed those into the existing +translate/transcribe pipelines. + +### Architecture + +``` +[TEE Boundary] ++----------------------------------------------------------------------+ +| Signal user --> signal-bot --!pact--> PactoClient (single-flight) -\ | +| | | +| Pacto user <--reply-- pacto_agent <== PactoAgent (full-duplex) <===+ | +| ^ | | | +| | same handlers as [Unix socket, | +| | Signal (chat/verify/...) shared volume] | +| | v | +| +------------------------------------ pacto-bot-api daemon | +| (Nostr keys, NIP-17/44/59) | ++----------------------------------------------------------------------+ + | + v + Nostr relays <--> Pacto users +``` + +The `pacto-client` crate speaks JSON-RPC 2.0 (newline-delimited frames) over the +daemon's Unix socket and provides two connection models: + +- **`PactoClient`** — single-flight request/response for outbound sends + (`!pact`) and progress pings. Registers as an outbound-only handler + (`SendMessages`, no event subscriptions) and publishes DMs with + `agent.send_dm`. Does not retry non-idempotent sends after a timeout. +- **`PactoAgent`** — a long-lived full-duplex connection that registers for + `dm_received` (`ReadMessages` + `SendMessages`), streams inbound DMs, and + replies via `handler.response{action:"reply"}` (the daemon auto-addresses the + reply to the sender). Auto-reconnects and re-registers if the daemon restarts. + +Inbound DMs are turned into a synthetic `signal_client::BotMessage` (author npub +as `source`, `is_group=false`) and dispatched through the *same* command/AI-chat +handlers the Signal loop uses, so behavior stays in lockstep. The only transport +seam is `ProgressSink` (the "🔧 Using ..." pings), implemented for both Signal +(reply) and Pacto (`send_dm`). Signal-quote `!translate`, voice, and +group-translate handlers are Signal-specific and are not reused; Pacto users +translate by asking the AI. + +Running the daemon inside the same TEE keeps the bot's Nostr signing key and +plaintext Pacto messages in protected memory, consistent with the rest of the +architecture. + +### Commands + +Outbound (Signal user → Pacto): + +| Command | Description | +|---------|-------------| +| `!pact ` | DM a Pacto user (npub or 64-char hex pubkey) | +| `!pact ` | DM the configured default recipient | +| `!pact status` | Check daemon reachability and version | +| `!pact` / `!pact help` | Usage | + +Inbound (Pacto user → bot): free-text → AI chat; `!verify`, `!clear`, `!models`, +`!help`, `!privacy`, `!list-langs`. + +Cross-network (Pacto user → Signal user), when enabled: + +| Command | Description | +|---------|-------------| +| `!signal <+number> ` | DM a Signal user (allowlist-gated) | + +The relayed Signal message is prefixed with the sender's Pacto npub, so the +Signal recipient knows who it's from and can reply with `!pact ` +— closing the loop without any stateful session mapping. + +**Abuse surface (why it's gated):** unlike the other directions, `!signal` +lets an anonymous Pacto npub reach arbitrary Signal numbers, i.e. an open relay +if left unguarded. It is therefore off by default, requires the bot's own +`SIGNAL__PHONE_NUMBER`, and an explicit allowlist of reachable recipients +(empty allowlist = deny all; `*` = allow any, opt-in open relay). A per-sender +rate limit is a sensible future addition. + +### Configuration + +| Variable | Default | Description | +|----------|---------|-------------| +| `PACTO__ENABLED` | `false` | Master switch for Pacto integration | +| `PACTO__AGENT_ENABLED` | `true` | Run the inbound DM agent (parity for Pacto users); false = outbound `!pact` only | +| `PACTO__SOCKET_PATH` | `/var/run/pacto/pacto-bot-api.sock` | Daemon Unix socket path | +| `PACTO__BOT_ID` | `sigstack` | Bot id from the daemon's `pacto-bot-api.toml` | +| `PACTO__DEFAULT_RECIPIENT` | (none) | npub used when `!pact` gets no recipient | +| `PACTO__SIGNAL_RELAY_ENABLED` | `false` | Enable `!signal` (Pacto user → Signal user) | +| `PACTO__SIGNAL_RELAY_ALLOWLIST` | (empty) | Comma-separated E.164 numbers reachable via `!signal`; empty = deny all, `*` = any | +| `SIGNAL__PHONE_NUMBER` | (none) | Bot's own Signal number; required for `!signal` | +| `PACTO__TIMEOUT` | `15s` | Max time per daemon request | + +### Local Setup + +1. Generate a bot identity: `pacto-bot-admin new sigstack --backend nsec` + (prints an npub, an nsec, and a config snippet) +2. `cp docker/pacto-bot-api.toml.example docker/pacto-bot-api.toml` and set + your bot's npub (the file is gitignored) +3. In `docker/.env` set `PACTO_BOT_NSEC=` and `PACTO_ENABLED=true` +4. Start the stack with the daemon: `docker compose --profile pacto up -d` + +The daemon's data dir (including its socket) lives in the `pacto-data` volume, +mounted read-write into signal-bot at `/var/run/pacto`. Both containers run as +uid 1000, matching the daemon's 0600 socket permissions. + +### Phala Deployment + +Phala CVMs cannot bind-mount local files, so bake the daemon config into a +derived image with `docker/Dockerfile.pacto` (builds the pinned upstream tag +from source and copies in your `pacto-bot-api.toml`). Push it as linux/amd64, +set `PACTO_BOT_API_IMAGE` plus the `PACTO_BOT_NSEC` encrypted secret, and +uncomment the `pacto-bot-api` service block in `docker/phala-compose.yaml`. + ## Testing ```bash diff --git a/Cargo.lock b/Cargo.lock index 180cebd..86e8916 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4585,6 +4585,18 @@ dependencies = [ "hashbrown 0.14.5", ] +[[package]] +name = "pacto-client" +version = "0.1.0" +dependencies = [ + "serde", + "serde_json", + "tempfile", + "thiserror 1.0.69", + "tokio", + "tracing", +] + [[package]] name = "parity-scale-codec" version = "3.7.5" @@ -6073,6 +6085,7 @@ dependencies = [ "humantime-serde", "mockall", "near-ai-client", + "pacto-client", "rand 0.8.5", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 91f058b..97d3450 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,6 +10,7 @@ members = [ "crates/tools", "crates/x402-payments", "crates/whisper-client", + "crates/pacto-client", ] [workspace.package] diff --git a/README.md b/README.md index 90a399d..9493b42 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,15 @@ This project implements a Signal bot that runs inside a Dstack-powered TEE (Inte | `!clear` | Clear conversation history | | `!models` | List available AI models | | `!help` | Show help message | +| `!pact ` | Send an encrypted DM into Pacto (optional, requires pacto-bot-api daemon) | + +When Pacto is enabled, the bot also works **the other way**: Pacto users can DM +the bot directly and get the same DM experience Signal users get (AI chat with +tools, `!verify`, `!clear`, `!models`, `!help`, `!privacy`, `!list-langs`). +Pacto users can also DM Signal users through the bot with `!signal <+number> +` — an optional, off-by-default, allowlist-gated cross-network relay. +Group and voice features are Signal-only (the Pacto daemon does not yet deliver +inbound group messages or audio). See `CLAUDE.md` → "Pacto Messaging". Any other message is sent to the AI for a response. diff --git a/crates/pacto-client/Cargo.toml b/crates/pacto-client/Cargo.toml new file mode 100644 index 0000000..b6a7ee5 --- /dev/null +++ b/crates/pacto-client/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "pacto-client" +version.workspace = true +edition.workspace = true + +[dependencies] +serde.workspace = true +serde_json.workspace = true +thiserror.workspace = true +tokio.workspace = true +tracing.workspace = true + +[dev-dependencies] +tempfile = "3" diff --git a/crates/pacto-client/src/agent.rs b/crates/pacto-client/src/agent.rs new file mode 100644 index 0000000..56289e7 --- /dev/null +++ b/crates/pacto-client/src/agent.rs @@ -0,0 +1,239 @@ +//! Full-duplex inbound agent for the pacto-bot-api daemon. +//! +//! Where [`crate::PactoClient`] is a single-flight request/response client for +//! *outbound* sends, `PactoAgent` maintains a long-lived connection that +//! registers for `dm_received` events, streams inbound DMs to a channel, and +//! replies via `handler.response`. It reconnects automatically if the daemon +//! restarts. + +use crate::error::PactoError; +use serde_json::{Value, json}; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixStream; +use tokio::net::unix::{OwnedReadHalf, OwnedWriteHalf}; +use tokio::sync::{Mutex, mpsc}; +use tracing::{debug, info, warn}; + +/// A decrypted direct message delivered by the daemon (`agent.event`). +#[derive(Debug, Clone)] +pub struct InboundDm { + /// Bot identity the message was addressed to. + pub bot_id: String, + /// Hex id of the enclosing gift-wrap event — used to reply via + /// `handler.response`. + pub event_id: String, + /// Conversation identifier (usually the sender's npub). + pub chat_id: Option, + /// Public key of the message author (the reply recipient). + pub author: String, + /// Decrypted message text. + pub content: String, + /// Hex id of the decrypted rumor. + pub rumor_id: String, + /// Unix timestamp of the rumor. + pub timestamp: i64, +} + +/// Long-lived inbound agent. Clone-cheap handle used to send replies; the +/// connection is owned by a background supervisor task spawned in [`Self::spawn`]. +pub struct PactoAgent { + write: Arc>>, + bot_id: String, +} + +impl PactoAgent { + /// Connect (with auto-reconnect) and register for `dm_received`. Returns a + /// reply handle plus the receiver of inbound DMs. The background task ends + /// only when the returned receiver is dropped. + pub fn spawn( + socket_path: impl Into, + bot_id: impl Into, + reconnect_backoff: Duration, + ) -> (Arc, mpsc::Receiver) { + let socket_path = socket_path.into(); + let bot_id = bot_id.into(); + let write = Arc::new(Mutex::new(None)); + let agent = Arc::new(PactoAgent { + write: write.clone(), + bot_id: bot_id.clone(), + }); + let (tx, rx) = mpsc::channel(64); + tokio::spawn(supervise(socket_path, bot_id, write, tx, reconnect_backoff)); + (agent, rx) + } + + /// Bot identity this agent serves. + pub fn bot_id(&self) -> &str { + &self.bot_id + } + + /// Reply to a delivered event. The daemon addresses the DM to the original + /// sender and threads it to the source rumor. + pub async fn reply(&self, event_id: &str, content: &str) -> Result<(), PactoError> { + self.respond(json!({ + "event_id": event_id, + "action": "reply", + "content": content, + })) + .await + } + + /// Terminate an event without replying (`ack` / `ignore` / `defer`). + pub async fn finish(&self, event_id: &str, action: &str) -> Result<(), PactoError> { + self.respond(json!({ "event_id": event_id, "action": action })) + .await + } + + async fn respond(&self, params: Value) -> Result<(), PactoError> { + let frame = json!({ + "jsonrpc": "2.0", + "method": "handler.response", + "params": params, + }); + let mut guard = self.write.lock().await; + let writer = guard + .as_mut() + .ok_or_else(|| PactoError::Protocol("Pacto agent not connected".into()))?; + match write_frame(writer, &frame).await { + Ok(()) => Ok(()), + Err(e) => { + // Drop the dead writer so the supervisor's reconnect can install + // a fresh one; surface the error to the caller. + *guard = None; + Err(PactoError::Io(e)) + } + } + } +} + +/// Connect, register, and pump inbound events — reconnecting forever until the +/// inbound receiver is dropped. +async fn supervise( + socket_path: PathBuf, + bot_id: String, + write: Arc>>, + tx: mpsc::Sender, + backoff: Duration, +) { + loop { + match connect_and_register(&socket_path, &bot_id, &write).await { + Ok(read_half) => { + info!(bot_id = %bot_id, "Pacto agent registered for dm_received"); + let consumer_gone = pump_events(read_half, &bot_id, &tx).await; + *write.lock().await = None; + if consumer_gone { + debug!("Pacto agent inbound receiver dropped; stopping"); + return; + } + warn!("Pacto agent connection closed; reconnecting"); + } + Err(e) => { + if tx.is_closed() { + return; + } + warn!(error = %e, "Pacto agent connect/register failed; retrying"); + } + } + tokio::time::sleep(backoff).await; + } +} + +async fn connect_and_register( + socket_path: &PathBuf, + bot_id: &str, + write: &Arc>>, +) -> Result { + let stream = UnixStream::connect(socket_path).await?; + let (read_half, mut write_half) = stream.into_split(); + + let register = json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "handler.register", + "params": { + "bot_ids": [bot_id], + "event_types": ["dm_received"], + "capabilities": ["ReadMessages", "SendMessages"], + }, + }); + write_frame(&mut write_half, ®ister).await?; + + // Publish the write half so replies can be sent once events arrive. The + // registration acknowledgement is consumed by the event pump below. + *write.lock().await = Some(write_half); + Ok(read_half) +} + +/// Read frames until EOF/error. Returns `true` if the inbound receiver was +/// dropped (caller should stop), `false` if the connection just died. +async fn pump_events(read_half: OwnedReadHalf, bot_id: &str, tx: &mpsc::Sender) -> bool { + let mut reader = BufReader::new(read_half); + let mut line = String::new(); + loop { + line.clear(); + match reader.read_line(&mut line).await { + Ok(0) | Err(_) => return false, + Ok(_) => {} + } + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + let msg: Value = match serde_json::from_str(trimmed) { + Ok(v) => v, + Err(e) => { + debug!(error = %e, "Pacto agent skipping unparseable frame"); + continue; + } + }; + + // An error frame is a failed `handler.register` (the only id-correlated + // request this agent sends). The connection will never deliver events, + // so treat it as a connection failure: return so the supervisor clears + // the published write half and reconnects after a backoff, rather than + // blocking forever on a dead registration. + if let Some(err) = msg.get("error") { + warn!(error = %err, "Pacto daemon rejected registration; reconnecting"); + return false; + } + match msg.get("method").and_then(Value::as_str) { + Some("agent.event") => { + if let Some(dm) = parse_inbound(bot_id, msg.get("params")) { + if tx.send(dm).await.is_err() { + return true; + } + } + } + Some(other) => debug!(method = other, "Pacto agent ignoring notification"), + None => debug!("Pacto agent registered (ack received)"), + } + } +} + +fn parse_inbound(bot_id: &str, params: Option<&Value>) -> Option { + let p = params?; + if p.get("type").and_then(Value::as_str) != Some("dm_received") { + return None; + } + let get = |k: &str| p.get(k).and_then(Value::as_str).map(String::from); + Some(InboundDm { + bot_id: get("bot_id").unwrap_or_else(|| bot_id.to_string()), + event_id: get("event_id")?, + chat_id: get("chat_id"), + author: get("author")?, + content: get("content")?, + rumor_id: get("rumor_id")?, + timestamp: p.get("timestamp").and_then(Value::as_i64).unwrap_or(0), + }) +} + +async fn write_frame(writer: &mut OwnedWriteHalf, frame: &Value) -> std::io::Result<()> { + let mut line = serde_json::to_string(frame) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + line.push('\n'); + writer.write_all(line.as_bytes()).await?; + writer.flush().await +} diff --git a/crates/pacto-client/src/client.rs b/crates/pacto-client/src/client.rs new file mode 100644 index 0000000..5fd13ca --- /dev/null +++ b/crates/pacto-client/src/client.rs @@ -0,0 +1,215 @@ +//! JSON-RPC 2.0 client over the pacto-bot-api Unix socket. + +use crate::error::PactoError; +use crate::types::{DaemonVersion, Registration}; +use serde_json::{Value, json}; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufStream}; +use tokio::net::UnixStream; +use tokio::sync::Mutex; +use tracing::{debug, warn}; + +/// A registered connection to the daemon. +struct Connection { + stream: BufStream, + next_id: u64, + registration: Registration, +} + +/// Client for sending messages into Pacto via a co-located pacto-bot-api daemon. +/// +/// Connects lazily on first use, registering as a handler with the +/// `SendMessages` capability for the configured bot. If the daemon restarts, +/// the next call transparently reconnects and re-registers. +pub struct PactoClient { + socket_path: PathBuf, + bot_id: String, + timeout: Duration, + conn: Mutex>, +} + +impl PactoClient { + pub fn new(socket_path: impl Into, bot_id: impl Into, timeout: Duration) -> Self { + Self { + socket_path: socket_path.into(), + bot_id: bot_id.into(), + timeout, + conn: Mutex::new(None), + } + } + + /// Bot identity used for outgoing messages. + pub fn bot_id(&self) -> &str { + &self.bot_id + } + + /// Whether the daemon socket exists (daemon likely running). + pub fn is_available(&self) -> bool { + Path::new(&self.socket_path).exists() + } + + /// Send an encrypted DM into Pacto. Returns the published event id. + /// + /// `recipient` is a Nostr public key (npub or hex). + pub async fn send_dm(&self, recipient: &str, content: &str) -> Result { + let params = json!({ + "bot_id": self.bot_id, + "recipient": recipient, + "content": content, + }); + let result = self.call_with_retry("agent.send_dm", params).await?; + result + .as_str() + .map(String::from) + .ok_or_else(|| PactoError::Protocol("agent.send_dm result was not a string".into())) + } + + /// Fetch the daemon version (doubles as a health check). + pub async fn version(&self) -> Result { + let result = self.call_with_retry("agent.version", Value::Null).await?; + serde_json::from_value(result) + .map_err(|e| PactoError::Protocol(format!("bad agent.version result: {e}"))) + } + + /// Issue a call, reconnecting once if the connection went stale + /// (e.g. the daemon restarted since the last call). + async fn call_with_retry(&self, method: &str, params: Value) -> Result { + let mut guard = self.conn.lock().await; + + if guard.is_none() { + *guard = Some(self.connect_and_register().await?); + } else if let Some(conn) = guard.as_mut() { + match self.call(conn, method, params.clone()).await { + Ok(result) => return Ok(result), + // A connection drop (EOF / broken pipe, e.g. the daemon + // restarted) means the request never completed, so it is safe + // to reconnect and retry it once below. + Err(PactoError::Io(e)) => { + warn!("Pacto daemon connection dropped ({e}), reconnecting"); + *guard = Some(self.connect_and_register().await?); + } + // A timeout is ambiguous: the daemon may have already acted on + // a non-idempotent call (agent.send_dm publishes a DM), so we + // must NOT silently retry and risk sending twice. Tear down the + // connection so the next call starts clean, and surface the error. + Err(e @ PactoError::Timeout(_)) => { + *guard = None; + return Err(e); + } + Err(e) => return Err(e), + } + } + + let conn = guard.as_mut().expect("connection established above"); + let result = self.call(conn, method, params).await; + if matches!(result, Err(PactoError::Io(_) | PactoError::Timeout(_))) { + *guard = None; + } + result + } + + /// Open the socket and complete `handler.register` for our bot. + async fn connect_and_register(&self) -> Result { + if !self.is_available() { + return Err(PactoError::SocketNotFound( + self.socket_path.display().to_string(), + )); + } + + let stream = UnixStream::connect(&self.socket_path).await?; + let mut conn = Connection { + stream: BufStream::new(stream), + next_id: 1, + registration: Registration { + handler_id: String::new(), + reconnect_token: String::new(), + registered_events: Vec::new(), + }, + }; + + // Outbound-only handler: no event subscriptions, just send capability. + let params = json!({ + "bot_ids": [self.bot_id], + "event_types": [], + "capabilities": ["SendMessages"], + }); + let result = self.call(&mut conn, "handler.register", params).await?; + conn.registration = serde_json::from_value(result) + .map_err(|e| PactoError::Protocol(format!("bad handler.register result: {e}")))?; + + debug!( + handler_id = %conn.registration.handler_id, + bot_id = %self.bot_id, + "Registered with Pacto daemon" + ); + Ok(conn) + } + + /// Write one request frame and read frames until our response arrives. + async fn call( + &self, + conn: &mut Connection, + method: &str, + params: Value, + ) -> Result { + let id = conn.next_id; + conn.next_id += 1; + + let mut request = json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + }); + if !params.is_null() { + request["params"] = params; + } + + let mut line = serde_json::to_string(&request) + .map_err(|e| PactoError::Protocol(format!("failed to serialize request: {e}")))?; + line.push('\n'); + + tokio::time::timeout(self.timeout, async { + conn.stream.write_all(line.as_bytes()).await?; + conn.stream.flush().await?; + + // Frames that aren't our response (daemon notifications like + // agent.status) are skipped. + loop { + let mut frame = String::new(); + let n = conn.stream.read_line(&mut frame).await?; + if n == 0 { + return Err(PactoError::Io(std::io::Error::new( + std::io::ErrorKind::UnexpectedEof, + "daemon closed connection", + ))); + } + let msg: Value = match serde_json::from_str(frame.trim()) { + Ok(v) => v, + Err(_) => continue, + }; + if msg.get("id").and_then(Value::as_u64) != Some(id) { + let method = msg.get("method").and_then(Value::as_str); + debug!(?method, "Skipping daemon frame"); + continue; + } + if let Some(error) = msg.get("error") { + return Err(PactoError::Rpc { + code: error.get("code").and_then(Value::as_i64).unwrap_or(0), + message: error + .get("message") + .and_then(Value::as_str) + .unwrap_or("unknown error") + .to_string(), + }); + } + return msg + .get("result") + .cloned() + .ok_or_else(|| PactoError::Protocol("response missing result".into())); + } + }) + .await + .map_err(|_| PactoError::Timeout(self.timeout))? + } +} diff --git a/crates/pacto-client/src/error.rs b/crates/pacto-client/src/error.rs new file mode 100644 index 0000000..164f2b9 --- /dev/null +++ b/crates/pacto-client/src/error.rs @@ -0,0 +1,21 @@ +//! Pacto client error types. + +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum PactoError { + #[error("Pacto daemon socket not found at {0}")] + SocketNotFound(String), + + #[error("Pacto daemon I/O error: {0}")] + Io(#[from] std::io::Error), + + #[error("Pacto daemon request timed out after {0:?}")] + Timeout(std::time::Duration), + + #[error("Pacto daemon returned error {code}: {message}")] + Rpc { code: i64, message: String }, + + #[error("Pacto protocol error: {0}")] + Protocol(String), +} diff --git a/crates/pacto-client/src/lib.rs b/crates/pacto-client/src/lib.rs new file mode 100644 index 0000000..023624f --- /dev/null +++ b/crates/pacto-client/src/lib.rs @@ -0,0 +1,16 @@ +//! Client for the pacto-bot-api daemon (https://github.com/covenant-gov/pacto-bot-api). +//! +//! Speaks JSON-RPC 2.0 over the daemon's Unix socket (newline-delimited +//! frames). The client registers as a handler with the `SendMessages` +//! capability and can then publish encrypted DMs into Pacto as the +//! configured bot identity. + +mod agent; +mod client; +mod error; +mod types; + +pub use agent::{InboundDm, PactoAgent}; +pub use client::PactoClient; +pub use error::PactoError; +pub use types::{DaemonVersion, Registration}; diff --git a/crates/pacto-client/src/types.rs b/crates/pacto-client/src/types.rs new file mode 100644 index 0000000..628d2d5 --- /dev/null +++ b/crates/pacto-client/src/types.rs @@ -0,0 +1,19 @@ +//! Types for the pacto-bot-api JSON-RPC contract (schemas/jsonrpc.json). + +use serde::Deserialize; + +/// Result of `handler.register`. +#[derive(Debug, Clone, Deserialize)] +pub struct Registration { + pub handler_id: String, + pub reconnect_token: String, + pub registered_events: Vec, +} + +/// Result of `agent.version`. +#[derive(Debug, Clone, Deserialize)] +pub struct DaemonVersion { + pub version: String, + #[serde(default)] + pub commit: Option, +} diff --git a/crates/pacto-client/tests/agent_test.rs b/crates/pacto-client/tests/agent_test.rs new file mode 100644 index 0000000..0c2dc14 --- /dev/null +++ b/crates/pacto-client/tests/agent_test.rs @@ -0,0 +1,238 @@ +//! Tests for the inbound PactoAgent against a mock pacto-bot-api daemon. + +use pacto_client::PactoAgent; +use serde_json::{Value, json}; +use std::path::PathBuf; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixListener; + +async fn read_frame(reader: &mut R) -> Value { + let mut line = String::new(); + reader.read_line(&mut line).await.unwrap(); + serde_json::from_str(line.trim()).unwrap() +} + +fn socket_path(dir: &tempfile::TempDir) -> PathBuf { + dir.path().join("pacto.sock") +} + +fn agent_event(event_id: &str, author: &str, content: &str) -> Value { + json!({ + "jsonrpc": "2.0", + "method": "agent.event", + "params": { + "bot_id": "test-bot", + "event_id": event_id, + "type": "dm_received", + "chat_id": author, + "content": content, + "rumor_id": "rumor-1", + "author": author, + "timestamp": 1234, + }, + }) +} + +#[tokio::test] +async fn registers_receives_dm_and_replies() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + let listener = UnixListener::bind(&path).unwrap(); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + // Split so we can read the register + response while pushing an event. + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + + // 1. Handler must register for dm_received with read+send capability. + let register = read_frame(&mut reader).await; + assert_eq!(register["method"], "handler.register"); + assert_eq!(register["params"]["bot_ids"], json!(["test-bot"])); + assert_eq!(register["params"]["event_types"], json!(["dm_received"])); + assert_eq!( + register["params"]["capabilities"], + json!(["ReadMessages", "SendMessages"]) + ); + + // Ack the registration. + let ack = json!({ + "jsonrpc": "2.0", + "id": register["id"], + "result": { + "handler_id": "h1", + "reconnect_token": "t1", + "registered_events": ["dm_received"], + }, + }); + let mut line = serde_json::to_string(&ack).unwrap(); + line.push('\n'); + write.write_all(line.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + + // 2. Deliver an inbound DM. + let ev = agent_event("evt-1", "npub1sender", "hello bot"); + let mut evline = serde_json::to_string(&ev).unwrap(); + evline.push('\n'); + write.write_all(evline.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + + // 3. Expect the agent's handler.response reply, threaded to the event. + let response = read_frame(&mut reader).await; + assert_eq!(response["method"], "handler.response"); + assert_eq!(response["params"]["event_id"], "evt-1"); + assert_eq!(response["params"]["action"], "reply"); + assert_eq!(response["params"]["content"], "echo: hello bot"); + }); + + let (agent, mut inbound) = PactoAgent::spawn(&path, "test-bot", Duration::from_millis(50)); + + let dm = tokio::time::timeout(Duration::from_secs(5), inbound.recv()) + .await + .expect("timed out waiting for inbound DM") + .expect("inbound channel closed"); + assert_eq!(dm.author, "npub1sender"); + assert_eq!(dm.content, "hello bot"); + assert_eq!(dm.event_id, "evt-1"); + + agent + .reply(&dm.event_id, &format!("echo: {}", dm.content)) + .await + .unwrap(); + + server.await.unwrap(); +} + +#[tokio::test] +async fn reconnects_and_reregisters_after_daemon_restart() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + + // First daemon: register, then drop the connection. + let listener = UnixListener::bind(&path).unwrap(); + let (agent, mut inbound) = PactoAgent::spawn(&path, "test-bot", Duration::from_millis(50)); + + let first = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + let register = read_frame(&mut reader).await; + assert_eq!(register["method"], "handler.register"); + let ack = json!({ + "jsonrpc": "2.0", "id": register["id"], + "result": {"handler_id": "h1", "reconnect_token": "t1", "registered_events": ["dm_received"]}, + }); + let mut line = serde_json::to_string(&ack).unwrap(); + line.push('\n'); + write.write_all(line.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + // Drop everything → simulate daemon restart. + }); + // The first task owns and drops the listener when it returns. + first.await.unwrap(); + let _ = std::fs::remove_file(&path); + + // Second daemon on the same path: the agent must reconnect and re-register, + // then deliver an event we can receive. + let listener = UnixListener::bind(&path).unwrap(); + let second = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + let register = read_frame(&mut reader).await; + assert_eq!(register["method"], "handler.register"); + let ack = json!({ + "jsonrpc": "2.0", "id": register["id"], + "result": {"handler_id": "h2", "reconnect_token": "t2", "registered_events": ["dm_received"]}, + }); + let mut line = serde_json::to_string(&ack).unwrap(); + line.push('\n'); + write.write_all(line.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + + let ev = agent_event("evt-2", "npub1again", "after restart"); + let mut evline = serde_json::to_string(&ev).unwrap(); + evline.push('\n'); + write.write_all(evline.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + // Keep the connection open briefly so the reply can be observed by the + // agent side; the test only asserts inbound delivery here. + tokio::time::sleep(Duration::from_millis(200)).await; + }); + + let dm = tokio::time::timeout(Duration::from_secs(5), inbound.recv()) + .await + .expect("timed out waiting for inbound DM after reconnect") + .expect("inbound channel closed"); + assert_eq!(dm.event_id, "evt-2"); + assert_eq!(dm.content, "after restart"); + + // Keep the agent alive until the inbound arrives. + drop(agent); + second.await.unwrap(); +} + +#[tokio::test] +async fn registration_error_triggers_reconnect() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + + // First daemon: reject the registration with a JSON-RPC error frame. + let listener = UnixListener::bind(&path).unwrap(); + let (_agent, mut inbound) = PactoAgent::spawn(&path, "test-bot", Duration::from_millis(50)); + + let first = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + let register = read_frame(&mut reader).await; + assert_eq!(register["method"], "handler.register"); + let err = json!({ + "jsonrpc": "2.0", + "id": register["id"], + "error": {"code": -32602, "message": "unknown bot"}, + }); + let mut line = serde_json::to_string(&err).unwrap(); + line.push('\n'); + write.write_all(line.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + // Hold briefly so the agent observes the error before we drop. + tokio::time::sleep(Duration::from_millis(100)).await; + }); + first.await.unwrap(); + let _ = std::fs::remove_file(&path); + + // Second daemon: the agent must reconnect and re-register despite the prior + // registration error, then deliver an event. + let listener = UnixListener::bind(&path).unwrap(); + let second = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let (read, mut write) = stream.into_split(); + let mut reader = BufReader::new(read); + let register = read_frame(&mut reader).await; + assert_eq!(register["method"], "handler.register"); + let ack = json!({ + "jsonrpc": "2.0", "id": register["id"], + "result": {"handler_id": "h", "reconnect_token": "t", "registered_events": ["dm_received"]}, + }); + let mut line = serde_json::to_string(&ack).unwrap(); + line.push('\n'); + write.write_all(line.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + + let ev = agent_event("evt-ok", "npub1ok", "recovered"); + let mut evline = serde_json::to_string(&ev).unwrap(); + evline.push('\n'); + write.write_all(evline.as_bytes()).await.unwrap(); + write.flush().await.unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + }); + + let dm = tokio::time::timeout(Duration::from_secs(5), inbound.recv()) + .await + .expect("timed out; agent did not reconnect after a registration error") + .expect("inbound channel closed"); + assert_eq!(dm.event_id, "evt-ok"); + second.await.unwrap(); +} diff --git a/crates/pacto-client/tests/client_test.rs b/crates/pacto-client/tests/client_test.rs new file mode 100644 index 0000000..2b3fc90 --- /dev/null +++ b/crates/pacto-client/tests/client_test.rs @@ -0,0 +1,266 @@ +//! Round-trip tests against a mock pacto-bot-api daemon on a Unix socket. + +use pacto_client::{PactoClient, PactoError}; +use serde_json::{Value, json}; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufStream}; +use tokio::net::{UnixListener, UnixStream}; + +struct MockDaemon { + listener: UnixListener, +} + +impl MockDaemon { + fn bind(path: &Path) -> Self { + Self { + listener: UnixListener::bind(path).unwrap(), + } + } + + /// Accept one connection and answer `handler.register`, then hand back + /// the stream for scripted follow-ups. + async fn accept_registered(&self) -> BufStream { + let (stream, _) = self.listener.accept().await.unwrap(); + let mut stream = BufStream::new(stream); + + let register = read_frame(&mut stream).await; + assert_eq!(register["method"], "handler.register"); + assert_eq!(register["params"]["bot_ids"], json!(["test-bot"])); + assert_eq!(register["params"]["capabilities"], json!(["SendMessages"])); + + write_frame( + &mut stream, + &json!({ + "jsonrpc": "2.0", + "id": register["id"], + "result": { + "handler_id": "handler-1", + "reconnect_token": "token-1", + "registered_events": [], + }, + }), + ) + .await; + stream + } +} + +async fn read_frame(stream: &mut BufStream) -> Value { + let mut line = String::new(); + stream.read_line(&mut line).await.unwrap(); + serde_json::from_str(line.trim()).unwrap() +} + +async fn write_frame(stream: &mut BufStream, msg: &Value) { + let mut line = serde_json::to_string(msg).unwrap(); + line.push('\n'); + stream.write_all(line.as_bytes()).await.unwrap(); + stream.flush().await.unwrap(); +} + +fn socket_path(dir: &tempfile::TempDir) -> PathBuf { + dir.path().join("pacto.sock") +} + +fn client(path: &Path) -> PactoClient { + PactoClient::new(path, "test-bot", Duration::from_secs(2)) +} + +#[tokio::test] +async fn send_dm_registers_then_returns_event_id() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + let daemon = MockDaemon::bind(&path); + + let server = tokio::spawn(async move { + let mut stream = daemon.accept_registered().await; + + let send = read_frame(&mut stream).await; + assert_eq!(send["method"], "agent.send_dm"); + assert_eq!(send["params"]["bot_id"], "test-bot"); + assert_eq!(send["params"]["recipient"], "npub1recipient"); + assert_eq!(send["params"]["content"], "hello pacto"); + + write_frame( + &mut stream, + &json!({"jsonrpc": "2.0", "id": send["id"], "result": "eventid123"}), + ) + .await; + }); + + let client = client(&path); + let event_id = client.send_dm("npub1recipient", "hello pacto").await.unwrap(); + assert_eq!(event_id, "eventid123"); + server.await.unwrap(); +} + +#[tokio::test] +async fn daemon_notifications_are_skipped_while_waiting() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + let daemon = MockDaemon::bind(&path); + + let server = tokio::spawn(async move { + let mut stream = daemon.accept_registered().await; + let send = read_frame(&mut stream).await; + + // Unsolicited notification lands before the response. + write_frame( + &mut stream, + &json!({"jsonrpc": "2.0", "method": "agent.status", "params": {"state": "ready"}}), + ) + .await; + write_frame( + &mut stream, + &json!({"jsonrpc": "2.0", "id": send["id"], "result": "eventid456"}), + ) + .await; + }); + + let client = client(&path); + let event_id = client.send_dm("npub1recipient", "hi").await.unwrap(); + assert_eq!(event_id, "eventid456"); + server.await.unwrap(); +} + +#[tokio::test] +async fn rpc_errors_are_surfaced() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + let daemon = MockDaemon::bind(&path); + + let server = tokio::spawn(async move { + let mut stream = daemon.accept_registered().await; + let send = read_frame(&mut stream).await; + write_frame( + &mut stream, + &json!({ + "jsonrpc": "2.0", + "id": send["id"], + "error": {"code": -32000, "message": "unknown bot"}, + }), + ) + .await; + }); + + let client = client(&path); + let err = client.send_dm("npub1recipient", "hi").await.unwrap_err(); + match err { + PactoError::Rpc { code, message } => { + assert_eq!(code, -32000); + assert_eq!(message, "unknown bot"); + } + other => panic!("expected Rpc error, got {other:?}"), + } + server.await.unwrap(); +} + +#[tokio::test] +async fn reconnects_after_daemon_restart() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + + // First daemon: serve one send_dm, then drop the connection. + let daemon = MockDaemon::bind(&path); + let client = client(&path); + + let server = tokio::spawn(async move { + let mut stream = daemon.accept_registered().await; + let send = read_frame(&mut stream).await; + write_frame( + &mut stream, + &json!({"jsonrpc": "2.0", "id": send["id"], "result": "first"}), + ) + .await; + // Daemon "restarts": connection and socket go away. + drop(stream); + drop(daemon); + }); + + assert_eq!(client.send_dm("npub1r", "one").await.unwrap(), "first"); + server.await.unwrap(); + std::fs::remove_file(&path).unwrap(); + + // Second daemon on the same path: client must re-register. + let daemon = MockDaemon::bind(&path); + let server = tokio::spawn(async move { + let mut stream = daemon.accept_registered().await; + let send = read_frame(&mut stream).await; + write_frame( + &mut stream, + &json!({"jsonrpc": "2.0", "id": send["id"], "result": "second"}), + ) + .await; + }); + + assert_eq!(client.send_dm("npub1r", "two").await.unwrap(), "second"); + server.await.unwrap(); +} + +#[tokio::test] +async fn timeout_does_not_resend_nonidempotent_dm() { + // A timeout is ambiguous: the daemon may already have published the DM. + // The client must surface the timeout and NOT silently retry (which would + // send the DM twice). It must also tear the connection down. + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + let daemon = MockDaemon::bind(&path); + + let server = tokio::spawn(async move { + let mut stream = daemon.accept_registered().await; + // Receive the send_dm but deliberately never respond. + let send = read_frame(&mut stream).await; + assert_eq!(send["method"], "agent.send_dm"); + // After the client times out it should close the connection rather than + // resend: the next read must be EOF (0 bytes), not a second frame. + let mut line = String::new(); + let n = stream.read_line(&mut line).await.unwrap(); + (n, line) + }); + + let client = PactoClient::new(&path, "test-bot", Duration::from_millis(200)); + let err = client.send_dm("npub1recipient", "once").await.unwrap_err(); + assert!(matches!(err, PactoError::Timeout(_)), "expected Timeout, got {err:?}"); + + let (n, second) = server.await.unwrap(); + assert_eq!(n, 0, "client resent after a timeout; got a second frame: {second}"); +} + +#[tokio::test] +async fn missing_socket_is_a_clear_error() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + + let client = client(&path); + let err = client.send_dm("npub1r", "hi").await.unwrap_err(); + assert!(matches!(err, PactoError::SocketNotFound(_))); +} + +#[tokio::test] +async fn version_parses_daemon_info() { + let dir = tempfile::tempdir().unwrap(); + let path = socket_path(&dir); + let daemon = MockDaemon::bind(&path); + + let server = tokio::spawn(async move { + let mut stream = daemon.accept_registered().await; + let req = read_frame(&mut stream).await; + assert_eq!(req["method"], "agent.version"); + write_frame( + &mut stream, + &json!({ + "jsonrpc": "2.0", + "id": req["id"], + "result": {"version": "0.6.0", "commit": "abcd1234"}, + }), + ) + .await; + }); + + let client = client(&path); + let version = client.version().await.unwrap(); + assert_eq!(version.version, "0.6.0"); + assert_eq!(version.commit.as_deref(), Some("abcd1234")); + server.await.unwrap(); +} diff --git a/crates/signal-bot/Cargo.toml b/crates/signal-bot/Cargo.toml index b9801e8..74066f8 100644 --- a/crates/signal-bot/Cargo.toml +++ b/crates/signal-bot/Cargo.toml @@ -16,6 +16,7 @@ signal-client = { path = "../signal-client" } tools = { path = "../tools" } whisper-client = { path = "../whisper-client" } x402-payments = { path = "../x402-payments" } +pacto-client = { path = "../pacto-client" } # Workspace dependencies tokio.workspace = true diff --git a/crates/signal-bot/src/commands/chat.rs b/crates/signal-bot/src/commands/chat.rs index d0ddc3d..f308545 100644 --- a/crates/signal-bot/src/commands/chat.rs +++ b/crates/signal-bot/src/commands/chat.rs @@ -1,6 +1,7 @@ //! Chat command - proxies messages to NEAR AI. use crate::commands::CommandHandler; +use crate::commands::progress::ProgressSink; use crate::error::AppResult; use async_trait::async_trait; use conversation_store::{ConversationStore, StoredToolCall}; @@ -8,7 +9,7 @@ use near_ai_client::{ FunctionDefinitionApi, Message, NearAiClient, NearAiError, Role, ToolDefinition as NearToolDefinition, }; -use signal_client::{BotMessage, SignalClient}; +use signal_client::BotMessage; use std::sync::Arc; use tools::{FunctionCall as ToolsFunctionCall, ToolCall as ToolsToolCall, ToolExecutor, ToolRegistry}; use tracing::{debug, error, info, instrument, warn}; @@ -20,7 +21,7 @@ use x402_payments::{ pub struct ChatHandler { near_ai: Arc, conversations: Arc, - signal_client: Arc, + progress: Arc, tool_executor: Arc, tool_registry: Arc, system_prompt: String, @@ -39,7 +40,7 @@ impl ChatHandler { pub fn new( near_ai: Arc, conversations: Arc, - signal_client: Arc, + progress: Arc, tool_registry: Arc, system_prompt: String, max_tool_iterations: usize, @@ -49,7 +50,7 @@ impl ChatHandler { Self { near_ai, conversations, - signal_client, + progress, tool_executor: Arc::new(ToolExecutor::new(tool_registry.clone())), tool_registry, system_prompt, @@ -65,7 +66,7 @@ impl ChatHandler { pub fn with_payments( near_ai: Arc, conversations: Arc, - signal_client: Arc, + progress: Arc, tool_registry: Arc, system_prompt: String, max_tool_iterations: usize, @@ -77,7 +78,7 @@ impl ChatHandler { Self { near_ai, conversations, - signal_client, + progress, tool_executor: Arc::new(ToolExecutor::new(tool_registry.clone())), tool_registry, system_prompt, @@ -310,15 +311,9 @@ impl ChatHandler { // Execute each tool call for tool_call in tool_calls { - // Send progress message + // Send progress message (best-effort, transport-agnostic) let progress_msg = format!("🔧 Using {}...", tool_call.function.name); - if let Err(e) = self - .signal_client - .reply(message, &progress_msg) - .await - { - warn!("Failed to send progress message: {}", e); - } + self.progress.notify(message, &progress_msg).await; // Convert to tools crate format and execute let tools_call = ToolsToolCall { @@ -433,7 +428,7 @@ impl CommandHandler for ChatHandler { #[cfg(test)] mod tests { use super::*; - use signal_client::BotMessage; + use signal_client::{BotMessage, SignalClient}; use std::time::Duration; use tools::ToolRegistry; @@ -460,7 +455,7 @@ mod tests { let handler = ChatHandler::new( Arc::new(NearAiClient::new("key", "http://localhost", "model", Duration::from_secs(30)).unwrap()), Arc::new(ConversationStore::new(50, Duration::from_secs(3600))), - Arc::new(SignalClient::new("http://localhost").unwrap()), + Arc::new(SignalClient::new("http://localhost").unwrap()) as Arc, Arc::new(ToolRegistry::new()), String::new(), 5, @@ -475,7 +470,7 @@ mod tests { let handler = ChatHandler::new( Arc::new(NearAiClient::new("key", "http://localhost", "model", Duration::from_secs(30)).unwrap()), Arc::new(ConversationStore::new(50, Duration::from_secs(3600))), - Arc::new(SignalClient::new("http://localhost").unwrap()), + Arc::new(SignalClient::new("http://localhost").unwrap()) as Arc, Arc::new(ToolRegistry::new()), String::new(), 5, diff --git a/crates/signal-bot/src/commands/mod.rs b/crates/signal-bot/src/commands/mod.rs index d51dbf6..ff6689a 100644 --- a/crates/signal-bot/src/commands/mod.rs +++ b/crates/signal-bot/src/commands/mod.rs @@ -9,8 +9,11 @@ mod help; mod manual_transcribe; mod menu_locale; mod models; +mod pact; mod privacy; +pub mod progress; mod set_language; +mod signal_relay; mod transcribe; mod translate; mod translate_all; @@ -28,8 +31,11 @@ pub use deposit::DepositHandler; pub use help::HelpHandler; pub use manual_transcribe::ManualTranscribeHandler; pub use models::ModelsHandler; +pub use pact::PactHandler; pub use privacy::PrivacyHandler; +pub use progress::ProgressSink; pub use set_language::SetLanguageHandler; +pub use signal_relay::SignalRelayHandler; pub use transcribe::TranscribeHandler; pub use translate::TranslateHandler; pub use translate_all::TranslateAllHandler; diff --git a/crates/signal-bot/src/commands/pact.rs b/crates/signal-bot/src/commands/pact.rs new file mode 100644 index 0000000..6fc30a1 --- /dev/null +++ b/crates/signal-bot/src/commands/pact.rs @@ -0,0 +1,157 @@ +//! Pact command - sends an encrypted DM into Pacto via the pacto-bot-api daemon. + +use crate::commands::CommandHandler; +use crate::error::AppResult; +use async_trait::async_trait; +use pacto_client::{PactoClient, PactoError}; +use signal_client::BotMessage; +use std::sync::Arc; +use tracing::info; + +pub struct PactHandler { + pacto: Arc, + default_recipient: Option, +} + +impl PactHandler { + pub fn new(pacto: Arc, default_recipient: Option) -> Self { + Self { + pacto, + default_recipient, + } + } + + fn usage(&self) -> String { + let default_line = match &self.default_recipient { + Some(r) => format!("\nDefault recipient: {}", shorten_key(r)), + None => String::new(), + }; + format!( + "**Pacto Messaging**\n\n\ + Send a message into Pacto:\n\ + - !pact — DM a Pacto user\n\ + - !pact — DM the default recipient{}\n\n\ + Messages are sent as bot `{}` via the pacto-bot-api daemon in this TEE.", + default_line, + self.pacto.bot_id(), + ) + } +} + +/// A token is treated as a recipient if it looks like a Nostr pubkey. +fn is_pubkey(token: &str) -> bool { + (token.starts_with("npub1") && token.len() >= 60) + || (token.len() == 64 && token.chars().all(|c| c.is_ascii_hexdigit())) +} + +fn shorten_key(key: &str) -> String { + if key.len() > 16 { + format!("{}…{}", &key[..8], &key[key.len() - 4..]) + } else { + key.to_string() + } +} + +#[async_trait] +impl CommandHandler for PactHandler { + fn trigger(&self) -> Option<&str> { + Some("!pact") + } + + fn label(&self) -> &'static str { + "pact" + } + + async fn execute(&self, message: &BotMessage) -> AppResult { + let args = message + .text + .strip_prefix("!pact") + .unwrap_or_default() + .trim(); + + if args.is_empty() || args == "help" { + return Ok(self.usage()); + } + + if args == "status" { + return Ok(match self.pacto.version().await { + Ok(v) => format!( + "Pacto daemon reachable (v{}, commit {}). Sending as bot `{}`.", + v.version, + v.commit.as_deref().unwrap_or("unknown"), + self.pacto.bot_id(), + ), + Err(e) => format!("Pacto daemon unreachable: {e}"), + }); + } + + let (recipient, content) = match args.split_once(char::is_whitespace) { + Some((first, rest)) if is_pubkey(first) => (first.to_string(), rest.trim()), + _ => match &self.default_recipient { + Some(default) => (default.clone(), args), + None => { + return Ok( + "No recipient given and no default configured.\n\ + Usage: !pact " + .to_string(), + ); + } + }, + }; + + if content.is_empty() { + return Ok("Message is empty. Usage: !pact ".to_string()); + } + + match self.pacto.send_dm(&recipient, content).await { + Ok(event_id) => { + info!( + recipient = %shorten_key(&recipient), + event_id = %event_id, + "Sent Pacto DM" + ); + Ok(format!( + "✅ Sent to {} on Pacto (event {})", + shorten_key(&recipient), + shorten_key(&event_id), + )) + } + Err(PactoError::SocketNotFound(_)) => Ok( + "Pacto messaging is not available: the pacto-bot-api daemon is not running." + .to_string(), + ), + Err(PactoError::Rpc { message, .. }) => { + Ok(format!("Pacto daemon rejected the message: {message}")) + } + Err(e) => Err(e.into()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pubkey_detection() { + assert!(is_pubkey( + "npub1qqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqqq" + )); + assert!(is_pubkey( + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + )); + assert!(!is_pubkey("hello")); + assert!(!is_pubkey("npub1short")); + // 63 hex chars is not a key + assert!(!is_pubkey(&"a".repeat(63))); + } + + #[test] + fn shorten_key_truncates_long_keys() { + assert_eq!( + shorten_key("aaaaaaaabbbbbbbbccccccccdddd"), + "aaaaaaaa…dddd" + ); + assert_eq!(shorten_key("short"), "short"); + } +} diff --git a/crates/signal-bot/src/commands/progress.rs b/crates/signal-bot/src/commands/progress.rs new file mode 100644 index 0000000..80f432a --- /dev/null +++ b/crates/signal-bot/src/commands/progress.rs @@ -0,0 +1,26 @@ +//! Transport-agnostic progress notifications. +//! +//! The chat handler emits best-effort progress pings (e.g. "🔧 Using +//! web_search...") while it runs tools. On Signal these go out as a reply; on +//! Pacto they go out as a DM. `ProgressSink` abstracts that so the same chat +//! logic drives both transports. + +use async_trait::async_trait; +use signal_client::{BotMessage, SignalClient}; +use tracing::warn; + +/// Sends best-effort, non-final progress messages back to the user. +#[async_trait] +pub trait ProgressSink: Send + Sync { + /// Deliver an intermediate progress message. Failures are non-fatal. + async fn notify(&self, message: &BotMessage, text: &str); +} + +#[async_trait] +impl ProgressSink for SignalClient { + async fn notify(&self, message: &BotMessage, text: &str) { + if let Err(e) = self.reply(message, text).await { + warn!("Failed to send progress message: {}", e); + } + } +} diff --git a/crates/signal-bot/src/commands/signal_relay.rs b/crates/signal-bot/src/commands/signal_relay.rs new file mode 100644 index 0000000..ff4bfc1 --- /dev/null +++ b/crates/signal-bot/src/commands/signal_relay.rs @@ -0,0 +1,184 @@ +//! `!signal` — lets a Pacto user DM a Signal user through the bot. +//! +//! This is the mirror of `!pact`. Because it can deliver to arbitrary Signal +//! numbers, it is off by default and gated by an allowlist. The relayed message +//! is prefixed with the sender's Pacto npub so the Signal recipient knows who +//! it is from and can reply with `!pact `. + +use crate::commands::CommandHandler; +use crate::error::AppResult; +use async_trait::async_trait; +use signal_client::{BotMessage, SignalClient}; +use std::sync::Arc; +use tracing::info; + +pub struct SignalRelayHandler { + signal: Arc, + /// The bot's own Signal number (the "from" for relayed messages). + from_number: String, + /// Permitted recipient numbers (E.164), or `["*"]` for any. + allowlist: Vec, +} + +impl SignalRelayHandler { + pub fn new(signal: Arc, from_number: String, allowlist: Vec) -> Self { + Self { + signal, + from_number, + allowlist, + } + } + + fn usage(&self) -> String { + "**Message a Signal user**\n\n\ + - !signal <+number> — DM a Signal user\n\n\ + They'll see your Pacto npub and can reply with !pact." + .to_string() + } + + fn is_allowed(&self, recipient: &str) -> bool { + self.allowlist.iter().any(|a| a == "*" || a == recipient) + } +} + +/// A plausible E.164 number: `+` followed by 7–15 digits. +fn is_e164(s: &str) -> bool { + let digits = s.strip_prefix('+').unwrap_or(""); + !digits.is_empty() + && digits.len() >= 7 + && digits.len() <= 15 + && digits.chars().all(|c| c.is_ascii_digit()) +} + +#[async_trait] +impl CommandHandler for SignalRelayHandler { + fn trigger(&self) -> Option<&str> { + Some("!signal") + } + + fn label(&self) -> &'static str { + "signal-relay" + } + + async fn execute(&self, message: &BotMessage) -> AppResult { + let args = message + .text + .strip_prefix("!signal") + .unwrap_or_default() + .trim(); + + if args.is_empty() || args == "help" { + return Ok(self.usage()); + } + + let Some((recipient, body)) = args.split_once(char::is_whitespace) else { + return Ok(self.usage()); + }; + let recipient = recipient.trim(); + let body = body.trim(); + + if !is_e164(recipient) { + return Ok(format!( + "'{recipient}' is not a valid phone number. Use E.164, e.g. +14155551234." + )); + } + if body.is_empty() { + return Ok("Message is empty. Usage: !signal <+number> ".to_string()); + } + if !self.is_allowed(recipient) { + return Ok( + "That Signal number isn't reachable from Pacto (not on the allowlist).".to_string(), + ); + } + + // `message.source` is the sender's Pacto npub. Prefix it so the Signal + // user knows who it's from and can reply via `!pact `. + let relayed = format!("💬 Pacto DM from {}:\n\n{}", message.source, body); + + self.signal + .send(&self.from_number, recipient, &relayed) + .await?; + + info!(recipient = %recipient, "Relayed Pacto→Signal DM"); + Ok(format!( + "✅ Delivered to {}. They can reply with !pact {}.", + recipient, message.source + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn handler(allowlist: &[&str]) -> SignalRelayHandler { + SignalRelayHandler::new( + Arc::new(SignalClient::new("http://localhost").unwrap()), + "+15550000000".to_string(), + allowlist.iter().map(|s| s.to_string()).collect(), + ) + } + + fn dm(text: &str) -> BotMessage { + BotMessage { + source: "npub1sender".into(), + text: text.into(), + timestamp: 0, + message_timestamp: 0, + is_group: false, + group_id: None, + receiving_account: "sigstack".into(), + attachments: vec![], + quote: None, + } + } + + #[test] + fn e164_validation() { + assert!(is_e164("+14155551234")); + assert!(is_e164("+15550000000")); + assert!(!is_e164("14155551234")); // missing + + assert!(!is_e164("+abc")); // non-digits + assert!(!is_e164("+123")); // too short + assert!(!is_e164("+1234567890123456")); // too long + } + + #[test] + fn allowlist_gates_recipients() { + let h = handler(&["+14155551234"]); + assert!(h.is_allowed("+14155551234")); + assert!(!h.is_allowed("+19998887777")); + + let open = handler(&["*"]); + assert!(open.is_allowed("+19998887777")); + + let empty = handler(&[]); + assert!(!empty.is_allowed("+14155551234")); // empty = deny all + } + + #[tokio::test] + async fn usage_and_rejections_do_not_send() { + let h = handler(&["+14155551234"]); + // No network is touched on any of these paths. + assert!(h.execute(&dm("!signal")).await.unwrap().contains("Message a Signal user")); + assert!( + h.execute(&dm("!signal notaphone hi")) + .await + .unwrap() + .contains("not a valid phone number") + ); + assert!( + h.execute(&dm("!signal +19998887777 hi")) + .await + .unwrap() + .contains("allowlist") + ); + assert!( + h.execute(&dm("!signal +14155551234")) + .await + .unwrap() + .contains("Message a Signal user") // no body → usage + ); + } +} diff --git a/crates/signal-bot/src/config.rs b/crates/signal-bot/src/config.rs index 19c6887..7c601bb 100644 --- a/crates/signal-bot/src/config.rs +++ b/crates/signal-bot/src/config.rs @@ -45,6 +45,10 @@ pub struct Config { /// Encrypted persistence for per-group bot preferences #[serde(default)] pub group_preferences: GroupPreferencesConfig, + + /// Pacto messaging (via pacto-bot-api daemon) configuration + #[serde(default)] + pub pacto: PactoConfig, } #[derive(Debug, Clone, Deserialize)] @@ -53,6 +57,11 @@ pub struct SignalConfig { #[serde(default = "default_signal_service")] pub service_url: String, + /// The bot's own Signal number (E.164). Needed to send proactively, e.g. + /// relaying a Pacto user's message to a Signal user. + #[serde(default)] + pub phone_number: Option, + /// Poll interval for messages #[serde(default = "default_poll_interval", with = "humantime_serde")] pub poll_interval: Duration, @@ -206,11 +215,54 @@ pub struct GroupPreferencesConfig { pub storage_path: String, } +#[derive(Debug, Clone, Deserialize)] +pub struct PactoConfig { + /// Master switch for Pacto integration (requires a pacto-bot-api daemon). + /// Enables the `!pact` outbound command for Signal users. + #[serde(default)] + pub enabled: bool, + + /// When enabled (and `enabled`), also run the inbound DM agent so Pacto + /// users get the full bot experience (AI chat + commands). Set false for + /// outbound-only (`!pact`) deployments. + #[serde(default = "default_true")] + pub agent_enabled: bool, + + /// pacto-bot-api daemon Unix socket path (shared Docker volume in production) + #[serde(default = "default_pacto_socket")] + pub socket_path: String, + + /// Bot identity id configured in the daemon's pacto-bot-api.toml + #[serde(default = "default_pacto_bot_id")] + pub bot_id: String, + + /// Recipient (npub or hex pubkey) used when `!pact` is called without one + #[serde(default)] + pub default_recipient: Option, + + /// Allow Pacto users to DM Signal users via `!signal `. + /// Off by default — this can relay to arbitrary Signal numbers, so it is + /// gated by an allowlist (see `signal_relay_allowlist`). + #[serde(default)] + pub signal_relay_enabled: bool, + + /// Comma-separated E.164 numbers a Pacto user may reach via `!signal`. + /// Empty = deny all (the safe default); `*` = allow any number (open relay, + /// abuse risk — use deliberately). + #[serde(default)] + pub signal_relay_allowlist: String, + + /// Max time per daemon request + #[serde(default = "default_pacto_timeout", with = "humantime_serde")] + pub timeout: Duration, +} + // Default implementations impl Default for SignalConfig { fn default() -> Self { Self { service_url: default_signal_service(), + phone_number: None, poll_interval: default_poll_interval(), } } @@ -313,6 +365,21 @@ impl Default for GroupPreferencesConfig { } } +impl Default for PactoConfig { + fn default() -> Self { + Self { + enabled: false, + agent_enabled: default_true(), + socket_path: default_pacto_socket(), + bot_id: default_pacto_bot_id(), + default_recipient: None, + signal_relay_enabled: false, + signal_relay_allowlist: String::new(), + timeout: default_pacto_timeout(), + } + } +} + // Default value functions fn default_signal_service() -> String { "http://signal-api:8080".into() @@ -443,6 +510,18 @@ fn default_group_preferences_path() -> String { "/data/group_prefs.enc".into() } +fn default_pacto_socket() -> String { + "/var/run/pacto/pacto-bot-api.sock".into() +} + +fn default_pacto_bot_id() -> String { + "sigstack".into() +} + +fn default_pacto_timeout() -> Duration { + Duration::from_secs(15) +} + impl Config { /// Load configuration from environment variables. pub fn load() -> Result { diff --git a/crates/signal-bot/src/error.rs b/crates/signal-bot/src/error.rs index 2bedfa6..ae5ce5c 100644 --- a/crates/signal-bot/src/error.rs +++ b/crates/signal-bot/src/error.rs @@ -22,6 +22,9 @@ pub enum AppError { #[error("Whisper error: {0}")] Whisper(#[from] whisper_client::WhisperError), + + #[error("Pacto error: {0}")] + Pacto(#[from] pacto_client::PactoError), } /// Result type alias for application errors. diff --git a/crates/signal-bot/src/lib.rs b/crates/signal-bot/src/lib.rs index 9aa7957..c602946 100644 --- a/crates/signal-bot/src/lib.rs +++ b/crates/signal-bot/src/lib.rs @@ -3,5 +3,6 @@ pub mod config; pub mod error; pub mod group_preferences_store; pub mod menu_language; +pub mod pacto_agent; pub mod transcribe_store; pub mod voice_attachment_cache; diff --git a/crates/signal-bot/src/main.rs b/crates/signal-bot/src/main.rs index b9f5e41..aeedda6 100644 --- a/crates/signal-bot/src/main.rs +++ b/crates/signal-bot/src/main.rs @@ -10,6 +10,8 @@ use anyhow::Context; use conversation_store::ConversationStore; use dstack_client::DstackClient; use near_ai_client::NearAiClient; +use pacto_client::{PactoAgent, PactoClient}; +use signal_bot::pacto_agent; use signal_client::{MessageReceiver, SignalClient}; use whisper_client::WhisperClient; use std::path::PathBuf; @@ -162,6 +164,101 @@ async fn main() -> AppResult<()> { } info!("Signal API healthy"); + let pacto_client = if config.pacto.enabled { + let pacto = Arc::new(PactoClient::new( + &config.pacto.socket_path, + config.pacto.bot_id.clone(), + config.pacto.timeout, + )); + + match pacto.version().await { + Ok(v) => info!( + "Pacto daemon healthy (v{}) - sending as bot '{}'", + v.version, + pacto.bot_id() + ), + Err(e) => warn!( + "Pacto daemon not reachable at {} ({e}) - !pact will retry on use", + config.pacto.socket_path + ), + } + + Some(pacto) + } else { + info!("Pacto messaging disabled"); + None + }; + + // Inbound Pacto DM agent — gives Pacto users the same DM experience as + // Signal users (AI chat + core commands). Outbound `!pact` still works + // without it; disable via PACTO__AGENT_ENABLED=false. + if let Some(ref pacto) = pacto_client { + if config.pacto.agent_enabled { + let (agent, inbound) = PactoAgent::spawn( + config.pacto.socket_path.clone(), + config.pacto.bot_id.clone(), + std::time::Duration::from_secs(3), + ); + + // Optional Pacto→Signal relay (!signal). Requires the bot's own + // Signal number and an allowlist of reachable recipients. + let signal_relay = if config.pacto.signal_relay_enabled { + match &config.signal.phone_number { + Some(from_number) => { + let allowlist: Vec = config + .pacto + .signal_relay_allowlist + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(String::from) + .collect(); + if allowlist.is_empty() { + warn!( + "Pacto→Signal relay enabled but allowlist is empty - all \ + recipients will be denied (set PACTO__SIGNAL_RELAY_ALLOWLIST)" + ); + } + Some(pacto_agent::SignalRelay { + signal: signal.clone(), + from_number: from_number.clone(), + allowlist, + }) + } + None => { + warn!( + "Pacto→Signal relay enabled but SIGNAL__PHONE_NUMBER is not set - \ + disabling !signal" + ); + None + } + } + } else { + None + }; + + pacto_agent::spawn( + inbound, + agent, + pacto_agent::PactoAgentDeps { + near_ai: near_ai.clone(), + conversations: conversations.clone(), + tool_registry: tool_registry.clone(), + dstack: dstack.clone(), + pacto_client: pacto.clone(), + system_prompt: config.bot.system_prompt.clone(), + max_tool_calls: config.tools.max_tool_calls, + signal_username: config.bot.signal_username.clone(), + github_repo: config.bot.github_repo.clone(), + signal_relay, + }, + ); + info!("Pacto DM agent enabled: Pacto users get AI chat + commands"); + } else { + info!("Pacto DM agent disabled (outbound !pact only)"); + } + } + let whisper_client = if config.whisper.enabled { let whisper = Arc::new( WhisperClient::new( @@ -300,6 +397,17 @@ async fn main() -> AppResult<()> { info!("Payment commands enabled: !balance, !deposit"); } + // Add Pacto messaging if enabled + if let Some(ref pacto) = pacto_client { + let default_recipient = config + .pacto + .default_recipient + .clone() + .filter(|r| !r.trim().is_empty()); + handlers.push(Box::new(PactHandler::new(pacto.clone(), default_recipient))); + info!("Pacto messaging enabled: !pact"); + } + info!("Registered {} command handlers", handlers.len()); info!("NEAR AI endpoint: {}", config.near_ai.base_url); info!("Listening for messages..."); diff --git a/crates/signal-bot/src/pacto_agent.rs b/crates/signal-bot/src/pacto_agent.rs new file mode 100644 index 0000000..bff092b --- /dev/null +++ b/crates/signal-bot/src/pacto_agent.rs @@ -0,0 +1,350 @@ +//! Inbound Pacto DM agent — gives Pacto users the same DM experience Signal +//! users get. +//! +//! A background task consumes decrypted DMs delivered by the pacto-bot-api +//! daemon, runs each one through the *same* command/AI-chat handlers the Signal +//! side uses (via a synthetic [`BotMessage`]), and replies over Pacto. +//! +//! ## Parity ceiling +//! +//! The daemon only delivers `dm_received` events — it does not expose inbound +//! MLS group messages or audio attachments to handlers. So Pacto parity covers +//! the full **DM** experience (AI chat with tools, `!verify`, `!clear`, +//! `!models`, `!help`, `!privacy`, `!list-langs`, and AI-driven translation), +//! but **group translation and voice transcription are not possible** until the +//! daemon gains inbound group/attachment delivery. + +use crate::commands::{ + ChatHandler, ClearHandler, CommandHandler, ModelsHandler, ProgressSink, SignalRelayHandler, + TranslateLangsHandler, VerifyHandler, +}; +use async_trait::async_trait; +use conversation_store::ConversationStore; +use dstack_client::DstackClient; +use near_ai_client::NearAiClient; +use pacto_client::{InboundDm, PactoAgent, PactoClient}; +use signal_client::{BotMessage, SignalClient}; +use std::sync::Arc; +use tokio::sync::mpsc; +use tools::ToolRegistry; +use tracing::{debug, error, info, warn}; + +/// Everything the agent loop needs to construct the Pacto DM handler set. +pub struct PactoAgentDeps { + pub near_ai: Arc, + pub conversations: Arc, + pub tool_registry: Arc, + pub dstack: Arc, + /// Outbound request/response client, reused to deliver progress pings. + pub pacto_client: Arc, + pub system_prompt: String, + pub max_tool_calls: usize, + pub signal_username: Option, + pub github_repo: Option, + /// Optional `!signal` relay (Pacto user → Signal user). `None` disables it. + pub signal_relay: Option, +} + +/// Configuration for the `!signal` relay handler. +pub struct SignalRelay { + pub signal: Arc, + /// The bot's own Signal number (E.164) to send from. + pub from_number: String, + /// Permitted recipient numbers, or `["*"]` for any. + pub allowlist: Vec, +} + +/// Spawn the inbound Pacto DM loop. Returns immediately; the loop runs until the +/// inbound channel closes (daemon supervisor gives up / process shuts down). +pub fn spawn(inbound: mpsc::Receiver, agent: Arc, deps: PactoAgentDeps) { + let handlers = build_handlers(deps); + tokio::spawn(run(inbound, agent, handlers)); +} + +fn build_handlers(deps: PactoAgentDeps) -> Vec> { + let progress: Arc = Arc::new(PactoProgress { + pacto: deps.pacto_client, + }); + let chat = ChatHandler::new( + deps.near_ai.clone(), + deps.conversations.clone(), + progress, + deps.tool_registry, + deps.system_prompt, + deps.max_tool_calls, + deps.signal_username, + deps.github_repo, + ); + + // Command handlers first, AI chat (default) last — same precedence as the + // Signal dispatch loop. Only transport-clean handlers are reused; voice, + // group-translate, and Signal-quote `!translate` are Signal-specific. + let relay_enabled = deps.signal_relay.is_some(); + let mut handlers: Vec> = vec![ + Box::new(VerifyHandler::new(deps.dstack)), + Box::new(ClearHandler::new(deps.conversations)), + Box::new(ModelsHandler::new(deps.near_ai)), + Box::new(PactoHelpHandler { relay_enabled }), + Box::new(PactoPrivacyHandler), + Box::new(TranslateLangsHandler::new()), + ]; + + // Optional: let Pacto users DM Signal users via `!signal` (allowlist-gated). + if let Some(relay) = deps.signal_relay { + handlers.push(Box::new(SignalRelayHandler::new( + relay.signal, + relay.from_number, + relay.allowlist, + ))); + info!("Pacto→Signal relay enabled: !signal"); + } + + // AI chat is the default (matches non-command free text), so it goes last. + handlers.push(Box::new(chat)); + handlers +} + +async fn run( + mut inbound: mpsc::Receiver, + agent: Arc, + handlers: Vec>, +) { + info!("Pacto DM agent ready ({} handlers)", handlers.len()); + while let Some(dm) = inbound.recv().await { + let message = to_bot_message(&dm); + debug!(author = %shorten(&dm.author), "Pacto DM received"); + + match dispatch(&handlers, &message).await { + Some(reply) => { + if let Err(e) = agent.reply(&dm.event_id, &reply).await { + warn!(error = %e, "Failed to send Pacto reply"); + } + } + None => { + // Nothing to say (empty text or unknown !command): terminate the + // event cleanly so the daemon doesn't consider it unhandled. + let _ = agent.finish(&dm.event_id, "ignore").await; + } + } + } + info!("Pacto DM agent stopped"); +} + +/// Find the first matching handler and run it, mirroring the Signal main loop. +async fn dispatch(handlers: &[Box], message: &BotMessage) -> Option { + let handler = handlers.iter().find(|h| h.matches(message))?; + match handler.execute(message).await { + Ok(reply) => Some(reply), + Err(e) => { + error!(handler = handler.label(), error = %e, "Pacto handler error"); + Some("Sorry, something went wrong.".to_string()) + } + } +} + +/// Present a Pacto DM as a Signal-shaped [`BotMessage`] so existing handlers +/// work unchanged. The author's pubkey is the `source`, so `reply_target()` +/// (and thus the conversation key) is per-sender. +fn to_bot_message(dm: &InboundDm) -> BotMessage { + BotMessage { + source: dm.author.clone(), + text: dm.content.clone(), + timestamp: dm.timestamp, + message_timestamp: dm.timestamp, + is_group: false, + group_id: None, + receiving_account: dm.bot_id.clone(), + attachments: Vec::new(), + quote: None, + } +} + +fn shorten(key: &str) -> String { + if key.len() > 12 { + format!("{}…", &key[..12]) + } else { + key.to_string() + } +} + +/// Progress sink that delivers "🔧 Using ..." pings as plain Pacto DMs via the +/// request/response client (the final reply goes over the agent connection). +struct PactoProgress { + pacto: Arc, +} + +#[async_trait] +impl ProgressSink for PactoProgress { + async fn notify(&self, message: &BotMessage, text: &str) { + if let Err(e) = self.pacto.send_dm(&message.source, text).await { + debug!(error = %e, "Failed to send Pacto progress ping"); + } + } +} + +/// Pacto-accurate `!help` (omits Signal-only voice/group features). +struct PactoHelpHandler { + /// Whether the `!signal` relay is active (so help only lists it when usable). + relay_enabled: bool, +} + +#[async_trait] +impl CommandHandler for PactoHelpHandler { + fn trigger(&self) -> Option<&str> { + Some("!help") + } + + fn label(&self) -> &'static str { + "pacto-help" + } + + async fn execute(&self, _message: &BotMessage) -> crate::error::AppResult { + let mut help = PACTO_HELP.to_string(); + if self.relay_enabled { + help.push_str("\n- !signal <+number> — DM a Signal user"); + } + Ok(help) + } +} + +/// Pacto-accurate `!privacy`. +struct PactoPrivacyHandler; + +#[async_trait] +impl CommandHandler for PactoPrivacyHandler { + fn trigger(&self) -> Option<&str> { + Some("!privacy") + } + + fn label(&self) -> &'static str { + "pacto-privacy" + } + + async fn execute(&self, _message: &BotMessage) -> crate::error::AppResult { + Ok(PACTO_PRIVACY.to_string()) + } +} + +const PACTO_HELP: &str = r#"**Bread Coop AI on Pacto** (Private & Verifiable) + +**AI chat:** +- Just message me — I reply with private AI inference +- I can search the web, do math, and check the weather +- Ask me to translate, e.g. "translate 'good morning' to Spanish" + +**Commands:** +- !verify — cryptographic TEE attestation +- !models — list available AI models +- !clear — clear our conversation history +- !list-langs — supported translation languages +- !privacy — privacy & security details +- !help — this menu"#; + +const PACTO_PRIVACY: &str = r#"**Bread Coop AI on Pacto** (Private & Verifiable) + +Your messages are end-to-end encrypted over Nostr (NIP-17 gift wraps) and +decrypted only inside an Intel TDX Trusted Execution Environment. The bot's +Nostr key and your plaintext never leave the TEE. + +AI inference runs on NEAR AI Cloud's private GPU TEE. Neither the bot operator +nor the AI provider can read your messages. + +Send `!verify ` for a fresh hardware attestation proving this bot +runs in a genuine TEE."#; + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn test_handlers() -> Vec> { + build_handlers(PactoAgentDeps { + near_ai: Arc::new( + NearAiClient::new("k", "http://localhost", "m", Duration::from_secs(5)).unwrap(), + ), + conversations: Arc::new(ConversationStore::new(50, Duration::from_secs(3600))), + tool_registry: Arc::new(ToolRegistry::new()), + dstack: Arc::new(DstackClient::new("/tmp/nonexistent-dstack.sock")), + pacto_client: Arc::new(PactoClient::new( + "/tmp/nonexistent-pacto.sock", + "test-bot", + Duration::from_secs(5), + )), + system_prompt: String::new(), + max_tool_calls: 5, + signal_username: None, + github_repo: None, + signal_relay: None, + }) + } + + fn dm(text: &str) -> InboundDm { + InboundDm { + bot_id: "test-bot".into(), + event_id: "evt".into(), + chat_id: Some("npub1author".into()), + author: "npub1author".into(), + content: text.into(), + rumor_id: "rumor".into(), + timestamp: 0, + } + } + + #[test] + fn bot_message_is_a_dm_keyed_by_author() { + let m = to_bot_message(&dm("hi")); + assert_eq!(m.source, "npub1author"); + assert!(!m.is_group); + assert_eq!(m.group_id, None); + // Conversation key is the sender, so history is per-Pacto-user. + assert_eq!(m.reply_target(), "npub1author"); + assert_eq!(m.receiving_account, "test-bot"); + } + + #[tokio::test] + async fn help_routes_to_pacto_specific_menu() { + let handlers = test_handlers(); + let reply = dispatch(&handlers, &to_bot_message(&dm("!help"))) + .await + .expect("help should produce a reply"); + assert!(reply.contains("Bread Coop AI on Pacto")); + // Must not advertise Signal-only features Pacto can't offer. + assert!(!reply.contains("!transcribe")); + assert!(!reply.contains("!translate-on")); + } + + #[tokio::test] + async fn privacy_and_list_langs_route() { + let handlers = test_handlers(); + let privacy = dispatch(&handlers, &to_bot_message(&dm("!privacy"))) + .await + .expect("privacy reply"); + assert!(privacy.contains("TDX")); + assert!( + dispatch(&handlers, &to_bot_message(&dm("!list-langs"))) + .await + .is_some() + ); + } + + #[tokio::test] + async fn empty_message_is_ignored() { + let handlers = test_handlers(); + assert!( + dispatch(&handlers, &to_bot_message(&dm(" "))) + .await + .is_none() + ); + } + + #[tokio::test] + async fn unknown_command_is_ignored() { + let handlers = test_handlers(); + // A "!"-prefixed token matches no command and is excluded from chat. + assert!( + dispatch(&handlers, &to_bot_message(&dm("!nope"))) + .await + .is_none() + ); + } +} diff --git a/docker/Dockerfile.pacto b/docker/Dockerfile.pacto new file mode 100644 index 0000000..a57be5f --- /dev/null +++ b/docker/Dockerfile.pacto @@ -0,0 +1,44 @@ +# Derived pacto-bot-api image for Phala deployment. +# +# Phala CVMs cannot bind-mount local files, so the daemon config is baked in. +# The nsec is NOT baked in — it stays in the PACTO_BOT_NSEC encrypted secret, +# which the config references via ${PACTO_BOT_NSEC}. +# +# Build (from the docker/ directory, after copying pacto-bot-api.toml.example +# to pacto-bot-api.toml and filling in your bot's npub): +# docker buildx build --platform linux/amd64 \ +# -t YOUR_DOCKERHUB/pacto-bot-api:v0.6.0 \ +# -f Dockerfile.pacto --push . + +# Upstream publishes no container image yet, so the base is built from the +# pinned source tag using the upstream Dockerfile. +FROM rust:1.96-bookworm AS builder +ARG PACTO_VERSION=v0.6.0 +RUN git clone --depth 1 --branch ${PACTO_VERSION} \ + https://github.com/covenant-gov/pacto-bot-api /usr/src/pacto-bot-api +WORKDIR /usr/src/pacto-bot-api +RUN cargo build --release --bins + +FROM debian:bookworm-slim AS runtime + +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates \ + && rm -rf /var/lib/apt/lists/* + +RUN groupadd --system --gid 1000 pacto \ + && useradd --system --uid 1000 --gid pacto \ + --home-dir /var/lib/pacto-bot-api --shell /sbin/nologin pacto + +RUN mkdir -p /etc/pacto /var/lib/pacto-bot-api \ + && chown -R pacto:pacto /etc/pacto /var/lib/pacto-bot-api + +COPY --from=builder /usr/src/pacto-bot-api/target/release/pacto-bot-api /usr/local/bin/ +COPY --from=builder /usr/src/pacto-bot-api/target/release/pacto-bot-admin /usr/local/bin/ +COPY --chown=pacto:pacto pacto-bot-api.toml /etc/pacto/pacto-bot-api.toml + +USER pacto +WORKDIR /var/lib/pacto-bot-api + +VOLUME ["/var/lib/pacto-bot-api"] + +CMD ["pacto-bot-api", "--config", "/etc/pacto/pacto-bot-api.toml", "--data-dir", "/var/lib/pacto-bot-api"] diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index ec14288..290b9ff 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -47,9 +47,20 @@ services: - DSTACK__SOCKET_PATH=/var/run/dstack.sock - GROUP_PREFERENCES__PERSIST=${GROUP_PREFERENCES_PERSIST:-true} - GROUP_PREFERENCES__STORAGE_PATH=/data/group_prefs.enc + # Pacto messaging (requires the pacto-bot-api service: --profile pacto) + - PACTO__ENABLED=${PACTO_ENABLED:-false} + # Inbound DM agent: Pacto users get AI chat + commands (parity with Signal) + - PACTO__AGENT_ENABLED=${PACTO_AGENT_ENABLED:-true} + - PACTO__SOCKET_PATH=/var/run/pacto/pacto-bot-api.sock + - PACTO__BOT_ID=${PACTO_BOT_ID:-sigstack} + - PACTO__DEFAULT_RECIPIENT=${PACTO_DEFAULT_RECIPIENT:-} + # Pacto→Signal relay (!signal): off by default, allowlist-gated + - PACTO__SIGNAL_RELAY_ENABLED=${PACTO_SIGNAL_RELAY_ENABLED:-false} + - PACTO__SIGNAL_RELAY_ALLOWLIST=${PACTO_SIGNAL_RELAY_ALLOWLIST:-} volumes: - /var/run/dstack.sock:/var/run/dstack.sock:ro - group-prefs-data:/data + - pacto-data:/var/run/pacto networks: - internal depends_on: @@ -79,6 +90,27 @@ services: start_period: 300s restart: unless-stopped + # Pacto messaging daemon (https://github.com/covenant-gov/pacto-bot-api). + # Opt-in: enable with `docker compose --profile pacto up -d` after copying + # pacto-bot-api.toml.example to pacto-bot-api.toml and setting PACTO_BOT_NSEC + # (plus PACTO_ENABLED=true on signal-bot). + pacto-bot-api: + build: + # Pinned to a release tag; bump deliberately. + context: https://github.com/covenant-gov/pacto-bot-api.git#v0.6.0 + container_name: pacto-bot-api + profiles: ["pacto"] + environment: + - PACTO_BOT_NSEC=${PACTO_BOT_NSEC} + - RUST_LOG=${LOG_LEVEL:-info} + volumes: + - ./pacto-bot-api.toml:/etc/pacto/pacto-bot-api.toml:ro + # Shared with signal-bot: the daemon's Unix socket lives in this volume + - pacto-data:/var/lib/pacto-bot-api + networks: + - internal + restart: unless-stopped + signal-registration-proxy: build: context: .. @@ -121,6 +153,8 @@ volumes: driver: local group-prefs-data: driver: local + pacto-data: + driver: local networks: internal: diff --git a/docker/pacto-bot-api.toml.example b/docker/pacto-bot-api.toml.example new file mode 100644 index 0000000..d444719 --- /dev/null +++ b/docker/pacto-bot-api.toml.example @@ -0,0 +1,24 @@ +# pacto-bot-api daemon config for the sigstack deployment. +# +# Copy to docker/pacto-bot-api.toml (gitignored), fill in the bot identity, +# and start the daemon with: docker compose --profile pacto up -d +# +# Generate a bot identity with: +# pacto-bot-admin new sigstack --backend nsec +# (or: docker compose --profile pacto run --rm pacto-bot-api pacto-bot-admin new sigstack --backend nsec) +# +# The nsec is injected via the PACTO_BOT_NSEC environment variable — never +# write it into this file. For production, prefer a NIP-46 bunker backend. + +[daemon] +data_dir = "/var/lib/pacto-bot-api" +# The socket lives in the shared Docker volume so signal-bot can reach it. +socket_path = "/var/lib/pacto-bot-api/pacto-bot-api.sock" + +[[bots]] +# Must match PACTO__BOT_ID on the signal-bot service. +id = "sigstack" +npub = "npub1REPLACE_WITH_YOUR_BOT_NPUB" +signing = { backend = "nsec", nsec = "${PACTO_BOT_NSEC}" } +relays = ["wss://relay.pacto.chat"] +capabilities = ["ReadMessages", "SendMessages"] diff --git a/docker/phala-compose.yaml b/docker/phala-compose.yaml index 3d4dd93..cbb03e2 100644 --- a/docker/phala-compose.yaml +++ b/docker/phala-compose.yaml @@ -48,6 +48,8 @@ services: container_name: signal-bot environment: - SIGNAL__SERVICE_URL=http://signal-api:8080 + # Bot's own Signal number — required for the Pacto→Signal !signal relay + - SIGNAL__PHONE_NUMBER=${SIGNAL_PHONE:-} - NEAR_AI__API_KEY=${NEAR_AI_API_KEY} - NEAR_AI__BASE_URL=${NEAR_AI_BASE_URL:-https://cloud-api.near.ai/v1} - NEAR_AI__MODEL=${NEAR_AI_MODEL:-deepseek-ai/DeepSeek-V3.1} @@ -69,14 +71,42 @@ services: - DSTACK__SOCKET_PATH=/var/run/dstack.sock - GROUP_PREFERENCES__PERSIST=true - GROUP_PREFERENCES__STORAGE_PATH=/data/group_prefs.enc + # Pacto messaging — set PACTO_ENABLED=true and uncomment the + # pacto-bot-api service below to activate + - PACTO__ENABLED=${PACTO_ENABLED:-false} + # Inbound DM agent: Pacto users get AI chat + commands (parity with Signal) + - PACTO__AGENT_ENABLED=${PACTO_AGENT_ENABLED:-true} + - PACTO__SOCKET_PATH=/var/run/pacto/pacto-bot-api.sock + - PACTO__BOT_ID=${PACTO_BOT_ID:-sigstack} + - PACTO__DEFAULT_RECIPIENT=${PACTO_DEFAULT_RECIPIENT:-} + # Pacto→Signal relay (!signal): off by default, allowlist-gated + - PACTO__SIGNAL_RELAY_ENABLED=${PACTO_SIGNAL_RELAY_ENABLED:-false} + - PACTO__SIGNAL_RELAY_ALLOWLIST=${PACTO_SIGNAL_RELAY_ALLOWLIST:-} volumes: - /var/run/dstack.sock:/var/run/dstack.sock:ro - group-prefs-data:/data + - pacto-data:/var/run/pacto depends_on: - signal-api - whisper-api restart: unless-stopped + # Pacto messaging daemon (https://github.com/covenant-gov/pacto-bot-api). + # Phala cannot bind-mount local files, so build a derived linux/amd64 image + # that bakes in /etc/pacto/pacto-bot-api.toml (see docker/Dockerfile.pacto + # and docker/pacto-bot-api.toml.example), push it, set PACTO_BOT_API_IMAGE + # and the PACTO_BOT_NSEC encrypted secret, then uncomment: + # + # pacto-bot-api: + # image: ${PACTO_BOT_API_IMAGE} + # container_name: pacto-bot-api + # environment: + # - PACTO_BOT_NSEC=${PACTO_BOT_NSEC} + # - RUST_LOG=${LOG_LEVEL:-info} + # volumes: + # - pacto-data:/var/lib/pacto-bot-api + # restart: unless-stopped + signal-registration-proxy: image: ${SIGNAL_PROXY_IMAGE} container_name: signal-registration-proxy @@ -107,3 +137,5 @@ volumes: driver: local whisper-models: driver: local + pacto-data: + driver: local