From fe6e7aabf0d428975f7ab3635ec82ebe6b165883 Mon Sep 17 00:00:00 2001
From: New1Direction <285551516+New1Direction@users.noreply.github.com>
Date: Sun, 14 Jun 2026 14:41:06 -0700
Subject: [PATCH 1/6] docs(swarm): SP2 collaboration design+plan (role-aware
 provider, data-flow, permissions, self-healing)

---
 ...-14-korg-swarm-sp2-collaboration-design.md | 62 +++++++++++++++++++
 1 file changed, 62 insertions(+)
 create mode 100644 docs/superpowers/specs/2026-06-14-korg-swarm-sp2-collaboration-design.md

diff --git a/docs/superpowers/specs/2026-06-14-korg-swarm-sp2-collaboration-design.md b/docs/superpowers/specs/2026-06-14-korg-swarm-sp2-collaboration-design.md
new file mode 100644
index 0000000..2d260d3
--- /dev/null
+++ b/docs/superpowers/specs/2026-06-14-korg-swarm-sp2-collaboration-design.md
@@ -0,0 +1,62 @@
+# Korg Swarm — SP2: Real collaboration (Track B)
+
+**Status:** Design+plan / approved-by-delegation ("do it all now"), grounded against real code 2026-06-14
+**Branch:** `feat/swarm-collaboration` (stacked on `feat/swarm-honest-demo`)
+**Sub-project:** SP2 of Track B.
+
+## 1. The honest problem (grounded)
+- The DAG (`build_campaign_dag`) encodes the collaboration **graph** (benjamin depends on captain+harper; lucas on benjamin) but never passes **data** — `packages_map` is immutable and each `WorkPackage.description` is a static `"Plan/Research/Implement/Synthesize: {root_task}"` string. **Downstream personas never see upstream output.**
+- Permissions are plumbed (`RouteWork.permissions`) but **ignored** (`_permissions`, hardcoded `fs:write:worktree` for everyone). No read-only/write distinction.
+- Benjamin's worker is **deliberately crash-simulated**: `decompose_into_persona_packages` bakes `"Implement (simulate-crash)"` (leader.rs:2552) and `harness.rs:362` honors it → Benjamin never does real work in a campaign.
+- `run_self_healing_loop` is a **no-op** (targets a worktree the worker child already deleted) and never re-plumbs `files_changed` after a heal.
+- **Honesty constraint:** the default `DeterministicProvider` only emits real applyable content for Benjamin on the *fixture* task; every other (persona, task) returns honest-null. So collaboration plumbing alone yields empty downstream diffs offline. Real per-persona work offline requires a **role-aware** provider keyed on fixture-class tasks; arbitrary tasks need `--provider ollama`.
+
+## 2. Goal
+Make the collaboration **mechanism** genuinely real and verifiable: upstream persona output flows into downstream payloads, personas have real per-role behavior + permissions, the workers actually run (no fake crash), and the ledger count stays truthful through healing. Demonstrated deterministically on the **fixture task** (each persona emits a real role-shaped artifact); honest-null on tasks the stub can't do.
+
+**Acid test:** in a campaign on the fixture task, (a) Benjamin's payload contains Captain's plan text (data-flow is real), (b) a read-only persona that emits mutations does NOT mutate the worktree (permissions enforced), (c) the workers complete without the fake crash, (d) the attested `mutations_this_round` equals the real summed diff including any heal.
+
+## 3. Slices (each independently shippable + testable)
+
+### Slice 1 — Role-aware DeterministicProvider + de-theater the workers
+- Extend `korg-llm/src/deterministic.rs`: `role_marker` recognizes all 5 personas (Captain/Harper/Benjamin/Lucas/Evaluator via their prompt markers). For the **fixture-class task**, each emits a real, deterministic, role-shaped artifact:
+  - Captain → `{work_packages, acceptance_criteria}` (a real plan referencing `src/lib.rs`),
+  - Harper → `{concerns, risk_assessment}`,
+  - Benjamin → the applyable patch (existing),
+  - Lucas → `{resolutions}` (synthesis referencing the implement step),
+  - Evaluator → `{passed_rubrics, recommended_action}`.
+  - Any non-fixture task → honest-null per role (`[]`/empty + low confidence) — never fabricated.
+- Remove the `(simulate-crash)` directive from `decompose_into_persona_packages` (leader.rs:2552) → `"Implement: {root_task}"`. (Leave the `simulate-crash` handler in harness for explicit fault-injection tests, but the default decomposition must not trigger it.)
+- **Test:** each persona's provider output for the fixture task parses to a non-empty role-shaped artifact; non-fixture → honest-null. A campaign-level smoke test confirms Benjamin's worker no longer crash-simulates by default.
+
+### Slice 2 — Real upstream→downstream data-flow
+- In `dispatch_concurrent` (leader.rs): make `packages_map` mutable; after each level completes, rewrite each downstream node's `description` to append serialized upstream `PersonaResult.output` from its DAG dependencies (Captain's plan + Harper's concerns → Benjamin; Benjamin's output → Lucas). Carry it in the existing `RouteWork.payload` String (no ACP schema change), **size-capped** (reuse the 8000-char Heavy-Consciousness ceiling).
+- **Test:** after L1, the benjamin node's payload contains Captain's plan marker; after Benjamin, lucas's payload contains Benjamin's output marker. (Mirror `test_build_campaign_dag_produces_four_levels`.)
+
+### Slice 3 — Per-persona permissions + apply/analyze policy
+- New `fn permissions_for(persona) -> Vec<String>` resolved at spawn (session.rs, replacing the hardcoded vec): Benjamin/Lucas → `fs:write:worktree`; Harper/Captain/Evaluator → `fs:read`.
+- Stop ignoring `_permissions` in `handle_route_work`; in `run_task_in_worktree` gate the `apply_mutations` call on a write capability. A read-only persona that emits mutations → recorded as applied=0 (analyze-only), not mutating.
+- **Test:** Harper (read-only) emitting a mutation yields `files_changed==0` and does not write the file; Benjamin (write) applies normally.
+
+### Slice 4 — Self-healing re-plumb (follow-up #1)
+- Fix `run_self_healing_loop`: heal in the worker child before worktree cleanup (or defer cleanup for heal-eligible nodes), then re-run `numstat` and update `files_changed` so the leader's `real_files_changed` sum (leader.rs:1484) reflects heals. Plumb the post-heal count via the existing `SubmitTransaction.files_changed`.
+- **Test:** the existing `test_self_healing_loop_success` exercises a non-no-op heal and asserts the re-measured count flows through.
+
+## 4. Autonomous decisions (per "do it all")
+- **Carrier = the existing `payload` String** (thin, no ACP schema change), size-capped at 8000 chars.
+- **Permission model = flat `Vec<String>` capability list**, derived from persona (matches the existing `RouteWork.permissions` shape).
+- **Read-only violation response = analyze-only** (don't apply; record `files_changed=0`) + a logged note — NOT a hard task failure (keeps the campaign progressing honestly).
+- **Lucas applies** the synthesized patch (gets `fs:write`); **Harper/Captain/Evaluator do not**.
+- **Honesty boundary stated explicitly:** offline real work is demonstrated on fixture-class tasks via the role-aware stub; arbitrary tasks honestly honest-null offline and need `--provider ollama`. Do NOT claim "all personas do real work on any task."
+- Keep upstream-context serialization deterministic/ordered (so the campaign Merkle root stays reproducible).
+
+## 5. Verification
+- `cargo test -p korg-llm` (role-aware provider tests) + `cargo test -p korg-runtime` (data-flow, permissions, self-healing) green.
+- A campaign smoke test on the fixture task: workers complete (no fake crash), Benjamin's payload carries Captain's plan, the attested count reflects real diffs, permissions enforced.
+- Full `cargo test --workspace` green; fmt + clippy clean.
+- Honesty: a non-fixture campaign task still completes honestly with honest-null personas (attests what it really did, no fabrication).
+
+## 6. Out of scope (later)
+- Making arbitrary-task campaigns produce real work (needs a real model; `--provider ollama` is the path).
+- SP3 (warm boot).
+- The deeper worker-subprocess robustness (timeouts/idle) beyond removing the deliberate crash.

From 52fbddd4d77e5aa3ff4ab9ae091a20cd3c16ca9a Mon Sep 17 00:00:00 2001
From: New1Direction <285551516+New1Direction@users.noreply.github.com>
Date: Sun, 14 Jun 2026 14:46:59 -0700
Subject: [PATCH 2/6] feat(swarm): role-aware DeterministicProvider (5
 personas) + remove fake-crash from decomposition

role_marker now recognizes all 5 personas from their Prompts/*.md role titles
(Captain=Swarm Orchestrator & Planner, Harper=Adversarial Researcher & Reviewer,
Benjamin=Builder & Implementer, Lucas=Synthesizer & Reconciler,
Evaluator=Guardrail Evaluator & Critic), with persona names as fallback.

For the fixture-class task (add bug on src/lib.rs) each persona emits a real,
deterministic, role-shaped JSON artifact matching its documented schema:
captain->work_packages/acceptance_criteria, harper->concerns/risk_assessment,
benjamin->applyable mutations (unchanged), lucas->resolutions,
evaluator->passed_rubrics/recommended_action. Any non-fixture task ->
honest-null (empty role-shaped artifact + 0.2 confidence), never fabricated.

complete() stays pure/deterministic. Adds role-aware fixture + honest-null tests.

De-theater: decompose_into_persona_packages no longer bakes 'simulate-crash'
into Benjamin's package (now 'Implement: {root_task}'); the harness handler is
retained for explicit fault-injection. Adds a unit test asserting the default
benjamin package description does not contain 'simulate-crash'.
---
 crates/korg-llm/src/deterministic.rs | 354 +++++++++++++++++++++++++--
 crates/korg-runtime/src/leader.rs    |  26 +-
 2 files changed, 362 insertions(+), 18 deletions(-)

diff --git a/crates/korg-llm/src/deterministic.rs b/crates/korg-llm/src/deterministic.rs
index 6788a89..9c13e96 100644
--- a/crates/korg-llm/src/deterministic.rs
+++ b/crates/korg-llm/src/deterministic.rs
@@ -19,8 +19,10 @@ fn estimate_tokens(s: &str) -> u32 {
 }
 
 /// Recover the persona role from the System message text. `LlmRequest` carries
-/// no structured role field, so we match a stable marker the persona prompt
-/// contains (Benjamin's prompt says "Builder & Implementer").
+/// no structured role field, so we match a stable marker each persona prompt
+/// contains. The markers below are the `role:` lines from `Prompts/*.md`
+/// (e.g. Benjamin = "Builder & Implementer", Captain = "Swarm Orchestrator &
+/// Planner"); the persona name is a secondary fallback.
 fn role_marker(req: &LlmRequest) -> &'static str {
     let system = req
         .messages
@@ -28,8 +30,18 @@ fn role_marker(req: &LlmRequest) -> &'static str {
         .find(|m| matches!(m.role, Role::System))
         .map(|m| m.content.as_str())
         .unwrap_or("");
+    // Check the most-specific role titles first; persona names are a fallback for
+    // truncated/overridden prompts that drop the `role:` frontmatter line.
     if system.contains("Builder & Implementer") || system.contains("Benjamin") {
         "benjamin"
+    } else if system.contains("Swarm Orchestrator & Planner") || system.contains("Captain") {
+        "captain"
+    } else if system.contains("Adversarial Researcher & Reviewer") || system.contains("Harper") {
+        "harper"
+    } else if system.contains("Synthesizer & Reconciler") || system.contains("Lucas") {
+        "lucas"
+    } else if system.contains("Guardrail Evaluator & Critic") || system.contains("Evaluator") {
+        "evaluator"
     } else {
         "unknown"
     }
@@ -49,14 +61,31 @@ fn task_text(req: &LlmRequest) -> &str {
 /// rewrite src/lib.rs so `add` actually adds. Full file body (applyable as-is).
 const FIXTURE_LIB_RS: &str = "/// Adds two numbers.\npub fn add(a: i64, b: i64) -> i64 {\n    a + b\n}\n\n#[cfg(test)]\nmod tests {\n    use super::*;\n\n    #[test]\n    fn adds() {\n        assert_eq!(add(2, 3), 5);\n    }\n}\n";
 
-/// Build the structured response text (markdown frontmatter + ```json block),
-/// matching what `parse_structured_response` in korg-runtime expects.
-fn render(confidence: f32, mutations_json: &str) -> String {
+/// Build the structured response text (markdown frontmatter + a JSON fence),
+/// matching what `parse_structured_response` in korg-runtime expects. `body`
+/// is the *complete* JSON object body (already serialized) to place inside the
+/// `json` fence, letting each persona emit its own role-shaped schema.
+fn render_artifact(confidence: f32, body: &str) -> String {
     format!(
-        "---\nconfidence: {confidence}\nself_score: {confidence}\n---\n\n```json\n{{\n  \"mutations\": {mutations_json}\n}}\n```\n\nDeterministic honest provider output.\n"
+        "---\nconfidence: {confidence}\nself_score: {confidence}\n---\n\n```json\n{body}\n```\n\nDeterministic honest provider output.\n"
+    )
+}
+
+/// Convenience for Benjamin's mutations-only artifact (kept byte-for-byte
+/// compatible with the SP1 output the apply path already understands).
+fn render(confidence: f32, mutations_json: &str) -> String {
+    render_artifact(
+        confidence,
+        &format!("{{\n  \"mutations\": {mutations_json}\n}}"),
     )
 }
 
+/// Serialize a JSON value into the body string for `render_artifact`,
+/// pretty-printed for stability and human-auditability.
+fn artifact_body(value: &serde_json::Value) -> String {
+    serde_json::to_string_pretty(value).unwrap_or_else(|_| "{}".to_string())
+}
+
 pub struct DeterministicProvider {
     name: &'static str,
 }
@@ -69,24 +98,137 @@ impl DeterministicProvider {
     }
 
     /// Pure rendering core (no async, no I/O) so it is trivially testable.
+    ///
+    /// For the *fixture-class* task (the `add`-bug fix on `src/lib.rs`) every
+    /// persona emits a real, deterministic, role-shaped artifact matching its
+    /// documented `Prompts/*.md` output schema. For any other task each persona
+    /// returns an honest null (empty role-shaped artifact + low confidence) —
+    /// it never fabricates success for a task this stub cannot actually do.
     fn render_for(&self, req: &LlmRequest) -> String {
         let role = role_marker(req);
         let task = task_text(req).to_ascii_lowercase();
-        // Recognize the fixture task by a stable signature.
-        let is_fixture =
-            role == "benjamin" && task.contains("add function") && task.contains("src/lib.rs");
+        // Recognize the fixture task by a stable signature, independent of role,
+        // so the whole swarm collaborates on it (not just Benjamin).
+        let is_fixture = task.contains("add function") && task.contains("src/lib.rs");
         if is_fixture {
-            let content = serde_json::Value::String(FIXTURE_LIB_RS.to_string());
-            let mutations = format!(
-                "[{{\"target\":\"src/lib.rs\",\"action\":\"update\",\"content\":{},\"description\":\"Fix add to use addition\"}}]",
-                content
-            );
-            render(0.95, &mutations)
+            self.fixture_artifact(role)
         } else {
-            // Honest null: no fabricated mutations, low confidence.
-            render(0.20, "[]")
+            self.honest_null(role)
+        }
+    }
+
+    /// Real role-shaped artifact for the fixture task. Each branch mirrors the
+    /// JSON schema documented in `Prompts/{persona}.md`.
+    fn fixture_artifact(&self, role: &str) -> String {
+        match role {
+            "captain" => {
+                let body = serde_json::json!({
+                    "work_packages": [
+                        {
+                            "id": 1,
+                            "title": "Fix the broken `add` in src/lib.rs",
+                            "assigned_to": "Benjamin",
+                            "description": "Rewrite `add` in src/lib.rs so it returns a + b, and keep a unit test asserting add(2,3)==5.",
+                            "dependencies": []
+                        }
+                    ],
+                    "acceptance_criteria": [
+                        "src/lib.rs compiles cleanly",
+                        "add(2, 3) returns 5",
+                        "the adds() unit test passes"
+                    ]
+                });
+                render_artifact(0.95, &artifact_body(&body))
+            }
+            "harper" => {
+                let body = serde_json::json!({
+                    "concerns": [
+                        {
+                            "severity": "high",
+                            "description": "The current `add` in src/lib.rs does not perform addition; the fix must use a + b, not a - b or a hardcoded constant.",
+                            "file_path": "src/lib.rs"
+                        }
+                    ],
+                    "risk_assessment": "low",
+                    "prior_art_checked": [
+                        "fixtures/honest-demo-repo src/lib.rs add bug"
+                    ],
+                    "recommendations": [
+                        "Keep the change to src/lib.rs minimal — only the add body and its test"
+                    ]
+                });
+                render_artifact(0.9, &artifact_body(&body))
+            }
+            "benjamin" => {
+                let content = serde_json::Value::String(FIXTURE_LIB_RS.to_string());
+                let mutations = format!(
+                    "[{{\"target\":\"src/lib.rs\",\"action\":\"update\",\"content\":{},\"description\":\"Fix add to use addition\"}}]",
+                    content
+                );
+                render(0.95, &mutations)
+            }
+            "lucas" => {
+                let body = serde_json::json!({
+                    "synthesis": "Captain's plan and Harper's concern both point at the same one-file fix; Benjamin's implement step rewrites src/lib.rs add to return a + b. No conflicts to reconcile.",
+                    "hybrid_ready": true,
+                    "resolutions": [
+                        {
+                            "topic": "src/lib.rs add fix",
+                            "decision": "Adopt Benjamin's implement step (add returns a + b) — consistent with Captain's acceptance criteria and Harper's concern."
+                        }
+                    ]
+                });
+                render_artifact(0.9, &artifact_body(&body))
+            }
+            "evaluator" => {
+                let body = serde_json::json!({
+                    "overall": "PASS",
+                    "passed_rubrics": [
+                        "correctness",
+                        "completeness",
+                        "minimal_diff",
+                        "provenance_strength"
+                    ],
+                    "total_rubrics": 5,
+                    "justifications": [
+                        "Correctness: add now returns a + b and the adds() test asserts 5.",
+                        "Minimal diff: only src/lib.rs is touched."
+                    ],
+                    "recommended_action": "hold"
+                });
+                render_artifact(0.9, &artifact_body(&body))
+            }
+            _ => self.honest_null(role),
         }
     }
+
+    /// Honest null for tasks this stub cannot do: an empty role-shaped artifact
+    /// with low confidence. Never fabricated success.
+    fn honest_null(&self, role: &str) -> String {
+        let body = match role {
+            "captain" => serde_json::json!({ "work_packages": [], "acceptance_criteria": [] }),
+            "harper" => serde_json::json!({
+                "concerns": [],
+                "risk_assessment": "unknown",
+                "prior_art_checked": [],
+                "recommendations": []
+            }),
+            "benjamin" => serde_json::json!({ "mutations": [] }),
+            "lucas" => {
+                serde_json::json!({ "synthesis": "", "hybrid_ready": false, "resolutions": [] })
+            }
+            "evaluator" => serde_json::json!({
+                "overall": "NEEDS_REVISION",
+                "passed_rubrics": [],
+                "total_rubrics": 5,
+                "justifications": [],
+                "recommended_action": "hold"
+            }),
+            // Unknown persona: fall back to the historical mutations-only null.
+            _ => serde_json::json!({ "mutations": [] }),
+        };
+        render_artifact(0.20, &artifact_body(&body))
+    }
 }
 
 impl Default for DeterministicProvider {
@@ -253,6 +395,184 @@ mod tests {
         assert!(conf < 0.5, "honest null reports low confidence, got {conf}");
     }
 
+    // --- Slice 1: role-aware fixture artifacts (one per persona) ---
+
+    /// A persona system message + the fixture task. Helper keeps the role
+    /// markers in one place so the assertions below read cleanly.
+    async fn fixture_output_for(system: &str) -> serde_json::Value {
+        let p = DeterministicProvider::new();
+        let r = p
+            .complete(req(
+                system,
+                "Plan the work to fix the add function in src/lib.rs so it adds",
+            ))
+            .await
+            .unwrap();
+        let (output, conf, _fm) = crate::deterministic::parse_for_test(&r.content);
+        assert!(
+            conf > 0.5,
+            "fixture artifact should carry real (high) confidence, got {conf}"
+        );
+        output
+    }
+
+    #[tokio::test]
+    async fn captain_fixture_output_has_nonempty_work_packages() {
+        let out =
+            fixture_output_for("You are the Captain, the Swarm Orchestrator & Planner.").await;
+        let wps = out
+            .get("work_packages")
+            .and_then(|v| v.as_array())
+            .expect("captain artifact has work_packages array");
+        assert!(
+            !wps.is_empty(),
+            "captain fixture work_packages must be non-empty"
+        );
+        assert!(
+            out.get("acceptance_criteria")
+                .and_then(|v| v.as_array())
+                .map(|a| !a.is_empty())
+                .unwrap_or(false),
+            "captain fixture must include acceptance_criteria"
+        );
+    }
+
+    #[tokio::test]
+    async fn harper_fixture_output_has_nonempty_concerns() {
+        let out =
+            fixture_output_for("You are Harper, the Adversarial Researcher & Reviewer.").await;
+        let concerns = out
+            .get("concerns")
+            .and_then(|v| v.as_array())
+            .expect("harper artifact has concerns array");
+        assert!(
+            !concerns.is_empty(),
+            "harper fixture concerns must be non-empty"
+        );
+        assert!(
+            out.get("risk_assessment")
+                .and_then(|v| v.as_str())
+                .is_some(),
+            "harper fixture must include a risk_assessment string"
+        );
+    }
+
+    #[tokio::test]
+    async fn benjamin_fixture_output_has_applyable_mutation() {
+        let out = fixture_output_for("You are Benjamin, the Builder & Implementer.").await;
+        let muts = out
+            .get("mutations")
+            .and_then(|v| v.as_array())
+            .expect("benjamin artifact has mutations array");
+        assert_eq!(
+            muts.len(),
+            1,
+            "benjamin emits exactly one applyable mutation"
+        );
+        let content = muts[0]
+            .get("content")
+            .and_then(|v| v.as_str())
+            .expect("applyable content field");
+        assert!(content.contains("a + b"), "benjamin patch must fix the bug");
+    }
+
+    #[tokio::test]
+    async fn lucas_fixture_output_has_nonempty_resolutions() {
+        let out = fixture_output_for("You are Lucas, the Synthesizer & Reconciler.").await;
+        let resolutions = out
+            .get("resolutions")
+            .and_then(|v| v.as_array())
+            .expect("lucas artifact has resolutions array");
+        assert!(
+            !resolutions.is_empty(),
+            "lucas fixture resolutions must be non-empty"
+        );
+    }
+
+    #[tokio::test]
+    async fn evaluator_fixture_output_has_nonempty_passed_rubrics() {
+        let out =
+            fixture_output_for("You are the Evaluator, the Guardrail Evaluator & Critic.").await;
+        let passed = out
+            .get("passed_rubrics")
+            .and_then(|v| v.as_array())
+            .expect("evaluator artifact has passed_rubrics array");
+        assert!(
+            !passed.is_empty(),
+            "evaluator fixture passed_rubrics must be non-empty"
+        );
+        assert!(
+            out.get("recommended_action")
+                .and_then(|v| v.as_str())
+                .is_some(),
+            "evaluator fixture must include a recommended_action"
+        );
+    }
+
+    // --- Slice 1: honest-null for an unknown task, per persona ---
+
+    async fn honest_null_output_for(system: &str) -> (serde_json::Value, f32) {
+        let p = DeterministicProvider::new();
+        let r = p
+            .complete(req(
+                system,
+                "Build a real-time multiplayer physics engine from scratch",
+            ))
+            .await
+            .unwrap();
+        let (output, conf, _fm) = crate::deterministic::parse_for_test(&r.content);
+        (output, conf)
+    }
+
+    fn array_is_empty(out: &serde_json::Value, key: &str) -> bool {
+        out.get(key)
+            .and_then(|v| v.as_array())
+            .map(|a| a.is_empty())
+            .unwrap_or(false)
+    }
+
+    #[tokio::test]
+    async fn every_persona_returns_honest_null_for_unknown_task() {
+        let (cap, c) =
+            honest_null_output_for("You are the Captain, the Swarm Orchestrator & Planner.").await;
+        assert!(c < 0.5, "captain honest-null low confidence, got {c}");
+        assert!(
+            array_is_empty(&cap, "work_packages"),
+            "captain unknown → no work_packages"
+        );
+
+        let (har, c) =
+            honest_null_output_for("You are Harper, the Adversarial Researcher & Reviewer.").await;
+        assert!(c < 0.5, "harper honest-null low confidence, got {c}");
+        assert!(
+            array_is_empty(&har, "concerns"),
+            "harper unknown → no concerns"
+        );
+
+        let (ben, c) = honest_null_output_for("You are Benjamin, the Builder & Implementer.").await;
+        assert!(c < 0.5, "benjamin honest-null low confidence, got {c}");
+        assert!(
+            array_is_empty(&ben, "mutations"),
+            "benjamin unknown → no mutations"
+        );
+
+        let (luc, c) = honest_null_output_for("You are Lucas, the Synthesizer & Reconciler.").await;
+        assert!(c < 0.5, "lucas honest-null low confidence, got {c}");
+        assert!(
+            array_is_empty(&luc, "resolutions"),
+            "lucas unknown → no resolutions"
+        );
+
+        let (eva, c) =
+            honest_null_output_for("You are the Evaluator, the Guardrail Evaluator & Critic.")
+                .await;
+        assert!(c < 0.5, "evaluator honest-null low confidence, got {c}");
+        assert!(
+            array_is_empty(&eva, "passed_rubrics"),
+            "evaluator unknown → no passed_rubrics"
+        );
+    }
+
     #[tokio::test]
     async fn output_is_byte_identical_for_same_inputs() {
         let p = DeterministicProvider::new();
diff --git a/crates/korg-runtime/src/leader.rs b/crates/korg-runtime/src/leader.rs
index da18b13..a391d2d 100644
--- a/crates/korg-runtime/src/leader.rs
+++ b/crates/korg-runtime/src/leader.rs
@@ -2549,7 +2549,7 @@ impl LeaderOrchestrator {
             "work_packages": [
                 {"id": "pkg-captain", "personas": ["captain"], "description": format!("Plan: {}{}", self.root_task, heavy_ctx.as_ref().map(|c| format!("\n\nContext:\n{}", c)).unwrap_or_default())},
                 {"id": "pkg-harper",  "personas": ["harper"],  "description": format!("Research: {}{}", self.root_task, heavy_ctx.as_ref().map(|c| format!("\n\nContext:\n{}", c)).unwrap_or_default())},
-                {"id": "pkg-benjamin","personas": ["benjamin"],"description": format!("Implement (simulate-crash): {}{}", self.root_task, heavy_ctx.as_ref().map(|c| format!("\n\nContext:\n{}", c)).unwrap_or_default())},
+                {"id": "pkg-benjamin","personas": ["benjamin"],"description": format!("Implement: {}{}", self.root_task, heavy_ctx.as_ref().map(|c| format!("\n\nContext:\n{}", c)).unwrap_or_default())},
                 {"id": "pkg-lucas",   "personas": ["lucas"],   "description": format!("Synthesize: {}{}", self.root_task, heavy_ctx.as_ref().map(|c| format!("\n\nContext:\n{}", c)).unwrap_or_default())}
             ]
         })
@@ -3422,6 +3422,30 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn default_decomposition_does_not_simulate_crash() {
+        // De-theater: the default Benjamin package must do real work, not bake
+        // the fake-crash directive. (The simulate-crash handler stays in the
+        // harness for explicit fault-injection tests.)
+        let leader = LeaderOrchestrator::new("Fix the add bug".to_string(), None);
+        let plan = leader.decompose_into_persona_packages();
+        let benjamin = plan["work_packages"]
+            .as_array()
+            .expect("work_packages array")
+            .iter()
+            .find(|p| p["id"] == "pkg-benjamin")
+            .expect("benjamin package present");
+        let desc = benjamin["description"].as_str().unwrap_or("");
+        assert!(
+            !desc.contains("simulate-crash"),
+            "benjamin's default package must not contain 'simulate-crash', got: {desc}"
+        );
+        assert!(
+            desc.starts_with("Implement:"),
+            "benjamin's default package should be a real implement task, got: {desc}"
+        );
+    }
+
     #[test]
     fn real_mutation_count_replaces_synthetic_formula() {
         // The honest count is the sum of per-worker files_changed, NOT

From d6326576cefe817a4766adc095b61557462cf486 Mon Sep 17 00:00:00 2001
From: New1Direction <285551516+New1Direction@users.noreply.github.com>
Date: Sun, 14 Jun 2026 14:50:57 -0700
Subject: [PATCH 3/6] feat(swarm): thread upstream persona output into
 downstream payloads (real data-flow)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

dispatch_concurrent now makes packages_map mutable and, after each DAG level
completes, rewrites every still-pending downstream node whose dependencies
include a just-completed node so its worker payload carries the real upstream
PersonaResult.output (Captain+Harper -> Benjamin; Benjamin -> Lucas). The reverse
dependency map is read from the DAG node .dependencies; the next level's
dispatch_level reads pkg.description as the worker payload, so the flow is real
(not a no-op).

The payload rewrite lives in a pure helper compose_downstream_payload(base,
upstream) in workers.rs: upstream entries are sorted by (persona, node_id) for
determinism, and the appended upstream context is capped at
UPSTREAM_CONTEXT_BUDGET (8000 chars) with an explicit …[truncated] marker on a
UTF-8 char boundary. Adds unit tests proving the downstream payload carries
upstream content (captain plan -> benjamin, benjamin output -> lucas),
order-independence, the size cap, and the empty-upstream no-op.
---
 crates/korg-runtime/src/leader.rs  |  60 +++++++++++-
 crates/korg-runtime/src/workers.rs | 148 +++++++++++++++++++++++++++++
 2 files changed, 207 insertions(+), 1 deletion(-)

diff --git a/crates/korg-runtime/src/leader.rs b/crates/korg-runtime/src/leader.rs
index a391d2d..b14005e 100644
--- a/crates/korg-runtime/src/leader.rs
+++ b/crates/korg-runtime/src/leader.rs
@@ -2571,11 +2571,25 @@ impl LeaderOrchestrator {
         let start_time = std::time::Instant::now();
 
         // Build the DAG + packages_map via workers module
-        let (mut dag, levels, packages_map) =
+        let (mut dag, levels, mut packages_map) =
             crate::workers::build_campaign_dag(&self.root_task, plan, self.tui_tx.as_ref()).await?;
 
         tracing::info!(levels = levels.len(), "campaign_dag_compiled");
 
+        // Reverse dependency map: node_id -> its direct dependencies. Read from
+        // the DAG nodes so downstream payloads can carry upstream output.
+        let node_dependencies: std::collections::HashMap<String, Vec<String>> = dag
+            .nodes
+            .iter()
+            .map(|(id, node)| (id.clone(), node.dependencies.clone()))
+            .collect();
+
+        // Accumulator of completed upstream outputs, keyed by routing/node id.
+        // Used to rewrite each downstream package's description after its
+        // dependencies finish. (persona_name, output_json) per node.
+        let mut completed_outputs: std::collections::HashMap<String, (String, serde_json::Value)> =
+            std::collections::HashMap::new();
+
         let mut results = vec![];
 
         for (idx, level) in levels.iter().enumerate() {
@@ -2627,6 +2641,50 @@ impl LeaderOrchestrator {
                 }
             }
 
+            // --- Real upstream→downstream data-flow ---
+            // Record this level's outputs, then rewrite every still-pending
+            // downstream node whose dependencies include a just-completed node
+            // so its payload carries the real upstream output (Captain+Harper →
+            // Benjamin; Benjamin → Lucas). Deterministic + size-capped via
+            // compose_downstream_payload.
+            for res in &level_result.completed {
+                completed_outputs.insert(
+                    res.routing_id.clone(),
+                    (res.persona.name().to_string(), res.output.clone()),
+                );
+            }
+            let just_completed: std::collections::HashSet<&String> = level_result
+                .completed
+                .iter()
+                .map(|r| &r.routing_id)
+                .collect();
+            for (node_id, deps) in &node_dependencies {
+                // Only rewrite nodes that depend on something we just finished
+                // (and are not themselves done yet).
+                if completed_outputs.contains_key(node_id) {
+                    continue;
+                }
+                if !deps.iter().any(|d| just_completed.contains(d)) {
+                    continue;
+                }
+                // Gather all completed upstream outputs for this node's deps.
+                let upstream: Vec<(String, String, serde_json::Value)> = deps
+                    .iter()
+                    .filter_map(|dep_id| {
+                        completed_outputs.get(dep_id).map(|(persona, output)| {
+                            (persona.clone(), dep_id.clone(), output.clone())
+                        })
+                    })
+                    .collect();
+                if upstream.is_empty() {
+                    continue;
+                }
+                if let Some(pkg) = packages_map.get_mut(node_id) {
+                    pkg.description =
+                        crate::workers::compose_downstream_payload(&pkg.description, &upstream);
+                }
+            }
+
             tracing::info!(
                 level = idx + 1,
                 completed = level_result.completed.len(),
diff --git a/crates/korg-runtime/src/workers.rs b/crates/korg-runtime/src/workers.rs
index 63e56fa..762b848 100644
--- a/crates/korg-runtime/src/workers.rs
+++ b/crates/korg-runtime/src/workers.rs
@@ -524,6 +524,67 @@ pub struct WorkPackage {
     pub routing_id: String,
 }
 
+/// Total character budget for upstream context appended to a downstream
+/// payload — mirrors the 8000-char Heavy-Consciousness ceiling so payloads
+/// can't grow unbounded across a deep DAG.
+pub const UPSTREAM_CONTEXT_BUDGET: usize = 8000;
+
+/// Compose a downstream persona's payload from its base description plus the
+/// serialized outputs of its just-completed upstream dependencies.
+///
+/// This is the heart of the real data-flow: an upstream `PersonaResult.output`
+/// (e.g. Captain's `work_packages`) is appended to the downstream node's
+/// payload so Benjamin/Lucas actually *see* what their dependencies produced.
+///
+/// Guarantees:
+/// - **Deterministic order:** upstream entries are sorted by `(persona, node_id)`
+///   before appending, so the campaign stays reproducible regardless of the
+///   completion order the JoinSet surfaced.
+/// - **Size-capped:** the *appended* upstream context is capped at
+///   [`UPSTREAM_CONTEXT_BUDGET`] characters; if it would overflow, it is
+///   truncated with an explicit `…[truncated]` marker. The base payload is
+///   never truncated.
+///
+/// `upstream` entries are `(persona_name, node_id, output_json)`.
+pub fn compose_downstream_payload(
+    base: &str,
+    upstream: &[(String, String, serde_json::Value)],
+) -> String {
+    if upstream.is_empty() {
+        return base.to_string();
+    }
+
+    // Stable order: sort by persona name then node id so byte-identical inputs
+    // (in any completion order) yield byte-identical payloads.
+    let mut sorted: Vec<&(String, String, serde_json::Value)> = upstream.iter().collect();
+    sorted.sort_by(|a, b| (a.0.as_str(), a.1.as_str()).cmp(&(b.0.as_str(), b.1.as_str())));
+
+    let mut appended = String::new();
+    for (persona, node_id, output) in sorted {
+        let json = serde_json::to_string_pretty(output).unwrap_or_else(|_| output.to_string());
+        let block = format!("\n\n## Upstream from {persona} ({node_id}):\n{json}");
+        // Enforce the budget on the *appended* context only.
+        if appended.len() + block.len() > UPSTREAM_CONTEXT_BUDGET {
+            let remaining = UPSTREAM_CONTEXT_BUDGET.saturating_sub(appended.len());
+            if remaining > 0 {
+                let marker = "\n…[truncated]";
+                let take = remaining.saturating_sub(marker.len()).min(block.len());
+                // Truncate on a char boundary to avoid splitting a UTF-8 scalar.
+                let mut end = take;
+                while end > 0 && !block.is_char_boundary(end) {
+                    end -= 1;
+                }
+                appended.push_str(&block[..end]);
+                appended.push_str(marker);
+            }
+            break;
+        }
+        appended.push_str(&block);
+    }
+
+    format!("{base}{appended}")
+}
+
 /// Build the canonical 4-persona campaign DAG and return topological levels.
 /// Speculative pre-warm is gated on the `speculative_execution` capability.
 #[tracing::instrument(skip(root_task, tui_tx))]
@@ -907,4 +968,91 @@ mod tests {
         assert!(packages_map.contains_key("pkg-captain"));
         assert_eq!(dag.nodes.len(), 4);
     }
+
+    // --- Slice 2: real upstream→downstream data-flow (pure helper) ---
+
+    #[test]
+    fn compose_downstream_payload_carries_upstream_output() {
+        // Captain's plan output flows into Benjamin's payload.
+        let captain_out = serde_json::json!({
+            "work_packages": [{"id": 1, "title": "Fix add"}],
+            "acceptance_criteria": ["add(2,3)==5"]
+        });
+        let benjamin_payload = compose_downstream_payload(
+            "Implement: fix the add bug",
+            &[(
+                "Captain".to_string(),
+                "pkg-captain".to_string(),
+                captain_out,
+            )],
+        );
+        // Base is preserved.
+        assert!(benjamin_payload.starts_with("Implement: fix the add bug"));
+        // Upstream content is actually present (data-flow is real, not a no-op).
+        assert!(
+            benjamin_payload.contains("work_packages"),
+            "benjamin payload must contain captain's plan marker"
+        );
+        assert!(benjamin_payload.contains("Upstream from Captain (pkg-captain)"));
+    }
+
+    #[test]
+    fn compose_downstream_payload_then_benjamin_into_lucas() {
+        // The two-hop chain: Benjamin's output appears in Lucas's payload.
+        let benjamin_out = serde_json::json!({
+            "mutations": [{"target": "src/lib.rs", "action": "update"}]
+        });
+        let lucas_payload = compose_downstream_payload(
+            "Synthesize: fix the add bug",
+            &[(
+                "Benjamin".to_string(),
+                "pkg-benjamin".to_string(),
+                benjamin_out,
+            )],
+        );
+        assert!(
+            lucas_payload.contains("mutations") && lucas_payload.contains("src/lib.rs"),
+            "lucas payload must contain benjamin's output marker"
+        );
+    }
+
+    #[test]
+    fn compose_downstream_payload_is_order_independent() {
+        // Captain + Harper into Benjamin — completion order must not matter.
+        let cap = (
+            "Captain".to_string(),
+            "pkg-captain".to_string(),
+            serde_json::json!({"work_packages": [1]}),
+        );
+        let har = (
+            "Harper".to_string(),
+            "pkg-harper".to_string(),
+            serde_json::json!({"concerns": [2]}),
+        );
+        let a = compose_downstream_payload("base", &[cap.clone(), har.clone()]);
+        let b = compose_downstream_payload("base", &[har, cap]);
+        assert_eq!(a, b, "payload must be byte-identical regardless of order");
+    }
+
+    #[test]
+    fn compose_downstream_payload_respects_size_cap() {
+        // A huge upstream output must not blow the payload past base + budget.
+        let big = serde_json::json!({ "blob": "x".repeat(50_000) });
+        let out = compose_downstream_payload(
+            "base",
+            &[("Captain".to_string(), "pkg-captain".to_string(), big)],
+        );
+        assert!(
+            out.len() <= "base".len() + UPSTREAM_CONTEXT_BUDGET,
+            "appended upstream context must be capped at the budget, got {} chars",
+            out.len()
+        );
+        assert!(out.contains("…[truncated]"), "truncation must be marked");
+    }
+
+    #[test]
+    fn compose_downstream_payload_empty_upstream_is_noop() {
+        let out = compose_downstream_payload("base only", &[]);
+        assert_eq!(out, "base only");
+    }
 }

From 85b90f6479dadeffb13ec048c8ab20a16c577aba Mon Sep 17 00:00:00 2001
From: New1Direction <285551516+New1Direction@users.noreply.github.com>
Date: Sun, 14 Jun 2026 15:05:50 -0700
Subject: [PATCH 4/6] =?UTF-8?q?feat(swarm):=20per-persona=20permissions=20?=
 =?UTF-8?q?=E2=80=94=20read-only=20personas=20analyze,=20never=20mutate?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Add a permission policy module (permissions_for / may_write): benjamin & lucas
get fs:write:worktree; harper, captain, evaluator (and any unknown persona)
get fs:read and are analyze-only.

- session.rs: SubprocessBackend::spawn now derives RouteWork.permissions from
  permissions_for(&spec.persona) instead of hardcoding fs:write:worktree.
- harness.rs: handle_route_work stops ignoring permissions (_permissions ->
  permissions) and threads them into run_task_in_worktree, which gates
  apply_mutations on may_write(). A read-only persona that emits mutations is
  analyzed (numstat/cargo_check on the existing tree), never applied, and
  records files_changed = 0 honestly, with a one-line log note.

TDD: 8 policy tests + an apply-gate integration test proving benjamin's real
fixture patch is blocked under fs:read (file unchanged, files_changed == 0) but
applied under fs:write:worktree (src/lib.rs rewritten to a + b).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 crates/korg-runtime/src/harness.rs     | 153 +++++++++++++++++++++++--
 crates/korg-runtime/src/lib.rs         |   1 +
 crates/korg-runtime/src/permissions.rs | 114 ++++++++++++++++++
 crates/korg-runtime/src/session.rs     |   5 +-
 4 files changed, 262 insertions(+), 11 deletions(-)
 create mode 100644 crates/korg-runtime/src/permissions.rs

diff --git a/crates/korg-runtime/src/harness.rs b/crates/korg-runtime/src/harness.rs
index ec7f8e5..24e9a59 100644
--- a/crates/korg-runtime/src/harness.rs
+++ b/crates/korg-runtime/src/harness.rs
@@ -191,9 +191,12 @@ impl SingleWorkerHarness {
         payload: String,
         base_snapshot: String,
         codebase_merkle_root: String,
-        _permissions: Vec<String>,
+        permissions: Vec<String>,
     ) -> Result<()> {
-        eprintln!("[Harness] Received RouteWork {}: {}", routing_id, payload);
+        eprintln!(
+            "[Harness] Received RouteWork {}: {} (permissions={:?})",
+            routing_id, payload, permissions
+        );
 
         // Save original working directory to restore it during cleanup
         let original_dir = std::env::current_dir()?;
@@ -353,8 +356,10 @@ impl SingleWorkerHarness {
             }
         });
 
-        // 2. Run the actual persona task (emitter runs in parallel)
-        let result = self.run_task_in_worktree(&payload).await?;
+        // 2. Run the actual persona task (emitter runs in parallel).
+        //    Permissions gate whether emitted mutations are applied: a read-only
+        //    persona (no fs:write:worktree) is analyzed, never mutated.
+        let result = self.run_task_in_worktree(&payload, &permissions).await?;
 
         // Wait for emitter to finish (or abort it)
         let _ = emitter_handle.await;
@@ -518,7 +523,11 @@ impl SingleWorkerHarness {
         Ok(())
     }
 
-    async fn run_task_in_worktree(&mut self, payload: &str) -> Result<TaskResult> {
+    async fn run_task_in_worktree(
+        &mut self,
+        payload: &str,
+        permissions: &[String],
+    ) -> Result<TaskResult> {
         // Route persona from worker_id when possible (real 4-persona topology)
         let persona = self.infer_persona_from_worker_id();
         eprintln!(
@@ -529,11 +538,29 @@ impl SingleWorkerHarness {
 
         let persona_result = run_persona(persona, payload, "worker-task").await;
 
-        // Honest observation: apply the persona's patch to THIS worktree, then
-        // measure reality. The worktree is the process CWD (set at harness.rs:267).
+        // Honest observation. The worktree is the process CWD (set at harness.rs:267).
         let cwd = std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from("."));
         let started = std::time::Instant::now();
-        let apply = crate::observation::apply_mutations(&cwd, &persona_result.mutations).await;
+
+        // --- Apply gate (SP2 Slice 3) ---
+        // Only personas holding `fs:write:worktree` may mutate. A read-only
+        // persona (harper/captain/evaluator) that nonetheless emitted mutations
+        // is analyzed, NEVER applied: we run numstat/cargo_check on the existing
+        // (unmutated) tree and record files_changed = 0 honestly.
+        let may_write = crate::permissions::may_write(permissions);
+        let apply = if may_write {
+            crate::observation::apply_mutations(&cwd, &persona_result.mutations).await
+        } else {
+            if !persona_result.mutations.is_empty() {
+                eprintln!(
+                    "[Harness] {} is read-only (no fs:write:worktree) but emitted {} mutation(s); analyzing only — NOT applying (files_changed=0).",
+                    persona.name(),
+                    persona_result.mutations.len()
+                );
+            }
+            // Analyze-only: nothing applied, nothing rejected.
+            crate::observation::ApplyOutcome::default()
+        };
         let numstat = crate::observation::numstat(&cwd).await;
         let check = crate::observation::cargo_check(&cwd).await;
         let elapsed = started.elapsed().as_secs_f64().max(1e-3);
@@ -554,7 +581,9 @@ impl SingleWorkerHarness {
         self.last_observation = Some(crate::observation::honest_metrics(
             &apply, &check, &numstat, tokens, elapsed, cpu, &surface,
         ));
-        let files_changed = numstat.files;
+        // A read-only persona did not apply anything, so it changed nothing:
+        // record files_changed = 0 honestly (don't count analyze as apply).
+        let files_changed = if may_write { numstat.files } else { 0 };
 
         // Stage all modifications so git write-tree will capture them
         let _ = tokio::process::Command::new("git")
@@ -677,7 +706,7 @@ mod tests {
                 payload,
                 "HEAD".to_string(),
                 "".to_string(),
-                vec![],
+                crate::permissions::permissions_for(&worker_id),
             )
             .await;
 
@@ -687,4 +716,108 @@ mod tests {
         let worktree_path = korg_core::paths::worktree_dir_harness(&worker_id, &routing_id);
         assert!(!worktree_path.exists());
     }
+
+    /// CWD is process-global, so tests that `set_current_dir` into a sandbox
+    /// must not run concurrently with each other. This mutex serializes them.
+    static CWD_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
+
+    /// Build a throwaway git repo containing the fixture `src/lib.rs` (broken
+    /// `add`) and return its path. The caller cd's into it.
+    fn setup_fixture_git_repo(tag: &str) -> std::path::PathBuf {
+        let dir =
+            std::env::temp_dir().join(format!("korg-slice3-{}-{}", tag, uuid::Uuid::now_v7()));
+        std::fs::create_dir_all(dir.join("src")).unwrap();
+        // A compiling crate whose `add` is intentionally wrong (fixture-shaped).
+        std::fs::write(
+            dir.join("Cargo.toml"),
+            "[package]\nname = \"fix\"\nversion = \"0.1.0\"\nedition = \"2021\"\n",
+        )
+        .unwrap();
+        std::fs::write(
+            dir.join("src/lib.rs"),
+            "pub fn add(a: i32, b: i32) -> i32 { a - b }\n",
+        )
+        .unwrap();
+        for args in [vec!["init", "-q"], vec!["add", "-A"]] {
+            let _ = std::process::Command::new("git")
+                .args(&args)
+                .current_dir(&dir)
+                .output();
+        }
+        // commit needs identity; set it locally for the throwaway repo
+        let _ = std::process::Command::new("git")
+            .args([
+                "-c",
+                "user.email=t@t",
+                "-c",
+                "user.name=t",
+                "commit",
+                "-q",
+                "-m",
+                "init",
+            ])
+            .current_dir(&dir)
+            .output();
+        dir
+    }
+
+    /// Slice 3: a read-only persona that emits an applyable mutation must NOT
+    /// mutate the worktree and must record files_changed == 0. The SAME persona
+    /// with write capability DOES apply — proving the gate (not the persona) is
+    /// what blocks the write.
+    #[tokio::test]
+    async fn read_only_persona_does_not_mutate_worktree() {
+        let _guard = CWD_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let original_dir = std::env::current_dir().unwrap();
+
+        // Benjamin's worker emits a real applyable src/lib.rs patch on the
+        // fixture task, so the ONLY thing that can stop the write is the gate.
+        let payload = "Fix the add function in src/lib.rs so it adds";
+
+        // --- Read-only run: pass fs:read perms; expect NO mutation ---
+        let ro_repo = setup_fixture_git_repo("ro");
+        std::env::set_current_dir(&ro_repo).unwrap();
+        let mut ro_harness = SingleWorkerHarness::new("benjamin-ro-test".to_string());
+        let ro_result = ro_harness
+            .run_task_in_worktree(payload, &[crate::permissions::CAP_FS_READ.to_string()])
+            .await
+            .unwrap();
+        let ro_lib = std::fs::read_to_string(ro_repo.join("src/lib.rs")).unwrap();
+        std::env::set_current_dir(&original_dir).unwrap();
+
+        assert_eq!(
+            ro_result.files_changed, 0,
+            "read-only persona must record files_changed == 0 (analyze-only)"
+        );
+        assert!(
+            ro_lib.contains("a - b"),
+            "read-only persona must NOT have rewritten src/lib.rs (file unchanged)"
+        );
+
+        // --- Write run: pass fs:write:worktree; expect the patch to apply ---
+        let rw_repo = setup_fixture_git_repo("rw");
+        std::env::set_current_dir(&rw_repo).unwrap();
+        let mut rw_harness = SingleWorkerHarness::new("benjamin-rw-test".to_string());
+        let rw_result = rw_harness
+            .run_task_in_worktree(
+                payload,
+                &[crate::permissions::CAP_FS_WRITE_WORKTREE.to_string()],
+            )
+            .await
+            .unwrap();
+        let rw_lib = std::fs::read_to_string(rw_repo.join("src/lib.rs")).unwrap();
+        std::env::set_current_dir(&original_dir).unwrap();
+
+        assert!(
+            rw_result.files_changed >= 1,
+            "write persona must record the applied change (files_changed >= 1)"
+        );
+        assert!(
+            rw_lib.contains("a + b"),
+            "write persona must have applied the fixture patch (add now uses a + b)"
+        );
+
+        let _ = std::fs::remove_dir_all(&ro_repo);
+        let _ = std::fs::remove_dir_all(&rw_repo);
+    }
 }
diff --git a/crates/korg-runtime/src/lib.rs b/crates/korg-runtime/src/lib.rs
index 88e7d9e..a97fc73 100644
--- a/crates/korg-runtime/src/lib.rs
+++ b/crates/korg-runtime/src/lib.rs
@@ -25,6 +25,7 @@ pub mod harness;
 pub mod identity;
 pub mod leader;
 pub mod observation;
+pub mod permissions;
 pub mod personas;
 pub mod provenance;
 pub mod recovery;
diff --git a/crates/korg-runtime/src/permissions.rs b/crates/korg-runtime/src/permissions.rs
new file mode 100644
index 0000000..516e12d
--- /dev/null
+++ b/crates/korg-runtime/src/permissions.rs
@@ -0,0 +1,114 @@
+//! Per-persona permission policy (SP2 Slice 3).
+//!
+//! Personas have real, role-shaped capabilities instead of a single hardcoded
+//! `fs:write:worktree` grant for everyone:
+//!
+//! - **Benjamin / Lucas** implement and synthesize patches → they may write the
+//!   worktree (`fs:write:worktree`).
+//! - **Harper / Captain / Evaluator** plan, research, and evaluate → they are
+//!   read-only (`fs:read`). A read-only persona that nonetheless *emits*
+//!   mutations must NOT mutate the worktree; it is analyzed (numstat / cargo
+//!   check on the existing tree) and recorded as `files_changed = 0` honestly.
+//!
+//! The capability list is the flat `Vec<String>` shape already carried by
+//! `RouteWork.permissions`, so this drops in without an ACP schema change.
+
+/// Capability granting write access to the worker's isolated worktree.
+pub const CAP_FS_WRITE_WORKTREE: &str = "fs:write:worktree";
+
+/// Capability granting read-only access (analyze, never mutate).
+pub const CAP_FS_READ: &str = "fs:read";
+
+/// Resolve the capability list for a persona by name.
+///
+/// Matching is case-insensitive and tolerant of the decorated worker id form
+/// (e.g. `"benjamin-019ec8…"`), so callers can pass either the bare persona
+/// name (`spec.persona`) or a worker id without surprises.
+///
+/// Implementers (write the worktree): benjamin, lucas.
+/// Read-only (analyze only): harper, captain, evaluator.
+/// Unknown personas default to read-only — the safe, least-privilege choice.
+pub fn permissions_for(persona: &str) -> Vec<String> {
+    let p = persona.to_lowercase();
+    if p.contains("benjamin") || p.contains("lucas") {
+        vec![CAP_FS_WRITE_WORKTREE.to_string()]
+    } else {
+        // harper / captain / evaluator — and any unrecognized persona — are
+        // read-only by default (least privilege).
+        vec![CAP_FS_READ.to_string()]
+    }
+}
+
+/// The apply gate: may this permission set mutate the worktree?
+///
+/// `true` iff the capability list contains `fs:write:worktree`. A persona
+/// without that capability is analyze-only — its emitted mutations are observed
+/// but never written.
+pub fn may_write(permissions: &[String]) -> bool {
+    permissions.iter().any(|c| c == CAP_FS_WRITE_WORKTREE)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn benjamin_has_write_capability() {
+        let perms = permissions_for("benjamin");
+        assert!(
+            may_write(&perms),
+            "benjamin (implementer) must be able to write the worktree"
+        );
+        assert_eq!(perms, vec![CAP_FS_WRITE_WORKTREE.to_string()]);
+    }
+
+    #[test]
+    fn lucas_has_write_capability() {
+        // Lucas synthesizes and applies the synthesized patch → write.
+        assert!(may_write(&permissions_for("lucas")));
+    }
+
+    #[test]
+    fn harper_is_read_only() {
+        let perms = permissions_for("harper");
+        assert!(
+            !may_write(&perms),
+            "harper (researcher) must be read-only — analyze, never mutate"
+        );
+        assert_eq!(perms, vec![CAP_FS_READ.to_string()]);
+    }
+
+    #[test]
+    fn captain_is_read_only() {
+        assert!(!may_write(&permissions_for("captain")));
+    }
+
+    #[test]
+    fn evaluator_is_read_only() {
+        assert!(!may_write(&permissions_for("evaluator")));
+    }
+
+    #[test]
+    fn matching_is_case_insensitive_and_tolerates_worker_id_form() {
+        // Bare name, capitalized name, and decorated worker-id form all resolve.
+        assert!(may_write(&permissions_for("Benjamin")));
+        assert!(may_write(&permissions_for("benjamin-019ec826-9422-7cc2")));
+        assert!(!may_write(&permissions_for("Harper")));
+        assert!(!may_write(&permissions_for("harper-019ec826-9422-7cc2")));
+    }
+
+    #[test]
+    fn unknown_persona_defaults_to_read_only() {
+        // Least privilege: an unrecognized persona must not get write.
+        assert!(!may_write(&permissions_for("some-unknown-persona")));
+    }
+
+    #[test]
+    fn may_write_is_capability_driven_not_persona_driven() {
+        // The gate keys on the capability, so an explicit write grant works
+        // regardless of how it was derived.
+        assert!(may_write(&[CAP_FS_WRITE_WORKTREE.to_string()]));
+        assert!(!may_write(&[CAP_FS_READ.to_string()]));
+        assert!(!may_write(&[]));
+    }
+}
diff --git a/crates/korg-runtime/src/session.rs b/crates/korg-runtime/src/session.rs
index 5c918c6..b61cb28 100644
--- a/crates/korg-runtime/src/session.rs
+++ b/crates/korg-runtime/src/session.rs
@@ -347,7 +347,10 @@ impl SessionBackend for SubprocessBackend {
             payload: spec.payload.clone(),
             base_snapshot: "latest-from-blackboard".into(),
             codebase_merkle_root: codebase_root,
-            permissions: vec!["fs:write:worktree".into()],
+            // Per-persona capability list (SP2 Slice 3): implementers (benjamin/
+            // lucas) get fs:write:worktree; read-only personas (harper/captain/
+            // evaluator) get fs:read and analyze-only — they never mutate.
+            permissions: crate::permissions::permissions_for(&spec.persona),
         };
         crate::acp::write_signed_acp_envelope(&mut stdin, signing_key, route_work).await?;
 

From 1d9a4fa13b67aa8855cbe4a8e41856fdcc5197dc Mon Sep 17 00:00:00 2001
From: New1Direction <285551516+New1Direction@users.noreply.github.com>
Date: Sun, 14 Jun 2026 15:09:35 -0700
Subject: [PATCH 5/6] fix(swarm): self-healing re-measures files_changed
 (truthful ledger count after heal)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

run_self_healing_loop previously never re-plumbed files_changed after a heal,
so the leader's real_files_changed sum (leader.rs ~1484) missed healed changes.

- After a real heal (direct heal-success or a speculative repair that mutated
  the worktree), re-run observation::numstat on the heal target and set
  current_result.files_changed to the re-measured count. Gated on a
  healed_this_loop flag so a no-op pass never re-measures or clobbers the
  worker's reported count — no fabrication.
- Made the honest no-op explicit: when no worktree exists (clean run — the
  worker already cleaned up), return the worker's result unchanged.

TDD:
- test_self_healing_loop_success now sets up a git worktree with a committed
  missing-semicolon error so the heal path actually runs (non-vacuous), and
  asserts the re-measured files_changed == 1 (the healed src/main.rs;
  build artifacts gitignored, not counted).
- Added test_self_healing_loop_no_op_preserves_count_when_no_worktree: with no
  worktree, the worker's reported count (5) is preserved, not re-measured.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 crates/korg-runtime/src/leader.rs | 152 +++++++++++++++++++++++++-----
 1 file changed, 131 insertions(+), 21 deletions(-)

diff --git a/crates/korg-runtime/src/leader.rs b/crates/korg-runtime/src/leader.rs
index b14005e..a63f0d7 100644
--- a/crates/korg-runtime/src/leader.rs
+++ b/crates/korg-runtime/src/leader.rs
@@ -3249,9 +3249,13 @@ impl LeaderOrchestrator {
         .to_string();
         let worktree_path = std::path::Path::new(&worktree_dir);
 
+        // Honest no-op: if no worktree exists there is nothing to heal. The
+        // worker child deletes its worktree on success (harness cleanup), so a
+        // clean run legitimately leaves nothing here — return the worker's
+        // result unchanged. We do NOT fabricate a heal or a count.
         if !worktree_path.exists() {
             println!(
-                "[Self-Healing] Worktree path {} does not exist. Skipping compilation check.",
+                "[Self-Healing] Worktree path {} does not exist — nothing to heal (clean run). Returning the worker's result unchanged.",
                 worktree_dir
             );
             return Ok(current_result);
@@ -3263,6 +3267,11 @@ impl LeaderOrchestrator {
         );
         println!("Checking compilation in worktree: {}", worktree_dir);
 
+        // Tracks whether a real heal mutated the worktree this loop. If so, we
+        // re-measure files_changed before returning so the leader's
+        // real_files_changed sum (leader.rs ~1484) reflects the heal honestly.
+        let mut healed_this_loop = false;
+
         for iteration in 1..=3 {
             println!(
                 "[Self-Healing] Iteration {}/3: Running compiler check (cargo check)...",
@@ -3314,6 +3323,7 @@ impl LeaderOrchestrator {
 
                         if let Ok(true) = heal_res {
                             self.last_round_healed = true;
+                            healed_this_loop = true;
                             korg_llm::HEALS_RESOLVED
                                 .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                             println!("\x1b[38;2;0;255;128m🔧 [Self-Healing] Thumper auto-healed the compilation failure in sub-seconds!\x1b[0m");
@@ -3380,6 +3390,9 @@ impl LeaderOrchestrator {
                                     )
                                     .await;
                                 }
+                                // A speculative repair really mutated the worktree;
+                                // the post-loop re-measure will count it truthfully.
+                                healed_this_loop = true;
                             }
                         }
 
@@ -3396,6 +3409,21 @@ impl LeaderOrchestrator {
             }
         }
 
+        // --- Re-plumb files_changed after a real heal (SP2 Slice 4) ---
+        // If a heal (or speculative repair) actually mutated the worktree, the
+        // worker's reported count no longer reflects reality. Re-run numstat on
+        // the heal target so the leader's real_files_changed sum is truthful.
+        // We only re-measure when a heal happened — a no-op pass leaves the
+        // worker's count untouched (no fabrication).
+        if healed_this_loop {
+            let n = crate::observation::numstat(worktree_path).await;
+            println!(
+                "[Self-Healing] Re-measured worktree after heal: {} file(s) changed (was {}). Plumbing truthful count into the ledger.",
+                n.files, current_result.files_changed
+            );
+            current_result.files_changed = n.files;
+        }
+
         Ok(current_result)
     }
 
@@ -3868,39 +3896,121 @@ mod tests {
         assert_eq!(arena_outcome["confidence"].as_f64().unwrap(), 0.88);
     }
 
+    /// Slice 4: the self-healing loop must exercise a REAL (non-no-op) heal and
+    /// re-measure files_changed so the leader's ledger sum is truthful after a
+    /// heal. The worktree is a git repo whose committed `src/main.rs` has a
+    /// missing-semicolon error that fails `cargo check`; the loop heals it
+    /// (inserts `;`), and the re-measured numstat count must flow into the
+    /// returned PersonaResult.
     #[tokio::test]
     async fn test_self_healing_loop_success() {
+        // Unique routing id so this test's worktree path can't collide with
+        // other runs/tests sharing the cache dir.
+        let routing_id = format!("pkg-benjamin-heal-{}", uuid::Uuid::now_v7());
         let mut leader = LeaderOrchestrator::new("Test self-healing success".to_string(), None);
         let mut benjamin_res =
-            crate::personas::PersonaResult::new(Persona::Benjamin, "pkg-benjamin".to_string());
-        benjamin_res.mutations = vec![json!({
-            "target": "src/main.rs",
-            "action": "modify",
-            "content": "// speculative self-healing repair dummy mutation"
-        })];
+            crate::personas::PersonaResult::new(Persona::Benjamin, routing_id.clone());
+        // The worker reported 0 files (or the count was lost); after a real heal
+        // the loop must re-measure a truthful, non-zero count.
+        benjamin_res.files_changed = 0;
 
-        // Create the worktree path manually so the test passes
-        let worktree_dir =
-            korg_core::paths::worktree_dir("benjamin", "pkg-benjamin", "pkg-benjamin")
-                .display()
-                .to_string();
-        let _ = std::fs::create_dir_all(&worktree_dir);
-
-        // Run cargo init to make it a valid compiling crate
-        let _ = std::process::Command::new("cargo")
-            .arg("init")
-            .arg("--bin")
-            .arg("--name")
-            .arg("dummy")
+        // The self-healing loop targets worktree_dir(persona, routing, routing).
+        let worktree_dir = korg_core::paths::worktree_dir("benjamin", &routing_id, &routing_id)
+            .display()
+            .to_string();
+        let _ = std::fs::remove_dir_all(&worktree_dir);
+        std::fs::create_dir_all(format!("{}/src", worktree_dir)).unwrap();
+
+        // A compiling-shaped crate, but with a deliberate missing semicolon so
+        // `cargo check` fails with an actionable error the healer can fix.
+        std::fs::write(
+            format!("{}/Cargo.toml", worktree_dir),
+            "[package]\nname = \"dummy\"\nversion = \"0.1.0\"\nedition = \"2021\"\n",
+        )
+        .unwrap();
+        // Mirror a real repo: build artifacts are gitignored, so the re-measured
+        // count reflects only source changes (not target/ from cargo check).
+        std::fs::write(
+            format!("{}/.gitignore", worktree_dir),
+            "/target\nCargo.lock\n",
+        )
+        .unwrap();
+        std::fs::write(
+            format!("{}/src/main.rs", worktree_dir),
+            "fn main() {\n    let x = 42\n    println!(\"{}\", x);\n}\n",
+        )
+        .unwrap();
+
+        // Commit the broken baseline so numstat measures the heal's real diff.
+        for args in [vec!["init", "-q"], vec!["add", "-A"]] {
+            let _ = std::process::Command::new("git")
+                .args(&args)
+                .current_dir(&worktree_dir)
+                .output();
+        }
+        let _ = std::process::Command::new("git")
+            .args([
+                "-c",
+                "user.email=t@t",
+                "-c",
+                "user.name=t",
+                "commit",
+                "-q",
+                "-m",
+                "broken baseline",
+            ])
             .current_dir(&worktree_dir)
-            .status();
+            .output();
 
         let healed = leader.run_self_healing_loop(&benjamin_res).await.unwrap();
         assert_eq!(healed.persona, Persona::Benjamin);
 
+        // The heal actually fixed the file (inserted the missing semicolon).
+        let repaired = std::fs::read_to_string(format!("{}/src/main.rs", worktree_dir)).unwrap();
+        assert!(
+            repaired.contains("let x = 42;"),
+            "self-healing should have inserted the missing semicolon, got:\n{repaired}"
+        );
+
+        // The re-measured count flows into the result so the leader's
+        // real_files_changed sum is truthful: exactly one source file
+        // (src/main.rs) was healed — build artifacts are gitignored, not counted.
+        assert_eq!(
+            healed.files_changed, 1,
+            "after a real heal, files_changed must be re-measured to the heal's real source diff (1), got {}",
+            healed.files_changed
+        );
+
         let _ = std::fs::remove_dir_all(&worktree_dir);
     }
 
+    /// Slice 4 honesty guard: when no worktree exists (the clean-run case — the
+    /// worker already cleaned up its worktree), the loop is a clean no-op and
+    /// must return the worker's result UNCHANGED. It must never fabricate a heal
+    /// or a re-measured count.
+    #[tokio::test]
+    async fn test_self_healing_loop_no_op_preserves_count_when_no_worktree() {
+        let routing_id = format!("pkg-benjamin-noop-{}", uuid::Uuid::now_v7());
+        let mut leader = LeaderOrchestrator::new("Test self-healing no-op".to_string(), None);
+        let mut benjamin_res =
+            crate::personas::PersonaResult::new(Persona::Benjamin, routing_id.clone());
+        // The worker honestly reported 5 changed files; with nothing to heal,
+        // the loop must leave that count intact (no fabrication, no clobber).
+        benjamin_res.files_changed = 5;
+
+        // Ensure the target worktree path does not exist.
+        let worktree_dir = korg_core::paths::worktree_dir("benjamin", &routing_id, &routing_id)
+            .display()
+            .to_string();
+        let _ = std::fs::remove_dir_all(&worktree_dir);
+
+        let result = leader.run_self_healing_loop(&benjamin_res).await.unwrap();
+        assert_eq!(
+            result.files_changed, 5,
+            "no-op heal must preserve the worker's reported count, not re-measure or zero it"
+        );
+    }
+
     #[tokio::test]
     async fn test_goal_mode_auto_approval() {
         let mut leader = LeaderOrchestrator::new("Test goal mode".to_string(), None);

From 27f66b78b469af76872ee226a0357672425247f0 Mon Sep 17 00:00:00 2001
From: New1Direction <285551516+New1Direction@users.noreply.github.com>
Date: Sun, 14 Jun 2026 15:11:49 -0700
Subject: [PATCH 6/6] test(swarm): integration test for the dispatch_concurrent
 data-flow rewrite
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The Slice-2 data-flow wiring previously had only the pure compose_downstream_payload
helper under test — the per-level rewrite loop in dispatch_concurrent had no
regression test. Extract that loop into a pure, testable
workers::apply_upstream_to_pending(packages_map, node_dependencies,
completed_outputs, just_completed_ids) and call it from dispatch_concurrent
(behavior unchanged — the inline loop is now a single call).

Add an integration test on the canonical 4-node DAG that asserts the REAL
rewrite (not a tautology):
- After captain+harper complete, benjamin's payload contains BOTH 'work_packages'
  / 'Upstream from Captain (pkg-captain)' AND 'concerns' / 'Upstream from Harper',
  while lucas (dep not yet done) is untouched and L1 peers are not rewritten.
- After benjamin completes, lucas's payload contains 'mutations' / 'src/lib.rs' /
  'Upstream from Benjamin (pkg-benjamin)', and benjamin is not re-rewritten.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 crates/korg-runtime/src/leader.rs  |  36 ++----
 crates/korg-runtime/src/workers.rs | 194 +++++++++++++++++++++++++++++
 2 files changed, 202 insertions(+), 28 deletions(-)

diff --git a/crates/korg-runtime/src/leader.rs b/crates/korg-runtime/src/leader.rs
index a63f0d7..167b533 100644
--- a/crates/korg-runtime/src/leader.rs
+++ b/crates/korg-runtime/src/leader.rs
@@ -2653,37 +2653,17 @@ impl LeaderOrchestrator {
                     (res.persona.name().to_string(), res.output.clone()),
                 );
             }
-            let just_completed: std::collections::HashSet<&String> = level_result
+            let just_completed: std::collections::HashSet<String> = level_result
                 .completed
                 .iter()
-                .map(|r| &r.routing_id)
+                .map(|r| r.routing_id.clone())
                 .collect();
-            for (node_id, deps) in &node_dependencies {
-                // Only rewrite nodes that depend on something we just finished
-                // (and are not themselves done yet).
-                if completed_outputs.contains_key(node_id) {
-                    continue;
-                }
-                if !deps.iter().any(|d| just_completed.contains(d)) {
-                    continue;
-                }
-                // Gather all completed upstream outputs for this node's deps.
-                let upstream: Vec<(String, String, serde_json::Value)> = deps
-                    .iter()
-                    .filter_map(|dep_id| {
-                        completed_outputs.get(dep_id).map(|(persona, output)| {
-                            (persona.clone(), dep_id.clone(), output.clone())
-                        })
-                    })
-                    .collect();
-                if upstream.is_empty() {
-                    continue;
-                }
-                if let Some(pkg) = packages_map.get_mut(node_id) {
-                    pkg.description =
-                        crate::workers::compose_downstream_payload(&pkg.description, &upstream);
-                }
-            }
+            crate::workers::apply_upstream_to_pending(
+                &mut packages_map,
+                &node_dependencies,
+                &completed_outputs,
+                &just_completed,
+            );
 
             tracing::info!(
                 level = idx + 1,
diff --git a/crates/korg-runtime/src/workers.rs b/crates/korg-runtime/src/workers.rs
index 762b848..1f65489 100644
--- a/crates/korg-runtime/src/workers.rs
+++ b/crates/korg-runtime/src/workers.rs
@@ -585,6 +585,56 @@ pub fn compose_downstream_payload(
     format!("{base}{appended}")
 }
 
+/// Rewrite every still-pending downstream package whose dependencies include a
+/// node that just completed, appending that upstream node's real
+/// `PersonaResult.output` to the downstream payload (Captain+Harper → Benjamin;
+/// Benjamin → Lucas).
+///
+/// This is the per-level data-flow step extracted from `dispatch_concurrent` so
+/// it can be unit-tested in isolation. Behavior is identical to the inline loop:
+///
+/// - A node already present in `completed_outputs` is skipped (it's done).
+/// - A node is only rewritten if at least one of its dependencies is in
+///   `just_completed_ids` (we just produced new upstream context for it).
+/// - Upstream context is gathered from `completed_outputs` for ALL of the node's
+///   dependencies (so a node with two upstreams sees both once both finish), and
+///   composed deterministically + size-capped via [`compose_downstream_payload`].
+///
+/// `completed_outputs` maps `node_id -> (persona_name, output_json)`.
+/// `node_dependencies` maps `node_id -> [dependency_node_id, …]`.
+pub fn apply_upstream_to_pending(
+    packages_map: &mut HashMap<String, WorkPackage>,
+    node_dependencies: &HashMap<String, Vec<String>>,
+    completed_outputs: &HashMap<String, (String, serde_json::Value)>,
+    just_completed_ids: &std::collections::HashSet<String>,
+) {
+    for (node_id, deps) in node_dependencies {
+        // Only rewrite nodes that depend on something we just finished
+        // (and are not themselves done yet).
+        if completed_outputs.contains_key(node_id) {
+            continue;
+        }
+        if !deps.iter().any(|d| just_completed_ids.contains(d)) {
+            continue;
+        }
+        // Gather all completed upstream outputs for this node's deps.
+        let upstream: Vec<(String, String, serde_json::Value)> = deps
+            .iter()
+            .filter_map(|dep_id| {
+                completed_outputs
+                    .get(dep_id)
+                    .map(|(persona, output)| (persona.clone(), dep_id.clone(), output.clone()))
+            })
+            .collect();
+        if upstream.is_empty() {
+            continue;
+        }
+        if let Some(pkg) = packages_map.get_mut(node_id) {
+            pkg.description = compose_downstream_payload(&pkg.description, &upstream);
+        }
+    }
+}
+
 /// Build the canonical 4-persona campaign DAG and return topological levels.
 /// Speculative pre-warm is gated on the `speculative_execution` capability.
 #[tracing::instrument(skip(root_task, tui_tx))]
@@ -1055,4 +1105,148 @@ mod tests {
         let out = compose_downstream_payload("base only", &[]);
         assert_eq!(out, "base only");
     }
+
+    // --- Slice 2 integration: the dispatch_concurrent per-level rewrite loop ---
+
+    fn dataflow_fixture() -> (HashMap<String, WorkPackage>, HashMap<String, Vec<String>>) {
+        // The canonical 4-node campaign DAG:
+        //   captain, harper (no deps) → benjamin (deps: captain, harper) → lucas (deps: benjamin)
+        let mk = |id: &str, persona: Persona, desc: &str| WorkPackage {
+            node_id: id.to_string(),
+            persona,
+            description: desc.to_string(),
+            routing_id: id.to_string(),
+        };
+        let mut packages_map = HashMap::new();
+        packages_map.insert(
+            "pkg-captain".into(),
+            mk("pkg-captain", Persona::Captain, "Plan: root task"),
+        );
+        packages_map.insert(
+            "pkg-harper".into(),
+            mk("pkg-harper", Persona::Harper, "Research: root task"),
+        );
+        packages_map.insert(
+            "pkg-benjamin".into(),
+            mk("pkg-benjamin", Persona::Benjamin, "Implement: root task"),
+        );
+        packages_map.insert(
+            "pkg-lucas".into(),
+            mk("pkg-lucas", Persona::Lucas, "Synthesize: root task"),
+        );
+
+        let mut node_dependencies = HashMap::new();
+        node_dependencies.insert("pkg-captain".to_string(), vec![]);
+        node_dependencies.insert("pkg-harper".to_string(), vec![]);
+        node_dependencies.insert(
+            "pkg-benjamin".to_string(),
+            vec!["pkg-captain".to_string(), "pkg-harper".to_string()],
+        );
+        node_dependencies.insert("pkg-lucas".to_string(), vec!["pkg-benjamin".to_string()]);
+
+        (packages_map, node_dependencies)
+    }
+
+    #[test]
+    fn apply_upstream_to_pending_threads_captain_into_benjamin_then_benjamin_into_lucas() {
+        let (mut packages_map, node_dependencies) = dataflow_fixture();
+        let mut completed_outputs: HashMap<String, (String, serde_json::Value)> = HashMap::new();
+
+        // --- L1 completes: captain + harper produce real outputs ---
+        completed_outputs.insert(
+            "pkg-captain".to_string(),
+            (
+                "Captain".to_string(),
+                serde_json::json!({
+                    "work_packages": [{"id": 1, "title": "Fix add"}],
+                    "acceptance_criteria": ["add(2,3)==5"]
+                }),
+            ),
+        );
+        completed_outputs.insert(
+            "pkg-harper".to_string(),
+            (
+                "Harper".to_string(),
+                serde_json::json!({ "concerns": [{"id": "c1", "file_path": "src/lib.rs"}] }),
+            ),
+        );
+        let l1_completed: std::collections::HashSet<String> =
+            ["pkg-captain".to_string(), "pkg-harper".to_string()]
+                .into_iter()
+                .collect();
+
+        apply_upstream_to_pending(
+            &mut packages_map,
+            &node_dependencies,
+            &completed_outputs,
+            &l1_completed,
+        );
+
+        // Benjamin's payload now carries BOTH captain's plan and harper's concerns
+        // (the real rewrite — downstream payload carries upstream content).
+        let benjamin_desc = &packages_map["pkg-benjamin"].description;
+        assert!(
+            benjamin_desc.starts_with("Implement: root task"),
+            "base payload must be preserved"
+        );
+        assert!(
+            benjamin_desc.contains("work_packages")
+                && benjamin_desc.contains("Upstream from Captain (pkg-captain)"),
+            "benjamin payload must contain Captain's plan marker, got:\n{benjamin_desc}"
+        );
+        assert!(
+            benjamin_desc.contains("concerns")
+                && benjamin_desc.contains("Upstream from Harper (pkg-harper)"),
+            "benjamin payload must also contain Harper's concerns marker"
+        );
+        // Lucas not yet rewritten — its dep (benjamin) hasn't completed.
+        assert_eq!(
+            packages_map["pkg-lucas"].description, "Synthesize: root task",
+            "lucas must NOT be rewritten before benjamin completes"
+        );
+        // L1 peers (captain/harper) are not themselves rewritten.
+        assert_eq!(packages_map["pkg-captain"].description, "Plan: root task");
+
+        // --- L2 completes: benjamin produces output referencing the implement step ---
+        completed_outputs.insert(
+            "pkg-benjamin".to_string(),
+            (
+                "Benjamin".to_string(),
+                serde_json::json!({
+                    "mutations": [{"target": "src/lib.rs", "action": "update"}]
+                }),
+            ),
+        );
+        let l2_completed: std::collections::HashSet<String> =
+            ["pkg-benjamin".to_string()].into_iter().collect();
+
+        apply_upstream_to_pending(
+            &mut packages_map,
+            &node_dependencies,
+            &completed_outputs,
+            &l2_completed,
+        );
+
+        // Now Lucas's payload carries Benjamin's real output marker.
+        let lucas_desc = &packages_map["pkg-lucas"].description;
+        assert!(
+            lucas_desc.starts_with("Synthesize: root task"),
+            "lucas base payload must be preserved"
+        );
+        assert!(
+            lucas_desc.contains("mutations")
+                && lucas_desc.contains("src/lib.rs")
+                && lucas_desc.contains("Upstream from Benjamin (pkg-benjamin)"),
+            "lucas payload must contain Benjamin's output marker, got:\n{lucas_desc}"
+        );
+        // Benjamin is now done; it must not be re-rewritten as if pending.
+        assert!(
+            packages_map["pkg-benjamin"]
+                .description
+                .matches("Upstream from Captain")
+                .count()
+                == 1,
+            "benjamin must not be rewritten again after it completes"
+        );
+    }
 }