feat(mesh): mesh-AI assistant scheduler + config panel (#50)

Adds the assistant scheduler, MeshAssistantPanel UI, and the remaining
config-RPC / live-toggle / Ollama-detect wiring on top of Phase 1.x.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

core/archipelago/src/api/rpc/dispatcher.rs

@@ -383,6 +383,9 @@ impl RpcHandler {
             "mesh.deadman-checkin" => self.handle_mesh_deadman_checkin().await,
             "mesh.assistant-status" => self.handle_mesh_assistant_status().await,
             "mesh.assistant-configure" => self.handle_mesh_assistant_configure(params).await,
+            "mesh.schedule-message" => self.handle_mesh_schedule_message(params).await,
+            "mesh.list-scheduled" => self.handle_mesh_list_scheduled().await,
+            "mesh.cancel-scheduled" => self.handle_mesh_cancel_scheduled(params).await,
             "mesh.test-send" => self.handle_mesh_test_send(params).await,
 
             // Transport layer (unified routing)

core/archipelago/src/api/rpc/mesh/assistant.rs

@@ -23,12 +23,19 @@ impl RpcHandler {
         };
 
         let (ollama_detected, models) = detect_ollama().await;
+        let claude_available = tokio::fs::metadata(
+            self.config.data_dir.join("secrets/claude-api-key"),
+        )
+        .await
+        .is_ok();
         Ok(serde_json::json!({
             "enabled": cfg.enabled,
             "model": cfg.model,
             "trusted_only": cfg.trusted_only,
+            "backend": cfg.backend,
             "default_model": DEFAULT_MODEL,
             "ollama_detected": ollama_detected,
+            "claude_available": claude_available,
             "models": models,
         }))
     }
@@ -48,6 +55,10 @@ impl RpcHandler {
 
         let enabled = params.get("enabled").and_then(|v| v.as_bool());
         let trusted_only = params.get("trusted_only").and_then(|v| v.as_bool());
+        let backend = params
+            .get("backend")
+            .and_then(|v| v.as_str())
+            .map(|s| s.to_string());
         // model: key present + string => set; present + null => clear; absent => leave
         let model = if let Some(v) = params.get("model") {
             Some(v.as_str().map(|s| s.to_string()))
@@ -55,14 +66,77 @@ impl RpcHandler {
             None
         };
 
-        svc.configure_assistant(enabled, model, trusted_only).await?;
+        svc.configure_assistant(enabled, model, trusted_only, backend)
+            .await?;
         let cfg = svc.assistant_config().await;
         Ok(serde_json::json!({
             "enabled": cfg.enabled,
             "model": cfg.model,
             "trusted_only": cfg.trusted_only,
+            "backend": cfg.backend,
         }))
     }
+
+    /// mesh.schedule-message — queue a message to send at a future time.
+    /// Params: `body: string`, `fire_at: i64` (unix secs), and one of
+    /// `contact_id: u32` (DM) or `channel: u8` (broadcast).
+    pub(in crate::api::rpc) async fn handle_mesh_schedule_message(
+        &self,
+        params: Option<serde_json::Value>,
+    ) -> Result<serde_json::Value> {
+        let p = params.ok_or_else(|| anyhow::anyhow!("Missing params"))?;
+        let body = p
+            .get("body")
+            .and_then(|v| v.as_str())
+            .ok_or_else(|| anyhow::anyhow!("body is required"))?
+            .to_string();
+        let fire_at = p
+            .get("fire_at")
+            .and_then(|v| v.as_i64())
+            .ok_or_else(|| anyhow::anyhow!("fire_at (unix seconds) is required"))?;
+        let contact_id = p.get("contact_id").and_then(|v| v.as_u64()).map(|v| v as u32);
+        let channel = p.get("channel").and_then(|v| v.as_u64()).map(|v| v as u8);
+        if contact_id.is_none() && channel.is_none() {
+            anyhow::bail!("either contact_id or channel is required");
+        }
+
+        let service = self.mesh_service.read().await;
+        let svc = service
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
+        let msg = svc.scheduler.add(contact_id, channel, body, fire_at).await?;
+        Ok(serde_json::to_value(msg)?)
+    }
+
+    /// mesh.list-scheduled — list queued messages (sorted by fire time).
+    pub(in crate::api::rpc) async fn handle_mesh_list_scheduled(
+        &self,
+    ) -> Result<serde_json::Value> {
+        let service = self.mesh_service.read().await;
+        let svc = service
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
+        let messages = svc.scheduler.list().await;
+        Ok(serde_json::json!({ "messages": messages }))
+    }
+
+    /// mesh.cancel-scheduled — remove a queued message by id.
+    pub(in crate::api::rpc) async fn handle_mesh_cancel_scheduled(
+        &self,
+        params: Option<serde_json::Value>,
+    ) -> Result<serde_json::Value> {
+        let id = params
+            .as_ref()
+            .and_then(|p| p.get("id"))
+            .and_then(|v| v.as_u64())
+            .ok_or_else(|| anyhow::anyhow!("id is required"))?;
+        let service = self.mesh_service.read().await;
+        let svc = service
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("Mesh service not running"))?;
+        let cancelled = svc.scheduler.cancel(id).await?;
+        Ok(serde_json::json!({ "cancelled": cancelled }))
+    }
 }
 
 /// Probe the local Ollama HTTP API; return (detected, model_names).

