From 374321f8f44c0323b10cee892ba95ce893b33ce6 Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 15:44:45 -0600 Subject: [PATCH 01/11] Add marmotd (MLS over Nostr) as optional messaging provider Introduce a Messenger trait abstraction that allows Sage to use either Signal (default) or Marmot for message transport. Marmot uses the marmotd sidecar daemon from pika, communicating via JSONL over stdio to provide MLS-encrypted messaging over Nostr relays. Key changes: - messenger.rs: Messenger trait with shared IncomingMessage/IncomingAttachment types - marmot.rs: MarmotClient spawning marmotd subprocess, JSONL protocol, npub support - config.rs: MESSENGER env var (signal|marmot) with marmot-specific config fields - main.rs: Messenger-polymorphic event loop using Arc> - signal.rs: Refactored to implement Messenger trait, uses shared types - Dockerfile: Build marmotd from pika source to avoid GLIBC mismatch - justfile: Messenger-aware start/restart/stop with conditional signal-cli Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- Dockerfile | 16 +- crates/sage-core/src/config.rs | 75 +++++ crates/sage-core/src/lib.rs | 2 + crates/sage-core/src/main.rs | 251 +++++++++-------- crates/sage-core/src/marmot.rs | 445 ++++++++++++++++++++++++++++++ crates/sage-core/src/messenger.rs | 35 +++ crates/sage-core/src/signal.rs | 38 ++- justfile | 131 ++++++--- 8 files changed, 816 insertions(+), 177 deletions(-) create mode 100644 crates/sage-core/src/marmot.rs create mode 100644 crates/sage-core/src/messenger.rs diff --git a/Dockerfile b/Dockerfile index 171f740..3a832c8 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.3.2 +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/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..9b52bea 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,119 @@ 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 - ); - } + // Determine context type for agent management + let context_type = match config.messenger_type { + MessengerType::Signal => ContextType::Direct, + MessengerType::Marmot => ContextType::Group, + }; + + // 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)); + + let receive_handle = + tokio::spawn(async move { run_receive_loop(reader, tx).await }); + + (messenger, receive_handle) + } + } + MessengerType::Marmot => { + let marmot_config = config.marmot_config(); - tokio::time::sleep(backoff).await; - backoff = (backoff * 2).min(backoff_max); + if marmot_config.relays.is_empty() { + return Err(anyhow::anyhow!( + "MARMOT_RELAYS must be set when MESSENGER=marmot" + )); } - }); - (signal_client, receive_handle) - } else { - info!("Starting Signal interface (subprocess mode)..."); + info!("Starting Marmot interface..."); + info!(" Relays: {:?}", marmot_config.relays); + info!(" State dir: {}", marmot_config.state_dir); - let signal_client = SignalClient::spawn_subprocess(&signal_phone)?; - let reader = signal_client.take_reader()?; - let signal_client = Arc::new(Mutex::new(signal_client)); + let (client, stdout, _child) = marmot::spawn_marmot(&marmot_config)?; + let writer = marmot::writer_handle(&client); + let messenger: Arc> = Arc::new(Mutex::new(client)); - let receive_handle = tokio::spawn(async move { run_receive_loop(reader, tx).await }); + let receive_handle = tokio::spawn(async move { + marmot::run_marmot_receive_loop(stdout, writer, tx, marmot_config).await + }); - (signal_client, receive_handle) + (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 +250,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 +298,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 +316,47 @@ 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 group ID (reply_to == nostr_group_id) 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) + // 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 +378,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 +391,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 +401,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 +421,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 +433,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 +446,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 +488,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 +500,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 +512,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 +528,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..d77bdfc --- /dev/null +++ b/crates/sage-core/src/marmot.rs @@ -0,0 +1,445 @@ +use anyhow::{anyhow, Context, Result}; +use serde_json::json; +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)] +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, +} + +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 Messenger for MarmotClient { + fn send_message(&self, recipient: &str, message: &str) -> Result<()> { + let id = self.next_request_id(); + // Find valid UTF-8 boundary for preview + 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 group {}: {}...", + id, + recipient, + &message[..preview_end] + ); + self.send_cmd(json!({ + "cmd": "send_message", + "request_id": id, + "nostr_group_id": recipient, + "content": message + })) + } + + fn send_typing(&self, recipient: &str, stop: bool) -> Result<()> { + if stop { + return Ok(()); + } + let id = self.next_request_id(); + self.send_cmd(json!({ + "cmd": "send_typing", + "request_id": id, + "nostr_group_id": recipient + })) + } +} + +/// Spawn marmotd daemon and return the client, stdout reader, and child process handle. +pub fn spawn_marmot( + config: &MarmotConfig, +) -> Result<(MarmotClient, std::process::ChildStdout, Child)> { + 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); + + for pk in &config.allowed_pubkeys { + let hex_pk = normalize_pubkey(pk) + .with_context(|| format!("invalid MARMOT_ALLOWED_PUBKEYS entry: {}", 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); + } + }); + + let writer = Arc::new(Mutex::new(BufWriter::new(stdin))); + + let client = MarmotClient { + writer: writer.clone(), + request_id: AtomicU64::new(1), + }; + + Ok((client, stdout, child)) +} + +/// Run the marmot receive loop: waits for daemon ready, publishes keypackage, +/// then listens for incoming messages and auto-accepts welcomes. +pub async fn run_marmot_receive_loop( + stdout: std::process::ChildStdout, + writer: Arc>>, + tx: mpsc::Sender, + config: MarmotConfig, +) -> Result<()> { + tokio::task::spawn_blocking(move || { + let mut reader = BufReader::new(stdout); + let mut line = String::new(); + + let send_cmd = |cmd: serde_json::Value| -> Result<()> { + let mut w = 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(()) + }; + + // 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 = serde_json::from_str(line.trim()) + .with_context(|| format!("parse marmotd json: {}", line.trim()))?; + + 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 = serde_json::from_str(line.trim()) + .with_context(|| format!("parse marmotd json: {}", line.trim()))?; + + 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) => { + warn!("marmotd closed stdout"); + break; + } + 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] + ); + + let msg = IncomingMessage { + source: from_pubkey.to_string(), + source_name: None, + message: content.to_string(), + attachments: vec![], + timestamp: created_at, + reply_to: group_id.to_string(), + }; + + if tx.blocking_send(msg).is_err() { + error!("Failed to send marmot message to channel"); + break; + } + } + "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) => { + error!("Error reading from marmotd: {}", e); + break; + } + } + } + + warn!("Marmot receive loop ended"); + Ok::<_, anyhow::Error>(()) + }) + .await??; + + Ok(()) +} + +/// Get the shared writer handle from a MarmotClient (for the receive loop). +pub fn writer_handle(client: &MarmotClient) -> Arc>> { + client.writer.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..05ed075 --- /dev/null +++ b/crates/sage-core/src/messenger.rs @@ -0,0 +1,35 @@ +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, + /// Where to send replies (same as source for Signal, nostr_group_id for Marmot) + pub reply_to: String, +} + +/// 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/signal.rs b/crates/sage-core/src/signal.rs index 62b1821..920ac30 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,6 +432,7 @@ 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(), 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." From 25c3dce7aa0d6b08e129164ba5b578f735d9b33f Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 15:51:18 -0600 Subject: [PATCH 02/11] Key Marmot agents by sender pubkey instead of group ID Treat each Nostr pubkey as a stable identity (like Signal UUID) so the same user gets the same agent regardless of which MLS group they message from. MarmotClient maintains a pubkey -> latest group_id routing table so replies reach the correct group. This keeps parity with Signal's 1:1 identity model. When multi-agent support lands, per-group agent threads can be layered on top while sharing a parent identity for cross-thread memory. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- crates/sage-core/src/main.rs | 14 +++++---- crates/sage-core/src/marmot.rs | 54 ++++++++++++++++++++++++++++++---- 2 files changed, 57 insertions(+), 11 deletions(-) diff --git a/crates/sage-core/src/main.rs b/crates/sage-core/src/main.rs index 9b52bea..3b8eb59 100644 --- a/crates/sage-core/src/main.rs +++ b/crates/sage-core/src/main.rs @@ -121,11 +121,11 @@ async fn main() -> Result<()> { // Create channel for incoming messages let (tx, mut rx) = mpsc::channel::(100); - // Determine context type for agent management - let context_type = match config.messenger_type { - MessengerType::Signal => ContextType::Direct, - MessengerType::Marmot => ContextType::Group, - }; + // 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 { @@ -207,10 +207,12 @@ async fn main() -> Result<()> { let (client, stdout, _child) = marmot::spawn_marmot(&marmot_config)?; let writer = marmot::writer_handle(&client); + let group_routes = marmot::group_routes_handle(&client); let messenger: Arc> = Arc::new(Mutex::new(client)); let receive_handle = tokio::spawn(async move { - marmot::run_marmot_receive_loop(stdout, writer, tx, marmot_config).await + marmot::run_marmot_receive_loop(stdout, writer, tx, marmot_config, group_routes) + .await }); (messenger, receive_handle) diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs index d77bdfc..79b48a3 100644 --- a/crates/sage-core/src/marmot.rs +++ b/crates/sage-core/src/marmot.rs @@ -1,5 +1,6 @@ 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}; @@ -74,6 +75,13 @@ pub struct MarmotConfig { 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>>, } impl MarmotClient { @@ -93,10 +101,23 @@ impl MarmotClient { } } +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(); - // Find valid UTF-8 boundary for preview let preview_end = { let max_len = 50.min(message.len()); let mut end = max_len; @@ -106,15 +127,16 @@ impl Messenger for MarmotClient { end }; info!( - "Sending marmot message (req #{}) to group {}: {}...", + "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": recipient, + "nostr_group_id": group_id, "content": message })) } @@ -123,11 +145,15 @@ impl Messenger for MarmotClient { 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": recipient + "nostr_group_id": group_id })) } } @@ -184,9 +210,11 @@ pub fn spawn_marmot( let writer = Arc::new(Mutex::new(BufWriter::new(stdin))); + let group_routes = Arc::new(Mutex::new(HashMap::new())); let client = MarmotClient { writer: writer.clone(), request_id: AtomicU64::new(1), + group_routes, }; Ok((client, stdout, child)) @@ -199,6 +227,7 @@ pub async fn run_marmot_receive_loop( writer: Arc>>, tx: mpsc::Sender, config: MarmotConfig, + group_routes: Arc>>, ) -> Result<()> { tokio::task::spawn_blocking(move || { let mut reader = BufReader::new(stdout); @@ -362,13 +391,23 @@ pub async fn run_marmot_receive_loop( &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: group_id.to_string(), + reply_to: from_pubkey.to_string(), }; if tx.blocking_send(msg).is_err() { @@ -411,6 +450,11 @@ pub fn writer_handle(client: &MarmotClient) -> Arc Arc>> { + client.group_routes.clone() +} + #[cfg(test)] mod tests { use super::*; From 83ad3a1f26b63b9b6d9f2668d44e3c10631fe59d Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 17:52:03 -0600 Subject: [PATCH 03/11] Skip --allow-pubkey flags when wildcard (*) is set When MARMOT_ALLOWED_PUBKEYS=* is configured, don't pass any --allow-pubkey args to marmotd so it accepts messages from everyone. Previously normalize_pubkey("*") would error and crash on startup. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- crates/sage-core/src/marmot.rs | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs index 79b48a3..ae50d80 100644 --- a/crates/sage-core/src/marmot.rs +++ b/crates/sage-core/src/marmot.rs @@ -171,10 +171,13 @@ pub fn spawn_marmot( cmd.arg("--state-dir").arg(&config.state_dir); - for pk in &config.allowed_pubkeys { - let hex_pk = normalize_pubkey(pk) - .with_context(|| format!("invalid MARMOT_ALLOWED_PUBKEYS entry: {}", pk))?; - cmd.arg("--allow-pubkey").arg(&hex_pk); + let is_wildcard = config.allowed_pubkeys.iter().any(|p| p == "*"); + if !is_wildcard { + for pk in &config.allowed_pubkeys { + let hex_pk = normalize_pubkey(pk) + .with_context(|| format!("invalid MARMOT_ALLOWED_PUBKEYS entry: {}", pk))?; + cmd.arg("--allow-pubkey").arg(&hex_pk); + } } cmd.stdin(Stdio::piped()) From d8d6cb1d43495aaca0a815a40979ea3238260900 Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 18:34:57 -0600 Subject: [PATCH 04/11] Kill marmotd child process on MarmotClient drop Store the Child handle inside MarmotClient and implement Drop to kill + wait on the process, preventing orphaned marmotd daemons on shutdown. Mirrors Signal's subprocess cleanup pattern. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- crates/sage-core/src/main.rs | 2 +- crates/sage-core/src/marmot.rs | 17 +++++++++++++---- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/crates/sage-core/src/main.rs b/crates/sage-core/src/main.rs index 3b8eb59..f3d7daa 100644 --- a/crates/sage-core/src/main.rs +++ b/crates/sage-core/src/main.rs @@ -205,7 +205,7 @@ async fn main() -> Result<()> { info!(" Relays: {:?}", marmot_config.relays); info!(" State dir: {}", marmot_config.state_dir); - let (client, stdout, _child) = marmot::spawn_marmot(&marmot_config)?; + let (client, stdout) = marmot::spawn_marmot(&marmot_config)?; let writer = marmot::writer_handle(&client); let group_routes = marmot::group_routes_handle(&client); let messenger: Arc> = Arc::new(Mutex::new(client)); diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs index ae50d80..fb5e994 100644 --- a/crates/sage-core/src/marmot.rs +++ b/crates/sage-core/src/marmot.rs @@ -82,6 +82,16 @@ pub struct MarmotClient { /// to route per-group (each group ID = separate agent thread) while still /// sharing a parent identity for cross-thread memory. group_routes: Arc>>, + child: Mutex, +} + +impl Drop for MarmotClient { + fn drop(&mut self) { + if let Ok(mut child) = self.child.lock() { + let _ = child.kill(); + let _ = child.wait(); + } + } } impl MarmotClient { @@ -159,9 +169,7 @@ impl Messenger for MarmotClient { } /// Spawn marmotd daemon and return the client, stdout reader, and child process handle. -pub fn spawn_marmot( - config: &MarmotConfig, -) -> Result<(MarmotClient, std::process::ChildStdout, Child)> { +pub fn spawn_marmot(config: &MarmotConfig) -> Result<(MarmotClient, std::process::ChildStdout)> { let mut cmd = Command::new(&config.binary_path); cmd.arg("daemon"); @@ -218,9 +226,10 @@ pub fn spawn_marmot( writer: writer.clone(), request_id: AtomicU64::new(1), group_routes, + child: Mutex::new(child), }; - Ok((client, stdout, child)) + Ok((client, stdout)) } /// Run the marmot receive loop: waits for daemon ready, publishes keypackage, From 7c1fe4747630016110e46a0a7ac61556095aeffb Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 18:36:18 -0600 Subject: [PATCH 05/11] Bump marmotd to v0.5.1 Adds send_typing support (no more unknown variant errors), --version flag, audio file streaming, and video call rejection. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- Dockerfile | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Dockerfile b/Dockerfile index 3a832c8..d09523a 100644 --- a/Dockerfile +++ b/Dockerfile @@ -42,7 +42,7 @@ RUN cargo build --release # Stage 4: Build marmotd (optional MLS messaging sidecar) FROM chef AS marmotd-builder -ARG MARMOTD_VERSION=0.3.2 +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 From b002e3e17d8c238c7903a584078f64a2c7933c2f Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 18:45:41 -0600 Subject: [PATCH 06/11] Fix stale comments: reply_to is sender pubkey, not group ID Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- crates/sage-core/src/main.rs | 2 +- crates/sage-core/src/messenger.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/sage-core/src/main.rs b/crates/sage-core/src/main.rs index f3d7daa..1c67e16 100644 --- a/crates/sage-core/src/main.rs +++ b/crates/sage-core/src/main.rs @@ -328,7 +328,7 @@ async fn main() -> Result<()> { // Get or create agent for this conversation // For Signal: keyed by user UUID (reply_to == source) - // For Marmot: keyed by group ID (reply_to == nostr_group_id) + // For Marmot: keyed by sender pubkey (reply_to == from_pubkey) let (agent_id, agent) = match agent_manager.get_or_create_agent( &msg.reply_to, context_type, diff --git a/crates/sage-core/src/messenger.rs b/crates/sage-core/src/messenger.rs index 05ed075..0584199 100644 --- a/crates/sage-core/src/messenger.rs +++ b/crates/sage-core/src/messenger.rs @@ -19,7 +19,7 @@ pub struct IncomingMessage { pub attachments: Vec, #[allow(dead_code)] pub timestamp: u64, - /// Where to send replies (same as source for Signal, nostr_group_id for Marmot) + /// Identity key for agent lookup and reply routing (Signal UUID or Marmot pubkey) pub reply_to: String, } From afb4987b57e885ebc4377630bb1e0490e8032b51 Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 19:11:44 -0600 Subject: [PATCH 07/11] Persist Marmot reply routes in PostgreSQL for restart survival Add reply_context column to chat_contexts table to store transport- specific routing info (e.g. Marmot nostr_group_id for a pubkey). Updated on each incoming message, restored into group_routes on startup. Ensures scheduled tasks can deliver after restarts. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- .../down.sql | 1 + .../up.sql | 1 + crates/sage-core/src/agent_manager.rs | 39 +++++++++++++++++++ crates/sage-core/src/main.rs | 23 +++++++++++ crates/sage-core/src/marmot.rs | 1 + crates/sage-core/src/messenger.rs | 3 ++ crates/sage-core/src/schema.rs | 1 + crates/sage-core/src/signal.rs | 1 + 8 files changed, 70 insertions(+) create mode 100644 crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/down.sql create mode 100644 crates/sage-core/migrations/2026-02-22-010953-0000_add_reply_context_to_chat_contexts/up.sql 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/main.rs b/crates/sage-core/src/main.rs index 1c67e16..5480e17 100644 --- a/crates/sage-core/src/main.rs +++ b/crates/sage-core/src/main.rs @@ -208,6 +208,22 @@ async fn main() -> Result<()> { let (client, stdout) = marmot::spawn_marmot(&marmot_config)?; let writer = marmot::writer_handle(&client); let group_routes = marmot::group_routes_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 messenger: Arc> = Arc::new(Mutex::new(client)); let receive_handle = tokio::spawn(async move { @@ -343,6 +359,13 @@ async fn main() -> Result<()> { info!("Using agent {} for user {}", agent_id, user_name); + // 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 = messenger.lock().await; diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs index fb5e994..3b8ebee 100644 --- a/crates/sage-core/src/marmot.rs +++ b/crates/sage-core/src/marmot.rs @@ -420,6 +420,7 @@ pub async fn run_marmot_receive_loop( 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() { diff --git a/crates/sage-core/src/messenger.rs b/crates/sage-core/src/messenger.rs index 0584199..2d34679 100644 --- a/crates/sage-core/src/messenger.rs +++ b/crates/sage-core/src/messenger.rs @@ -21,6 +21,9 @@ pub struct IncomingMessage { 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 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 920ac30..4b917a0 100644 --- a/crates/sage-core/src/signal.rs +++ b/crates/sage-core/src/signal.rs @@ -438,6 +438,7 @@ pub fn parse_incoming_message(line: &str) -> Option { message: message.to_string(), attachments, timestamp, + reply_context: None, }) } From 13a114403cdfaffcabb4620cd91248fcbf0f4696 Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 20:33:04 -0600 Subject: [PATCH 08/11] Document Marmot/Pika as alternative messaging provider in README Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- README.md | 57 +++++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 51 insertions(+), 6 deletions(-) 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 From 0c571264187e6c6d7e26ff7b3404a701321e978d Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 20:45:10 -0600 Subject: [PATCH 09/11] Gracefully skip non-JSON stdout lines during marmotd startup phases Phases 1 (wait for ready) and 2 (keypackage publish) now use the same continue-on-parse-error pattern as Phase 3, preventing silent receive loop death if marmotd outputs non-JSON lines during startup. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- crates/sage-core/src/marmot.rs | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs index 3b8ebee..e842fad 100644 --- a/crates/sage-core/src/marmot.rs +++ b/crates/sage-core/src/marmot.rs @@ -259,8 +259,13 @@ pub async fn run_marmot_receive_loop( if reader.read_line(&mut line)? == 0 { return Err(anyhow!("marmotd closed stdout before ready")); } - let event: serde_json::Value = serde_json::from_str(line.trim()) - .with_context(|| format!("parse marmotd json: {}", line.trim()))?; + 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 @@ -289,8 +294,13 @@ pub async fn run_marmot_receive_loop( if reader.read_line(&mut line)? == 0 { return Err(anyhow!("marmotd closed stdout during keypackage publish")); } - let event: serde_json::Value = serde_json::from_str(line.trim()) - .with_context(|| format!("parse marmotd json: {}", line.trim()))?; + 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" From 7e2825d27456b6ab9d7346d4b4aa17196fd84c47 Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 20:46:48 -0600 Subject: [PATCH 10/11] Add supervisor loop with exponential backoff for marmotd receive Respawns marmotd on any failure (process exit, read error) with exponential backoff (250ms to 60s). Each restart re-initializes stdin/stdout handles, kills the old process, and re-runs all three phases (ready, keypackage, message loop). Mirrors Signal TCP mode's supervisor pattern. Stops retrying only when the message channel is closed (Sage shutting down). Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- crates/sage-core/src/main.rs | 6 +- crates/sage-core/src/marmot.rs | 541 ++++++++++++++++++++------------- 2 files changed, 332 insertions(+), 215 deletions(-) diff --git a/crates/sage-core/src/main.rs b/crates/sage-core/src/main.rs index 5480e17..6d802f9 100644 --- a/crates/sage-core/src/main.rs +++ b/crates/sage-core/src/main.rs @@ -205,9 +205,10 @@ async fn main() -> Result<()> { info!(" Relays: {:?}", marmot_config.relays); info!(" State dir: {}", marmot_config.state_dir); - let (client, stdout) = marmot::spawn_marmot(&marmot_config)?; + let (client, _initial_stdout) = marmot::spawn_marmot(&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() { @@ -226,8 +227,9 @@ async fn main() -> Result<()> { let messenger: Arc> = Arc::new(Mutex::new(client)); + // Supervisor loop: respawns marmotd on failure with exponential backoff let receive_handle = tokio::spawn(async move { - marmot::run_marmot_receive_loop(stdout, writer, tx, marmot_config, group_routes) + marmot::run_marmot_receive_loop(tx, marmot_config, group_routes, writer, child) .await }); diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs index e842fad..265b371 100644 --- a/crates/sage-core/src/marmot.rs +++ b/crates/sage-core/src/marmot.rs @@ -64,6 +64,7 @@ pub fn normalize_pubkey(input: &str) -> Result { } #[derive(Debug, Clone)] +#[allow(dead_code)] pub struct MarmotConfig { pub binary_path: String, pub relays: Vec, @@ -82,7 +83,7 @@ pub struct MarmotClient { /// to route per-group (each group ID = separate agent thread) while still /// sharing a parent identity for cross-thread memory. group_routes: Arc>>, - child: Mutex, + child: Arc>, } impl Drop for MarmotClient { @@ -226,246 +227,355 @@ pub fn spawn_marmot(config: &MarmotConfig) -> Result<(MarmotClient, std::process writer: writer.clone(), request_id: AtomicU64::new(1), group_routes, - child: Mutex::new(child), + child: Arc::new(Mutex::new(child)), }; Ok((client, stdout)) } -/// Run the marmot receive loop: waits for daemon ready, publishes keypackage, -/// then listens for incoming messages and auto-accepts welcomes. -pub async fn run_marmot_receive_loop( - stdout: std::process::ChildStdout, - writer: Arc>>, - tx: mpsc::Sender, - config: MarmotConfig, - group_routes: Arc>>, +/// 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<()> { - tokio::task::spawn_blocking(move || { - let mut reader = BufReader::new(stdout); - let mut line = String::new(); - - let send_cmd = |cmd: serde_json::Value| -> Result<()> { - let mut w = 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(()) - }; - - // 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; + // 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()); - // 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; + 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_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; + }; + + 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..."); + info!("Marmot receive loop started, listening for messages..."); - // Phase 3: Main receive loop - loop { - line.clear(); - match reader.read_line(&mut line) { - Ok(0) => { - warn!("marmotd closed stdout"); - break; - } - 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); - } + // 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); + } + "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; } - "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 preview_end = { + let max_len = 100.min(content.len()); + let mut end = max_len; + while end > 0 && !content.is_char_boundary(end) { + end -= 1; } - - 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"); - break; + 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()); } } - "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()); + + 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")); } } - } - Err(e) => { - error!("Error reading from marmotd: {}", e); - break; + "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)); + } } + } +} - warn!("Marmot receive loop ended"); - Ok::<_, anyhow::Error>(()) - }) - .await??; +/// 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 + ); + } + } - Ok(()) + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(backoff_max); + } } /// Get the shared writer handle from a MarmotClient (for the receive loop). @@ -478,6 +588,11 @@ pub fn group_routes_handle(client: &MarmotClient) -> Arc Arc> { + client.child.clone() +} + #[cfg(test)] mod tests { use super::*; From 43619053e3e742437c81f78134539d4d46f1b890 Mon Sep 17 00:00:00 2001 From: Tony Giorgio Date: Sat, 21 Feb 2026 21:01:22 -0600 Subject: [PATCH 11/11] Fix double-spawn: create MarmotClient without spawning marmotd Replace spawn_marmot() with new_marmot_client() which only validates config and creates placeholder handles. The supervisor loop is now solely responsible for spawning marmotd, eliminating the race where two instances ran against the same state directory. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- crates/sage-core/src/main.rs | 2 +- crates/sage-core/src/marmot.rs | 68 ++++++++++------------------------ 2 files changed, 21 insertions(+), 49 deletions(-) diff --git a/crates/sage-core/src/main.rs b/crates/sage-core/src/main.rs index 6d802f9..0b2145e 100644 --- a/crates/sage-core/src/main.rs +++ b/crates/sage-core/src/main.rs @@ -205,7 +205,7 @@ async fn main() -> Result<()> { info!(" Relays: {:?}", marmot_config.relays); info!(" State dir: {}", marmot_config.state_dir); - let (client, _initial_stdout) = marmot::spawn_marmot(&marmot_config)?; + 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); diff --git a/crates/sage-core/src/marmot.rs b/crates/sage-core/src/marmot.rs index 265b371..d050601 100644 --- a/crates/sage-core/src/marmot.rs +++ b/crates/sage-core/src/marmot.rs @@ -169,68 +169,40 @@ impl Messenger for MarmotClient { } } -/// Spawn marmotd daemon and return the client, stdout reader, and child process handle. -pub fn spawn_marmot(config: &MarmotConfig) -> Result<(MarmotClient, std::process::ChildStdout)> { - 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); - +/// 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 { - let hex_pk = normalize_pubkey(pk) + normalize_pubkey(pk) .with_context(|| format!("invalid MARMOT_ALLOWED_PUBKEYS entry: {}", 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 + // 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 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); - } - }); + .context("Failed to get placeholder stdin")?; let writer = Arc::new(Mutex::new(BufWriter::new(stdin))); - let group_routes = Arc::new(Mutex::new(HashMap::new())); - let client = MarmotClient { + + Ok(MarmotClient { writer: writer.clone(), request_id: AtomicU64::new(1), group_routes, - child: Arc::new(Mutex::new(child)), - }; - - Ok((client, stdout)) + child: Arc::new(Mutex::new(placeholder)), + }) } /// Single iteration of the marmot receive loop: spawn marmotd, run through