forge_workspace/mcp/workers/

facade.rs

//! Narrow workspace-state surface the workers MCP Tool impls
//! consume. Trait + mock for unit-testing the Tool impls without
//! spinning up a real Workspace. Mirrors `crate::mcp::peers::facade`.
//!
//! The production impl is `ProdWorkerFacade` (below).

use std::sync::{Arc, Weak};

use forge_agent::client::SpawnFailureKind;
use forge_primitives::WorkerStatus;

use crate::SessionSlot;
use crate::mcp::peers::types::WrappedPrompt;
use crate::protocol::{Command, SessionChoice, WorkerSpawnReply};
use crate::workspace::{ResumeTarget, Workspace};

/// A delivery's own answer, and it has one shape: the prompt was accepted
/// (handed to the target's session queue, held for its turn end, or parked
/// for its connect). A prompt that reached no session is an error instead,
/// so "accepted" is the whole of what this claims.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WorkerTargetStatus {
    Delivered,
}

/// Error from `deliver_worker_prompt`. A prompt not yet taken is not one:
/// a worker that never finishes connecting leaves a parked message that
/// `Workspace::expire_parked_for_slot` acknowledges back to its sender.
#[derive(Debug, PartialEq, Eq)]
pub enum WorkerDeliverError {
    /// No worker by `label` in `project_key` at all.
    UnknownLabel { project_key: String, label: String },
    /// The label's row exists but is not live - it crashed or was closed,
    /// and the row is kept so the label can be respawned. Distinct from
    /// [`Self::UnknownLabel`], which claims the label was never there.
    NotLive { project_key: String, label: String },
    /// The delivery did not happen: the handler found nothing that could
    /// take the prompt, the workspace channel failed, or the handler
    /// dropped its reply, so whether anything took the message is
    /// unknown. Distinct from [`Self::UnknownLabel`], which claims there
    /// is no live match.
    DispatchFailed { message: String },
}

/// Error from `deliver_prompt_to_lead`. The caller wants to send a
/// prompt back to its lead via the reserved `"lead"` addressing
/// keyword.
#[derive(Debug, PartialEq, Eq)]
pub enum WorkerLeadDeliverError {
    /// Caller resolves to no known session (defensive - should not
    /// happen with a valid caller slot).
    UnknownCaller,
    /// Caller is a project lead - leads have no lead to talk back
    /// to. The `"lead"` keyword is worker-only.
    LeadCallerHasNoLead,
    /// Worker's recorded lead session no longer resolves to a session
    /// in the pool (lead session ended). The worker can't reach its
    /// lead anymore.
    LeadGone,
    /// The delivery did not happen (see
    /// [`WorkerDeliverError::DispatchFailed`]).
    DispatchFailed { message: String },
}

/// The label reserved for a project's own agent. A caller addresses it
/// as the `label` of an `agents__send_message` target;
/// `agents__spawn` rejects the label so no live worker can shadow the
/// keyword.
pub use forge_primitives::LEAD_LABEL;

/// Synchronous error from `spawn_worker`. All gating happens before
/// the workspace dispatch is even issued.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum WorkerSpawnError {
    /// Caller is a worker, not a project lead. v1 is lead-only.
    NotLeadCaller,
    /// `label` is empty after trim.
    EmptyLabel,
    /// `label` collides with the reserved `"lead"` keyword
    /// (workers MCP uses it as an addressing target for the worker's
    /// own lead - see [`LEAD_LABEL`]).
    ReservedLabel,
    /// `label` carries a slash. claude rewrites one before deriving its
    /// worktree path and branch, so forge's own composition from the raw
    /// label names a directory and a branch claude never uses (#1603).
    LabelWithPathSeparator,
    /// `charter` is empty after trim.
    EmptyCharter,
    /// Caller's session key resolves to no known project (defensive;
    /// the tools closure-bind a real key at spawn time, so this
    /// should never fire in practice).
    UnknownCallerProject,
    /// Spawn was dispatched but the workspace returned an error
    /// (e.g. tag-write failed). Forwarded as-is to the LLM.
    DispatchFailed { message: String },
    /// The worker's git worktree is missing or unusable, so nothing was
    /// spawned: claude could not create `--worktree <label>` (its reason
    /// passed through verbatim), or the composed path belongs to another
    /// project's checkout or worker and was refused before anything
    /// could touch it.
    WorktreeCreationFailed { reason: String },
    /// `resume_session` was set and the label's prior session could not
    /// be looked up: the transcript directory, or a transcript in it, is
    /// there and unreadable. Nothing was spawned. A lookup that found
    /// nothing is not this - that one falls back to a fresh session and
    /// reports it.
    ResumeLookupFailed { label: String, message: String },
}

/// Synchronous error from `update_worker`. Gating (lead-only, non-empty
/// label, at least one supplied field) happens in the tool before the
/// call, so these are the outcomes only the store can decide.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum WorkerUpdateError {
    /// Caller's session key resolves to no known project (defensive).
    UnknownCallerProject,
    /// No persisted dynamic-worker row for `label` in the caller's
    /// project. Update never creates one.
    NoSuchWorker { label: String, project_key: String },
    /// The store read or write failed, or the store isn't open.
    StoreFailed { message: String },
}

/// Synchronous outcome of `despawn_worker` - the success-shaped half
/// of a `Command::DespawnWorker`. Gating + dispatch errors are the
/// `Err` arm of the result; these two variants are what the
/// `agents__despawn` Tool renders as `status: "despawned" | "blocked"`.
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum DespawnOutcome {
    /// Worker torn down. `worktree_cleanup_warning` is `Some` when the
    /// post-teardown `git worktree remove` failed (the worker is still
    /// gone; only the worktree directory lingers).
    /// `branch_cleanup_warning` is `Some` when the `worktree-<label>`
    /// branch was left in place because it holds commits reachable from
    /// no other ref, or the check failed.
    Despawned { worktree_cleanup_warning: Option<String>, branch_cleanup_warning: Option<String> },
    /// Despawn refused: the worktree is dirty (uncommitted/untracked or
    /// unpushed) and `force` was not set. `reason` names what is dirty;
    /// the worker stays live.
    Blocked { reason: String },
}

/// Synchronous error from `despawn_worker`. Gating (lead-only,
/// non-empty label) happens before dispatch; `UnknownLabel` is
/// surfaced from the handler's `NotFound`, and `DispatchFailed` from
/// its `Failed`.
#[derive(Debug, PartialEq, Eq)]
pub enum WorkerDespawnError {
    /// Caller is a worker, not a project lead. Despawn is lead-only.
    NotLeadCaller,
    /// `label` is empty after trim.
    EmptyLabel,
    /// Caller's session key resolves to no known project (defensive).
    UnknownCallerProject,
    /// No live worker matched `label` in the caller's project.
    UnknownLabel { label: String, project_key: String },
    /// The despawn did not happen: the workspace channel failed, the
    /// handler dropped the reply, or the store could not be read or
    /// written, so whether a worker's row exists is unknown. Distinct
    /// from [`Self::UnknownLabel`], which claims there is none.
    DispatchFailed { message: String },
}

/// Heuristic mapping from a raw spawn-error message + the worker's
/// `is_git_repo_at_spawn` flag to a typed [`WorkerSpawnError`].
///
/// Routing rules:
///
/// - The message mentions "worktree" verbatim, OR matches a known
///   worktree-creation failure pattern ("failed to resolve base branch",
///   "already used by worktree") AND the worker was actually spawned
///   in a git repo (so `--worktree=<label>` was on the argv) →
///   [`WorkerSpawnError::WorktreeCreationFailed`]. The original message
///   is forwarded verbatim as `reason`.
/// - Everything else → [`WorkerSpawnError::DispatchFailed`].
///
/// The `is_git_repo_at_spawn` guard exists because a worker spawned in
/// a non-git-repo project (no `--worktree` on the argv) could not have
/// produced a worktree-creation failure. A "worktree" word arriving for
/// such a worker is suspicious - some other layer of the spawn pipe -
/// and falls through to [`WorkerSpawnError::DispatchFailed`] so the
/// LLM sees the raw message rather than a misclassified variant.
///
/// Matching is case-insensitive. The reason string passed back keeps
/// the original casing for verbatim LLM display.
///
/// ## Bridge-prefix contract (#245 Layer C blocker 2)
///
/// Async spawn failures arrive here pre-wrapped by the bridge in
/// `forge_agent::forge_sdk_bridge`:
///
/// - `"forge-sdk session spawn failed: {err}"`
/// - `"forge-sdk session resume failed: {err}"`
/// - `"forge-sdk session spawn failed after resume fallback (resume err: {a}; new err: {b})"`
///
/// The wrapper prefixes themselves carry no discriminator, but the
/// wrapped error can: `Error::CwdNotFound` renders the path it could not
/// enter, and for a worker that path runs through
/// `.claude/worktrees/<label>`. That word is the worktree heuristic's own
/// key, so a message-only classifier reads a resume with nowhere to start
/// as a worktree that could not be CREATED, deletes the worker's row, and
/// tells the lead something untrue. `kind` carries the SDK's own variant
/// for exactly that case; the substring search below is what answers for
/// everything the bridge could not name.
pub fn classify_worker_spawn_failure(
    message: &str,
    is_git_repo_at_spawn: bool,
    kind: SpawnFailureKind,
) -> WorkerSpawnError {
    // A named variant is the truth and the message is only how it
    // renders; searching the render cannot overrule it.
    if !matches!(kind, SpawnFailureKind::Unclassified) {
        return WorkerSpawnError::DispatchFailed { message: message.to_owned() };
    }
    // The cap refusal embeds the project key, which derives from a
    // directory path, so it can carry "worktree" without being a
    // worktree failure.
    if message.contains("worker limit reached") {
        return WorkerSpawnError::DispatchFailed { message: message.to_owned() };
    }
    // The cleanup refusal is forge's own sentence, so it is recognised by
    // its own closing phrase - the part no label interpolated into the
    // sentence can split - and whatever that label happens to spell cannot
    // turn it into a worktree-creation failure.
    if message.contains("is still cleaning up; retry once it finishes") {
        return WorkerSpawnError::DispatchFailed { message: message.to_owned() };
    }
    // The known-project refusal is forge's own sentence and carries a
    // project's name - whatever that project is called, "worktree"
    // included - so it is recognised by the phrase no name can split.
    if message.contains("which forge.toml declares") {
        return WorkerSpawnError::DispatchFailed { message: message.to_owned() };
    }
    let lower = message.to_lowercase();
    let mentions_worktree = lower.contains("worktree");
    let resembles_branch_resolve = lower.contains("failed to resolve base branch");
    let resembles_already_exists = lower.contains("already used by worktree");
    if (mentions_worktree || resembles_branch_resolve || resembles_already_exists)
        && is_git_repo_at_spawn
    {
        return WorkerSpawnError::WorktreeCreationFailed { reason: message.to_owned() };
    }
    WorkerSpawnError::DispatchFailed { message: message.to_owned() }
}

/// Caller's project + lead-or-not flag. Returned by
/// [`WorkerFacade::caller_project`].
#[derive(Debug, Clone)]
pub struct CallerProject {
    pub project_key: crate::ProjectKey,
    pub is_lead: bool,
}

/// Whether a project's worker cap came from its
/// `[[orgs.projects]]` `max_workers` override or the built-in default.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WorkerCapSource {
    ProjectMaxWorkers,
    Default,
}

/// Point-in-time worker-capacity read for the caller's project: the
/// cap the spawn path enforces, how many workers are live under it,
/// and where the cap came from. Produced by
/// [`WorkerFacade::capacity`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct WorkerCapacity {
    pub project: String,
    pub cap: usize,
    pub live: usize,
    pub cap_source: WorkerCapSource,
}

