wsb-daemon 0.248.6265

use wsb::prelude::*;
use axum::{
    extract::{Query, State},
    http::StatusCode,
    response::sse::{Event, KeepAlive, Sse},
    Json,
};
use futures::stream::Stream;
use std::convert::Infallible;

use super::sessions::ErrorResponse;
use wsb::fleet::git::tree::SubmoduleTree;
use wsb::fleet::git::GitOps;
use wsb::git_backend::{self, FileStatusKind};
use crate::web::ops::OpKind;
use crate::web::state::WebState;

// ---------------------------------------------------------------------------
// Shared types
// ---------------------------------------------------------------------------

#[derive(Serialize, Clone, Debug)]
pub struct GitFileStatus {
    pub path: String,
    pub status: FileStatusKind,
}

#[derive(Serialize, Clone, Debug)]
pub struct RepoDetailedStatus {
    pub name: String,
    pub path: String,
    pub branch: Option<String>,
    pub is_clean: bool,
    pub is_detached: bool,
    pub ahead: usize,
    pub behind: usize,
    pub files: Vec<GitFileStatus>,
    pub exists: bool,
    pub is_superproject: bool,
    /// Nesting level in the submodule tree: 0 for the superproject, 1 for its
    /// direct submodules, deeper for submodules declared inside those.
    pub depth: usize,
    /// Workspace-relative path of the repository that declares this one — "."
    /// for a direct submodule of the superproject. None for the superproject.
    pub parent_path: Option<String>,
}

/// Legacy compat-free: the old RepoStatus is replaced by RepoDetailedStatus.
/// The fleet_status endpoint now returns full file-level status.

// ---------------------------------------------------------------------------
// Request / Response types for mutation endpoints
// ---------------------------------------------------------------------------

#[derive(Deserialize)]
pub struct RepoPathRequest {
    pub repo_path: Option<String>,
}

#[derive(Deserialize)]
pub struct CommitRequest {
    pub repo_path: Option<String>,
    pub message: String,
}

#[derive(Deserialize)]
pub struct CheckoutRequest {
    pub repo_path: Option<String>,
    pub branch: String,
    pub create: Option<bool>,
}

#[derive(Deserialize)]
pub struct CreateBranchRequest {
    pub repo_path: Option<String>,
    pub branch: String,
    pub from_ref: Option<String>,
}

#[derive(Deserialize)]
pub struct FleetCommitRequest {
    pub message: String,
}

#[derive(Deserialize)]
pub struct FleetCheckoutRequest {
    pub branch: String,
    pub create: Option<bool>,
}

#[derive(Deserialize)]
pub struct FleetMergeBackRequest {
    pub message: Option<String>,
}

#[derive(Deserialize)]
pub struct BranchesQuery {
    pub repo_path: Option<String>,
}

#[derive(Deserialize)]
pub struct DiffQuery {
    /// Workspace-relative path to the file or directory (e.g. "src/main.rs"
    /// or "floom-engine/src/main.rs" for a submodule file).
    pub path: String,
    /// Optional repo root (workspace-relative). Defaults to superproject.
    pub repo_path: Option<String>,
}

#[derive(Deserialize)]
pub struct FocusHintRequest {
    pub repo_path: Option<String>,
}

#[derive(Serialize)]
pub struct GitOpResult {
    pub success: bool,
    pub message: String,
}

#[derive(Serialize)]
pub struct BranchListResponse {
    pub branches: Vec<String>,
    pub current: Option<String>,
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// The workspace these routes act on.
///
/// Taken from the state the route was reached through. It used to be the
/// process's own directory, which was the same thing only while a daemon
/// served exactly one workspace — afterwards it would have been whichever
/// directory the daemon was started in, for every workspace on the machine.
pub(crate) fn workspace_root_of(state: &WebState) -> PathBuf {
    state.workspace_root.clone()
}


/// Resolve an optional relative repo_path to an absolute path under the workspace root.
/// If repo_path is None or ".", returns the workspace root itself.
/// Sync version of `resolve_repo_path` for use inside `spawn_blocking` closures.
fn resolve_repo_path_sync(
    root: &Path,
    repo_path: Option<&str>,
) -> Result<PathBuf, (StatusCode, Json<ErrorResponse>)> {
    resolve_repo_path(root, repo_path)
}

fn resolve_repo_path(
    root: &Path,
    repo_path: Option<&str>,
) -> Result<PathBuf, (StatusCode, Json<ErrorResponse>)> {
    let rel = repo_path.unwrap_or(".");
    if rel == "." || rel.is_empty() {
        return Ok(root.to_path_buf());
    }
    let resolved = root.join(rel);
    // Security: ensure resolved path stays under workspace root
    let canonical_root = wsb::platform::canonicalize(&root).map_err(|e| {
        (
            StatusCode::INTERNAL_SERVER_ERROR,
            Json(ErrorResponse {
                error: format!("cannot canonicalize workspace root: {}", e),
            }),
        )
    })?;
    let canonical_resolved = wsb::platform::canonicalize(&resolved).map_err(|e| {
        (
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse {
                error: format!("cannot resolve repo path '{}': {}", rel, e),
            }),
        )
    })?;
    if !canonical_resolved.starts_with(&canonical_root) {
        return Err((
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse {
                error: format!("repo path '{}' escapes workspace root", rel),
            }),
        ));
    }
    Ok(canonical_resolved)
}

fn internal_error(error: wsb::fleet::error::SuperGitError) -> (StatusCode, Json<ErrorResponse>) {
    tracing::error!(error = %error, error_debug = ?error, "fleet handler error");
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(ErrorResponse {
            error: format!("{:#}", error),
        }),
    )
}

fn anyhow_error(error: anyhow::Error) -> (StatusCode, Json<ErrorResponse>) {
    tracing::error!(error = %error, "fleet handler error");
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(ErrorResponse {
            error: format!("{:#}", error),
        }),
    )
}

/// What a pull did, as a sentence — or the reason it did nothing, as an error.
///
/// The split is the point: every outcome except a divergence leaves the branch
/// holding everything the remote has, and a divergence leaves it behind.
fn pull_message(outcome: git_backend::PullOutcome) -> anyhow::Result<String> {
    Ok(match outcome {
        git_backend::PullOutcome::FastForwarded { commits, .. } => {
            format!("pulled {commits} commit(s)")
        }
        git_backend::PullOutcome::Rebased { commits } => {
            format!("rebased onto {commits} incoming commit(s)")
        }
        git_backend::PullOutcome::UpToDate => "already up to date".to_string(),
        git_backend::PullOutcome::Ahead { outgoing } => {
            format!("already up to date ({outgoing} local commit(s) to push)")
        }
        git_backend::PullOutcome::NoRemoteBranch => {
            "this branch has no counterpart on origin".to_string()
        }
        git_backend::PullOutcome::Diverged {
            incoming,
            outgoing,
        } => {
            return Err(anyhow::anyhow!(
                "the branch has diverged: {incoming} incoming commit(s) and {outgoing} \
                 local commit(s), so a fast-forward is impossible. Rebase the local \
                 commits onto the incoming ones, or merge them."
            ))
        }
        git_backend::PullOutcome::LocalChangesBlock {
            incoming,
            paths,
            untracked,
        } => {
            let mut message = format!(
                "{incoming} incoming commit(s) rewrite files this tree has changed: {}",
                paths.join(", ")
            );
            if !untracked.is_empty() {
                message.push_str(&format!(
                    ". These are untracked and git holds no copy of them, so they must be \
                     moved or removed by hand: {}",
                    untracked.join(", ")
                ));
            } else {
                message.push_str(
                    ". Run `wsb pull --autostash` to set them aside and put them back on \
                     top, or `wsb pull --overwrite-local` to throw them away.",
                );
            }
            return Err(anyhow::anyhow!(message));
        }
        git_backend::PullOutcome::AutostashConflict {
            incoming,
            paths,
            stash,
        } => {
            return Err(anyhow::anyhow!(
                "took {incoming} incoming commit(s), but your changes to {} could not be \
                 replayed on top of them. They are kept in {stash}.",
                paths.join(", ")
            ))
        }
    })
}

fn ok_result(msg: impl Into<String>) -> Json<GitOpResult> {
    Json(GitOpResult {
        success: true,
        message: msg.into(),
    })
}

