slipcase_open/

flow.rs

//! Concept 5's steps, joined into a session.
//
// Author: David M. Anderson
// Built with AI assistance (Claude, Anthropic)
//
//! Open and validate, decide, extract, mark, launch, watch, write back, close.
//! Everything security-relevant that this tool does happens on the path through
//! [`open`], which is what concept 8 means by the engine being one body of code
//! on three platforms.
//!
//! **The policy check is here and immediately before the launch.** Concept 10
//! says enforcement lives in the launch path: a value read at startup, held
//! across a policy push, or handed in over IPC is a bypass. So [`open`] resolves
//! policy itself, from sources it is given rather than from an answer somebody
//! else computed, and nothing between that decision and the launch can change
//! what runs.

use std::fmt;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use crate::outside::Outside;
use crate::policy::{self, Decision};
use crate::session::{self, Session};
use crate::watch::{Change, Watch};
use crate::{content, extract, recover, writeback};

/// Why a container did not open.
#[derive(Debug)]
pub enum Error {
    /// It is not a container, or not one this build can read.
    Container(slpc::Error),
    /// The payload is a program wearing a document's name, so it was not
    /// opened. Concept 5.1: the one content check there is, and the only thing
    /// it can do is refuse something policy had already allowed.
    Misrepresented(content::Executable),
    /// Policy will not have it opened. Carries the decision, so the refusal can
    /// say which of the several reasons applies.
    Refused(Decision),
    /// Policy could not be established. Distinct from a refusal: nothing has
    /// decided that this payload may not be opened, and the remedy is to fix
    /// the source rather than to change the lists.
    Policy(policy::Error),
    /// The session directory could not be made.
    Session(std::io::Error),
    /// The payload did not reach the session directory.
    Extract(extract::Error),
    /// The desktop would not open it.
    Launch(std::io::Error),
    /// The payload directory could not be watched. Fatal rather than
    /// degraded: concept 6 already concedes that detection is unreliable, and a
    /// session with no watch at all would write back only at close while
    /// looking like one that writes back on every save.
    Watch(notify::Error),
}

impl fmt::Display for Error {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Container(e) => write!(f, "{e}"),
            Self::Misrepresented(what) => write!(
                f,
                "the payload is {}, not the document its name claims, so it was not opened",
                what.describes()
            ),
            Self::Refused(d) => match d {
                Decision::Denied { key } => write!(f, "{key} is on the deny list"),
                Decision::NotPermitted { key } => write!(f, "{key} is not in the allowed set"),
                Decision::NoUsableExtension => write!(
                    f,
                    "the payload has no usable extension, so the desktop would ask which \
                     application to run it with"
                ),
                Decision::Open { .. } => write!(f, "permitted"),
            },
            Self::Policy(e) => write!(f, "policy could not be read: {e}"),
            Self::Session(e) => write!(f, "the session could not be started: {e}"),
            Self::Extract(e) => write!(f, "{e}"),
            Self::Launch(e) => write!(f, "the payload could not be opened: {e}"),
            Self::Watch(e) => write!(f, "the payload directory could not be watched: {e}"),
        }
    }
}

impl std::error::Error for Error {}

/// A session that is open, with its payload launched and its directory watched.
pub struct Opened {
    session: Session,
    watch: Watch,
    /// What the platform recorded about where the container came from, carried
    /// onto the payload.
    pub mark: slpc::provenance::Mark,
    saw_payload_change: bool,
}

/// What closing a session did.
pub enum Closed {
    /// Written back where asked, and the session directory removed.
    Cleared,
    /// The target application still has things of its own in the payload
    /// directory, so the session was handed to recovery instead of being
    /// removed. Concept 6.2: the close is honoured, but deleting the directory
    /// underneath a running editor sends its next save nowhere this tool will
    /// ever look.
    ///
    /// The watch comes with it. Concept 8: what a resident process is good for
    /// on this path is noticing the application's last save when it happens,
    /// rather than leaving the question until somebody next opens a container.
    ///
    /// Boxed because of what the watch weighs on macOS. The first time the
    /// gate ran on a Mac, 2026-09-07, clippy refused this enum for a variant
    /// carrying nothing beside one carrying 256 bytes, and `size_of` put the
    /// numbers on it there: `Lingering` 256, of which `Watch` is 144, of which
    /// the platform's watcher is 128. A close is not a hot path, so the
    /// indirection costs nothing anybody would notice.
    LeftForRecovery(Box<Lingering>),
}

/// A closed session the target application has not finished with, still
/// watched.
///
/// **Nothing here writes back, and that is concept 6.3 rather than an
/// omission.** The session is closed; a save arriving now is one this tool was
/// not watching for when the user said they were done, and it cannot tell a
/// complete save from a half-written one. So the watch is used to know when the
/// application has stopped, which is the moment at which asking is worth
/// anything, and the answer comes from the person.
pub struct Lingering {
    session: Session,
    watch: Watch,
    quiet_since: Instant,
}

impl Lingering {
    /// The session on disk.
    #[must_use]
    pub fn session(&self) -> &Session {
        &self.session
    }