/// The narrow workspace-state surface workers MCP Tool impls
/// depend on.
///
/// `spawn_worker` is async because the production impl awaits a
/// oneshot receiver wired into `Command::SpawnWorker`. The remaining
/// methods are synchronous (one mutex acquire over in-memory state).
#[async_trait::async_trait]
pub trait WorkerFacade: Send + Sync {
    /// Resolve the caller's project key + lead/worker flag.
    /// Returns `None` when `caller` matches no known session.
    fn caller_project(&self, caller: &SessionSlot) -> Option<CallerProject>;

    /// Dispatch a `Command::SpawnWorker` and await its synchronous
    /// reply. Gating (lead-only, non-empty label, non-empty charter)
    /// happens before dispatch. `resume_kick` is persisted rather than
    /// delivered now; it replaces the generic restart note when this
    /// worker resumes after a forge restart with a turn that was cut off.
    /// A worker that was idle when forge stopped resumes with nothing
    /// delivered. `interactive` keeps the built-in
    /// `AskUserQuestion` tool, which every other worker is denied.
    /// `resume_session` resumes the label's most recent prior session
    /// instead of starting fresh. A label with none resolves to
    /// `ResumeTarget::None`, which starts a fresh session and says so in
    /// the reply's [`SessionChoice`]; only a lookup that failed is an
    /// error.
    ///
    /// `cwd` names the tree the worker should run in when it is not the
    /// project's own - an absolute or `~`-prefixed path, resolved to its
    /// repo root. `None` keeps the project's path for a fresh spawn, and
    /// adopts the row's recorded tree for a resume that states none.
    async fn spawn_worker(
        &self,
        caller: &SessionSlot,
        label: String,
        charter: String,
        kick: Option<String>,
        resume_kick: Option<String>,
        interactive: bool,
        resume_session: bool,
        mcp_families: Option<Vec<String>>,
        cwd: Option<String>,
    ) -> Result<WorkerSpawnReply, WorkerSpawnError>;

    /// Merge the supplied fields onto the stored dynamic-worker row for
    /// `label` in the caller's project, leaving a `None` field at its
    /// stored value. Errors with `NoSuchWorker` when no row exists:
    /// update revises an existing worker, it never creates one. Takes
    /// effect on that worker's next respawn, since a session's system
    /// prompt and tool list are fixed when the session spawns.
    /// `mcp_families` follows the same contract: `None` leaves the
    /// stored selection alone, `Some(names)` replaces it - canonicalised
    /// by the tool - and `Some(empty)` resets it to every family.
    fn update_worker(
        &self,
        caller: &SessionSlot,
        label: &str,
        charter: Option<String>,
        kick: Option<String>,
        resume_kick: Option<String>,
        mcp_families: Option<Vec<String>>,
    ) -> Result<(), WorkerUpdateError>;

    /// Dispatch a `Command::DespawnWorker` and await its result.
    /// Gating (lead-only, non-empty label) happens before dispatch.
    /// `force` removes a dirty worktree; without it a dirty worktree
    /// blocks the despawn (`DespawnOutcome::Blocked`).
    async fn despawn_worker(
        &self,
        caller: &SessionSlot,
        label: &str,
        force: bool,
    ) -> Result<DespawnOutcome, WorkerDespawnError>;

    /// Snapshot of every worker in the caller's project. Returns an
    /// empty Vec when the caller resolves to no project.
    fn list_workers(&self, caller: &SessionSlot) -> Vec<WorkerStatus>;

    /// Snapshot of every project's workers: the real workspace walks
    /// `list_projects` in config order, the mock sorts its project keys.
    fn list_all_workers(&self) -> Vec<WorkerStatus>;

    /// The caller's project worker capacity: the cap the spawn path
    /// enforces (`max_workers` in the project's `[[orgs.projects]]`
    /// entry, else `DEFAULT_MAX_WORKERS_PER_PROJECT`) paired with the
    /// live count.
    /// A plain data read - the same one the spawn path's cap check
    /// performs. Returns `None` when the caller resolves to no
    /// project.
    fn capacity(&self, caller: &SessionSlot) -> Option<WorkerCapacity>;

