wsb-daemon 0.258.6441

//! Reconciliation: desired state + git state + filesystem state + peer state
//! → the next safe action.
//!
//! Not a queue of imperative copies. Every input — a file changed, a branch
//! changed, a peer connected, an operator paused — is a reason to recompute
//! what should be true, and then to do the one thing that moves reality
//! towards it. That is what makes the special cases disappear: a peer that
//! disconnects mid-transfer and a peer that was never up take the same path,
//! because both are just "not connected" the next time round.
//!
//! # The order of the checks is the safety argument
//!
//! ```text
//! desired?          not enabled → idle/paused, do nothing
//! peer connected?   no          → blocked/peer_offline
//! git compatible?   no          → blocked/<reason>
//! policy agreed?    no          → blocked/policy_mismatch
//! scan both sides   →           converge against the baseline
//! conflicts?        yes         → conflict, and nothing moves for those paths
//! apply             →           re-validate the epoch, then land it
//! verify            →           rescan, then and only then say SYNCED
//! ```

use std::collections::BTreeSet;
use std::path::Path;
use std::sync::Arc;

use wsb::peer::compatibility::{eligibility, RepositoryFacts};
use wsb::peer::converge::{self, Action};
use wsb::peer::model::{
    Blocked, Continuity, Desired, Observed, PeerId, RepositoryId, RepositoryStatus, Service,
    WorkspaceId,
};
use wsb::peer::protocol::{
    baseline_digest, RefusalCode, RepositoryReport, Request, Response, Write,
};
use wsb::peer::scan::{self, Policy};
use wsb::peer::transport::Session;

use super::service::{now, PeerService};
use super::store::MachineStore;

/// One workspace's repositories, as this machine sees them.
pub struct LocalWorkspace {
    pub id: WorkspaceId,
    /// Every repository in the workspace, in fleet order (parent before
    /// child), with the git directory each one actually uses.
    pub repositories: Vec<LocalRepository>,
    pub policy: Policy,
}

pub struct LocalRepository {
    pub id: RepositoryId,
    pub workdir: std::path::PathBuf,
    /// A submodule's git directory is inside the superproject, and reading
    /// state markers beside its `.git` FILE would find none, for ever.
    pub git_dir: std::path::PathBuf,
}

impl LocalWorkspace {
    /// Discover a workspace's repository tree through the fleet's own model,
    /// so continuity and every git command agree about what a repository is.
    pub fn discover(
        id: WorkspaceId,
        root: &Path,
        policy: Policy,
    ) -> Result<Self, String> {
        let tree = wsb::fleet::git::SubmoduleTree::discover_allow_empty(root)
            .map_err(|e| format!("cannot read the repository tree at {}: {e}", root.display()))?;

        let mut repositories = vec![LocalRepository {
            id: RepositoryId::root(),
            workdir: root.to_path_buf(),
            git_dir: root.join(".git"),
        }];
        for node in tree.top_down() {
            if !node.initialized {
                // An uninitialised submodule has no working tree to
                // synchronise. It is reported as missing by the eligibility
                // check rather than skipped silently.
                continue;
            }
            let Ok(repository) = RepositoryId::parse(&node.display_path) else {
                continue;
            };
            repositories.push(LocalRepository {
                id: repository,
                git_dir: resolve_git_dir(&node.abs_path),
                workdir: node.abs_path.clone(),
            });
        }
        Ok(Self {
            id,
            repositories,
            policy,
        })
    }

    pub fn repository(&self, id: &RepositoryId) -> Option<&LocalRepository> {
        self.repositories.iter().find(|repo| &repo.id == id)
    }

    /// What this machine reports about the named repositories.
    pub fn report(&self, wanted: &[RepositoryId]) -> Vec<RepositoryReport> {
        wanted
            .iter()
            .map(|id| match self.repository(id) {
                Some(repo) => {
                    let facts = RepositoryFacts::read(id.clone(), &repo.workdir, &repo.git_dir)
                        .unwrap_or_else(|_| RepositoryFacts::absent(id.clone()));
                    let entries = if facts.present {
                        scan::read(&repo.workdir, &self.policy).unwrap_or_default()
                    } else {
                        Default::default()
                    };
                    RepositoryReport {
                        facts,
                        policy: (&self.policy).into(),
                        entries,
                    }
                }
                None => RepositoryReport {
                    facts: RepositoryFacts::absent(id.clone()),
                    policy: (&self.policy).into(),
                    entries: Default::default(),
                },
            })
            .collect()
    }
}

/// A submodule's real git directory: the `.git` FILE points at it.
fn resolve_git_dir(workdir: &Path) -> std::path::PathBuf {
    let marker = workdir.join(".git");
    let Ok(text) = std::fs::read_to_string(&marker) else {
        // A directory rather than a file: an ordinary repository.
        return marker;
    };
    let Some(rest) = text.trim().strip_prefix("gitdir:") else {
        return marker;
    };
    let target = std::path::PathBuf::from(rest.trim());
    if target.is_absolute() {
        target
    } else {
        workdir.join(target)
    }
}

/// What one reconciliation pass did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Outcome {
    pub statuses: Vec<RepositoryStatus>,
    /// What each repository intended to move, before it moved. Empty for the
    /// repositories that never got as far as a plan — a blocked repository
    /// has no plan, which is different from a plan that is empty.
    pub plans: Vec<(RepositoryId, Vec<crate::machine::PlannedMove>)>,
}

