newt-agent 0.8.0

//! `newt crew` — the human front door to the multi-LLM crew control loop
//! (`newt_scheduler::run_crew`). This module provides the **effects** side: a
//! [`WorktreeWorkspace`] that runs the crew against an *isolated git worktree*
//! (never the live tree — the harness-owns-verification guardrail), plus the
//! test-command inference the front door uses.
//!
//! Design: `docs/design/crew-front-door-and-workflow-tui.md`,
//! `docs/design/config-scaling-deployment-and-trust.md`.

use std::path::{Path, PathBuf};

use newt_core::scope_grounding::{definition_grep_pattern, grep_terms, ground_scope};
use newt_scheduler::{Edit, Workspace};

/// Infer the verification command for a repo at `dir`, in priority order:
/// a `justfile` → `just check`; a `Cargo.toml` → `cargo test`; a Python project
/// (`pyproject.toml` / `pytest.ini` / `tox.ini`) → `pytest -x`. Returns `None`
/// when none is found — the front door then **refuses** rather than running a
/// silent no-op "test" (a crew that never verified would be a false success).
pub fn infer_test_command(dir: &Path) -> Option<String> {
    let has = |name: &str| dir.join(name).exists();
    if has("justfile") || has("Justfile") {
        Some("just check".to_string())
    } else if has("Cargo.toml") {
        Some("cargo test".to_string())
    } else if has("pyproject.toml") || has("pytest.ini") || has("tox.ini") {
        Some("pytest -x".to_string())
    } else {
        None
    }
}

/// The normalize/format commands the crew finalize runs (in the worktree) before
/// it commits, so a model quirk (e.g. a dropped trailing newline) doesn't bounce
/// off the gate's `fmt --check` (#880).
///
/// De-hard-coded: the toolchain → formatter mapping is DATA — composable
/// [`newt_core::tooling`] packs (built-in + `~/.newt/tooling/*.toml` drop-ins),
/// so a polyglot repo runs SEVERAL formatters and an unknown repo runs NONE. The
/// `NEWT_CREW_NORMALIZE` env override wins (a single command, or `none`/empty to
/// DISABLE) — configuration over convention.
pub fn crew_normalize_commands(dir: &Path) -> Vec<String> {
    match std::env::var("NEWT_CREW_NORMALIZE") {
        Ok(v) if v.trim().is_empty() || v.trim().eq_ignore_ascii_case("none") => Vec::new(),
        Ok(v) => vec![v],
        Err(_) => {
            newt_core::tooling::resolved_phase_commands(dir, newt_core::tooling::Phase::Format)
        }
    }
}

/// Is `path` a safe in-worktree edit target — built only from `Normal` (and `.`)
/// components, so no root/drive prefix and no `..` escape? `Path::join` discards
/// the base for an absolute path, so this guard (not the `fs_write` caveat, which
/// is `Scope::All` on the crew/plan path) is the real worktree boundary.
///
/// Checked by components, NOT `is_absolute()`: on Windows `/etc/passwd` is
/// root-relative and `is_absolute()` is `false`, so an `is_absolute` guard would
/// wrongly admit it. A `RootDir` / drive `Prefix` / `ParentDir` component is
/// refused on every platform.
fn is_safe_worktree_path(path: &str) -> bool {
    use std::path::Component;
    Path::new(path)
        .components()
        .all(|c| matches!(c, Component::Normal(_) | Component::CurDir))
}

/// Run a repo-configured build / test / format command **confined** inside the
/// crew worktree (P4 / `p4-constrained-executor`). The command string comes from
/// the repo's tooling config (`resolved_phase_commands`) or the operator's
/// `--verify`, so it is attacker-influenced; it no longer runs as a raw `sh -c`.
///
/// Through [`ConstrainedExecutor`] the child starts env-EMPTY — only
/// `PATH`/`HOME`/`TMPDIR`/`CARGO_TARGET_DIR` are granted, no credentials or
/// authority switches (#8) — its writes are fenced to the worktree plus the crew
/// build infrastructure (the shared `CARGO_TARGET_DIR` and the operator's Cargo
/// cache; crew is an operator-run build context), its reads are the calibrated
/// toolchain/cache set (not `~/.ssh` etc.), its **network is denied** so nothing
/// exfiltrates (#9), and it **fails closed** off the kernel fence (#10). Returns
/// `(success, combined stdout+stderr)` or an `Err` describing the refusal.
fn run_confined_build(worktree: &Path, base: &Path, cmd: &str) -> Result<(bool, String), String> {
    use newt_core::confined_exec::{
        build_tool_caveats_with_writes, ConstrainedExecutor, ExecOrigin, ExecRequest, NetGrant,
    };
    let target = crew_shared_target_dir(base);
    let target_str = target.to_string_lossy().into_owned();
    // Grant the shared crew target + the operator's Cargo cache as write roots so
    // incremental builds and the `~/.cargo/.package-cache` lock work. Both are
    // operator/build infrastructure, never repository-controlled paths.
    let mut extra_writes = vec![target_str.clone()];
    if let Some(cargo_home) = std::env::var_os("CARGO_HOME")
        .map(PathBuf::from)
        .or_else(|| std::env::var_os("HOME").map(|h| PathBuf::from(h).join(".cargo")))
    {
        extra_writes.push(cargo_home.to_string_lossy().into_owned());
    }
    let caveats = build_tool_caveats_with_writes(worktree, &extra_writes);

    #[cfg(unix)]
    let (program, args): (&str, [String; 2]) = ("sh", ["-c".into(), cmd.into()]);
    #[cfg(windows)]
    let (program, args): (&str, [String; 2]) = ("cmd", ["/C".into(), cmd.into()]);

    let mut req = ExecRequest::new(
        ExecOrigin::AgentInfluenced,
        program,
        args,
        worktree,
        caveats,
    )
    // Seccomp egress floor on the crew build/test child: the `net: none` caveat
    // already Landlock-denies TCP; `DenyAll` additionally closes UDP/DNS/raw, so
    // a hostile repo's `build.rs`/test cannot exfiltrate. Pure hardening — the
    // crew build already ran with no network (deps are pre-fetched under the
    // shared CARGO_TARGET_DIR / cargo cache), so nothing legitimate regresses.
    .with_build_git_reads(worktree)
    .net_grant(NetGrant::DenyAll)
    .env("CARGO_TARGET_DIR", &target_str)
    .env("TMPDIR", worktree.to_string_lossy());
    // Real HOME so cargo/tools find ~/.cargo (read via the calibrated set,
    // written via the grant above); nothing else credential-bearing is granted.
    if let Some(home) = std::env::var_os("HOME") {
        req = req.env("HOME", home.to_string_lossy());
    }
    if let Ok(path) = std::env::var("PATH") {
        req = req.env("PATH", path);
    }
    match ConstrainedExecutor::run(&req) {
        Ok(out) => Ok((
            out.success,
            format!(
                "{}{}",
                String::from_utf8_lossy(&out.stdout),
                String::from_utf8_lossy(&out.stderr)
            ),
        )),
        Err(e) => Err(e.to_string()),
    }
}