    /// Whether the application appears to have finished: nothing of its own
    /// left in the payload directory (concept 6.1), and nothing written there
    /// for `quiet`.
    ///
    /// **Both halves, because either alone is wrong.** Siblings gone is the
    /// signal concept 6.1 settles on, and it says the application cleaned up;
    /// it says nothing about a save still being flushed. A quiet period alone
    /// would fire in the middle of somebody's afternoon, between two edits.
    ///
    /// Takes `&mut self` because asking is also draining: a change seen here is
    /// what resets the quiet period.
    pub fn has_settled(&mut self, quiet: Duration) -> bool {
        if self.watch.drain().next().is_some() {
            self.quiet_since = Instant::now();
        }
        if self.quiet_since.elapsed() < quiet {
            return false;
        }
        // Unreadable means the answer is not known, and the safe reading of not
        // known is that the application is still there. A directory that has
        // gone is the other case and settles: there is nothing left to wait
        // for, and what remains is a recovery record naming a payload that is
        // not on disk, which `recover` reports.
        !crate::watch::siblings_present(&self.session.payload_dir(), &self.session.record().payload)
            .unwrap_or(true)
    }

    /// Give up the watch and hand back the session, for the caller that is
    /// about to act on it.
    #[must_use]
    pub fn into_session(self) -> Session {
        self.session
    }
}

/// Open a container: concept 5, steps 1 through 7.
///
/// # Errors
///
/// See [`Error`]. Nothing is left behind on any of them except
/// [`Error::Extract`] carrying [`extract::Error::Unmarked`] with
/// `payload_removed` false, which says so.
pub fn open(root: &Path, container_path: &Path, outside: &Outside<'_>) -> Result<Opened, Error> {
    // Step 1. Opening is validating: `Container::open` applies SPEC 3 and the
    // limits SPEC 6 asks for before it will answer any question about the file.
    let mut container = slpc::Container::open(container_path).map_err(Error::Container)?;

    // Step 2, and step 3's refusals. Resolved here rather than passed in.
    let decision =
        policy::decide(outside.policy, container.payload_name()).map_err(Error::Policy)?;
    if !matches!(decision, Decision::Open { .. }) {
        return Err(Error::Refused(decision));
    }

    // Concept 5.1's content check, and it refuses.
    //
    // **A veto, not a control, and the distinction is what keeps 5.1's argument
    // standing.** The extension still decides what may be opened — the
    // allowlist above is the control, this admits nothing, and a payload that
    // gets past here has been permitted by policy and not by inspection. All
    // this can do is say *no* to something already permitted. 5.1's reasoning
    // about why sniffing cannot be the control is untouched; what changed is
    // the last line of it, which had this telling the person and standing
    // aside.
    //
    // **Before the session, so nothing reaches the disk.** The bytes are read
    // out of the container, so a refusal here means the executable was never
    // written anywhere outside it — no session directory, no payload file, no
    // mark, and nothing for a later sweep to find. That is worth more than the
    // warning it replaces.
    if let Some(what) = misrepresentation(&mut container, &decision) {
        return Err(Error::Misrepresented(what));
    }

    // Step 4.
    let mut session =
        session::create(root, container_path, container.payload_name()).map_err(Error::Session)?;

    // Steps 5 and 6. A failure here takes the session directory with it rather
    // than leaving a half-made one for recovery to ask about.
    let mark = match extract::extract(&mut container, &mut session) {
        Ok(m) => m,
        Err(e) => {
            let payload = session.payload_path();
            let _ = session.clone().remove();
            // `extract` reports whether *it* managed to take the ungated
            // payload back off disk, and then this removes the whole session
            // directory, which usually succeeds where the single unlink did
            // not. Left alone, the message tells somebody there is an ungated
            // executable on disk after the file has gone. Re-asked of the
            // filesystem, after the cleanup, so the sentence is true when it is
            // printed.
            return Err(Error::Extract(match e {
                extract::Error::Unmarked { cause, .. } => extract::Error::Unmarked {
                    cause,
                    payload_removed: !payload.exists(),
                },
                other => other,
            }));
        }
    };

    // Step 8 before step 7: the watch is registered before the application is
    // told the file exists, or a save that arrives quickly enough is a save
    // nothing was listening for.
    let watch = match Watch::on(&session.payload_dir(), &session.record().payload) {
        Ok(w) => w,
        Err(e) => {
            let _ = session.clone().remove();
            return Err(Error::Watch(e));
        }
    };

    if let Err(e) = outside.launcher.launch(&session.payload_path()) {
        let _ = session.clone().remove();
        return Err(Error::Launch(e));
    }

    Ok(Opened {
        session,
        watch,
        mark,
        saw_payload_change: false,
    })
}

