Merge feat/a2a-rooms-delegation-ingress into main

Brings a2a rooms, delegation, and A2A ingress work back into main. Prod's
DB has migrations 0026 (group_rooms) and 0027 (a2a) applied from an
earlier hand-tagged fleet21 build cut from this branch, but main never got
them — so the CI-built server image from main refused to start against
prod's DB with "migration 26 was previously applied but is missing".
Landing the branch closes that gap: main + prod DB now share the same
migration state, so images built from main can safely roll onto gw-04.

Included:
- 0026_group_rooms.sql / 0027_a2a.sql — align main with prod's schema
- cm-api routes/a2a.rs + mcp_door.rs updates — A2A ingress and MCP door
- cm-runtime tools/delegate.rs + tools/chat.rs — delegation + N-way rooms
- frontend TeamObserver + FleetPanels + AgentObserver updates
- taxonomy.ts — delegation + A2A signals in the live world feed

Not included (still WIP on the local checkout):
- 0028_backfill_on_delete.sql / 0029_hot_query_indexes.sql
- broker pool + team-run quota changes
- Dashboard.tsx UI rename (Large World → Visualizations)
- Node write.send timeout (aaab663) — separate concern

The SSE resume fix (202e853) is preserved by auto-merge — approvals.rs
still awaits resume_run inline so the channel is ready before reply.

crates/cm-api/src/routes/a2a.rs