    /// Dispatch a worker-bound wrapped prompt. `Delivered` means the
    /// prompt was accepted - handed to the target's session queue, held
    /// for its turn end, or parked for its connect - which is the whole
    /// claim the sender's ack makes; anything that reached no session is
    /// an error. `UnknownLabel` when no live worker matches.
    async fn deliver_worker_prompt(
        &self,
        caller: &SessionSlot,
        target_label: &str,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerDeliverError>;

    /// Dispatch a worker-bound wrapped prompt to a project named by
    /// its `(org, project)` pair rather than the caller's own. The
    /// caller lives in another project, so the target cannot be
    /// resolved from it. Mirrors `deliver_worker_prompt` otherwise:
    /// an unknown label is an error and a sleeping target is not
    /// auto-spawned.
    async fn deliver_worker_prompt_to_project(
        &self,
        caller: &SessionSlot,
        org: &str,
        project: &str,
        target_label: &str,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerDeliverError>;

    /// Dispatch a wrapped prompt from a worker back to its lead.
    /// Caller MUST be a worker. The target lead is resolved from the
    /// caller's `WorkerEntry::spawned_by_session_id`. Same
    /// wire-shape contract as `deliver_worker_prompt`: `Delivered`
    /// means the lead's session accepted it, and one the session could
    /// not take is an error.
    async fn deliver_prompt_to_lead(
        &self,
        caller: &SessionSlot,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerLeadDeliverError>;
}

/// The refusal for a label that cannot take a delivery: `None` when a live
/// worker matches, else the error naming which absence this is - a label
/// that is not in the project's pool at all, or a row kept because the
/// worker crashed. The two read very differently to the sender: one says
/// "check the label", the other "respawn it".
fn refused_delivery(
    workers: &[crate::mcp::workers::types::WorkerEntry],
    project_key: &str,
    label: &str,
) -> Option<WorkerDeliverError> {
    if crate::mcp::workers::types::live_worker_with_label(workers, label).is_some() {
        return None;
    }
    if workers.iter().any(|w| w.label == label) {
        return Some(WorkerDeliverError::NotLive {
            project_key: project_key.to_owned(),
            label: label.to_owned(),
        });
    }
    Some(WorkerDeliverError::UnknownLabel {
        project_key: project_key.to_owned(),
        label: label.to_owned(),
    })
}

/// Await a delivery's acceptance answer, mapping it to the caller's
/// ack. The handler answers `Ok(())` once the prompt was accepted and
/// `Err(reason)` when nothing could take it, so this is the one place
/// the ack is decided.
async fn await_delivery(
    rx: tokio::sync::oneshot::Receiver<Result<(), String>>,
) -> Result<WorkerTargetStatus, WorkerDeliverError> {
    match rx.await {
        Ok(Ok(())) => Ok(WorkerTargetStatus::Delivered),
        Ok(Err(message)) => Err(WorkerDeliverError::DispatchFailed { message }),
        Err(_) => Err(WorkerDeliverError::DispatchFailed {
            message: "the delivery handler dropped its reply".to_owned(),
        }),
    }
}

impl WorkerSpawnError {
    /// The sentence a caller reads. Shared with the spawn handler, which the
    /// command bus and the socket reach without the tool's own validation,
    /// so both seams refuse a label with the same words.
    pub(crate) fn message(&self) -> String {
        match self {
            Self::NotLeadCaller => {
                "agents__spawn is lead-only; this session is a worker. Workers cannot \
                 spawn other workers in v1."
                    .to_owned()
            }
            Self::EmptyLabel => "label must be non-empty after trim".to_owned(),
            Self::ReservedLabel => format!(
                "label '{LEAD_LABEL}' is reserved - agents__send_message uses it as \
                 the addressing keyword for the caller's project's own agent. Pick a different label."
            ),
            Self::LabelWithPathSeparator => {
                "label must not contain a slash ('/'): claude rewrites it in the worktree name, \
                 so forge's path and branch would point somewhere it never creates. Pick a label \
                 without one."
                    .to_owned()
            }
            Self::EmptyCharter => "charter must be non-empty after trim".to_owned(),
            Self::UnknownCallerProject => {
                "could not resolve caller to a known project (forge bug)".to_owned()
            }
            Self::DispatchFailed { message } => format!("worker spawn failed: {message}"),
            Self::WorktreeCreationFailed { reason } => {
                format!("worktree creation failed: {reason}")
            }
            Self::ResumeLookupFailed { label, message } => format!(
                "could not look up a prior session for '{label}': {message}. Nothing was spawned, \
                 because a failed lookup is not the same answer as no prior session; retry, or \
                 spawn without resume_session to start fresh"
            ),
        }
    }
}

/// The label rules a spawn must pass wherever it enters: empty, the
/// reserved `lead`, and a slash. Shared by the tool's validation and the
/// spawn handler so the MCP tool, the command bus and the socket cannot
/// disagree about which labels mint a worker.
pub(crate) fn validate_worker_label(label: &str) -> Result<(), WorkerSpawnError> {
    if label.trim().is_empty() {
        return Err(WorkerSpawnError::EmptyLabel);
    }
    if label.trim() == LEAD_LABEL {
        return Err(WorkerSpawnError::ReservedLabel);
    }
    if label.contains('/') {
        return Err(WorkerSpawnError::LabelWithPathSeparator);
    }
    Ok(())
}

/// Validation chain shared by the production and mock `spawn_worker`
/// impls so tests exercise the real rules rather than a hand-copied
/// duplicate. `is_lead` is the resolved caller role.
pub(super) fn validate_worker_spawn(
    is_lead: bool,
    label: &str,
    charter: &str,
) -> Result<(), WorkerSpawnError> {
    if !is_lead {
        return Err(WorkerSpawnError::NotLeadCaller);
    }
    validate_worker_label(label)?;
    if charter.trim().is_empty() {
        return Err(WorkerSpawnError::EmptyCharter);
    }
    Ok(())
}

/// Undo the worktree a refused spawn's `ensure_worker_worktree` step
/// created, so a refusal strands neither worktree nor branch. `ensured`
/// is `None` when this spawn never touched a worktree.
///
/// A concurrent spawn anywhere in the repository that won this tree holds
/// it now - a claim, or a live seat, in ANY project - so the rollback
/// leaves it alone.
fn discard_refused_worktree(
    ws: &Workspace,
    project_key: &crate::target::ProjectKey,
    repo: &std::path::Path,
    label: &str,
    ensured: Option<(std::path::PathBuf, forge_agent::env::worktree::WorktreeEnsure)>,
) {
    let Some((worktree, outcome)) = ensured else { return };
    if let Some(holder) = ws.worktree_held_elsewhere(&worktree, project_key, label) {
        tracing::debug!(
            target: "forge_workspace::mcp::workers",
            %label,
            worktree = %worktree.display(),
            %holder,
            "a refused spawn's worktree is held by another seat; the rollback leaves it",
        );
        return;
    }
    if let Some(warning) =
        forge_agent::env::worktree::discard_worker_worktree(repo, label, &worktree, outcome)
    {
        tracing::warn!(
            target: "forge_workspace::mcp::workers",
            %label,
            worktree = %worktree.display(),
            %warning,
            "a refused spawn's fresh worktree stayed behind"
        );
    }
}

/// Production impl. Holds a `Weak<Workspace>` so construction doesn't
/// close a strong cycle through the Workspace -> bridge -> MCP ->
/// Tool -> facade -> Workspace path. Every method starts with
/// `upgrade()` and short-circuits when the workspace has been
/// dropped (only possible during shutdown).
pub struct ProdWorkerFacade {
    workspace: Weak<Workspace>,
}

impl ProdWorkerFacade {
    pub fn from_arc(workspace: &Arc<Workspace>) -> Arc<dyn WorkerFacade> {
        Arc::new(Self { workspace: Arc::downgrade(workspace) })
    }
}

#[async_trait::async_trait]
impl WorkerFacade for ProdWorkerFacade {
    fn caller_project(&self, caller: &SessionSlot) -> Option<CallerProject> {
        let ws = self.workspace.upgrade()?;
        let cx = crate::mcp::caller_context::caller_context(&ws, caller)?;
        Some(CallerProject { project_key: cx.project_key, is_lead: cx.is_lead })
    }

    async fn spawn_worker(
        &self,
        caller: &SessionSlot,
        label: String,
        charter: String,
        kick: Option<String>,
        resume_kick: Option<String>,
        interactive: bool,
        resume_session: bool,
        mcp_families: Option<Vec<String>>,
        cwd: Option<String>,
    ) -> Result<WorkerSpawnReply, WorkerSpawnError> {
        let cp = self.caller_project(caller).ok_or(WorkerSpawnError::UnknownCallerProject)?;
        validate_worker_spawn(cp.is_lead, &label, &charter)?;
        let ws = self.workspace.upgrade().ok_or_else(|| WorkerSpawnError::DispatchFailed {
            message: "workspace dropped".into(),
        })?;
        // A label whose despawn cleanup is still running is refused here,
        // before anything mints a worktree: that cleanup is deleting the
        // directory a resume would re-create, and re-attaching its branch
        // would leave the despawn reporting a branch it could not reap - a
        // failure this spawn made. The spawn handler reads the same mark;
        // this half is what keeps a refused spawn from minting anything.
        if ws.despawn_cleanup_pending(&cp.project_key, &label) {
            return Err(WorkerSpawnError::DispatchFailed {
                message:
                    "this label's previous despawn is still cleaning up; retry once it finishes"
                        .to_owned(),
            });
        }
        // The project view's path answers the is_git_repo probe the
        // failure classifier reads (the WorkerEntry is gone by the time
        // we see a spawn error).
        let Some(view) = ws.list_projects().into_iter().find(|v| v.key == cp.project_key) else {
            return Err(WorkerSpawnError::UnknownCallerProject);
        };
        // D7's fallback: a resume that states no tree resumes the one the
        // row records, so the session lands back where its JSONL lives.
        // A fresh spawn stating none keeps the project's own path (D9).
        // The row is read once here: its tree is both that fallback and
        // what a stated tree is judged against below.
        let row = if resume_session {
            ws.stored_worker_row(&cp.project_key, &label).unwrap_or_else(|error| {
                tracing::warn!(
                    target: "forge_workspace::mcp::workers",
                    event_name = "worker_row_cwd_unreadable",
                    project = %cp.project_key.as_str(),
                    label = %label,
                    %error,
                    "reading the worker's recorded tree failed; resuming in the project \
                     path, which may be a different directory than the row names",
                );
                None
            })
        } else {
            None
        };
        let stated = cwd
            .as_deref()
            .map(str::to_owned)
            .or_else(|| row.as_ref().and_then(|row| row.cwd.clone()));
        // The tree this spawn runs the worker in: the stated path resolved
        // to its repo root, else the project's own path. Resolving here is
        // what makes the ensure, the run directory and the record agree -
        // they all compose from `root` below.
        let (root, stated_is_git) =
            crate::mcp::workers::types::worker_root_and_git(&view.path, stated.as_deref())
                .map_err(|message| WorkerSpawnError::DispatchFailed { message })?;
        // D3 again, before anything is ensured: the handler refuses this
        // as the shared core, but it runs after this half has already
        // recreated a worktree - and a refused spawn must not write into
        // a declared project's tree even transiently.
        if stated.is_some()
            && let Some(known) = ws.project_name_for_resolved_path(&root.to_string_lossy())
        {
            return Err(WorkerSpawnError::DispatchFailed {
                message: crate::spawn::known_project_refusal(&known, &view.name),
            });
        }
        // A resume that states a DIFFERENT tree than the row's cannot
        // carry the history: claude finds a conversation by the project
        // key derived from the directory it runs in, so a moved resume
        // dies "No conversation found". The worker's history is keyed to
        // its tree, so moving means despawn and a fresh spawn.
        if resume_session && let (Some(_), Some(row)) = (cwd.as_deref(), row.as_ref()) {
            let history =
                row.cwd.as_deref().map_or_else(|| view.path.clone(), std::path::PathBuf::from);
            let history = std::fs::canonicalize(&history).unwrap_or(history);
            if root != history {
                return Err(WorkerSpawnError::DispatchFailed {
                    message: format!(
                        "this resume states a different tree ('{}') than the one this \
                         worker's history lives in ('{}'); a conversation is found by the \
                         directory it ran in, so a worker moves trees only by being \
                         despawned and spawned fresh",
                        root.display(),
                        history.display(),
                    ),
                });
            }
        }
        // A row that already exists answers for gitness, exactly as it
        // does in `handle_spawn_worker`, so the worktree ensured below is
        // the one the spawn will actually run in. Probing here instead
        // would diverge on a project whose repo-ness changed since the
        // worker was spawned: the ensure would create a worktree nothing
        // uses and nothing cleans up, or skip one the resume needs. A
        // stated tree answers for itself: the row's gitness belongs to the
        // directory the row names.
        let is_git_repo_at_spawn = if let Some(stated_is_git) = stated_is_git {
            stated_is_git
        } else {
            ws.recorded_worker_is_git_repo(&cp.project_key, &label)
                .unwrap_or_else(|error| {
                    tracing::warn!(
                        target: "forge_workspace::mcp::workers",
                        event_name = "worker_row_gitness_unreadable",
                        project = %cp.project_key.as_str(),
                        label = %label,
                        %error,
                        "reading the worker's recorded gitness failed; probing the project path, \
                         which may ensure a different worktree than the row names",
                    );
                    None
                })
                .unwrap_or_else(|| forge_agent::env::worktree::is_git_repo(&view.path))
        };
        // The one-live-worker-per-label guard lives in the shared
        // `handle_spawn_worker` core, so a boot re-spawn is deduped
        // against this one; a duplicate MCP spawn surfaces from there as
        // a dispatch error.
        //
        // The worktree is recreated before the spawn: a despawn removes
        // it, and `--worktree <label>` needs it to exist so the resumed
        // session lands back in the directory it first ran in. What
        // ensure did is kept so a refusal can undo exactly that work -
        // and, on the fallback below, so the fresh session inherits it.
        let mut ensured = None;
        // The checkout the worker's worktree belongs to: the CLI's
        // canonical root, which is the main checkout when the project is
        // itself a linked worktree (#2460). Every worktree operation here
        // - the ensure, the leftover clear under it, and a refusal's
        // discard - aims at it, not at the project path.
        let base = forge_agent::env::worktree::worktree_base(&root);
        // Held until this function returns, after the core has inserted
        // the worker or refused it.
        let mut _worktree_claim: Option<crate::workspace::WorktreeClaimGuard> = None;
        let (resume_existing, session_choice) = if resume_session {
            // The directory the label's sessions ran in: its worktree, or
            // the project root for a worker spawned without one. The
            // ensure, the tag write and the lookup below all take it from
            // `worker_tag_dir`, so they cannot disagree about where a
            // worker lives.
            let run_dir =
                crate::mcp::workers::types::worker_tag_dir(&root, &label, is_git_repo_at_spawn);
            if is_git_repo_at_spawn {
                // Claim the path before the ensure or the transcript
                // lookup can touch it, and hold it through the dispatch:
                // the claim is the atomic half of the collision check, so
                // a concurrent spawn in another project cannot pass
                // between the check and this spawn's insert.
                let claim = ws
                    .claim_worktree(
                        &run_dir,
                        &cp.project_key,
                        &label,
                        crate::workspace::WorktreeOp::Spawn,
                    )
                    .map_err(|reason| WorkerSpawnError::WorktreeCreationFailed {
                        reason: format!("this worker's worktree cannot be shared: {reason}"),
                    })?;
                // Refuse before the ensure or the transcript lookup can
                // touch a tree that belongs to another project (#2460):
                // the base is repository-global, so the label alone does
                // not identify this project's worktree. The ensure is not
                // read-only - it clears a deregistered leftover and
                // rewrites a stale link - and the lookup would read that
                // project's conversation.
                if let Some(collision) = ws.worktree_collision(
                    &run_dir,
                    &cp.project_key,
                    &label,
                    crate::workspace::WorktreeOp::Spawn,
                ) {
                    return Err(WorkerSpawnError::WorktreeCreationFailed {
                        reason: format!("this worker's worktree cannot be shared: {collision}"),
                    });
                }
                // Released when this function returns, after the core has
                // inserted the worker or refused it.
                _worktree_claim = Some(claim);
                match forge_agent::env::worktree::ensure_worker_worktree(&base, &label, &run_dir) {
                    Ok(outcome) => ensured = Some((run_dir.clone(), outcome)),
                    Err(err) => {
                        return Err(WorkerSpawnError::WorktreeCreationFailed {
                            reason: err.to_string(),
                        });
                    }
                }
            }
            match ws.resolve_worker_resume_session(&view.org, &view.name, &run_dir, &label) {
                Ok(ResumeTarget::Found { session_id, restamp }) => {
                    if restamp {
                        // A pre-upgrade tag row carries no project; stamp
                        // it now, before the spawn, so the migration
                        // window is this one adoption and not every
                        // project's to reuse.
                        if let Err(error) = ws.stamp_adopted_worker_transcript(
                            &cp.project_key,
                            &label,
                            &session_id,
                            &run_dir,
                        ) {
                            // Like every other failure after the ensure:
                            // undo the worktree it minted.
                            discard_refused_worktree(
                                &ws,
                                &cp.project_key,
                                &base,
                                &label,
                                ensured.take(),
                            );
                            return Err(WorkerSpawnError::DispatchFailed {
                                message: format!(
                                    "stamping the adopted transcript's project failed: {error}"
                                ),
                            });
                        }
                    }
                    (Some(session_id), SessionChoice::Resumed)
                }
                // Best effort: a label with nothing to resume starts a
                // new session and the reply says so. The worktree the
                // ensure minted stays - it is where the new session runs.
                Ok(ResumeTarget::None) => (None, SessionChoice::FreshWithoutPrior),
                // A lookup that failed is not an absence. Starting fresh
                // on it would answer a question the lookup never
                // answered, leaving the caller holding context it does
                // not have.
                Err(error) => {
                    discard_refused_worktree(&ws, &cp.project_key, &base, &label, ensured.take());
                    return Err(WorkerSpawnError::ResumeLookupFailed {
                        label,
                        message: error.to_string(),
                    });
                }
            }
        } else {
            (None, SessionChoice::Fresh)
        };

        // The worker's row is written by the spawn handler itself, from
        // the same arguments, so there is nothing to persist here - and
        // one write path covers the boot re-spawn too.
        let spawn_label = label.clone();
        let (tx, rx) = tokio::sync::oneshot::channel();
        let cmd = Command::SpawnWorker {
            project_key: cp.project_key.clone(),
            label,
            charter,
            spawned_by: caller.clone(),
            // Fresh unless the caller asked to resume; the resume
            // resolution above fills in the real session id.
            resume_existing,
            kick,
            resume_kick,
            interactive,
            mcp_families,
            // The resolved root, which the handler records on the row: a
            // spawn that stated no tree records none, so the row keeps
            // meaning the project's own path.
            cwd: stated.as_ref().map(|_| root.to_string_lossy().into_owned()),
            from_boot_respawn: false,
            return_to: Some(tx),
        };
        if let Err(err) = ws.dispatch(cmd) {
            discard_refused_worktree(&ws, &cp.project_key, &base, &spawn_label, ensured.take());
            return Err(WorkerSpawnError::DispatchFailed {
                message: format!("dispatch failed: {err:?}"),
            });
        }
        match rx.await {
            Ok(Ok(mut reply)) => {
                reply.session_choice = session_choice;
                if let Some(warning) = reply.durability_warning.take() {
                    tracing::error!(
                        target: "forge_workspace::mcp::workers",
                        project = %cp.project_key.as_str(),
                        label = %spawn_label,
                        warning = %warning,
                        "the worker's row could not be recorded; it will not survive a forge restart",
                    );
                    reply.durability_warning = Some(warning);
                }
                Ok(reply)
            }
            Ok(Err(message)) => {
                discard_refused_worktree(&ws, &cp.project_key, &base, &spawn_label, ensured.take());
                // The sync path's reply is a string the spawn handler
                // built, not an SDK error, so there is no variant to
                // name and the heuristics answer.
                Err(classify_worker_spawn_failure(
                    &message,
                    is_git_repo_at_spawn,
                    SpawnFailureKind::Unclassified,
                ))
            }
            // No rollback here: a dropped reply means the command may
            // still run, and the spawn it would do needs the worktree.
            Err(_) => Err(WorkerSpawnError::DispatchFailed {
                message: "spawn handler dropped reply channel".into(),
            }),
        }
    }

    fn update_worker(
        &self,
        caller: &SessionSlot,
        label: &str,
        charter: Option<String>,
        kick: Option<String>,
        resume_kick: Option<String>,
        mcp_families: Option<Vec<String>>,
    ) -> Result<(), WorkerUpdateError> {
        let cp = self.caller_project(caller).ok_or(WorkerUpdateError::UnknownCallerProject)?;
        let ws = self.workspace.upgrade().ok_or_else(|| WorkerUpdateError::StoreFailed {
            message: "workspace dropped".into(),
        })?;
        match ws.update_worker_row(&cp.project_key, label, charter, kick, resume_kick, mcp_families)
        {
            Ok(true) => Ok(()),
            Ok(false) => Err(WorkerUpdateError::NoSuchWorker {
                label: label.to_owned(),
                project_key: cp.project_key.as_str().to_owned(),
            }),
            Err(err) => Err(WorkerUpdateError::StoreFailed { message: err.to_string() }),
        }
    }

    async fn despawn_worker(
        &self,
        caller: &SessionSlot,
        label: &str,
        force: bool,
    ) -> Result<DespawnOutcome, WorkerDespawnError> {
        let cp = self.caller_project(caller).ok_or(WorkerDespawnError::UnknownCallerProject)?;
        if !cp.is_lead {
            return Err(WorkerDespawnError::NotLeadCaller);
        }
        if label.trim().is_empty() {
            return Err(WorkerDespawnError::EmptyLabel);
        }
        let ws = self.workspace.upgrade().ok_or_else(|| WorkerDespawnError::DispatchFailed {
            message: "workspace dropped".into(),
        })?;
        let (tx, rx) = tokio::sync::oneshot::channel();
        let cmd = Command::DespawnWorker {
            project_key: cp.project_key.clone(),
            label: label.to_owned(),
            force,
            respond: Some(tx),
        };
        if let Err(err) = ws.dispatch(cmd) {
            return Err(WorkerDespawnError::DispatchFailed {
                message: format!("dispatch failed: {err:?}"),
            });
        }
        match rx.await {
            Ok(crate::protocol::DespawnResult::Despawned {
                worktree_cleanup_warning,
                branch_cleanup_warning,
            }) => {
                Ok(DespawnOutcome::Despawned { worktree_cleanup_warning, branch_cleanup_warning })
            }
            Ok(crate::protocol::DespawnResult::Blocked { reason }) => {
                Ok(DespawnOutcome::Blocked { reason })
            }
            Ok(crate::protocol::DespawnResult::NotFound) => Err(WorkerDespawnError::UnknownLabel {
                label: label.to_owned(),
                project_key: cp.project_key.as_str().to_owned(),
            }),
            Ok(crate::protocol::DespawnResult::Failed { reason }) => {
                Err(WorkerDespawnError::DispatchFailed { message: reason })
            }
            Err(_) => Err(WorkerDespawnError::DispatchFailed {
                message: "despawn handler dropped reply channel".into(),
            }),
        }
    }

    fn list_workers(&self, caller: &SessionSlot) -> Vec<WorkerStatus> {
        let Some(cp) = self.caller_project(caller) else {
            return Vec::new();
        };
        let Some(ws) = self.workspace.upgrade() else {
            return Vec::new();
        };
        ws.list_live_workers(&cp.project_key)
            .iter()
            .map(|entry| ws.worker_status_snapshot(entry))
            .collect()
    }

    fn list_all_workers(&self) -> Vec<WorkerStatus> {
        let Some(ws) = self.workspace.upgrade() else {
            return Vec::new();
        };
        let mut all = Vec::new();
        for project in ws.list_projects() {
            for entry in ws.list_live_workers(&project.key) {
                all.push(ws.worker_status_snapshot(&entry));
            }
        }
        all
    }

    fn capacity(&self, caller: &SessionSlot) -> Option<WorkerCapacity> {
        let cp = self.caller_project(caller)?;
        let ws = self.workspace.upgrade()?;
        let (cap, cap_source) = ws
            .project_for_key(&cp.project_key)
            .and_then(|project| {
                project.max_workers.map(|cap| (cap, WorkerCapSource::ProjectMaxWorkers))
            })
            .unwrap_or((crate::config::DEFAULT_MAX_WORKERS_PER_PROJECT, WorkerCapSource::Default));
        Some(WorkerCapacity {
            project: cp.project_key.as_str().to_owned(),
            cap,
            live: ws.count_live_workers(&cp.project_key),
            cap_source,
        })
    }

    async fn deliver_worker_prompt(
        &self,
        caller: &SessionSlot,
        target_label: &str,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerDeliverError> {
        let cp = self.caller_project(caller).ok_or_else(|| WorkerDeliverError::UnknownLabel {
            project_key: "<unknown>".into(),
            label: target_label.into(),
        })?;
        let Some(ws) = self.workspace.upgrade() else {
            return Err(WorkerDeliverError::UnknownLabel {
                project_key: cp.project_key.as_str().into(),
                label: target_label.into(),
            });
        };
        // A crashed worker keeps its row (Failed) so the label stays
        // respawnable; that row is not addressable, and the refusal says so
        // rather than acking a prompt no session can take.
        if let Some(refusal) = refused_delivery(
            &ws.list_live_workers(&cp.project_key),
            cp.project_key.as_str(),
            target_label,
        ) {
            return Err(refusal);
        }
        let (tx, rx) = tokio::sync::oneshot::channel();
        if let Err(err) = ws.dispatch(Command::DeliverWorkerPrompt {
            caller: caller.clone(),
            project_key: cp.project_key,
            target_label: target_label.into(),
            wrapped,
            respond: Some(tx),
        }) {
            return Err(WorkerDeliverError::DispatchFailed {
                message: format!("dispatch failed: {err:?}"),
            });
        }
        await_delivery(rx).await
    }

    async fn deliver_worker_prompt_to_project(
        &self,
        caller: &SessionSlot,
        org: &str,
        project: &str,
        target_label: &str,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerDeliverError> {
        let unknown = || WorkerDeliverError::UnknownLabel {
            project_key: project.into(),
            label: target_label.into(),
        };
        let Some(ws) = self.workspace.upgrade() else {
            return Err(unknown());
        };
        let Some(view) = ws.list_projects().into_iter().find(|v| v.org == org && v.name == project)
        else {
            return Err(unknown());
        };
        if let Some(refusal) =
            refused_delivery(&ws.list_live_workers(&view.key), project, target_label)
        {
            return Err(refusal);
        }
        let (tx, rx) = tokio::sync::oneshot::channel();
        if let Err(err) = ws.dispatch(Command::DeliverWorkerPrompt {
            caller: caller.clone(),
            project_key: view.key,
            target_label: target_label.into(),
            wrapped,
            respond: Some(tx),
        }) {
            return Err(WorkerDeliverError::DispatchFailed {
                message: format!("dispatch failed: {err:?}"),
            });
        }
        await_delivery(rx).await
    }

    async fn deliver_prompt_to_lead(
        &self,
        caller: &SessionSlot,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerLeadDeliverError> {
        let Some(ws) = self.workspace.upgrade() else {
            return Err(WorkerLeadDeliverError::UnknownCaller);
        };
        // Resolve the caller's project + current lead via the shared
        // helper. The previous impl read `entry.spawned_by_session_id`
        // (set once at spawn time, never rewritten on lead-resume) -
        // tells routed to the dead pre-resume session id and were
        // silently dropped. #298 Cause 2.
        let Some(cx) = crate::mcp::caller_context::caller_context(&ws, caller) else {
            return Err(WorkerLeadDeliverError::UnknownCaller);
        };
        if cx.is_lead {
            return Err(WorkerLeadDeliverError::LeadCallerHasNoLead);
        }
        let target_lead_key = cx.lead.clone();
        // Defensive: confirm the lead's session is still in the pool
        // before dispatching. If it closed since the worker was
        // spawned, surface a clear error so the worker LLM can adapt.
        if !cx.lead_running {
            return Err(WorkerLeadDeliverError::LeadGone);
        }
        let (tx, rx) = tokio::sync::oneshot::channel();
        if let Err(err) = ws.dispatch(Command::DeliverWorkerPromptToLead {
            caller: caller.clone(),
            target_lead_key,
            wrapped,
            respond: Some(tx),
        }) {
            return Err(WorkerLeadDeliverError::DispatchFailed {
                message: format!("dispatch failed: {err:?}"),
            });
        }
        match rx.await {
            Ok(Ok(())) => Ok(WorkerTargetStatus::Delivered),
            Ok(Err(message)) => Err(WorkerLeadDeliverError::DispatchFailed { message }),
            Err(_) => Err(WorkerLeadDeliverError::DispatchFailed {
                message: "the delivery handler dropped its reply".to_owned(),
            }),
        }
    }
}

/// A captured `MockWorkerFacade::spawn_worker` call:
/// `(caller, label, resolved charter, kick, resume_kick, interactive,
/// resume_session, mcp_families, cwd)`.
#[cfg(any(test, feature = "testing"))]
type RecordedSpawnCall = (
    SessionSlot,
    String,
    String,
    Option<String>,
    Option<String>,
    bool,
    bool,
    Option<Vec<String>>,
    Option<String>,
);

/// A captured `MockWorkerFacade::update_worker` call:
/// `(caller, label, charter, kick, resume_kick)`.
#[cfg(any(test, feature = "testing"))]
type RecordedUpdateCall =
    (SessionSlot, String, Option<String>, Option<String>, Option<String>, Option<Vec<String>>);

/// One captured `deliver_worker_prompt_to_project` call: the address it
/// was given, and the prompt it carried.
#[cfg(any(test, feature = "testing"))]
#[derive(Debug)]
pub struct RecordedProjectDelivery {
    pub caller: SessionSlot,
    pub org: String,
    pub project: String,
    pub label: String,
    pub wrapped: WrappedPrompt,
}

/// Mock for unit-testing the four Tool impls. Captures every
/// dispatched call into a Vec so tests can assert "tool X
/// dispatched spawn with these args" without spinning up a real
/// Workspace.
#[cfg(any(test, feature = "testing"))]
#[derive(Default)]
pub struct MockWorkerFacade {
    /// Pre-loaded caller -> project mapping. Tests insert entries
    /// before invoking the tool.
    pub callers: parking_lot::Mutex<std::collections::HashMap<SessionSlot, CallerProject>>,
    /// Pre-loaded `live_workers` snapshot per project_key string.
    /// `list_workers` and `deliver_worker_prompt` both read from
    /// this.
    pub workers: parking_lot::Mutex<std::collections::HashMap<String, Vec<WorkerStatus>>>,
    /// Captured `spawn_worker` calls.
    pub spawn_calls: parking_lot::Mutex<Vec<RecordedSpawnCall>>,
    /// Pre-loaded reply for `spawn_worker`. When `None`, the mock
    /// returns `DispatchFailed { message: "no preloaded reply" }`.
    pub spawn_reply: parking_lot::Mutex<Option<Result<WorkerSpawnReply, WorkerSpawnError>>>,
    /// Captured `update_worker` calls.
    pub update_calls: parking_lot::Mutex<Vec<RecordedUpdateCall>>,
    /// Pre-loaded result for `update_worker`. When `None`, the mock
    /// reports success, so a test that cares only about the args passed
    /// through does not have to set it.
    pub update_result: parking_lot::Mutex<Option<Result<(), WorkerUpdateError>>>,
    /// Captured `deliver_worker_prompt` calls.
    pub deliver_calls: parking_lot::Mutex<Vec<(SessionSlot, String, WrappedPrompt)>>,
    /// Captured `deliver_worker_prompt_to_project` calls.
    pub deliver_to_project_calls: parking_lot::Mutex<Vec<RecordedProjectDelivery>>,
    /// Captured `despawn_worker` calls: (caller, label, force).
    pub despawn_calls: parking_lot::Mutex<Vec<(SessionSlot, String, bool)>>,
    /// Pre-loaded outcome for `despawn_worker` on a known label. When
    /// `None`, the mock reports `Despawned` with neither warning set.
    pub despawn_outcome: parking_lot::Mutex<Option<DespawnOutcome>>,
    /// Pre-loaded reply for `capacity`. When `None`, the mock derives
    /// it from the workers map: default cap, live count from the
    /// caller's project's entries.
    pub capacity_reply: parking_lot::Mutex<Option<WorkerCapacity>>,
}

#[cfg(any(test, feature = "testing"))]
impl MockWorkerFacade {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn into_arc(self) -> std::sync::Arc<dyn WorkerFacade> {
        std::sync::Arc::new(self)
    }
}

#[cfg(any(test, feature = "testing"))]
#[async_trait::async_trait]
impl WorkerFacade for MockWorkerFacade {
    fn caller_project(&self, caller: &SessionSlot) -> Option<CallerProject> {
        self.callers.lock().get(caller).cloned()
    }

    async fn spawn_worker(
        &self,
        caller: &SessionSlot,
        label: String,
        charter: String,
        kick: Option<String>,
        resume_kick: Option<String>,
        interactive: bool,
        resume_session: bool,
        mcp_families: Option<Vec<String>>,
        cwd: Option<String>,
    ) -> Result<WorkerSpawnReply, WorkerSpawnError> {
        let cp = self.caller_project(caller).ok_or(WorkerSpawnError::UnknownCallerProject)?;
        validate_worker_spawn(cp.is_lead, &label, &charter)?;
        self.spawn_calls.lock().push((
            caller.clone(),
            label,
            charter,
            kick,
            resume_kick,
            interactive,
            resume_session,
            mcp_families,
            cwd,
        ));
        self.spawn_reply.lock().clone().unwrap_or(Err(WorkerSpawnError::DispatchFailed {
            message: "no preloaded reply".into(),
        }))
    }

    fn update_worker(
        &self,
        caller: &SessionSlot,
        label: &str,
        charter: Option<String>,
        kick: Option<String>,
        resume_kick: Option<String>,
        mcp_families: Option<Vec<String>>,
    ) -> Result<(), WorkerUpdateError> {
        self.caller_project(caller).ok_or(WorkerUpdateError::UnknownCallerProject)?;
        self.update_calls.lock().push((
            caller.clone(),
            label.to_owned(),
            charter,
            kick,
            resume_kick,
            mcp_families,
        ));
        self.update_result.lock().clone().unwrap_or(Ok(()))
    }

    async fn despawn_worker(
        &self,
        caller: &SessionSlot,
        label: &str,
        force: bool,
    ) -> Result<DespawnOutcome, WorkerDespawnError> {
        let cp = self.caller_project(caller).ok_or(WorkerDespawnError::UnknownCallerProject)?;
        if !cp.is_lead {
            return Err(WorkerDespawnError::NotLeadCaller);
        }
        if label.trim().is_empty() {
            return Err(WorkerDespawnError::EmptyLabel);
        }
        let known = self
            .workers
            .lock()
            .get(cp.project_key.as_str())
            .is_some_and(|ws| ws.iter().any(|w| w.label == label));
        if !known {
            return Err(WorkerDespawnError::UnknownLabel {
                label: label.to_owned(),
                project_key: cp.project_key.as_str().to_owned(),
            });
        }
        self.despawn_calls.lock().push((caller.clone(), label.to_owned(), force));
        Ok(self.despawn_outcome.lock().clone().unwrap_or(DespawnOutcome::Despawned {
            worktree_cleanup_warning: None,
            branch_cleanup_warning: None,
        }))
    }

    fn list_workers(&self, caller: &SessionSlot) -> Vec<WorkerStatus> {
        let Some(cp) = self.caller_project(caller) else {
            return Vec::new();
        };
        self.workers.lock().get(cp.project_key.as_str()).cloned().unwrap_or_default()
    }

    fn list_all_workers(&self) -> Vec<WorkerStatus> {
        let workers = self.workers.lock();
        let mut projects: Vec<(&String, &Vec<WorkerStatus>)> = workers.iter().collect();
        projects.sort_by(|left, right| left.0.cmp(right.0));
        projects.into_iter().flat_map(|(_, list)| list).cloned().collect()
    }

    fn capacity(&self, caller: &SessionSlot) -> Option<WorkerCapacity> {
        let cp = self.caller_project(caller)?;
        Some(self.capacity_reply.lock().clone().unwrap_or(WorkerCapacity {
            project: cp.project_key.as_str().to_owned(),
            cap: crate::config::DEFAULT_MAX_WORKERS_PER_PROJECT,
            live: self.workers.lock().get(cp.project_key.as_str()).map_or(0, Vec::len),
            cap_source: WorkerCapSource::Default,
        }))
    }

    async fn deliver_worker_prompt(
        &self,
        caller: &SessionSlot,
        target_label: &str,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerDeliverError> {
        let cp = self.caller_project(caller).ok_or_else(|| WorkerDeliverError::UnknownLabel {
            project_key: "<unknown>".into(),
            label: target_label.into(),
        })?;
        let known = self
            .workers
            .lock()
            .get(cp.project_key.as_str())
            .is_some_and(|ws| ws.iter().any(|w| w.label == target_label));
        if !known {
            return Err(WorkerDeliverError::UnknownLabel {
                project_key: cp.project_key.as_str().into(),
                label: target_label.into(),
            });
        }
        self.deliver_calls.lock().push((caller.clone(), target_label.into(), wrapped));
        Ok(WorkerTargetStatus::Delivered)
    }

    async fn deliver_worker_prompt_to_project(
        &self,
        caller: &SessionSlot,
        org: &str,
        project: &str,
        target_label: &str,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerDeliverError> {
        let unknown = || WorkerDeliverError::UnknownLabel {
            project_key: project.into(),
            label: target_label.into(),
        };
        let known = self
            .workers
            .lock()
            .get(project)
            .is_some_and(|ws| ws.iter().any(|w| w.label == target_label));
        if !known {
            return Err(unknown());
        }
        self.deliver_to_project_calls.lock().push(RecordedProjectDelivery {
            caller: caller.clone(),
            org: org.into(),
            project: project.into(),
            label: target_label.into(),
            wrapped,
        });
        Ok(WorkerTargetStatus::Delivered)
    }

    async fn deliver_prompt_to_lead(
        &self,
        caller: &SessionSlot,
        wrapped: WrappedPrompt,
    ) -> Result<WorkerTargetStatus, WorkerLeadDeliverError> {
        let cp = self.caller_project(caller).ok_or(WorkerLeadDeliverError::UnknownCaller)?;
        if cp.is_lead {
            return Err(WorkerLeadDeliverError::LeadCallerHasNoLead);
        }
        // Mock surfaces a synthetic spawned_by lookup via the
        // preloaded workers map (entries carry spawned_by via
        // WorkerStatus). The Tool tests preload that field; the
        // failure modes (LeadGone, UnknownCaller) are still
        // reachable when the test omits the entry.
        let lead_slot = self
            .workers
            .lock()
            .get(cp.project_key.as_str())
            .and_then(|ws| ws.iter().find(|w| w.slot == *caller))
            .map(|w| w.spawned_by.clone());
        let Some(lead_slot) = lead_slot else {
            return Err(WorkerLeadDeliverError::UnknownCaller);
        };
        // Mock has no pool to consult; treat an unset `spawned_by` as
        // "lead gone" so tests can exercise that path explicitly.
        if lead_slot.label().is_empty() {
            return Err(WorkerLeadDeliverError::LeadGone);
        }
        // Record under the synthetic label `<lead>` so tests can
        // assert "the lead-bound delivery happened" without colliding
        // with a real worker label.
        self.deliver_calls.lock().push((caller.clone(), "<lead>".to_owned(), wrapped));
        Ok(WorkerTargetStatus::Delivered)
    }
}

#[cfg(test)]
mod mock_tests {
    use super::*;
    use crate::mcp::peers::types::WrappedKind;

    #[test]
    fn mock_caller_project_returns_preloaded() {
        let mock = MockWorkerFacade::new();
        mock.callers.lock().insert(
            SessionSlot::from_str_for_test("k1"),
            CallerProject { project_key: crate::ProjectKey::new("forge"), is_lead: true },
        );
        let cp = mock.caller_project(&SessionSlot::from_str_for_test("k1")).unwrap();
        assert!(cp.is_lead);
        assert_eq!(cp.project_key.as_str(), "forge");
    }

    #[tokio::test]
    async fn mock_spawn_rejects_non_lead() {
        let mock = MockWorkerFacade::new();
        mock.callers.lock().insert(
            SessionSlot::from_str_for_test("k1"),
            CallerProject { project_key: crate::ProjectKey::new("forge"), is_lead: false },
        );
        let res = mock
            .spawn_worker(
                &SessionSlot::from_str_for_test("k1"),
                "reviewer".into(),
                "charter".into(),
                None,
                None,
                false,
                false,
                None,
                None,
            )
            .await;
        assert!(matches!(res, Err(WorkerSpawnError::NotLeadCaller)));
    }

    #[tokio::test]
    async fn mock_spawn_records_call_and_returns_preloaded_reply() {
        let mock = MockWorkerFacade::new();
        mock.callers.lock().insert(
            SessionSlot::from_str_for_test("lead-key"),
            CallerProject { project_key: crate::ProjectKey::new("forge"), is_lead: true },
        );
        *mock.spawn_reply.lock() = Some(Ok(WorkerSpawnReply {
            session_id: "new-uuid".into(),
            tag: "forge:worker:reviewer".into(),
            mcp_families: None,
            rate_limited_account: None,
            durability_warning: None,
            worktree: None,
            session_choice: SessionChoice::Fresh,
        }));
        let res = mock
            .spawn_worker(
                &SessionSlot::from_str_for_test("lead-key"),
                "reviewer".into(),
                "charter".into(),
                None,
                None,
                false,
                false,
                None,
                None,
            )
            .await
            .unwrap();
        assert_eq!(res.session_id, "new-uuid");
        assert_eq!(mock.spawn_calls.lock().len(), 1);
    }

    #[test]
    fn validate_worker_spawn_enforces_lead_and_nonempty_fields() {
        // Prod + mock spawn_worker both route through this, so the rules
        // are covered against the shipping code, not a copy.
        assert!(matches!(
            validate_worker_spawn(false, "reviewer", "c"),
            Err(WorkerSpawnError::NotLeadCaller)
        ));
        assert!(matches!(
            validate_worker_spawn(true, "   ", "c"),
            Err(WorkerSpawnError::EmptyLabel)
        ));
        assert!(matches!(
            validate_worker_spawn(true, LEAD_LABEL, "c"),
            Err(WorkerSpawnError::ReservedLabel)
        ));
        // claude rewrites a slash before deriving the worktree's path and
        // branch, so a label carrying one names a directory and a branch
        // forge would never find (#1603).
        assert!(matches!(
            validate_worker_spawn(true, "a/b", "c"),
            Err(WorkerSpawnError::LabelWithPathSeparator)
        ));
        assert!(matches!(
            validate_worker_spawn(true, "reviewer", "   "),
            Err(WorkerSpawnError::EmptyCharter)
        ));
        assert!(validate_worker_spawn(true, "reviewer", "c").is_ok());
    }

    #[tokio::test]
    async fn mock_spawn_empty_inline_charter_errors_empty_charter() {
        let mock = MockWorkerFacade::new();
        let lead = SessionSlot::from_str_for_test("lead-key");
        mock.callers.lock().insert(
            lead.clone(),
            CallerProject { project_key: crate::ProjectKey::new("forge"), is_lead: true },
        );
        let res = mock
            .spawn_worker(
                &lead,
                "reviewer".into(),
                "   ".into(),
                None,
                None,
                false,
                false,
                None,
                None,
            )
            .await;
        assert!(matches!(res, Err(WorkerSpawnError::EmptyCharter)));
        assert_eq!(
            mock.spawn_calls.lock().len(),
            0,
            "empty inline charter must not record a spawn"
        );
    }

    #[test]
    fn dedup_finds_non_failed_label_ignores_failed() {
        use crate::mcp::workers::types::WorkerEntry;
        use forge_primitives::WorkerLiveness;
        use std::time::SystemTime;
        let entry = |label: &str, status: WorkerLiveness| WorkerEntry {
            label: label.to_owned(),
            charter: "c".into(),
            slot: SessionSlot::from_str_for_test("w-uuid"),
            session_id: None,
            status,
            spawned_at: SystemTime::UNIX_EPOCH,
            spawned_by: SessionSlot::from_str_for_test("lead"),
            needs_tag: false,
            is_git_repo_at_spawn: false,
            cwd: None,
            diagnostic: None,
            kick: None,
        };
        // Running / Spawning workers hold the label -> dedup blocks re-spawn.
        let running = vec![entry("implementer", WorkerLiveness::Running)];
        assert!(
            crate::mcp::workers::types::live_worker_with_label(&running, "implementer").is_some()
        );
        let spawning = vec![entry("implementer", WorkerLiveness::Spawning)];
        assert!(
            crate::mcp::workers::types::live_worker_with_label(&spawning, "implementer").is_some()
        );
        // A Failed entry is ignored -> its label can be re-spawned.
        let failed = vec![entry("implementer", WorkerLiveness::Failed)];
        assert!(
            crate::mcp::workers::types::live_worker_with_label(&failed, "implementer").is_none()
        );
        // No matching label.
        assert!(crate::mcp::workers::types::live_worker_with_label(&running, "tester").is_none());
    }

    #[tokio::test]
    async fn mock_deliver_unknown_label_errors() {
        let mock = MockWorkerFacade::new();
        let caller = SessionSlot::from_str_for_test("k1");
        mock.callers.lock().insert(
            caller.clone(),
            CallerProject { project_key: crate::ProjectKey::new("forge"), is_lead: true },
        );
        let wrapped = WrappedPrompt {
            id: crate::mcp::peers::types::MessageId::mint(),
            kind: WrappedKind::Message,
            sender_name: "forge".into(),
            sender_org: "Personal".into(),
            body: "hi".into(),
        };
        let res = mock.deliver_worker_prompt(&caller, "missing", wrapped).await;
        assert!(matches!(res, Err(WorkerDeliverError::UnknownLabel { .. })));
    }

    fn status(label: &str) -> WorkerStatus {
        WorkerStatus {
            label: label.into(),
            charter: format!("{label}'s charter"),
            status: forge_primitives::WorkerLiveness::Running,
            session_id: format!("{label}-uuid"),
            slot: SessionSlot::from_str_for_test(label),
            spawned_at: std::time::SystemTime::UNIX_EPOCH,
            spawned_by: SessionSlot::from_str_for_test("lead"),
            diagnostic: None,
            heal_attempt: None,
            heal_next_at: None,
            activity: None,
        }
    }

    #[test]
    fn mock_list_all_workers_flattens_every_project() {
        let mock = MockWorkerFacade::new();
        mock.workers.lock().insert("zeta".into(), vec![status("z1")]);
        mock.workers.lock().insert("alpha".into(), vec![status("a1"), status("a2")]);
        let mut labels: Vec<String> =
            mock.list_all_workers().into_iter().map(|worker| worker.label).collect();
        labels.sort();
        assert_eq!(labels, ["a1", "a2", "z1"], "every project's pool is flattened");
    }

    #[test]
    fn mock_capacity_fallback_derives_live_from_workers_map() {
        let mock = MockWorkerFacade::new();
        mock.callers.lock().insert(
            SessionSlot::from_str_for_test("k1"),
            CallerProject { project_key: crate::ProjectKey::new("forge"), is_lead: true },
        );
        mock.workers.lock().insert(
            "forge".into(),
            vec![
                WorkerStatus {
                    label: "a".into(),
                    charter: "c".into(),
                    status: forge_primitives::WorkerLiveness::Running,
                    session_id: "a-uuid".into(),
                    slot: SessionSlot::from_str_for_test("a-uuid"),
                    spawned_at: std::time::SystemTime::UNIX_EPOCH,
                    spawned_by: SessionSlot::from_str_for_test("lead"),
                    diagnostic: None,
                    heal_attempt: None,
                    heal_next_at: None,
                    activity: None,
                },
                WorkerStatus {
                    label: "b".into(),
                    charter: "c".into(),
                    status: forge_primitives::WorkerLiveness::Running,
                    session_id: "b-uuid".into(),
                    slot: SessionSlot::from_str_for_test("b-uuid"),
                    spawned_at: std::time::SystemTime::UNIX_EPOCH,
                    spawned_by: SessionSlot::from_str_for_test("lead"),
                    diagnostic: None,
                    heal_attempt: None,
                    heal_next_at: None,
                    activity: None,
                },
            ],
        );
        let capacity =
            mock.capacity(&SessionSlot::from_str_for_test("k1")).expect("caller resolves");
        assert_eq!(capacity.project, "forge");
        assert_eq!(capacity.cap, crate::config::DEFAULT_MAX_WORKERS_PER_PROJECT);
        assert_eq!(capacity.cap_source, WorkerCapSource::Default);
        assert_eq!(capacity.live, 2);
    }
}

/// The `agents__list` read path is the only producer of
/// `WorkerStatus::activity`, so it needs pinning here rather than only
/// on the projection it calls - reverting this one `map` to
/// `WorkerEntry::to_status` makes the whole field inert.
#[cfg(test)]
mod prod_list_workers_tests {
    use super::*;
    use forge_primitives::{SessionLifecycleState, WorkerLiveness};

    #[test]
    fn list_workers_populates_activity() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project("forge", "/tmp/forge");
        let project = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .expect("seeded project present")
            .key;

        // The caller IS the worker, which is enough for
        // `caller_context` to resolve it into the project.
        let caller = SessionSlot::worker("TestOrg", "forge", "implementer");
        ws.insert_live_worker(
            &project,
            crate::mcp::workers::types::WorkerEntry {
                label: "implementer".into(),
                charter: "test charter".into(),
                slot: caller.clone(),
                session_id: None,
                status: WorkerLiveness::Running,
                spawned_at: std::time::SystemTime::UNIX_EPOCH,
                spawned_by: SessionSlot::lead("TestOrg", "forge"),
                needs_tag: false,
                is_git_repo_at_spawn: false,
                cwd: None,
                diagnostic: None,
                kick: None,
            },
        );
        let domain = ws.register_domain_session(caller.clone(), None);

        let facade = ProdWorkerFacade::from_arc(&ws);
        let listed = facade.list_workers(&caller);
        assert_eq!(listed.len(), 1);
        assert_eq!(
            listed[0].activity,
            Some(SessionLifecycleState::Idle),
            "agents__list must report the derived activity, not leave it unset",
        );

        domain.lock().turn_pending = true;
        assert_eq!(
            facade.list_workers(&caller)[0].activity,
            Some(SessionLifecycleState::Running),
            "and it must be re-derived per call, not cached at spawn",
        );
    }

    /// The real walk pairs `list_projects` with the `live_workers` map, so
    /// this pins that both projects' entries reach the listing. A split
    /// project-key derivation would otherwise show an empty listing while
    /// every mock-backed handler test stays green.
    #[test]
    fn list_all_workers_walks_every_project() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project("forge", "/tmp/forge");
        ws.seed_test_project("other", "/tmp/other");
        let projects = ws.list_projects();
        let key_for = |name: &str| {
            projects
                .iter()
                .find(|view| view.name == name)
                .expect("seeded project present")
                .key
                .clone()
        };
        let entry = |project_name: &str, label: &str, status: WorkerLiveness| {
            crate::mcp::workers::types::WorkerEntry {
                label: label.into(),
                charter: "test charter".into(),
                slot: SessionSlot::worker("TestOrg", project_name, label),
                session_id: None,
                status,
                spawned_at: std::time::SystemTime::UNIX_EPOCH,
                spawned_by: SessionSlot::lead("TestOrg", project_name),
                needs_tag: false,
                is_git_repo_at_spawn: false,
                cwd: None,
                diagnostic: None,
                kick: None,
            }
        };
        ws.insert_live_worker(
            &key_for("forge"),
            entry("forge", "implementer", WorkerLiveness::Running),
        );
        ws.insert_live_worker(
            &key_for("other"),
            entry("other", "reviewer", WorkerLiveness::Running),
        );
        // A failed worker keeps its row; the real listing must keep it too.
        ws.insert_live_worker(&key_for("forge"), entry("forge", "crashed", WorkerLiveness::Failed));

        let implementer = SessionSlot::worker("TestOrg", "forge", "implementer");
        let domain = ws.register_domain_session(implementer, None);

        let facade = ProdWorkerFacade::from_arc(&ws);
        let listed = facade.list_all_workers();
        let pairs: Vec<(String, String)> = listed
            .iter()
            .map(|worker| (worker.slot.project().to_owned(), worker.label.clone()))
            .collect();
        assert_eq!(
            pairs,
            vec![
                ("forge".to_owned(), "implementer".to_owned()),
                ("forge".to_owned(), "crashed".to_owned()),
                ("other".to_owned(), "reviewer".to_owned()),
            ],
            "every project's workers, in config order",
        );
        assert_eq!(
            listed[0].activity,
            Some(SessionLifecycleState::Idle),
            "the listing carries the derived activity, not the bare status",
        );
        domain.lock().turn_pending = true;
        assert_eq!(
            facade.list_all_workers()[0].activity,
            Some(SessionLifecycleState::Running),
            "and it is re-derived per call, not cached at spawn",
        );
    }
}

#[cfg(test)]
mod deliver_tests {
    use super::*;
    use crate::mcp::peers::types::{MessageId, WrappedKind};
    use crate::mcp::workers::types::WorkerEntry;
    use forge_primitives::WorkerLiveness;
    use std::time::SystemTime;

    fn entry(label: &str, status: WorkerLiveness, session_id: Option<&str>) -> WorkerEntry {
        WorkerEntry {
            label: label.to_owned(),
            charter: "test charter".into(),
            slot: SessionSlot::worker("TestOrg", "forge", label),
            session_id: session_id.map(forge_primitives::SessionId::new),
            status,
            spawned_at: SystemTime::UNIX_EPOCH,
            spawned_by: SessionSlot::lead("TestOrg", "forge"),
            needs_tag: false,
            is_git_repo_at_spawn: false,
            cwd: None,
            diagnostic: None,
            kick: None,
        }
    }

    fn message() -> WrappedPrompt {
        WrappedPrompt {
            id: MessageId::mint(),
            kind: WrappedKind::Message,
            sender_name: "forge".to_owned(),
            sender_org: "Busytools".to_owned(),
            body: "status?".to_owned(),
        }
    }

    /// A live worker whose seat has no session behind it: the prompt reaches
    /// nothing, so the ack must say so. This is the overclaim the issue names -
    /// `Delivered` for a delivery that never left forge. The entry is
    /// Running with a session id, so the delivery is not parked or held: it
    /// goes to the session dispatch, which finds no session task.
    #[tokio::test]
    async fn a_prompt_the_session_cannot_take_is_not_acked_as_delivered() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project("forge", "/tmp/forge");
        let project = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .expect("seeded project present")
            .key;
        let slot = SessionSlot::worker("TestOrg", "forge", "implementer");
        ws.insert_live_worker(
            &project,
            entry("implementer", WorkerLiveness::Running, Some("live")),
        );
        // A connected domain and no session task: the seat looks reachable
        // and nothing can receive the prompt.
        ws.mark_session_connected_for_test(&slot, "live");

        let facade = ProdWorkerFacade::from_arc(&ws);
        let caller = SessionSlot::lead("TestOrg", "forge");
        let res = facade.deliver_worker_prompt(&caller, "implementer", message()).await;
        assert!(
            matches!(res, Err(WorkerDeliverError::DispatchFailed { .. })),
            "a prompt no session took must not be acked as delivered; got {res:?}",
        );
    }

    /// A crashed worker keeps its row (Failed) so its label can be respawned
    /// and the crash reason stays visible. Addressing that row is addressing
    /// a seat with no session: the prompt can never reach a live worker, so
    /// the ack must refuse it rather than say "sent" - and the refusal names
    /// the dead row, not a label that was never there.
    #[tokio::test]
    async fn a_failed_workers_label_is_refused_as_not_live() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project("forge", "/tmp/forge");
        let project = ws
            .list_projects()
            .into_iter()
            .find(|v| v.name == "forge")
            .expect("seeded project present")
            .key;
        ws.insert_live_worker(&project, entry("implementer", WorkerLiveness::Failed, Some("dead")));

        let facade = ProdWorkerFacade::from_arc(&ws);
        let caller = SessionSlot::lead("TestOrg", "forge");
        let res = facade.deliver_worker_prompt(&caller, "implementer", message()).await;
        assert!(
            matches!(res, Err(WorkerDeliverError::NotLive { .. })),
            "a Failed worker is not a live worker; the delivery must refuse it rather \
             than ack it as sent, and say the row is dead; got {res:?}",
        );

        // The control: a label with no entry at all is the other refusal, so
        // the two absences cannot collapse into one message.
        let ghost = facade.deliver_worker_prompt(&caller, "ghost", message()).await;
        assert!(
            matches!(ghost, Err(WorkerDeliverError::UnknownLabel { .. })),
            "a label with no row at all is UnknownLabel, not NotLive; got {ghost:?}",
        );
    }
}

#[cfg(test)]
mod capacity_tests {
    use super::*;
    use forge_primitives::WorkerLiveness;

    /// The slot the `forge` project's spawn names for `label`.
    fn worker_slot(label: &str) -> SessionSlot {
        SessionSlot::worker("TestOrg", "forge", label)
    }

    fn entry(label: &str) -> crate::mcp::workers::types::WorkerEntry {
        crate::mcp::workers::types::WorkerEntry {
            label: label.to_owned(),
            charter: "test charter".into(),
            slot: worker_slot(label),
            session_id: None,
            status: WorkerLiveness::Running,
            spawned_at: std::time::SystemTime::UNIX_EPOCH,
            spawned_by: SessionSlot::lead("TestOrg", "forge"),
            needs_tag: false,
            is_git_repo_at_spawn: false,
            cwd: None,
            diagnostic: None,
            kick: None,
        }
    }

    fn project_key_of(ws: &Workspace, name: &str) -> crate::ProjectKey {
        ws.list_projects().into_iter().find(|v| v.name == name).expect("seeded project present").key
    }

    #[test]
    fn capacity_reports_project_override_when_set() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project_with_max_workers("forge", "/tmp/forge", 1);
        let project = project_key_of(&ws, "forge");
        let caller = worker_slot("implementer");
        ws.insert_live_worker(&project, entry("implementer"));

        let facade = ProdWorkerFacade::from_arc(&ws);
        let capacity = facade.capacity(&caller).expect("live worker resolves to its project");
        assert_eq!(capacity.project, project.as_str());
        assert_eq!(capacity.cap, 1);
        assert_eq!(capacity.cap_source, WorkerCapSource::ProjectMaxWorkers);
        assert_eq!(capacity.live, 1);
    }

    #[test]
    fn capacity_falls_back_to_default_without_override() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project("forge", "/tmp/forge");
        let project = project_key_of(&ws, "forge");
        let caller = worker_slot("implementer");
        ws.insert_live_worker(&project, entry("implementer"));

        let facade = ProdWorkerFacade::from_arc(&ws);
        let capacity = facade.capacity(&caller).expect("live worker resolves to its project");
        assert_eq!(capacity.cap, crate::config::DEFAULT_MAX_WORKERS_PER_PROJECT);
        assert_eq!(capacity.cap_source, WorkerCapSource::Default);
        assert_eq!(capacity.live, 1);
    }

    #[test]
    fn capacity_live_count_matches_project_workers() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project("forge", "/tmp/forge");
        let project = project_key_of(&ws, "forge");
        let caller = worker_slot("a");
        ws.insert_live_worker(&project, entry("a"));
        ws.insert_live_worker(&project, entry("b"));

        let facade = ProdWorkerFacade::from_arc(&ws);
        let capacity = facade.capacity(&caller).expect("live worker resolves to its project");
        assert_eq!(capacity.live, 2);
    }

    #[test]
    fn capacity_none_for_unknown_caller() {
        let (ws, _rx) = Workspace::testing_stub();
        let facade = ProdWorkerFacade::from_arc(&ws);
        assert!(facade.capacity(&SessionSlot::from_str_for_test("ghost")).is_none());
    }

    #[test]
    fn capacity_excludes_failed_workers_from_live_count() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project_with_max_workers("forge", "/tmp/forge", 2);
        let project = project_key_of(&ws, "forge");
        let caller = worker_slot("running");
        ws.insert_live_worker(&project, entry("running"));
        let mut failed = entry("failed");
        failed.status = WorkerLiveness::Failed;
        ws.insert_live_worker(&project, failed);

        let facade = ProdWorkerFacade::from_arc(&ws);
        let capacity = facade.capacity(&caller).expect("live worker resolves to its project");
        assert_eq!(
            capacity.live, 1,
            "a Failed entry stays in the map but must not count against the cap"
        );
    }