/// Reconcile one workspace against one peer, over an open session.
///
/// Everything is decided per repository, so a submodule on the wrong branch
/// blocks itself and nothing else.
pub fn reconcile(
    service: &PeerService,
    store: &Arc<MachineStore>,
    session: &mut Session,
    peer: &PeerId,
    local: &LocalWorkspace,
    staging_root: &Path,
) -> Result<Outcome, String> {
    let desired = store.desired(peer, &local.id)?;
    let continuity = Continuity::of(
        store.service(peer, Service::Files)?,
        desired,
        store.mode(peer, &local.id)?,
    );

    // Off is off: no report is asked for and none is answered, so two machines
    // whose file sync is switched off are not scanning each other's working
    // trees every few seconds. Everything else COMPARES — a paused or manual
    // relationship is one somebody wants to look at, and looking is the whole
    // of what it does.
    //
    // Nothing here touches the other services. A peer whose files are off
    // still answers for a secret, which is the entire reason these are
    // separate switches.
    if !continuity.compares() {
        return Ok(Outcome {
            statuses: local
                .repositories
                .iter()
                .map(|repo| RepositoryStatus {
                    repository: repo.id.clone(),
                    desired,
                    continuity,
                    observed: Observed::Idle,
                    blocked: None,
                    pending: 0,
                    incoming: 0,
                    outgoing: 0,
                    conflicts: 0,
                    undecided: 0,
                    branch: None,
                    head: None,
                })
                .collect(),
            plans: Vec::new(),
        });
    }

    let wanted: Vec<RepositoryId> = local.repositories.iter().map(|r| r.id.clone()).collect();

    // What this machine believed had been agreed, AT THE MOMENT it asked the
    // peer what it holds.
    //
    // The peer's answer is a snapshot, and a baseline is not still while a
    // pass runs: the other machine's acknowledgement advances it on this side
    // from a different thread, mid-pass. Planning from a report taken before
    // that and a baseline read after it mixes two readings that were never
    // true together — and the difference between them reads exactly like the
    // peer having deleted something.
    //
    // It did, once, and this is what it cost: the laptop fetched a new file
    // and both baselines advanced to record it; six milliseconds later the
    // desktop planned with a report taken BEFORE that fetch and a baseline
    // read after it, concluded "the peer no longer has what we agreed on", and
    // deleted the file from the machine that had just created it.
    let agreed_when_asked: std::collections::BTreeMap<RepositoryId, String> = wanted
        .iter()
        .map(|repository| {
            let baseline = store.baseline(peer, &local.id, repository)?;
            Ok((repository.clone(), baseline_digest(&baseline)))
        })
        .collect::<Result<_, String>>()?;

    let response: Response = session.exchange(&Request::Report {
        workspace: local.id.clone(),
        repositories: wanted.clone(),
    })?;
    let (present, theirs) = match response {
        Response::Report {
            workspace_present,
            repositories,
        } => (workspace_present.unwrap_or(true), repositories),
        Response::Refused { code, detail } => {
            return Err(format!("the peer refused a report ({}): {detail}", code.as_str()))
        }
        other => return Err(format!("the peer answered a report with {other:?}")),
    };

    if !present {
        return Ok(Outcome {
            statuses: wanted
                .into_iter()
                .map(|repository| RepositoryStatus {
                    repository,
                    desired,
                    continuity,
                    observed: Observed::Blocked,
                    blocked: Some(Blocked::WorkspaceMissing),
                    pending: 0,
                    incoming: 0,
                    outgoing: 0,
                    conflicts: 0,
                    undecided: 0,
                    branch: None,
                    head: None,
                })
                .collect(),
            plans: Vec::new(),
        });
    }

    let mine = local.report(&wanted);
    let mut statuses = Vec::new();
    let mut plans = Vec::new();
    for (here, there) in mine.iter().zip(theirs.iter()) {
        let (status, planned) = reconcile_one(
            service,
            store,
            session,
            peer,
            local,
            here,
            there,
            agreed_when_asked.get(&here.facts.repository),
            staging_root,
            desired,
            continuity,
        )?;
        plans.push((status.repository.clone(), planned));
        statuses.push(status);
    }
    Ok(Outcome { statuses, plans })
}

