feat(observe): surface delegation + A2A in the live world feed via audit poll
Edge-initiated inter-agent events (gated delegation, A2A ingress) bypass the
run loop, so the world SSE now polls the append-only audit log (cursor on the
BIGINT id, seeded to max on first pass) and emits:
- delegation.invoked -> agent.delegate {fromAgentId,toAgentId,toName,task}
- a2a.invoked -> a2a.invoked {agentId}
TeamObserver renders agent.delegate as an A->B handoff in the team timeline;
adds the agent.delegate taxonomy type. Reliable live (the synthesized-run path
never streamed — active_runs + first-sight cursor jump skip it).
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
7352ed47ab
commit
85b0e1ac33
@@ -287,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();
|
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.
|
// 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();
|
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 {
|
loop {
|
||||||
let roster = match cm_db::repo::agents::roster(&pool, ws).await {
|
let roster = match cm_db::repo::agents::roster(&pool, ws).await {
|
||||||
Ok(r) => r,
|
Ok(r) => r,
|
||||||
@@ -387,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() }),
|
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;
|
first = false;
|
||||||
tokio::time::sleep(Duration::from_secs(2)).await;
|
tokio::time::sleep(Duration::from_secs(2)).await;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -7,7 +7,7 @@
|
|||||||
// A2A invocations among team members as they happen. No composer — observe only.
|
// A2A invocations among team members as they happen. No composer — observe only.
|
||||||
|
|
||||||
import { useEffect, useRef, useState } from "react";
|
import { useEffect, useRef, useState } from "react";
|
||||||
import { Users, MessagesSquare, Hash, Globe } from "lucide-react";
|
import { Users, MessagesSquare, Hash, Globe, GitBranch } from "lucide-react";
|
||||||
|
|
||||||
import type { Agent } from "@/lib/api/schemas";
|
import type { Agent } from "@/lib/api/schemas";
|
||||||
import { useFetchJson } from "@/lib/api/use-fetch";
|
import { useFetchJson } from "@/lib/api/use-fetch";
|
||||||
@@ -24,9 +24,10 @@ interface Room {
|
|||||||
type FeedItem = {
|
type FeedItem = {
|
||||||
key: number;
|
key: number;
|
||||||
ts: string;
|
ts: string;
|
||||||
kind: "msg" | "room" | "a2a";
|
kind: "msg" | "room" | "a2a" | "delegate";
|
||||||
fromId: string;
|
fromId: string;
|
||||||
toId?: string;
|
toId?: string;
|
||||||
|
toName?: string;
|
||||||
subject?: string;
|
subject?: string;
|
||||||
text: string;
|
text: string;
|
||||||
};
|
};
|
||||||
@@ -110,6 +111,11 @@ export function TeamObserver({ agent }: { agent: Agent }) {
|
|||||||
push({ kind: "a2a", fromId: d.agentId, text: d.skill ? `skill: ${d.skill}` : "external task" });
|
push({ kind: "a2a", fromId: d.agentId, text: d.skill ? `skill: ${d.skill}` : "external task" });
|
||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
useLiveEvent("agent.delegate", (d) => {
|
||||||
|
if (members.has(d.fromAgentId) || members.has(d.toAgentId)) {
|
||||||
|
push({ kind: "delegate", fromId: d.fromAgentId, toId: d.toAgentId, toName: d.toName, text: d.task || "sub-task" });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
const rooms = (useFetchJson<Room[]>("/api/claw-chat/rooms").data ?? []).filter(
|
const rooms = (useFetchJson<Room[]>("/api/claw-chat/rooms").data ?? []).filter(
|
||||||
(r) => r.kind === "room" && r.participants.some((id) => members.has(id)),
|
(r) => r.kind === "room" && r.participants.some((id) => members.has(id)),
|
||||||
@@ -157,6 +163,8 @@ export function TeamObserver({ agent }: { agent: Agent }) {
|
|||||||
<Hash aria-hidden size={11} className="text-muted-foreground" />
|
<Hash aria-hidden size={11} className="text-muted-foreground" />
|
||||||
) : f.kind === "a2a" ? (
|
) : f.kind === "a2a" ? (
|
||||||
<Globe aria-hidden size={11} className="text-[#c98af0]" />
|
<Globe aria-hidden size={11} className="text-[#c98af0]" />
|
||||||
|
) : f.kind === "delegate" ? (
|
||||||
|
<GitBranch aria-hidden size={11} className="text-coral" />
|
||||||
) : (
|
) : (
|
||||||
<MessagesSquare aria-hidden size={11} className="text-muted-foreground" />
|
<MessagesSquare aria-hidden size={11} className="text-muted-foreground" />
|
||||||
)}
|
)}
|
||||||
@@ -165,7 +173,9 @@ export function TeamObserver({ agent }: { agent: Agent }) {
|
|||||||
? `${nameOf(f.fromId)} → #${f.subject || "room"}`
|
? `${nameOf(f.fromId)} → #${f.subject || "room"}`
|
||||||
: f.kind === "a2a"
|
: f.kind === "a2a"
|
||||||
? `external A2A → ${nameOf(f.fromId)}`
|
? `external A2A → ${nameOf(f.fromId)}`
|
||||||
: `${nameOf(f.fromId)} → ${f.toId ? nameOf(f.toId) : "?"}`}
|
: f.kind === "delegate"
|
||||||
|
? `${nameOf(f.fromId)} ⇒ ${f.toName || (f.toId ? nameOf(f.toId) : "?")}`
|
||||||
|
: `${nameOf(f.fromId)} → ${f.toId ? nameOf(f.toId) : "?"}`}
|
||||||
</span>
|
</span>
|
||||||
<span className="ml-auto text-xxs text-muted-foreground">{relativeTime(f.ts)}</span>
|
<span className="ml-auto text-xxs text-muted-foreground">{relativeTime(f.ts)}</span>
|
||||||
</span>
|
</span>
|
||||||
|
|||||||
@@ -44,6 +44,8 @@ export interface TaxonomyEvents {
|
|||||||
};
|
};
|
||||||
/** Observe → an external A2A caller invoked this agent (a new ingress). */
|
/** Observe → an external A2A caller invoked this agent (a new ingress). */
|
||||||
"a2a.invoked": { agentId: string; caller?: string; skill?: string; ts?: string };
|
"a2a.invoked": { agentId: string; caller?: string; skill?: string; ts?: string };
|
||||||
|
/** Observe → a claw delegated a sub-task to another claw (A→B handoff). */
|
||||||
|
"agent.delegate": { fromAgentId: string; toAgentId: string; toName?: string; task?: string; ts?: string };
|
||||||
/** World → Live (Gource): the agent pawn retargets to nodeId and beams. */
|
/** World → Live (Gource): the agent pawn retargets to nodeId and beams. */
|
||||||
"world.touch": { agentId: string; nodeId: string; kind?: "service" | "event"; weight?: number };
|
"world.touch": { agentId: string; nodeId: string; kind?: "service" | "event"; weight?: number };
|
||||||
/** World → node glow/pulse intensity. */
|
/** World → node glow/pulse intensity. */
|
||||||
@@ -86,6 +88,7 @@ export const TAXONOMY_TYPES: TaxonomyType[] = [
|
|||||||
"agent.message",
|
"agent.message",
|
||||||
"room.message",
|
"room.message",
|
||||||
"a2a.invoked",
|
"a2a.invoked",
|
||||||
|
"agent.delegate",
|
||||||
"world.touch",
|
"world.touch",
|
||||||
"node.activity",
|
"node.activity",
|
||||||
"topology.update",
|
"topology.update",
|
||||||
|
|||||||
Reference in New Issue
Block a user