drove 0.1.3

//! Executes tasks and lifecycle hooks: host argv Drove runs directly,
//! approval-gated on content digest, recorded in the local state journal
//! (spec §5, D11-D13, D19).
//!
//! Reconciles backend resources and runs task and lifecycle hook argv on
//! the host, in the repo root. Pane startup hooks gate command launch.

use std::{
    collections::{BTreeMap, BTreeSet},
    path::Path,
    process::Command,
};

use anyhow::Result;

use crate::{
    backend::{Backend, PaneSpec, SessionState},
    ir::{Ir, Resource},
    model::{Profile, Task, canonical_digest},
    planner::{Action, CoreAction, HerdrAction, Plan, PlannedAction},
    state::{LocalState, ManagedProfile, ManagedResource},
};

/// Runs host argv. A real [`HostCommandRunner`] shells out; tests substitute
/// a fake that records calls instead of touching the filesystem or network.
pub trait CommandRunner {
    fn run(&self, argv: &[String], cwd: &Path, env: &BTreeMap<String, String>) -> Result<bool>;
}

pub struct HostCommandRunner;

impl CommandRunner for HostCommandRunner {
    fn run(&self, argv: &[String], cwd: &Path, env: &BTreeMap<String, String>) -> Result<bool> {
        anyhow::ensure!(!argv.is_empty(), "cannot run an empty argv");
        let mut command = Command::new(&argv[0]);
        command.args(&argv[1..]).current_dir(cwd);
        for (key, value) in env {
            command.env(key, value);
        }
        Ok(command.status()?.success())
    }
}

pub struct ExecutionContext<'a> {
    pub repo_root: &'a Path,
    pub profile: &'a str,
    pub runner: &'a dyn CommandRunner,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TaskOutcome {
    /// `check` passed; `run` never executed.
    Skipped,
    /// `run` executed; carries its exit status.
    Ran(bool),
    /// `run` needed approval that isn't recorded yet.
    Blocked,
    /// Never attempted: an `after` prerequisite failed or was blocked (D52
    /// point 2 — a task after a failed task is skipped, not run anyway).
    DependencySkipped,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum HookEvent {
    Start,
    Stop,
}

impl HookEvent {
    fn label(self) -> &'static str {
        match self {
            HookEvent::Start => "on_start",
            HookEvent::Stop => "on_stop",
        }
    }
}

/// Runs one `task()`: `check` first (early cutoff), otherwise the
/// approval-gated `run`, then its `on_start` hook if `run` executed at all.
/// `resource_digest` is the task's IR content digest, recorded in local
/// state so the next `build_plan` sees this task as converged.
pub fn run_task(
    task: &Task,
    resource_digest: &str,
    ctx: &ExecutionContext<'_>,
    state: &mut LocalState,
    approve: bool,
) -> Result<TaskOutcome> {
    if let Some(check) = &task.check {
        // A check that cannot even start (missing executable, permission
        // denied) answers "not satisfied" rather than aborting `run_task`
        // outright: the task should still get a chance to run.
        let satisfied = ctx
            .runner
            .run(check, ctx.repo_root, &BTreeMap::new())
            .unwrap_or(false);
        if satisfied {
            record_task_resource(state, ctx, &task.name, resource_digest, "skipped", true)?;
            return Ok(TaskOutcome::Skipped);
        }
    }

    let approval_digest = canonical_digest(&task.run)?;
    if approve {
        state.approve(approval_digest.clone());
    }
    if !state.is_approved(&approval_digest) {
        return Ok(TaskOutcome::Blocked);
    }

    state.begin_action(&format!("task:{}", task.name), &approval_digest)?;
    let success = ctx.runner.run(&task.run, ctx.repo_root, &BTreeMap::new())?;
    state.finish_action(&approval_digest, success)?;

    if let Some(hook) = &task.on_start {
        // The hook's own outcome is intentionally not folded into this
        // task's `TaskOutcome`: it already gets its own approval gate and
        // journal entry (same as `down`'s hooks), but a wrapped task ran
        // (or didn't) independently of whether its post-run notification
        // succeeded.
        run_hook(
            hook,
            &task.name,
            None,
            HookEvent::Start,
            ctx,
            state,
            approve,
        )?;
    }

    // D18: only a successful `run` converges the task. Recording the
    // declared digest on failure would make the very next `build_plan` see
    // this task as in sync, so a failing `run` would never be retried.
    record_task_resource(
        state,
        ctx,
        &task.name,
        resource_digest,
        if success { "ok" } else { "failed" },
        success,
    )?;
    Ok(TaskOutcome::Ran(success))
}

/// Runs one `on_start`/`on_stop` argv hook (D13): approval-gated like a
/// task's `run`, with `DROVE_RESOURCE` and (when known) `DROVE_BACKEND_ID`
/// in the environment.
pub fn run_hook(
    argv: &[String],
    resource: &str,
    backend_id: Option<&str>,
    event: HookEvent,
    ctx: &ExecutionContext<'_>,
    state: &mut LocalState,
    approve: bool,
) -> Result<TaskOutcome> {
    if argv.is_empty() {
        return Ok(TaskOutcome::Skipped);
    }

    let mut env = BTreeMap::new();
    env.insert("DROVE_RESOURCE".to_owned(), resource.to_owned());
    if let Some(id) = backend_id {
        env.insert("DROVE_BACKEND_ID".to_owned(), id.to_owned());
    }

    let approval_digest = canonical_digest(&argv.to_vec())?;
    if approve {
        state.approve(approval_digest.clone());
    }
    if !state.is_approved(&approval_digest) {
        return Ok(TaskOutcome::Blocked);
    }

    state.begin_action(
        &format!("hook:{resource}:{}", event.label()),
        &approval_digest,
    )?;
    let success = ctx.runner.run(argv, ctx.repo_root, &env)?;
    state.finish_action(&approval_digest, success)?;
    Ok(TaskOutcome::Ran(success))
}

/// Records the task's last outcome unconditionally, but only records
/// `digest` as its *observed* digest when `converged` is true. On a failed
/// `run`, `converged` is false, so the previously recorded digest (or none,
/// if this is the task's first run) is kept: the task stays out of sync and
/// `build_plan` proposes it again on the next `drove up`/`plan`/`status`.
fn record_task_resource(
    state: &mut LocalState,
    ctx: &ExecutionContext<'_>,
    name: &str,
    digest: &str,
    outcome: &str,
    converged: bool,
) -> Result<()> {
    let profile = state.profile_mut(ctx.profile);
    let observed_digest = if converged {
        digest.to_owned()
    } else {
        profile
            .resources
            .get(name)
            .map(|resource| resource.digest.clone())
            .unwrap_or_default()
    };
    profile.resources.insert(
        name.to_owned(),
        ManagedResource {
            kind: "task".into(),
            backend_id: String::new(),
            parent: None,
            digest: observed_digest,
            label: None,
            cwd: None,
            adopted: None,
            command_started: None,
            last_outcome: Some(outcome.to_owned()),
        },
    );
    state.save()
}

/// `name`, last recorded outcome (`None` if it has never run) for every
/// declared task, in declaration order — what `drove run` with no argument
/// prints.
pub fn list_tasks(profile: &Profile, state: &LocalState) -> Vec<(String, Option<String>)> {
    let managed = state.profile(profile.name.as_str());
    profile
        .tasks
        .iter()
        .map(|task| {
            let outcome = managed
                .and_then(|managed| managed.resources.get(&task.name))
                .and_then(|resource| resource.last_outcome.clone());
            (task.name.clone(), outcome)
        })
        .collect()
}

/// Runs `target` and every task it transitively depends on through `after`,
/// in dependency order, and nothing else declared in the profile.
pub fn run_named_task(
    profile: &Profile,
    target: &str,
    ctx: &ExecutionContext<'_>,
    state: &mut LocalState,
    approve: bool,
) -> Result<Vec<(String, TaskOutcome)>> {
    let tasks_by_name: BTreeMap<&str, &Task> = profile
        .tasks
        .iter()
        .map(|task| (task.name.as_str(), task))
        .collect();
    anyhow::ensure!(
        tasks_by_name.contains_key(target),
        "no task named `{target}`"
    );

    let mut needed: BTreeSet<String> = BTreeSet::new();
    let mut stack = vec![target.to_owned()];
    while let Some(name) = stack.pop() {
        if !needed.insert(name.clone()) {
            continue;
        }
        if let Some(task) = tasks_by_name.get(name.as_str()) {
            for dep in &task.after {
                if tasks_by_name.contains_key(dep.as_str()) {
                    stack.push(dep.clone());
                }
            }
        }
    }

    let mut edges: BTreeMap<String, Vec<String>> = BTreeMap::new();
    for name in &needed {
        let after = tasks_by_name[name.as_str()].after.clone();
        edges.insert(name.clone(), after);
    }
    let order = topo_forward(&needed, &edges);

    let ir = profile.to_ir();
    let mut results = Vec::new();
    let mut blocked: BTreeSet<String> = BTreeSet::new();
    for name in order {
        let task = tasks_by_name[name.as_str()];
        // A prerequisite that failed or was itself blocked means this task
        // never runs either (D52 point 2): running it anyway would build on
        // a state its own `after` says isn't ready.
        if task.after.iter().any(|dep| blocked.contains(dep.as_str())) {
            blocked.insert(name.clone());
            results.push((name, TaskOutcome::DependencySkipped));
            continue;
        }
        let digest = digest_of(&ir, "task", &name)?;
        let outcome = run_task(task, digest, ctx, state, approve)?;
        if matches!(outcome, TaskOutcome::Ran(false) | TaskOutcome::Blocked) {
            blocked.insert(name.clone());
        }
        results.push((name, outcome));
    }
    Ok(results)
}

/// Runs every `RunTask` action a [`Plan`] proposes, in the plan's own
/// (already `after`-ordered) order. Every other action kind is left for a
/// future PR once the `Backend` methods it needs (PR 3/5) exist.
///
/// A task whose `after` names one that failed or was itself blocked is never
/// run: it's reported as [`TaskOutcome::DependencySkipped`] in the returned
/// task list and collected into the returned [`SkippedAction`] list (D52
/// point 2), the same shape `apply_plan_gated` uses for a skipped backend
/// action.
type PlanTaskResults = (Vec<(String, TaskOutcome)>, Vec<SkippedAction>);

pub fn execute_plan_tasks(
    profile: &Profile,
    plan: &Plan,
    ctx: &ExecutionContext<'_>,
    state: &mut LocalState,
    approve: bool,
) -> Result<PlanTaskResults> {
    let tasks_by_name: BTreeMap<&str, &Task> = profile
        .tasks
        .iter()
        .map(|task| (task.name.as_str(), task))
        .collect();
    let ir = profile.to_ir();
    let mut results = Vec::new();
    let mut skipped = Vec::new();
    let mut blocked: BTreeSet<String> = BTreeSet::new();
    for action in &plan.actions {
        if action.kind != Action::Core(CoreAction::RunTask) {
            continue;
        }
        let Some(task) = tasks_by_name.get(action.address.as_str()) else {
            continue;
        };
        let blocking_dependency = task.after.iter().find(|dep| blocked.contains(dep.as_str()));
        if let Some(depends_on) = blocking_dependency {
            skipped.push(SkippedAction {
                address: action.address.clone(),
                depends_on: depends_on.clone(),
            });
            blocked.insert(action.address.clone());
            results.push((action.address.clone(), TaskOutcome::DependencySkipped));
            continue;
        }
        let digest = digest_of(&ir, "task", &action.address)?;
        let outcome = run_task(task, digest, ctx, state, approve)?;
        if matches!(outcome, TaskOutcome::Ran(false) | TaskOutcome::Blocked) {
            blocked.insert(action.address.clone());
        }
        results.push((action.address.clone(), outcome));
    }
    Ok((results, skipped))
}

#[derive(Debug, Clone, Default)]
pub struct DownReport {
    /// Resource identities detached, in the order they were torn down.
    pub detached: Vec<String>,
    /// `(resource identity, hook succeeded)` for every `on_stop` hook run.
    pub hooks_run: Vec<(String, bool)>,
    /// `(resource identity, error message)` for every `close_pane` call that
    /// failed (D50): the resource is still detached and the failure does not
    /// abort the teardown, so a pane the session already lost is treated as
    /// already gone rather than blocking `down`.
    pub close_failed: Vec<(String, String)>,
}

/// `drove down` (D19): runs each owned resource's `on_stop` hook (if the
/// profile still declares one), then detaches it — with `purge`, also
/// closes owned panes on the backend — in reverse dependency order.
/// Resources the backend doesn't recognize as owned by this profile (an
/// unmanaged pane) are never touched, because they are never in
/// `state`'s managed set to begin with. A `close_pane` failure (D50) no
/// longer aborts the loop: the resource is still detached and saved, and
/// the failure is collected in [`DownReport::close_failed`] instead.
pub fn down(
    profile: &Profile,
    ctx: &ExecutionContext<'_>,
    state: &mut LocalState,
    approve: bool,
    purge: bool,
    backend: Option<&dyn Backend>,
) -> Result<DownReport> {
    let ir = profile.to_ir();
    let on_stop_hooks = collect_hooks(profile, HookEvent::Stop);
    let managed = state.profile(ctx.profile).cloned().unwrap_or_default();
    let order = teardown_order(&managed.resources, &ir);

    let mut report = DownReport::default();
    for id in order {
        let Some(resource) = managed.resources.get(&id) else {
            continue;
        };
        if let Some(hook) = on_stop_hooks.get(id.as_str()) {
            let backend_id = if resource.backend_id.is_empty() {
                None
            } else {
                Some(resource.backend_id.as_str())
            };
            // A blocked (unapproved) or failed `on_stop` does not stop the
            // teardown below: `down`'s job is to stop tracking a resource,
            // not to hold it hostage to hook approval. A hook that must run
            // before teardown (e.g. one that kills a background process)
            // needs its digest pre-approved, the same way a task's `run`
            // does.
            let outcome = run_hook(hook, &id, backend_id, HookEvent::Stop, ctx, state, approve)?;
            if let TaskOutcome::Ran(success) = outcome {
                report.hooks_run.push((id.clone(), success));
            }
        }

        if purge
            && resource.kind == "pane"
            && let Some(backend) = backend
            && let Err(error) = backend.close_pane(&resource.backend_id)
        {
            report.close_failed.push((id.clone(), error.to_string()));
        }

        state.profile_mut(ctx.profile).resources.remove(&id);
        state.save()?;
        report.detached.push(id);
    }
    Ok(report)
}

fn collect_hooks(profile: &Profile, event: HookEvent) -> BTreeMap<&str, &[String]> {
    let mut hooks = BTreeMap::new();
    for workspace in &profile.workspaces {
        for group in &workspace.tabs {
            for pane in &group.panes {
                let hook = match event {
                    HookEvent::Start => &pane.on_start,
                    HookEvent::Stop => &pane.on_stop,
                };
                if let Some(argv) = hook {
                    hooks.insert(pane.name.as_str(), argv.as_slice());
                }
            }
        }
    }
    for task in &profile.tasks {
        let hook = match event {
            HookEvent::Start => &task.on_start,
            HookEvent::Stop => &task.on_stop,
        };
        if let Some(argv) = hook {
            hooks.insert(task.name.as_str(), argv.as_slice());
        }
    }
    hooks
}

/// A resource's identity for the shared namespace (D5): its own declared
/// name. Placement groups are not core resources (D29), so they never appear
/// here.
fn identity(resource: &Resource) -> String {
    resource.name.clone()
}

/// Dependent -> its dependencies: a resource's structural parent (a pane
/// depends on its placement group, an agent on its pane) plus whatever it
/// names in a declared `after`.
fn dependency_edges(ir: &Ir) -> BTreeMap<String, Vec<String>> {
    let mut edges = BTreeMap::new();
    for resource in &ir.resources {
        let mut deps = Vec::new();
        if let Some(parent) = &resource.parent {
            deps.push(parent.clone());
        }
        if let Some(after) = resource.fields.get("after").and_then(|v| v.as_array()) {
            for value in after {
                if let Some(name) = value.as_str() {
                    deps.push(name.to_owned());
                }
            }
        }
        edges.insert(identity(resource), deps);
    }
    edges
}

fn digest_of<'a>(ir: &'a Ir, kind: &str, name: &str) -> Result<&'a str> {
    ir.resources
        .iter()
        .find(|resource| resource.kind == kind && resource.name == name)
        .map(|resource| resource.digest.as_str())
        .ok_or_else(|| anyhow::anyhow!("no {kind} resource named `{name}`"))
}

