fix(workspaces): make a handover atomic and stop two stores clobbering one file

The takeover moves two things — the `WorkspaceStore`'s record and the
server's `AttachRegistry` handle — and each was internally locked, which
is not the same as the pair moving together. Two clients attaching one
workspace at the same instant could each win a different table, after
which the store named a session the registry had already evicted and no
`detach` could clear it: the workspace reported a takeover against a
client that had disconnected hours ago. Both moves now happen under one
handover lock, dropped before the displaced client is written to so a
peer that has stopped reading still cannot hold up the next attach.

`WorkspaceDelete` had the same split with no race needed at all: the
store dropped its attachment and the registry kept its handle, so the
next client to attach that id evicted a session nobody displaced — and,
that entry being dedicated, closed its whole link.

Two `tty7-server --stdio` sessions arriving while no daemon was up each
served in-process, each with its own store over the one file. `persist`
writes the whole document, so the second to save silently dropped the
first's changes, and their separate registries made takeover a no-op
between them. The probe path now starts the daemon and bridges to it —
the rule `bridge_panes` already follows one dialect over — and the store
re-reads when the file has moved underneath it, which covers the cases
where two writers are deliberate.

`MAX_RECORD_BYTES` and `MAX_WORKSPACES` did not bound their product:
seventeen maximal records put the array past `MAX_FRAME`, after which
every `WorkspaceList` was unencodable and every client showed an empty
list. The document is now bounded at the save, and only when growing, so
an over-large file can still be deleted back under the limit.

crates/tty7-core/src/core/workspace_store.rs

@@ -72,6 +72,20 @@ pub const MAX_RECORD_BYTES: usize = 4 * 1024 * 1024;
 /// error rather than grow the file until the disk fills.
 pub const MAX_WORKSPACES: usize = 1024;
 
+/// Ceiling on the whole document, which is what a single `WorkspaceList` reply
+/// has to fit into.
+///
+/// The per-record and per-count ceilings above are independent of each other,
+/// and their product is 4 GiB — sixty-four times the frame limit. Seventeen
+/// accepted `WorkspacePut`s of a maximal record are enough to put the array
+/// past it, and from then on *every* `WorkspaceList` on the machine is a reply
+/// that cannot be encoded: every client shows an empty workspace list, and the
+/// only repair is editing the file by hand. So the total is bounded where it is
+/// actually known — at the save — with room to spare under
+/// [`MAX_FRAME`](crate::daemon::protocol::MAX_FRAME), since what is measured
+/// here is the pretty-printed form and the wire carries the compact one.
+pub const MAX_STORE_BYTES: usize = 32 * 1024 * 1024;
+
 /// Ceiling on a record key, which is a workspace uuid in every non-hostile
 /// case.
 const MAX_ID_BYTES: usize = 128;