core/archipelago/src/mesh/listener/assist.rs

@@ -11,6 +11,7 @@ use super::super::message_types::{self, AssistResponsePayload, MeshMessageType};
 use super::bitcoin::send_to_peer;
 use super::{MeshCommand, MeshState};
 use crate::federation::TrustLevel;
+use std::path::Path;
 use std::sync::Arc;
 use std::time::Duration;
 use tracing::{info, warn};
@@ -19,6 +20,10 @@ use tracing::{info, warn};
 const OLLAMA_URL: &str = "http://localhost:11434/api/generate";
 /// Default model when the node hasn't configured one (matches Meshroller).
 const DEFAULT_MODEL: &str = "qwen2.5-coder";
+/// Anthropic Messages API (called with the shared proxy token).
+const CLAUDE_URL: &str = "https://api.anthropic.com/v1/messages";
+/// Default Claude model — Haiku 4.5: fast + cheap, ideal for short mesh answers.
+const CLAUDE_DEFAULT_MODEL: &str = "claude-haiku-4-5-20251001";
 /// Max time to wait on the model before giving up.
 const OLLAMA_TIMEOUT: Duration = Duration::from_secs(60);
 /// Hard cap on answer length sent over the radio — keeps airtime sane.
@@ -82,14 +87,23 @@ pub(super) async fn run_assist(
         prompt: prompt.clone(),
     });
 
-    let configured_model = state.assistant.read().await.model.clone();
+    let (backend, configured_model) = {
+        let a = state.assistant.read().await;
+        (a.backend.clone(), a.model.clone())
+    };
+    let is_claude = backend == "claude";
+    let default_model = if is_claude { CLAUDE_DEFAULT_MODEL } else { DEFAULT_MODEL };
     let model = model_override
         .or(configured_model)
-        .unwrap_or_else(|| DEFAULT_MODEL.to_string());
+        .unwrap_or_else(|| default_model.to_string());
 
-    info!(from = asker, req_id, model = %model, "Answering AI query over mesh");
+    info!(from = asker, req_id, backend = %backend, model = %model, "Answering AI query over mesh");
 
-    let result = call_ollama(&model, &prompt).await;
+    let result = if is_claude {
+        call_claude(&state.data_dir, &model, &prompt).await
+    } else {
+        call_ollama(&model, &prompt).await
+    };
 
     match result {
         Ok(answer) => {
@@ -283,3 +297,51 @@ async fn call_ollama(model: &str, prompt: &str) -> anyhow::Result<String> {
     }
     Ok(text)
 }
+
+/// Call Claude via the Anthropic Messages API using the node's shared proxy
+/// token at `secrets/claude-api-key`. Keeps answers short for radio airtime.
+async fn call_claude(data_dir: &Path, model: &str, prompt: &str) -> anyhow::Result<String> {
+    let key = tokio::fs::read_to_string(data_dir.join("secrets/claude-api-key"))
+        .await
+        .map_err(|_| anyhow::anyhow!("Claude API key not configured on this node"))?;
+    let key = key.trim();
+    if key.is_empty() {
+        anyhow::bail!("Claude API key is empty");
+    }
+    let client = reqwest::Client::builder().timeout(OLLAMA_TIMEOUT).build()?;
+    let body = serde_json::json!({
+        "model": model,
+        "max_tokens": 512,
+        "system": "You answer questions over a low-bandwidth radio mesh. Reply in at most two short sentences. No markdown, no preamble.",
+        "messages": [{ "role": "user", "content": prompt }],
+    });
+    let resp = client
+        .post(CLAUDE_URL)
+        .header("x-api-key", key)
+        .header("anthropic-version", "2023-06-01")
+        .header("content-type", "application/json")
+        .json(&body)
+        .send()
+        .await?;
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let txt = resp.text().await.unwrap_or_default();
+        anyhow::bail!(
+            "Claude API HTTP {}: {}",
+            status,
+            txt.chars().take(180).collect::<String>()
+        );
+    }
+    let json: serde_json::Value = resp.json().await?;
+    // `content` is an array of blocks; take the first text block.
+    let text = json
+        .get("content")
+        .and_then(|c| c.as_array())
+        .and_then(|arr| arr.iter().find_map(|b| b.get("text").and_then(|t| t.as_str())))
+        .unwrap_or("")
+        .to_string();
+    if text.trim().is_empty() {
+        anyhow::bail!("Claude returned an empty response");
+    }
+    Ok(text)
+}

