feat(control): attachment and takeover ride the machine tree too

WorkspaceAttach / WorkspaceDetach (and the hello-names-a-workspace
shorthand) now record their data half on whichever workspace stores the
server carries: the retired record store, the machine tree, or — on a
full daemon while clients migrate — both, since they describe the same
workspace. The behavioural contract is untouched and now survives the
record store's retirement: newcomer always wins, the displaced session
is pushed Preempted (and closed only when its link was dedicated), and
a preempted session's tidy-up detach cannot evict the usurper — the
token check lives in the tree's runtime-only attachment exactly as it
did in the store's. A server carrying neither store answers the same
refusal a store-less server always has.

WorkspaceId gains FromStr (the inverse of its Display) because the
attach verbs predate the typed tree and carry the id as a string. The
end-to-end test drives a takeover on a server serving the tree and no
record store at all, asserting the tree's own attachment record moves
with it.

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

@@ -146,6 +146,17 @@ impl std::fmt::Display for WorkspaceId {
     }
 }
 
+impl std::str::FromStr for WorkspaceId {
+    type Err = uuid::Error;
+
+    /// The inverse of `Display`, for the places a workspace id crosses a
+    /// string-keyed boundary (the control dialect's attach verbs, which
+    /// predate the typed tree) and has to come back out as itself.
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        s.parse().map(WorkspaceId)
+    }
+}
+
 // ---------------------------------------------------------------------------
 // Remote references
 // ---------------------------------------------------------------------------

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