/// Run `git <args>` in `dir`, returning trimmed stdout on success.
///
/// Confused-deputy-safe (step-7.4): a crew worktree can hold hostile repo
/// content, so this routes through `hardened_git`, which disarms the git
/// config-gadget surface (`core.fsmonitor`, hooks, `diff.external`, pager, …).
/// Diff callers additionally pass `--no-ext-diff --no-textconv`.
pub(crate) fn git(dir: &Path, args: &[&str]) -> anyhow::Result<String> {
    let out = newt_core::git_hardening::hardened_git(dir, args)?.output()?;
    if !out.status.success() {
        anyhow::bail!(
            "git {} (status={}): {}",
            args.join(" "),
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok(String::from_utf8_lossy(&out.stdout).trim().to_string())
}

/// Build a `def_sites` closure backed by `git grep` in `repo_dir` — the SAME
/// grounding crew mode's [`author_plan_to_plan`] and team mode's dispatch
/// (`newt-cli/src/crew_runner.rs`) both call, so the two modes ground
/// identically and cannot drift. `--full-name` prints paths relative to the
/// repo TOPLEVEL regardless of `repo_dir` — the same basis as worktree edit
/// paths, so a scope derived here matches what the fence partitions on even
/// when planning from a subdir. `core.quotepath=false` keeps non-ASCII paths
/// raw instead of C-quoted (a quoted path would be an unusable fence entry).
/// (#812, #840)
pub(crate) fn def_sites_grep(repo_dir: &Path) -> impl Fn(&str) -> Vec<String> + '_ {
    move |sym: &str| -> Vec<String> {
        git(
            repo_dir,
            &[
                "-c",
                "core.quotepath=false",
                "grep",
                "--full-name",
                "-nE",
                "-e",
                &definition_grep_pattern(sym),
                "--",
                "*.rs",
            ],
        )
        .map(|s| s.lines().map(str::to_string).collect())
        .unwrap_or_default()
    }
}

/// A [`Workspace`] backed by an **isolated git worktree** under
/// `<base>/.scratch/worktrees/<id>/` (self-ignored, inside cwd so writes stay
/// within the fs confinement). Edits land in the worktree, never the live tree;
/// `run_test` shells the verification command there. `Drop` removes the worktree.
pub struct WorktreeWorkspace {
    /// The original repo (where `git worktree` commands run).
    base: PathBuf,
    /// The isolated worktree path.
    worktree: PathBuf,
    /// The verification command (e.g. `just check`).
    test_cmd: String,
}

impl WorktreeWorkspace {
    /// Create a detached worktree at `<base>/.scratch/worktrees/<id>` off
    /// `base_ref` (any git commit-ish — a sha, branch, or `HEAD`). `base` must be
    /// a git repo with at least one commit.
    pub fn create(base: &Path, id: &str, base_ref: &str, test_cmd: String) -> anyhow::Result<Self> {
        // #844: crew scratch (worktrees + the shared cargo target) lives under
        // `.scratch/`, which `ensure_scratch` self-ignores (`.scratch/.gitignore`
        // = `*`) so it is never committable — no more `.newt/crew-target` swept
        // into a commit by `git add -A`.
        let scratch = newt_core::scratch::ensure_scratch(base)?;
        let worktree = scratch.join("worktrees").join(id);
        if let Some(parent) = worktree.parent() {
            std::fs::create_dir_all(parent)?;
        }
        // --detach: a free-floating checkout of the executor-supplied
        // `base_ref`, not a new branch.
        git(
            base,
            &[
                "worktree",
                "add",
                "--detach",
                worktree.to_str().unwrap_or_default(),
                base_ref,
            ],
        )?;
        Ok(Self {
            base: base.to_path_buf(),
            worktree,
            test_cmd,
        })
    }

    /// The isolated worktree path (e.g. to show touched files relative to it).
    pub fn path(&self) -> &Path {
        &self.worktree
    }

    /// The crew's changes as a unified diff (all edits incl. new files), for the
    /// overseer to review. Stages everything in the throwaway worktree, then diffs
    /// the index against HEAD. Empty string if nothing changed or git errs.
    pub fn diff(&self) -> String {
        let _ = git(&self.worktree, &["add", "-A"]);
        git(
            &self.worktree,
            &["diff", "--no-ext-diff", "--no-textconv", "--cached", "HEAD"],
        )
        .unwrap_or_default()
    }

    /// #880: run the repo-convention formatter (or the `NEWT_CREW_NORMALIZE`
    /// override) in the worktree before committing, so a model quirk (e.g. a
    /// dropped trailing newline) doesn't bounce off the gate's `fmt --check`.
    /// Best-effort: a missing/failing formatter warns and the commit still lands.
    fn normalize(&self) {
        // A polyglot repo can match several tooling packs → run every formatter.
        // Each is repo-configured, so it runs CONFINED (P4, see `run_confined_build`).
        for cmd in crew_normalize_commands(&self.worktree) {
            match run_confined_build(&self.worktree, &self.base, &cmd) {
                Ok((true, _)) => {}
                Ok((false, _)) => eprintln!(
                    "  crew normalize (`{cmd}`) failed — committing unformatted; the gate may flag it"
                ),
                Err(e) => eprintln!(
                    "  crew normalize (`{cmd}`) could not run ({e}) — committing unformatted"
                ),
            }
        }
    }

    /// **Land** the crew's work as a commit on a new `branch` in the SHARED object
    /// store, so verified work persists (a reviewable, mergeable branch the base
    /// repo sees) instead of being thrown away with the worktree. The worktree is a
    /// linked `git worktree`, so the commit + branch ref live in the common `.git`
    /// and survive `cleanup()`. Returns `(branch, short_sha)`; errs if nothing
    /// changed. (Provenance: a content-addressed commit, authored by the agent
    /// identity — the seam where agent-mesh signing later attests the crew member.)
    pub fn commit_to_branch(
        &self,
        branch: &str,
        author_name: &str,
        author_email: &str,
        message: &str,
    ) -> anyhow::Result<(String, String)> {
        // #880: normalize (format) the model's edits before staging, so a quirk
        // like a dropped trailing newline doesn't bounce off the gate.
        self.normalize();
        git(&self.worktree, &["checkout", "-q", "-b", branch])?;
        git(&self.worktree, &["add", "-A"])?;
        // `diff --cached --quiet` exits 0 when the index matches HEAD (nothing to
        // land) → don't manufacture an empty commit.
        if git(
            &self.worktree,
            &[
                "diff",
                "--no-ext-diff",
                "--no-textconv",
                "--cached",
                "--quiet",
            ],
        )
        .is_ok()
        {
            anyhow::bail!("no changes to land");
        }
        git(
            &self.worktree,
            &[
                "-c",
                &format!("user.name={author_name}"),
                "-c",
                &format!("user.email={author_email}"),
                "commit",
                "-q",
                "-m",
                message,
            ],
        )?;
        let sha = git(&self.worktree, &["rev-parse", "--short", "HEAD"])?;
        Ok((branch.to_string(), sha))
    }

    /// Merge already-landed leaf branches into one review branch. The merge is
    /// performed in another detached worktree, so the operator's checkout is
    /// never switched or edited.
    pub fn consolidate_branches(
        base: &Path,
        id: &str,
        base_ref: &str,
        branches: &[String],
        author_name: &str,
        author_email: &str,
    ) -> anyhow::Result<(String, String)> {
        if branches.is_empty() {
            anyhow::bail!("no leaf branches to consolidate");
        }
        let ws = Self::create(base, id, base_ref, "true".to_string())?;
        let branch = format!("crew/{id}");
        git(&ws.worktree, &["checkout", "-q", "-b", &branch])?;
        for leaf in branches {
            git(
                &ws.worktree,
                &[
                    "-c",
                    &format!("user.name={author_name}"),
                    "-c",
                    &format!("user.email={author_email}"),
                    "merge",
                    "--no-edit",
                    leaf,
                ],
            )?;
        }
        let sha = git(&ws.worktree, &["rev-parse", "--short", "HEAD"])?;
        Ok((branch, sha))
    }

    /// Remove the worktree (best-effort). Called by `Drop`; also callable early.
    pub fn cleanup(&self) {
        let _ = git(
            &self.base,
            &[
                "worktree",
                "remove",
                "--force",
                self.worktree.to_str().unwrap_or_default(),
            ],
        );
    }
}

impl Drop for WorktreeWorkspace {
    fn drop(&mut self) {
        self.cleanup();
    }
}

impl Workspace for WorktreeWorkspace {
    fn files(&self) -> Vec<String> {
        // Tracked files in the worktree (gitignore-respecting, no junk).
        git(&self.worktree, &["ls-files"])
            .map(|o| {
                o.lines()
                    .map(str::to_string)
                    .filter(|l| !l.is_empty())
                    .collect()
            })
            .unwrap_or_default()
    }

    fn read(&self, path: &str) -> Option<String> {
        // Confine reads to the worktree (#521): refuse absolute / `..`-escaping
        // paths — the same structural boundary as `apply` (#637). `Path::join`
        // discards the base for an absolute path, so without this guard
        // `read("/etc/passwd")` would escape the worktree and read the host.
        if !is_safe_worktree_path(path) {
            return None;
        }
        std::fs::read_to_string(self.worktree.join(path)).ok()
    }

    fn apply(&mut self, edits: &[Edit]) -> Vec<String> {
        let mut written = Vec::new();
        for e in edits {
            // STRUCTURAL worktree boundary: refuse an absolute or `..`-escaping
            // edit path. `Path::join` silently discards the base for an absolute
            // path (`worktree.join("/etc/passwd") == "/etc/passwd"`), so the
            // fs_write caveat — often `Scope::All` on the crew/plan path — is NOT
            // the boundary; this guard is. Closes the escape for `newt crew` and
            // `newt plan` alike.
            if !is_safe_worktree_path(&e.path) {
                continue;
            }
            let full = self.worktree.join(&e.path);
            if let Some(parent) = full.parent() {
                let _ = std::fs::create_dir_all(parent);
            }
            if std::fs::write(&full, &e.new_content).is_ok() {
                written.push(e.path.clone());
            }
        }
        written
    }

    fn run_test(&self) -> (bool, String) {
        // The verify command (`just check` / `cargo test` / operator `--verify`)
        // runs the repo's own code in a worktree that may hold hostile content, so
        // it runs CONFINED (P4, see `run_confined_build`): env-empty, fs fenced to
        // the worktree + crew build dirs, network denied, fail-closed off the
        // kernel fence. `CARGO_TARGET_DIR` is a per-RUN shared target under the
        // crew root so the sequential leaves build incrementally (#697).
        match run_confined_build(&self.worktree, &self.base, &self.test_cmd) {
            Ok((ok, out)) => (ok, out),
            Err(e) => (false, format!("failed to run `{}`: {e}", self.test_cmd)),
        }
    }

    fn set_test_command(&mut self, cmd: &str) {
        self.test_cmd = cmd.to_string();
    }
}

/// The per-run shared cargo target for crew leaf verifies (#697): one dir under
/// the crew root that every (sequential, ephemeral) leaf reuses, so verifies build
/// incrementally instead of each cold. `base` is absolute (the `--dir`/cwd root),
/// so the path stays absolute and a leaf's own cwd never relativizes it.
fn crew_shared_target_dir(base: &Path) -> PathBuf {
    newt_core::scratch::scratch_root(base).join("crew-target")
}

// ---------------------------------------------------------------------------
// `newt crew` command (B2)
// ---------------------------------------------------------------------------

use newt_core::Config;
use newt_scheduler::{
    run_crew, BackendPool, CrewConfig, CrewOutcome, CrewStatus, Dispatcher, LocalDispatcher,
    StaticSource,
};

/// Parsed `newt crew` arguments.
pub struct CrewArgs {
    pub task: String,
    pub crew: Option<String>,
    pub dir: Option<PathBuf>,
    pub test: Option<String>,
    pub max_attempts: Option<u32>,
    pub dry_run: bool,
}

/// Entry point for `newt crew`: resolve config, run the crew with the real
/// (local-HTTP) dispatcher, return the process exit code (0 passed, 2 needs
/// human review, errors bubble up as `Err` → exit 1).
pub async fn run_cli(args: CrewArgs) -> anyhow::Result<i32> {
    let cfg = crate::migration_notices::read(|report| Config::resolve(report))
        .map_err(|e| anyhow::anyhow!("config: {e}"))?;
    run_with(&cfg, args, &LocalDispatcher).await
}

/// Args for `newt plan <file>` — preview, or `--execute` an overseer-authored plan.
pub struct PlanArgs {
    pub file: PathBuf,
    pub dir: Option<PathBuf>,
    /// Actually dispatch the crews. The default is preview-only, so autonomous
    /// multi-crew execution always needs this second, explicit affirmation.
    pub execute: bool,
    /// `--one-shot`: this run carries the human's tacit approval, so grant each
    /// leaf the fs authority it needs (`grant_one_shot_authority`). A plain
    /// `--execute` of a reviewed file does NOT — the human granted caveats by
    /// editing the TOML.
    pub one_shot: bool,
    /// Refuse to execute a plan with more leaves than this without an explicit
    /// raise — each leaf is an autonomous crew with no per-leaf human review.
    pub max_leaves: usize,
}

/// A one-screen preview of a plan: goal, subtask count, and the **leaves** (the
/// dispatch units) with their deps. A dep that names a non-leaf (a branch, never
/// dispatched) or an absent id is flagged `[will stall]` — it can never reach
/// `Done`, so a leaf waiting on it would never run. Pure (no fs, no dispatch) so
/// the preview and the unit tests share it.
pub fn render_plan_preview(plan: &newt_core::plan::Plan) -> String {
    use std::collections::HashSet;
    let leaves = plan.leaves();
    let leaf_ids: HashSet<&str> = leaves.iter().map(|s| s.id.as_str()).collect();
    let mut out = String::new();
    if let Some(g) = &plan.goal {
        out.push_str(&format!("goal: {g}\n"));
    }
    out.push_str(&format!(
        "{} subtask(s); {} leaf/leaves to dispatch:\n",
        plan.subtasks.len(),
        leaves.len()
    ));
    for leaf in &leaves {
        let after = if leaf.deps.is_empty() {
            String::new()
        } else {
            let deps: Vec<String> = leaf
                .deps
                .iter()
                .map(|d| {
                    if leaf_ids.contains(d.as_str()) {
                        d.clone()
                    } else {
                        // not a leaf (a branch, or an absent id) → never Done.
                        format!("{d} [will stall]")
                    }
                })
                .collect();
            format!("  (after {})", deps.join(", "))
        };
        out.push_str(&format!("  • {} — {}{after}\n", leaf.id, leaf.instruction));
    }
    out
}

/// Pre-execution structural sanity (B3): the problems that would doom an
/// autonomous run — a dep naming a non-existent subtask (stalls forever), an
/// empty-instruction leaf, or a plan with no dispatchable leaves (nothing runs).
/// Empty result = structurally runnable. Pure (no fs / dispatch).
fn plan_sanity(plan: &newt_core::plan::Plan) -> Vec<String> {
    use std::collections::HashSet;
    let ids: HashSet<&str> = plan.subtasks.iter().map(|s| s.id.as_str()).collect();
    let mut problems = Vec::new();
    for s in &plan.subtasks {
        for d in &s.deps {
            if !ids.contains(d.as_str()) {
                problems.push(format!(
                    "subtask `{}` depends on `{}`, which no subtask defines — it can never run",
                    s.id, d
                ));
            }
        }
        if s.instruction.trim().is_empty() {
            problems.push(format!("subtask `{}` has an empty instruction", s.id));
        }
    }
    if plan.leaves().is_empty() && !plan.subtasks.is_empty() {
        problems.push(
            "the plan has no dispatchable leaves (every subtask is a parent) — nothing would run"
                .to_string(),
        );
    }
    problems
}

/// Path-like tokens in an instruction (e.g. `newt-cli/src/crew.rs`): a token with
/// a `/` and a short alphanumeric extension — the file a subtask CLAIMS a symbol
/// lives in. Used by [`plan_grounding_contradictions`].
fn path_tokens(text: &str) -> Vec<String> {
    text.split(|c: char| c.is_whitespace() || "()[]{}<>,;:\"'`".contains(c))
        .filter(|t| {
            t.contains('/')
                && t.rsplit_once('.').is_some_and(|(_, ext)| {
                    (1..=4).contains(&ext.len()) && ext.chars().all(|c| c.is_ascii_alphanumeric())
                })
        })
        .map(str::to_string)
        .collect()
}

/// Claim-check an authored plan against ground truth (#691) — a backstop to the
/// #687 grounding helper. When a subtask's instruction co-locates a `symbol` and
/// a `file/path` but the symbol is DEFINED elsewhere, that is a provable
/// mis-ground. `def_sites(symbol)` returns the symbol's definition grep lines
/// (`path:line:…`), injected so the contradiction logic stays pure/testable.
///
/// Conservative: refutes ONLY on positive contradiction (defined somewhere, but
/// not at the claimed path). A symbol defined nowhere (to be created) or a
/// pathless instruction yields nothing — precision over recall.
fn plan_grounding_contradictions(
    plan: &newt_core::plan::Plan,
    def_sites: impl Fn(&str) -> Vec<String>,
) -> Vec<String> {
    let mut out = Vec::new();
    for s in &plan.subtasks {
        let paths = path_tokens(&s.instruction);
        if paths.is_empty() {
            continue;
        }
        for sym in grep_terms(&s.instruction) {
            let sites = def_sites(&sym);
            if sites.is_empty() {
                continue; // defined nowhere — maybe to be created; never refute
            }
            let def_files: Vec<String> = sites
                .iter()
                .filter_map(|l| l.split(':').next().map(str::to_string))
                .collect();
            for claimed in &paths {
                let agrees = def_files
                    .iter()
                    .any(|f| f == claimed || f.ends_with(claimed) || claimed.ends_with(f.as_str()));
                if !agrees {
                    out.push(format!(
                        "subtask `{}`: `{sym}` is defined at {}, not `{claimed}` — make the edit there.",
                        s.id,
                        def_files.join(", "),
                    ));
                }
            }
        }
    }
    out.sort();
    out.dedup();
    out
}

/// #812/#840: stamp every subtask's `context` as ground truth ∪ declaration,
/// via the shared [`newt_core::scope_grounding::ground_scope`] — the same
/// implementation team mode's `run_team` calls, so the two paths cannot drift.
fn ground_subtask_scopes(
    plan: &mut newt_core::plan::Plan,
    def_sites: &impl Fn(&str) -> Vec<String>,
) {
    for s in &mut plan.subtasks {
        s.context = ground_scope(&s.instruction, &s.context, def_sites);
    }
}

/// `newt plan <file>` — PREVIEW an overseer-authored plan, or (`--execute`)
/// dispatch it leaf-by-leaf via a crew (each leaf in its own worktree, through the
/// same `LocalCrewRunner` the in-session `crew` tool uses).
///
/// Preview is the **default** because `--execute` runs an autonomous DAG of crews
/// with no per-leaf human review (the per-leaf `verify` is the only gate); it is
/// bounded by `--max-leaves`. The run-log (statuses + results) is written to a
/// sibling `<file>.run.toml` — the authored source file is **never modified**.
/// Exit 0 = preview / complete, 1 = incomplete (failed or stalled).
pub async fn run_plan_cli(args: PlanArgs) -> anyhow::Result<i32> {
    let toml = std::fs::read_to_string(&args.file)
        .map_err(|e| anyhow::anyhow!("read {}: {e}", args.file.display()))?;
    let mut plan = newt_core::plan::Plan::from_toml_str(&toml)
        .map_err(|e| anyhow::anyhow!("parse plan {}: {e}", args.file.display()))?;
    println!("plan: {}", args.file.display());
    print!("{}", render_plan_preview(&plan));
    if !args.execute {
        println!("\n(preview only — re-run with --execute to dispatch one crew per leaf)");
        return Ok(0);
    }
    // `--one-shot` on a FILE is tacit approval to grant the leaves their fs
    // authority; a plain `--execute` respects the caveats the human reviewed in.
    if args.one_shot {
        grant_one_shot_authority(&mut plan);
    }
    // Run-log lands in a SIBLING artifact — never modify the authored source.
    let log_path = args.file.with_extension("run.toml");
    execute_plan(&mut plan, args.dir, args.max_leaves, Some(log_path), None).await
}

/// `--one-shot`'s tacit approval, made real: grant every subtask the filesystem
/// authority an autonomous run needs — `fs_read` + `fs_write` (worktree-bounded
/// by the runner's apply-path guard). `exec` + `net` stay DENIED: verification
/// runs via the runner's TRUSTED inferred command (`just check` / `cargo test`),
/// not the model-authored `verify` shell, and the crew needs no network. Without
/// this, every authored leaf is default-deny and can edit nothing.
fn grant_one_shot_authority(plan: &mut newt_core::plan::Plan) {
    use newt_core::role_profile::ScopeSpec;
    for s in &mut plan.subtasks {
        s.caveat_policy.fs_read = ScopeSpec::default(); // "all"
        s.caveat_policy.fs_write = ScopeSpec::default(); // "all"
    }
    println!(
        "--one-shot: granted each leaf fs_read + fs_write (worktree-bounded); \
         exec + net stay denied (verify runs via the runner's trusted command)."
    );
}

/// Re-ground a failed leaf from its build error (#692): parse the unresolved
/// symbol rustc names, grep its DEFINITION (#687's pattern), and — if the
/// instruction doesn't already target that file — append a grounding correction
/// so the retried leaf edits the real seam.
struct DefGroundReground {
    dir: std::path::PathBuf,
}
impl newt_core::agentic::Reground for DefGroundReground {
    fn reground(&self, error: &str, instruction: &str) -> Option<String> {
        let sym = unresolved_symbol(error)?;
        let sites = git(
            &self.dir,
            &[
                "grep",
                "-nE",
                "-e",
                &definition_grep_pattern(&sym),
                "--",
                "*.rs",
            ],
        )
        .ok()?;
        let file = sites.lines().next()?.split(':').next()?.to_string();
        if file.is_empty() || instruction.contains(&file) {
            return None;
        }
        Some(format!(
            "{instruction}\n\nGROUNDING: `{sym}` is defined at {file} — make the edit there, do not invent paths."
        ))
    }
}

/// The unresolved symbol named in a rustc error, if any (#692) — `cannot find
/// function/value/type` `X`, `no method named` `X`, etc.
fn unresolved_symbol(error: &str) -> Option<String> {
    const MARKERS: &[&str] = &[
        "cannot find function `",
        "cannot find value `",
        "cannot find type `",
        "cannot find macro `",
        "cannot find struct, variant or union type `",
        "no method named `",
        "no function or associated item named `",
        "use of undeclared `",
    ];
    for m in MARKERS {
        if let Some(i) = error.find(m) {
            let rest = &error[i + m.len()..];
            if let Some(j) = rest.find('`') {
                let sym = &rest[..j];
                if !sym.is_empty() {
                    return Some(sym.to_string());
                }
            }
        }
    }
    None
}

/// Execute a parsed/authored plan autonomously: enforce the `--max-leaves`
/// autonomy bound, then drive `run_plan` via a `LocalCrewRunner`
/// (`Presence::Prompt` — the human's `--execute`/`--one-shot` gesture). Prints
/// per-leaf progress; writes a run-log to `log_path` when work actually ran.
/// Returns 0 = plan complete, 1 = incomplete. The source is never modified.
pub async fn execute_plan(
    plan: &mut newt_core::plan::Plan,
    dir: Option<PathBuf>,
    max_leaves: usize,
    log_path: Option<PathBuf>,
    locked_verify: Option<String>,
) -> anyhow::Result<i32> {
    // B3: refuse a structurally-doomed plan BEFORE spending crew runs on it — a
    // dep on a non-existent subtask, an empty-instruction leaf, or no
    // dispatchable leaves all guarantee a stall / wasted dispatch.
    let problems = plan_sanity(plan);
    if !problems.is_empty() {
        eprintln!("✗ plan is not structurally runnable — refusing to dispatch:");
        for p in &problems {
            eprintln!("  - {p}");
        }
        return Ok(2);
    }
    // Autonomy bound: refuse an oversized autonomous fan-out unless the human
    // explicitly raises the cap (each leaf runs with no per-leaf review).
    let leaf_count = plan.leaves().len();
    if leaf_count > max_leaves {
        return Err(anyhow::anyhow!(
            "plan has {leaf_count} leaves (> --max-leaves {max_leaves}); each is an autonomous \
             crew with no per-leaf review. Re-run with `--max-leaves {leaf_count}` to confirm you \
             intend to run them all."
        ));
    }
    let cfg = crate::migration_notices::read(|report| Config::resolve(report))
        .map_err(|e| anyhow::anyhow!("config: {e}"))?;
    let dir = dir.unwrap_or_else(|| std::env::current_dir().unwrap_or_default());
    println!(
        "executing {leaf_count} leaf/leaves autonomously (one crew each; per-leaf verify gates each)…"
    );
    // Honest, non-top session caveats (mirrors `newt crew`): fs_write=All — the
    // real boundary is the per-leaf worktree + the apply-path guard
    // (`is_safe_worktree_path`), NOT this caveat — exec/net locked, so a
    // plan-authored `verify` needing exec is dropped fail-closed (#634). The
    // `--execute`/`--one-shot` gesture is modelled as `Presence::Prompt` (the
    // BOOT attest ceremony is #472).
    let caveats = newt_acp_worker::worker_session_caveats(None);
    let reground = DefGroundReground { dir: dir.clone() };
    let runner =
        crate::crew_runner::LocalCrewRunner::new(cfg, dir, newt_core::agentic::Presence::Prompt)
            .with_locked_verify(locked_verify);
    let run = newt_core::agentic::run_plan_with_reground(plan, &caveats, &runner, &reground).await;
    for id in &run.dispatched {
        if let Some(s) = plan.subtask(id) {
            println!("  [{:?}] {}", s.status, id);
        }
    }
    if let Some(e) = &run.failed {
        println!("✗ stopped at a failed leaf: {e}");
    }
    if !run.remaining.is_empty() {
        println!("remaining (blocked/stalled): {}", run.remaining.join(", "));
    }
    if let Some(consolidated) = &run.consolidated {
        println!("{consolidated}");
    }
    println!(
        "{}",
        if run.complete {
            "✓ plan complete"
        } else {
            "plan incomplete"
        }
    );
    // Write the run-log only when work actually ran.
    if !run.dispatched.is_empty() {
        if let Some(log) = log_path {
            std::fs::write(&log, plan.to_toml_string()?)
                .map_err(|e| anyhow::anyhow!("write run-log {}: {e}", log.display()))?;
            println!("run-log → {}", log.display());
        }
    }
    Ok(i32::from(!run.complete))
}

// ── Plan authoring (the overseer's decompose step) ────────────────────────────

/// The overseer's decompose prompt: turn a goal into a dependency-ordered
/// `plan::Plan`. It asks for the WORK only — never authority — so an authored
/// plan carries default-deny caveats the human grants by editing the TOML.
const PLAN_AUTHOR_SYSTEM: &str = "You are a planning lead. Decompose the GOAL into a \
    dependency-ordered plan of the FEWEST engineering subtasks that accomplish it — each \
    subtask MUST change code (produce a file edit). Reply with ONLY JSON: \
    {\"goal\":\"<the goal>\",\"subtasks\":[{\"id\":\"<short-kebab-id>\",\
    \"instruction\":\"<imperative code-changing step>\",\
    \"deps\":[\"<id of a step that must finish first>\"],\
    \"files\":[\"<relative path of a REAL existing file this step edits; omit if unknown>\"],\
    \"verify\":\"<shell command that exits 0 once THIS step is done; omit if none>\"}]}. \
    A small, single-file change is ONE subtask. Do NOT create separate \
    inspect/understand/explore/locate/verify/test/run-tests subtasks — the harness reads \
    the repo for you and automatically verifies EVERY subtask after it runs (put a step's \
    own check in its `verify` FIELD, never as a standalone subtask). That prohibition covers \
    REPHRASINGS of the same non-actionable pattern too — \"add edge-case tests\", \"clean up \
    the code\", \"update the comments\", \"validate the fix\" are inspect/verify/test steps \
    wearing different words; fold any such follow-up into the ONE subtask that changes the \
    behavior, or drop it. Example: the goal \"Fix `humanize_duration` so 90 seconds returns \
    '1m 30s' and the test passes\" is ONE subtask, not six — \
    {\"goal\":\"Fix humanize_duration so 90 seconds returns '1m 30s'\",\
    \"subtasks\":[{\"id\":\"fix-duration-format\",\"instruction\":\"Fix the minutes/seconds \
    formatting in humanize_duration so it is correct for every input: zero, under a minute, \
    and a mixed minutes+seconds value\",\"deps\":[],\
    \"verify\":\"cargo test humanize_duration\"}]} — NOT \
    fix-then-adjust-then-format-then-test-edge-cases-then-clean-up-then-update-comments; a \
    failed later subtask strands the earlier ones half-fixed. Use `deps` for ordering (a \
    step lists the ids it waits on). Ids: short, stable, unique. Do NOT grant permissions or \
    describe authority — only the work.";

/// The first balanced `{…}` span in `s` (string-aware), so a model reply wrapped
/// in ```json fences or prose still parses. `None` if there is no balanced object.
fn extract_json_object(s: &str) -> Option<&str> {
    let start = s.find('{')?;
    let bytes = s.as_bytes();
    let (mut depth, mut in_str, mut esc) = (0i32, false, false);
    for i in start..bytes.len() {
        let c = bytes[i];
        if in_str {
            match c {
                b'\\' if !esc => esc = true,
                b'"' if !esc => in_str = false,
                _ => esc = false,
            }
        } else if c == b'"' {
            in_str = true;
        } else if c == b'{' {
            depth += 1;
        } else if c == b'}' {
            depth -= 1;
            if depth == 0 {
                return Some(&s[start..=i]);
            }
        }
    }
    None
}

/// Parse a model-authored plan (tolerating fences/prose) into a `plan::Plan`.
/// Every subtask gets **default-deny** caveats — the model proposes the WORK, never
/// the authority; the human grants it by editing the TOML. `None` when no parseable
/// `{…}` object with a non-empty `subtasks` list (each with `id` + `instruction`)
/// is found. Parsed with `serde_json::Value` (newt-cli has no `serde` derive dep).
fn parse_authored_plan(raw: &str) -> Option<newt_core::plan::Plan> {
    use newt_core::plan::{Aggregation, CaveatPolicy, NodeKind, Plan, Subtask, SubtaskStatus};
    use std::collections::HashSet;
    let v: serde_json::Value = serde_json::from_str(extract_json_object(raw)?).ok()?;
    let arr = v.get("subtasks")?.as_array()?;
    if arr.is_empty() {
        return None;
    }
    let mut seen = HashSet::new();
    let mut subtasks = Vec::with_capacity(arr.len());
    for s in arr {
        // Reject empty / whitespace-only ids+instructions, and DUPLICATE ids.
        // ids must be unique: `Plan::mark` updates the first match while
        // `next_ready_leaf` finds the next unmarked one, so duplicate ids would
        // desync the execute-time cursor into an unbounded re-dispatch loop.
        let id = s.get("id")?.as_str()?.trim().to_string();
        let instruction = s.get("instruction")?.as_str()?.trim().to_string();
        if id.is_empty() || instruction.is_empty() || !seen.insert(id.clone()) {
            return None;
        }
        let deps = s
            .get("deps")
            .and_then(|d| d.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|x| x.as_str().map(str::to_string))
                    .collect()
            })
            .unwrap_or_default();
        // #812: the model's self-declared files are read into `context` as a
        // FALLBACK only — author_plan_to_plan overrides them with the
        // harness's own def-site derivation wherever grounding finds the real
        // seam (the model is untrusted to widen, and unreliable at declaring).
        let context = s
            .get("files")
            .and_then(|f| f.as_array())
            .map(|a| {
                a.iter()
                    .filter_map(|x| x.as_str().map(str::trim).map(str::to_string))
                    .filter(|p| !p.is_empty())
                    .collect()
            })
            .unwrap_or_default();
        subtasks.push(Subtask {
            id,
            instruction,
            deps,
            parallel_ok: false,
            context,
            verify: s.get("verify").and_then(|x| x.as_str()).map(str::to_string),
            status: SubtaskStatus::Pending,
            result: None,
            parent: None,
            kind: NodeKind::Task,
            conversation_id: None,
            artifact_ref: None,
            caveat_policy: CaveatPolicy::default(), // default-DENY: model proposes work, not authority
        });
    }
    Some(Plan {
        goal: v.get("goal").and_then(|x| x.as_str()).map(str::to_string),
        aggregation: Aggregation::default(),
        subtasks,
    })
}