#[allow(clippy::too_many_arguments)]
fn reconcile_one(
    _service: &PeerService,
    store: &Arc<MachineStore>,
    session: &mut Session,
    peer: &PeerId,
    local: &LocalWorkspace,
    here: &RepositoryReport,
    there: &RepositoryReport,
    // The digest of what had been agreed when `there` was asked for.
    agreed_when_asked: Option<&String>,
    staging_root: &Path,
    desired: Desired,
    continuity: Continuity,
) -> Result<(RepositoryStatus, Vec<crate::machine::PlannedMove>), String> {
    let repository = here.facts.repository.clone();
    let mut status = RepositoryStatus {
        repository: repository.clone(),
        desired,
        continuity,
        observed: Observed::Checking,
        blocked: None,
        pending: 0,
        incoming: 0,
        outgoing: 0,
        conflicts: 0,
        undecided: 0,
        branch: here.facts.branch.clone(),
        head: here.facts.head.clone(),
    };

    // Git first. Nothing about files matters until both sides stand on the
    // same base.
    if let Err(reason) = eligibility(&here.facts, &there.facts) {
        status.observed = Observed::Blocked;
        status.blocked = Some(reason);
        return Ok((status, Vec::new()));
    }

    // TWO epochs, and they are not interchangeable.
    //
    // An epoch identifies one moment of one repository ON ONE MACHINE: it
    // includes the index, which is that machine's own file. `mine` is what
    // this side re-validates before writing anything here; `theirs` is what
    // the peer re-validates before writing anything there, and it is what
    // travels in a request. Sending this machine's epoch for the peer to check
    // against its own repository compares two things that were never the same,
    // and every transfer is refused as "the repository moved".
    let mine = here.facts.epoch();
    let theirs = there.facts.epoch();

    // Then policy: two machines that disagree about what may move must not
    // move anything.
    if let Err(detail) = policy_of(here).agrees_with(&policy_of(there)) {
        status.observed = Observed::Blocked;
        status.blocked = Some(Blocked::PolicyMismatch { detail });
        return Ok((status, Vec::new()));
    }

    // The first baseline is the commit both sides are standing on.
    //
    // Without it, every committed file reads as "created independently on
    // both machines" — a conflict for the whole repository on the first pass.
    // The two machines are on the same branch at the same commit, which is
    // exactly a state they have agreed on, and both compute the same seed from
    // it without exchanging anything.
    let seeded_already = store.is_seeded(peer, &local.id, &repository)?;
    if !seeded_already {
        let repo = local
            .repository(&repository)
            .ok_or_else(|| format!("{repository} vanished while seeding"))?;
        let seeded = wsb::peer::seed::from_head(&repo.workdir)?;
        store.put_baseline(peer, &local.id, &repository, &seeded)?;
        store.mark_seeded(peer, &local.id, &repository, now())?;
        tracing::debug!(
            %peer,
            repository = %repository,
            paths = seeded.len(),
            "seeded a baseline from the commit both machines share"
        );
    }

    let baseline = store.baseline(peer, &local.id, &repository)?;

    // The peer's answer and this agreement have to be readings of one moment.
    //
    // If the baseline moved between asking the peer what it holds and reading
    // it here, the two describe different instants — and every path where they
    // disagree looks like a change the peer made. A path this side holds, that
    // the report predates, and that the baseline has since gained, reads as
    // "the peer deleted it": a deletion decided entirely from the gap between
    // two clocks.
    //
    // Nothing is done about it beyond not acting: the next pass asks again,
    // both readings are current, and the plan is about files rather than about
    // timing. `Checking` is the honest state for "this machine does not yet
    // know", which is what it is.
    //
    // A pass that SEEDED the baseline is excused, because it is what moved it:
    // the seed is this side's own record of the commit both machines already
    // share, and it can only add paths both of them have.
    if !readings_are_of_one_moment(agreed_when_asked, &baseline, !seeded_already) {
        tracing::debug!(
            %peer,
            repository = %repository,
            "what had been agreed moved while the peer was being asked; \
             planning again next pass"
        );
        status.observed = Observed::Checking;
        return Ok((status, Vec::new()));
    }

    let mut convergence = converge::plan(&here.entries, &there.entries, &baseline);

    // The three inputs of every decision, at the moment they were read. When
    // two machines disagree about what should have crossed, this is the line
    // that says which of them saw what — and no amount of "synced" or
    // "checking" afterwards can answer that.
    tracing::debug!(
        %peer,
        repository = %repository,
        here = here.entries.len(),
        there = there.entries.len(),
        baseline = baseline.len(),
        actions = convergence.actions.len(),
        // What it decided, path by path. A count says two machines disagree;
        // only the paths say about what.
        moves = %convergence
            .actions
            .iter()
            .filter(|action| !matches!(action, Action::Agree { .. }))
            .map(|action| format!("{}:{}", action.kind(), action.path()))
            .collect::<Vec<_>>()
            .join(" "),
        "compared"
    );

    // A difference a person has settled becomes an ordinary change.
    //
    // The decision is turned into a Send or a Receive here rather than into a
    // baseline edit, so it travels through the same transfer, the same epoch
    // check and the same acknowledgement as everything else — and both
    // machines' baselines move together at the end of it.
    //
    // This settles ANY difference, not only a conflict. In manual mode the two
    // are the same act from the operator's side: a path differs, and they say
    // which version wins. A decision that agrees with what convergence had
    // already worked out costs nothing and produces the same action; one that
    // disagrees — "no, keep MINE" over a file the peer changed — is the whole
    // reason manual mode exists, and it has to be able to reverse the
    // direction convergence would have chosen.
    let decisions = store.resolutions(peer, &local.id, &repository)?;
    if !decisions.is_empty() {
        for action in convergence.actions.iter_mut() {
            if matches!(action, Action::Agree { .. }) {
                continue;
            }
            let path = action.path().to_string();
            let Some(keep) = decisions.get(path.as_str()) else {
                continue;
            };
            // What each side holds right now, from the two scans — the same
            // source the conflict's own versions came from, so a decision
            // about a plain difference and one about a conflict are carried
            // out identically.
            let mine = here.entries.get(&path).cloned();
            let theirs = there.entries.get(&path).cloned();
            *action = match (keep.as_str(), mine, theirs) {
                ("here", Some(entry), _) => Action::Send { path, entry },
                // Keeping this machine's side of a path this machine deleted
                // means the deletion is the decision.
                ("here", None, _) => Action::SendDeletion { path },
                (_, _, Some(entry)) => Action::Receive { path, entry },
                (_, _, None) => Action::ReceiveDeletion { path },
            };
        }
    }

    // A difference somebody chose to leave alone, as it stood at the time.
    //
    // Matched on the two hashes rather than the path: dismissing is "this
    // file, as these two machines have it right now, is not interesting", and
    // a dismissal that outlived the versions it was about would hide the next
    // edit as well. Either side changing brings it straight back.
    let dismissals = store.dismissals(peer, &local.id, &repository)?;
    let still_dismissed = |path: &str| -> bool {
        let Some((here_hash, there_hash)) = dismissals.get(path) else {
            return false;
        };
        let mine = here.entries.get(path).map(|entry| entry.hash.clone());
        let theirs = there.entries.get(path).map(|entry| entry.hash.clone());
        mine.as_deref() == here_hash.as_deref() && theirs.as_deref() == there_hash.as_deref()
    };

    // What this pass intends to move, named path by path, before any of it
    // moves. A count says something differs; only the paths say what, and a
    // person deciding whether to let a working tree cross machines is
    // deciding about files.
    let planned: Vec<crate::machine::PlannedMove> = convergence
        .actions
        .iter()
        .filter_map(|action| {
            let (direction, size) = match action {
                Action::Agree { .. } => return None,
                Action::Send { entry, .. } => ("send", Some(entry.size)),
                Action::Receive { entry, .. } => ("receive", Some(entry.size)),
                Action::SendDeletion { .. } => ("send_deletion", None),
                Action::ReceiveDeletion { .. } => ("receive_deletion", None),
                Action::Conflict { .. } => ("conflict", None),
            };
            let path = action.path().to_string();
            let dismissed = still_dismissed(&path);
            let decision = match decisions.get(path.as_str()).map(String::as_str) {
                Some("here") => Some("here"),
                Some("there") => Some("there"),
                _ => None,
            };
            let here_hash = here.entries.get(&path).map(|entry| entry.hash.clone());
            let there_hash = there.entries.get(&path).map(|entry| entry.hash.clone());
            Some(crate::machine::PlannedMove {
                path,
                direction,
                size,
                decision,
                dismissed,
                here_hash,
                there_hash,
            })
        })
        .collect();

    // Conflicts are recorded before anything else happens, so a crash between
    // here and the apply leaves them known rather than lost. A conflict that
    // has been decided is no longer one — the loop above has already turned it
    // into a transfer.
    let mut conflicts = 0usize;
    for action in convergence.conflicts() {
        if let Action::Conflict {
            path,
            here: mine,
            there: theirs,
            base,
        } = action
        {
            conflicts += 1;
            tracing::debug!(%repository, %path, "conflict");
            store.record_conflict(
                peer,
                &local.id,
                &repository,
                path,
                mine.as_ref(),
                theirs.as_ref(),
                base.as_ref(),
                now(),
            )?;
        }
    }
    // A path that used to conflict and now agrees is no longer a conflict.
    for stale in store.conflicts(peer, &local.id)? {
        if stale.repository != repository.as_str() {
            continue;
        }
        let still = convergence
            .conflicts()
            .any(|action| action.path() == stale.path);
        if !still {
            store.clear_conflict(peer, &local.id, &repository, &stale.path)?;
        }
    }

    // How many differences nobody has spoken about. Counted before anything is
    // filtered, because this is what the surfaces show as the size of the
    // decision waiting to be made.
    //
    // A dismissed one has been spoken about: the answer was "leave it". It is
    // still a difference and still listed, and it does not count towards the
    // number asking for attention — otherwise dismissing everything would
    // leave the badge exactly where it was.
    let is_undecided = |action: &Action| {
        action.moves_data()
            && !decisions.contains_key(action.path())
            && !still_dismissed(action.path())
    };
    let undecided = convergence.actions.iter().filter(|a| is_undecided(a)).count();
    // The ones this machine will not move by itself whatever its mode: they
    // would land on the PEER, and that is the peer's decision to make.
    let undecided_outgoing = convergence
        .actions
        .iter()
        .filter(|a| is_undecided(a) && a.leaves_here())
        .count();
    // Captured BEFORE the plan is filtered, because what this machine will not
    // move by itself is still a difference and still has a direction.
    let incoming_paths: BTreeSet<String> = convergence
        .actions
        .iter()
        .filter(|a| a.moves_data() && a.arrives_here())
        .map(|a| a.path().to_string())
        .collect();
    let outgoing_paths: BTreeSet<String> = convergence
        .actions
        .iter()
        .filter(|a| a.moves_data() && a.leaves_here())
        .map(|a| a.path().to_string())
        .collect();
    let differences = convergence.pending();

    // Paused stops here, with everything OBSERVED and nothing done. The
    // baseline is untouched, so resuming carries exactly what accumulated
    // while it was paused rather than starting from a new agreement nobody
    // made — and a decision taken before the pause waits for the resume
    // rather than acting during it.
    //
    // Read again here, not only at the start of the pass: scanning both trees
    // takes seconds, and a pause asked for meanwhile is a pause — a file
    // written after it must not cross because the pass began before it.
    let now = Continuity::of(
        store.service(peer, Service::Files)?,
        store.desired(peer, &local.id)?,
        store.mode(peer, &local.id)?,
    );
    if !continuity.carries_decided() || !now.carries_decided() {
        status.observed = Observed::Paused;
        status.pending = differences;
        status.undecided = undecided;
        status.conflicts = conflicts;
        return Ok((status, planned));
    }

    // What may cross without anybody having said so — and in which direction.
    //
    // `auto` is a statement this machine makes about its OWN working tree:
    // what arrives here is applied without asking. It says nothing about the
    // peer's tree. Whether this machine's changes are applied there is the
    // peer's decision, made by the peer's own pass with the peer's own mode —
    // so the two can differ and changes travel automatically one way only.
    //
    // This filter therefore runs in every mode, not only manual. Set to
    // `auto`, this machine used to push undecided changes onto a peer that had
    // deliberately stayed manual, which is the one thing manual exists to
    // prevent.
    //
    // The undecided differences are dropped from the action list rather than
    // turned into agreement — dropping them leaves the baseline exactly where
    // it was for those paths, so the difference is still there to be decided
    // next pass. Turning them into agreement would record "both machines have
    // settled this" about a file nobody looked at, and the difference would
    // vanish without either version moving.
    //
    // Agreement stays: two machines that already hold the same bytes have
    // agreed whatever the mode, and their baseline may advance to say so.
    convergence.actions.retain(|action| {
        converge::crosses(action, continuity, decisions.contains_key(action.path()))
    });

    let digest = baseline_digest(&baseline);
    let mut applied: BTreeSet<String> = BTreeSet::new();

    // What this side must take.
    let wanted: Vec<String> = convergence
        .actions
        .iter()
        .filter_map(|action| match action {
            Action::Receive { path, .. } => Some(path.clone()),
            _ => None,
        })
        .collect();
    let deletions_here: Vec<String> = convergence
        .actions
        .iter()
        .filter_map(|action| match action {
            Action::ReceiveDeletion { path } => Some(path.clone()),
            _ => None,
        })
        .collect();

    if !wanted.is_empty() || !deletions_here.is_empty() {
        let repo = local
            .repository(&repository)
            .ok_or_else(|| format!("{repository} vanished while reconciling"))?;
        // Before DELETING anything here, both sides confirm they still agree
        // about what they last agreed on.
        //
        // A deletion is the one action decided from the baseline alone. "The
        // peer no longer has this" is read off this machine's own copy, and if
        // that copy has drifted from the peer's, the conclusion can be about a
        // file the peer never had — which is how a file was deleted from the
        // machine that created it, while the peer sat there never having seen
        // it. A pass whose plan was only deletions sent no message at all, so
        // there was no moment at which the peer could have said so.
        //
        // Asked ONLY for deletions, and that limit is the point. Everything
        // else a pass does is additive — fetching what the peer has, agreeing
        // on what both hold — and those are what bring two drifted baselines
        // back together. Gating the whole pass on the confirmation stopped the
        // agreements as well, and the two sides then re-seeded past each other
        // for ever without a single file crossing.
        if !deletions_here.is_empty() {
            match session.exchange::<_, Response>(&Request::Confirm {
                workspace: local.id.clone(),
                repository: repository.clone(),
                baseline_digest: digest.clone(),
            })? {
                Response::Acknowledged => {}
                Response::Refused { code, detail } => {
                    if code == RefusalCode::BaselineMismatch {
                        reseed(store, peer, local, &repository)?;
                    }
                    return Ok((refused_status(status, code, detail), planned));
                }
                other => {
                    return Err(format!(
                        "the peer answered a baseline confirmation with {other:?}"
                    ))
                }
            }
        }

        let writes = if wanted.is_empty() {
            Vec::new()
        } else {
            match session.exchange::<_, Response>(&Request::Fetch {
                workspace: local.id.clone(),
                repository: repository.clone(),
                epoch: theirs.clone(),
                paths: wanted.clone(),
            })? {
                Response::Fetched { writes } => writes,
                Response::Refused { code, detail } => {
                    // An epoch that moved is ordinary: the peer is being used.
                    // The next pass sees the new state.
                    if code == RefusalCode::BaselineMismatch {
                    reseed(store, peer, local, &repository)?;
                }
                return Ok((refused_status(status, code, detail), planned));
                }
                other => return Err(format!("the peer answered a fetch with {other:?}")),
            }
        };

        match super::transfer::apply(
            &repo.workdir,
            &repository,
            &repo.git_dir,
            &mine,
            &writes,
            &deletions_here,
            // What this pass read here, which the plan was made from.
            &here.entries,
            staging_root,
        ) {
            Ok(landed) => applied.extend(landed),
            Err(super::transfer::ApplyRefusal::EpochMoved) => {
                status.observed = Observed::Checking;
                return Ok((status, planned));
            }
            Err(other) => return Err(format!("{repository}: {other:?}")),
        }
    }

    // What the peer must take.
    let sending: Vec<Write> = convergence
        .actions
        .iter()
        .filter_map(|action| match action {
            Action::Send { path, entry } => Some((path.clone(), entry.clone())),
            _ => None,
        })
        .map(|(path, entry)| -> Result<Write, String> {
            let repo = local
                .repository(&repository)
                .ok_or_else(|| format!("{repository} vanished while reading {path}"))?;
            let absolute = scan::local_path(&repo.workdir, &path)?;
            let content = match entry.kind {
                converge::EntryKind::Symlink => {
                    let target = std::fs::read_link(&absolute)
                        .map_err(|e| format!("cannot read the link {path}: {e}"))?;
                    wsb::peer::identity::base64_encode(scan::to_slash(&target).as_bytes())
                }
                converge::EntryKind::File => {
                    let bytes = std::fs::read(&absolute)
                        .map_err(|e| format!("cannot read {path}: {e}"))?;
                    wsb::peer::identity::base64_encode(&bytes)
                }
            };
            Ok(Write {
                path,
                entry,
                content,
            })
        })
        .collect::<Result<Vec<_>, _>>()?;
    let deletions_there: Vec<String> = convergence
        .actions
        .iter()
        .filter_map(|action| match action {
            Action::SendDeletion { path } => Some(path.clone()),
            _ => None,
        })
        .collect();

    if !sending.is_empty() || !deletions_there.is_empty() {
        // What the peer's own report held at each of those paths.
        let found = sending
            .iter()
            .map(|write| &write.path)
            .chain(deletions_there.iter())
            .filter_map(|path| there.entries.get(path).map(|entry| (path.clone(), entry.clone())))
            .collect();
        match session.exchange::<_, Response>(&Request::Apply {
            workspace: local.id.clone(),
            repository: repository.clone(),
            epoch: theirs.clone(),
            writes: sending,
            deletions: deletions_there,
            found,
            baseline_digest: digest.clone(),
        })? {
            Response::Applied { applied: landed } => applied.extend(landed),
            Response::Refused { code, detail } => {
                if code == RefusalCode::BaselineMismatch {
                    reseed(store, peer, local, &repository)?;
                }
                return Ok((refused_status(status, code, detail), planned));
            }
            other => return Err(format!("the peer answered an apply with {other:?}")),
        }
    }

    // The peer's baseline moves FIRST, and this side's only if it did.
    //
    // A baseline is what both machines agree on, so advancing it before the
    // other side has agreed makes it a private opinion — and two machines with
    // different private baselines refuse each other's transfers for ever,
    // because every plan is made against a state the other does not share.
    if !applied.is_empty() {
        let acknowledged: Vec<String> = applied.iter().cloned().collect();
        match session.exchange::<_, Response>(&Request::Acknowledge {
            workspace: local.id.clone(),
            repository: repository.clone(),
            epoch: theirs,
            paths: acknowledged,
            baseline_digest: digest,
        })? {
            Response::Acknowledged => {}
            Response::Refused { code, detail } => {
                // Nothing is advanced here. The next pass plans against the
                // same baseline the peer still holds, which is the state both
                // sides can still agree about.
                if code == RefusalCode::BaselineMismatch {
                    reseed(store, peer, local, &repository)?;
                }
                return Ok((refused_status(status, code, detail), planned));
            }
            other => return Err(format!("the peer answered an acknowledge with {other:?}")),
        }
    }

    // Agreement advances the baseline without an acknowledgement, because
    // nothing travelled for it to acknowledge. Both machines compute the same
    // agreement from the same two scans, so both baselines move the same way —
    // which is exactly what keeps their digests equal.
    let mut settled = applied.clone();
    for action in &convergence.actions {
        if matches!(action, Action::Agree { .. }) {
            settled.insert(action.path().to_string());
        }
    }
    let next = converge::advance(&baseline, &convergence, &settled);
    store.put_baseline(peer, &local.id, &repository, &next)?;

    // A decision that has now been carried is spent, and the conflict it
    // settled is over.
    for path in applied.iter() {
        if decisions.contains_key(path.as_str()) {
            store.clear_resolution(peer, &local.id, &repository, path)?;
            store.clear_conflict(peer, &local.id, &repository, path)?;
        }
    }

    status.pending = differences.saturating_sub(applied.len());
    // Which way the differences that are STILL there would go.
    //
    // Counted from the full plan rather than from what was allowed to cross:
    // an outgoing difference this machine will not push by itself is exactly
    // the thing a person needs to see, and counting after the filter reported
    // zero of them in `auto` — the mode in which they are the only thing left
    // to do.
    status.incoming = incoming_paths.difference(&applied).count();
    status.outgoing = outgoing_paths.difference(&applied).count();
    // What is still waiting for a person ON THIS MACHINE.
    //
    // In `manual` that is every undecided difference: this person can `take`
    // the peer's version or `keep` their own, so both directions are theirs to
    // settle.
    //
    // In `auto` the incoming ones are not waiting — they were just applied.
    // The OUTGOING ones still are: this machine will not push them by itself,
    // and `keep` is how a person says to. Reporting zero there hid every
    // outgoing difference behind a workspace that called itself idle.
    status.undecided = if continuity.takes_undecided() {
        undecided_outgoing
    } else {
        undecided
    };
    status.conflicts = conflicts;
    status.observed = if conflicts > 0 {
        Observed::Conflict
    } else if status.undecided > 0 {
        // Differences are known and listed, and the machine is doing exactly
        // what it was asked to: waiting. Not `syncing` — nothing is moving —
        // and not an error.
        Observed::Undecided
    } else if status.pending > 0 {
        Observed::Syncing
    } else if convergence.is_converged() && applied.is_empty() {
        // Nothing to do and nothing done: both trees stand on the same
        // baseline, verified by this pass's scan of both sides.
        Observed::Synced
    } else {
        // Something landed. The next pass verifies it rather than this one
        // claiming it: "synced" is an observation, and the observation has not
        // been made yet.
        Observed::Ready
    };
    Ok((status, planned))
}