@@ -590,21 +590,38 @@ fn attach_workspace(
     workspace: &str,
     dedicated: bool,
 ) -> io::Result<Option<String>> {
-    let store = conn.workspaces()?;
+    // The attach verbs predate the machine tree, so the id arrives as a
+    // string; while the record store and the tree coexist, the attachment's
+    // data half lands on **whichever of the two this server carries** (both,
+    // on a full daemon — they describe the same workspace). A server with
+    // neither answers exactly what a store-less server always has.
+    if conn.workspaces.is_none() && conn.machine.is_none() {
+        return Err(io::Error::other(
+            "this server does not serve the workspace store",
+        ));
+    }
+    let tree_id: Option<crate::core::session::WorkspaceId> = workspace.parse().ok();
     let (displaced, evicted) = {
-        // Both tables move under one lock. Held only across the two moves —
+        // Every table moves under one lock. Held only across the 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 attachment = Attachment::new(conn.holder.token.clone(), conn.holder.hostname.clone());
+        let displaced_record = conn
+            .workspaces
+            .as_ref()
+            .and_then(|store| store.attach(workspace, attachment.clone()));
+        let displaced_tree = match (&conn.machine, tree_id) {
+            (Some(machine), Some(id)) => machine.attach(id, attachment),
+            _ => None,
+        };
         let evicted = conn
             .attachments
             .claim(workspace, conn.id, &conn.holder, dedicated);
-        (displaced, evicted)
+        // On a server carrying both, the two answers name the same session;
+        // the record store's wins only in the sense that it is asked first.
+        (displaced_record.or(displaced_tree), evicted)
     };
 
     if let Some(evicted) = evicted {
@@ -640,11 +657,22 @@ fn attach_workspace(
 /// are the same guard seen from both halves: a session that was preempted and
 /// 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()?;
+    if conn.workspaces.is_none() && conn.machine.is_none() {
+        return Err(io::Error::other(
+            "this server does not serve the workspace store",
+        ));
+    }
     let _handover = conn.attachments.handover();
     let released = conn.attachments.release(workspace, conn.id);
-    let forgotten = store.detach(workspace, &conn.holder.token);
-    Ok(released || forgotten)
+    let forgotten = conn
+        .workspaces
+        .as_ref()
+        .is_some_and(|store| store.detach(workspace, &conn.holder.token));
+    let forgotten_tree = match (&conn.machine, workspace.parse().ok()) {
+        (Some(machine), Some(id)) => machine.detach(id, &conn.holder.token),
+        _ => false,
+    };
+    Ok(released || forgotten || forgotten_tree)
 }
 
 /// This machine's home directory, for the handshake's `home` field — the value
@@ -1166,11 +1194,13 @@ impl Conn {
     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;
-        };
         for workspace in released {
-            store.detach(&workspace, &self.holder.token);
+            if let Some(store) = self.workspaces.as_ref() {
+                store.detach(&workspace, &self.holder.token);
+            }
+            if let (Some(machine), Some(id)) = (&self.machine, workspace.parse().ok()) {
+                machine.detach(id, &self.holder.token);
+            }
         }
     }
 

crates/tty7-server/tests/machine_tree.rs

@@ -394,6 +394,85 @@ fn an_operation_from_one_client_reaches_the_other_as_a_delta() {
     assert_eq!(watcher.delta_count(), 2, "still only the writer's two ops");
 }
 
+/// Takeover semantics on the new tree, with **no record store served at
+/// all**: the attach verbs predate the tree, and their contract — newcomer
+/// wins, the displaced session is told, a stale detach cannot evict the
+/// usurper — must survive the record store's retirement.
+#[test]
+fn attachment_rides_the_tree_when_no_record_store_is_served() {
+    use tty7_core::host::local::LocalHost;
+    use tty7_core::host::server;
+
+    let dir = data_dir();
+    let machine = tty7_core::core::machine::MachineStore::open(machine_file(&dir));
+    let sock = dir.path().join("control.sock");
+    let listener = server::bind_control_socket(&sock).unwrap();
+    {
+        let machine = Arc::clone(&machine);
+        std::thread::spawn(move || {
+            server::serve_listener_with(
+                listener,
+                LocalHost::new(),
+                server::Services::none().and_machine(machine),
+            )
+        });
+    }
+    let ws = machine
+        .workspace_create(Some("shared".into()), None)
+        .unwrap();
+
+    let laptop = bridged(&sock, "laptop");
+    let desktop = bridged(&sock, "desktop");
+
+    let attach = |client: &Client| {
+        client.control.call(ControlRequest::WorkspaceAttach {
+            id: ws.id.to_string(),
+        })
+    };
+    match attach(&laptop).expect("first attach") {
+        ReplyOk::Attached { took_over_from } => assert_eq!(took_over_from, None),
+        other => panic!("{other:?}"),
+    }
+    assert_eq!(
+        machine.attachment(ws.id).map(|a| a.hostname),
+        Some("laptop".into()),
+        "the tree's own record says who holds the workspace"
+    );
+
+    // The newcomer wins, learns whom it displaced, and the displaced session
+    // is pushed a Preempted notice.
+    match attach(&desktop).expect("takeover") {
+        ReplyOk::Attached { took_over_from } => {
+            assert_eq!(took_over_from.as_deref(), Some("laptop"));
+        }
+        other => panic!("{other:?}"),
+    }
+    let deadline = Instant::now() + Duration::from_secs(10);
+    loop {
+        let seen = laptop.events.lock().unwrap().clone();
+        if seen.iter().any(|e| {
+            matches!(e, ControlEvent::Preempted { workspace, by }
+                if *workspace == ws.id.to_string() && by == "desktop")
+        }) {
+            break;
+        }
+        assert!(Instant::now() < deadline, "no Preempted push; saw {seen:?}");
+        std::thread::sleep(Duration::from_millis(20));
+    }
+
+    // The preempted session tidying up must not evict the usurper.
+    laptop
+        .control
+        .call(ControlRequest::WorkspaceDetach {
+            id: ws.id.to_string(),
+        })
+        .expect("a stale detach is success, not eviction");
+    assert_eq!(
+        machine.attachment(ws.id).map(|a| a.hostname),
+        Some("desktop".into())
+    );
+}
+
 /// A `--stdio --bridge` child connected to an already-listening control
 /// socket — the two-hop shape a real multi-client machine has.
 fn bridged(sock: &Path, token: &str) -> Client {