/// Dependencies before dependents (a resource's creation order), restricted
/// to `ids`. The DAG is already enforced by `Profile::validate`, so a stray
/// cycle among `ids` alone (there isn't one in practice) just falls back to
/// appending the unresolved remainder in name order.
fn topo_forward(ids: &BTreeSet<String>, edges: &BTreeMap<String, Vec<String>>) -> Vec<String> {
    let mut indegree: BTreeMap<&str, usize> = ids.iter().map(|id| (id.as_str(), 0)).collect();
    let mut children: BTreeMap<&str, Vec<&str>> = BTreeMap::new();
    for id in ids {
        if let Some(deps) = edges.get(id) {
            for dep in deps {
                if ids.contains(dep) {
                    *indegree
                        .get_mut(id.as_str())
                        .expect("every id in `ids` seeds `indegree`") += 1;
                    children.entry(dep.as_str()).or_default().push(id.as_str());
                }
            }
        }
    }

    let mut frontier: BTreeSet<&str> = indegree
        .iter()
        .filter(|(_, degree)| **degree == 0)
        .map(|(id, _)| *id)
        .collect();
    let mut order: Vec<String> = Vec::new();
    while let Some(id) = frontier.iter().next().copied() {
        frontier.remove(id);
        order.push(id.to_owned());
        if let Some(kids) = children.get(id) {
            for kid in kids {
                let degree = indegree
                    .get_mut(kid)
                    .expect("`children` only ever names ids seeded into `indegree`");
                *degree -= 1;
                if *degree == 0 {
                    frontier.insert(kid);
                }
            }
        }
    }
    for id in ids {
        if !order.contains(id) {
            order.push(id.clone());
        }
    }
    order
}

fn teardown_order(managed: &BTreeMap<String, ManagedResource>, ir: &Ir) -> Vec<String> {
    let ids: BTreeSet<String> = managed.keys().cloned().collect();
    let edges = dependency_edges(ir);
    let mut order = topo_forward(&ids, &edges);
    order.reverse();
    order
}

/// One backend call inside an apply loop that failed; collected instead of
/// aborting so independent actions still apply and dependents can be skipped
/// (D52 point 2).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FailedAction {
    pub address: String,
    pub error: String,
}

/// One action left unattempted because an action it structurally depends on
/// (its placement group, its workspace, its pane) failed or was itself
/// skipped earlier in the same apply (D52 point 2).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SkippedAction {
    pub address: String,
    pub depends_on: String,
}

/// What became of one planned action when applied to a backend (D29, D52).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Outcome {
    /// The verb ran against the backend.
    Applied,
    /// The action is recorded/handled outside the backend apply loop (a task
    /// run, a detach, a task conflict report), or held back for want of
    /// `--yes` (a destructive action, D22).
    Skipped,
    /// A flavor action whose flavor this backend does not implement. `drove
    /// plan` prints it and `drove status` counts it; it is never dropped.
    Unsupported {
        flavor: &'static str,
        action: Action,
    },
    /// The backend call for this action failed; the error is also collected
    /// into the apply's `failed` list. The loop does not abort (D52 point 2).
    Failed(String),
    /// Not attempted: it depends on an action that failed or was itself
    /// skipped this run (D52 point 2).
    DependencySkipped { depends_on: String },
}

/// Running backend ids gathered while a plan applies: creating a workspace,
/// group or pane yields the id later actions address.
#[derive(Default)]
struct ApplyState {
    workspace_ids: BTreeMap<String, String>,
    group_ids: BTreeMap<String, String>,
    pane_ids: BTreeMap<String, String>,
    created_workspaces: BTreeSet<String>,
    created_groups: BTreeSet<String>,
    created_panes: BTreeSet<String>,
    command_started: BTreeMap<String, bool>,
    blocked: BTreeSet<String>,
}

impl ApplyState {
    /// Seeds the resolver with the backend ids a previous run recorded, so an
    /// action against a parent that already converged (and so needs no action
    /// this run) still resolves that parent's id. Without this, a later `up`
    /// that only adds a pane to an existing group cannot find the group's
    /// workspace, since nothing populated `workspace_ids` for it this run.
    fn seeded_from(managed: Option<&crate::state::ManagedProfile>) -> Self {
        let mut state = Self::default();
        let Some(managed) = managed else {
            return state;
        };
        for (address, resource) in &managed.resources {
            if resource.backend_id.is_empty() {
                continue;
            }
            let map = match resource.kind.as_str() {
                "workspace" => &mut state.workspace_ids,
                "placement" => &mut state.group_ids,
                "pane" => &mut state.pane_ids,
                _ => continue,
            };
            map.insert(address.clone(), resource.backend_id.clone());
        }
        state
    }
}

/// A fresh Herdr tab launches every pane in one backend call. Its hooks must all
/// succeed before that call; new panes have no backend id at this point.
fn starting_panes(ir: &Ir, action: &PlannedAction) -> Vec<String> {
    match action.kind {
        Action::Core(CoreAction::CreatePane | CoreAction::RestartCommand)
        | Action::Herdr(HerdrAction::SplitPane) => vec![action.address.clone()],
        Action::Herdr(HerdrAction::CreateTab) => placement_group(ir, &action.address)
            .map(|group| group.panes.clone())
            .unwrap_or_default(),
        _ => Vec::new(),
    }
}

/// Every id this action's target structurally depends on, nearest first: a
/// pane's placement group and that group's workspace, a group's workspace, an
/// agent's pane and that pane's group and workspace, followed by transitive
/// declared `after` prerequisites. A fresh group also depends on the external
/// prerequisites of the panes it starts.
fn action_dependencies(ir: &Ir, action: &PlannedAction) -> Vec<String> {
    let mut dependencies = match action.kind {
        Action::Herdr(HerdrAction::CreateTab | HerdrAction::RenameTab | HerdrAction::SetRatio) => {
            group_workspace(ir, &action.address)
                .map(|workspace| vec![workspace.to_owned()])
                .unwrap_or_default()
        }
        Action::Core(
            CoreAction::CreatePane | CoreAction::RenamePane | CoreAction::RestartCommand,
        )
        | Action::Herdr(HerdrAction::SplitPane) => {
            let Ok(group) = pane_group_id(ir, &action.address) else {
                return Vec::new();
            };
            let mut chain = vec![group.clone()];
            if let Some(workspace) = group_workspace(ir, &group) {
                chain.push(workspace.to_owned());
            }
            chain
        }
        Action::Herdr(HerdrAction::StartAgent) | Action::Core(CoreAction::PromptAgent) => {
            let Ok(pane) = agent_parent(ir, &action.address) else {
                return Vec::new();
            };
            let mut chain = vec![pane.clone()];
            if let Ok(group) = pane_group_id(ir, &pane) {
                chain.push(group.clone());
                if let Some(workspace) = group_workspace(ir, &group) {
                    chain.push(workspace.to_owned());
                }
            }
            chain
        }
        _ => Vec::new(),
    };
    let edges = dependency_edges(ir);
    let mut pending = starting_panes(ir, action);
    pending.push(action.address.clone());
    let mut visited = BTreeSet::new();
    while let Some(id) = pending.pop() {
        if visited.insert(id.clone())
            && let Some(deps) = edges.get(&id)
        {
            dependencies.extend(deps.iter().cloned());
            pending.extend(deps.iter().cloned());
        }
    }
    // A fresh group's panes are created together, so they cannot gate their
    // own containing action. External prerequisites still gate the group.
    let internal = starting_panes(ir, action);
    dependencies.retain(|id| id != &action.address && !internal.contains(id));
    dependencies
}

/// Applies every action in `plan` against `backend`, resolving each verb's
/// concrete arguments from `ir`, and returns each action's [`Outcome`] in
/// order. A flavor action on a backend without that flavor is surfaced as
/// [`Outcome::Unsupported`] and the loop continues, so core resources in the
/// same plan are still created (D29). Destructive actions are always applied;
/// [`up`] uses `apply_plan_gated` instead to hold them behind `--yes`.
pub fn apply_plan(backend: &dyn Backend, ir: &Ir, plan: &Plan) -> Vec<(String, Outcome)> {
    apply_plan_gated(backend, ir, plan, true, ApplyState::default(), |_, _, _| {
        Ok(())
    })
    .0
}

/// Like [`apply_plan`], but when `approve` is false every destructive action
/// (a topology-change `ClosePane`, D22) is left unapplied and reported as
/// [`Outcome::Skipped`] — the same `--yes` gate a task's `run` sits behind.
///
/// A failing action's error is collected into the returned `failed` list
/// instead of aborting the loop (D52 point 2): every other independent
/// action still applies. An action that depends on one that failed, or was
/// itself skipped this run, is reported as [`Outcome::DependencySkipped`] and
/// collected into `skipped` rather than attempted. `on_action` runs before
/// an attempted action with `None` (an error gates the backend call), then
/// with `Some(outcome)` to record ownership before the next action. An error
/// from that second call stops the loop so nothing more applies without a
/// record of it (D52 point 1).
/// Every action's outcome, the backend ids created along the way, and what
/// failed or was skipped — [`apply_plan_gated`]'s result.
type ApplyPlanResult = (
    Vec<(String, Outcome)>,
    ApplyState,
    Vec<FailedAction>,
    Vec<SkippedAction>,
);

fn apply_plan_gated(
    backend: &dyn Backend,
    ir: &Ir,
    plan: &Plan,
    approve: bool,
    seed: ApplyState,
    mut on_action: impl FnMut(&PlannedAction, Option<&Outcome>, &ApplyState) -> Result<()>,
) -> ApplyPlanResult {
    let mut state = seed;
    let mut outcomes: Vec<Option<(String, Outcome)>> =
        (0..plan.actions.len()).map(|_| None).collect();
    let mut failed = Vec::new();
    let mut skipped = Vec::new();
    let mut blocked = std::mem::take(&mut state.blocked);

    // A `SetRatio` addresses a split gap, so it must run after the panes that
    // create the group's gaps. The plan orders every Herdr tab action ahead of
    // the pane splits (its rank sorts before the pane rank), which is right for
    // `plan`/`status` output but would apply a ratio before its gap exists when
    // a pane is added to an existing group. So apply the ratios last, keeping
    // each action's outcome in its original plan position.
    let is_deferred =
        |action: &PlannedAction| matches!(action.kind, Action::Herdr(HerdrAction::SetRatio));
    let mut pending: Vec<_> = plan
        .actions
        .iter()
        .enumerate()
        .filter(|(_, action)| !is_deferred(action))
        .chain(
            plan.actions
                .iter()
                .enumerate()
                .filter(|(_, action)| is_deferred(action)),
        )
        .collect();
    let dependencies: Vec<_> = plan
        .actions
        .iter()
        .map(|action| action_dependencies(ir, action))
        .collect();

    while !pending.is_empty() {
        // Preserve plan order wherever possible, but wait for declared
        // prerequisites even when their names sort after their dependents.
        // SetRatio is an end-of-apply adjustment, never a startup prerequisite.
        let ready = pending.iter().position(|(index, action)| {
            !pending.iter().any(|(other_index, other)| {
                other_index != index
                    && other.address != action.address
                    && !is_deferred(other)
                    && (dependencies[*index].contains(&other.address)
                        || starting_panes(ir, other)
                            .iter()
                            .any(|pane| dependencies[*index].contains(pane)))
            })
        });
        let (index, action) = pending.remove(ready.unwrap_or(0));
        let blocking_dependency = std::iter::once(action.address.clone())
            .chain(dependencies[index].iter().cloned())
            .find(|id| blocked.contains(id));
        let outcome = if action.kind == Action::Core(CoreAction::RunTask)
            || (action.destructive && !approve)
        {
            Outcome::Skipped
        } else if let Some(depends_on) = blocking_dependency {
            skipped.push(SkippedAction {
                address: action.address.clone(),
                depends_on: depends_on.clone(),
            });
            Outcome::DependencySkipped { depends_on }
        } else {
            let before = if ready.is_none() {
                Err(anyhow::anyhow!(
                    "cyclic startup dependencies between placement groups"
                ))
            } else {
                on_action(action, None, &state)
            };
            match before.and_then(|()| apply_action(backend, ir, &mut state, action)) {
                Ok(outcome) => outcome,
                Err(error) => {
                    let message = error.to_string();
                    failed.push(FailedAction {
                        address: action.address.clone(),
                        error: message.clone(),
                    });
                    Outcome::Failed(message)
                }
            }
        };
        if matches!(
            outcome,
            Outcome::Failed(_) | Outcome::DependencySkipped { .. }
        ) || (action.destructive && !approve)
        {
            blocked.insert(action.address.clone());
            blocked.extend(starting_panes(ir, action));
        }
        let keep_going = on_action(action, Some(&outcome), &state).is_ok();
        outcomes[index] = Some((action.address.clone(), outcome));
        if !keep_going {
            break;
        }
    }

    // Ordinarily every slot is filled, but an ownership save error
    // stops the loop before the rest run, so
    // trailing entries stay `None` rather than lying about an outcome they
    // never got.
    let outcomes = outcomes.into_iter().flatten().collect();
    (outcomes, state, failed, skipped)
}

/// What `drove up` did, reported as one summary line (D43 step 5).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum UpOutcome {
    /// The session was reachable (or just started headlessly) and the plan's
    /// tasks and backend actions were applied.
    Reconciled {
        created: usize,
        changed: usize,
        tasks_run: usize,
    },
    /// Nothing was out of sync; the workspace was only brought to the front.
    AlreadyRunning,
    /// The Herdr session's server was not reachable and could not be started
    /// headlessly (D43 step 2); `hint` is the command to run by hand.
    CannotStart { hint: String },
}

/// The full result of one [`up`] run.
#[derive(Debug, Clone)]
pub struct UpReport {
    pub outcome: UpOutcome,
    /// Per-task run results, for `--json` output and the process exit code.
    pub tasks: Vec<(String, TaskOutcome)>,
    /// The backend id of the workspace brought to the front, if any.
    pub focused: Option<String>,
    /// A destructive action was left unapplied for want of `--yes` (D22).
    pub blocked_destructive: bool,
    /// Backend actions whose call or startup hook failed (including hooks
    /// awaiting approval); `up` exits 1 when this is non-empty.
    pub failed: Vec<FailedAction>,
    /// Actions left unattempted because an action they depend on failed or
    /// was itself skipped this run (D52 point 2).
    pub skipped: Vec<SkippedAction>,
}

