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
3 changes: 3 additions & 0 deletions codex-rs/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 3 additions & 0 deletions codex-rs/app-server-client/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -13,17 +13,20 @@ doctest = false
workspace = true

[dependencies]
codex-api = { workspace = true }
codex-app-server = { workspace = true }
codex-app-server-protocol = { workspace = true }
codex-app-server-transport = { workspace = true }
codex-arg0 = { workspace = true }
codex-backend-client = { workspace = true }
codex-cloud-client = { path = "../cloud-client" }
codex-config = { workspace = true }
codex-login = { workspace = true }
codex-core = { workspace = true }
codex-exec-server = { workspace = true }
codex-feedback = { workspace = true }
codex-http-client = { workspace = true }
codex-model-provider = { workspace = true }
codex-protocol = { workspace = true }
codex-uds = { workspace = true }
codex-utils-absolute-path = { workspace = true }
Expand Down
66 changes: 57 additions & 9 deletions codex-rs/app-server-client/src/cloud.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,14 @@
use crate::AppServerEvent;
use crate::RequestResult;
use crate::TypedRequestError;
use crate::cloud_backend::CloudTitleClient;
use codex_app_server_protocol::ClientNotification;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::ModelListParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::Result as JsonRpcResult;
use codex_app_server_protocol::ThreadSetNameParams;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnStartParams;
use std::collections::HashMap;
Expand Down Expand Up @@ -40,6 +42,8 @@ pub enum CloudOperation {
project: codex_cloud_client::ProjectEnvironment,
},
SubmitUserMessage(Box<TurnStartParams>),
/// Apply the result of a name write that already succeeded outside the command queue.
ThreadNameSaved(ThreadSetNameParams),
Notify(ClientNotification),
Resolve {
request_id: RequestId,
Expand Down Expand Up @@ -81,7 +85,8 @@ pub(super) enum IdleReconnect {
pub struct CloudRequestHandle {
pub(super) reconnect: tokio::sync::watch::Sender<IdleReconnect>,
commands: mpsc::Sender<CloudCommand>,
catalog: Option<Arc<codex_cloud_client::Client>>,
native: Option<Arc<codex_cloud_client::Client>>,
title_client: Option<Arc<CloudTitleClient>>,
}

/// Tracks only losses queued for delivery. The worker remains the sole owner
Expand Down Expand Up @@ -147,12 +152,13 @@ impl CloudClient {
mpsc::Receiver<CloudCommand>,
mpsc::UnboundedSender<AppServerEvent>,
) {
Self::channel_with_catalog(capacity, /*catalog*/ None)
Self::channel_with_native(capacity, /*native*/ None, /*title_client*/ None)
}

pub(super) fn channel_with_catalog(
pub(super) fn channel_with_native(
capacity: usize,
catalog: Option<codex_cloud_client::Client>,
native: Option<codex_cloud_client::Client>,
title_client: Option<CloudTitleClient>,
) -> (
Self,
mpsc::Receiver<CloudCommand>,
Expand All @@ -166,7 +172,8 @@ impl CloudClient {
handle: CloudRequestHandle {
commands,
reconnect: Default::default(),
catalog: catalog.map(Arc::new),
native: native.map(Arc::new),
title_client: title_client.map(Arc::new),
},
events: event_receiver,
},
Expand Down Expand Up @@ -237,6 +244,47 @@ impl CloudClient {
}

impl CloudRequestHandle {
/// Generate metadata with Cloud's own account, outside its conversation command queue.
pub async fn generate_thread_title(
&self,
model: &str,
effort: Option<codex_protocol::openai_models::ReasoningEffort>,
prompt: String,
schema: serde_json::Value,
) -> color_eyre::Result<String> {
let client = self
.title_client
.as_ref()
.ok_or_else(|| io::Error::other("Cloud title generation is unavailable"))?;
client.generate(model, effort, prompt, schema).await
}

/// Read the saved title even when another Cloud conversation owns the attachment.
pub async fn read_native_thread_name(
&self,
thread_id: &str,
) -> Option<Result<Option<String>, codex_cloud_client::Error>> {
Some(self.native.as_ref()?.read_thread_name(thread_id).await)
}

/// Save an automatic name without occupying the conversation command queue.
pub async fn save_native_thread_name(
&self,
params: ThreadSetNameParams,
) -> Option<color_eyre::Result<()>> {
let client = self.native.as_ref()?;
Some(
async {
client
.set_thread_name(&params.thread_id, &params.name)
.await?;
self.unit(CloudOperation::ThreadNameSaved(params)).await?;
Ok(())
}
.await,
)
}

/// Accept a quiet Attach only for the conversation displayed by the caller.
pub fn reconnect_idle_thread(&self, thread_id: &str) -> bool {
self.reconnect.send_if_modified(|state| {
Expand All @@ -252,7 +300,7 @@ impl CloudRequestHandle {

/// Whether a completed model read still belongs to the bound Cloud account.
pub fn model_catalog_is_current(&self) -> bool {
self.catalog
self.native
.as_ref()
.is_none_or(|catalog| catalog.model_catalog_is_current())
}
Expand Down Expand Up @@ -338,7 +386,8 @@ impl CloudRequestHandle {
Self {
commands,
reconnect: self.reconnect.clone(),
catalog: None,
native: None,
title_client: None,
},
))
}
Expand Down Expand Up @@ -375,8 +424,7 @@ impl CloudRequestHandle {

pub(crate) async fn request(&self, request: ClientRequest) -> io::Result<RequestResult> {
// Catalog reads must not occupy the serialized send and attachment lane.
if let (Some(catalog), ClientRequest::ModelList { params, .. }) = (&self.catalog, &request)
{
if let (Some(catalog), ClientRequest::ModelList { params, .. }) = (&self.native, &request) {
// Keep the catalog HTTP state out of the shared request future.
return Ok(Box::pin(Self::model_list(catalog, params)).await);
}
Expand Down
9 changes: 9 additions & 0 deletions codex-rs/app-server-client/src/cloud_backend/auth.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
//! Opens the internal Cloud preview with the selected ChatGPT credentials and local transport policy.
use super::CloudTitleClient;
use super::start_worker;
use crate::AppServerClient;
use crate::legacy_core::config::Config;
Expand Down Expand Up @@ -39,6 +40,13 @@ pub async fn connect(config: &Config) -> Result<AppServerClient> {
let permit = factory.network_policy().acquire(&origin)?;
let credentials = cloud::Credentials::new(&token, &account_id)?;
let client = cloud::Client::new(&factory, ORIGIN, credentials)?;
let title_client = CloudTitleClient::new(
factory,
&auth,
account_id.clone(),
config.chatgpt_backend_url(),
)
.ok();
Ok(start_worker(
client,
super::idle::Login {
Expand All @@ -58,5 +66,6 @@ pub async fn connect(config: &Config) -> Result<AppServerClient> {
requires_openai_auth: true,
workspace_routing: None,
},
title_client,
))
}
10 changes: 9 additions & 1 deletion codex-rs/app-server-client/src/cloud_backend/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ mod auth;
mod events;
mod idle;
mod requests;
mod thread_title;

use crate::AppServerClient;
use crate::AppServerEvent;
Expand All @@ -27,6 +28,7 @@ use tokio::sync::mpsc;
use tokio::sync::oneshot;

pub use auth::connect;
pub(super) use thread_title::CloudTitleClient;

type ReplyStream = BoxStream<'static, Result<cloud::Event, cloud::Error>>;
type OpenReplyStream = (ReplyStream, Option<cloud::Event>);
Expand Down Expand Up @@ -291,9 +293,10 @@ fn start_worker(
client: cloud::Client,
login: idle::Login,
account: protocol::GetAccountResponse,
title_client: Option<CloudTitleClient>,
) -> AppServerClient {
let (app_client, commands, output) =
CloudClient::channel_with_catalog(/*capacity*/ 32, Some(client.clone()));
CloudClient::channel_with_native(/*capacity*/ 32, Some(client.clone()), title_client);
let client = Arc::new(client);
let mut requests = requests::Requests {
client: Arc::clone(&client),
Expand Down Expand Up @@ -386,6 +389,11 @@ async fn run_worker_with_prepared(
let _ = command.response.send(Ok(Ok(Value::Null)));
return;
}
CloudOperation::ThreadNameSaved(params) => {
requests.thread_name_saved(params, output);
let _ = command.response.send(Ok(Ok(Value::Null)));
continue;
}
CloudOperation::Request(request) => {
Ok((*request, cloud::SubmissionMode::StartIfIdle, None))
}
Expand Down
44 changes: 27 additions & 17 deletions codex-rs/app-server-client/src/cloud_backend/requests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,27 @@ pub(super) enum Response {
}

impl Requests {
pub(super) fn thread_name_saved(
&mut self,
params: protocol::ThreadSetNameParams,
output: &mpsc::UnboundedSender<crate::AppServerEvent>,
) {
if let Some(loaded) = &mut self.loaded
&& loaded.thread.id == params.thread_id
{
loaded.thread.name = Some(params.name.clone());
}
send_notification(
output,
protocol::ServerNotification::ThreadNameUpdated(
protocol::ThreadNameUpdatedNotification {
thread_id: params.thread_id,
thread_name: Some(params.name),
},
),
);
}

pub(super) fn read_thread(
&self,
params: &protocol::ThreadReadParams,
Expand Down Expand Up @@ -72,23 +93,12 @@ impl Requests {
}
serde_json::to_value(page)?
}
Request::ThreadSetName { params, .. } => {
let name = params.name.trim();
self.client.set_thread_name(&params.thread_id, name).await?;
if let Some(loaded) = &mut self.loaded
&& loaded.thread.id == params.thread_id
{
loaded.thread.name = Some(name.to_owned());
}
send_notification(
output,
protocol::ServerNotification::ThreadNameUpdated(
protocol::ThreadNameUpdatedNotification {
thread_id: params.thread_id,
thread_name: Some(name.to_owned()),
},
),
);
Request::ThreadSetName { mut params, .. } => {
params.name = params.name.trim().to_owned();
self.client
.set_thread_name(&params.thread_id, &params.name)
.await?;
self.thread_name_saved(params, output);
serde_json::to_value(protocol::ThreadSetNameResponse {})?
}
Request::ThreadArchive { params, .. } => {
Expand Down
Loading
Loading