    #[test]
    fn capacity_zero_live_workers_after_failure() {
        let (ws, _rx) = Workspace::testing_stub();
        ws.seed_test_project("forge", "/tmp/forge");
        let project = project_key_of(&ws, "forge");
        let caller = worker_slot("only");
        let mut failed = entry("only");
        failed.status = WorkerLiveness::Failed;
        ws.insert_live_worker(&project, failed);

        let facade = ProdWorkerFacade::from_arc(&ws);
        let capacity = facade.capacity(&caller).expect("a Failed entry still resolves the project");
        assert_eq!(capacity.live, 0);
        assert_eq!(capacity.cap, crate::config::DEFAULT_MAX_WORKERS_PER_PROJECT);
    }
}

#[cfg(test)]
mod worktree_creation_failed_tests {
    use super::*;

    #[test]
    fn worktree_creation_failed_carries_reason() {
        let err = WorkerSpawnError::WorktreeCreationFailed {
            reason: "fatal: 'worktree-reviewer' is already used by worktree at /a".into(),
        };
        let WorkerSpawnError::WorktreeCreationFailed { reason } = err else {
            panic!("expected WorktreeCreationFailed");
        };
        assert!(reason.contains("already used by worktree"));
    }

    // ---------------------------------------------------------------
    // classify_worker_spawn_failure: heuristic mapping from claude's
    // raw error text + the WorkerEntry's is_git_repo_at_spawn flag
    // into either WorktreeCreationFailed (when the message implicates
    // the --worktree machinery AND the worker was actually spawned in
    // a git repo) or DispatchFailed (everything else). The classifier
    // is pure so it's unit-testable without spinning up a workspace.
    // ---------------------------------------------------------------

