vivac 0.15.8

//! The store: one directory, several files.
//!
//! ```text
//! .vivac/
//!   events    append-only log, one JSON per line   <- SOURCE OF TRUTH
//!   config    project_id and opaque actor
//!   lane      which lane this folder is (`t594` §2.3), when it is one
//!   index     derived projection of `events`       <- DISPOSABLE, REGENERABLE
//! ```
//!
//! `index` is not SQLite and not a second home for any state `events` does
//! not already hold: deleting it changes no command's output, only how long
//! building a `Tree` takes. `index.rs` owns its format and every rule about
//! when it is trusted, refreshed or thrown away; this module only names
//! where it lives.

use crate::anchor;
use crate::failure::Failure;
use crate::{clock, id};
use serde::{Deserialize, Serialize};
use std::ffi::OsStr;
use std::fs::{self, File, OpenOptions};
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};

pub const DIR: &str = ".vivac";
pub const LOG: &str = "events";
pub const CONFIG: &str = "config";
pub const INDEX: &str = "index";
pub const LOCK: &str = "lock";
/// The lane file's own name (`d595`). `lane::FILE` reexports it, so the
/// literal is written here and nowhere else.
pub const LANE: &str = "lane";

/// How long a writer waits for another one before it gives up (`d598`).
pub(crate) const LOCK_DEADLINE: std::time::Duration = std::time::Duration::from_secs(5);
/// How long it keeps retrying with a bare yield before it starts sleeping a
/// millisecond between tries: a handoff between two writers takes
/// microseconds, and a sleep would round that up to a timer tick.
const LOCK_SPIN: std::time::Duration = std::time::Duration::from_millis(50);

/// Whether this process has made either of the two writes the copy warning
/// (`registry::warn_if_wrote`) cares about: an event appended (`append`,
/// below) or a `.vivac/lane` file written (`lane::write`). The warning
/// hangs off this fact and off nothing else -- not which verb ran, and not
/// whether the verb is merely capable of writing (`t594`). A usage
/// failure that never reaches either write leaves this
/// `false`, and a write through any path -- an ordinary `push`, `setup
/// --join`, `relocate` -- sets it the same way.
static WROTE: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

/// Marks the one fact `WROTE` exists to carry. Called from `append`,
/// below, and from `lane::write`; nowhere else.
pub(crate) fn mark_write() {
    WROTE.store(true, std::sync::atomic::Ordering::Relaxed);
}

/// Whether `mark_write` has run at least once in this process.
pub(crate) fn wrote() -> bool {
    WROTE.load(std::sync::atomic::Ordering::Relaxed)
}

/// Whether the brief has already put the copy notice in front of whoever
/// is reading, somewhere earlier in this same process. `session start`
/// prints it this way before it ever writes anything (`t594`):
/// the stderr echo checks this so that a folder which already saw the block once, on `stdout`, never sees it said again on `stderr`
/// for the very same reason.
static SHOWN: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

/// Marks the one fact `SHOWN` exists to carry. Called from `brief`'s own
/// rendering, and nowhere else.
pub(crate) fn mark_shown() {
    SHOWN.store(true, std::sync::atomic::Ordering::Relaxed);
}

/// Whether `mark_shown` has run at least once in this process.
pub(crate) fn shown() -> bool {
    SHOWN.load(std::sync::atomic::Ordering::Relaxed)
}

/// Whether this process is a resident server (`vivac mcp`) rather than a
/// one-shot command. Its own `stderr` reaches nobody once it is running
/// headless -- never a terminal a person is reading, never the stream an
/// agent parses either -- so the copy warning is not this process's own to
/// print on that stream at all: its seat is the brief instead, recomputed
/// fresh on every `vivac_brief` call for as long as the server lives
/// (`t594`).
static RESIDENT: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false);

/// Declares this process resident. Called once, from `mcp::serve`, before
/// it starts answering calls.
pub(crate) fn mark_resident() {
    RESIDENT.store(true, std::sync::atomic::Ordering::Relaxed);
}

/// Whether `mark_resident` has run in this process.
pub(crate) fn is_resident() -> bool {
    RESIDENT.load(std::sync::atomic::Ordering::Relaxed)
}

/// `t594` §4.9: every `.vivac/` ignores itself. One line, `*`, which a git
/// reads as "everything here, this file included", so no file of the
/// user's is touched and a clone never carries a copy of the log.
pub const GITIGNORE: &str = ".gitignore";

/// Where the global store lives, read from the environment.
///
/// `VIVAC_HOME` names the directory itself, the same shape as `CARGO_HOME`:
/// unset, it defaults to `$HOME/.cargo` and, set, *is* the directory. A Rust
/// developer already knows the rule.
///
/// Under `cfg(test)`, an unset `VIVAC_HOME` is refused outright rather than
/// answered with this machine's real home. `tests/` binaries are separate
/// processes that always set `VIVAC_HOME` through `Sandbox` before spawning
/// the compiled tool, so nothing there can reach this; a unit test inside
/// `src/` shares this very process, and reading past a missing
/// `VIVAC_HOME` here is exactly what let two of them write six entries
/// into this machine's real `~/.vivac/projects`, in a version an installed
/// release could not read, over two rounds of this same task before
/// anyone noticed. A quiet fallback is what made that invisible; refusing
/// loudly is what makes it impossible instead of merely unlikely.
pub fn store_dir() -> Option<PathBuf> {
    let vivac_home = std::env::var_os("VIVAC_HOME");
    #[cfg(test)]
    if non_blank(vivac_home.as_deref()).is_none() {
        panic!(
            "store::store_dir() called with no VIVAC_HOME set. A test that reaches \
             this would read or write this machine's real registry. Set VIVAC_HOME to \
             a fresh temporary directory before calling anything that resolves the \
             store -- store::locate, registry::note and its own callers, relocate -- \
             the same way every test under tests/ already does through Sandbox."
        );
    }
    resolve_store_dir(
        vivac_home.as_deref(),
        std::env::var_os("HOME").as_deref(),
        std::env::var_os("USERPROFILE").as_deref(),
    )
}

/// Pure: given the three variables, where does the store go?
///
/// Split from `store_dir` so the tests never mutate the environment.
/// `std::env::set_var` is process-global and the test harness runs threads in
/// parallel; two tests setting `VIVAC_HOME` would race and the failure would
/// be intermittent, which is worse than no test at all.
fn resolve_store_dir(
    vivac_home: Option<&OsStr>,
    home: Option<&OsStr>,
    userprofile: Option<&OsStr>,
) -> Option<PathBuf> {
    if let Some(v) = non_blank(vivac_home) {
        return Some(PathBuf::from(v));
    }
    resolve_home_dir(home, userprofile).map(|h| h.join(DIR))
}

/// Where the user's home directory is, from the environment: `HOME` and
/// then `USERPROFILE`, the same two variables `store_dir` falls back to once
/// `VIVAC_HOME` is not set. Split out so `setup` can refuse to run there
/// without a second search of its own (`t579` §4.1).
pub fn home_dir() -> Option<PathBuf> {
    resolve_home_dir(
        std::env::var_os("HOME").as_deref(),
        std::env::var_os("USERPROFILE").as_deref(),
    )
}

/// Pure half of `home_dir`, for the same reason `resolve_store_dir` is split
/// from `store_dir`.
fn resolve_home_dir(home: Option<&OsStr>, userprofile: Option<&OsStr>) -> Option<PathBuf> {
    if let Some(h) = non_blank(home) {
        return Some(PathBuf::from(h));
    }
    if let Some(u) = non_blank(userprofile) {
        return Some(PathBuf::from(u));
    }
    None
}

/// `None` for a variable that is unset, empty or made only of whitespace: an
/// exported-but-empty variable is a common shell accident, and treating it as
/// "the store is at the filesystem root" would be actively harmful.
fn non_blank(v: Option<&OsStr>) -> Option<&OsStr> {
    let v = v?;
    match v.to_str() {
        Some(s) if s.trim().is_empty() => None,
        _ => Some(v),
    }
}

/// `config`'s `version`, once it is known to be one of the three shapes this
/// release can act on. `d444`: a tree that gains its first pillar or rule
/// turns this from `One` to `Locked`, in place, before the event that
/// creates it is appended -- and a release earlier than that fails to parse
/// `Locked`'s own sentence, which is the whole point. `Lanes` does the same
/// the moment a tree gains a lane.
///
/// No `#[derive(Serialize, Deserialize)]`: none of the three is an enum tag
/// in the usual sense, one is the bare integer `1` and the other two are
/// strings, and `check_config_version` -- not this type -- is what tells a
/// genuinely unknown version apart from one of these three.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ConfigVersion {
    One,
    Locked,
    Lanes,
}

/// The sentence a config's `version` becomes the moment its tree gains a
/// pillar or a rule. Literal and without a number: release-plz decides the
/// number when it publishes, and every release from here on reads the
/// sentence exactly as it reads `1`. `d444`.
pub const LOCK_SENTENCE: &str =
    "this tree holds pillars and rules, and this vivac is too old to read them: update vivac";

/// What a tree's `config` version becomes the moment it holds lanes. Same
/// mechanism as `d444`'s own sentence and for the same reason: a release
/// that does not know lanes must stop with a sentence a person can act on,
/// not read half a tree and act on it.
///
/// 0.12 reads both sentences, so a tree that holds pillars and lanes says
/// this one and loses nothing.
pub const LANE_SENTENCE: &str =
    "this tree holds lanes, and this vivac is too old to read them: update vivac";

impl Serialize for ConfigVersion {
    fn serialize<S>(&self, s: S) -> Result<S::Ok, S::Error>
    where
        S: serde::Serializer,
    {
        match self {
            ConfigVersion::One => s.serialize_u32(1),
            ConfigVersion::Locked => s.serialize_str(LOCK_SENTENCE),
            ConfigVersion::Lanes => s.serialize_str(LANE_SENTENCE),
        }
    }
}

impl<'de> Deserialize<'de> for ConfigVersion {
    /// Only ever reached once `check_config_version` has already let the raw
    /// value through: a `1`, the lock sentence, or the lane sentence.
    /// Anything else refuses generically here, which is `d444`'s "como hoy"
    /// for a version this deserializer was never meant to explain --
    /// negative, a float, an object, `null`.
    fn deserialize<D>(d: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        let v = serde_json::Value::deserialize(d)?;
        match &v {
            serde_json::Value::Number(n) if n.as_u64() == Some(1) => Ok(ConfigVersion::One),
            serde_json::Value::String(s) if s == LOCK_SENTENCE => Ok(ConfigVersion::Locked),
            serde_json::Value::String(s) if s == LANE_SENTENCE => Ok(ConfigVersion::Lanes),
            _ => Err(serde::de::Error::custom("unsupported config version")),
        }
    }
}

#[derive(Debug, Serialize, Deserialize)]
pub struct Config {
    pub version: ConfigVersion,
    pub project_id: String,
    /// Opaque identifier for this install. **It carries no email and no name**:
    /// the security pillar forbids it, and vetoes `MODEL.md` §3.4.
    pub actor: String,
}

