wsb-daemon 0.248.6265

//! The machine this daemon serves, and every worktree on it.
//!
//! One daemon, one machine, N worktrees. Each worktree keeps everything it
//! kept before — its database, its watchers, its operations — and the daemon
//! holds them side by side rather than being one of them.
//!
//! # Why the worktree is explicit everywhere
//!
//! It used to be implicit in the PROCESS: whichever directory started the
//! daemon was the workspace, and nothing had to name it. Then it was named by
//! its repository, and every worktree of a project answered to the first one
//! opened. So the worktree is a parameter, a route segment and a map key —
//! [`WorktreeId`], which is what `prj` calls it — and there is no "current"
//! one anywhere in this process. What two machines compare, the
//! [`WorkspaceId`], is a property of a worktree, looked up through it.

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{Context, Result};
use tokio::sync::RwLock;

use wsb::daemon::{Continuity, RegisteredWorktree, WorktreeId, WorktreePlace};
use wsb::peer::identity::MachineIdentity;
use wsb::peer::model::WorkspaceId;
use wsb::peer::protocol::WorkspaceSummary;
use wsb::peer::scan::Policy;

use crate::peer::coordinator::LocalWorkspace;
use crate::peer::service::{now, PeerService, DEFAULT_PEER_PORT};
use crate::peer::store::MachineStore;
use crate::web::state::WebState;

/// One worktree, open.
pub struct WorktreeRuntime {
    pub registered: RegisteredWorktree,
    /// Everything the workbench needs for this worktree.
    pub web: WebState,
    /// The routes, already bound to that state. Held behind a mutex because a
    /// `Router` is `Send` and not `Sync`, and the machine it hangs from is
    /// shared across threads; each request clones it out (which is cheap) and
    /// serves from the clone. Held at all so a worktree opened while the
    /// daemon runs is reachable immediately, without rebuilding the machine's
    /// own router.
    router: std::sync::Mutex<axum::Router>,
    /// The repository tree, for continuity — or why this worktree has none: a
    /// repository with no commits, or a name two machines could not share.
    pub local: Result<Arc<LocalWorkspace>, String>,
}

impl WorktreeRuntime {
    /// A clone of this worktree's routes, to serve one request from.
    pub fn router(&self) -> axum::Router {
        self.router
            .lock()
            .expect("a worktree router is unusable after a panic")
            .clone()
    }
}

/// The machine.
pub struct MachineState {
    pub identity: Arc<MachineIdentity>,
    pub store: Arc<MachineStore>,
    pub peers: Arc<PeerService>,
    pub machine_dir: PathBuf,
    pub listen: std::net::IpAddr,
    pub port: u16,
    pub started_at: u64,
    /// The secret that proves a caller is this machine's operator.
    ///
    /// Published in the 0600 rendezvous file and nowhere else, so reading it
    /// is the proof — see [`wsb::daemon::MachineRendezvous::control_token`].
    /// New every time the daemon starts: a token from a daemon that is gone
    /// opens nothing.
    pub control_token: String,
    worktrees: RwLock<BTreeMap<WorktreeId, Arc<WorktreeRuntime>>>,
    /// What the last reconciliation observed, per relationship.
    ///
    /// Observed state is never persisted — see the machine store — but it is
    /// remembered for as long as this process runs, so a page asking "what is
    /// happening" gets the last answer rather than waiting for the next sweep.
    observed: std::sync::Mutex<
        BTreeMap<(String, String), Vec<wsb::peer::model::RepositoryStatus>>,
    >,
    /// What the last reconciliation planned to move, per repository.
    ///
    /// Counts answer "is anything different"; a person deciding whether to
    /// let it move wants to know WHICH files, and a count cannot be read that
    /// way. Held for the life of the process for the same reason as
    /// `observed`: it is what was last seen, never an assumption about now.
    plans: std::sync::Mutex<BTreeMap<(String, String, String), Vec<PlannedMove>>>,
    /// "Look again now", raised by the filesystem watcher.
    pub nudge: Arc<crate::peer::watch::Nudge>,
    /// One gate per worktree being opened for the first time.
    ///
    /// Opening is the expensive path — it opens the worktree database, runs
    /// whatever migrations it is behind on, and walks the repository tree —
    /// and several commands starting at once all resolve the same worktree.
    /// Without a gate each of them builds it: the same database opened N
    /// times, N sets of migrations serialising against each other, and every
    /// one of those callers waiting on the slowest. With it, the first caller
    /// builds and the rest wait for that build.
    registering: tokio::sync::Mutex<BTreeMap<WorktreeId, Arc<tokio::sync::Mutex<()>>>>,
    /// The roots the continuity watcher currently covers.
    ///
    /// Held so opening a worktree can tell whether the watcher needs to be
    /// started again over a wider set — a watcher covering three roots says
    /// nothing about a fourth.
    watching: std::sync::Mutex<Vec<PathBuf>>,
    /// The machine's GitHub mirror, which every worktree shares: GitHub
    /// belongs to accounts, not to checkouts. Or why there is none.
    pub github: Result<Arc<crate::web::github::GitHub>, String>,
    /// The plugins running on this machine and the services they answer.
    pub plugins: Arc<crate::plugin::PluginRegistry>,
    /// The workbench's machine-wide state: the machine's agent sessions and
    /// their one sync engine.
    pub web: crate::web::machine_web::MachineWeb,
    /// The machine's projects and worktrees, as last read.
    pub listing: Arc<crate::web::handlers::prj::Listing>,
}

