diff --git a/Dockerfile b/Dockerfile index 171f740..d09523a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -40,7 +40,14 @@ COPY crates/ crates/ RUN cargo build --release -# Stage 4: Runtime +# Stage 4: Build marmotd (optional MLS messaging sidecar) +FROM chef AS marmotd-builder +ARG MARMOTD_VERSION=0.5.1 +RUN git clone --depth 1 --branch marmotd-v${MARMOTD_VERSION} https://github.com/sledtools/pika.git /marmotd-src +WORKDIR /marmotd-src +RUN cargo build -p marmotd --release + +# Stage 5: Runtime FROM docker.io/debian:bookworm-slim # Install runtime dependencies and comprehensive CLI toolset @@ -140,6 +147,9 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ toml \ rich +# Copy marmotd binary (MLS messaging sidecar, built from pika source) +COPY --from=marmotd-builder /marmotd-src/target/release/marmotd /usr/local/bin/marmotd + # Create non-root user RUN useradd -m -u 1000 sage @@ -151,8 +161,8 @@ COPY --from=builder /app/target/release/sage /app/sage # Copy migrations for diesel COPY --from=builder /app/crates/sage-core/migrations /app/migrations -# Create workspace directory -RUN mkdir -p /workspace && chown sage:sage /workspace +# Create workspace and marmot state directories +RUN mkdir -p /workspace /data/marmot-state && chown sage:sage /workspace /data/marmot-state # Run as non-root user USER sage diff --git a/README.md b/README.md index 74d652d..3cd5c81 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,7 @@ Sage is an AI assistant that prioritizes **privacy** and **data sovereignty**. It's designed to be a trusted companion that remembers your conversations, learns about you over time, and can take actions on your behalf - all while keeping your data under your control. **Key Features:** -- **End-to-end encrypted messaging** via Signal +- **End-to-end encrypted messaging** via Signal or [Pika](https://github.com/sledtools/pika) (MLS over Nostr) - **Image understanding** - send photos and Sage can see and describe them - **Long-term memory** that persists across conversations - **Confidential compute** - LLM inference runs in a TEE (Trusted Execution Environment) @@ -22,7 +22,7 @@ Most AI assistants are stateless - they forget everything after each conversatio - Your conversations stay on **your** PostgreSQL instance - LLM inference happens in **confidential compute** (Maple/TEE) - the inference provider can't see your prompts -- Communication happens over **Signal's E2E encryption** +- Communication happens over **Signal's E2E encryption** or **MLS-encrypted Nostr** via [Pika](https://github.com/sledtools/pika) - The agent runs in **your container** on your infrastructure ## Technical Highlights @@ -250,7 +250,7 @@ DSRs parses this back into a typed `AgentResponse` struct that Sage uses to exec | LLM | **Kimi K2** (thinking variant) | Strong tool use, 128k context | | Inference | **Maple** | TEE-based confidential compute | | Embeddings | **nomic-embed-text** | Via Maple | -| Messaging | **Signal** (signal-cli) | E2E encrypted, works on mobile | +| Messaging | **Signal** (signal-cli) or **Pika** (marmotd) | E2E encrypted; Signal for mobile, Pika for decentralized Nostr | | Database | **PostgreSQL + pgvector** | Structured data + vector search | | Framework | **DSRs** (dspy-rs) | Typed signatures, BAML parsing | @@ -266,12 +266,41 @@ DSRs parses this back into a typed `AgentResponse` struct that Sage uses to exec | `schedule_task` | Reminders (cron or one-off) | | `set_preference` | User preferences (timezone, etc.) | +## Messaging Providers + +Sage supports two messaging backends. Set the `MESSENGER` environment variable to choose (`signal` is the default). + +### Signal (Default) + +Uses [signal-cli](https://github.com/AsamK/signal-cli) for E2E encrypted messaging over the Signal protocol. Requires a registered phone number and runs signal-cli as a sidecar container. + +```bash +MESSENGER=signal +SIGNAL_PHONE_NUMBER=+1234567890 +SIGNAL_ALLOWED_USERS=* # Or comma-separated Signal UUIDs +``` + +### Marmot / Pika (Decentralized) + +Uses [marmotd](https://github.com/sledtools/pika) for MLS-encrypted messaging over Nostr relays. No phone number required — identity is a Nostr keypair. Message Sage from the [Pika](https://github.com/sledtools/pika) app. + +```bash +MESSENGER=marmot +MARMOT_RELAYS=wss://relay.damus.io,wss://nos.lol,wss://relay.primal.net +MARMOT_ALLOWED_PUBKEYS=npub1... # Or * for all +MARMOT_AUTO_ACCEPT_WELCOMES=true +``` + +On first startup, marmotd generates a Nostr keypair and prints its `npub` in the logs. Use this npub to start a conversation from Pika. The keypair and MLS state persist in a Docker volume (`sage-marmot-state`). + +marmotd is built from source during `docker build` (included in the Dockerfile). + ## Quick Start ### Prerequisites - [Podman](https://podman.io/) or Docker -- signal-cli registered with a phone number +- signal-cli registered with a phone number (if using Signal) or [Pika](https://github.com/sledtools/pika) app (if using Marmot) - Maple API access (or compatible OpenAI endpoint) ### Option 1: Docker (Recommended) @@ -336,12 +365,22 @@ just start # Start all services MAPLE_API_URL=https://your-maple-endpoint MAPLE_API_KEY=your-api-key MAPLE_MODEL=maple/kimi-k2-5 + +# Messenger (choose one) +MESSENGER=signal # "signal" (default) or "marmot" + +# Signal config (when MESSENGER=signal) SIGNAL_PHONE_NUMBER=+1234567890 +SIGNAL_ALLOWED_USERS=* # Or comma-separated UUIDs + +# Marmot config (when MESSENGER=marmot) +MARMOT_RELAYS=wss://relay.damus.io,wss://nos.lol,wss://relay.primal.net +MARMOT_ALLOWED_PUBKEYS=npub1... # Or * for all +MARMOT_AUTO_ACCEPT_WELCOMES=true # Optional BRAVE_API_KEY=your-brave-key # For web search MAPLE_VISION_MODEL=maple/kimi-k2-5 # For image understanding (defaults to MAPLE_MODEL) -SIGNAL_ALLOWED_USERS=* # Or comma-separated UUIDs ``` ## Architecture @@ -359,6 +398,11 @@ SIGNAL_ALLOWED_USERS=* # Or comma-separated UUIDs │ │ Manager │ │ System │ │ │ │ │ └─────────────┘ └─────────────┘ └────────────┘ │ └─────────────────────────┬───────────────────────────┘ + ▲ + │ JSONL/stdio +┌─────────────────┐ MLS/Nostr ┌────────┴────────┐ +│ Pika App │◄──────────────►│ marmotd │ +└─────────────────┘ (encrypted) └─────────────────┘ │ ┌─────────────────┼─────────────────┐ ▼ ▼ ▼ @@ -372,7 +416,7 @@ SIGNAL_ALLOWED_USERS=* # Or comma-separated UUIDs | Layer | Protection | |-------|------------| -| **Transport** | Signal E2E encryption | +| **Transport** | Signal E2E encryption or MLS-encrypted Nostr (via Pika) | | **Inference** | Maple TEE (confidential compute) | | **Embeddings** | Maple TEE (memory vectors generated privately) | | **Storage** | Local PostgreSQL (your machine) | @@ -444,6 +488,7 @@ Current categories: first-time users, casual chat, web search, memory storage, t - [Letta](https://github.com/letta-ai/letta) - Memory management inspiration - [DSRs](https://github.com/krypticmouse/DSRs) - DSPy in Rust - [signal-cli](https://github.com/AsamK/signal-cli) - Signal CLI interface +- [Pika](https://github.com/sledtools/pika) - MLS-encrypted messaging over Nostr - [Maple](https://www.trymaple.ai/) - Confidential compute LLM inference ## License diff --git a/crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/down.sql b/crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/down.sql new file mode 100644 index 0000000..874274a --- /dev/null +++ b/crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/down.sql @@ -0,0 +1 @@ +ALTER TABLE chat_contexts DROP COLUMN reply_context; diff --git a/crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/up.sql b/crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/up.sql new file mode 100644 index 0000000..a153ecc --- /dev/null +++ b/crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/up.sql @@ -0,0 +1 @@ +ALTER TABLE chat_contexts ADD COLUMN reply_context TEXT; diff --git a/crates/sage-core/src/agent_manager.rs b/crates/sage-core/src/agent_manager.rs index b6003a0..92c43bd 100644 --- a/crates/sage-core/src/agent_manager.rs +++ b/crates/sage-core/src/agent_manager.rs @@ -34,6 +34,7 @@ pub struct ChatContext { pub context_type: String, pub display_name: Option, pub created_at: chrono::DateTime, + pub reply_context: Option, } /// New chat context for insertion @@ -217,6 +218,7 @@ impl AgentManager { context_type: context_type.as_str().to_string(), display_name: display_name.map(|s| s.to_string()), created_at: Utc::now(), + reply_context: None, }) } @@ -322,6 +324,43 @@ impl AgentManager { Ok(result) } + /// Update the reply_context for a given identifier (e.g. Marmot group_id for a pubkey) + pub fn update_reply_context(&self, signal_identifier: &str, reply_ctx: &str) -> Result<()> { + let mut conn = self + .db_conn + .lock() + .map_err(|_| anyhow::anyhow!("Failed to acquire database lock"))?; + + diesel::update( + chat_contexts::table.filter(chat_contexts::signal_identifier.eq(signal_identifier)), + ) + .set(chat_contexts::reply_context.eq(Some(reply_ctx))) + .execute(&mut *conn)?; + + Ok(()) + } + + /// Load all reply_context mappings (identifier -> reply_context) for route restoration + pub fn load_reply_contexts(&self) -> Result> { + let mut conn = self + .db_conn + .lock() + .map_err(|_| anyhow::anyhow!("Failed to acquire database lock"))?; + + let results: Vec<(String, Option)> = chat_contexts::table + .select(( + chat_contexts::signal_identifier, + chat_contexts::reply_context, + )) + .filter(chat_contexts::reply_context.is_not_null()) + .load(&mut *conn)?; + + Ok(results + .into_iter() + .filter_map(|(id, ctx)| ctx.map(|c| (id, c))) + .collect()) + } + /// Get all chat contexts #[allow(dead_code)] pub fn list_contexts(&self) -> Result> { diff --git a/crates/sage-core/src/config.rs b/crates/sage-core/src/config.rs index 94131cb..77a0b25 100644 --- a/crates/sage-core/src/config.rs +++ b/crates/sage-core/src/config.rs @@ -1,5 +1,13 @@ use anyhow::{Context, Result}; +use crate::marmot::MarmotConfig; + +#[derive(Debug, Clone, PartialEq)] +pub enum MessengerType { + Signal, + Marmot, +} + #[derive(Debug, Clone)] #[allow(dead_code)] pub struct Config { @@ -11,12 +19,23 @@ pub struct Config { pub database_url: String, + /// Which messaging provider to use + pub messenger_type: MessengerType, + + // Signal-specific config pub signal_phone_number: Option, pub signal_allowed_users: Vec, /// If set, connect to signal-cli daemon via TCP instead of spawning subprocess pub signal_cli_host: Option, pub signal_cli_port: u16, + // Marmot-specific config + pub marmot_binary: String, + pub marmot_relays: Vec, + pub marmot_state_dir: String, + pub marmot_allowed_pubkeys: Vec, + pub marmot_auto_accept_welcomes: bool, + pub brave_api_key: Option, /// Workspace directory for shell commands and file operations @@ -40,6 +59,15 @@ impl Config { database_url: std::env::var("DATABASE_URL").context("DATABASE_URL must be set")?, + messenger_type: match std::env::var("MESSENGER") + .unwrap_or_else(|_| "signal".to_string()) + .to_lowercase() + .as_str() + { + "marmot" => MessengerType::Marmot, + _ => MessengerType::Signal, + }, + signal_phone_number: std::env::var("SIGNAL_PHONE_NUMBER").ok(), signal_allowed_users: std::env::var("SIGNAL_ALLOWED_USERS") .map(|s| s.split(',').map(|u| u.trim().to_string()).collect()) @@ -50,6 +78,36 @@ impl Config { .parse() .unwrap_or(7583), + marmot_binary: std::env::var("MARMOT_BINARY").unwrap_or_else(|_| "marmotd".to_string()), + marmot_relays: std::env::var("MARMOT_RELAYS") + .map(|s| { + s.split(',') + .map(|r| r.trim().to_string()) + .filter(|r| !r.is_empty()) + .collect() + }) + .unwrap_or_default(), + marmot_state_dir: std::env::var("MARMOT_STATE_DIR") + .unwrap_or_else(|_| "/data/marmot-state".to_string()), + marmot_allowed_pubkeys: std::env::var("MARMOT_ALLOWED_PUBKEYS") + .map(|s| { + s.split(',') + .map(|p| p.trim().to_string()) + .filter(|p| !p.is_empty()) + .map(|p| { + if p == "*" { + p + } else { + crate::marmot::normalize_pubkey(&p).unwrap_or(p) + } + }) + .collect() + }) + .unwrap_or_default(), + marmot_auto_accept_welcomes: std::env::var("MARMOT_AUTO_ACCEPT_WELCOMES") + .map(|s| s != "false" && s != "0") + .unwrap_or(true), + brave_api_key: std::env::var("BRAVE_API_KEY").ok(), workspace_path: std::env::var("SAGE_WORKSPACE") @@ -61,4 +119,21 @@ impl Config { .context("HTTP_PORT must be a valid port number")?, }) } + + pub fn marmot_config(&self) -> MarmotConfig { + MarmotConfig { + binary_path: self.marmot_binary.clone(), + relays: self.marmot_relays.clone(), + state_dir: self.marmot_state_dir.clone(), + allowed_pubkeys: self.marmot_allowed_pubkeys.clone(), + auto_accept_welcomes: self.marmot_auto_accept_welcomes, + } + } + + pub fn allowed_users(&self) -> &[String] { + match self.messenger_type { + MessengerType::Signal => &self.signal_allowed_users, + MessengerType::Marmot => &self.marmot_allowed_pubkeys, + } + } } diff --git a/crates/sage-core/src/lib.rs b/crates/sage-core/src/lib.rs index 852a451..4ff8236 100644 --- a/crates/sage-core/src/lib.rs +++ b/crates/sage-core/src/lib.rs @@ -4,7 +4,9 @@ pub mod agent_manager; pub mod config; +pub mod marmot; pub mod memory; +pub mod messenger; pub mod sage_agent; pub mod scheduler; pub mod scheduler_tools; diff --git a/crates/sage-core/src/main.rs b/crates/sage-core/src/main.rs index 6e93755..0b2145e 100644 --- a/crates/sage-core/src/main.rs +++ b/crates/sage-core/src/main.rs @@ -9,7 +9,9 @@ use uuid::Uuid; mod agent_manager; mod config; +mod marmot; mod memory; +mod messenger; mod sage_agent; mod scheduler; mod scheduler_tools; @@ -20,8 +22,10 @@ mod storage; mod vision; use agent_manager::{AgentManager, ContextType}; +use config::MessengerType; +use messenger::{IncomingMessage, Messenger}; use sage_agent::SageAgent; -use signal::{run_receive_loop, run_receive_loop_tcp, IncomingMessage, SignalClient}; +use signal::{run_receive_loop, run_receive_loop_tcp, SignalClient}; /// Health check response #[derive(Serialize)] @@ -114,82 +118,139 @@ async fn main() -> Result<()> { config.workspace_path ); - // Check if Signal is configured - let signal_phone = match &config.signal_phone_number { - Some(phone) => phone.clone(), - None => { - warn!("SIGNAL_PHONE_NUMBER not set - cannot start Signal interface"); - info!("Set SIGNAL_PHONE_NUMBER in .env to enable messaging."); - tokio::signal::ctrl_c().await?; - return Ok(()); - } - }; - // Create channel for incoming messages let (tx, mut rx) = mpsc::channel::(100); - // Start Signal client - let (signal_client, receive_handle) = if let Some(ref host) = config.signal_cli_host { - info!( - "Starting Signal interface (TCP mode: {}:{})...", - host, config.signal_cli_port - ); - - let signal_client = SignalClient::connect_tcp(&signal_phone, host, config.signal_cli_port)?; - let signal_client = Arc::new(Mutex::new(signal_client)); - - let host = host.clone(); - let port = config.signal_cli_port; - let account = signal_phone.clone(); - // Supervise the receive loop: if the TCP subscription drops, reconnect + resubscribe. - let receive_handle = tokio::spawn(async move { - let mut backoff = std::time::Duration::from_millis(250); - let backoff_max = std::time::Duration::from_secs(60); - - loop { - match run_receive_loop_tcp(&host, port, &account, tx.clone()).await { - Ok(()) => { - warn!( - "Signal TCP receive loop exited unexpectedly; restarting in {:?}", - backoff - ); - } - Err(e) => { - warn!( - "Signal TCP receive loop error; restarting in {:?}: {}", - backoff, e - ); - } + // Agent keyed by identity (Signal UUID or Marmot pubkey). + // Both messengers currently use Direct (1:1 identity = 1 agent). + // TODO: With multi-agent support, Marmot groups could each get their own + // agent thread while sharing a parent identity for cross-thread memory. + let context_type = ContextType::Direct; + + // Start messenger based on config + let (messenger, receive_handle): (Arc>, _) = match config.messenger_type { + MessengerType::Signal => { + let signal_phone = match &config.signal_phone_number { + Some(phone) => phone.clone(), + None => { + warn!("SIGNAL_PHONE_NUMBER not set - cannot start Signal interface"); + info!("Set SIGNAL_PHONE_NUMBER in .env to enable messaging."); + tokio::signal::ctrl_c().await?; + return Ok(()); } + }; + + if let Some(ref host) = config.signal_cli_host { + info!( + "Starting Signal interface (TCP mode: {}:{})...", + host, config.signal_cli_port + ); + + let signal_client = + SignalClient::connect_tcp(&signal_phone, host, config.signal_cli_port)?; + let messenger: Arc> = Arc::new(Mutex::new(signal_client)); + + let host = host.clone(); + let port = config.signal_cli_port; + let account = signal_phone.clone(); + let receive_handle = tokio::spawn(async move { + let mut backoff = std::time::Duration::from_millis(250); + let backoff_max = std::time::Duration::from_secs(60); + + loop { + match run_receive_loop_tcp(&host, port, &account, tx.clone()).await { + Ok(()) => { + warn!( + "Signal TCP receive loop exited unexpectedly; restarting in {:?}", + backoff + ); + } + Err(e) => { + warn!( + "Signal TCP receive loop error; restarting in {:?}: {}", + backoff, e + ); + } + } + + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(backoff_max); + } + }); + + (messenger, receive_handle) + } else { + info!("Starting Signal interface (subprocess mode)..."); + + let signal_client = SignalClient::spawn_subprocess(&signal_phone)?; + let reader = signal_client.take_reader()?; + let messenger: Arc> = Arc::new(Mutex::new(signal_client)); - tokio::time::sleep(backoff).await; - backoff = (backoff * 2).min(backoff_max); + let receive_handle = + tokio::spawn(async move { run_receive_loop(reader, tx).await }); + + (messenger, receive_handle) } - }); + } + MessengerType::Marmot => { + let marmot_config = config.marmot_config(); - (signal_client, receive_handle) - } else { - info!("Starting Signal interface (subprocess mode)..."); + if marmot_config.relays.is_empty() { + return Err(anyhow::anyhow!( + "MARMOT_RELAYS must be set when MESSENGER=marmot" + )); + } - let signal_client = SignalClient::spawn_subprocess(&signal_phone)?; - let reader = signal_client.take_reader()?; - let signal_client = Arc::new(Mutex::new(signal_client)); + info!("Starting Marmot interface..."); + info!(" Relays: {:?}", marmot_config.relays); + info!(" State dir: {}", marmot_config.state_dir); + + let client = marmot::new_marmot_client(&marmot_config)?; + let writer = marmot::writer_handle(&client); + let group_routes = marmot::group_routes_handle(&client); + let child = marmot::child_handle(&client); + + // Restore persisted pubkey -> group_id routes from DB + match agent_manager.load_reply_contexts() { + Ok(routes) => { + if !routes.is_empty() { + info!("Restored {} Marmot route(s) from database", routes.len()); + if let Ok(mut map) = group_routes.lock() { + for (pubkey, group_id) in routes { + map.insert(pubkey, group_id); + } + } + } + } + Err(e) => warn!("Failed to load reply contexts: {}", e), + } - let receive_handle = tokio::spawn(async move { run_receive_loop(reader, tx).await }); + let messenger: Arc> = Arc::new(Mutex::new(client)); - (signal_client, receive_handle) + // Supervisor loop: respawns marmotd on failure with exponential backoff + let receive_handle = tokio::spawn(async move { + marmot::run_marmot_receive_loop(tx, marmot_config, group_routes, writer, child) + .await + }); + + (messenger, receive_handle) + } }; // Log allowed users configuration - if config.signal_allowed_users.iter().any(|u| u == "*") { + let allowed_users = config.allowed_users(); + if allowed_users.iter().any(|u| u == "*") { info!("Allowed users: * (all users)"); - } else if config.signal_allowed_users.is_empty() { - warn!("⚠️ No SIGNAL_ALLOWED_USERS configured - Sage will respond to ANYONE!"); + } else if allowed_users.is_empty() { + warn!("No allowed users configured - Sage will respond to ANYONE!"); } else { - info!("Allowed users: {:?}", config.signal_allowed_users); + info!("Allowed users: {:?}", allowed_users); } - info!("🌿 Sage is awake and listening on Signal!"); + info!( + "Sage is awake and listening via {:?}!", + config.messenger_type + ); // Start HTTP health check server let health_port: u16 = std::env::var("HEALTH_PORT") @@ -209,48 +270,43 @@ async fn main() -> Result<()> { let mut scheduler_rx = scheduler::spawn_scheduler(scheduler_db.clone(), 30); info!("Background scheduler started (polling every 30s)"); - // Signal health check interval (every 60 minutes) - // This refreshes prekeys to prevent silent send failures - let mut signal_health_interval = tokio::time::interval(std::time::Duration::from_secs(60 * 60)); - signal_health_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); - // Skip the first immediate tick - signal_health_interval.tick().await; - info!("Signal health check scheduled (every 60 minutes)"); + // Messenger health check interval (every 60 minutes) + let mut health_interval = tokio::time::interval(std::time::Duration::from_secs(60 * 60)); + health_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + health_interval.tick().await; + info!("Messenger health check scheduled (every 60 minutes)"); // Main event loop loop { tokio::select! { - // Periodic Signal health check (refresh prekeys) - _ = signal_health_interval.tick() => { - info!("🔄 Running Signal health check..."); - let client = signal_client.lock().await; - if let Err(e) = client.refresh_account() { - warn!("Signal health check failed: {} - will retry next interval", e); + // Periodic messenger health check + _ = health_interval.tick() => { + let client = messenger.lock().await; + if let Err(e) = client.refresh() { + warn!("Messenger health check failed: {} - will retry next interval", e); } } // Handle scheduled task events Some(event) = scheduler_rx.recv() => { let task = event.task; - info!("⏰ Processing scheduled task: {} ({})", task.description, task.task_type.as_str()); + info!("Processing scheduled task: {} ({})", task.description, task.task_type.as_str()); - // Look up the signal_identifier for this agent_id let signal_identifier = match agent_manager.get_signal_identifier(task.agent_id) { Ok(Some(id)) => id, Ok(None) => { - error!("No signal_identifier found for agent_id {} - cannot deliver scheduled task", task.agent_id); + error!("No identifier found for agent_id {} - cannot deliver scheduled task", task.agent_id); continue; } Err(e) => { - error!("Failed to look up signal_identifier for agent_id {}: {}", task.agent_id, e); + error!("Failed to look up identifier for agent_id {}: {}", task.agent_id, e); continue; } }; - // Handle different task types based on payload let task_result: Result<(), String> = match &task.payload { scheduler::TaskPayload::Message(msg_payload) => { - info!("📤 Sending scheduled message to {}: {}", signal_identifier, msg_payload.message); - let client = signal_client.lock().await; + info!("Sending scheduled message to {}: {}", signal_identifier, msg_payload.message); + let client = messenger.lock().await; if let Err(e) = client.send_message(&signal_identifier, &msg_payload.message) { Err(format!("Failed to send scheduled message: {}", e)) } else { @@ -262,7 +318,6 @@ async fn main() -> Result<()> { } }; - // Update task status based on result match task_result { Ok(()) => { if let Err(e) = scheduler::complete_task(&scheduler_db, &task) { @@ -281,47 +336,54 @@ async fn main() -> Result<()> { // Handle incoming messages Some(msg) = rx.recv() => { // Check if sender is allowed - if !is_user_allowed(&msg.source, &config.signal_allowed_users) { - warn!("🚫 Ignoring message from unauthorized user: {}", msg.source); + if !is_user_allowed(&msg.source, config.allowed_users()) { + warn!("Ignoring message from unauthorized user: {}", msg.source); continue; } let user_name = msg.source_name.as_deref().unwrap_or(&msg.source); info!("Processing message from {}...", user_name); - // Get or create agent for this user + // Get or create agent for this conversation + // For Signal: keyed by user UUID (reply_to == source) + // For Marmot: keyed by sender pubkey (reply_to == from_pubkey) let (agent_id, agent) = match agent_manager.get_or_create_agent( - &msg.source, - ContextType::Direct, + &msg.reply_to, + context_type, msg.source_name.as_deref(), ).await { Ok(result) => result, Err(e) => { - error!("Failed to get/create agent for {}: {}", msg.source, e); + error!("Failed to get/create agent for {}: {}", msg.reply_to, e); continue; } }; info!("Using agent {} for user {}", agent_id, user_name); - // Send typing indicator early (image processing can be slow) + // Persist reply context (e.g. Marmot group_id) for route restoration after restart + if let Some(ref ctx) = msg.reply_context { + if let Err(e) = agent_manager.update_reply_context(&msg.reply_to, ctx) { + warn!("Failed to persist reply context: {}", e); + } + } + + // Send typing indicator early { - let client = signal_client.lock().await; - let _ = client.send_typing(&msg.source, false); + let client = messenger.lock().await; + let _ = client.send_typing(&msg.reply_to, false); } // Check for image attachments and run vision pre-processing let attachment_text = { let image_attachment = msg.attachments.iter().find(|a| vision::is_supported_image(&a.content_type)); if let Some(attachment) = image_attachment { - // Construct full path to the attachment file inside the container let attachment_path = format!( "/signal-cli-data/.local/share/signal-cli/attachments/{}", attachment.file ); - info!("🖼️ Image attachment detected: {} ({}) at {}", attachment.file, attachment.content_type, attachment_path); + info!("Image attachment detected: {} ({}) at {}", attachment.file, attachment.content_type, attachment_path); - // Get recent conversation context for the vision agent (last 6 messages) let recent_context = { let agent_guard = agent.lock().await; match agent_guard.get_recent_messages_for_vision(6) { @@ -343,7 +405,7 @@ async fn main() -> Result<()> { &recent_context, ).await { Ok(description) => { - info!("🖼️ Image described ({} chars)", description.len()); + info!("Image described ({} chars)", description.len()); Some(description) } Err(e) => { @@ -356,7 +418,6 @@ async fn main() -> Result<()> { } }; - // Build the user message content that the agent will see let user_message = if let Some(ref desc) = attachment_text { if msg.message.is_empty() { format!("[Uploaded Image: {}]", desc) @@ -367,7 +428,7 @@ async fn main() -> Result<()> { msg.message.clone() }; - // Store incoming message (sync, no embedding) and update embedding in background + // Store incoming message let user_msg_id = { let agent_guard = agent.lock().await; match agent_guard.store_message_sync_with_attachment( @@ -387,7 +448,6 @@ async fn main() -> Result<()> { } }; - // Update embedding in background (embed the full content the agent sees) if let Some(msg_id) = user_msg_id { let agent_clone = agent.clone(); let embed_content = user_message.clone(); @@ -400,7 +460,7 @@ async fn main() -> Result<()> { } // Process message with agent - let recipient = msg.source.clone(); + let recipient = msg.reply_to.clone(); let mut had_error = false; let max_steps = 10; @@ -413,44 +473,37 @@ async fn main() -> Result<()> { match step_result { Ok(result) => { - // Send messages from this step IMMEDIATELY (don't wait for storage) let msg_count = result.messages.len(); let mut messages_to_store: Vec = Vec::new(); for (i, response) in result.messages.iter().enumerate() { let log_preview: String = response.chars().take(50).collect(); - info!("📤 Sending response ({}/{}): {}...", i + 1, msg_count, log_preview); + info!("Sending response ({}/{}): {}...", i + 1, msg_count, log_preview); - // Send reply FIRST (don't block on storage) { - let client = signal_client.lock().await; + let client = messenger.lock().await; if let Err(e) = client.send_message(&recipient, response) { error!("Failed to send reply: {}", e); } } - // Queue for background storage messages_to_store.push(response.clone()); - // Delay between multiple messages for natural feel if i < msg_count - 1 { tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; { - let client = signal_client.lock().await; + let client = messenger.lock().await; let _ = client.send_typing(&recipient, false); } tokio::time::sleep(tokio::time::Duration::from_millis(1450)).await; } } - // Stop typing indicator if msg_count > 0 { - let client = signal_client.lock().await; + let client = messenger.lock().await; let _ = client.send_typing(&recipient, true); } - // Store messages SYNCHRONOUSLY so next step sees them in conversation history - // Only the embedding update is async (slow API call) let mut msg_ids_for_embedding: Vec<(Uuid, String)> = Vec::new(); for response in &messages_to_store { let msg_id = { @@ -462,7 +515,6 @@ async fn main() -> Result<()> { } } - // Async embedding updates (don't block next step) if !msg_ids_for_embedding.is_empty() { let agent_clone = agent.clone(); tokio::spawn(async move { @@ -475,7 +527,6 @@ async fn main() -> Result<()> { }); } - // Store executed tool calls in background (still uses async store_tool_message) if !result.executed_tools.is_empty() { let agent_clone = agent.clone(); let recipient_clone = recipient.clone(); @@ -488,10 +539,9 @@ async fn main() -> Result<()> { } } }); - info!("🔧 Queued {} tool calls for storage", result.executed_tools.len()); + info!("Queued {} tool calls for storage", result.executed_tools.len()); } - // If done, break if result.done { break; } @@ -505,7 +555,7 @@ async fn main() -> Result<()> { } if had_error { - let client = signal_client.lock().await; + let client = messenger.lock().await; let _ = client.send_message( &recipient, "Sorry, I encountered an error processing your message." diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs new file mode 100644 index 0000000..d050601 --- /dev/null +++ b/crates/sage-core/src/marmot.rs @@ -0,0 +1,599 @@ +use anyhow::{anyhow, Context, Result}; +use serde_json::json; +use std::collections::HashMap; +use std::io::{BufRead, BufReader, BufWriter, Write}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use tokio::sync::mpsc; +use tracing::{debug, error, info, warn}; + +use crate::messenger::{IncomingMessage, Messenger}; + +const BECH32_CHARSET: &str = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"; + +/// Decode a bech32-encoded string (npub1...) into its raw bytes. +fn bech32_decode_payload(s: &str) -> Option> { + let pos = s.rfind('1')?; + let data_part = &s[pos + 1..]; + if data_part.len() < 6 { + return None; + } + let values: Vec = data_part + .chars() + .map(|c| BECH32_CHARSET.find(c).map(|i| i as u8)) + .collect::>>()?; + let data_values = &values[..values.len() - 6]; + let mut acc: u32 = 0; + let mut bits: u32 = 0; + let mut result = Vec::new(); + for &v in data_values { + acc = (acc << 5) | (v as u32); + bits += 5; + if bits >= 8 { + bits -= 8; + result.push((acc >> bits) as u8); + acc &= (1 << bits) - 1; + } + } + Some(result) +} + +/// Convert an npub (bech32) or hex pubkey string to hex. +/// Accepts both "npub1..." and raw 64-char hex. +pub fn normalize_pubkey(input: &str) -> Result { + let trimmed = input.trim(); + if trimmed.starts_with("npub1") { + let bytes = + bech32_decode_payload(trimmed).ok_or_else(|| anyhow!("invalid npub: {}", trimmed))?; + if bytes.len() != 32 { + return Err(anyhow!( + "npub decoded to {} bytes, expected 32", + bytes.len() + )); + } + Ok(bytes.iter().map(|b| format!("{:02x}", b)).collect()) + } else if trimmed.len() == 64 && trimmed.chars().all(|c| c.is_ascii_hexdigit()) { + Ok(trimmed.to_lowercase()) + } else { + Err(anyhow!( + "invalid pubkey (expected npub1... or 64-char hex): {}", + trimmed + )) + } +} + +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub struct MarmotConfig { + pub binary_path: String, + pub relays: Vec, + pub state_dir: String, + pub allowed_pubkeys: Vec, + pub auto_accept_welcomes: bool, +} + +pub struct MarmotClient { + writer: Arc>>, + request_id: AtomicU64, + /// Maps sender pubkey -> latest nostr_group_id for routing replies. + /// Currently treats each pubkey as a single identity (like Signal UUID), + /// collapsing all groups from the same sender into one agent context. + /// TODO: When multi-agent/subagent support lands, this could be extended + /// to route per-group (each group ID = separate agent thread) while still + /// sharing a parent identity for cross-thread memory. + group_routes: Arc>>, + child: Arc>, +} + +impl Drop for MarmotClient { + fn drop(&mut self) { + if let Ok(mut child) = self.child.lock() { + let _ = child.kill(); + let _ = child.wait(); + } + } +} + +impl MarmotClient { + fn send_cmd(&self, cmd: serde_json::Value) -> Result<()> { + let mut writer = self + .writer + .lock() + .map_err(|e| anyhow!("Lock error: {}", e))?; + let cmd_str = serde_json::to_string(&cmd)? + "\n"; + writer.write_all(cmd_str.as_bytes())?; + writer.flush()?; + Ok(()) + } + + fn next_request_id(&self) -> String { + self.request_id.fetch_add(1, Ordering::SeqCst).to_string() + } +} + +impl MarmotClient { + fn resolve_group(&self, pubkey: &str) -> Result { + let routes = self + .group_routes + .lock() + .map_err(|e| anyhow!("Lock error: {}", e))?; + routes + .get(pubkey) + .cloned() + .ok_or_else(|| anyhow!("No group route for pubkey {}", pubkey)) + } +} + +impl Messenger for MarmotClient { + fn send_message(&self, recipient: &str, message: &str) -> Result<()> { + let group_id = self.resolve_group(recipient)?; + let id = self.next_request_id(); + let preview_end = { + let max_len = 50.min(message.len()); + let mut end = max_len; + while end > 0 && !message.is_char_boundary(end) { + end -= 1; + } + end + }; + info!( + "Sending marmot message (req #{}) to {} via group {}: {}...", + id, + recipient, + group_id, + &message[..preview_end] + ); + self.send_cmd(json!({ + "cmd": "send_message", + "request_id": id, + "nostr_group_id": group_id, + "content": message + })) + } + + fn send_typing(&self, recipient: &str, stop: bool) -> Result<()> { + if stop { + return Ok(()); + } + let group_id = match self.resolve_group(recipient) { + Ok(gid) => gid, + Err(_) => return Ok(()), + }; + let id = self.next_request_id(); + self.send_cmd(json!({ + "cmd": "send_typing", + "request_id": id, + "nostr_group_id": group_id + })) + } +} + +/// Create a MarmotClient without spawning marmotd. The supervisor loop +/// (`run_marmot_receive_loop`) handles spawning and respawning the daemon. +/// Validates config (pubkey format) eagerly so startup fails fast on bad config. +pub fn new_marmot_client(config: &MarmotConfig) -> Result { + // Validate pubkeys eagerly + let is_wildcard = config.allowed_pubkeys.iter().any(|p| p == "*"); + if !is_wildcard { + for pk in &config.allowed_pubkeys { + normalize_pubkey(pk) + .with_context(|| format!("invalid MARMOT_ALLOWED_PUBKEYS entry: {}", pk))?; + } + } + + // Placeholder process -- the supervisor replaces writer and child on first spawn. + // Using `cat` which blocks on stdin and exits cleanly when stdin closes. + let mut placeholder = Command::new("cat") + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .spawn() + .context("Failed to spawn placeholder process")?; + let stdin = placeholder + .stdin + .take() + .context("Failed to get placeholder stdin")?; + + let writer = Arc::new(Mutex::new(BufWriter::new(stdin))); + let group_routes = Arc::new(Mutex::new(HashMap::new())); + + Ok(MarmotClient { + writer: writer.clone(), + request_id: AtomicU64::new(1), + group_routes, + child: Arc::new(Mutex::new(placeholder)), + }) +} + +/// Single iteration of the marmot receive loop: spawn marmotd, run through +/// all three phases (ready, keypackage, message loop), and return on any exit. +/// The caller (supervisor) handles retry with backoff. +fn run_marmot_receive_once( + config: &MarmotConfig, + tx: &mpsc::Sender, + group_routes: &Arc>>, + client_writer: &Arc>>, + client_child: &Mutex, +) -> Result<()> { + // Spawn a fresh marmotd process + let mut cmd = Command::new(&config.binary_path); + cmd.arg("daemon"); + for relay in &config.relays { + cmd.arg("--relay").arg(relay); + } + cmd.arg("--state-dir").arg(&config.state_dir); + let is_wildcard = config.allowed_pubkeys.iter().any(|p| p == "*"); + if !is_wildcard { + for pk in &config.allowed_pubkeys { + if let Ok(hex_pk) = normalize_pubkey(pk) { + cmd.arg("--allow-pubkey").arg(&hex_pk); + } + } + } + cmd.stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + + info!( + "Spawning marmotd: {} daemon --relay {} --state-dir {}", + config.binary_path, + config.relays.join(","), + config.state_dir + ); + + let mut child = cmd.spawn().context("Failed to spawn marmotd")?; + let stdin = child.stdin.take().context("Failed to get marmotd stdin")?; + let stdout = child + .stdout + .take() + .context("Failed to get marmotd stdout")?; + let stderr = child + .stderr + .take() + .context("Failed to get marmotd stderr")?; + + // Forward stderr to tracing + std::thread::spawn(move || { + let reader = BufReader::new(stderr); + for line in reader.lines().map_while(Result::ok) { + info!(target: "marmotd", "{}", line); + } + }); + + // Update the shared writer and child so MarmotClient can send to the new process + { + let mut w = client_writer + .lock() + .map_err(|e| anyhow!("Lock error: {}", e))?; + *w = BufWriter::new(stdin); + } + { + let mut c = client_child + .lock() + .map_err(|e| anyhow!("Lock error: {}", e))?; + // Kill old process if still running + let _ = c.kill(); + let _ = c.wait(); + *c = child; + } + + let send_cmd = |cmd: serde_json::Value| -> Result<()> { + let mut w = client_writer + .lock() + .map_err(|e| anyhow!("Lock error: {}", e))?; + let s = serde_json::to_string(&cmd)? + "\n"; + w.write_all(s.as_bytes())?; + w.flush()?; + Ok(()) + }; + + let mut reader = BufReader::new(stdout); + let mut line = String::new(); + + // Phase 1: Wait for ready + loop { + line.clear(); + if reader.read_line(&mut line)? == 0 { + return Err(anyhow!("marmotd closed stdout before ready")); + } + let event: serde_json::Value = match serde_json::from_str(line.trim()) { + Ok(v) => v, + Err(_) => { + debug!("marmotd non-json output (startup): {}", line.trim()); + continue; + } + }; + + if event.get("type").and_then(|t| t.as_str()) == Some("ready") { + let pubkey = event + .get("pubkey") + .and_then(|p| p.as_str()) + .unwrap_or("unknown"); + let npub = event + .get("npub") + .and_then(|n| n.as_str()) + .unwrap_or("unknown"); + info!("marmotd ready: pubkey={} npub={}", pubkey, npub); + break; + } + } + + // Phase 2: Publish keypackage + let relays: Vec<&str> = config.relays.iter().map(|s| s.as_str()).collect(); + send_cmd(json!({ + "cmd": "publish_keypackage", + "request_id": "init_kp", + "relays": relays + }))?; + + loop { + line.clear(); + if reader.read_line(&mut line)? == 0 { + return Err(anyhow!("marmotd closed stdout during keypackage publish")); + } + let event: serde_json::Value = match serde_json::from_str(line.trim()) { + Ok(v) => v, + Err(_) => { + debug!("marmotd non-json output (keypackage): {}", line.trim()); + continue; + } + }; + + let event_type = event.get("type").and_then(|t| t.as_str()).unwrap_or(""); + if event_type == "ok" + && event.get("request_id").and_then(|id| id.as_str()) == Some("init_kp") + { + info!("marmotd keypackage published"); + break; + } + if event_type == "error" + && event.get("request_id").and_then(|id| id.as_str()) == Some("init_kp") + { + let msg = event + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("unknown error"); + warn!("marmotd keypackage publish failed: {}", msg); + break; + } + } + + info!("Marmot receive loop started, listening for messages..."); + + // Phase 3: Main receive loop + loop { + line.clear(); + match reader.read_line(&mut line) { + Ok(0) => { + return Err(anyhow!("marmotd closed stdout unexpectedly")); + } + Ok(_) => { + let event: serde_json::Value = match serde_json::from_str(line.trim()) { + Ok(v) => v, + Err(e) => { + debug!("marmotd non-json output: {} ({})", line.trim(), e); + continue; + } + }; + let event_type = event.get("type").and_then(|t| t.as_str()).unwrap_or(""); + + match event_type { + "welcome_received" => { + let wrapper_id = event + .get("wrapper_event_id") + .and_then(|x| x.as_str()) + .unwrap_or(""); + let from = event + .get("from_pubkey") + .and_then(|x| x.as_str()) + .unwrap_or("unknown"); + let group_name = event + .get("group_name") + .and_then(|x| x.as_str()) + .unwrap_or(""); + info!( + "Marmot welcome received from {} (group: {}, wrapper: {})", + from, group_name, wrapper_id + ); + + if config.auto_accept_welcomes && !wrapper_id.is_empty() { + let req_id = format!("auto_{}", wrapper_id); + if let Err(e) = send_cmd(json!({ + "cmd": "accept_welcome", + "request_id": req_id, + "wrapper_event_id": wrapper_id + })) { + warn!("Failed to auto-accept welcome: {}", e); + } + } + } + "group_joined" => { + let group_id = event + .get("nostr_group_id") + .and_then(|x| x.as_str()) + .unwrap_or("unknown"); + info!("Marmot joined group: {}", group_id); + } + "message_received" => { + let from_pubkey = event + .get("from_pubkey") + .and_then(|x| x.as_str()) + .unwrap_or(""); + let content = event.get("content").and_then(|x| x.as_str()).unwrap_or(""); + let group_id = event + .get("nostr_group_id") + .and_then(|x| x.as_str()) + .unwrap_or(""); + let created_at = event + .get("created_at") + .and_then(|x| x.as_u64()) + .unwrap_or(0); + + if content.is_empty() { + continue; + } + + let preview_end = { + let max_len = 100.min(content.len()); + let mut end = max_len; + while end > 0 && !content.is_char_boundary(end) { + end -= 1; + } + end + }; + info!( + "Marmot message from {} in group {}: {}", + from_pubkey, + group_id, + &content[..preview_end] + ); + + // Track pubkey -> latest group for reply routing. + // This means the most recent group a user messages from + // becomes the reply target. When we add multi-agent support, + // each group could maintain its own agent thread instead. + if !from_pubkey.is_empty() && !group_id.is_empty() { + if let Ok(mut routes) = group_routes.lock() { + routes.insert(from_pubkey.to_string(), group_id.to_string()); + } + } + + let msg = IncomingMessage { + source: from_pubkey.to_string(), + source_name: None, + message: content.to_string(), + attachments: vec![], + timestamp: created_at, + reply_to: from_pubkey.to_string(), + reply_context: Some(group_id.to_string()), + }; + + if tx.blocking_send(msg).is_err() { + error!("Failed to send marmot message to channel (receiver dropped)"); + return Err(anyhow!("message channel closed")); + } + } + "ok" | "keypackage_published" => { + debug!("marmotd: {}", line.trim()); + } + "error" => { + let msg = event + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("unknown"); + warn!("marmotd error: {}", msg); + } + _ => { + debug!("marmotd event: {}", line.trim()); + } + } + } + Err(e) => { + return Err(anyhow!("Error reading from marmotd: {}", e)); + } + } + } +} + +/// Supervised marmot receive loop with exponential backoff on failures. +/// Respawns marmotd and re-initializes stdin/stdout handles on each retry. +pub async fn run_marmot_receive_loop( + tx: mpsc::Sender, + config: MarmotConfig, + group_routes: Arc>>, + client_writer: Arc>>, + client_child: Arc>, +) -> Result<()> { + let mut backoff = std::time::Duration::from_millis(250); + let backoff_max = std::time::Duration::from_secs(60); + + loop { + let config = config.clone(); + let tx = tx.clone(); + let group_routes = group_routes.clone(); + let client_writer = client_writer.clone(); + let client_child = client_child.clone(); + + let result = tokio::task::spawn_blocking(move || { + run_marmot_receive_once(&config, &tx, &group_routes, &client_writer, &client_child) + }) + .await; + + match result { + Ok(Ok(())) => { + warn!( + "Marmot receive loop exited unexpectedly; restarting in {:?}", + backoff + ); + } + Ok(Err(e)) => { + let msg = format!("{}", e); + if msg.contains("message channel closed") { + error!("Message channel closed, stopping marmot supervisor"); + return Err(e); + } + warn!( + "Marmot receive loop error; restarting in {:?}: {}", + backoff, e + ); + } + Err(e) => { + warn!( + "Marmot receive task panicked; restarting in {:?}: {}", + backoff, e + ); + } + } + + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(backoff_max); + } +} + +/// Get the shared writer handle from a MarmotClient (for the receive loop). +pub fn writer_handle(client: &MarmotClient) -> Arc>> { + client.writer.clone() +} + +/// Get the shared group routes handle from a MarmotClient (for the receive loop). +pub fn group_routes_handle(client: &MarmotClient) -> Arc>> { + client.group_routes.clone() +} + +/// Get the shared child process handle from a MarmotClient (for the supervisor loop). +pub fn child_handle(client: &MarmotClient) -> Arc> { + client.child.clone() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_normalize_npub() { + let hex = + normalize_pubkey("npub1gx8my906z8urmgzpcynjlj43ehwc5jket0mc70pkvzkg6k636hmqnwunq7") + .unwrap(); + assert_eq!( + hex, + "418fb215fa11f83da041c1272fcab1cddd8a4ad95bf78f3c3660ac8d5b51d5f6" + ); + } + + #[test] + fn test_normalize_hex_passthrough() { + let hex = + normalize_pubkey("418fb215fa11f83da041c1272fcab1cddd8a4ad95bf78f3c3660ac8d5b51d5f6") + .unwrap(); + assert_eq!( + hex, + "418fb215fa11f83da041c1272fcab1cddd8a4ad95bf78f3c3660ac8d5b51d5f6" + ); + } + + #[test] + fn test_normalize_invalid() { + assert!(normalize_pubkey("not_a_valid_key").is_err()); + assert!(normalize_pubkey("npub1invalid").is_err()); + } +} diff --git a/crates/sage-core/src/messenger.rs b/crates/sage-core/src/messenger.rs new file mode 100644 index 0000000..2d34679 --- /dev/null +++ b/crates/sage-core/src/messenger.rs @@ -0,0 +1,38 @@ +use anyhow::Result; + +/// An attachment received from a messaging provider +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub struct IncomingAttachment { + pub file: String, + pub content_type: String, + pub size: Option, +} + +/// A message received from a messaging provider +#[derive(Debug, Clone)] +pub struct IncomingMessage { + /// Unique identifier of the sender (Signal UUID, Nostr pubkey, etc.) + pub source: String, + pub source_name: Option, + pub message: String, + pub attachments: Vec, + #[allow(dead_code)] + pub timestamp: u64, + /// Identity key for agent lookup and reply routing (Signal UUID or Marmot pubkey) + pub reply_to: String, + /// Transport-specific routing context to persist (e.g. Marmot nostr_group_id). + /// Used to restore reply routing after restarts. + pub reply_context: Option, +} + +/// Trait for sending messages via a messaging provider +pub trait Messenger: Send + Sync { + fn send_message(&self, recipient: &str, message: &str) -> Result<()>; + fn send_typing(&self, recipient: &str, stop: bool) -> Result<()>; + + /// Periodic health/refresh check (no-op by default) + fn refresh(&self) -> Result<()> { + Ok(()) + } +} diff --git a/crates/sage-core/src/schema.rs b/crates/sage-core/src/schema.rs index da646b3..0de680f 100644 --- a/crates/sage-core/src/schema.rs +++ b/crates/sage-core/src/schema.rs @@ -128,6 +128,7 @@ diesel::table! { context_type -> Varchar, display_name -> Nullable, created_at -> Timestamptz, + reply_context -> Nullable, } } diff --git a/crates/sage-core/src/signal.rs b/crates/sage-core/src/signal.rs index 62b1821..4b917a0 100644 --- a/crates/sage-core/src/signal.rs +++ b/crates/sage-core/src/signal.rs @@ -16,28 +16,7 @@ use std::time::{Duration, Instant}; use tokio::sync::mpsc; use tracing::{debug, error, info, warn}; -/// An attachment received from Signal -#[derive(Debug, Clone)] -#[allow(dead_code)] -pub struct IncomingAttachment { - /// Local file path where signal-cli downloaded the attachment - pub file: String, - /// MIME content type (e.g., "image/jpeg", "image/png") - pub content_type: String, - /// File size in bytes (if reported) - pub size: Option, -} - -/// A message received from Signal -#[derive(Debug, Clone)] -pub struct IncomingMessage { - pub source: String, - pub source_name: Option, - pub message: String, - pub attachments: Vec, - #[allow(dead_code)] - pub timestamp: u64, -} +use crate::messenger::{IncomingAttachment, IncomingMessage, Messenger}; /// Connection mode for signal-cli #[allow(dead_code)] @@ -350,6 +329,20 @@ impl SignalClient { } } +impl Messenger for SignalClient { + fn send_message(&self, recipient: &str, message: &str) -> Result<()> { + SignalClient::send_message(self, recipient, message) + } + + fn send_typing(&self, recipient: &str, stop: bool) -> Result<()> { + SignalClient::send_typing(self, recipient, stop) + } + + fn refresh(&self) -> Result<()> { + self.refresh_account() + } +} + impl Drop for SignalClient { fn drop(&mut self) { if let Ok(mut mode) = self.mode.lock() { @@ -439,11 +432,13 @@ pub fn parse_incoming_message(line: &str) -> Option { let timestamp = data_message.get("timestamp")?.as_u64()?; Some(IncomingMessage { + reply_to: source.clone(), source, source_name, message: message.to_string(), attachments, timestamp, + reply_context: None, }) } diff --git a/justfile b/justfile index 1e38687..50f372e 100644 --- a/justfile +++ b/justfile @@ -14,7 +14,7 @@ default: build: podman build -f Dockerfile -t sage:latest . -# Start all containers (postgres, signal-cli, sage) +# Start all containers (postgres + messenger + sage) start: #!/usr/bin/env bash set -e @@ -22,7 +22,8 @@ start: source .env set +a - echo "Starting Sage stack..." + MESSENGER="${MESSENGER:-signal}" + echo "Starting Sage stack (messenger: $MESSENGER)..." # Start PostgreSQL if not running if ! podman ps --format '{{{{.Names}}}}' | grep -q '^sage-postgres$'; then @@ -36,50 +37,81 @@ start: echo "PostgreSQL already running" fi - # Start signal-cli if not running - if ! podman ps --format '{{{{.Names}}}}' | grep -q '^sage-signal-cli$'; then - echo "Starting signal-cli..." - podman run -d --name sage-signal-cli \ - -p 7583:7583 -v signal-cli-data:/var/lib/signal-cli --tmpfs /tmp:exec \ - registry.gitlab.com/packaging/signal-cli/signal-cli-jre:latest \ - daemon --tcp 0.0.0.0:7583 --send-read-receipts --ignore-stories - sleep 2 + # Messenger-specific setup + MESSENGER_VOLUMES="" + MESSENGER_ENV="" + + if [ "$MESSENGER" = "signal" ]; then + # Start signal-cli if not running + if ! podman ps --format '{{{{.Names}}}}' | grep -q '^sage-signal-cli$'; then + echo "Starting signal-cli..." + podman run -d --name sage-signal-cli \ + -p 7583:7583 -v signal-cli-data:/var/lib/signal-cli --tmpfs /tmp:exec \ + registry.gitlab.com/packaging/signal-cli/signal-cli-jre:latest \ + daemon --tcp 0.0.0.0:7583 --send-read-receipts --ignore-stories + sleep 2 + else + echo "signal-cli already running" + fi + + # Ensure signal-cli attachments are readable by sage + podman run --rm -v signal-cli-data:/signal-cli-data \ + docker.io/alpine:latest \ + sh -c "chmod o+rX /signal-cli-data/.local/share/signal-cli/attachments 2>/dev/null || true" + + MESSENGER_VOLUMES="-v signal-cli-data:/signal-cli-data:ro" + MESSENGER_ENV="\ + -e MESSENGER=signal \ + -e SIGNAL_CLI_HOST=localhost -e SIGNAL_CLI_PORT=7583 \ + -e SIGNAL_PHONE_NUMBER=$SIGNAL_PHONE_NUMBER \ + -e SIGNAL_ALLOWED_USERS=$SIGNAL_ALLOWED_USERS" + elif [ "$MESSENGER" = "marmot" ]; then + echo "Using Marmot (MLS over Nostr) - no signal-cli needed" + podman volume create sage-marmot-state 2>/dev/null || true + + MESSENGER_VOLUMES="-v sage-marmot-state:/data/marmot-state:U,z" + MESSENGER_ENV="\ + -e MESSENGER=marmot \ + -e MARMOT_RELAYS=${MARMOT_RELAYS:-} \ + -e MARMOT_STATE_DIR=/data/marmot-state \ + -e MARMOT_ALLOWED_PUBKEYS=${MARMOT_ALLOWED_PUBKEYS:-} \ + -e MARMOT_AUTO_ACCEPT_WELCOMES=${MARMOT_AUTO_ACCEPT_WELCOMES:-true}" else - echo "signal-cli already running" + echo "Unknown MESSENGER=$MESSENGER (expected 'signal' or 'marmot')" + exit 1 fi - # Ensure signal-cli attachments are readable by sage - podman run --rm -v signal-cli-data:/signal-cli-data \ - docker.io/alpine:latest \ - sh -c "chmod o+rX /signal-cli-data/.local/share/signal-cli/attachments 2>/dev/null || true" - # Create workspace directory if it doesn't exist mkdir -p ~/.sage/workspace # Remove old sage container and start fresh podman rm -f sage 2>/dev/null || true echo "Starting Sage..." - podman run -d --name sage --network host \ + eval podman run -d --name sage --network host \ -v ~/.sage/workspace:/workspace:U,z \ - -v signal-cli-data:/signal-cli-data:ro \ + $MESSENGER_VOLUMES \ -e DATABASE_URL=postgres://sage:sage@localhost:5434/sage \ -e MAPLE_API_URL="$MAPLE_API_URL" \ -e MAPLE_API_KEY="$MAPLE_API_KEY" \ -e MAPLE_MODEL="$MAPLE_MODEL" \ -e MAPLE_EMBEDDING_MODEL="$MAPLE_EMBEDDING_MODEL" \ - -e SIGNAL_CLI_HOST=localhost -e SIGNAL_CLI_PORT=7583 \ - -e SIGNAL_PHONE_NUMBER="$SIGNAL_PHONE_NUMBER" \ - -e SIGNAL_ALLOWED_USERS="$SIGNAL_ALLOWED_USERS" \ + $MESSENGER_ENV \ -e BRAVE_API_KEY="$BRAVE_API_KEY" \ -e SAGE_WORKSPACE=/workspace \ + -e HEALTH_PORT="${HEALTH_PORT:-8080}" \ -e RUST_LOG=info \ sage:latest sleep 2 echo "" - echo "🌿 Sage stack started!" - echo " - PostgreSQL: localhost:5434 (data: sage-pgdata volume)" - echo " - signal-cli: localhost:7583 (data: signal-cli-data volume)" + echo "Sage stack started! (messenger: $MESSENGER)" + if [ "$MESSENGER" = "signal" ]; then + echo " - PostgreSQL: localhost:5434 (data: sage-pgdata volume)" + echo " - signal-cli: localhost:7583 (data: signal-cli-data volume)" + else + echo " - PostgreSQL: localhost:5434 (data: sage-pgdata volume)" + echo " - Marmot state: sage-marmot-state volume" + fi echo " - Sage: running" echo " - Workspace: ~/.sage/workspace" echo "" @@ -93,38 +125,60 @@ stop: podman rm -f sage 2>/dev/null || true podman rm -f sage-signal-cli 2>/dev/null || true podman rm -f sage-postgres 2>/dev/null || true - echo "Containers stopped. Data preserved in volumes (sage-pgdata, signal-cli-data)." + echo "Containers stopped. Data preserved in volumes (sage-pgdata, signal-cli-data, sage-marmot-state)." -# Restart Sage only (keeps postgres and signal-cli running) +# Restart Sage only (keeps postgres running) restart: #!/usr/bin/env bash set -a source .env set +a - # Ensure signal-cli attachments are readable by sage - podman run --rm -v signal-cli-data:/signal-cli-data \ - docker.io/alpine:latest \ - sh -c "chmod o+rX /signal-cli-data/.local/share/signal-cli/attachments 2>/dev/null || true" + MESSENGER="${MESSENGER:-signal}" + + MESSENGER_VOLUMES="" + MESSENGER_ENV="" + + if [ "$MESSENGER" = "signal" ]; then + podman run --rm -v signal-cli-data:/signal-cli-data \ + docker.io/alpine:latest \ + sh -c "chmod o+rX /signal-cli-data/.local/share/signal-cli/attachments 2>/dev/null || true" + + MESSENGER_VOLUMES="-v signal-cli-data:/signal-cli-data:ro" + MESSENGER_ENV="\ + -e MESSENGER=signal \ + -e SIGNAL_CLI_HOST=localhost -e SIGNAL_CLI_PORT=7583 \ + -e SIGNAL_PHONE_NUMBER=$SIGNAL_PHONE_NUMBER \ + -e SIGNAL_ALLOWED_USERS=$SIGNAL_ALLOWED_USERS" + elif [ "$MESSENGER" = "marmot" ]; then + podman volume create sage-marmot-state 2>/dev/null || true + + MESSENGER_VOLUMES="-v sage-marmot-state:/data/marmot-state:U,z" + MESSENGER_ENV="\ + -e MESSENGER=marmot \ + -e MARMOT_RELAYS=${MARMOT_RELAYS:-} \ + -e MARMOT_STATE_DIR=/data/marmot-state \ + -e MARMOT_ALLOWED_PUBKEYS=${MARMOT_ALLOWED_PUBKEYS:-} \ + -e MARMOT_AUTO_ACCEPT_WELCOMES=${MARMOT_AUTO_ACCEPT_WELCOMES:-true}" + fi mkdir -p ~/.sage/workspace podman rm -f sage 2>/dev/null || true - podman run -d --name sage --network host \ + eval podman run -d --name sage --network host \ -v ~/.sage/workspace:/workspace:U,z \ - -v signal-cli-data:/signal-cli-data:ro \ + $MESSENGER_VOLUMES \ -e DATABASE_URL=postgres://sage:sage@localhost:5434/sage \ -e MAPLE_API_URL="$MAPLE_API_URL" \ -e MAPLE_API_KEY="$MAPLE_API_KEY" \ -e MAPLE_MODEL="$MAPLE_MODEL" \ -e MAPLE_EMBEDDING_MODEL="$MAPLE_EMBEDDING_MODEL" \ - -e SIGNAL_CLI_HOST=localhost -e SIGNAL_CLI_PORT=7583 \ - -e SIGNAL_PHONE_NUMBER="$SIGNAL_PHONE_NUMBER" \ - -e SIGNAL_ALLOWED_USERS="$SIGNAL_ALLOWED_USERS" \ + $MESSENGER_ENV \ -e BRAVE_API_KEY="$BRAVE_API_KEY" \ -e SAGE_WORKSPACE=/workspace \ + -e HEALTH_PORT="${HEALTH_PORT:-8080}" \ -e RUST_LOG=info \ sage:latest - echo "Sage restarted" + echo "Sage restarted (messenger: $MESSENGER)" # View Sage logs logs: @@ -207,7 +261,7 @@ lint: # List all Sage-related volumes volumes: - podman volume ls | grep -E "sage|signal" + podman volume ls | grep -E "sage|signal|marmot" # DANGER: Delete all data and start fresh nuke: @@ -215,11 +269,12 @@ nuke: echo "⚠️ This will DELETE ALL SAGE DATA including:" echo " - PostgreSQL database (memory, conversations, archival)" echo " - signal-cli registration" + echo " - Marmot state (MLS keys, identity)" echo "" read -p "Type 'DELETE' to confirm: " confirm if [ "$confirm" = "DELETE" ]; then just stop - podman volume rm -f sage-pgdata signal-cli-data 2>/dev/null || true + podman volume rm -f sage-pgdata signal-cli-data sage-marmot-state 2>/dev/null || true echo "All data deleted." else echo "Aborted."