/// Extract a plain error string from the handler error tuple, for use in op failure messages.
fn handler_err_msg(e: &(StatusCode, Json<ErrorResponse>)) -> String {
    e.1.error.clone()
}

// ---------------------------------------------------------------------------
// Status collection
// ---------------------------------------------------------------------------

fn collect_detailed_repo_status(
    _repo_root: &Path,
    repo_path: &Path,
    name: &str,
    relative_path: &str,
    is_superproject: bool,
    depth: usize,
    parent_path: Option<String>,
) -> Result<RepoDetailedStatus, wsb::fleet::error::SuperGitError> {
    if !repo_path.exists() {
        return Ok(RepoDetailedStatus {
            name: name.to_string(),
            path: relative_path.to_string(),
            branch: None,
            is_clean: true,
            is_detached: false,
            ahead: 0,
            behind: 0,
            files: Vec::new(),
            exists: false,
            is_superproject,
            depth,
            parent_path,
        });
    }

    let repo = GitOps::open_repo(repo_path)?;
    let is_detached = GitOps::is_detached_head(&repo)?;
    let branch = if is_detached {
        None
    } else {
        Some(GitOps::current_branch(&repo)?)
    };

    let entries = git_backend::status_entries(&repo)?;
    let is_clean = !git_backend::has_uncommitted_changes(&entries);
    let files: Vec<GitFileStatus> = entries
        .into_iter()
        .map(|e| GitFileStatus {
            path: e.path,
            status: e.status,
        })
        .collect();

    let (ahead, behind) = if is_detached {
        (0, 0)
    } else {
        GitOps::get_ahead_behind(&repo).unwrap_or((0, 0))
    };

    Ok(RepoDetailedStatus {
        name: name.to_string(),
        path: relative_path.to_string(),
        branch,
        is_clean,
        is_detached,
        ahead,
        behind,
        files,
        exists: true,
        is_superproject,
        depth,
        parent_path,
    })
}

/// Existing repositories, deepest first. Use for operations whose result in a
/// repository depends on its submodules having acted first — stage, commit,
/// push, merge-back — because a parent records its submodules' commits as
/// gitlinks and cannot record a commit that does not exist yet.
fn paths_bottom_up(repos: &[RepoDetailedStatus]) -> Vec<String> {
    let mut ordered: Vec<&RepoDetailedStatus> = repos.iter().filter(|r| r.exists).collect();
    // Stable: siblings keep discovery order while every child precedes its parent.
    ordered.sort_by_key(|repo| std::cmp::Reverse(repo.depth));
    ordered.iter().map(|repo| repo.path.clone()).collect()
}

/// Existing repositories, shallowest first. Use for operations that push state
/// outward from the superproject — pull, checkout — where a parent's new
/// gitlinks name the commits its submodules must move to.
fn paths_top_down(repos: &[RepoDetailedStatus]) -> Vec<String> {
    let mut ordered: Vec<&RepoDetailedStatus> = repos.iter().filter(|r| r.exists).collect();
    ordered.sort_by_key(|repo| repo.depth);
    ordered.iter().map(|repo| repo.path.clone()).collect()
}

/// Sort key that lays repositories out in pre-order: the superproject first,
/// then each repository immediately before the ones nested inside it.
fn tree_order_key(repo: &RepoDetailedStatus) -> (u8, Vec<String>) {
    if repo.is_superproject {
        return (0, Vec::new());
    }
    (
        1,
        repo.path.split('/').map(|part| part.to_string()).collect(),
    )
}

/// Workspace-relative path of the repository that declares `node`.
/// A top-level submodule ("capdag") is declared by the superproject (".");
/// a nested one ("capdag/capdag-rs") by its enclosing submodule.
///
/// Taken from the node's own `parent_path` — the repository git says declared
/// it — and not from cutting the last segment off the display path. Those two
/// agree only where a submodule sits directly in its parent's root: a
/// submodule at `libs/core` is declared by the superproject, while the string
/// form answers `libs`, which is a directory and not a repository at all. The
/// fleet then hung the row under a parent that was never in the list.
fn declaring_repo_path(root: &Path, node: &wsb::fleet::git::tree::SubmoduleNode) -> String {
    let relative = node
        .parent_path
        .strip_prefix(root)
        .unwrap_or_else(|_| Path::new(""));
    if relative.as_os_str().is_empty() {
        return ".".to_string();
    }
    relative
        .components()
        .map(|c| c.as_os_str().to_string_lossy())
        .collect::<Vec<_>>()
        .join("/")
}

pub(crate) fn collect_repo_statuses(
    repo_root: &Path,
) -> wsb::fleet::error::Result<Vec<RepoDetailedStatus>> {
    let super_name = repo_root
        .file_name()
        .map(|name| name.to_string_lossy().into_owned())
        .unwrap_or_else(|| repo_root.display().to_string());

    let mut repos = vec![collect_detailed_repo_status(
        repo_root,
        repo_root,
        &super_name,
        ".",
        true,
        0,
        None,
    )?];

    // The whole nesting tree, not just the superproject's own .gitmodules: a
    // submodule declared inside a submodule is as much a repository of this
    // fleet as a top-level one, and the tree walk is what gives every repo its
    // depth and declaring parent.
    let tree = SubmoduleTree::discover_allow_empty(repo_root)?;

    // Pre-order, so a parent always precedes the submodules it declares.
    for node in tree.top_down() {
        repos.push(collect_detailed_repo_status(
            repo_root,
            &node.abs_path,
            &node.name,
            &node.display_path,
            false,
            node.depth + 1,
            Some(declaring_repo_path(repo_root, node)),
        )?);
    }

    Ok(repos)
}

// ---------------------------------------------------------------------------
// GET endpoints
// ---------------------------------------------------------------------------

/// GET /api/v1/fleet/status — full detailed status for all repos.
///
/// Served from the in-memory snapshot maintained by the git watcher — instant,
/// never blocks, always up to date (watcher pushes on every change).
pub async fn fleet_status(
    State(state): State<WebState>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
    let snapshot = state.git_status_snapshot.read().await;
    // The snapshot is a map, so its iteration order is arbitrary; clients render
    // a tree, which needs the superproject first and every repo next to the one
    // that declares it.
    let mut repos: Vec<&RepoDetailedStatus> = snapshot.values().collect();
    repos.sort_by(|left, right| tree_order_key(left).cmp(&tree_order_key(right)));
    Ok(Json(serde_json::json!({ "repos": repos })))
}

/// GET /api/v1/fleet/repos — list GitHub repos for issues
pub async fn fleet_repos(
    State(state): State<WebState>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
    let repos = crate::web::github::repository_names(&state).await?;
    Ok(Json(serde_json::json!({ "repos": repos })))
}

/// GET /api/v1/fleet/detailed-status — enriched status with per-file FileStatusKind
pub async fn fleet_detailed_status(
    State(state): State<WebState>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
    let repo_root = workspace_root_of(&state);
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&repo_root))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;
    Ok(Json(serde_json::json!({ "repos": repos })))
}

/// GET /api/v1/fleet/branches — list branches for a repo
pub async fn fleet_branches(
    State(state): State<WebState>,
    Query(query): Query<BranchesQuery>,
) -> Result<Json<BranchListResponse>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, query.repo_path.as_deref())?;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        let branches = git_backend::local_branches(&repo).map_err(anyhow_error)?;
        let is_detached = git_backend::is_detached_head(&repo).map_err(anyhow_error)?;
        let current = if is_detached {
            None
        } else {
            Some(git_backend::current_branch(&repo).map_err(anyhow_error)?)
        };
        Ok::<BranchListResponse, (StatusCode, Json<ErrorResponse>)>(BranchListResponse {
            branches,
            current,
        })
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))??;
    Ok(Json(result))
}

/// GET /api/v1/fleet/diff — unified diff for a path vs HEAD
///
/// `path` is the repo-relative path of the file or directory to diff.
/// `repo_path` is the workspace-relative repo root (defaults to superproject).
/// Returns plain text unified diff, or empty string if path has no changes.
pub async fn fleet_diff(
    State(state): State<WebState>,
    Query(query): Query<DiffQuery>,
) -> Result<axum::response::Response, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, query.repo_path.as_deref())?;
    let path = query.path.clone();
    let diff = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::diff_file(&repo, &path).map_err(anyhow_error)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))??;
    Ok(axum::response::Response::builder()
        .status(StatusCode::OK)
        .header("content-type", "text/plain; charset=utf-8")
        .body(axum::body::boxed(axum::body::Full::from(diff)))
        .unwrap())
}