/// Whether two paths name the same directory.
///
/// Compared through the filesystem where both exist, because the same
/// directory has more than one spelling — a symlinked home, `/tmp` against
/// `/private/tmp` on macOS — and a textual comparison would call one worktree
/// two. Where a path cannot be canonicalised (it has been removed) the literal
/// paths are all there is to compare.
fn same_directory(one: &Path, other: &Path) -> bool {
    match (wsb::platform::canonicalize(&one), wsb::platform::canonicalize(&other)) {
        (Ok(one), Ok(other)) => one == other,
        _ => one == other,
    }
}

/// One path the last reconciliation intended to move, and which way.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub struct PlannedMove {
    pub path: String,
    /// `send` | `receive` | `send_deletion` | `receive_deletion` | `conflict`.
    pub direction: &'static str,
    /// The size of the version that would land. Absent for a deletion, which
    /// has no bytes, and for a conflict, which has two.
    pub size: Option<u64>,
    /// Which side somebody chose to keep — `here` or `there` — when they have.
    ///
    /// In manual mode this is the difference between a listed difference and
    /// one that is about to move, and the surfaces have to be able to say
    /// which is which before it happens rather than after.
    pub decision: Option<&'static str>,
    /// Left alone deliberately, as it stands. Reappears by itself if either
    /// machine changes the path again.
    pub dismissed: bool,
    /// What each side holds for this path, as of the scan this plan came
    /// from. Absent means that machine does not have the file.
    ///
    /// Carried so that dismissing a difference can record the two versions it
    /// was about: a dismissal is a statement about these bytes, and the ones
    /// the operator was looking at are the ones in the list they clicked.
    pub here_hash: Option<String>,
    pub there_hash: Option<String>,
}