fn policy_of(report: &RepositoryReport) -> Policy {
    Policy {
        include_ignored: report.policy.include_ignored.clone(),
        exclude: report.policy.exclude.clone(),
    }
}

/// Put this repository's baseline back to something both machines compute the
/// same way: the commit they are both standing on.
///
/// Reached when the peer refuses a request because its baseline is not the one
/// this side planned against. Two baselines that have drifted apart can never
/// be reconciled by more transfers — every one of them is refused for the same
/// reason — so the relationship would stay stuck for ever. The commit is the
/// one state both machines already agree on without exchanging anything, and
/// re-seeding from it costs nothing but a re-comparison: identical files agree,
/// and anything genuinely different becomes a conflict, which is the honest
/// answer rather than a silent overwrite.
/// Whether the peer's answer and this side's record of agreement describe the
/// same moment.
///
/// A pass reads three things: what is here, what the peer says is there, and
/// what the two last agreed on. The first two are a snapshot taken when the
/// peer was asked. The third is read afterwards, and it does not hold still —
/// the peer's own pass acknowledges transfers on a different thread of this
/// process, and that advances it.
///
/// Where those two instants differ, every path they disagree about looks like
/// a change the peer made. The one that costs something is a path this side
/// holds which the report predates and the baseline has since gained: it reads
/// as "the peer deleted what we agreed on", and the answer to that is to
/// delete it here. A file was removed from the machine that created it exactly
/// that way, six milliseconds after the other machine had successfully
/// fetched it.
///
/// A pass that SEEDED the baseline is excused, because it is the thing that
/// moved it. A seed is this side's own record of the commit both machines are
/// already standing on, so it can only add paths both of them have — and those
/// converge as agreement, never as a deletion.
fn readings_are_of_one_moment(
    agreed_when_asked: Option<&String>,
    baseline_now: &wsb::peer::converge::Baseline,
    this_pass_seeded: bool,
) -> bool {
    if this_pass_seeded {
        return true;
    }
    match agreed_when_asked {
        Some(when_asked) => &baseline_digest(baseline_now) == when_asked,
        // No snapshot means nobody can say, and a pass that cannot establish
        // this must not be the one that deletes something.
        None => false,
    }
}

