feat(viz): kind-specific choreography and a finished mission you can read

Security: the pawns already orbited their destination, so homing them at
the security station gave circling for free. This adds the radial
press-and-retreat — an agent closing on the target and backing off reads
as probing it, where a fixed radius reads as waiting — and holds the
stochastic target release while probing, or the circling breaks up into
stray trips that look like distraction rather than a scan.

Findings are `mission_tasks` rows, one orb each, popped once. There is
deliberately no severity anywhere in the path: the scanner keeps
severity, file and line as substrings inside `title`, so a severity
parsed out of prose and rendered as an orb's RADIUS would be the picture
asserting a measurement the data never contained. Count only.

Benchmarks annotate the station, as text. `delta` has no schema —
compute_delta emits `{kind:"opaque"}` whenever the before/after metrics
were not structurally comparable, which is most drivers. The server
formats the shape it can parse and COUNTS the rest; an unparseable driver
reports "3 sample(s)" rather than an invented improvement, and an opaque
delta says nothing at all.

The finished map: the live label rule gates service/event nodes on
`heat > 0.12`, which is exactly backwards once everything has cooled — a
static map would be unlabelled dots. Frozen, the 25 most-touched nodes
label regardless of heat, phase stations carry a second line counting
what they produced, and the camera is released ONCE so it frames the
result even if the user panned during the run.

Every count in a caption is read off the drawn scene rather than a
parallel tally, so the words and the picture cannot disagree.

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

crates/cm-api/src/routes/world.rs

@@ -283,6 +283,137 @@ async fn world_acts(pool: &PgPool, ws: WorkspaceId, only: Option<Uuid>, after: i
         .collect()
 }
 
+/// One security finding, as the scanner recorded it.
+struct FindingRow {
+    mission_id: String,
+    phase_id: String,
+    task_id: String,
+    title: String,
+}
+
+/// Findings raised by a security phase.
+///
+/// They are `mission_tasks` rows — the scanner's own store — not a new table.
+/// There is deliberately NO severity field here: severity, file and line are
+/// substrings inside `title` (see `security_scan.rs`), and a severity parsed
+/// out of prose and then encoded as an orb's RADIUS would be an invented fact
+/// rendered as a measurement. The title is shown as written.
+async fn world_findings(pool: &PgPool, ws: WorkspaceId, only: Option<Uuid>) -> Vec<FindingRow> {
+    let rows = sqlx::query(
+        "SELECT t.mission_id::text AS mission_id,
+                t.phase_id::text   AS phase_id,
+                t.id::text         AS task_id,
+                t.title
+           FROM mission_tasks t
+           JOIN mission_phases p ON p.id = t.phase_id
+           JOIN missions m ON m.id = t.mission_id
+          WHERE m.workspace_id = $1
+            AND p.kind = 'security_scan'
+            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)
+          LIMIT 300",
+    )
+    .bind(ws.as_uuid())
+    .bind(only)
+    .fetch_all(pool)
+    .await
+    .unwrap_or_default();
+    rows.into_iter()
+        .map(|r| FindingRow {
+            mission_id: r.get::<String, _>("mission_id"),
+            phase_id: r.get::<String, _>("phase_id"),
+            task_id: r.get::<String, _>("task_id"),
+            title: r.get::<String, _>("title"),
+        })
+        .collect()
+}
+
+/// A benchmark phase's result, summarised.
+struct BenchRow {
+    mission_id: String,
+    phase_id: String,
+    note: String,
+}
+
+/// Benchmark deltas, as an annotation on the phase — not as nodes.
+///
+/// `delta` is a `Record<string, unknown>` with no schema: `compute_delta`
+/// produces `{kind:"bencher_diff", samples:[…]}` when the before/after shapes
+/// are structurally comparable, and `{kind:"opaque"}` when they are not. Only
+/// the shape that can be parsed is formatted; the rest is COUNTED, never
+/// guessed at, so a driver whose output we do not understand reports "3
+/// samples" rather than an invented improvement.
+async fn world_benchmarks(pool: &PgPool, ws: WorkspaceId, only: Option<Uuid>) -> Vec<BenchRow> {
+    let rows = sqlx::query(
+        "SELECT DISTINCT ON (b.phase_id)
+                b.mission_id::text AS mission_id,
+                b.phase_id::text   AS phase_id,
+                b.delta
+           FROM benchmark_snapshots b
+           JOIN missions m ON m.id = b.mission_id
+          WHERE m.workspace_id = $1
+            AND b.delta IS NOT NULL
+            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 b.phase_id, b.iteration DESC",
+    )
+    .bind(ws.as_uuid())
+    .bind(only)
+    .fetch_all(pool)
+    .await
+    .unwrap_or_default();
+    rows.into_iter()
+        .filter_map(|r| {
+            let note = benchmark_note(&r.get::<serde_json::Value, _>("delta"))?;
+            Some(BenchRow {
+                mission_id: r.get::<String, _>("mission_id"),
+                phase_id: r.get::<String, _>("phase_id"),
+                note,
+            })
+        })
+        .collect()
+}
+
+/// One line describing a benchmark delta, or `None` if there is nothing
+/// truthful to say about it.
+fn benchmark_note(delta: &serde_json::Value) -> Option<String> {
+    let samples = delta.get("samples").and_then(|s| s.as_array())?;
+    if samples.is_empty() {
+        return None;
+    }
+    let mut improved = 0usize;
+    let mut regressed = 0usize;
+    // Best (most negative) percent change, since that is the one claim the
+    // shape actually supports.
+    let mut best: Option<f64> = None;
+    for s in samples {
+        match s.get("direction").and_then(|d| d.as_str()) {
+            Some("improved") => improved += 1,
+            Some("regressed") => regressed += 1,
+            // Counted in the total but claimed for neither side.
+            _ => {}
+        }
+        if let Some(pct) = s.get("delta_pct").and_then(serde_json::Value::as_f64) {
+            best = Some(best.map_or(pct, |b: f64| b.min(pct)));
+        }
+    }
+    let mut parts = vec![format!("{} sample(s)", samples.len())];
+    if improved > 0 {
+        parts.push(format!("{improved} faster"));
+    }
+    if regressed > 0 {
+        parts.push(format!("{regressed} slower"));
+    }
+    if let Some(b) = best.filter(|b| *b < 0.0) {
+        parts.push(format!("best {:.1}%", b));
+    }
+    Some(parts.join(", "))
+}
+
 /// 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(