@@ -172,6 +186,23 @@ struct State {
     /// because the file's array order is what a client lists, and a hash map
     /// would reshuffle the picker on every save for no reason.
     records: Vec<(String, Value)>,
+    /// `(mtime, len)` of the file as this snapshot last saw it, or `None` when
+    /// there was no file.
+    ///
+    /// This store is not always the only writer. The design's answer is one
+    /// server per machine, and `tty7-server --stdio` now starts the daemon
+    /// rather than serving in-process for exactly that reason — but an explicit
+    /// `--serve`, or a daemon that could not be started, still leaves two
+    /// processes over one file. `persist` writes the *whole* document, so
+    /// without noticing that the file moved underneath it, the second to save
+    /// silently drops everything the first did.
+    stamp: Option<(std::time::SystemTime, u64)>,
+}
+
+/// The file's identity as far as [`State::stamp`] is concerned.
+fn stamp_of(path: &Path) -> Option<(std::time::SystemTime, u64)> {
+    let meta = std::fs::metadata(path).ok()?;
+    Some((meta.modified().ok()?, meta.len()))
 }
 
 impl WorkspaceStore {
@@ -186,9 +217,10 @@ impl WorkspaceStore {
     pub fn open(path: impl Into<PathBuf>) -> Arc<WorkspaceStore> {
         let path = path.into();
         let records = load_records(&path);
+        let stamp = stamp_of(&path);
         Arc::new(WorkspaceStore {
+            state: Mutex::new(State { records, stamp }),
             path,
-            state: Mutex::new(State { records }),
             attachments: Mutex::new(Vec::new()),
             subscribers: Mutex::new(Vec::new()),
             next_subscriber: AtomicU64::new(1),
@@ -302,7 +334,7 @@ impl WorkspaceStore {
                     Undo::Remove(st.records.len() - 1)
                 }
             };
-            if let Err(e) = self.persist(&st) {
+            if let Err(e) = self.persist(&st, true) {
                 match undo {
                     Undo::Restore(i, old) => st.records[i].1 = old,
                     Undo::Remove(i) => {
@@ -311,6 +343,7 @@ impl WorkspaceStore {
                 }
                 return Err(e);
             }
+            self.restamp(&mut st);
         }
 
         self.notify(id, origin);
@@ -329,10 +362,11 @@ impl WorkspaceStore {
                 return Ok(false);
             };
             let removed = st.records.remove(i);
-            if let Err(e) = self.persist(&st) {
+            if let Err(e) = self.persist(&st, false) {
                 st.records.insert(i, removed);
                 return Err(e);
             }
+            self.restamp(&mut st);
         }
         // The attachment goes with it: nothing can be attached to a workspace
         // that no longer exists, and leaving the entry would have M6 report a
@@ -440,7 +474,24 @@ impl WorkspaceStore {
         // in-memory state is still a valid state (the undo path restores it
         // before returning) and the file is either the old or the new one, so
         // carrying on is strictly better than taking the server down.
-        self.state.lock().unwrap_or_else(|e| e.into_inner())
+        let mut st = self.state.lock().unwrap_or_else(|e| e.into_inner());
+
+        // Re-read when the file moved under us. Cheap — one `stat` — and it is
+        // what keeps a second writer's changes from being overwritten by this
+        // store's whole-document save, since the base we mutate is then theirs
+        // rather than a snapshot from before their write. It also lets a read
+        // see their changes at all: `notify` reaches subscribers in *this*
+        // process only.
+        let on_disk = stamp_of(&self.path);
+        if on_disk != st.stamp {
+            log::debug!(
+                "{} changed underneath this store; re-reading",
+                self.path.display()
+            );
+            st.records = load_records(&self.path);
+            st.stamp = on_disk;
+        }
+        st
     }
 
     fn attachments_locked(&self) -> std::sync::MutexGuard<'_, Vec<(String, Attachment)>> {
@@ -453,7 +504,14 @@ impl WorkspaceStore {
     /// [`Workspaces`](crate::core::session::Workspaces) parses, so this file is
     /// readable by the same code that reads a client's `session.json` and a
     /// human can diff the two.
-    fn persist(&self, st: &State) -> io::Result<()> {
+    /// Write the whole document.
+    ///
+    /// `bounded` asks for [`MAX_STORE_BYTES`] to be enforced. Set by the paths
+    /// that *grow* the file and clear by the ones that shrink it: a store that
+    /// came up holding an over-large file — written by an older build, or by
+    /// hand — must still be able to delete its way back under the limit rather
+    /// than refusing every operation including the repair.
+    fn persist(&self, st: &State, bounded: bool) -> io::Result<()> {
         #[derive(Serialize)]
         struct Doc<'a> {
             workspaces: Vec<&'a Value>,
@@ -462,11 +520,28 @@ impl WorkspaceStore {
             workspaces: st.records.iter().map(|(_, v)| v).collect(),
         };
         let bytes = serde_json::to_vec_pretty(&doc).map_err(io::Error::other)?;
+        if bounded && bytes.len() > MAX_STORE_BYTES {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                format!(
+                    "the workspace store would be {} bytes; the limit is {MAX_STORE_BYTES}, \
+                     which is what one WorkspaceList reply has to fit into",
+                    bytes.len()
+                ),
+            ));
+        }
         if let Some(parent) = self.path.parent() {
             std::fs::create_dir_all(parent)?;
         }
         crate::core::config::write_atomic(&self.path, &bytes)
     }
+
+    /// Record the file's identity after this store wrote it, so the next
+    /// [`WorkspaceStore::locked`] does not mistake its own save for someone
+    /// else's and re-read it.
+    fn restamp(&self, st: &mut State) {
+        st.stamp = stamp_of(&self.path);
+    }
 }
 
 /// How to undo a mutation whose write failed.
@@ -632,6 +707,102 @@ mod tests {
         })
     }
 
+    /// Two stores over one file — an explicit `--serve` alongside a daemon, or
+    /// a daemon that could not be started — must not silently undo each other.
+    ///
+    /// `persist` writes the whole document, so a store that mutates a snapshot
+    /// taken before the other's write puts that stale snapshot back. This is
+    /// how a workspace rename made on the laptop vanishes the next time the
+    /// desktop reorders a tab, with nothing reported to either.
+    #[test]
+    fn a_second_writer_does_not_get_overwritten_by_a_stale_snapshot() {
+        let dir = tempfile::TempDir::new().unwrap();
+        let path = dir.path().join(STORE_FILE);
+        let first = WorkspaceStore::open(&path);
+        let second = WorkspaceStore::open(&path);
+
+        first.put("w1", record("w1", "one"), None).unwrap();
+        first.put("w2", record("w2", "two"), None).unwrap();
+
+        // `second` last read the file when it was empty. It has to notice.
+        second
+            .put("w2", record("w2", "two, renamed"), None)
+            .unwrap();
+
+        let names: Vec<String> = WorkspaceStore::open(&path)
+            .list()
+            .iter()
+            .map(|r| r["name"].as_str().unwrap_or_default().to_string())
+            .collect();
+        assert_eq!(
+            names,
+            ["one", "two, renamed"],
+            "the second writer's save dropped what the first had written"
+        );
+    }
+
+    /// The same, one layer down: a read sees another process's write, because
+    /// `notify` only ever reaches subscribers inside this process.
+    #[test]
+    fn a_read_sees_a_change_another_store_made_to_the_file() {
+        let dir = tempfile::TempDir::new().unwrap();
+        let path = dir.path().join(STORE_FILE);
+        let reader = WorkspaceStore::open(&path);
+        let writer = WorkspaceStore::open(&path);
+
+        assert!(reader.get("w1").is_none());
+        writer.put("w1", record("w1", "one"), None).unwrap();
+        assert_eq!(
+            reader.get("w1").map(|r| r["name"].clone()),
+            Some(serde_json::json!("one")),
+            "a read answered from a snapshot older than the file"
+        );
+    }
+
+    /// The per-record and per-count ceilings do not bound their product, so the
+    /// document is bounded where it is known — at the save.
+    ///
+    /// Past `MAX_FRAME` the store is not merely large, it is unreadable: every
+    /// `WorkspaceList` becomes a reply that cannot be encoded, so every client
+    /// shows an empty list and the only repair is editing the file by hand.
+    #[test]
+    fn a_put_that_would_outgrow_one_reply_is_refused_and_undone() {
+        let (store, _dir) = store();
+        // Records big enough that a handful crosses the limit, and small enough
+        // that the test stays quick.
+        let chunk = "x".repeat(2 * 1024 * 1024);
+        let big = |id: &str| {
+            let mut r = record(id, "big");
+            r["padding"] = Value::String(chunk.clone());
+            r
+        };
+
+        let mut accepted = 0;
+        let refusal = loop {
+            let id = format!("w{accepted}");
+            match store.put(&id, big(&id), None) {
+                Ok(()) => accepted += 1,
+                Err(e) => break e,
+            }
+            assert!(accepted < 64, "the total was never bounded");
+        };
+        assert_eq!(refusal.kind(), io::ErrorKind::InvalidInput);
+        assert!(
+            refusal.to_string().contains("WorkspaceList"),
+            "the refusal has to say what the limit is for: {refusal}"
+        );
+
+        // Refused, not half-applied: the record that did not fit is not in the
+        // store and is not in the file.
+        assert_eq!(store.len(), accepted);
+        assert!(store.get(&format!("w{accepted}")).is_none());
+        assert_eq!(WorkspaceStore::open(store.path()).len(), accepted);
+
+        // And a delete still works, so a store that came up over the limit can
+        // be repaired rather than being wedged.
+        assert!(store.delete("w0", None).unwrap());
+    }
+
     // ── The basics ──────────────────────────────────────────────────────────
 
     #[test]

crates/tty7-core/src/host/server.rs

@@ -144,6 +144,21 @@ impl Services {
 #[derive(Default)]
 pub struct AttachRegistry {
     live: Mutex<Vec<Live>>,
+    /// Held across *both* tables for the length of one handover.
+    ///
+    /// A takeover moves two things that live in different places: this
+    /// registry's handles, and the `WorkspaceStore`'s record. Each is
+    /// internally locked, and that is not enough — two clients attaching to one
+    /// workspace at the same moment can each win a different table, after which
+    /// the store names a session the registry has already evicted and no
+    /// `detach` can ever clear it, because the token no longer matches. From
+    /// then on the workspace reports a takeover against a client that
+    /// disconnected hours ago.
+    ///
+    /// Coarse on purpose: attach and detach happen once per workspace opened or
+    /// closed, so serializing them costs nothing worth measuring. Always the
+    /// outermost lock of the two, and never held while writing to a peer.
+    handover: Mutex<()>,
 }
 
 struct Live {
@@ -182,6 +197,11 @@ struct Evicted {
 }
 
 impl AttachRegistry {
+    /// Take the handover lock. See [`AttachRegistry::handover`].
+    fn handover(&self) -> std::sync::MutexGuard<'_, ()> {
+        self.handover.lock().unwrap_or_else(|e| e.into_inner())
+    }
+
     /// Who holds `workspace` — `(token, hostname)`. Diagnostics and tests.
     pub fn holder(&self, workspace: &str) -> Option<(String, String)> {
         self.locked()
@@ -244,6 +264,14 @@ impl AttachRegistry {
         evicted
     }
 
+    /// Forget `workspace` whoever holds it — the workspace itself is gone.
+    ///
+    /// Unconditional, unlike [`AttachRegistry::release`]: a delete is not one
+    /// session giving something up, it is the thing ceasing to exist.
+    fn forget_workspace(&self, workspace: &str) {
+        self.locked().retain(|l| l.workspace != workspace);
+    }
+
     /// Release `workspace`, but only if `conn` still holds it. `false` means it
     /// had already been taken over, which is success as far as the caller is
     /// concerned — and the reason releasing is conditional at all.
@@ -520,13 +548,21 @@ fn attach_workspace(
     dedicated: bool,
 ) -> io::Result<Option<String>> {
     let store = conn.workspaces()?;
-    let displaced = store.attach(
-        workspace,
-        Attachment::new(conn.holder.token.clone(), conn.holder.hostname.clone()),
-    );
-    let evicted = conn
-        .attachments
-        .claim(workspace, conn.id, &conn.holder, dedicated);
+    let (displaced, evicted) = {
+        // Both tables move under one lock. Held only across the two moves —
+        // the notice below goes out with nothing held, because writing to a
+        // peer that has stopped reading must not hold up the next client's
+        // attach.
+        let _handover = conn.attachments.handover();
+        let displaced = store.attach(
+            workspace,
+            Attachment::new(conn.holder.token.clone(), conn.holder.hostname.clone()),
+        );
+        let evicted = conn
+            .attachments
+            .claim(workspace, conn.id, &conn.holder, dedicated);
+        (displaced, evicted)
+    };
 
     if let Some(evicted) = evicted {
         log::info!(
@@ -562,6 +598,7 @@ fn attach_workspace(
 /// then tidied up must not evict the client that took over from it.
 fn detach_workspace(conn: &Arc<Conn>, workspace: &str) -> io::Result<bool> {
     let store = conn.workspaces()?;
+    let _handover = conn.attachments.handover();
     let released = conn.attachments.release(workspace, conn.id);
     let forgotten = store.detach(workspace, &conn.holder.token);
     Ok(released || forgotten)
@@ -814,7 +851,19 @@ fn run_request(
         ControlRequest::WorkspaceDelete { id } => {
             // Deleting what is not there is success — a delete that raced
             // another client's delete has got what it asked for.
-            conn.workspaces()?.delete(&id, conn.workspace_origin)?;
+            let store = conn.workspaces()?;
+            {
+                // The store drops its own attachment on delete; the registry
+                // has to be told, and under the same lock, or the two disagree
+                // with no race needed at all. Left behind, the stale `Live`
+                // entry means the *next* client to attach a workspace with this
+                // id evicts a session nobody displaced — and, that entry being
+                // dedicated, closes its whole link, taking every other
+                // workspace on it down too.
+                let _handover = conn.attachments.handover();
+                store.delete(&id, conn.workspace_origin)?;
+                conn.attachments.forget_workspace(&id);
+            }
             (ReplyOk::Unit, Vec::new())
         }
 
@@ -894,6 +943,7 @@ impl Conn {
     /// earlier is already gone from the registry and is not touched — the exact
     /// case the store's token check exists for, seen from the other side.
     fn release_all_workspaces(&self) {
+        let _handover = self.attachments.handover();
         let released = self.attachments.release_conn(self.id);
         let Some(store) = self.workspaces.as_ref() else {
             return;
@@ -3015,6 +3065,98 @@ mod tests {
         );
     }
 
+    /// Deleting a workspace clears it from *both* tables.
+    ///
+    /// The store drops its own attachment on delete. If the registry keeps its
+    /// handle, the two disagree with no race needed, and the next client to
+    /// attach that id evicts a session nobody displaced — closing its whole
+    /// link, since a dedicated entry takes every other workspace on that
+    /// connection down with it.
+    #[test]
+    fn deleting_a_workspace_clears_both_attachment_tables() {
+        let (services, _dir) = workspace_services();
+        let registry = Arc::clone(&services.attachments);
+        let store = services.workspaces.clone().unwrap();
+
+        let ((mut laptop, _), _l) =
+            raw_hello(services.clone(), hello_for("w", "tok-laptop", "laptop"));
+        await_holder(&registry, "w", "laptop");
+        assert!(store.attachment("w").is_some());
+
+        ask(
+            &mut laptop,
+            1,
+            ControlRequest::WorkspacePut {
+                id: "w".to_string(),
+                json: ws_record("w", "the workspace"),
+            },
+        );
+        let reply = ask(
+            &mut laptop,
+            2,
+            ControlRequest::WorkspaceDelete {
+                id: "w".to_string(),
+            },
+        );
+        assert!(
+            matches!(reply, ControlReply::Ok(ReplyOk::Unit)),
+            "{reply:?}"
+        );
+
+        assert!(
+            store.attachment("w").is_none(),
+            "the store still names a holder for a workspace that is gone"
+        );
+        assert!(
+            registry.holder("w").is_none(),
+            "the registry still holds a workspace that is gone"
+        );
+    }
+
+    /// The store's record and the registry's handle move under **one** lock.
+    ///
+    /// They are separate tables with separate locks, and taking them one after
+    /// the other is not enough: two clients attaching the same workspace at the
+    /// same instant can each win a different one, after which the store names a
+    /// session the registry has already evicted. No `detach` can clear it — its
+    /// token no longer matches — so from then on the workspace reports a
+    /// takeover against a client that disconnected hours ago.
+    ///
+    /// Held from the test rather than raced, because the window is a few
+    /// instructions wide and a racing test passes against the broken ordering
+    /// far more often than it fails. Holding the handover proves the stronger
+    /// thing anyway: with it held, an attach reaches *neither* table.
+    #[test]
+    fn an_attach_moves_both_tables_under_one_lock() {
+        let (services, _dir) = workspace_services();
+        let registry = Arc::clone(&services.attachments);
+        let store = services.workspaces.clone().unwrap();
+
+        let held = registry.handover();
+        // The handshake replies before the attach, so this returns rather than
+        // blocking on the lock we are holding.
+        let ((_laptop, _ok), _served) =
+            raw_hello(services.clone(), hello_for("w", "tok-laptop", "laptop"));
+
+        std::thread::sleep(Duration::from_millis(150));
+        assert!(
+            registry.holder("w").is_none(),
+            "the registry was moved while a handover was in flight"
+        );
+        assert!(
+            store.attachment("w").is_none(),
+            "the store was moved while a handover was in flight"
+        );
+
+        drop(held);
+        await_holder(&registry, "w", "laptop");
+        assert_eq!(
+            store.attachment("w").map(|a| a.token).as_deref(),
+            Some("tok-laptop"),
+            "both tables have to name the same session once the handover is done"
+        );
+    }
+
     /// The other half of that rule. A client holds **one connection per
     /// machine**, so closing the link on a takeover would drop windows nobody
     /// preempted; the push still goes out, the link stays up.

crates/tty7-server/src/main.rs

@@ -163,6 +163,7 @@ fn run_stdio(args: &[String]) -> io::Result<()> {
     {
         use std::os::unix::net::UnixStream;
         use tty7_core::daemon::duplex::StdioDuplex;
+        use tty7_core::daemon::spawn;
         use tty7_core::host::local::LocalHost;
         use tty7_core::host::server;
 
@@ -201,10 +202,43 @@ fn run_stdio(args: &[String]) -> io::Result<()> {
                 Err(e) if force_bridge => return Err(e),
                 Err(e) => {
                     log_stderr(format_args!(
-                        "no control server at {} ({e}); serving in this process",
+                        "no control server at {} ({e})",
                         sock.display()
                     ));
-                    None
+                    // One control server per machine, started if nobody has —
+                    // the same rule `bridge_panes` follows one dialect over,
+                    // and for the same reason. Two `--stdio` sessions both
+                    // falling through to serving in-process would each hold
+                    // their own `WorkspaceStore` over the one file, and
+                    // `persist` writes the whole document: the second to save
+                    // silently drops the first's changes. Their attachment
+                    // registries would be separate too, which makes design
+                    // §10's takeover a no-op between them — both clients would
+                    // hold the same workspace and neither would be told.
+                    //
+                    // Not attempted when the caller named a socket: starting a
+                    // daemon binds the machine's default endpoint, not theirs,
+                    // so it would be a daemon nobody asked for and nobody uses.
+                    if may_start_daemon(args) {
+                        match spawn::ensure_running()
+                            .map_err(io::Error::other)
+                            .and_then(|()| UnixStream::connect(&sock))
+                        {
+                            Ok(s) => {
+                                log_stderr(format_args!("started one; bridging to it"));
+                                Some(s)
+                            }
+                            Err(e) => {
+                                log_stderr(format_args!(
+                                    "could not start one ({e}); serving in this process"
+                                ));
+                                None
+                            }
+                        }
+                    } else {
+                        log_stderr(format_args!("serving in this process"));
+                        None
+                    }
                 }
             }
         };
@@ -358,6 +392,17 @@ fn control_services() -> tty7_core::host::server::Services {
 }
 
 /// `--flag <value>` or `--flag=<value>`, first occurrence wins.
+/// Whether a failed control probe may start the machine's daemon.
+///
+/// Only when the caller did not name a socket. `--control-sock` says "this
+/// endpoint", and `spawn::ensure_running` binds the machine's default one — so
+/// starting a daemon there would leave a process nobody asked for and nobody
+/// reaches. It is also what keeps the test suite, and any `--config-dir`
+/// isolation built on it, from spraying daemons across a developer's machine.
+fn may_start_daemon(args: &[String]) -> bool {
+    flag_value(args, "--control-sock").is_none()
+}
+
 fn flag_value(args: &[String], flag: &str) -> Option<String> {
     let with_eq = format!("{flag}=");
     let mut it = args.iter();
@@ -385,3 +430,23 @@ fn apply_config_dir_arg(args: &[String]) {
         tty7_core::core::config::set_config_dir(path.into());
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn argv(args: &[&str]) -> Vec<String> {
+        args.iter().map(|a| a.to_string()).collect()
+    }
+
+    /// A named socket suppresses the daemon start, in both spellings of the
+    /// flag. Without this guard every `--stdio` in the test suite that points
+    /// at a temp socket would start a real daemon on the developer's machine.
+    #[test]
+    fn a_named_control_socket_suppresses_starting_a_daemon() {
+        assert!(may_start_daemon(&argv(&[])));
+        assert!(may_start_daemon(&argv(&["--serve"])));
+        assert!(!may_start_daemon(&argv(&["--control-sock", "/tmp/x.sock"])));
+        assert!(!may_start_daemon(&argv(&["--control-sock=/tmp/x.sock"])));
+    }
+}