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

fix(world): missions were never really on the wire

Browse Source 
Three bugs in one query, each hiding the next, plus one that made the whole
rich layer dead code.

`active_missions` joined `team_members` on `missions.team_id` — the LEGACY
pointer at the first minted team, superseded by the `mission_teams` junction in
0056. It was an INNER JOIN, and `mission_orchestrator::on_launch` deliberately
mints no team for a microVM mission, so the platform's primary execution tier
was dropped by a join and the World has been showing nothing at all for it. And
it selected only `status='running'`, while missions finish in minutes, so the
scene was empty almost always.

Now: join `mission_teams`, LEFT so teamless missions survive (their `agent_id`
is NULL and no pawn beams at them, which is the truth — nothing on this
platform ran that phase except a VM), and include missions finished in the last
24h carrying `status`/`template_kind` so the client can draw a finished map
instead of animating a corpse. `?mission=` scopes the feed server-side.

The whole phase plan now ships as `mission.phase`, including phases that have
not started: a phase list that appeared only as phases began made a five-phase
mission look like a one-phase mission until it was nearly over. Attribution
reuses `phase_runner::purposes_for` rather than copying it — two copies would
let the picture disagree with the machine about who is working on what, which
presents as a rendering bug and is really a lie.

Deleted the checkpoint tail. It read `topology_runs` keyed by an `agent_runs`
id; mission phases live in `topology_runs` under independently generated ids,
so it ran every poll and matched nothing, for every mission, forever. That is
why no mission has ever shown tool or file activity. Not repointed at
`topology_runs`: its only per-step content is the agent's own prose, and a tool
name in prose cannot be told from an agent talking about a tool. The a2a
`run_events` tail is kept — it genuinely works for the path that writes it.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Omar Sobh
2026-08-10 17:58:38 -07:00
co-authored by Claude Opus 5
parent b210acf3c2
commit 5f85dbb718
3 changed files with 416 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/cm-api/src/routes/world.rs
+348 -83
View File
@@ -23,6 +23,7 @@ use cm_domain::WorkspaceId;
use serde::Deserialize;
use serde_json::{json, Value};
use sqlx::{PgPool, Row};
use uuid::Uuid;
use crate::{ApiError, AppState, Authed};
@@ -44,36 +45,169 @@ async fn working_agents(pool: &PgPool, ws: WorkspaceId) -> HashSet<String> {
rows.into_iter().map(|r| r.get::<String, _>("id")).collect()
}
/// Active missions (status='running') with their assigned team members —
/// one row per (mission, agent) pair. The World SSE loop emits each as
/// a `mission:<id>` landmark orb + `world.touch` beams from every team
/// member. Replaces the retired research/loops landmarks (commit
/// fdb8cfe) with the missions-era equivalent.
async fn active_missions(pool: &PgPool, ws: WorkspaceId) -> Vec<(String, String, String)> {
/// One row per (mission, agent) pair for every mission the World should draw.
///
/// Three bugs were fixed here at once, and each hid the next:
///
/// 1. **The join was on the wrong column.** It read
/// `JOIN team_members tm ON tm.team_id = m.team_id`, but `missions.team_id`
/// is a legacy pointer at the FIRST minted team — `0056_mission_teams.sql`
/// superseded it with the `mission_teams(mission_id, team_id, purpose)`
/// junction, which is what everything else (including `/api/workforce`)
/// joins through. A multi-team mission showed only its first team.
///
/// 2. **It was an INNER JOIN, and microVM missions have no team at all.**
/// `mission_orchestrator::on_launch` deliberately mints none for them, so
/// the platform's primary execution tier was dropped by the join and the
/// World has been showing nothing whatsoever for it. LEFT JOIN, and
/// `agent_id` is `None` for those — a phase that ran with no platform
/// agents draws no pawns, which is the truth rather than a gap.
///
/// 3. **Only `running` missions were selected.** Missions finish in minutes,
/// so the World was empty almost always. Recently-finished ones come back
/// too, carrying `status` so the client can draw a finished map instead of
/// animating a corpse.
struct MissionRow {
mission_id: String,
title: String,
status: String,
template_kind: String,
completed_at: Option<String>,
/// `None` for a mission with no team — see (2) above.
agent_id: Option<String>,
}
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,
tm.claw_id::text AS agent_id
"SELECT m.id::text AS mission_id,
m.title AS title,
m.status AS status,
m.template_kind AS template_kind,
to_char(m.completed_at AT TIME ZONE 'UTC',
'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') AS completed_at,
tm.claw_id::text AS agent_id
FROM missions m
JOIN team_members tm ON tm.team_id = m.team_id
LEFT JOIN mission_teams mt ON mt.mission_id = m.id
LEFT JOIN team_members tm ON tm.team_id = mt.team_id
WHERE m.workspace_id = $1
AND m.status = 'running'",
AND ( m.status = 'running'
OR ( m.status IN ('completed', 'failed')
AND m.completed_at > now() - interval '24 hours' ) )
AND ($2::uuid IS NULL OR m.id = $2)
ORDER BY m.created_at DESC",
)
.bind(ws.as_uuid())
.bind(only)
.fetch_all(pool)
.await
.unwrap_or_default();
rows.into_iter()
.map(|r| {
(
r.get::<String, _>("mission_id"),
r.get::<String, _>("title"),
r.get::<String, _>("agent_id"),
)
.map(|r| MissionRow {
mission_id: r.get::<String, _>("mission_id"),
title: r.get::<String, _>("title"),
status: r.get::<String, _>("status"),
template_kind: r.get::<String, _>("template_kind"),
completed_at: r.get::<Option<String>, _>("completed_at"),
agent_id: r.get::<Option<String>, _>("agent_id"),
})
.collect()
}
/// Every phase of the given missions, in plan order, with the agents that
/// execute it.
///
/// The whole plan is emitted — including `pending` phases that have not
/// started — because the World draws the mission's shape upfront and lights it
/// as it progresses. A phase list that only appeared as phases began would make
/// a five-phase mission indistinguishable from a one-phase mission until it was
/// nearly over.
struct PhaseRow {
mission_id: String,
phase_id: String,
kind: String,
order_idx: i32,
status: String,
iteration: i32,
started_at: Option<String>,
completed_at: Option<String>,
}
async fn world_phases(pool: &PgPool, ws: WorkspaceId, only: Option<Uuid>) -> Vec<PhaseRow> {
let rows = sqlx::query(
"SELECT mp.mission_id::text AS mission_id,
mp.id::text AS phase_id,
mp.kind AS kind,
mp.order_idx AS order_idx,
mp.status AS status,
COALESCE(mp.iteration, 0) AS iteration,
to_char(mp.started_at AT TIME ZONE 'UTC',
'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') AS started_at,
to_char(mp.completed_at AT TIME ZONE 'UTC',
'YYYY-MM-DD\"T\"HH24:MI:SS\"Z\"') AS completed_at
FROM mission_phases mp
JOIN missions m ON m.id = mp.mission_id
WHERE m.workspace_id = $1
AND ( m.status = 'running'
OR ( m.status IN ('completed', 'failed')
AND m.completed_at > now() - interval '24 hours' ) )
AND ($2::uuid IS NULL OR m.id = $2)
ORDER BY mp.mission_id, mp.order_idx",
)
.bind(ws.as_uuid())
.bind(only)
.fetch_all(pool)
.await
.unwrap_or_default();
rows.into_iter()
.map(|r| PhaseRow {
mission_id: r.get::<String, _>("mission_id"),
phase_id: r.get::<String, _>("phase_id"),
kind: r.get::<String, _>("kind"),
order_idx: r.get::<i32, _>("order_idx"),
status: r.get::<String, _>("status"),
iteration: r.get::<i32, _>("iteration"),
started_at: r.get::<Option<String>, _>("started_at"),
completed_at: r.get::<Option<String>, _>("completed_at"),
})
.collect()
}
/// 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> {
let Ok(mid) = Uuid::parse_str(mission_id) else {
return Vec::new();
};
// `purposes_for` is the runner's own mapping, shared rather than copied —
// see its doc comment.
let purposes: Vec<String> = crate::phase_runner::purposes_for(kind)
.iter()
.map(|s| s.to_string())
.collect();
let rows = sqlx::query(
"SELECT DISTINCT tm.claw_id::text AS agent_id
FROM mission_teams mt
JOIN team_members tm ON tm.team_id = mt.team_id
WHERE mt.mission_id = $1 AND mt.purpose = ANY($2)",
)
.bind(mid)
.bind(&purposes)
.fetch_all(pool)
.await
.unwrap_or_default();
rows.into_iter()
.map(|r| r.get::<String, _>("agent_id"))
.collect()
}
/// Currently-running runs in the workspace as (run_id, agent_id) — each is a
/// real "this agent is converging on its active work" signal (Gource).
async fn active_runs(pool: &PgPool, ws: WorkspaceId) -> Vec<(String, String)> {
@@ -330,10 +464,24 @@ async fn agent_telemetry(
m
}
/// Query for `GET /api/world/live`.
#[derive(serde::Deserialize)]
pub struct LiveQuery {
/// Scope the feed to one mission. The client already filters events by
/// mission, but doing it here means the server stops querying and sending
/// what the viewer will discard.
pub mission: Option<Uuid>,
}
/// `GET /api/world/live` — the taxonomy SSE feed.
pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) -> impl IntoResponse {
pub async fn world_live(
State(state): State<AppState>,
Authed(user): Authed,
Query(q): Query<LiveQuery>,
) -> impl IntoResponse {
let pool = state.pool.clone();
let ws = user.workspace_id;
let only_mission = q.mission;
let stream = async_stream::stream! {
let mut first = true;
@@ -341,10 +489,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();
// How many checkpoint step-records of each run have already been sent.
// Separate from `cursors`: that tracks `run_events.seq`, and the two
// sources are populated by different paths — see the loop below.
let mut step_cursors: std::collections::HashMap<String, usize> =
// Last emitted signature per phase, so the plan is sent once and then
// only when a phase actually moves.
let mut last_phase: std::collections::HashMap<String, String> =
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.
@@ -396,26 +543,98 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
}
}
// Mission landmarks: one `mission:<id>` orb per running mission,
// with `world.touch` beams from every assigned team member. Missions
// outlive individual runs, so the orb gives the World a persistent
// pin for "this is what the team is working on right now" even when
// no run is claimed. Replaces the retired repo:{topic}/loop:{id}
// landmarks after commit fdb8cfe.
let missions = active_missions(&pool, ws).await;
// Mission landmarks, and the whole phase plan behind them.
//
// The orb pins "this is the work"; the phases give it shape. Every
// phase is emitted, including ones that have not started, because
// the World draws the plan upfront and lights it as it progresses —
// a phase list that appeared only as phases began would make a
// five-phase mission indistinguishable from a one-phase mission
// until it was nearly over.
let missions = world_missions(&pool, ws, only_mission).await;
let mut seen_missions: HashSet<String> = HashSet::new();
for (mission_id, title, agent_id) in &missions {
if seen_missions.insert(mission_id.clone()) {
let node_id = format!("mission:{mission_id}");
for m in &missions {
if seen_missions.insert(m.mission_id.clone()) {
yield sse(
"mission.update",
json!({
"missionId": m.mission_id,
"title": m.title,
"status": m.status,
"templateKind": m.template_kind,
"completedAt": m.completed_at,
}),
);
// A terminal mission is a map to read, not a scene to
// animate: it still gets an orb, but no heat.
let running = m.status == "running";
yield sse(
"node.activity",
json!({ "nodeId": node_id, "label": title, "kind": "mission", "heat": 0.75 }),
json!({
"nodeId": format!("mission:{}", m.mission_id),
"label": m.title,
"kind": "mission",
"heat": if running { 0.75 } else { 0.0 },
}),
);
}
let node_id = format!("mission:{mission_id}");
// `agent_id` is None for a teamless microVM mission. No pawn
// beams at it, which is accurate — nothing on this platform ran
// that phase except a VM — and is why the join had to become a
// LEFT JOIN rather than being left to drop the mission entirely.
if let (Some(agent_id), true) = (&m.agent_id, m.status == "running") {
yield sse(
"world.touch",
json!({
"agentId": agent_id,
"nodeId": format!("mission:{}", m.mission_id),
"kind": "mission",
"weight": 0.6,
}),
);
}
}
for p in world_phases(&pool, ws, only_mission).await {
// Re-emit only on change. `iteration` and `completed_at` are in
// the signature because a phase that loops re-enters `running`
// from `running` — status alone would hide the retry.
let sig = format!(
"{}|{}|{}",
p.status,
p.iteration,
p.completed_at.as_deref().unwrap_or("")
);
let changed = last_phase.get(&p.phase_id).map(|s| s != &sig).unwrap_or(true);
if !changed {
continue;
}
last_phase.insert(p.phase_id.clone(), sig);
let agents = phase_agents(&pool, &p.mission_id, &p.kind).await;
yield sse(
"world.touch",
json!({ "agentId": agent_id, "nodeId": node_id, "kind": "mission", "weight": 0.6 }),
"mission.phase",
json!({
"missionId": p.mission_id,
"phaseId": p.phase_id,
"kind": p.kind,
"orderIdx": p.order_idx,
"status": p.status,
"iteration": p.iteration,
"startedAt": p.started_at,
"completedAt": p.completed_at,
"agentIds": agents,
}),
);
yield sse(
"node.activity",
json!({
"nodeId": format!("phase:{}", p.phase_id),
"label": p.kind,
"kind": "phase",
// The client owns the lit/dim encoding via `status`;
// heat here is only the arrival pulse.
"heat": if p.status == "running" { 0.8 } else { 0.0 },
}),
);
}
@@ -431,54 +650,23 @@ pub async fn world_live(State(state): State<AppState>, Authed(user): Authed) ->
// Tail the run's journal for richer real events (reasoning, tool
// convergence, doors). On first sight, jump the cursor to the
// current max so we stream forward without replaying the backlog.
// Mission runs write NO `run_events`: `run_events::append` is called
// only from the a2a path. Their per-step detail lives in
// `topology_runs.checkpoint.records`, which is what
// `topology::run_events_sse` has always streamed. So the World's
// rich layer was empty for every mission on the platform — not
// because the data was missing, but because this read the one
// source missions never write.
//
// Both sources are tailed, because both are real: a2a runs
// populate the table, topology runs populate the checkpoint.
let steps: Vec<Value> = sqlx::query_scalar::<_, Option<Value>>(
"SELECT checkpoint FROM topology_runs WHERE id = $1::uuid",
)
.bind(run_id)
.fetch_optional(&pool)
.await
.ok()
.flatten()
.flatten()
.and_then(|c| c.get("records").cloned())
.and_then(|r| r.as_array().cloned())
.unwrap_or_default();
let already = *step_cursors.get(run_id).unwrap_or(&0);
if already < steps.len() {
for rec in &steps[already..] {
// A node turn IS a step. `step_started` is what
// `normalize_run_event` already maps to a tool call, so
// the viz needs no new event vocabulary.
let role = rec
.get("role")
.or_else(|| rec.get("node_id"))
.and_then(|v| v.as_str())
.unwrap_or("turn");
for (t, d) in normalize_run_event(
agent_id,
"step_started",
&json!({ "tool": role, "input": rec.get("output").cloned().unwrap_or(Value::Null) }),
) {
yield sse(t, d);
}
}
step_cursors.insert(run_id.clone(), steps.len());
} else if !step_cursors.contains_key(run_id) {
// First sight: jump to the end rather than replaying the
// backlog, matching how the run_events cursor below behaves.
step_cursors.insert(run_id.clone(), steps.len());
}
// THIS BLOCK IS THE a2a/door LAYER ONLY, and it works: those
// runs are the ones that write `run_events` (`run_events::append`
// is reached from `routes/a2a.rs` and `mcp_door.rs`).
//
// A sibling block used to read `topology_runs.checkpoint.records`
// here, on the theory that it covered missions. It could never
// have: `run_id` comes from `active_runs`, which selects
// `agent_runs.id`, and mission phases live in `topology_runs`
// under independently generated ids. The query ran every poll and
// matched nothing, for every mission, forever — which is why the
// World has never shown a mission's tool or file activity.
// Removed rather than repointed: pointing it at `topology_runs`
// would resurrect a path whose only per-step content is the
// agent's own prose output, and a tool name in prose is
// indistinguishable from an agent talking about a tool. Mission
// detail arrives as structured `mission_events` instead.
if let Some(&after) = cursors.get(run_id) {
let rows = sqlx::query(
"SELECT seq, event_type, payload FROM run_events
@@ -639,3 +827,80 @@ pub async fn world_replay(
json!({ "events": events, "hours": hours, "count": rows.len() }),
))
}
#[cfg(test)]
mod mission_feed_tests {
/// The mission query must not re-acquire any of the three bugs it fixed.
///
/// All three were invisible: an INNER JOIN does not error when it drops a
/// row, the legacy `missions.team_id` is a real column that resolves fine,
/// and `status = 'running'` is a defensible-looking filter. The World simply
/// showed less than the truth and nothing anywhere said so. A source walk is
/// the only guard available — these are runtime `sqlx::query` calls, so the
/// compiler checks nothing about them.
#[test]
fn the_mission_query_keeps_teamless_missions() {
let src = include_str!("world.rs");
let q = src
.split("async fn world_missions")
.nth(1)
.and_then(|s| s.split("async fn").next())
.expect("world_missions body");
assert!(
q.contains("LEFT JOIN mission_teams"),
"must LEFT JOIN mission_teams: a microVM mission has no team rows at \
all, and an INNER JOIN silently drops the platform's primary \
execution tier"
);
assert!(
q.contains("LEFT JOIN team_members"),
"the second join must be LEFT too, or the mission reappears and its \
agents take it back out"
);
assert!(
!q.contains("tm.team_id = m.team_id"),
"must not join on the legacy missions.team_id — 0056 superseded it \
with mission_teams, and it points at only the FIRST minted team"
);
assert!(
q.contains("m.completed_at >"),
"recently-finished missions must be included or the World is empty \
almost always: missions finish in minutes"
);
assert!(
q.contains("m.status") && q.contains("m.template_kind"),
"status and template_kind must be carried: one decides finished-map \
vs live, the other the palette"
);
}
/// The checkpoint tail read `topology_runs` keyed by an `agent_runs` id and
/// could never match. If it comes back, missions silently lose their tool
/// and file detail again — and the query looks entirely reasonable.
#[test]
fn the_dead_checkpoint_tail_stays_dead() {
let src = include_str!("world.rs");
// Split so this assertion's own literal is not what the source walk
// 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");
assert!(
!src.contains(needle),
"the checkpoint tail is keyed on an agent_runs id and cannot match a \
topology_runs row; mission detail comes from structured events"
);
}
/// Agent→phase attribution must come from the runner's own mapping.
#[test]
fn phase_attribution_reuses_the_runners_mapping() {
let src = include_str!("world.rs");
assert!(
src.contains("crate::phase_runner::purposes_for"),
"a second copy of the kind→purpose mapping would let the picture \
disagree with the machine about who is working on what"
);
}
}