use std::collections::HashSet; use std::sync::Arc; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::Project; use codex_app_server_protocol::ProjectChangeType; use codex_app_server_protocol::ProjectChangedNotification; use codex_app_server_protocol::ProjectCreateParams; use codex_app_server_protocol::ProjectCreateResponse; use codex_app_server_protocol::ProjectDeleteParams; use codex_app_server_protocol::ProjectDeleteResponse; use codex_app_server_protocol::ProjectImportParams; use codex_app_server_protocol::ProjectImportResponse; use codex_app_server_protocol::ProjectListParams; use codex_app_server_protocol::ProjectListResponse; use codex_app_server_protocol::ProjectMoveParams; use codex_app_server_protocol::ProjectMoveResponse; use codex_app_server_protocol::ProjectReadParams; use codex_app_server_protocol::ProjectReadResponse; use codex_app_server_protocol::ProjectRoot; use codex_app_server_protocol::ProjectSortKey; use codex_app_server_protocol::ProjectUpdateParams; use codex_app_server_protocol::ProjectUpdateResponse; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::SortDirection; use codex_app_server_protocol::ThreadProjectUpdatedNotification; use codex_thread_store::CreateProjectParams as StoreCreateProjectParams; use codex_thread_store::ListProjectsParams as StoreListProjectsParams; use codex_thread_store::MoveProjectParams as StoreMoveProjectParams; use codex_thread_store::ProjectMoveOutcome; use codex_thread_store::ProjectSortKey as StoreProjectSortKey; use codex_thread_store::SortDirection as StoreSortDirection; use codex_thread_store::StoredProject; use codex_thread_store::StoredProjectRoot; use codex_thread_store::ThreadStore; use codex_thread_store::ThreadStoreError; use codex_thread_store::UpdateProjectParams as StoreUpdateProjectParams; use tokio::sync::Semaphore; use tokio::sync::SemaphorePermit; use super::thread_processor::THREAD_LIST_DEFAULT_LIMIT; use super::thread_processor::THREAD_LIST_MAX_LIMIT; use crate::error_code::internal_error; use crate::error_code::invalid_params; use crate::error_code::method_not_found; use crate::outgoing_message::OutgoingMessageSender; #[derive(Clone)] pub(crate) struct ProjectRequestProcessor { thread_store: Arc, outgoing: Arc, thread_list_state_permit: Arc, } impl ProjectRequestProcessor { pub(crate) fn new( thread_store: Arc, outgoing: Arc, thread_list_state_permit: Arc, ) -> Self { Self { thread_store, outgoing, thread_list_state_permit, } } pub(crate) async fn project_list( &self, params: ProjectListParams, ) -> Result, JSONRPCErrorError> { if params.sort_key.is_none() && params.sort_direction.is_some() { return Err(invalid_params("sortDirection requires sortKey")); } let sort_key = match params.sort_key.unwrap_or(ProjectSortKey::Position) { ProjectSortKey::Position => StoreProjectSortKey::Position, ProjectSortKey::RecencyAt => StoreProjectSortKey::RecencyAt, }; let sort_direction = match params.sort_direction { Some(SortDirection::Asc) => StoreSortDirection::Asc, Some(SortDirection::Desc) => StoreSortDirection::Desc, None => match sort_key { StoreProjectSortKey::Position => StoreSortDirection::Asc, StoreProjectSortKey::RecencyAt => StoreSortDirection::Desc, }, }; let page = self .thread_store .list_projects(StoreListProjectsParams { cursor: params.cursor, sort_key, sort_direction, limit: params .limit .map(|limit| limit as usize) .unwrap_or(THREAD_LIST_DEFAULT_LIMIT) .clamp(1, THREAD_LIST_MAX_LIMIT), }) .await .map_err(|error| project_store_error("project/list", error))?; Ok(Some( ProjectListResponse { data: page .projects .into_iter() .map(api_project) .collect::>()?, next_cursor: page.next_cursor, } .into(), )) } pub(crate) async fn project_read( &self, params: ProjectReadParams, ) -> Result, JSONRPCErrorError> { let project = self .thread_store .read_project(params.project_id.clone()) .await .map_err(|error| project_store_error("project/read", error))? .ok_or_else(|| invalid_params(format!("project not found: {}", params.project_id)))?; Ok(Some( ProjectReadResponse { project: api_project(project)?, } .into(), )) } pub(crate) async fn project_create( &self, params: ProjectCreateParams, ) -> Result, JSONRPCErrorError> { let (project, created, _) = self .create_project( params.name, params.roots, params.metadata, Vec::new(), params.idempotency_key, "project/create", ) .await?; if created { self.notify_project_changed(&project.id, ProjectChangeType::Created) .await; } Ok(Some(ProjectCreateResponse { project }.into())) } pub(crate) async fn project_import( &self, params: ProjectImportParams, ) -> Result, JSONRPCErrorError> { let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; let (project, created, thread_ids) = self .create_project( params.name, params.roots, params.metadata, params.threads.unwrap_or_default(), params.idempotency_key, "project/import", ) .await?; if created { self.notify_project_changed(&project.id, ProjectChangeType::Created) .await; self.notify_thread_projects(thread_ids, Some(project.id.clone())) .await; } Ok(Some(ProjectImportResponse { project }.into())) } pub(crate) async fn project_update( &self, params: ProjectUpdateParams, ) -> Result, JSONRPCErrorError> { let name = params.name.map(validate_name).transpose()?; let roots = params.roots.map(validate_roots).transpose()?; let updated = self .thread_store .update_project(StoreUpdateProjectParams { project_id: params.project_id.clone(), name, roots, metadata: params.metadata, }) .await .map_err(|error| project_store_error("project/update", error))? .ok_or_else(|| invalid_params(format!("project not found: {}", params.project_id)))?; let project = api_project(updated.project)?; if updated.changed { self.notify_project_changed(&project.id, ProjectChangeType::Updated) .await; } Ok(Some(ProjectUpdateResponse { project }.into())) } pub(crate) async fn project_move( &self, params: ProjectMoveParams, ) -> Result, JSONRPCErrorError> { let project_id = params.project_id; let outcome = self .thread_store .move_project(StoreMoveProjectParams { project_id: project_id.clone(), before_project_id: params.before_project_id, }) .await .map_err(|error| project_store_error("project/move", error))? .ok_or_else(|| invalid_params(format!("project not found: {project_id}")))?; if outcome == ProjectMoveOutcome::Moved { self.notify_project_changed(&project_id, ProjectChangeType::Updated) .await; } Ok(Some(ProjectMoveResponse {}.into())) } pub(crate) async fn project_delete( &self, params: ProjectDeleteParams, ) -> Result, JSONRPCErrorError> { let _thread_list_state_permit = self.acquire_thread_list_state_permit().await?; let deleted = self .thread_store .delete_project(params.project_id.clone()) .await .map_err(|error| project_store_error("project/delete", error))? .ok_or_else(|| invalid_params(format!("project not found: {}", params.project_id)))?; self.notify_project_changed(¶ms.project_id, ProjectChangeType::Deleted) .await; self.notify_thread_projects(deleted.affected_active_thread_ids, /*project_id*/ None) .await; Ok(Some(ProjectDeleteResponse {}.into())) } async fn create_project( &self, name: String, roots: Vec, metadata: Option>, thread_ids: Vec, idempotency_key: String, operation: &'static str, ) -> Result<(Project, bool, Vec), JSONRPCErrorError> { let idempotency_key = validate_idempotency_key(idempotency_key)?; let thread_ids = validate_thread_ids(thread_ids)?; let created = self .thread_store .create_project(StoreCreateProjectParams { name: validate_name(name)?, roots: validate_roots(roots)?, metadata: metadata.unwrap_or_default(), thread_ids: thread_ids.clone(), idempotency_key, }) .await .map_err(|error| project_store_error(operation, error))?; Ok((api_project(created.project)?, created.created, thread_ids)) } async fn notify_project_changed(&self, project_id: &str, change_type: ProjectChangeType) { self.outgoing .send_server_notification(ServerNotification::ProjectChanged( ProjectChangedNotification { project_id: project_id.to_string(), change_type, }, )) .await; } async fn notify_thread_projects(&self, thread_ids: Vec, project_id: Option) { for thread_id in thread_ids { self.outgoing .send_server_notification(ServerNotification::ThreadProjectUpdated( ThreadProjectUpdatedNotification { thread_id, project_id: project_id.clone(), }, )) .await; } } async fn acquire_thread_list_state_permit( &self, ) -> Result, JSONRPCErrorError> { self.thread_list_state_permit .acquire() .await .map_err(|err| { internal_error(format!("failed to acquire thread list state permit: {err}")) }) } } fn validate_name(name: String) -> Result { let name = name.trim().to_string(); if name.is_empty() { return Err(invalid_params("project name must not be empty")); } Ok(name) } fn validate_idempotency_key(key: String) -> Result { if key.trim().is_empty() { return Err(invalid_params("idempotencyKey must not be empty")); } if key.len() > 512 { return Err(invalid_params("idempotencyKey must be at most 512 bytes")); } Ok(key) } fn validate_roots(roots: Vec) -> Result, JSONRPCErrorError> { let mut logical = HashSet::new(); let mut canonical = HashSet::new(); roots .into_iter() .map(|root| { let path = codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path_checked( root.path.into_path_buf(), ) .map_err(|error| invalid_params(format!("invalid project root: {error}")))? .into_path_buf(); if !logical.insert(path.clone()) { return Err(invalid_params(format!( "duplicate project root: {}", path.display() ))); } if let Ok(resolved) = std::fs::canonicalize(&path) && !canonical.insert(resolved) { return Err(invalid_params(format!( "duplicate resolved project root: {}", path.display() ))); } Ok(StoredProjectRoot { path: path.display().to_string(), }) }) .collect() } fn validate_thread_ids(thread_ids: Vec) -> Result, JSONRPCErrorError> { let mut seen = HashSet::new(); for thread_id in &thread_ids { if !seen.insert(thread_id.clone()) { return Err(invalid_params(format!("duplicate thread id: {thread_id}"))); } } Ok(thread_ids) } fn api_project(project: StoredProject) -> Result { Ok(Project { id: project.id, name: project.name, roots: project .roots .into_iter() .map(|root| { Ok(ProjectRoot { path: codex_utils_absolute_path::AbsolutePathBuf::from_absolute_path(root.path) .map_err(|error| { internal_error(format!("stored project root is not absolute: {error}")) })?, }) }) .collect::>()?, metadata: project.metadata, position: project.position, created_at: project.created_at_ms / 1000, updated_at: project.updated_at_ms / 1000, recency_at: project .recency_at_ms .map(|timestamp| timestamp.div_euclid(/*rhs*/ 1000)), }) } fn project_store_error(operation: &'static str, error: ThreadStoreError) -> JSONRPCErrorError { match error { ThreadStoreError::Unsupported { .. } => { method_not_found(format!("{operation} is unavailable without sqlite state")) } ThreadStoreError::InvalidRequest { message } => invalid_params(message), ThreadStoreError::Internal { message } if message.contains("thread not found") || message.contains("project not found") => { invalid_params(message) } error => internal_error(format!("failed to run {operation}: {error}")), } }