car-server-core 0.24.1

//! Foreman delegation: verified parallel coding inside a coder session.
//!
//! Composition of the two systems, each keeping its own boundary:
//!
//! - **Foreman** (car-multi `patterns/foreman`, #274) decomposes the intent,
//!   farms subtasks to an external CLI in per-subtask worktrees, and gates
//!   each patch plus the integrated union (AST containment + build/test +
//!   policy). Its repo root is the **coder session's worktree** — clean at
//!   HEAD when delegation starts — so subtask worktrees and the staging tree
//!   never see the user's checkout.
//! - **The coder** applies the gate-accepted union into its session worktree
//!   and then evaluates the **outcome contract** itself. Foreman's gate is an
//!   inner filter; the contract stays the outer trust boundary, exactly as
//!   with the single-session external engine.
//!
//! Fallback ladder (driven by [`super::rpc`]): foreman declining
//! (`prefer_single_session`, invalid plan, nothing accepted, integration
//! rejected) → single-session external CLI → native loop. A red contract
//! *after* foreman applied work falls to the native loop too — which then
//! repairs **on top of** foreman's changes rather than starting over.
//!
//! When the daemon's MCP listener is bound, its URL is threaded into
//! [`car_multi::FarmOutConfig::mcp_endpoint`] so the farmed-out CLI workers'
//! CAR-namespace tool calls (`memory_*`, `verify`, `skill_*`) route back
//! through car-server's policy + memgine — gated and audited. The workers' own
//! built-in tools (Edit, Bash) stay ungoverned (the residual upstream stage-4b
//! limitation), contained by the per-worktree gate and the outer contract.
//!
//! Cancellation is honored between stages (plan / farm / integrate); an
//! in-flight farm-out cannot be killed mid-stage yet (same limitation as the
//! single-session external engine).

use std::path::Path;
use std::sync::atomic::Ordering;
use std::sync::Arc;

use serde_json::json;

use super::contract::{evaluate_contract, OutcomeContract};
use super::native_loop::{LoopOutcome, TurnGenerator};
use super::session::{CancelFlag, CoderEventKind, EventSink};
use super::shell_tool::{tail, WorktreeExecutor};

/// Why foreman declined, so the caller can fall down the ladder. Not an
/// error: every variant has a working next step.
#[derive(Debug)]
pub enum ForemanFallback {
    /// The plan says farming out buys nothing (≤1 subtask / no parallelism).
    SingleSessionPreferred,
    /// The planner could not produce a valid decomposition.
    PlanInvalid(String),
    /// No subtask survived the per-worktree gate.
    NothingAccepted(String),
    /// The accepted union failed the integration gate or did not apply to
    /// the session worktree.
    IntegrationRejected(String),
}

impl ForemanFallback {
    pub fn reason(&self) -> String {
        match self {
            Self::SingleSessionPreferred => {
                "plan prefers a single session (no parallel speedup)".into()
            }
            Self::PlanInvalid(e) => format!("decomposition invalid: {e}"),
            Self::NothingAccepted(e) => format!("no subtask passed the merge gate: {e}"),
            Self::IntegrationRejected(e) => format!("union integration rejected: {e}"),
        }
    }
}

/// The union gate's **goal** leg (#275 `union_verify_command`), derived from
/// the contract: every plain exit-zero check chained with `&&`. The contract
/// is a goal check by construction — it must only ever gate the integrated
/// union, never a single subtask (a subtask legitimately implements part of
/// the goal). Checks that assert on output substrings (or invert the exit
/// code) can't be expressed as an argv exit status — they are *omitted here*
/// and still enforced by the coder's own contract evaluation afterwards,
/// which is the outer boundary anyway.
fn union_goal_command(contract: &OutcomeContract) -> Option<Vec<String>> {
    let chain: Vec<&str> = contract
        .checks
        .iter()
        .filter(|c| c.expect_exit_zero && c.output_contains.is_none())
        .map(|c| c.command.as_str())
        .collect();
    if chain.is_empty() {
        return None;
    }
    Some(vec!["sh".into(), "-lc".into(), chain.join(" && ")])
}