/// The kind of non-actionable subtask [`ACTION_MARKERS`] identifies.
#[derive(Clone, Copy)]
enum MarkerKind {
    /// A read-only "understand the code" verb. The harness already reads the repo
    /// for the planner, so an inspect leaf produces no diff — prune it wherever it
    /// sits in the plan.
    Inspect,
    /// A verification verb. The harness auto-verifies EVERY subtask, so a *terminal*
    /// gate leaf (no subtask depends on it) lands no diff — prune it ONLY when
    /// terminal; a mid-plan gate a real leaf depends on is kept.
    Gate,
}

/// The anti-pattern lexicon for the decompose prune (#801): `(leading verb, kind)`
/// pairs that mark a subtask as non-actionable — it produces no diff and so feeds
/// the crew executor's "nothing-to-land" fail-stop. Pure DATA (the three Cs): a new
/// anti-pattern is a one-line edit here (the flat sibling of
/// `agentic::routing::READ_ROUTES`). Polysemous verbs (build / run / check) are
/// deliberately EXCLUDED — "Build the parser", "run the migration", "check the
/// bounds" are real diff-producing work.
const ACTION_MARKERS: &[(&str, MarkerKind)] = &[
    ("inspect", MarkerKind::Inspect),
    ("examine", MarkerKind::Inspect),
    ("explore", MarkerKind::Inspect),
    ("investigate", MarkerKind::Inspect),
    ("understand", MarkerKind::Inspect),
    ("locate", MarkerKind::Inspect),
    ("identify", MarkerKind::Inspect),
    ("review", MarkerKind::Inspect),
    ("analyze", MarkerKind::Inspect),
    ("verify", MarkerKind::Gate),
    ("validate", MarkerKind::Gate),
    ("test", MarkerKind::Gate),
    ("confirm", MarkerKind::Gate),
    ("ensure", MarkerKind::Gate),
];