@@ -0,0 +1,387 @@
+//! A2A tenant-aware ingress (§ topology platform). ZeroClaw 0.8.2 ships a
+//! spec-conforming Agent2Agent server, but its auth is a single global bearer
+//! and its discovery cards are public — so the raw daemon (`:42617/a2a/*`) is
+//! NEVER exposed. This module is the only front door: cm-api authenticates the
+//! external caller per-workspace (`a2a_tokens`), maps to the workspace's daemon,
+//! injects the internal `ZEROCLAW_TOKEN`, and journals the turn.
+//!
+//! A2A is a new INGRESS, not a new egress: published claws are still tool-free
+//! behind the MCP door, so an A2A-invoked turn can only act via the gated door.
+
+use axum::body::Bytes;
+use axum::extract::{Path, State};
+use axum::http::header::AUTHORIZATION;
+use axum::http::{HeaderMap, StatusCode};
+use axum::Json;
+use serde::Deserialize;
+use serde_json::{json, Value};
+use uuid::Uuid;
+
+use crate::runtime_provision::RuntimeProvisioner;
+use crate::{ApiError, AppState, Authed};
+
+/// The internal daemon base URL + bearer (single daemon today; a per-workspace
+/// resolver slots in here when multi-daemon lands).
+fn daemon(_workspace: Uuid) -> Option<(String, String)> {
+    let url = std::env::var("ZEROCLAW_GATEWAY_URL").ok().filter(|u| !u.is_empty())?;
+    let token = std::env::var("ZEROCLAW_TOKEN").ok().filter(|t| !t.is_empty())?;
+    Some((url, token))
+}
+
+// ── Settings (operator, session-authed) ────────────────────────────────────
+
+#[derive(Deserialize)]
+pub struct SettingsBody {
+    enabled: bool,
+    #[serde(rename = "publicBaseUrl")]
+    public_base_url: Option<String>,
+    /// Claws to publish, each with the skills to advertise.
+    #[serde(default)]
+    publish: Vec<PublishEntry>,
+}
+
+#[derive(Deserialize)]
+pub struct PublishEntry {
+    #[serde(rename = "clawId")]
+    claw_id: Uuid,
+    #[serde(default)]
+    skills: Vec<String>,
+}
+
+/// POST /api/a2a/settings — opt the workspace in/out + publish a curated set of
+/// claws. Persists the opt-in and pushes the config to the runtime daemon.
+pub async fn settings(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Json(body): Json<SettingsBody>,
+) -> Result<Json<Value>, ApiError> {
+    let base = body
+        .public_base_url
+        .clone()
+        .unwrap_or_else(|| format!("/api/a2a/{}", user.workspace_id.as_uuid()));
+    sqlx::query(
+        "INSERT INTO workspace_a2a (workspace_id, enabled, public_base_url)
+         VALUES ($1, $2, $3)
+         ON CONFLICT (workspace_id)
+         DO UPDATE SET enabled = EXCLUDED.enabled,
+                       public_base_url = EXCLUDED.public_base_url,
+                       updated_at = now()",
+    )
+    .bind(user.workspace_id.as_uuid())
+    .bind(body.enabled)
+    .bind(&base)
+    .execute(&state.pool)
+    .await
+    .map_err(|_| ApiError::Internal)?;
+
+    // Push config to the daemon (best-effort: settings persist even if the
+    // runtime is momentarily unreachable).
+    if body.enabled {
+        if let Some(p) = RuntimeProvisioner::from_env() {
+            p.enable_a2a_server(&base).await.map_err(|_| ApiError::Internal)?;
+            for entry in &body.publish {
+                // Scope: each claw must be in the caller's workspace.
+                let agent = cm_db::repo::agents::get(&state.pool, entry.claw_id.into())
+                    .await
+                    .map_err(|_| ApiError::NotFound)?;
+                if agent.workspace_id != user.workspace_id {
+                    return Err(ApiError::Forbidden);
+                }
+                p.publish_claw(entry.claw_id, &entry.skills)
+                    .await
+                    .map_err(|_| ApiError::Internal)?;
+            }
+        }
+    }
+    Ok(Json(json!({ "enabled": body.enabled, "publicBaseUrl": base })))
+}
+
+/// GET /api/a2a/settings — current A2A opt-in state for the workspace.
+pub async fn get_settings(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+) -> Result<Json<Value>, ApiError> {
+    let row = sqlx::query_as::<_, (bool, Option<String>)>(
+        "SELECT enabled, public_base_url FROM workspace_a2a WHERE workspace_id = $1",
+    )
+    .bind(user.workspace_id.as_uuid())
+    .fetch_optional(&state.pool)
+    .await
+    .map_err(|_| ApiError::Internal)?;
+    let (enabled, base) = row.unwrap_or((false, None));
+    Ok(Json(json!({ "enabled": enabled, "publicBaseUrl": base })))
+}
+
+#[derive(Deserialize)]
+pub struct MintTokenBody {
+    /// Restrict the token to one claw alias, or `None` for any published claw.
+    #[serde(default)]
+    alias: Option<String>,
+    #[serde(default, rename = "exposedSkills")]
+    exposed_skills: Vec<String>,
+    #[serde(default)]
+    label: Option<String>,
+}
+
+/// POST /api/a2a/tokens — mint an external A2A bearer for the workspace.
+pub async fn mint_token(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Json(body): Json<MintTokenBody>,
+) -> Result<Json<Value>, ApiError> {
+    let id = Uuid::now_v7();
+    let token = Uuid::new_v4(); // unguessable external bearer
+    sqlx::query(
+        "INSERT INTO a2a_tokens (id, workspace_id, alias, token, exposed_skills, label)
+         VALUES ($1, $2, $3, $4, $5, $6)",
+    )
+    .bind(id)
+    .bind(user.workspace_id.as_uuid())
+    .bind(&body.alias)
+    .bind(token)
+    .bind(&body.exposed_skills)
+    .bind(&body.label)
+    .execute(&state.pool)
+    .await
+    .map_err(|_| ApiError::Internal)?;
+    let ws = user.workspace_id.as_uuid();
+    Ok(Json(json!({
+        "token": token.to_string(),
+        "discoveryUrl": format!("/api/a2a/{ws}/.well-known/agents-card.json"),
+        "taskUrlTemplate": format!("/api/a2a/{ws}/{{alias}}"),
+    })))
+}
+
+/// GET /api/a2a/tokens — list the workspace's A2A tokens (token value hidden).
+pub async fn list_tokens(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+) -> Result<Json<Value>, ApiError> {
+    let rows = sqlx::query_as::<_, (Uuid, Option<String>, bool, Option<String>, Option<time::OffsetDateTime>)>(
+        "SELECT id, alias, enabled, label, last_used_at FROM a2a_tokens
+         WHERE workspace_id = $1 ORDER BY created_at DESC",
+    )
+    .bind(user.workspace_id.as_uuid())
+    .fetch_all(&state.pool)
+    .await
+    .map_err(|_| ApiError::Internal)?;
+    let tokens: Vec<Value> = rows
+        .into_iter()
+        .map(|(id, alias, enabled, label, last_used)| {
+            json!({
+                "id": id.to_string(),
+                "alias": alias,
+                "enabled": enabled,
+                "label": label,
+                "lastUsedAt": last_used.map(|t| t.unix_timestamp()),
+            })
+        })
+        .collect();
+    Ok(Json(json!({ "tokens": tokens })))
+}
+
+/// DELETE /api/a2a/tokens/{id} — revoke (disable) a token.
+pub async fn revoke_token(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Value>, ApiError> {
+    let res = sqlx::query(
+        "UPDATE a2a_tokens SET enabled = false WHERE id = $1 AND workspace_id = $2",
+    )
+    .bind(id)
+    .bind(user.workspace_id.as_uuid())
+    .execute(&state.pool)
+    .await
+    .map_err(|_| ApiError::Internal)?;
+    if res.rows_affected() == 0 {
+        return Err(ApiError::NotFound);
+    }
+    Ok(Json(json!({ "ok": true })))
+}
+
+// ── Public ingress (external callers; NO session auth) ──────────────────────
+
+/// Whether a workspace has opted into A2A. Gates discovery so non-published
+/// workspaces can't be enumerated.
+async fn workspace_enabled(state: &AppState, workspace: Uuid) -> bool {
+    sqlx::query_scalar::<_, bool>("SELECT enabled FROM workspace_a2a WHERE workspace_id = $1")
+        .bind(workspace)
+        .fetch_optional(&state.pool)
+        .await
+        .ok()
+        .flatten()
+        .unwrap_or(false)
+}
+
+/// Rewrite any daemon-internal URL in a discovery card to the edge base, so
+/// external callers only ever learn our proxied endpoints.
+fn rewrite_card(card: &str, daemon_base: &str, edge_base: &str) -> String {
+    card.replace(daemon_base, edge_base.trim_end_matches('/'))
+}
+
+async fn proxy_card(state: &AppState, workspace: Uuid, path: &str) -> Result<Json<Value>, StatusCode> {
+    if !workspace_enabled(state, workspace).await {
+        return Err(StatusCode::NOT_FOUND);
+    }
+    let (base, _token) = daemon(workspace).ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
+    let edge = format!("/api/a2a/{workspace}");
+    let resp = reqwest::Client::new()
+        .get(format!("{base}{path}"))
+        .send()
+        .await
+        .map_err(|_| StatusCode::BAD_GATEWAY)?;
+    if !resp.status().is_success() {
+        return Err(StatusCode::NOT_FOUND);
+    }
+    let text = resp.text().await.map_err(|_| StatusCode::BAD_GATEWAY)?;
+    let rewritten = rewrite_card(&text, &base, &edge);
+    let value: Value = serde_json::from_str(&rewritten).map_err(|_| StatusCode::BAD_GATEWAY)?;
+    Ok(Json(value))
+}
+
+/// GET /api/a2a/{workspace}/.well-known/agents-card.json — the catalog card.
+pub async fn discovery_catalog(
+    State(state): State<AppState>,
+    Path(workspace): Path<Uuid>,
+) -> Result<Json<Value>, StatusCode> {
+    proxy_card(&state, workspace, "/.well-known/agents-card.json").await
+}
+
+/// GET /api/a2a/{workspace}/{alias}/.well-known/agent-card.json — per-claw card.
+pub async fn discovery_card(
+    State(state): State<AppState>,
+    Path((workspace, alias)): Path<(Uuid, String)>,
+) -> Result<Json<Value>, StatusCode> {
+    proxy_card(&state, workspace, &format!("/a2a/{alias}/.well-known/agent-card.json")).await
+}
+
+/// POST /api/a2a/{workspace}/{alias} — authenticated task invocation. Verifies
+/// the external bearer against `a2a_tokens`, scopes it to (workspace, alias),
+/// then proxies to the daemon with the internal bearer injected.
+pub async fn task(
+    State(state): State<AppState>,
+    Path((workspace, alias)): Path<(Uuid, String)>,
+    headers: HeaderMap,
+    body: Bytes,
+) -> Result<Json<Value>, StatusCode> {
+    // Kill switch.
+    if std::env::var("CLAWMATES_A2A_POLICY").as_deref() == Ok("deny") {
+        return Err(StatusCode::SERVICE_UNAVAILABLE);
+    }
+    // External bearer → token row.
+    let bearer = headers
+        .get(AUTHORIZATION)
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.strip_prefix("Bearer "))
+        .and_then(|t| Uuid::parse_str(t).ok())
+        .ok_or(StatusCode::UNAUTHORIZED)?;
+    let row = sqlx::query_as::<_, (Uuid, Uuid, Option<String>, bool)>(
+        "SELECT id, workspace_id, alias, enabled FROM a2a_tokens WHERE token = $1",
+    )
+    .bind(bearer)
+    .fetch_optional(&state.pool)
+    .await
+    .ok()
+    .flatten()
+    .ok_or(StatusCode::UNAUTHORIZED)?;
+    let (token_id, token_ws, token_alias, enabled) = row;
+    if !enabled {
+        return Err(StatusCode::UNAUTHORIZED);
+    }
+    // Scope: token's workspace must match the path; alias scope (if set).
+    if token_ws != workspace {
+        return Err(StatusCode::FORBIDDEN);
+    }
+    if let Some(scoped) = &token_alias {
+        if scoped != &alias {
+            return Err(StatusCode::FORBIDDEN);
+        }
+    }
+    let ws = cm_domain::WorkspaceId::from(workspace);
+
+    // Hourly ingress rate cap (counts a2a.invoked).
+    if let Some(cap) = std::env::var("CLAWMATES_A2A_RATE_LIMIT")
+        .ok()
+        .and_then(|v| v.parse::<i64>().ok())
+    {
+        let used: i64 = sqlx::query_scalar(
+            "SELECT count(*) FROM audit_log WHERE workspace_id = $1
+               AND event_type = 'a2a.invoked' AND created_at > now() - interval '1 hour'",
+        )
+        .bind(workspace)
+        .fetch_one(&state.pool)
+        .await
+        .unwrap_or(0);
+        if used >= cap {
+            return Err(StatusCode::TOO_MANY_REQUESTS);
+        }
+    }
+
+    let (base, internal_token) = daemon(workspace).ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
+    let resp = reqwest::Client::new()
+        .post(format!("{base}/a2a/{alias}"))
+        .bearer_auth(&internal_token)
+        .header("Content-Type", "application/json")
+        .body(body.clone())
+        .send()
+        .await
+        .map_err(|_| StatusCode::BAD_GATEWAY)?;
+    let status = resp.status();
+    let text = resp.text().await.map_err(|_| StatusCode::BAD_GATEWAY)?;
+    let value: Value = serde_json::from_str(&text).unwrap_or_else(|_| json!({ "raw": text }));
+
+    let _ = sqlx::query("UPDATE a2a_tokens SET last_used_at = now() WHERE id = $1")
+        .bind(token_id)
+        .execute(&state.pool)
+        .await;
+    let _ = cm_db::repo::audit::append(
+        &state.pool,
+        ws,
+        cm_db::repo::audit::Actor::System,
+        "a2a.invoked",
+        "agent",
+        &alias,
+        json!({ "ok": status.is_success(), "status": status.as_u16() }),
+    )
+    .await;
+    // Best-effort: surface the inbound turn in Observe/history.
+    let _ = synthesize_run(&state, ws, &alias).await;
+
+    if !status.is_success() {
+        return Err(StatusCode::BAD_GATEWAY);
+    }
+    Ok(Json(value))
+}
+
+/// Create a tiny session+run for the invoked claw so an A2A turn (which bypasses
+/// cm-api's run loop) still appears in the world feed / history. Best-effort.
+async fn synthesize_run(
+    state: &AppState,
+    workspace: cm_domain::WorkspaceId,
+    alias: &str,
+) -> Result<(), ()> {
+    let agent_id = alias
+        .strip_prefix("claw_")
+        .and_then(|hex| Uuid::parse_str(hex).ok())
+        .map(cm_domain::AgentId::from)
+        .ok_or(())?;
+    let session = cm_db::repo::sessions::create(&state.pool, agent_id, workspace, "a2a")
+        .await
+        .map_err(|_| ())?;
+    let run_id = cm_db::repo::runs::create(&state.pool, session.id)
+        .await
+        .map_err(|_| ())?;
+    let _ = cm_db::repo::run_events::append(&state.pool, run_id, 1, "run_started", json!({ "run_id": run_id })).await;
+    let _ = cm_db::repo::run_events::append(
+        &state.pool,
+        run_id,
+        2,
+        "a2a_invoked",
+        json!({ "alias": alias }),
+    )
+    .await;
+    let _ = cm_db::repo::run_events::append(&state.pool, run_id, 3, "run_completed", json!({ "message_id": "" })).await;
+    let _ = cm_db::repo::runs::set_state(&state.pool, run_id, cm_domain::RunState::Completed, None).await;
+    Ok(())
+}

