feat(missions): attribute a phase's tool calls to the agent that made them

`record_vm_tools` wrote `agent_id: None` for every call. The container
tap is per-CONTAINER and every role in a phase shares one, so a phase
arrived as one undifferentiated stream: every Skill-Use score was
per-mission rather than per-role, and the World's per-agent view got
nothing from this tier.

One `claude -p` invocation is one turn is one agent, and Claude Code
stamps each invocation with a `session_id` the tap was discarding. So the
distinct sessions, in order of first appearance, are the phase's turns in
the order they ran — and `prompt.composed` already records the agent of
each turn in that same order, written by the tier as it sends each turn,
so it IS the running order rather than a reconstruction of it.

**It attributes nothing rather than guessing.** Only when the counts
match exactly. A phase whose sessions and turns differ has something this
correlation does not model — a retry, a turn that called no tool, two
genuinely concurrent agents — and a plausible-looking wrong attribution
is worse than none here: it puts one agent's `git push` on another
agent's record, and a person later reasons from that. One call missing a
session id refuses the whole batch, because a hole shifts every later
session onto the wrong turn.

The microVM call sites pass no turn agents and so keep today's
behaviour exactly. Resolving a graph node to an agent uuid is the fix
there, it cannot be tested while the fleet is offline, and guessing would
put one node's actions on another node's record.

Also restores the `#[cfg(test)]` gate on `repo_less_text_tests`, which my
own insertion had taken — those tests would have compiled into release
builds.

Co-Authored-By: Claude Opus 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_018i9Ten1LU4jUr5d7TAWda9

crates/cm-api/src/phase_runner.rs

@@ -309,6 +309,80 @@ mod skill_delivery_wiring_tests {
     }
 }
 