/// `drove up` end to end (D43): ensure the session is reachable, run the
/// plan's tasks, apply its backend actions behind the `--yes` gate, record
/// what was created, and bring the target workspace to the front. The caller
/// (`src/cli.rs`) is responsible for the `Conflict` early exit before calling
/// this, for printing the summary, and for the `exec herdr session attach`
/// step, which is not exercised here.
#[allow(clippy::too_many_arguments)]
pub fn up(
    backend: &dyn Backend,
    profile: &Profile,
    ir: &Ir,
    plan: &Plan,
    ctx: &ExecutionContext<'_>,
    state: &mut LocalState,
    approve: bool,
    session: &str,
    focus_workspace: Option<&str>,
    do_focus: bool,
) -> Result<UpReport> {
    // Step 2: make the session reachable. Herdr starts its own server
    // headlessly; a flavorless backend (Radiator) has no such verb, so the
    // caller checks its reachability separately (D43 step 2, D44).
    if let Some(ext) = backend.herdr()
        && let SessionState::CannotStart { hint } = ext.ensure_session(session)?
    {
        return Ok(UpReport {
            outcome: UpOutcome::CannotStart { hint },
            tasks: Vec::new(),
            focused: None,
            blocked_destructive: false,
            failed: Vec::new(),
            skipped: Vec::new(),
        });
    }

    let was_in_sync = plan.actions.is_empty();

    // Step 3: run the plan's tasks, then apply its backend actions, recording
    // ownership and saving state right after each action succeeds (D52
    // point 1): a killed process leaves state describing exactly what was
    // created, instead of discarding it the way a single end-of-loop
    // `record_ownership` pass would.
    let (tasks, task_skipped) = execute_plan_tasks(profile, plan, ctx, state, approve)?;
    // Seed the resolver with what a previous run recorded, so an action
    // against a parent that already converged still finds its backend id.
    let mut seed = ApplyState::seeded_from(state.profile(ctx.profile));
    seed.blocked
        .extend(tasks.iter().filter_map(|(name, outcome)| {
            matches!(
                outcome,
                TaskOutcome::Ran(false) | TaskOutcome::Blocked | TaskOutcome::DependencySkipped
            )
            .then_some(name.clone())
        }));
    let hooks = collect_hooks(profile, HookEvent::Start);
    let mut started = BTreeSet::new();
    let existing = state.profile(ctx.profile).cloned().unwrap_or_default();
    let mut save_error: Option<anyhow::Error> = None;
    let (outcomes, applied, failed, backend_skipped) = apply_plan_gated(
        backend,
        ir,
        plan,
        approve,
        seed,
        |action, outcome, applied| {
            let Some(outcome) = outcome else {
                // Flavorless backends must still report Unsupported without
                // running hooks for a verb they cannot apply.
                if matches!(action.kind, Action::Herdr(_)) && backend.herdr().is_none() {
                    return Ok(());
                }
                for pane in starting_panes(ir, action) {
                    if started.contains(&pane) {
                        continue;
                    }
                    if let Some(argv) = hooks.get(pane.as_str()) {
                        let backend_id = if action.kind == Action::Core(CoreAction::RestartCommand)
                        {
                            action
                                .backend_id
                                .as_deref()
                                .or_else(|| applied.pane_ids.get(&pane).map(String::as_str))
                        } else {
                            None
                        };
                        let outcome = run_hook(
                            argv,
                            &pane,
                            backend_id,
                            HookEvent::Start,
                            ctx,
                            state,
                            approve,
                        )
                        .map_err(|error| anyhow::anyhow!("pane `{pane}` on_start: {error}"))?;
                        match outcome {
                            TaskOutcome::Skipped | TaskOutcome::Ran(true) => {}
                            TaskOutcome::Blocked => {
                                anyhow::bail!("pane `{pane}` on_start requires approval")
                            }
                            _ => anyhow::bail!("pane `{pane}` on_start failed"),
                        }
                    }
                    started.insert(pane);
                }
                return Ok(());
            };
            let managed = state.profile_mut(ctx.profile);
            let changed = record_action_ownership(managed, &existing, ir, action, outcome, applied);
            if changed && let Err(error) = state.save() {
                save_error = Some(error);
                anyhow::bail!("cannot save action ownership");
            }
            Ok(())
        },
    );
    if let Some(error) = save_error {
        return Err(error);
    }
    let mut skipped = task_skipped;
    skipped.extend(backend_skipped);

    let blocked_destructive = !approve && plan.has_destructive_actions();
    let (created, changed) = count_applied(plan, &outcomes);
    let tasks_run = tasks
        .iter()
        .filter(|(_, outcome)| matches!(outcome, TaskOutcome::Ran(_)))
        .count();

    // Step 4: bring the target workspace to the front.
    let focused = if do_focus {
        focus_first_workspace(backend, state, ctx.profile, &applied, focus_workspace)?
    } else {
        None
    };

    let outcome = if was_in_sync {
        UpOutcome::AlreadyRunning
    } else {
        UpOutcome::Reconciled {
            created,
            changed,
            tasks_run,
        }
    };
    Ok(UpReport {
        outcome,
        tasks,
        focused,
        blocked_destructive,
        failed,
        skipped,
    })
}

/// Splits the plan's applied actions into a created count and a changed count
/// for the summary line. Only [`Outcome::Applied`] actions count; a skipped,
/// unsupported, task, detach, or conflict action does not.
fn count_applied(plan: &Plan, outcomes: &[(String, Outcome)]) -> (usize, usize) {
    let mut created = 0;
    let mut changed = 0;
    for (action, (_, outcome)) in plan.actions.iter().zip(outcomes) {
        if *outcome != Outcome::Applied {
            continue;
        }
        match action.kind {
            Action::Core(CoreAction::CreateWorkspace | CoreAction::CreatePane)
            | Action::Herdr(
                HerdrAction::CreateTab | HerdrAction::SplitPane | HerdrAction::StartAgent,
            ) => created += 1,
            Action::Core(
                CoreAction::RenameWorkspace
                | CoreAction::RenamePane
                | CoreAction::RestartCommand
                | CoreAction::ClosePane
                | CoreAction::PromptAgent,
            )
            | Action::Herdr(HerdrAction::RenameTab | HerdrAction::SetRatio) => changed += 1,
            _ => {}
        }
    }
    (created, changed)
}

/// Brings the profile's target workspace to the front through
/// `workspace.focus` (D43 step 4). The workspace's backend id comes from what
/// this run just created, else from what a previous run recorded in local
/// state (the already-in-sync case). A flavorless backend has no
/// `focus_workspace` verb, so this is a no-op there.
fn focus_first_workspace(
    backend: &dyn Backend,
    state: &LocalState,
    profile: &str,
    applied: &ApplyState,
    workspace: Option<&str>,
) -> Result<Option<String>> {
    let Some(name) = workspace else {
        return Ok(None);
    };
    let Some(ext) = backend.herdr() else {
        return Ok(None);
    };
    let backend_id = applied
        .workspace_ids
        .get(name)
        .cloned()
        .or_else(|| {
            state
                .profile(profile)
                .and_then(|managed| managed.resources.get(name))
                .map(|resource| resource.backend_id.clone())
        })
        .filter(|id| !id.is_empty());
    let Some(backend_id) = backend_id else {
        return Ok(None);
    };
    ext.focus_workspace(&backend_id)?;
    Ok(Some(backend_id))
}

/// Records one applied action's ownership into `managed.resources` — the
/// per-action split of the old whole-plan pass (D52 point 1), called right
/// after each action's outcome is known so a save right after leaves state
/// describing exactly what succeeded. The backend id comes from what this
/// action's own apply created (`applied`), else the action's own already-known
/// backend id (a rename or restart of an already-known resource), else what
/// local state already held (`existing`, a snapshot from before this apply
/// started). A `Detach` drops the resource; an `AdoptPane` records the caller
/// pane even though no backend verb ran (D24) — both regardless of `outcome`,
/// since neither ever touches the backend. Returns whether `managed` actually
/// changed, so the caller only pays for a `state.save()` when there is
/// something to save.
fn record_action_ownership(
    managed: &mut ManagedProfile,
    existing: &ManagedProfile,
    ir: &Ir,
    action: &PlannedAction,
    outcome: &Outcome,
    applied: &ApplyState,
) -> bool {
    let resolve = |name: &str, ids: &BTreeMap<String, String>, own: Option<&str>| -> String {
        ids.get(name)
            .cloned()
            .or_else(|| own.map(str::to_owned))
            .or_else(|| {
                existing
                    .resources
                    .get(name)
                    .map(|resource| resource.backend_id.clone())
            })
            .unwrap_or_default()
    };

    let address = action.address.as_str();
    match action.kind {
        Action::Core(CoreAction::AdoptPane) => {
            let (Ok(digest), Ok(parent)) =
                (digest_of(ir, "pane", address), pane_group_id(ir, address))
            else {
                return false;
            };
            let backend_id = resolve(address, &applied.pane_ids, action.backend_id.as_deref());
            let cwd = resource_fields(ir, "pane", address)
                .ok()
                .and_then(|fields| string_field(fields, "cwd"));
            managed.resources.insert(
                address.to_owned(),
                ManagedResource {
                    kind: "pane".into(),
                    backend_id,
                    parent: Some(parent),
                    digest: digest.to_owned(),
                    label: None,
                    cwd,
                    adopted: Some(true),
                    command_started: None,
                    last_outcome: None,
                },
            );
            return true;
        }
        Action::Core(CoreAction::Detach) => {
            return managed.resources.remove(address).is_some();
        }
        _ => {}
    }

    let recorded_physical_create = match action.kind {
        Action::Core(CoreAction::CreateWorkspace) => applied.created_workspaces.contains(address),
        Action::Core(CoreAction::CreatePane) | Action::Herdr(HerdrAction::SplitPane) => {
            applied.created_panes.contains(address)
        }
        Action::Herdr(HerdrAction::CreateTab) => applied.created_groups.contains(address),
        _ => false,
    };

    if *outcome != Outcome::Applied && !recorded_physical_create {
        return false;
    }

    match action.kind {
        Action::Core(CoreAction::CreateWorkspace | CoreAction::RenameWorkspace) => {
            let Ok(digest) = digest_of(ir, "workspace", address) else {
                return false;
            };
            let backend_id = resolve(
                address,
                &applied.workspace_ids,
                action.backend_id.as_deref(),
            );
            let label = resource_fields(ir, "workspace", address)
                .ok()
                .and_then(|fields| string_field(fields, "label"));
            managed.resources.insert(
                address.to_owned(),
                ManagedResource {
                    kind: "workspace".into(),
                    backend_id,
                    parent: None,
                    digest: digest.to_owned(),
                    label,
                    cwd: None,
                    adopted: None,
                    command_started: None,
                    last_outcome: None,
                },
            );
            true
        }
        Action::Herdr(HerdrAction::CreateTab | HerdrAction::RenameTab | HerdrAction::SetRatio) => {
            let mut changed = false;
            if let (Some(digest), Some(workspace)) = (
                group_topology_digest(ir, address),
                group_workspace(ir, address),
            ) {
                let backend_id = resolve(address, &applied.group_ids, action.backend_id.as_deref());
                let label = ir
                    .placements
                    .iter()
                    .find(|group| group.id == address)
                    .map(|group| group.label.clone());
                managed.resources.insert(
                    address.to_owned(),
                    ManagedResource {
                        kind: "placement".into(),
                        backend_id,
                        parent: Some(workspace.to_owned()),
                        digest: digest.to_owned(),
                        label,
                        cwd: None,
                        adopted: None,
                        command_started: None,
                        last_outcome: None,
                    },
                );
                changed = true;
            }
            // A fresh `CreateTab` builds every pane in the group with no
            // per-pane action, so record each one here (its parent is the
            // group) — otherwise the next run would see them unobserved and
            // split them in again.
            if action.kind == Action::Herdr(HerdrAction::CreateTab)
                && let Ok(group) = placement_group(ir, address)
            {
                for pane in &group.panes {
                    if let Ok(digest) = digest_of(ir, "pane", pane) {
                        let backend_id = resolve(pane, &applied.pane_ids, None);
                        let cwd = resource_fields(ir, "pane", pane)
                            .ok()
                            .and_then(|fields| string_field(fields, "cwd"));
                        managed.resources.insert(
                            pane.clone(),
                            ManagedResource {
                                kind: "pane".into(),
                                backend_id,
                                parent: Some(address.to_owned()),
                                digest: digest.to_owned(),
                                label: None,
                                cwd,
                                adopted: None,
                                command_started: command_started(ir, pane, applied, existing),
                                last_outcome: None,
                            },
                        );
                        changed = true;
                    }
                }
            }
            changed
        }
        Action::Core(
            CoreAction::CreatePane | CoreAction::RenamePane | CoreAction::RestartCommand,
        )
        | Action::Herdr(HerdrAction::SplitPane) => {
            let (Ok(digest), Ok(parent)) =
                (digest_of(ir, "pane", address), pane_group_id(ir, address))
            else {
                return false;
            };
            let backend_id = resolve(address, &applied.pane_ids, action.backend_id.as_deref());
            let adopted = existing
                .resources
                .get(address)
                .and_then(|resource| resource.adopted);
            let cwd = resource_fields(ir, "pane", address)
                .ok()
                .and_then(|fields| string_field(fields, "cwd"));
            managed.resources.insert(
                address.to_owned(),
                ManagedResource {
                    kind: "pane".into(),
                    backend_id,
                    parent: Some(parent),
                    digest: digest.to_owned(),
                    label: None,
                    cwd,
                    adopted,
                    command_started: command_started(ir, address, applied, existing),
                    last_outcome: None,
                },
            );
            true
        }
        Action::Herdr(HerdrAction::StartAgent) | Action::Core(CoreAction::PromptAgent) => {
            let (Ok(digest), Ok(pane)) =
                (digest_of(ir, "agent", address), agent_parent(ir, address))
            else {
                return false;
            };
            let backend_id = resolve(&pane, &applied.pane_ids, action.backend_id.as_deref());
            managed.resources.insert(
                address.to_owned(),
                ManagedResource {
                    kind: "agent".into(),
                    backend_id,
                    parent: Some(pane),
                    digest: digest.to_owned(),
                    label: None,
                    cwd: None,
                    adopted: None,
                    command_started: None,
                    last_outcome: None,
                },
            );
            true
        }
        // The following split may be blocked by its startup hook. Persist
        // that the old pane is gone even when its replacement never starts.
        Action::Core(CoreAction::ClosePane) => managed.resources.remove(address).is_some(),
        _ => false,
    }
}

fn group_topology_digest<'a>(ir: &'a Ir, id: &str) -> Option<&'a str> {
    ir.placements
        .iter()
        .find(|group| group.id == id)
        .map(|group| group.topology_digest.as_str())
}

fn command_started(
    ir: &Ir,
    pane: &str,
    applied: &ApplyState,
    existing: &ManagedProfile,
) -> Option<bool> {
    if pane_command(ir, pane).is_empty() {
        return None;
    }
    Some(
        applied
            .command_started
            .get(pane)
            .copied()
            .or_else(|| {
                existing
                    .resources
                    .get(pane)
                    .and_then(|resource| resource.command_started)
            })
            .unwrap_or(true),
    )
}

fn group_workspace<'a>(ir: &'a Ir, id: &str) -> Option<&'a str> {
    ir.placements
        .iter()
        .find(|group| group.id == id)
        .map(|group| group.workspace.as_str())
}

/// Routes one planned action to the backend through a single exhaustive
/// match (D29). A `Herdr(..)`/`Radiator(..)` action whose accessor returns
/// `None` yields [`Outcome::Unsupported`] without touching the backend.
fn apply_action(
    backend: &dyn Backend,
    ir: &Ir,
    state: &mut ApplyState,
    action: &PlannedAction,
) -> Result<Outcome> {
    match action.kind {
        Action::Core(core) => apply_core(backend, ir, state, core, action),
        Action::Herdr(herdr) => {
            let Some(ext) = backend.herdr() else {
                return Ok(Outcome::Unsupported {
                    flavor: "herdr",
                    action: action.kind,
                });
            };
            apply_herdr(backend, ext, ir, state, herdr, action)
        }
        // `RadiatorAction` is empty (spec §8, D37); this arm keeps the match
        // exhaustive so adding a variant forces every backend to answer it.
        Action::Radiator(radiator) => match radiator {},
    }
}

fn apply_core(
    backend: &dyn Backend,
    ir: &Ir,
    state: &mut ApplyState,
    core: CoreAction,
    action: &PlannedAction,
) -> Result<Outcome> {
    match core {
        CoreAction::CreateWorkspace => {
            let fields = resource_fields(ir, "workspace", &action.address)?;
            let label = string_field(fields, "label").unwrap_or_else(|| action.address.clone());
            let cwd = string_field(fields, "cwd").unwrap_or_else(|| ".".to_owned());
            let id = backend.create_workspace(&label, Path::new(&cwd))?;
            state.workspace_ids.insert(action.address.clone(), id);
            state.created_workspaces.insert(action.address.clone());
            Ok(Outcome::Applied)
        }
        CoreAction::RenameWorkspace => {
            let id = backend_id(state.workspace_ids.get(&action.address), action)?;
            let fields = resource_fields(ir, "workspace", &action.address)?;
            let label = string_field(fields, "label").unwrap_or_else(|| action.address.clone());
            backend.rename_workspace(&id, &label)?;
            Ok(Outcome::Applied)
        }
        CoreAction::CreatePane => {
            let (workspace_id, mut spec) = pane_create_inputs(ir, state, &action.address)?;
            let command = backend
                .herdr()
                .is_some()
                .then(|| spec.command.take())
                .flatten();
            let pane_id = backend.create_pane(&workspace_id, &spec)?;
            state
                .pane_ids
                .insert(action.address.clone(), pane_id.clone());
            state.created_panes.insert(action.address.clone());
            if let Some(argv) = command {
                state.command_started.insert(action.address.clone(), false);
                backend.start_command(&pane_id, &argv)?;
                state.command_started.insert(action.address.clone(), true);
            }
            Ok(Outcome::Applied)
        }
        CoreAction::ClosePane => {
            let id = backend_id(action.backend_id.as_ref(), action)?;
            backend.close_pane(&id)?;
            Ok(Outcome::Applied)
        }
        CoreAction::RenamePane => {
            let id = backend_id(action.backend_id.as_ref(), action)?;
            let fields = resource_fields(ir, "pane", &action.address)?;
            let label = string_field(fields, "label").unwrap_or_else(|| action.address.clone());
            backend.rename_pane(&id, &label)?;
            Ok(Outcome::Applied)
        }
        CoreAction::RestartCommand => {
            let id = backend_id(action.backend_id.as_ref(), action)?;
            let argv = pane_command(ir, &action.address);
            backend.restart_command(&id, &argv)?;
            if !argv.is_empty() {
                state.command_started.insert(action.address.clone(), true);
            }
            Ok(Outcome::Applied)
        }
        CoreAction::PromptAgent => {
            let id = backend_id(action.backend_id.as_ref(), action)?;
            let fields = resource_fields(ir, "agent", &action.address)?;
            if let Some(prompt) = string_field(fields, "prompt") {
                backend.prompt_agent(&id, &prompt)?;
            }
            Ok(Outcome::Applied)
        }
        // Adoption records ownership of the caller pane (D24); it needs no
        // backend verb. Detach, RunTask and Conflict are handled outside the
        // backend apply loop (local state, `execute_plan_tasks`, reporting).
        CoreAction::AdoptPane | CoreAction::Detach | CoreAction::RunTask | CoreAction::Conflict => {
            Ok(Outcome::Skipped)
        }
    }
}