fn reseed(
    store: &Arc<MachineStore>,
    peer: &PeerId,
    local: &LocalWorkspace,
    repository: &RepositoryId,
) -> Result<(), String> {
    let repo = local
        .repository(repository)
        .ok_or_else(|| format!("{repository} vanished while re-seeding"))?;
    let seeded = wsb::peer::seed::from_head(&repo.workdir)?;
    store.put_baseline(peer, &local.id, repository, &seeded)?;
    store.mark_seeded(peer, &local.id, repository, now())?;
    tracing::warn!(
        %peer,
        repository = %repository,
        paths = seeded.len(),
        "the two machines no longer shared a baseline; re-seeded from the commit they share"
    );
    Ok(())
}

/// A refusal that is a STATE rather than a failure: the peer is fine, and
/// something about the moment was not.
fn refused_status(mut status: RepositoryStatus, code: RefusalCode, detail: String) -> RepositoryStatus {
    status.observed = match code {
        // Both mean "try again against what is true now".
        RefusalCode::EpochMoved | RefusalCode::BaselineMismatch => Observed::Checking,
        RefusalCode::NotEnabled => Observed::Idle,
        // The peer has switched this service off on its side. Not an error
        // here and not something this machine can fix: it is a decision
        // somebody made over there, and the honest local state is idle.
        RefusalCode::ServiceOff => Observed::Idle,
        RefusalCode::UnknownWorkspace => {
            status.blocked = Some(Blocked::WorkspaceMissing);
            Observed::Blocked
        }
        RefusalCode::UnknownRepository => {
            status.blocked = Some(Blocked::SubmoduleMissing {
                here: true,
                there: false,
            });
            Observed::Blocked
        }
        RefusalCode::PolicyMismatch => {
            status.blocked = Some(Blocked::PolicyMismatch { detail });
            Observed::Blocked
        }
        // A forbidden path, a malformed message or an internal failure are
        // this software's problem, not a state to sit in quietly. So is a
        // plugin's refusal arriving here: continuity sends no plugin request,
        // and a peer answering one with a plugin's words answered something
        // else.
        RefusalCode::ForbiddenPath
        | RefusalCode::Malformed
        | RefusalCode::Internal
        | RefusalCode::NoSuchService
        | RefusalCode::PluginAbsent
        | RefusalCode::PluginRefused => Observed::Error,
    };
    status
}

