acpmux: move a chat to another daemon with its history, queue and draft (acpmux.transfer.v1, cx-xtg0.6)

cmux-tui/crates/acpmux/src/cli/command.rs

@@ -299,6 +299,18 @@ pub enum Command {
         name: Option<String>,
     },
     #[command(hide = true)]
+    Transfer {
+        session: String,
+        #[arg(long)]
+        to: PathBuf,
+        #[arg(long)]
+        cwd: Option<String>,
+        #[arg(long, short)]
+        name: Option<String>,
+        #[arg(long)]
+        interrupt: bool,
+    },
+    #[command(hide = true)]
     Harnesses,
     #[command(hide = true)]
     Reload,
@@ -528,6 +540,21 @@ pub enum SessionCmd {
         #[arg(long, short)]
         name: Option<String>,
     },
+    /// Move a chat, with its history, queue and draft, to the daemon at another socket.
+    Transfer {
+        session: String,
+        /// The other daemon's socket (a sibling build's).
+        #[arg(long)]
+        to: PathBuf,
+        /// The chat's folder there, when it has another path.
+        #[arg(long)]
+        cwd: Option<String>,
+        #[arg(long, short)]
+        name: Option<String>,
+        /// Stop a running turn instead of refusing the move.
+        #[arg(long)]
+        interrupt: bool,
+    },
     /// Print the last raw events as JSON lines; -f keeps following.
     Tail {
         session: String,
@@ -735,6 +762,9 @@ pub fn flatten(c: Command) -> Command {
             SessionCmd::Answer { session, answer } => Command::Answer { session, answer },
             SessionCmd::Export { session, dest } => Command::Export { session, dest },
             SessionCmd::Import { path, name } => Command::Import { path, name },
+            SessionCmd::Transfer { session, to, cwd, name, interrupt } => {
+                Command::Transfer { session, to, cwd, name, interrupt }
+            }
             SessionCmd::Tail { session, last, follow, since } => {
                 Command::Tail { session, last, follow, since }
             }

cmux-tui/crates/acpmux/src/cli/run.rs

@@ -15,6 +15,7 @@ pub(crate) use permission::{answer_permission, answer_question};
 
 mod extra;
 mod standard;
+mod transfer;
 
 pub(crate) async fn run_client(cmd: Command, json_out: bool, suppress_reads: bool) -> Result<()> {
     if standard::handles(&cmd) {

cmux-tui/crates/acpmux/src/cli/run/extra.rs

@@ -171,6 +171,9 @@ pub(super) async fn run(cmd: Command, json_out: bool, _suppress_reads: bool) ->
             }
             Ok(())
         }
+        Command::Transfer { session, to, cwd, name, interrupt } => {
+            super::transfer::run(&session, &to, cwd, name, interrupt, json_out).await
+        }
         Command::Import { path, name } => {
             let client = connect(true).await?;
             let mut p = json!({"path": std::path::absolute(path)?});

cmux-tui/crates/acpmux/src/cli/run/transfer.rs

@@ -0,0 +1,123 @@
+//! `acpmux transfer SESSION --to SOCKET`: moves a chat from this daemon to another one (a sibling
+//! build's) through `_acpmux/transfer_*` (`hub/transfer_out.rs`). The reference client of the
+//! move: it pages the bundle from one daemon to the other, and gives the chat back on a failure.
+
+use super::*;
+
+/// Moves `session` to the daemon at `to`.
+pub(super) async fn run(
+    session: &str,
+    to: &std::path::Path,
+    cwd: Option<String>,
+    name: Option<String>,
+    interrupt: bool,
+    json_out: bool,
+) -> Result<()> {
+    let src = connect(true).await?;
+    let dst = Client::connect(to).await?;
+    dst.request(
+        method::INITIALIZE,
+        json!({"protocolVersion": 1, "clientInfo": {"name": "acpmux-transfer"}}),
+    )
+    .await?;
+    let id = resolve_id(&src, session).await?;
+    let begin = src
+        .request("_acpmux/transfer_begin", json!({"sessionId": id, "interrupt": interrupt}))
+        .await?;
+    let transfer_id = begin["transferId"].as_str().unwrap_or_default().to_owned();
+    let abort = async || {
+        let abort = json!({"sessionId": id, "transferId": transfer_id});
+        if let Err(e) = src.request("_acpmux/transfer_abort", abort).await {
+            eprintln!("acpmux: the move failed and the chat stays frozen here: {e}");
+        }
+    };
+    let receive = match dst.request("_acpmux/transfer_receive", json!({"manifest": begin})).await {
+        Ok(receive) => receive,
+        Err(e) => {
+            abort().await;
+            return Err(e);
+        }
+    };
+    let moved = match copy(&src, &dst, &begin, &receive["receiveId"], cwd, name).await {
+        Ok(moved) => moved,
+        // A finish whose answer was lost may have landed, and the target runs the queue then:
+        // the source commits. Only a target that shows no such move lets the chat come back.
+        Err(e) => match landed(&dst, &id, &transfer_id).await {
+            Some(true) => dst.request(method::MUX_INFO, json!({"sessionId": id})).await?,
+            Some(false) => {
+                abort().await;
+                return Err(e);
+            }
+            None => {
+                eprintln!(
+                    "acpmux: the move failed and the target cannot say whether it took the chat; it stays frozen here until the move is retried or aborted"
+                );
+                return Err(e);
+            }
+        },
+    };
+    let to_meta = json!({"sessionId": moved["sessionId"], "socket": to});
+    src.request(
+        "_acpmux/transfer_commit",
+        json!({"sessionId": id, "transferId": transfer_id, "to": to_meta}),
+    )
+    .await?;
+    if json_out {
+        print_json(&moved);
+    } else {
+        println!("moved {} to {}", moved["name"].as_str().unwrap_or(&id), to.display());
+    }
+    Ok(())
+}
+
+/// Whether the target took this move (its log has the `transferred_in` of `transfer_id`); `None`
+/// when it cannot say.
+async fn landed(dst: &Client, id: &str, transfer_id: &str) -> Option<bool> {
+    let kinds = json!({"sessionId": id, "afterSeq": 0, "kinds": ["transferred_in"]});
+    match dst.request(method::MUX_EVENTS, kinds).await {
+        Ok(page) => Some(
+            page["events"]
+                .as_array()
+                .into_iter()
+                .flatten()
+                .any(|e| e["msg"]["transferId"] == transfer_id),
+        ),
+        Err(e) => {
+            let m = e.to_string().to_lowercase();
+            (m.contains("no session") || m.contains("not found")).then_some(false)
+        }
+    }
+}
+
+/// Pages every bundle file from `src` into the receive on `dst`, and finishes it there.
+async fn copy(
+    src: &Client,
+    dst: &Client,
+    begin: &Value,
+    receive_id: &Value,
+    cwd: Option<String>,
+    name: Option<String>,
+) -> Result<Value> {
+    for file in begin["files"].as_array().into_iter().flatten() {
+        let path = &file["path"];
+        let mut offset = 0u64;
+        loop {
+            let read = json!({"transferId": begin["transferId"], "path": path, "offset": offset});
+            let page = src.request("_acpmux/transfer_read", read).await?;
+            let write = json!({"receiveId": receive_id, "path": path, "offset": offset, "data": page["data"]});
+            dst.request("_acpmux/transfer_write", write).await?;
+            offset = page["next"].as_u64().ok_or_else(|| anyhow!("transfer_read: no next"))?;
+            if page["eof"] == true {
+                break;
+            }
+        }
+    }
+    let mut finish = json!({"receiveId": receive_id, "from": {"sessionId": begin["sessionId"]}});
+    if let Some(cwd) = cwd {
+        finish["cwd"] = json!(cwd);
+    }
+    if let Some(name) = name {
+        finish["name"] = json!(name);
+    }
+    dst.request("_acpmux/transfer_finish", finish).await
+}

cmux-tui/crates/acpmux/src/hub/mod.rs

@@ -57,6 +57,8 @@ mod remote_sandbox;
 pub use permission_groups::PERMISSION_GROUP_OPERATIONS;
 pub mod rules;
 mod transfer;
+mod transfer_in;
+mod transfer_out;
 mod turns;
 mod warm;
 mod xai;
@@ -532,6 +534,7 @@ impl Hub {
             queued: AtomicU64::new(0),
             queue: StdMutex::new(Vec::new()),
             queue_moved: Notify::new(),
+            moving: StdMutex::new(None),
             restored: StdMutex::new(Vec::new()),
             stream: StdMutex::new(StreamState::default()),
             permissions: StdMutex::new(permission_groups::PermissionState::default()),
@@ -770,6 +773,9 @@ impl Hub {
                 Self::MAX_COMPOSER_DRAFT_CHARS
             ));
         }
+        if session.moving().is_some() {
+            return Err(queue::moving_error().message);
+        }
         let draft = (!text.trim().is_empty()).then(|| text.to_owned());
         {
             let mut meta = session.meta.lock().unwrap_or_else(|poisoned| poisoned.into_inner());