    #[test]
    fn classify_routes_worktree_word_to_worktree_creation_failed_when_git() {
        // claude's worktree failures usually mention the word
        // "worktree" verbatim (e.g. "Error creating worktree: ...").
        let err = classify_worker_spawn_failure(
            "Error creating worktree: something went wrong",
            true,
            SpawnFailureKind::Unclassified,
        );
        let WorkerSpawnError::WorktreeCreationFailed { reason } = err else {
            panic!("expected WorktreeCreationFailed; got {err:?}");
        };
        assert!(reason.contains("worktree"));
    }

    #[test]
    fn classify_routes_base_branch_resolve_to_worktree_creation_failed_when_git() {
        // The empty-repo (unborn HEAD) case from spike 0.3: claude
        // refuses with "Failed to resolve base branch \"HEAD\": git
        // rev-parse failed". The word "worktree" does not appear, so
        // the classifier matches on the resolve-base-branch substring
        // instead.
        let err = classify_worker_spawn_failure(
            "Failed to resolve base branch \"HEAD\": git rev-parse failed",
            true,
            SpawnFailureKind::Unclassified,
        );
        assert!(
            matches!(err, WorkerSpawnError::WorktreeCreationFailed { .. }),
            "expected WorktreeCreationFailed; got {err:?}",
        );
    }