impl MachineState {
    pub fn new(
        machine_dir: PathBuf,
        listen: std::net::IpAddr,
        port: u16,
        // Where PEERS reach this machine, which is not where the workbench is
        // served: one is authenticated and encrypted, the other is neither.
        peer_listen: std::net::IpAddr,
        peer_port: u16,
    ) -> Result<Arc<Self>> {
        let identity = Arc::new(
            MachineIdentity::load_or_create(&machine_dir, &default_machine_name())
                .map_err(|e| anyhow::anyhow!(e))?,
        );
        let store = Arc::new(MachineStore::open(&machine_dir).map_err(|e| anyhow::anyhow!(e))?);
        let peers = Arc::new(PeerService::new(
            identity.clone(),
            store.clone(),
            peer_listen,
            if peer_port == 0 { DEFAULT_PEER_PORT } else { peer_port },
        ));
        let plugins = Arc::new(crate::plugin::PluginRegistry::new(machine_dir.clone()));
        let web = crate::web::machine_web::MachineWeb::open(&machine_dir)?;
        let prj = match wsb_prj::Prj::from_env() {
            Ok(prj) => Some(prj),
            Err(wsb_prj::PrjError::NotSet) => None,
            Err(error) => return Err(anyhow::anyhow!(error)),
        };
        Ok(Arc::new(Self {
            identity,
            store,
            peers,
            machine_dir,
            listen,
            port,
            started_at: now(),
            control_token: wsb::daemon::new_control_token(),
            worktrees: RwLock::new(BTreeMap::new()),
            observed: std::sync::Mutex::new(BTreeMap::new()),
            plans: std::sync::Mutex::new(BTreeMap::new()),
            registering: tokio::sync::Mutex::new(BTreeMap::new()),
            nudge: Arc::new(crate::peer::watch::Nudge::default()),
            watching: std::sync::Mutex::new(Vec::new()),
            github: crate::web::github::GitHub::open(),
            plugins,
            web,
            listing: Arc::new(crate::web::handlers::prj::Listing::new(prj)),
        }))
    }

    /// Something the projects listing shows may have moved: read it again.
    pub fn note_projects_changed(&self) {
        self.listing.want();
    }

    /// Every worktree open right now.
    pub async fn all(&self) -> Vec<Arc<WorktreeRuntime>> {
        self.worktrees.read().await.values().cloned().collect()
    }

    pub async fn get(&self, id: &WorktreeId) -> Option<Arc<WorktreeRuntime>> {
        self.worktrees.read().await.get(id).cloned()
    }

    /// The open worktree a peer means by `workspace`, with its repository
    /// tree; none when no open worktree carries it.
    ///
    /// Two open worktrees carrying the same continuity id — two clones of one
    /// repository whose worktrees share a name, in two projects or outside
    /// `$PRJ` — are refused, naming both: carrying one machine's files into
    /// whichever of them came first is the mistake keying by worktree exists
    /// to stop.
    pub async fn continuity(
        &self,
        workspace: &WorkspaceId,
    ) -> Result<Option<(Arc<WorktreeRuntime>, Arc<LocalWorkspace>)>, String> {
        let carrying: Vec<Arc<WorktreeRuntime>> = self
            .all()
            .await
            .into_iter()
            .filter(|runtime| runtime.registered.workspace.as_ref() == Some(workspace))
            .collect();
        match carrying.as_slice() {
            [] => Ok(None),
            [one] => {
                let local = one.local.clone().map_err(|why| {
                    format!("{} has no continuity: {why}", one.registered.display_name())
                })?;
                Ok(Some((one.clone(), local)))
            }
            several => Err(ambiguous(workspace, several)),
        }
    }

    /// The worktree `id` names, opened if it is not.
    ///
    /// A `$PRJ` worktree is found where `prj` keeps it; any other by where
    /// this machine last saw it. Either way it is then resolved like a
    /// command's directory, and the answer must be the worktree asked for.
    pub async fn open(&self, id: &WorktreeId) -> Result<Arc<WorktreeRuntime>, String> {
        if let Some(open) = self.get(id).await {
            return Ok(open);
        }
        let root = match id.in_project() {
            Some((project, worktree)) => {
                let prj = wsb_prj::Prj::from_env().map_err(|e| e.to_string())?;
                prj.project(project)
                    .map_err(|e| e.to_string())?
                    .dir()
                    .join(wsb_prj::worktree_folder(worktree))
            }
            None => self
                .store
                .worktrees()?
                .into_iter()
                .find(|recorded| &recorded.id == id)
                .map(|recorded| recorded.super_root)
                .ok_or_else(|| format!("this machine has never opened a worktree {id}"))?,
        };
        if !root.is_dir() {
            return Err(format!("worktree {id} is not there: {} does not exist", root.display()));
        }
        let runtime = self.register(&root).await?;
        if &runtime.registered.id != id {
            return Err(format!(
                "{} is worktree {}, not {id}",
                root.display(),
                runtime.registered.id
            ));
        }
        Ok(runtime)
    }