/// The EFFECTIVE prune lexicon: the compiled [`ACTION_MARKERS`] composed with
/// the operator's droppable `[plan.prune]` override (#819) — removals first,
/// then additions (lowercased, deduped, empties dropped). Pure; `None` config
/// yields exactly the compiled defaults.
fn effective_markers(cfg: Option<&newt_core::PlanPruneConfig>) -> Vec<(String, MarkerKind)> {
    let mut out: Vec<(String, MarkerKind)> = ACTION_MARKERS
        .iter()
        .map(|(v, k)| ((*v).to_string(), *k))
        .collect();
    if let Some(c) = cfg {
        let removed: Vec<String> = c.remove.iter().map(|w| w.to_ascii_lowercase()).collect();
        out.retain(|(v, _)| !removed.contains(v));
        let mut add = |words: &[String], kind: MarkerKind| {
            for w in words {
                let w = w.trim().to_ascii_lowercase();
                if !w.is_empty() && !removed.contains(&w) && !out.iter().any(|(v, _)| *v == w) {
                    out.push((w, kind));
                }
            }
        };
        add(&c.add_inspect, MarkerKind::Inspect);
        add(&c.add_gate, MarkerKind::Gate);
    }
    out
}

/// Classify a subtask by the LEADING verb of its instruction against a
/// lexicon. Returns `None` for a real (diff-producing) subtask — "Extract
/// helper", "Add validation", "Rename fn" survive because their leading verb
/// is not a marker. Only the leading token is consulted, so a marker word
/// later in the sentence never trips a prune.
fn marker_kind_in(markers: &[(String, MarkerKind)], instruction: &str) -> Option<MarkerKind> {
    let head = instruction
        .split(|c: char| !c.is_alphanumeric())
        .find(|w| !w.is_empty())
        .unwrap_or("")
        .to_ascii_lowercase();
    markers
        .iter()
        .find(|(verb, _)| *verb == head)
        .map(|(_, kind)| *kind)
}

