fix(server): purge a retired request's step manifest

A deferred-matrix node is dispatched as a placeholder and gets a manifest
seeded, then retire_node_requests(Purge) removes the request when
expansion replaces it with real legs. The manifest stayed, so every
dynamic expansion accumulated an entry no run projection could reach.
It is now dropped alongside the request's other correlation state.

Durable rows cannot leak the same way: dispatch-time seeding is not
persisted and a placeholder never reports, so it never reaches
store_job_steps; rows for attempts that did report go with their run
through the runs foreign key.

The test asserts every retained manifest belongs to a live request, and
fails with the removal commented out.

Reported on #212 by Codex and Cubic.

Author: Bnjoroge1

crates/preloop-runner-server/src/lib_tests.rs

@@ -2843,6 +2843,84 @@ async fn log_run_logs_unknown_job_is_404_not_whole_run() {
     );
 }
 
+/// Expansion does not leave the placeholder's manifest behind.
+///
+/// A deferred-matrix node is dispatched as a placeholder, gets a manifest
+/// seeded, and is purged when expansion replaces it with real legs. Without
+/// cleanup every dynamic expansion accumulates an entry no run projection can
+/// reach.
+#[tokio::test]
+async fn expansion_purges_the_placeholder_step_manifest() {
+    let temp = tempfile::tempdir().unwrap();
+    let state = AppState::new(temp.path().to_path_buf()).await.unwrap();
+    let app = app(state.clone(), CancellationToken::new());
+
+    let accepted = submit_yaml(
+        &app,
+        r#"
+on: push
+jobs:
+  gen:
+    runs-on: ubuntu-latest
+    outputs:
+      matrix: ${{ steps.build.outputs.matrix }}
+    steps:
+      - id: build
+        run: echo matrix
+  fan:
+    needs: [gen]
+    runs-on: ubuntu-latest
+    strategy:
+      matrix: ${{ fromJson(needs.gen.outputs.matrix) }}
+    steps:
+      - run: echo leg
+"#,
+        "owner/repo",
+    )
+    .await;
+    let run_id: RunId = accepted["run_id"].as_str().unwrap().parse().unwrap();
+
+    let before = {
+        let inner = state.inner.lock().await;
+        inner.job_steps.len()
+    };
+
+    // Finish `gen` with a matrix so the deferred node expands.
+    let _ = crate::distributed_task::complete_job_inner(
+        state.shared(),
+        preloop_gha_protocol::JobCompletion {
+            run_id,
+            job_id: preloop_gha_protocol::JobId("gen".to_owned()),
+            agent_job_id: None,
+            status: ExecutionStatus::Success,
+            outputs: [("matrix".to_owned(), serde_json::json!("{\"leg\":[1,2]}"))]
+                .into_iter()
+                .collect(),
+            annotations: Vec::new(),
+            step_results: Vec::new(),
+        },
+    )
+    .await;
+
+    let inner = state.inner.lock().await;
+    // Every retained manifest must belong to a request that still exists.
+    let live: std::collections::BTreeSet<uuid::Uuid> = inner
+        .job_requests
+        .values()
+        .map(|record| record.agent_job_id)
+        .collect();
+    let orphans: Vec<&uuid::Uuid> = inner
+        .job_steps
+        .keys()
+        .filter(|agent_job_id| !live.contains(agent_job_id))
+        .collect();
+    assert!(
+        orphans.is_empty(),
+        "expansion left {} unreachable manifest(s) (had {before} before)",
+        orphans.len()
+    );
+}
+
 /// The list endpoint projects steps like the single-run endpoint.
 ///
 /// Step records live in the attempt manifest, not in the stored run, so a

crates/preloop-runner-server/src/runtime_scheduling.rs

@@ -2678,6 +2678,17 @@ pub(crate) fn retire_node_requests(
                 inner.live_log_lines.remove(&agent_key);
                 inner.live_log_tx.remove(&agent_key);
                 inner.live_log_closed.remove(&agent_key);
+                // The step manifest is attempt-scoped, so it belongs to the
+                // request being purged. A deferred-matrix placeholder gets one
+                // seeded at dispatch and is then purged when expansion
+                // replaces it, so without this every dynamic expansion leaves
+                // an entry no run projection can reach.
+                //
+                // Durable rows cannot leak the same way: seeding is not
+                // persisted, and a placeholder never reports, so it never
+                // reaches `store_job_steps`. Rows for attempts that did report
+                // are removed with their run through the `runs` foreign key.
+                inner.job_steps.remove(&record.agent_job_id);
             }
         }
     }