/// How often the coordinator looks at every enabled relationship.
///
/// A watcher nudge is what makes an edit appear on the other machine in a
/// second; this is the floor under it. It exists because a watcher can miss an
/// event, a peer can come back while nothing local is changing, and a
/// relationship that only reconciled on local activity would never notice the
/// other side's work.
pub const SWEEP: std::time::Duration = std::time::Duration::from_secs(3);

/// Run reconciliation for the life of the daemon.
///
/// One thread, working through every enabled relationship in turn. Not one per
/// peer: two relationships that both want the same repository's baseline must
/// not write it at the same instant, and one worker makes that impossible
/// rather than merely unlikely.
pub fn spawn_reconcile_loop(
    machine: std::sync::Arc<crate::machine::MachineState>,
    runtime: tokio::runtime::Handle,
) {
    std::thread::Builder::new()
        .name("wsb-peer-reconcile".to_string())
        .spawn(move || loop {
            let relationships = machine.store.relationships().unwrap_or_default();
            for (peer, workspace, desired) in relationships {
                // Paused is swept too: it observes without moving, so an
                // operator can see what is waiting before they resume.
                // Disabled is the only state that costs nothing.
                if desired == Desired::Disabled {
                    continue;
                }
                // The repository tree is re-read each pass: a submodule
                // initialised since the daemon started changes what continuity
                // covers, and a tree cached for the life of the process would
                // carry the superproject and silently ignore it.
                let Some(local) = futures::executor::block_on({
                    let machine = machine.clone();
                    let workspace = workspace.clone();
                    async move {
                        if let Err(error) = machine.refresh_tree(&workspace).await {
                            tracing::debug!(%workspace, %error, "could not re-read the repository tree");
                        }
                        machine.continuity(&workspace).await
                    }
                })
                .unwrap_or_else(|error| {
                    // Two open worktrees claim it; neither is carried until
                    // one of them is closed or renamed.
                    tracing::warn!(%workspace, %error, "not reconciled");
                    None
                }) else {
                    // A relationship, not live here: no worktree carrying it
                    // has been opened on this machine since the daemon
                    // started. Nothing to reconcile yet.
                    continue;
                };

                let outcome = reconcile_once(&machine, &peer, &local.1, &runtime);
                if let Err(error) = outcome {
                    tracing::debug!(%peer, %workspace, %error, "reconciliation did not complete");
                }
            }
            // Woken by a change, or by the sweep — whichever comes first. The
            // sweep is what makes the watcher optional rather than load-bearing.
            machine.nudge.wait(SWEEP);
        })
        .expect("the reconcile thread starts");
}