/// The per-worktree **regression** leg (#275 `verify_command`): "does this
/// one subtask's change still build?" — derived from the repo's detected
/// build system, NOT from the contract. Conservative: only build systems
/// with an unambiguous cheap check are mapped; `None` otherwise, in which
/// case the per-worktree gate is `Inconclusive` (fail-closed, no waiver) and
/// the session falls down the ladder to single-session delegation.
fn regression_command(worktree: &Path) -> Option<Vec<String>> {
    let candidates: [(&str, &str); 3] = [
        ("Cargo.toml", "cargo check"),
        ("go.mod", "go build ./..."),
        ("Package.swift", "swift build"),
    ];
    candidates
        .iter()
        .find(|(marker, _)| worktree.join(marker).exists())
        .map(|(_, cmd)| vec!["sh".into(), "-lc".into(), (*cmd).into()])
}

/// Apply a gate-accepted patch into the session worktree.
fn apply_patch(worktree: &Path, subtask_id: &str, patch: &str) -> Result<(), String> {
    use std::io::Write;
    let mut file = tempfile::NamedTempFile::new()
        .map_err(|e| format!("temp patch file for {subtask_id}: {e}"))?;
    file.write_all(patch.as_bytes())
        .map_err(|e| format!("write patch {subtask_id}: {e}"))?;
    let out = std::process::Command::new("git")
        .arg("-C")
        .arg(worktree)
        .args(["apply", "--whitespace=nowarn"])
        .arg(file.path())
        .output()
        .map_err(|e| format!("git apply {subtask_id}: {e}"))?;
    if out.status.success() {
        Ok(())
    } else {
        Err(format!(
            "git apply {subtask_id} failed: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        ))
    }
}

/// Run foreman delegation to a contract-evaluated outcome, or decline with a
/// fallback the caller can act on.
pub async fn run_foreman_loop(
    adapter_id: &str,
    intent: &str,
    contract: &OutcomeContract,
    executor: &WorktreeExecutor,
    sink: &Arc<EventSink>,
    cancel: &CancelFlag,
    generator: &Arc<dyn TurnGenerator>,
    // Daemon MCP URL, when bound. Routes the farmed-out CLI workers'
    // CAR-namespace tool calls through the daemon's policy + memgine.
    // `None` degrades cleanly.
    mcp_endpoint: Option<&str>,
) -> Result<LoopOutcome, ForemanFallback> {
    let worktree = executor.worktree().to_path_buf();
    let cancelled = || LoopOutcome {
        passed: false,
        iterations: 0,
        last_results: Vec::new(),
        error: Some("cancelled".into()),
    };

    // 1. Plan — decompose the intent against the session worktree.
    if cancel.load(Ordering::SeqCst) {
        return Ok(cancelled());
    }
    sink.emit(CoderEventKind::ExternalEvent {
        raw: json!({ "foreman": "planning", "adapter": adapter_id }),
    });
    let plan_generator = generator.clone();
    let plan = car_multi::decompose(&worktree, intent, 3, move |prompt| {
        let generator = plan_generator.clone();
        async move {
            generator
                .generate(car_inference::GenerateRequest {
                    prompt,
                    ..Default::default()
                })
                .await
                .map(|r| r.text)
        }
    })
    .await;

    if !plan.is_valid() {
        return Err(ForemanFallback::PlanInvalid(plan.issues.join("; ")));
    }
    sink.emit(CoderEventKind::ExternalEvent {
        raw: json!({
            "foreman": "planned",
            "subtasks": plan.subtasks.len(),
            "levels": plan.levels.len(),
            "prefer_single_session": plan.prefer_single_session,
        }),
    });
    if plan.prefer_single_session {
        return Err(ForemanFallback::SingleSessionPreferred);
    }

    // 2. Farm out — per-subtask worktrees + per-patch gate, against the
    //    contract-derived build/test leg.
    if cancel.load(Ordering::SeqCst) {
        return Ok(cancelled());
    }
    let agent = car_external_agents::ForemanExternalAgent::new(adapter_id.to_string());
    let infra = car_multi::SharedInfra::new();
    let config = car_multi::FarmOutConfig {
        // Regression vs goal split (#275): per-worktree gets the build-system
        // check; the integrated union gets the contract.
        verify_command: regression_command(&worktree),
        union_verify_command: union_goal_command(contract),
        // Gate + audit the farmed-out workers' CAR-namespace tool calls
        // through the daemon when its MCP listener is bound; None degrades
        // cleanly (the workers' own built-in tools stay ungoverned — the
        // residual upstream stage-4b limitation).
        mcp_endpoint: mcp_endpoint.map(String::from),
        ..Default::default()
    };
    // Stream each subtask's worktree lifecycle (started / gated) so a live UI
    // can show the parallel run advancing instead of only the run-level
    // milestones. Bridges `ForemanProgress` → the coder's `external_event`
    // channel, tagged `foreman` like the run-level stages.
    let progress_sink: car_multi::ForemanProgressSink = {
        let sink = Arc::clone(sink);
        Arc::new(move |ev: car_multi::ForemanProgress| {
            let raw = match ev {
                car_multi::ForemanProgress::SubtaskStarted {
                    subtask_id,
                    index,
                    level,
                    total,
                } => json!({
                    "foreman": "subtask_started",
                    "subtask_id": subtask_id,
                    "index": index,
                    "level": level,
                    "total": total,
                }),
                car_multi::ForemanProgress::SubtaskVerifying { subtask_id } => json!({
                    "foreman": "subtask_verifying",
                    "subtask_id": subtask_id,
                }),
                car_multi::ForemanProgress::SubtaskGated {
                    subtask_id,
                    accepted,
                    status,
                } => json!({
                    "foreman": "subtask_gated",
                    "subtask_id": subtask_id,
                    "accepted": accepted,
                    "status": status,
                }),
            };
            sink.emit(CoderEventKind::ExternalEvent { raw });
        })
    };
    let farmed = car_multi::run_farm_out_with_progress(
        &worktree,
        &plan.subtasks,
        &agent,
        &config,
        &infra,
        progress_sink,
    )
    .await;

    let accepted: Vec<(String, String)> = farmed
        .outcomes
        .iter()
        .filter(|o| o.is_accepted())
        .filter_map(|o| o.patch.clone().map(|p| (o.subtask_id.clone(), p)))
        .collect();
    sink.emit(CoderEventKind::ExternalEvent {
        raw: json!({
            "foreman": "farmed",
            "accepted": accepted.len(),
            "total": farmed.outcomes.len(),
        }),
    });
    if accepted.is_empty() {
        let detail = farmed
            .outcomes
            .iter()
            .filter_map(|o| o.error.as_deref())
            .take(3)
            .collect::<Vec<_>>()
            .join("; ");
        return Err(ForemanFallback::NothingAccepted(if detail.is_empty() {
            format!("{} subtask(s) all rejected or inconclusive", farmed.outcomes.len())
        } else {
            detail
        }));
    }

    // 3. Gate the integrated union in foreman's staging tree.
    if cancel.load(Ordering::SeqCst) {
        return Ok(cancelled());
    }
    let label = format!("coder-{}", sink_label(&worktree));
    let integration =
        car_multi::integrate_and_verify(&worktree, &label, &accepted, &config, &infra)
            .await
            .map_err(|e| ForemanFallback::IntegrationRejected(e.to_string()))?;
    if !integration.integrated_cleanly() {
        // Surface WHY the union failed (structured blame) so a UI can show which
        // subtasks are implicated — the subtasks all gated green individually, so
        // without this the board would read as success while the run failed.
        if let Some(blame) = &integration.blame {
            let reason = if !blame.apply_conflicts.is_empty() {
                "patch conflict"
            } else if !blame.duplicate_conflicts.is_empty() {
                "duplicate declaration"
            } else if blame.build_test.is_some() {
                "build/test failed"
            } else {
                "rejected"
            };
            // Same precedence as `reason` above (apply → duplicate → build_test)
            // so when more than one cause is ever populated, the banner's reason
            // and detail describe the SAME cause rather than two different ones.
            let detail = blame
                .apply_conflicts
                .first()
                .map(|c| format!("{} did not apply", c.subtask_id))
                .or_else(|| {
                    blame
                        .duplicate_conflicts
                        .first()
                        .map(|d| format!("duplicate `{}` in {}", d.symbol, d.file))
                })
                .or_else(|| blame.build_test.as_ref().map(|b| tail(&b.output_tail, 200)));
            let implicated: Vec<String> = blame.implicated_subtasks().into_iter().collect();
            sink.emit(CoderEventKind::ExternalEvent {
                raw: json!({
                    "foreman": "union_rejected",
                    "reason": reason,
                    "implicated": implicated,
                    "detail": detail,
                }),
            });
        }
        return Err(ForemanFallback::IntegrationRejected(format!(
            "applied {}, conflicts: [{}], union verdict accepting: {}",
            integration.applied,
            integration.apply_conflicts.join(", "),
            integration
                .verdict
                .as_ref()
                .is_some_and(|v| v.is_accepted()),
        )));
    }
    sink.emit(CoderEventKind::ExternalEvent {
        raw: json!({ "foreman": "union_verified", "applied": integration.applied }),
    });

    // 4. Land the verified union in the session worktree (clean at HEAD, the
    //    same base the staging tree gated, so application is deterministic).
    for (subtask_id, patch) in &accepted {
        apply_patch(&worktree, subtask_id, patch)
            .map_err(ForemanFallback::IntegrationRejected)?;
        sink.emit(CoderEventKind::ToolResult {
            tool: "foreman.apply".into(),
            ok: true,
            preview: format!("applied {subtask_id}"),
        });
    }

    // 5. The outer boundary: the coder's own contract evaluation.
    let last_results = evaluate_contract(contract, executor, sink).await;
    let passed = last_results.iter().all(|r| r.passed);
    Ok(LoopOutcome {
        passed,
        iterations: 1,
        last_results,
        error: None,
    })
}

/// Stable per-session label fragment for foreman's staging worktree, derived
/// from the session worktree's directory name (which embeds the session id).
fn sink_label(worktree: &Path) -> String {
    worktree
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_else(|| "session".into())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::coder::contract::ContractCheck;

    fn check(name: &str, command: &str, exit_zero: bool, contains: Option<&str>) -> ContractCheck {
        ContractCheck {
            name: name.into(),
            command: command.into(),
            expect_exit_zero: exit_zero,
            output_contains: contains.map(String::from),
            timeout_secs: 60,
        }
    }

    #[test]
    fn union_goal_chains_plain_exit_zero_checks_only() {
        let contract = OutcomeContract {
            description: "d".into(),
            checks: vec![
                check("build", "cargo build", true, None),
                check("tests", "cargo test", true, None),
                check("output", "cat x.txt", true, Some("needle")), // not expressible
                check("inverted", "grep -q bad src/", false, Some("x")), // not expressible
            ],
        };
        let cmd = union_goal_command(&contract).unwrap();
        assert_eq!(cmd[0], "sh");
        assert_eq!(cmd[2], "cargo build && cargo test");
    }

    #[test]
    fn no_expressible_checks_means_no_union_goal_command() {
        let contract = OutcomeContract {
            description: "d".into(),
            checks: vec![check("output", "cat x.txt", true, Some("needle"))],
        };
        assert!(union_goal_command(&contract).is_none());
    }

    #[test]
    fn regression_command_maps_known_build_systems_only() {
        let dir = tempfile::tempdir().unwrap();
        assert!(regression_command(dir.path()).is_none(), "unknown repo → None (fail-closed)");
        std::fs::write(dir.path().join("Cargo.toml"), "[package]").unwrap();
        let cmd = regression_command(dir.path()).unwrap();
        assert_eq!(cmd[2], "cargo check");
    }

    #[test]
    fn apply_patch_lands_changes_in_worktree() {
        let dir = tempfile::tempdir().unwrap();
        for args in [
            vec!["init", "-q", "-b", "main"],
            vec!["-c", "user.name=t", "-c", "user.email=t@t", "commit", "-q", "--allow-empty", "-m", "init"],
        ] {
            assert!(std::process::Command::new("git")
                .arg("-C")
                .arg(dir.path())
                .args(&args)
                .output()
                .unwrap()
                .status
                .success());
        }
        let patch = "diff --git a/new.txt b/new.txt\nnew file mode 100644\n--- /dev/null\n+++ b/new.txt\n@@ -0,0 +1 @@\n+from foreman\n";
        apply_patch(dir.path(), "s1", patch).unwrap();
        assert_eq!(
            std::fs::read_to_string(dir.path().join("new.txt")).unwrap(),
            "from foreman\n"
        );
    }

    #[test]
    fn apply_patch_conflict_is_reported_not_panicked() {
        let dir = tempfile::tempdir().unwrap();
        assert!(std::process::Command::new("git")
            .arg("-C")
            .arg(dir.path())
            .args(["init", "-q"])
            .output()
            .unwrap()
            .status
            .success());
        let err = apply_patch(dir.path(), "s1", "not a patch").unwrap_err();
        assert!(err.contains("git apply s1 failed"), "{err}");
    }

    #[test]
    fn fallback_reasons_are_descriptive() {
        assert!(ForemanFallback::SingleSessionPreferred.reason().contains("single session"));
        assert!(ForemanFallback::PlanInvalid("x".into()).reason().contains("decomposition"));
        assert!(ForemanFallback::NothingAccepted("y".into()).reason().contains("merge gate"));
        assert!(ForemanFallback::IntegrationRejected("z".into()).reason().contains("integration"));
    }
}