feat(tui): stream staged uploads over websocket sftp

This commit is contained in:
Edward Frazer
2026-05-07 13:40:43 -07:00
parent 80628db404
commit 7bd95aa648
10 changed files with 823 additions and 2 deletions
+105
View File
@@ -5,6 +5,11 @@
//! the main event loop remains single-threaded.
use super::*;
use base64::Engine;
use codex_app_server_protocol::FsCreateUploadParams;
use codex_app_server_protocol::FsCreateUploadResponse;
use codex_app_server_protocol::FsWriteFileParams;
use codex_app_server_protocol::FsWriteFileResponse;
use codex_app_server_protocol::HookTrustStatus;
use codex_app_server_protocol::MarketplaceAddParams;
use codex_app_server_protocol::MarketplaceAddResponse;
@@ -16,6 +21,16 @@ use codex_app_server_protocol::MarketplaceUpgradeResponse;
use codex_app_server_protocol::RequestId;
use codex_utils_absolute_path::AbsolutePathBuf;
use serde::Deserialize;
use tokio::io::AsyncReadExt;
const MAX_UPLOAD_FILE_BYTES: u64 = 45 * 1024 * 1024;
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct RemoteFsCreateUploadResponse {
path: String,
}
impl App {
pub(super) fn fetch_mcp_inventory(
@@ -141,6 +156,17 @@ impl App {
});
}
pub(super) fn upload_local_file(&mut self, app_server: &AppServerSession, local_path: PathBuf) {
let request_handle = app_server.request_handle();
let app_event_tx = self.app_event_tx.clone();
tokio::spawn(async move {
let result = upload_local_file_request(request_handle, &local_path)
.await
.map_err(|err| err.to_string());
app_event_tx.send(AppEvent::LocalFileUploaded { local_path, result });
});
}
pub(super) fn fetch_hooks_list(&mut self, app_server: &AppServerSession, cwd: PathBuf) {
let request_handle = app_server.request_handle();
let app_event_tx = self.app_event_tx.clone();
@@ -604,6 +630,74 @@ pub(super) async fn fetch_all_mcp_server_statuses(
Ok(statuses)
}
pub(super) async fn upload_local_file_request(
request_handle: AppServerRequestHandle,
local_path: &Path,
) -> Result<PathBuf> {
let file_name = local_path
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| color_eyre::eyre::eyre!("Upload path must name a file."))?
.to_string();
let metadata = tokio::fs::metadata(local_path).await?;
if !metadata.is_file() {
color_eyre::eyre::bail!("Upload path must name a regular file.");
}
ensure_upload_file_size(metadata.len())?;
let file = tokio::fs::File::open(local_path).await?;
let mut bytes = Vec::new();
file.take(MAX_UPLOAD_FILE_BYTES + 1)
.read_to_end(&mut bytes)
.await?;
ensure_upload_file_size(bytes.len() as u64)?;
match &request_handle {
AppServerRequestHandle::Remote(_) => {
let request_id = RequestId::String(format!("upload-local-file-{}", Uuid::new_v4()));
let response: RemoteFsCreateUploadResponse = request_handle
.request_typed(ClientRequest::FsCreateUpload {
request_id,
params: FsCreateUploadParams { file_name },
})
.await
.wrap_err("fs/createUpload failed in TUI")?;
request_handle
.upload_file_over_sftp(&response.path, &bytes)
.await
.wrap_err("staged upload stream failed in TUI")?;
Ok(PathBuf::from(response.path))
}
AppServerRequestHandle::InProcess(_) => {
let request_id = RequestId::String(format!("upload-local-file-{}", Uuid::new_v4()));
let response: FsCreateUploadResponse = request_handle
.request_typed(ClientRequest::FsCreateUpload {
request_id,
params: FsCreateUploadParams { file_name },
})
.await
.wrap_err("fs/createUpload failed in TUI")?;
let request_id = RequestId::String(format!("write-local-upload-{}", Uuid::new_v4()));
request_handle
.request_typed::<FsWriteFileResponse>(ClientRequest::FsWriteFile {
request_id,
params: FsWriteFileParams {
path: response.path.clone(),
data_base64: base64::engine::general_purpose::STANDARD.encode(bytes),
},
})
.await
.wrap_err("fs/writeFile failed for local staged upload in TUI")?;
Ok(response.path.into())
}
}
}
fn ensure_upload_file_size(file_size: u64) -> Result<()> {
if file_size > MAX_UPLOAD_FILE_BYTES {
color_eyre::eyre::bail!("Uploads accept files up to {MAX_UPLOAD_FILE_BYTES} bytes.");
}
Ok(())
}
pub(super) async fn fetch_account_rate_limits(
request_handle: AppServerRequestHandle,
) -> Result<Vec<RateLimitSnapshot>> {
@@ -1040,6 +1134,17 @@ mod tests {
);
}
#[test]
fn upload_rejects_files_larger_than_the_upload_limit() {
let err = ensure_upload_file_size(MAX_UPLOAD_FILE_BYTES + 1)
.expect_err("file should exceed upload limit");
assert_eq!(
err.to_string(),
format!("Uploads accept files up to {MAX_UPLOAD_FILE_BYTES} bytes.")
);
}
#[test]
fn mcp_inventory_maps_prefix_tool_names_by_server() {
let statuses = vec![
+15
View File
@@ -666,6 +666,21 @@ impl App {
AppEvent::FileSearchResult { query, matches } => {
self.chat_widget.apply_file_search_result(query, matches);
}
AppEvent::UploadLocalFile { path } => {
self.upload_local_file(app_server, path);
}
AppEvent::LocalFileUploaded { local_path, result } => match result {
Ok(remote_path) => {
self.chat_widget.insert_uploaded_file_path(&remote_path);
self.chat_widget.maybe_send_next_queued_input();
}
Err(err) => {
self.chat_widget.add_error_message(format!(
"Failed to upload '{}': {err}",
local_path.display()
));
}
},
AppEvent::RefreshRateLimits { origin } => {
self.refresh_rate_limits(app_server, origin);
}
+11
View File
@@ -229,6 +229,17 @@ pub(crate) enum AppEvent {
matches: Vec<FileMatch>,
},
/// Upload a local client file into Codex-managed app-server storage.
UploadLocalFile {
path: PathBuf,
},
/// Result of uploading a local client file into Codex-managed app-server storage.
LocalFileUploaded {
local_path: PathBuf,
result: Result<PathBuf, String>,
},
/// Refresh account rate limits in the background.
RefreshRateLimits {
origin: RateLimitRefreshOrigin,
+21
View File
@@ -10472,6 +10472,27 @@ impl ChatWidget {
self.bottom_pane.insert_str(text);
}
pub(crate) fn insert_uploaded_file_path(&mut self, path: &Path) {
let path = path.to_string_lossy();
let queued_message_index = self.queued_user_messages.iter().position(|queued_message| {
self.queued_message_accepts_uploaded_file_path(queued_message)
});
if let Some(queued_message) =
queued_message_index.and_then(|index| self.queued_user_messages.get_mut(index))
{
if !queued_message.user_message.text.is_empty() {
queued_message.user_message.text.push(' ');
}
queued_message.user_message.text.push_str(&path);
self.refresh_pending_input_preview();
return;
}
if !self.bottom_pane.composer_text().is_empty() {
self.bottom_pane.insert_str(" ");
}
self.bottom_pane.insert_str(&path);
}
/// Replace the composer content with the provided text and reset cursor.
pub(crate) fn set_composer_text(
&mut self,
@@ -33,6 +33,7 @@ const SIDE_SLASH_COMMAND_UNAVAILABLE_HINT: &str = "Press Esc to return to the ma
const GOAL_USAGE: &str = "Usage: /goal <objective>";
const GOAL_USAGE_HINT: &str = "Example: /goal improve benchmark coverage";
const RAW_USAGE: &str = "Usage: /raw [on|off]";
const UPLOAD_USAGE: &str = "Usage: /upload <local-path>";
impl ChatWidget {
/// Dispatch a bare slash command and record its staged local-history entry.
@@ -393,6 +394,9 @@ impl ChatWidget {
SlashCommand::Ps => {
self.add_ps_output();
}
SlashCommand::Upload => {
self.add_info_message(UPLOAD_USAGE.to_string(), /*hint*/ None);
}
SlashCommand::Stop => {
self.clean_background_terminals();
}
@@ -776,6 +780,16 @@ impl ChatWidget {
self.app_event_tx
.send(AppEvent::BeginWindowsSandboxGrantReadRoot { path: args });
}
SlashCommand::Upload if !trimmed.is_empty() => {
let path = PathBuf::from(trimmed);
self.app_event_tx.send(AppEvent::UploadLocalFile {
path: if path.is_absolute() {
path
} else {
self.config.cwd.join(path).to_path_buf()
},
});
}
_ => self.dispatch_command(cmd),
}
if source == SlashCommandDispatchSource::Live && cmd != SlashCommand::Goal {
@@ -826,6 +840,9 @@ impl ChatWidget {
if rest.is_empty() {
self.dispatch_command(cmd);
if cmd == SlashCommand::Upload {
return QueueDrain::Continue;
}
return self.queued_command_drain_result(cmd);
}
@@ -890,6 +907,35 @@ impl ChatWidget {
}
}
pub(super) fn queued_message_accepts_uploaded_file_path(
&self,
queued_message: &QueuedUserMessage,
) -> bool {
match queued_message.action {
QueuedInputAction::Plain => true,
QueuedInputAction::RunShell => false,
QueuedInputAction::ParseSlash => {
let Some((name, rest, _rest_offset)) = parse_slash_name(&queued_message.text)
else {
return true;
};
if name.contains('/') {
return true;
}
let Some(cmd) =
slash_commands::find_builtin_command(name, self.builtin_command_flags())
else {
return false;
};
if rest.trim().is_empty() {
return false;
}
matches!(cmd, SlashCommand::Plan | SlashCommand::Side)
|| !cmd.supports_inline_args()
}
}
}
fn queued_command_drain_result(&self, cmd: SlashCommand) -> QueueDrain {
if self.is_user_turn_pending_or_running() || !self.bottom_pane.no_modal_or_popup_active() {
return QueueDrain::Stop;
@@ -938,6 +984,7 @@ impl ChatWidget {
| SlashCommand::Experimental
| SlashCommand::AutoReview
| SlashCommand::Memories
| SlashCommand::Upload
| SlashCommand::Quit
| SlashCommand::Exit
| SlashCommand::Logout
@@ -1950,6 +1950,163 @@ async fn queued_fast_slash_applies_before_next_queued_message() {
}
}
#[tokio::test]
async fn queued_upload_stops_before_follow_up_prompt() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
handle_turn_started(&mut chat, "turn-1");
queue_composer_text_with_tab(&mut chat, "/upload /tmp/demo.txt");
queue_composer_text_with_tab(&mut chat, "use the upload");
complete_turn_with_message(&mut chat, "turn-1", Some("done"));
let events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
assert!(
events.iter().any(|event| matches!(
event,
AppEvent::UploadLocalFile { path } if path == &test_path_buf("/tmp/demo.txt")
)),
"expected queued /upload to start upload; events: {events:?}"
);
assert_eq!(chat.queued_user_messages.len(), 1);
assert_eq!(
chat.queued_user_messages.front().unwrap().text,
"use the upload"
);
assert_matches!(op_rx.try_recv(), Err(TryRecvError::Empty));
}
#[tokio::test]
async fn upload_resolves_relative_paths_against_session_cwd() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
let tempdir = tempdir().unwrap();
chat.config.cwd = tempdir.path().to_path_buf().abs();
submit_composer_text(&mut chat, "/upload demo.txt");
let events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
assert!(
events.iter().any(|event| matches!(
event,
AppEvent::UploadLocalFile { path }
if path == &chat.config.cwd.join("demo.txt").to_path_buf()
)),
"expected relative upload to use the session cwd; events: {events:?}"
);
}
#[tokio::test]
async fn queued_upload_resumes_follow_up_prompt_after_upload_succeeds() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
handle_turn_started(&mut chat, "turn-1");
queue_composer_text_with_tab(&mut chat, "/upload /tmp/demo.txt");
queue_composer_text_with_tab(&mut chat, "use the upload");
complete_turn_with_message(&mut chat, "turn-1", Some("done"));
let _events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
chat.insert_uploaded_file_path(Path::new("/tmp/.codex/uploads/demo.txt"));
chat.maybe_send_next_queued_input();
match next_submit_op(&mut op_rx) {
Op::UserTurn { items, .. } => assert_eq!(
items,
vec![UserInput::Text {
text: "use the upload /tmp/.codex/uploads/demo.txt".to_string(),
text_elements: Vec::new(),
}]
),
other => panic!("expected queued follow-up prompt to submit, got {other:?}"),
}
}
#[tokio::test]
async fn queued_upload_attaches_to_follow_up_plan_prompt() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
chat.set_feature_enabled(Feature::CollaborationModes, /*enabled*/ true);
handle_turn_started(&mut chat, "turn-1");
queue_composer_text_with_tab(&mut chat, "/upload /tmp/demo.txt");
queue_composer_text_with_tab(&mut chat, "/plan use the upload");
complete_turn_with_message(&mut chat, "turn-1", Some("done"));
let _events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
chat.insert_uploaded_file_path(Path::new("/tmp/.codex/uploads/demo.txt"));
chat.maybe_send_next_queued_input();
match next_submit_op(&mut op_rx) {
Op::UserTurn { items, .. } => assert_eq!(
items,
vec![UserInput::Text {
text: "use the upload /tmp/.codex/uploads/demo.txt".to_string(),
text_elements: Vec::new(),
}]
),
other => panic!("expected queued /plan prompt to submit, got {other:?}"),
}
}
#[tokio::test]
async fn queued_upload_skips_command_only_follow_ups_before_plain_prompt() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
handle_turn_started(&mut chat, "turn-1");
queue_composer_text_with_tab(&mut chat, "/upload /tmp/first.txt");
queue_composer_text_with_tab(&mut chat, "/upload /tmp/second.txt");
queue_composer_text_with_tab(&mut chat, "use both uploads");
complete_turn_with_message(&mut chat, "turn-1", Some("done"));
let _events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
chat.insert_uploaded_file_path(Path::new("/tmp/.codex/uploads/first.txt"));
assert_eq!(chat.queued_user_messages.len(), 2);
assert_eq!(
chat.queued_user_messages.front().unwrap().text,
"/upload /tmp/second.txt"
);
assert_eq!(
chat.queued_user_messages.back().unwrap().text,
"use both uploads /tmp/.codex/uploads/first.txt"
);
}
#[tokio::test]
async fn queued_upload_without_args_keeps_draining() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
handle_turn_started(&mut chat, "turn-1");
queue_composer_text_with_tab(&mut chat, "/upload");
queue_composer_text_with_tab(&mut chat, "after upload help");
complete_turn_with_message(&mut chat, "turn-1", Some("done"));
let events = std::iter::from_fn(|| rx.try_recv().ok()).collect::<Vec<_>>();
assert!(
!events
.iter()
.any(|event| matches!(event, AppEvent::UploadLocalFile { .. })),
"bare queued /upload should not start an upload; events: {events:?}"
);
match next_submit_op(&mut op_rx) {
Op::UserTurn { items, .. } => assert_eq!(
items,
vec![UserInput::Text {
text: "after upload help".to_string(),
text_elements: Vec::new(),
}]
),
other => panic!("expected follow-up prompt after bare /upload, got {other:?}"),
}
}
#[tokio::test]
async fn user_turn_sends_standard_override_after_fast_is_turned_off() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(Some("gpt-5.3-codex")).await;
+4
View File
@@ -58,6 +58,7 @@ pub enum SlashCommand {
Feedback,
Rollout,
Ps,
Upload,
#[strum(to_string = "stop", serialize = "clean")]
Stop,
Clear,
@@ -100,6 +101,7 @@ impl SlashCommand {
SlashCommand::Statusline => "configure which items appear in the status line",
SlashCommand::Theme => "choose a syntax highlighting theme",
SlashCommand::Ps => "list background terminals",
SlashCommand::Upload => "upload a local file to the app-server host",
SlashCommand::Stop => "stop all background terminals",
SlashCommand::MemoryDrop => "DO NOT USE",
SlashCommand::MemoryUpdate => "DO NOT USE",
@@ -159,6 +161,7 @@ impl SlashCommand {
| SlashCommand::Side
| SlashCommand::Resume
| SlashCommand::SandboxReadRoot
| SlashCommand::Upload
)
}
@@ -209,6 +212,7 @@ impl SlashCommand {
| SlashCommand::Status
| SlashCommand::DebugConfig
| SlashCommand::Ps
| SlashCommand::Upload
| SlashCommand::Stop
| SlashCommand::Goal
| SlashCommand::Mcp