feat(telemetry): record per-agent usage for mission turns

The command centre's SPEND, ACTIVITY and THROUGHPUT cards read `usage_events`,
and nothing on the mission path ever wrote a row: `cm_billing::charge` was
called only from the agent-run path. Measured mid-mission with 14 agents live,
`usage_events` was 0 while a crew had just burned 15k tokens — so an agent that
had done real work reported zero cost and zero activity.

The worker already knew everything needed: it logs node, role and token count
per step, and the node's `attrs.agent` carries the `claw_<uuid>` binding the
runtime dispatches on. This routes that to the ledger.

`charge`'s run_id is now Option. `usage_events.run_id` references `agent_runs`,
and a topology turn has no row there — passing its `topology_runs` id was a
foreign-key violation, which is exactly what the first attempt hit. NULL is the
honest value; the agent-run caller still passes its real id.

The executor reports one total rather than an in/out split, so the cost is right
(credits price the sum) and the columns record it as output rather than
inventing a split.

Verified end to end on a real mission: 4 agents, 1046-8670 tokens each, credits
attributed per agent, and the SPEND/ACTIVITY queries now return real numbers.

Co-Authored-By: Claude Opus 5 <[email protected]>

crates/cm-api/src/topology_worker.rs

@@ -201,12 +201,14 @@ async fn run_job(
     // to no mission — a bare topology run has no phase to hang tool calls on.
     let tap = mission_binding
         .as_ref()
-        .map(|(_, _, mission_id, phase_id)| crate::topology_exec::MissionTap {
-            pool: pool.clone(),
-            mission_id: *mission_id,
-            phase_id: *phase_id,
-            run_id: Some(id),
-        });
+        .map(
+            |(_, _, mission_id, phase_id)| crate::topology_exec::MissionTap {
+                pool: pool.clone(),
+                mission_id: *mission_id,
+                phase_id: *phase_id,
+                run_id: Some(id),
+            },
+        );
     let leaf_result = match mission_binding {
         Some((Some(url), Some(code), _, _)) => {
             ZeroClawDriveExecutor::from_env_for_gateway_with_code(url, code)
@@ -246,9 +248,29 @@ async fn run_job(
                 id,
                 Arc::new(leaf),
             );
-            drive(pool, id, &graph, &job.task, progress, &exec).await
+            drive(
+                pool,
+                id,
+                job.workspace_id,
+                &graph,
+                &job.task,
+                progress,
+                &exec,
+            )
+            .await
+        }
+        _ => {
+            drive(
+                pool,
+                id,
+                job.workspace_id,
+                &graph,
+                &job.task,
+                progress,
+                &leaf,
+            )
+            .await
         }
-        _ => drive(pool, id, &graph, &job.task, progress, &leaf).await,
     };
 
     finish(pool, id, result).await;
@@ -303,15 +325,14 @@ async fn run_composed(
         OrchestratorError::Executor("a composed run must belong to a mission phase".into())
     })?;
 
-    let mission: (Option<Uuid>, Option<String>, Option<String>, bool) =
-        sqlx::query_as(
-            "SELECT target_node_id, backend, team_engine, (repo_id IS NOT NULL) \
+    let mission: (Option<Uuid>, Option<String>, Option<String>, bool) = sqlx::query_as(
+        "SELECT target_node_id, backend, team_engine, (repo_id IS NOT NULL) \
              FROM missions WHERE id = $1",
-        )
-            .bind(mission_id)
-            .fetch_one(pool)
-            .await
-            .map_err(|e| OrchestratorError::Executor(format!("load mission {mission_id}: {e}")))?;
+    )
+    .bind(mission_id)
+    .fetch_one(pool)
+    .await
+    .map_err(|e| OrchestratorError::Executor(format!("load mission {mission_id}: {e}")))?;
 
     // The phase's completion gate, read here rather than carried on the run row
     // so an edited `done_when_check` takes effect on the next node instead of at
@@ -347,7 +368,16 @@ async fn run_composed(
             completed_steps: progress.completed as u32,
         },
     );
-    drive(pool, job.id, graph, &job.task, progress, &exec).await
+    drive(
+        pool,
+        job.id,
+        job.workspace_id,
+        graph,
+        &job.task,
+        progress,
+        &exec,
+    )
+    .await
 }
 
 /// Post-terminal hook: if this run's team is `ephemeral` and no siblings are
@@ -406,14 +436,30 @@ async fn maybe_teardown_ephemeral_team(pool: &PgPool, runtime: &cm_runtime::Runt
 async fn drive<E: TurnExecutor>(
     pool: &PgPool,
     id: Uuid,
+    workspace_id: Uuid,
     graph: &TopologyGraph,
     task: &str,
     progress: RunProgress,
     executor: &E,
 ) -> Result<RunRecord, OrchestratorError> {
     let pool_cb = pool.clone();
+    // node_id -> agent id, resolved once. The binding lives in the node's
+    // attrs (`agent = claw_<uuid>`), which is also what the runtime dispatches
+    // on — so usage is attributed to exactly the claw that did the work.
+    let agent_of: std::sync::Arc<std::collections::HashMap<String, Uuid>> = std::sync::Arc::new(
+        graph
+            .nodes
+            .iter()
+            .filter_map(|n| {
+                let alias = n.attrs.get("agent")?;
+                let uuid = alias.strip_prefix("claw_")?;
+                Some((n.id.clone(), Uuid::parse_str(uuid).ok()?))
+            })
+            .collect(),
+    );
     execute_resumable(graph, task, executor, progress, move |snap| {
         let pool = pool_cb.clone();
+        let agent_of = agent_of.clone();
         async move {
             // 2026-07-15: verbose per-step trace so `docker logs
             // clawmates_server_1` shows which topology node just fired,
@@ -439,6 +485,34 @@ async fn drive<E: TurnExecutor>(
                     last.tokens,
                     last.gated.len(),
                 );
+                // Per-agent usage. Without this the command centre's SPEND,
+                // ACTIVITY and THROUGHPUT cards read `usage_events`, which
+                // nothing on the mission path ever wrote — so they showed 0 for
+                // an agent that had just burned 15k tokens.
+                //
+                // `charge` also decrements credit lots, which is the point: a
+                // mission turn costs what it costs. It clamps at the available
+                // balance and still records the full obligation, so an empty
+                // wallet cannot fail a turn.
+                if let Some(agent_id) = agent_of.get(&last.node_id).copied() {
+                    if last.tokens > 0 {
+                        // The executor reports ONE total, not an in/out split.
+                        // Credits price the sum, so cost is right; the columns
+                        // record it as output rather than inventing a split.
+                        if let Err(e) = cm_billing::charge(
+                            &pool,
+                            cm_domain::WorkspaceId::from(workspace_id),
+                            cm_domain::AgentId::from(agent_id),
+                            None,
+                            0,
+                            last.tokens as u64,
+                        )
+                        .await
+                        {
+                            eprintln!("topology_worker: usage for {agent_id} failed: {e}");
+                        }
+                    }
+                }
             }
             // Best-effort checkpoint: a failed write just means we re-run the
             // step on resume (idempotent — topology turns are pure reads here).