/// What concept 5.1's check makes of the payload, read out of the container
/// rather than off disk so the answer is available before anything is written.
fn misrepresentation<R: std::io::Read + std::io::Seek>(
    container: &mut slpc::Container<R>,
    decision: &Decision,
) -> Option<content::Executable> {
    let key = match decision {
        Decision::Open { key } => Some(key.as_str()),
        _ => None,
    };
    let mut head = [0u8; content::HEAD];
    let mut payload = container.payload().ok()?;
    let mut at = 0;
    while at < head.len() {
        match std::io::Read::read(&mut payload, &mut head[at..]) {
            Ok(0) | Err(_) => break,
            Ok(n) => at += n,
        }
    }
    content::misrepresents(&head[..at], key)
}

impl Opened {
    /// The session on disk.
    #[must_use]
    pub fn session(&self) -> &Session {
        &self.session
    }

    /// Where the payload was put.
    #[must_use]
    pub fn payload_path(&self) -> PathBuf {
        self.session.payload_path()
    }

    /// Whether the payload has been seen to change since the session opened.
    #[must_use]
    pub fn saw_a_change(&self) -> bool {
        self.saw_payload_change
    }

    /// Whether the target application has anything of its own in the payload
    /// directory (concept 6.1).
    ///
    /// # Errors
    ///
    /// Where the payload directory cannot be read.
    pub fn application_is_working(&self) -> std::io::Result<bool> {
        crate::watch::siblings_present(&self.session.payload_dir(), &self.session.record().payload)
    }

    /// Take whatever the watch has to say, and write back once if the payload
    /// was among it.
    ///
    /// Once, rather than once per event. A single save arrives as several
    /// events — a temporary sibling, a rename, a metadata touch — and repacking
    /// per event would rebuild the container three times to the same end.
    ///
    /// # Errors
    ///
    /// Where the write-back failed. The session stays open: concept 6.2 puts
    /// the close at the user's hand, and a failed save is a reason to tell them
    /// rather than to give up on the container.
    pub fn pump(&mut self) -> Result<bool, writeback::Error> {
        self.pump_including(None)
    }

    /// [`pump`](Self::pump), counting a change already taken off the channel.
    ///
    /// **A change that has been received is a change that has happened.**
    /// `wait_and_pump` blocks by taking one change off the channel, so passing
    /// it in here is what stops that one being dropped on the floor. No save is
    /// known to have been lost to the earlier version — every save measured
    /// emits more than one event, and the next drain collects the rest — but it
    /// relied on that being true of every application on three platforms, which
    /// is not a thing this code is in a position to know.
    fn pump_including(&mut self, first: Option<Change>) -> Result<bool, writeback::Error> {
        let mut payload_changed = first == Some(Change::Payload);
        for change in self.watch.drain() {
            if change == Change::Payload {
                payload_changed = true;
            }
        }
        if !payload_changed {
            return Ok(false);
        }
        self.saw_payload_change = true;
        self.save_if_changed()
    }

    /// Write the payload back, unless it already matches what the container
    /// holds.
    ///
    /// **Asked of the bytes rather than of the events.** One save arrives as
    /// several events — a temporary sibling, a rename, a metadata touch — and
    /// they do not reliably land in one drain, so counting events makes the
    /// number of repacks a function of how busy the machine is. A quiet period
    /// before repacking would trade that for latency on every save and still
    /// only make the guess better. `recover` answers the real question by
    /// comparing against the CRC-32 the container already records (concept
    /// 6.3), so a redundant event costs one comparison instead of one rebuild.
    ///
    /// **Only the two quiet states are silent.** An earlier version returned
    /// *nothing to do* for every state that was not `Edited`, which meant a
    /// container deleted or replaced underneath a live session stopped it
    /// saving without saying anything — the user edits, nothing is written, and
    /// no error appears. Those states go to the write-back to be refused and
    /// reported, which is where the refusal belongs anyway.
    ///
    /// # Errors
    ///
    /// Where the write-back failed, or cannot be attempted at all.
    pub fn save_if_changed(&mut self) -> Result<bool, writeback::Error> {
        match recover::state(&self.session) {
            // Nothing to write, and nothing wrong.
            recover::State::Unchanged | recover::State::NothingExtracted => Ok(false),
            // `Edited`, and every state that means this session can no longer
            // reach its container. `write_back` refuses the ones it must and
            // names the reason.
            _ => {
                writeback::write_back(&mut self.session)?;
                Ok(true)
            }
        }
    }

    /// Wait up to `within` for something to happen, then [`pump`](Self::pump).
    ///
    /// # Errors
    ///
    /// As [`pump`](Self::pump).
    pub fn wait_and_pump(&mut self, within: Duration) -> Result<bool, writeback::Error> {
        let first = self.watch.next_change(within);
        self.pump_including(first)
    }