fn apply_herdr(
    backend: &dyn Backend,
    ext: &dyn crate::backend::HerdrExt,
    ir: &Ir,
    state: &mut ApplyState,
    herdr: HerdrAction,
    action: &PlannedAction,
) -> Result<Outcome> {
    match herdr {
        HerdrAction::CreateTab => {
            let group = placement_group(ir, &action.address)?;
            let workspace_id = backend_id(state.workspace_ids.get(&group.workspace), action)?;
            // A fresh group plans one `CreateTab` and no per-pane splits, so
            // this builds the whole tab: every declared pane, then the ratios.
            let mut commands = BTreeMap::new();
            let specs = group
                .panes
                .iter()
                .map(|pane| {
                    let mut spec = pane_spec(ir, pane)?;
                    if let Some(command) = spec.command.take() {
                        commands.insert(pane.clone(), command);
                    }
                    Ok(spec)
                })
                .collect::<Result<Vec<_>>>()?;
            // The root Herdr tab id is only ever present for a workspace
            // this same apply created, and only until the first `CreateTab`
            // for it consumes it (D49) — an adopted or pre-existing
            // workspace never has one.
            let existing_tab = ext.take_root_tab(&workspace_id);
            let layout = ext.create_tab(
                &workspace_id,
                &group.label,
                group.split,
                &group.ratios,
                &specs,
                existing_tab.as_deref(),
            )?;
            state
                .group_ids
                .insert(action.address.clone(), layout.tab_id);
            state.created_groups.insert(action.address.clone());
            for (pane, pane_id) in group.panes.iter().zip(layout.pane_ids) {
                state.pane_ids.insert(pane.clone(), pane_id);
                state.created_panes.insert(pane.clone());
                if commands.contains_key(pane) {
                    state.command_started.insert(pane.clone(), false);
                }
            }
            for pane in &group.panes {
                if let Some(argv) = commands.get(pane) {
                    let pane_id = state
                        .pane_ids
                        .get(pane)
                        .ok_or_else(|| anyhow::anyhow!("pane `{pane}` has no backend id yet"))?;
                    backend.start_command(pane_id, argv)?;
                    state.command_started.insert(pane.clone(), true);
                }
            }
            Ok(Outcome::Applied)
        }
        HerdrAction::RenameTab => {
            let group = placement_group(ir, &action.address)?;
            let tab_id = backend_id(state.group_ids.get(&action.address), action)?;
            ext.rename_tab(&tab_id, &group.label)?;
            Ok(Outcome::Applied)
        }
        HerdrAction::SetRatio => {
            let group = placement_group(ir, &action.address)?;
            let tab_id = backend_id(state.group_ids.get(&action.address), action)?;
            ext.set_ratio(&tab_id, &group.ratios)?;
            Ok(Outcome::Applied)
        }
        HerdrAction::SplitPane => {
            let (_workspace_id, mut spec) = pane_create_inputs(ir, state, &action.address)?;
            let command = spec.command.take();
            let group_id = pane_group_id(ir, &action.address)?;
            let tab_id = backend_id(state.group_ids.get(&group_id), action)?;
            let group = placement_group(ir, &group_id)?;
            let pane_id = ext.split_pane(&tab_id, &spec, group.split)?;
            state
                .pane_ids
                .insert(action.address.clone(), pane_id.clone());
            state.created_panes.insert(action.address.clone());
            if let Some(argv) = command {
                state.command_started.insert(action.address.clone(), false);
                backend.start_command(&pane_id, &argv)?;
                state.command_started.insert(action.address.clone(), true);
            }
            Ok(Outcome::Applied)
        }
        HerdrAction::StartAgent => {
            let fields = resource_fields(ir, "agent", &action.address)?;
            let pane = agent_parent(ir, &action.address)?;
            let pane_id = backend_id(state.pane_ids.get(&pane), action)?;
            let kind = string_field(fields, "kind").unwrap_or_default();
            let args = string_array(fields, "args");
            ext.start_agent(&pane_id, &action.address, &kind, &args)?;
            Ok(Outcome::Applied)
        }
    }
}

fn resource_fields<'a>(ir: &'a Ir, kind: &str, name: &str) -> Result<&'a serde_json::Value> {
    ir.resources
        .iter()
        .find(|resource| resource.kind == kind && resource.name == name)
        .map(|resource| &resource.fields)
        .ok_or_else(|| anyhow::anyhow!("no {kind} resource named `{name}` in the IR"))
}

fn placement_group<'a>(ir: &'a Ir, id: &str) -> Result<&'a crate::ir::PlacementGroup> {
    ir.placements
        .iter()
        .find(|group| group.id == id)
        .ok_or_else(|| anyhow::anyhow!("no placement group `{id}` in the IR"))
}

fn pane_group_id(ir: &Ir, pane: &str) -> Result<String> {
    ir.resources
        .iter()
        .find(|resource| resource.kind == "pane" && resource.name == pane)
        .and_then(|resource| resource.parent.clone())
        .ok_or_else(|| anyhow::anyhow!("pane `{pane}` has no placement group"))
}

fn agent_parent(ir: &Ir, agent: &str) -> Result<String> {
    ir.resources
        .iter()
        .find(|resource| resource.kind == "agent" && resource.name == agent)
        .and_then(|resource| resource.parent.clone())
        .ok_or_else(|| anyhow::anyhow!("agent `{agent}` has no pane"))
}

fn pane_create_inputs(ir: &Ir, state: &ApplyState, pane: &str) -> Result<(String, PaneSpec)> {
    let group_id = pane_group_id(ir, pane)?;
    let group = placement_group(ir, &group_id)?;
    let workspace_id = state
        .workspace_ids
        .get(&group.workspace)
        .cloned()
        .ok_or_else(|| anyhow::anyhow!("workspace `{}` has no backend id yet", group.workspace))?;
    Ok((workspace_id, pane_spec(ir, pane)?))
}

/// The [`PaneSpec`] for one declared pane: its label, cwd, command and env,
/// independent of any workspace backend id (used when building the panes of a
/// fresh Herdr tab up front, before their splits run).
fn pane_spec(ir: &Ir, pane: &str) -> Result<PaneSpec> {
    let fields = resource_fields(ir, "pane", pane)?;
    Ok(PaneSpec {
        label: string_field(fields, "label").or_else(|| Some(pane.to_owned())),
        cwd: string_field(fields, "cwd").map(std::path::PathBuf::from),
        command: {
            let argv = pane_command(ir, pane);
            (!argv.is_empty()).then_some(argv)
        },
        env: string_map(fields, "env"),
    })
}

/// The first `serve` candidate's argv (D8: `any_of` tries them in order; the
/// backend runs the first).
fn pane_command(ir: &Ir, pane: &str) -> Vec<String> {
    let Ok(fields) = resource_fields(ir, "pane", pane) else {
        return Vec::new();
    };
    fields
        .get("serve")
        .and_then(|v| v.as_array())
        .and_then(|candidates| candidates.first())
        .and_then(|v| v.as_array())
        .map(|argv| {
            argv.iter()
                .filter_map(|v| v.as_str().map(ToOwned::to_owned))
                .collect()
        })
        .unwrap_or_default()
}

fn backend_id(id: Option<&String>, action: &PlannedAction) -> Result<String> {
    id.cloned()
        .ok_or_else(|| anyhow::anyhow!("no backend id for `{}` yet", action.address))
}

fn string_field(fields: &serde_json::Value, key: &str) -> Option<String> {
    fields
        .get(key)
        .and_then(|v| v.as_str())
        .map(ToOwned::to_owned)
}

fn string_array(fields: &serde_json::Value, key: &str) -> Vec<String> {
    fields
        .get(key)
        .and_then(|v| v.as_array())
        .map(|values| {
            values
                .iter()
                .filter_map(|v| v.as_str().map(ToOwned::to_owned))
                .collect()
        })
        .unwrap_or_default()
}