crates/cm-api/src/routes/claw_chat.rs

@@ -1,8 +1,9 @@
-use axum::extract::{Query, State};
+use axum::extract::{Path, Query, State};
 use axum::Json;
 use cm_db::repo::threads::{Thread, ThreadMessage};
 use cm_domain::AgentId;
 use serde::Deserialize;
+use serde_json::{json, Value};
 use uuid::Uuid;
 
 use crate::routes::claws::workspace_agent;
@@ -48,3 +49,83 @@ pub async fn messages(
         cm_db::repo::threads::messages(&state.pool, query.thread_id).await?,
     ))
 }
+
+/// Confirms a thread is in the caller's workspace (room API scoping).
+async fn workspace_thread(
+    state: &AppState,
+    user: &cm_auth::AuthedUser,
+    thread_id: Uuid,
+) -> Result<(), ApiError> {
+    match cm_db::repo::threads::workspace_of(&state.pool, thread_id).await? {
+        Some(ws) if ws == user.workspace_id.as_uuid() => Ok(()),
+        _ => Err(ApiError::NotFound),
+    }
+}
+
+#[derive(Deserialize)]
+pub struct CreateRoomBody {
+    subject: String,
+    members: Vec<AgentId>,
+}
+
+/// POST /api/claw-chat/rooms — create an N-way group room (operator action).
+pub async fn create_room(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Json(body): Json<CreateRoomBody>,
+) -> Result<Json<Value>, ApiError> {
+    // Every member must be a claw in the caller's workspace.
+    for id in &body.members {
+        workspace_agent(&state, &user, *id).await?;
+    }
+    let thread_id = cm_db::repo::threads::create_room(
+        &state.pool,
+        user.workspace_id,
+        &body.subject,
+        None,
+        &body.members,
+    )
+    .await?;
+    Ok(Json(json!({ "threadId": thread_id })))
+}
+
+/// GET /api/claw-chat/rooms — list the workspace's group rooms.
+pub async fn rooms(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+) -> Result<Json<Vec<Thread>>, ApiError> {
+    Ok(Json(
+        cm_db::repo::threads::list_rooms_for_workspace(&state.pool, user.workspace_id).await?,
+    ))
+}
+
+#[derive(Deserialize)]
+pub struct AddParticipantBody {
+    #[serde(rename = "clawId")]
+    claw_id: AgentId,
+}
+
+/// POST /api/claw-chat/rooms/{threadId}/participants — add a claw to a room.
+pub async fn add_participant(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Path(thread_id): Path<Uuid>,
+    Json(body): Json<AddParticipantBody>,
+) -> Result<Json<Value>, ApiError> {
+    workspace_thread(&state, &user, thread_id).await?;
+    workspace_agent(&state, &user, body.claw_id).await?;
+    cm_db::repo::threads::add_participant(&state.pool, thread_id, body.claw_id, None).await?;
+    Ok(Json(json!({ "ok": true })))
+}
+
+/// DELETE /api/claw-chat/rooms/{threadId}/participants/{clawId} — remove a claw.
+pub async fn remove_participant(
+    State(state): State<AppState>,
+    Authed(user): Authed,
+    Path((thread_id, claw_id)): Path<(Uuid, AgentId)>,
+) -> Result<Json<Value>, ApiError> {
+    workspace_thread(&state, &user, thread_id).await?;
+    workspace_agent(&state, &user, claw_id).await?;
+    cm_db::repo::threads::remove_participant(&state.pool, thread_id, claw_id).await?;
+    Ok(Json(json!({ "ok": true })))
+}

