Merge pull request #6545 from singlerider/feat/6272-multi-agent-runtime

feat(runtime): multi-agent runtime

Author: singlerider

crates/zeroclaw-api/src/channel.rs

@@ -144,6 +144,51 @@ pub trait Channel: Send + Sync {
         false
     }
 
+    /// Self-loop guard for multi-agent runs.
+    ///
+    /// Returns the bot's own handle/identity on this channel
+    /// (e.g. `@my_bot` for Telegram, the bot's user ID for Discord)
+    /// when known, so the orchestrator can drop inbound events whose
+    /// `sender` matches: a bot must never respond to its own
+    /// messages, even if a misconfigured peer group lists the bot's
+    /// handle as an external peer.
+    ///
+    /// **Channels that handle inbound traffic must override this.**
+    /// The default `None` makes both layers of the orchestrator's
+    /// self-loop guard (the SDK-side `drop_self_messages` here, and
+    /// the agent-loop fallback `peers::should_drop_self_loop`) into
+    /// no-ops — both layers consult the same `self_handle`, so a
+    /// channel that returns `None` has no protection from looping on
+    /// its own outbound. Outbound-only channels (webhook, gmail-push,
+    /// voice-call) never see inbound and can keep the default. The
+    /// in-tree overrides currently cover Telegram (`bot_username`
+    /// cache), IRC (configured nickname), Discord (decoded from token),
+    /// Slack (cached `auth.test` user_id); other inbound channels
+    /// remain on the default and rely on per-impl filtering instead
+    /// of the shared guard.
+    fn self_handle(&self) -> Option<String> {
+        None
+    }
+
+    /// Whether the orchestrator should drop an inbound message as
+    /// self-authored (multi-agent self-loop guard).
+    ///
+    /// Default implementation compares `msg.sender` against
+    /// [`Self::self_handle`] case-insensitively, after stripping a
+    /// leading `@` from each side so Telegram-style handles match
+    /// regardless of which form the SDK delivers. Override only for
+    /// platforms whose identity comparison is non-string (e.g. a
+    /// numeric Discord user ID is `as_str` already; this default
+    /// works there too).
+    fn drop_self_messages(&self, msg: &ChannelMessage) -> bool {
+        let Some(handle) = self.self_handle() else {
+            return false;
+        };
+        let handle_norm = handle.trim_start_matches('@').to_ascii_lowercase();
+        let sender_norm = msg.sender.trim_start_matches('@').to_ascii_lowercase();
+        !handle_norm.is_empty() && handle_norm == sender_norm
+    }
+
     /// Whether this channel supports multi-message streaming delivery.
     fn supports_multi_message_streaming(&self) -> bool {
         false
@@ -285,3 +330,87 @@ pub trait Channel: Send + Sync {
         true
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Stub channel that overrides `self_handle` so the default
+    /// `drop_self_messages` implementation can be exercised.
+    struct StubChannel {
+        handle: Option<String>,
+    }
+
+    #[async_trait]
+    impl Channel for StubChannel {
+        fn name(&self) -> &str {
+            "stub"
+        }
+        async fn send(&self, _message: &SendMessage) -> anyhow::Result<()> {
+            Ok(())
+        }
+        async fn listen(
+            &self,
+            _tx: tokio::sync::mpsc::Sender<ChannelMessage>,
+        ) -> anyhow::Result<()> {
+            Ok(())
+        }
+        fn self_handle(&self) -> Option<String> {
+            self.handle.clone()
+        }
+    }
+
+    fn msg_from(sender: &str) -> ChannelMessage {
+        ChannelMessage {
+            id: "1".into(),
+            sender: sender.into(),
+            reply_target: String::new(),
+            content: "hi".into(),
+            channel: "stub".into(),
+            timestamp: 0,
+            thread_ts: None,
+            interruption_scope_id: None,
+            attachments: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn drop_self_messages_default_returns_false_when_handle_unknown() {
+        let channel = StubChannel { handle: None };
+        assert!(!channel.drop_self_messages(&msg_from("@anyone")));
+    }
+
+    #[test]
+    fn drop_self_messages_matches_exact_handle() {
+        let channel = StubChannel {
+            handle: Some("@my_bot".into()),
+        };
+        assert!(channel.drop_self_messages(&msg_from("@my_bot")));
+        assert!(!channel.drop_self_messages(&msg_from("@other_bot")));
+    }
+
+    #[test]
+    fn drop_self_messages_normalizes_at_prefix_and_case() {
+        let channel = StubChannel {
+            handle: Some("My_Bot".into()),
+        };
+        // SDK delivered with @ prefix, handle stored without. Match.
+        assert!(channel.drop_self_messages(&msg_from("@my_bot")));
+        // Both with @, mixed case. Match.
+        let channel = StubChannel {
+            handle: Some("@My_Bot".into()),
+        };
+        assert!(channel.drop_self_messages(&msg_from("@MY_BOT")));
+    }
+
+    #[test]
+    fn drop_self_messages_does_not_match_empty_handle() {
+        // A handle of "@" (effectively empty after normalization) must
+        // not match every inbound message; the guard only fires when
+        // the bot has a real handle to compare against.
+        let channel = StubChannel {
+            handle: Some("@".into()),
+        };
+        assert!(!channel.drop_self_messages(&msg_from("@anyone")));
+    }
+}

crates/zeroclaw-api/src/memory_traits.rs

@@ -44,6 +44,15 @@ pub struct MemoryEntry {
     /// If this entry was superseded by a newer conflicting entry.
     #[serde(default)]
     pub superseded_by: Option<String>,
+    /// The UUID of the agent this row is attributed to, when the
+    /// backend tracks per-agent attribution. Backends without per-agent
+    /// columns (Markdown, Qdrant payload-less variants, None) leave
+    /// this `None`; SQL-backed stores populate it from the
+    /// `memories.agent_id` column. `AgentScopedMemory` uses this field
+    /// to enforce the bound + allowlist boundary on read paths the
+    /// trait does not expose an agent-aware form for (`get`, `list`).
+    #[serde(default)]
+    pub agent_id: Option<String>,
 }
 
 fn default_namespace() -> String {
@@ -61,6 +70,7 @@ impl std::fmt::Debug for MemoryEntry {
             .field("score", &self.score)
             .field("namespace", &self.namespace)
             .field("importance", &self.importance)
+            .field("agent_id", &self.agent_id)
             .finish_non_exhaustive()
     }
 }
@@ -289,6 +299,68 @@ pub trait Memory: Send + Sync {
     ) -> anyhow::Result<()> {
         self.store(key, content, category, session_id).await
     }
+
+    /// Store a memory entry attributed to an explicit agent UUID.
+    /// Every backend must implement this explicitly so the agent_id
+    /// is never silently dropped at storage time. Backends with
+    /// native agent_id columns (SqliteMemory, PostgresMemory,
+    /// LucidMemory) persist the attribution in SQL; MarkdownMemory
+    /// attributes via the per-agent directory path; QdrantMemory
+    /// persists in the vector payload; NoneMemory is a no-op stub.
+    /// `AgentScopedMemory` is the canonical caller.
+    async fn store_with_agent(
+        &self,
+        key: &str,
+        content: &str,
+        category: MemoryCategory,
+        session_id: Option<&str>,
+        namespace: Option<&str>,
+        importance: Option<f64>,
+        agent_id: Option<&str>,
+    ) -> anyhow::Result<()>;
+
+    /// Recall memory entries scoped to a specific set of agent UUIDs.
+    /// When `allowed_agent_ids` is non-empty, the backend filters its
+    /// result set to rows whose `agent_id` matches one of the listed
+    /// UUIDs (or is NULL, for legacy rows written before the agent_id
+    /// column existed). Every backend must implement this explicitly
+    /// so the allowlist is never silently dropped at read time.
+    ///
+    /// For SQL-backed stores the filter is `WHERE agent_id IN (...)`.
+    /// For Markdown the implementation walks the allowed agents'
+    /// per-agent directories. For Qdrant it's a payload filter on
+    /// the `agent_id` field. For None it returns an empty list.
+    /// `AgentScopedMemory` is the canonical caller; direct invocation
+    /// is also valid for read-only cross-agent queries that bypass
+    /// the wrapper.
+    ///
+    /// Cross-backend allowlist entries are rejected at config load
+    /// (`agents.<alias>.workspace.read_memory_from` cannot point at a
+    /// sibling on a different memory backend); backends therefore
+    /// never need to handle a cross-backend recall.
+    async fn recall_for_agents(
+        &self,
+        allowed_agent_ids: &[&str],
+        query: &str,
+        limit: usize,
+        session_id: Option<&str>,
+        since: Option<&str>,
+        until: Option<&str>,
+    ) -> anyhow::Result<Vec<MemoryEntry>>;
+
+    /// Look up (or create) the identifier the backend uses to refer
+    /// to the agent named by `alias`.
+    ///
+    /// Backends with an `agents` table (SqliteMemory, PostgresMemory,
+    /// LucidMemory) return the row's UUID, inserting if absent.
+    /// Backends without (MarkdownMemory, QdrantMemory, NoneMemory)
+    /// return the alias verbatim — there is no UUID indirection at
+    /// the storage layer, so the alias serves as the agent_id.
+    /// Default impl returns the alias unchanged; SQL backends
+    /// override to do the real lookup.
+    async fn ensure_agent_uuid(&self, alias: &str) -> anyhow::Result<String> {
+        Ok(alias.to_string())
+    }
 }
 
 #[cfg(test)]
@@ -339,6 +411,7 @@ mod tests {
             namespace: "default".into(),
             importance: Some(0.7),
             superseded_by: None,
+            agent_id: None,
         };
 
         let json = serde_json::to_string(&entry).unwrap();

crates/zeroclaw-api/src/observability_traits.rs

@@ -83,20 +83,6 @@ pub enum ObserverEvent {
         /// Human-readable error description. Must not contain secrets or tokens.
         message: String,
     },
-    /// A hand has started execution.
-    HandStarted { hand_name: String },
-    /// A hand has completed execution successfully.
-    HandCompleted {
-        hand_name: String,
-        duration_ms: u64,
-        findings_count: usize,
-    },
-    /// A hand has failed during execution.
-    HandFailed {
-        hand_name: String,
-        error: String,
-        duration_ms: u64,
-    },
     /// A deployment has started.
     DeploymentStarted {
         /// Identifier for the deployment (e.g., commit SHA or release tag).
@@ -132,15 +118,6 @@ pub enum ObserverMetric {
     ActiveSessions(u64),
     /// Current depth of the inbound message queue.
     QueueDepth(u64),
-    /// Duration of a single hand run.
-    HandRunDuration {
-        hand_name: String,
-        duration: Duration,
-    },
-    /// Number of findings produced by a hand run.
-    HandFindingsCount { hand_name: String, count: u64 },
-    /// Records a hand run outcome for success-rate tracking.
-    HandSuccessRate { hand_name: String, success: bool },
     /// Time elapsed from commit to deployment (lead time for changes).
     DeploymentLeadTime(Duration),
     /// Time elapsed to recover from a failed deployment.
@@ -261,67 +238,4 @@ mod tests {
         assert!(matches!(cloned_event, ObserverEvent::ToolCall { .. }));
         assert!(matches!(cloned_metric, ObserverMetric::RequestLatency(_)));
     }
-
-    #[test]
-    fn hand_events_recordable() {
-        let observer = DummyObserver::default();
-
-        observer.record_event(&ObserverEvent::HandStarted {
-            hand_name: "review".into(),
-        });
-        observer.record_event(&ObserverEvent::HandCompleted {
-            hand_name: "review".into(),
-            duration_ms: 1500,
-            findings_count: 3,
-        });
-        observer.record_event(&ObserverEvent::HandFailed {
-            hand_name: "review".into(),
-            error: "timeout".into(),
-            duration_ms: 5000,
-        });
-
-        assert_eq!(*observer.events.lock(), 3);
-    }
-
-    #[test]
-    fn hand_metrics_recordable() {
-        let observer = DummyObserver::default();
-
-        observer.record_metric(&ObserverMetric::HandRunDuration {
-            hand_name: "review".into(),
-            duration: Duration::from_millis(1500),
-        });
-        observer.record_metric(&ObserverMetric::HandFindingsCount {
-            hand_name: "review".into(),
-            count: 3,
-        });
-        observer.record_metric(&ObserverMetric::HandSuccessRate {
-            hand_name: "review".into(),
-            success: true,
-        });
-
-        assert_eq!(*observer.metrics.lock(), 3);
-    }
-
-    #[test]
-    fn hand_event_and_metric_are_cloneable() {
-        let event = ObserverEvent::HandCompleted {
-            hand_name: "review".into(),
-            duration_ms: 500,
-            findings_count: 2,
-        };
-        let metric = ObserverMetric::HandRunDuration {
-            hand_name: "review".into(),
-            duration: Duration::from_millis(500),
-        };
-
-        let cloned_event = event.clone();
-        let cloned_metric = metric.clone();
-
-        assert!(matches!(cloned_event, ObserverEvent::HandCompleted { .. }));
-        assert!(matches!(
-            cloned_metric,
-            ObserverMetric::HandRunDuration { .. }
-        ));
-    }
 }

crates/zeroclaw-channels/src/discord.rs

@@ -901,6 +901,17 @@ impl Channel for DiscordChannel {
         "discord"
     }
 
+    /// Discord bot tokens encode the bot's user ID in the first
+    /// segment (`base64(user_id).timestamp.hmac`); decode on demand
+    /// rather than caching since the result is deterministic and the
+    /// orchestrator only calls `self_handle` on the inbound path.
+    /// Returning the user ID engages the SDK self-loop guard against
+    /// gateway events the bot itself produced (typing indicators,
+    /// echoed message events from intent overlap, etc.).
+    fn self_handle(&self) -> Option<String> {
+        Self::bot_user_id_from_token(&self.bot_token)
+    }
+
     async fn send(&self, message: &SendMessage) -> anyhow::Result<()> {
         let raw_content = crate::util::strip_tool_call_tags(&message.content);
         let (cleaned_content, parsed_attachments) = parse_attachment_markers(&raw_content);

crates/zeroclaw-channels/src/irc.rs

@@ -354,6 +354,18 @@ impl Channel for IrcChannel {
         "irc"
     }
 
+    /// IRC echoes the bot's own PRIVMSGs back through the same socket
+    /// for any channel the bot is JOINed to. Returning the configured
+    /// nickname here engages the SDK self-loop guard so those echoes
+    /// drop before reaching the agent loop. The nickname is set at
+    /// construction (`config.nickname`) and used as the preferred nick
+    /// during NICK negotiation; if the server forces a different nick
+    /// (collision fallback in `listen`), the agent-loop fallback
+    /// catches the gap.
+    fn self_handle(&self) -> Option<String> {
+        Some(self.nickname.clone())
+    }
+
     async fn send(&self, message: &SendMessage) -> anyhow::Result<()> {
         let mut guard = self.writer.lock().await;
         let writer = guard