From e5407029e56fbd813bfea48ca6951dfd1d8e473f Mon Sep 17 00:00:00 2001 From: Ron Turetzky Date: Mon, 6 Jul 2026 21:38:33 -0400 Subject: [PATCH 1/5] =?UTF-8?q?feat:=20integrate=20covenant-gov/pacto-bot-?= =?UTF-8?q?api=20for=20!pact=20Signal=E2=86=92Pacto=20messaging?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a new `pacto-client` crate speaking JSON-RPC 2.0 over the pacto-bot-api Unix socket, a `!pact ` command in signal-bot, and opt-in Docker service wiring (profile `pacto`) for both local and Phala deployments. --- .env.example | 10 + .gitignore | 3 + CLAUDE.md | 74 +++++++ Cargo.lock | 13 ++ Cargo.toml | 1 + README.md | 1 + crates/pacto-client/Cargo.toml | 14 ++ crates/pacto-client/src/client.rs | 204 +++++++++++++++++++ crates/pacto-client/src/error.rs | 21 ++ crates/pacto-client/src/lib.rs | 14 ++ crates/pacto-client/src/types.rs | 19 ++ crates/pacto-client/tests/client_test.rs | 237 +++++++++++++++++++++++ crates/signal-bot/Cargo.toml | 1 + crates/signal-bot/src/commands/mod.rs | 2 + crates/signal-bot/src/commands/pact.rs | 157 +++++++++++++++ crates/signal-bot/src/config.rs | 51 +++++ crates/signal-bot/src/error.rs | 3 + crates/signal-bot/src/main.rs | 37 ++++ docker/Dockerfile.pacto | 44 +++++ docker/docker-compose.yaml | 29 +++ docker/pacto-bot-api.toml.example | 24 +++ docker/phala-compose.yaml | 25 +++ 22 files changed, 984 insertions(+) create mode 100644 crates/pacto-client/Cargo.toml create mode 100644 crates/pacto-client/src/client.rs create mode 100644 crates/pacto-client/src/error.rs create mode 100644 crates/pacto-client/src/lib.rs create mode 100644 crates/pacto-client/src/types.rs create mode 100644 crates/pacto-client/tests/client_test.rs create mode 100644 crates/signal-bot/src/commands/pact.rs create mode 100644 docker/Dockerfile.pacto create mode 100644 docker/pacto-bot-api.toml.example diff --git a/.env.example b/.env.example index 5c8577c..7ce191a 100644 --- a/.env.example +++ b/.env.example @@ -38,6 +38,16 @@ 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 the !pact command (requires a running pacto-bot-api +# daemon; see docker/pacto-bot-api.toml.example) +PACTO__ENABLED=false +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 + # 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..2d3855b 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,79 @@ crates/tools/ The tool system uses OpenAI-compatible function calling schema, which NEAR AI supports. +## Pacto Messaging (!pact) + +The bot can send encrypted DMs into [Pacto](https://github.com/covenant-gov/pacto-app) +via a co-located [pacto-bot-api](https://github.com/covenant-gov/pacto-bot-api) +daemon. Disabled by default. + +### Architecture + +``` +[TEE Boundary] ++--------------------------------------------------------------------+ +| Signal user --> signal-bot --!pact--> pacto-client | +| | | +| [Unix socket, 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. On first use it registers as an outbound-only +handler (`handler.register` with the `SendMessages` capability and no event +subscriptions), then publishes DMs with `agent.send_dm`. If the daemon +restarts, the next call reconnects and re-registers transparently. + +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 + +| 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 | + +### Configuration + +| Variable | Default | Description | +|----------|---------|-------------| +| `PACTO__ENABLED` | `false` | Master switch for the `!pact` command | +| `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__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..7c45203 100644 --- a/README.md +++ b/README.md @@ -36,6 +36,7 @@ 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) | 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/client.rs b/crates/pacto-client/src/client.rs new file mode 100644 index 0000000..0fb363e --- /dev/null +++ b/crates/pacto-client/src/client.rs @@ -0,0 +1,204 @@ +//! 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), + Err(e @ (PactoError::Io(_) | PactoError::Timeout(_))) => { + warn!("Pacto daemon connection stale ({e}), reconnecting"); + *guard = Some(self.connect_and_register().await?); + } + 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..b036ca3 --- /dev/null +++ b/crates/pacto-client/src/lib.rs @@ -0,0 +1,14 @@ +//! 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 client; +mod error; +mod types; + +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/client_test.rs b/crates/pacto-client/tests/client_test.rs new file mode 100644 index 0000000..969cb04 --- /dev/null +++ b/crates/pacto-client/tests/client_test.rs @@ -0,0 +1,237 @@ +//! 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 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/mod.rs b/crates/signal-bot/src/commands/mod.rs index d51dbf6..287c9dc 100644 --- a/crates/signal-bot/src/commands/mod.rs +++ b/crates/signal-bot/src/commands/mod.rs @@ -9,6 +9,7 @@ mod help; mod manual_transcribe; mod menu_locale; mod models; +mod pact; mod privacy; mod set_language; mod transcribe; @@ -28,6 +29,7 @@ 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 set_language::SetLanguageHandler; pub use transcribe::TranscribeHandler; 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/config.rs b/crates/signal-bot/src/config.rs index 19c6887..f01b0da 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)] @@ -206,6 +210,29 @@ pub struct GroupPreferencesConfig { pub storage_path: String, } +#[derive(Debug, Clone, Deserialize)] +pub struct PactoConfig { + /// Master switch for the `!pact` command (requires a pacto-bot-api daemon) + #[serde(default)] + pub 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, + + /// 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 { @@ -313,6 +340,18 @@ impl Default for GroupPreferencesConfig { } } +impl Default for PactoConfig { + fn default() -> Self { + Self { + enabled: false, + socket_path: default_pacto_socket(), + bot_id: default_pacto_bot_id(), + default_recipient: None, + timeout: default_pacto_timeout(), + } + } +} + // Default value functions fn default_signal_service() -> String { "http://signal-api:8080".into() @@ -443,6 +482,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/main.rs b/crates/signal-bot/src/main.rs index b9f5e41..33d4900 100644 --- a/crates/signal-bot/src/main.rs +++ b/crates/signal-bot/src/main.rs @@ -10,6 +10,7 @@ use anyhow::Context; use conversation_store::ConversationStore; use dstack_client::DstackClient; use near_ai_client::NearAiClient; +use pacto_client::PactoClient; use signal_client::{MessageReceiver, SignalClient}; use whisper_client::WhisperClient; use std::path::PathBuf; @@ -162,6 +163,31 @@ 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 + }; + let whisper_client = if config.whisper.enabled { let whisper = Arc::new( WhisperClient::new( @@ -300,6 +326,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/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..8efd099 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -47,9 +47,15 @@ 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} + - PACTO__SOCKET_PATH=/var/run/pacto/pacto-bot-api.sock + - PACTO__BOT_ID=${PACTO_BOT_ID:-sigstack} + - PACTO__DEFAULT_RECIPIENT=${PACTO_DEFAULT_RECIPIENT:-} 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 +85,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 +148,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..0879126 100644 --- a/docker/phala-compose.yaml +++ b/docker/phala-compose.yaml @@ -69,14 +69,37 @@ 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} + - PACTO__SOCKET_PATH=/var/run/pacto/pacto-bot-api.sock + - PACTO__BOT_ID=${PACTO_BOT_ID:-sigstack} + - PACTO__DEFAULT_RECIPIENT=${PACTO_DEFAULT_RECIPIENT:-} 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 +130,5 @@ volumes: driver: local whisper-models: driver: local + pacto-data: + driver: local From 0084a401de54eb52ba01571aa01fcbcb5121416f Mon Sep 17 00:00:00 2001 From: Ron Turetzky Date: Tue, 7 Jul 2026 05:44:31 -0400 Subject: [PATCH 2/5] fix(pacto-client): don't resend non-idempotent DMs after a timeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit call_with_retry previously reconnected and retried on both Io and Timeout errors. A timeout is ambiguous — the daemon may already have published the gift-wrap for agent.send_dm — so retrying could send the DM twice. Only a connection drop (Io/EOF) guarantees the request never completed and is safe to retry; on timeout, tear down the connection and surface the error instead. Adds a test asserting a timeout returns PactoError::Timeout and does not resend (the mock daemon confirms it never receives a second frame). --- crates/pacto-client/src/client.rs | 15 ++++++++++-- crates/pacto-client/tests/client_test.rs | 29 ++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/crates/pacto-client/src/client.rs b/crates/pacto-client/src/client.rs index 0fb363e..5fd13ca 100644 --- a/crates/pacto-client/src/client.rs +++ b/crates/pacto-client/src/client.rs @@ -82,10 +82,21 @@ impl PactoClient { } else if let Some(conn) = guard.as_mut() { match self.call(conn, method, params.clone()).await { Ok(result) => return Ok(result), - Err(e @ (PactoError::Io(_) | PactoError::Timeout(_))) => { - warn!("Pacto daemon connection stale ({e}), reconnecting"); + // 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), } } diff --git a/crates/pacto-client/tests/client_test.rs b/crates/pacto-client/tests/client_test.rs index 969cb04..2b3fc90 100644 --- a/crates/pacto-client/tests/client_test.rs +++ b/crates/pacto-client/tests/client_test.rs @@ -198,6 +198,35 @@ async fn reconnects_after_daemon_restart() { 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(); From 0f6314ab39f9b52f01da527743980d1d7f4b465a Mon Sep 17 00:00:00 2001 From: Ron Turetzky Date: Tue, 7 Jul 2026 15:57:32 -0400 Subject: [PATCH 3/5] =?UTF-8?q?feat:=20two-way=20Pacto=20bridge=20?= =?UTF-8?q?=E2=80=94=20DM=20parity=20for=20Pacto=20users?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds an inbound Pacto DM agent so Pacto users get the same DM experience Signal users get (AI chat with tools, !verify/!clear/!models/!help/!privacy/ !list-langs, AI-driven translation), not just outbound !pact. - pacto-client: new full-duplex PactoAgent that registers for dm_received (ReadMessages+SendMessages), streams inbound DMs, replies via handler.response{reply} (daemon auto-addresses the sender), and auto-reconnects. Existing single-flight PactoClient unchanged. - signal-bot: ProgressSink trait extracts the one Signal-coupled progress ping from ChatHandler (impl for Signal reply + Pacto send_dm), so the same chat/command handlers drive both transports via a synthetic BotMessage. - pacto_agent module dispatches inbound DMs through the reused handlers; Pacto-accurate !help/!privacy. Gated by PACTO__AGENT_ENABLED (default true). - Tests: PactoAgent register/receive/reply + reconnect; dispatch routing. - Docs: two-way architecture and the DM-only parity ceiling (groups/voice are blocked upstream — the daemon delivers only dm_received, no inbound MLS group messages or audio attachments). --- .env.example | 6 +- CLAUDE.md | 84 ++++-- README.md | 6 + crates/pacto-client/src/agent.rs | 235 +++++++++++++++ crates/pacto-client/src/lib.rs | 2 + crates/pacto-client/tests/agent_test.rs | 181 ++++++++++++ crates/signal-bot/src/commands/chat.rs | 29 +- crates/signal-bot/src/commands/mod.rs | 2 + crates/signal-bot/src/commands/progress.rs | 26 ++ crates/signal-bot/src/config.rs | 10 +- crates/signal-bot/src/lib.rs | 1 + crates/signal-bot/src/main.rs | 34 ++- crates/signal-bot/src/pacto_agent.rs | 317 +++++++++++++++++++++ docker/docker-compose.yaml | 2 + docker/phala-compose.yaml | 2 + 15 files changed, 895 insertions(+), 42 deletions(-) create mode 100644 crates/pacto-client/src/agent.rs create mode 100644 crates/pacto-client/tests/agent_test.rs create mode 100644 crates/signal-bot/src/commands/progress.rs create mode 100644 crates/signal-bot/src/pacto_agent.rs diff --git a/.env.example b/.env.example index 7ce191a..a94b791 100644 --- a/.env.example +++ b/.env.example @@ -39,9 +39,13 @@ GROUP_PREFERENCES__PERSIST=true GROUP_PREFERENCES__STORAGE_PATH=/data/group_prefs.enc # Pacto Messaging (via pacto-bot-api daemon) -# Set to true to enable the !pact command (requires a running pacto-bot-api +# 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 diff --git a/CLAUDE.md b/CLAUDE.md index 2d3855b..8f5df9f 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -493,34 +493,68 @@ crates/tools/ The tool system uses OpenAI-compatible function calling schema, which NEAR AI supports. -## Pacto Messaging (!pact) +## Pacto Messaging (!pact + two-way DM agent) -The bot can send encrypted DMs into [Pacto](https://github.com/covenant-gov/pacto-app) -via a co-located [pacto-bot-api](https://github.com/covenant-gov/pacto-bot-api) +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. + +### 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--> pacto-client | -| | | -| [Unix socket, 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. On first use it registers as an outbound-only -handler (`handler.register` with the `SendMessages` capability and no event -subscriptions), then publishes DMs with `agent.send_dm`. If the daemon -restarts, the next call reconnects and re-registers transparently. ++----------------------------------------------------------------------+ +| 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 @@ -528,6 +562,8 @@ architecture. ### Commands +Outbound (Signal user → Pacto): + | Command | Description | |---------|-------------| | `!pact ` | DM a Pacto user (npub or 64-char hex pubkey) | @@ -535,11 +571,15 @@ architecture. | `!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`. + ### Configuration | Variable | Default | Description | |----------|---------|-------------| -| `PACTO__ENABLED` | `false` | Master switch for the `!pact` command | +| `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 | diff --git a/README.md b/README.md index 7c45203..4b25ad5 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,12 @@ This project implements a Signal bot that runs inside a Dstack-powered TEE (Inte | `!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`). +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. ## Group Chat Support diff --git a/crates/pacto-client/src/agent.rs b/crates/pacto-client/src/agent.rs new file mode 100644 index 0000000..28d33d1 --- /dev/null +++ b/crates/pacto-client/src/agent.rs @@ -0,0 +1,235 @@ +//! 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; + } + }; + + // Registration ack / other id-correlated responses. + if let Some(err) = msg.get("error") { + warn!(error = %err, "Pacto daemon returned an error frame"); + continue; + } + 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/lib.rs b/crates/pacto-client/src/lib.rs index b036ca3..023624f 100644 --- a/crates/pacto-client/src/lib.rs +++ b/crates/pacto-client/src/lib.rs @@ -5,10 +5,12 @@ //! 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/tests/agent_test.rs b/crates/pacto-client/tests/agent_test.rs new file mode 100644 index 0000000..edfa57a --- /dev/null +++ b/crates/pacto-client/tests/agent_test.rs @@ -0,0 +1,181 @@ +//! 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, UnixStream}; + +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() +} + +async fn write_frame(stream: &mut UnixStream, 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 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(); +} 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 287c9dc..b6f6516 100644 --- a/crates/signal-bot/src/commands/mod.rs +++ b/crates/signal-bot/src/commands/mod.rs @@ -11,6 +11,7 @@ mod menu_locale; mod models; mod pact; mod privacy; +pub mod progress; mod set_language; mod transcribe; mod translate; @@ -31,6 +32,7 @@ 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 transcribe::TranscribeHandler; pub use translate::TranslateHandler; 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/config.rs b/crates/signal-bot/src/config.rs index f01b0da..5a1f23a 100644 --- a/crates/signal-bot/src/config.rs +++ b/crates/signal-bot/src/config.rs @@ -212,10 +212,17 @@ pub struct GroupPreferencesConfig { #[derive(Debug, Clone, Deserialize)] pub struct PactoConfig { - /// Master switch for the `!pact` command (requires a pacto-bot-api daemon) + /// 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, @@ -344,6 +351,7 @@ 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, 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 33d4900..8d5b1de 100644 --- a/crates/signal-bot/src/main.rs +++ b/crates/signal-bot/src/main.rs @@ -10,7 +10,8 @@ use anyhow::Context; use conversation_store::ConversationStore; use dstack_client::DstackClient; use near_ai_client::NearAiClient; -use pacto_client::PactoClient; +use pacto_client::{PactoAgent, PactoClient}; +use signal_bot::pacto_agent; use signal_client::{MessageReceiver, SignalClient}; use whisper_client::WhisperClient; use std::path::PathBuf; @@ -188,6 +189,37 @@ async fn main() -> AppResult<()> { 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), + ); + 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(), + }, + ); + 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( diff --git a/crates/signal-bot/src/pacto_agent.rs b/crates/signal-bot/src/pacto_agent.rs new file mode 100644 index 0000000..c7939bb --- /dev/null +++ b/crates/signal-bot/src/pacto_agent.rs @@ -0,0 +1,317 @@ +//! 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, 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; +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, +} + +/// 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. + vec![ + Box::new(VerifyHandler::new(deps.dstack)), + Box::new(ClearHandler::new(deps.conversations)), + Box::new(ModelsHandler::new(deps.near_ai)), + Box::new(PactoHelpHandler), + Box::new(PactoPrivacyHandler), + Box::new(TranslateLangsHandler::new()), + Box::new(chat), + ] +} + +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; + +#[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 { + Ok(PACTO_HELP.to_string()) + } +} + +/// 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, + }) + } + + 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/docker-compose.yaml b/docker/docker-compose.yaml index 8efd099..85d6e4a 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -49,6 +49,8 @@ services: - 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:-} diff --git a/docker/phala-compose.yaml b/docker/phala-compose.yaml index 0879126..90f651d 100644 --- a/docker/phala-compose.yaml +++ b/docker/phala-compose.yaml @@ -72,6 +72,8 @@ services: # 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:-} From dc4ebc153acfa87a227c268a662c9919d0468c02 Mon Sep 17 00:00:00 2001 From: Ron Turetzky Date: Tue, 7 Jul 2026 16:16:40 -0400 Subject: [PATCH 4/5] fix(pacto-client): reconnect when handler.register is rejected Previously the agent's read loop logged a JSON-RPC error frame and continued, leaving the connection stuck (no events will arrive after a failed registration) with the write half still published, so reply() would fail with confusing protocol errors instead of reconnecting. Treat an error frame as a connection failure so the supervisor clears the writer and retries after the backoff. Adds a test rejecting the first register and asserting reconnect. --- crates/pacto-client/src/agent.rs | 10 +++- crates/pacto-client/tests/agent_test.rs | 73 ++++++++++++++++++++++--- 2 files changed, 72 insertions(+), 11 deletions(-) diff --git a/crates/pacto-client/src/agent.rs b/crates/pacto-client/src/agent.rs index 28d33d1..56289e7 100644 --- a/crates/pacto-client/src/agent.rs +++ b/crates/pacto-client/src/agent.rs @@ -190,10 +190,14 @@ async fn pump_events(read_half: OwnedReadHalf, bot_id: &str, tx: &mpsc::Sender { diff --git a/crates/pacto-client/tests/agent_test.rs b/crates/pacto-client/tests/agent_test.rs index edfa57a..0c2dc14 100644 --- a/crates/pacto-client/tests/agent_test.rs +++ b/crates/pacto-client/tests/agent_test.rs @@ -5,7 +5,7 @@ use serde_json::{Value, json}; use std::path::PathBuf; use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::{UnixListener, UnixStream}; +use tokio::net::UnixListener; async fn read_frame(reader: &mut R) -> Value { let mut line = String::new(); @@ -13,13 +13,6 @@ async fn read_frame(reader: &mut R) -> Value { serde_json::from_str(line.trim()).unwrap() } -async fn write_frame(stream: &mut UnixStream, 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") } @@ -179,3 +172,67 @@ async fn reconnects_and_reregisters_after_daemon_restart() { 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(); +} From 62944ce2704bc3d0b5de637975ef1d410ed184d8 Mon Sep 17 00:00:00 2001 From: Ron Turetzky Date: Wed, 8 Jul 2026 16:15:35 -0400 Subject: [PATCH 5/5] =?UTF-8?q?feat:=20!signal=20=E2=80=94=20let=20Pacto?= =?UTF-8?q?=20users=20DM=20Signal=20users=20through=20the=20bot?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mirror of !pact for the fourth direction (Pacto user -> Signal user). A Pacto user DMs the bot "!signal <+number> " and it's relayed to the Signal recipient, prefixed with the sender's npub so they can reply via !pact (no stateful session mapping needed). Because this can relay to arbitrary Signal numbers (an open relay if unguarded), it is off by default and allowlist-gated: - PACTO__SIGNAL_RELAY_ENABLED (default false) - PACTO__SIGNAL_RELAY_ALLOWLIST (empty = deny all, "*" = any) - requires SIGNAL__PHONE_NUMBER (the bot's own number to send from) The !signal handler lives only in the Pacto inbound dispatch and is added to the handler set + help only when enabled. Adds SignalConfig.phone_number, unit tests for E.164 validation, allowlist gating, and the no-send rejection paths. --- .env.example | 8 + CLAUDE.md | 22 +++ README.md | 2 + crates/signal-bot/src/commands/mod.rs | 2 + .../signal-bot/src/commands/signal_relay.rs | 184 ++++++++++++++++++ crates/signal-bot/src/config.rs | 20 ++ crates/signal-bot/src/main.rs | 39 ++++ crates/signal-bot/src/pacto_agent.rs | 51 ++++- docker/docker-compose.yaml | 3 + docker/phala-compose.yaml | 5 + 10 files changed, 327 insertions(+), 9 deletions(-) create mode 100644 crates/signal-bot/src/commands/signal_relay.rs diff --git a/.env.example b/.env.example index a94b791..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 @@ -51,6 +52,13 @@ 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 diff --git a/CLAUDE.md b/CLAUDE.md index 8f5df9f..eb022ed 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -503,6 +503,8 @@ daemon. Disabled by default. - **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 @@ -574,6 +576,23 @@ Outbound (Signal user → Pacto): 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 | @@ -583,6 +602,9 @@ Inbound (Pacto user → bot): free-text → AI chat; `!verify`, `!clear`, `!mode | `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 diff --git a/README.md b/README.md index 4b25ad5..9493b42 100644 --- a/README.md +++ b/README.md @@ -41,6 +41,8 @@ This project implements a Signal bot that runs inside a Dstack-powered TEE (Inte 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". diff --git a/crates/signal-bot/src/commands/mod.rs b/crates/signal-bot/src/commands/mod.rs index b6f6516..ff6689a 100644 --- a/crates/signal-bot/src/commands/mod.rs +++ b/crates/signal-bot/src/commands/mod.rs @@ -13,6 +13,7 @@ mod pact; mod privacy; pub mod progress; mod set_language; +mod signal_relay; mod transcribe; mod translate; mod translate_all; @@ -34,6 +35,7 @@ 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/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 5a1f23a..7c601bb 100644 --- a/crates/signal-bot/src/config.rs +++ b/crates/signal-bot/src/config.rs @@ -57,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, @@ -235,6 +240,18 @@ pub struct PactoConfig { #[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, @@ -245,6 +262,7 @@ impl Default for SignalConfig { fn default() -> Self { Self { service_url: default_signal_service(), + phone_number: None, poll_interval: default_poll_interval(), } } @@ -355,6 +373,8 @@ impl Default for PactoConfig { 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(), } } diff --git a/crates/signal-bot/src/main.rs b/crates/signal-bot/src/main.rs index 8d5b1de..aeedda6 100644 --- a/crates/signal-bot/src/main.rs +++ b/crates/signal-bot/src/main.rs @@ -199,6 +199,44 @@ async fn main() -> AppResult<()> { 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, @@ -212,6 +250,7 @@ async fn main() -> AppResult<()> { 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"); diff --git a/crates/signal-bot/src/pacto_agent.rs b/crates/signal-bot/src/pacto_agent.rs index c7939bb..bff092b 100644 --- a/crates/signal-bot/src/pacto_agent.rs +++ b/crates/signal-bot/src/pacto_agent.rs @@ -15,15 +15,15 @@ //! daemon gains inbound group/attachment delivery. use crate::commands::{ - ChatHandler, ClearHandler, CommandHandler, ModelsHandler, ProgressSink, TranslateLangsHandler, - VerifyHandler, + 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; +use signal_client::{BotMessage, SignalClient}; use std::sync::Arc; use tokio::sync::mpsc; use tools::ToolRegistry; @@ -41,6 +41,17 @@ pub struct PactoAgentDeps { 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 @@ -68,15 +79,29 @@ fn build_handlers(deps: PactoAgentDeps) -> Vec> { // 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. - vec![ + 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), + Box::new(PactoHelpHandler { relay_enabled }), Box::new(PactoPrivacyHandler), Box::new(TranslateLangsHandler::new()), - Box::new(chat), - ] + ]; + + // 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( @@ -158,7 +183,10 @@ impl ProgressSink for PactoProgress { } /// Pacto-accurate `!help` (omits Signal-only voice/group features). -struct PactoHelpHandler; +struct PactoHelpHandler { + /// Whether the `!signal` relay is active (so help only lists it when usable). + relay_enabled: bool, +} #[async_trait] impl CommandHandler for PactoHelpHandler { @@ -171,7 +199,11 @@ impl CommandHandler for PactoHelpHandler { } async fn execute(&self, _message: &BotMessage) -> crate::error::AppResult { - Ok(PACTO_HELP.to_string()) + let mut help = PACTO_HELP.to_string(); + if self.relay_enabled { + help.push_str("\n- !signal <+number> — DM a Signal user"); + } + Ok(help) } } @@ -242,6 +274,7 @@ mod tests { max_tool_calls: 5, signal_username: None, github_repo: None, + signal_relay: None, }) } diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index 85d6e4a..290b9ff 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -54,6 +54,9 @@ services: - 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 diff --git a/docker/phala-compose.yaml b/docker/phala-compose.yaml index 90f651d..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} @@ -77,6 +79,9 @@ services: - 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