core/archipelago/src/mesh/listener/mod.rs

@@ -140,10 +140,12 @@ pub struct MeshState {
 pub struct AssistantConfig {
     /// Answer AssistQuery messages with the local LLM.
     pub enabled: bool,
-    /// Ollama model to use; None → the built-in default.
+    /// Model to use; None → the backend's built-in default.
     pub model: Option<String>,
     /// Restrict asking to federation-Trusted peers (vs. anyone on the mesh).
     pub trusted_only: bool,
+    /// AI backend: "claude" (shared proxy token) or "ollama" (local model).
+    pub backend: String,
 }
 
 /// Contact metadata kept alongside MeshState.peers. Pinned contacts sort to

core/archipelago/src/mesh/mod.rs

@@ -14,6 +14,7 @@ pub mod message_types;
 pub mod outbox;
 pub mod protocol;
 pub mod ratchet;
+pub mod scheduler;
 pub mod serial;
 pub mod session;
 pub mod steganography;
@@ -187,6 +188,15 @@ pub struct MeshConfig {
     /// any peer on the mesh may ask (spends this node's compute + airtime).
     #[serde(default = "default_true")]
     pub assistant_trusted_only: bool,
+    /// Which AI backend answers queries: "claude" (the shared Claude proxy
+    /// token at secrets/claude-api-key — default for now, works without a
+    /// local GPU) or "ollama" (a local model on this node).
+    #[serde(default = "default_assistant_backend")]
+    pub assistant_backend: String,
+}
+
+fn default_assistant_backend() -> String {
+    "claude".to_string()
 }
 
 fn default_true() -> bool {
@@ -208,6 +218,7 @@ impl Default for MeshConfig {
             assistant_enabled: false,
             assistant_model: None,
             assistant_trusted_only: true,
+            assistant_backend: default_assistant_backend(),
         }
     }
 }
@@ -336,6 +347,7 @@ pub struct MeshService {
     deadman_handle: Option<tokio::task::JoinHandle<()>>,
     block_announcer_handle: Option<tokio::task::JoinHandle<()>>,
     presence_handle: Option<tokio::task::JoinHandle<()>>,
+    scheduler_handle: Option<tokio::task::JoinHandle<()>>,
     cmd_rx: Option<tokio::sync::mpsc::Receiver<listener::MeshCommand>>,
     // Crypto identity for this node
     our_did: String,
@@ -349,6 +361,8 @@ pub struct MeshService {
     pub block_header_cache: Arc<BlockHeaderCache>,
     pub relay_tracker: Arc<RelayTracker>,
     pub dead_man_switch: Arc<DeadManSwitch>,
+    /// Scheduled / queued outbound mesh messages (issue #50, phase 1.7).
+    pub scheduler: Arc<scheduler::MeshScheduler>,
 }
 
 impl MeshService {
@@ -380,6 +394,7 @@ impl MeshService {
                 enabled: config.assistant_enabled,
                 model: config.assistant_model.clone(),
                 trusted_only: config.assistant_trusted_only,
+                backend: config.assistant_backend.clone(),
             },
             data_dir.to_path_buf(),
         );
@@ -438,6 +453,7 @@ impl MeshService {
             deadman_handle: None,
             block_announcer_handle: None,
             presence_handle: None,
+            scheduler_handle: None,
             cmd_rx: Some(cmd_rx),
             our_did: did.to_string(),
             our_ed_pubkey_hex: ed_pubkey_hex.to_string(),
@@ -448,6 +464,7 @@ impl MeshService {
             block_header_cache,
             relay_tracker,
             dead_man_switch,
+            scheduler: Arc::new(scheduler::MeshScheduler::load(data_dir).await),
         })
     }
 
@@ -525,6 +542,20 @@ impl MeshService {
         });
         self.deadman_handle = Some(dms_handle);
 