@@ -612,6 +743,11 @@ pub async fn world_live(
         // opening a finished mission would replay an hour of tool calls as a
         // burst storm and read as a mission that just did all of it at once.
         let mut event_cursor: i64 = -1;
+        // Findings already announced. A finding is raised once.
+        let mut last_finding: HashSet<String> = HashSet::new();
+        // Last benchmark summary per phase; a looping phase produces a new one.
+        let mut last_bench: 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.
         let mut audit_cursor: i64 = -1;
@@ -792,6 +928,42 @@ pub async fn world_live(
                 );
             }
 
+            // Benchmark results, as an annotation on the station. Re-emitted
+            // only when the summary changes: a phase that loops produces a new
+            // snapshot per iteration, and that IS news.
+            for b in world_benchmarks(&pool, ws, only_mission).await {
+                if last_bench.get(&b.phase_id).map(|n| n == &b.note).unwrap_or(false) {
+                    continue;
+                }
+                last_bench.insert(b.phase_id.clone(), b.note.clone());
+                yield sse(
+                    "mission.benchmark",
+                    json!({
+                        "missionId": b.mission_id,
+                        "phaseId": b.phase_id,
+                        "note": b.note,
+                    }),
+                );
+            }
+
+            // Findings, once each, hanging off the security station. Same rule
+            // as files: a finding is a fact that was raised once, and
+            // re-sending it would pop the orb again on every poll.
+            for f in world_findings(&pool, ws, only_mission).await {
+                if !last_finding.insert(f.task_id.clone()) {
+                    continue;
+                }
+                yield sse(
+                    "mission.finding",
+                    json!({
+                        "missionId": f.mission_id,
+                        "phaseId": f.phase_id,
+                        "findingId": f.task_id,
+                        "title": f.title,
+                    }),
+                );
+            }
+
             // Structured activity — the motion channel. Tool calls and file
             // touches recorded at the source by the container tap and the
             // microVM `PostToolUse` hook. Never parsed from prose: a tool name
@@ -1110,6 +1282,48 @@ mod mission_feed_tests {
         );
     }
 
+    /// A benchmark delta is described only as far as its shape allows.
+    ///
+    /// `compute_delta` emits `{kind:"opaque"}` whenever the before/after
+    /// metrics were not structurally comparable — which is most drivers. The
+    /// temptation is to say something anyway; the result would be a performance
+    /// claim the picture makes and the data does not support.
+    #[test]
+    fn a_benchmark_is_described_only_as_far_as_its_shape_allows() {
+        use serde_json::json;
+        assert_eq!(
+            super::benchmark_note(&json!({
+                "kind": "bencher_diff",
+                "samples": [
+                    {"name": "a", "delta_pct": -12.5, "direction": "improved"},
+                    {"name": "b", "delta_pct": 3.0, "direction": "regressed"},
+                    // No direction and no percentage: counted, claimed for
+                    // neither side.
+                    {"name": "c"},
+                ],
+            }))
+            .as_deref(),
+            Some("3 sample(s), 1 faster, 1 slower, best -12.5%")
+        );
+        // Nothing comparable happened, so nothing is said.
+        assert_eq!(
+            super::benchmark_note(&json!({ "kind": "opaque", "note": "not comparable" })),
+            None
+        );
+        assert_eq!(
+            super::benchmark_note(&json!({ "kind": "bencher_diff", "samples": [] })),
+            None
+        );
+        // Everything got slower: no "best" claim at all.
+        assert_eq!(
+            super::benchmark_note(&json!({
+                "samples": [{"name": "a", "delta_pct": 8.0, "direction": "regressed"}],
+            }))
+            .as_deref(),
+            Some("1 sample(s), 1 slower")
+        );
+    }
+
     /// Agent→phase attribution must come from the runner's own mapping.
     #[test]
     fn phase_attribution_reuses_the_runners_mapping() {