agent-joule/crates/joule_coded/tests/fanout.rs

//! The fan-out driver, end to end over real git worktrees: decompose a plan into
//! dependency-ordered parallel waves, drive the real L3 loop in an isolated
//! worktree session per task (with a scripted on-device model standing in for the
//! GGUF runtime), then commit, verify, and merge each branch into the base before
//! the next wave runs. Proves the second wave's two independent tasks both land in
//! the base on top of the first wave's merged work — parallel subagents, grounded
//! in the spec's task graph, with intact signed ledgers throughout.

use ed25519_dalek::SigningKey;
use joule_code_schema::ContentId;
use joule_coded::{run_plan_fanout, seed_from};
use joule_core::fabric::{DeviceClass, FabricInventory, IsolationTier};
use joule_core::methodology::{Plan, PlanTask};
use joule_core::sessions::{SessionGrant, SessionManager};
use std::collections::BTreeMap;
use std::process::Command;

fn task(id: &str, clauses: &[&str], deps: &[&str]) -> PlanTask {
    PlanTask {
        id: id.into(),
        title: format!("rewrite for {id}"),
        clauses: clauses.iter().map(|c| c.to_string()).collect(),
        deps: deps.iter().map(|d| d.to_string()).collect(),
    }
}

/// A throwaway git repo with three committed files the agents will rewrite.
/// `label` keeps parallel tests (same PID) from sharing a directory.
// Sync test setup: blocking on `git` is intended (no async executor in a test).
#[allow(clippy::disallowed_methods)]
fn scratch_repo(label: &str) -> (std::path::PathBuf, std::path::PathBuf) {
    let base = std::env::temp_dir().join(format!("jc-fanout-{}-{label}", std::process::id()));
    let _ = std::fs::remove_dir_all(&base);
    let repo = base.join("repo");
    let worktrees = base.join("worktrees");
    std::fs::create_dir_all(&repo).unwrap();
    std::fs::create_dir_all(&worktrees).unwrap();
    let git = |args: &[&str]| {
        assert!(
            Command::new("git").args(args).current_dir(&repo).output().unwrap().status.success(),
            "git {args:?}"
        );
    };
    git(&["init", "-q"]);
    git(&["config", "user.email", "t@t"]);
    git(&["config", "user.name", "t"]);
    for name in ["a.txt", "b.txt", "c.txt"] {
        std::fs::write(repo.join(name), "0\n").unwrap();
    }
    git(&["add", "-A"]);
    git(&["commit", "-qm", "init"]);
    (repo, worktrees)
}