impl Config {
    fn new_seeded() -> Config {
        Config {
            version: ConfigVersion::One,
            project_id: id::ulid(),
            actor: format!("a_{}", &id::ulid()[..12]),
        }
    }
}

pub struct Store {
    pub root: PathBuf,
    pub config: Config,
    /// Whether `events` was already there the moment this store opened it.
    /// A process that opens a tree whose log is already gone still recreates
    /// it on the next append, the same as planting would. What this guards
    /// against is narrower: a process that had the log open while it was
    /// there, and then had it moved out from underneath it, fails instead
    /// of silently starting a fresh one in its place.
    log_present: bool,
}

/// Walks up from `from_dir` looking for a `.vivac/`. No daemon and no environment
/// variable: the same rule as git, already in everyone's fingers.
///
/// The global store is a `.vivac/` as well, and it sits in the home directory,
/// so without this it answers the walk: any directory under a home and outside
/// a project resolves to the home itself, and a `push` there writes into the
/// global store instead of refusing. `d206` already said the upward search must
/// not find it, and this is that sentence. It asks what the directory holds
/// rather than where it sits, because `VIVAC_HOME` can move the store and a
/// rule that compared paths would fail exactly when somebody moved it.
///
/// The same question settles two more cases the same way (`f719`), and
/// neither is decided by guessing. A `.vivac/` that holds neither a tree
/// (`already_planted`) nor a lane file is not a map -- exactly what an
/// `--undo` that took a lane's own folder back used to leave behind, and
/// also what a half-deleted tree leaves behind, and the bytes on disk
/// cannot tell the two apart. With nothing above it either, there is
/// nothing to confuse it with: the walk passes it by, finds no map at
/// all, and `Ok(None)` reaches the ordinary "no store" answer, same as an
/// empty `.vivac/` always has (`f566`). With a real map above it, though,
/// stopping at the hollow one would silently treat somebody's half-gone
/// tree as its own -- `f566`'s own fear -- and passing it by would just as
/// silently fold this folder into the tree above, orphaning whatever it
/// was for -- `f719`'s own measurement. Both are a guess dressed as an
/// answer, so this refuses instead, the same as a half-written Codex
/// marker block does rather than guessing where it was meant to close.
pub fn find_root(from_dir: &Path) -> Result<Option<PathBuf>, Failure> {
    find_root_impl(from_dir, true)
}

/// [`find_root`], but passing a hollow `.vivac/` by instead of refusing
/// over it: `init`'s own walk (`d734`), scoped here rather than in
/// `find_root` itself so every other command keeps `f719`'s refusal
/// exactly as it always has. The hollow one's own remedy already reads
/// "Make it a tree of its own: vivac init" (`hollow_vivac_refusal`), and
/// `init` is the one command whose job is exactly that -- with a plan
/// that shows the tree above (`above_warning`) and asks before writing,
/// the choice is the person's, not a guess `f719` still refuses to make
/// for a command that only reads. `refuse_hollow` false means
/// `find_root_impl` never raises `Err` itself, but the return type still
/// carries one: `Located`'s own walk, past this point, still can.
pub fn find_root_for_planting(from_dir: &Path) -> Result<Option<PathBuf>, Failure> {
    find_root_impl(from_dir, false)
}

/// `find_root`'s own walk, shared with [`find_root_for_planting`]:
/// `refuse_hollow` is the only difference between the two, checked in
/// exactly one place, `d734`'s own guard against the two drifting apart
/// the way two hand copies would (`f724`).
fn find_root_impl(from_dir: &Path, refuse_hollow: bool) -> Result<Option<PathBuf>, Failure> {
    let mut d = from_dir.to_path_buf();
    let mut passed_a_hollow_one = false;
    loop {
        let candidate = d.join(DIR);
        if candidate.is_dir() && !crate::registry::marks_global_store(&candidate) {
            if already_planted(&d) || candidate.join(LANE).is_file() {
                if passed_a_hollow_one {
                    if !refuse_hollow {
                        // `init`: the hollow one already answered "plant
                        // here" for whichever folder is `from_dir`, not
                        // "join what is above it" -- the same folder the
                        // hollow one's own remedy names as "this folder".
                        return Ok(None);
                    }
                    return Err(hollow_vivac_refusal(&d));
                }
                return Ok(Some(d));
            }
            if !refuse_hollow {
                return Ok(None);
            }
            passed_a_hollow_one = true;
        }
        if !d.pop() {
            return Ok(None);
        }
    }
}

/// The same product-wide ceiling `f720` gave the setup plan (`PLAN_WIDTH`,
/// `claude_code.rs`) and the registry gives its own notices (`NOTICE_WIDTH`,
/// `registry.rs`): no printed line is wider than this, the indent counted
/// in. `hollow_vivac_refusal` is the one place in this module that embeds
/// text of unbounded length -- a folder name a person chose -- so it is the
/// one place here that needs it.
const REFUSAL_WIDTH: usize = 76;
const REFUSAL_INDENT: &str = "  ";

/// `f719`'s third case: a hollow `.vivac/` sat somewhere below `tree_above`
/// in the walk that found it -- holding neither a tree nor a lane. Which
/// of the two the hollow one belongs to is not on the filesystem to read,
/// so this names both ways out rather than picking one. The hollow
/// folder's own path is not named in the text: the sentence already says
/// "this folder", which is wherever the command was run from, and that is
/// not always the hollow folder itself -- a subfolder with no `.vivac/` of
/// its own, sitting under one that is hollow, reaches this exactly the
/// same way.
///
/// The paragraph goes through `render::wrap` rather than a hand-split
/// literal (`f720`, measured a second time here): `tree_above`'s own name
/// is a folder name a person chose, with no bound on how long it runs, and
/// splicing it into lines split by hand is wrong by construction the
/// moment that name is not the one this was tried against. The two lines
/// after it are not prose but the two ways out, one of them a command, and
/// a command broken across two lines is not one anybody can paste -- the
/// same exemption `sub_line` already has, so they are pushed whole.
fn hollow_vivac_refusal(tree_above: &Path) -> Failure {
    let name = tree_above
        .file_name()
        .map(|n| n.to_string_lossy().into_owned())
        .and_then(|n| match crate::redact::check_field("folder name", &n) {
            Some(_) => None,
            None => Some(n),
        });
    let label = crate::registry::label_for(name.as_deref());
    let prose = format!(
        "This folder has a .vivac/ that is neither a tree nor a lane: no log, no \
         config, no lane file. A tree sits above it, in {label}. Which of the two \
         this folder belongs to is a guess, and this tool does not guess."
    );
    let mut lines =
        crate::render::wrap(&prose, REFUSAL_WIDTH - REFUSAL_INDENT.len(), REFUSAL_INDENT);
    lines.push(String::new());
    lines.push(format!(
        "{REFUSAL_INDENT}Make it a tree of its own:  vivac init"
    ));
    lines.push(format!(
        "{REFUSAL_INDENT}Or hand it back to the tree above by deleting the empty .vivac/ here."
    ));
    Failure::Model(lines.join("\n"))
}

/// What resolving a working folder answers: not just where the tree is,
/// but whose thread this folder is.
#[derive(Debug)]
pub struct Located {
    /// The folder whose `.vivac/` holds `events` and `config`.
    pub root: PathBuf,
    /// The folder whose thread this is. Equal to `root` for a tree whose
    /// own folder is its founding lane, which is every tree today.
    ///
    /// The folder a lane file gets written to, which is setup's job (`t594`
    /// §4.5). Resolution answers it here so that nobody has to walk up
    /// twice.
    pub lane_dir: PathBuf,
    /// The lane as `.vivac/lane` names it. `None` when the folder holding
    /// the tree carries no lane file: the implicit `main` of rule 2.
    pub lane: Option<crate::lane::Lane>,
    /// The root of the linked worktree the command was run inside, when
    /// there is one. Whether it is a lane of its own is not a question the
    /// filesystem can answer -- it depends on the repositories the lane
    /// declared, which are in the log -- so it is answered elsewhere.
    ///
    /// A submodule inside a linked worktree comes out as that worktree, not
    /// as `None`: `anchor::linked_worktree` keeps looking above a
    /// submodule's own `.git` -- a file with no `commondir` -- for the
    /// worktree that contains it, rather than stopping at the submodule's
    /// own identity the way the anchor's other questions correctly do
    /// (`f606`).
    pub worktree: Option<PathBuf>,
}

/// Resolves `from_dir` to the tree it belongs to and the lane it is.
///
/// Walks up looking for a `.vivac/`, the same walk `find_root` always did,
/// and reads only small files past that: a lane file, and the log's own
/// first line when a lane's project has to be checked against an ancestor.
/// **No git and no folded log**: this runs on every process start, and the
/// write budget it shares the process with is 5 ms.
///
/// `None` for exactly the case `find_root` used to answer that way: no
/// `.vivac/` anywhere above `from_dir`, nor -- for a linked worktree --
/// above the main copy its history lives in either.
pub fn locate(from_dir: &Path) -> Result<Option<Located>, Failure> {
    locate_from(from_dir, store_dir().as_deref())
}

/// [`locate`], but for `init`'s own resolution (`d734`): the same walk,
/// through `find_root_for_planting` rather than `find_root`, so a hollow
/// `.vivac/` is passed by instead of refused over -- see that function's
/// own doc for why, and for whom. Used only by `setup::resolve_roots`.
pub fn locate_for_planting(from_dir: &Path) -> Result<Option<Located>, Failure> {
    locate_from_with(from_dir, store_dir().as_deref(), find_root_for_planting)
}

/// `locate`'s own algorithm, with the registry's directory taken as an
/// argument rather than read from the environment: `store_dir` reads
/// `VIVAC_HOME`, and mutating that in a test races every other test in the
/// same process, the same reason `resolve_store_dir` above is split from
/// `store_dir`.
fn locate_from(from_dir: &Path, registry_dir: Option<&Path>) -> Result<Option<Located>, Failure> {
    locate_from_with(from_dir, registry_dir, find_root)
}

/// `locate_from`'s own body, and [`locate_for_planting`]'s: which of the
/// two walks over a `.vivac/` -- `find_root`'s or
/// `find_root_for_planting`'s -- is the one parameter the two do not
/// share, so this is the one place either ever has to change (`f724`).
fn locate_from_with(
    from_dir: &Path,
    registry_dir: Option<&Path>,
    find_root: impl Fn(&Path) -> Result<Option<PathBuf>, Failure> + Copy,
) -> Result<Option<Located>, Failure> {
    let worktree = anchor::linked_worktree(from_dir);
    let mut found = locate_here(from_dir, registry_dir, find_root)?;
    if found.is_none() {
        // `git worktree add ../feature`: the worktree lives outside the
        // folder that holds the product, and nothing above it will ever
        // carry a `.vivac/` of the tree's own.
        if let Some(worktree_root) = &worktree {
            if let Some(main_root) = anchor::main_copy_of(worktree_root) {
                found = locate_here(&main_root, registry_dir, find_root)?;
            }
        }
    }
    Ok(found.map(|mut l| {
        l.worktree = worktree;
        l
    }))
}