// ---------------------------------------------------------------------------
// SSE — real-time git status stream with focus tracking
// ---------------------------------------------------------------------------

/// GET /api/v1/fleet/watch — SSE stream that pushes git status updates.
///
/// The background watcher thread continuously polls git state and broadcasts
/// changes.  Clients receive the full fleet status on each change.
pub async fn fleet_watch(
    State(state): State<WebState>,
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
    let mut rx = state.git_status_tx.subscribe();

    let stream = async_stream::stream! {
        // Send the current snapshot as individual per-repo events so new
        // subscribers see the full picture immediately without waiting for
        // the next watcher cycle.
        {
            let snapshot = state.git_status_snapshot.read().await;
            for repo in snapshot.values() {
                if let Ok(json) = serde_json::to_string(repo) {
                    yield Ok(Event::default().event("repo").data(json));
                }
            }
        }

        loop {
            match rx.recv().await {
                Ok(repo) => {
                    if let Ok(json) = serde_json::to_string(&repo) {
                        yield Ok(Event::default().event("repo").data(json));
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
                    // We fell behind; re-send the full snapshot so the client
                    // is consistent, then continue from the live stream.
                    tracing::warn!("git SSE client lagged by {} events, re-sending snapshot", n);
                    let snapshot = state.git_status_snapshot.read().await;
                    for repo in snapshot.values() {
                        if let Ok(json) = serde_json::to_string(repo) {
                            yield Ok(Event::default().event("repo").data(json));
                        }
                    }
                }
                Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
            }
        }
    };

    Sse::new(stream).keep_alive(KeepAlive::default())
}

/// POST /api/v1/fleet/focus — hint the watcher about user's current focus
pub async fn fleet_focus(
    State(state): State<WebState>,
    Json(body): Json<FocusHintRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let hint = body.repo_path.unwrap_or_else(|| ".".to_string());
    // Send focus hint — non-blocking, ignore if channel is full
    let _ = state.git_focus_tx.send(hint.clone());
    Ok(ok_result(format!("focus set to '{}'", hint)))
}

// ---------------------------------------------------------------------------
// Single-repo mutation endpoints
// ---------------------------------------------------------------------------

/// POST /api/v1/fleet/stage — stage all changes in a repo
pub async fn fleet_stage(
    State(state): State<WebState>,
    Json(body): Json<RepoPathRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let label = format!("Stage: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::stage_all(&repo).map_err(anyhow_error)?;
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?;
    match result {
        Ok(()) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            Ok(ok_result("staged all changes"))
        }
        Err(e) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
    }
}

/// POST /api/v1/fleet/commit — commit staged changes in a repo
pub async fn fleet_commit(
    State(state): State<WebState>,
    Json(body): Json<CommitRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let message = body.message.clone();
    let label = format!("Commit: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let sha = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::commit_all(&repo, &message).map_err(anyhow_error)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match sha {
        Ok(Ok(sha)) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            let msg = match sha {
                Some(sha) => format!("committed: {}", sha),
                None => "nothing to commit".to_string(),
            };
            Ok(ok_result(msg))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/push — push current branch
/// What a discard would throw away in one repository, or the discarding of it.
///
/// Two answers from one route because they are one question asked twice: the
/// preview has to be produced by exactly the code that then acts, or the list
/// somebody approved is not the list that goes. `wsb discard` does the same —
/// find every worktree with changes, say what is in them, then ask.
#[derive(Deserialize)]
pub struct DiscardRequest {
    pub repo_path: Option<String>,
    /// Remove untracked files too. They are the ones git has no copy of at
    /// all, so they are never included unless they are asked for.
    #[serde(default)]
    pub untracked: bool,
    /// Report what would go and change nothing.
    #[serde(default)]
    pub dry_run: bool,
}

#[derive(Serialize)]
pub struct DiscardAnswer {
    /// Tracked files with changes — what a hard reset throws away. The content
    /// was never committed, so git holds no copy to go back to.
    pub tracked: Vec<String>,
    /// Files git is not tracking, listed whether or not they were asked for so
    /// the answer says what a second click would also take.
    pub untracked: Vec<String>,
    /// Whether anything was actually discarded.
    pub discarded: bool,
    pub message: String,
}

pub async fn fleet_discard(
    State(state): State<WebState>,
    Json(body): Json<DiscardRequest>,
) -> Result<Json<DiscardAnswer>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let dry_run = body.dry_run;
    let untracked_too = body.untracked;

    let op = if dry_run {
        None
    } else {
        Some(
            state
                .ops
                .start_op(OpKind::GitMutation, format!("Discard: {rel}"))
                .await,
        )
    };

    let answer = tokio::task::spawn_blocking(move || -> anyhow::Result<DiscardAnswer> {
        let repo = git_backend::open_repo(&repo_path)?;
        let entries = git_backend::status_entries(&repo)?;
        let mut tracked = Vec::new();
        let mut untracked = Vec::new();
        for entry in entries {
            match entry.status {
                git_backend::FileStatusKind::Untracked => untracked.push(entry.path),
                _ => tracked.push(entry.path),
            }
        }

        if dry_run {
            return Ok(DiscardAnswer {
                message: format!(
                    "{} tracked, {} untracked",
                    tracked.len(),
                    untracked.len()
                ),
                tracked,
                untracked,
                discarded: false,
            });
        }

        if !tracked.is_empty() {
            git_backend::discard_tracked_changes(&repo)?;
        }
        if untracked_too && !untracked.is_empty() {
            git_backend::remove_untracked_files(&repo)?;
        }
        Ok(DiscardAnswer {
            message: format!(
                "discarded {} tracked change(s){}",
                tracked.len(),
                if untracked_too {
                    format!(" and {} untracked file(s)", untracked.len())
                } else {
                    String::new()
                }
            ),
            tracked,
            untracked,
            discarded: true,
        })
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?;

    match answer {
        Ok(answer) => {
            if let Some(op) = op {
                op.complete().await;
            }
            if answer.discarded {
                notify_git_change_repo(&state, &rel).await;
            }
            Ok(Json(answer))
        }
        Err(error) => {
            if let Some(op) = op {
                op.fail(error.to_string()).await;
            }
            Err(anyhow_error(error))
        }
    }
}

/// Bring the remote's refs up to date without touching the working tree.
pub async fn fleet_fetch(
    State(state): State<WebState>,
    Json(body): Json<RepoPathRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let op = state
        .ops
        .start_op(OpKind::GitMutation, format!("Fetch: {rel}"))
        .await;
    let result = tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
        let repo = git_backend::open_repo(&repo_path)?;
        git_backend::fetch(&repo, Some("origin"))
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));

    match result {
        Ok(Ok(())) => {
            op.complete().await;
            // The working tree did not move, but what the page says about
            // ahead/behind did.
            notify_git_change_repo(&state, &rel).await;
            Ok(ok_result("fetched"))
        }
        Ok(Err(error)) => {
            op.fail(error.to_string()).await;
            Err(anyhow_error(error))
        }
        Err(failure) => Err(failure),
    }
}

#[derive(Deserialize)]
pub struct DeleteBranchRequest {
    pub repo_path: Option<String>,
    pub branch: String,
    /// Delete a branch whose work is not merged anywhere.
    #[serde(default)]
    pub force: bool,
}

/// Delete a local branch in one repository.
pub async fn fleet_delete_branch(
    State(state): State<WebState>,
    Json(body): Json<DeleteBranchRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let branch = body.branch.clone();
    if branch.trim().is_empty() {
        return Err((
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse {
                error: "a branch to delete has to be named".to_string(),
            }),
        ));
    }
    let force = body.force;
    let op = state
        .ops
        .start_op(OpKind::GitMutation, format!("Delete branch {branch}: {rel}"))
        .await;
    let named = branch.clone();
    let result = tokio::task::spawn_blocking(move || -> anyhow::Result<()> {
        let repo = git_backend::open_repo(&repo_path)?;
        // The branch somebody is standing on cannot be deleted from under
        // them: git refuses, and the refusal here says why rather than
        // reporting whatever git says about a ref that is checked out.
        let current = git_backend::current_branch(&repo)?;
        if current == named {
            anyhow::bail!(
                "'{named}' is the branch this repository is on — check out another one first"
            );
        }
        git_backend::delete_local_branch(&repo, &named, force)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));

    match result {
        Ok(Ok(())) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            Ok(ok_result(format!("deleted branch {branch}")))
        }
        Ok(Err(error)) => {
            op.fail(error.to_string()).await;
            Err(anyhow_error(error))
        }
        Err(failure) => Err(failure),
    }
}