fn string_map(fields: &serde_json::Value, key: &str) -> BTreeMap<String, String> {
    fields
        .get(key)
        .and_then(|v| v.as_object())
        .map(|object| {
            object
                .iter()
                .filter_map(|(k, v)| v.as_str().map(|v| (k.clone(), v.to_owned())))
                .collect()
        })
        .unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use std::{fs, path::PathBuf, sync::Mutex};

    use anyhow::bail;
    use serde_json::json;

    use super::*;
    use crate::planner::{Snapshot, build_plan};

    type Call = (Vec<String>, BTreeMap<String, String>);

    #[derive(Default)]
    struct FakeRunner {
        /// argv joined with a space -> whether it should succeed.
        outcomes: Mutex<BTreeMap<String, bool>>,
        calls: Mutex<Vec<Call>>,
    }

    impl FakeRunner {
        fn succeed(self, argv: &[&str]) -> Self {
            self.outcomes
                .lock()
                .expect("outcomes mutex")
                .insert(argv.join(" "), true);
            self
        }

        fn fail(self, argv: &[&str]) -> Self {
            self.outcomes
                .lock()
                .expect("outcomes mutex")
                .insert(argv.join(" "), false);
            self
        }

        fn calls(&self) -> Vec<Vec<String>> {
            self.calls
                .lock()
                .expect("calls mutex")
                .iter()
                .map(|(argv, _)| argv.clone())
                .collect()
        }

        fn envs_for(&self, argv: &[&str]) -> Option<BTreeMap<String, String>> {
            let key = argv.join(" ");
            self.calls
                .lock()
                .expect("calls mutex")
                .iter()
                .find(|(call, _)| call.join(" ") == key)
                .map(|(_, env)| env.clone())
        }
    }

    impl CommandRunner for FakeRunner {
        fn run(
            &self,
            argv: &[String],
            _cwd: &Path,
            env: &BTreeMap<String, String>,
        ) -> Result<bool> {
            self.calls
                .lock()
                .expect("calls mutex")
                .push((argv.to_vec(), env.clone()));
            Ok(*self
                .outcomes
                .lock()
                .expect("outcomes mutex")
                .get(&argv.join(" "))
                .unwrap_or(&true))
        }
    }

    /// Builds a `LocalState` pointed at a fresh temp file directly, rather
    /// than through `LocalState::load`'s env-var-based directory lookup:
    /// that lookup is process-global, and mutating it from a test would
    /// need `std::env::set_var`, which this crate denies (`unsafe_code =
    /// "deny"`) and which would race other tests regardless.
    fn temp_state() -> (LocalState, tempfile::TempDir) {
        let dir = tempfile::tempdir().expect("tempdir");
        let state = LocalState {
            schema_version: 1,
            repo_root: PathBuf::from("/repo"),
            profiles: BTreeMap::new(),
            approvals: BTreeSet::new(),
            journal: Vec::new(),
            path: dir.path().join("state.json"),
        };
        (state, dir)
    }

    fn profile_from(value: serde_json::Value) -> Profile {
        serde_json::from_value(value).expect("profile fixture")
    }

    fn ctx<'a>(root: &'a Path, runner: &'a dyn CommandRunner) -> ExecutionContext<'a> {
        ExecutionContext {
            repo_root: root,
            profile: "default",
            runner,
        }
    }

    #[test]
    fn check_passing_skips_run() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default().succeed(&["check"]);
        let task = Task {
            name: "scaffold".into(),
            run: vec!["run".into()],
            check: Some(vec!["check".into()]),
            inputs: vec![],
            after: vec![],
            auto: true,
            on_start: None,
            on_stop: None,
        };
        let root = PathBuf::from("/repo");
        let outcome =
            run_task(&task, "digest-1", &ctx(&root, &runner), &mut state, false).expect("run_task");
        assert_eq!(outcome, TaskOutcome::Skipped);
        assert_eq!(runner.calls(), vec![vec!["check".to_owned()]]);
        assert_eq!(
            state
                .profile("default")
                .expect("profile recorded")
                .resources
                .get("scaffold")
                .expect("scaffold recorded")
                .last_outcome
                .as_deref(),
            Some("skipped")
        );
    }

    #[test]
    fn failing_check_runs_the_task_when_approved() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default().fail(&["check"]).succeed(&["run"]);
        let task = Task {
            name: "scaffold".into(),
            run: vec!["run".into()],
            check: Some(vec!["check".into()]),
            inputs: vec![],
            after: vec![],
            auto: true,
            on_start: None,
            on_stop: None,
        };
        let root = PathBuf::from("/repo");
        let outcome =
            run_task(&task, "digest-1", &ctx(&root, &runner), &mut state, true).expect("run_task");
        assert_eq!(outcome, TaskOutcome::Ran(true));
        assert_eq!(
            runner.calls(),
            vec![vec!["check".to_owned()], vec!["run".to_owned()]]
        );
    }

    #[test]
    fn a_failed_run_is_not_recorded_as_converged() {
        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default",
            "tasks": [{"name": "scaffold", "run": ["run"]}]
        }));
        let digest = digest_of(&profile.to_ir(), "task", "scaffold")
            .expect("scaffold resource")
            .to_owned();
        let runner = FakeRunner::default().fail(&["run"]);
        let root = PathBuf::from("/repo");
        let outcome = run_task(
            &profile.tasks[0],
            &digest,
            &ctx(&root, &runner),
            &mut state,
            true,
        )
        .expect("run_task");
        assert_eq!(outcome, TaskOutcome::Ran(false));

        // D18: a failed `run` must not look converged to the next
        // `build_plan` — recording `scaffold`'s real IR digest as observed
        // here would be a false convergence.
        let snapshot = state
            .profile("default")
            .expect("profile recorded")
            .to_snapshot("default", None);
        let plan = build_plan(&profile, &snapshot).expect("plan");
        assert!(
            plan.actions
                .iter()
                .any(|action| action.address == "scaffold"),
            "a failed task must still be proposed to run again: {plan:?}"
        );
    }

    #[test]
    fn unapproved_task_is_blocked() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let task = Task {
            name: "scaffold".into(),
            run: vec!["run".into()],
            check: None,
            inputs: vec![],
            after: vec![],
            auto: true,
            on_start: None,
            on_stop: None,
        };
        let root = PathBuf::from("/repo");
        let outcome =
            run_task(&task, "digest-1", &ctx(&root, &runner), &mut state, false).expect("run_task");
        assert_eq!(outcome, TaskOutcome::Blocked);
        assert!(runner.calls().is_empty(), "blocked task must not run");
    }

    #[test]
    fn approving_once_covers_a_later_unattended_run() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let task = Task {
            name: "scaffold".into(),
            run: vec!["run".into()],
            check: None,
            inputs: vec![],
            after: vec![],
            auto: true,
            on_start: None,
            on_stop: None,
        };
        let root = PathBuf::from("/repo");
        run_task(&task, "digest-1", &ctx(&root, &runner), &mut state, true).expect("first run");
        let second = run_task(&task, "digest-1", &ctx(&root, &runner), &mut state, false)
            .expect("second run");
        assert_eq!(second, TaskOutcome::Ran(true));
    }

    #[test]
    fn task_on_start_hook_fires_after_run() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let task = Task {
            name: "scaffold".into(),
            run: vec!["run".into()],
            check: None,
            inputs: vec![],
            after: vec![],
            auto: true,
            on_start: Some(vec!["notify".into()]),
            on_stop: None,
        };
        let root = PathBuf::from("/repo");
        run_task(&task, "digest-1", &ctx(&root, &runner), &mut state, true).expect("run");
        assert_eq!(
            runner.calls(),
            vec![vec!["run".to_owned()], vec!["notify".to_owned()]]
        );
        let env = runner.envs_for(&["notify"]).expect("hook env recorded");
        assert_eq!(env.get("DROVE_RESOURCE"), Some(&"scaffold".to_owned()));
    }

    #[test]
    fn hook_reports_backend_id_in_env() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");
        run_hook(
            &["notify".into()],
            "gitlog",
            Some("w1:p2"),
            HookEvent::Stop,
            &ctx(&root, &runner),
            &mut state,
            true,
        )
        .expect("hook");
        let env = runner.envs_for(&["notify"]).expect("env recorded");
        assert_eq!(env.get("DROVE_RESOURCE"), Some(&"gitlog".to_owned()));
        assert_eq!(env.get("DROVE_BACKEND_ID"), Some(&"w1:p2".to_owned()));
    }

    #[test]
    fn run_named_task_runs_only_target_and_its_prerequisites() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let profile = profile_from(json!({
            "name": "default",
            "tasks": [
                {"name": "a", "run": ["run-a"]},
                {"name": "b", "run": ["run-b"], "after": ["a"]},
                {"name": "unrelated", "run": ["run-unrelated"]}
            ]
        }));
        let root = PathBuf::from("/repo");
        let results =
            run_named_task(&profile, "b", &ctx(&root, &runner), &mut state, true).expect("run");
        assert_eq!(
            results.iter().map(|(n, _)| n.clone()).collect::<Vec<_>>(),
            vec!["a".to_owned(), "b".to_owned()]
        );
        assert_eq!(
            runner.calls(),
            vec![vec!["run-a".to_owned()], vec!["run-b".to_owned()]]
        );
    }

    #[test]
    fn run_named_task_rejects_unknown_name() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let profile = profile_from(json!({"name": "default", "tasks": []}));
        let root = PathBuf::from("/repo");
        let error = run_named_task(&profile, "missing", &ctx(&root, &runner), &mut state, true)
            .expect_err("unknown task");
        assert!(error.to_string().contains("no task named"));
    }

    #[test]
    fn run_named_task_skips_a_task_whose_prerequisite_failed() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default().fail(&["run-a"]);
        let profile = profile_from(json!({
            "name": "default",
            "tasks": [
                {"name": "a", "run": ["run-a"]},
                {"name": "b", "run": ["run-b"], "after": ["a"]}
            ]
        }));
        let root = PathBuf::from("/repo");
        let results =
            run_named_task(&profile, "b", &ctx(&root, &runner), &mut state, true).expect("run");

        assert_eq!(
            results,
            vec![
                ("a".to_owned(), TaskOutcome::Ran(false)),
                ("b".to_owned(), TaskOutcome::DependencySkipped),
            ]
        );
        // `b`'s own `run` must never have been invoked.
        assert_eq!(runner.calls(), vec![vec!["run-a".to_owned()]]);
    }

    #[test]
    fn list_tasks_reports_last_outcome() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let profile = profile_from(json!({
            "name": "default",
            "tasks": [{"name": "scaffold", "run": ["run"]}, {"name": "never-run", "run": ["run"]}]
        }));
        let root = PathBuf::from("/repo");
        run_named_task(&profile, "scaffold", &ctx(&root, &runner), &mut state, true).expect("run");
        let listed = list_tasks(&profile, &state);
        assert_eq!(
            listed,
            vec![
                ("scaffold".to_owned(), Some("ok".to_owned())),
                ("never-run".to_owned(), None),
            ]
        );
    }

    #[test]
    fn execute_plan_tasks_runs_ready_tasks_from_the_plan() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default();
        let profile = profile_from(json!({
            "name": "default",
            "tasks": [{"name": "scaffold", "run": ["run"]}]
        }));
        let plan = build_plan(&profile, &Snapshot::default()).expect("plan");
        let root = PathBuf::from("/repo");
        let (results, skipped) =
            execute_plan_tasks(&profile, &plan, &ctx(&root, &runner), &mut state, true)
                .expect("execute");
        assert_eq!(
            results,
            vec![("scaffold".to_owned(), TaskOutcome::Ran(true))]
        );
        assert!(skipped.is_empty());
    }

    #[test]
    fn execute_plan_tasks_skips_a_task_after_a_failed_prerequisite() {
        let (mut state, _dir) = temp_state();
        let runner = FakeRunner::default().fail(&["run-a"]);
        let profile = profile_from(json!({
            "name": "default",
            "tasks": [
                {"name": "a", "run": ["run-a"]},
                {"name": "b", "run": ["run-b"], "after": ["a"]}
            ]
        }));
        let plan = build_plan(&profile, &Snapshot::default()).expect("plan");
        let root = PathBuf::from("/repo");

        let (results, skipped) =
            execute_plan_tasks(&profile, &plan, &ctx(&root, &runner), &mut state, true)
                .expect("execute");

        assert_eq!(
            results,
            vec![
                ("a".to_owned(), TaskOutcome::Ran(false)),
                ("b".to_owned(), TaskOutcome::DependencySkipped),
            ]
        );
        assert_eq!(skipped.len(), 1);
        assert_eq!(skipped[0].address, "b");
        assert_eq!(skipped[0].depends_on, "a");
        // `b`'s own `run` must never have been invoked.
        assert_eq!(runner.calls(), vec![vec!["run-a".to_owned()]]);
    }

    fn down_test_profile() -> Profile {
        profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [{
                    "name": "main",
                    "panes": [{"name": "gitlog", "serve": [["lazygit"]], "on_stop": ["notify-stop"]}]
                }]
            }]
        }))
    }

    fn seed_managed(state: &mut LocalState, entries: &[(&str, &str, Option<&str>)]) {
        let profile = state.profile_mut("default");
        for (id, kind, parent) in entries {
            profile.resources.insert(
                (*id).to_owned(),
                ManagedResource {
                    kind: (*kind).to_owned(),
                    backend_id: format!("backend-{id}"),
                    parent: parent.map(str::to_owned),
                    digest: "any-digest".into(),
                    label: None,
                    cwd: None,
                    adopted: None,
                    command_started: None,
                    last_outcome: None,
                },
            );
        }
        state.save().expect("save seeded state");
    }

    #[test]
    fn down_runs_on_stop_before_detaching_and_leaves_unmanaged_alone() {
        let (mut state, _dir) = temp_state();
        let profile = down_test_profile();
        seed_managed(
            &mut state,
            &[
                ("dev", "workspace", None),
                ("gitlog", "pane", Some("dev/main")),
            ],
        );

        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");
        let report = down(
            &profile,
            &ctx(&root, &runner),
            &mut state,
            true,
            false,
            None,
        )
        .expect("down");

        // Reverse dependency order: pane before workspace. The placement
        // group is derived, not a managed resource (D29), so it is not torn
        // down on its own.
        assert_eq!(report.detached, vec!["gitlog", "dev"]);
        assert_eq!(report.hooks_run, vec![("gitlog".to_owned(), true)]);
        assert_eq!(runner.calls(), vec![vec!["notify-stop".to_owned()]]);

        let managed = state.profile("default").expect("profile recorded");
        assert!(
            managed.resources.is_empty(),
            "every managed resource must be detached"
        );
    }

    #[test]
    fn down_never_touches_a_resource_it_never_managed() {
        let (mut state, _dir) = temp_state();
        let profile = down_test_profile();
        // Nothing seeded: an unmanaged pane the backend might report is
        // simply absent from local state, so `down` has nothing to iterate.
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");
        let report = down(
            &profile,
            &ctx(&root, &runner),
            &mut state,
            true,
            false,
            None,
        )
        .expect("down");
        assert!(report.detached.is_empty());
        assert!(runner.calls().is_empty());
    }

    #[test]
    fn down_blocks_on_stop_hook_without_approval() {
        let (mut state, _dir) = temp_state();
        let profile = down_test_profile();
        seed_managed(&mut state, &[("gitlog", "pane", Some("dev/main"))]);
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");
        let report = down(
            &profile,
            &ctx(&root, &runner),
            &mut state,
            false,
            false,
            None,
        )
        .expect("down");
        assert!(runner.calls().is_empty(), "blocked hook must not run");
        // The resource is still detached even though its hook was blocked:
        // down's job is to stop tracking it, not to force approval.
        assert_eq!(report.detached, vec!["gitlog"]);
        assert!(report.hooks_run.is_empty());
    }

    /// A backend with no Herdr flavor: `herdr()` is `None`, so every
    /// `Herdr(..)` action must come back `Unsupported`. It records the panes
    /// it is asked to create so the test can prove core actions still run.
    #[derive(Default)]
    struct FlavorlessBackend {
        created_panes: Mutex<Vec<String>>,
        created_commands: Mutex<Vec<Option<Vec<String>>>>,
        start_commands: Mutex<Vec<Vec<String>>>,
    }

    impl Backend for FlavorlessBackend {
        fn snapshot(&self) -> Result<crate::backend::herdr::SessionSnapshot> {
            Ok(Default::default())
        }
        fn caller_pane_id(&self) -> Option<String> {
            None
        }
        fn create_workspace(&self, _label: &str, _cwd: &Path) -> Result<String> {
            Ok("w1".into())
        }
        fn rename_workspace(&self, _id: &str, _label: &str) -> Result<()> {
            Ok(())
        }
        fn create_pane(&self, _workspace_id: &str, spec: &PaneSpec) -> Result<String> {
            let name = spec.label.clone().unwrap_or_default();
            self.created_panes.lock().expect("mutex").push(name);
            self.created_commands
                .lock()
                .expect("mutex")
                .push(spec.command.clone());
            Ok("p1".into())
        }
        fn close_pane(&self, _id: &str) -> Result<()> {
            Ok(())
        }
        fn rename_pane(&self, _id: &str, _label: &str) -> Result<()> {
            Ok(())
        }
        fn start_command(&self, _id: &str, argv: &[String]) -> Result<()> {
            self.start_commands
                .lock()
                .expect("mutex")
                .push(argv.to_vec());
            Ok(())
        }
        fn restart_command(&self, _id: &str, _argv: &[String]) -> Result<()> {
            Ok(())
        }
        fn prompt_agent(&self, _id: &str, _prompt: &str) -> Result<()> {
            Ok(())
        }
        fn process_info(&self, _id: &str) -> Result<Option<crate::backend::ProcessInfo>> {
            Ok(None)
        }
        fn report_tokens(&self, _address: &str, _tokens: &BTreeMap<String, String>) -> Result<()> {
            Ok(())
        }
        fn output(&self, _id: &str, _timeout: std::time::Duration) -> Result<String> {
            Ok(String::new())
        }
        fn capabilities(&self) -> crate::backend::Capabilities {
            crate::backend::Capabilities {
                workspace_env: false,
                pane_command_at_create: true,
                metadata_tokens: false,
                process_info: false,
                events: false,
                readiness_output: false,
            }
        }
        // No `herdr()` override: it inherits the default `None`.
    }

    /// A backend whose `close_pane` fails for one chosen backend id and
    /// succeeds for every other (D50), so a `down --purge` test can prove a
    /// lost pane no longer aborts the teardown.
    struct CloseFailsBackend {
        fails_for: &'static str,
    }

    impl Backend for CloseFailsBackend {
        fn snapshot(&self) -> Result<crate::backend::herdr::SessionSnapshot> {
            Ok(Default::default())
        }
        fn caller_pane_id(&self) -> Option<String> {
            None
        }
        fn create_workspace(&self, _label: &str, _cwd: &Path) -> Result<String> {
            Ok("w1".into())
        }
        fn rename_workspace(&self, _id: &str, _label: &str) -> Result<()> {
            Ok(())
        }
        fn create_pane(&self, _workspace_id: &str, _spec: &PaneSpec) -> Result<String> {
            Ok("p1".into())
        }
        fn close_pane(&self, id: &str) -> Result<()> {
            if id == self.fails_for {
                bail!("pane_not_found: {id}");
            }
            Ok(())
        }
        fn rename_pane(&self, _id: &str, _label: &str) -> Result<()> {
            Ok(())
        }
        fn restart_command(&self, _id: &str, _argv: &[String]) -> Result<()> {
            Ok(())
        }
        fn prompt_agent(&self, _id: &str, _prompt: &str) -> Result<()> {
            Ok(())
        }
        fn process_info(&self, _id: &str) -> Result<Option<crate::backend::ProcessInfo>> {
            Ok(None)
        }
        fn report_tokens(&self, _address: &str, _tokens: &BTreeMap<String, String>) -> Result<()> {
            Ok(())
        }
        fn output(&self, _id: &str, _timeout: std::time::Duration) -> Result<String> {
            Ok(String::new())
        }
        fn capabilities(&self) -> crate::backend::Capabilities {
            crate::backend::Capabilities {
                workspace_env: false,
                pane_command_at_create: true,
                metadata_tokens: false,
                process_info: false,
                events: false,
                readiness_output: false,
            }
        }
    }

    #[test]
    fn down_collects_a_close_pane_failure_without_aborting_the_teardown() {
        let (mut state, _dir) = temp_state();
        let profile = down_test_profile();
        seed_managed(
            &mut state,
            &[
                ("dev", "workspace", None),
                ("gitlog", "pane", Some("dev/main")),
            ],
        );

        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");
        let backend = CloseFailsBackend {
            fails_for: "backend-gitlog",
        };
        let report = down(
            &profile,
            &ctx(&root, &runner),
            &mut state,
            true,
            true,
            Some(&backend),
        )
        .expect("down");

        assert_eq!(report.detached, vec!["gitlog", "dev"]);
        assert_eq!(report.close_failed.len(), 1);
        let (id, message) = &report.close_failed[0];
        assert_eq!(id, "gitlog");
        assert!(message.contains("pane_not_found"));

        let managed = state.profile("default").expect("profile recorded");
        assert!(
            managed.resources.is_empty(),
            "every managed resource must still be detached and saved"
        );
    }

    #[test]
    fn core_create_pane_preserves_direct_command_create_for_non_herdr_backends() {
        let profile = up_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Core(CoreAction::CreatePane), "editor"),
        ]);
        let backend = FlavorlessBackend::default();
        let outcomes = apply_plan(&backend, &ir, &plan);

        assert_eq!(outcomes[1].1, Outcome::Applied);
        assert_eq!(
            *backend.created_commands.lock().expect("mutex"),
            vec![Some(vec!["bash".to_owned()])],
            "non-Herdr backends keep structured create-time argv"
        );
        assert!(
            backend.start_commands.lock().expect("mutex").is_empty(),
            "non-Herdr first launch must not fall through to typed restart"
        );
    }

    /// A fake Herdr for the `up` flow (D43): it hands out backend ids for
    /// every workspace, Herdr tab and pane it is asked to create, records the
    /// verbs it receives, and answers `ensure_session` with a state the test
    /// sets.
    struct RecordingHerdr {
        session: SessionState,
        calls: Mutex<Vec<String>>,
        next_id: Mutex<u32>,
        /// When set, every workspace this fake creates is given a root
        /// Herdr tab id (`<workspace>-root`), simulating what a real
        /// `workspace.create` always returns alongside it (D49).
        auto_root_tab: bool,
        root_tabs: Mutex<BTreeMap<String, String>>,
        /// Calls (matched against the same string [`RecordingHerdr::record`]
        /// logs) that fail instead of succeeding, so a test can prove `up`
        /// collects one action's error and keeps applying the rest (D52).
        fail_calls: BTreeSet<String>,
    }

    impl RecordingHerdr {
        fn running() -> Self {
            Self {
                session: SessionState::Running,
                calls: Mutex::new(Vec::new()),
                next_id: Mutex::new(1),
                auto_root_tab: false,
                root_tabs: Mutex::new(BTreeMap::new()),
                fail_calls: BTreeSet::new(),
            }
        }

        /// Like [`RecordingHerdr::running`], but the given calls fail
        /// instead of succeeding (D52). A failing `create_workspace`/
        /// `create_tab` still records the call and consumes an id, matching
        /// what recording-then-failing looks like against a real backend.
        fn running_failing(fail_calls: &[&'static str]) -> Self {
            Self {
                fail_calls: fail_calls.iter().map(|call| (*call).to_owned()).collect(),
                ..Self::running()
            }
        }

        fn running_failing_owned(fail_calls: impl IntoIterator<Item = String>) -> Self {
            Self {
                fail_calls: fail_calls.into_iter().collect(),
                ..Self::running()
            }
        }

        /// Like [`RecordingHerdr::running`], but simulates Herdr's own
        /// behavior of always returning a root Herdr tab alongside a
        /// freshly created workspace (D49).
        fn running_with_root_tabs() -> Self {
            Self {
                auto_root_tab: true,
                ..Self::running()
            }
        }

        fn cannot_start(hint: &str) -> Self {
            Self {
                session: SessionState::CannotStart { hint: hint.into() },
                calls: Mutex::new(Vec::new()),
                next_id: Mutex::new(1),
                auto_root_tab: false,
                root_tabs: Mutex::new(BTreeMap::new()),
                fail_calls: BTreeSet::new(),
            }
        }

        fn id(&self, prefix: &str) -> String {
            let mut next = self.next_id.lock().expect("id lock");
            let id = format!("{prefix}{next}");
            *next += 1;
            id
        }

        fn record(&self, call: String) {
            self.calls.lock().expect("calls lock").push(call);
        }

        fn calls(&self) -> Vec<String> {
            self.calls.lock().expect("calls lock").clone()
        }
    }

    impl Backend for RecordingHerdr {
        fn snapshot(&self) -> Result<crate::backend::herdr::SessionSnapshot> {
            Ok(Default::default())
        }
        fn caller_pane_id(&self) -> Option<String> {
            None
        }
        fn create_workspace(&self, label: &str, _cwd: &Path) -> Result<String> {
            let call = format!("create_workspace:{label}");
            self.record(call.clone());
            if self.fail_calls.contains(&call) {
                bail!("boom: {call}");
            }
            let workspace_id = self.id("w");
            if self.auto_root_tab {
                self.root_tabs
                    .lock()
                    .expect("root tabs lock")
                    .insert(workspace_id.clone(), format!("{workspace_id}-root"));
            }
            Ok(workspace_id)
        }
        fn rename_workspace(&self, id: &str, label: &str) -> Result<()> {
            self.record(format!("rename_workspace:{id}:{label}"));
            Ok(())
        }
        fn create_pane(&self, workspace_id: &str, spec: &PaneSpec) -> Result<String> {
            let label = spec.label.clone().unwrap_or_default();
            self.record(format!("create_pane:{workspace_id}:{label}"));
            Ok(self.id("p"))
        }
        fn close_pane(&self, id: &str) -> Result<()> {
            self.record(format!("close_pane:{id}"));
            Ok(())
        }
        fn rename_pane(&self, id: &str, label: &str) -> Result<()> {
            self.record(format!("rename_pane:{id}:{label}"));
            Ok(())
        }
        fn start_command(&self, id: &str, _argv: &[String]) -> Result<()> {
            let call = format!("start_command:{id}");
            self.record(call.clone());
            if self.fail_calls.contains(&call) {
                bail!("boom: {call}");
            }
            Ok(())
        }
        fn restart_command(&self, id: &str, _argv: &[String]) -> Result<()> {
            let call = format!("restart_command:{id}");
            self.record(call.clone());
            if self.fail_calls.contains(&call) {
                bail!("boom: {call}");
            }
            Ok(())
        }
        fn prompt_agent(&self, id: &str, _prompt: &str) -> Result<()> {
            self.record(format!("prompt_agent:{id}"));
            Ok(())
        }
        fn process_info(&self, _id: &str) -> Result<Option<crate::backend::ProcessInfo>> {
            Ok(None)
        }
        fn report_tokens(&self, _address: &str, _tokens: &BTreeMap<String, String>) -> Result<()> {
            Ok(())
        }
        fn output(&self, _id: &str, _timeout: std::time::Duration) -> Result<String> {
            Ok(String::new())
        }
        fn capabilities(&self) -> crate::backend::Capabilities {
            crate::backend::Capabilities {
                workspace_env: true,
                pane_command_at_create: true,
                metadata_tokens: true,
                process_info: true,
                events: true,
                readiness_output: true,
            }
        }
        fn herdr(&self) -> Option<&dyn crate::backend::HerdrExt> {
            Some(self)
        }
    }

    impl crate::backend::HerdrExt for RecordingHerdr {
        fn create_tab(
            &self,
            workspace_id: &str,
            label: &str,
            _split: crate::backend::Split,
            ratios: &[f64],
            panes: &[PaneSpec],
            existing_tab: Option<&str>,
        ) -> Result<crate::backend::TabLayout> {
            self.record(format!(
                "create_tab:{workspace_id}:{label}:existing={existing_tab:?}"
            ));
            let fail_key = format!("create_tab:{label}");
            if self.fail_calls.contains(fail_key.as_str()) {
                bail!("boom: {fail_key}");
            }
            let tab_id = existing_tab.map_or_else(|| self.id("t"), ToOwned::to_owned);
            let pane_ids = panes
                .iter()
                .map(|spec| {
                    let pane_label = spec.label.clone().unwrap_or_default();
                    self.record(format!("tab_pane:{tab_id}:{pane_label}"));
                    self.id("p")
                })
                .collect();
            if !ratios.is_empty() {
                self.record(format!("set_ratio:{tab_id}"));
            }
            Ok(crate::backend::TabLayout { tab_id, pane_ids })
        }

        fn take_root_tab(&self, workspace_id: &str) -> Option<String> {
            self.root_tabs
                .lock()
                .expect("root tabs lock")
                .remove(workspace_id)
        }
        fn split_pane(
            &self,
            tab_id: &str,
            spec: &PaneSpec,
            _split: crate::backend::Split,
        ) -> Result<String> {
            let label = spec.label.clone().unwrap_or_default();
            self.record(format!("split_pane:{tab_id}:{label}"));
            Ok(self.id("p"))
        }
        fn set_ratio(&self, tab_id: &str, _ratios: &[f64]) -> Result<()> {
            self.record(format!("set_ratio:{tab_id}"));
            Ok(())
        }
        fn rename_tab(&self, tab_id: &str, label: &str) -> Result<()> {
            self.record(format!("rename_tab:{tab_id}:{label}"));
            Ok(())
        }
        fn start_agent(
            &self,
            pane_id: &str,
            name: &str,
            _kind: &str,
            _args: &[String],
        ) -> Result<()> {
            self.record(format!("start_agent:{pane_id}:{name}"));
            Ok(())
        }
        fn focus_workspace(&self, id: &str) -> Result<()> {
            self.record(format!("focus:{id}"));
            Ok(())
        }
        fn ensure_session(&self, _name: &str) -> Result<SessionState> {
            Ok(self.session.clone())
        }
        fn stop_session(&self, name: &str) -> Result<crate::backend::SessionStop> {
            self.record(format!("stop_session:{name}"));
            Ok(crate::backend::SessionStop {
                stopped: true,
                deleted: true,
            })
        }
    }

    fn up_profile() -> Profile {
        profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [{"name": "main", "panes": [{"name": "editor", "serve": [["bash"]]}]}]
            }]
        }))
    }

    fn up_plan(kinds: &[(Action, &str)]) -> Plan {
        use crate::planner::SyncStatus;
        Plan {
            profile: "default".into(),
            desired_digest: String::new(),
            status: SyncStatus::OutOfSync,
            adopted: BTreeMap::new(),
            actions: kinds
                .iter()
                .map(|(kind, address)| PlannedAction {
                    kind: *kind,
                    address: (*address).to_owned(),
                    backend_id: None,
                    destructive: false,
                    reason: String::new(),
                })
                .collect(),
        }
    }

    /// Shares the backend trace so assertions prove ordering across the host
    /// runner and the backend calls that can launch pane commands.
    struct PaneHookRunner<'a> {
        backend: &'a RecordingHerdr,
        runner: FakeRunner,
        error: bool,
    }

    impl CommandRunner for PaneHookRunner<'_> {
        fn run(&self, argv: &[String], cwd: &Path, env: &BTreeMap<String, String>) -> Result<bool> {
            if let Some(pane) = env.get("DROVE_RESOURCE") {
                self.backend.record(format!("hook:{pane}"));
            }
            if self.error {
                bail!("hook executable missing");
            }
            self.runner.run(argv, cwd, env)
        }
    }

    fn pane_hook_up(
        profile: &Profile,
        plan: &Plan,
        backend: &RecordingHerdr,
        runner: &dyn CommandRunner,
        state: &mut LocalState,
        approve: bool,
    ) -> UpReport {
        up(
            backend,
            profile,
            &profile.to_ir(),
            plan,
            &ctx(Path::new("/repo"), runner),
            state,
            approve,
            "test",
            None,
            false,
        )
        .expect("up")
    }

    fn pane_hook_plan(profile: &Profile, state: &LocalState) -> Plan {
        let snapshot = state
            .profile("default")
            .map(|managed| managed.to_snapshot("default", None))
            .unwrap_or_default();
        build_plan(profile, &snapshot).expect("plan")
    }

    #[test]
    fn pane_hooks_gate_fresh_tab_once_then_converged_up_and_hook_edit_do_not_run_them() {
        let (mut state, _dir) = temp_state();
        let mut profile = profile_from(json!({
            "name": "default", "workspaces": [{"name": "dev", "tabs": [{
                "name": "main", "panes": [
                    {"name": "editor", "serve": [["serve-editor"]], "on_start": ["register"]},
                    {"name": "tests", "serve": [["serve-tests"]], "on_start": ["register"]}
                ]
            }]}]
        }));
        let backend = RecordingHerdr::running();
        let runner = PaneHookRunner {
            backend: &backend,
            runner: FakeRunner::default(),
            error: false,
        };
        let report = pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &runner,
            &mut state,
            true,
        );
        assert!(report.failed.is_empty());
        let calls = backend.calls();
        assert_eq!(
            &calls[..4],
            [
                "create_workspace:dev",
                "hook:editor",
                "hook:tests",
                "create_tab:w1:main:existing=None"
            ]
        );
        assert_eq!(
            runner.runner.calls().len(),
            2,
            "same argv still runs for each pane identity"
        );
        for (_, env) in runner.runner.calls.lock().expect("calls").iter() {
            assert!(
                !env.contains_key("DROVE_BACKEND_ID"),
                "new pane has no id yet"
            );
        }
        assert_eq!(state.journal.len(), 2);
        assert!(
            state
                .journal
                .iter()
                .all(|entry| entry.completed && entry.success == Some(true))
        );
        let plan = pane_hook_plan(&profile, &state);
        assert!(plan.actions.is_empty());
        assert_eq!(
            pane_hook_up(&profile, &plan, &backend, &runner, &mut state, false).outcome,
            UpOutcome::AlreadyRunning
        );
        assert_eq!(backend.calls(), calls);

        profile.workspaces[0].tabs[0].panes[0].on_start = Some(vec!["unapproved-new-hook".into()]);
        let plan = pane_hook_plan(&profile, &state);
        assert_eq!(plan.actions[0].kind, Action::Core(CoreAction::RenamePane));
        assert!(
            pane_hook_up(&profile, &plan, &backend, &runner, &mut state, false)
                .failed
                .is_empty()
        );
        assert_eq!(runner.runner.calls().len(), 2);
        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn pane_hooks_gate_core_create_and_split_and_do_not_duplicate_within_apply() {
        let (mut state, _dir) = temp_state();
        let mut profile = up_profile();
        profile.workspaces[0].tabs[0].panes[0].on_start = Some(vec!["register".into()]);
        let backend = RecordingHerdr::running();
        let runner = PaneHookRunner {
            backend: &backend,
            runner: FakeRunner::default(),
            error: false,
        };
        let mut plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Core(CoreAction::CreatePane), "editor"),
            (Action::Core(CoreAction::RestartCommand), "editor"),
        ]);
        plan.actions[2].backend_id = Some("p2".into());
        assert!(
            pane_hook_up(&profile, &plan, &backend, &runner, &mut state, true)
                .failed
                .is_empty()
        );
        assert_eq!(
            backend.calls(),
            [
                "create_workspace:dev",
                "hook:editor",
                "create_pane:w1:editor",
                "start_command:p2",
                "restart_command:p2"
            ]
        );
        assert_eq!(runner.runner.calls().len(), 1);

        // A separate fresh group makes a recorded group to exercise an actual
        // planner-generated SplitPane when a pane is added on a later up.
        let (mut state, _dir) = temp_state();
        pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &runner,
            &mut state,
            true,
        );
        let mut pane = profile.workspaces[0].tabs[0].panes[0].clone();
        pane.name = "tests".into();
        profile.workspaces[0].tabs[0].panes.push(pane);
        let plan = pane_hook_plan(&profile, &state);
        assert!(
            plan.actions
                .iter()
                .any(|a| a.kind == Action::Herdr(HerdrAction::SplitPane))
        );
        let report = pane_hook_up(&profile, &plan, &backend, &runner, &mut state, false);
        assert!(
            report.failed.is_empty(),
            "recorded approval permits unattended hook"
        );
        let calls = backend.calls();
        let hook = calls
            .iter()
            .position(|call| call == "hook:tests")
            .expect("hook");
        assert!(calls[hook + 1].starts_with("split_pane:"));
        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn pane_hook_restart_failures_preserve_digest_and_id_and_retry_before_command() {
        for mode in ["blocked", "failed", "error"] {
            let (mut state, _dir) = temp_state();
            let mut profile = up_profile();
            let backend = RecordingHerdr::running();
            pane_hook_up(
                &profile,
                &pane_hook_plan(&profile, &state),
                &backend,
                &FakeRunner::default(),
                &mut state,
                true,
            );
            let original = state.profile("default").expect("managed").resources["editor"].clone();
            profile.workspaces[0].tabs[0].panes[0].serve = vec![vec!["new-serve".into()]];
            profile.workspaces[0].tabs[0].panes[0].on_start = Some(vec!["register".into()]);
            let runner = PaneHookRunner {
                backend: &backend,
                runner: FakeRunner::default().fail(&["register"]),
                error: mode == "error",
            };
            let plan = pane_hook_plan(&profile, &state);
            assert_eq!(
                plan.actions[0].kind,
                Action::Core(CoreAction::RestartCommand)
            );
            let report = pane_hook_up(
                &profile,
                &plan,
                &backend,
                &runner,
                &mut state,
                mode != "blocked",
            );
            assert_eq!(report.failed.len(), 1, "{mode}");
            assert!(report.failed[0].error.contains("on_start"));
            assert!(
                !backend
                    .calls()
                    .iter()
                    .any(|call| call.starts_with("restart_command:"))
            );
            let managed = &state.profile("default").expect("managed").resources["editor"];
            assert_eq!(managed.digest, original.digest);
            assert_eq!(managed.backend_id, original.backend_id);
            match mode {
                "blocked" => assert!(state.journal.is_empty()),
                "failed" => assert_eq!(state.journal[0].success, Some(false)),
                _ => assert!(!state.journal[0].completed),
            }
            let retry = pane_hook_plan(&profile, &state);
            assert_eq!(
                retry.actions[0].kind,
                Action::Core(CoreAction::RestartCommand)
            );
            let runner = PaneHookRunner {
                backend: &backend,
                runner: FakeRunner::default(),
                error: false,
            };
            assert!(
                pane_hook_up(&profile, &retry, &backend, &runner, &mut state, true)
                    .failed
                    .is_empty()
            );
            let calls = backend.calls();
            assert_eq!(
                &calls[calls.len() - 2..],
                [
                    "hook:editor",
                    &format!("restart_command:{}", original.backend_id)
                ]
            );
            let env = runner.runner.envs_for(&["register"]).expect("env");
            assert_eq!(env["DROVE_RESOURCE"], "editor");
            assert_eq!(env["DROVE_BACKEND_ID"], original.backend_id);
            assert!(state.journal.iter().all(|entry| entry.completed));
            assert!(pane_hook_plan(&profile, &state).actions.is_empty());
        }
    }

    #[test]
    fn post_create_command_failure_records_core_pane_id_before_reporting_failure() {
        let (mut state, _dir) = temp_state();
        let profile = up_profile();
        seed_managed(
            &mut state,
            &[
                ("dev", "workspace", None),
                ("dev/main", "placement", Some("dev")),
            ],
        );
        state
            .profile_mut("default")
            .resources
            .get_mut("dev")
            .expect("workspace")
            .backend_id = "w1".into();
        state
            .profile_mut("default")
            .resources
            .get_mut("dev/main")
            .expect("group")
            .backend_id = "t1".into();
        let backend = RecordingHerdr::running_failing(&["start_command:p1"]);
        let report = pane_hook_up(
            &profile,
            &up_plan(&[(Action::Core(CoreAction::CreatePane), "editor")]),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );

        assert_eq!(report.failed.len(), 1);
        assert_eq!(report.failed[0].address, "editor");
        let managed = state.profile("default").expect("managed");
        assert_eq!(managed.resources["editor"].backend_id, "p1");
        assert_eq!(managed.resources["editor"].command_started, Some(false));
        assert_eq!(
            backend.calls(),
            ["create_pane:w1:editor", "start_command:p1"]
        );
        let retry = pane_hook_plan(&profile, &state);
        assert_eq!(retry.actions.len(), 1);
        assert_eq!(
            retry.actions[0].kind,
            Action::Core(CoreAction::RestartCommand)
        );
        assert_eq!(retry.actions[0].backend_id.as_deref(), Some("p1"));
        let backend = RecordingHerdr::running();
        let report = pane_hook_up(
            &profile,
            &retry,
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        assert!(report.failed.is_empty());
        assert_eq!(backend.calls(), ["restart_command:p1"]);
        assert_eq!(
            state.profile("default").expect("managed").resources["editor"].command_started,
            Some(true)
        );
        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn post_create_command_failure_records_fresh_tab_ids_before_reporting_failure() {
        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default", "workspaces": [{"name": "dev", "tabs": [{
                "name": "main", "panes": [
                    {"name": "editor", "serve": [["serve-editor"]]},
                    {"name": "tests", "serve": [["serve-tests"]]}
                ]
            }]}]
        }));
        let backend = RecordingHerdr::running_failing(&["start_command:p3"]);
        let report = pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );

        assert_eq!(report.failed.len(), 1);
        assert_eq!(report.failed[0].address, "dev/main");
        let managed = state.profile("default").expect("managed");
        assert_eq!(managed.resources["dev/main"].backend_id, "t2");
        assert_eq!(managed.resources["editor"].backend_id, "p3");
        assert_eq!(managed.resources["tests"].backend_id, "p4");
        assert_eq!(managed.resources["editor"].command_started, Some(false));
        assert_eq!(managed.resources["tests"].command_started, Some(false));
        let retry = pane_hook_plan(&profile, &state);
        assert_eq!(retry.actions.len(), 2);
        assert!(
            retry
                .actions
                .iter()
                .all(|action| action.kind == Action::Core(CoreAction::RestartCommand))
        );
        assert_eq!(retry.actions[0].backend_id.as_deref(), Some("p3"));
        assert_eq!(retry.actions[1].backend_id.as_deref(), Some("p4"));
        let backend = RecordingHerdr::running();
        let report = pane_hook_up(
            &profile,
            &retry,
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        assert!(report.failed.is_empty());
        assert_eq!(
            backend.calls(),
            ["restart_command:p3", "restart_command:p4"]
        );
        let managed = state.profile("default").expect("managed");
        assert_eq!(managed.resources["editor"].command_started, Some(true));
        assert_eq!(managed.resources["tests"].command_started, Some(true));
        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn post_split_command_failure_records_split_pane_id_before_reporting_failure() {
        let (mut state, _dir) = temp_state();
        let mut profile = up_profile();
        let backend = RecordingHerdr::running();
        pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        let mut pane = profile.workspaces[0].tabs[0].panes[0].clone();
        pane.name = "tests".into();
        pane.serve = vec![vec!["serve-tests".into()]];
        profile.workspaces[0].tabs[0].panes.push(pane);
        let backend = RecordingHerdr {
            fail_calls: ["start_command:p4".to_owned()].into_iter().collect(),
            ..backend
        };
        let report = pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );

        assert_eq!(report.failed.len(), 1);
        assert_eq!(report.failed[0].address, "tests");
        assert_eq!(
            state.profile("default").expect("managed").resources["tests"].backend_id,
            "p4"
        );
        assert_eq!(
            state.profile("default").expect("managed").resources["tests"].command_started,
            Some(false)
        );
        let retry = pane_hook_plan(&profile, &state);
        assert_eq!(retry.actions.len(), 1);
        assert_eq!(
            retry.actions[0].kind,
            Action::Core(CoreAction::RestartCommand)
        );
        assert_eq!(retry.actions[0].backend_id.as_deref(), Some("p4"));
        let backend = RecordingHerdr::running();
        let report = pane_hook_up(
            &profile,
            &retry,
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        assert!(report.failed.is_empty());
        assert_eq!(backend.calls(), ["restart_command:p4"]);
        assert_eq!(
            state.profile("default").expect("managed").resources["tests"].command_started,
            Some(true)
        );
        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn rename_pane_with_pending_command_start_restarts_in_the_same_up() {
        let (mut state, _dir) = temp_state();
        let mut profile = up_profile();
        let backend = RecordingHerdr::running();
        pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        {
            let pane = state
                .profile_mut("default")
                .resources
                .get_mut("editor")
                .expect("managed pane");
            pane.command_started = Some(false);
        }
        let pane_id = state.profile("default").expect("managed").resources["editor"]
            .backend_id
            .clone();
        state.save().expect("save pending command start");
        profile.workspaces[0].tabs[0].panes[0].label = Some("Editor".into());

        let plan = pane_hook_plan(&profile, &state);
        assert_eq!(plan.actions.len(), 2);
        assert_eq!(plan.actions[0].kind, Action::Core(CoreAction::RenamePane));
        assert_eq!(
            plan.actions[1].kind,
            Action::Core(CoreAction::RestartCommand)
        );
        let backend = RecordingHerdr::running();
        let report = pane_hook_up(
            &profile,
            &plan,
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        assert!(report.failed.is_empty());
        assert_eq!(
            backend.calls(),
            [
                format!("rename_pane:{pane_id}:Editor"),
                format!("restart_command:{pane_id}")
            ]
        );
        assert_eq!(
            report.outcome,
            UpOutcome::Reconciled {
                created: 0,
                changed: 2,
                tasks_run: 0
            }
        );
        assert_eq!(
            state.profile("default").expect("managed").resources["editor"].command_started,
            Some(true)
        );

        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn failed_pending_restart_after_rename_reports_partial_and_remains_pending() {
        let (mut state, _dir) = temp_state();
        let mut profile = up_profile();
        let backend = RecordingHerdr::running();
        pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        {
            let pane = state
                .profile_mut("default")
                .resources
                .get_mut("editor")
                .expect("managed pane");
            pane.command_started = Some(false);
        }
        let pane_id = state.profile("default").expect("managed").resources["editor"]
            .backend_id
            .clone();
        state.save().expect("save pending command start");
        profile.workspaces[0].tabs[0].panes[0].label = Some("Editor".into());

        let plan = pane_hook_plan(&profile, &state);
        assert_eq!(plan.actions.len(), 2);
        assert_eq!(plan.actions[0].kind, Action::Core(CoreAction::RenamePane));
        assert_eq!(
            plan.actions[1].kind,
            Action::Core(CoreAction::RestartCommand)
        );
        let backend = RecordingHerdr::running_failing_owned([format!("restart_command:{pane_id}")]);
        let report = pane_hook_up(
            &profile,
            &plan,
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        assert_eq!(report.failed.len(), 1);
        assert_eq!(report.failed[0].address, "editor");
        assert_eq!(
            report.outcome,
            UpOutcome::Reconciled {
                created: 0,
                changed: 1,
                tasks_run: 0
            }
        );
        assert_eq!(
            state.profile("default").expect("managed").resources["editor"].command_started,
            Some(false)
        );
        let retry = pane_hook_plan(&profile, &state);
        assert_eq!(retry.actions.len(), 1);
        assert_eq!(
            retry.actions[0].kind,
            Action::Core(CoreAction::RestartCommand)
        );
        assert_eq!(
            retry.actions[0].backend_id.as_deref(),
            Some(pane_id.as_str())
        );
    }

    #[test]
    fn pane_hook_failure_gates_earlier_named_dependent_tab_and_agent_but_keeps_independent_work() {
        for approve in [false, true] {
            let (mut state, _dir) = temp_state();
            let profile = profile_from(json!({
                "name": "default", "workspaces": [{"name": "dev", "tabs": [
                    {"name": "a-dependent", "panes": [{"name": "dependent", "after": ["prerequisite"], "on_start": ["dependent-hook"], "serve": [["dependent-serve"]]}]},
                    {"name": "z-prerequisite", "panes": [{"name": "prerequisite", "on_start": ["register"], "serve": [["serve"]], "agent": {"kind": "claude", "name": "worker"}}]},
                    {"name": "independent", "panes": [{"name": "independent", "serve": [["independent-serve"]]}]}
                ]}]
            }));
            let backend = RecordingHerdr::running();
            let runner = PaneHookRunner {
                backend: &backend,
                runner: FakeRunner::default().fail(&["register"]),
                error: false,
            };
            let report = pane_hook_up(
                &profile,
                &pane_hook_plan(&profile, &state),
                &backend,
                &runner,
                &mut state,
                approve,
            );
            assert_eq!(report.failed.len(), 1);
            assert_eq!(report.failed[0].address, "dev/z-prerequisite");
            assert!(
                report
                    .skipped
                    .iter()
                    .any(|skip| skip.address == "dev/a-dependent")
            );
            assert!(report.skipped.iter().any(|skip| skip.address == "worker"));
            let managed = state.profile("default").expect("managed");
            assert!(managed.resources.contains_key("dev"));
            assert!(managed.resources.contains_key("independent"));
            assert!(!managed.resources.contains_key("prerequisite"));
            assert!(!managed.resources.contains_key("dev/z-prerequisite"));
            assert!(!managed.resources.contains_key("dependent"));
            assert!(
                !backend
                    .calls()
                    .iter()
                    .any(|call| call == "hook:dependent" || call.starts_with("start_agent:"))
            );
            let runner = PaneHookRunner {
                backend: &backend,
                runner: FakeRunner::default(),
                error: false,
            };
            let retry = pane_hook_plan(&profile, &state);
            assert!(
                pane_hook_up(&profile, &retry, &backend, &runner, &mut state, true)
                    .failed
                    .is_empty()
            );
            assert!(pane_hook_plan(&profile, &state).actions.is_empty());
        }
    }

    #[test]
    fn pane_hook_recreate_failure_removes_closed_ownership_and_retries_only_the_split() {
        let (mut state, _dir) = temp_state();
        let mut profile = up_profile();
        let backend = RecordingHerdr::running();
        pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        let original_id = state.profile("default").expect("managed").resources["editor"]
            .backend_id
            .clone();
        profile.workspaces[0].tabs[0].panes[0].cwd = Some("/new-cwd".into());
        profile.workspaces[0].tabs[0].panes[0].on_start = Some(vec!["register".into()]);
        let runner = PaneHookRunner {
            backend: &backend,
            runner: FakeRunner::default().fail(&["register"]),
            error: false,
        };
        let plan = pane_hook_plan(&profile, &state);
        assert!(plan.has_destructive_actions());
        let before = backend.calls();
        let report = pane_hook_up(&profile, &plan, &backend, &runner, &mut state, false);
        assert!(report.blocked_destructive);
        assert_eq!(
            backend.calls(),
            before,
            "unapproved close also gates the replacement"
        );
        let report = pane_hook_up(&profile, &plan, &backend, &runner, &mut state, true);
        assert_eq!(report.failed.len(), 1);
        assert!(
            !state
                .profile("default")
                .expect("managed")
                .resources
                .contains_key("editor")
        );
        let calls = backend.calls();
        assert_eq!(
            &calls[calls.len() - 2..],
            [format!("close_pane:{original_id}"), "hook:editor".into()]
        );
        let retry = pane_hook_plan(&profile, &state);
        assert_eq!(retry.actions.len(), 1);
        assert_eq!(retry.actions[0].kind, Action::Herdr(HerdrAction::SplitPane));
        assert!(
            pane_hook_up(
                &profile,
                &retry,
                &backend,
                &FakeRunner::default(),
                &mut state,
                false
            )
            .failed
            .is_empty()
        );
        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn pane_hooks_gate_imported_idle_pane_restarts_and_keep_successful_ownership() {
        let (mut state, _dir) = temp_state();
        let mut profile = profile_from(json!({
            "name": "default", "workspaces": [{"name": "dev", "tabs": [{"name": "main", "panes": [
                {"name": "commit", "serve": [["idle"]]},
                {"name": "docs", "serve": [["idle"]], "after": ["commit"]}
            ]}]}]
        }));
        let backend = RecordingHerdr::running();
        pane_hook_up(
            &profile,
            &pane_hook_plan(&profile, &state),
            &backend,
            &FakeRunner::default(),
            &mut state,
            true,
        );
        for (pane, id) in [("commit", "w2:p2"), ("docs", "w2:p3")] {
            state
                .profile_mut("default")
                .resources
                .get_mut(pane)
                .expect("pane")
                .backend_id = id.into();
        }
        state.save().expect("import ownership");
        for pane in &mut profile.workspaces[0].tabs[0].panes {
            pane.serve = vec![vec![format!("serve-{}", pane.name)]];
            pane.on_start = Some(vec![format!("register-{}", pane.name)]);
        }
        let runner = PaneHookRunner {
            backend: &backend,
            runner: FakeRunner::default().fail(&["register-docs"]),
            error: false,
        };
        let plan = pane_hook_plan(&profile, &state);
        assert_eq!(plan.actions.len(), 2);
        assert!(
            plan.actions
                .iter()
                .all(|action| action.kind == Action::Core(CoreAction::RestartCommand))
        );
        let report = pane_hook_up(&profile, &plan, &backend, &runner, &mut state, true);
        assert_eq!(report.failed[0].address, "docs");
        let calls = backend.calls();
        assert_eq!(
            &calls[calls.len() - 3..],
            ["hook:commit", "restart_command:w2:p2", "hook:docs"]
        );
        let retry = pane_hook_plan(&profile, &state);
        assert_eq!(
            retry.actions.len(),
            1,
            "successful first restart is already persisted"
        );
        assert_eq!(retry.actions[0].address, "docs");
        let runner = PaneHookRunner {
            backend: &backend,
            runner: FakeRunner::default(),
            error: false,
        };
        assert!(
            pane_hook_up(&profile, &retry, &backend, &runner, &mut state, false)
                .failed
                .is_empty()
        );
        let calls = backend.calls();
        assert_eq!(
            &calls[calls.len() - 2..],
            ["hook:docs", "restart_command:w2:p3"]
        );
        assert_eq!(
            runner.runner.envs_for(&["register-docs"]).expect("env")["DROVE_BACKEND_ID"],
            "w2:p3"
        );
        assert!(pane_hook_plan(&profile, &state).actions.is_empty());
    }

    #[test]
    fn pane_hooks_preserve_adopted_caller_and_unrelated_manual_resources() {
        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default", "workspaces": [{"name": "dev", "tabs": [{
                "name": "main", "panes": [{"name": "controller", "adopt": "caller", "on_start": ["must-not-run"], "serve": [["must-not-start"]]}]
            }]}]
        }));
        let snapshot = Snapshot::default()
            .with_caller("w2:p1")
            .unmanaged("workspace", "manual-workspace", "w9", None)
            .unmanaged(
                "placement",
                "manual-group",
                "w9:t1",
                Some("manual-workspace"),
            )
            .unmanaged("pane", "manual-pane", "w9:p1", Some("manual-group"));
        let plan = build_plan(&profile, &snapshot).expect("plan");
        let backend = RecordingHerdr::running();
        let runner = PaneHookRunner {
            backend: &backend,
            runner: FakeRunner::default(),
            error: false,
        };
        let report = pane_hook_up(&profile, &plan, &backend, &runner, &mut state, false);
        assert!(report.failed.is_empty());
        let caller = &state.profile("default").expect("managed").resources["controller"];
        assert_eq!(caller.backend_id, "w2:p1");
        assert_eq!(caller.adopted, Some(true));
        assert!(runner.runner.calls().is_empty());
        assert!(backend.calls().iter().all(|call| !call.contains("w2:p1")
            && !call.contains("w9")
            && !call.contains("manual")
            && !call.contains("pane:")));
    }

    #[test]
    fn up_applies_workspace_and_pane_then_focuses_the_first_workspace() {
        let (mut state, _dir) = temp_state();
        let profile = up_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Core(CoreAction::CreatePane), "editor"),
        ]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up");

        let calls = backend.calls();
        assert!(
            calls.iter().any(|c| c == "create_workspace:dev"),
            "workspace must be created: {calls:?}"
        );
        assert!(
            calls.iter().any(|c| c.starts_with("create_pane:")),
            "pane must be created: {calls:?}"
        );
        // The first workspace is brought to the front with the id its create
        // returned.
        assert_eq!(report.focused.as_deref(), Some("w1"));
        assert!(
            calls.iter().any(|c| c == "focus:w1"),
            "the first workspace must be focused: {calls:?}"
        );
        assert_eq!(
            report.outcome,
            UpOutcome::Reconciled {
                created: 2,
                changed: 0,
                tasks_run: 0
            }
        );
        // Ownership is recorded so the next run sees the resources in sync.
        assert!(
            state
                .profile("default")
                .expect("profile recorded")
                .resources
                .contains_key("dev")
        );
    }

    #[test]
    fn up_with_no_focus_applies_but_never_focuses() {
        let (mut state, _dir) = temp_state();
        let profile = up_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Core(CoreAction::CreatePane), "editor"),
        ]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            false,
        )
        .expect("up");

        assert_eq!(report.focused, None);
        let calls = backend.calls();
        assert!(
            !calls.iter().any(|c| c.starts_with("focus:")),
            "--no-focus must not focus anything: {calls:?}"
        );
    }

    #[test]
    fn up_already_in_sync_brings_the_workspace_to_the_front() {
        let (mut state, _dir) = temp_state();
        let profile = up_profile();
        let ir = profile.to_ir();
        // A previous run recorded the workspace's backend id; nothing is out
        // of sync now.
        state.profile_mut("default").resources.insert(
            "dev".to_owned(),
            ManagedResource {
                kind: "workspace".into(),
                backend_id: "w1".into(),
                parent: None,
                digest: "any".into(),
                label: None,
                cwd: None,
                adopted: None,
                command_started: None,
                last_outcome: None,
            },
        );
        let plan = up_plan(&[]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up");

        assert_eq!(report.outcome, UpOutcome::AlreadyRunning);
        assert_eq!(report.focused.as_deref(), Some("w1"));
        let calls = backend.calls();
        assert_eq!(calls, vec!["focus:w1".to_owned()], "only focus, no creates");
    }

    #[test]
    fn up_reports_the_hint_when_the_session_cannot_be_started() {
        let (mut state, _dir) = temp_state();
        let profile = up_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[(Action::Core(CoreAction::CreateWorkspace), "dev")]);
        let backend = RecordingHerdr::cannot_start("herdr --session dev-session");
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up");

        assert_eq!(
            report.outcome,
            UpOutcome::CannotStart {
                hint: "herdr --session dev-session".to_owned()
            }
        );
        assert!(
            backend.calls().is_empty(),
            "an unstartable session applies nothing and focuses nothing"
        );
    }

    #[test]
    fn up_builds_a_fresh_multi_pane_tab_and_records_every_pane() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [{
                    "name": "main",
                    "split": "right",
                    "ratios": [0.67],
                    "panes": [{"name": "editor"}, {"name": "tests"}],
                }]
            }]
        }));
        let ir = profile.to_ir();
        // A fresh group plans one CreateTab and no per-pane splits.
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Herdr(HerdrAction::CreateTab), "dev/main"),
        ]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up");

        let calls = backend.calls();
        // Both declared panes are built as part of the one CreateTab, and the
        // ratio is applied once (the executor never emits a bare set_ratio on
        // a one-pane Herdr tab).
        assert!(
            calls.iter().any(|c| c.starts_with("create_tab:")),
            "the Herdr tab must be created: {calls:?}"
        );
        assert_eq!(
            calls.iter().filter(|c| c.starts_with("tab_pane:")).count(),
            2,
            "both panes must be built into the tab: {calls:?}"
        );
        assert!(
            calls.iter().any(|c| c.starts_with("set_ratio:")),
            "the ratio must be applied: {calls:?}"
        );

        // The group and each pane are recorded, so a second run sees them
        // owned instead of splitting them in again.
        let managed = state.profile("default").expect("profile recorded");
        assert!(managed.resources.contains_key("dev/main"), "group recorded");
        assert!(
            managed.resources.contains_key("editor"),
            "first pane recorded"
        );
        assert!(
            managed.resources.contains_key("tests"),
            "second pane recorded"
        );
        assert_eq!(
            report.outcome,
            UpOutcome::Reconciled {
                created: 2,
                changed: 0,
                tasks_run: 0
            }
        );
    }

    /// Two independent workspaces, each with one Herdr tab and one pane — the
    /// shape D52's repro needs: three create-style backend calls
    /// (`CreateWorkspace(dev)`, `CreateWorkspace(ops)`, `CreateTab(dev/main)`)
    /// where the second can be made to fail without touching the third.
    fn two_workspace_profile() -> Profile {
        profile_from(json!({
            "name": "default",
            "workspaces": [
                {"name": "dev", "tabs": [{"name": "main", "panes": [{"name": "editor"}]}]},
                {"name": "ops", "tabs": [{"name": "main", "panes": [{"name": "shell"}]}]},
            ]
        }))
    }

    #[test]
    fn up_records_the_first_action_and_still_applies_an_independent_third_past_a_failed_second() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = two_workspace_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Core(CoreAction::CreateWorkspace), "ops"),
            (Action::Herdr(HerdrAction::CreateTab), "dev/main"),
        ]);
        let backend = RecordingHerdr::running_failing(&["create_workspace:ops"]);
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            None,
            false,
        )
        .expect("up does not abort on a single failed action");

        assert_eq!(
            report.failed.len(),
            1,
            "the second action must be reported failed"
        );
        assert_eq!(report.failed[0].address, "ops");
        assert!(report.failed[0].error.contains("boom"));
        assert!(
            report.skipped.is_empty(),
            "the third action does not depend on `ops`, so nothing is skipped: {:?}",
            report.skipped
        );

        // The independent third action still ran, despite the second one
        // failing (D52 point 2, audit finding 1).
        let calls = backend.calls();
        assert!(
            calls.iter().any(|c| c.starts_with("create_tab:")),
            "the independent CreateTab must still apply: {calls:?}"
        );

        // The first action's ownership was recorded (D52 point 1): a killed
        // process after the failure would leave `dev` and `dev/main`
        // describing exactly what succeeded.
        let managed = state.profile("default").expect("profile recorded");
        assert!(managed.resources.contains_key("dev"), "dev recorded");
        assert!(
            managed.resources.contains_key("dev/main"),
            "dev/main recorded"
        );
        assert!(managed.resources.contains_key("editor"), "editor recorded");
        assert!(
            !managed.resources.contains_key("ops"),
            "the failed workspace must not be recorded"
        );
    }

    #[test]
    fn up_skips_an_action_that_depends_on_one_that_failed() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = two_workspace_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "ops"),
            (Action::Herdr(HerdrAction::CreateTab), "ops/main"),
        ]);
        let backend = RecordingHerdr::running_failing(&["create_workspace:ops"]);
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            None,
            false,
        )
        .expect("up");

        assert_eq!(report.failed.len(), 1);
        assert_eq!(report.failed[0].address, "ops");
        assert_eq!(
            report.skipped.len(),
            1,
            "the dependent Herdr tab must be skipped"
        );
        assert_eq!(report.skipped[0].address, "ops/main");
        assert_eq!(report.skipped[0].depends_on, "ops");

        // The skipped action never touched the backend at all.
        let calls = backend.calls();
        assert!(
            !calls.iter().any(|c| c.starts_with("create_tab:")),
            "a dependency-skipped action must never call the backend: {calls:?}"
        );
        assert!(
            state
                .profile("default")
                .map(|managed| managed.resources.is_empty())
                .unwrap_or(true),
            "neither the failed nor the skipped resource is recorded"
        );
    }

    #[test]
    fn a_rerun_after_a_partial_failure_plans_only_what_is_still_missing() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = two_workspace_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Core(CoreAction::CreateWorkspace), "ops"),
            (Action::Herdr(HerdrAction::CreateTab), "dev/main"),
        ]);
        let backend = RecordingHerdr::running_failing(&["create_workspace:ops"]);
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            None,
            false,
        )
        .expect("up");

        // The next `build_plan` sees `dev` and its Herdr tab as owned and
        // converged, and proposes only what actually failed last time.
        let managed = state.profile("default").cloned().unwrap_or_default();
        let snapshot = managed.to_snapshot("default", None);
        let rerun = build_plan(&profile, &snapshot).expect("plan");

        let addresses: Vec<&str> = rerun
            .actions
            .iter()
            .map(|action| action.address.as_str())
            .collect();
        assert_eq!(
            addresses,
            vec!["ops", "ops/main"],
            "only the failed workspace (and what depends on it) is replanned: {addresses:?}"
        );
    }

    #[test]
    fn up_stops_applying_further_actions_once_a_state_save_fails() {
        let (mut state, dir) = temp_state();
        // Force every `state.save()` to fail: its parent directory component
        // is actually a plain file, so `fs::create_dir_all` cannot create it.
        let blocker = dir.path().join("blocker");
        fs::write(&blocker, b"not a directory").expect("write blocker file");
        state.path = blocker.join("state.json");

        let profile = two_workspace_profile();
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Core(CoreAction::CreateWorkspace), "ops"),
        ]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let result = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            None,
            false,
        );

        assert!(
            result.is_err(),
            "a state save failure must surface as an error, not a silent partial apply"
        );

        // The first action's backend call happened (its outcome had to be
        // known before the save that failed), but the loop must have stopped
        // there instead of going on to apply `ops` without any record of
        // `dev` or a chance to record `ops` either.
        let calls = backend.calls();
        assert!(
            calls.iter().any(|c| c == "create_workspace:dev"),
            "the first action still applies before the save fails: {calls:?}"
        );
        assert!(
            !calls.iter().any(|c| c == "create_workspace:ops"),
            "no action after the save failure may reach the backend: {calls:?}"
        );
    }

    #[test]
    fn up_skips_a_task_that_depends_on_one_that_failed() {
        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default",
            "tasks": [
                {"name": "a", "run": ["run-a"]},
                {"name": "b", "run": ["run-b"], "after": ["a"]}
            ]
        }));
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::RunTask), "a"),
            (Action::Core(CoreAction::RunTask), "b"),
        ]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default().fail(&["run-a"]);
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            None,
            false,
        )
        .expect("up does not abort on a failed task");

        assert_eq!(
            report.tasks,
            vec![
                ("a".to_owned(), TaskOutcome::Ran(false)),
                ("b".to_owned(), TaskOutcome::DependencySkipped),
            ],
            "a task after a failed task must not run (D52 point 2)"
        );
        assert_eq!(report.skipped.len(), 1);
        assert_eq!(report.skipped[0].address, "b");
        assert_eq!(report.skipped[0].depends_on, "a");
        assert_eq!(runner.calls(), vec![vec!["run-a".to_owned()]]);
    }

    #[test]
    fn up_reuses_the_freshly_created_workspaces_root_tab_for_its_first_tab_only() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [
                    {"name": "main", "panes": [{"name": "editor"}]},
                    {"name": "second", "panes": [{"name": "logs"}]},
                ]
            }]
        }));
        let ir = profile.to_ir();
        let plan = up_plan(&[
            (Action::Core(CoreAction::CreateWorkspace), "dev"),
            (Action::Herdr(HerdrAction::CreateTab), "dev/main"),
            (Action::Herdr(HerdrAction::CreateTab), "dev/second"),
        ]);
        let backend = RecordingHerdr::running_with_root_tabs();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up");

        let calls = backend.calls();
        assert!(
            calls.iter().any(|c| c.starts_with("create_tab:")
                && c.contains(":main:")
                && c.contains("existing=Some")),
            "the workspace's own root tab must be reused for its first declared tab: {calls:?}"
        );
        assert!(
            calls.iter().any(|c| c.starts_with("create_tab:")
                && c.contains(":second:")
                && c.contains("existing=None")),
            "only the first declared tab may reuse the root tab: {calls:?}"
        );
    }

    #[test]
    fn up_never_reuses_a_root_tab_for_an_adopted_or_pre_existing_workspace() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [{"name": "main", "panes": [{"name": "editor"}]}]
            }]
        }));
        let ir = profile.to_ir();
        // The workspace already exists from a previous run, so this plan has
        // no `CreateWorkspace` action for it — only the Herdr tab is being
        // added.
        {
            let managed = state.profile_mut("default");
            managed.resources.insert(
                "dev".to_owned(),
                ManagedResource {
                    kind: "workspace".into(),
                    backend_id: "w1".into(),
                    parent: None,
                    digest: "d".into(),
                    label: None,
                    cwd: None,
                    adopted: None,
                    command_started: None,
                    last_outcome: None,
                },
            );
        }
        let plan = up_plan(&[(Action::Herdr(HerdrAction::CreateTab), "dev/main")]);
        let backend = RecordingHerdr::running_with_root_tabs();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up");

        let calls = backend.calls();
        assert!(
            calls
                .iter()
                .any(|c| c.starts_with("create_tab:") && c.contains("existing=None")),
            "an adopted or pre-existing workspace must never reuse a root tab: {calls:?}"
        );
    }

    #[test]
    fn up_splits_a_pane_into_a_converged_group_using_recorded_ids() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [{"name": "main", "panes": [{"name": "editor"}, {"name": "tests"}]}]
            }]
        }));
        let ir = profile.to_ir();
        // A previous run recorded the workspace, the group and the first pane;
        // only `tests` is being added now, so its parents need no action this
        // run and their ids live only in recorded state.
        {
            let managed = state.profile_mut("default");
            for (address, kind, backend, parent) in [
                ("dev", "workspace", "w1", None),
                ("dev/main", "placement", "t1", Some("dev")),
                ("editor", "pane", "p1", Some("dev/main")),
            ] {
                managed.resources.insert(
                    address.to_owned(),
                    ManagedResource {
                        kind: kind.into(),
                        backend_id: backend.into(),
                        parent: parent.map(ToOwned::to_owned),
                        digest: "d".into(),
                        label: None,
                        cwd: None,
                        adopted: None,
                        command_started: None,
                        last_outcome: None,
                    },
                );
            }
        }
        let plan = up_plan(&[(Action::Herdr(HerdrAction::SplitPane), "tests")]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        let report = up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up must resolve the converged workspace from recorded state");

        let calls = backend.calls();
        // The new pane splits into the group's recorded Herdr tab, and no
        // error is raised for the workspace that was never touched this run.
        assert!(
            calls.iter().any(|c| c.starts_with("split_pane:t1:")),
            "the pane must split into the recorded tab: {calls:?}"
        );
        assert!(
            state
                .profile("default")
                .expect("profile")
                .resources
                .contains_key("tests"),
            "the new pane is recorded"
        );
        assert!(matches!(report.outcome, UpOutcome::Reconciled { .. }));
    }

    #[test]
    fn set_ratio_is_applied_after_the_split_that_creates_its_gap() {
        use crate::planner::HerdrAction;

        let (mut state, _dir) = temp_state();
        let profile = profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [{
                    "name": "main",
                    "split": "right",
                    "ratios": [0.6],
                    "panes": [{"name": "editor"}, {"name": "tests"}]
                }]
            }]
        }));
        let ir = profile.to_ir();
        {
            let managed = state.profile_mut("default");
            for (address, kind, backend, parent) in [
                ("dev", "workspace", "w1", None),
                ("dev/main", "placement", "t1", Some("dev")),
                ("editor", "pane", "p1", Some("dev/main")),
            ] {
                managed.resources.insert(
                    address.to_owned(),
                    ManagedResource {
                        kind: kind.into(),
                        backend_id: backend.into(),
                        parent: parent.map(ToOwned::to_owned),
                        digest: "d".into(),
                        label: None,
                        cwd: None,
                        adopted: None,
                        command_started: None,
                        last_outcome: None,
                    },
                );
            }
        }
        // The planner orders every Herdr tab action ahead of the pane splits,
        // so the ratio comes first in the plan — but applying it before the
        // split exists would fail against a real backend.
        let plan = up_plan(&[
            (Action::Herdr(HerdrAction::SetRatio), "dev/main"),
            (Action::Herdr(HerdrAction::SplitPane), "tests"),
        ]);
        let backend = RecordingHerdr::running();
        let runner = FakeRunner::default();
        let root = PathBuf::from("/repo");

        up(
            &backend,
            &profile,
            &ir,
            &plan,
            &ctx(&root, &runner),
            &mut state,
            true,
            "dev-session",
            Some("dev"),
            true,
        )
        .expect("up");

        let calls = backend.calls();
        let split_at = calls
            .iter()
            .position(|c| c.starts_with("split_pane:"))
            .expect("a split happened");
        let ratio_at = calls
            .iter()
            .position(|c| c.starts_with("set_ratio:"))
            .expect("a ratio was set");
        assert!(
            ratio_at > split_at,
            "the ratio must be applied after the split, whatever the plan order: {calls:?}"
        );
    }

    #[test]
    fn herdr_action_on_a_flavorless_backend_is_unsupported_but_core_panes_still_run() {
        use crate::planner::{Action, CoreAction, HerdrAction, PlannedAction, SyncStatus};

        let profile = profile_from(json!({
            "name": "default",
            "workspaces": [{
                "name": "dev",
                "tabs": [{"name": "main", "panes": [{"name": "editor", "serve": [["bash"]]}]}]
            }]
        }));
        let ir = profile.to_ir();

        let action = |kind, address: &str| PlannedAction {
            kind,
            address: address.to_owned(),
            backend_id: None,
            destructive: false,
            reason: String::new(),
        };
        let plan = Plan {
            profile: "default".into(),
            desired_digest: String::new(),
            status: SyncStatus::OutOfSync,
            adopted: BTreeMap::new(),
            actions: vec![
                action(Action::Core(CoreAction::CreateWorkspace), "dev"),
                action(Action::Herdr(HerdrAction::CreateTab), "dev/main"),
                action(Action::Core(CoreAction::CreatePane), "editor"),
            ],
        };

        let backend = FlavorlessBackend::default();
        let outcomes = apply_plan(&backend, &ir, &plan);

        assert_eq!(outcomes[0].1, Outcome::Applied);
        assert_eq!(
            outcomes[1].1,
            Outcome::Unsupported {
                flavor: "herdr",
                action: Action::Herdr(HerdrAction::CreateTab),
            },
            "a Herdr action on a backend without the flavor must be Unsupported"
        );
        assert_eq!(outcomes[2].1, Outcome::Applied);
        assert_eq!(
            *backend.created_panes.lock().expect("mutex"),
            vec!["editor".to_owned()],
            "the core pane must still be created despite the unsupported Herdr action"
        );
    }
}