/// The upward walk for the nearest `.vivac/`, and what it means once found:
/// a lane to resolve, or the implicit `main` every tree with none is.
fn locate_here(
    from_dir: &Path,
    registry_dir: Option<&Path>,
    find_root: impl Fn(&Path) -> Result<Option<PathBuf>, Failure>,
) -> Result<Option<Located>, Failure> {
    let Some(d) = find_root(from_dir)? else {
        return Ok(None);
    };
    match crate::lane::read(&d.join(DIR))? {
        Some(l) => resolve_lane(&d, l, registry_dir).map(Some),
        None => Ok(Some(Located {
            root: d.clone(),
            lane_dir: d,
            lane: None,
            worktree: None,
        })),
    }
}

/// Where the tree is for a folder that carries `.vivac/lane`: itself, when
/// it holds `events` or `config` **and** its own first event is the lane's
/// own project; otherwise the nearest ancestor with the same two things
/// true of it -- a folder that fails the second half, itself included, is
/// some other tree's and is walked past, not stopped at -- and only then
/// the registry, keyed by that same project.
///
/// The revalidation on `lane_dir` itself is not optional the way it might
/// look: a folder can carry both a lane file and a live `config` without
/// the two agreeing. `Store::open` writes a fresh, empty `config` the
/// moment one is missing, and a concurrent reader can land here in the
/// narrow window `relocate` opens between renaming the origin's `config`
/// away and its `events` -- no crash required, just a read. Trusting
/// `already_planted` alone there would read that orphaned config as this
/// folder's own tree and answer an empty one; checking its first event
/// against what the lane file names is what tells the two apart.
fn resolve_lane(
    lane_dir: &Path,
    lane: crate::lane::Lane,
    registry_dir: Option<&Path>,
) -> Result<Located, Failure> {
    if already_planted(lane_dir)
        && first_event_id(lane_dir).as_deref() == Some(lane.project.as_str())
    {
        return Ok(Located {
            root: lane_dir.to_path_buf(),
            lane_dir: lane_dir.to_path_buf(),
            lane: Some(lane),
            worktree: None,
        });
    }
    let mut up = lane_dir.to_path_buf();
    while up.pop() {
        if already_planted(&up) && first_event_id(&up).as_deref() == Some(lane.project.as_str()) {
            return Ok(Located {
                root: up,
                lane_dir: lane_dir.to_path_buf(),
                lane: Some(lane),
                worktree: None,
            });
        }
    }
    if let Some(registry_dir) = registry_dir {
        if let Some(root) = crate::registry::root_of(registry_dir, &lane.project) {
            return Ok(Located {
                root,
                lane_dir: lane_dir.to_path_buf(),
                lane: Some(lane),
                worktree: None,
            });
        }
    }
    // `t594`: writing this folder's own `.vivac/lane` is
    // what takes away the one path that used to resolve it. A linked
    // worktree with *no* lane file at all never reaches this function --
    // `locate_here` answers `None` for it, and `locate_from`'s own
    // fallback retries from the worktree's main copy. Once a lane file
    // exists here, resolution comes through this function instead, and
    // that fallback is never reached: neither the ancestor walk above nor
    // the registry knows anything about a path found by retrying from a
    // main copy. So this tries the retry itself, last, with the same
    // fingerprint check the ancestor walk above already uses -- a main
    // copy of some other repository entirely does not carry this lane's
    // `project` as its first event, so it is walked past rather than
    // mistaken for the right one. The registry stays a convenience that
    // can fail quietly (`registry::note` swallows its own errors): this
    // is what keeps a folder from going unusable just because it could
    // not be written to.
    if let Some(worktree_root) = crate::anchor::linked_worktree(lane_dir) {
        if let Some(main_root) = crate::anchor::main_copy_of(&worktree_root) {
            let mut up = main_root;
            loop {
                if already_planted(&up)
                    && first_event_id(&up).as_deref() == Some(lane.project.as_str())
                {
                    return Ok(Located {
                        root: up,
                        lane_dir: lane_dir.to_path_buf(),
                        lane: Some(lane),
                        worktree: None,
                    });
                }
                if !up.pop() {
                    break;
                }
            }
        }
    }
    Err(Failure::tree_not_found())
}

/// Whether `root/.vivac/` already holds a tree worth opening rather than
/// creating: a config or a log. `f566`: an empty `.vivac/` -- one that exists
/// as a directory but holds neither -- is planted like a new one, and `init`
/// and `setup` share this one check rather than each guessing it their own
/// way.
pub fn already_planted(root: &Path) -> bool {
    let dir = root.join(DIR);
    dir.join(CONFIG).is_file() || dir.join(LOG).is_file()
}

/// `t594` §4.9: every `.vivac/` ignores itself, tree or lane. Creates the
/// directory if it is not there, and writes nothing over a file that
/// already exists -- somebody may have added a line of their own.
pub fn write_gitignore(vivac_dir: &Path) -> std::io::Result<()> {
    fs::create_dir_all(vivac_dir)?;
    let ignore = vivac_dir.join(GITIGNORE);
    if !ignore.exists() {
        fs::write(&ignore, "*\n")?;
    }
    Ok(())
}

/// Builds `vivac_dir` whole rather than empty-then-filled, for the one case
/// it does not exist yet: everything `fill` writes lands in a sibling
/// temporary directory first -- named `<DIR>.<ulid>.tmp`, so it sits beside
/// `vivac_dir` on the same filesystem and two writers never pick the same
/// name -- and only once `fill` returns `Ok` is that directory renamed
/// onto `vivac_dir`, in one filesystem call. A reader resolving `vivac_dir`
/// while this runs therefore sees either nothing at all, or a folder that
/// already holds everything `fill` put there -- never a directory that
/// exists but is still empty, or half of what `fill` meant to write
/// (`f735`). `lane::write` and `Store::create` are its two callers: the
/// mechanics used to be copied between them, which is exactly the mistake
/// `f724` warns about -- closing one instance of a class of bug and finding
/// it again, unfixed, the same day.
///
/// Returns `Ok(true)` once `vivac_dir` holds this write. Returns `Ok(false)`
/// without ever calling `fill` when `vivac_dir` already exists -- an
/// empty or hollow `.vivac/` included (`f566`, `d734`): there is no fresh
/// appearance left to make atomic at that point, so the caller falls back
/// to writing in place, exactly as every caller did before this existed.
/// Returns `Ok(false)` a second way too, after `fill` has already run:
/// the rename itself can still lose a race with another writer whose own
/// `vivac_dir` appeared in the gap between the check above and the rename
/// below. On Unix, renaming a directory onto one that is empty succeeds and
/// replaces it outright, so a loser that finds only an empty directory
/// there quietly wins anyway, landing its own complete write; but once the
/// other writer's content has landed the directory is no longer empty, and
/// the rename fails there the same way it always fails on Windows, empty or
/// not. Either branch ends with `vivac_dir` holding someone's whole write,
/// never a hollow one -- which is all this call ever promises about the
/// folder it does not itself finish.
///
/// Every path out of here that is not the successful rename removes the
/// temporary directory, best effort. The one thing it cannot clean up after
/// is a literal crash between two of its own steps, not an `Err` it can
/// catch -- the most that leaves behind is the `<DIR>.<ulid>.tmp` sibling
/// itself, never a hollow `vivac_dir`: nothing is ever renamed there until
/// `fill` has already finished it.
pub(crate) fn build_fresh(
    vivac_dir: &Path,
    fill: impl FnOnce(&Path) -> std::io::Result<()>,
) -> std::io::Result<bool> {
    if vivac_dir.is_dir() {
        return Ok(false);
    }
    let Some(parent) = vivac_dir.parent() else {
        return Ok(false);
    };
    let tmp_dir = parent.join(format!("{DIR}.{}.tmp", id::ulid()));
    if let Err(e) = (|| -> std::io::Result<()> {
        fs::create_dir_all(&tmp_dir)?;
        fill(&tmp_dir)
    })() {
        fs::remove_dir_all(&tmp_dir).ok();
        return Err(e);
    }
    match fs::rename(&tmp_dir, vivac_dir) {
        Ok(()) => Ok(true),
        Err(_) => {
            fs::remove_dir_all(&tmp_dir).ok();
            Ok(false)
        }
    }
}

/// The log's length and modification time: the whole change detector.
/// The log only grows, so a different length is exact; the time rides
/// along for a rewrite that lands on the same byte count.
pub(crate) fn fingerprint(log: &Path) -> (u64, Option<std::time::SystemTime>) {
    match fs::metadata(log) {
        Ok(m) => (m.len(), m.modified().ok()),
        Err(_) => (0, None),
    }
}

/// The same pair as `fingerprint`, read off a handle already open, so the
/// file it is still reading can be compared against whatever the path
/// names now.
pub(crate) fn fingerprint_in(f: &File) -> (u64, Option<std::time::SystemTime>) {
    match f.metadata() {
        Ok(m) => (m.len(), m.modified().ok()),
        Err(_) => (0, None),
    }
}

/// One tree's write lock, held for as long as this value lives (`d598`).
/// Dropping it releases the lock, and so does the process dying: the
/// operating system lets go of every lock a dead process held, so there
/// is never a stale lock to clean up by hand.
///
/// It locks `.vivac/lock` and never the log: on Windows a lock taken with
/// `LockFileEx` is mandatory, and a locked log would refuse its readers.
pub struct WriteLock {
    file: File,
    path: PathBuf,
}

impl WriteLock {
    /// Whether this lock is the one that covers `lock_path`. `Store::append`
    /// asks before it writes: a lock is only a lock over the tree whose file
    /// it holds, and taking one tree's lock to write another's would look
    /// exactly like holding no lock at all (`f602`).
    pub fn covers(&self, lock_path: &Path) -> bool {
        self.path == lock_path
    }
}

impl Drop for WriteLock {
    fn drop(&mut self) {
        let _ = self.file.unlock();
    }
}

