feat(mesh): '!ai' over mesh runs the assistant's shared tool loop

Mesh AssistQuery answered with a bare LLM call — no tools, no actions.
The CallerScope::Mesh variant was designed for this wiring ('the variant
exists so the shape is right when a future plan wires mesh callers into
the shared loop'); this is that plan. A trusted/allowlisted asker's prompt
now runs assistant::chat with CallerScope::Mesh { authorized } — the
operator's persisted grants cap what the model may touch (never wider),
and writes suspend on the node's own confirm gate. The reply is capped
for airtime as before, with a brevity instruction for mesh turns.

Wiring follows the blob_store pattern: RpcHandler::set_mesh_service (now
&Arc<Self>) forward-propagates an Arc<RpcHandler> into the mesh state's
new assistant_handler slot; absent (early boot) falls back to the legacy
bare-LLM answer.

Test: mesh_caller_authority_is_capped_at_operator_grants.

Co-Authored-By: Claude <noreply@anthropic.com>

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

@@ -200,13 +200,18 @@ impl RpcHandler {
     }
 
     /// Set the mesh service (called after identity is loaded).
-    pub async fn set_mesh_service(&self, service: crate::mesh::MeshService) {
+    pub async fn set_mesh_service(self: &Arc<Self>, service: crate::mesh::MeshService) {
         // If the blob store is already initialised, propagate it into the
         // freshly-started mesh state so the listener can persist inline
         // attachments. Mirrors `set_blob_store`'s forward-propagation.
         if let Some(store) = self.blob_store.read().await.as_ref().cloned() {
             *service.shared_state().blob_store.write().await = Some(store);
         }
+        // Wire the mesh `!ai` path into the assistant's shared tool loop: a
+        // trusted mesh peer's question runs the same tool registry with the
+        // operator's persisted grants, and writes still suspend on this
+        // node's confirm gate. Same forward-propagation pattern as above.
+        *service.shared_state().assistant_handler.write().await = Some(Arc::clone(self));
         *self.mesh_service.write().await = Some(service);
     }
 

core/archipelago/src/assistant/mod.rs

@@ -944,6 +944,16 @@ pub async fn chat_with_surfaces(
     tracing::info!(backend = %backend_id, "assistant.chat: backend selected for this turn");
 
     let system_prompt = build_system_prompt(&visible_tools, &disabled_tools);
+    // Radio airtime is scarce: a mesh caller's answer must stay short.
+    let system_prompt = if matches!(caller, CallerScope::Mesh { .. }) {
+        format!(
+            "{system_prompt}\n\nThis question arrives over a low-bandwidth radio mesh. Reply \
+             in at most two short sentences, no markdown, no preamble — and if a write was \
+             confirmed and started, say only that it is underway."
+        )
+    } else {
+        system_prompt
+    };
 
     let key = history::HistoryKey::from_caller(&caller);
 
@@ -1297,6 +1307,41 @@ mod tests {
         );
     }
 
+    /// Mesh callers (#50 wiring): an AUTHORIZED mesh peer inherits exactly
+    /// the operator's persisted grants — never more — and an unauthorized
+    /// one gets nothing, even when the operator has categories open.
+    #[tokio::test]
+    async fn mesh_caller_authority_is_capped_at_operator_grants() {
+        let (handler, _tmp) = test_rpc_handler().await;
+        // Open one category as the operator would.
+        let mut grants = crate::assistant::grants::Grants::load(handler.data_dir()).await;
+        grants.set(PermissionCategory::Apps, true);
+        grants
+            .save(handler.data_dir())
+            .await
+            .expect("persist test grants");
+
+        let authorized = CallerScope::Mesh {
+            peer_id: "peer-a".to_string(),
+            authorized: true,
+        };
+        let granted = authorized.granted_categories(handler.data_dir()).await;
+        assert!(granted.contains(&PermissionCategory::Apps));
+        assert!(!granted.contains(&PermissionCategory::Media));
+
+        let unauthorized = CallerScope::Mesh {
+            peer_id: "peer-b".to_string(),
+            authorized: false,
+        };
+        assert!(
+            unauthorized
+                .granted_categories(handler.data_dir())
+                .await
+                .is_empty(),
+            "an unauthorized mesh caller must resolve to zero grants"
+        );
+    }
+
     /// The prompt's AVAILABLE section must list only granted-category tools.
     /// Ungranted tools appear exclusively under DISABLED — listed so the
     /// model's call attempt hits the execution gate and records the refusal

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

@@ -129,10 +129,30 @@ pub(super) async fn run_assist(
 
     info!(from = asker, req_id, backend = %backend, model = %model, "Answering AI query over mesh");
 
-    let result = if is_claude {
-        call_claude(&state.data_dir, &model, &prompt).await
-    } else {
-        call_ollama(&model, &prompt).await
+    // Same tool surface as the embedded assistant (task: mesh `!ai` must
+    // ACTION, not just chat): when the server has wired the shared loop in,
+    // the asker's prompt runs through `assistant::chat` with a Mesh caller
+    // scope — the operator's persisted grants gate what the model may touch,
+    // and any write suspends on the node's confirm gate for the operator to
+    // approve. The asker passed `is_sender_allowed` above, so `authorized`
+    // is true here. Without the handler (early boot), the legacy bare-LLM
+    // path still answers.
+    let shared = state.assistant_handler.read().await.clone();
+    let result = match shared {
+        Some(handler) => {
+            let caller = crate::assistant::CallerScope::Mesh {
+                peer_id: asker.to_string(),
+                authorized: true,
+            };
+            crate::assistant::chat(handler, caller, prompt.clone()).await
+        }
+        None => {
+            if is_claude {
+                call_claude(&state.data_dir, &model, &prompt).await
+            } else {
+                call_ollama(&model, &prompt).await
+            }
+        }
     };
 
     match result {

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

@@ -223,6 +223,11 @@ pub struct MeshState {
     /// by `RpcHandler` after startup so the mesh listener can persist inline
     /// file bytes into the same store the HTTP layer serves.
     pub blob_store: RwLock<Option<Arc<crate::blobs::BlobStore>>>,
+    /// The assistant's shared tool loop, reached through the same RpcHandler
+    /// every caller dispatches on. Populated by the server after startup
+    /// (same pattern as blob_store). `None` on early boot → the legacy
+    /// bare-LLM path in assist.rs answers instead.
+    pub assistant_handler: RwLock<Option<Arc<crate::api::rpc::RpcHandler>>>,
     /// Firmware-pubkey-hex of radio contacts the user has chosen to ignore
     /// (via mesh.clear-all). `refresh_contacts` skips any device contact
     /// whose pubkey is in this set, preventing the meshcore firmware's
@@ -354,6 +359,7 @@ impl MeshState {
             contacts: RwLock::new(HashMap::new()),
             our_ed_pubkey_hex,
             blob_store: RwLock::new(None),
+            assistant_handler: RwLock::new(None),
             radio_contact_blocklist: RwLock::new(HashSet::new()),
             assistant: RwLock::new(assistant),
             data_dir,