    #[test]
    fn classify_routes_already_used_to_worktree_creation_failed_when_git() {
        // "fatal: '<label>' is already used by worktree at <path>" -
        // the conflicting-branch case (label collision with an
        // existing worktree).
        let err = classify_worker_spawn_failure(
            "fatal: 'reviewer' is already used by worktree at /a/b/c",
            true,
            SpawnFailureKind::Unclassified,
        );
        assert!(
            matches!(err, WorkerSpawnError::WorktreeCreationFailed { .. }),
            "expected WorktreeCreationFailed; got {err:?}",
        );
    }

    #[test]
    fn classify_falls_through_to_dispatch_failed_for_unrelated_messages() {
        // A generic spawn-side error that has nothing to do with
        // worktrees should stay DispatchFailed so the LLM gets the
        // raw text verbatim, not a misclassified worktree variant.
        let err = classify_worker_spawn_failure(
            "agent spawn failed: subprocess exited with code 2",
            true,
            SpawnFailureKind::Unclassified,
        );
        let WorkerSpawnError::DispatchFailed { message } = err else {
            panic!("expected DispatchFailed; got {err:?}");
        };
        assert!(message.contains("subprocess exited"));
    }

    #[test]
    fn classify_with_worktree_word_but_non_git_repo_falls_through() {
        // Defensive: a worktree-mentioning message arriving for a
        // worker that was NEVER spawned with --worktree (non-git
        // project, is_git_repo_at_spawn=false) is suspicious - the
        // claude binary couldn't have emitted a worktree-creation
        // error since it wasn't asked to create one. Fall through to
        // DispatchFailed rather than mislabel it.
        let err = classify_worker_spawn_failure(
            "Error creating worktree: something went wrong",
            false,
            SpawnFailureKind::Unclassified,
        );
        assert!(
            matches!(err, WorkerSpawnError::DispatchFailed { .. }),
            "expected DispatchFailed; got {err:?}",
        );
    }