/// Takes the lock at `path`, retrying until `deadline`. A handoff is
/// microseconds, so it yields for `LOCK_SPIN` before it starts sleeping.
pub(crate) fn lock_with_deadline(
    path: &Path,
    deadline: std::time::Duration,
) -> Result<WriteLock, Failure> {
    let file = OpenOptions::new()
        .create(true)
        .read(true)
        .write(true)
        .truncate(false)
        .open(path)?;
    let start = std::time::Instant::now();
    loop {
        match file.try_lock() {
            Ok(()) => {
                return Ok(WriteLock {
                    file,
                    path: path.to_path_buf(),
                })
            }
            Err(std::fs::TryLockError::WouldBlock) => {}
            Err(std::fs::TryLockError::Error(e)) => return Err(e.into()),
        }
        let waited = start.elapsed();
        if waited >= deadline {
            return Err(Failure::busy(deadline));
        }
        if waited < LOCK_SPIN {
            std::thread::yield_now();
        } else {
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
    }
}

/// The `id` of line 1 of `<root>/.vivac/events`, without folding the rest of
/// the log. An empty log, an unreadable file or a first line that will not
/// parse all come back `None`; the caller decides what that means.
pub fn first_event_id(root: &Path) -> Option<String> {
    let f = File::open(root.join(DIR).join(LOG)).ok()?;
    let mut line = String::new();
    BufReader::new(f).read_line(&mut line).ok()?;
    if line.trim().is_empty() {
        return None;
    }
    let e: crate::event::Event = serde_json::from_str(line.trim_end()).ok()?;
    Some(e.id)
}

impl Store {
    pub fn open(root: PathBuf) -> Result<Store, Failure> {
        let p = root.join(DIR).join(CONFIG);
        let config = match fs::read_to_string(&p) {
            Ok(s) => read_config(&s)?,
            Err(_) => {
                // A `.vivac/` with no config comes from an earlier version or a
                // half-finished delete. Fill it in rather than fail: the tree,
                // which is what matters, lives in `events`. `d444`: if the log
                // already carries a pillar, a rule or a lane, the regenerated
                // config is born locked to whichever of those sentences the
                // log backs up -- deleting one file must never hand an older
                // release a config that looks readable over a tree it is not.
                let c = Config {
                    version: regenerated_version(&root),
                    ..Config::new_seeded()
                };
                write_config(&root, &c)?;
                c
            }
        };
        let log_present = root.join(DIR).join(LOG).is_file();
        Ok(Store {
            root,
            config,
            log_present,
        })
    }

    /// For a `.vivac/` that does not exist yet. `f566`: `init` on one that
    /// already does calls `open` instead, so this always writes a fresh
    /// config -- calling it over an existing tree would hand it a new
    /// `project_id` and drop `d444`'s lock back to `1`.
    ///
    /// Built whole through `build_fresh` (`f735`): `config`, `events` and
    /// `.gitignore` all land in a sibling temporary directory before it is
    /// ever renamed onto `root/.vivac`, so a reader resolving this folder
    /// mid-plant sees either nothing yet or the finished tree, never a
    /// directory that exists but holds none of the three. When `d` is
    /// already there -- an empty or hollow `.vivac/`, which `d734` lets
    /// `init` plant straight over -- `build_fresh` calls back `Ok(false)`
    /// without touching anything, and this falls back to the same three
    /// writes, in the same order, `create` has always done in place: there
    /// is no fresh appearance left to make atomic once the folder already
    /// exists.
    pub fn create(root: &Path) -> std::io::Result<Store> {
        let d = root.join(DIR);
        let config = Config::new_seeded();
        let built = build_fresh(&d, |tmp_dir| {
            write_config_into(tmp_dir, &config)?;
            File::create(tmp_dir.join(LOG))?;
            write_gitignore(tmp_dir)
        })?;
        if !built {
            fs::create_dir_all(&d)?;
            write_config(root, &config)?;
            if !d.join(LOG).exists() {
                File::create(d.join(LOG))?;
            }
            write_gitignore(&d)?;
        }
        Ok(Store {
            root: root.to_path_buf(),
            config,
            log_present: true,
        })
    }

    pub fn log(&self) -> PathBuf {
        self.root.join(DIR).join(LOG)
    }

    pub fn index_path(&self) -> PathBuf {
        self.root.join(DIR).join(INDEX)
    }

    pub fn lock_path(&self) -> PathBuf {
        self.root.join(DIR).join(LOCK)
    }

    /// Takes this tree's write lock (`d598`), waiting for another writer
    /// for up to five seconds. Hold it from the moment the tree is brought
    /// up to date until the append is done.
    pub fn lock_for_write(&self) -> Result<WriteLock, Failure> {
        lock_with_deadline(&self.lock_path(), LOCK_DEADLINE)
    }
}

fn write_config(root: &Path, c: &Config) -> std::io::Result<()> {
    write_config_into(&root.join(DIR), c)
}

/// `write_config`, given the `.vivac/` directory itself rather than its
/// parent: the shape `build_fresh`'s `fill` needs, since the directory it
/// hands back while planting a tree fresh is a temporary sibling, not
/// `root.join(DIR)` yet.
fn write_config_into(vivac_dir: &Path, c: &Config) -> std::io::Result<()> {
    let mut f = File::create(vivac_dir.join(CONFIG))?;
    f.write_all(serde_json::to_string_pretty(c)?.as_bytes())?;
    f.write_all(b"\n")
}

/// `d444`'s own protection: the config is written to a sibling temporary
/// file and renamed over the real one, never edited in place. A process
/// that dies between the two steps leaves the old config exactly as it
/// was -- there is no window where `config` itself is half-written.
fn write_config_atomic(root: &Path, c: &Config) -> std::io::Result<()> {
    let dir = root.join(DIR);
    let tmp = dir.join("config.tmp");
    {
        let mut f = File::create(&tmp)?;
        f.write_all(serde_json::to_string_pretty(c)?.as_bytes())?;
        f.write_all(b"\n")?;
    }
    fs::rename(&tmp, dir.join(CONFIG))
}

/// Parses `config`'s text into a `Config`, refusing the two shapes of
/// `version` `t411` §27 gives a name to before letting `serde_json` see the
/// rest: a non-negative integer that is not `1`, or a string that is not
/// `d444`'s own sentence. Every other shape -- absent, negative, a float, an
/// object, `null` -- is left to `Config`'s own `Deserialize`, which fails
/// exactly as it did before `d444`.
fn read_config(raw: &str) -> Result<Config, Failure> {
    let v: serde_json::Value =
        serde_json::from_str(raw).map_err(|e| Failure::Io(std::io::Error::other(e)))?;
    check_config_version(v.get("version"))?;
    serde_json::from_value(v).map_err(|e| Failure::Io(std::io::Error::other(e)))
}

fn check_config_version(version: Option<&serde_json::Value>) -> Result<(), Failure> {
    match version {
        Some(serde_json::Value::Number(n)) => match n.as_u64() {
            Some(1) => Ok(()),
            Some(other) => Err(Failure::newer_vivac(format!(
                "This tree was written by a newer vivac: its config has version {other}, \
                 which this version does not know. Update vivac to read it. A session or \
                 vivac web opened before an update keeps the old vivac until it restarts. \
                 Nothing was written."
            ))),
            // Negative or non-integer: not one of the two known shapes, and
            // not a value worth a friendly message either. Falls through to
            // the generic config-read failure, same as before `d444`.
            None => Ok(()),
        },
        Some(serde_json::Value::String(s)) if s == LOCK_SENTENCE => Ok(()),
        Some(serde_json::Value::String(s)) if s == LANE_SENTENCE => Ok(()),
        Some(serde_json::Value::String(s)) => Err(Failure::newer_vivac(format!(
            "This tree was written by a newer vivac: its config says {s:?}. Update vivac \
             to read it. A session or vivac web opened before an update keeps the old \
             vivac until it restarts. Nothing was written."
        ))),
        _ => Ok(()),
    }
}

/// The rare path `Store::open` takes when `config` itself is missing: which
/// sentence, if any, the log already backs up. Reads the whole log --
/// something no ordinary read ever pays for -- because a vanished config is
/// itself the unusual case, and regenerating one that looks readable by any
/// release over a tree that already governs something, or already holds a
/// lane, would undo the very lock `d444` and `t594` §2.6 exist to keep.
///
/// A lane wins over a pillar or a rule when a log somehow carries both: 0.12
/// reads both sentences, so a tree that holds pillars and lanes still says
/// this one and loses nothing, and there is no third sentence for "both" to
/// pick instead.
fn regenerated_version(root: &Path) -> ConfigVersion {
    let Ok(f) = File::open(root.join(DIR).join(LOG)) else {
        return ConfigVersion::One;
    };
    let mut governed = false;
    for line in BufReader::new(f).lines().map_while(Result::ok) {
        if line.trim().is_empty() {
            continue;
        }
        let Ok(v) = serde_json::from_str::<serde_json::Value>(&line) else {
            continue;
        };
        match v["payload"]["type"].as_str() {
            Some("lane.declared") | Some("lane.claimed") => return ConfigVersion::Lanes,
            Some("node.created")
                if matches!(v["payload"]["kind"].as_str(), Some("pillar") | Some("rule")) =>
            {
                governed = true;
            }
            _ => {}
        }
    }
    if governed {
        ConfigVersion::Locked
    } else {
        ConfigVersion::One
    }
}

/// `config`'s version, read directly and without writing anything: `None`
/// for a tree with no config at all, or one this release cannot make sense
/// of. `Store::open` would fill a missing one in, and that write is exactly
/// what a caller that must never write -- `setup`'s own `--dry-run` -- is
/// not allowed to trigger just by asking what version a tree is on
/// (`t594`).
pub(crate) fn peek_config_version(root: &Path) -> Option<ConfigVersion> {
    let raw = fs::read_to_string(root.join(DIR).join(CONFIG)).ok()?;
    read_config(&raw).ok().map(|c| c.version)
}

/// What an append left behind: the events as written, plus where the last
/// of them begins and where the file now ends. A caller that keeps a tree
/// in memory needs those two numbers to stay current without reading back
/// what it just wrote -- and reading it back is not free: opening the log
/// again right after writing it costs about six milliseconds a write on
/// this machine, more than the whole write budget (`f599`).
pub struct Appended {
    pub events: Vec<crate::event::Event>,
    /// The log's length right before this append, so a caller can tell
    /// whether these lines are a clean continuation of what it already
    /// folded, or landed behind bytes it never saw (`f599`).
    pub previous_len: u64,
    pub last_line_offset: u64,
    pub end_offset: u64,
}

/// Whether `path`'s log already ends with `\n`, so `append` knows whether
/// it can write straight behind it or has to close the torn line first
/// (`f604`). One open plus one seek to the log's own end and one byte read
/// back -- never a read of what came before it, so this stays flat no
/// matter how long the log has grown. An empty log has no line to tear, so
/// it answers `true`.
fn log_ends_with_newline(path: &Path) -> std::io::Result<bool> {
    let mut f = File::open(path)?;
    let len = f.metadata()?.len();
    if len == 0 {
        return Ok(true);
    }
    f.seek(SeekFrom::End(-1))?;
    let mut last = [0u8; 1];
    f.read_exact(&mut last)?;
    Ok(last[0] == b'\n')
}

impl Store {
    /// Reads the whole log. An unreadable line **does not abort**: it is
    /// counted and skipped. A half-written log has to stay readable, or the
    /// tool that keeps the thread becomes the one that loses it.
    ///
    /// One case refuses instead of skipping: `t411` §13, a line that is
    /// well-formed JSON but names an event type or a node kind this version
    /// does not know. That line was written by a newer vivac, and reading
    /// past it in silence would mean acting on a tree this version cannot
    /// actually see all of.
    pub fn read_all(&self) -> Result<(Vec<crate::event::Event>, usize), Failure> {
        read_all_from(&self.log())
    }

    /// Appends events at the end. One line per event, rewriting nothing.
    ///
    /// This is the critical path of the agent's turn: a p99 < 5 ms budget.
    /// That is why there is no `fsync` --on Windows it costs more than the
    /// whole budget-- and why it opens in `append` mode, which makes each
    /// single-line write atomic. Atomic lines do not make two writers agree
    /// on `seq` and `num`, though: the write lock is an argument here, not a
    /// convention a caller could forget or take twice. Without one this does
    /// not compile, and `lock.covers` refuses one taken on another tree's
    /// `.vivac/lock` (`f602`).
    ///
    /// `lane` is an argument for the same reason: which thread this write
    /// signs as is a fact about *this* write, not a property of the folder
    /// where it lands, and a `Store` that carried it as a field could be
    /// asked to sign with whatever it happened to be left holding
    /// (`f608`, third time -- the write lock in task 1, the tree's own
    /// lane in task 6, and this one). `Ctx` is the only place that ever
    /// knew which lane it was; now it is also the only place that can
    /// forget.
    ///
    /// `tree_already_governed` is `d444`'s own check, paid before any of
    /// `body` reaches disk: the config locks in place, first, so a process
    /// that dies between the two leaves an unlocked config over a tree with
    /// no pillar and no rule, which is harmless.
    ///
    /// Returns the events as written and where their bytes landed, so a
    /// caller that keeps the tree in memory applies exactly those and never
    /// stamps them a second time (`f590`), and a caller that keeps a
    /// resident tree never has to read the log back to learn where its own
    /// write landed (`f599`).
    pub fn append(
        &mut self,
        lock: &WriteLock,
        lane: &str,
        body: Vec<crate::event::Body>,
        from_seq: u64,
        tree_already_governed: bool,
    ) -> std::io::Result<Appended> {
        if !lock.covers(&self.lock_path()) {
            return Err(std::io::Error::other("write lock does not cover this tree"));
        }
        self.lock_if_needed(&body, tree_already_governed)?;
        let mut buf = String::with_capacity(256 * body.len());
        let mut written = Vec::with_capacity(body.len());
        let mut last_line_start = 0usize;
        for (i, c) in body.into_iter().enumerate() {
            let e = crate::event::Event {
                seq: from_seq + i as u64 + 1,
                id: id::ulid(),
                ts: clock::now_rfc3339(),
                actor: self.config.actor.clone(),
                lane: lane.to_string(),
                payload: c,
            };
            last_line_start = buf.len();
            buf.push_str(&serde_json::to_string(&e).map_err(std::io::Error::other)?);
            buf.push('\n');
            written.push(e);
        }
        // `f604`: a crash mid-write can leave the log's last line with no
        // closing `\n`. Opening in `append` mode writes straight behind
        // whatever is already there, so without this the next write glues
        // its own first line onto the torn one -- and the merged line
        // fails to parse, taking that first line down with the one
        // already lost. Checked with one seek and one byte read from the
        // log's own end, never a read of what came before it, so the cost
        // stays flat as the log grows: `append`'s budget is p99 < 5 ms.
        let needs_newline_first = self.log_present && !log_ends_with_newline(&self.log())?;
        let mut f = OpenOptions::new()
            .create(!self.log_present)
            .append(true)
            .open(self.log())?;
        let previous_len = f.metadata()?.len();
        let prefix_len: u64 = if needs_newline_first {
            f.write_all(b"\n")?;
            1
        } else {
            0
        };
        f.write_all(buf.as_bytes())?;
        self.log_present = true;
        if !written.is_empty() {
            mark_write();
        }
        Ok(Appended {
            previous_len,
            last_line_offset: previous_len + prefix_len + last_line_start as u64,
            end_offset: previous_len + prefix_len + buf.len() as u64,
            events: written,
        })
    }

    /// `d444`: locks the config in place the moment this tree gains its
    /// first pillar or rule -- before the event that creates one is
    /// appended. A no-op once the config is already locked, and a no-op for
    /// every write that neither creates a pillar or a rule nor lands on a
    /// tree that already has one.
    fn lock_if_needed(
        &mut self,
        body: &[crate::event::Body],
        tree_already_governed: bool,
    ) -> std::io::Result<()> {
        if self.config.version != ConfigVersion::One {
            return Ok(());
        }
        let creates_governance = body.iter().any(|b| {
            matches!(
                b,
                crate::event::Body::NodeCreated {
                    kind: crate::event::Kind::Pillar | crate::event::Kind::Rule,
                    ..
                }
            )
        });
        if !tree_already_governed && !creates_governance {
            return Ok(());
        }
        let locked = Config {
            version: ConfigVersion::Locked,
            project_id: self.config.project_id.clone(),
            actor: self.config.actor.clone(),
        };
        write_config_atomic(&self.root, &locked)?;
        self.config = locked;
        Ok(())
    }

    /// Locks the config in place the moment this tree gains a lane, the same
    /// mechanism `lock_if_needed` uses for a pillar or a rule and by the same
    /// `write_config_atomic`. A no-op once the config already says `Lanes`.
    ///
    /// Takes the write lock as an argument for the same reason `append`
    /// does: without one this does not compile, and `lock.covers` refuses
    /// one taken on another tree's `.vivac/lock` (`f602`) -- `config.tmp`'s
    /// own name is fixed, so it is only safe with nobody else writing at
    /// the same time.
    pub fn lock_lanes_in_config(&mut self, lock: &WriteLock) -> std::io::Result<()> {
        if !lock.covers(&self.lock_path()) {
            return Err(std::io::Error::other("write lock does not cover this tree"));
        }
        if self.config.version == ConfigVersion::Lanes {
            return Ok(());
        }
        let locked = Config {
            version: ConfigVersion::Lanes,
            project_id: self.config.project_id.clone(),
            actor: self.config.actor.clone(),
        };
        write_config_atomic(&self.root, &locked)?;
        self.config = locked;
        Ok(())
    }
}

/// The read `Store::read_all` runs, taken as a free function of a path
/// rather than a method: `index.rs`'s own tail read (`read_tracked`) keeps a
/// separate implementation for its own reasons (`LOADING.md` §4), but when
/// it hits a line `t411` §13 refuses over, it falls back to a full read from
/// byte zero rather than reconstructing this file's own line count -- and
/// that full read is this function, so the two paths report the very same
/// line number for the very same line.
pub(crate) fn read_all_from(path: &Path) -> Result<(Vec<crate::event::Event>, usize), Failure> {
    let f = match File::open(path) {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((vec![], 0)),
        Err(e) => return Err(e.into()),
    };
    let mut reader = BufReader::new(f);
    let mut events = Vec::new();
    let mut broken = 0usize;
    let mut line_no = 0usize;
    let mut raw = Vec::new();
    loop {
        raw.clear();
        let n = reader.read_until(b'\n', &mut raw)?;
        if n == 0 {
            break;
        }
        line_no += 1;
        if raw.last() != Some(&b'\n') {
            // An append that stopped mid-write is not a line until it
            // ends -- the same rule `index::read_tracked` follows, so a
            // log the two of them read agrees on what it holds (`f599`).
            if !String::from_utf8_lossy(&raw).trim().is_empty() {
                broken += 1;
            }
            break;
        }
        let mut bytes = raw.as_slice();
        if bytes.last() == Some(&b'\n') {
            bytes = &bytes[..bytes.len() - 1];
        }
        if bytes.last() == Some(&b'\r') {
            bytes = &bytes[..bytes.len() - 1];
        }
        let line = String::from_utf8(bytes.to_vec()).map_err(|_| {
            std::io::Error::new(
                std::io::ErrorKind::InvalidData,
                "stream did not contain valid UTF-8",
            )
        })?;
        if line.trim().is_empty() {
            continue;
        }
        match serde_json::from_str(&line) {
            Ok(e) => events.push(e),
            Err(_) => match crate::event::unknown_reason_for(&line) {
                Some(reason) => return Err(newer_vivac_failure(line_no, reason)),
                None => broken += 1,
            },
        }
    }
    Ok((events, broken))
}

/// One `seq` two different writes both believed was theirs (`f610`): unlike
/// `num`, whose repeat the fold already tracks with one entry per second
/// claimant, nothing walks the raw log with a line number in hand, and
/// `check` is the only caller that needs one.
pub(crate) struct RepeatedSeq {
    pub(crate) seq: u64,
    pub(crate) first_line: usize,
    pub(crate) second_line: usize,
}

/// What `scan_log` finds: the two log corruptions `check`'s own fold never
/// named before `f604`/`f610`, each with the line number a text editor
/// would show.
pub(crate) struct LogScan {
    pub(crate) repeated_seqs: Vec<RepeatedSeq>,
    /// The log's own last line, when it never got its closing `\n`: a
    /// crash mid-write, or a write that landed behind one and is gone for
    /// good (`f604`). `None` on a log that ends cleanly, or has none.
    pub(crate) torn_tail: Option<usize>,
}

/// A full read of the log for `check` alone, in the shape `read_all_from`
/// already takes: `check` carries no budget of its own, so one pass that
/// finds both of `f604`/`f610`'s corruptions costs less than two separate
/// ones would. A line neither valid JSON nor a readable event is already
/// named by `broken_lines`, so this pass skips it rather than naming it
/// again.
pub(crate) fn scan_log(path: &Path) -> Result<LogScan, Failure> {
    let f = match File::open(path) {
        Ok(f) => f,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
            return Ok(LogScan {
                repeated_seqs: Vec::new(),
                torn_tail: None,
            })
        }
        Err(e) => return Err(e.into()),
    };
    let mut reader = BufReader::new(f);
    let mut seen: std::collections::HashMap<u64, usize> = std::collections::HashMap::new();
    let mut repeated_seqs = Vec::new();
    let mut torn_tail = None;
    let mut line_no = 0usize;
    let mut raw = Vec::new();
    loop {
        raw.clear();
        let n = reader.read_until(b'\n', &mut raw)?;
        if n == 0 {
            break;
        }
        line_no += 1;
        if raw.last() != Some(&b'\n') {
            if !String::from_utf8_lossy(&raw).trim().is_empty() {
                torn_tail = Some(line_no);
            }
            break;
        }
        let Ok(line) = std::str::from_utf8(&raw) else {
            continue;
        };
        let line = line.trim();
        if line.is_empty() {
            continue;
        }
        let Ok(e) = serde_json::from_str::<crate::event::Event>(line) else {
            continue;
        };
        if let Some(&first_line) = seen.get(&e.seq) {
            repeated_seqs.push(RepeatedSeq {
                seq: e.seq,
                first_line,
                second_line: line_no,
            });
        } else {
            seen.insert(e.seq, line_no);
        }
    }
    Ok(LogScan {
        repeated_seqs,
        torn_tail,
    })
}