+#[cfg(test)]
+mod attribution_tests {
+    use super::attribute_sessions;
+    use crate::vm_tool_tap::Observed;
+    use uuid::Uuid;
+
+    fn call(session: Option<&str>) -> Observed {
+        Observed {
+            tool: "Bash".into(),
+            path: None,
+            session: session.map(str::to_string),
+            input: serde_json::json!({"command": "ls"}),
+        }
+    }
+
+    /// The ordinary case: three turns, three sessions, in order.
+    #[test]
+    fn each_session_lands_on_the_turn_that_ran_it() {
+        let (a, b, c) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7());
+        let tools = [
+            call(Some("s1")),
+            call(Some("s1")),
+            call(Some("s2")),
+            call(Some("s3")),
+            call(Some("s2")),
+        ];
+        assert_eq!(
+            attribute_sessions(&tools, &[a, b, c]),
+            vec![Some(a), Some(a), Some(b), Some(c), Some(b)],
+            "sessions are ordered by FIRST appearance, so a later call from an \
+             earlier session still belongs to that earlier turn"
+        );
+    }
+
+    /// More sessions than turns — something happened this model does not
+    /// describe, so it must not produce a confident answer.
+    #[test]
+    fn a_count_mismatch_attributes_nothing() {
+        let a = Uuid::now_v7();
+        let tools = [call(Some("s1")), call(Some("s2"))];
+        assert_eq!(
+            attribute_sessions(&tools, &[a]),
+            vec![None, None],
+            "a plausible-looking wrong attribution puts one agent's actions on \
+             another agent's record, and a person later reasons from it"
+        );
+        // And the other direction.
+        assert_eq!(
+            attribute_sessions(&[call(Some("s1"))], &[a, Uuid::now_v7()]),
+            vec![None]
+        );
+    }
+
+    /// One call with no session id poisons the ORDER, not just itself.
+    #[test]
+    fn a_single_missing_session_refuses_the_whole_batch() {
+        let (a, b) = (Uuid::now_v7(), Uuid::now_v7());
+        let tools = [call(Some("s1")), call(None), call(Some("s2"))];
+        assert_eq!(
+            attribute_sessions(&tools, &[a, b]),
+            vec![None, None, None],
+            "a hole shifts every later session onto the wrong turn"
+        );
+    }
+
+    /// The pre-session tap, and the microVM path that supplies no turns.
+    #[test]
+    fn no_turns_and_no_sessions_stay_unattributed() {
+        assert_eq!(attribute_sessions(&[call(Some("s1"))], &[]), vec![None]);
+        assert_eq!(attribute_sessions(&[call(None)], &[]), vec![None]);
+        assert!(attribute_sessions(&[], &[]).is_empty());
+    }
+}
+
 #[cfg(test)]
 mod repo_less_text_tests {
     use super::*;
@@ -567,12 +641,26 @@ async fn drain_finished_container_phases(pool: &PgPool) -> Result<(), String> {
                 .await
                 .ok()
                 .flatten();
+        // The agent of each turn, in the order the turns ran. `prompt.composed`
+        // is written by the tier as it sends each turn, so this IS the running
+        // order — not a reconstruction of it.
+        let turn_agents: Vec<Uuid> = sqlx::query_scalar(
+            "SELECT agent_id FROM mission_events
+              WHERE phase_id = $1 AND kind = $2 AND agent_id IS NOT NULL
+              ORDER BY id",
+        )
+        .bind(phase_id)
+        .bind(crate::mission_events::PROMPT_COMPOSED)
+        .fetch_all(pool)
+        .await
+        .unwrap_or_default();
         record_vm_tools(
             pool,
             mission_id,
             phase_id,
             run_id.unwrap_or(phase_id),
             &tools,
+            &turn_agents,
         )
         .await;
         eprintln!(
@@ -1598,7 +1686,7 @@ async fn launch_microvm_phase(
         // still records; recording the same calls twice is what the empty
         // contract exists to prevent.
         if let Ok(o) = &outcome {
-            record_vm_tools(&pool2, mission_id, phase_id, run_id, &o.tools).await;
+            record_vm_tools(&pool2, mission_id, phase_id, run_id, &o.tools, &[]).await;
         }
         let (status, note) = match outcome {
             // The gate gave up. It is the ONLY thing that runs a
@@ -2052,29 +2140,94 @@ pub(crate) fn vm_tool_recorder(
     let pool = pool.clone();
     tokio::spawn(async move {
         while let Some(batch) = rx.recv().await {
-            record_vm_tools(&pool, mission_id, phase_id, run_id, &batch).await;
+            record_vm_tools(&pool, mission_id, phase_id, run_id, &batch, &[]).await;
         }
     });
     tx
 }
 
+/// Which agent each observed call belongs to, by session.
+///
+/// The tap is per-CONTAINER and every role in a phase shares one, so a phase's
+/// calls arrive as one undifferentiated stream and `agent_id` was written
+/// `None` for all of them. That is why every Skill-Use score is per-mission
+/// rather than per-role, and why the World's per-agent view gets nothing from
+/// this tier.
+///
+/// One `claude -p` invocation is one turn is one agent, and Claude Code stamps
+/// each invocation with a `session_id`. So the distinct sessions, in the order
+/// they first appear, are the phase's turns in the order they ran — and
+/// `prompt.composed` already records the agent of each turn in that same order.
+///
+/// # It attributes nothing rather than guessing
+///
+/// Only when the counts match exactly. A phase whose sessions and turns differ
+/// in number has something this correlation does not model — a retry, a turn
+/// that called no tool, two agents genuinely concurrent — and a
+/// plausible-looking wrong attribution is worse here than none: it would put
+/// one agent's `git push` on another agent's record, which is the sort of thing
+/// a person later reasons from.
+pub(crate) fn attribute_sessions(
+    tools: &[crate::vm_tool_tap::Observed],
+    turn_agents: &[Uuid],
+) -> Vec<Option<Uuid>> {
+    let mut order: Vec<&str> = Vec::new();
+    for t in tools {
+        let Some(sid) = t.session.as_deref() else {
+            // A single unattributable call means the sequence has a hole in it,
+            // and a hole shifts every later session onto the wrong turn.
+            return vec![None; tools.len()];
+        };
+        if !order.contains(&sid) {
+            order.push(sid);
+        }
+    }
+    if order.len() != turn_agents.len() || order.is_empty() {
+        return vec![None; tools.len()];
+    }
+    tools
+        .iter()
+        .map(|t| {
+            let sid = t.session.as_deref()?;
+            let idx = order.iter().position(|s| *s == sid)?;
+            turn_agents.get(idx).copied()
+        })
+        .collect()
+}
+
 pub(crate) async fn record_vm_tools(
     pool: &PgPool,
     mission_id: Uuid,
     phase_id: Uuid,
     run_id: Uuid,
     tools: &[crate::vm_tool_tap::Observed],
+    turn_agents: &[Uuid],
 ) {
     if tools.is_empty() {
         return;
     }
+    let owners = attribute_sessions(tools, turn_agents);
+    if owners.iter().all(Option::is_none) && !turn_agents.is_empty() {
+        eprintln!(
+            "phase_runner: {} tool call(s) for phase {phase_id} could not be \
+             attributed to an agent ({} session(s) across {} turn(s)) — recorded \
+             unattributed rather than guessed",
+            tools.len(),
+            tools
+                .iter()
+                .filter_map(|t| t.session.as_deref())
+                .collect::<std::collections::HashSet<_>>()
+                .len(),
+            turn_agents.len()
+        );
+    }
     let mut events = Vec::new();
-    for t in tools {
+    for (t, owner) in tools.iter().zip(owners) {
         events.push(crate::mission_events::MissionEvent {
             mission_id,
             phase_id: Some(phase_id),
             run_id: Some(run_id),
-            agent_id: None,
+            agent_id: owner,
             kind: crate::mission_events::TOOL_CALL.to_string(),
             target: Some(t.tool.clone()),
             // `path` because the World's SSE reads `detail.path` for this kind
@@ -2088,7 +2241,7 @@ pub(crate) async fn record_vm_tools(
                 mission_id,
                 phase_id: Some(phase_id),
                 run_id: Some(run_id),
-                agent_id: None,
+                agent_id: owner,
                 kind: crate::mission_events::FILE_TOUCH.to_string(),
                 target: Some(crate::mission_events::repo_relative(
                     path,