diff --git a/docs/architecture/agent-runtime-services-design.md b/docs/architecture/agent-runtime-services-design.md index fe94c4e1e3..6050c250e4 100644 --- a/docs/architecture/agent-runtime-services-design.md +++ b/docs/architecture/agent-runtime-services-design.md @@ -58,9 +58,9 @@ Agent Runtime SDK 和“对外能力门面”不是同一发布包必须同时 复用后，才允许两者共享稳定 DTO 或版本边界。 SDK 公共接口以 `AGENT_RUNTIME_SDK_API_VERSION` 标记兼容边界。当前接口版本为 v1 preview： -小版本更新允许增加可选 builder hook、DTO 字段或注册表查询能力，但不得改变既有端口语义、 -错误分类、session / turn 标识含义或默认 feature 依赖。任何需要调用方改写现有嵌入代码的变更， -必须提升接口版本并提供兼容迁移路径。 +小版本更新允许增加可选 builder hook、有默认实现的端口方法或注册表查询能力，但不得向外部可用 +Rust 结构体字面量（struct literal）构造的 DTO 直接增加字段，也不得改变既有端口语义、错误分类、session / turn 标识含义或 +默认 feature 依赖。任何需要调用方改写现有嵌入代码的变更，必须提升接口版本并提供兼容迁移路径。 只要外部调用方仍必须导入 `bitfun-core`、启用 `product-full`、持有具体服务管理器、读取产品命令 注册表或依赖全局可变状态，SDK 发布边界就不成立。 @@ -709,8 +709,10 @@ pub struct HarnessExecutionContext { Agent Runtime SDK 门面，不注册未实现的 `RuntimeServices` 能力，也不宣称完整 Desktop profile 可用。CLI 通过 一个调用级上下文把 Agent Runtime SDK、Harness、能力注册、调用级权限和 Agentic 事件广播交给 TUI、Exec、Session、Usage 与 交互模式下的 Peer Host。SDK 已承接会话创建/列举/删除/恢复、类型化转录读取、轮次提交/取消，以及 CLI/TUI -的工具确认、拒绝和用户问题回答；Peer Host 当前只消费同一 SDK 的工具确认与拒绝。SDK v1 尚未覆盖的固定 ID -会话创建、模型更新、分支、用量、快照和 +的工具确认、拒绝和用户问题回答；固定 ID 创建使用独立的 `create_session_with_id` 方法，普通创建 DTO 保持 v1 +字段集合。未实现该能力的提供方返回类型化不支持错误；实现成功时 Runtime 必须校验返回 ID 与请求完全一致，不能 +替换为自动生成的 ID。`SessionSelector::Create` 仍保持自动生成。Peer Host 当前只消费同一 SDK 的工具确认与拒绝。 +SDK v1 尚未覆盖的模型更新、分支、用量、快照和 持久化维护等操作仍由 `assembly/core` 的单一兼容门面转发。`doctor` 与 `health` 校验真实组装结果及必需注册完整性； Core 的 Network、Git 和 MCP Catalog 当前仍含兼容 marker，因此该诊断不等于对这些外部服务做实时探活。 @@ -799,8 +801,9 @@ Core 的 Network、Git 和 MCP Catalog 当前仍含兼容 marker，因此该诊 扩张通用 transcript。CLI/TUI 的工具确认、拒绝和用户问题回答，以及 ACP 服务端 / Peer Host 的工具确认与拒绝，通过类型化 `AgentInteractionResponsePort` 回到 Core 的工具管线或用户输入 owner，不改变审批策略或交互所有权。 `CoreAgentRuntimeCompatibility` 仍承载未迁移的 -固定 ID 会话创建、模型更新、持久化、分支、用量和快照等操作；不能据此把整个兼容门面一次性删除，也不能把这些 -操作提前声明为跨宿主稳定接口。 +模型更新、持久化、分支、用量和快照等操作；不能据此把整个兼容门面一次性删除，也不能把这些 +操作提前声明为跨宿主稳定接口。旧固定 ID 创建方法仅作为标记废弃的源码迁移转发保留一个兼容周期，生产 CLI +调用已经迁移到 Runtime SDK；该转发不得重新成为新调用入口。 ### 4.3 Product Capability 设计 diff --git a/docs/architecture/cli-product-line-design.md b/docs/architecture/cli-product-line-design.md index 57ae1d453b..1763d985b1 100644 --- a/docs/architecture/cli-product-line-design.md +++ b/docs/architecture/cli-product-line-design.md @@ -93,8 +93,10 @@ BitFun CLI 应成为可独立安装和发布的 Agent 产品，而不是 Desktop - BitFun 原生插件目录的发现、内容校验、来源确认，以及 OpenCode custom tool 静态名称预览。 - CLI 本地 Agent 入口以类型化 `RuntimeServices` 调用 `ProductAssembler`，选择 `DeliveryProfile::Cli`， 并把 `ProductRuntimeParts`、Agent Runtime SDK、事件源和调用级审批策略保存在一个 `CliRuntimeContext` 中。 -- TUI、`exec`、会话、用量和交互模式下的 Peer Host 复用同一上下文。SDK 已承接会话创建/列举/删除、 - 轮次提交和取消；SDK v1 尚未覆盖的固定 ID、恢复视图、消息、分支、用量、快照和工具确认由一个 Core +- TUI、`exec`、会话、用量和交互模式下的 Peer Host 复用同一上下文。SDK 已承接会话创建（包括 + `exec --session-id` 和缺失后端会话通过独立固定 ID 方法按原 ID 重建）/列举/删除、轮次提交和取消；普通创建 + DTO 保持 v1 字段集合，固定 ID 冲突返回 `InvalidRequest`。SDK v1 尚未覆盖的恢复视图、 + 消息、分支、用量、快照和工具确认由一个 Core 兼容门面转发给原 owner。 - Agentic Event Queue 仍是唯一事件 owner；TUI、`exec` 与 Peer Host 使用独立广播订阅，不互相消费事件。 - 有界旧队列只承担兼容存储；达到容量时不得抑制广播。CLI 保持一个后台 drain，订阅方一旦报告 lag/closed， diff --git a/docs/plans/core-decomposition-plan.md b/docs/plans/core-decomposition-plan.md index 0fa5d40f1d..b21512e0e1 100644 --- a/docs/plans/core-decomposition-plan.md +++ b/docs/plans/core-decomposition-plan.md @@ -22,7 +22,7 @@ | 事实 | 当前状态 | 结论 | |---|---|---| | 产品能力组装 | `DeliveryProfile`、`ProductAssembler`、能力计划、服务可用性和测试已存在 | 这些是可测试的 assembly facts，不代表产品入口已接入 | -| CLI / Desktop / ACP | 三者仍启用 `bitfun-core/product-full`；CLI 已提交 `DeliveryProfile::Cli` 并消费 Runtime Parts/SDK，Desktop 与 ACP 尚未切换 | CLI 已建立产品组装边界但仍保留 Core owner；三个入口均未完成 owner 迁移 | +| CLI / Desktop / ACP | 三者仍按需启用 `bitfun-core/product-full`；CLI 与 ACP 已分别提交对应 `DeliveryProfile` 并消费 Runtime Parts/SDK，Desktop 主交互已消费由现有 owner 构造的窄口径 SDK 门面 | 三个入口均复用单一 Core owner；完整 Desktop profile 和剩余兼容操作仍需逐项迁移 | | Server | 当前生产路由只形成 health/info/ping 基线 | 没有插件状态或独立产品组装闭环 | | Server / Remote / Web / Mobile Web / SDK profile | 当前为空计划、未接入入口或仅有 preview 测试 | 不得据枚举值宣称产品能力已交付 | | Agent Runtime SDK | 已有无 `bitfun-core` 依赖的 v1 preview 门面和 smoke test | 发布边界仍需真实嵌入方证明 | @@ -59,7 +59,8 @@ assembly → apps。embedded 的 bind、静态 fallback 和任务生命周期移 CLI 是首个入口迁移对象，因为它已有独立产品诉求、显式设计和最小 CI 命令。 当前纵向切片已经完成：入口只提交一次 `DeliveryProfile::Cli`，通过现有 `ProductAssembler` 获得计划、服务可用性、 -Harness 和禁用的插件 binding；TUI、Exec、Session 与 Usage 共用一个 `CliRuntimeContext`。会话创建/列举/ +Harness 和禁用的插件 binding；TUI、Exec、Session 与 Usage 共用一个 `CliRuntimeContext`。会话创建（包括 +`exec --session-id` 和缺失后端会话通过独立固定 ID 方法按原 ID 重建）/列举/ 删除、轮次提交和取消走 Agent Runtime SDK；SDK v1 缺口集中在一个 Core 兼容门面。Agentic Event Queue 仍是唯一 owner，各入口只建立独立广播订阅，有界兼容队列满载不再阻断广播。TUI 与 Exec 审批均为调用级策略，不写全局 配置；CLI 本地路径不获取具体 PersistenceManager。交互、执行和管理入口分别控制 Peer Host/MCP 生命周期，管理查询不启动 @@ -71,8 +72,8 @@ Peer Host 的 Runtime 接入和跨 Relay/Desktop/Web 的协议切换保持独立 下一步按独立纵向切片推进： -1. 以真实调用方和行为等价测试补齐 SDK 端口，逐项缩小固定 ID 会话创建、模型更新、分支、用量、快照和持久化维护兼容面。 -2. 迁移 ACP 的会话/权限/事件投影，但保留 ACP stdio 生命周期在接口入口。 +1. 以真实调用方和行为等价测试补齐 SDK 端口，逐项缩小模型更新、分支、用量、快照和持久化维护兼容面。 +2. 继续迁移 ACP 尚未接入 SDK 的持久化历史、模型/模式和 MCP 操作；ACP stdio 与协议投影生命周期保留在接口入口。 3. 继续拆分 TUI 副作用边界并补 package smoke，不以大规模重写替代现有回归保护。 当前 assembly 切换条件已经满足：CLI 生产入口消费真实组装结果，目标链路没有第二套状态，独立测试与三平台 diff --git a/docs/plans/product-architecture-evolution-plan.md b/docs/plans/product-architecture-evolution-plan.md index a6f7bdd9d4..ab1d9e57e3 100644 --- a/docs/plans/product-architecture-evolution-plan.md +++ b/docs/plans/product-architecture-evolution-plan.md @@ -99,8 +99,9 @@ HarmonyOS 手机 Remote App 不在该平台执行范围内。 ## 7. 工作流五：入口逐项迁移 -- CLI：会话创建/列举/删除/恢复、类型化转录和轮次提交/取消已由真实入口消费 Runtime SDK；TUI 通过 SDK 处理 - 工具确认、拒绝和用户问题回答，Exec 与交互模式下的 Peer Host 处理工具确认与拒绝。固定 ID 会话创建、模型 +- CLI：会话创建（包括 `exec --session-id` 和缺失后端会话通过独立固定 ID 方法按原 ID 重建）/列举/删除/恢复、类型化转录和轮次 + 提交/取消已由真实入口消费 Runtime SDK； + TUI 通过 SDK 处理工具确认、拒绝和用户问题回答，Exec 与交互模式下的 Peer Host 处理工具确认与拒绝。模型 更新、分支、用量、快照和持久化维护等操作在补齐端口与行为测试前继续由 现有单一兼容路径转发。 - ACP：CLI 行为稳定后单独迁移会话、权限和事件投影；ACP stdio 生命周期留在接口入口。 diff --git a/src/apps/cli/src/agent/core_adapter.rs b/src/apps/cli/src/agent/core_adapter.rs index 4dd718d72d..f471ed7ccb 100644 --- a/src/apps/cli/src/agent/core_adapter.rs +++ b/src/apps/cli/src/agent/core_adapter.rs @@ -257,14 +257,20 @@ impl CoreAgentAdapter { } } - self.compatibility + self.runtime .create_session_with_id( session_id.to_string(), - session_name, - effective_agent_type, - self.workspace_path_string(), + AgentSessionCreateRequest { + session_name, + agent_type: effective_agent_type, + workspace_path: Some(self.workspace_path_string()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, ) - .await?; + .await + .map_err(|error| anyhow::anyhow!(error.into_message()))?; tracing::info!("Recreated backend session with existing id: {}", session_id); Ok(()) @@ -316,18 +322,24 @@ impl CoreAgentAdapter { let mut session_id_guard = self.session_id.lock().await; let session = self - .compatibility + .runtime .create_session_with_id( - session_id.clone(), - Self::build_default_session_name(), - agent_type.to_string(), - self.workspace_path_string(), + session_id, + AgentSessionCreateRequest { + session_name: Self::build_default_session_name(), + agent_type: agent_type.to_string(), + workspace_path: Some(self.workspace_path_string()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, ) - .await?; + .await + .map_err(|error| anyhow::anyhow!(error.into_message()))?; let id = session.session_id.clone(); *session_id_guard = Some(id.clone()); - tracing::info!("Created core session with fixed id: {}", id); + tracing::info!("Created runtime session with fixed id: {}", id); Ok(id) } diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index 0664e19cc8..d61d15d902 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -7474,46 +7474,78 @@ fn resolve_agent_session_create_created_by( .map(ToOwned::to_owned) } +fn runtime_port_backend_error(error: BitFunError) -> bitfun_runtime_ports::PortError { + bitfun_runtime_ports::PortError::new( + bitfun_runtime_ports::PortErrorKind::Backend, + error.to_string(), + ) +} + +async fn create_agent_session_from_runtime_request( + coordinator: &ConversationCoordinator, + session_id: Option, + request: bitfun_runtime_ports::AgentSessionCreateRequest, + map_core_error: fn(BitFunError) -> bitfun_runtime_ports::PortError, +) -> bitfun_runtime_ports::PortResult { + let workspace_path = request.workspace_path.clone().ok_or_else(|| { + bitfun_runtime_ports::PortError::new( + bitfun_runtime_ports::PortErrorKind::InvalidRequest, + "workspace_path is required to create an agent session", + ) + })?; + let created_by = resolve_agent_session_create_created_by(&request.metadata); + let session = coordinator + .create_session_with_workspace_and_creator( + session_id, + request.session_name, + request.agent_type, + SessionConfig { + workspace_path: Some(workspace_path.clone()), + remote_connection_id: request.remote_connection_id, + remote_ssh_host: request.remote_ssh_host, + ..Default::default() + }, + workspace_path, + created_by, + ) + .await + .map_err(map_core_error)?; + + Ok(bitfun_runtime_ports::AgentSessionCreateResult { + session_id: session.session_id, + session_name: session.session_name, + agent_type: session.agent_type, + }) +} + #[async_trait::async_trait] impl bitfun_runtime_ports::AgentSubmissionPort for ConversationCoordinator { async fn create_session( &self, request: bitfun_runtime_ports::AgentSessionCreateRequest, ) -> bitfun_runtime_ports::PortResult { - let workspace_path = request.workspace_path.clone().ok_or_else(|| { + create_agent_session_from_runtime_request(self, None, request, runtime_port_backend_error) + .await + } + + async fn create_session_with_id( + &self, + session_id: String, + request: bitfun_runtime_ports::AgentSessionCreateRequest, + ) -> bitfun_runtime_ports::PortResult { + bitfun_core_types::validate_session_id(&session_id).map_err(|message| { bitfun_runtime_ports::PortError::new( bitfun_runtime_ports::PortErrorKind::InvalidRequest, - "workspace_path is required to create an agent session", + message, ) })?; - - let session = self - .create_session_with_workspace_and_creator( - None, - request.session_name, - request.agent_type, - SessionConfig { - workspace_path: Some(workspace_path.clone()), - remote_connection_id: request.remote_connection_id.clone(), - remote_ssh_host: request.remote_ssh_host.clone(), - ..Default::default() - }, - workspace_path, - resolve_agent_session_create_created_by(&request.metadata), - ) - .await - .map_err(|error| { - bitfun_runtime_ports::PortError::new( - bitfun_runtime_ports::PortErrorKind::Backend, - error.to_string(), - ) - })?; - - Ok(bitfun_runtime_ports::AgentSessionCreateResult { - session_id: session.session_id, - session_name: session.session_name, - agent_type: session.agent_type, - }) + create_agent_session_from_runtime_request( + self, + Some(session_id), + request, + runtime_port_error_from_bitfun, + ) + .await } async fn submit_message( @@ -8254,7 +8286,9 @@ mod tests { } use tokio::sync::RwLock as TokioRwLock; - fn test_coordinator() -> (ConversationCoordinator, Arc) { + fn test_coordinator_with_max_active_sessions( + max_active_sessions: usize, + ) -> (ConversationCoordinator, Arc) { let event_queue = Arc::new(EventQueue::new(EventQueueConfig::default())); let session_manager = Arc::new(SessionManager::new( Arc::new(SessionContextStore::new()), @@ -8263,7 +8297,7 @@ mod tests { .expect("persistence manager"), ), SessionManagerConfig { - max_active_sessions: 100, + max_active_sessions, session_idle_timeout: Duration::from_secs(3600), auto_save_interval: Duration::from_secs(300), enable_persistence: false, @@ -8303,6 +8337,10 @@ mod tests { (coordinator, session_manager) } + fn test_coordinator() -> (ConversationCoordinator, Arc) { + test_coordinator_with_max_active_sessions(100) + } + #[test] fn conversation_coordinator_exposes_remote_runtime_ports() { fn assert_cancellation_port() {} @@ -9117,6 +9155,107 @@ mod tests { let _ = std::fs::remove_dir_all(workspace_path); } + #[tokio::test] + async fn agent_submission_create_session_preserves_v1_backend_error_classification() { + let (coordinator, _) = test_coordinator_with_max_active_sessions(0); + let error = AgentSubmissionPort::create_session( + &coordinator, + AgentSessionCreateRequest { + session_name: "Over capacity".to_string(), + agent_type: "agentic".to_string(), + workspace_path: Some(std::env::temp_dir().to_string_lossy().into_owned()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, + ) + .await + .expect_err("v1 create should preserve its backend error classification"); + + assert_eq!(error.kind, bitfun_runtime_ports::PortErrorKind::Backend); + } + + #[tokio::test] + async fn agent_submission_create_session_preserves_requested_session_id() { + let (coordinator, session_manager) = test_coordinator(); + let workspace_path = std::env::temp_dir().join(format!( + "bitfun-agent-session-fixed-id-port-test-{}", + uuid::Uuid::new_v4() + )); + std::fs::create_dir_all(&workspace_path).expect("workspace dir should exist"); + + let result = AgentSubmissionPort::create_session_with_id( + &coordinator, + "fixed-session-id".to_string(), + AgentSessionCreateRequest { + session_name: "Fixed worker".to_string(), + agent_type: "agentic".to_string(), + workspace_path: Some(workspace_path.to_string_lossy().into_owned()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, + ) + .await + .expect("fixed-id session creation should succeed"); + + assert_eq!(result.session_id, "fixed-session-id"); + assert!(session_manager.get_session("fixed-session-id").is_some()); + + let duplicate_error = AgentSubmissionPort::create_session_with_id( + &coordinator, + "fixed-session-id".to_string(), + AgentSessionCreateRequest { + session_name: "Duplicate worker".to_string(), + agent_type: "agentic".to_string(), + workspace_path: Some(workspace_path.to_string_lossy().into_owned()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, + ) + .await + .expect_err("duplicate fixed session id should be rejected"); + assert_eq!( + duplicate_error.kind, + bitfun_runtime_ports::PortErrorKind::InvalidRequest + ); + assert!(duplicate_error.message.contains("already exists")); + assert_eq!( + session_manager + .get_session("fixed-session-id") + .expect("original fixed-id session should remain") + .session_name, + "Fixed worker" + ); + + let _ = std::fs::remove_dir_all(workspace_path); + } + + #[tokio::test] + async fn agent_submission_create_session_rejects_invalid_requested_session_id() { + let (coordinator, _) = test_coordinator(); + let error = AgentSubmissionPort::create_session_with_id( + &coordinator, + "../other-session".to_string(), + AgentSessionCreateRequest { + session_name: "Invalid worker".to_string(), + agent_type: "agentic".to_string(), + workspace_path: Some(std::env::temp_dir().to_string_lossy().into_owned()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, + ) + .await + .expect_err("invalid fixed session id should be rejected"); + + assert_eq!( + error.kind, + bitfun_runtime_ports::PortErrorKind::InvalidRequest + ); + } + #[tokio::test] async fn subagent_session_config_preserves_registered_remote_workspace_identity() { let manager = init_remote_workspace_manager(); diff --git a/src/crates/assembly/core/src/agentic/persistence/manager.rs b/src/crates/assembly/core/src/agentic/persistence/manager.rs index 3da1413a15..d1e88b9b6f 100644 --- a/src/crates/assembly/core/src/agentic/persistence/manager.rs +++ b/src/crates/assembly/core/src/agentic/persistence/manager.rs @@ -284,6 +284,8 @@ struct ParsedTranscriptTurnSelector { pub struct PersistenceManager { path_manager: Arc, runtime_service: Arc, + #[cfg(test)] + fail_next_session_state_write: std::sync::Mutex>, } impl PersistenceManager { @@ -291,6 +293,8 @@ impl PersistenceManager { Ok(Self { runtime_service: Arc::new(WorkspaceRuntimeService::new(path_manager.clone())), path_manager, + #[cfg(test)] + fail_next_session_state_write: std::sync::Mutex::new(None), }) } @@ -307,6 +311,14 @@ impl PersistenceManager { &self.runtime_service } + #[cfg(test)] + pub(crate) fn fail_next_session_state_write_for_test(&self, session_id: &str) { + *self + .fail_next_session_state_write + .lock() + .expect("session state fault lock") = Some(session_id.to_string()); + } + /// Resolve the on-disk sessions directory for `workspace_path`. /// /// Callers may pass either a logical workspace root or an already-resolved @@ -1360,12 +1372,78 @@ impl PersistenceManager { // ============ Session Persistence ============ + /// Persist a newly created session without overwriting an existing session ID. + /// + /// The final session directory is created exclusively so this manager owns any + /// cleanup required by a failed first write. This also prevents a losing + /// creator in another runtime or process from deleting the winning session. + pub(crate) async fn create_session_if_absent( + &self, + workspace_path: &Path, + session: &Session, + ) -> BitFunResult<()> { + Self::validate_session_id(&session.session_id)?; + self.ensure_runtime_for_write(workspace_path).await?; + + let sessions_dir = self.project_sessions_dir(workspace_path); + fs::create_dir_all(&sessions_dir).await.map_err(|error| { + BitFunError::io(format!( + "Failed to create sessions directory {}: {}", + sessions_dir.display(), + error + )) + })?; + let session_dir = self + .session_layout(workspace_path) + .session_dir(&session.session_id); + match fs::create_dir(&session_dir).await { + Ok(()) => {} + Err(error) if error.kind() == ErrorKind::AlreadyExists => { + return Err(BitFunError::Validation(format!( + "Persisted session ID already exists: {}", + session.session_id + ))); + } + Err(error) => { + return Err(BitFunError::io(format!( + "Failed to claim session directory {}: {}", + session_dir.display(), + error + ))); + } + } + + if let Err(error) = self.save_session_files(workspace_path, session).await { + if let Err(cleanup_error) = self + .session_metadata_store(workspace_path) + .delete_session_dir_and_index(&session.session_id) + .await + { + warn!( + "Failed to clean up partial session persistence: session_id={}, error={}", + session.session_id, cleanup_error + ); + } + return Err(error); + } + + Ok(()) + } + /// Save session pub async fn save_session(&self, workspace_path: &Path, session: &Session) -> BitFunResult<()> { Self::validate_session_id(&session.session_id)?; self.ensure_runtime_for_write(workspace_path).await?; self.ensure_session_dir(workspace_path, &session.session_id) .await?; + self.save_session_files(workspace_path, session).await + } + + async fn save_session_files( + &self, + workspace_path: &Path, + session: &Session, + ) -> BitFunResult<()> { let existing_metadata = self .load_session_metadata(workspace_path, &session.session_id) .await?; @@ -1384,6 +1462,17 @@ impl PersistenceManager { compression_state: session.compression_state.clone(), runtime_state: sanitize_persisted_session_state(&session.state), }; + #[cfg(test)] + { + let mut fault = self + .fail_next_session_state_write + .lock() + .expect("session state fault lock"); + if fault.as_deref() == Some(session.session_id.as_str()) { + *fault = None; + return Err(BitFunError::io("Injected session state write failure")); + } + } self.save_stored_session_state(workspace_path, &session.session_id, &state) .await } @@ -2654,6 +2743,7 @@ mod tests { SessionRelationshipKind, SessionTranscriptExportOptions, StoredSessionIndexFile, TextItemData, UserMessageData, }; + use crate::BitFunError; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Instant; @@ -2702,6 +2792,82 @@ mod tests { assert!(error.to_string().contains("session_id"), "{error}"); } + #[tokio::test] + async fn concurrent_first_session_persistence_keeps_the_winner() { + let workspace = TestWorkspace::new(); + let manager_a = Arc::new( + PersistenceManager::new(workspace.path_manager()).expect("first persistence manager"), + ); + let manager_b = Arc::new( + PersistenceManager::new(workspace.path_manager()).expect("second persistence manager"), + ); + let session_id = format!("concurrent-session-{}", Uuid::new_v4()); + let config = SessionConfig { + workspace_path: Some(workspace.path().to_string_lossy().to_string()), + ..Default::default() + }; + let session_a = Session::new_with_id( + session_id.clone(), + "First contender".to_string(), + "agent".to_string(), + config.clone(), + ); + let session_b = Session::new_with_id( + session_id.clone(), + "Second contender".to_string(), + "agent".to_string(), + config, + ); + let workspace_path = workspace.path().to_path_buf(); + let barrier = Arc::new(tokio::sync::Barrier::new(3)); + + let first = tokio::spawn({ + let manager = manager_a.clone(); + let barrier = barrier.clone(); + let workspace_path = workspace_path.clone(); + async move { + barrier.wait().await; + let result = manager + .create_session_if_absent(&workspace_path, &session_a) + .await; + ("First contender", result) + } + }); + let second = tokio::spawn({ + let manager = manager_b.clone(); + let barrier = barrier.clone(); + let workspace_path = workspace_path.clone(); + async move { + barrier.wait().await; + let result = manager + .create_session_if_absent(&workspace_path, &session_b) + .await; + ("Second contender", result) + } + }); + barrier.wait().await; + + let first = first.await.expect("first contender should finish"); + let second = second.await.expect("second contender should finish"); + let outcomes = [first, second]; + let winner = outcomes + .iter() + .find_map(|(name, result)| result.is_ok().then_some(*name)) + .expect("one contender must persist the session"); + let failures = outcomes + .iter() + .filter_map(|(_, result)| result.as_ref().err()) + .collect::>(); + + assert_eq!(failures.len(), 1, "exactly one contender must fail"); + assert!(matches!(failures[0], BitFunError::Validation(_))); + let persisted = manager_a + .load_session(workspace.path(), &session_id) + .await + .expect("the winning session must remain persisted"); + assert_eq!(persisted.session_name, winner); + } + #[tokio::test] async fn token_anchors_save_load_and_delete_roundtrip() { let workspace = TestWorkspace::new(); diff --git a/src/crates/assembly/core/src/agentic/session/session_manager.rs b/src/crates/assembly/core/src/agentic/session/session_manager.rs index c09b570232..273a1626e7 100644 --- a/src/crates/assembly/core/src/agentic/session/session_manager.rs +++ b/src/crates/assembly/core/src/agentic/session/session_manager.rs @@ -1864,8 +1864,6 @@ impl SessionManager { ))); } } - self.commit_session_storage_path_claim(&session_id, &session_storage_path, storage_claim); - // 2. Initialize the in-memory context cache. self.context_store.create_session(&session_id); self.token_anchor_store.create_session(&session_id); @@ -1877,10 +1875,26 @@ impl SessionManager { // Use the local `session` directly -- no need to re-fetch from DashMap, // which would hold a Ref guard across the async save_session call. if self.config.enable_persistence && Self::should_persist_session(&session) { - self.persistence_manager - .save_session(&session_storage_path, &session) - .await?; + if let Err(error) = self + .persistence_manager + .create_session_if_absent(&session_storage_path, &session) + .await + { + self.sessions.remove(&session_id); + self.context_store.delete_session(&session_id); + self.token_anchor_store.delete_session(&session_id); + self.turn_skill_agent_snapshot_store + .delete_session(&session_id); + self.file_read_state_store.delete_session(&session_id); + self.release_failed_session_storage_path_claim( + &session_id, + &session_storage_path, + storage_claim, + ); + return Err(error); + } } + self.commit_session_storage_path_claim(&session_id, &session_storage_path, storage_claim); info!("Session created: session_name={}", session.session_name); @@ -6183,6 +6197,52 @@ mod tests { assert!(manager.session_storage_path_index.get(invalid_id).is_none()); } + #[tokio::test] + async fn persistent_session_creation_failure_does_not_publish_runtime_state() { + let workspace = TestWorkspace::new(); + let persistence_manager = Arc::new( + PersistenceManager::new(workspace.path_manager()).expect("persistence manager"), + ); + let session_id = "failed-persistent-session"; + persistence_manager.fail_next_session_state_write_for_test(session_id); + let manager = test_manager(persistence_manager.clone()); + let config = SessionConfig { + workspace_path: Some(workspace.path().to_string_lossy().to_string()), + ..Default::default() + }; + + manager + .create_session_with_id( + Some(session_id.to_string()), + "Must not become visible".to_string(), + "agentic".to_string(), + config.clone(), + ) + .await + .expect_err("state persistence failure must fail session creation"); + + assert!(manager.get_session(session_id).is_none()); + assert!(manager.session_storage_path_index.get(session_id).is_none()); + assert!(!persistence_manager + .session_storage_exists(workspace.path(), session_id) + .expect("session storage existence")); + assert!(persistence_manager + .load_session_metadata(workspace.path(), session_id) + .await + .expect("load session metadata") + .is_none()); + + manager + .create_session_with_id( + Some(session_id.to_string()), + "Retry succeeds".to_string(), + "agentic".to_string(), + config, + ) + .await + .expect("retry should not be blocked by partial persistence"); + } + #[tokio::test] async fn background_title_update_cannot_recreate_storage_during_deletion() { let workspace = TestWorkspace::new(); diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index ccd18c29c8..f466d331da 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -138,6 +138,8 @@ impl CoreAgentRuntimeCompatibility { } } + /// Compatibility shim for callers migrating to the Agent Runtime SDK. + #[deprecated(note = "use AgentRuntime::create_session_with_id")] pub async fn create_session_with_id( &self, session_id: String, @@ -680,6 +682,7 @@ mod tests { } #[test] + #[allow(deprecated)] fn compatibility_operations_have_one_core_owned_facade() { fn build( coordinator: Arc, diff --git a/src/crates/contracts/runtime-ports/src/lib.rs b/src/crates/contracts/runtime-ports/src/lib.rs index 7eea6ec17d..dbd3580a90 100644 --- a/src/crates/contracts/runtime-ports/src/lib.rs +++ b/src/crates/contracts/runtime-ports/src/lib.rs @@ -1650,6 +1650,23 @@ pub trait AgentSubmissionPort: Send + Sync { request: AgentSessionCreateRequest, ) -> PortResult; + /// Creates a session with an exact caller-provided identity. + /// + /// Providers that do not support exact identity creation keep the default + /// typed unsupported response. A successful response must preserve + /// `session_id` exactly. + async fn create_session_with_id( + &self, + session_id: String, + request: AgentSessionCreateRequest, + ) -> PortResult { + let _ = (session_id, request); + Err(PortError::new( + PortErrorKind::NotAvailable, + "exact session identity creation is not supported by this provider", + )) + } + async fn submit_message( &self, request: AgentSubmissionRequest, @@ -1955,6 +1972,35 @@ impl SubagentContextMode { mod tests { use super::*; + #[test] + fn agent_session_create_request_keeps_rust_literal_compatible() { + let request = AgentSessionCreateRequest { + session_name: "Generated session".to_string(), + agent_type: "agentic".to_string(), + workspace_path: Some("/workspace/project".to_string()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }; + + let json = serde_json::to_value(request).expect("serialize create request"); + + assert!(json.get("sessionId").is_none()); + } + + #[test] + fn agent_session_create_request_keeps_legacy_payload_compatible() { + let request: AgentSessionCreateRequest = serde_json::from_value(serde_json::json!({ + "sessionName": "Generated session", + "agentType": "agentic", + "workspacePath": "/workspace/project" + })) + .expect("deserialize legacy create request"); + + let json = serde_json::to_value(request).expect("serialize create request"); + assert!(json.get("sessionId").is_none()); + } + #[test] fn port_error_display_keeps_kind_and_message() { let error = PortError::new(PortErrorKind::NotAvailable, "coordinator missing"); diff --git a/src/crates/execution/agent-runtime/src/runtime.rs b/src/crates/execution/agent-runtime/src/runtime.rs index 01ef50157b..53357543b3 100644 --- a/src/crates/execution/agent-runtime/src/runtime.rs +++ b/src/crates/execution/agent-runtime/src/runtime.rs @@ -18,9 +18,9 @@ use bitfun_runtime_ports::{ AgentSubmissionResult, AgentSubmissionSource, AgentThreadGoalCreateRequest, AgentThreadGoalDeliveryRequest, AgentThreadGoalGetRequest, AgentThreadGoalManagementPort, AgentThreadGoalUpdateStatusRequest, AgentTurnCancellationPort, AgentTurnCancellationRequest, - AgentTurnCancellationResult, DialogSubmitOutcome, PluginRuntimeBinding, PortError, PortResult, - RuntimeEventEnvelope, SessionTranscript, SessionTranscriptReader, SessionTranscriptRequest, - ThreadGoal, + AgentTurnCancellationResult, DialogSubmitOutcome, PluginRuntimeBinding, PortError, + PortErrorKind, PortResult, RuntimeEventEnvelope, SessionTranscript, SessionTranscriptReader, + SessionTranscriptRequest, ThreadGoal, }; use bitfun_runtime_services::RuntimeServices; @@ -673,6 +673,29 @@ impl AgentRuntime { .map_err(RuntimeError::from) } + pub async fn create_session_with_id( + &self, + session_id: String, + request: AgentSessionCreateRequest, + ) -> Result { + let result = self + .submission + .create_session_with_id(session_id.clone(), request) + .await + .map_err(RuntimeError::from)?; + if result.session_id != session_id { + return Err(PortError::new( + PortErrorKind::Backend, + format!( + "agent submission provider returned session_id '{}' for requested session_id '{}'", + result.session_id, session_id + ), + ) + .into()); + } + Ok(result) + } + pub async fn list_sessions( &self, request: AgentSessionListRequest, @@ -949,6 +972,7 @@ mod tests { #[derive(Debug, Default)] struct FakeAgentRuntimePorts { created_sessions: Mutex>, + exact_session_result_id: Mutex>, submitted_messages: Mutex>, cancelled_turns: Mutex>, listed_sessions: Mutex>, @@ -1134,6 +1158,24 @@ mod tests { }) } + async fn create_session_with_id( + &self, + session_id: String, + request: AgentSessionCreateRequest, + ) -> PortResult { + self.created_sessions.lock().unwrap().push(request.clone()); + Ok(AgentSessionCreateResult { + session_id: self + .exact_session_result_id + .lock() + .unwrap() + .clone() + .unwrap_or(session_id), + session_name: request.session_name, + agent_type: request.agent_type, + }) + } + async fn submit_message( &self, request: AgentSubmissionRequest, @@ -1314,6 +1356,65 @@ mod tests { assert_eq!(err, RuntimeBuildError::UnsupportedPluginRuntimeHostBinding); } + #[tokio::test] + async fn create_session_with_id_uses_exact_identity() { + let ports = Arc::new(FakeAgentRuntimePorts::default()); + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(ports.clone()) + .build() + .expect("runtime"); + + let created = runtime + .create_session_with_id( + "fixed-session-id".to_string(), + AgentSessionCreateRequest { + session_name: "Fixed session".to_string(), + agent_type: "agentic".to_string(), + workspace_path: Some("/workspace/project".to_string()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, + ) + .await + .expect("create session"); + + assert_eq!(created.session_id, "fixed-session-id"); + } + + #[tokio::test] + async fn create_session_with_id_rejects_provider_identity_mismatch() { + let ports = Arc::new(FakeAgentRuntimePorts::default()); + *ports.exact_session_result_id.lock().unwrap() = Some("other-session-id".to_string()); + let runtime = AgentRuntimeBuilder::new() + .with_submission_port(ports) + .build() + .expect("runtime"); + + let error = runtime + .create_session_with_id( + "fixed-session-id".to_string(), + AgentSessionCreateRequest { + session_name: "Fixed session".to_string(), + agent_type: "agentic".to_string(), + workspace_path: Some("/workspace/project".to_string()), + remote_connection_id: None, + remote_ssh_host: None, + metadata: serde_json::Map::new(), + }, + ) + .await + .expect_err("runtime must reject a provider identity mismatch"); + + assert!(matches!( + error, + RuntimeError::Port(PortError { + kind: PortErrorKind::Backend, + .. + }) + )); + } + #[tokio::test] async fn run_creates_session_and_submits_turn_through_ports() { let ports = Arc::new(FakeAgentRuntimePorts::default()); diff --git a/src/crates/execution/agent-runtime/src/sdk.rs b/src/crates/execution/agent-runtime/src/sdk.rs index f2ffaec2b3..2d8250389a 100644 --- a/src/crates/execution/agent-runtime/src/sdk.rs +++ b/src/crates/execution/agent-runtime/src/sdk.rs @@ -240,6 +240,14 @@ impl AgentRuntime { self.inner.create_session(request).await } + pub async fn create_session_with_id( + &self, + session_id: String, + request: AgentSessionCreateRequest, + ) -> Result { + self.inner.create_session_with_id(session_id, request).await + } + pub async fn list_sessions( &self, request: AgentSessionListRequest,