/// Deterministically prune non-actionable subtasks from a freshly-authored plan
/// (#801). A read-only `inspect` leaf and a *terminal* `gate` leaf both produce no
/// diff, so under the crew executor they trip the "nothing-to-land" fail-stop
/// (`newt-core/src/agentic/plan_exec.rs`) — the #1 observed crew-mode failure in the
/// DGX planner-strength sweep. This is the deterministic backstop to
/// [`PLAN_AUTHOR_SYSTEM`]'s prose, which a weaker planner ignores. Pure and
/// in-memory: it runs BEFORE any authority grant, only removes subtasks and rewires
/// their `deps`, and never touches `caveat_policy` — so it cannot widen authority
/// nor make a gate easier to pass (a surviving leaf still faces the full per-leaf
/// verify). It removes only a false-NEGATIVE (no-diff) failure source.
fn prune_non_actionable_subtasks_in(
    plan: &mut newt_core::plan::Plan,
    markers: &[(String, MarkerKind)],
) {
    use std::collections::HashSet;
    // Empty-guard: if NOTHING in the plan is actionable — every subtask is a marker
    // — leave it entirely untouched. Pruning would empty (or half-prune) it; let
    // `plan_sanity` / re-author handle a degenerate plan rather than fabricate one.
    if !plan
        .subtasks
        .iter()
        .any(|s| marker_kind_in(markers, &s.instruction).is_none())
    {
        return;
    }
    // Fixpoint: remove one prunable subtask at a time (Inspect anywhere; Gate only
    // when terminal), splicing the removed subtask's own deps into every survivor
    // that named it — a dangling dep would STALL the survivor, since
    // `Plan::next_ready_leaf` treats an absent dep as never-`Done`.
    loop {
        let depended: HashSet<String> = plan
            .subtasks
            .iter()
            .flat_map(|s| s.deps.iter().cloned())
            .collect();
        let victim =
            plan.subtasks
                .iter()
                .position(|s| match marker_kind_in(markers, &s.instruction) {
                    Some(MarkerKind::Inspect) => true,
                    Some(MarkerKind::Gate) => !depended.contains(&s.id),
                    None => false,
                });
        let Some(idx) = victim else { return };
        let removed = plan.subtasks.remove(idx);
        for s in &mut plan.subtasks {
            if let Some(pos) = s.deps.iter().position(|d| *d == removed.id) {
                s.deps.remove(pos);
                for vd in &removed.deps {
                    if !s.deps.contains(vd) {
                        s.deps.push(vd.clone());
                    }
                }
            }
        }
    }
}