    /// Resolve a directory to the worktree that contains it, opening it if
    /// this daemon has not.
    ///
    /// This is what replaces "the daemon's workspace". A command sends the
    /// directory it was run in; the daemon walks to the super root exactly as
    /// every other part of wsb does, and answers with that worktree. A
    /// command run in a submodule therefore reaches its worktree's state, not
    /// a second daemon — and a command run in another worktree of the same
    /// project reaches that one.
    pub async fn resolve(&self, cwd: &Path) -> Result<Arc<WorktreeRuntime>, String> {
        let root = wsb::workspace::find_workspace_root(cwd).ok_or_else(|| {
            format!(
                "{} is not inside a wsb worktree: no repository above it holds .wsb/state.json \
                 (`wsb init` makes one)",
                cwd.display()
            )
        })?;
        self.register(&root).await
    }

    /// Open a worktree by its SUPER root, or return the one already open.
    ///
    /// Private on purpose. A worktree is a super root, and the only way to
    /// name one is to walk a directory up to it — [`MachineState::resolve`].
    /// A caller that could hand this an arbitrary directory could open a
    /// submodule as a worktree of its own: two worktrees where the machine
    /// has one, each with its own database, and neither knowing about the
    /// other.
    async fn register(&self, root: &Path) -> Result<Arc<WorktreeRuntime>, String> {
        // The invariant, checked rather than trusted: everything below is
        // keyed by this path, and a wrong one is not recoverable from once its
        // database exists.
        let super_root = wsb::workspace::find_super_root(root).ok_or_else(|| {
            format!(
                "{} is not inside a git repository, so it is not part of any worktree",
                root.display()
            )
        })?;
        if !same_directory(&super_root, root) {
            return Err(format!(
                "{} is not a worktree root: it is inside the worktree at {}",
                root.display(),
                super_root.display()
            ));
        }

        let place = wsb::daemon::place_worktree(root)?;
        if let Some(existing) = self.touch(&place.id).await {
            return Ok(existing);
        }

        // One build per worktree, however many callers arrive at once. The
        // gate is taken before the second look, so a caller that waited here
        // finds the finished worktree rather than starting a second build of
        // it.
        let gate = {
            let mut gates = self.registering.lock().await;
            gates.entry(place.id.clone()).or_default().clone()
        };
        let _building = gate.lock().await;
        if let Some(existing) = self.touch(&place.id).await {
            return Ok(existing);
        }

        let continuity = wsb::daemon::identify_workspace(&place)?;
        let id = place.id.clone();
        let opened_at = place.root.clone();
        let runtime = self
            .build(place, continuity)
            .await
            // `{e:#}` rather than `{e}`: anyhow's `Display` prints only the
            // OUTERMOST context, and every reason a worktree fails to open is
            // underneath one. A database written by a different build of wsb
            // refuses with a paragraph naming both migration sets and saying to
            // delete the file; through `{e}` all of that became "cannot open
            // the worktree at <root>: cannot open <root>" -- the same path
            // twice and no cause.
            .map_err(|e| format!("cannot open the worktree at {}: {e:#}", opened_at.display()))?;
        let runtime = Arc::new(runtime);
        self.store.register_worktree(&runtime.registered)?;
        self.worktrees.write().await.insert(id, runtime.clone());
        self.watch_worktrees().await;
        self.note_projects_changed();
        Ok(runtime)
    }

    /// A worktree already open, with its place in the list refreshed.
    ///
    /// `last_seen` is what orders the list a person reads, so it moves on
    /// every resolve rather than only on the first one.
    async fn touch(&self, id: &WorktreeId) -> Option<Arc<WorktreeRuntime>> {
        let existing = self.get(id).await?;
        let mut registered = existing.registered.clone();
        registered.last_seen = now();
        if let Err(error) = self.store.register_worktree(&registered) {
            tracing::warn!(%error, worktree = %id, "cannot record that the worktree was used");
        }
        Some(existing)
    }