    /// `Error::CwdNotFound` renders the path it could not enter, and for
    /// a worker that path runs through `.claude/worktrees/<label>` - the
    /// one word the worktree heuristic keys on. Reading that as a
    /// worktree-creation failure deletes the worker's row and tells the
    /// lead its worktree could not be created, which is the opposite of
    /// what happened and of what this change promises.
    ///
    /// The second half is the control: the identical message unclassified
    /// still routes to the worktree variant, so the first assertion is
    /// reading the named kind and not a message that stopped matching.
    #[test]
    fn classify_reads_a_named_variant_over_a_worktrees_path_in_the_message() {
        let message = "forge-sdk session resume failed: claude subprocess working directory \
                       `/repo/.claude/worktrees/reviewer` does not exist";
        let named =
            classify_worker_spawn_failure(message, true, SpawnFailureKind::WorkingDirMissing);
        assert!(
            matches!(named, WorkerSpawnError::DispatchFailed { .. }),
            "a named missing-working-directory failure is not a worktree-creation \
             failure, however its path renders; got {named:?}",
        );

        let unclassified =
            classify_worker_spawn_failure(message, true, SpawnFailureKind::Unclassified);
        assert!(
            matches!(unclassified, WorkerSpawnError::WorktreeCreationFailed { .. }),
            "control: the same message with no named variant must still route on the \
             heuristic; got {unclassified:?}",
        );
    }