/// Decompose `goal` into a `plan::Plan` by asking `model` (the overseer seat).
/// The model proposes the work; caveats stay default-deny.
pub async fn author_plan(
    pool: &newt_scheduler::BackendPool,
    dispatcher: &dyn Dispatcher,
    model: &str,
    goal: &str,
    max_subtasks: usize,
    prune: Option<&newt_core::PlanPruneConfig>,
) -> anyhow::Result<newt_core::plan::Plan> {
    let req = newt_scheduler::ChatRequest::new()
        .system(PLAN_AUTHOR_SYSTEM)
        .user(format!("GOAL:\n{goal}\n\nAt most {max_subtasks} subtasks."));
    let reply = pool
        .run_role(dispatcher, newt_core::Tier::Complex, model, req)
        .await
        .ok_or_else(|| {
            anyhow::anyhow!("no live model reachable to author the plan (model {model})")
        })?;
    parse_authored_plan(&reply.result.content)
        .map(|mut plan| {
            // #801: deterministically drop the non-actionable (inspect /
            // terminal-gate) leaves a weaker planner pads the plan with, before any
            // authority is granted or the plan is dispatched. The lexicon composes
            // with the operator's droppable [plan.prune] override (#819), which can
            // also disable the prune entirely.
            if !prune.map(|c| c.disabled).unwrap_or(false) {
                prune_non_actionable_subtasks_in(&mut plan, &effective_markers(prune));
            }
            plan
        })
        .ok_or_else(|| {
            anyhow::anyhow!(
                "the model did not return a parseable JSON plan. Raw reply:\n{}",
                reply.result.content
            )
        })
}

/// `newt plan --goal "<g>"` — author a plan from a goal, write it (or print it),
/// and show the preview. Does NOT execute: the human reviews/edits, then runs
/// `newt plan <file> --execute`.
/// Author a plan from a goal (the overseer's decompose step) and RETURN it.
/// Shared by [`author_plan_cli`] (which writes/prints it) and
/// [`one_shot_goal_cli`] (which executes it).
pub async fn author_plan_to_plan(
    goal: &str,
    max_subtasks: usize,
    repo_dir: &Path,
) -> anyhow::Result<newt_core::plan::Plan> {
    let cfg = crate::migration_notices::read(|report| Config::resolve(report))
        .map_err(|e| anyhow::anyhow!("config: {e}"))?;
    let model = cfg
        .backends
        .first()
        .and_then(|b| b.effective_model().map(str::to_string))
        .ok_or_else(|| {
            anyhow::anyhow!(
                "no backend with a model configured to author a plan                  (add a [[backends]] with `model`, or run against an adopted backend)"
            )
        })?;
    let pool = BackendPool::from_source(&StaticSource::from_configs(cfg.backends.iter()));
    let mut effective_goal = goal.to_string();
    // Phase 1a (#646): READ any GitHub issue/PR the goal references, via `gh` (a
    // harness-available tool), so the planner decomposes the ACTUAL document
    // instead of a bare link it cannot fetch. Best-effort — a missing /
    // unauthenticated `gh` just leaves the goal text alone.
    let docs = fetch_referenced_docs(goal);
    if !docs.is_empty() {
        println!(
            "read referenced GitHub doc(s) via gh ({} chars)",
            docs.len()
        );
        effective_goal.push_str(&format!(
            "\n\nThe referenced document(s) below are the TASK to implement — \
             decompose THESE into concrete engineering subtasks:{docs}"
        ));
    }
    // Phase 1b: READ the TARGET REPO's structure (language, build, layout) so the
    // planner authors subtasks that fit THIS codebase — e.g. Rust crates, not a
    // hallucinated Python stack. Best-effort: a non-repo dir adds nothing.
    let repo = fetch_repo_context(repo_dir);
    if !repo.is_empty() {
        println!("read target-repo context ({} chars)", repo.len());
        effective_goal.push_str(&repo);
    }
    // Phase 1c (B2): GREP the repo for the commands/symbols the TASK references
    // (slash-commands, backtick-quoted identifiers), so the planner targets REAL
    // files instead of guessing paths. Greps the goal+issue text, not the repo
    // layout. Best-effort: a non-repo dir / no matches adds nothing.
    let hits = fetch_code_grep_hits(&format!("{goal}{docs}"), repo_dir);
    if !hits.is_empty() {
        println!(
            "found relevant code location(s) via grep ({} chars)",
            hits.len()
        );
        effective_goal.push_str(&hits);
    }
    // Phase 2: decompose the (doc + repo-augmented) goal into a plan.
    println!("authoring a plan for: {goal}  (model {model})…");
    let prune_cfg = cfg.plan.as_ref().map(|p| p.prune.clone());
    let mut plan = author_plan(
        &pool,
        &LocalDispatcher,
        &model,
        &effective_goal,
        max_subtasks,
        prune_cfg.as_ref(),
    )
    .await?;
    // Phase 2b (#691): claim-check the plan against ground truth. A subtask that
    // targets a symbol's WRONG file (it's defined elsewhere) is a provable
    // mis-ground — re-author ONCE with the citation. A backstop to the #687
    // grounding helper; refutes only on positive contradiction, so a clean plan
    // is never re-authored.
    let def_sites = def_sites_grep(repo_dir);
    let contradictions = plan_grounding_contradictions(&plan, &def_sites);
    if contradictions.is_empty() {
        // #812: stamp each leaf's file scope from the SAME ground truth the
        // claim-check uses, so the worker fence bites deterministically.
        ground_subtask_scopes(&mut plan, &def_sites);
        return Ok(plan);
    }
    println!(
        "plan grounding contradictions — re-authoring with corrections:\n  {}",
        contradictions.join("\n  ")
    );
    effective_goal.push_str(&format!(
        "\n\nGROUNDING CORRECTIONS — the prior plan targeted the wrong file(s); \
         fix these and do NOT invent paths:\n{}",
        contradictions.join("\n")
    ));
    let mut plan = author_plan(
        &pool,
        &LocalDispatcher,
        &model,
        &effective_goal,
        max_subtasks,
        prune_cfg.as_ref(),
    )
    .await?;
    ground_subtask_scopes(&mut plan, &def_sites);
    Ok(plan)
}