/// One pass over one relationship, on a connection this call opens and closes.
///
/// A connection per pass rather than one held open: a peer that went away
/// between passes is discovered by failing to connect, which is the state to
/// report, and a held connection would have to be revalidated anyway.
fn reconcile_once(
    machine: &std::sync::Arc<crate::machine::MachineState>,
    peer: &PeerId,
    local: &LocalWorkspace,
    _runtime: &tokio::runtime::Handle,
) -> Result<Outcome, String> {
    let mut session = machine.peers.connect(peer)?;
    let outcome = reconcile(
        &machine.peers,
        &machine.store,
        &mut session,
        peer,
        local,
        &machine.staging_root(),
    );
    session.shutdown();
    let outcome = outcome?;
    // What each repository came to, every pass. At debug because a machine
    // that is working says this every few seconds; when something is wrong it
    // is the first thing anybody needs.
    for status in &outcome.statuses {
        tracing::debug!(
            %peer,
            repository = %status.repository,
            observed = status.observed.as_str(),
            blocked = status.blocked.as_ref().map(|reason| reason.code()).unwrap_or(""),
            pending = status.pending,
            conflicts = status.conflicts,
            "reconciled"
        );
    }
    machine.record_statuses(peer, &local.id, &outcome.statuses);
    for (repository, planned) in &outcome.plans {
        machine.record_plan(peer, &local.id, repository, planned.clone());
    }
    Ok(outcome)
}

