fix(terminal): retry a two-stage send without repeating paste or Enter

cmux-tui/crates/cmux-tui/src/cli/mcp/terminal_talk_tools.rs

@@ -38,7 +38,9 @@ const SEND_DESCRIPTION: &str = "Send a message to the agent (or shell) in a cmux
     passes timeout_seconds (default 300, at most 900) answers timed_out true with the output \
     so far. An agent that already waits on a dialog is refused unless force is true, since \
     the text would answer the dialog. target is a term_ id (or unique prefix), a tab_ id or \
-    an exact terminal title; agents_terminal_list names them.";
+    an exact terminal title; agents_terminal_list names them. Optional idempotency_key \
+    identifies the complete paste-and-Enter send: a lost reply reports its \
+    phase and the same key for a safe retry.";
 
 pub(super) fn descriptors() -> [Value; 2] {
     [
@@ -83,6 +85,8 @@ fn send_schema() -> Value {
             "settle_seconds": {"type": "number", "minimum": 0.2, "maximum": 60, "default": 3},
             "force": {"type": "boolean", "default": false,
                       "description": "Type even when the agent waits on a dialog."},
+            "idempotency_key": {"type": "string", "minLength": 1, "maxLength": 128,
+                      "description": "Stable send identity for retries; reuse after an uncertain result."},
             "session": session_schema(),
         },
         "required": ["target", "text"],
@@ -131,6 +135,11 @@ pub(super) fn send_request(
     let text = string(arguments, "text")?.ok_or_else(|| format!("{SEND}: text is required"))?;
     let mut request = send::SendRequest::new(target, text);
     request.timeout = Duration::from_secs_f64(DEFAULT_SEND_SECONDS);
+    if let Some(key) = string(arguments, "idempotency_key")? {
+        cmux_tui_core::resource::validate_idempotency_key(&key)
+            .map_err(|error| format!("{SEND}: {}", error.message))?;
+        request.idempotency_key = Some(key);
+    }
     if let Some(wait) = string(arguments, "wait")? {
         request.wait = send::Wait::parse(&wait)
             .ok_or_else(|| format!("{SEND}: wait is idle, prompt or none, not {wait:?}"))?;
@@ -209,13 +218,19 @@ mod tests {
         assert_eq!(send["inputSchema"]["required"], json!(["target", "text"]));
         assert_eq!(list["annotations"]["readOnlyHint"], true);
         assert_eq!(send["annotations"]["readOnlyHint"], false);
+        assert_eq!(
+            send["inputSchema"]["properties"]["idempotency_key"]["maxLength"],
+            128,
+            "a composed send needs a stable retry identity: {send}"
+        );
     }
 
     #[test]
     fn send_arguments_become_the_cli_request() {
         let (request, session) = send_request(&arguments(json!({
             "target": "term_a", "text": "hi\nthere", "wait": "prompt",
             "timeout_seconds": 20, "settle_seconds": 0.5, "force": true, "session": "work",
+            "idempotency_key": "repeatable-send",
         })))
         .expect("valid");
         assert_eq!(request.target, "term_a");
@@ -224,6 +239,7 @@ mod tests {
         assert_eq!(request.timeout.as_secs(), 20);
         assert_eq!(request.settle.as_millis(), 500);
         assert!(request.force);
+        assert_eq!(request.idempotency_key.as_deref(), Some("repeatable-send"));
         assert_eq!(session.as_deref(), Some("work"));
         let (defaults, _) =
             send_request(&arguments(json!({"target": "t", "text": "x"}))).expect("valid");
@@ -238,6 +254,8 @@ mod tests {
             json!({"target": "t", "text": "x", "wait": "later"}),
             json!({"target": "t", "text": "x", "timeout_seconds": 5000}),
             json!({"target": "t", "text": "x", "force": "yes"}),
+            json!({"target": "t", "text": "x", "idempotency_key": ""}),
+            json!({"target": "t", "text": "x", "idempotency_key": "x".repeat(129)}),
         ] {
             assert!(send_request(&arguments(bad.clone())).is_err(), "{bad}");
         }

cmux-tui/crates/cmux-tui/src/cli/terminal_talk.rs

@@ -45,7 +45,10 @@ pub(in crate::cli) fn run(global: &GlobalArgs, words: &[String]) -> Option<i32>
             })
         }
         [target, verb, flags @ ..] if verb == "send" => Some(match parse_send(target, flags) {
-            Ok(request) => send_command(global, &request),
+            Ok(mut request) => {
+                request.idempotency_key = global.idempotency_key.clone();
+                send_command(global, &request)
+            }
             Err(error) => usage(global, &error),
         }),
         _ => None,

cmux-tui/crates/cmux-tui/src/cli/terminal_talk/send.rs

@@ -23,6 +23,7 @@ use std::time::{Duration, Instant};
 
 use base64::Engine as _;
 use serde_json::{Value, json};
+use sha2::{Digest, Sha256};
 
 use super::super::Surface;
 use super::super::command::{self, CommandPlan, RequestPlan};
@@ -74,6 +75,8 @@ pub(in crate::cli) struct SendRequest {
     pub timeout: Duration,
     pub settle: Duration,
     pub force: bool,
+    /// Stable identity for retrying the whole paste-and-Enter send.
+    pub idempotency_key: Option<String>,
     pub enter_delay: Duration,
     pub poll: Duration,
 }
@@ -87,6 +90,7 @@ impl SendRequest {
             timeout: DEFAULT_TIMEOUT,
             settle: DEFAULT_SETTLE,
             force: false,
+            idempotency_key: None,
             enter_delay: DEFAULT_ENTER_DELAY,
             poll: DEFAULT_POLL,
         }
@@ -143,15 +147,47 @@ fn plan(words: &[&str]) -> Result<RequestPlan, SendFailure> {
     }
 }
 
-fn mutation(words: &[&str]) -> Result<RequestPlan, SendFailure> {
+fn mutation(words: &[&str], key: String) -> Result<RequestPlan, SendFailure> {
     let mut plan = plan(words)?;
-    plan.idempotency_key = Some(
-        super::super::wire::random_idempotency_key()
-            .map_err(|error| SendFailure::local("usage.invalid", error.0, json!({}), 2))?,
-    );
+    plan.idempotency_key = Some(key);
     Ok(plan)
 }
 
+/// Both terminal writes derive stable, distinct daemon keys from one caller
+/// key and the resolved terminal. A lost reply can then replay the original
+/// paste or Enter without repeating that input on the terminal.
+fn stage_key(send_id: &str, terminal: &str, stage: &str) -> String {
+    let mut hash = Sha256::new();
+    for part in ["cmux.terminal.send.v1", send_id, terminal, stage] {
+        hash.update(part.as_bytes());
+        hash.update([0]);
+    }
+    format!("terminal-send-{:x}", hash.finalize())
+}
+
+/// The daemon's write key is an implementation detail. Return the caller's
+/// identity and the uncertain phase, so the *entire send* can be retried.
+fn stage_failure(mut failure: SendFailure, send_id: &str, phase: &str) -> SendFailure {
+    if !failure.failure.error["details"].is_object() {
+        failure.failure.error.set("details", json!({}));
+    }
+    failure.failure.error.set_path(&["details", "send_id"], json!(send_id));
+    failure.failure.error.set_path(&["details", "phase"], json!(phase));
+    if failure.failure.kind == FailureKind::InProgress {
+        let original = failure.failure.error["message"]
+            .as_str()
+            .unwrap_or("terminal write acknowledgement lost")
+            .to_owned();
+        failure.failure.error.set(
+            "message",
+            json!(format!("{original}; {phase} may have applied; retry with --idempotency-key {send_id}")),
+        );
+        failure.failure.error.set("retryable", json!(true));
+    }
+    failure.failure.idempotency_key = Some(send_id.to_owned());
+    failure
+}
+
 /// The daemon's session snapshot (`session.snapshot`).
 pub(in crate::cli) fn snapshot<B: Daemon + ?Sized>(
     backend: &B,
@@ -336,6 +372,15 @@ pub(in crate::cli) fn run<B: Daemon + ?Sized>(
             1,
         ));
     }
+    let send_id = match request.idempotency_key.as_deref() {
+        Some(key) => {
+            cmux_tui_core::resource::validate_idempotency_key(key)
+                .map_err(|error| SendFailure::local("usage.invalid", error.message, json!({}), 2))?;
+            key.to_owned()
+        }
+        None => super::super::wire::random_idempotency_key()
+            .map_err(|error| SendFailure::local("usage.invalid", error.0, json!({}), 2))?,
+    };
     let state = backend.request(session, plan(&["terminal", &id, "state", "read"])?)?;
     let vt = state["state_base64"]
         .as_str()
@@ -344,18 +389,23 @@ pub(in crate::cli) fn run<B: Daemon + ?Sized>(
     let paste = reply::paste_mode_on(&vt);
     let engine = &base64::engine::general_purpose::STANDARD;
     let typed = engine.encode(reply::input_bytes(&request.text, paste));
-    let write = |encoded: &str| -> Result<(), SendFailure> {
-        let plan = mutation(&["terminal", &id, "write", "--bytes-base64", encoded])?;
-        backend.request(session, plan)?;
+    let write = |encoded: &str, stage: &str| -> Result<(), SendFailure> {
+        let plan = mutation(
+            &["terminal", &id, "write", "--bytes-base64", encoded],
+            stage_key(&send_id, &id, stage),
+        )?;
+        backend.request(session, plan)
+            .map_err(|error| stage_failure(error.into(), &send_id, stage))?;
         Ok(())
     };
     let sent_ms = super::now_ms();
-    write(&typed)?;
+    write(&typed, "paste")?;
     std::thread::sleep(request.enter_delay);
-    write(&engine.encode(b"\r"))?;
+    write(&engine.encode(b"\r"), "enter")?;
     let sent_at = Instant::now();
     let mut receipt = json!({
         "terminal": id,
+        "idempotency_key": send_id,
         "harness": agent.as_ref().map(|agent| agent.harness.clone()),
         "bracketed_paste": paste,
         "wait": request.wait.as_str(),

cmux-tui/crates/cmux-tui/src/cli/terminal_talk/tests.rs

@@ -1,11 +1,12 @@
-use std::cell::RefCell;
+use std::cell::{Cell, RefCell};
+use std::collections::HashSet;
 use std::time::{Duration, Instant};
 
 use base64::Engine as _;
 use serde_json::{Value, json};
 
 use super::super::command::RequestPlan;
-use super::super::mcp::CallFailure;
+use super::super::mcp::{CallFailure, FailureKind};
 use super::super::wire::request_value;
 use super::reply::{self, Reply};
 use super::roster::{self, AgentState};
@@ -294,6 +295,94 @@ fn quick(text: &str) -> SendRequest {
     }
 }
 
+/// Simulate the daemon applying one write then dropping its acknowledgement.
+struct LostAckDaemon {
+    inner: FakeDaemon,
+    keys: RefCell<HashSet<String>>,
+    applied: Cell<usize>,
+    lose_on: usize,
+}
+
+impl LostAckDaemon {
+    fn new(lose_on: usize) -> Self {
+        Self {
+            inner: FakeDaemon::new(snapshot(), &[(1, "")]),
+            keys: RefCell::new(HashSet::new()),
+            applied: Cell::new(0),
+            lose_on,
+        }
+    }
+}
+
+impl Daemon for LostAckDaemon {
+    fn request(&self, session: Option<&str>, plan: RequestPlan) -> Result<Value, CallFailure> {
+        let wire = request_value(&plan).expect("valid request");
+        if wire["operation"] != "terminal.input.write" {
+            return self.inner.request(session, plan);
+        }
+        let key = plan.idempotency_key.clone().expect("each write carries a replay key");
+        if !self.keys.borrow_mut().insert(key.clone()) {
+            // The owner replays the accepted result instead of applying the bytes twice.
+            return Ok(json!({}));
+        }
+        let result = self.inner.request(session, plan)?;
+        let count = self.applied.get() + 1;
+        self.applied.set(count);
+        if count == self.lose_on {
+            return Err(CallFailure {
+                kind: FailureKind::InProgress,
+                error: json!({"code": "transport.failed", "message": "reply lost",
+                              "details": {}, "retryable": true}),
+                idempotency_key: Some(key),
+            });
+        }
+        Ok(result)
+    }
+}
+
+#[test]
+fn terminal_send_retry_replays_both_writes_after_either_lost_ack() {
+    for lost_at in [1, 2] {
+        let daemon = LostAckDaemon::new(lost_at);
+        let request = SendRequest {
+            wait: Wait::None,
+            enter_delay: Duration::ZERO,
+            idempotency_key: Some("repeat-this-send".into()),
+            ..quick("echo one")
+        };
+        let failed = send::run(&daemon, None, &request).expect_err("lost acknowledgement");
+        assert_eq!(failed.failure.idempotency_key.as_deref(), Some("repeat-this-send"));
+        assert_eq!(failed.failure.error["details"]["send_id"], "repeat-this-send");
+        assert_eq!(
+            failed.failure.error["details"]["phase"],
+            if lost_at == 1 { "paste" } else { "enter" }
+        );
+        let replay = send::run(&daemon, None, &request).expect("replayed the original send");
+        assert_eq!(replay["idempotency_key"], "repeat-this-send");
+        assert_eq!(replay["sent"], true);
+        assert_eq!(daemon.applied.get(), 2, "one paste and one Enter, exactly");
+        assert_eq!(daemon.keys.borrow().len(), 2);
+        assert_eq!(
+            daemon.inner.writes(),
+            [b"\x1b[200~echo one\x1b[201~".to_vec(), b"\r".to_vec()],
+            "each stage applies once after an uncertain reply"
+        );
+    }
+}
+
+#[test]
+fn terminal_send_generated_key_is_returned_on_a_lost_reply() {
+    let daemon = LostAckDaemon::new(1);
+    let request = SendRequest { wait: Wait::None, enter_delay: Duration::ZERO, ..quick("hello") };
+    let failed = send::run(&daemon, None, &request).expect_err("lost acknowledgement");
+    let key = failed.failure.idempotency_key.expect("whole-send recovery identity");
+    assert_eq!(failed.failure.error["details"]["send_id"], key);
+    let replay = SendRequest { idempotency_key: Some(key.clone()), ..request };
+    let finished = send::run(&daemon, None, &replay).expect("same send retried");
+    assert_eq!(finished["idempotency_key"], key);
+    assert_eq!(daemon.applied.get(), 2, "one paste and one Enter");
+}
+
 #[test]
 fn send_pastes_presses_enter_waits_and_answers_only_the_reply() {
     let answer = "> what is 2+2\n\n⏺ 4\n\n──────────────────────\n> \n──────────────────────";