/// Bounded structural context about the target repo at `dir` — its language /
/// build system and top-level layout — so the planner authors subtasks that fit
/// THIS codebase (e.g. Rust crates `newt-cli`/`newt-core`, not a hallucinated
/// FastAPI/Django stack). Best-effort: a non-repo / unreadable `dir` yields "".
fn fetch_repo_context(dir: &Path) -> String {
    let mut facts: Vec<String> = Vec::new();
    if dir.join("Cargo.toml").exists() {
        facts.push("Language/build: Rust (a cargo workspace).".into());
    } else if dir.join("package.json").exists() {
        facts.push("Language/build: JavaScript/TypeScript (npm).".into());
    } else if dir.join("pyproject.toml").exists() || dir.join("setup.py").exists() {
        facts.push("Language/build: Python.".into());
    } else if dir.join("go.mod").exists() {
        facts.push("Language/build: Go.".into());
    }
    if let Some(cmd) = infer_test_command(dir) {
        facts.push(format!("Verify/build command: `{cmd}`."));
    }
    // Top-level tracked entries (the crate / dir layout), bounded.
    if let Ok(top) = git(dir, &["ls-tree", "--name-only", "HEAD"]) {
        let entries: Vec<&str> = top.lines().filter(|l| !l.is_empty()).take(40).collect();
        if !entries.is_empty() {
            facts.push(format!("Top-level entries: {}.", entries.join(", ")));
        }
    }
    if facts.is_empty() {
        return String::new();
    }
    format!(
        "\n\nTarget repository context — author subtasks that fit THIS codebase \
         (use its real language, build command, and directories; do NOT invent a \
         different stack):\n- {}",
        facts.join("\n- ")
    )
}

/// GitHub issue/PR references in `goal`: `(owner, repo, kind, number)` for each
/// `github.com/<owner>/<repo>/(issues|pull)/<n>` URL. Tolerant of trailing
/// punctuation on the number and of the URL sitting in surrounding prose.
fn github_refs(goal: &str) -> Vec<(String, String, String, String)> {
    let mut refs = Vec::new();
    for token in goal.split_whitespace() {
        let Some(idx) = token.find("github.com/") else {
            continue;
        };
        let parts: Vec<&str> = token[idx + "github.com/".len()..].split('/').collect();
        if parts.len() < 4 {
            continue;
        }
        let num: String = parts[3]
            .chars()
            .take_while(|c| c.is_ascii_digit())
            .collect();
        if !num.is_empty() && (parts[2] == "issues" || parts[2] == "pull") {
            refs.push((
                parts[0].to_string(),
                parts[1].to_string(),
                parts[2].to_string(),
                num,
            ));
        }
    }
    refs
}

/// One grep block for a single task term: the DEFINITION sites (the real seam)
/// and the other mention sites, kept apart so [`format_grounding_hits`] can rank
/// definitions FIRST and never truncate them. The #687 bug was `git grep |
/// take(3)`: because `git grep` is path-alphabetical, an earlier-sorting
/// same-named decoy (e.g. `newt-cli/crew.rs` mentions of `help_lines`) buried
/// the real `fn help_lines()` in `newt-tui`, so the planner was told to edit a
/// file where the symbol does not exist.
struct GroundingBlock {
    term: String,
    defs: Vec<String>,
    mentions: Vec<String>,
}

fn truncate_grep_line(line: &str) -> String {
    line.chars().take(160).collect()
}

/// Render grounding blocks for the plan-author prompt: DEFINITION sites first
/// (marked `[def]`, never truncated, per-term cap), so the real seam is always
/// surfaced, then a couple of non-definition mentions for context, under an
/// overall budget. Pure — unit-testable without git.
fn format_grounding_hits(blocks: &[GroundingBlock]) -> String {
    let mut out = String::new();
    let mut total = 0usize;
    for b in blocks {
        if total >= 25 {
            break;
        }
        // Definitions FIRST and untruncated — the real seam can never be buried
        // by an alphabetically-earlier decoy (the #687 fix).
        for d in b.defs.iter().take(6) {
            if total >= 25 {
                break;
            }
            out.push_str(&format!("  {} [def]: {}\n", b.term, truncate_grep_line(d)));
            total += 1;
        }
        // Then a couple of mentions not already shown as definitions.
        let mut shown_mentions = 0usize;
        for m in &b.mentions {
            if total >= 25 || shown_mentions >= 2 {
                break;
            }
            if b.defs.contains(m) {
                continue;
            }
            out.push_str(&format!("  {}: {}\n", b.term, truncate_grep_line(m)));
            total += 1;
            shown_mentions += 1;
        }
    }
    if out.is_empty() {
        return String::new();
    }
    format!(
        "\n\nRelevant existing code — task terms already appear here. A `[def]` \
         line is the DEFINITION site (the real seam to edit); implement AT these \
         sites, do NOT invent file paths:\n{out}"
    )
}

/// Grep the target repo for the task's high-signal terms and return where they
/// already appear (bounded), so the planner implements AT real sites instead of
/// guessing paths. Definition sites are surfaced first and never truncated
/// (#687). Best-effort: a non-repo `dir` / no matches yields "".
fn fetch_code_grep_hits(task: &str, dir: &Path) -> String {
    let lines = |res: anyhow::Result<String>| -> Vec<String> {
        res.map(|s| s.lines().map(str::to_string).collect())
            .unwrap_or_default()
    };
    let mut blocks = Vec::new();
    for term in grep_terms(task) {
        let defs = lines(git(
            dir,
            &[
                "grep",
                "-nE",
                "-e",
                &definition_grep_pattern(&term),
                "--",
                "*.rs",
            ],
        ));
        let mentions = lines(git(dir, &["grep", "-n", "-F", "-e", &term, "--", "*.rs"]));
        if !defs.is_empty() || !mentions.is_empty() {
            blocks.push(GroundingBlock {
                term,
                defs,
                mentions,
            });
        }
    }
    format_grounding_hits(&blocks)
}

/// Read each GitHub issue/PR referenced in `goal` with `gh … view --json
/// title,body` and return their text. Best-effort: a `gh` that is missing,
/// unauthenticated, or errors contributes nothing (and prints a note), so
/// authoring still proceeds from the goal text alone.
fn fetch_referenced_docs(goal: &str) -> String {
    let mut out = String::new();
    for (owner, repo, kind, num) in github_refs(goal) {
        let sub = if kind == "pull" { "pr" } else { "issue" };
        let slug = format!("{owner}/{repo}");
        match std::process::Command::new("gh")
            .args([sub, "view", &num, "--repo", &slug, "--json", "title,body"])
            .output()
        {
            Ok(o) if o.status.success() => {
                if let Ok(v) = serde_json::from_slice::<serde_json::Value>(&o.stdout) {
                    let title = v["title"].as_str().unwrap_or("");
                    let body = v["body"].as_str().unwrap_or("");
                    out.push_str(&format!(
                        "\n\n--- {sub} {slug}#{num}: {title} ---\n{body}\n"
                    ));
                }
            }
            Ok(o) => eprintln!(
                "note: `gh {sub} view {num}` failed ({}) — planning from the goal text alone",
                String::from_utf8_lossy(&o.stderr).trim()
            ),
            Err(e) => {
                eprintln!("note: could not run `gh` ({e}) — planning from the goal text alone");
            }
        }
    }
    out
}

/// `newt plan --goal "<text>" --one-shot`: author a plan from the goal AND
/// execute it autonomously in one gesture — the headless autonomous drive (e.g.
/// the #548 evaluator). The `--one-shot` flag is the approval, like `--execute`.
pub async fn one_shot_goal_cli(
    goal: &str,
    dir: Option<PathBuf>,
    max_leaves: usize,
    locked_verify: Option<String>,
) -> anyhow::Result<i32> {
    // The repo we're about to modify IS the planning context (--dir, else cwd).
    let repo_dir = dir
        .clone()
        .unwrap_or_else(|| std::env::current_dir().unwrap_or_default());
    let mut plan = author_plan_to_plan(goal, max_leaves, &repo_dir).await?;
    print!("{}", render_plan_preview(&plan));
    grant_one_shot_authority(&mut plan);
    println!("\n--one-shot: executing the authored plan autonomously…");
    // No source file to shadow — write the run-log into the cwd.
    execute_plan(
        &mut plan,
        dir,
        max_leaves,
        Some(PathBuf::from("plan.run.toml")),
        locked_verify,
    )
    .await
}