/// The exact wording of `t411` §13's refusal, for the one line that earned
/// it. `line_no` is 1-based, matching what a text editor would show.
pub(crate) fn newer_vivac_failure(line_no: usize, reason: crate::event::UnknownReason) -> Failure {
    let path = format!("{DIR}/{LOG}");
    let detail = match reason {
        crate::event::UnknownReason::EventType(t) => {
            format!("is an event this version does not know ({t})")
        }
        crate::event::UnknownReason::NodeKind(k) => {
            format!("creates a node of a type this version does not know ({k})")
        }
        crate::event::UnknownReason::Shape(t) => {
            format!("is a {t} event whose fields this version cannot read")
        }
    };
    Failure::newer_vivac(format!(
        "This tree was written by a newer vivac: line {line_no} of {path} {detail}. \
         Update vivac to read it. A session or vivac web opened before an update keeps \
         the old vivac until it restarts. Nothing was written."
    ))
}

impl Store {
    /// Writes already-built events, keeping their original timestamp. Only
    /// `import` uses it: a tree from elsewhere keeps its dates, because
    /// otherwise the migration flattens the only timeline it had. Takes the
    /// write lock as an argument for the same reason `append` does: without
    /// one this does not compile, and `lock.covers` refuses one taken on
    /// another tree's `.vivac/lock` (`f602`).
    pub fn write_raw(
        &self,
        lock: &WriteLock,
        events: &[crate::event::Event],
    ) -> std::io::Result<()> {
        if !lock.covers(&self.lock_path()) {
            return Err(std::io::Error::other("write lock does not cover this tree"));
        }
        let mut buf = String::with_capacity(256 * events.len());
        for e in events {
            buf.push_str(&serde_json::to_string(e).map_err(std::io::Error::other)?);
            buf.push('\n');
        }
        let mut f = OpenOptions::new()
            .create(!self.log_present)
            .append(true)
            .open(self.log())?;
        f.write_all(buf.as_bytes())
    }
}

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

    #[test]
    fn search_upward() {
        let tmp = std::env::temp_dir().join(format!("vivac-t-{}", id::ulid()));
        let depth_of = tmp.join("a").join("b").join("c");
        fs::create_dir_all(&depth_of).unwrap();
        assert!(find_root(&depth_of).unwrap().is_none());
        Store::create(&tmp).unwrap();
        assert_eq!(find_root(&depth_of).unwrap().unwrap(), tmp);
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn the_global_store_does_not_answer_the_walk() {
        // The collision as it shipped: the global store is a `.vivac/` too, so
        // a directory with no project above it resolved to the home directory
        // and wrote there without saying so.
        let tmp = std::env::temp_dir().join(format!("vivac-t-{}", id::ulid()));
        let deep = tmp.join("a").join("b");
        fs::create_dir_all(&deep).unwrap();
        Store::create(&tmp).unwrap();
        assert_eq!(find_root(&deep).unwrap().unwrap(), tmp);
        crate::registry::note(
            &tmp.join(DIR),
            "01aaaaaaaaaaaaaaaaaaaaaaaa",
            crate::registry::Sighting {
                root: &deep,
                lane: None,
                repos: None,
            },
        );
        assert_ne!(find_root(&deep).unwrap(), Some(tmp.clone()));
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn a_project_under_the_global_store_still_wins() {
        // Skipping the global store must not cost a real project below it.
        let tmp = std::env::temp_dir().join(format!("vivac-t-{}", id::ulid()));
        let project = tmp.join("work");
        let deep = project.join("src").join("deep");
        fs::create_dir_all(&deep).unwrap();
        Store::create(&tmp).unwrap();
        crate::registry::note(
            &tmp.join(DIR),
            "01aaaaaaaaaaaaaaaaaaaaaaaa",
            crate::registry::Sighting {
                root: &project,
                lane: None,
                repos: None,
            },
        );
        Store::create(&project).unwrap();
        assert_eq!(find_root(&deep).unwrap().unwrap(), project);
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn the_actor_carries_no_personal_data() {
        let c = Config::new_seeded();
        assert!(c.actor.starts_with("a_"));
        assert!(!c.actor.contains('@'));
        assert_ne!(c.actor, whoami_ish());
    }

    fn whoami_ish() -> String {
        std::env::var("USERNAME")
            .or_else(|_| std::env::var("USER"))
            .unwrap_or_default()
    }

    #[test]
    fn vivac_home_wins_and_is_used_as_is() {
        let got = resolve_store_dir(
            Some(OsStr::new("/somewhere/store")),
            Some(OsStr::new("/home/anyone")),
            Some(OsStr::new("C:\\Users\\anyone")),
        );
        assert_eq!(got, Some(PathBuf::from("/somewhere/store")));
    }

    #[test]
    fn blank_vivac_home_falls_through() {
        let got = resolve_store_dir(
            Some(OsStr::new("   ")),
            Some(OsStr::new("/home/anyone")),
            None,
        );
        assert_eq!(got, Some(PathBuf::from("/home/anyone").join(DIR)));
    }

    #[test]
    fn home_alone_appends_dir() {
        let got = resolve_store_dir(None, Some(OsStr::new("/home/anyone")), None);
        assert_eq!(got, Some(PathBuf::from("/home/anyone").join(DIR)));
    }

    #[test]
    fn userprofile_used_when_home_is_absent() {
        let got = resolve_store_dir(None, None, Some(OsStr::new("C:\\Users\\anyone")));
        assert_eq!(got, Some(PathBuf::from("C:\\Users\\anyone").join(DIR)));
    }

    #[test]
    fn home_wins_over_userprofile() {
        let got = resolve_store_dir(
            None,
            Some(OsStr::new("/home/anyone")),
            Some(OsStr::new("C:\\Users\\anyone")),
        );
        assert_eq!(got, Some(PathBuf::from("/home/anyone").join(DIR)));
    }

    #[test]
    fn nothing_set_means_no_global_store() {
        assert_eq!(resolve_store_dir(None, None, None), None);
    }

    #[test]
    fn first_event_id_on_an_empty_log_is_none() {
        let tmp = std::env::temp_dir().join(format!("vivac-fe-{}", id::ulid()));
        Store::create(&tmp).unwrap();
        assert_eq!(first_event_id(&tmp), None);
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn first_event_id_reads_line_one_without_folding() {
        let tmp = std::env::temp_dir().join(format!("vivac-fe-{}", id::ulid()));
        let mut s = Store::create(&tmp).unwrap();
        let lock = s.lock_for_write().unwrap();
        // A log large enough that folding the whole thing would be visible
        // in the timing, if this ever regressed into calling `read_all`.
        for _ in 0..500 {
            s.append(
                &lock,
                crate::lane::MAIN,
                vec![crate::event::Body::NodeNoted {
                    node: "t1".into(),
                    note: "filler".into(),
                }],
                0,
                false,
            )
            .unwrap();
        }
        let first_line = fs::read_to_string(s.log())
            .unwrap()
            .lines()
            .next()
            .unwrap()
            .to_string();
        let want: crate::event::Event = serde_json::from_str(&first_line).unwrap();
        assert_eq!(first_event_id(&tmp), Some(want.id));
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn a_second_writer_waits_for_the_lock_and_then_gives_up() {
        let tmp = std::env::temp_dir().join(format!("vivac-lock-{}", id::ulid()));
        fs::create_dir_all(&tmp).unwrap();
        Store::create(&tmp).unwrap();
        let s = Store::open(tmp.clone()).unwrap();
        let held = s.lock_for_write().unwrap();
        let second = lock_with_deadline(&s.lock_path(), std::time::Duration::from_millis(200));
        assert!(
            matches!(second, Err(Failure::Busy(_))),
            "the lock let a second writer in"
        );
        drop(held);
        assert!(
            lock_with_deadline(&s.lock_path(), std::time::Duration::from_millis(200)).is_ok(),
            "dropping the first lock did not release it"
        );
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn the_busy_failure_names_the_deadline_it_was_given() {
        let f = Failure::busy(std::time::Duration::from_secs(5));
        assert_eq!(f.code(), 5);
        assert!(
            f.message().contains("held this tree for 5 seconds"),
            "{}",
            f.message()
        );
    }

    #[test]
    fn append_never_recreates_a_log_that_vanished() {
        let tmp = std::env::temp_dir().join(format!("vivac-vanished-{}", id::ulid()));
        fs::create_dir_all(&tmp).unwrap();
        Store::create(&tmp).unwrap();
        let mut s = Store::open(tmp.clone()).unwrap();
        let lock = s.lock_for_write().unwrap();
        fs::remove_file(s.log()).unwrap();
        let body = vec![crate::event::Body::NodeNoted {
            node: "01VANISHEDAAAAAAAAAAAAAAAA".into(),
            note: "x".into(),
        }];
        assert!(
            s.append(&lock, crate::lane::MAIN, body, 0, false).is_err(),
            "append wrote into a log that is gone"
        );
        assert!(
            !s.log().exists(),
            "append created a new log where the old one was"
        );
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn appending_with_another_trees_lock_is_refused() {
        let a = std::env::temp_dir().join(format!("vivac-locka-{}", id::ulid()));
        let b = std::env::temp_dir().join(format!("vivac-lockb-{}", id::ulid()));
        Store::create(&a).unwrap();
        Store::create(&b).unwrap();
        let mut sa = Store::open(a.clone()).unwrap();
        let sb = Store::open(b.clone()).unwrap();
        let wrong = sb.lock_for_write().unwrap();
        let body = vec![crate::event::Body::NodeNoted {
            node: "t1".into(),
            note: "x".into(),
        }];
        assert!(
            sa.append(&wrong, crate::lane::MAIN, body, 0, false)
                .is_err(),
            "append accepted a lock taken on a different tree"
        );
        fs::remove_dir_all(&a).ok();
        fs::remove_dir_all(&b).ok();
    }

    fn locate_tmp(prefix: &str) -> PathBuf {
        std::env::temp_dir().join(format!("vivac-locate-{prefix}-{}", id::ulid()))
    }

    /// Writes the `.git` file a linked worktree and a submodule both carry:
    /// a file, at `working_dir`, naming a `gitdir` elsewhere.
    fn write_git_file(working_dir: &Path, gitdir: &Path) {
        fs::create_dir_all(working_dir).unwrap();
        fs::create_dir_all(gitdir).unwrap();
        fs::write(
            working_dir.join(".git"),
            format!("gitdir: {}\n", gitdir.display()),
        )
        .unwrap();
    }

    // `locate_from(_, None)`, not `locate`, in every test below that has
    // no real use for the registry: `locate` reads it from `VIVAC_HOME`,
    // and this whole module's own unit tests run under `cfg(test)`, where
    // `store_dir` now refuses to answer with this machine's real home at
    // all -- `t594`, the same reason it refuses for a unit test
    // anywhere in the crate. `None` says outright that these are testing
    // the walk itself, not the fallback.
    #[test]
    fn the_tree_in_this_very_folder() {
        let tmp = locate_tmp("here");
        Store::create(&tmp).unwrap();
        let located = locate_from(&tmp, None).unwrap().unwrap();
        assert_eq!(located.root, tmp);
        assert_eq!(located.lane_dir, tmp);
        assert!(located.lane.is_none());
        assert!(located.worktree.is_none());
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn an_empty_vivac_directory_with_nothing_above_answers_as_no_store() {
        // `f566` first drew this shape -- a `.vivac/` with no log and no
        // config, what a half-finished delete leaves behind -- and made it
        // resolve to itself, worried that walking past it would quietly
        // move somebody's work to another tree. `f719` measured the
        // mirror shape: the very same bytes are also what an undone join
        // leaves behind, and *that* folder belongs to the tree above, not
        // to itself. The two cannot be told apart by what is on disk, so
        // neither guess is made any more (`hollow_vivac_refusal` is where
        // the guess would live, and it only fires with a real tree above
        // to be confused with). With nothing above to confuse it with,
        // there is nothing to guess: this is the ordinary "no store"
        // answer, the same one an empty `.vivac/` already gets
        // everywhere else the walk finds no map at all. `f566`'s own
        // outcome is unaffected: `init` never calls this walk, checks
        // only `cwd` itself, and plants a tree over an empty `.vivac/`
        // exactly as it always has.
        let tmp = locate_tmp("empty-vivac");
        fs::create_dir_all(tmp.join(DIR)).unwrap();
        assert!(locate_from(&tmp, None).unwrap().is_none());
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn an_empty_vivac_directory_under_a_real_tree_refuses_rather_than_guessing() {
        // `f719`'s third case: which of `f566`'s worry and `f719`'s own
        // measurement this shape is cannot be read off the filesystem, so
        // resolving it either way -- to itself, or to the tree above --
        // would be a guess dressed as an answer.
        let tmp = locate_tmp("hollow-under-tree");
        let hollow = tmp.join("hollow");
        Store::create(&tmp).unwrap();
        fs::create_dir_all(hollow.join(DIR)).unwrap();
        let err = locate_from(&hollow, None).unwrap_err();
        assert_eq!(err.code(), 1);
        assert!(
            err.message()
                .contains("This folder has a .vivac/ that is neither a tree nor a lane"),
            "{}",
            err.message()
        );
        assert!(err.message().contains("vivac init"), "{}", err.message());
        fs::remove_dir_all(&tmp).ok();
    }

    /// `f720`, measured a second time (`hollow_vivac_refusal`'s doc comment
    /// says where): a folder name has no bound of its own, and the text
    /// this refusal used to splice it into was split by hand at columns
    /// that only ever fit the one name it was tried against -- anything
    /// past nine characters already ran past 76. This one runs well over a
    /// hundred, in the neighbourhood of `t640`'s own `NAME_MAX_LEN` --
    /// the longest a name in this product is ever let to run -- built out
    /// of repeated words rather than typed by hand, so nobody shortens it
    /// for the sake of a tidier diff: the point is that `wrap` needs no
    /// shortening to hold it, and a real name carries spaces to break on
    /// the same as this one does.
    #[test]
    fn hollow_vivac_refusal_never_prints_a_line_past_76_columns() {
        let long_name = "folder name ".repeat(9).trim().to_string();
        let tree_above = PathBuf::from("/tmp").join(&long_name);
        let message = hollow_vivac_refusal(&tree_above).message();
        // The name's own spaces are exactly where `wrap` may break a
        // line, the same as any other run of words, so this checks the
        // name arrived whole with the wrapping folded back out rather
        // than demanding it landed on one physical line.
        let unwrapped: String = message.split_whitespace().collect::<Vec<_>>().join(" ");
        assert!(unwrapped.contains(&long_name), "{message}");
        for line in message.lines() {
            let trimmed = line.trim_start();
            // The two ways out, one of them a command: whole on purpose,
            // the same exemption `sub_line` already has (`f720`).
            let is_a_way_out = trimmed.starts_with("Make it a tree of its own")
                || trimmed.starts_with("Or hand it back to the tree above");
            assert!(
                is_a_way_out || line.chars().count() <= 76,
                "a wrapped line ran past 76 columns ({} chars): {line:?}\nfull message:\n{message}",
                line.chars().count()
            );
        }
    }

    #[test]
    fn a_lane_below_the_tree_finds_it_walking_up() {
        let tmp = locate_tmp("below");
        let mut s = Store::create(&tmp).unwrap();
        let lock = s.lock_for_write().unwrap();
        s.append(
            &lock,
            crate::lane::MAIN,
            vec![crate::event::Body::NodeNoted {
                node: "t1".into(),
                note: "seed".into(),
            }],
            0,
            false,
        )
        .unwrap();
        drop(lock);
        let project = first_event_id(&tmp).unwrap();
        let lane_dir = tmp.join("lane");
        let lane = crate::lane::Lane {
            version: 1,
            id: crate::lane::new_id(),
            project: project.clone(),
        };
        crate::lane::write(&lane_dir.join(DIR), &lane).unwrap();

        let located = locate_from(&lane_dir, None).unwrap().unwrap();
        assert_eq!(located.root, tmp);
        assert_eq!(located.lane_dir, lane_dir);
        assert_eq!(located.lane.unwrap().project, project);
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn a_lane_outside_the_tree_finds_it_through_the_registry() {
        let lane_dir = locate_tmp("outside");
        let lane = crate::lane::Lane {
            version: 1,
            id: crate::lane::new_id(),
            project: "01OUTSIDEPROJECTAAAAAAAAAA".into(),
        };
        crate::lane::write(&lane_dir.join(DIR), &lane).unwrap();

        let registry_dir = locate_tmp("outside-registry");
        let noted_root = locate_tmp("outside-fake-root");
        crate::registry::note(
            &registry_dir,
            &lane.project,
            crate::registry::Sighting {
                root: &noted_root,
                lane: None,
                repos: None,
            },
        );

        let located = locate_from(&lane_dir, Some(&registry_dir))
            .unwrap()
            .unwrap();
        assert_eq!(located.root, noted_root);
        assert_eq!(located.lane_dir, lane_dir);

        fs::remove_dir_all(&lane_dir).ok();
        fs::remove_dir_all(&registry_dir).ok();
    }

    /// `t594`: a folder can carry both a lane file and a stray
    /// `config` of its own without the two agreeing. `Store::open` writes
    /// a fresh, empty config the moment `events` is missing where `config`
    /// is not -- no crash needed, a plain read does it -- and this is
    /// exactly the state `relocate` can leave a reader looking at between
    /// renaming the origin's own `config` away and its `events`. The
    /// orphaned config here carries no events at all, so its own first
    /// event id is `None`, and `None` never equals the lane's `project`.
    #[test]
    fn a_lane_folder_with_an_orphaned_config_is_not_read_as_its_own_tree() {
        let lane_dir = locate_tmp("orphan-config");
        let lane = crate::lane::Lane {
            version: 1,
            id: crate::lane::new_id(),
            project: "01ORPHANPROJECTAAAAAAAAAA".into(),
        };
        crate::lane::write(&lane_dir.join(DIR), &lane).unwrap();
        // The orphaned config: `Store::open` regenerates one the moment it
        // finds this folder's own `.vivac/` with no `config` in it, and
        // the fresh one it mints carries a project id of its own, never
        // `lane.project`.
        Store::open(lane_dir.clone()).unwrap();
        assert!(
            lane_dir.join(DIR).join(CONFIG).is_file(),
            "the setup itself must have regenerated a config here"
        );
        assert!(!lane_dir.join(DIR).join(LOG).is_file());

        let registry_dir = locate_tmp("orphan-config-registry");
        let noted_root = locate_tmp("orphan-config-fake-root");
        crate::registry::note(
            &registry_dir,
            &lane.project,
            crate::registry::Sighting {
                root: &noted_root,
                lane: None,
                repos: None,
            },
        );

        let located = locate_from(&lane_dir, Some(&registry_dir))
            .unwrap()
            .unwrap();
        assert_eq!(
            located.root, noted_root,
            "an orphaned config must not make this folder answer as its own tree"
        );
        assert_eq!(located.lane_dir, lane_dir);

        fs::remove_dir_all(&lane_dir).ok();
        fs::remove_dir_all(&registry_dir).ok();
    }

    #[test]
    fn a_lane_whose_tree_the_registry_does_not_know_refuses_with_exit_4() {
        let lane_dir = locate_tmp("unknown");
        let lane = crate::lane::Lane {
            version: 1,
            id: crate::lane::new_id(),
            project: "01UNKNOWNPROJECTAAAAAAAAAA".into(),
        };
        crate::lane::write(&lane_dir.join(DIR), &lane).unwrap();
        let registry_dir = locate_tmp("unknown-registry");

        let err = locate_from(&lane_dir, Some(&registry_dir)).unwrap_err();
        assert_eq!(err.code(), 4);
        assert!(
            err.message().contains("registry does not know"),
            "{}",
            err.message()
        );
        fs::remove_dir_all(&lane_dir).ok();
    }

    #[test]
    fn a_subfolder_belongs_to_the_nearest_lane_above() {
        let outer = locate_tmp("nearest-outer");
        Store::create(&outer).unwrap(); // a distractor tree, further up
        let lane_dir = outer.join("consumer");
        let lane = crate::lane::Lane {
            version: 1,
            id: crate::lane::new_id(),
            project: "01NEARESTPROJECTAAAAAAAAAA".into(),
        };
        crate::lane::write(&lane_dir.join(DIR), &lane).unwrap();
        let deep = lane_dir.join("x").join("y");
        fs::create_dir_all(&deep).unwrap();

        let registry_dir = locate_tmp("nearest-registry");
        let noted_root = locate_tmp("nearest-fake-root");
        crate::registry::note(
            &registry_dir,
            &lane.project,
            crate::registry::Sighting {
                root: &noted_root,
                lane: None,
                repos: None,
            },
        );

        let located = locate_from(&deep, Some(&registry_dir)).unwrap().unwrap();
        assert_eq!(
            located.lane_dir, lane_dir,
            "picked a farther .vivac/ than the nearest one"
        );
        assert_eq!(located.root, noted_root);

        fs::remove_dir_all(&outer).ok();
        fs::remove_dir_all(&registry_dir).ok();
    }

    #[test]
    fn a_linked_worktree_inside_the_lane_is_reported_as_a_worktree() {
        let tmp = locate_tmp("wt-inside");
        let worktree_dir = tmp.join("feature");
        let gitdir = tmp
            .join("main")
            .join(".git")
            .join("worktrees")
            .join("feature");
        write_git_file(&worktree_dir, &gitdir);
        fs::write(gitdir.join("commondir"), "../..\n").unwrap();

        let lane = crate::lane::Lane {
            version: 1,
            id: crate::lane::new_id(),
            project: "01WTINSIDEPROJECTAAAAAAAAA".into(),
        };
        crate::lane::write(&worktree_dir.join(DIR), &lane).unwrap();

        let registry_dir = locate_tmp("wt-inside-registry");
        let noted_root = locate_tmp("wt-inside-fake-root");
        crate::registry::note(
            &registry_dir,
            &lane.project,
            crate::registry::Sighting {
                root: &noted_root,
                lane: None,
                repos: None,
            },
        );

        let deep = worktree_dir.join("src").join("deep");
        fs::create_dir_all(&deep).unwrap();

        let located = locate_from(&deep, Some(&registry_dir)).unwrap().unwrap();
        assert_eq!(located.root, noted_root);
        assert_eq!(
            located.lane_dir, worktree_dir,
            "who the lane really is gets decided elsewhere, not here"
        );
        assert_eq!(located.worktree, Some(worktree_dir.clone()));

        fs::remove_dir_all(&tmp).ok();
        fs::remove_dir_all(&registry_dir).ok();
    }

    #[test]
    fn a_linked_worktree_outside_any_lane_finds_the_tree_through_its_main_copy() {
        let tmp = locate_tmp("wt-outside");
        let main_dir = tmp.join("main");
        Store::create(&main_dir).unwrap();
        let worktree_dir = tmp.join("feature");
        let gitdir = main_dir.join(".git").join("worktrees").join("feature");
        write_git_file(&worktree_dir, &gitdir);
        fs::write(gitdir.join("commondir"), "../..\n").unwrap();

        let located = locate_from(&worktree_dir, None).unwrap().unwrap();
        assert_eq!(located.root, main_dir);
        assert_eq!(located.lane_dir, main_dir);
        assert!(located.lane.is_none());
        assert_eq!(located.worktree, Some(worktree_dir));

        fs::remove_dir_all(&tmp).ok();
    }

    /// `t594`: once the folder inside a linked worktree
    /// carries its own `.vivac/lane`, resolution takes `resolve_lane`
    /// rather than the plain "no lane" fallback the test above exercises
    /// -- and until this fix, that path never tried the worktree's main
    /// copy at all, so a lane file with no registry entry to back it up
    /// used to leave the folder unable to find its own tree.
    #[test]
    fn a_lane_file_inside_a_linked_worktree_still_resolves_through_its_main_copy() {
        let tmp = locate_tmp("wt-lane-fallback");
        let main_dir = tmp.join("main");
        let mut s = Store::create(&main_dir).unwrap();
        let lock = s.lock_for_write().unwrap();
        s.append(
            &lock,
            crate::lane::MAIN,
            vec![crate::event::Body::NodeNoted {
                node: "t1".into(),
                note: "seed".into(),
            }],
            0,
            false,
        )
        .unwrap();
        drop(lock);
        let project = first_event_id(&main_dir).unwrap();

        let worktree_dir = tmp.join("feature");
        let gitdir = main_dir.join(".git").join("worktrees").join("feature");
        write_git_file(&worktree_dir, &gitdir);
        fs::write(gitdir.join("commondir"), "../..\n").unwrap();

        let lane = crate::lane::Lane {
            version: 1,
            id: crate::lane::new_id(),
            project: project.clone(),
        };
        crate::lane::write(&worktree_dir.join(DIR), &lane).unwrap();

        // No registry at all: the only path left back to the tree is the
        // main-copy retry inside `resolve_lane` itself.
        let located = locate_from(&worktree_dir, None).unwrap().unwrap();
        assert_eq!(located.root, main_dir);
        assert_eq!(located.lane_dir, worktree_dir);
        assert_eq!(located.lane.unwrap().project, project);

        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn a_submodule_is_not_a_worktree() {
        let tmp = locate_tmp("submodule");
        Store::create(&tmp).unwrap();
        let sub_dir = tmp.join("vendor").join("lib");
        let gitdir = tmp.join(".git-modules").join("lib");
        write_git_file(&sub_dir, &gitdir);
        // No `commondir` written: a submodule owns its own repository.

        let located = locate_from(&sub_dir, None).unwrap().unwrap();
        assert_eq!(located.root, tmp);
        assert!(
            located.worktree.is_none(),
            "a submodule was reported as a linked worktree"
        );

        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn the_global_store_still_does_not_answer_the_walk() {
        let tmp = locate_tmp("global");
        let deep = tmp.join("a").join("b");
        fs::create_dir_all(&deep).unwrap();
        Store::create(&tmp).unwrap();
        crate::registry::note(
            &tmp.join(DIR),
            "01aaaaaaaaaaaaaaaaaaaaaaaa",
            crate::registry::Sighting {
                root: &deep,
                lane: None,
                repos: None,
            },
        );
        assert!(locate_from(&deep, None).unwrap().is_none());
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn an_event_is_signed_by_whatever_lane_append_is_given() {
        // `f608`, third time: the lane used to live on the `Store` itself
        // (`with_lane`), and a `Store` freshly opened by `refold` carried
        // `main` again no matter which lane the `Ctx` around it answered
        // as. It is an argument now, exactly like the write lock, so
        // there is nothing left on `Store` for a caller to leave stale.
        let tmp = std::env::temp_dir().join(format!("vivac-sign-{}", id::ulid()));
        Store::create(&tmp).unwrap();
        let mut s = Store::open(tmp.clone()).unwrap();
        let lock = s.lock_for_write().unwrap();
        let w = s
            .append(
                &lock,
                "01M2XYZ",
                vec![crate::event::Body::NodeNoted {
                    node: "t1".into(),
                    note: "x".into(),
                }],
                0,
                false,
            )
            .unwrap();
        assert_eq!(w.events[0].lane, "01M2XYZ");
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn a_caller_that_passes_main_signs_main() {
        // Every tree that exists today, and every tree where nobody has run
        // setup: the log has to stay byte for byte what 0.11 wrote, and
        // `Ctx::emit` is the caller that keeps that true by passing
        // `lane::MAIN` on its own once `self.lane` is `None`.
        let tmp = std::env::temp_dir().join(format!("vivac-signmain-{}", id::ulid()));
        Store::create(&tmp).unwrap();
        let mut s = Store::open(tmp.clone()).unwrap();
        let lock = s.lock_for_write().unwrap();
        let w = s
            .append(
                &lock,
                crate::lane::MAIN,
                vec![crate::event::Body::NodeNoted {
                    node: "t1".into(),
                    note: "x".into(),
                }],
                0,
                false,
            )
            .unwrap();
        assert_eq!(w.events[0].lane, crate::lane::MAIN);
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn locking_the_config_for_lanes_is_idempotent_and_atomic() {
        // Same mechanism as `d444`'s own sentence: the config is written to a
        // sibling and renamed, and saying it twice writes once.
        let tmp = std::env::temp_dir().join(format!("vivac-lanelock-{}", id::ulid()));
        let mut s = Store::create(&tmp).unwrap();
        let lock = s.lock_for_write().unwrap();
        s.lock_lanes_in_config(&lock).unwrap();
        assert_eq!(s.config.version, ConfigVersion::Lanes);
        let text = fs::read_to_string(tmp.join(DIR).join(CONFIG)).unwrap();
        assert!(text.contains(LANE_SENTENCE));

        s.lock_lanes_in_config(&lock).unwrap();
        assert_eq!(s.config.version, ConfigVersion::Lanes);
        assert!(
            fs::read_dir(tmp.join(DIR))
                .unwrap()
                .filter_map(|e| e.ok())
                .all(|e| !e.file_name().to_string_lossy().ends_with(".tmp")),
            "a temporary file was left behind"
        );
        fs::remove_dir_all(&tmp).ok();
    }

    /// `t594`: a config that vanishes over a log that
    /// already holds a `lane.declared` must come back locked to the lanes
    /// sentence, the same as `d444` already does for a pillar or a rule --
    /// `log_already_governed`'s blind spot before this test existed.
    #[test]
    fn a_missing_config_regenerates_the_lanes_sentence_when_the_log_has_a_lane_event() {
        let tmp = std::env::temp_dir().join(format!("vivac-relock-{}", id::ulid()));
        let mut s = Store::create(&tmp).unwrap();
        let lock = s.lock_for_write().unwrap();
        s.append(
            &lock,
            crate::lane::MAIN,
            vec![crate::event::Body::LaneDeclared {
                lane: crate::lane::MAIN.to_string(),
                name: crate::lane::MAIN.to_string(),
                repos: vec![],
            }],
            0,
            false,
        )
        .unwrap();
        drop(lock);
        fs::remove_file(tmp.join(DIR).join(CONFIG)).unwrap();

        let reopened = Store::open(tmp.clone()).unwrap();
        assert_eq!(reopened.config.version, ConfigVersion::Lanes);
        fs::remove_dir_all(&tmp).ok();
    }

    #[test]
    fn build_fresh_writes_everything_fill_wrote_and_leaves_no_temp_sibling() {
        let root = std::env::temp_dir().join(format!("vivac-fresh-{}", id::ulid()));
        let dir = root.join(DIR);
        assert!(
            !dir.exists(),
            "the fixture must start with no .vivac/ at all"
        );
        let built = build_fresh(&dir, |tmp_dir| fs::write(tmp_dir.join("marker"), b"x")).unwrap();
        assert!(built);
        assert!(dir.join("marker").is_file());
        let siblings: Vec<String> = fs::read_dir(&root)
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| e.file_name().to_string_lossy().into_owned())
            .collect();
        assert_eq!(
            siblings,
            vec![DIR.to_string()],
            "a temporary sibling directory was left behind: {siblings:?}"
        );
        fs::remove_dir_all(&root).ok();
    }

    #[test]
    fn build_fresh_falls_back_without_running_fill_when_the_directory_is_already_there() {
        // `f566`, `d734`: an empty or hollow `.vivac/` that is already
        // there is planted in place, not through this helper -- there is
        // no fresh appearance left to make atomic. `fill` never runs at
        // all in that case, which this pins with a flag rather than
        // trusting the return value alone.
        let root = std::env::temp_dir().join(format!("vivac-exists-{}", id::ulid()));
        let dir = root.join(DIR);
        fs::create_dir_all(&dir).unwrap();
        let ran = std::cell::Cell::new(false);
        let built = build_fresh(&dir, |_tmp_dir| {
            ran.set(true);
            Ok(())
        })
        .unwrap();
        assert!(!built);
        assert!(!ran.get(), "fill ran even though vivac_dir already existed");
        fs::remove_dir_all(&root).ok();
    }

    /// The other half of `f735`: `Store::create` used to `create_dir_all(&d)`
    /// and only *afterwards* write `config`, `events` and `.gitignore`, so a
    /// reader resolving this same folder mid-plant could see the identical
    /// hollow shape `lane::write` could. Same proof, same shape: a reader
    /// spins in a tight loop against a writer planting a tree into a folder
    /// that does not exist yet, round after round. `already_planted` is
    /// enough to tell hollow apart here -- `Store::create` never writes a
    /// `lane` file, so `find_root`'s other way out of "hollow" never applies.
    #[test]
    fn a_reader_racing_a_fresh_plant_never_sees_vivac_dir_hollow() {
        const ROUNDS: usize = 500;
        for round in 0..ROUNDS {
            let root = std::env::temp_dir().join(format!("vivac-plant-race-{}", id::ulid()));
            let dir = root.join(DIR);
            let hollow_seen = std::sync::atomic::AtomicBool::new(false);
            let stop = std::sync::atomic::AtomicBool::new(false);
            std::thread::scope(|s| {
                s.spawn(|| {
                    while !stop.load(std::sync::atomic::Ordering::Relaxed) {
                        if dir.is_dir() && !already_planted(&root) {
                            hollow_seen.store(true, std::sync::atomic::Ordering::Relaxed);
                            break;
                        }
                    }
                });
                Store::create(&root).unwrap();
                stop.store(true, std::sync::atomic::Ordering::Relaxed);
            });
            fs::remove_dir_all(&root).ok();
            assert!(
                !hollow_seen.load(std::sync::atomic::Ordering::Relaxed),
                "a reader observed a hollow .vivac/ on round {round} of {ROUNDS}"
            );
        }
    }
}