crates/cm-api/src/routes/mod.rs

@@ -1,3 +1,4 @@
+pub mod a2a;
 pub mod approvals;
 pub mod apps;
 pub mod auth;

crates/cm-api/src/routes/world.rs

@@ -152,6 +152,41 @@ fn normalize_run_event(
                 }),
             ));
         }
+        "room_message" => {
+            let s = |k: &str| {
+                payload
+                    .get(k)
+                    .and_then(|v| v.as_str())
+                    .unwrap_or("")
+                    .to_owned()
+            };
+            let participant_ids = payload
+                .get("participant_ids")
+                .and_then(|v| v.as_array())
+                .map(|a| {
+                    a.iter()
+                        .filter_map(|x| x.as_str().map(|s| s.to_owned()))
+                        .collect::<Vec<_>>()
+                })
+                .unwrap_or_default();
+            out.push((
+                "room.message",
+                json!({
+                    "fromAgentId": agent_id,
+                    "threadId": s("thread_id"),
+                    "subject": s("subject"),
+                    "text": s("text"),
+                    "participantIds": participant_ids,
+                }),
+            ));
+        }
+        "a2a_invoked" => {
+            // An external A2A caller started a turn on this agent (a new ingress).
+            out.push((
+                "a2a.invoked",
+                json!({ "agentId": agent_id }),
+            ));
+        }
         _ => {}
     }
     out