pub async fn author_plan_cli(
    goal: String,
    output: Option<PathBuf>,
    max_subtasks: usize,
    dir: Option<PathBuf>,
) -> anyhow::Result<i32> {
    let repo_dir = dir.unwrap_or_else(|| std::env::current_dir().unwrap_or_default());
    let plan = author_plan_to_plan(&goal, max_subtasks, &repo_dir).await?;
    let toml = plan
        .to_toml_string()
        .map_err(|e| anyhow::anyhow!("serialize plan: {e}"))?;
    print!("{}", render_plan_preview(&plan));
    match &output {
        Some(path) => {
            // Don't clobber an existing plan the human may have edited (granted
            // caveats into). Authoring writes a fresh draft only.
            if path.exists() {
                anyhow::bail!(
                    "{} already exists — choose another -o path or remove it first \
                     (authoring won't overwrite a plan you may have edited)",
                    path.display()
                );
            }
            std::fs::write(path, &toml)
                .map_err(|e| anyhow::anyhow!("write {}: {e}", path.display()))?;
            println!(
                "\nwrote plan → {} — review/edit (grant caveats where needed), then \
                 `newt plan {} --execute`",
                path.display(),
                path.display()
            );
        }
        None => println!(
            "\n--- plan.toml (write to a file, then `newt plan <file> --execute`) ---\n{toml}"
        ),
    }
    Ok(0)
}

/// The testable core: same as [`run_cli`] but with the inference `Dispatcher`
/// and resolved `cfg` injected (tests pass an in-memory config + a mock).
async fn run_with(
    cfg: &Config,
    args: CrewArgs,
    dispatcher: &dyn Dispatcher,
) -> anyhow::Result<i32> {
    let crew_name = resolve_crew_name(cfg, args.crew.as_deref())?;
    let crew = &cfg.crews[&crew_name];
    crew.validate(cfg).map_err(|e| anyhow::anyhow!("{e}"))?;

    let planner_model = model_for_role(cfg, &crew.planner)?;
    let navigator_model = match &crew.navigator {
        Some(n) => model_for_role(cfg, n)?,
        None => planner_model.clone(),
    };
    let triage_model = match &crew.triage {
        Some(t) => model_for_role(cfg, t)?,
        None => planner_model.clone(),
    };

    let dir = args
        .dir
        .clone()
        .unwrap_or_else(|| std::env::current_dir().unwrap_or_default());
    let test_cmd = args
        .test
        .clone()
        .or_else(|| crew.test.clone())
        .or_else(|| infer_test_command(&dir))
        .ok_or_else(|| {
            anyhow::anyhow!(
                "no verification command — pass --test, set the crew's `test`, or add a \
                 justfile / Cargo.toml / pyproject.toml to {}",
                dir.display()
            )
        })?;
    let max_attempts = args
        .max_attempts
        .or_else(|| crew.budgets.as_ref().and_then(|b| b.max_attempts))
        .unwrap_or(3);

    println!(
        "crew '{crew_name}': planner▸{planner_model}  navigator▸{navigator_model}  \
         triage▸{triage_model}  (max {max_attempts}, test `{test_cmd}`)"
    );
    if args.dry_run {
        return Ok(0);
    }

    // Build the pool from config. We do NOT pre-probe: the dispatcher's own error
    // drives failover (an unreachable backend errors → next candidate → `None`),
    // which keeps the loop honest without a probe round-trip.
    let pool = BackendPool::from_source(&StaticSource::from_configs(cfg.backends.iter()));
    let mut ws = WorktreeWorkspace::create(&dir, &worktree_id(), "HEAD", test_cmd)?;
    let crew_cfg = CrewConfig {
        navigator_model,
        planner_model,
        triage_model,
        max_attempts,
        role_timeout: None,
        // #883: the standalone `newt crew` leaf calibrates — a verify already
        // green on the pre-edit tree cannot prove the crew's edits.
        calibrate_baseline: true,
    };
    // Honest, non-top session caveats (the #94 guardrail forbids `Caveats::top()`
    // in dispatch code): exec/net are locked down. The REAL fs boundary here is
    // the throwaway git worktree, NOT a caveat leash: `worker_session_caveats(None)`
    // returns `fs_write = Scope::All` (identity.rs), so `permits_fs_write` is true
    // for every path and `run_crew`'s REFUSED branch is dead code. A scoped
    // `fs_write` leash would be a real second boundary, but isn't wired yet.
    //
    // NOTE: the CWD fs-lock (newt_core::caveats::apply_cli_fs_grants, applied for
    // `newt code`) is NOT applied here yet — `run_crew` enforces `permits_fs_write`
    // by EXACT match on the planner's (relative) edit paths, which an
    // absolute-workspace-root grant can't match, so the lock would deny
    // legitimate edits. The crew is already sandboxed by its isolated git
    // worktree. Reconciling the scheduler's path-enforcement with the lock is a
    // follow-up.
    let caveats = newt_acp_worker::worker_session_caveats(None);
    // Direct `newt crew` runs are a single whole-task crew (no plan, no leaf
    // lanes) — there is no leaf scope to fence (#812), so pass none.
    let outcome = run_crew(
        &pool,
        dispatcher,
        &mut ws,
        &crew_cfg,
        &caveats,
        &args.task,
        &[],
    )
    .await;
    // Drop ws (removes the worktree) BEFORE we may process::exit upstream.
    let touched_in = ws.path().to_path_buf();
    drop(ws);
    Ok(render(&outcome, &touched_in))
}

/// Pick the crew: the explicit `--crew`, else the sole crew, else an error.
fn resolve_crew_name(cfg: &Config, explicit: Option<&str>) -> anyhow::Result<String> {
    if let Some(n) = explicit {
        if cfg.crews.contains_key(n) {
            return Ok(n.to_string());
        }
        anyhow::bail!("no crew named '{n}' (known: {})", names(cfg.crews.keys()));
    }
    match cfg.crews.len() {
        1 => Ok(cfg.crews.keys().next().unwrap().clone()),
        0 => anyhow::bail!("no crews defined — add a [crews.<name>] or ~/.newt/crews/<name>.toml"),
        _ => anyhow::bail!(
            "multiple crews defined ({}) — pick one with --crew",
            names(cfg.crews.keys())
        ),
    }
}

/// The model a role-loadout pins: its `model` (with any `@variant` stripped for
/// pool pinning), else its provider backend's default model.
pub(crate) fn model_for_role(cfg: &Config, loadout_name: &str) -> anyhow::Result<String> {
    let l = cfg
        .loadouts
        .get(loadout_name)
        .ok_or_else(|| anyhow::anyhow!("role loadout '{loadout_name}' not found"))?;
    if let Some(m) = &l.model {
        return Ok(m.split('@').next().unwrap_or(m).to_string());
    }
    if let Some(p) = &l.provider {
        if let Some(b) = cfg.backends.iter().find(|b| &b.name == p) {
            if let Some(m) = b.effective_model() {
                return Ok(m.to_string());
            }
        }
    }
    anyhow::bail!(
        "role loadout '{loadout_name}' has no model (set `model` or a `provider` with a backend)"
    )
}

fn names<'a>(keys: impl Iterator<Item = &'a String>) -> String {
    let v: Vec<&str> = keys.map(String::as_str).collect();
    if v.is_empty() {
        "none".to_string()
    } else {
        v.join(", ")
    }
}

/// A unique worktree id for this run (pid + nanos — no extra deps).
pub(crate) fn worktree_id() -> String {
    let nanos = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!("crew-{}-{nanos}", std::process::id())
}

/// Print the outcome; return the exit code (0 passed, 2 needs human review).
fn render(o: &CrewOutcome, worktree: &Path) -> i32 {
    match o.status {
        CrewStatus::Passed => {
            let touched = if o.touched.is_empty() {
                "(none)".to_string()
            } else {
                o.touched.join(", ")
            };
            println!(
                "✓ crew passed in {} attempt(s) — touched: {touched}\n  worktree: {}",
                o.attempts,
                worktree.display()
            );
            0
        }
        CrewStatus::NeedsHumanReview => {
            println!(
                "⚠ crew needs human review — {} attempt(s) without a green check",
                o.attempts
            );
            2
        }
        CrewStatus::VacuousVerify => {
            println!(
                "⚠ vacuous verify — the check passed on the unedited baseline, so a green \
                 cannot prove the change ({} attempt(s)); not landed",
                o.attempts
            );
            2
        }
    }
}

#[cfg(test)]
#[path = "crew/tests.rs"]
mod tests;