#[cfg(test)]
mod one_moment_tests {
    use super::*;
    use wsb::peer::converge::{Baseline, Entry, EntryKind};

    fn entry(bytes: &[u8]) -> Entry {
        Entry {
            kind: EntryKind::File,
            hash: wsb::peer::converge::hash_bytes(bytes),
            size: bytes.len() as u64,
            executable: false,
        }
    }

    /// The two baselines from the run that lost a file: what the desktop had
    /// agreed when it asked the laptop what it held, and what it had agreed by
    /// the time it read that agreement back.
    fn before_and_after() -> (Baseline, Baseline) {
        let mut before = Baseline::new();
        for (path, bytes) in [
            (".gitignore", &b"target\n"[..]),
            ("README.md", &b"# fixture\n"[..]),
            ("child/child.txt", &b"child\n"[..]),
            ("committed.txt", &b"to be committed\n"[..]),
        ] {
            before.insert(path.to_string(), entry(bytes));
        }
        let mut after = before.clone();
        // The path the laptop had just fetched, which advanced both baselines
        // while this side's pass was already in flight.
        after.insert("after-commit.txt".to_string(), entry(b"carried again\n"));
        (before, after)
    }

    /// TEST11591: a pass whose agreement moved under it does not plan.
    ///
    /// The exact shape of the data loss. The report said the peer held four
    /// paths; by the time the agreement was read it held five, because the
    /// peer had fetched the fifth in between and said so. Planning from those
    /// two readings makes the fifth path look like something the peer deleted,
    /// and the answer to a deletion is to delete it here — on the machine that
    /// had created it seconds earlier.
    #[test]
    fn test11591_an_agreement_that_moved_mid_pass_is_not_planned_from() {
        let (before, after) = before_and_after();
        let asked_with = baseline_digest(&before);

        assert!(
            !readings_are_of_one_moment(Some(&asked_with), &after, false),
            "a pass planned from a report and an agreement taken at different \
             moments, which is what deleted the file"
        );
    }

    /// TEST11592: a pass whose agreement held still plans normally.
    ///
    /// The control, and the reason this cannot simply refuse: every ordinary
    /// pass reaches here, and one that never planned would carry nothing at
    /// all while looking perfectly healthy.
    #[test]
    fn test11592_an_agreement_that_held_still_is_planned_from() {
        let (before, _) = before_and_after();
        let asked_with = baseline_digest(&before);
        assert!(readings_are_of_one_moment(
            Some(&asked_with),
            &before,
            false
        ));
    }

    /// TEST11593: the pass that seeded the agreement is the one that moved it.
    ///
    /// A first pass has nothing agreed when it asks, and seeds from the commit
    /// both machines are standing on before it plans. Refusing that would mean
    /// two paired machines never got past their first pass.
    #[test]
    fn test11593_the_seeding_pass_is_not_refused_for_seeding() {
        let (before, after) = before_and_after();
        let asked_with = baseline_digest(&before);
        assert!(
            readings_are_of_one_moment(Some(&asked_with), &after, true),
            "the pass that seeds an agreement was refused for having seeded it"
        );
    }

    /// TEST11594: a pass that cannot say what was agreed does not plan.
    ///
    /// There is no reading to compare against, so there is no way to know the
    /// two are of one moment. A deletion decided on that basis is decided on
    /// nothing.
    #[test]
    fn test11594_a_pass_with_no_snapshot_does_not_plan() {
        let (_, after) = before_and_after();
        assert!(!readings_are_of_one_moment(None, &after, false));
    }

    /// TEST11595: the digest actually distinguishes these two baselines.
    ///
    /// The guard is a digest comparison, so it is worth nothing if the digest
    /// does not change when the agreement gains a path. Without this the three
    /// tests above would pass against a digest that returned a constant.
    #[test]
    fn test11595_a_gained_path_changes_the_digest() {
        let (before, after) = before_and_after();
        assert_ne!(
            baseline_digest(&before),
            baseline_digest(&after),
            "an agreement that gained a path has the same digest, so nothing \
             above is detecting anything"
        );
    }
}