    /// Build everything one worktree needs.
    async fn build(&self, place: WorktreePlace, continuity: Continuity) -> Result<WorktreeRuntime> {
        // Opening a worktree is blocking work — reading `.gitmodules`,
        // writing git's exclude file, opening the database and running every
        // migration it is behind on — and it runs on the same runtime that
        // answers every other request. On the blocking pool it cannot starve
        // the async workers, so a daemon opening a large worktree for the
        // first time still answers everything else while it does.
        //
        // A malformed `.gitmodules` is refused here rather than at startup:
        // the fleet's every operation reads it, and a worktree whose
        // repository tree cannot be trusted is one this daemon must not serve.
        // Scoped to the worktree — one bad checkout must not stop the machine.
        //
        // Opening it is also where a worktree acquires a `.wsb/` and its
        // database. Git has to be told to ignore it before it exists:
        // untracked, it turns up in every status and blocks a pull that
        // touches those paths.
        let root = place.root.clone();
        let opening = root.clone();
        let ctx = tokio::task::spawn_blocking(move || -> Result<wsb::app::AppContext> {
            crate::web::validate_gitmodules_if_present(&opening)?;
            wsb::workspace::exclude_state_dir(&opening)
                .with_context(|| format!("cannot exclude {}/.wsb from git", opening.display()))?;
            wsb::app::AppContext::bootstrap(&opening)
                .with_context(|| format!("cannot open {}", opening.display()))
        })
        .await
        .context("the task opening the worktree did not finish")??;
        // The sessions that ran here: this worktree of this project. A
        // checkout outside `$PRJ` has none placed in it.
        let sessions_here = place.project.clone().map(|project| wsb::persist::global::models::PlaceFilter {
            project: Some(project),
            worktree: Some(place.name.clone()),
            placed: false,
        });
        let web = WebState::new(
            ctx.db.clone(),
            ctx.db_service.clone(),
            self.web.db.clone(),
            sessions_here,
            root.clone(),
            self.listen,
            self.port,
            self.github.clone(),
        )
        .await?;

        // Reads the operation logs already on disk back into history, and
        // keeps the directory under its limit from here on.
        crate::web::log_janitor::spawn(web.clone());

        // Blocking for the same reason: it walks the repository tree, which
        // on a superrepo means one git open per submodule.
        let local = match continuity.workspace.clone() {
            Ok(workspace) => {
                let walking = root.clone();
                tokio::task::spawn_blocking(move || {
                    LocalWorkspace::discover(workspace, &walking, continuity_policy(&walking)?)
                        .map(Arc::new)
                })
                .await
                .context("the task reading the repository tree did not finish")?
            }
            Err(why) => Err(why),
        };
        if let (Ok(_), Err(why)) = (&continuity.workspace, &local) {
            tracing::warn!(worktree = %place.id, error = %why, "continuity is off for this worktree");
        }

        Ok(WorktreeRuntime {
            registered: RegisteredWorktree {
                id: place.id,
                project: place.project,
                name: place.name,
                workspace: continuity.workspace.ok(),
                super_root: root,
                root_commit: continuity.root_commit,
                last_seen: now(),
                available: true,
            },
            router: std::sync::Mutex::new(crate::web::create_router(web.clone())),
            web,
            local,
        })
    }

    /// Forget every recorded worktree that is inside another one.
    ///
    /// A worktree is a super root, so no worktree can contain another: a
    /// record for a submodule of a recorded worktree is the trace of a
    /// resolution that should never have happened — a directory taken for a
    /// root instead of being walked to one. That record outlives the bug, in
    /// the machine's own database, and a person reading the list sees two
    /// worktrees where the machine has one. Corrected on the way up rather
    /// than left to be read.
    ///
    /// Reports what it dropped, so the daemon's log says it happened rather
    /// than a worktree quietly disappearing from a list.
    pub fn forget_contained_worktrees(&self) -> Result<Vec<PathBuf>, String> {
        let registered = self.store.worktrees()?;
        let mut dropped = Vec::new();
        for candidate in &registered {
            let contained = registered.iter().any(|other| {
                other.id != candidate.id
                    && !same_directory(&other.super_root, &candidate.super_root)
                    && candidate.super_root.starts_with(&other.super_root)
            });
            if contained {
                self.store.forget_worktree(&candidate.id)?;
                dropped.push(candidate.super_root.clone());
            }
        }
        Ok(dropped)
    }

