diff --git a/src/apps/desktop/src/api/agentic_api.rs b/src/apps/desktop/src/api/agentic_api.rs index bf972e443b..aa8447ea92 100644 --- a/src/apps/desktop/src/api/agentic_api.rs +++ b/src/apps/desktop/src/api/agentic_api.rs @@ -9,7 +9,9 @@ use tauri::{AppHandle, State}; use crate::api::app_state::AppState; use crate::api::session_storage_path::desktop_effective_session_storage_path; -use crate::runtime::DesktopRuntimeContext; +use crate::runtime::{ + DesktopRuntimeContext, DesktopSessionApplicationError, DesktopSessionScopeRequest, +}; use crate::startup_trace::DesktopStartupTrace; use bitfun_agent_runtime::sdk::{ AgentDialogTurnRequest, AgentInputAttachment, AgentSessionModelUpdateRequest, @@ -50,6 +52,18 @@ const SESSION_VIEW_TOOL_RESULT_STRING_CHAR_LIMIT: usize = 16 * 1024; const SESSION_VIEW_TRUNCATED_MARKER: &str = "\n... Output truncated for session preview"; const SESSION_VIEW_OMITTED_MARKER: &str = "Output omitted from session preview"; +fn desktop_session_scope( + workspace_path: String, + remote_connection_id: Option, + remote_ssh_host: Option, +) -> DesktopSessionScopeRequest { + DesktopSessionScopeRequest { + workspace_path, + remote_connection_id, + remote_ssh_host, + } +} + #[derive(Debug, Deserialize)] #[serde(rename_all = "camelCase")] pub struct CreateSessionRequest { @@ -917,8 +931,7 @@ pub async fn update_session_model( #[tauri::command] pub async fn update_session_title( - coordinator: State<'_, Arc>, - app_state: State<'_, AppState>, + runtime: State<'_, DesktopRuntimeContext>, request: UpdateSessionTitleRequest, ) -> Result { let session_id = request.session_id.trim(); @@ -926,88 +939,53 @@ pub async fn update_session_title( return Err("session_id is required".to_string()); } - if coordinator - .get_session_manager() - .get_session(session_id) - .is_none() - { - let workspace_path = request - .workspace_path - .as_deref() - .map(str::trim) - .filter(|value| !value.is_empty()) - .ok_or_else(|| { - "workspace_path is required when the session is not loaded".to_string() - })?; - - let effective = desktop_effective_session_storage_path( - &app_state, - workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - - coordinator - .restore_session_from_storage_path(&effective, session_id) - .await - .map_err(|e| format!("Failed to restore session before renaming: {}", e))?; - } - - let result = coordinator - .update_session_title(session_id, &request.title) + let scope = request + .workspace_path + .filter(|workspace_path| !workspace_path.trim().is_empty()) + .map(|workspace_path| { + desktop_session_scope( + workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ) + }); + runtime + .session_application() + .rename_session(scope, session_id.to_string(), request.title) .await - .map_err(|e| format!("Failed to update session title: {}", e)); - - // Notify auto-sync: metadata changed (title) - let wp = request.workspace_path.as_deref().unwrap_or(""); - crate::api::remote_connect_api::notify_session_changed(session_id, wp); - - result + .map_err(|error| match error { + DesktopSessionApplicationError::Validation(message) => message, + DesktopSessionApplicationError::RestoreBeforeRename(message) => { + format!("Failed to restore session before renaming: {message}") + } + error => format!("Failed to update session title: {error}"), + }) } /// Load the session into the coordinator process when it exists on disk but is not in memory. /// Uses the same remote→local session path mapping as `restore_session`. #[tauri::command] pub async fn ensure_coordinator_session( - coordinator: State<'_, Arc>, - app_state: State<'_, AppState>, + runtime: State<'_, DesktopRuntimeContext>, request: EnsureCoordinatorSessionRequest, ) -> Result<(), String> { let session_id = request.session_id.trim(); if session_id.is_empty() { return Err("session_id is required".to_string()); } - if coordinator - .get_session_manager() - .get_session(session_id) - .is_some() - { - return Ok(()); - } - - let wp = request.workspace_path.trim(); - if wp.is_empty() { - return Err("workspace_path is required when the session is not loaded".to_string()); - } - - let effective = desktop_effective_session_storage_path( - &app_state, - wp, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let restore_result = if request.include_internal { - coordinator - .restore_internal_session_from_storage_path(&effective, session_id) - .await - } else { - coordinator - .restore_session_from_storage_path(&effective, session_id) - .await - }; - restore_result.map(|_| ()).map_err(|e| e.to_string()) + runtime + .session_application() + .ensure_session_loaded( + desktop_session_scope( + request.workspace_path.clone(), + request.remote_connection_id.clone(), + request.remote_ssh_host.clone(), + ), + session_id, + request.include_internal, + ) + .await + .map_err(|error| error.to_string()) } #[tauri::command] @@ -2022,63 +2000,48 @@ pub async fn cancel_tool( #[tauri::command] pub async fn delete_session( - coordinator: State<'_, Arc>, - app_state: State<'_, AppState>, + runtime: State<'_, DesktopRuntimeContext>, request: DeleteSessionRequest, ) -> Result<(), String> { - let effective_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - if let Some(acp_client_service) = app_state.acp_client_service.as_ref() { - acp_client_service - .release_bitfun_session(&request.session_id) - .await; - } - coordinator - .delete_session(&effective_path, &request.session_id) + runtime + .session_application() + .delete_session( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.session_id, + ) .await - .map_err(|e| format!("Failed to delete session: {}", e))?; - - // Notify auto-sync: tombstone this session on the relay - crate::api::remote_connect_api::notify_session_deleted(&request.session_id); - Ok(()) + .map_err(|error| format!("Failed to delete session: {error}")) } #[tauri::command] pub async fn restore_session( - coordinator: State<'_, Arc>, - app_state: State<'_, AppState>, + runtime: State<'_, DesktopRuntimeContext>, request: RestoreSessionRequest, ) -> Result { - let effective_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let session = if request.include_internal { - coordinator - .restore_internal_session_from_storage_path(&effective_path, &request.session_id) - .await - } else { - coordinator - .restore_session_from_storage_path(&effective_path, &request.session_id) - .await - } - .map_err(|e| format!("Failed to restore session: {}", e))?; + let session = runtime + .session_application() + .restore_session( + desktop_session_scope( + request.workspace_path.clone(), + request.remote_connection_id.clone(), + request.remote_ssh_host.clone(), + ), + &request.session_id, + request.include_internal, + ) + .await + .map_err(|error| format!("Failed to restore session: {error}"))?; Ok(session_to_response(session)) } #[tauri::command] pub async fn restore_session_view( - coordinator: State<'_, Arc>, - app_state: State<'_, AppState>, + runtime: State<'_, DesktopRuntimeContext>, startup_trace: State<'_, DesktopStartupTrace>, request: RestoreSessionRequest, ) -> Result { @@ -2089,72 +2052,36 @@ pub async fn restore_session_view( "restore_session_view request received: trace_id={}, session_id={}", trace_id, request.session_id ); - let path_started_at = Instant::now(); - let effective_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let resolve_storage_path_duration_ms = - path_started_at.elapsed().as_millis().min(u64::MAX as u128) as u64; - debug!( - "restore_session_view storage path resolved: trace_id={}, session_id={}, duration_ms={}", - trace_id, - request.session_id, - resolve_storage_path_duration_ms - ); - - let session_storage_path = effective_path; let tail_turn_count = request .tail_turn_count .filter(|count| *count > 0) .map(|count| count.min(16)); - let (session, mut turns, total_turn_count, mut timings) = - if let Some(tail_turn_count) = tail_turn_count { - if request.include_internal { - coordinator - .restore_internal_session_view_from_storage_path_tail_timed( - &session_storage_path, - &request.session_id, - tail_turn_count, - ) - .await - } else { - coordinator - .restore_session_view_from_storage_path_tail_timed( - &session_storage_path, - &request.session_id, - tail_turn_count, - ) - .await - } - } else if request.include_internal { - coordinator - .restore_internal_session_view_from_storage_path_timed( - &session_storage_path, - &request.session_id, - ) - .await - .map(|(session, turns, timings)| { - let total_turn_count = turns.len(); - (session, turns, total_turn_count, timings) - }) - } else { - coordinator - .restore_session_view_from_storage_path_timed( - &session_storage_path, - &request.session_id, - ) - .await - .map(|(session, turns, timings)| { - let total_turn_count = turns.len(); - (session, turns, total_turn_count, timings) - }) - } - .map_err(|e| format!("Failed to restore session view: {}", e))?; - timings.resolve_storage_path_duration_ms = resolve_storage_path_duration_ms; + let restored = runtime + .session_application() + .restore_session_view( + desktop_session_scope( + request.workspace_path.clone(), + request.remote_connection_id.clone(), + request.remote_ssh_host.clone(), + ), + &request.session_id, + request.include_internal, + tail_turn_count, + |resolve_storage_path_duration_ms| { + debug!( + "restore_session_view storage path resolved: trace_id={}, session_id={}, duration_ms={}", + trace_id, + request.session_id, + resolve_storage_path_duration_ms + ); + }, + ) + .await + .map_err(|error| format!("Failed to restore session view: {error}"))?; + let session = restored.session; + let mut turns = restored.turns; + let total_turn_count = restored.total_turn_count; + let timings = restored.timings; let loaded_turn_count = turns.len(); let is_partial = loaded_turn_count < total_turn_count; @@ -2209,8 +2136,7 @@ pub async fn restore_session_view( #[tauri::command] pub async fn restore_session_with_turns( - coordinator: State<'_, Arc>, - app_state: State<'_, AppState>, + runtime: State<'_, DesktopRuntimeContext>, request: RestoreSessionRequest, ) -> Result { let started_at = std::time::Instant::now(); @@ -2219,33 +2145,29 @@ pub async fn restore_session_with_turns( "restore_session_with_turns request received: trace_id={}, session_id={}", trace_id, request.session_id ); - let path_started_at = std::time::Instant::now(); - let effective_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - debug!( - "restore_session_with_turns storage path resolved: trace_id={}, session_id={}, duration_ms={}", - trace_id, - request.session_id, - path_started_at.elapsed().as_millis() - ); - let (session, turns) = if request.include_internal { - coordinator - .restore_internal_session_with_turns_from_storage_path( - &effective_path, - &request.session_id, - ) - .await - } else { - coordinator - .restore_session_with_turns_from_storage_path(&effective_path, &request.session_id) - .await - } - .map_err(|e| format!("Failed to restore session: {}", e))?; + let restored = runtime + .session_application() + .restore_session_with_turns( + desktop_session_scope( + request.workspace_path.clone(), + request.remote_connection_id.clone(), + request.remote_ssh_host.clone(), + ), + &request.session_id, + request.include_internal, + |resolve_storage_path_duration_ms| { + debug!( + "restore_session_with_turns storage path resolved: trace_id={}, session_id={}, duration_ms={}", + trace_id, + request.session_id, + resolve_storage_path_duration_ms + ); + }, + ) + .await + .map_err(|error| format!("Failed to restore session: {error}"))?; + let session = restored.session; + let turns = restored.turns; if log::log_enabled!(log::Level::Debug) { let payload_stats = restore_turn_payload_stats(&turns); diff --git a/src/apps/desktop/src/api/session_api.rs b/src/apps/desktop/src/api/session_api.rs index 2dd98b178a..89f8ba1bd9 100644 --- a/src/apps/desktop/src/api/session_api.rs +++ b/src/apps/desktop/src/api/session_api.rs @@ -2,80 +2,39 @@ use crate::api::app_state::AppState; use crate::api::session_storage_path::desktop_effective_session_storage_path; +use crate::runtime::{ + DesktopRuntimeContext, DesktopSessionApplicationError, DesktopSessionScopeRequest, + UiSessionMetadataField, +}; use crate::startup_trace::DesktopStartupTrace; use bitfun_core::agentic::persistence::{ - PersistenceManager, SessionBranchRequest, SessionBranchResult, SessionMetadataPage, + PersistenceManager, SessionBranchResult, SessionMetadataPage, }; use bitfun_core::infrastructure::PathManager; use bitfun_core::service::session::{ DialogTurnData, SessionKind, SessionMetadata, SessionStatus, SessionTranscriptExport, SessionTranscriptExportOptions, }; -use bitfun_core::service::session_usage::{ - generate_session_usage_report, SessionUsageReport, SessionUsageReportRequest, -}; +use bitfun_core::service::session_usage::SessionUsageReport; use serde::{Deserialize, Serialize}; use std::sync::Arc; use std::time::Instant; use tauri::State; -const UI_CUSTOM_METADATA_KEYS: [&str; 3] = ["titleSource", "titleKey", "titleParams"]; - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase")] -pub enum UiSessionMetadataField { - SessionName, - Tags, - Todos, - ReviewActionState, - UnreadCompletion, - NeedsUserAttention, - TitleMetadata, -} - -fn merge_ui_owned_session_metadata( - current: &mut SessionMetadata, - incoming: &SessionMetadata, - fields: &[UiSessionMetadataField], -) { - if fields.contains(&UiSessionMetadataField::SessionName) { - current.session_name = incoming.session_name.clone(); - } - if fields.contains(&UiSessionMetadataField::Tags) { - current.tags = incoming.tags.clone(); - } - if fields.contains(&UiSessionMetadataField::Todos) { - current.todos = incoming.todos.clone(); - } - if fields.contains(&UiSessionMetadataField::ReviewActionState) { - current.review_action_state = incoming.review_action_state.clone(); - } - if fields.contains(&UiSessionMetadataField::UnreadCompletion) { - current.unread_completion = incoming.unread_completion.clone(); - } - if fields.contains(&UiSessionMetadataField::NeedsUserAttention) { - current.needs_user_attention = incoming.needs_user_attention.clone(); +fn desktop_session_scope( + workspace_path: String, + remote_connection_id: Option, + remote_ssh_host: Option, +) -> DesktopSessionScopeRequest { + DesktopSessionScopeRequest { + workspace_path, + remote_connection_id, + remote_ssh_host, } +} - if fields.contains(&UiSessionMetadataField::TitleMetadata) { - let mut custom = current - .custom_metadata - .as_ref() - .and_then(serde_json::Value::as_object) - .cloned() - .unwrap_or_default(); - let incoming_custom = incoming - .custom_metadata - .as_ref() - .and_then(serde_json::Value::as_object); - for key in UI_CUSTOM_METADATA_KEYS { - custom.remove(key); - if let Some(value) = incoming_custom.and_then(|metadata| metadata.get(key)) { - custom.insert(key.to_string(), value.clone()); - } - } - current.custom_metadata = (!custom.is_empty()).then(|| serde_json::Value::Object(custom)); - } +fn desktop_session_error(error: DesktopSessionApplicationError) -> String { + error.to_string() } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -254,50 +213,49 @@ pub struct DeleteAllArchivedSessionsRequest { #[tauri::command] pub async fn list_persisted_sessions( request: ListPersistedSessionsRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result, String> { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - manager - .list_session_metadata(&workspace_path) + runtime + .session_application() + .list_persisted_sessions(desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + )) .await - .map_err(|e| format!("Failed to list persisted sessions: {}", e)) + .map_err(|error| { + format!( + "Failed to list persisted sessions: {}", + desktop_session_error(error) + ) + }) } #[tauri::command] pub async fn list_persisted_sessions_page( request: ListPersistedSessionsPageRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, startup_trace: State<'_, DesktopStartupTrace>, ) -> Result { let trace_started = Instant::now(); - let result = async { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), + let result = runtime + .session_application() + .list_persisted_sessions_page( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.cursor.as_deref(), + request.limit, ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - manager - .list_session_metadata_page(&workspace_path, request.cursor.as_deref(), request.limit) - .await - .map_err(|e| format!("Failed to list persisted session page: {}", e)) - } - .await; + .await + .map_err(|error| { + format!( + "Failed to list persisted session page: {}", + desktop_session_error(error) + ) + }); startup_trace.record_tauri_command_elapsed("list_persisted_sessions_page", None, trace_started); result } @@ -305,8 +263,7 @@ pub async fn list_persisted_sessions_page( #[tauri::command] pub async fn load_session_turns( request: LoadSessionTurnsRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, startup_trace: State<'_, DesktopStartupTrace>, ) -> Result, String> { let trace_started = Instant::now(); @@ -315,30 +272,24 @@ pub async fn load_session_turns( } else { "full" }; - let result = async { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), + let result = runtime + .session_application() + .load_session_turns( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + &request.session_id, + request.limit, ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - let turns = if let Some(limit) = request.limit { - manager - .load_recent_turns(&workspace_path, &request.session_id, limit) - .await - } else { - manager - .load_session_turns(&workspace_path, &request.session_id) - .await - }; - - turns.map_err(|e| format!("Failed to load session turns: {}", e)) - } - .await; + .await + .map_err(|error| { + format!( + "Failed to load session turns: {}", + desktop_session_error(error) + ) + }); startup_trace.record_tauri_command_elapsed( "load_session_turns", Some(trace_target), @@ -350,38 +301,26 @@ pub async fn load_session_turns( #[tauri::command] pub async fn get_session_usage_report( request: GetSessionUsageReportRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result { - let storage_workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - let mut report = generate_session_usage_report( - &manager, - Some(app_state.token_usage_service.as_ref()), - SessionUsageReportRequest { - session_id: request.session_id, - workspace_path: Some(storage_workspace_path.to_string_lossy().to_string()), - remote_connection_id: request.remote_connection_id.clone(), - remote_ssh_host: request.remote_ssh_host.clone(), - include_hidden_subagents: request.include_hidden_subagents, - }, - ) - .await - .map_err(|e| format!("Failed to generate session usage report: {}", e))?; - - report.workspace.path_label = Some(request.workspace_path); - report.workspace.remote_connection_id = request.remote_connection_id; - report.workspace.remote_ssh_host = request.remote_ssh_host; - - Ok(report) + runtime + .session_application() + .generate_usage_report( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.session_id, + request.include_hidden_subagents, + ) + .await + .map_err(|error| { + format!( + "Failed to generate session usage report: {}", + desktop_session_error(error) + ) + }) } #[tauri::command] @@ -416,33 +355,27 @@ pub async fn save_session_turn( #[tauri::command] pub async fn save_session_metadata( request: SaveSessionMetadataRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result<(), String> { - if request.fields.is_empty() { - return Err("At least one session metadata field is required".to_string()); - } - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - let session_id = request.metadata.session_id.clone(); - manager - .update_session_metadata(&workspace_path, &session_id, |metadata| { - merge_ui_owned_session_metadata(metadata, &request.metadata, &request.fields); - }) + runtime + .session_application() + .save_ui_metadata( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.metadata, + request.fields, + ) .await - .map_err(|e| format!("Failed to save session metadata: {}", e))?; - - // Notify the auto-sync background task - crate::api::remote_connect_api::notify_session_changed(&session_id, &request.workspace_path); - Ok(()) + .map_err(|error| match error { + DesktopSessionApplicationError::Validation(message) => message, + error => format!( + "Failed to save session metadata: {}", + desktop_session_error(error) + ), + }) } #[tauri::command] @@ -479,54 +412,51 @@ pub async fn export_session_transcript( #[tauri::command] pub async fn delete_persisted_session( request: DeletePersistedSessionRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result<(), String> { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - manager - .delete_session(&workspace_path, &request.session_id) + runtime + .session_application() + .delete_session( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.session_id, + ) .await - .map_err(|e| format!("Failed to delete persisted session: {}", e))?; - - // Notify auto-sync: tombstone this session on the relay - crate::api::remote_connect_api::notify_session_deleted(&request.session_id); - Ok(()) + .map_err(|error| { + format!( + "Failed to delete persisted session: {}", + desktop_session_error(error) + ) + }) } #[tauri::command] pub async fn touch_session_activity( request: TouchSessionActivityRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, startup_trace: State<'_, DesktopStartupTrace>, ) -> Result<(), String> { let trace_started = Instant::now(); - let result = async { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), + let result = runtime + .session_application() + .touch_session( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + &request.session_id, ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - manager - .touch_session(&workspace_path, &request.session_id) - .await - .map_err(|e| format!("Failed to update session activity: {}", e)) - } - .await; + .await + .map_err(|error| { + format!( + "Failed to update session activity: {}", + desktop_session_error(error) + ) + }); startup_trace.record_tauri_command_elapsed("touch_session_activity", None, trace_started); result } @@ -534,32 +464,29 @@ pub async fn touch_session_activity( #[tauri::command] pub async fn load_persisted_session_metadata( request: LoadPersistedSessionMetadataRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, startup_trace: State<'_, DesktopStartupTrace>, ) -> Result, String> { let trace_started = Instant::now(); - let result = async { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), + // Direct metadata lookups are used by persistence flows that must be able + // to read hidden subagent sessions without list-level visibility filtering. + let result = runtime + .session_application() + .load_session_metadata( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + &request.session_id, ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - let metadata = manager - .load_session_metadata(&workspace_path, &request.session_id) - .await - .map_err(|e| format!("Failed to load persisted session metadata: {}", e))?; - - // Direct metadata lookups are used by persistence flows that must be able - // to read hidden subagent sessions without list-level visibility filtering. - Ok(metadata) - } - .await; + .await + .map_err(|error| { + format!( + "Failed to load persisted session metadata: {}", + desktop_session_error(error) + ) + }); startup_trace.record_tauri_command_elapsed( "load_persisted_session_metadata", None, @@ -571,77 +498,71 @@ pub async fn load_persisted_session_metadata( #[tauri::command] pub async fn fork_session( request: ForkSessionRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - manager - .branch_session( - &workspace_path, - &SessionBranchRequest { - source_session_id: request.source_session_id, - source_turn_id: request.source_turn_id, - }, + runtime + .session_application() + .fork_session( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.source_session_id, + request.source_turn_id, ) .await - .map_err(|e| format!("Failed to fork session: {}", e)) + .map_err(|error| format!("Failed to fork session: {}", desktop_session_error(error))) } #[tauri::command] pub async fn archive_session( request: ArchiveSessionRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result<(), String> { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - manager - .update_session_metadata(&workspace_path, &request.session_id, |metadata| { - metadata.status = SessionStatus::Archived; - }) + runtime + .session_application() + .set_session_archived( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.session_id, + true, + ) .await - .map_err(|e| format!("Failed to save session metadata: {}", e)) + .map_err(|error| { + format!( + "Failed to save session metadata: {}", + desktop_session_error(error) + ) + }) } #[tauri::command] pub async fn unarchive_session( request: UnarchiveSessionRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result<(), String> { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - manager - .update_session_metadata(&workspace_path, &request.session_id, |metadata| { - metadata.status = SessionStatus::Active; - }) + runtime + .session_application() + .set_session_archived( + desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ), + request.session_id, + false, + ) .await - .map_err(|e| format!("Failed to save session metadata: {}", e)) + .map_err(|error| { + format!( + "Failed to save session metadata: {}", + desktop_session_error(error) + ) + }) } #[tauri::command] @@ -689,30 +610,17 @@ pub async fn archive_all_sessions( #[tauri::command] pub async fn list_archived_sessions( request: ListPersistedSessionsRequest, - app_state: State<'_, AppState>, - path_manager: State<'_, Arc>, + runtime: State<'_, DesktopRuntimeContext>, ) -> Result, String> { - let workspace_path = desktop_effective_session_storage_path( - &app_state, - &request.workspace_path, - request.remote_connection_id.as_deref(), - request.remote_ssh_host.as_deref(), - ) - .await; - let manager = PersistenceManager::new(path_manager.inner().clone()) - .map_err(|e| format!("Failed to create persistence manager: {}", e))?; - - let sessions = manager - .list_session_metadata(&workspace_path) + runtime + .session_application() + .list_archived_sessions(desktop_session_scope( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + )) .await - .map_err(|e| format!("Failed to list sessions: {}", e))?; - - let archived: Vec = sessions - .into_iter() - .filter(|s| s.status == SessionStatus::Archived) - .collect(); - - Ok(archived) + .map_err(|error| format!("Failed to list sessions: {}", desktop_session_error(error))) } #[tauri::command] @@ -750,117 +658,3 @@ pub async fn delete_all_archived_sessions( Ok(deleted_count) } - -#[cfg(test)] -mod tests { - use super::{merge_ui_owned_session_metadata, UiSessionMetadataField}; - use bitfun_core::service::session::{ - SessionKind, SessionMemoryMode, SessionMetadata, SessionStatus, - }; - use serde_json::json; - - #[test] - fn ui_metadata_merge_preserves_core_authoritative_fields_and_custom_keys() { - let mut current = SessionMetadata::new( - "session".to_string(), - "Current".to_string(), - "plan".to_string(), - "model-a".to_string(), - ); - current.last_submitted_agent_type = Some("plan".to_string()); - current.memory_mode = SessionMemoryMode::Polluted; - current.session_kind = SessionKind::Standard; - current.status = SessionStatus::Archived; - current.turn_count = 7; - current.custom_metadata = Some(json!({ - "threadGoal": { "objective": "preserve" }, - "titleSource": "i18n", - "titleKey": "old" - })); - - let mut incoming = current.clone(); - incoming.session_name = "Renamed".to_string(); - incoming.agent_type = "agentic".to_string(); - incoming.model_name = "stale-model".to_string(); - incoming.memory_mode = SessionMemoryMode::Enabled; - incoming.status = SessionStatus::Active; - incoming.turn_count = 1; - incoming.review_action_state = Some(json!({ "phase": "fixing" })); - incoming.custom_metadata = Some(json!({ - "titleSource": "i18n", - "titleKey": "new", - "untrustedCoreKey": "drop" - })); - - merge_ui_owned_session_metadata( - &mut current, - &incoming, - &[ - UiSessionMetadataField::SessionName, - UiSessionMetadataField::Tags, - UiSessionMetadataField::Todos, - UiSessionMetadataField::ReviewActionState, - UiSessionMetadataField::UnreadCompletion, - UiSessionMetadataField::NeedsUserAttention, - UiSessionMetadataField::TitleMetadata, - ], - ); - - assert_eq!(current.session_name, "Renamed"); - assert_eq!(current.agent_type, "plan"); - assert_eq!(current.model_name, "model-a"); - assert_eq!(current.memory_mode, SessionMemoryMode::Polluted); - assert_eq!(current.status, SessionStatus::Archived); - assert_eq!(current.turn_count, 7); - assert_eq!(current.review_action_state, incoming.review_action_state); - let custom = current.custom_metadata.unwrap(); - assert_eq!(custom["threadGoal"]["objective"], "preserve"); - assert_eq!(custom["titleKey"], "new"); - assert!(custom.get("untrustedCoreKey").is_none()); - } - - #[test] - fn ui_metadata_field_mask_keeps_independent_writers_isolated() { - let mut current = SessionMetadata::new( - "session".to_string(), - "Current".to_string(), - "agentic".to_string(), - "auto".to_string(), - ); - current.review_action_state = Some(json!({ "phase": "review_completed" })); - current.unread_completion = Some("completed".to_string()); - current.needs_user_attention = Some("ask_user".to_string()); - - let mut stale_general_update = current.clone(); - stale_general_update.session_name = "Renamed".to_string(); - stale_general_update.review_action_state = None; - merge_ui_owned_session_metadata( - &mut current, - &stale_general_update, - &[UiSessionMetadataField::SessionName], - ); - assert_eq!(current.session_name, "Renamed"); - assert_eq!( - current.review_action_state, - Some(json!({ "phase": "review_completed" })) - ); - assert_eq!(current.unread_completion.as_deref(), Some("completed")); - assert_eq!(current.needs_user_attention.as_deref(), Some("ask_user")); - - let mut review_update = current.clone(); - review_update.review_action_state = Some(json!({ "phase": "fixing" })); - review_update.unread_completion = None; - review_update.needs_user_attention = None; - merge_ui_owned_session_metadata( - &mut current, - &review_update, - &[UiSessionMetadataField::ReviewActionState], - ); - assert_eq!( - current.review_action_state, - Some(json!({ "phase": "fixing" })) - ); - assert_eq!(current.unread_completion.as_deref(), Some("completed")); - assert_eq!(current.needs_user_attention.as_deref(), Some("ask_user")); - } -} diff --git a/src/apps/desktop/src/lib.rs b/src/apps/desktop/src/lib.rs index a41c61504b..e8b5251218 100644 --- a/src/apps/desktop/src/lib.rs +++ b/src/apps/desktop/src/lib.rs @@ -387,14 +387,20 @@ pub async fn run() { startup_trace.record_elapsed_step("native_pre_tauri", "initialize_app_state", step_started); let step_started = Instant::now(); - let desktop_runtime = - match runtime::DesktopRuntimeContext::build(coordinator.clone(), scheduler.clone()) { - Ok(runtime) => runtime, - Err(error) => { - log::error!("Failed to initialize Desktop Agent Runtime: {}", error); - return; - } - }; + let desktop_runtime = match runtime::DesktopRuntimeContext::build( + coordinator.clone(), + scheduler.clone(), + app_state.token_usage_service.clone(), + app_state.workspace_service.clone(), + app_state.ssh_manager.clone(), + app_state.acp_client_service.clone(), + ) { + Ok(runtime) => runtime, + Err(error) => { + log::error!("Failed to initialize Desktop Agent Runtime: {}", error); + return; + } + }; startup_timings.record_elapsed("initialize_desktop_agent_runtime", step_started); startup_trace.record_elapsed_step( "native_pre_tauri", diff --git a/src/apps/desktop/src/runtime/mod.rs b/src/apps/desktop/src/runtime/mod.rs index 7b3a04af69..64fd1524df 100644 --- a/src/apps/desktop/src/runtime/mod.rs +++ b/src/apps/desktop/src/runtime/mod.rs @@ -1,12 +1,23 @@ use std::sync::Arc; -use bitfun_agent_runtime::sdk::{ - AgentDialogTurnPort, AgentInteractionResponsePort, AgentRuntime, AgentRuntimeBuilder, - AgentSessionModelPort, AgentSubmissionPort, AgentTurnCancellationPort, RuntimeBuildError, -}; +use bitfun_agent_runtime::sdk::AgentRuntime; use bitfun_core::agentic::coordination::{ConversationCoordinator, DialogScheduler}; use bitfun_core::product_runtime::CoreLocalWorkspaceSnapshot; +use bitfun_core::service::remote_ssh::SSHConnectionManager; +use bitfun_core::service::token_usage::TokenUsageService; +use bitfun_core::service::workspace::WorkspaceService; use bitfun_runtime_ports::LocalWorkspaceSnapshotPort; +use tokio::sync::RwLock; + +mod session_application; +mod session_host_effects; + +use session_host_effects::ProductionDesktopSessionHostEffects; + +pub(crate) use session_application::{ + DesktopSessionApplication, DesktopSessionApplicationError, DesktopSessionScopeRequest, + UiSessionMetadataField, +}; /// Desktop-owned access to the Agent Runtime SDK interaction facade. /// @@ -15,7 +26,7 @@ use bitfun_runtime_ports::LocalWorkspaceSnapshotPort; /// ports used by current Tauri commands; it does not claim that the complete /// Desktop delivery profile or its product services have been assembled. pub struct DesktopRuntimeContext { - agent_runtime: AgentRuntime, + session_application: DesktopSessionApplication, local_workspace_snapshot: Arc, } @@ -23,29 +34,34 @@ impl DesktopRuntimeContext { pub(crate) fn build( coordinator: Arc, scheduler: Arc, - ) -> Result { - let submission: Arc = coordinator.clone(); - let session_model: Arc = coordinator.clone(); - let interaction_response: Arc = coordinator; - let dialog_turn: Arc = scheduler.clone(); - let cancellation: Arc = scheduler; - let agent_runtime = AgentRuntimeBuilder::new() - .with_submission_port(submission) - .with_session_model_port(session_model) - .with_dialog_turn_port(dialog_turn) - .with_cancellation_port(cancellation) - .with_interaction_response_port(interaction_response) - .build()?; + token_usage_service: Arc, + workspace_service: Arc, + ssh_manager: Arc>>, + acp_client_service: Option>, + ) -> Result { + let host_effects = Arc::new(ProductionDesktopSessionHostEffects::new(acp_client_service)); + let session_application = DesktopSessionApplication::build( + coordinator, + scheduler, + token_usage_service, + workspace_service, + ssh_manager, + host_effects, + )?; let local_workspace_snapshot = CoreLocalWorkspaceSnapshot::build(); Ok(Self { - agent_runtime, + session_application, local_workspace_snapshot, }) } pub(crate) fn agent_runtime(&self) -> &AgentRuntime { - &self.agent_runtime + self.session_application.agent_runtime() + } + + pub(crate) fn session_application(&self) -> &DesktopSessionApplication { + &self.session_application } pub(crate) fn local_workspace_snapshot(&self) -> &dyn LocalWorkspaceSnapshotPort { @@ -67,12 +83,16 @@ mod tests { assert!(app_source.contains("DesktopRuntimeContext::build(")); assert!(app_source.contains(".manage(desktop_runtime)")); - assert!(runtime_source.contains("with_dialog_turn_port")); - assert!(runtime_source.contains("with_cancellation_port")); - assert!(runtime_source.contains("with_interaction_response_port")); - assert!(runtime_source.contains("with_session_model_port")); + assert!(runtime_source.contains("DesktopSessionApplication::build(")); assert!(runtime_source.contains("CoreLocalWorkspaceSnapshot::build()")); + let session_commands = include_str!("../api/session_api.rs"); + assert_eq!( + session_commands.matches("PersistenceManager::new").count(), + 4, + "only raw turn save, transcript export, and the two excluded bulk operations keep direct persistence" + ); + let snapshot_commands = include_str!("../api/snapshot_service.rs"); assert_eq!( snapshot_commands diff --git a/src/apps/desktop/src/runtime/session_application.rs b/src/apps/desktop/src/runtime/session_application.rs new file mode 100644 index 0000000000..efe73b55a3 --- /dev/null +++ b/src/apps/desktop/src/runtime/session_application.rs @@ -0,0 +1,990 @@ +//! Framework-neutral Desktop session use cases. +//! +//! Tauri commands map their transport DTOs into this application boundary. +//! Rich Desktop persistence views remain on Core's compatibility facade while +//! stable lifecycle operations use the Agent Runtime SDK. + +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Instant; + +use async_trait::async_trait; +use bitfun_agent_runtime::sdk::{ + AgentRuntime, AgentSessionArchiveStateRequest, AgentSessionDeleteRequest, + AgentSessionForkAtTurnRequest, AgentSessionRenameRequest, AgentSessionUsageRequest, +}; +use bitfun_core::agentic::coordination::{ConversationCoordinator, DialogScheduler}; +use bitfun_core::agentic::core::Session; +use bitfun_core::agentic::persistence::{SessionBranchResult, SessionMetadataPage}; +use bitfun_core::agentic::session::SessionViewRestoreTiming; +use bitfun_core::product_runtime::{CoreAgentRuntimeCompatibility, CoreProductAgentRuntime}; +use bitfun_core::service::remote_ssh::workspace_state::get_effective_session_path; +use bitfun_core::service::remote_ssh::SSHConnectionManager; +use bitfun_core::service::session::{DialogTurnData, SessionMetadata, SessionStatus}; +use bitfun_core::service::session_usage::SessionUsageReport; +use bitfun_core::service::token_usage::TokenUsageService; +use bitfun_core::service::workspace::WorkspaceService; +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; + +const UI_CUSTOM_METADATA_KEYS: [&str; 3] = ["titleSource", "titleKey", "titleParams"]; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum UiSessionMetadataField { + SessionName, + Tags, + Todos, + ReviewActionState, + UnreadCompletion, + NeedsUserAttention, + TitleMetadata, +} + +#[derive(Debug, Clone)] +pub(crate) struct DesktopSessionScopeRequest { + pub workspace_path: String, + pub remote_connection_id: Option, + pub remote_ssh_host: Option, +} + +#[derive(Debug, thiserror::Error)] +pub(crate) enum DesktopSessionApplicationError { + #[error("{0}")] + Validation(String), + #[error("{0}")] + Core(String), + #[error("{0}")] + Runtime(String), + #[error("{0}")] + RestoreBeforeRename(String), +} + +pub(crate) type DesktopSessionApplicationResult = Result; + +#[derive(Debug)] +pub(crate) struct DesktopSessionViewRestore { + pub session: Session, + pub turns: Vec, + pub total_turn_count: usize, + pub timings: SessionViewRestoreTiming, +} + +#[derive(Debug)] +pub(crate) struct DesktopSessionWithTurnsRestore { + pub session: Session, + pub turns: Vec, +} + +#[derive(Clone)] +struct ResolvedDesktopSessionScope { + workspace_path: String, + effective_storage_path: PathBuf, + remote_connection_id: Option, + requested_remote_ssh_host: Option, + resolved_remote_ssh_host: Option, +} + +#[derive(Clone)] +struct DesktopSessionScopeResolver { + workspace_service: Arc, + ssh_manager: Arc>>, +} + +impl DesktopSessionScopeResolver { + async fn resolve(&self, request: DesktopSessionScopeRequest) -> ResolvedDesktopSessionScope { + let remote_connection_id = normalized_optional(request.remote_connection_id.as_deref()); + let requested_remote_ssh_host = normalized_optional(request.remote_ssh_host.as_deref()); + let mut registered_remote_ssh_host = None; + if requested_remote_ssh_host.is_none() { + if let Some(connection_id) = remote_connection_id.as_deref() { + registered_remote_ssh_host = self + .workspace_service + .remote_ssh_host_for_remote_workspace(connection_id, &request.workspace_path) + .await; + } + } + let mut saved_remote_ssh_host = None; + if requested_remote_ssh_host.is_none() && registered_remote_ssh_host.is_none() { + if let Some(connection_id) = remote_connection_id.as_deref() { + let manager = self.ssh_manager.read().await.clone(); + if let Some(manager) = manager { + saved_remote_ssh_host = manager + .get_saved_host_for_connection_id(connection_id) + .await; + } + } + } + let resolved_remote_ssh_host = choose_remote_ssh_host( + requested_remote_ssh_host.as_deref(), + registered_remote_ssh_host.as_deref(), + saved_remote_ssh_host.as_deref(), + ); + let effective_storage_path = get_effective_session_path( + &request.workspace_path, + remote_connection_id.as_deref(), + resolved_remote_ssh_host.as_deref(), + ) + .await; + + ResolvedDesktopSessionScope { + workspace_path: request.workspace_path, + effective_storage_path, + remote_connection_id, + requested_remote_ssh_host, + resolved_remote_ssh_host, + } + } +} + +fn normalized_optional(value: Option<&str>) -> Option { + value + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(ToOwned::to_owned) +} + +fn choose_remote_ssh_host( + requested: Option<&str>, + registered: Option<&str>, + saved: Option<&str>, +) -> Option { + normalized_optional(requested) + .or_else(|| normalized_optional(registered)) + .or_else(|| normalized_optional(saved)) +} + +#[async_trait] +pub(crate) trait DesktopSessionHostEffects: Send + Sync { + async fn release_session(&self, session_id: &str); + fn notify_session_changed(&self, session_id: &str, workspace_path: &str); + fn notify_session_deleted(&self, session_id: &str); +} + +#[derive(Clone)] +pub(crate) struct DesktopSessionApplication { + agent_runtime: AgentRuntime, + compatibility: CoreAgentRuntimeCompatibility, + scope_resolver: DesktopSessionScopeResolver, + host_effects: Arc, +} + +impl DesktopSessionApplication { + pub(crate) fn build( + coordinator: Arc, + scheduler: Arc, + token_usage_service: Arc, + workspace_service: Arc, + ssh_manager: Arc>>, + host_effects: Arc, + ) -> Result { + let agent_runtime = CoreProductAgentRuntime::build_session_surface( + coordinator.clone(), + scheduler.clone(), + token_usage_service, + )?; + let compatibility = CoreAgentRuntimeCompatibility::build(coordinator, scheduler); + + Ok(Self { + agent_runtime, + compatibility, + scope_resolver: DesktopSessionScopeResolver { + workspace_service, + ssh_manager, + }, + host_effects, + }) + } + + pub(crate) fn agent_runtime(&self) -> &AgentRuntime { + &self.agent_runtime + } + + async fn resolved_scope( + &self, + request: DesktopSessionScopeRequest, + ) -> ResolvedDesktopSessionScope { + self.scope_resolver.resolve(request).await + } + + fn storage_path(&self, scope: &ResolvedDesktopSessionScope) -> PathBuf { + scope.effective_storage_path.clone() + } + + pub(crate) async fn list_persisted_sessions( + &self, + request: DesktopSessionScopeRequest, + ) -> DesktopSessionApplicationResult> { + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + self.compatibility + .list_persisted_sessions(&storage_path) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + } + + pub(crate) async fn list_persisted_sessions_page( + &self, + request: DesktopSessionScopeRequest, + cursor: Option<&str>, + limit: usize, + ) -> DesktopSessionApplicationResult { + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + self.compatibility + .list_persisted_sessions_page(&storage_path, cursor, limit) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + } + + pub(crate) async fn list_archived_sessions( + &self, + request: DesktopSessionScopeRequest, + ) -> DesktopSessionApplicationResult> { + let sessions = self.list_persisted_sessions(request).await?; + Ok(sessions + .into_iter() + .filter(|session| session.status == SessionStatus::Archived) + .collect()) + } + + pub(crate) async fn load_session_turns( + &self, + request: DesktopSessionScopeRequest, + session_id: &str, + limit: Option, + ) -> DesktopSessionApplicationResult> { + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + self.compatibility + .load_persisted_session_turns(&storage_path, session_id, limit) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + } + + pub(crate) async fn load_session_metadata( + &self, + request: DesktopSessionScopeRequest, + session_id: &str, + ) -> DesktopSessionApplicationResult> { + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + self.compatibility + .load_persisted_session_metadata(&storage_path, session_id) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + } + + pub(crate) async fn touch_session( + &self, + request: DesktopSessionScopeRequest, + session_id: &str, + ) -> DesktopSessionApplicationResult<()> { + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + self.compatibility + .touch_persisted_session(&storage_path, session_id) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + } + + pub(crate) async fn save_ui_metadata( + &self, + request: DesktopSessionScopeRequest, + incoming: SessionMetadata, + fields: Vec, + ) -> DesktopSessionApplicationResult<()> { + if fields.is_empty() { + return Err(DesktopSessionApplicationError::Validation( + "At least one session metadata field is required".to_string(), + )); + } + let workspace_path = request.workspace_path.clone(); + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + let session_id = incoming.session_id.clone(); + self.compatibility + .update_persisted_session_metadata(&storage_path, &session_id, |current| { + merge_ui_owned_session_metadata(current, &incoming, &fields); + }) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))?; + self.host_effects + .notify_session_changed(&session_id, &workspace_path); + Ok(()) + } + + pub(crate) async fn generate_usage_report( + &self, + request: DesktopSessionScopeRequest, + session_id: String, + include_hidden_subagents: bool, + ) -> DesktopSessionApplicationResult { + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + let mut report = self + .agent_runtime + .generate_session_usage(AgentSessionUsageRequest { + session_id, + workspace_path: Some(storage_path.to_string_lossy().to_string()), + remote_connection_id: scope.remote_connection_id.clone(), + remote_ssh_host: scope.requested_remote_ssh_host.clone(), + include_hidden_subagents, + }) + .await + .map_err(|error| DesktopSessionApplicationError::Runtime(error.into_message()))?; + report.workspace.path_label = Some(scope.workspace_path); + report.workspace.remote_connection_id = scope.remote_connection_id; + report.workspace.remote_ssh_host = scope.requested_remote_ssh_host; + Ok(report) + } + + pub(crate) async fn fork_session( + &self, + request: DesktopSessionScopeRequest, + source_session_id: String, + source_turn_id: String, + ) -> DesktopSessionApplicationResult { + let scope = self.resolved_scope(request).await; + let result = self + .agent_runtime + .fork_session_at_turn(AgentSessionForkAtTurnRequest { + workspace_path: scope.effective_storage_path.to_string_lossy().into_owned(), + source_session_id, + source_turn_id, + remote_connection_id: scope.remote_connection_id, + remote_ssh_host: scope.resolved_remote_ssh_host, + }) + .await + .map_err(|error| DesktopSessionApplicationError::Runtime(error.into_message()))?; + Ok(SessionBranchResult { + session_id: result.session_id, + session_name: result.session_name, + agent_type: result.agent_type, + }) + } + + pub(crate) async fn set_session_archived( + &self, + request: DesktopSessionScopeRequest, + session_id: String, + archived: bool, + ) -> DesktopSessionApplicationResult<()> { + let scope = self.resolved_scope(request).await; + self.agent_runtime + .set_session_archived(AgentSessionArchiveStateRequest { + workspace_path: scope.effective_storage_path.to_string_lossy().into_owned(), + session_id, + archived, + remote_connection_id: scope.remote_connection_id, + remote_ssh_host: scope.resolved_remote_ssh_host, + }) + .await + .map_err(|error| DesktopSessionApplicationError::Runtime(error.into_message())) + } + + pub(crate) async fn delete_session( + &self, + request: DesktopSessionScopeRequest, + session_id: String, + ) -> DesktopSessionApplicationResult<()> { + let scope = self.resolved_scope(request).await; + delete_session_with_host_effects( + &self.agent_runtime, + self.host_effects.as_ref(), + scope, + session_id, + ) + .await + } + + pub(crate) async fn rename_session( + &self, + request: Option, + session_id: String, + title: String, + ) -> DesktopSessionApplicationResult { + let normalized_title = title.trim().to_string(); + if let Some(request) = request { + let scope = self.resolved_scope(request).await; + if !self + .compatibility + .is_session_loaded_in_memory(&session_id) + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))? + { + let storage_path = self.storage_path(&scope); + self.compatibility + .restore_session_from_storage_path(&storage_path, &session_id, false) + .await + .map_err(|error| { + DesktopSessionApplicationError::RestoreBeforeRename(error.to_string()) + })?; + } + self.agent_runtime + .rename_session(AgentSessionRenameRequest { + workspace_path: scope.effective_storage_path.to_string_lossy().into_owned(), + session_id: session_id.clone(), + session_name: title, + remote_connection_id: scope.remote_connection_id, + remote_ssh_host: scope.resolved_remote_ssh_host, + }) + .await + .map_err(|error| DesktopSessionApplicationError::Runtime(error.into_message()))?; + self.host_effects + .notify_session_changed(&session_id, &scope.workspace_path); + return Ok(normalized_title); + } + + if !self + .compatibility + .is_session_loaded_in_memory(&session_id) + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))? + { + return Err(DesktopSessionApplicationError::Validation( + "workspace_path is required when the session is not loaded".to_string(), + )); + } + let updated_title = self + .compatibility + .update_loaded_session_title(&session_id, &title) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))?; + self.host_effects.notify_session_changed(&session_id, ""); + Ok(updated_title) + } + + pub(crate) async fn ensure_session_loaded( + &self, + request: DesktopSessionScopeRequest, + session_id: &str, + include_internal: bool, + ) -> DesktopSessionApplicationResult<()> { + if self + .compatibility + .is_session_loaded_in_memory(session_id) + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))? + { + return Ok(()); + } + if request.workspace_path.trim().is_empty() { + return Err(DesktopSessionApplicationError::Validation( + "workspace_path is required when the session is not loaded".to_string(), + )); + } + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + self.compatibility + .ensure_session_loaded_from_storage_path(&storage_path, session_id, include_internal) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + } + + pub(crate) async fn restore_session( + &self, + request: DesktopSessionScopeRequest, + session_id: &str, + include_internal: bool, + ) -> DesktopSessionApplicationResult { + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + self.compatibility + .restore_session_from_storage_path(&storage_path, session_id, include_internal) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + } + + pub(crate) async fn restore_session_view( + &self, + request: DesktopSessionScopeRequest, + session_id: &str, + include_internal: bool, + tail_turn_count: Option, + on_storage_path_resolved: F, + ) -> DesktopSessionApplicationResult + where + F: FnOnce(u64) + Send, + { + let path_started_at = Instant::now(); + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + let resolve_storage_path_duration_ms = + path_started_at.elapsed().as_millis().min(u64::MAX as u128) as u64; + on_storage_path_resolved(resolve_storage_path_duration_ms); + let (session, turns, total_turn_count, mut timings) = self + .compatibility + .restore_session_view_from_storage_path( + &storage_path, + session_id, + include_internal, + tail_turn_count, + ) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))?; + timings.resolve_storage_path_duration_ms = resolve_storage_path_duration_ms; + Ok(DesktopSessionViewRestore { + session, + turns, + total_turn_count, + timings, + }) + } + + pub(crate) async fn restore_session_with_turns( + &self, + request: DesktopSessionScopeRequest, + session_id: &str, + include_internal: bool, + on_storage_path_resolved: F, + ) -> DesktopSessionApplicationResult + where + F: FnOnce(u64) + Send, + { + let path_started_at = Instant::now(); + let scope = self.resolved_scope(request).await; + let storage_path = self.storage_path(&scope); + let resolve_storage_path_duration_ms = + path_started_at.elapsed().as_millis().min(u64::MAX as u128) as u64; + on_storage_path_resolved(resolve_storage_path_duration_ms); + let (session, turns) = self + .compatibility + .restore_session_with_turns_from_storage_path( + &storage_path, + session_id, + include_internal, + ) + .await + .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))?; + Ok(DesktopSessionWithTurnsRestore { session, turns }) + } +} + +async fn delete_session_with_host_effects( + agent_runtime: &AgentRuntime, + host_effects: &dyn DesktopSessionHostEffects, + scope: ResolvedDesktopSessionScope, + session_id: String, +) -> DesktopSessionApplicationResult<()> { + host_effects.release_session(&session_id).await; + agent_runtime + .delete_session(AgentSessionDeleteRequest { + workspace_path: scope.effective_storage_path.to_string_lossy().into_owned(), + session_id: session_id.clone(), + remote_connection_id: scope.remote_connection_id, + remote_ssh_host: scope.resolved_remote_ssh_host, + }) + .await + .map_err(|error| DesktopSessionApplicationError::Runtime(error.into_message()))?; + host_effects.notify_session_deleted(&session_id); + Ok(()) +} + +fn merge_ui_owned_session_metadata( + current: &mut SessionMetadata, + incoming: &SessionMetadata, + fields: &[UiSessionMetadataField], +) { + if fields.contains(&UiSessionMetadataField::SessionName) { + current.session_name = incoming.session_name.clone(); + } + if fields.contains(&UiSessionMetadataField::Tags) { + current.tags = incoming.tags.clone(); + } + if fields.contains(&UiSessionMetadataField::Todos) { + current.todos = incoming.todos.clone(); + } + if fields.contains(&UiSessionMetadataField::ReviewActionState) { + current.review_action_state = incoming.review_action_state.clone(); + } + if fields.contains(&UiSessionMetadataField::UnreadCompletion) { + current.unread_completion = incoming.unread_completion.clone(); + } + if fields.contains(&UiSessionMetadataField::NeedsUserAttention) { + current.needs_user_attention = incoming.needs_user_attention.clone(); + } + if fields.contains(&UiSessionMetadataField::TitleMetadata) { + let mut custom = current + .custom_metadata + .as_ref() + .and_then(serde_json::Value::as_object) + .cloned() + .unwrap_or_default(); + let incoming_custom = incoming + .custom_metadata + .as_ref() + .and_then(serde_json::Value::as_object); + for key in UI_CUSTOM_METADATA_KEYS { + custom.remove(key); + if let Some(value) = incoming_custom.and_then(|metadata| metadata.get(key)) { + custom.insert(key.to_string(), value.clone()); + } + } + current.custom_metadata = (!custom.is_empty()).then(|| serde_json::Value::Object(custom)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bitfun_agent_runtime::sdk::{ + AgentRuntimeBuilder, AgentSessionCreateRequest, AgentSessionCreateResult, + AgentSessionListRequest, AgentSessionManagementPort, AgentSessionSummary, + AgentSessionWorkspaceBinding, AgentSessionWorkspaceRequest, AgentSubmissionPort, + AgentSubmissionRequest, AgentSubmissionResult, PortError, PortErrorKind, PortResult, + }; + use bitfun_core::service::session::{SessionKind, SessionMemoryMode}; + use serde_json::json; + use std::sync::Mutex; + + struct RecordingDeletePort { + events: Arc>>, + fail_delete: bool, + } + + struct NoopSubmissionPort; + + #[async_trait] + impl AgentSubmissionPort for NoopSubmissionPort { + async fn create_session( + &self, + request: AgentSessionCreateRequest, + ) -> PortResult { + Ok(AgentSessionCreateResult { + session_id: "unused".to_string(), + session_name: request.session_name, + agent_type: request.agent_type, + }) + } + + async fn submit_message( + &self, + _request: AgentSubmissionRequest, + ) -> PortResult { + Ok(AgentSubmissionResult { + turn_id: "unused".to_string(), + accepted: true, + }) + } + + async fn resolve_session_agent_type( + &self, + _session_id: &str, + ) -> PortResult> { + Ok(None) + } + } + + #[async_trait] + impl AgentSessionManagementPort for RecordingDeletePort { + async fn list_sessions( + &self, + _request: AgentSessionListRequest, + ) -> PortResult> { + Ok(Vec::new()) + } + + async fn delete_session(&self, _request: AgentSessionDeleteRequest) -> PortResult<()> { + self.events.lock().unwrap().push("durable_delete"); + if self.fail_delete { + return Err(PortError::new(PortErrorKind::Backend, "delete failed")); + } + Ok(()) + } + + async fn resolve_session_workspace_binding( + &self, + _request: AgentSessionWorkspaceRequest, + ) -> PortResult> { + Ok(None) + } + } + + struct RecordingHostEffects { + events: Arc>>, + } + + #[async_trait] + impl DesktopSessionHostEffects for RecordingHostEffects { + async fn release_session(&self, _session_id: &str) { + self.events.lock().unwrap().push("release"); + } + + fn notify_session_changed(&self, _session_id: &str, _workspace_path: &str) {} + + fn notify_session_deleted(&self, _session_id: &str) { + self.events.lock().unwrap().push("relay_delete"); + } + } + + fn delete_test_scope() -> ResolvedDesktopSessionScope { + ResolvedDesktopSessionScope { + workspace_path: "D:/workspace/project".to_string(), + effective_storage_path: PathBuf::from("D:/managed/project/sessions"), + remote_connection_id: None, + requested_remote_ssh_host: None, + resolved_remote_ssh_host: None, + } + } + + fn delete_test_runtime( + events: Arc>>, + fail_delete: bool, + ) -> AgentRuntime { + AgentRuntimeBuilder::new() + .with_submission_port(Arc::new(NoopSubmissionPort)) + .with_session_management_port(Arc::new(RecordingDeletePort { + events, + fail_delete, + })) + .build() + .expect("delete test runtime") + } + + #[test] + fn optional_scope_values_are_trimmed_without_inventing_identity() { + assert_eq!( + normalized_optional(Some(" host ")), + Some("host".to_string()) + ); + assert_eq!(normalized_optional(Some(" ")), None); + assert_eq!(normalized_optional(None), None); + } + + #[test] + fn remote_host_resolution_preserves_request_registry_and_offline_saved_precedence() { + assert_eq!( + choose_remote_ssh_host(Some("request-host"), Some("live-host"), Some("saved-host")), + Some("request-host".to_string()) + ); + assert_eq!( + choose_remote_ssh_host(None, Some("live-host"), Some("saved-host")), + Some("live-host".to_string()) + ); + assert_eq!( + choose_remote_ssh_host(None, None, Some(" saved-host ")), + Some("saved-host".to_string()) + ); + } + + #[tokio::test] + async fn local_session_storage_identity_survives_workspace_removal() { + let root = std::env::temp_dir().join(format!( + "bitfun-desktop-session-path-test-{}", + uuid::Uuid::new_v4() + )); + let workspace_path = root.join("project"); + std::fs::create_dir_all(&workspace_path).expect("workspace directory"); + let workspace_path = workspace_path + .canonicalize() + .expect("canonical workspace path") + .to_string_lossy() + .into_owned(); + + let before_removal = get_effective_session_path(&workspace_path, None, None).await; + std::fs::remove_dir_all(&workspace_path).expect("remove workspace directory"); + let after_removal = get_effective_session_path(&workspace_path, None, None).await; + + assert_eq!(after_removal, before_removal); + assert_eq!( + after_removal.file_name().and_then(|value| value.to_str()), + Some("sessions") + ); + let _ = std::fs::remove_dir_all(root); + } + + #[tokio::test] + async fn offline_saved_ssh_host_resolves_to_the_same_remote_session_tree() { + let connection_id = "offline-saved-host-test"; + let workspace_path = "/srv/offline-project"; + let saved_host = + choose_remote_ssh_host(None, None, Some("saved.example")).expect("saved SSH host"); + + let resolved = get_effective_session_path( + workspace_path, + Some(connection_id), + Some(saved_host.as_str()), + ) + .await; + let unresolved = + get_effective_session_path(workspace_path, Some(connection_id), None).await; + + assert_ne!(resolved, unresolved); + assert_eq!( + resolved.file_name().and_then(|value| value.to_str()), + Some("sessions") + ); + assert!(!resolved + .components() + .any(|component| component.as_os_str() == std::ffi::OsStr::new("_unresolved"))); + } + + #[test] + fn application_boundary_stays_framework_neutral() { + let source = include_str!("session_application.rs"); + let tauri_namespace = ["tauri", "::"].concat(); + let tauri_state = ["tauri", "::", "State"].concat(); + assert!(!source.contains(&tauri_namespace)); + assert!(!source.contains(&tauri_state)); + for forbidden in [["crate", "::", "api"].concat(), ["bitfun", "_acp"].concat()] { + assert!(!source.contains(&forbidden), "unexpected {forbidden}"); + } + for forbidden in [ + ["Product", "Assembler"].concat(), + ["Runtime", "Services"].concat(), + ["Harness", "Registry"].concat(), + ] { + assert!(!source.contains(&forbidden), "unexpected {forbidden}"); + } + } + + #[tokio::test] + async fn delete_orders_host_release_durable_delete_and_relay_tombstone() { + let events = Arc::new(Mutex::new(Vec::new())); + let runtime = delete_test_runtime(events.clone(), false); + let host_effects = RecordingHostEffects { + events: events.clone(), + }; + + delete_session_with_host_effects( + &runtime, + &host_effects, + delete_test_scope(), + "session-1".to_string(), + ) + .await + .expect("delete should succeed"); + + assert_eq!( + events.lock().unwrap().as_slice(), + ["release", "durable_delete", "relay_delete"] + ); + } + + #[tokio::test] + async fn delete_failure_does_not_publish_relay_tombstone() { + let events = Arc::new(Mutex::new(Vec::new())); + let runtime = delete_test_runtime(events.clone(), true); + let host_effects = RecordingHostEffects { + events: events.clone(), + }; + + let error = delete_session_with_host_effects( + &runtime, + &host_effects, + delete_test_scope(), + "session-1".to_string(), + ) + .await + .expect_err("durable delete should fail"); + + assert!(matches!(error, DesktopSessionApplicationError::Runtime(_))); + assert_eq!( + events.lock().unwrap().as_slice(), + ["release", "durable_delete"] + ); + } + + #[test] + fn ui_metadata_merge_preserves_core_authoritative_fields_and_custom_keys() { + let mut current = SessionMetadata::new( + "session".to_string(), + "Current".to_string(), + "plan".to_string(), + "model-a".to_string(), + ); + current.last_submitted_agent_type = Some("plan".to_string()); + current.memory_mode = SessionMemoryMode::Polluted; + current.session_kind = SessionKind::Standard; + current.status = SessionStatus::Archived; + current.turn_count = 7; + current.custom_metadata = Some(json!({ + "threadGoal": { "objective": "preserve" }, + "titleSource": "i18n", + "titleKey": "old" + })); + + let mut incoming = current.clone(); + incoming.session_name = "Renamed".to_string(); + incoming.agent_type = "agentic".to_string(); + incoming.model_name = "stale-model".to_string(); + incoming.memory_mode = SessionMemoryMode::Enabled; + incoming.status = SessionStatus::Active; + incoming.turn_count = 1; + incoming.review_action_state = Some(json!({ "phase": "fixing" })); + incoming.custom_metadata = Some(json!({ + "titleSource": "i18n", + "titleKey": "new", + "untrustedCoreKey": "drop" + })); + + merge_ui_owned_session_metadata( + &mut current, + &incoming, + &[ + UiSessionMetadataField::SessionName, + UiSessionMetadataField::Tags, + UiSessionMetadataField::Todos, + UiSessionMetadataField::ReviewActionState, + UiSessionMetadataField::UnreadCompletion, + UiSessionMetadataField::NeedsUserAttention, + UiSessionMetadataField::TitleMetadata, + ], + ); + + assert_eq!(current.session_name, "Renamed"); + assert_eq!(current.agent_type, "plan"); + assert_eq!(current.model_name, "model-a"); + assert_eq!(current.memory_mode, SessionMemoryMode::Polluted); + assert_eq!(current.status, SessionStatus::Archived); + assert_eq!(current.turn_count, 7); + assert_eq!(current.review_action_state, incoming.review_action_state); + let custom = current.custom_metadata.unwrap(); + assert_eq!(custom["threadGoal"]["objective"], "preserve"); + assert_eq!(custom["titleKey"], "new"); + assert!(custom.get("untrustedCoreKey").is_none()); + } + + #[test] + fn ui_metadata_field_mask_keeps_independent_writers_isolated() { + let mut current = SessionMetadata::new( + "session".to_string(), + "Current".to_string(), + "agentic".to_string(), + "auto".to_string(), + ); + current.review_action_state = Some(json!({ "phase": "review_completed" })); + current.unread_completion = Some("completed".to_string()); + current.needs_user_attention = Some("ask_user".to_string()); + + let mut stale_general_update = current.clone(); + stale_general_update.session_name = "Renamed".to_string(); + stale_general_update.review_action_state = None; + merge_ui_owned_session_metadata( + &mut current, + &stale_general_update, + &[UiSessionMetadataField::SessionName], + ); + assert_eq!(current.session_name, "Renamed"); + assert_eq!( + current.review_action_state, + Some(json!({ "phase": "review_completed" })) + ); + assert_eq!(current.unread_completion.as_deref(), Some("completed")); + assert_eq!(current.needs_user_attention.as_deref(), Some("ask_user")); + + let mut review_update = current.clone(); + review_update.review_action_state = Some(json!({ "phase": "fixing" })); + review_update.unread_completion = None; + review_update.needs_user_attention = None; + merge_ui_owned_session_metadata( + &mut current, + &review_update, + &[UiSessionMetadataField::ReviewActionState], + ); + assert_eq!( + current.review_action_state, + Some(json!({ "phase": "fixing" })) + ); + assert_eq!(current.unread_completion.as_deref(), Some("completed")); + assert_eq!(current.needs_user_attention.as_deref(), Some("ask_user")); + } +} diff --git a/src/apps/desktop/src/runtime/session_host_effects.rs b/src/apps/desktop/src/runtime/session_host_effects.rs new file mode 100644 index 0000000000..ccf6960161 --- /dev/null +++ b/src/apps/desktop/src/runtime/session_host_effects.rs @@ -0,0 +1,32 @@ +use std::sync::Arc; + +use async_trait::async_trait; + +use super::session_application::DesktopSessionHostEffects; + +pub(super) struct ProductionDesktopSessionHostEffects { + acp_client_service: Option>, +} + +impl ProductionDesktopSessionHostEffects { + pub(super) fn new(acp_client_service: Option>) -> Self { + Self { acp_client_service } + } +} + +#[async_trait] +impl DesktopSessionHostEffects for ProductionDesktopSessionHostEffects { + async fn release_session(&self, session_id: &str) { + if let Some(service) = self.acp_client_service.as_ref() { + service.release_bitfun_session(session_id).await; + } + } + + fn notify_session_changed(&self, session_id: &str, workspace_path: &str) { + crate::api::remote_connect_api::notify_session_changed(session_id, workspace_path); + } + + fn notify_session_deleted(&self, session_id: &str) { + crate::api::remote_connect_api::notify_session_deleted(session_id); + } +} diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index 28f89bd13d..5cd83c2db6 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -8011,6 +8011,23 @@ impl bitfun_runtime_ports::AgentSessionManagementPort for ConversationCoordinato async fn archive_session( &self, request: bitfun_runtime_ports::AgentSessionArchiveRequest, + ) -> bitfun_runtime_ports::PortResult<()> { + bitfun_runtime_ports::AgentSessionManagementPort::set_session_archived( + self, + bitfun_runtime_ports::AgentSessionArchiveStateRequest { + workspace_path: request.workspace_path, + session_id: request.session_id, + archived: true, + remote_connection_id: request.remote_connection_id, + remote_ssh_host: request.remote_ssh_host, + }, + ) + .await + } + + async fn set_session_archived( + &self, + request: bitfun_runtime_ports::AgentSessionArchiveStateRequest, ) -> bitfun_runtime_ports::PortResult<()> { bitfun_core_types::validate_session_id(&request.session_id).map_err(|message| { bitfun_runtime_ports::PortError::new( @@ -8037,7 +8054,11 @@ impl bitfun_runtime_ports::AgentSessionManagementPort for ConversationCoordinato session_manager .persistence_manager() .update_session_metadata(&effective_storage_path, &request.session_id, |metadata| { - metadata.status = SessionStatus::Archived + metadata.status = if request.archived { + SessionStatus::Archived + } else { + SessionStatus::Active + } }) .await .map_err(runtime_port_error_preserving_message) @@ -9731,7 +9752,7 @@ mod tests { } #[tokio::test] - async fn agent_session_management_port_renames_and_archives_persisted_session() { + async fn agent_session_management_port_renames_and_sets_persisted_archive_state() { let (coordinator, session_manager) = test_coordinator(); let workspace_path = std::env::temp_dir().join(format!( "bitfun-agent-session-management-port-test-{}", @@ -9791,7 +9812,7 @@ mod tests { AgentSessionManagementPort::archive_session( &coordinator, AgentSessionArchiveRequest { - workspace_path: workspace, + workspace_path: workspace.clone(), session_id: created.session_id.clone(), remote_connection_id: None, remote_ssh_host: None, @@ -9807,6 +9828,26 @@ mod tests { .expect("metadata should exist"); assert_eq!(metadata.status, SessionStatus::Archived); + AgentSessionManagementPort::set_session_archived( + &coordinator, + bitfun_runtime_ports::AgentSessionArchiveStateRequest { + workspace_path: workspace.clone(), + session_id: created.session_id.clone(), + archived: false, + remote_connection_id: None, + remote_ssh_host: None, + }, + ) + .await + .expect("session unarchive should succeed"); + let metadata = session_manager + .persistence_manager() + .load_session_metadata(&storage_path, &created.session_id) + .await + .expect("metadata should load") + .expect("metadata should exist"); + assert_eq!(metadata.status, SessionStatus::Active); + let _ = std::fs::remove_dir_all(storage_path); let _ = std::fs::remove_dir_all(workspace_path); } diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index d8be37c6b2..c31679cb00 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -11,9 +11,9 @@ use std::sync::Arc; use std::time::Duration; use bitfun_agent_runtime::sdk::{ - AgentEventSource, AgentRuntime, AgentSessionForkPort, AgentSessionForkRequest, - AgentSessionForkResult, AgentSessionUsagePort, AgentSessionUsageRequest, - AgentTurnSettlementPort, AgentTurnSettlementRequest, + AgentEventSource, AgentRuntime, AgentSessionForkAtTurnRequest, AgentSessionForkPort, + AgentSessionForkRequest, AgentSessionForkResult, AgentSessionUsagePort, + AgentSessionUsageRequest, AgentTurnSettlementPort, AgentTurnSettlementRequest, }; use bitfun_harness::HarnessRegistry; use bitfun_runtime_ports::{ @@ -206,6 +206,26 @@ impl LocalWorkspaceSnapshotPort for CoreLocalWorkspaceSnapshot { pub struct CoreProductAgentRuntime; impl CoreProductAgentRuntime { + /// Build a narrow session and interaction facade for an existing product + /// owner. This does not assemble runtime services, harnesses, events, or a + /// complete delivery profile. + pub fn build_session_surface( + coordinator: Arc, + scheduler: Arc, + token_usage_service: Arc, + ) -> Result { + let session_operations = Arc::new(CoreSessionOperationsPort::new( + coordinator.clone(), + token_usage_service, + )); + CoreServiceAgentRuntime::session_surface_agent_runtime( + coordinator, + scheduler, + session_operations.clone(), + session_operations, + ) + } + pub fn build( coordinator: Arc, scheduler: Arc, @@ -273,6 +293,88 @@ impl CoreAgentRuntimeCompatibility { } } + pub async fn restore_session_from_storage_path( + &self, + storage_path: &Path, + session_id: &str, + include_internal: bool, + ) -> BitFunResult { + validate_persisted_session_id(session_id)?; + if include_internal { + self.coordinator + .restore_internal_session_from_storage_path(storage_path, session_id) + .await + } else { + self.coordinator + .restore_session_from_storage_path(storage_path, session_id) + .await + } + } + + pub async fn restore_session_view_from_storage_path( + &self, + storage_path: &Path, + session_id: &str, + include_internal: bool, + tail_turn_count: Option, + ) -> BitFunResult<( + Session, + Vec, + usize, + SessionViewRestoreTiming, + )> { + validate_persisted_session_id(session_id)?; + if let Some(tail_turn_count) = tail_turn_count { + if include_internal { + self.coordinator + .restore_internal_session_view_from_storage_path_tail_timed( + storage_path, + session_id, + tail_turn_count, + ) + .await + } else { + self.coordinator + .restore_session_view_from_storage_path_tail_timed( + storage_path, + session_id, + tail_turn_count, + ) + .await + } + } else { + let (session, turns, timing) = if include_internal { + self.coordinator + .restore_internal_session_view_from_storage_path_timed(storage_path, session_id) + .await? + } else { + self.coordinator + .restore_session_view_from_storage_path_timed(storage_path, session_id) + .await? + }; + let total_turn_count = turns.len(); + Ok((session, turns, total_turn_count, timing)) + } + } + + pub async fn restore_session_with_turns_from_storage_path( + &self, + storage_path: &Path, + session_id: &str, + include_internal: bool, + ) -> BitFunResult<(Session, Vec)> { + validate_persisted_session_id(session_id)?; + if include_internal { + self.coordinator + .restore_internal_session_with_turns_from_storage_path(storage_path, session_id) + .await + } else { + self.coordinator + .restore_session_with_turns_from_storage_path(storage_path, session_id) + .await + } + } + pub async fn restore_session_view_for_workspace( &self, request: SessionStoragePathRequest, @@ -356,6 +458,29 @@ impl CoreAgentRuntimeCompatibility { .await } + pub async fn load_persisted_session_metadata( + &self, + workspace_path: &Path, + session_id: &str, + ) -> BitFunResult> { + validate_persisted_session_id(session_id)?; + self.persistence + .load_session_metadata(workspace_path, session_id) + .await + } + + pub async fn update_persisted_session_metadata( + &self, + workspace_path: &Path, + session_id: &str, + update: impl FnOnce(&mut SessionMetadata), + ) -> BitFunResult<()> { + validate_persisted_session_id(session_id)?; + self.persistence + .update_session_metadata(workspace_path, session_id, update) + .await + } + pub fn is_session_loaded_in_memory(&self, session_id: &str) -> BitFunResult { validate_persisted_session_id(session_id)?; Ok(self @@ -365,6 +490,17 @@ impl CoreAgentRuntimeCompatibility { .is_some()) } + pub async fn update_loaded_session_title( + &self, + session_id: &str, + title: &str, + ) -> BitFunResult { + validate_persisted_session_id(session_id)?; + self.coordinator + .update_session_title(session_id, title) + .await + } + pub async fn resolve_persisted_session_storage_path( &self, request: SessionStoragePathRequest, @@ -568,6 +704,52 @@ impl CoreSessionOperationsPort { token_usage_service, } } + + async fn resolve_fork_storage_path( + &self, + workspace_path: String, + remote_connection_id: Option, + remote_ssh_host: Option, + ) -> PortResult { + CoreSessionStorePort::with_path_manager(self.persistence.path_manager().clone()) + .resolve_session_storage_path(SessionStoragePathRequest { + workspace_path: PathBuf::from(workspace_path), + remote_connection_id, + remote_ssh_host, + }) + .await + .map(|resolution| resolution.effective_storage_path) + } + + async fn fork_at_persisted_turn( + &self, + storage_path: &Path, + source_session_id: String, + source_turn_id: String, + ) -> PortResult { + if source_turn_id.trim().is_empty() { + return Err(PortError::new( + PortErrorKind::InvalidRequest, + "source turn id is required", + )); + } + let result = self + .persistence + .branch_session( + storage_path, + &SessionBranchRequest { + source_session_id, + source_turn_id, + }, + ) + .await + .map_err(runtime_port_error)?; + Ok(AgentSessionForkResult { + session_id: result.session_id, + session_name: result.session_name, + agent_type: result.agent_type, + }) + } } fn runtime_port_error(error: BitFunError) -> PortError { @@ -582,7 +764,7 @@ fn runtime_port_error(error: BitFunError) -> PortError { PortError::new(kind, error.to_string()) } -fn validate_local_session_fork_request(request: &AgentSessionForkRequest) -> PortResult<()> { +fn validate_latest_turn_fork_scope(request: &AgentSessionForkRequest) -> PortResult<()> { if request.remote_connection_id.is_some() || request.remote_ssh_host.is_some() { return Err(PortError::new( PortErrorKind::NotAvailable, @@ -598,30 +780,43 @@ impl AgentSessionForkPort for CoreSessionOperationsPort { &self, request: AgentSessionForkRequest, ) -> PortResult { - validate_local_session_fork_request(&request)?; - let workspace_path = Path::new(&request.workspace_path); - let (_, turns) = self + validate_latest_turn_fork_scope(&request)?; + let AgentSessionForkRequest { + workspace_path, + source_session_id, + remote_connection_id, + remote_ssh_host, + } = request; + let storage_path = self + .resolve_fork_storage_path(workspace_path, remote_connection_id, remote_ssh_host) + .await?; + let (_, turns, _) = self .coordinator - .restore_session_view(workspace_path, &request.source_session_id) + .restore_session_view_from_storage_path_timed(&storage_path, &source_session_id) .await .map_err(runtime_port_error)?; let source_turn_id = latest_persisted_turn_id(&turns).map_err(runtime_port_error)?; - let result = self - .persistence - .branch_session( - workspace_path, - &SessionBranchRequest { - source_session_id: request.source_session_id, - source_turn_id, - }, - ) + self.fork_at_persisted_turn(&storage_path, source_session_id, source_turn_id) .await - .map_err(runtime_port_error)?; - Ok(AgentSessionForkResult { - session_id: result.session_id, - session_name: result.session_name, - agent_type: result.agent_type, - }) + } + + async fn fork_session_at_turn( + &self, + request: AgentSessionForkAtTurnRequest, + ) -> PortResult { + let storage_path = self + .resolve_fork_storage_path( + request.workspace_path, + request.remote_connection_id, + request.remote_ssh_host, + ) + .await?; + self.fork_at_persisted_turn( + &storage_path, + request.source_session_id, + request.source_turn_id, + ) + .await } } @@ -668,6 +863,7 @@ impl AgentTurnSettlementPort for CoreSessionOperationsPort { mod tests { use std::path::{Path, PathBuf}; use std::sync::Arc; + use std::time::Duration; use bitfun_agent_runtime::sdk::AgentRuntime; use bitfun_harness::HarnessRegistry; @@ -679,12 +875,22 @@ mod tests { use super::{ generate_core_session_usage_report, latest_persisted_turn_id, runtime_port_error, - validate_local_session_fork_request, validate_persisted_session_id, + validate_latest_turn_fork_scope, validate_persisted_session_id, CoreAgentRuntimeCompatibility, CoreLocalWorkspaceSnapshot, CoreProductAgentRuntime, CoreSessionOperationsPort, }; use crate::agentic::coordination::{ConversationCoordinator, DialogScheduler}; + use crate::agentic::events::{EventQueue, EventQueueConfig, EventRouter}; + use crate::agentic::execution::{ + ExecutionEngine, ExecutionEngineConfig, RoundExecutor, StreamProcessor, + }; use crate::agentic::persistence::PersistenceManager; + use crate::agentic::session::{ + compression::{CompressionConfig, ContextCompressor}, + PromptCachePolicy, SessionContextStore, SessionManager, SessionManagerConfig, + }; + use crate::agentic::tools::registry::ToolRegistry; + use crate::agentic::tools::{ToolPipeline, ToolStateManager}; use crate::infrastructure::PathManager; use crate::service::session::{DialogTurnData, SessionMetadata, UserMessageData}; use crate::service::session_usage::UsageTokenSource; @@ -695,8 +901,9 @@ mod tests { }; use crate::util::errors::BitFunError; use bitfun_agent_runtime::sdk::{ - AgentSessionForkRequest, AgentSessionUsageRequest, PortErrorKind, + AgentSessionForkPort, AgentSessionForkRequest, AgentSessionUsageRequest, PortErrorKind, }; + use tokio::sync::RwLock as TokioRwLock; struct TestWorkspace { path: PathBuf, @@ -731,7 +938,7 @@ mod tests { } #[test] - fn product_agent_runtime_has_one_sdk_safe_builder_boundary() { + fn product_agent_runtime_exposes_reviewed_full_and_narrow_builders() { fn build( coordinator: Arc, scheduler: Arc, @@ -749,6 +956,7 @@ mod tests { } let _ = build; + let _ = CoreProductAgentRuntime::build_session_surface; let _ = CoreProductAgentRuntime::build_acp; } @@ -863,19 +1071,6 @@ mod tests { assert!(error.message.contains("session-1"), "{error}"); } - #[test] - fn local_session_fork_rejects_remote_identity() { - let error = validate_local_session_fork_request(&AgentSessionForkRequest { - workspace_path: "D:/workspace/project".to_string(), - source_session_id: "session-1".to_string(), - remote_connection_id: Some("remote-1".to_string()), - remote_ssh_host: None, - }) - .expect_err("the local fork provider must reject remote identity"); - - assert_eq!(error.kind, PortErrorKind::NotAvailable); - } - #[test] fn local_session_fork_uses_latest_persisted_turn_and_preserves_empty_error() { let turns = [ @@ -918,6 +1113,123 @@ mod tests { ); } + #[test] + fn latest_turn_session_fork_keeps_remote_identity_unsupported() { + for (remote_connection_id, remote_ssh_host) in [ + (Some("remote-1".to_string()), None), + (None, Some("host-1".to_string())), + ] { + let request = AgentSessionForkRequest { + workspace_path: "D:/workspace/project".to_string(), + source_session_id: "session-1".to_string(), + remote_connection_id, + remote_ssh_host, + }; + let error = validate_latest_turn_fork_scope(&request) + .expect_err("latest-turn remote fork must remain unsupported"); + + assert_eq!(error.kind, PortErrorKind::NotAvailable); + } + } + + #[tokio::test] + async fn latest_turn_fork_restores_from_the_resolved_storage_path() { + let workspace = TestWorkspace::new(); + let workspace_root = workspace.path().join("project"); + std::fs::create_dir_all(&workspace_root).expect("workspace root"); + let path_manager = workspace.path_manager(); + let storage_path = WorkspaceRuntimeService::new(path_manager.clone()) + .context_for_local_workspace(&workspace_root) + .sessions_dir; + std::fs::create_dir_all(&storage_path).expect("resolved session storage"); + let persistence = + Arc::new(PersistenceManager::new(path_manager).expect("persistence manager")); + let session_manager = Arc::new(SessionManager::new( + Arc::new(SessionContextStore::new()), + persistence.clone(), + SessionManagerConfig { + max_active_sessions: 100, + session_idle_timeout: Duration::from_secs(3600), + auto_save_interval: Duration::from_secs(300), + enable_persistence: false, + prompt_cache_policy: PromptCachePolicy::default(), + }, + )); + let event_queue = Arc::new(EventQueue::new(EventQueueConfig::default())); + let tool_pipeline = Arc::new(ToolPipeline::new( + Arc::new(TokioRwLock::new(ToolRegistry::new())), + Arc::new(ToolStateManager::new(event_queue.clone())), + None, + )); + let execution_engine = Arc::new(ExecutionEngine::new( + Arc::new(RoundExecutor::new( + Arc::new(StreamProcessor::new(event_queue.clone())), + event_queue.clone(), + tool_pipeline.clone(), + )), + event_queue.clone(), + session_manager.clone(), + Arc::new(ContextCompressor::new(CompressionConfig::default())), + ExecutionEngineConfig::default(), + )); + let coordinator = Arc::new(ConversationCoordinator::new( + session_manager, + execution_engine, + tool_pipeline, + event_queue, + Arc::new(EventRouter::new()), + )); + let token_usage_service = Arc::new( + TokenUsageService::new_in_base_dir(workspace.path().join("tokens")) + .await + .expect("token usage service"), + ); + let port = CoreSessionOperationsPort::new(coordinator, token_usage_service); + + let session_id = "session-latest-fork"; + persistence + .save_session_metadata( + &storage_path, + &SessionMetadata::new( + session_id.to_string(), + "Latest fork".to_string(), + "agentic".to_string(), + "model-a".to_string(), + ), + ) + .await + .expect("session metadata"); + let mut turn = DialogTurnData::new( + "turn-latest".to_string(), + 0, + session_id.to_string(), + UserMessageData { + id: "user-latest".to_string(), + content: "fork here".to_string(), + timestamp: 1, + metadata: None, + }, + ); + turn.mark_completed(); + persistence + .save_dialog_turn(&storage_path, &turn) + .await + .expect("persisted turn"); + + let result = port + .fork_session(AgentSessionForkRequest { + workspace_path: workspace_root.to_string_lossy().into_owned(), + source_session_id: session_id.to_string(), + remote_connection_id: None, + remote_ssh_host: None, + }) + .await + .expect("latest-turn fork should restore through the resolved storage path"); + + assert_ne!(result.session_id, session_id); + assert_eq!(result.agent_type, "agentic"); + } + #[tokio::test] async fn sdk_usage_provider_validates_ids_and_keeps_live_token_enrichment() { let workspace = TestWorkspace::new(); diff --git a/src/crates/assembly/core/src/service_agent_runtime.rs b/src/crates/assembly/core/src/service_agent_runtime.rs index 97fe62ce22..bf67cb1981 100644 --- a/src/crates/assembly/core/src/service_agent_runtime.rs +++ b/src/crates/assembly/core/src/service_agent_runtime.rs @@ -537,6 +537,13 @@ impl AgentSessionManagementPort for ScheduledSessionManagementPort { AgentSessionManagementPort::archive_session(self.coordinator.as_ref(), request).await } + async fn set_session_archived( + &self, + request: bitfun_runtime_ports::AgentSessionArchiveStateRequest, + ) -> bitfun_runtime_ports::PortResult<()> { + AgentSessionManagementPort::set_session_archived(self.coordinator.as_ref(), request).await + } + async fn resolve_session_workspace_binding( &self, request: bitfun_runtime_ports::AgentSessionWorkspaceRequest, @@ -905,6 +912,35 @@ impl CoreServiceAgentRuntime { .map_err(|error| error.to_string()) } + /// Builds the narrow interaction and session-operation surface used by a + /// product entrypoint without claiming a complete delivery profile. + pub(crate) fn session_surface_agent_runtime( + coordinator: Arc, + scheduler: Arc, + session_fork: Arc, + session_usage: Arc, + ) -> Result { + let submission: Arc = coordinator.clone(); + let session_management = + scheduled_session_management_port(coordinator.clone(), scheduler.clone()); + let session_model: Arc = coordinator.clone(); + let interaction_response: Arc = coordinator; + let dialog_turn: Arc = scheduler.clone(); + let cancellation: Arc = scheduler; + + AgentRuntimeBuilder::new() + .with_submission_port(submission) + .with_session_management_port(session_management) + .with_session_model_port(session_model) + .with_dialog_turn_port(dialog_turn) + .with_cancellation_port(cancellation) + .with_interaction_response_port(interaction_response) + .with_session_fork_port(session_fork) + .with_session_usage_port(session_usage) + .build() + .map_err(|error| error.to_string()) + } + pub(crate) fn agent_runtime_with_scheduler_ports( coordinator: Arc, scheduler: Arc, diff --git a/src/crates/contracts/runtime-ports/src/lib.rs b/src/crates/contracts/runtime-ports/src/lib.rs index a199e22b0e..a59b546ced 100644 --- a/src/crates/contracts/runtime-ports/src/lib.rs +++ b/src/crates/contracts/runtime-ports/src/lib.rs @@ -1056,6 +1056,22 @@ pub struct AgentSessionArchiveRequest { pub remote_ssh_host: Option, } +/// Sets the persisted archive state without exposing product-specific archive UI. +/// +/// This is separate from [`AgentSessionArchiveRequest`] so existing archive-only +/// consumers keep their current request shape and behavior. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentSessionArchiveStateRequest { + pub workspace_path: String, + pub session_id: String, + pub archived: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub remote_connection_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub remote_ssh_host: Option, +} + /// Records one completed local command result in user-visible session history. /// /// This request intentionally cannot select another turn kind or opt the turn @@ -1098,6 +1114,22 @@ pub struct AgentSessionForkRequest { pub remote_ssh_host: Option, } +/// Forks a session at an explicitly selected persisted turn. +/// +/// This is additive to [`AgentSessionForkRequest`] so existing Rust SDK +/// consumers keep the source-compatible latest-turn request shape. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentSessionForkAtTurnRequest { + pub workspace_path: String, + pub source_session_id: String, + pub source_turn_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub remote_connection_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub remote_ssh_host: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct AgentSessionForkResult { @@ -1815,6 +1847,26 @@ pub trait AgentSessionManagementPort: Send + Sync { )) } + async fn set_session_archived( + &self, + request: AgentSessionArchiveStateRequest, + ) -> PortResult<()> { + if request.archived { + return self + .archive_session(AgentSessionArchiveRequest { + workspace_path: request.workspace_path, + session_id: request.session_id, + remote_connection_id: request.remote_connection_id, + remote_ssh_host: request.remote_ssh_host, + }) + .await; + } + Err(PortError::new( + PortErrorKind::NotAvailable, + "session unarchive is not supported by this provider", + )) + } + async fn resolve_session_workspace_binding( &self, request: AgentSessionWorkspaceRequest, @@ -1846,6 +1898,17 @@ pub trait AgentSessionForkPort: Send + Sync { &self, request: AgentSessionForkRequest, ) -> PortResult; + + async fn fork_session_at_turn( + &self, + request: AgentSessionForkAtTurnRequest, + ) -> PortResult { + let _ = request; + Err(PortError::new( + PortErrorKind::NotAvailable, + "exact-turn session fork is not supported by this provider", + )) + } } #[async_trait::async_trait] @@ -2143,6 +2206,108 @@ impl SubagentContextMode { #[cfg(test)] mod tests { use super::*; + use std::sync::Mutex; + + #[derive(Default)] + struct ArchiveOnlySessionProvider { + archived_requests: Mutex>, + } + + #[async_trait::async_trait] + impl AgentSessionManagementPort for ArchiveOnlySessionProvider { + async fn list_sessions( + &self, + _request: AgentSessionListRequest, + ) -> PortResult> { + Ok(Vec::new()) + } + + async fn delete_session(&self, _request: AgentSessionDeleteRequest) -> PortResult<()> { + Ok(()) + } + + async fn archive_session(&self, request: AgentSessionArchiveRequest) -> PortResult<()> { + self.archived_requests.lock().unwrap().push(request); + Ok(()) + } + + async fn resolve_session_workspace_binding( + &self, + _request: AgentSessionWorkspaceRequest, + ) -> PortResult> { + Ok(None) + } + } + + struct LatestTurnForkOnlyProvider; + + #[async_trait::async_trait] + impl AgentSessionForkPort for LatestTurnForkOnlyProvider { + async fn fork_session( + &self, + request: AgentSessionForkRequest, + ) -> PortResult { + Ok(AgentSessionForkResult { + session_id: format!("{}-fork", request.source_session_id), + session_name: "Fork".to_string(), + agent_type: "agentic".to_string(), + }) + } + } + + fn archive_state_request(archived: bool) -> AgentSessionArchiveStateRequest { + AgentSessionArchiveStateRequest { + workspace_path: "/workspace/project".to_string(), + session_id: "session_1".to_string(), + archived, + remote_connection_id: Some("conn-1".to_string()), + remote_ssh_host: Some("host-1".to_string()), + } + } + + #[tokio::test] + async fn archive_state_default_preserves_archive_only_provider_compatibility() { + let provider = ArchiveOnlySessionProvider::default(); + + AgentSessionManagementPort::set_session_archived(&provider, archive_state_request(true)) + .await + .expect("archive=true should delegate to the legacy provider"); + let requests = provider.archived_requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].workspace_path, "/workspace/project"); + assert_eq!(requests[0].session_id, "session_1"); + assert_eq!(requests[0].remote_connection_id.as_deref(), Some("conn-1")); + assert_eq!(requests[0].remote_ssh_host.as_deref(), Some("host-1")); + drop(requests); + + let error = AgentSessionManagementPort::set_session_archived( + &provider, + archive_state_request(false), + ) + .await + .expect_err("legacy providers must reject unarchive by default"); + assert_eq!(error.kind, PortErrorKind::NotAvailable); + assert_eq!(provider.archived_requests.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn exact_turn_fork_default_preserves_latest_turn_only_provider_compatibility() { + let provider = LatestTurnForkOnlyProvider; + let error = AgentSessionForkPort::fork_session_at_turn( + &provider, + AgentSessionForkAtTurnRequest { + workspace_path: "/workspace/project".to_string(), + source_session_id: "session_1".to_string(), + source_turn_id: "turn_1".to_string(), + remote_connection_id: None, + remote_ssh_host: None, + }, + ) + .await + .expect_err("legacy providers must reject exact-turn fork by default"); + + assert_eq!(error.kind, PortErrorKind::NotAvailable); + } #[test] fn agent_session_create_request_keeps_rust_literal_compatible() { @@ -2888,6 +3053,26 @@ mod tests { remote_connection_id: Some("conn-1".to_string()), remote_ssh_host: Some("host-1".to_string()), }; + let archive_state_request = AgentSessionArchiveStateRequest { + workspace_path: "/workspace/project".to_string(), + session_id: "session_1".to_string(), + archived: false, + remote_connection_id: Some("conn-1".to_string()), + remote_ssh_host: Some("host-1".to_string()), + }; + let fork_request = AgentSessionForkRequest { + workspace_path: "/workspace/project".to_string(), + source_session_id: "session_1".to_string(), + remote_connection_id: None, + remote_ssh_host: None, + }; + let fork_at_turn_request = AgentSessionForkAtTurnRequest { + workspace_path: "/workspace/project".to_string(), + source_session_id: "session_1".to_string(), + source_turn_id: "turn_2".to_string(), + remote_connection_id: Some("conn-1".to_string()), + remote_ssh_host: Some("host-1".to_string()), + }; let model_request = AgentSessionModelUpdateRequest { session_id: "session_1".to_string(), model_id: "provider/model".to_string(), @@ -2912,6 +3097,11 @@ mod tests { let rename_json = serde_json::to_value(rename_request).expect("serialize rename request"); let archive_json = serde_json::to_value(archive_request).expect("serialize archive request"); + let archive_state_json = + serde_json::to_value(archive_state_request).expect("serialize archive-state request"); + let fork_json = serde_json::to_value(fork_request).expect("serialize fork request"); + let fork_at_turn_json = + serde_json::to_value(fork_at_turn_request).expect("serialize exact-turn fork request"); let model_json = serde_json::to_value(model_request).expect("serialize model request"); let mode_json = serde_json::to_value(mode_request).expect("serialize mode request"); let workspace_json = @@ -2936,6 +3126,9 @@ mod tests { assert_eq!(rename_json["remoteConnectionId"], "conn-1"); assert_eq!(archive_json["sessionId"], "session_1"); assert_eq!(archive_json["remoteSshHost"], "host-1"); + assert_eq!(archive_state_json["archived"], false); + assert!(fork_json.get("sourceTurnId").is_none()); + assert_eq!(fork_at_turn_json["sourceTurnId"], "turn_2"); assert_eq!(model_json["sessionId"], "session_1"); assert_eq!(model_json["modelId"], "provider/model"); assert_eq!(mode_json["sessionId"], "session_1"); diff --git a/src/crates/execution/agent-runtime/src/runtime.rs b/src/crates/execution/agent-runtime/src/runtime.rs index 3542ad1a36..c94a055ace 100644 --- a/src/crates/execution/agent-runtime/src/runtime.rs +++ b/src/crates/execution/agent-runtime/src/runtime.rs @@ -12,8 +12,9 @@ use bitfun_harness::HarnessRegistry; use bitfun_runtime_ports::{ AgentBackgroundResultRequest, AgentDialogTurnPort, AgentDialogTurnRequest, AgentInputAttachment, AgentLifecycleDeliveryPort, AgentLocalCommandTurnPort, - AgentLocalCommandTurnRecordRequest, AgentSessionArchiveRequest, AgentSessionCreateRequest, - AgentSessionCreateResult, AgentSessionDeleteRequest, AgentSessionForkPort, + AgentLocalCommandTurnRecordRequest, AgentSessionArchiveRequest, + AgentSessionArchiveStateRequest, AgentSessionCreateRequest, AgentSessionCreateResult, + AgentSessionDeleteRequest, AgentSessionForkAtTurnRequest, AgentSessionForkPort, AgentSessionForkRequest, AgentSessionForkResult, AgentSessionListRequest, AgentSessionManagementPort, AgentSessionModePort, AgentSessionModeUpdateRequest, AgentSessionModelPort, AgentSessionModelUpdateRequest, AgentSessionRenameRequest, @@ -860,6 +861,20 @@ impl AgentRuntime { .map_err(RuntimeError::from) } + pub async fn set_session_archived( + &self, + request: AgentSessionArchiveStateRequest, + ) -> Result<(), RuntimeError> { + let session_management = self + .session_management + .as_ref() + .ok_or(RuntimeError::MissingSessionManagementPort)?; + session_management + .set_session_archived(request) + .await + .map_err(RuntimeError::from) + } + pub async fn record_completed_local_command_turn( &self, request: AgentLocalCommandTurnRecordRequest, @@ -918,6 +933,21 @@ impl AgentRuntime { port.fork_session(request).await.map_err(RuntimeError::from) } + pub async fn fork_session_at_turn( + &self, + request: AgentSessionForkAtTurnRequest, + ) -> Result { + let port = self.session_fork.as_ref().ok_or_else(|| { + RuntimeError::Port(PortError::new( + PortErrorKind::NotAvailable, + "agent session fork port is not registered", + )) + })?; + port.fork_session_at_turn(request) + .await + .map_err(RuntimeError::from) + } + pub async fn generate_session_usage( &self, request: AgentSessionUsageRequest, @@ -1206,6 +1236,7 @@ mod tests { deleted_sessions: Mutex>, renamed_sessions: Mutex>, archived_sessions: Mutex>, + archive_state_updates: Mutex>, local_command_turns: Mutex>, restored_sessions: Mutex>, mode_updates: Mutex>, @@ -1327,6 +1358,14 @@ mod tests { Ok(()) } + async fn set_session_archived( + &self, + request: AgentSessionArchiveStateRequest, + ) -> PortResult<()> { + self.archive_state_updates.lock().unwrap().push(request); + Ok(()) + } + async fn resolve_session_workspace_binding( &self, request: AgentSessionWorkspaceRequest, @@ -1876,6 +1915,16 @@ mod tests { }) .await .expect("archive session"); + runtime + .set_session_archived(AgentSessionArchiveStateRequest { + workspace_path: "/workspace/project".to_string(), + session_id: "session_1".to_string(), + archived: false, + remote_connection_id: None, + remote_ssh_host: None, + }) + .await + .expect("unarchive session"); let workspace_binding = runtime .resolve_session_workspace_binding(AgentSessionWorkspaceRequest { session_id: "session_1".to_string(), @@ -1898,6 +1947,7 @@ mod tests { assert_eq!(ports.deleted_sessions.lock().unwrap().len(), 1); assert_eq!(ports.renamed_sessions.lock().unwrap().len(), 1); assert_eq!(ports.archived_sessions.lock().unwrap().len(), 1); + assert_eq!(ports.archive_state_updates.lock().unwrap().len(), 1); assert_eq!(ports.workspace_binding_requests.lock().unwrap().len(), 1); } diff --git a/src/crates/execution/agent-runtime/src/sdk.rs b/src/crates/execution/agent-runtime/src/sdk.rs index 6c32cffc45..22f54071eb 100644 --- a/src/crates/execution/agent-runtime/src/sdk.rs +++ b/src/crates/execution/agent-runtime/src/sdk.rs @@ -57,8 +57,9 @@ pub use bitfun_harness::{ pub use bitfun_runtime_ports::{ AgentBackgroundResultRequest, AgentDialogTurnPort, AgentDialogTurnRequest, AgentInputAttachment, AgentLifecycleDeliveryPort, AgentLocalCommandTurnPort, - AgentLocalCommandTurnRecordRequest, AgentSessionArchiveRequest, AgentSessionCreateRequest, - AgentSessionCreateResult, AgentSessionDeleteRequest, AgentSessionForkPort, + AgentLocalCommandTurnRecordRequest, AgentSessionArchiveRequest, + AgentSessionArchiveStateRequest, AgentSessionCreateRequest, AgentSessionCreateResult, + AgentSessionDeleteRequest, AgentSessionForkAtTurnRequest, AgentSessionForkPort, AgentSessionForkRequest, AgentSessionForkResult, AgentSessionListRequest, AgentSessionManagementPort, AgentSessionModePort, AgentSessionModeUpdateRequest, AgentSessionModelPort, AgentSessionModelUpdateRequest, AgentSessionRenameRequest, @@ -315,6 +316,13 @@ impl AgentRuntime { self.inner.archive_session(request).await } + pub async fn set_session_archived( + &self, + request: AgentSessionArchiveStateRequest, + ) -> Result<(), RuntimeError> { + self.inner.set_session_archived(request).await + } + pub async fn record_completed_local_command_turn( &self, request: AgentLocalCommandTurnRecordRequest, @@ -345,6 +353,13 @@ impl AgentRuntime { self.inner.fork_session(request).await } + pub async fn fork_session_at_turn( + &self, + request: AgentSessionForkAtTurnRequest, + ) -> Result { + self.inner.fork_session_at_turn(request).await + } + pub async fn generate_session_usage( &self, request: AgentSessionUsageRequest, diff --git a/src/crates/execution/agent-runtime/tests/session_operation_ports.rs b/src/crates/execution/agent-runtime/tests/session_operation_ports.rs index 11867efcd6..7424d83168 100644 --- a/src/crates/execution/agent-runtime/tests/session_operation_ports.rs +++ b/src/crates/execution/agent-runtime/tests/session_operation_ports.rs @@ -1,10 +1,11 @@ use std::sync::{Arc, Mutex}; use bitfun_agent_runtime::sdk::{ - AgentRuntimeBuilder, AgentSessionForkPort, AgentSessionForkRequest, AgentSessionForkResult, - AgentSessionUsagePort, AgentSessionUsageRequest, AgentSubmissionPort, AgentSubmissionRequest, - AgentSubmissionResult, AgentTurnSettlementPort, AgentTurnSettlementRequest, PortErrorKind, - PortResult, SessionUsageReport, + AgentRuntimeBuilder, AgentSessionForkAtTurnRequest, AgentSessionForkPort, + AgentSessionForkRequest, AgentSessionForkResult, AgentSessionUsagePort, + AgentSessionUsageRequest, AgentSubmissionPort, AgentSubmissionRequest, AgentSubmissionResult, + AgentTurnSettlementPort, AgentTurnSettlementRequest, PortErrorKind, PortResult, + SessionUsageReport, }; use bitfun_agent_runtime::sdk::{AgentSessionCreateRequest, AgentSessionCreateResult}; @@ -60,6 +61,22 @@ impl AgentSessionForkPort for RecordingSessionOperations { agent_type: "agentic".to_string(), }) } + + async fn fork_session_at_turn( + &self, + request: AgentSessionForkAtTurnRequest, + ) -> PortResult { + assert_eq!(request.workspace_path, "D:/workspace/project"); + assert_eq!(request.source_session_id, "session-1"); + assert_eq!(request.source_turn_id, "turn-1"); + assert_eq!(request.remote_connection_id, None); + assert_eq!(request.remote_ssh_host, None); + Ok(AgentSessionForkResult { + session_id: "session-2".to_string(), + session_name: "Main (fork)".to_string(), + agent_type: "agentic".to_string(), + }) + } } #[async_trait::async_trait] @@ -98,9 +115,10 @@ async fn runtime_delegates_narrow_session_operations_to_registered_ports() { .expect("runtime"); let fork = runtime - .fork_session(AgentSessionForkRequest { + .fork_session_at_turn(AgentSessionForkAtTurnRequest { workspace_path: "D:/workspace/project".to_string(), source_session_id: "session-1".to_string(), + source_turn_id: "turn-1".to_string(), remote_connection_id: None, remote_ssh_host: None, }) @@ -108,6 +126,17 @@ async fn runtime_delegates_narrow_session_operations_to_registered_ports() { .expect("fork session"); assert_eq!(fork.session_id, "session-2"); + let latest_turn_fork = runtime + .fork_session(AgentSessionForkRequest { + workspace_path: "D:/workspace/project".to_string(), + source_session_id: "session-1".to_string(), + remote_connection_id: None, + remote_ssh_host: None, + }) + .await + .expect("latest-turn fork session"); + assert_eq!(latest_turn_fork.session_id, "session-2"); + let report = runtime .generate_session_usage(AgentSessionUsageRequest { session_id: "session-1".to_string(),