    /// Close the session: catch up on the watch, then clean up.
    ///
    /// Concept 6.2's question — *write it back anyway?* — is the caller's, and
    /// so is the answer: it asks, and calls
    /// [`save_if_changed`](Self::save_if_changed) if the answer is yes. This
    /// used to take a `bool` and repack unconditionally on it, which rebuilt
    /// the container even when the payload matched it byte for byte, and
    /// rebuilt it twice when the final pump had just done so.
    ///
    /// # Errors
    ///
    /// Where the final catch-up write-back failed, in which case nothing is
    /// removed and the session stays recoverable.
    pub fn close(mut self) -> Result<Closed, writeback::Error> {
        // Anything the watch has not been asked about yet. A save arriving
        // between the last pump and the close is a save.
        self.pump()?;

        // Concept 6.2. The close is honoured either way; what changes is
        // whether the directory goes now or is handed to recovery, so that an
        // editor still holding the payload has somewhere for its next save to
        // land and the next launch asks about it.
        if self.application_is_working().unwrap_or(true) {
            return Ok(Closed::LeftForRecovery(Box::new(Lingering {
                session: self.session,
                watch: self.watch,
                quiet_since: Instant::now(),
            })));
        }
        // **The watch goes before the directory does, and the order is load
        // bearing.** `Watch::drop` waits for the platform to finish stopping
        // it, so by the line below there is no watcher left to collide with the
        // removal. Removing first, which is what this used to do, is what a
        // 300-second hang was caught doing; `docs/windows-save-test-hang.md`
        // has the stack.
        drop(self.watch);

        // A failure to remove leaves a session recovery will pick up, which is
        // the same outcome by another road and not worth a second error type.
        let _ = self.session.remove();
        Ok(Closed::Cleared)
    }
}

#[cfg(test)]
mod tests {
    use super::{open, Closed, Error};
    use crate::outside::Outside;
    use crate::platform::testing::Recording;
    use crate::policy::{Layer, Origin, Source};
    use crate::present::testing::Silent;
    use crate::writeback;
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::time::Duration;

    /// Says nothing at every layer, so the shipped set answers.
    struct Default_;
    impl Source for Default_ {
        fn layer(&self, _o: Origin) -> crate::policy::Read {
            Ok(None)
        }
    }

    /// Denies everything, for the refusal arms.
    struct DenyAll;
    impl Source for DenyAll {
        fn layer(&self, o: Origin) -> crate::policy::Read {
            Ok((o == Origin::MachinePolicy).then(|| Layer {
                allowed: Some(Vec::new()),
                ..Layer::default()
            }))
        }
    }

    fn container(at: &Path, name: &str, payload: &[u8]) -> PathBuf {
        let doc: slpc::toml_edit::DocumentMut =
            format!("slipcase_version = \"1.0\"\n\n[payload]\nfile = \"{name}\"\n")
                .parse()
                .unwrap();
        let path = at.join(format!("{name}.slpc"));
        slpc::pack_reader(name, payload, doc, fs::File::create(&path).unwrap()).unwrap();
        path
    }