    #[test]
    fn classify_is_case_insensitive_on_match_terms() {
        // Be lenient about casing - claude has shipped messages with
        // varying capitalization across CLI versions. Match on the
        // lowercased message body.
        let err = classify_worker_spawn_failure(
            "WORKTREE creation failed",
            true,
            SpawnFailureKind::Unclassified,
        );
        assert!(
            matches!(err, WorkerSpawnError::WorktreeCreationFailed { .. }),
            "expected case-insensitive match to WorktreeCreationFailed; got {err:?}",
        );
    }

    #[test]
    fn classify_preserves_original_casing_in_reason() {
        // The classifier may lowercase for matching but the reason
        // string handed back to the LLM should preserve the original
        // casing - the LLM sees claude's verbatim error text.
        let original = "Error creating Worktree: nope";
        let err = classify_worker_spawn_failure(original, true, SpawnFailureKind::Unclassified);
        let WorkerSpawnError::WorktreeCreationFailed { reason } = err else {
            panic!("expected WorktreeCreationFailed; got {err:?}");
        };
        assert_eq!(reason, original);
    }

    /// #245 Layer C blocker 2: pin the bridge-prefix contract.
    /// `forge_sdk_bridge` wraps every async ConnectionFailed message
    /// with a literal "forge-sdk session ..." prefix; none of those
    /// prefixes contain the words "worktree" / "base branch" /
    /// "already used by worktree" that the classifier matches on, so
    /// a generic dispatch failure can't accidentally classify as
    /// `WorktreeCreationFailed` just because the bridge happened to
    /// wrap it.
    ///
    /// If a future bridge change introduces "worktree" into the
    /// wrapper prose, this test fails - that's the signal to either
    /// rename the wrapper or plumb a typed reason variant through
    /// `AgentEvent::ConnectionFailed`.
    #[test]
    fn bridge_prefix_does_not_collide_with_worktree_predicate() {
        // The three wrapper prefixes verbatim from forge_sdk_bridge.
        let prefixes = [
            "forge-sdk session spawn failed: ",
            "forge-sdk session resume failed: ",
            "forge-sdk session spawn failed after resume fallback (resume err: ",
        ];
        for prefix in prefixes {
            let lower = prefix.to_lowercase();
            assert!(
                !lower.contains("worktree"),
                "bridge prefix {prefix:?} must not contain 'worktree' or the \
                 classifier will misclassify generic errors as worktree failures",
            );
            assert!(
                !lower.contains("failed to resolve base branch"),
                "bridge prefix {prefix:?} must not contain the 'failed to resolve base branch' \
                 discriminator",
            );
            assert!(
                !lower.contains("already used by worktree"),
                "bridge prefix {prefix:?} must not contain the 'already used by worktree' \
                 discriminator",
            );
        }

        // Concretely: a non-worktree generic failure wrapped by the
        // bridge stays classified as DispatchFailed.
        let wrapped = "forge-sdk session resume failed: subprocess exited with code 2";
        let err = classify_worker_spawn_failure(
            wrapped,
            /* is_git_repo_at_spawn */ true,
            SpawnFailureKind::Unclassified,
        );
        assert!(
            matches!(err, WorkerSpawnError::DispatchFailed { .. }),
            "wrapped generic failure must stay DispatchFailed; got {err:?}",
        );
    }

    #[test]
    fn the_cap_refusal_stays_a_dispatch_failure() {
        let message = crate::spawn::worker_limit_reached_message("my-worktree-proj", 2, 2);
        assert_eq!(
            classify_worker_spawn_failure(&message, true, SpawnFailureKind::Unclassified),
            WorkerSpawnError::DispatchFailed { message },
            "the cap refusal stays DispatchFailed even when the project key carries 'worktree'",
        );
    }

    /// The known-project refusal is recognised by its own phrase too, and
    /// that guard is what keeps it a dispatch failure: the sentence
    /// interpolates a project NAME, and a project may be called anything -
    /// `worktree-lab` here - which the classifier would otherwise route to
    /// a worktree-creation failure the spawn never suffered.
    #[test]
    fn the_known_project_refusal_stays_a_dispatch_failure_whatever_the_project_is_called() {
        let message = crate::spawn::known_project_refusal("worktree-lab", "forge");
        assert_eq!(
            classify_worker_spawn_failure(&message, true, SpawnFailureKind::Unclassified),
            WorkerSpawnError::DispatchFailed { message: message.clone() },
            "the known-project refusal stays DispatchFailed however its project reads",
        );

        // The other arm too: a spawn from the project named that way.
        let own = crate::spawn::known_project_refusal("worktree-lab", "worktree-lab");
        assert_eq!(
            classify_worker_spawn_failure(&own, true, SpawnFailureKind::Unclassified),
            WorkerSpawnError::DispatchFailed { message: own.clone() },
            "and the same for a project whose own name carries the word",
        );
    }

    /// The despawn-cleanup refusal is recognised by its own phrase, so a
    /// label the sentence might come to carry ("worktree-fix" here, the
    /// shape the LabelLive refusal already has) cannot route it to a
    /// worktree-creation failure it did not suffer.
    #[test]
    fn the_cleanup_refusal_stays_a_dispatch_failure_whatever_the_label_spells() {
        let message =
            "the previous despawn of 'worktree-fix' is still cleaning up; retry once it finishes";
        assert_eq!(
            classify_worker_spawn_failure(message, true, SpawnFailureKind::Unclassified),
            WorkerSpawnError::DispatchFailed { message: message.to_owned() },
            "the cleanup refusal stays DispatchFailed however its label reads",
        );
    }
}

#[cfg(test)]
mod discard_tests {
    use super::*;
    use std::sync::Arc;

    fn run_git(dir: &std::path::Path, args: &[&str]) {
        let status =
            std::process::Command::new("git").arg("-C").arg(dir).args(args).status().expect("git");
        assert!(status.success(), "git {args:?} in {}", dir.display());
    }

    /// A main checkout with proj-a and proj-b linked worktrees declared,
    /// plus an account.
    fn linked_projects() -> (Arc<Workspace>, std::path::PathBuf, Vec<tempfile::TempDir>) {
        let repo = tempfile::tempdir().expect("repo");
        let main = repo.path().join("main");
        std::fs::create_dir_all(&main).expect("main dir");
        run_git(&main, &["init", "-q"]);
        run_git(&main, &["config", "user.email", "t@example.com"]);
        run_git(&main, &["config", "user.name", "Test"]);
        run_git(&main, &["config", "commit.gpgsign", "false"]);
        std::fs::write(main.join("README.md"), "seed").expect("write seed");
        run_git(&main, &["add", "."]);
        run_git(&main, &["commit", "-q", "-m", "init"]);
        let main = std::fs::canonicalize(&main).expect("canonical main");
        let holder = tempfile::tempdir().expect("holder");
        let linked_a = holder.path().join("proj-a");
        let linked_b = holder.path().join("proj-b");
        run_git(
            &main,
            &["worktree", "add", "-q", "-b", "proj-a", linked_a.to_str().expect("utf8")],
        );
        run_git(
            &main,
            &["worktree", "add", "-q", "-b", "proj-b", linked_b.to_str().expect("utf8")],
        );
        let cfg = tempfile::tempdir().expect("cfg");
        let forge_dir = cfg.path().join("forge");
        std::fs::create_dir_all(&forge_dir).expect("forge dir");
        std::fs::write(
            forge_dir.join("forge.toml"),
            format!(
                "[[orgs]]\nname = \"Default\"\naccounts = [\"a\"]\n\
                 [[orgs.projects]]\nname = \"proj-a\"\npath = \"{}\"\n\
                 [[orgs.projects]]\nname = \"proj-b\"\npath = \"{}\"\n\
                 [[accounts]]\ndisplay_name = \"a\"\ntoken = \"t\"\n\
                 models = [\"claude-sonnet-5\"]\nprovider = \"anthropic\"\n",
                linked_a.display(),
                linked_b.display()
            ),
        )
        .expect("write forge.toml");
        let ws = Arc::new(Workspace::new_for_test(cfg.path().to_owned()).expect("boot"));
        (ws, main, vec![repo, holder, cfg])
    }

    fn key_of(ws: &Arc<Workspace>, name: &str) -> crate::ProjectKey {
        ws.list_projects().into_iter().find(|v| v.name == name).expect("project").key
    }

    fn winner_entry(project: &str, label: &str) -> crate::mcp::workers::types::WorkerEntry {
        crate::mcp::workers::types::WorkerEntry {
            label: label.to_owned(),
            charter: "test".to_owned(),
            slot: SessionSlot::worker("Default", project, label),
            session_id: None,
            status: forge_primitives::WorkerLiveness::Running,
            spawned_at: std::time::SystemTime::UNIX_EPOCH,
            spawned_by: SessionSlot::lead("Default", project),
            needs_tag: false,
            is_git_repo_at_spawn: true,
            cwd: None,
            diagnostic: None,
            kick: None,
        }
    }

    fn shared_worktree(main: &std::path::Path) -> std::path::PathBuf {
        let shared = main.join(".claude").join("worktrees").join("shared");
        std::fs::create_dir_all(shared.parent().expect("worktrees root")).expect("mkdir");
        run_git(
            main,
            &["worktree", "add", "-q", "-b", "worktree-shared", shared.to_str().expect("utf8")],
        );
        shared
    }

    fn branch_stands(main: &std::path::Path) -> bool {
        let branches = std::process::Command::new("git")
            .args(["branch", "--list", "worktree-shared"])
            .current_dir(main)
            .output()
            .expect("git branch");
        !String::from_utf8_lossy(&branches.stdout).trim().is_empty()
    }

    /// A rollback never discards a tree another project's live worker
    /// holds: a scan of the refused caller's own project misses the
    /// cross-project winner and deletes its tree and branch.
    #[test]
    fn a_rollback_keeps_a_tree_another_project_holds() {
        let (ws, main, _keep) = linked_projects();
        let shared = shared_worktree(&main);
        ws.insert_live_worker(&key_of(&ws, "proj-a"), winner_entry("proj-a", "shared"));

        discard_refused_worktree(
            &ws,
            &key_of(&ws, "proj-b"),
            &main,
            "shared",
            Some((
                shared.clone(),
                forge_agent::env::worktree::WorktreeEnsure::CreatedWithNewBranch,
            )),
        );

        assert!(shared.exists(), "another project's live worker's tree must survive the rollback");
        assert!(branch_stands(&main), "and its branch is not reaped");
    }

    /// Two resumes of one label: the loser mints the tree, the winner takes
    /// the registry under the same project and label, and the loser's
    /// rollback must leave the winner's tree and branch alone. The sibling
    /// test's holder is another project; this is the same-project winner
    /// the skip exists for (a non-git holder's tree is the project root and
    /// never matches).
    #[test]
    fn a_losing_resume_leaves_the_winners_worktree_alone() {
        let (ws, main, _keep) = linked_projects();
        let shared = shared_worktree(&main);
        ws.insert_live_worker(&key_of(&ws, "proj-a"), winner_entry("proj-a", "shared"));

        discard_refused_worktree(
            &ws,
            &key_of(&ws, "proj-a"),
            &main,
            "shared",
            Some((
                shared.clone(),
                forge_agent::env::worktree::WorktreeEnsure::CreatedWithNewBranch,
            )),
        );

        assert!(shared.exists(), "the winner's tree survives the losing resume's rollback");
        assert!(branch_stands(&main), "and the branch the loser minted is not reaped");
    }
}