feat(telemetry): the reasoning stream actually streams
deploy / test (push) Successful in 4m18s
deploy / build (push) Successful in 5m17s

`agent.reasoning.delta` and `agent.tool.call` have been declared in the taxonomy
and listened for by the command centre since it shipped — and NOTHING ever
emitted them. The world feed emitted five types; neither was among them, so
REASONING STREAM could not populate no matter what an agent did.

The feed is a database poll, not a push bus, so a live card can only show what
was persisted. The worker already holds each step's output text and the claw
that produced it, so it records a `reasoning` mission_event (truncated — the
card renders a tail, not a transcript, and mission_events is capped per phase),
and the feed emits it forward from a cursor that starts at the current max so a
page load streams rather than replaying history.

`tool.call` is emitted from the same place. On the container tier it will stay
empty, and that is correct rather than broken: those agents are tool-free behind
the §15 door. Tool lines appear where agents actually hold tools.

Verified on a live mission: agent.reasoning.delta observed on /api/world/live
carrying the agent's own text, keyed by agentId.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Omar Sobh
2026-08-15 15:58:56 -07:00
co-authored by Claude Opus 5
parent 8be7b3c9b2
commit bf40d10064
2 changed files with 103 additions and 11 deletions
+72 -11
View File
@@ -77,11 +77,7 @@ struct MissionRow {
agent_id: Option<String>,
}
async fn world_missions(
pool: &PgPool,
ws: WorkspaceId,
only: Option<Uuid>,
) -> Vec<MissionRow> {
async fn world_missions(pool: &PgPool, ws: WorkspaceId, only: Option<Uuid>) -> Vec<MissionRow> {
let rows = sqlx::query(
"SELECT m.id::text AS mission_id,
m.title AS title,
@@ -416,11 +412,7 @@ fn benchmark_note(delta: &serde_json::Value) -> Option<String> {
/// Agents that execute a phase of this kind, via the purposes the phase runner
/// itself uses. Returns empty for a teamless (microVM) mission.
async fn phase_agents(
pool: &PgPool,
mission_id: &str,
kind: &str,
) -> Vec<String> {
async fn phase_agents(pool: &PgPool, mission_id: &str, kind: &str) -> Vec<String> {
let Ok(mid) = Uuid::parse_str(mission_id) else {
return Vec::new();
};
@@ -751,6 +743,10 @@ pub async fn world_live(
// 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;
// Cursor over mission_events for the per-agent LIVE column. Starts at
// -1 and jumps to the current max on first sight, so a page load
// streams forward instead of replaying every past turn.
let mut agent_ev_cursor: i64 = -1;
// Track the brain-file size we last announced per agent so we only
// emit `agent.memory` when the file has actually grown (or shrunk).
// On first seed we still emit — the World engine needs the initial
@@ -1106,6 +1102,68 @@ pub async fn world_live(
}));
}
// Per-agent LIVE column: REASONING STREAM + tool lines.
//
// These two taxonomy types were declared and listened for since the
// command centre shipped, and NOTHING ever emitted them — the cards
// could not populate no matter what an agent did. mission_events is
// the durable source: `reasoning` rows carry the agent's own step
// output, `tool.call` rows its actions.
//
// NOTE: on the container tier `tool.call` legitimately stays empty —
// those agents are tool-free behind the §15 door. Reasoning flows on
// every tier; tool lines appear where agents actually hold tools.
if agent_ev_cursor < 0 {
agent_ev_cursor = sqlx::query_scalar(
"SELECT coalesce(max(id), 0) FROM mission_events",
)
.fetch_one(&pool)
.await
.unwrap_or(0);
} else {
let rows = sqlx::query(
"SELECT e.id, e.agent_id, e.kind, e.target, e.detail
FROM mission_events e
JOIN missions m ON m.id = e.mission_id
WHERE m.workspace_id = $1
AND e.id > $2
AND e.agent_id IS NOT NULL
AND e.kind IN ('reasoning', 'tool.call')
ORDER BY e.id
LIMIT 200",
)
.bind(ws.as_uuid())
.bind(agent_ev_cursor)
.fetch_all(&pool)
.await
.unwrap_or_default();
for r in &rows {
let eid: i64 = r.get("id");
agent_ev_cursor = agent_ev_cursor.max(eid);
let agent_id: Option<uuid::Uuid> = r.get("agent_id");
let Some(agent_id) = agent_id else { continue };
let kind: String = r.get("kind");
let detail: serde_json::Value = r.get("detail");
let target: Option<String> = r.get("target");
if kind == "tool.call" {
yield sse("agent.tool.call", json!({
"agentId": agent_id.to_string(),
"tool": target.clone().unwrap_or_default(),
"target": detail.get("path").and_then(|v| v.as_str()),
}));
} else {
let text = detail.get("text").and_then(|v| v.as_str()).unwrap_or("");
if !text.is_empty() {
yield sse("agent.reasoning.delta", json!({
"agentId": agent_id.to_string(),
"text": text,
"channel": "say",
}));
}
}
}
}
// Workspace-wide telemetry (top-bar pills / Observe system strip).
yield sse(
"telemetry",
@@ -1274,7 +1332,10 @@ mod mission_feed_tests {
// finds. Written whole, the test fails on itself — which it did, and
// which is the same self-match that makes `pkill -f <pattern>` kill the
// shell carrying the pattern.
let needle = concat!("SELECT checkpoint FROM topology_", "runs WHERE id = $1::uuid");
let needle = concat!(
"SELECT checkpoint FROM topology_",
"runs WHERE id = $1::uuid"
);
assert!(
!src.contains(needle),
"the checkpoint tail is keyed on an agent_runs id and cannot match a \