Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions cmux-tui/crates/cmux-tui/src/cli/app.rs
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ mod page;
mod run;
mod settings;
mod skew;
mod wait;

/// Scopes that belong to the app, whatever follows.
pub(super) const APP_SCOPES: &[&str] = &[
Expand Down Expand Up @@ -602,10 +603,10 @@ fn call(global: &GlobalArgs, stream: &mut UnixStream, command: AppCommand) -> Ra
{
eprintln!("{note}");
}
let response = match request_with_retry(stream, method, &params, timeout) {
let response = match wait::request(global, stream, method, &params, timeout) {
Ok(response) => response,
Err(error) => {
let code = failure("app.transport", &error, global.output, 3);
let code = super::wire::print_local_error(&error, global.output, 3);
report.finish(global.output);
return Ran::Done(code);
}
Expand Down Expand Up @@ -890,10 +891,9 @@ pub(super) fn send_line(stream: &mut UnixStream, value: &Value) -> Result<(), St
}

fn read_error(error: &std::io::Error, timeout: Option<Duration>) -> String {
let messages = &crate::localization::catalog().app_control;
match (error.kind(), timeout) {
(std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut, Some(timeout)) => {
messages.timeout.replace("{seconds}", &timeout.as_secs().to_string())
wait::timeout_message(timeout)
}
_ => error.to_string(),
}
Expand Down
205 changes: 205 additions & 0 deletions cmux-tui/crates/cmux-tui/src/cli/app/wait.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,205 @@
//! A request that may wait long for the app (an `action.run` that waits for
//! its work, a page wait, a screenshot): progress on stderr while it waits,
//! and a timeout that says who did not answer, where, and what to run next.

use std::io::IsTerminal as _;
use std::os::unix::net::UnixStream;
use std::path::Path;
use std::sync::mpsc::{self, RecvTimeoutError};
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

use serde_json::{Value, json};

use super::{GlobalArgs, OutputMode, SCREENSHOT_TIMEOUT, command_socket, request_with_retry};

/// Only the long waits (a screenshot's 35 s, a waiting run's 45 s) show
/// notes; a 5 s control-plane read answers or fails before a note helps.
const LONG_WAIT: Duration = SCREENSHOT_TIMEOUT;
/// No note for a wait shorter than this.
const FIRST_NOTE_AFTER: Duration = Duration::from_secs(2);
const NOTE_EVERY: Duration = Duration::from_secs(5);

/// [`request_with_retry`] with progress notes while it waits; a transport
/// failure is the `app.transport` error to print, typed by [`timed_out`]
/// when the app took the request and never answered.
pub(super) fn request(
global: &GlobalArgs,
stream: &mut UnixStream,
method: &str,
params: &Value,
timeout: Option<Duration>,
) -> Result<Result<Value, Value>, Value> {
let long = timeout.is_some_and(|timeout| timeout >= LONG_WAIT);
let timeout = effective_timeout(timeout);
let what = waiting_for(method, params);
let progress = Progress::start(&what, timeout.filter(|_| long), global.output);
let response = request_with_retry(stream, method, params, timeout);
let waited = progress.finish();
response.map_err(|error| match timeout {
Some(timeout) if error == timeout_message(timeout) => {
let socket = command_socket(global).ok().map(|socket| socket.path);
timed_out(socket.as_deref(), &next_command(global), method, &what, timeout, waited)
}
_ => {
json!({ "code": "app.transport", "message": error, "details": {}, "retryable": false })
}
})
}

/// The "app did not answer within N s" text of a read that timed out.
pub(super) fn timeout_message(timeout: Duration) -> String {
let messages = &crate::localization::catalog().app_control;
messages.timeout.replace("{seconds}", &timeout.as_secs().to_string())
}

/// `timeout`, or the test build's `CMUX_TEST_APP_WAIT_TIMEOUT_MS` for a
/// request that has one (a release build ignores it).
fn effective_timeout(timeout: Option<Duration>) -> Option<Duration> {
#[cfg(debug_assertions)]
if timeout.is_some()
&& let Some(ms) = std::env::var("CMUX_TEST_APP_WAIT_TIMEOUT_MS")
.ok()
.and_then(|value| value.parse::<u64>().ok())
{
return Some(Duration::from_millis(ms));
}
timeout
}

/// What to run when the app does not answer: `cmux identify` asks the app
/// the cheapest question there is, over the route this call took
/// (`--app-socket`, `--socket` or `--session`), so it reaches the same app.
fn next_command(global: &GlobalArgs) -> String {
let route = if let Some(path) = &global.app_socket {
format!(" --app-socket {}", quote(&path.to_string_lossy()))
} else if let Some(path) = &global.socket {
format!(" --socket {}", quote(&path.to_string_lossy()))
} else if let Some(session) = &global.session {
format!(" --session {}", quote(session))
} else {
String::new()
};
format!("cmux{route} identify")
}

/// `value` as one shell word: unchanged when it has only safe characters.
fn quote(value: &str) -> String {
let safe = |c: char| c.is_ascii_alphanumeric() || "/._-:@%+=,".contains(c);
if !value.is_empty() && value.chars().all(safe) {
return value.to_owned();
}
format!("'{}'", value.replace('\'', "'\\''"))
}

/// What the app is asked to do: the action of an `action.run`, else the
/// method.
fn waiting_for(method: &str, params: &Value) -> String {
match params.get("action").and_then(Value::as_str) {
Some(action) if method == "action.run" => action.to_owned(),
_ => method.to_owned(),
}
}

/// Prints a progress note every few seconds while a request waits, until it
/// is dropped. Notes go to stderr in human output, and in JSON output only
/// when stderr is a terminal, so a piped JSON error is never mixed with them.
struct Progress {
started: Instant,
stop: Option<mpsc::Sender<()>>,
thread: Option<JoinHandle<()>>,
}

impl Progress {
fn start(what: &str, timeout: Option<Duration>, output: OutputMode) -> Self {
let started = Instant::now();
let shows = match output {
OutputMode::Human => true,
OutputMode::Json | OutputMode::JsonLines => std::io::stderr().is_terminal(),
OutputMode::Quiet => false,
};
let Some(total) = timeout.filter(|total| shows && *total > FIRST_NOTE_AFTER) else {
return Self { started, stop: None, thread: None };
};
let (stop, stopped) = mpsc::channel::<()>();
let template = crate::localization::catalog().app_control.waiting_progress;
let what = what.to_owned();
let spawned = std::thread::Builder::new().name("app-wait-notes".into()).spawn(move || {
let mut next = FIRST_NOTE_AFTER;
while next < total {
let left = next.saturating_sub(started.elapsed());
match stopped.recv_timeout(left) {
Err(RecvTimeoutError::Timeout) => {}
Ok(()) | Err(RecvTimeoutError::Disconnected) => return,
}
eprintln!(
"{}",
template
.replace("{what}", &what)
.replace("{elapsed}", &started.elapsed().as_secs().to_string())
.replace("{total}", &total.as_secs().to_string())
);
next += NOTE_EVERY;
}
});
// No thread, no notes: the request itself still runs.
match spawned {
Ok(thread) => Self { started, stop: Some(stop), thread: Some(thread) },
Err(_) => Self { started, stop: None, thread: None },
}
}

/// Stops the notes before the answer or the error is printed.
fn finish(mut self) -> Duration {
self.stop_notes();
self.started.elapsed()
}

fn stop_notes(&mut self) {
drop(self.stop.take());
if let Some(thread) = self.thread.take() {
let _ = thread.join();
}
}
}

impl Drop for Progress {
fn drop(&mut self) {
self.stop_notes();
}
}

/// The `app.transport` error of a request the app took and never answered:
/// the owner, the socket, the request, the wait and the next command, as
/// fields of `details` beside the message.
fn timed_out(
socket: Option<&Path>,
next: &str,
method: &str,
what: &str,
timeout: Duration,
waited: Duration,
) -> Value {
let messages = &crate::localization::catalog().app_control;
let socket = socket.map(|path| path.display().to_string());
let message = messages
.timeout_next
.replace("{path}", socket.as_deref().unwrap_or("?"))
.replace("{what}", what)
.replace("{seconds}", &timeout.as_secs().to_string())
.replace("{next}", next);
json!({
"code": "app.transport",
"message": message,
"details": {
"owner": "app",
"socket": socket,
"method": method,
"waiting_for": what,
"timeout_seconds": timeout.as_secs(),
"waited_seconds": waited.as_secs(),
"next_command": next,
},
"retryable": false,
})
}
6 changes: 6 additions & 0 deletions cmux-tui/crates/cmux-tui/src/localization/app_control.rs
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ pub(crate) struct AppControlMessages {
pub wrong_kind: &'static str,
pub explicit_no_app: &'static str,
pub timeout: &'static str,
pub timeout_next: &'static str,
pub waiting_progress: &'static str,
pub closed: &'static str,
pub invalid_response: &'static str,
pub root_scopes: &'static str,
Expand Down Expand Up @@ -66,6 +68,8 @@ pub(super) const ENGLISH: AppControlMessages = AppControlMessages {
wrong_kind: "{path} is not a cmux app control socket (it did not answer as the cmux app); app commands need the app's socket: pass --app-socket PATH. cmux tried no other socket",
explicit_no_app: "this command needs the cmux app, and no cmux app owns the session that {flag} names; pass --app-socket PATH. cmux tried no other socket",
timeout: "the cmux app did not answer within {seconds} s",
timeout_next: "the cmux app at {path} did not answer {what} within {seconds} s. Run `{next}` to check that cmux is running and responsive, or reopen cmux.",
waiting_progress: "waiting for the cmux app to run {what} ({elapsed} s of {total} s)",
closed: "the cmux app closed the connection without an answer",
invalid_response: "the cmux app sent an answer this cmux cannot read",
acp_open_usage: "usage: cmux acp open SESSION [--pane PANE]",
Expand Down Expand Up @@ -107,6 +111,8 @@ pub(super) const JAPANESE: AppControlMessages = AppControlMessages {
wrong_kind: "{path} は cmux アプリの制御ソケットではありません (cmux アプリとして応答しませんでした)。アプリのコマンドにはアプリのソケットが必要です: --app-socket パス を指定してください。cmux は他のソケットを試していません",
explicit_no_app: "このコマンドには cmux アプリが必要ですが、{flag} で指定したセッションを持つ cmux アプリはありません。--app-socket パス を指定してください。cmux は他のソケットを試していません",
timeout: "cmux アプリが {seconds} 秒以内に応答しませんでした",
timeout_next: "{path} の cmux アプリは {seconds} 秒以内に {what} に応答しませんでした。`{next}` を実行して cmux が起動して応答しているか確認するか、cmux を開き直してください。",
waiting_progress: "cmux アプリが {what} を実行するのを待っています ({total} 秒中 {elapsed} 秒経過)",
closed: "cmux アプリが応答せずに接続を閉じました",
invalid_response: "cmux アプリの応答を読み取れません",
acp_open_usage: "使い方: cmux acp open セッション [--pane ペイン]",
Expand Down
95 changes: 95 additions & 0 deletions cmux-tui/crates/cmux-tui/tests/app_wait_progress.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
//! A waiting app command against an app that takes the request and never
//! answers: a person sees progress on stderr while the CLI waits, and the
//! timeout names the owner that did not answer, the socket tried and the next
//! command to run, also as fields of the JSON error (cx-cmux-agents item 3).
#![cfg(unix)]

use std::os::unix::net::UnixListener;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::{SystemTime, UNIX_EPOCH};

use serde_json::Value;

/// The test build's wait for `action.run`, instead of 45 s.
const WAIT_MS: &str = "3500";

fn spawn(socket: &Path, output: &[&str]) -> Child {
spawn_command(socket, output, &["action", "run", "window.new"])
}

fn spawn_command(socket: &Path, output: &[&str], command: &[&str]) -> Child {
Command::new(env!("CARGO_BIN_EXE_cmux-tui"))
.args(output)
.arg("--app-socket")
.arg(socket)
.args(command)
.env("CMUX_TEST_APP_WAIT_TIMEOUT_MS", WAIT_MS)
.env_remove("CMUX_TUI_TERMINAL_ID")
.env_remove("CMUX_AGENT_SESSION")
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap()
}

#[test]
fn a_silent_app_shows_progress_and_times_out_with_the_next_command() {
let stamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos();
let dir: PathBuf = Path::new("/tmp").join(format!("cmux-wait-{}-{stamp}", std::process::id()));
std::fs::create_dir_all(&dir).unwrap();
let socket = dir.join("app.sock");
// Never accepted: the connection waits in the backlog, the request is
// buffered, and no answer ever comes.
let listener = UnixListener::bind(&socket).unwrap();

let human = spawn(&socket, &[]);
let json = spawn(&socket, &["--json"]);
// A control-plane read (5 s outside tests) is short: no progress notes.
let identify = spawn_command(&socket, &[], &["identify"]);
let human = human.wait_with_output().unwrap();
let json = json.wait_with_output().unwrap();
let identify = identify.wait_with_output().unwrap();
drop(listener);
let _ = std::fs::remove_dir_all(&dir);

let stderr = String::from_utf8_lossy(&human.stderr);
assert_eq!(human.status.code(), Some(3), "{stderr}");
assert!(
stderr.contains("waiting for the cmux app to run window.new (2 s of 3 s)"),
"no progress line: {stderr}"
);
// The next command keeps the route this call took.
let next = format!("cmux --app-socket {} identify", socket.display());
assert!(stderr.contains(&next), "no routed next command: {stderr}");
let identify_stderr = String::from_utf8_lossy(&identify.stderr);
assert_eq!(identify.status.code(), Some(3), "{identify_stderr}");
assert!(
!identify_stderr.contains("waiting for"),
"a short read shows progress: {identify_stderr}"
);
assert!(human.stdout.is_empty());

// JSON output: stdout stays empty and every stderr line is a JSON object
// (a pipe is not a terminal, so no progress lines are mixed in): the
// error, then the run's idempotency key report.
assert_eq!(json.status.code(), Some(3));
assert!(json.stdout.is_empty());
let text = String::from_utf8_lossy(&json.stderr);
let lines = text
.lines()
.map(|line| serde_json::from_str::<Value>(line).unwrap_or_else(|_| panic!("{text}")))
.collect::<Vec<_>>();
assert_eq!(lines.len(), 2, "{text}");
assert_eq!(lines[1]["code"], "mutation.outcome_unknown", "{text}");
let error = &lines[0];
assert_eq!(error["code"], "app.transport", "{error}");
let details = &error["details"];
assert_eq!(details["owner"], "app", "{error}");
assert_eq!(details["socket"], socket.to_string_lossy().as_ref(), "{error}");
assert_eq!(details["method"], "action.run", "{error}");
assert_eq!(details["waiting_for"], "window.new", "{error}");
assert_eq!(details["timeout_seconds"], 3, "{error}");
assert_eq!(details["next_command"], next.as_str(), "{error}");
}
Loading