@@ -252,6 +287,9 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
         let mut last: std::collections::HashMap<String, String> = std::collections::HashMap::new();
         // Per-run journal cursor so we stream only NEW run_events each poll.
         let mut cursors: std::collections::HashMap<String, i64> = std::collections::HashMap::new();
+        // Audit-log cursor for edge-initiated inter-agent events (delegation,
+        // A2A) that bypass the run loop. -1 until seeded on the first pass.
+        let mut audit_cursor: i64 = -1;
         loop {
             let roster = match cm_db::repo::agents::roster(&pool, ws).await {
                 Ok(r) => r,
@@ -352,6 +390,62 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
                 json!({ "doorsPending": doors_pending(&pool, ws).await, "loops": runs.len() }),
             );
 
+            // Edge-initiated inter-agent events (gated delegation, A2A ingress)
+            // bypass the run loop, so surface them from the append-only audit log.
+            // On first sight jump the cursor to the current max so we stream
+            // forward instead of replaying history.
+            if audit_cursor < 0 {
+                audit_cursor = sqlx::query_scalar(
+                    "SELECT coalesce(max(id), 0) FROM audit_log WHERE workspace_id = $1",
+                )
+                .bind(ws.as_uuid())
+                .fetch_one(&pool)
+                .await
+                .unwrap_or(0);
+            } else {
+                let rows = sqlx::query(
+                    "SELECT id, actor_id, event_type, subject_id, detail FROM audit_log
+                     WHERE workspace_id = $1 AND id > $2
+                       AND event_type IN ('delegation.invoked', 'a2a.invoked')
+                     ORDER BY id ASC LIMIT 100",
+                )
+                .bind(ws.as_uuid())
+                .bind(audit_cursor)
+                .fetch_all(&pool)
+                .await
+                .unwrap_or_default();
+                for row in &rows {
+                    let id: i64 = row.get("id");
+                    let et: String = row.get("event_type");
+                    let actor: Option<uuid::Uuid> = row.get("actor_id");
+                    let subject: String = row.get("subject_id");
+                    let detail: Value = row.get("detail");
+                    match et.as_str() {
+                        "delegation.invoked" => {
+                            yield sse("agent.delegate", json!({
+                                "fromAgentId": actor.map(|u| u.to_string()).unwrap_or_default(),
+                                "toAgentId": detail.get("to_id").and_then(|v| v.as_str()).unwrap_or(""),
+                                "toName": subject,
+                                "task": detail.get("task").and_then(|v| v.as_str()).unwrap_or(""),
+                            }));
+                        }
+                        "a2a.invoked" => {
+                            // subject_id is the claw_<id> alias → surface the target agent.
+                            let agent_id = subject
+                                .strip_prefix("claw_")
+                                .and_then(|h| uuid::Uuid::parse_str(h).ok())
+                                .map(|u| u.to_string())
+                                .unwrap_or_else(|| subject.clone());
+                            yield sse("a2a.invoked", json!({ "agentId": agent_id }));
+                        }
+                        _ => {}
+                    }
+                    if id > audit_cursor {
+                        audit_cursor = id;
+                    }
+                }
+            }
+
             first = false;
             tokio::time::sleep(Duration::from_secs(2)).await;
         }