+        // Scheduled-message task (issue #50, phase 1.7): fires queued messages
+        // when due, retrying peer DMs until the peer is back in range.
+        let sched = Arc::clone(&self.scheduler);
+        let sched_state = Arc::clone(&self.state);
+        let sched_shutdown = self
+            .shutdown_tx
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("Shutdown channel not initialized"))?
+            .subscribe();
+        let sched_handle = tokio::spawn(async move {
+            scheduler::run_scheduler(sched, sched_state, sched_shutdown).await;
+        });
+        self.scheduler_handle = Some(sched_handle);
+
         // Spawn block header announcer (internet-connected nodes only)
         if self.config.announce_block_headers {
             let bha_state = Arc::clone(&self.state);
@@ -675,6 +706,10 @@ impl MeshService {
             handle.abort();
             let _ = handle.await;
         }
+        if let Some(handle) = self.scheduler_handle.take() {
+            handle.abort();
+            let _ = handle.await;
+        }
         if let Some(handle) = self.block_announcer_handle.take() {
             handle.abort();
             let _ = handle.await;
@@ -1359,6 +1394,7 @@ impl MeshService {
         enabled: Option<bool>,
         model: Option<Option<String>>,
         trusted_only: Option<bool>,
+        backend: Option<String>,
     ) -> Result<()> {
         {
             let mut a = self.state.assistant.write().await;
@@ -1371,6 +1407,9 @@ impl MeshService {
             if let Some(t) = trusted_only {
                 a.trusted_only = t;
             }
+            if let Some(b) = backend {
+                a.backend = b;
+            }
         }
         // Persist by updating the on-disk config (the in-memory `self.config`
         // snapshot stays as-is; the live `state.assistant` is the runtime
@@ -1381,6 +1420,7 @@ impl MeshService {
             cfg.assistant_enabled = a.enabled;
             cfg.assistant_model = a.model.clone();
             cfg.assistant_trusted_only = a.trusted_only;
+            cfg.assistant_backend = a.backend.clone();
         }
         save_config(&self.data_dir, &cfg).await?;
         Ok(())

core/archipelago/src/mesh/scheduler.rs

@@ -0,0 +1,213 @@
+//! Scheduled / queued mesh messages (issue #50, phase 1.7).
+//!
+//! A small persisted queue of messages to send at a future time. A background
+//! task fires due messages via the listener. A message addressed to a peer that
+//! isn't currently in the contact table stays queued and retries on later ticks
+//! — i.e. it sends itself when the peer comes back in range.
+
+use super::listener::{MeshCommand, MeshState};
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use tokio::fs;
+use tokio::sync::{watch, RwLock};
+use tracing::warn;
+
+const SCHEDULER_FILE: &str = "mesh-scheduled.json";
+/// Wake interval for firing due messages.
+const TICK_SECS: u64 = 10;
+/// Drop a still-undeliverable message after this many attempts (~1h at 10s).
+const MAX_ATTEMPTS: u32 = 360;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ScheduledMessage {
+    pub id: u64,
+    /// Direct-message target (peer contact_id), or None for a channel broadcast.
+    #[serde(default)]
+    pub contact_id: Option<u32>,
+    /// Channel to broadcast on, or None for a direct message.
+    #[serde(default)]
+    pub channel: Option<u8>,
+    pub body: String,
+    /// Unix seconds when the message becomes due.
+    pub fire_at: i64,
+    #[serde(default)]
+    pub attempts: u32,
+}
+
+pub struct MeshScheduler {
+    path: PathBuf,
+    queue: RwLock<Vec<ScheduledMessage>>,
+    next_id: RwLock<u64>,
+}
+
+impl MeshScheduler {
+    pub async fn load(data_dir: &Path) -> Self {
+        let path = data_dir.join(SCHEDULER_FILE);
+        let queue: Vec<ScheduledMessage> = match fs::read_to_string(&path).await {
+            Ok(s) => serde_json::from_str(&s).unwrap_or_default(),
+            Err(_) => Vec::new(),
+        };
+        let next = queue.iter().map(|m| m.id).max().unwrap_or(0) + 1;
+        Self {
+            path,
+            queue: RwLock::new(queue),
+            next_id: RwLock::new(next),
+        }
+    }
+
+    async fn save(&self) -> Result<()> {
+        let json = {
+            let q = self.queue.read().await;
+            serde_json::to_string_pretty(&*q).context("serialize scheduled queue")?
+        };
+        if let Some(parent) = self.path.parent() {
+            fs::create_dir_all(parent).await.ok();
+        }
+        fs::write(&self.path, json)
+            .await
+            .context("write scheduled queue")?;
+        Ok(())
+    }
+
+    pub async fn add(
+        &self,
+        contact_id: Option<u32>,
+        channel: Option<u8>,
+        body: String,
+        fire_at: i64,
+    ) -> Result<ScheduledMessage> {
+        let id = {
+            let mut n = self.next_id.write().await;
+            let id = *n;
+            *n += 1;
+            id
+        };
+        let msg = ScheduledMessage {
+            id,
+            contact_id,
+            channel,
+            body,
+            fire_at,
+            attempts: 0,
+        };
+        self.queue.write().await.push(msg.clone());
+        self.save().await?;
+        Ok(msg)
+    }
+
+    pub async fn list(&self) -> Vec<ScheduledMessage> {
+        let mut v = self.queue.read().await.clone();
+        v.sort_by_key(|m| m.fire_at);
+        v
+    }
+
+    pub async fn cancel(&self, id: u64) -> Result<bool> {
+        let removed = {
+            let mut q = self.queue.write().await;
+            let before = q.len();
+            q.retain(|m| m.id != id);
+            q.len() != before
+        };
+        if removed {
+            self.save().await?;
+        }
+        Ok(removed)
+    }
+}
+
+/// Background loop: every `TICK_SECS`, fire any due messages.
+pub async fn run_scheduler(
+    scheduler: Arc<MeshScheduler>,
+    state: Arc<MeshState>,
+    mut shutdown: watch::Receiver<bool>,
+) {
+    let mut interval = tokio::time::interval(std::time::Duration::from_secs(TICK_SECS));
+    loop {
+        tokio::select! {
+            _ = interval.tick() => fire_due(&scheduler, &state).await,
+            _ = shutdown.changed() => {
+                if *shutdown.borrow() { return; }
+            }
+        }
+    }
+}
+
+async fn fire_due(scheduler: &Arc<MeshScheduler>, state: &Arc<MeshState>) {
+    let now = chrono::Utc::now().timestamp();
+    let due: Vec<ScheduledMessage> = scheduler
+        .queue
+        .read()
+        .await
+        .iter()
+        .filter(|m| m.fire_at <= now)
+        .cloned()
+        .collect();
+    if due.is_empty() {
+        return;
+    }
+
+    let mut delivered: Vec<u64> = Vec::new();
+    let mut failed: Vec<u64> = Vec::new();
+    for msg in &due {
+        if try_send(state, msg).await {
+            delivered.push(msg.id);
+        } else {
+            failed.push(msg.id);
+        }
+    }
+
+    let mut to_remove = delivered;
+    {
+        let mut q = scheduler.queue.write().await;
+        for m in q.iter_mut() {
+            if failed.contains(&m.id) {
+                m.attempts += 1;
+                if m.attempts >= MAX_ATTEMPTS {
+                    warn!(id = m.id, attempts = m.attempts, "Dropping undeliverable scheduled message");
+                    to_remove.push(m.id);
+                }
+            }
+        }
+        q.retain(|m| !to_remove.contains(&m.id));
+    }
+    let _ = scheduler.save().await;
+}
+
+/// Hand a due message to the radio. Returns true if it was sent (or should be
+/// dropped); false to keep it queued for a later retry (peer not in range yet).
+async fn try_send(state: &Arc<MeshState>, msg: &ScheduledMessage) -> bool {
+    let payload = msg.body.clone().into_bytes();
+    if let Some(channel) = msg.channel {
+        return state
+            .send_cmd(MeshCommand::BroadcastChannel { channel, payload })
+            .await
+            .is_ok();
+    }
+    if let Some(contact_id) = msg.contact_id {
+        let pubkey = {
+            let peers = state.peers.read().await;
+            peers.get(&contact_id).and_then(|p| p.pubkey_hex.clone())
+        };
+        if let Some(pk) = pubkey {
+            if let Ok(bytes) = hex::decode(&pk) {
+                if bytes.len() >= 6 {
+                    let mut dest = [0u8; 6];
+                    dest.copy_from_slice(&bytes[..6]);
+                    return state
+                        .send_cmd(MeshCommand::SendText {
+                            dest_pubkey_prefix: dest,
+                            payload,
+                        })
+                        .await
+                        .is_ok();
+                }
+            }
+        }
+        // Peer unknown / not in range yet — keep queued, retry next tick.
+        return false;
+    }
+    warn!("Scheduled message has neither channel nor contact_id — dropping");
+    true
+}