pub async fn fleet_push(
    State(state): State<WebState>,
    Json(body): Json<RepoPathRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let label = format!("Push: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::push_current_branch(&repo, Some("origin"), true, false).map_err(anyhow_error)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(outcome)) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            let msg = match outcome {
                git_backend::PushOutcome::Updated => "pushed successfully",
                git_backend::PushOutcome::UpToDate => "already up to date",
            };
            Ok(ok_result(msg))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/pull — pull current branch
pub async fn fleet_pull(
    State(state): State<WebState>,
    Json(body): Json<RepoPathRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let label = format!("Pull: {}", rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        let outcome =
            git_backend::pull(&repo, git_backend::PullOptions::default()).map_err(anyhow_error)?;
        // A pull that could not take what is there is not a pull that
        // succeeded. The page shows this message; "pulled successfully" over a
        // branch that stayed behind is the one thing it must never say.
        pull_message(outcome).map_err(anyhow_error)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(message)) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            Ok(ok_result(message))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/checkout — checkout a branch
pub async fn fleet_checkout(
    State(state): State<WebState>,
    Json(body): Json<CheckoutRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let branch = body.branch.clone();
    let create = body.create.unwrap_or(false);
    let label = format!("Checkout {}: {}", body.branch, rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let branch_name = body.branch.clone();
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        if create {
            git_backend::create_branch(&repo, &branch, None).map_err(anyhow_error)?;
        } else {
            git_backend::checkout_branch(&repo, &branch).map_err(anyhow_error)?;
        }
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(())) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            Ok(ok_result(format!("checked out '{}'", branch_name)))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/branch/create — create a new branch
pub async fn fleet_create_branch(
    State(state): State<WebState>,
    Json(body): Json<CreateBranchRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let repo_path = resolve_repo_path(&root, body.repo_path.as_deref())?;
    let rel = body.repo_path.clone().unwrap_or_else(|| ".".to_string());
    let branch = body.branch.clone();
    let from_ref = body.from_ref.clone();
    let label = format!("Create branch {}: {}", body.branch, rel);
    let op = state.ops.start_op(OpKind::GitMutation, label).await;
    let branch_name = body.branch.clone();
    let result = tokio::task::spawn_blocking(move || {
        let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
        git_backend::create_branch(&repo, &branch, from_ref.as_deref()).map_err(anyhow_error)?;
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(())) => {
            op.complete().await;
            notify_git_change_repo(&state, &rel).await;
            Ok(ok_result(format!("created branch '{}'", branch_name)))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

// ---------------------------------------------------------------------------
// Fleet-wide mutation endpoints
// ---------------------------------------------------------------------------

/// POST /api/v1/fleet/stage-all — stage all changes in every dirty repo
pub async fn fleet_stage_all(
    State(state): State<WebState>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let root_clone = root.clone();
    let op = state
        .ops
        .start_op(OpKind::GitMutation, "Stage all repos")
        .await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Every existing repository, deepest first. Staging a clean repository is a
    // no-op, and a repository that looked clean when the statuses were read can
    // have a gitlink to stage by the time its submodules have been staged, so
    // the dirty set read up front is not a safe filter.
    let ordered_paths = paths_bottom_up(&repos);

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &ordered_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            git_backend::stage_all(&repo).map_err(anyhow_error)?;
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(staged)) => {
            op.complete().await;
            notify_git_change_all(&state).await;
            Ok(ok_result(format!("staged {} repos", staged)))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/commit-all — commit in every dirty repo
pub async fn fleet_commit_all(
    State(state): State<WebState>,
    Json(body): Json<FleetCommitRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let root_clone = root.clone();
    let op = state
        .ops
        .start_op(OpKind::GitMutation, "Commit all repos")
        .await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Deepest first: a parent's commit records its submodules' commits, so they
    // must exist before it runs. Repositories with nothing staged commit
    // nothing, which is why every existing one is visited rather than only
    // those that were dirty when the statuses were read.
    let ordered_paths = paths_bottom_up(&repos);
    let message = body.message.clone();

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &ordered_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            git_backend::stage_all(&repo).map_err(anyhow_error)?;
            if git_backend::commit_all(&repo, &message)
                .map_err(anyhow_error)?
                .is_some()
            {
                count += 1;
            }
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(committed)) => {
            op.complete().await;
            notify_git_change_all(&state).await;
            Ok(ok_result(format!("committed in {} repos", committed)))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/push-all — push submodules first, then superproject
pub async fn fleet_push_all(
    State(state): State<WebState>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let root_clone = root.clone();
    let op = state
        .ops
        .start_op(OpKind::GitMutation, "Push all repos")
        .await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Deepest first: a pushed gitlink must name a commit its remote already has.
    let ordered_paths = paths_bottom_up(&repos);

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &ordered_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            git_backend::push_current_branch(&repo, Some("origin"), true, false)
                .map_err(anyhow_error)?;
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(pushed)) => {
            op.complete().await;
            notify_git_change_all(&state).await;
            Ok(ok_result(format!("pushed {} repos", pushed)))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/pull-all — pull superproject first, then submodules
pub async fn fleet_pull_all(
    State(state): State<WebState>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let root_clone = root.clone();
    let op = state
        .ops
        .start_op(OpKind::GitMutation, "Pull all repos")
        .await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Shallowest first: each repository pulls after the one that declares it.
    let ordered_paths = paths_top_down(&repos);

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &ordered_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            let outcome = git_backend::pull(&repo, git_backend::PullOptions::default())
                .map_err(anyhow_error)?;
            // One diverged repository fails the whole sweep, naming itself.
            // Counting it as pulled would report a number that is not true of
            // any repository in the list.
            pull_message(outcome)
                .map_err(|e| anyhow_error(e.context(format!("in '{rel_path}'"))))?;
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(pulled)) => {
            op.complete().await;
            notify_git_change_all(&state).await;
            Ok(ok_result(format!("pulled {} repos", pulled)))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/checkout-all — checkout same branch in all repos
pub async fn fleet_checkout_all(
    State(state): State<WebState>,
    Json(body): Json<FleetCheckoutRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let root_clone = root.clone();
    let op = state
        .ops
        .start_op(
            OpKind::GitMutation,
            format!("Checkout all: {}", body.branch),
        )
        .await;
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    // Shallowest first, so a repository is on the requested branch before the
    // submodules it declares are switched under it.
    let existing_paths = paths_top_down(&repos);
    let branch = body.branch.clone();
    let create = body.create.unwrap_or(false);
    let branch_name = body.branch.clone();

    let result = tokio::task::spawn_blocking(move || {
        let mut count = 0usize;
        for rel_path in &existing_paths {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            if create {
                git_backend::create_branch(&repo, &branch, None).map_err(anyhow_error)?;
            } else {
                git_backend::checkout_branch(&repo, &branch).map_err(anyhow_error)?;
            }
            count += 1;
        }
        Ok::<usize, (StatusCode, Json<ErrorResponse>)>(count)
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(switched)) => {
            op.complete().await;
            notify_git_change_all(&state).await;
            Ok(ok_result(format!(
                "checked out '{}' in {} repos",
                branch_name, switched
            )))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

/// POST /api/v1/fleet/merge-back — 5-phase merge-back:
/// stage → commit → push feature, checkout main, ff-merge, push main
pub async fn fleet_merge_back(
    State(state): State<WebState>,
    Json(body): Json<FleetMergeBackRequest>,
) -> Result<Json<GitOpResult>, (StatusCode, Json<ErrorResponse>)> {
    let root = workspace_root_of(&state);
    let root_clone = root.clone();
    let repos = tokio::task::spawn_blocking(move || collect_repo_statuses(&root_clone))
        .await
        .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)))?
        .map_err(internal_error)?;

    let commit_msg = body
        .message
        .unwrap_or_else(|| "merge-back: merge feature into main".to_string());

    // Build per-repo inputs: (rel_path, branch), deepest first so a repository
    // merges back only after the submodules whose commits it records have.
    // Validate branch availability before entering spawn_blocking.
    let mut ordered: Vec<&RepoDetailedStatus> = repos.iter().filter(|r| r.exists).collect();
    ordered.sort_by_key(|repo| std::cmp::Reverse(repo.depth));

    let mut repo_inputs: Vec<(String, String)> = Vec::new();
    for repo_status in ordered {
        let current_branch = match &repo_status.branch {
            Some(b) => b.clone(),
            None => {
                return Err((
                    StatusCode::BAD_REQUEST,
                    Json(ErrorResponse {
                        error: format!(
                            "repo '{}' is in detached HEAD state, cannot merge-back",
                            repo_status.name
                        ),
                    }),
                ));
            }
        };
        repo_inputs.push((repo_status.path.clone(), current_branch));
    }

    let op = state
        .ops
        .start_op(OpKind::GitMutation, "Merge-back all repos")
        .await;
    let result = tokio::task::spawn_blocking(move || {
        for (rel_path, current_branch) in &repo_inputs {
            let repo_path = resolve_repo_path_sync(&root, Some(rel_path))?;
            let repo = git_backend::open_repo(&repo_path).map_err(anyhow_error)?;
            let default_branch = git_backend::get_default_branch(&repo).map_err(anyhow_error)?;

            if current_branch == &default_branch {
                continue;
            }

            // Stage unconditionally: the cleanliness read before this run does
            // not account for the gitlinks this repository's own submodules
            // just moved.
            git_backend::stage_all(&repo).map_err(anyhow_error)?;
            git_backend::commit_all(&repo, &commit_msg).map_err(anyhow_error)?;
            git_backend::push_current_branch(&repo, Some("origin"), true, false)
                .map_err(anyhow_error)?;
            git_backend::checkout_branch(&repo, &default_branch).map_err(anyhow_error)?;
            git_backend::merge_ff(&repo, current_branch).map_err(anyhow_error)?;
            git_backend::push_current_branch(&repo, Some("origin"), false, false)
                .map_err(anyhow_error)?;
        }
        Ok::<(), (StatusCode, Json<ErrorResponse>)>(())
    })
    .await
    .map_err(|e| anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e)));
    match result {
        Ok(Ok(())) => {
            op.complete().await;
            notify_git_change_all(&state).await;
            Ok(ok_result("merge-back complete"))
        }
        Ok(Err(e)) => {
            op.fail(handler_err_msg(&e)).await;
            Err(e)
        }
        Err(e) => {
            let err = anyhow_error(anyhow::anyhow!("spawn_blocking panicked: {:?}", e));
            op.fail(handler_err_msg(&err)).await;
            Err(err)
        }
    }
}

// ---------------------------------------------------------------------------
// Background git watcher
// ---------------------------------------------------------------------------

/// Trigger a targeted re-scan of a single repo after a mutation.
async fn notify_git_change_repo(state: &WebState, repo_path: &str) {
    let _ = state.git_nudge_tx.send(Some(repo_path.to_string()));
}

/// Trigger a full re-scan of all repos after a fleet-wide mutation.
async fn notify_git_change_all(state: &WebState) {
    let _ = state.git_nudge_tx.send(None);
}

/// Spawn a per-repo filesystem watcher on each repo's working tree.
///
/// When any tracked file in a repo's working tree changes (edits, saves,
/// creates, deletes), the OS event fires and we send a targeted nudge
/// `Some(rel_path)` so only that repo is re-scanned — not the entire fleet.
///
/// Design:
/// - One scoped watcher per repo (`web::fs_scope`), each with a 200ms
///   debounce window so rapid saves (autosave, formatter) collapse into a
///   single git status call.
/// - The scope excludes git-ignored trees, `.git/` and `.wsb/` at the WATCH
///   layer — a recursive watch costs one inotify watch per directory, and
///   ignored artifact trees exhausted `fs.inotify.max_user_watches` — and
///   stops at nested-repo boundaries, since every repo has its own watcher.
/// - `.git/` churn (refs, index, COMMIT_EDITMSG) is already covered by the
///   mutation handlers calling notify_git_change.
/// The repositories an event path can belong to, innermost first.
///
/// Sorted by path length descending so the FIRST match is the innermost
/// repository: a submodule's event must never be attributed to the
/// superproject that contains it, which would rescan the wrong tree and leave
/// the right one stale.
fn repo_owners(repos: &[RepoDescriptor]) -> Vec<(PathBuf, String)> {
    let mut owners: Vec<(PathBuf, String)> = repos
        .iter()
        .map(|repo| (repo.abs_path.clone(), repo.rel_path.clone()))
        .collect();
    owners.sort_by(|a, b| b.0.as_os_str().len().cmp(&a.0.as_os_str().len()));
    owners
}

/// Which repository an event path belongs to, or `None` when it is outside
/// every one of them.
fn owning_repo<'a>(owners: &'a [(PathBuf, String)], path: &Path) -> Option<&'a str> {
    owners
        .iter()
        .find(|(abs, _)| path.starts_with(abs))
        .map(|(_, rel)| rel.as_str())
}

pub fn spawn_git_fs_watcher(state: WebState) {
    use std::time::Duration;

    let workspace_root = state.workspace_root.clone();
    let nudge_tx = state.git_nudge_tx.clone();

    std::thread::Builder::new()
        .name("git-fs-watcher".to_string())
        .spawn(move || {
            let repos = match build_repo_list(&workspace_root) {
                Ok(r) => r,
                Err(e) => {
                    tracing::warn!(error = %e, "git fs watcher: failed to build repo list");
                    return;
                }
            };

            if repos.is_empty() {
                tracing::warn!("git fs watcher: no repos found, watcher idle");
                return;
            }

            // ONE watcher over every repo's working tree — not one per repo.
            //
            // Two inotify budgets exist and they are not alike. Watches
            // (`max_user_watches`, commonly 65536) are spent per DIRECTORY, so
            // the scope below still excludes ignored trees, `.git/` and
            // `.wsb/`, and stops at nested-repo boundaries. INSTANCES
            // (`max_user_instances`, commonly 128) are spent per watcher and
            // are shared with every other program the person runs: one watcher
            // per repository spent forty of them on this workspace alone, and
            // a second wsb beside an editor then failed with "Too many open
            // files" and silently fell back to the periodic scan.
            //
            // Repo scopes are disjoint (each stops where the next repo
            // begins), so an event belongs to exactly one repo: the one whose
            // working tree is the longest prefix of the event path.
            let owners = repo_owners(&repos);

            let roots: Vec<PathBuf> = repos.iter().map(|repo| repo.abs_path.clone()).collect();
            let repo_count = repos.len();
            let nudge = nudge_tx.clone();
            let outcome = crate::web::fs_scope::spawn_scoped_watcher(
                "git-fs".to_string(),
                roots,
                false,
                Duration::from_millis(200),
                move |events| {
                    // Events inside .git/ and .wsb/ never arrive — those
                    // directories are outside the watch scope — but a FILE
                    // named .git (a submodule pointer) can still produce one,
                    // and wsb's own writes must not feed back into a scan that
                    // causes more writes.
                    let mut touched: std::collections::HashSet<&str> =
                        std::collections::HashSet::new();
                    for event in events {
                        if event.path.components().any(|c| {
                            let s = c.as_os_str();
                            s == ".git" || s == ".wsb"
                        }) {
                            continue;
                        }
                        if let Some(rel) = owning_repo(&owners, &event.path) {
                            touched.insert(rel);
                        }
                    }
                    // One nudge per repo per batch, not per event: a formatter
                    // rewriting thirty files in one repo is one rescan.
                    for rel in touched {
                        let _ = nudge.send(Some(rel.to_string()));
                    }
                },
            );
            let watching = match outcome {
                Ok(()) => repo_count,
                Err(e) => {
                    tracing::warn!(
                        error = %e,
                        "git fs watcher: no working tree is watched; the scan loop's \
                         periodic pass still covers every repo"
                    );
                    0
                }
            };

            tracing::info!(
                "git fs watcher: watching working trees of {} of {} repos",
                watching,
                repo_count
            );
            // Each scoped watcher owns its own lifetime (a process-lifetime
            // rescan thread keeps its debouncer alive); nothing to hold here.
        })
        .expect("git-fs-watcher thread failed to spawn");
}

/// A repo entry descriptor built once from .gitmodules, used to drive
/// the incremental per-repo scan loop.
#[derive(Clone)]
struct RepoDescriptor {
    /// Absolute filesystem path to the repo.
    abs_path: PathBuf,
    /// Repo name (directory / submodule name).
    name: String,
    /// Workspace-relative path used as the key in the status map.
    rel_path: String,
    is_superproject: bool,
    depth: usize,
    parent_path: Option<String>,
}

/// Build the ordered list of repo descriptors for the workspace.
/// Returns superproject first, then submodules in .gitmodules order.
fn build_repo_list(repo_root: &Path) -> wsb::fleet::error::Result<Vec<RepoDescriptor>> {
    let super_name = repo_root
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .unwrap_or_else(|| repo_root.display().to_string());

    let mut list = vec![RepoDescriptor {
        abs_path: repo_root.to_path_buf(),
        name: super_name,
        rel_path: ".".to_string(),
        is_superproject: true,
        depth: 0,
        parent_path: None,
    }];

    // Every repository in the nesting tree gets watched and scanned, not only
    // the superproject's direct submodules.
    let tree = SubmoduleTree::discover_allow_empty(repo_root)?;

    for node in tree.top_down() {
        list.push(RepoDescriptor {
            abs_path: node.abs_path.clone(),
            name: node.name.clone(),
            rel_path: node.display_path.clone(),
            is_superproject: false,
            depth: node.depth + 1,
            parent_path: Some(declaring_repo_path(repo_root, node)),
        });
    }

    Ok(list)
}

/// Spawn the background git status watcher.
///
/// Design:
/// - Nudge channel carries `Option<String>`: `None` = full scan all repos,
///   `Some(rel_path)` = scan only that repo (targeted, from FS watcher).
/// - The 30-second idle interval is a fallback for edge cases where the FS
///   watcher misses an event (e.g. network FS, NFS mounts). Normal operation
///   is entirely event-driven via the nudge channel.
/// - Builds the repo list fresh on every full scan so newly added/removed
///   submodules are picked up automatically.
/// - Each scan (full or targeted) runs all relevant repos concurrently via
///   JoinSet — no waiting for one slow repo before starting the next.
/// - Each repo result is broadcast as soon as it finishes — no accumulation.
/// - Only broadcasts a repo when its serialized status actually changed.
/// - RwLock on the snapshot: multiple SSE clients read simultaneously; the
///   watcher only takes a write lock briefly per changed repo.
pub fn spawn_git_watcher(state: WebState) {
    let workspace_root = state.workspace_root.clone();
    let tx = state.git_status_tx.clone();
    let snapshot = state.git_status_snapshot.clone();
    let ops = state.ops.clone();
    let mut focus_rx = state.git_focus_tx.subscribe();
    let mut nudge_rx = state.git_nudge_tx.subscribe();

    tokio::spawn(async move {
        // Fallback interval — only fires if no FS events arrive. Normal
        // operation is entirely event-driven.
        let idle_interval = tokio::time::Duration::from_secs(30);
        // Per-repo last-seen JSON for change detection, keyed by rel_path.
        let mut last_seen: std::collections::HashMap<String, String> =
            std::collections::HashMap::new();
        // Cached repo list — rebuilt on every full scan.
        let mut cached_repos: Vec<RepoDescriptor> = Vec::new();

        loop {
            // Collect targeted repo set from pending nudges.
            // None in the set means full scan; Some(rel_path) means targeted.
            let mut targeted: std::collections::HashSet<String> = std::collections::HashSet::new();
            let mut full_scan = false;

            // Wait for: idle timeout, focus change, or nudge.
            tokio::select! {
                _ = tokio::time::sleep(idle_interval) => {
                    full_scan = true;
                }
                Ok(nudge) = nudge_rx.recv() => {
                    match nudge {
                        None => full_scan = true,
                        Some(rel_path) => { targeted.insert(rel_path); }
                    }
                }
                Ok(_) = focus_rx.recv() => {
                    full_scan = true;
                }
            }
            // Drain all pending nudges and focus changes without blocking.
            while let Ok(nudge) = nudge_rx.try_recv() {
                match nudge {
                    None => full_scan = true,
                    Some(rel_path) => {
                        targeted.insert(rel_path);
                    }
                }
            }
            while focus_rx.try_recv().is_ok() {}

            // Rebuild repo list on full scans (picks up new/removed submodules).
            if full_scan {
                let root = workspace_root.clone();
                cached_repos =
                    match tokio::task::spawn_blocking(move || build_repo_list(&root)).await {
                        Ok(Ok(r)) => r,
                        Ok(Err(e)) => {
                            tracing::warn!(error = %e, "git watcher: failed to list repos");
                            continue;
                        }
                        Err(e) => {
                            tracing::warn!(error = %e, "git watcher: build_repo_list panicked");
                            continue;
                        }
                    };
            }

            // Determine which repos to scan this cycle.
            let repos_to_scan: Vec<RepoDescriptor> = if full_scan {
                cached_repos.clone()
            } else {
                // Targeted: only repos whose rel_path was nudged.
                // If the targeted path isn't in the cached list (e.g. new
                // submodule added since last full scan), trigger a full scan
                // instead so the list stays accurate.
                let found: Vec<RepoDescriptor> = cached_repos
                    .iter()
                    .filter(|r| targeted.contains(&r.rel_path))
                    .cloned()
                    .collect();
                if found.len() < targeted.len() {
                    // At least one nudged path is unknown — rebuild list.
                    let root = workspace_root.clone();
                    cached_repos = match tokio::task::spawn_blocking(move || build_repo_list(&root))
                        .await
                    {
                        Ok(Ok(r)) => r,
                        Ok(Err(e)) => {
                            tracing::warn!(error = %e, "git watcher: failed to list repos (targeted fallback)");
                            continue;
                        }
                        Err(e) => {
                            tracing::warn!(error = %e, "git watcher: build_repo_list panicked (targeted fallback)");
                            continue;
                        }
                    };
                    cached_repos.clone()
                } else {
                    found
                }
            };

            if repos_to_scan.is_empty() {
                continue;
            }

            let repo_count = repos_to_scan.len();
            let scan_label = if full_scan {
                format!("Scan {} repos", repo_count)
            } else {
                let names: Vec<&str> = repos_to_scan.iter().map(|r| r.name.as_str()).collect();
                format!("Scan: {}", names.join(", "))
            };
            let scan_op = ops.start_op(OpKind::GitScan, scan_label).await;

            // Scan all repos in this cycle concurrently.
            let mut join_set: tokio::task::JoinSet<(
                String,
                Result<RepoDetailedStatus, wsb::fleet::error::SuperGitError>,
            )> = tokio::task::JoinSet::new();

            for desc in repos_to_scan {
                let root = workspace_root.clone();
                join_set.spawn_blocking(move || {
                    let result = collect_detailed_repo_status(
                        &root,
                        &desc.abs_path,
                        &desc.name,
                        &desc.rel_path,
                        desc.is_superproject,
                        desc.depth,
                        desc.parent_path.clone(),
                    );
                    (desc.rel_path, result)
                });
            }

            // Stream each changed repo immediately as it finishes.
            let mut repos_done = 0usize;

            while let Some(join_result) = join_set.join_next().await {
                repos_done += 1;
                if repo_count > 0 {
                    scan_op
                        .progress(repos_done as f32 / repo_count as f32)
                        .await;
                }

                let (rel_path, status_result) = match join_result {
                    Ok(pair) => pair,
                    Err(e) => {
                        tracing::warn!(error = %e, "git watcher: spawn_blocking panicked");
                        continue;
                    }
                };

                let repo_status = match status_result {
                    Ok(s) => s,
                    Err(e) => {
                        tracing::warn!(
                            path = %rel_path,
                            error = %e,
                            "git watcher: failed to collect repo status"
                        );
                        continue;
                    }
                };

                // Only broadcast if this repo's status actually changed.
                let json = match serde_json::to_string(&repo_status) {
                    Ok(j) => j,
                    Err(_) => continue,
                };

                let changed = last_seen
                    .get(&repo_status.path)
                    .map_or(true, |prev| prev != &json);

                if !changed {
                    continue;
                }

                last_seen.insert(repo_status.path.clone(), json);

                {
                    let mut snap = snapshot.write().await;
                    snap.insert(repo_status.path.clone(), repo_status.clone());
                }

                // Broadcast immediately — clients receive each repo update
                // as soon as it is ready, not after all repos finish.
                let _ = tx.send(repo_status);
            }

            scan_op.complete().await;
        }
    });
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn descriptor(root: &Path, rel: &str) -> RepoDescriptor {
        RepoDescriptor {
            abs_path: if rel.is_empty() {
                root.to_path_buf()
            } else {
                root.join(rel)
            },
            name: rel.rsplit('/').next().unwrap_or("superproject").to_string(),
            rel_path: if rel.is_empty() { ".".to_string() } else { rel.to_string() },
            is_superproject: rel.is_empty(),
            depth: rel.matches('/').count(),
            parent_path: None,
        }
    }

    /// TEST9711: an event belongs to the innermost repository containing it.
    ///
    /// One watcher now covers every working tree, so attribution is done here
    /// rather than by having forty watchers each know their own tree. Matching
    /// the first prefix instead of the LONGEST one would hand a submodule's
    /// edit to the superproject: the wrong tree gets rescanned, and the one
    /// the person is editing stays stale on screen.
    #[test]
    fn test9711_an_event_belongs_to_the_innermost_repo() {
        let temp = TempDir::new().unwrap();
        let root = temp.path();
        let repos = vec![
            descriptor(root, ""),
            descriptor(root, "cartridges"),
            descriptor(root, "cartridges/mlxcartridge"),
        ];
        let owners = repo_owners(&repos);

        assert_eq!(
            owning_repo(&owners, &root.join("cartridges/mlxcartridge/Sources/x.swift")),
            Some("cartridges/mlxcartridge")
        );
        assert_eq!(
            owning_repo(&owners, &root.join("cartridges/README.md")),
            Some("cartridges")
        );
        assert_eq!(owning_repo(&owners, &root.join("workspace.json")), Some("."));
        // A path outside every tree belongs to none of them, rather than to
        // whichever happens to sort first.
        assert_eq!(owning_repo(&owners, Path::new("/tmp/elsewhere/file")), None);
        // A sibling whose name merely starts the same must not match by
        // string prefix — starts_with on components is what makes that safe.
        assert_eq!(
            owning_repo(&owners, &root.join("cartridges-old/file")),
            Some(".")
        );
    }

    #[test]
    fn workspace_root_resolves_from_nested_directory() {
        let temp_dir = init_repo();
        let nested = temp_dir.path().join("nested").join("dir");
        std::fs::create_dir_all(&nested).unwrap();

        let resolved = GitOps::discover_repo_root(&nested).unwrap();
        assert_eq!(
            wsb::platform::canonicalize(&resolved).unwrap(),
            wsb::platform::canonicalize(&temp_dir.path()).unwrap()
        );
    }

    #[test]
    fn collect_repo_statuses_returns_repos_shape() {
        let temp_dir = init_repo();
        std::fs::write(
            temp_dir.path().join(".gitmodules"),
            r#"[submodule "libs/core"]
	path = libs/core
	url = git@github.com:example/core.git
"#,
        )
        .unwrap();
        std::fs::create_dir_all(temp_dir.path().join("libs/core")).unwrap();
        wsb::git_backend::git_command()
            .arg("init")
            .arg(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["config", "user.email", "wsb-tests@example.com"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["config", "user.name", "wsb tests"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        std::fs::write(temp_dir.path().join("libs/core/README.md"), "core\n").unwrap();
        wsb::git_backend::git_command()
            .args(["add", "README.md"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["commit", "-m", "init"])
            .current_dir(temp_dir.path().join("libs/core"))
            .output()
            .unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let json = serde_json::to_value(&repos).unwrap();
        assert!(json.is_array());
        assert_eq!(json.as_array().unwrap().len(), 2);
        assert!(json.as_array().unwrap()[0]
            .get("is_superproject")
            .unwrap()
            .as_bool()
            .unwrap());
        // Verify files field exists and is an array
        assert!(json.as_array().unwrap()[0].get("files").unwrap().is_array());
    }

    #[test]
    fn detailed_status_classifies_file_states() {
        let temp_dir = init_repo();

        // Create an untracked file
        std::fs::write(temp_dir.path().join("untracked.txt"), "new file\n").unwrap();

        // Modify a tracked file
        std::fs::write(temp_dir.path().join("README.md"), "modified\n").unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];
        assert!(!super_repo.is_clean);
        assert!(super_repo.files.len() >= 2);

        // Find the untracked file
        let untracked = super_repo
            .files
            .iter()
            .find(|f| f.path == "untracked.txt")
            .expect("untracked.txt should be in status");
        assert_eq!(untracked.status, FileStatusKind::Untracked);

        // Find the modified file
        let modified = super_repo
            .files
            .iter()
            .find(|f| f.path == "README.md")
            .expect("README.md should be in status");
        assert_eq!(modified.status, FileStatusKind::Modified);
    }

    #[test]
    fn detailed_status_classifies_staged_files() {
        let temp_dir = init_repo();

        // Modify and stage a file
        std::fs::write(temp_dir.path().join("README.md"), "staged change\n").unwrap();
        wsb::git_backend::git_command()
            .args(["add", "README.md"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];
        assert!(!super_repo.is_clean);

        let staged = super_repo
            .files
            .iter()
            .find(|f| f.path == "README.md")
            .expect("README.md should be in status");
        assert_eq!(staged.status, FileStatusKind::Staged);
    }

    #[test]
    fn detailed_status_keeps_ignored_files_out_of_dirty_state() {
        let temp_dir = init_repo();

        std::fs::write(temp_dir.path().join(".gitignore"), ".wsb/\nignored.log\n").unwrap();
        wsb::git_backend::git_command()
            .args(["add", ".gitignore"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["commit", "-m", "add ignore rules"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();

        std::fs::create_dir_all(temp_dir.path().join(".wsb")).unwrap();
        std::fs::write(temp_dir.path().join(".wsb/cache.db"), "cache\n").unwrap();
        std::fs::write(temp_dir.path().join("ignored.log"), "log\n").unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];
        assert!(super_repo.is_clean);

        let ignored = super_repo
            .files
            .iter()
            .map(|f| (f.path.clone(), f.status))
            .collect::<std::collections::BTreeMap<_, _>>();
        assert_eq!(ignored.get("ignored.log"), Some(&FileStatusKind::Ignored));
        assert!(ignored.iter().any(
            |(path, status)| (path == ".wsb" || path.starts_with(".wsb/"))
                && *status == FileStatusKind::Ignored
        ));
    }

    #[test]
    fn untracked_dirs_with_only_ignored_or_empty_content_are_excluded() {
        let temp_dir = init_repo();

        // Set up gitignore rules
        std::fs::write(
            temp_dir.path().join(".gitignore"),
            "Package.resolved\n*.xcuserdatad/\n",
        )
        .unwrap();
        wsb::git_backend::git_command()
            .args(["add", ".gitignore"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["commit", "-m", "add gitignore"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();

        // Create a directory whose only file is gitignored (mirrors
        // xcshareddata/swiftpm/Package.resolved in Xcode projects)
        std::fs::create_dir_all(temp_dir.path().join("Foo.xcworkspace/xcshareddata/swiftpm"))
            .unwrap();
        std::fs::write(
            temp_dir
                .path()
                .join("Foo.xcworkspace/xcshareddata/swiftpm/Package.resolved"),
            "{}",
        )
        .unwrap();

        // Create a directory tree that is entirely empty (mirrors .swiftpm
        // directories with only empty subdirs)
        std::fs::create_dir_all(temp_dir.path().join(".swiftpm/xcode/configuration")).unwrap();

        // Create a directory with only ignored content nested deeper
        std::fs::create_dir_all(temp_dir.path().join("deep/nested.xcuserdatad/schemes")).unwrap();
        std::fs::write(
            temp_dir
                .path()
                .join("deep/nested.xcuserdatad/schemes/data.plist"),
            "<plist/>",
        )
        .unwrap();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let super_repo = &repos[0];

        // None of these phantom directories should appear as untracked
        let untracked_paths: Vec<&str> = super_repo
            .files
            .iter()
            .filter(|f| f.status == FileStatusKind::Untracked)
            .map(|f| f.path.as_str())
            .collect();

        assert!(
            !untracked_paths.iter().any(|p| p.contains("xcshareddata")),
            "directory with only gitignored files should not appear as untracked, got: {:?}",
            untracked_paths,
        );
        assert!(
            !untracked_paths.iter().any(|p| p.contains(".swiftpm")),
            "empty directory tree should not appear as untracked, got: {:?}",
            untracked_paths,
        );

        // The repo should be clean (only ignored content was added)
        assert!(
            super_repo.is_clean,
            "repo should be clean when only ignored/empty dirs exist, but found: {:?}",
            super_repo
                .files
                .iter()
                .filter(|f| f.status != FileStatusKind::Ignored)
                .map(|f| format!("{}: {:?}", f.path, f.status))
                .collect::<Vec<_>>(),
        );
    }

    #[test]
    fn resolve_repo_path_rejects_escape() {
        let temp_dir = TempDir::new().unwrap();
        let root = temp_dir.path();
        std::fs::create_dir_all(root.join("sub")).unwrap();
        // Create a directory outside root to escape to
        let outside = temp_dir.path().join("../escape_target");
        std::fs::create_dir_all(&outside).unwrap();

        let result = resolve_repo_path(root, Some("../escape_target"));
        assert!(result.is_err());
    }

    /// A repository at `path`, committed so it has a HEAD to report.
    fn init_repo_at(path: &Path) {
        std::fs::create_dir_all(path).unwrap();
        wsb::git_backend::git_command()
            .arg("init")
            .current_dir(path)
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["config", "user.email", "wsb-tests@example.com"])
            .current_dir(path)
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["config", "user.name", "wsb tests"])
            .current_dir(path)
            .output()
            .unwrap();
        std::fs::write(path.join("README.md"), "repo\n").unwrap();
        wsb::git_backend::git_command()
            .args(["add", "README.md"])
            .current_dir(path)
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["commit", "-m", "init"])
            .current_dir(path)
            .output()
            .unwrap();
    }

    /// Superproject → `libs/core` → `libs/core/vendor/dep`: two levels of
    /// nesting, which is what the flat `.gitmodules` read could not see.
    fn init_nested_fleet() -> TempDir {
        let temp_dir = init_repo();
        std::fs::write(
            temp_dir.path().join(".gitmodules"),
            "[submodule \"core\"]\n\tpath = libs/core\n\turl = git@github.com:example/core.git\n",
        )
        .unwrap();
        init_repo_at(&temp_dir.path().join("libs/core"));
        std::fs::write(
            temp_dir.path().join("libs/core/.gitmodules"),
            "[submodule \"dep\"]\n\tpath = vendor/dep\n\turl = git@github.com:example/dep.git\n",
        )
        .unwrap();
        init_repo_at(&temp_dir.path().join("libs/core/vendor/dep"));
        temp_dir
    }

    /// TEST9600: a submodule declared inside a submodule is a repository of
    /// this fleet. Only the superproject's own `.gitmodules` used to be read,
    /// so every repository below the first level was invisible to the web
    /// layer — absent from the fleet view, unwatched, and never scanned.
    #[test]
    fn test9600_nested_submodules_are_reported_with_their_ancestry() {
        let temp_dir = init_nested_fleet();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let by_path: std::collections::HashMap<&str, &RepoDetailedStatus> =
            repos.iter().map(|r| (r.path.as_str(), r)).collect();

        let superproject = by_path.get(".").expect("superproject is reported");
        assert!(superproject.is_superproject);
        assert_eq!(superproject.depth, 0);
        assert_eq!(superproject.parent_path, None);

        let core = by_path.get("libs/core").expect("submodule is reported");
        assert!(!core.is_superproject);
        assert!(core.exists);
        assert_eq!(core.depth, 1);
        assert_eq!(core.parent_path.as_deref(), Some("."));

        let dep = by_path
            .get("libs/core/vendor/dep")
            .expect("a submodule of a submodule is reported");
        assert!(dep.exists);
        assert_eq!(dep.name, "dep");
        assert_eq!(dep.depth, 2);
        assert_eq!(dep.parent_path.as_deref(), Some("libs/core"));
    }

    /// TEST9601: repositories come back in pre-order, so a client rendering
    /// them in order never has to draw a child before the row it hangs under.
    #[test]
    fn test9601_repo_statuses_are_reported_parent_before_child() {
        let temp_dir = init_nested_fleet();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let position = |path: &str| {
            repos
                .iter()
                .position(|r| r.path == path)
                .unwrap_or_else(|| panic!("{path} is reported"))
        };

        assert!(position(".") < position("libs/core"));
        assert!(position("libs/core") < position("libs/core/vendor/dep"));
    }

    /// TEST9602: the fleet status endpoint sorts the snapshot it serves. The
    /// snapshot is a map, so without this the rows arrive in whatever order
    /// hashing produced and the nesting cannot be drawn.
    #[test]
    fn test9602_snapshot_order_is_restored_by_the_tree_sort() {
        let temp_dir = init_nested_fleet();

        let repos = collect_repo_statuses(temp_dir.path()).unwrap();
        let mut shuffled: Vec<&RepoDetailedStatus> = repos.iter().rev().collect();
        shuffled.sort_by(|left, right| tree_order_key(left).cmp(&tree_order_key(right)));

        let paths: Vec<&str> = shuffled.iter().map(|r| r.path.as_str()).collect();
        assert_eq!(paths, vec![".", "libs/core", "libs/core/vendor/dep"]);
    }

    /// TEST9603: fleet-wide mutations run in an order git can accept. A parent
    /// records its submodules' commits, so staging, committing and pushing
    /// must reach the deepest repository first; pulling and checking out move
    /// outward from the superproject instead.
    #[test]
    fn test9603_mutation_order_follows_the_nesting() {
        let temp_dir = init_nested_fleet();
        let mut repos = collect_repo_statuses(temp_dir.path()).unwrap();
        repos.push(RepoDetailedStatus {
            name: "absent".to_string(),
            path: "libs/absent".to_string(),
            branch: None,
            is_clean: true,
            is_detached: false,
            ahead: 0,
            behind: 0,
            files: Vec::new(),
            exists: false,
            is_superproject: false,
            depth: 1,
            parent_path: Some(".".to_string()),
        });

        let bottom_up = paths_bottom_up(&repos);
        assert_eq!(
            bottom_up,
            vec![
                "libs/core/vendor/dep".to_string(),
                "libs/core".to_string(),
                ".".to_string(),
            ],
        );

        let top_down = paths_top_down(&repos);
        assert_eq!(
            top_down,
            vec![
                ".".to_string(),
                "libs/core".to_string(),
                "libs/core/vendor/dep".to_string(),
            ],
        );

        // A repository that is not checked out has no working tree to act on.
        assert!(!bottom_up.contains(&"libs/absent".to_string()));
        assert!(!top_down.contains(&"libs/absent".to_string()));
    }

    /// TEST9604: the watcher watches and scans every repository in the tree,
    /// not only the superproject's direct submodules — a nested repository
    /// whose files change must still produce a status update.
    #[test]
    fn test9604_watched_repo_list_covers_the_whole_tree() {
        let temp_dir = init_nested_fleet();

        let descriptors = build_repo_list(temp_dir.path()).unwrap();
        let paths: Vec<&str> = descriptors.iter().map(|d| d.rel_path.as_str()).collect();
        assert_eq!(paths, vec![".", "libs/core", "libs/core/vendor/dep"]);

        let dep = descriptors
            .iter()
            .find(|d| d.rel_path == "libs/core/vendor/dep")
            .unwrap();
        assert_eq!(dep.depth, 2);
        assert_eq!(dep.parent_path.as_deref(), Some("libs/core"));
        assert_eq!(
            dep.abs_path,
            temp_dir.path().join("libs/core/vendor/dep"),
            "the watcher must be pointed at the nested working tree itself"
        );
    }

    fn init_repo() -> TempDir {
        let temp_dir = TempDir::new().unwrap();
        wsb::git_backend::git_command()
            .arg("init")
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["config", "user.email", "wsb-tests@example.com"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["config", "user.name", "wsb tests"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        std::fs::write(temp_dir.path().join("README.md"), "workspace\n").unwrap();
        wsb::git_backend::git_command()
            .args(["add", "README.md"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .args(["commit", "-m", "init"])
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        wsb::git_backend::git_command()
            .arg("remote")
            .arg("add")
            .arg("origin")
            .arg("git@github.com:example/workspace.git")
            .current_dir(temp_dir.path())
            .output()
            .unwrap();
        temp_dir
    }
}