#[test]
fn fanout_runs_a_plan_across_waves_and_merges_each_into_the_base() {
    let (repo, worktrees) = scratch_repo("merge");

    let key = SigningKey::from_bytes(&seed_from(&repo.to_string_lossy()));
    let author = key.verifying_key().to_bytes();
    let grant = SessionGrant {
        tiers: vec!["l0".into(), "l1".into(), "l2".into(), "l3".into()],
        joule_ceiling_uj: 1_000_000_000,
    };
    let mut sessions = SessionManager::with_primary(
        repo.clone(),
        key,
        worktrees,
        grant.clone(),
        "fanout-worker",
    );

    // A diamond's head and its two independent branches: T0 must finish before
    // T1 and T2, which are independent of each other → wave 1 = [T0], wave 2 =
    // [T1, T2] (run in parallel). Each task rewrites a distinct file.
    let spec = ContentId::of_bytes(b"spec");
    let clauses = ["C0", "C1", "C2"];
    let plan = Plan::new(
        spec,
        vec![
            task("T0", &["C0"], &[]),
            task("T1", &["C1"], &["T0"]),
            task("T2", &["C2"], &["T0"]),
        ],
        &clauses,
    )
    .unwrap();
    assert_eq!(plan.max_parallelism(), 2, "the plan fans out 2-wide at its widest");

    // The scripted model per task: rewrite this task's file, then finish.
    let targets: BTreeMap<&str, (&str, &str)> =
        [("T0", ("a.txt", "alpha")), ("T1", ("b.txt", "bravo")), ("T2", ("c.txt", "charlie"))]
            .into_iter()
            .collect();
    let model_for = |task: &PlanTask| -> Box<dyn FnMut(&str) -> anyhow::Result<String> + Send> {
        let (path, content) = targets[task.id.as_str()];
        let turns = [
            format!("<<<TOOL edit_file>>>\n{path}\n{content}\n<<<END>>>"),
            "<<<TOOL finish>>>\n".to_string(),
        ];
        let mut step = 0usize;
        Box::new(move |_prompt: &str| {
            let turn = turns[step.min(turns.len() - 1)].clone();
            step += 1;
            Ok(turn)
        })
    };

    // An Apple-Silicon-shaped host: CPU + GPU + NPU, isolates up to S2.
    let inventory = FabricInventory {
        devices: vec![DeviceClass::Cpu, DeviceClass::Gpu, DeviceClass::Npu],
        max_tier: IsolationTier::S2,
    };

    let report = run_plan_fanout(
        &mut sessions,
        &plan,
        &grant,
        &inventory,
        8,
        author,
        model_for,
        |_trace| 10_000, // host meter stand-in: 10 mJ per task
        |trace| trace.error.is_none(),
    )
    .unwrap();

    // Two waves; the widest staffed two subagents; all three tasks merged.
    assert_eq!(report.waves, 2, "T0 alone, then T1+T2");
    assert_eq!(report.max_width, 2);
    assert_eq!(report.merged(), 3, "every task merged its branch into the base");
    assert_eq!(report.total_ops(), 3, "one edit sealed as an op per task");
    assert_eq!(report.total_energy_uj(), 30_000);

    // Fabric placement: every L3 task ran on the NPU (cheapest accelerator) at
    // the S2 isolation its untrusted shell work requires, and the per-device
    // energy rollup attributes all 30 mJ to that device. Its shell commands were
    // confined by a real OS sandbox (seatbelt on macOS), network-denied.
    for outcome in &report.outcomes {
        assert!(outcome.ran);
        assert_eq!(outcome.placement.device, DeviceClass::Npu);
        assert_eq!(outcome.placement.isolation, IsolationTier::S2);
        assert!(outcome.placement.feasible);
        assert!(outcome.enforcement.is_enforced(), "S2 shell work must be OS-confined");
    }
    let by_device = report.energy_by_device();
    assert_eq!(by_device.get(&DeviceClass::Npu).copied(), Some(30_000));

    // The base checkout now holds all three rewrites — wave 2's independent work
    // landed on top of wave 1's merged commit.
    for (_, (file, content)) in targets {
        let on_disk = std::fs::read_to_string(repo.join(file)).unwrap();
        assert!(on_disk.contains(content), "{file} not rewritten in base: {on_disk:?}");
    }

    // Every session's signed ledger is intact end to end through merge + close.
    for outcome in &report.outcomes {
        let session = sessions.get(&outcome.session_id).unwrap();
        assert!(session.verify_chain(), "ledger for {} must verify", outcome.task_id);
    }

    std::fs::remove_dir_all(&repo.parent().unwrap()).ok();
}

#[test]
fn fanout_refuses_every_task_when_the_host_cannot_isolate_untrusted_shell() {
    let (repo, worktrees) = scratch_repo("refuse");
    let key = SigningKey::from_bytes(&seed_from(&repo.to_string_lossy()));
    let author = key.verifying_key().to_bytes();
    let grant = SessionGrant {
        tiers: vec!["l0".into(), "l1".into(), "l2".into(), "l3".into()],
        joule_ceiling_uj: 1_000_000_000,
    };
    let mut sessions =
        SessionManager::with_primary(repo.clone(), key, worktrees, grant.clone(), "fanout-worker");

    let plan = Plan::new(
        ContentId::of_bytes(b"spec"),
        vec![task("T0", &["C0"], &[]), task("T1", &["C1"], &["T0"])],
        &["C0", "C1"],
    )
    .unwrap();

    // A host with no isolation (max_tier S0) cannot enforce the S2 floor an L3
    // task's untrusted shell requires. The model factory must never be reached.
    let bare = FabricInventory { devices: vec![DeviceClass::Cpu], max_tier: IsolationTier::S0 };
    let report = run_plan_fanout(
        &mut sessions,
        &plan,
        &grant,
        &bare,
        8,
        author,
        |_task| Box::new(|_p: &str| panic!("a refused task must not run the model")),
        |_trace| 10_000,
        |trace| trace.error.is_none(),
    )
    .unwrap();

    assert_eq!(report.waves, 0, "nothing executed");
    assert_eq!(report.merged(), 0);
    assert_eq!(report.outcomes.len(), 2, "both tasks accounted for, as refused");
    for outcome in &report.outcomes {
        assert!(!outcome.ran, "task {} must be refused, not run unsandboxed", outcome.task_id);
        assert!(!outcome.placement.feasible);
        assert_eq!(outcome.placement.isolation, IsolationTier::S2);
    }
    // The base checkout is untouched — no worktree was even opened.
    assert_eq!(std::fs::read_to_string(repo.join("a.txt")).unwrap(), "0\n");

    std::fs::remove_dir_all(&repo.parent().unwrap()).ok();
}