osobh/clawmates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: agent-to-agent platform on ZeroClaw 0.8.2 — rooms, delegation, A2A ingress

Browse Source 
Builds on the v0.8.2 runtime. Four workstreams, all behind the §15 MCP door:

- Group rooms (Phase 1): migration 0026; N-way threads repo with a DM/room
  count-guard; chat.send {room} + room.create/invite/leave tools; RoomMessage
  -> room.message SSE; /api/claw-chat/rooms* APIs; Observer room badge.
- Per-claw door identity: door caller_agent resolves the X-ZeroClaw-Agent
  header (set by the fork) to the specific claw, falling back to roster[0].
- Gated delegation bridge (Phase 3): clawmates__delegate door tool drives a
  sibling via the existing /ws/chat ZeroClawDriveExecutor (not A2A); self-deny,
  per-workspace hourly budget, audit trail, untrusted-banner result. Native
  in-daemon delegation stays off (it would bypass the door).
- A2A tenant ingress (Phase 2): migration 0027 (workspace_a2a + a2a_tokens);
  runtime_provision enable_a2a_server/publish_claw; routes/a2a.rs tenant-aware
  proxy (per-workspace tokens, injected internal bearer, daemon stays internal,
  cards URL-rewritten to the cm-api edge); a2a.invoked taxonomy.