    /// Cover every open worktree with the continuity watcher.
    ///
    /// Started again from scratch whenever the set of worktrees grows, which
    /// is rare, and cheap next to missing every change in a worktree that
    /// opened after the daemon did. Failure is logged and not fatal: the
    /// watcher only makes reconciliation prompt, and the sweep still runs.
    async fn watch_worktrees(&self) {
        let roots: Vec<PathBuf> = self
            .worktrees
            .read()
            .await
            .values()
            .filter(|runtime| runtime.local.is_ok())
            .map(|runtime| runtime.registered.super_root.clone())
            .collect();
        {
            let watching = self.watching.lock().expect("the watch set lock is poisoned");
            if *watching == roots {
                return;
            }
        }
        match crate::peer::watch::spawn(roots.clone(), self.nudge.clone()) {
            Ok(count) => {
                *self.watching.lock().expect("the watch set lock is poisoned") = roots;
                tracing::debug!(roots = count, "watching for working-tree changes");
            }
            Err(error) => {
                tracing::warn!(%error, "continuity will reconcile on its sweep alone");
            }
        }
    }

    /// Re-read one worktree's repository tree.
    ///
    /// A submodule that was initialised, or a worktree whose `.gitmodules`
    /// changed, changes what continuity covers. Recomputed rather than
    /// cached for the life of the daemon.
    pub async fn refresh_tree(&self, workspace: &WorkspaceId) -> Result<(), String> {
        let Some((runtime, _)) = self.continuity(workspace).await? else {
            return Ok(());
        };
        let local = Arc::new(LocalWorkspace::discover(
            workspace.clone(),
            &runtime.registered.super_root,
            continuity_policy(&runtime.registered.super_root)?,
        )?);
        let mut worktrees = self.worktrees.write().await;
        if let Some(current) = worktrees.get(&runtime.registered.id) {
            let replacement = Arc::new(WorktreeRuntime {
                registered: current.registered.clone(),
                web: current.web.clone(),
                router: std::sync::Mutex::new(current.router()),
                local: Ok(local),
            });
            worktrees.insert(runtime.registered.id.clone(), replacement);
        }
        Ok(())
    }

    /// What this machine advertises to a peer: every open worktree that has
    /// continuity, once.
    ///
    /// A continuity id two open worktrees share is advertised by neither —
    /// see [`MachineState::continuity`] — and the daemon's log says which.
    pub async fn summaries(&self) -> Vec<WorkspaceSummary> {
        let mut by_workspace: BTreeMap<WorkspaceId, Vec<Arc<WorktreeRuntime>>> = BTreeMap::new();
        for runtime in self.all().await {
            if runtime.local.is_err() {
                continue;
            }
            if let Some(workspace) = runtime.registered.workspace.clone() {
                by_workspace.entry(workspace).or_default().push(runtime);
            }
        }
        let mut summaries = Vec::new();
        for (workspace, runtimes) in by_workspace {
            match runtimes.as_slice() {
                [one] => match &one.registered.root_commit {
                    Some(root_commit) => summaries.push(WorkspaceSummary {
                        workspace,
                        display_name: one.registered.display_name(),
                        root_commit: root_commit.clone(),
                    }),
                    None => tracing::error!(
                        worktree = %one.registered.id,
                        "a worktree with a continuity id has no root commit"
                    ),
                },
                several => tracing::warn!(error = %ambiguous(&workspace, several), "not advertised"),
            }
        }
        summaries
    }

    pub fn staging_root(&self) -> PathBuf {
        self.machine_dir.join("staging")
    }