    #[test]
    fn opening_extracts_launches_and_watches() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"%PDF first");
        let launcher = Recording::default();

        let o = open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)).unwrap();
        assert_eq!(launcher.launched(), [o.payload_path()]);
        assert_eq!(fs::read(o.payload_path()).unwrap(), b"%PDF first");
        assert!(!o.saw_a_change());
    }

    #[test]
    fn a_save_reaches_the_container_without_anybody_closing_the_session() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let launcher = Recording::default();

        let mut o = open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)).unwrap();

        // The way a serious editor saves: a temporary sibling renamed over the
        // target, which is the case a watch on the file would miss.
        let scratch = o.payload_path().with_extension("pdf.tmp");
        fs::write(&scratch, b"edited").unwrap();
        fs::rename(&scratch, o.payload_path()).unwrap();

        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while std::time::Instant::now() < deadline && !o.saw_a_change() {
            o.wait_and_pump(Duration::from_millis(250)).unwrap();
        }
        assert!(o.saw_a_change(), "the save never reached the session");

        let mut back = slpc::Container::open(&c).unwrap();
        let mut got = Vec::new();
        std::io::copy(&mut back.payload().unwrap(), &mut got).unwrap();
        assert_eq!(got, b"edited");
    }

    #[test]
    fn a_save_that_emits_one_event_still_reaches_the_container() {
        // Concept 6 is written about editors that save atomically, and the
        // tests followed it there. This is the other shape: a plain write in
        // place, which emits fewer events. It passes either side of the
        // `pump_including` change rather than pinning it — what it pins is that
        // the simple save works at all, which nothing else asserted.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let mut o = open(
            &root,
            &c,
            &Outside::new(&Default_, &Recording::default(), &Silent),
        )
        .unwrap();

        fs::write(o.payload_path(), b"edited in place").unwrap();

        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while std::time::Instant::now() < deadline && !o.saw_a_change() {
            o.wait_and_pump(Duration::from_millis(250)).unwrap();
        }
        assert!(o.saw_a_change(), "a single-event save was never noticed");

        let mut back = slpc::Container::open(&c).unwrap();
        let mut got = Vec::new();
        std::io::copy(&mut back.payload().unwrap(), &mut got).unwrap();
        assert_eq!(got, b"edited in place");
    }

    #[test]
    fn one_save_is_one_write_back() {
        // A repack costs a full rebuild of the container, so the number of
        // them a session performs should follow the edits and not the event
        // traffic. Counting events cannot give that: one save arrives as
        // several, they do not reliably land in one drain, and this test was
        // flaky under a loaded suite for exactly that reason before `pump`
        // compared the bytes instead.
        //
        // Counted rather than inspected, because every redundant repack writes
        // the same bytes — a test asserting the container's contents passes
        // whatever the count is.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let launcher = Recording::default();

        let mut o = open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)).unwrap();

        let scratch = o.payload_path().with_extension("pdf.tmp");
        fs::write(&scratch, b"edited").unwrap();
        fs::rename(&scratch, o.payload_path()).unwrap();

        // Pump well past the point where the save has landed, so every event
        // it produced has arrived and been acted on.
        let deadline = std::time::Instant::now() + Duration::from_secs(3);
        while std::time::Instant::now() < deadline {
            o.wait_and_pump(Duration::from_millis(100)).unwrap();
        }

        assert!(o.saw_a_change(), "the save never reached the session");
        assert_eq!(
            o.session().record().write_backs,
            1,
            "one save produced more than one write-back"
        );
    }

    #[test]
    fn policy_refuses_before_a_session_directory_exists() {
        // Concept 10 puts enforcement in the launch path, and a refusal that
        // had already written the payload somewhere would be a refusal in name.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let launcher = Recording::default();

        assert!(matches!(
            open(&root, &c, &Outside::new(&DenyAll, &launcher, &Silent)),
            Err(Error::Refused(_))
        ));
        assert!(launcher.launched().is_empty());
        assert!(crate::session::scan(&root).unwrap().is_empty());
    }

    #[test]
    fn a_payload_with_no_usable_extension_is_refused() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "README", b"hello");
        let launcher = Recording::default();

        match open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)) {
            Err(e) => assert!(e.to_string().contains("no usable extension"), "{e}"),
            Ok(_) => panic!("a payload with no usable extension was opened"),
        }
        assert!(crate::session::scan(&root).unwrap().is_empty());
    }

    #[test]
    fn an_executable_wearing_a_documents_name_is_refused() {
        // Concept 5.1's check, as a veto. It admits nothing — policy had
        // already allowed `.pdf` — and all it does here is say no to something
        // policy allowed.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "invoice.pdf", b"MZ\x90\x00 not a pdf");
        let launcher = Recording::default();

        match open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)) {
            Err(Error::Misrepresented(what)) => {
                assert_eq!(what, crate::content::Executable::Pe);
            }
            Err(e) => panic!("refused for the wrong reason: {e}"),
            Ok(_) => panic!("a program wearing a document's name was opened"),
        }
        assert!(
            launcher.launched().is_empty(),
            "nothing was handed to the desktop"
        );
        // The refusal is before the session, so the bytes never left the
        // container: no session directory, no payload on disk, and nothing for
        // a later sweep to find.
        assert!(
            !root.exists() || crate::session::scan(&root).unwrap().is_empty(),
            "the executable reached the disk"
        );
    }

    #[test]
    fn a_program_under_its_own_name_is_left_to_policy() {
        // The other half of *veto, not control*: this check never admits
        // anything and never fires on a payload that is what it says. What
        // happens to a `.exe` is the allowlist's business, and here nothing
        // stands in its way.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "setup.exe", b"MZ\x90\x00 an installer");
        let launcher = Recording::default();

        let opened = open(&root, &c, &Outside::new(&Default_, &launcher, &Silent));
        assert!(
            !matches!(opened, Err(Error::Misrepresented(_))),
            "the content check refused a payload that is what its name says"
        );
    }

    #[test]
    fn a_desktop_that_will_not_open_it_leaves_nothing_behind() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");

        assert!(matches!(
            open(
                &root,
                &c,
                &Outside::new(&Default_, &Recording::refusing(), &Silent)
            ),
            Err(Error::Launch(_))
        ));
        assert!(crate::session::scan(&root).unwrap().is_empty());
    }

    #[test]
    fn closing_without_a_change_can_still_write_back() {
        // The only available answer to Save As: no event fires when somebody
        // saves elsewhere, so a session that saw nothing may still have an edit
        // that belongs in the container.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let launcher = Recording::default();

        let mut o = open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)).unwrap();
        fs::write(o.payload_path(), b"edited quietly").unwrap();
        // Deliberately not pumped: this is the path where nothing was seen.
        assert!(o.save_if_changed().unwrap());
        assert!(matches!(o.close().unwrap(), Closed::Cleared));

        let mut back = slpc::Container::open(&c).unwrap();
        let mut got = Vec::new();
        std::io::copy(&mut back.payload().unwrap(), &mut got).unwrap();
        assert_eq!(got, b"edited quietly");
        assert!(crate::session::scan(&root).unwrap().is_empty());
    }

    #[test]
    fn closing_while_the_application_is_working_hands_over_to_recovery() {
        // Concept 6.2: the close is honoured, but removing the directory under
        // a running editor sends its next save nowhere this tool will look.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let launcher = Recording::default();

        // No edit here, deliberately. `close` pumps before it decides, so a
        // save made in this test may or may not have reached the container by
        // the time the state is read — asserting on that state made this fail
        // about one run in six. What the handover rule guarantees is that the
        // directory survives, and that is what is checked.
        let o = open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)).unwrap();
        let payload = o.payload_path();
        fs::write(payload.with_file_name("~$report.pdf"), b"").unwrap();

        assert!(matches!(o.close().unwrap(), Closed::LeftForRecovery(_)));

        let left = crate::session::scan(&root).unwrap();
        assert_eq!(left.len(), 1);
        // Still there for the editor's next save to land in, which is the whole
        // point of not deleting it.
        assert!(payload.is_file());
    }

    #[test]
    fn a_container_deleted_under_a_live_session_is_reported_rather_than_ignored() {
        // Found in review, and it was a regression: once `pump` compared bytes,
        // every state that was not `Edited` returned *nothing to do*, so a
        // container removed underneath a session stopped it saving and said
        // nothing at all. The person keeps editing and no error ever appears.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let mut o = open(
            &root,
            &c,
            &Outside::new(&Default_, &Recording::default(), &Silent),
        )
        .unwrap();

        fs::write(o.payload_path(), b"edited").unwrap();
        fs::remove_file(&c).unwrap();

        assert!(matches!(
            o.save_if_changed(),
            Err(writeback::Error::Container(_))
        ));
    }

    #[test]
    fn a_different_container_at_the_recorded_path_refuses_the_write_back() {
        // The guard belongs on the acting side and not only in `recover`:
        // repacking here would rename the payload of a container this session
        // was never opened against.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let mut o = open(
            &root,
            &c,
            &Outside::new(&Default_, &Recording::default(), &Silent),
        )
        .unwrap();

        fs::write(o.payload_path(), b"edited").unwrap();
        let other = container(tmp.path(), "plan.dwg", b"unrelated");
        fs::rename(&other, &c).unwrap();

        match o.save_if_changed() {
            Err(writeback::Error::ContainerChanged { recorded, found }) => {
                assert_eq!(recorded, "report.pdf");
                assert_eq!(found, "plan.dwg");
            }
            other => panic!("{other:?}"),
        }
        // Untouched: still the other container, still its own payload name.
        assert_eq!(
            slpc::Container::open(&c).unwrap().payload_name(),
            "plan.dwg"
        );
    }

    #[test]
    fn saying_yes_to_an_unchanged_payload_rebuilds_nothing() {
        // `close` used to take the answer as a `bool` and repack on it without
        // asking whether anything had changed. That signature is gone, so this
        // cannot be made to fail by reverting the fix the way the two above
        // can; it pins the behaviour rather than the defect. What it is worth
        // is that rewriting the only copy of a container is not a free
        // operation, and answering *yes* to a question about a payload nobody
        // edited should cost nothing.
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let mut o = open(
            &root,
            &c,
            &Outside::new(&Default_, &Recording::default(), &Silent),
        )
        .unwrap();

        assert!(!o.save_if_changed().unwrap());
        assert_eq!(o.session().record().write_backs, 0);
    }

    #[test]
    fn an_edit_is_written_back_once_however_many_times_it_is_asked_for() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let mut o = open(
            &root,
            &c,
            &Outside::new(&Default_, &Recording::default(), &Silent),
        )
        .unwrap();

        fs::write(o.payload_path(), b"edited").unwrap();
        assert!(o.save_if_changed().unwrap());
        assert!(!o.save_if_changed().unwrap());
        assert!(!o.save_if_changed().unwrap());
        assert_eq!(o.session().record().write_backs, 1);
    }

    #[test]
    fn a_clean_close_leaves_nothing_for_recovery_to_ask_about() {
        let tmp = tempfile::tempdir().unwrap();
        let root = tmp.path().join("sessions");
        let c = container(tmp.path(), "report.pdf", b"first");
        let launcher = Recording::default();

        let o = open(&root, &c, &Outside::new(&Default_, &launcher, &Silent)).unwrap();
        assert!(matches!(o.close().unwrap(), Closed::Cleared));
        assert!(crate::session::scan(&root).unwrap().is_empty());
    }

    /// How wide and how long the two stress diagnostics below run.
    ///
    /// Threads, because the first single-threaded attempt at these was too
    /// gentle to wedge anything: 40,000 rounds of the exact shape that wedges
    /// in the suite produced nothing, while the suite itself wedges at roughly
    /// one run in twenty-seven. The suite runs its tests across threads and
    /// this did not, which makes concurrent watchers the first difference to
    /// put back.
    fn stress_shape() -> (usize, usize) {
        let total: usize = std::env::var("SLPC_CLOSE_ROUNDS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(500);
        let threads: usize = std::env::var("SLPC_CLOSE_THREADS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or_else(|| {
                std::thread::available_parallelism().map_or(8, std::num::NonZero::get)
            });
        (total.div_ceil(threads), threads)
    }

    /// **A diagnostic, not part of the gate**, which is why it is `ignore`d.
    ///
    /// `docs/windows-save-test-hang.md` caught a wedge in `TempDir::drop`:
    /// `remove_dir_all` on a watched directory against the watcher re-arming
    /// `ReadDirectoryChangesW` on it. `close` does that collision by
    /// construction rather than by luck — `session.remove()` removes the
    /// watched directory while `self.watch` is still alive, and the watch is
    /// dropped only afterwards. This loops the shipping close path so the
    /// question can be answered by measurement instead of by reading.
    ///
    /// Run it explicitly, under an external timeout, and read the stack of
    /// anything that stops:
    ///
    /// ```text
    /// cargo test --lib -- --ignored --exact flow::tests::close_alone_under_repetition
    /// ```
    ///
    /// A wedge whose stack shows `remove_dir_all` under `session::remove` is
    /// the product hanging on close. One under `tempfile` is the teardown
    /// already written up, and says nothing new.
    #[test]
    #[ignore = "diagnostic; run explicitly under an external timeout"]
    fn close_alone_under_repetition() {
        let (rounds, threads) = stress_shape();
        std::thread::scope(|s| {
            for t in 0..threads {
                s.spawn(move || {
                    for i in 0..rounds {
                        let tmp = tempfile::tempdir().unwrap();
                        let root = tmp.path().join("sessions");
                        let c = container(tmp.path(), "report.pdf", b"first");
                        let o = open(
                            &root,
                            &c,
                            &Outside::new(&Default_, &Recording::default(), &Silent),
                        )
                        .unwrap();

                        // A save immediately before the close, so the watcher
                        // has a completed notification in flight when the
                        // directory goes. That is the state the captured stack
                        // was in.
                        fs::write(o.payload_path(), b"edited").unwrap();

                        assert!(matches!(o.close().unwrap(), Closed::Cleared));

                        // Sparse on purpose: instrumentation is what hid this.
                        if t == 0 && i % 50 == 0 {
                            eprintln!("round {i}");
                        }
                    }
                });
            }
        });
    }

    /// **The positive control for [`close_alone_under_repetition`].** Same
    /// loop, minus the close: `o` and `tmp` both go at the end of the scope,
    /// so the watch is stopped while `remove_dir_all` walks the directory it
    /// was watching. That is the shape of the stack in
    /// `docs/windows-save-test-hang.md`, and this is the run that says whether
    /// the harness can catch it at all.
    ///
    /// Without this, "no wedge in N closes" is not evidence about `close`; it
    /// is only evidence that the loop is too gentle to wedge anything — which
    /// is exactly what the single-threaded version of both turned out to be.
    #[test]
    #[ignore = "diagnostic; run explicitly under an external timeout"]
    fn teardown_alone_under_repetition() {
        let (rounds, threads) = stress_shape();
        std::thread::scope(|s| {
            for t in 0..threads {
                s.spawn(move || {
                    for i in 0..rounds {
                        let tmp = tempfile::tempdir().unwrap();
                        let root = tmp.path().join("sessions");
                        let c = container(tmp.path(), "report.pdf", b"first");
                        let mut o = open(
                            &root,
                            &c,
                            &Outside::new(&Default_, &Recording::default(), &Silent),
                        )
                        .unwrap();

                        // Mirrors the test that wedged: a save, then the save
                        // that finds nothing to do, then the scope ends. No
                        // close.
                        fs::write(o.payload_path(), b"edited").unwrap();
                        assert!(o.save_if_changed().unwrap());
                        assert!(!o.save_if_changed().unwrap());

                        if t == 0 && i % 50 == 0 {
                            eprintln!("round {i}");
                        }
                        // `o` drops here, then `tmp`: stop_watch against
                        // remove_dir_all.
                    }
                });
            }
        });
    }

    /// **Does the collision cross directories?** The close capture could not
    /// say: eight workers and several watchers were alive, so the watch that
    /// collided with the close might have been the closing session's own or a
    /// neighbour's. This separates them.
    ///
    /// One worker runs the shipping close path on its own directory. One
    /// neighbour churns watches on a directory it **never removes** — so the
    /// neighbour can never wedge on a teardown of its own, and every
    /// `stop_watch` in flight belongs to it rather than to the worker, whose
    /// own watch is not stopping during its `remove` (`close` drops it after).
    ///
    /// So a wedge here is a `remove_dir_all` on one directory against a
    /// `stop_watch` on a *different* one, and the fix has to be wider than
    /// ordering a session's own teardown. Its control is
    /// [`close_alone_under_repetition`] run with `SLPC_CLOSE_THREADS=1`, which
    /// is the same worker with no neighbour at all.
    #[test]
    #[ignore = "diagnostic; run explicitly under an external timeout"]
    fn close_with_a_neighbouring_watch() {
        let rounds: usize = std::env::var("SLPC_CLOSE_ROUNDS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(400);

        let stop = std::sync::atomic::AtomicBool::new(false);
        std::thread::scope(|s| {
            s.spawn(|| {
                let ntmp = tempfile::tempdir().unwrap();
                let payload = ntmp.path().join("neighbour.pdf");
                fs::write(&payload, b"x").unwrap();
                while !stop.load(std::sync::atomic::Ordering::Relaxed) {
                    let w = crate::watch::Watch::on(ntmp.path(), "neighbour.pdf").unwrap();
                    fs::write(&payload, b"y").unwrap();
                    drop(w);
                }
            });

            for i in 0..rounds {
                let tmp = tempfile::tempdir().unwrap();
                let root = tmp.path().join("sessions");
                let c = container(tmp.path(), "report.pdf", b"first");
                let o = open(
                    &root,
                    &c,
                    &Outside::new(&Default_, &Recording::default(), &Silent),
                )
                .unwrap();
                fs::write(o.payload_path(), b"edited").unwrap();
                assert!(matches!(o.close().unwrap(), Closed::Cleared));
                if i % 50 == 0 {
                    eprintln!("round {i}");
                }
            }
            stop.store(true, std::sync::atomic::Ordering::Relaxed);
        });
    }

    /// **A `close` diagnostic that cannot wedge in its own teardown**, so what
    /// it counts is `close` and not the harness.
    ///
    /// [`close_alone_under_repetition`] drops a `TempDir` every round, and
    /// three of its four captured wedges were in that drop rather than in
    /// `close` — which is why its numbers say `close` can wedge but not how
    /// often. Here `TempDir::keep` hands the directory over undeleted, so the
    /// only `remove_dir_all` this process performs is the one inside
    /// `Session::remove` under `Opened::close`. A wedge is the product path by
    /// construction rather than by attribution.
    ///
    /// It leaves `slpc-leak-*` directories under the system temp directory on
    /// purpose. Removing them is the loop's job, between runs, when the
    /// process is gone and no watch is alive to collide with the removal.
    #[test]
    #[ignore = "diagnostic; run explicitly under an external timeout; leaks temp dirs by design"]
    fn close_alone_leaving_the_directory_behind() {
        let (rounds, threads) = stress_shape();
        std::thread::scope(|s| {
            for t in 0..threads {
                s.spawn(move || {
                    for i in 0..rounds {
                        let dir = tempfile::Builder::new()
                            .prefix("slpc-leak-")
                            .tempdir()
                            .unwrap()
                            .keep();
                        let root = dir.join("sessions");
                        let c = container(&dir, "report.pdf", b"first");
                        let o = open(
                            &root,
                            &c,
                            &Outside::new(&Default_, &Recording::default(), &Silent),
                        )
                        .unwrap();
                        fs::write(o.payload_path(), b"edited").unwrap();

                        // The only removal in the process.
                        assert!(matches!(o.close().unwrap(), Closed::Cleared));

                        if t == 0 && i % 50 == 0 {
                            eprintln!("round {i}");
                        }
                    }
                });
            }
        });
    }

    /// **The product's actual shape**, which none of the diagnostics above
    /// have. `Resident` serves every request on one thread: the accepting
    /// thread only forwards streams down a channel, and `handle`, `turn` and
    /// every `close` run in the single main loop. So the product never closes
    /// two sessions at once, and the eight concurrent closers the other
    /// diagnostics use correspond to nothing it does.
    ///
    /// What it *does* do is `stand_down`: several sessions open together, each
    /// holding a live watch, then closed one after another on that one thread.
    /// The exposure there was never two removals racing — it was one close's
    /// watch still stopping when the next close's removal began, because the
    /// stop used to be asynchronous. `Watch::drop` waiting is meant to close
    /// exactly that window, and this is the test of whether it does.
    ///
    /// `SLPC_SESSIONS` is how many are open at once. Directories are leaked
    /// with `TempDir::keep`, so the only removal in the process is `close`'s.
    #[test]
    #[ignore = "diagnostic; run explicitly under an external timeout; leaks temp dirs by design"]
    fn a_stand_down_shaped_close() {
        let rounds: usize = std::env::var("SLPC_CLOSE_ROUNDS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(400);
        let at_once: usize = std::env::var("SLPC_SESSIONS")
            .ok()
            .and_then(|v| v.parse().ok())
            .unwrap_or(8);

        for i in 0..rounds {
            // Several sessions live at the same time, as an instance holds
            // them, each with its own watch on its own directory.
            let mut open_now = Vec::with_capacity(at_once);
            for _ in 0..at_once {
                let dir = tempfile::Builder::new()
                    .prefix("slpc-leak-")
                    .tempdir()
                    .unwrap()
                    .keep();
                let root = dir.join("sessions");
                let c = container(&dir, "report.pdf", b"first");
                let o = open(
                    &root,
                    &c,
                    &Outside::new(&Default_, &Recording::default(), &Silent),
                )
                .unwrap();
                fs::write(o.payload_path(), b"edited").unwrap();
                open_now.push(o);
            }

            // `stand_down`: one thread, one after another, no concurrency.
            for o in open_now {
                assert!(matches!(o.close().unwrap(), Closed::Cleared));
            }

            if i % 25 == 0 {
                eprintln!("round {i}");
            }
        }
    }
}