Tests: cm-db room repos, cm-runtime chat tools, door units. sqlx cache updated.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Omar Sobh
2026-06-28 16:11:01 -07:00
co-authored by Claude Opus 4.8
parent 1a531e91cd
commit cbfa0ff24f
34 changed files with 1716 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/cm-db/src/repo/threads.rs
+168 -10
View File
@@ -5,13 +5,16 @@ use uuid::Uuid;
use crate::DbError;
/// An inter-agent conversation (§7.2 Claw Chat).
/// An inter-agent conversation (§7.2 Claw Chat). `kind` is `dm` for the 1:1
/// threads `find_or_create` makes, or `room` for N-way group rooms.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct Thread {
pub id: Uuid,
pub workspace_id: Uuid,
pub subject: String,
pub sensitivity: String,
pub kind: String,
pub created_by: Option<Uuid>,
pub participants: Vec<Uuid>,
pub last_preview: Option<String>,
#[serde(with = "time::serde::rfc3339")]
@@ -30,7 +33,9 @@ pub struct ThreadMessage {
}
/// Finds the 1:1 thread between two agents, or creates it with the given
/// subject.
/// subject. The match is constrained to `kind = 'dm'` threads with exactly the
/// two active participants, so a DM never resolves to a group room that happens
/// to contain both agents.
pub async fn find_or_create(
pool: &PgPool,
workspace_id: WorkspaceId,
@@ -41,10 +46,13 @@ pub async fn find_or_create(
let existing = sqlx::query_scalar!(
r#"SELECT t.id FROM threads t
WHERE t.workspace_id = $1
AND t.kind = 'dm'
AND EXISTS (SELECT 1 FROM thread_participants p
WHERE p.thread_id = t.id AND p.agent_id = $2)
WHERE p.thread_id = t.id AND p.agent_id = $2 AND p.active)
AND EXISTS (SELECT 1 FROM thread_participants p
WHERE p.thread_id = t.id AND p.agent_id = $3)
WHERE p.thread_id = t.id AND p.agent_id = $3 AND p.active)
AND (SELECT count(*) FROM thread_participants p
WHERE p.thread_id = t.id AND p.active) = 2
LIMIT 1"#,
workspace_id.as_uuid(),
a.as_uuid(),
@@ -59,7 +67,7 @@ pub async fn find_or_create(
let id = Uuid::now_v7();
let mut tx = pool.begin().await.map_err(DbError::from)?;
sqlx::query!(
"INSERT INTO threads (id, workspace_id, subject) VALUES ($1, $2, $3)",
"INSERT INTO threads (id, workspace_id, subject, kind) VALUES ($1, $2, $3, 'dm')",
id,
workspace_id.as_uuid(),
subject,
@@ -68,9 +76,10 @@ pub async fn find_or_create(
.await?;
for agent in [a, b] {
sqlx::query!(
"INSERT INTO thread_participants (thread_id, agent_id) VALUES ($1, $2)",
"INSERT INTO thread_participants (thread_id, agent_id, added_by) VALUES ($1, $2, $3)",
id,
agent.as_uuid(),
a.as_uuid(),
)
.execute(&mut *tx)
.await?;
@@ -79,6 +88,134 @@ pub async fn find_or_create(
Ok(id)
}
/// Creates an N-way group room with the given participants. `created_by` is the
/// agent who opened it (a claw via `room.create`) or `None` for an
/// operator-created room. Participants are deduped.
pub async fn create_room(
pool: &PgPool,
workspace_id: WorkspaceId,
subject: &str,
created_by: Option<AgentId>,
participants: &[AgentId],
) -> Result<Uuid, DbError> {
let id = Uuid::now_v7();
let creator = created_by.map(|a| a.as_uuid());
let mut tx = pool.begin().await.map_err(DbError::from)?;
sqlx::query!(
"INSERT INTO threads (id, workspace_id, subject, kind, created_by)
VALUES ($1, $2, $3, 'room', $4)",
id,
workspace_id.as_uuid(),
subject,
creator,
)
.execute(&mut *tx)
.await?;
let mut seen = std::collections::HashSet::new();
for agent in participants.iter().copied() {
if !seen.insert(agent.as_uuid()) {
continue;
}
sqlx::query!(
"INSERT INTO thread_participants (thread_id, agent_id, added_by) VALUES ($1, $2, $3)",
id,
agent.as_uuid(),
creator,
)
.execute(&mut *tx)
.await?;
}
tx.commit().await.map_err(DbError::from)?;
Ok(id)
}
/// Adds (or re-activates) a participant in a thread. `added_by` is the inviter,
/// or `None` for an operator action.
pub async fn add_participant(
pool: &PgPool,
thread_id: Uuid,
agent: AgentId,
added_by: Option<AgentId>,
) -> Result<(), DbError> {
sqlx::query!(
"INSERT INTO thread_participants (thread_id, agent_id, added_by, active)
VALUES ($1, $2, $3, true)
ON CONFLICT (thread_id, agent_id)
DO UPDATE SET active = true, added_by = EXCLUDED.added_by",
thread_id,
agent.as_uuid(),
added_by.map(|a| a.as_uuid()),
)
.execute(pool)
.await?;
Ok(())
}
/// Soft-removes a participant (keeps their messages attributable).
pub async fn remove_participant(
pool: &PgPool,
thread_id: Uuid,
agent: AgentId,
) -> Result<(), DbError> {
sqlx::query!(
"UPDATE thread_participants SET active = false
WHERE thread_id = $1 AND agent_id = $2",
thread_id,
agent.as_uuid(),
)
.execute(pool)
.await?;
Ok(())
}
/// Active participant ids in a thread (for message fan-out + event emission).
pub async fn participants(pool: &PgPool, thread_id: Uuid) -> Result<Vec<Uuid>, DbError> {
let rows = sqlx::query_scalar!(
"SELECT agent_id FROM thread_participants WHERE thread_id = $1 AND active",
thread_id,
)
.fetch_all(pool)
.await?;
Ok(rows)
}
/// All group rooms in a workspace (human/operator room list).
pub async fn list_rooms_for_workspace(
pool: &PgPool,
workspace_id: WorkspaceId,
) -> Result<Vec<Thread>, DbError> {
let rows = sqlx::query!(
r#"SELECT t.id, t.workspace_id, t.subject, t.sensitivity, t.kind,
t.created_by, t.created_at,
ARRAY(SELECT p2.agent_id FROM thread_participants p2
WHERE p2.thread_id = t.id AND p2.active) AS "participants!",
(SELECT m.content->>'text' FROM thread_messages m
WHERE m.thread_id = t.id
ORDER BY m.created_at DESC LIMIT 1) AS last_preview
FROM threads t
WHERE t.workspace_id = $1 AND t.kind = 'room'
ORDER BY (SELECT max(m.created_at) FROM thread_messages m
WHERE m.thread_id = t.id) DESC NULLS LAST"#,
workspace_id.as_uuid(),
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|r| Thread {
id: r.id,
workspace_id: r.workspace_id,
subject: r.subject,
sensitivity: r.sensitivity,
kind: r.kind,
created_by: r.created_by,
participants: r.participants,
last_preview: r.last_preview,
created_at: r.created_at,
})
.collect())
}
pub async fn add_message(
pool: &PgPool,
thread_id: Uuid,
@@ -105,15 +242,16 @@ pub async fn add_message(
/// last message preview (§7.2 thread list).
pub async fn list_for_agent(pool: &PgPool, agent_id: AgentId) -> Result<Vec<Thread>, DbError> {
let rows = sqlx::query!(
r#"SELECT t.id, t.workspace_id, t.subject, t.sensitivity, t.created_at,
r#"SELECT t.id, t.workspace_id, t.subject, t.sensitivity, t.kind,
t.created_by, t.created_at,
ARRAY(SELECT p2.agent_id FROM thread_participants p2
WHERE p2.thread_id = t.id) AS "participants!",
WHERE p2.thread_id = t.id AND p2.active) AS "participants!",
(SELECT m.content->>'text' FROM thread_messages m
WHERE m.thread_id = t.id
ORDER BY m.created_at DESC LIMIT 1) AS last_preview
FROM threads t
JOIN thread_participants p ON p.thread_id = t.id
WHERE p.agent_id = $1
WHERE p.agent_id = $1 AND p.active
ORDER BY (SELECT max(m.created_at) FROM thread_messages m
WHERE m.thread_id = t.id) DESC NULLS LAST"#,
agent_id.as_uuid(),
@@ -127,6 +265,8 @@ pub async fn list_for_agent(pool: &PgPool, agent_id: AgentId) -> Result<Vec<Thre
workspace_id: r.workspace_id,
subject: r.subject,
sensitivity: r.sensitivity,
kind: r.kind,
created_by: r.created_by,
participants: r.participants,
last_preview: r.last_preview,
created_at: r.created_at,
@@ -146,6 +286,24 @@ pub async fn messages(pool: &PgPool, thread_id: Uuid) -> Result<Vec<ThreadMessag
Ok(rows)
}
/// The workspace a thread belongs to, if it exists (API scoping).
pub async fn workspace_of(pool: &PgPool, thread_id: Uuid) -> Result<Option<Uuid>, DbError> {
Ok(
sqlx::query_scalar!("SELECT workspace_id FROM threads WHERE id = $1", thread_id)
.fetch_optional(pool)
.await?,
)
}
/// A thread's subject line, if it exists.
pub async fn subject(pool: &PgPool, thread_id: Uuid) -> Result<Option<String>, DbError> {
Ok(
sqlx::query_scalar!("SELECT subject FROM threads WHERE id = $1", thread_id)
.fetch_optional(pool)
.await?,
)
}
/// Whether an agent participates in a thread (API scoping).
pub async fn is_participant(
pool: &PgPool,
@@ -153,7 +311,7 @@ pub async fn is_participant(
agent_id: AgentId,
) -> Result<bool, DbError> {
let row = sqlx::query_scalar!(
"SELECT 1 AS x FROM thread_participants WHERE thread_id = $1 AND agent_id = $2",
"SELECT 1 AS x FROM thread_participants WHERE thread_id = $1 AND agent_id = $2 AND active",
thread_id,
agent_id.as_uuid(),
)