    /// Record what a reconciliation pass saw.
    pub fn record_statuses(
        &self,
        peer: &wsb::peer::model::PeerId,
        workspace: &WorkspaceId,
        statuses: &[wsb::peer::model::RepositoryStatus],
    ) {
        // Every status is checked before it is stored: "blocked with no
        // reason" is a state no surface can render, and the place to catch it
        // is where it is produced rather than where it is displayed.
        for status in statuses {
            if let Err(error) = status.check() {
                tracing::error!(%error, "a reconciliation produced an ill-formed status");
                return;
            }
        }
        if let Ok(mut observed) = self.observed.lock() {
            observed.insert(
                (peer.to_string(), workspace.to_string()),
                statuses.to_vec(),
            );
        }
    }

    /// Record what one repository's reconciliation intended to move.
    pub fn record_plan(
        &self,
        peer: &wsb::peer::model::PeerId,
        workspace: &WorkspaceId,
        repository: &wsb::peer::model::RepositoryId,
        moves: Vec<PlannedMove>,
    ) {
        if let Ok(mut plans) = self.plans.lock() {
            // Keyed by the repository's PATH, not its label: `to_string` on
            // a repository id renders the superproject as "superproject" for
            // a person to read, and a key that a route looks up by path must
            // be the path.
            let key = (
                peer.to_string(),
                workspace.to_string(),
                repository.as_str().to_string(),
            );
            // An empty plan is a fact worth storing: it is what "nothing
            // differs" looks like, and leaving the previous plan in place
            // would report movement that already happened.
            plans.insert(key, moves);
        }
    }

    /// What was last planned for one relationship, repository by repository.
    pub fn plans(
        &self,
        peer: &wsb::peer::model::PeerId,
        workspace: &WorkspaceId,
    ) -> BTreeMap<String, Vec<PlannedMove>> {
        let prefix = (peer.to_string(), workspace.to_string());
        self.plans
            .lock()
            .ok()
            .map(|plans| {
                plans
                    .iter()
                    .filter(|((p, w, _), _)| p == &prefix.0 && w == &prefix.1)
                    .map(|((_, _, repository), moves)| (repository.clone(), moves.clone()))
                    .collect()
            })
            .unwrap_or_default()
    }

    /// Forget everything observed about a peer that is no longer one.
    pub fn forget_peer(&self, peer: &wsb::peer::model::PeerId) {
        let peer = peer.to_string();
        if let Ok(mut observed) = self.observed.lock() {
            observed.retain(|(p, _), _| p != &peer);
        }
        if let Ok(mut plans) = self.plans.lock() {
            plans.retain(|(p, _, _), _| p != &peer);
        }
    }

    /// Every relationship's last observed statuses, as (peer, workspace, …).
    ///
    /// For the census, which is machine-wide: it merges what is waiting to
    /// move (observed, and gone when this process ends) with what is in
    /// conflict (persisted, and not).
    pub fn observed_statuses(
        &self,
    ) -> Vec<(String, String, Vec<wsb::peer::model::RepositoryStatus>)> {
        self.observed
            .lock()
            .map(|observed| {
                observed
                    .iter()
                    .map(|((peer, workspace), statuses)| {
                        (peer.clone(), workspace.clone(), statuses.clone())
                    })
                    .collect()
            })
            .unwrap_or_default()
    }

    /// What was last observed for one relationship.
    pub fn statuses(
        &self,
        peer: &wsb::peer::model::PeerId,
        workspace: &WorkspaceId,
    ) -> Vec<wsb::peer::model::RepositoryStatus> {
        self.observed
            .lock()
            .ok()
            .and_then(|observed| observed.get(&(peer.to_string(), workspace.to_string())).cloned())
            .unwrap_or_default()
    }

    /// Ask a peer what workspaces it has.
    pub async fn peer_inventory(
        self: &Arc<Self>,
        peer: &wsb::peer::model::PeerId,
    ) -> Result<Vec<WorkspaceSummary>, String> {
        let machine = self.clone();
        let peer = peer.clone();
        tokio::task::spawn_blocking(move || {
            let mut session = machine.peers.connect(&peer)?;
            let answer: wsb::peer::protocol::Response =
                session.exchange(&wsb::peer::protocol::Request::Inventory)?;
            session.shutdown();
            match answer {
                wsb::peer::protocol::Response::Inventory { workspaces } => Ok(workspaces),
                other => Err(format!("the peer answered an inventory with {other:?}")),
            }
        })
        .await
        .map_err(|e| format!("join error: {e}"))?
    }
}

/// A blocking view of the machine for the peer connection server, which runs
/// on its own threads rather than in the async runtime.
///
/// The peer protocol is request/response over a blocking socket, and the work
/// behind it — scanning trees, hashing files, writing them — is blocking too.
/// Running it on the runtime's threads would starve the workbench.
pub struct BlockingMachine {
    pub state: Arc<MachineState>,
}

impl crate::peer::connection::Machine for BlockingMachine {
    fn workspaces(&self) -> Vec<WorkspaceSummary> {
        let state = self.state.clone();
        // `block_on` inside a runtime thread would deadlock; this type is only
        // ever used from threads that are NOT the runtime's, and the handle is
        // how it reaches state the runtime owns.
        futures::executor::block_on(async move { state.summaries().await })
    }

    fn workspace(&self, id: &WorkspaceId) -> Result<Arc<LocalWorkspace>, String> {
        let state = self.state.clone();
        let id = id.clone();
        futures::executor::block_on(async move { state.continuity(&id).await })?
            .map(|(_, local)| local)
            .ok_or_else(|| "this machine does not have that worktree open".to_string())
    }

    fn store(&self) -> Arc<MachineStore> {
        self.state.store.clone()
    }

    fn staging_root(&self) -> PathBuf {
        self.state.staging_root()
    }

    fn plugins(&self) -> Arc<crate::plugin::PluginRegistry> {
        self.state.plugins.clone()
    }
}


/// Two open worktrees claiming one continuity id, named.
fn ambiguous(workspace: &WorkspaceId, runtimes: &[Arc<WorktreeRuntime>]) -> String {
    let places: Vec<String> = runtimes
        .iter()
        .map(|runtime| format!("{} ({})", runtime.registered.display_name(), runtime.registered.super_root.display()))
        .collect();
    format!(
        "{workspace} is claimed by {} worktrees on this machine: {}. Continuity carries one \
         worktree with one; rename one of them (or keep one closed) so each name is unique",
        runtimes.len(),
        places.join(", ")
    )
}

/// The peer configuration a worktree commits.
#[derive(Debug, Default, serde::Deserialize)]
#[serde(deny_unknown_fields)]
struct PeerConfig {
    #[serde(default)]
    continuity: Policy,
}

/// What this workspace declares about its continuity set.
///
/// Read from the workspace's committed `.wsb/peer.json`, which is where facts
/// about a workspace belong. A workspace that declares nothing gets the
/// default: tracked files plus untracked non-ignored ones, which is what git
/// already calls work in progress. A declaration that does not read is
/// refused, naming the file: carrying files under a policy nobody wrote is
/// the one thing continuity must not do.
pub fn continuity_policy(root: &Path) -> Result<Policy, String> {
    let path = root.join(wsb::peer::POLICY_FILE);
    let text = match std::fs::read_to_string(&path) {
        Ok(text) => text,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Policy::default()),
        Err(error) => return Err(format!("cannot read {}: {error}", path.display())),
    };
    serde_json::from_str::<PeerConfig>(&text)
        .map(|config| config.continuity)
        .map_err(|error| format!("{} is not a valid peer configuration: {error}", path.display()))
}

/// What this machine is called until somebody renames it.
fn default_machine_name() -> String {
    if let Ok(name) = std::env::var("WSB_MACHINE_NAME") {
        if !name.trim().is_empty() {
            return name.trim().to_string();
        }
    }
    hostname().unwrap_or_else(|| "this machine".to_string())
}

fn hostname() -> Option<String> {
    #[cfg(unix)]
    {
        let output = wsb::git_backend::command_without_repo_scope("hostname")
            .output()
            .ok()?;
        let name = String::from_utf8(output.stdout).ok()?;
        let name = name.trim().split('.').next()?.to_string();
        (!name.is_empty()).then_some(name)
    }
    #[cfg(windows)]
    {
        std::env::var("COMPUTERNAME").ok().filter(|n| !n.is_empty())
    }
}

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