diff --git a/docs/architecture/agent-runtime-deployment-design.md b/docs/architecture/agent-runtime-deployment-design.md index 2dd8af0e06..c03cf674fd 100644 --- a/docs/architecture/agent-runtime-deployment-design.md +++ b/docs/architecture/agent-runtime-deployment-design.md @@ -207,7 +207,7 @@ sequenceDiagram end ``` -当前私有协议（v14）只覆盖 TUI 已有用户旅程需要的窄操作： +当前私有协议（v15）只覆盖 TUI 已有用户旅程需要的窄操作： | 已支持 | 明确不支持 | |---|---| @@ -232,6 +232,7 @@ sequenceDiagram - v11 增加无请求体、无 Session lease 的只读 workspace diff operation。它只在当前连接没有活动 Turn 时查询 Runtime 启动时绑定的 canonical workspace，避免单连接请求排序阻塞流事件或 Turn 控制，并返回 Runtime Port DTO；Git 行为仍由 `services-integrations` provider 持有。文本 patch 总量限制为 3 MiB，为 JSON 转义和 envelope 预留既有 8 MiB response frame 的空间；该 operation 不隐式获得 stage/reset/commit、Remote 或公开 SDK 能力。 - v12 增加用户显式 Shell Turn；v13 增加活动 Turn steering。两者都复用 Agent Runtime 的原有准入、Tool、权限、持久化和取消 owner，不在 IPC 内复制执行状态机。 - v14 增加三个 current-root-controller 限定的 lineage operation：查询 Runtime 归一化后的扁平 lineage、读取已验证后代的权威 transcript，以及取消指定后代的活动执行子树。查询和读取可在根 Turn 活动时执行；取消复用现有 Session abort 语义，但不切换 controller，也不引入 observer、detach、分页或通用 Session RPC。 +- v15 为后代 transcript 读取增加 `required_settled_turn_ids` 一致性前置条件：Runtime 必须确认这些 Turn 已由 owner 持久化为终态，否则返回 `outcome_unknown`，由 TUI 在同一绝对期限内退避重试；TUI 只保留事件投影和该读屏障，不合并或重写权威 transcript。后代取消同时携带用户实际看到的 `expected_active_turn_id`，并在 owner 锁内拒绝已经切换的 Turn，避免迟到操作取消后续执行。lineage 查询和 transcript 读取是每连接至多一个的可抢占推测读取；更新的请求会取消旧读取，使后代取消和 Session 切换不会排在慢 transcript I/O 之后。该行为不放宽 controller 校验，不引入 observer 或通用多路复用。 - 一个连接最多控制一个 Session、同时最多提交一个活动 Turn；一个 Session 同时只有一个 controller。create/restore/fork 在完整结果通过大小检查后才原子切换控制权，失败时保留原 Session。fork 只接受当前 controller 的空闲 Session；无选中 Turn 时复制到最新持久化 Turn，指定 `before_turn_id` 时只复制该 Turn 之前的历史。活动 Turn 期间不能切换或 fork Session，也不能修改其名称、Agent mode 或 model；删除只作用于非当前且未被任何连接控制的 Session。 - Submit 与手动 context compaction 都使用调用方已有的 `turn_id` 标识不确定结果；若操作超时，返回 `outcome_unknown`、关闭连接并按该 ID 取消。手动 compaction 要求当前 controller 且 Session 空闲，由 Core 通过与普通对话 Turn 共用的原子准入路径创建一个可审计 maintenance Turn，并在取得所有权后读取压缩上下文：planning 阶段允许取消，atomic commit 开始后忽略晚到取消并保持 Processing 直至终态持久化完成。maintenance Turn 保留在权威 transcript 中但不进入模型上下文，live/restored payload 使用同一 compression ID 和 `applied` 事实；commit 后的持久化故障发布明确失败终态而不是遗留 Processing。断连取消只有得到确认后才释放 Session 控制权；无法确认时继续隔离该 Session，直到 Runtime 进程退出。 - Session delete/rename 和 Agent mode/model update 复用既有 Runtime 端口和校验，Runtime 对最终结果保持权威并拒绝无效目标。它们都是有副作用操作；发送前编码或 frame 上限失败表示请求未执行，连接仍可使用。rename 写入失败时恢复旧 metadata：确认恢复后返回明确失败，无法确认时返回 `outcome_unknown`。Shared Client 在请求写入后响应超时或丢失连接时也返回 `outcome_unknown` 并断开连接。两种情况都不自动重试：rename 由用户恢复 Session 并核对当前值；delete 由用户重新打开 `/sessions` 核对目标是否仍存在。模式与模型目录仍是同版本第一方产品事实，不加入 IPC。 @@ -263,7 +264,7 @@ flowchart LR | Shared request | Client 将 operation 编码一次并写入一个长度前缀 frame | 请求保持 128 KiB 上限；业务层只接收类型化 operation | | Shared response/event | Server 将结果或事件编码一次后写出 | 响应/事件保持 8 MiB 上限；超限使事件流明确失效，不能无界分配 | | Shared receive | 每个方向只有一个严格 transport decode 边界 | 未知信封字段和不兼容版本 fail closed；严格校验可以检查规范化 JSON，但不能把动态 JSON 传入 Runtime owner | -| 多 TUI | 一个 Runtime、最多 64 个连接；每个 Client 的 command channel 容量为 64、event channel 容量为 256 | request gate 使每个 Client 同时只有一个请求进入 channel；事件落后时失效而非无限缓存 | +| 多 TUI | 一个 Runtime、最多 64 个连接；每个 Client 的 command channel 容量为 64、event channel 容量为 256 | request gate 使每个 Client 同时只有一个控制请求进入 channel；lineage 推测读取写入后释放 gate，Server 同时只保留一个且允许更新请求抢占；Client 使用并发 reader 与单一有序 writer，避免大 transcript 响应和后续大请求互相阻塞；事件落后时失效而非无限缓存 | 协议只承载当前交互所需的小型控制请求、受 3 MiB 文本上限保护的 workspace diff 快照和既有事件。大 transcript 继续受 frame 上限约束；本阶段不为假设场景增加通用分页、二进制 side channel、压缩或批处理协议。 diff --git a/src/apps/cli/src/agent/runtime_client.rs b/src/apps/cli/src/agent/runtime_client.rs index 61ca352acd..f7d85aaf00 100644 --- a/src/apps/cli/src/agent/runtime_client.rs +++ b/src/apps/cli/src/agent/runtime_client.rs @@ -558,11 +558,13 @@ impl CliAgentRuntimeClient { &self, root_session_id: &str, session_id: &str, + required_settled_turn_ids: &[String], ) -> std::result::Result { let request = AgentSessionLineageTranscriptRequest { workspace_path: self.current_workspace_path().to_string_lossy().into_owned(), root_session_id: root_session_id.to_string(), session_id: session_id.to_string(), + required_settled_turn_ids: required_settled_turn_ids.to_vec(), remote_connection_id: None, remote_ssh_host: None, }; @@ -590,11 +592,13 @@ impl CliAgentRuntimeClient { &self, root_session_id: &str, session_id: &str, + expected_active_turn_id: &str, ) -> Result { let request = AgentSessionLineageCancellationRequest { workspace_path: self.current_workspace_path().to_string_lossy().into_owned(), root_session_id: root_session_id.to_string(), session_id: session_id.to_string(), + expected_active_turn_id: Some(expected_active_turn_id.to_string()), source: Some(AgentSubmissionSource::Cli), reason: Some("user_cancelled".to_string()), wait_timeout_ms: Some(5_000), diff --git a/src/apps/cli/src/dispatch/workspace.rs b/src/apps/cli/src/dispatch/workspace.rs index bad11e156a..bcefb1680e 100644 --- a/src/apps/cli/src/dispatch/workspace.rs +++ b/src/apps/cli/src/dispatch/workspace.rs @@ -751,6 +751,17 @@ fn bundle_commit_in_store( /// job's own branch, so a controller that never syncs leaves no trace here. pub(crate) fn sync(request: DispatchWorkspaceSyncRequest) -> Result { let store = DispatchStore::open_default()?; + start_sync_in_store(&store, request, super::runner::spawn_workspace_sync) +} + +fn start_sync_in_store( + store: &DispatchStore, + request: DispatchWorkspaceSyncRequest, + spawn_sync: F, +) -> Result +where + F: FnOnce(&str) -> Result, +{ super::store::validate_id("jobId", &request.job_id)?; super::store::validate_id("operationId", &request.operation_id)?; if request @@ -872,7 +883,7 @@ pub(crate) fn sync(request: DispatchWorkspaceSyncRequest) -> Result { operation.worker_pid = Some(pid); operation.updated_at = chrono::Utc::now().to_rfc3339(); @@ -1461,7 +1472,7 @@ fn path_arg(path: &Path) -> Result<&str> { } fn canonical_utf8(path: &Path) -> Result { - path.canonicalize() + dunce::canonicalize(path) .with_context(|| format!("resolve dispatch path {}", path.display()))? .to_str() .map(ToOwned::to_owned) @@ -2036,182 +2047,146 @@ mod tests { #[test] fn reported_sync_failure_allows_a_new_operation_to_take_over() { - const CHILD_ENV: &str = "BITFUN_DISPATCH_FAILED_SYNC_RETRY_CHILD"; - if let Some(bitfun_home) = std::env::var_os(CHILD_ENV) { - let store = DispatchStore::open_default().expect("open isolated default store"); - let source = PathBuf::from(bitfun_home).join("source"); - let base_commit = init_source_repository(&source); - provision_from_bundle(&store, &source, &base_commit); - let operation_path = store - .workspace_upload_dir("job-1") - .expect("workspace path") - .join(SYNC_OPERATION_FILE); - let failed_request = DispatchWorkspaceSyncRequest { - job_id: "job-1".to_string(), - operation_id: "sync-failed-generation".to_string(), - message: None, - known_head: Some(base_commit.clone()), - }; - atomic_write_json( - &operation_path, - &SyncOperationRecord { - request: failed_request.clone(), - state: WorkspaceOperationState::Failed, - worker_pid: None, - response: None, - last_error: Some("transient Git lock".to_string()), - failure_reported: false, - updated_at: chrono::Utc::now().to_rfc3339(), - }, - ) - .expect("seed failed operation"); - - let error = sync(failed_request).expect_err("the failed generation must be reported"); - assert!(error.to_string().contains("transient Git lock")); - let mut retained: SyncOperationRecord = - read_json(&operation_path).expect("read reported failure"); - assert_eq!(retained.state, WorkspaceOperationState::Failed); - assert!(retained.failure_reported); - assert_eq!(retained.last_error.as_deref(), Some("transient Git lock")); + let temp = tempfile::tempdir().expect("tempdir"); + let store = DispatchStore::open(temp.path().join("dispatch")).expect("store"); + let source = temp.path().join("source"); + let base_commit = init_source_repository(&source); + provision_from_bundle(&store, &source, &base_commit); + let operation_path = store + .workspace_upload_dir("job-1") + .expect("workspace path") + .join(SYNC_OPERATION_FILE); + let failed_request = DispatchWorkspaceSyncRequest { + job_id: "job-1".to_string(), + operation_id: "sync-failed-generation".to_string(), + message: None, + known_head: Some(base_commit.clone()), + }; + atomic_write_json( + &operation_path, + &SyncOperationRecord { + request: failed_request.clone(), + state: WorkspaceOperationState::Failed, + worker_pid: None, + response: None, + last_error: Some("transient Git lock".to_string()), + failure_reported: false, + updated_at: chrono::Utc::now().to_rfc3339(), + }, + ) + .expect("seed failed operation"); - let retry = DispatchWorkspaceSyncRequest { - job_id: "job-1".to_string(), - operation_id: "sync-retry-generation".to_string(), - message: None, - known_head: Some(base_commit), - }; - retained.worker_pid = Some(std::process::id()); - retained.updated_at = chrono::Utc::now().to_rfc3339(); - atomic_write_json(&operation_path, &retained).expect("seed active worker marker"); - let error = - sync(retry.clone()).expect_err("an active generation must retain ownership"); - assert!(error - .to_string() - .contains("already bound to a different request")); - - retained.worker_pid = None; - atomic_write_json(&operation_path, &retained).expect("retire failed worker marker"); - let response = sync(retry.clone()).expect("retry with a new operation id"); - assert!(response.pending); - let replacement: SyncOperationRecord = - read_json(&operation_path).expect("read replacement operation"); - assert_eq!(replacement.request, retry); - assert!(matches!( - replacement.state, - WorkspaceOperationState::Pending | WorkspaceOperationState::Running - )); - assert!(!replacement.failure_reported); - return; - } + let error = start_sync_in_store(&store, failed_request, |_| { + panic!("reporting a retained failure must not spawn a worker") + }) + .expect_err("the failed generation must be reported"); + assert!(error.to_string().contains("transient Git lock")); + let mut retained: SyncOperationRecord = + read_json(&operation_path).expect("read reported failure"); + assert_eq!(retained.state, WorkspaceOperationState::Failed); + assert!(retained.failure_reported); + assert_eq!(retained.last_error.as_deref(), Some("transient Git lock")); + + let retry = DispatchWorkspaceSyncRequest { + job_id: "job-1".to_string(), + operation_id: "sync-retry-generation".to_string(), + message: None, + known_head: Some(base_commit), + }; + retained.worker_pid = Some(std::process::id()); + retained.updated_at = chrono::Utc::now().to_rfc3339(); + atomic_write_json(&operation_path, &retained).expect("seed active worker marker"); + let error = start_sync_in_store(&store, retry.clone(), |_| { + panic!("an active generation must not spawn a replacement worker") + }) + .expect_err("an active generation must retain ownership"); + assert!(error + .to_string() + .contains("already bound to a different request")); - let dir = tempfile::tempdir().expect("tempdir"); - let bitfun_home = dir.path().join("bitfun-home"); - let user_root = dir.path().join("user-root"); - let output = std::process::Command::new(std::env::current_exe().expect("test executable")) - .args([ - "--exact", - "dispatch::workspace::tests::reported_sync_failure_allows_a_new_operation_to_take_over", - "--nocapture", - ]) - .env(CHILD_ENV, &bitfun_home) - .env("BITFUN_HOME", &bitfun_home) - .env("BITFUN_USER_ROOT", &user_root) - .env("BITFUN_E2E_STORAGE_GUARD", "1") - .env_remove("BITFUN_E2E_HOME") - .env_remove("BITFUN_E2E_USER_ROOT") - .output() - .expect("run isolated failed-sync retry test"); - assert!( - output.status.success(), - "isolated child failed:\nstdout:\n{}\nstderr:\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); + retained.worker_pid = None; + atomic_write_json(&operation_path, &retained).expect("retire failed worker marker"); + let response = start_sync_in_store(&store, retry.clone(), |job_id| { + assert_eq!(job_id, "job-1"); + Ok(42) + }) + .expect("retry with a new operation id"); + assert!(response.pending); + let replacement: SyncOperationRecord = + read_json(&operation_path).expect("read replacement operation"); + assert_eq!(replacement.request, retry); + assert_eq!(replacement.worker_pid, Some(42)); + assert!(matches!( + replacement.state, + WorkspaceOperationState::Pending | WorkspaceOperationState::Running + )); + assert!(!replacement.failure_reported); } #[test] fn legacy_sync_failure_without_operation_id_is_reported_then_retryable() { - const CHILD_ENV: &str = "BITFUN_DISPATCH_LEGACY_SYNC_RETRY_CHILD"; - if let Some(bitfun_home) = std::env::var_os(CHILD_ENV) { - let store = DispatchStore::open_default().expect("open isolated default store"); - let source = PathBuf::from(bitfun_home).join("source"); - let base_commit = init_source_repository(&source); - provision_from_bundle(&store, &source, &base_commit); - let operation_path = store - .workspace_upload_dir("job-1") - .expect("workspace path") - .join(SYNC_OPERATION_FILE); - // Early protocol-v3 development builds wrote neither operationId - // nor failureReported. Keep that exact JSON shape readable. - atomic_write_json( - &operation_path, - &serde_json::json!({ - "request": { - "jobId": "job-1", - "message": null, - "knownHead": base_commit, - }, - "state": "failed", - "workerPid": null, - "response": null, - "lastError": "legacy sync failure", - "updatedAt": chrono::Utc::now().to_rfc3339(), - }), - ) - .expect("seed legacy operation"); - - let first_retry = DispatchWorkspaceSyncRequest { - job_id: "job-1".to_string(), - operation_id: "sync-after-upgrade-1".to_string(), - message: None, - known_head: Some(base_commit.clone()), - }; - let error = sync(first_retry).expect_err("legacy failure must be surfaced once"); - assert!(error.to_string().contains("legacy sync failure")); - let reported: SyncOperationRecord = - read_json(&operation_path).expect("read upgraded legacy operation"); - assert!(reported.request.operation_id.is_empty()); - assert_eq!(reported.state, WorkspaceOperationState::Failed); - assert!(reported.failure_reported); - - let second_retry = DispatchWorkspaceSyncRequest { - job_id: "job-1".to_string(), - operation_id: "sync-after-upgrade-2".to_string(), - message: None, - known_head: Some(base_commit), - }; - let response = sync(second_retry.clone()).expect("replace legacy generation"); - assert!(response.pending); - let replacement: SyncOperationRecord = - read_json(&operation_path).expect("read replacement operation"); - assert_eq!(replacement.request, second_retry); - assert!(!replacement.failure_reported); - return; - } + let temp = tempfile::tempdir().expect("tempdir"); + let store = DispatchStore::open(temp.path().join("dispatch")).expect("store"); + let source = temp.path().join("source"); + let base_commit = init_source_repository(&source); + provision_from_bundle(&store, &source, &base_commit); + let operation_path = store + .workspace_upload_dir("job-1") + .expect("workspace path") + .join(SYNC_OPERATION_FILE); + // Early protocol-v3 development builds wrote neither operationId + // nor failureReported. Keep that exact JSON shape readable. + atomic_write_json( + &operation_path, + &serde_json::json!({ + "request": { + "jobId": "job-1", + "message": null, + "knownHead": base_commit, + }, + "state": "failed", + "workerPid": null, + "response": null, + "lastError": "legacy sync failure", + "updatedAt": chrono::Utc::now().to_rfc3339(), + }), + ) + .expect("seed legacy operation"); - let dir = tempfile::tempdir().expect("tempdir"); - let bitfun_home = dir.path().join("bitfun-home"); - let user_root = dir.path().join("user-root"); - let output = std::process::Command::new(std::env::current_exe().expect("test executable")) - .args([ - "--exact", - "dispatch::workspace::tests::legacy_sync_failure_without_operation_id_is_reported_then_retryable", - "--nocapture", - ]) - .env(CHILD_ENV, &bitfun_home) - .env("BITFUN_HOME", &bitfun_home) - .env("BITFUN_USER_ROOT", &user_root) - .env("BITFUN_E2E_STORAGE_GUARD", "1") - .env_remove("BITFUN_E2E_HOME") - .env_remove("BITFUN_E2E_USER_ROOT") - .output() - .expect("run isolated legacy-sync retry test"); - assert!( - output.status.success(), - "isolated child failed:\nstdout:\n{}\nstderr:\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); + let first_retry = DispatchWorkspaceSyncRequest { + job_id: "job-1".to_string(), + operation_id: "sync-after-upgrade-1".to_string(), + message: None, + known_head: Some(base_commit.clone()), + }; + let error = start_sync_in_store(&store, first_retry, |_| { + panic!("reporting a legacy failure must not spawn a worker") + }) + .expect_err("legacy failure must be surfaced once"); + assert!(error.to_string().contains("legacy sync failure")); + let reported: SyncOperationRecord = + read_json(&operation_path).expect("read upgraded legacy operation"); + assert!(reported.request.operation_id.is_empty()); + assert_eq!(reported.state, WorkspaceOperationState::Failed); + assert!(reported.failure_reported); + + let second_retry = DispatchWorkspaceSyncRequest { + job_id: "job-1".to_string(), + operation_id: "sync-after-upgrade-2".to_string(), + message: None, + known_head: Some(base_commit), + }; + let response = start_sync_in_store(&store, second_retry.clone(), |job_id| { + assert_eq!(job_id, "job-1"); + Ok(43) + }) + .expect("replace legacy generation"); + assert!(response.pending); + let replacement: SyncOperationRecord = + read_json(&operation_path).expect("read replacement operation"); + assert_eq!(replacement.request, second_retry); + assert_eq!(replacement.worker_pid, Some(43)); + assert!(!replacement.failure_reported); } #[test] diff --git a/src/apps/cli/src/modes/chat.rs b/src/apps/cli/src/modes/chat.rs index 46c43f6f9b..99df6fed81 100644 --- a/src/apps/cli/src/modes/chat.rs +++ b/src/apps/cli/src/modes/chat.rs @@ -23,7 +23,7 @@ use tokio::sync::broadcast::error::TryRecvError; use bitfun_agent_runtime::sdk::{ AgentLocalCommandTurnRecordRequest, AgentSessionComposerUpdate, AgentSessionLineageEntry, AgentSessionLineageInspection, AgentSessionLineageSnapshot, AgentSessionUsageRequest, - SessionTranscript, SessionUsageReport, + AgentTurnCancellationResult, SessionTranscript, SessionUsageReport, }; use bitfun_events::{AgenticEvent, ToolEventData, ToolEventIdentity}; use resize::ResizeRedrawState; @@ -367,6 +367,64 @@ struct PendingWorkspaceDiff { >, } +enum LineageInspectionTaskError { + Runtime(SessionOperationError), + Deadline, +} + +impl LineageInspectionTaskError { + fn outcome_unknown(&self) -> bool { + matches!(self, Self::Runtime(error) if error.outcome_unknown()) + } +} + +impl std::fmt::Display for LineageInspectionTaskError { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Runtime(error) => error.fmt(formatter), + Self::Deadline => formatter.write_str("the transcript settlement deadline elapsed"), + } + } +} + +enum PendingLineageOperation { + Query { + root_session_id: String, + handle: tokio::task::JoinHandle>>, + }, + Inspect { + entry: AgentSessionLineageEntry, + refresh: bool, + event_generation: u64, + handle: tokio::task::JoinHandle< + std::result::Result, + >, + }, +} + +impl PendingLineageOperation { + fn is_finished(&self) -> bool { + match self { + Self::Query { handle, .. } => handle.is_finished(), + Self::Inspect { handle, .. } => handle.is_finished(), + } + } + + fn abort(&self) { + match self { + Self::Query { handle, .. } => handle.abort(), + Self::Inspect { handle, .. } => handle.abort(), + } + } +} + +struct PendingLineageCancellation { + root_session_id: String, + session_id: String, + navigation_generation: u64, + handle: tokio::task::JoinHandle>, +} + #[derive(Debug, Clone)] struct ExternalPromptCommandInvocation { command_name: String, @@ -423,23 +481,12 @@ struct ChatEventContext<'a> { struct AgentSessionInspection { selected_session_id: String, chat_state: ChatState, - /// Terminal events invalidate this read model. Live chunks are projected - /// directly; Runtime transcript reads reconcile the settled turn. + /// Runtime events invalidate this read model. Live chunks are projected + /// directly; Runtime transcript reads replace it authoritatively. refresh_pending: bool, refresh_due_at: Instant, refresh_deadline: Option, - /// A terminal event precedes persistence. An otherwise successful Runtime - /// read is stale while it still reports one of these Turns as active. - settling_turn_ids: BTreeSet, - /// Failed/cancelled persistence omits partial model output, so settlement - /// confirmation must retain the already projected terminal view. - preserve_live_terminal: bool, -} - -#[derive(Clone)] -struct LineageSettlement { - turn_id: String, - preserve_live_terminal: bool, + refresh_retry_delay: Duration, } struct BufferedLineageEvent { @@ -449,7 +496,10 @@ struct BufferedLineageEvent { const LINEAGE_EVENT_BUFFER_MAX_BYTES: usize = 1024 * 1024; const LINEAGE_EVENT_BUFFER_MAX_EVENTS: usize = 4096; +const LINEAGE_READ_BARRIER_MAX_TURNS_PER_SESSION: usize = 256; const LINEAGE_SETTLEMENT_RETRY_WINDOW: Duration = Duration::from_secs(5); +const LINEAGE_SETTLEMENT_RETRY_MIN: Duration = Duration::from_millis(250); +const LINEAGE_SETTLEMENT_RETRY_MAX: Duration = Duration::from_secs(1); pub(crate) struct ChatMode { config: CliConfig, @@ -479,17 +529,31 @@ pub(crate) struct ChatMode { pending_workspace_diff: Option, pending_local_effect: Option, pending_workspace_reference_search: Option, + /// One lineage read in flight. Runtime I/O never blocks the TUI + /// event loop, and refresh requests cannot overlap. + pending_lineage_operation: Option, + /// Side-effecting cancellation outlives navigation resets, but its result + /// is only surfaced to the lineage generation that initiated it. + pending_lineage_cancellation: Option, /// Last authoritative flat lineage read. This is a presentation cache only; /// Services remains the membership/order owner. lineage_snapshot: Option, + /// Presentation-only index for the immutable ordering owned by Services. + lineage_session_index: HashMap, lineage_inspection: Option, /// Bounded presentation tail for active descendants. It bridges live /// broadcast output until the Runtime-owned transcript settles. lineage_event_buffer: VecDeque, lineage_event_buffer_bytes: usize, - /// At most one Turn can settle per Session before the next Turn starts. - /// Keeping this outside the selected inspection closes selector-time races. - lineage_settlements: BTreeMap, + /// Advances whenever a lineage event can make an in-flight transcript read + /// stale. Async inspection results are applied only to the generation they + /// observed, so a late read cannot erase newer live projection state. + lineage_event_generations: HashMap, + lineage_navigation_generation: u64, + /// Exact terminal Turns observed from Runtime events but not yet reflected + /// by an authoritative inspection. These are read-consistency tokens, not + /// a second settlement state machine. + lineage_required_settled_turns: BTreeMap>, workspace_reference_search_generation: u64, last_workspace_reference_query: Option, /// One explicit native slash-menu choice waiting for its parameterized submission. @@ -555,11 +619,16 @@ impl ChatMode { pending_workspace_diff: None, pending_local_effect: None, pending_workspace_reference_search: None, + pending_lineage_operation: None, + pending_lineage_cancellation: None, lineage_snapshot: None, + lineage_session_index: HashMap::new(), lineage_inspection: None, lineage_event_buffer: VecDeque::new(), lineage_event_buffer_bytes: 0, - lineage_settlements: BTreeMap::new(), + lineage_event_generations: HashMap::new(), + lineage_navigation_generation: 0, + lineage_required_settled_turns: BTreeMap::new(), workspace_reference_search_generation: 0, last_workspace_reference_query: None, selected_native_command_once: None, diff --git a/src/apps/cli/src/modes/chat/input.rs b/src/apps/cli/src/modes/chat/input.rs index 0d8bc6b3db..f34acd6f7f 100644 --- a/src/apps/cli/src/modes/chat/input.rs +++ b/src/apps/cli/src/modes/chat/input.rs @@ -524,6 +524,10 @@ impl ChatMode { } } + if key.code == KeyCode::Esc && self.cancel_pending_lineage_load(chat_view) { + return Ok(None); + } + if self.lineage_inspection.is_some() { match key.code { KeyCode::Up => self.navigate_lineage_parent(chat_view, rt_handle), @@ -786,11 +790,9 @@ impl ChatMode { } else { match mouse.kind { MouseEventKind::ScrollUp => { - let total = context - .chat_view - .count_message_lines( - context.this.displayed_chat_state(context.chat_state), - ); + let total = context.chat_view.count_message_lines( + context.this.displayed_chat_state(context.chat_state), + ); context.chat_view.scroll_up(3, total); } MouseEventKind::ScrollDown => { diff --git a/src/apps/cli/src/modes/chat/run.rs b/src/apps/cli/src/modes/chat/run.rs index 3ff5701078..953fb2f486 100644 --- a/src/apps/cli/src/modes/chat/run.rs +++ b/src/apps/cli/src/modes/chat/run.rs @@ -571,6 +571,10 @@ impl ChatMode { if self.poll_workspace_diff(&mut chat_view) { needs_redraw = true; } + if self.poll_lineage_operation_completion(&mut chat_view, &rt_handle) { + chat_view.invalidate_lines_cache(); + needs_redraw = true; + } chat_view.set_action_state( self.action_state(self.displayed_chat_state(&chat_state).is_processing, false), &self.keymap, diff --git a/src/apps/cli/src/modes/chat/session_lineage.rs b/src/apps/cli/src/modes/chat/session_lineage.rs index b4229cd615..d138ce2fee 100644 --- a/src/apps/cli/src/modes/chat/session_lineage.rs +++ b/src/apps/cli/src/modes/chat/session_lineage.rs @@ -10,35 +10,29 @@ impl ChatMode { let Some(event_session_id) = event.session_id() else { return false; }; - let Some(lineage_active_turn_id) = self.lineage_snapshot.as_ref().and_then(|snapshot| { - snapshot - .sessions - .iter() - .find(|entry| entry.session_id == event_session_id) - .map(|entry| entry.active_turn_id.clone()) - }) else { + let Some(entry_index) = self.lineage_session_index.get(event_session_id).copied() else { + return false; + }; + if is_buffered_lineage_event(event) + || matches!(event, AgenticEvent::SessionHistoryChanged { .. }) + { + let generation = self + .lineage_event_generations + .entry(event_session_id.to_string()) + .or_default(); + *generation = generation.wrapping_add(1); + } + let Some(lineage_active_turn_id) = self + .lineage_snapshot + .as_ref() + .and_then(|snapshot| snapshot.sessions.get(entry_index)) + .map(|entry| entry.active_turn_id.clone()) + else { return false; }; match event { AgenticEvent::DialogTurnStarted { turn_id, .. } => { - if lineage_terminal_reconciliation_pending( - self.lineage_inspection.as_ref(), - event_session_id, - ) { - if let Some(inspection) = self.lineage_inspection.as_mut() { - let now = Instant::now(); - inspection.refresh_pending = true; - inspection.refresh_due_at = now; - inspection.refresh_deadline = Some(now + LINEAGE_SETTLEMENT_RETRY_WINDOW); - } - } else { - self.lineage_settlements.remove(event_session_id); - self.retain_active_lineage_events(event_session_id, Some(turn_id)); - clear_lineage_settlement_for_new_turn( - self.lineage_inspection.as_mut(), - event_session_id, - ); - } + self.retain_lineage_events(event_session_id, Some(turn_id)); } AgenticEvent::DialogTurnCompleted { .. } | AgenticEvent::DialogTurnFailed { .. } @@ -55,27 +49,21 @@ impl ChatMode { inspection.selected_session_id == event_session_id && inspection.chat_state.current_turn_id() == Some(turn_id) }); - let Some(settlement) = lineage_settlement_from_event( - self.lineage_settlements.get(event_session_id), - observed_turn, - event, - ) else { - return false; - }; - self.lineage_settlements - .insert(event_session_id.to_string(), settlement); + if observed_turn { + record_required_settled_lineage_turn( + &mut self.lineage_required_settled_turns, + event_session_id, + turn_id, + ); + } } _ => {} } - let should_buffer = self.lineage_snapshot.as_ref().is_some_and(|snapshot| { - snapshot.sessions.iter().any(|entry| { - entry.session_id == event_session_id - && event.turn_id().is_some_and(|turn_id| { - matches!(event, AgenticEvent::DialogTurnStarted { .. }) - || entry.active_turn_id.as_deref() == Some(turn_id) - }) - }) && is_buffered_lineage_event(event) - }); + let should_buffer = matches!(event, AgenticEvent::SystemError { .. }) + || (event.turn_id().is_some_and(|turn_id| { + matches!(event, AgenticEvent::DialogTurnStarted { .. }) + || lineage_active_turn_id.as_deref() == Some(turn_id) + }) && is_buffered_lineage_event(event)); if should_buffer { push_bounded_lineage_event( &mut self.lineage_event_buffer, @@ -86,7 +74,7 @@ impl ChatMode { ); } if let Some(snapshot) = self.lineage_snapshot.as_mut() { - update_lineage_active_turn(snapshot, event); + update_lineage_active_turn(snapshot, &self.lineage_session_index, event); } let Some(inspection) = self .lineage_inspection @@ -107,24 +95,9 @@ impl ChatMode { if requires_authoritative_refresh { let now = Instant::now(); inspection.refresh_pending = true; - inspection.refresh_due_at = now; + inspection.refresh_due_at = now + LINEAGE_SETTLEMENT_RETRY_MIN; inspection.refresh_deadline = Some(now + LINEAGE_SETTLEMENT_RETRY_WINDOW); - if matches!( - event, - AgenticEvent::DialogTurnCompleted { .. } - | AgenticEvent::DialogTurnFailed { .. } - | AgenticEvent::DialogTurnCancelled { .. } - ) { - if let Some(turn_id) = event.turn_id() { - inspection.settling_turn_ids.insert(turn_id.to_string()); - } - } - if matches!( - event, - AgenticEvent::DialogTurnFailed { .. } | AgenticEvent::DialogTurnCancelled { .. } - ) { - inspection.preserve_live_terminal = true; - } + inspection.refresh_retry_delay = LINEAGE_SETTLEMENT_RETRY_MIN; } projection.changed || requires_authoritative_refresh } @@ -135,6 +108,12 @@ impl ChatMode { root_chat_state: &ChatState, rt_handle: &tokio::runtime::Handle, ) { + if self.pending_lineage_operation.is_some() { + chat_view.set_status(Some( + "A subagent Session operation is already in progress".to_string(), + )); + return; + } let root_session_id = self .lineage_snapshot .as_ref() @@ -142,27 +121,14 @@ impl ChatMode { .unwrap_or(&root_chat_state.core_session_id) .to_string(); let agent = self.agent.clone(); - let result = tokio::task::block_in_place(|| { - rt_handle.block_on(agent.session_lineage(&root_session_id)) + let task_root_session_id = root_session_id.clone(); + let handle = + rt_handle.spawn(async move { agent.session_lineage(&task_root_session_id).await }); + self.pending_lineage_operation = Some(PendingLineageOperation::Query { + root_session_id, + handle, }); - match result { - Ok(Some(snapshot)) - if snapshot - .sessions - .iter() - .any(|entry| entry.session_id != snapshot.root_session_id) => - { - chat_view.show_session_lineage_selector(&snapshot); - chat_view.set_status(Some( - "Select a subagent Session to inspect its transcript".to_string(), - )); - self.lineage_snapshot = Some(snapshot); - } - Ok(_) => chat_view.set_status(Some( - "No subagent Sessions are available for this conversation".to_string(), - )), - Err(error) => chat_view.set_status(Some(format!("Could not load subagents: {error}"))), - } + chat_view.set_status(Some("Loading subagent Sessions...".to_string())); } fn inspect_lineage_session( @@ -171,14 +137,28 @@ impl ChatMode { chat_view: &mut ChatView, rt_handle: &tokio::runtime::Handle, ) { + if matches!( + self.pending_lineage_operation, + Some(PendingLineageOperation::Inspect { refresh: true, .. }) + ) { + if let Some(pending) = self.pending_lineage_operation.take() { + pending.abort(); + } + } + if self.pending_lineage_operation.is_some() { + chat_view.set_status(Some( + "A subagent Session operation is already in progress".to_string(), + )); + return; + } let Some(snapshot) = self.lineage_snapshot.as_ref() else { chat_view.set_status(Some("Reopen View subagents and try again".to_string())); return; }; - let Some(entry) = snapshot - .sessions - .iter() - .find(|entry| entry.session_id == session_id) + let Some(entry) = self + .lineage_session_index + .get(session_id) + .and_then(|index| snapshot.sessions.get(*index)) .cloned() else { chat_view.set_status(Some( @@ -191,173 +171,31 @@ impl ChatMode { return; } let root_session_id = snapshot.root_session_id.clone(); + let required_settled_turn_ids = self + .lineage_required_settled_turns + .get(&entry.session_id) + .cloned() + .unwrap_or_default(); let agent = self.agent.clone(); - let result = tokio::task::block_in_place(|| { - rt_handle.block_on(agent.inspect_lineage_session(&root_session_id, session_id)) + let task_session_id = entry.session_id.clone(); + let event_generation = self.lineage_event_generation(&task_session_id); + let handle = rt_handle.spawn(async move { + agent + .inspect_lineage_session( + &root_session_id, + &task_session_id, + &required_settled_turn_ids, + ) + .await + .map_err(LineageInspectionTaskError::Runtime) }); - match result { - Ok(inspection) => { - let active_turn_id = inspection.active_turn_id.clone(); - let settlement = self.lineage_settlements.get(&entry.session_id).cloned(); - let stale = settlement.as_ref().is_some_and(|settlement| { - active_turn_id.as_deref() == Some(settlement.turn_id.as_str()) - }); - let candidate_settlement = settlement - .as_ref() - .filter(|settlement| stale || settlement.preserve_live_terminal); - let replay_turn_id = candidate_settlement - .map(|settlement| settlement.turn_id.as_str()) - .or(active_turn_id.as_deref()); - let replay_events = replay_turn_id - .map(|turn_id| self.buffered_lineage_events(&entry.session_id, turn_id)) - .unwrap_or_default(); - let display_settlement = candidate_settlement.filter(|settlement| { - replay_events - .iter() - .any(|event| is_terminal_lineage_event(event, &settlement.turn_id)) - }); - let preserve_cached_terminal = - display_settlement.is_some_and(|settlement| settlement.preserve_live_terminal); - let runtime_advanced_past_settlement = - display_settlement.is_some_and(|settlement| { - active_turn_id - .as_deref() - .is_some_and(|active_turn_id| active_turn_id != settlement.turn_id) - }); - let state = if runtime_advanced_past_settlement { - let runtime_replay_events = active_turn_id - .as_deref() - .map(|turn_id| self.buffered_lineage_events(&entry.session_id, turn_id)) - .unwrap_or_default(); - let mut runtime_state = build_lineage_chat_state( - &entry, - inspection.clone(), - &runtime_replay_events, - ); - let settlement = display_settlement.expect("settlement checked above"); - let mut replay_inspection = inspection; - replay_inspection.active_turn_id = Some(settlement.turn_id.clone()); - let replayed = - build_lineage_chat_state(&entry, replay_inspection, &replay_events); - merge_live_terminal_into_authoritative( - &mut runtime_state, - &replayed, - None, - &settlement.turn_id, - ); - runtime_state - } else { - let mut display_inspection = inspection; - if let Some(settlement) = display_settlement { - display_inspection.active_turn_id = Some(settlement.turn_id.clone()); - } - build_lineage_chat_state(&entry, display_inspection, &replay_events) - }; - if let Some(snapshot) = self.lineage_snapshot.as_mut() { - if let Some(snapshot_entry) = snapshot - .sessions - .iter_mut() - .find(|candidate| candidate.session_id == entry.session_id) - { - snapshot_entry.active_turn_id = - if stale { None } else { active_turn_id.clone() }; - } - } - let now = Instant::now(); - self.lineage_inspection = Some(AgentSessionInspection { - selected_session_id: entry.session_id.clone(), - chat_state: state, - refresh_pending: stale, - refresh_due_at: if stale { - now + Duration::from_millis(250) - } else { - now - }, - refresh_deadline: stale.then_some(now + LINEAGE_SETTLEMENT_RETRY_WINDOW), - settling_turn_ids: settlement - .as_ref() - .filter(|_| stale) - .map(|settlement| BTreeSet::from([settlement.turn_id.clone()])) - .unwrap_or_default(), - preserve_live_terminal: settlement - .as_ref() - .is_some_and(|_| stale && preserve_cached_terminal), - }); - if runtime_advanced_past_settlement { - self.lineage_settlements.remove(&entry.session_id); - self.retain_active_lineage_events(&entry.session_id, active_turn_id.as_deref()); - } else if !stale && !preserve_cached_terminal { - self.lineage_settlements.remove(&entry.session_id); - self.retain_active_lineage_events(&entry.session_id, active_turn_id.as_deref()); - } else if !stale { - self.retain_active_lineage_events( - &entry.session_id, - settlement.as_ref().map(|value| value.turn_id.as_str()), - ); - } - chat_view.set_lineage_inspection(Some(entry.session_name)); - chat_view.invalidate_lines_cache(); - chat_view.set_status(Some( - "Read-only subagent transcript; root input remains preserved".to_string(), - )); - } - Err(error) if error.outcome_unknown() => { - let now = Instant::now(); - let settlement = self.lineage_settlements.get(&entry.session_id).cloned(); - let settlement_turn_id = settlement - .as_ref() - .map(|settlement| settlement.turn_id.as_str()); - let settlement_events = settlement_turn_id - .map(|turn_id| self.buffered_lineage_events(&entry.session_id, turn_id)) - .unwrap_or_default(); - let has_terminal_event = settlement.as_ref().is_some_and(|settlement| { - settlement_events - .iter() - .any(|event| is_terminal_lineage_event(event, &settlement.turn_id)) - }); - let replay_turn_id = provisional_lineage_replay_turn( - settlement_turn_id, - has_terminal_event, - entry.active_turn_id.as_deref(), - ); - let replay_events = replay_turn_id - .map(|turn_id| self.buffered_lineage_events(&entry.session_id, turn_id)) - .unwrap_or_default(); - let state = build_lineage_chat_state( - &entry, - AgentSessionLineageInspection { - transcript: SessionTranscript { - session_id: entry.session_id.clone(), - messages: Vec::new(), - }, - active_turn_id: replay_turn_id.map(str::to_string), - }, - &replay_events, - ); - self.lineage_inspection = Some(AgentSessionInspection { - selected_session_id: entry.session_id.clone(), - chat_state: state, - refresh_pending: true, - refresh_due_at: now + Duration::from_millis(250), - refresh_deadline: Some(now + LINEAGE_SETTLEMENT_RETRY_WINDOW), - settling_turn_ids: settlement - .as_ref() - .map(|settlement| BTreeSet::from([settlement.turn_id.clone()])) - .unwrap_or_default(), - preserve_live_terminal: settlement.as_ref().is_some_and(|settlement| { - has_terminal_event && settlement.preserve_live_terminal - }), - }); - chat_view.set_lineage_inspection(Some(entry.session_name)); - chat_view.invalidate_lines_cache(); - chat_view.set_status(Some( - "Waiting for the subagent transcript to settle".to_string(), - )); - } - Err(error) => { - chat_view.set_status(Some(format!("Could not inspect subagent Session: {error}"))) - } - } + self.pending_lineage_operation = Some(PendingLineageOperation::Inspect { + entry, + refresh: false, + event_generation, + handle, + }); + chat_view.set_status(Some("Loading subagent transcript...".to_string())); } fn refresh_inspected_lineage_if_due( @@ -365,6 +203,9 @@ impl ChatMode { chat_view: &mut ChatView, rt_handle: &tokio::runtime::Handle, ) -> bool { + if self.pending_lineage_operation.is_some() { + return false; + } let now = Instant::now(); let Some(selected_session_id) = self.lineage_inspection.as_ref().and_then(|inspection| { (inspection.refresh_pending && inspection.refresh_due_at <= now) @@ -376,10 +217,10 @@ impl ChatMode { return false; }; let root_session_id = snapshot.root_session_id.clone(); - let Some(entry) = snapshot - .sessions - .iter() - .find(|entry| entry.session_id == selected_session_id) + let Some(entry) = self + .lineage_session_index + .get(&selected_session_id) + .and_then(|index| snapshot.sessions.get(*index)) .cloned() else { if let Some(inspection) = self.lineage_inspection.as_mut() { @@ -390,148 +231,409 @@ impl ChatMode { )); return true; }; - + let Some(deadline) = self + .lineage_inspection + .as_ref() + .and_then(|inspection| inspection.refresh_deadline) + else { + return false; + }; + if now >= deadline { + if let Some(inspection) = self.lineage_inspection.as_mut() { + inspection.refresh_pending = false; + inspection.refresh_deadline = None; + } + chat_view.set_status(Some( + "The subagent transcript is still settling; reopen View subagents to retry" + .to_string(), + )); + return true; + } + let required_settled_turn_ids = self + .lineage_required_settled_turns + .get(&selected_session_id) + .cloned() + .unwrap_or_default(); let agent = self.agent.clone(); - let result = tokio::task::block_in_place(|| { - rt_handle - .block_on(agent.inspect_lineage_session(&root_session_id, &selected_session_id)) + let timeout = deadline.saturating_duration_since(now); + let task_session_id = selected_session_id.clone(); + let handle = rt_handle.spawn(async move { + match tokio::time::timeout( + timeout, + agent.inspect_lineage_session( + &root_session_id, + &task_session_id, + &required_settled_turn_ids, + ), + ) + .await + { + Ok(result) => result.map_err(LineageInspectionTaskError::Runtime), + Err(_) => Err(LineageInspectionTaskError::Deadline), + } }); - match result { - Ok(inspection) => { - let active_turn_id = inspection.active_turn_id.clone(); - let refresh_action = self - .lineage_inspection - .as_ref() - .map(|current| { - lineage_refresh_action( - ¤t.settling_turn_ids, - active_turn_id.as_deref(), - current.preserve_live_terminal, - ) - }) - .unwrap_or(LineageRefreshAction::ReplaceFromRuntime); - if refresh_action == LineageRefreshAction::RetrySettlement { - if let Some(current) = self.lineage_inspection.as_mut() { - if current - .refresh_deadline - .is_some_and(|deadline| Instant::now() < deadline) - { - current.refresh_pending = true; - current.refresh_due_at = Instant::now() + Duration::from_millis(250); - return false; - } - current.refresh_pending = false; - current.refresh_deadline = None; - } + self.pending_lineage_operation = Some(PendingLineageOperation::Inspect { + entry, + refresh: true, + event_generation: self.lineage_event_generation(&selected_session_id), + handle, + }); + false + } + + fn poll_lineage_operation_completion( + &mut self, + chat_view: &mut ChatView, + rt_handle: &tokio::runtime::Handle, + ) -> bool { + let cancellation_changed = self.poll_lineage_cancellation_completion(chat_view, rt_handle); + let Some(pending) = self.pending_lineage_operation.as_ref() else { + return cancellation_changed; + }; + if !pending.is_finished() { + return cancellation_changed; + } + let pending = self + .pending_lineage_operation + .take() + .expect("finished lineage operation should remain present"); + match pending { + PendingLineageOperation::Query { + root_session_id, + handle, + } => match tokio::task::block_in_place(|| rt_handle.block_on(handle)) { + Ok(Ok(Some(snapshot))) + if snapshot.root_session_id == root_session_id + && snapshot + .sessions + .iter() + .any(|entry| entry.session_id != snapshot.root_session_id) => + { + self.lineage_session_index = lineage_session_index(&snapshot); + chat_view.show_session_lineage_selector(&snapshot); chat_view.set_status(Some( - "The subagent transcript is still settling; reopen View subagents to retry" - .to_string(), + "Select a subagent Session to inspect its transcript".to_string(), )); - return true; + self.lineage_snapshot = Some(snapshot); } - let preserved_turn_id = (refresh_action - == LineageRefreshAction::PreserveLiveTerminal) - .then(|| { - self.lineage_settlements - .get(&selected_session_id) - .map(|settlement| settlement.turn_id.clone()) - .or_else(|| { - self.lineage_inspection.as_ref().and_then(|current| { - current.settling_turn_ids.iter().next().cloned() - }) - }) - }) - .flatten(); - let replay_events = active_turn_id - .as_deref() - .map(|turn_id| self.buffered_lineage_events(&selected_session_id, turn_id)) - .unwrap_or_default(); - let preserved_projection = preserved_turn_id.as_ref().map(|turn_id| { - let preserved_events = - self.buffered_lineage_events(&selected_session_id, turn_id); - let mut preserved_inspection = inspection.clone(); - preserved_inspection.active_turn_id = Some(turn_id.clone()); - build_lineage_chat_state(&entry, preserved_inspection, &preserved_events) - }); - let mut runtime_state = - build_lineage_chat_state(&entry, inspection, &replay_events); - if let Some(snapshot_entry) = self.lineage_snapshot.as_mut().and_then(|snapshot| { - snapshot - .sessions - .iter_mut() - .find(|candidate| candidate.session_id == selected_session_id) - }) { - snapshot_entry.active_turn_id = active_turn_id.clone(); + Ok(Ok(_)) => { + self.lineage_snapshot = None; + self.lineage_session_index.clear(); + chat_view.set_status(Some( + "No subagent Sessions are available for this conversation".to_string(), + )); } - if let Some(current) = self - .lineage_inspection - .as_mut() - .filter(|inspection| inspection.selected_session_id == selected_session_id) - { - if refresh_action == LineageRefreshAction::PreserveLiveTerminal { - if let Some(turn_id) = preserved_turn_id.as_deref() { - merge_live_terminal_into_authoritative( - &mut runtime_state, - ¤t.chat_state, - preserved_projection.as_ref(), - turn_id, - ); - } - } - current.chat_state = runtime_state; - current.refresh_pending = false; - current.refresh_due_at = now; - current.refresh_deadline = None; - current.settling_turn_ids.clear(); - current.preserve_live_terminal = false; + Ok(Err(error)) => { + chat_view.set_status(Some(format!("Could not load subagents: {error}"))) } - if refresh_action == LineageRefreshAction::PreserveLiveTerminal - && active_turn_id.is_none() - { - self.retain_active_lineage_events( - &selected_session_id, - preserved_turn_id.as_deref(), - ); - } else { - self.lineage_settlements.remove(&selected_session_id); - self.retain_active_lineage_events( - &selected_session_id, - active_turn_id.as_deref(), - ); + Err(error) => { + chat_view.set_status(Some(format!("Subagent Session loading stopped: {error}"))) } - chat_view.set_status(Some( - "Read-only subagent transcript; root input remains preserved".to_string(), - )); - true - } - Err(error) => { - let retryable = error.outcome_unknown(); - if let Some(current) = self - .lineage_inspection - .as_mut() - .filter(|inspection| inspection.selected_session_id == selected_session_id) - { - // Only the Runtime's typed settlement uncertainty is - // retryable. Permanent storage/workspace errors stop here. - if retryable - && current - .refresh_deadline - .is_some_and(|deadline| Instant::now() < deadline) - { - current.refresh_pending = true; - current.refresh_due_at = Instant::now() + Duration::from_millis(250); + }, + PendingLineageOperation::Inspect { + entry, + refresh, + event_generation, + handle, + } => match tokio::task::block_in_place(|| rt_handle.block_on(handle)) { + Ok(Ok(inspection)) => self.apply_lineage_inspection( + entry, + inspection, + refresh, + event_generation, + chat_view, + ), + Ok(Err(error)) => self.apply_lineage_inspection_error( + entry, + refresh, + event_generation, + error, + chat_view, + ), + Err(error) => { + if !lineage_inspection_result_is_current( + &entry.session_id, + event_generation, + &self.lineage_event_generations, + ) { + self.handle_stale_lineage_inspection(&entry, refresh, chat_view); } else { - current.refresh_pending = false; - current.refresh_deadline = None; + if let Some(current) = self.lineage_inspection.as_mut().filter(|current| { + refresh && current.selected_session_id == entry.session_id + }) { + current.refresh_pending = false; + current.refresh_deadline = None; + } + chat_view.set_status(Some(format!( + "Subagent transcript loading stopped: {error}" + ))); } } - chat_view.set_status(Some(format!("Could not refresh subagent Session: {error}"))); - true + }, + } + true + } + + fn poll_lineage_cancellation_completion( + &mut self, + chat_view: &mut ChatView, + rt_handle: &tokio::runtime::Handle, + ) -> bool { + let Some(pending) = self.pending_lineage_cancellation.as_ref() else { + return false; + }; + if !pending.handle.is_finished() { + return false; + } + let pending = self + .pending_lineage_cancellation + .take() + .expect("finished lineage cancellation should remain present"); + let result = tokio::task::block_in_place(|| rt_handle.block_on(pending.handle)); + let belongs_to_current_navigation = lineage_cancellation_result_is_current( + pending.navigation_generation, + self.lineage_navigation_generation, + &pending.root_session_id, + self.lineage_snapshot + .as_ref() + .map(|snapshot| snapshot.root_session_id.as_str()), + &pending.session_id, + self.lineage_inspection + .as_ref() + .map(|inspection| inspection.selected_session_id.as_str()), + ); + if !belongs_to_current_navigation { + return false; + } + match result { + Ok(Ok(result)) if result.requested => chat_view.set_status(Some(format!( + "Interrupt requested for subagent Session {}", + pending.session_id + ))), + Ok(Ok(_)) => chat_view.set_status(Some("The subagent has no active turn".to_string())), + Ok(Err(error)) => chat_view.set_status(Some(format!( + "Could not interrupt subagent Session: {error}" + ))), + Err(error) => { + chat_view.set_status(Some(format!("Subagent interruption stopped: {error}"))) + } + } + true + } + + fn apply_lineage_inspection( + &mut self, + entry: AgentSessionLineageEntry, + inspection: AgentSessionLineageInspection, + refresh: bool, + event_generation: u64, + chat_view: &mut ChatView, + ) { + if !lineage_inspection_result_is_current( + &entry.session_id, + event_generation, + &self.lineage_event_generations, + ) { + self.handle_stale_lineage_inspection(&entry, refresh, chat_view); + return; + } + if refresh + && self + .lineage_inspection + .as_ref() + .is_none_or(|current| current.selected_session_id != entry.session_id) + { + return; + } + let active_turn_id = inspection.active_turn_id.clone(); + let state = self.build_authoritative_lineage_state(&entry, inspection); + if let Some(index) = self.lineage_session_index.get(&entry.session_id).copied() { + if let Some(snapshot_entry) = self + .lineage_snapshot + .as_mut() + .and_then(|snapshot| snapshot.sessions.get_mut(index)) + { + snapshot_entry.active_turn_id = active_turn_id.clone(); + } + } + self.lineage_required_settled_turns + .remove(&entry.session_id); + let now = Instant::now(); + self.lineage_inspection = Some(AgentSessionInspection { + selected_session_id: entry.session_id.clone(), + chat_state: state, + refresh_pending: false, + refresh_due_at: now, + refresh_deadline: None, + refresh_retry_delay: LINEAGE_SETTLEMENT_RETRY_MIN, + }); + self.retain_lineage_events(&entry.session_id, active_turn_id.as_deref()); + chat_view.set_lineage_inspection(Some(entry.session_name)); + chat_view.invalidate_lines_cache(); + chat_view.set_status(Some( + "Read-only subagent transcript; root input remains preserved".to_string(), + )); + } + + fn apply_lineage_inspection_error( + &mut self, + entry: AgentSessionLineageEntry, + refresh: bool, + event_generation: u64, + error: LineageInspectionTaskError, + chat_view: &mut ChatView, + ) { + if !lineage_inspection_result_is_current( + &entry.session_id, + event_generation, + &self.lineage_event_generations, + ) { + self.handle_stale_lineage_inspection(&entry, refresh, chat_view); + return; + } + if error.outcome_unknown() { + if refresh { + self.schedule_lineage_consistency_retry(&entry.session_id, chat_view); + } else { + self.open_provisional_lineage_inspection(&entry, chat_view); + } + return; + } + if let Some(current) = self + .lineage_inspection + .as_mut() + .filter(|current| refresh && current.selected_session_id == entry.session_id) + { + current.refresh_pending = false; + current.refresh_deadline = None; + } + let operation = if refresh { "refresh" } else { "inspect" }; + chat_view.set_status(Some(format!( + "Could not {operation} subagent Session: {error}" + ))); + } + + fn handle_stale_lineage_inspection( + &mut self, + entry: &AgentSessionLineageEntry, + refresh: bool, + chat_view: &mut ChatView, + ) { + if !refresh { + self.open_provisional_lineage_inspection(entry, chat_view); + return; + } + let now = Instant::now(); + let Some(current) = self + .lineage_inspection + .as_mut() + .filter(|current| current.selected_session_id == entry.session_id) + else { + return; + }; + current.refresh_pending = true; + current.refresh_due_at = now + LINEAGE_SETTLEMENT_RETRY_MIN; + current + .refresh_deadline + .get_or_insert(now + LINEAGE_SETTLEMENT_RETRY_WINDOW); + chat_view.set_status(Some( + "Waiting for newer subagent transcript state".to_string(), + )); + } + + fn lineage_event_generation(&self, session_id: &str) -> u64 { + self.lineage_event_generations + .get(session_id) + .copied() + .unwrap_or_default() + } + + fn schedule_lineage_consistency_retry(&mut self, session_id: &str, chat_view: &mut ChatView) { + let now = Instant::now(); + let Some(current) = self + .lineage_inspection + .as_mut() + .filter(|current| current.selected_session_id == session_id) + else { + return; + }; + if current + .refresh_deadline + .is_some_and(|deadline| now < deadline) + { + let delay = current.refresh_retry_delay; + current.refresh_pending = true; + current.refresh_due_at = now + delay; + current.refresh_retry_delay = + std::cmp::min(delay.saturating_mul(2), LINEAGE_SETTLEMENT_RETRY_MAX); + return; + } + current.refresh_pending = false; + current.refresh_deadline = None; + chat_view.set_status(Some( + "The subagent transcript is still settling; reopen View subagents to retry".to_string(), + )); + } + + fn open_provisional_lineage_inspection( + &mut self, + entry: &AgentSessionLineageEntry, + chat_view: &mut ChatView, + ) { + let now = Instant::now(); + let events = self.buffered_lineage_session_events(&entry.session_id); + let state = build_provisional_lineage_chat_state(entry, &events); + self.lineage_inspection = Some(AgentSessionInspection { + selected_session_id: entry.session_id.clone(), + chat_state: state, + refresh_pending: true, + refresh_due_at: now + LINEAGE_SETTLEMENT_RETRY_MIN, + refresh_deadline: Some(now + LINEAGE_SETTLEMENT_RETRY_WINDOW), + refresh_retry_delay: LINEAGE_SETTLEMENT_RETRY_MIN, + }); + chat_view.set_lineage_inspection(Some(entry.session_name.clone())); + chat_view.invalidate_lines_cache(); + chat_view.set_status(Some( + "Waiting for the subagent transcript to settle".to_string(), + )); + } + + fn build_authoritative_lineage_state( + &self, + entry: &AgentSessionLineageEntry, + inspection: AgentSessionLineageInspection, + ) -> ChatState { + let active_turn_id = inspection.active_turn_id.clone(); + let mut active_events = Vec::new(); + let mut session_events = Vec::new(); + for buffered in &self.lineage_event_buffer { + if buffered.event.session_id() != Some(entry.session_id.as_str()) { + continue; } + let Some(turn_id) = buffered.event.turn_id() else { + session_events.push(buffered.event.clone()); + continue; + }; + if active_turn_id.as_deref() == Some(turn_id) { + active_events.push(buffered.event.clone()); + } + } + let mut authoritative = build_lineage_chat_state(entry, inspection, &active_events); + for event in &session_events { + project_transcript_event(&mut authoritative, event, false); } + authoritative } fn leave_lineage_inspection(&mut self, chat_view: &mut ChatView) { + if matches!( + self.pending_lineage_operation, + Some(PendingLineageOperation::Inspect { refresh: true, .. }) + ) { + if let Some(pending) = self.pending_lineage_operation.take() { + pending.abort(); + } + } if self.lineage_inspection.take().is_some() { chat_view.set_lineage_inspection(None); chat_view.invalidate_lines_cache(); @@ -539,21 +641,44 @@ impl ChatMode { } } - fn buffered_lineage_events(&self, session_id: &str, turn_id: &str) -> Vec { + fn cancel_pending_lineage_load(&mut self, chat_view: &mut ChatView) -> bool { + let should_cancel = matches!( + self.pending_lineage_operation, + Some(PendingLineageOperation::Query { .. }) + | Some(PendingLineageOperation::Inspect { refresh: false, .. }) + ); + if !should_cancel { + return false; + } + if let Some(pending) = self.pending_lineage_operation.take() { + pending.abort(); + } + chat_view.set_status(Some("Subagent Session loading cancelled".to_string())); + true + } + + fn buffered_lineage_session_events(&self, session_id: &str) -> Vec { self.lineage_event_buffer .iter() - .filter(|buffered| { - buffered.event.session_id() == Some(session_id) - && buffered.event.turn_id() == Some(turn_id) - }) + .filter(|buffered| buffered.event.session_id() == Some(session_id)) .map(|buffered| buffered.event.clone()) .collect() } - fn retain_active_lineage_events(&mut self, session_id: &str, active_turn_id: Option<&str>) { + fn retain_lineage_events(&mut self, session_id: &str, active_turn_id: Option<&str>) { + let retained_turn_ids = self + .lineage_required_settled_turns + .get(session_id) + .into_iter() + .flatten() + .map(String::as_str) + .collect::>(); self.lineage_event_buffer.retain(|buffered| { buffered.event.session_id() != Some(session_id) - || active_turn_id.is_some_and(|turn_id| buffered.event.turn_id() == Some(turn_id)) + || buffered.event.turn_id().is_none() + || buffered.event.turn_id().is_some_and(|turn_id| { + active_turn_id == Some(turn_id) || retained_turn_ids.contains(turn_id) + }) }); self.lineage_event_buffer_bytes = self .lineage_event_buffer @@ -563,11 +688,23 @@ impl ChatMode { } fn reset_lineage_navigation(&mut self, chat_view: &mut ChatView) { + if matches!( + self.pending_lineage_operation, + Some(PendingLineageOperation::Query { .. }) + | Some(PendingLineageOperation::Inspect { .. }) + ) { + if let Some(pending) = self.pending_lineage_operation.take() { + pending.abort(); + } + } self.lineage_snapshot = None; + self.lineage_session_index.clear(); self.lineage_inspection = None; self.lineage_event_buffer.clear(); self.lineage_event_buffer_bytes = 0; - self.lineage_settlements.clear(); + self.lineage_event_generations.clear(); + self.lineage_navigation_generation = self.lineage_navigation_generation.wrapping_add(1); + self.lineage_required_settled_turns.clear(); chat_view.set_lineage_inspection(None); } @@ -626,6 +763,26 @@ impl ChatMode { chat_view: &mut ChatView, rt_handle: &tokio::runtime::Handle, ) { + if matches!( + self.pending_lineage_operation, + Some(PendingLineageOperation::Inspect { refresh: true, .. }) + ) { + if let Some(pending) = self.pending_lineage_operation.take() { + pending.abort(); + } + } + if self.pending_lineage_operation.is_some() { + chat_view.set_status(Some( + "A subagent Session operation is already in progress".to_string(), + )); + return; + } + if self.pending_lineage_cancellation.is_some() { + chat_view.set_status(Some( + "A subagent interruption is already in progress".to_string(), + )); + return; + } let (Some(snapshot), Some(inspection)) = ( self.lineage_snapshot.as_ref(), self.lineage_inspection.as_ref(), @@ -634,19 +791,34 @@ impl ChatMode { }; let agent = self.agent.clone(); let root_session_id = snapshot.root_session_id.clone(); + let task_root_session_id = root_session_id.clone(); let session_id = inspection.selected_session_id.clone(); - let result = tokio::task::block_in_place(|| { - rt_handle.block_on(agent.cancel_lineage_session(&root_session_id, &session_id)) + let Some(expected_active_turn_id) = self + .lineage_session_index + .get(&session_id) + .and_then(|index| snapshot.sessions.get(*index)) + .and_then(|entry| entry.active_turn_id.clone()) + else { + chat_view.set_status(Some("The subagent has no active turn".to_string())); + return; + }; + let task_session_id = session_id.clone(); + let handle = rt_handle.spawn(async move { + agent + .cancel_lineage_session( + &task_root_session_id, + &task_session_id, + &expected_active_turn_id, + ) + .await }); - match result { - Ok(result) if result.requested => chat_view.set_status(Some(format!( - "Interrupt requested for subagent Session {session_id}" - ))), - Ok(_) => chat_view.set_status(Some("The subagent has no active turn".to_string())), - Err(error) => chat_view.set_status(Some(format!( - "Could not interrupt subagent Session: {error}" - ))), - } + self.pending_lineage_cancellation = Some(PendingLineageCancellation { + root_session_id, + session_id, + navigation_generation: self.lineage_navigation_generation, + handle, + }); + chat_view.set_status(Some("Requesting subagent interruption...".to_string())); } } @@ -665,100 +837,48 @@ fn is_buffered_lineage_event(event: &AgenticEvent) -> bool { | AgenticEvent::ContextCompressionCompleted { .. } | AgenticEvent::ContextCompressionFailed { .. } | AgenticEvent::TokenUsageUpdated { .. } + | AgenticEvent::SystemError { .. } ) } -fn is_terminal_lineage_event(event: &AgenticEvent, turn_id: &str) -> bool { - matches!( - event, - AgenticEvent::DialogTurnCompleted { - turn_id: event_turn_id, - .. - } | AgenticEvent::DialogTurnFailed { - turn_id: event_turn_id, - .. - } | AgenticEvent::DialogTurnCancelled { - turn_id: event_turn_id, - .. - } if event_turn_id == turn_id - ) -} - -fn lineage_settlement_from_event( - existing: Option<&LineageSettlement>, - observed_turn: bool, - event: &AgenticEvent, -) -> Option { - let turn_id = event.turn_id()?; - if !observed_turn - || !is_terminal_lineage_event(event, turn_id) - || existing.is_some_and(|settlement| settlement.turn_id == turn_id) - { - return None; - } - Some(LineageSettlement { - turn_id: turn_id.to_string(), - preserve_live_terminal: matches!( - event, - AgenticEvent::DialogTurnFailed { .. } | AgenticEvent::DialogTurnCancelled { .. } - ), - }) -} - -fn clear_lineage_settlement_for_new_turn( - inspection: Option<&mut AgentSessionInspection>, +fn record_required_settled_lineage_turn( + required_turns: &mut BTreeMap>, session_id: &str, + turn_id: &str, ) { - let Some(inspection) = - inspection.filter(|inspection| inspection.selected_session_id == session_id) - else { + let session_turns = required_turns.entry(session_id.to_string()).or_default(); + if session_turns.iter().any(|required| required == turn_id) { return; - }; - inspection.settling_turn_ids.clear(); - inspection.preserve_live_terminal = false; + } + session_turns.push(turn_id.to_string()); + if session_turns.len() > LINEAGE_READ_BARRIER_MAX_TURNS_PER_SESSION { + session_turns.remove(0); + } } -fn lineage_terminal_reconciliation_pending( - inspection: Option<&AgentSessionInspection>, +fn lineage_inspection_result_is_current( session_id: &str, + request_generation: u64, + current_generations: &HashMap, ) -> bool { - inspection.is_some_and(|inspection| { - inspection.selected_session_id == session_id - && (!inspection.settling_turn_ids.is_empty() || inspection.preserve_live_terminal) - }) -} - -fn provisional_lineage_replay_turn<'a>( - settlement_turn_id: Option<&'a str>, - has_terminal_event: bool, - active_turn_id: Option<&'a str>, -) -> Option<&'a str> { - if has_terminal_event { - settlement_turn_id - } else { - active_turn_id - } + request_generation + == current_generations + .get(session_id) + .copied() + .unwrap_or_default() } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum LineageRefreshAction { - RetrySettlement, - PreserveLiveTerminal, - ReplaceFromRuntime, -} - -fn lineage_refresh_action( - settling_turn_ids: &BTreeSet, - active_turn_id: Option<&str>, - preserve_live_terminal: bool, -) -> LineageRefreshAction { - if active_turn_id.is_some_and(|turn_id| settling_turn_ids.contains(turn_id)) { - LineageRefreshAction::RetrySettlement - } else if preserve_live_terminal { - LineageRefreshAction::PreserveLiveTerminal - } else { - LineageRefreshAction::ReplaceFromRuntime - } +fn lineage_cancellation_result_is_current( + request_navigation_generation: u64, + current_navigation_generation: u64, + request_root_session_id: &str, + current_root_session_id: Option<&str>, + request_session_id: &str, + current_selected_session_id: Option<&str>, +) -> bool { + request_navigation_generation == current_navigation_generation + && current_root_session_id == Some(request_root_session_id) + && current_selected_session_id == Some(request_session_id) } fn push_bounded_lineage_event( @@ -789,6 +909,27 @@ fn push_bounded_lineage_event( *encoded_bytes = encoded_bytes.saturating_add(event_bytes); } +fn build_provisional_lineage_chat_state( + entry: &AgentSessionLineageEntry, + replay_events: &[AgenticEvent], +) -> ChatState { + let transcript = SessionTranscript { + session_id: entry.session_id.clone(), + messages: Vec::new(), + }; + let mut state = ChatState::from_session_transcript( + entry.session_id.clone(), + entry.session_name.clone(), + entry.agent_type.clone(), + entry.workspace_path.clone(), + &transcript, + ); + for event in replay_events { + project_transcript_event(&mut state, event, false); + } + state +} + fn build_lineage_chat_state( entry: &AgentSessionLineageEntry, mut inspection: AgentSessionLineageInspection, @@ -831,111 +972,26 @@ fn build_lineage_chat_state( state } -fn merge_live_terminal_into_authoritative( - authoritative: &mut ChatState, - live: &ChatState, - replayed: Option<&ChatState>, - terminal_turn_id: &str, -) { - let live_terminal_messages = live - .messages - .iter() - .filter(|message| message.turn_id.as_deref() == Some(terminal_turn_id)) - .cloned() - .collect::>(); - let replayed_terminal_messages = replayed - .into_iter() - .flat_map(|state| state.messages.iter()) - .filter(|message| message.turn_id.as_deref() == Some(terminal_turn_id)) - .cloned() - .collect::>(); - let has_terminal_marker = |messages: &[crate::chat_state::ChatMessage]| { - messages.iter().any(|message| { - message.flow_items.iter().any(|item| { - matches!( - item, - crate::chat_state::FlowItem::Text { content, .. } - if content == "[Cancelled]" || content.starts_with("[Error: ") - ) - }) - }) - }; - let projected_terminal_messages = - if has_terminal_marker(&live_terminal_messages) || replayed_terminal_messages.is_empty() { - &live_terminal_messages - } else { - &replayed_terminal_messages - }; - let authoritative_terminal_messages = authoritative - .messages - .iter() - .filter(|message| message.turn_id.as_deref() == Some(terminal_turn_id)) - .cloned() - .collect::>(); - let projected_has_user = projected_terminal_messages - .iter() - .any(|message| message.role == crate::chat_state::MessageRole::User); - let projected_has_assistant = projected_terminal_messages - .iter() - .any(|message| message.role == crate::chat_state::MessageRole::Assistant); - let mut replacement = Vec::new(); - if !projected_has_user { - replacement.extend( - authoritative_terminal_messages - .iter() - .filter(|message| message.role == crate::chat_state::MessageRole::User) - .cloned(), - ); - } - replacement.extend(projected_terminal_messages.iter().cloned()); - if !projected_has_assistant { - replacement.extend( - authoritative_terminal_messages - .iter() - .filter(|message| message.role == crate::chat_state::MessageRole::Assistant) - .cloned(), - ); - } - - let insertion_index = authoritative - .messages +fn lineage_session_index(snapshot: &AgentSessionLineageSnapshot) -> HashMap { + snapshot + .sessions .iter() - .position(|message| message.turn_id.as_deref() == Some(terminal_turn_id)) - .or_else(|| { - authoritative.current_turn_id().and_then(|active_turn_id| { - authoritative - .messages - .iter() - .position(|message| message.turn_id.as_deref() == Some(active_turn_id)) - }) - }) - .unwrap_or(authoritative.messages.len()); - let mut merged = Vec::with_capacity(authoritative.messages.len() + replacement.len()); - let mut inserted = false; - for (index, message) in authoritative.messages.drain(..).enumerate() { - if !inserted && index == insertion_index { - merged.append(&mut replacement); - inserted = true; - } - if message.turn_id.as_deref() != Some(terminal_turn_id) { - merged.push(message); - } - } - if !inserted { - merged.append(&mut replacement); - } - authoritative.messages = merged; - authoritative.metadata.message_count = authoritative.messages.len(); + .enumerate() + .map(|(index, entry)| (entry.session_id.clone(), index)) + .collect() } -fn update_lineage_active_turn(snapshot: &mut AgentSessionLineageSnapshot, event: &AgenticEvent) { +fn update_lineage_active_turn( + snapshot: &mut AgentSessionLineageSnapshot, + session_index: &HashMap, + event: &AgenticEvent, +) { let Some(session_id) = event.session_id() else { return; }; - let Some(entry) = snapshot - .sessions - .iter_mut() - .find(|entry| entry.session_id == session_id) + let Some(entry) = session_index + .get(session_id) + .and_then(|index| snapshot.sessions.get_mut(*index)) else { return; }; @@ -1005,15 +1061,14 @@ mod session_lineage_tests { use bitfun_events::AgenticEvent; use crate::chat_state::{FlowItem, MessageRole}; - use std::collections::{BTreeSet, VecDeque}; + use std::collections::{BTreeMap, HashMap, VecDeque}; use super::{ - build_lineage_chat_state, clear_lineage_settlement_for_new_turn, lineage_parent_session_id, - lineage_refresh_action, lineage_settlement_from_event, lineage_sibling_session_id, - lineage_terminal_reconciliation_pending, merge_live_terminal_into_authoritative, - project_transcript_event, provisional_lineage_replay_turn, push_bounded_lineage_event, - update_lineage_active_turn, AgentSessionInspection, BufferedLineageEvent, - LineageRefreshAction, LineageSettlement, + build_lineage_chat_state, lineage_cancellation_result_is_current, + lineage_inspection_result_is_current, lineage_parent_session_id, lineage_session_index, + lineage_sibling_session_id, project_transcript_event, push_bounded_lineage_event, + record_required_settled_lineage_turn, update_lineage_active_turn, BufferedLineageEvent, + LINEAGE_READ_BARRIER_MAX_TURNS_PER_SESSION, }; fn entry(id: &str, parent: Option<&str>) -> AgentSessionLineageEntry { @@ -1076,11 +1131,94 @@ mod session_lineage_tests { assert_eq!(lineage_sibling_session_id(&snapshot, "nested", 1), None); } + #[test] + fn inspection_freshness_is_scoped_to_the_selected_session() { + let generations = HashMap::from([ + ("selected".to_string(), 4), + ("streaming-sibling".to_string(), 99), + ]); + + assert!(lineage_inspection_result_is_current( + "selected", + 4, + &generations + )); + assert!(!lineage_inspection_result_is_current( + "selected", + 3, + &generations + )); + } + + #[test] + fn cancellation_result_is_hidden_after_switching_to_a_sibling() { + assert!(lineage_cancellation_result_is_current( + 4, + 4, + "root", + Some("root"), + "child-a", + Some("child-a"), + )); + assert!(!lineage_cancellation_result_is_current( + 4, + 4, + "root", + Some("root"), + "child-a", + Some("child-b"), + )); + assert!(!lineage_cancellation_result_is_current( + 4, + 4, + "root", + Some("root"), + "child-a", + None, + )); + } + + #[test] + fn lineage_runtime_io_is_started_off_the_tui_loop_and_polled_once() { + let source = include_str!("session_lineage.rs") + .split("#[cfg(test)]") + .next() + .expect("production lineage source"); + for method in [ + "show_session_lineage", + "inspect_lineage_session", + "refresh_inspected_lineage_if_due", + "cancel_inspected_lineage_session", + ] { + let body = source + .split(&format!("fn {method}")) + .nth(1) + .and_then(|source| source.split("\n }").next()) + .expect("lineage operation body"); + assert!(body.contains("rt_handle.spawn"), "{method} must spawn I/O"); + assert!( + !body.contains("block_in_place"), + "{method} must not synchronously wait on Runtime I/O" + ); + } + let poll = source + .split("fn poll_lineage_operation_completion") + .nth(1) + .and_then(|source| source.split("fn apply_lineage_inspection").next()) + .expect("lineage completion polling"); + assert!( + poll.find("is_finished").expect("finished gate") + < poll.find("block_in_place").expect("finished join") + ); + } + #[test] fn background_child_events_refresh_the_existing_lineage_snapshot() { let mut snapshot = snapshot(); + let session_index = lineage_session_index(&snapshot); update_lineage_active_turn( &mut snapshot, + &session_index, &AgenticEvent::DialogTurnStarted { session_id: "first".to_string(), turn_id: "turn-live".to_string(), @@ -1097,6 +1235,7 @@ mod session_lineage_tests { update_lineage_active_turn( &mut snapshot, + &session_index, &AgenticEvent::DialogTurnCancelled { session_id: "first".to_string(), turn_id: "turn-live".to_string(), @@ -1262,294 +1401,47 @@ mod session_lineage_tests { } #[test] - fn stale_terminal_cannot_replace_the_observed_active_turn() { - let terminal = AgenticEvent::DialogTurnCancelled { - session_id: "first".to_string(), - turn_id: "turn-old".to_string(), - }; + fn observed_terminal_turn_is_kept_as_a_runtime_read_barrier() { + let mut required_turns = BTreeMap::new(); + record_required_settled_lineage_turn(&mut required_turns, "first", "turn-old"); - assert!(lineage_settlement_from_event(None, false, &terminal).is_none()); - let settlement = lineage_settlement_from_event(None, true, &terminal).unwrap(); - assert_eq!(settlement.turn_id, "turn-old"); - assert!(settlement.preserve_live_terminal); + assert_eq!(required_turns["first"], ["turn-old"]); } #[test] fn duplicate_terminal_for_the_same_turn_is_ignored() { - let existing = LineageSettlement { - turn_id: "turn-live".to_string(), - preserve_live_terminal: true, - }; - let duplicate = AgenticEvent::DialogTurnCancelled { - session_id: "first".to_string(), - turn_id: "turn-live".to_string(), - }; - - assert!(lineage_settlement_from_event(Some(&existing), true, &duplicate).is_none()); - } + let mut required_turns = BTreeMap::new(); + record_required_settled_lineage_turn(&mut required_turns, "first", "turn-live"); + record_required_settled_lineage_turn(&mut required_turns, "first", "turn-live"); - #[test] - fn new_turn_keeps_terminal_reconciliation_until_authoritative_refresh() { - let mut inspection = AgentSessionInspection { - selected_session_id: "first".to_string(), - chat_state: crate::chat_state::ChatState::new( - "first".to_string(), - "first".to_string(), - "explore".to_string(), - None, - ), - refresh_pending: true, - refresh_due_at: std::time::Instant::now(), - refresh_deadline: Some(std::time::Instant::now()), - settling_turn_ids: BTreeSet::from(["turn-old".to_string()]), - preserve_live_terminal: true, - }; - - assert!(lineage_terminal_reconciliation_pending( - Some(&inspection), - "first" - )); - assert!(inspection.refresh_pending); - assert!(inspection.refresh_deadline.is_some()); - assert_eq!( - inspection.settling_turn_ids, - BTreeSet::from(["turn-old".to_string()]) - ); - assert!(inspection.preserve_live_terminal); - assert_eq!( - lineage_refresh_action( - &inspection.settling_turn_ids, - Some("turn-new"), - inspection.preserve_live_terminal, - ), - LineageRefreshAction::PreserveLiveTerminal - ); - - inspection.settling_turn_ids.clear(); - inspection.preserve_live_terminal = false; - assert!(!lineage_terminal_reconciliation_pending( - Some(&inspection), - "first" - )); - clear_lineage_settlement_for_new_turn(Some(&mut inspection), "first"); - assert!(inspection.refresh_pending); + assert_eq!(required_turns["first"], ["turn-live"]); } #[test] - fn settled_transcript_fills_provisional_history_without_losing_live_terminal() { - let entry = entry("first", Some("root")); - let replay_events = vec![ - AgenticEvent::TextChunk { - session_id: "first".to_string(), - turn_id: "turn-live".to_string(), - round_id: "round-live".to_string(), - attempt_id: None, - attempt_index: None, - text: "partial".to_string(), - }, - AgenticEvent::DialogTurnCancelled { - session_id: "first".to_string(), - turn_id: "turn-live".to_string(), - }, - ]; - let live = crate::chat_state::ChatState::new( - "first".to_string(), - "first".to_string(), - "explore".to_string(), - None, - ); - let inspection = AgentSessionLineageInspection { - transcript: SessionTranscript { - session_id: "first".to_string(), - messages: vec![TranscriptMessage { - id: Some("persisted-user".to_string()), - role: "user".to_string(), - turn_id: Some("turn-live".to_string()), - timestamp_ms: Some(1), - content: TranscriptContent::Text("continue".to_string()), - }], - }, - active_turn_id: None, - }; - let mut authoritative = build_lineage_chat_state(&entry, inspection.clone(), &[]); - let mut replay_inspection = inspection; - replay_inspection.active_turn_id = Some("turn-live".to_string()); - let replayed = build_lineage_chat_state(&entry, replay_inspection, &replay_events); - - merge_live_terminal_into_authoritative( - &mut authoritative, - &live, - Some(&replayed), - "turn-live", - ); + fn terminal_read_barriers_have_an_explicit_per_session_bound() { + let mut required_turns = BTreeMap::new(); + for index in 0..=LINEAGE_READ_BARRIER_MAX_TURNS_PER_SESSION { + record_required_settled_lineage_turn( + &mut required_turns, + "first", + &format!("turn-{index}"), + ); + } - assert!(authoritative.messages.iter().any(|message| { - message.role == MessageRole::User - && message.turn_id.as_deref() == Some("turn-live") - && message.flow_items.iter().any( - |item| matches!(item, FlowItem::Text { content, .. } if content == "continue"), - ) - })); - assert!(authoritative.messages.iter().any(|message| { - message.role == MessageRole::Assistant - && message.flow_items.iter().any( - |item| matches!(item, FlowItem::Text { content, .. } if content == "partial"), - ) - })); assert_eq!( - authoritative - .messages - .iter() - .flat_map(|message| message.flow_items.iter()) - .filter(|item| { - matches!(item, FlowItem::Text { content, .. } if content == "[Cancelled]") - }) - .count(), - 1 + required_turns["first"].len(), + LINEAGE_READ_BARRIER_MAX_TURNS_PER_SESSION ); + assert_eq!(required_turns["first"][0], "turn-1"); } #[test] - fn runtime_active_turn_stays_authoritative_while_previous_live_terminal_is_preserved() { - let entry = entry("first", Some("root")); - let live_events = vec![ - AgenticEvent::TextChunk { - session_id: "first".to_string(), - turn_id: "turn-a".to_string(), - round_id: "round-a".to_string(), - attempt_id: None, - attempt_index: None, - text: "partial a".to_string(), - }, - AgenticEvent::DialogTurnCancelled { - session_id: "first".to_string(), - turn_id: "turn-a".to_string(), - }, - ]; - let mut live = build_lineage_chat_state( - &entry, - AgentSessionLineageInspection { - transcript: SessionTranscript { - session_id: "first".to_string(), - messages: Vec::new(), - }, - active_turn_id: Some("turn-a".to_string()), - }, - &live_events, - ); - let next_start = AgenticEvent::DialogTurnStarted { - session_id: "first".to_string(), - turn_id: "turn-b".to_string(), - turn_index: 1, - user_input: "follow up".to_string(), - original_user_input: None, - user_message_metadata: None, - }; - assert!(project_transcript_event(&mut live, &next_start, false).changed); - let mut authoritative = build_lineage_chat_state( - &entry, - AgentSessionLineageInspection { - transcript: SessionTranscript { - session_id: "first".to_string(), - messages: vec![ - TranscriptMessage { - id: Some("user-a".to_string()), - role: "user".to_string(), - turn_id: Some("turn-a".to_string()), - timestamp_ms: Some(1), - content: TranscriptContent::Text("first".to_string()), - }, - TranscriptMessage { - id: Some("user-b".to_string()), - role: "user".to_string(), - turn_id: Some("turn-b".to_string()), - timestamp_ms: Some(2), - content: TranscriptContent::Text("follow up".to_string()), - }, - ], - }, - active_turn_id: Some("turn-b".to_string()), - }, - &[], - ); + fn consecutive_terminal_turns_remain_independent_read_barriers() { + let mut required_turns = BTreeMap::new(); + record_required_settled_lineage_turn(&mut required_turns, "first", "turn-a"); + record_required_settled_lineage_turn(&mut required_turns, "first", "turn-b"); - merge_live_terminal_into_authoritative(&mut authoritative, &live, None, "turn-a"); - - assert_eq!(authoritative.current_turn_id(), Some("turn-b")); - assert_eq!( - authoritative - .messages - .iter() - .filter_map(|message| message.turn_id.as_deref()) - .collect::>(), - vec!["turn-a", "turn-a", "turn-b", "turn-b"] - ); - assert!(authoritative.messages.iter().any(|message| { - message.role == MessageRole::User - && message.turn_id.as_deref() == Some("turn-a") - && message.flow_items.iter().any( - |item| matches!(item, FlowItem::Text { content, .. } if content == "first"), - ) - })); - assert!(authoritative.messages.iter().any(|message| { - message.role == MessageRole::Assistant - && message.turn_id.as_deref() == Some("turn-a") - && message.flow_items.iter().any( - |item| matches!(item, FlowItem::Text { content, .. } if content == "partial a"), - ) - })); - let delayed_start = project_transcript_event( - &mut authoritative, - &AgenticEvent::DialogTurnStarted { - session_id: "first".to_string(), - turn_id: "turn-b".to_string(), - turn_index: 1, - user_input: "follow up".to_string(), - original_user_input: None, - user_message_metadata: None, - }, - false, - ); - assert!(!delayed_start.changed); - assert_eq!( - authoritative - .messages - .iter() - .filter(|message| { - message.role == MessageRole::User - && message.turn_id.as_deref() == Some("turn-b") - }) - .count(), - 1 - ); - let chunk = project_transcript_event( - &mut authoritative, - &AgenticEvent::TextChunk { - session_id: "first".to_string(), - turn_id: "turn-b".to_string(), - round_id: "round-b".to_string(), - attempt_id: None, - attempt_index: None, - text: "live b".to_string(), - }, - false, - ); - assert!(chunk.changed); - assert!(authoritative.messages.iter().any(|message| { - message.role == MessageRole::Assistant - && message.turn_id.as_deref() == Some("turn-b") - && message.flow_items.iter().any( - |item| matches!(item, FlowItem::Text { content, .. } if content == "live b"), - ) - })); - } - - #[test] - fn provisional_replay_uses_the_lineage_active_turn_before_terminal_arrives() { - assert_eq!( - provisional_lineage_replay_turn(None, false, Some("turn-live")), - Some("turn-live") - ); + assert_eq!(required_turns["first"], vec!["turn-a", "turn-b"]); } #[test] @@ -1583,28 +1475,6 @@ mod session_lineage_tests { assert_eq!(buffer.len(), 1); } - #[test] - fn terminal_reconciliation_retries_while_the_settling_turn_is_still_active() { - let settling = BTreeSet::from(["turn-terminal".to_string()]); - - assert_eq!( - lineage_refresh_action(&settling, Some("turn-terminal"), true), - LineageRefreshAction::RetrySettlement - ); - assert_eq!( - lineage_refresh_action(&settling, None, true), - LineageRefreshAction::PreserveLiveTerminal - ); - assert_eq!( - lineage_refresh_action(&settling, Some("turn-new"), true), - LineageRefreshAction::PreserveLiveTerminal - ); - assert_eq!( - lineage_refresh_action(&settling, None, false), - LineageRefreshAction::ReplaceFromRuntime - ); - } - #[test] fn delayed_start_for_a_persisted_turn_is_ignored_after_inspection() { let entry = entry("first", Some("root")); diff --git a/src/apps/cli/src/ui/session_lineage_selector.rs b/src/apps/cli/src/ui/session_lineage_selector.rs index 9a4634bd15..ba48831e16 100644 --- a/src/apps/cli/src/ui/session_lineage_selector.rs +++ b/src/apps/cli/src/ui/session_lineage_selector.rs @@ -92,7 +92,7 @@ fn lineage_points(snapshot: &AgentSessionLineageSnapshot) -> Vec( depth } -fn lifecycle_label(status: AgentSessionLifecycleStatus) -> &'static str { - match status { - AgentSessionLifecycleStatus::Active => "active", +fn lineage_status_label(entry: &AgentSessionLineageEntry) -> &'static str { + if entry.active_turn_id.is_some() { + return "running"; + } + match entry.status { + AgentSessionLifecycleStatus::Active => "idle", AgentSessionLifecycleStatus::Archived => "archived", AgentSessionLifecycleStatus::Completed => "completed", } @@ -147,6 +150,17 @@ mod tests { } } + #[test] + fn active_turn_is_distinct_from_an_idle_active_session() { + let mut idle = entry("idle", Some("root"), 1); + idle.status = AgentSessionLifecycleStatus::Active; + let mut running = idle.clone(); + running.active_turn_id = Some("turn-live".to_string()); + + assert_eq!(lineage_status_label(&idle), "idle"); + assert_eq!(lineage_status_label(&running), "running"); + } + #[test] fn selector_preserves_owner_order_and_indents_nested_descendants() { let snapshot = AgentSessionLineageSnapshot { diff --git a/src/crates/adapters/agent-runtime-ipc/src/client.rs b/src/crates/adapters/agent-runtime-ipc/src/client.rs index b119734241..835f2ee41c 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/client.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/client.rs @@ -38,7 +38,36 @@ struct ClientCommand { frame_bytes: Vec, response: oneshot::Sender, deadline: tokio::time::Instant, - _request_gate: OwnedMutexGuard<()>, + release_gate_after_write: bool, + _request_gate: Option>, +} + +struct PendingRequest { + response: oneshot::Sender, + request_gate: Option>, + release_gate_after_write: bool, +} + +struct ClientWrite { + request_id: u64, + frame_bytes: Vec, + deadline: tokio::time::Instant, +} + +enum ClientWriteOutcome { + Complete { + request_id: u64, + }, + Timeout { + request_id: u64, + }, + FrameTooLarge { + request_id: u64, + error: RuntimeIpcIoError, + }, + Disconnected { + request_id: u64, + }, } enum PendingResponse { @@ -162,6 +191,7 @@ impl RuntimeIpcClient { operation: RuntimeIpcOperation, ) -> Result { let request_gate = self.request_gate.clone().lock_owned().await; + let release_gate_after_write = operation.is_interruptible_lineage_read(); let request_id = self .next_request_id .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| { @@ -185,7 +215,8 @@ impl RuntimeIpcClient { frame_bytes, response: sender, deadline, - _request_gate: request_gate, + release_gate_after_write, + _request_gate: Some(request_gate), }), ) .await @@ -229,12 +260,18 @@ impl RuntimeIpcClient { } } -async fn run_connection( - mut stream: crate::LocalIpcStream, +async fn run_connection( + stream: S, mut commands: mpsc::Receiver, events: broadcast::Sender, mut disconnect: watch::Receiver, -) { +) where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static, +{ + let (mut reader, writer) = tokio::io::split(stream); + let (writes, write_rx) = mpsc::channel(1); + let (write_outcomes, mut write_outcome_rx) = mpsc::unbounded_channel(); + let writer = tokio::spawn(run_writer(writer, write_rx, write_outcomes)); let mut pending = std::collections::HashMap::new(); let mut frames = RuntimeIpcFrameReader::new(MAX_RESPONSE_FRAME_BYTES); loop { @@ -254,34 +291,53 @@ async fn run_connection( let _ = command.response.send(PendingResponse::Timeout); continue; } - match tokio::time::timeout_at( - command.deadline, - write_serialized_frame_with_limit( - &mut stream, - &command.frame_bytes, - MAX_REQUEST_FRAME_BYTES, - ), - ).await { - Err(_) => { - let _ = command.response.send(PendingResponse::Timeout); - break; + pending.insert(request_id, PendingRequest { + response: command.response, + request_gate: command._request_gate, + release_gate_after_write: command.release_gate_after_write, + }); + if writes.try_send(ClientWrite { + request_id, + frame_bytes: command.frame_bytes, + deadline: command.deadline, + }).is_err() { + if let Some(request) = pending.remove(&request_id) { + let _ = request.response.send(PendingResponse::Disconnected); } - Ok(Err(error @ RuntimeIpcIoError::FrameTooLarge { .. })) => { - let _ = command.response.send(PendingResponse::Io(error)); - continue; + break; + } + }, + outcome = write_outcome_rx.recv() => match outcome { + Some(ClientWriteOutcome::Complete { request_id }) => { + if let Some(request) = pending.get_mut(&request_id) { + if request.release_gate_after_write { + request.request_gate.take(); + } } - Ok(Err(_)) => { - let _ = command.response.send(PendingResponse::Disconnected); - break; + } + Some(ClientWriteOutcome::FrameTooLarge { request_id, error }) => { + if let Some(request) = pending.remove(&request_id) { + let _ = request.response.send(PendingResponse::Io(error)); + } + } + Some(ClientWriteOutcome::Timeout { request_id }) => { + if let Some(request) = pending.remove(&request_id) { + let _ = request.response.send(PendingResponse::Timeout); + } + break; + } + Some(ClientWriteOutcome::Disconnected { request_id }) => { + if let Some(request) = pending.remove(&request_id) { + let _ = request.response.send(PendingResponse::Disconnected); } - Ok(Ok(())) => {} + break; } - pending.insert(request_id, (command.response, command._request_gate)); + None => break, }, - frame = frames.read_strict(&mut stream) => match frame { + frame = frames.read_strict(&mut reader) => match frame { Ok(RuntimeIpcFrame::Response { request_id, result }) => { - if let Some((sender, _request_gate)) = pending.remove(&request_id) { - let _ = sender.send(PendingResponse::Result(result)); + if let Some(request) = pending.remove(&request_id) { + let _ = request.response.send(PendingResponse::Result(result)); } else { break; } @@ -290,8 +346,8 @@ async fn run_connection( request_id: Some(request_id), error, }) => { - if let Some((sender, _request_gate)) = pending.remove(&request_id) { - let _ = sender.send(PendingResponse::Remote(error)); + if let Some(request) = pending.remove(&request_id) { + let _ = request.response.send(PendingResponse::Remote(error)); } else { break; } @@ -304,12 +360,51 @@ async fn run_connection( } } - for (_, (sender, _request_gate)) in pending.drain() { - let _ = sender.send(PendingResponse::Disconnected); + drop(writes); + writer.abort(); + let _ = writer.await; + for (_, request) in pending.drain() { + let _ = request.response.send(PendingResponse::Disconnected); } let _ = events.send(RuntimeIpcClientEvent::Disconnected); } +async fn run_writer( + mut writer: W, + mut writes: mpsc::Receiver, + outcomes: mpsc::UnboundedSender, +) where + W: tokio::io::AsyncWrite + Unpin, +{ + while let Some(write) = writes.recv().await { + let request_id = write.request_id; + let outcome = match tokio::time::timeout_at( + write.deadline, + write_serialized_frame_with_limit( + &mut writer, + &write.frame_bytes, + MAX_REQUEST_FRAME_BYTES, + ), + ) + .await + { + Err(_) => ClientWriteOutcome::Timeout { request_id }, + Ok(Err(error @ RuntimeIpcIoError::FrameTooLarge { .. })) => { + ClientWriteOutcome::FrameTooLarge { request_id, error } + } + Ok(Err(_)) => ClientWriteOutcome::Disconnected { request_id }, + Ok(Ok(())) => ClientWriteOutcome::Complete { request_id }, + }; + let terminal = matches!( + outcome, + ClientWriteOutcome::Timeout { .. } | ClientWriteOutcome::Disconnected { .. } + ); + if outcomes.send(outcome).is_err() || terminal { + break; + } + } +} + async fn timeout_io( timeout: Duration, future: impl std::future::Future>, @@ -352,3 +447,133 @@ pub enum RuntimeIpcClientError { #[error(transparent)] Io(#[from] RuntimeIpcIoError), } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{write_frame_with_limit, RuntimeSessionRenameRequest}; + + #[tokio::test] + async fn connection_reads_a_large_response_while_a_large_request_is_still_writing() { + let (client_stream, mut server_stream) = tokio::io::duplex(64); + let (commands, command_rx) = mpsc::channel(CLIENT_COMMAND_BUFFER); + let (events, _) = broadcast::channel(CLIENT_EVENT_BUFFER); + let (disconnect, disconnect_rx) = watch::channel(false); + let connection = tokio::spawn(run_connection( + client_stream, + command_rx, + events, + disconnect_rx, + )); + + let server = tokio::spawn(async move { + let mut frames = RuntimeIpcFrameReader::new(MAX_REQUEST_FRAME_BYTES); + let first = frames.read_strict(&mut server_stream).await.unwrap(); + assert!(matches!( + first, + RuntimeIpcFrame::Request { request_id: 2, .. } + )); + + // Observe the second write without draining it. With a single + // read/write task this fills both directions and deadlocks. + frames + .wait_for_frame_start(&mut server_stream) + .await + .unwrap(); + write_frame_with_limit( + &mut server_stream, + &RuntimeIpcFrame::Response { + request_id: 2, + result: RuntimeIpcOperationResult::Health { + instance_identity: "response".repeat(4_096), + process_id: 7, + }, + }, + MAX_RESPONSE_FRAME_BYTES, + ) + .await + .unwrap(); + + let second = frames.read_strict(&mut server_stream).await.unwrap(); + assert!(matches!( + second, + RuntimeIpcFrame::Request { request_id: 3, .. } + )); + write_frame_with_limit( + &mut server_stream, + &RuntimeIpcFrame::Response { + request_id: 3, + result: RuntimeIpcOperationResult::Unit, + }, + MAX_RESPONSE_FRAME_BYTES, + ) + .await + .unwrap(); + }); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + let request_gate = Arc::new(Mutex::new(())); + let (first_sender, first_response) = oneshot::channel(); + commands + .send(ClientCommand { + request_id: 2, + frame_bytes: serialize_frame_with_limit( + &RuntimeIpcFrame::Request { + request_id: 2, + operation: RuntimeIpcOperation::Health, + }, + MAX_REQUEST_FRAME_BYTES, + ) + .unwrap(), + response: first_sender, + deadline, + release_gate_after_write: true, + _request_gate: Some(request_gate.clone().lock_owned().await), + }) + .await + .unwrap(); + + let (second_sender, second_response) = oneshot::channel(); + let second_request_gate = request_gate.lock_owned().await; + commands + .send(ClientCommand { + request_id: 3, + frame_bytes: serialize_frame_with_limit( + &RuntimeIpcFrame::Request { + request_id: 3, + operation: RuntimeIpcOperation::RenameSession { + request: RuntimeSessionRenameRequest { + session_id: "session".to_string(), + session_name: "request".repeat(8_192), + }, + }, + }, + MAX_REQUEST_FRAME_BYTES, + ) + .unwrap(), + response: second_sender, + deadline, + release_gate_after_write: false, + _request_gate: Some(second_request_gate), + }) + .await + .unwrap(); + + tokio::time::timeout(Duration::from_millis(500), async { + assert!(matches!( + first_response.await.unwrap(), + PendingResponse::Result(RuntimeIpcOperationResult::Health { .. }) + )); + assert!(matches!( + second_response.await.unwrap(), + PendingResponse::Result(RuntimeIpcOperationResult::Unit) + )); + server.await.unwrap(); + }) + .await + .expect("full-duplex IPC must continue reading while a request write is backpressured"); + + let _ = disconnect.send(true); + connection.await.unwrap(); + } +} diff --git a/src/crates/adapters/agent-runtime-ipc/src/operation.rs b/src/crates/adapters/agent-runtime-ipc/src/operation.rs index 176db67e7c..c7cfb06792 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/operation.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/operation.rs @@ -131,6 +131,17 @@ pub enum RuntimeIpcOperation { } impl RuntimeIpcOperation { + /// These speculative reads may be superseded by a newer request from the + /// same TUI connection. The server keeps their execution outside the + /// connection's serial control path so cancellation and Session changes + /// are never queued behind transcript I/O. + pub(crate) fn is_interruptible_lineage_read(&self) -> bool { + matches!( + self, + Self::GetSessionLineage { .. } | Self::InspectLineageSession { .. } + ) + } + pub fn session_id(&self) -> Option<&str> { match self { Self::RestoreSession { request } => Some(&request.session_id), @@ -421,6 +432,7 @@ mod tests { workspace_path: "D:/workspace/project".to_string(), root_session_id: "root-1".to_string(), session_id: "child-1".to_string(), + required_settled_turn_ids: Vec::new(), remote_connection_id: None, remote_ssh_host: None, }, @@ -430,6 +442,7 @@ mod tests { workspace_path: "D:/workspace/project".to_string(), root_session_id: "root-1".to_string(), session_id: "child-1".to_string(), + expected_active_turn_id: Some("turn-child".to_string()), source: None, reason: None, wait_timeout_ms: None, diff --git a/src/crates/adapters/agent-runtime-ipc/src/protocol.rs b/src/crates/adapters/agent-runtime-ipc/src/protocol.rs index c6cf4c4d03..9ba79b7777 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/protocol.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/protocol.rs @@ -5,7 +5,7 @@ use crate::{RuntimeIpcOperation, RuntimeIpcOperationResult}; use bitfun_events::AgenticEventEnvelope; use bitfun_product_domains::tool_permissions::PermissionRequestEvent; -pub const PROTOCOL_VERSION: u32 = 14; +pub const PROTOCOL_VERSION: u32 = 15; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)] diff --git a/src/crates/adapters/agent-runtime-ipc/src/server.rs b/src/crates/adapters/agent-runtime-ipc/src/server.rs index 0efffa8f3d..998d266755 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/server.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/server.rs @@ -313,6 +313,8 @@ async fn run_initialized_connection( ) -> Result<(), RuntimeIpcServerError> { let mut frames = RuntimeIpcFrameReader::new(MAX_REQUEST_FRAME_BYTES); let mut frame_deadline = None; + let mut interruptible_lineage_reads = JoinSet::new(); + let mut interruptible_lineage_request_id = None; loop { match next_connection_input( config.request_timeout, @@ -321,9 +323,57 @@ async fn run_initialized_connection( &mut frame_deadline, events.as_mut(), availability.as_mut(), + &mut interruptible_lineage_reads, ) .await? { + ConnectionInput::InterruptibleLineageReadCompleted(completed) => { + let Some(request_id) = interruptible_lineage_request_id.take() else { + continue; + }; + match completed { + Ok(InterruptibleLineageReadResult::Completed(result)) => match result { + Ok(result) => { + send_operation_result( + config.request_timeout, + stream, + request_id, + result, + ) + .await?; + } + Err(error) => { + send_runtime_error( + stream, + config.request_timeout, + Some(request_id), + error, + ) + .await?; + } + }, + Ok(InterruptibleLineageReadResult::Deadline) => { + send_error( + stream, + config.request_timeout, + Some(request_id), + RuntimeIpcErrorCode::Unavailable, + "runtime lineage read exceeded its deadline", + ) + .await?; + } + Err(error) => { + send_error( + stream, + config.request_timeout, + Some(request_id), + RuntimeIpcErrorCode::Internal, + &format!("runtime lineage read task stopped: {error}"), + ) + .await?; + } + } + } ConnectionInput::Event(event) => { if matches!(event, RuntimeIpcEvent::StreamInvalidated { .. }) { timeout_write( @@ -378,6 +428,18 @@ async fn run_initialized_connection( request_id, operation, }) => { + if let Some(superseded_request_id) = interruptible_lineage_request_id.take() { + interruptible_lineage_reads.abort_all(); + while interruptible_lineage_reads.join_next().await.is_some() {} + send_error( + stream, + config.request_timeout, + Some(superseded_request_id), + RuntimeIpcErrorCode::Unavailable, + "runtime lineage read was superseded by a newer request", + ) + .await?; + } if matches!(operation, RuntimeIpcOperation::Health) { send_operation_result( config.request_timeout, @@ -460,6 +522,20 @@ async fn run_initialized_connection( continue; } }; + if operation.is_interruptible_lineage_read() { + let handler = handler.clone(); + let request_timeout = config.request_timeout; + interruptible_lineage_reads.spawn(async move { + match tokio::time::timeout(request_timeout, handler.execute(operation)) + .await + { + Ok(result) => InterruptibleLineageReadResult::Completed(result), + Err(_) => InterruptibleLineageReadResult::Deadline, + } + }); + interruptible_lineage_request_id = Some(request_id); + continue; + } let provisional_turn_id = match &operation { RuntimeIpcOperation::SubmitTurn { request } => { let Some(turn_id) = request.turn_id.clone() else { @@ -706,6 +782,14 @@ enum ConnectionInput { EventLagged, EventClosed, RuntimeUnavailable, + InterruptibleLineageReadCompleted( + Result, + ), +} + +enum InterruptibleLineageReadResult { + Completed(Result), + Deadline, } async fn next_connection_input( @@ -715,9 +799,13 @@ async fn next_connection_input( frame_deadline: &mut Option, events: Option<&mut broadcast::Receiver>, availability: Option<&mut watch::Receiver>, + interruptible_lineage_reads: &mut JoinSet, ) -> Result { tokio::select! { frame = read_connected(timeout, stream, frames, frame_deadline) => frame.map(ConnectionInput::Frame), + completed = receive_interruptible_lineage_read(interruptible_lineage_reads) => { + Ok(ConnectionInput::InterruptibleLineageReadCompleted(completed)) + }, event = receive_event(events) => match event { None => std::future::pending().await, Some(Ok(event)) => Ok(ConnectionInput::Event(event)), @@ -728,6 +816,18 @@ async fn next_connection_input( } } +async fn receive_interruptible_lineage_read( + reads: &mut JoinSet, +) -> Result { + if reads.is_empty() { + return std::future::pending().await; + } + reads + .join_next() + .await + .expect("a non-empty lineage read set should yield a task") +} + async fn receive_event( events: Option<&mut broadcast::Receiver>, ) -> Option> { diff --git a/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs b/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs index 1a447faeec..6499aa6f05 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/tests/protocol_contracts.rs @@ -75,7 +75,7 @@ fn protocol_round_trips_reviewed_permission_and_user_input_operations() { #[test] fn protocol_round_trips_exact_turn_steering_without_replacing_turn_admission() { - assert_eq!(PROTOCOL_VERSION, 14); + assert_eq!(PROTOCOL_VERSION, 15); let operation = RuntimeIpcOperation::SteerTurn { request: AgentDialogSteerRequest { session_id: "session-1".to_string(), @@ -160,6 +160,7 @@ fn protocol_round_trips_root_scoped_lineage_operations() { workspace_path: "D:/workspace/project".to_string(), root_session_id: "root-1".to_string(), session_id: "child-1".to_string(), + required_settled_turn_ids: vec!["turn-terminal".to_string()], remote_connection_id: None, remote_ssh_host: None, }, @@ -169,6 +170,7 @@ fn protocol_round_trips_root_scoped_lineage_operations() { workspace_path: "D:/workspace/project".to_string(), root_session_id: "root-1".to_string(), session_id: "child-1".to_string(), + expected_active_turn_id: Some("turn-child".to_string()), source: Some(AgentSubmissionSource::Cli), reason: Some("user_cancelled".to_string()), wait_timeout_ms: Some(5_000), @@ -189,7 +191,7 @@ fn protocol_round_trips_root_scoped_lineage_operations() { #[test] fn protocol_round_trips_workspace_diff_as_a_read_only_workspace_operation() { - assert_eq!(PROTOCOL_VERSION, 14); + assert_eq!(PROTOCOL_VERSION, 15); let operation = RuntimeIpcOperation::WorkspaceDiff; let encoded = serde_json::to_value(&operation).expect("serialize workspace diff operation"); @@ -310,7 +312,7 @@ fn protocol_round_trips_the_reviewed_session_model_operation() { #[test] fn protocol_round_trips_the_current_session_rename_operation() { - assert_eq!(PROTOCOL_VERSION, 14); + assert_eq!(PROTOCOL_VERSION, 15); let operation = RuntimeIpcOperation::RenameSession { request: RuntimeSessionRenameRequest { diff --git a/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs b/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs index f9c5b1f486..de4f9bda64 100644 --- a/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs +++ b/src/crates/adapters/agent-runtime-ipc/src/tests/shared_controller.rs @@ -9,7 +9,8 @@ use async_trait::async_trait; use bitfun_events::{AgenticEvent, AgenticEventEnvelope, AgenticEventPriority}; use bitfun_runtime_ports::{ AgentDialogTurnRequest, AgentSessionCompactionRequest, AgentSessionComposerUpdate, - AgentSessionCreateRequest, AgentSessionCreateResult, AgentSessionModeUpdateRequest, + AgentSessionCreateRequest, AgentSessionCreateResult, AgentSessionLineageCancellationRequest, + AgentSessionLineageTranscriptRequest, AgentSessionModeUpdateRequest, AgentSessionModelUpdateRequest, AgentSessionRevertRequest, AgentSessionRevertResult, AgentSessionSummary, AgentSubmissionSource, AgentUserShellCommandRequest, DialogSubmissionPolicy, SessionTranscript, @@ -90,6 +91,7 @@ struct FakeHandler { rename_delay: Option, delete_delay: Option, submit_delay: Option, + lineage_read_delay: Option, invalid_steer_result: bool, settle_cancel: bool, events: broadcast::Sender, @@ -115,6 +117,7 @@ impl Default for FakeHandler { rename_delay: None, delete_delay: None, submit_delay: None, + lineage_read_delay: None, invalid_steer_result: false, settle_cancel: true, events, @@ -205,6 +208,15 @@ impl RuntimeIpcRequestHandler for FakeHandler { tokio::time::sleep(delay).await; } } + if matches!( + operation, + RuntimeIpcOperation::GetSessionLineage { .. } + | RuntimeIpcOperation::InspectLineageSession { .. } + ) { + if let Some(delay) = self.lineage_read_delay { + tokio::time::sleep(delay).await; + } + } match operation { RuntimeIpcOperation::RestoreSession { request } => Ok(restored(&request.session_id)), RuntimeIpcOperation::ForkSession { .. } => { @@ -250,6 +262,26 @@ impl RuntimeIpcRequestHandler for FakeHandler { turn_id: request.turn_id, }) } + RuntimeIpcOperation::InspectLineageSession { request } => { + Ok(RuntimeIpcOperationResult::LineageSessionInspection { + inspection: bitfun_runtime_ports::AgentSessionLineageInspection { + transcript: SessionTranscript { + session_id: request.session_id, + messages: Vec::new(), + }, + active_turn_id: None, + }, + }) + } + RuntimeIpcOperation::CancelLineageSession { request } => { + Ok(RuntimeIpcOperationResult::TurnCancelled { + cancellation: bitfun_runtime_ports::AgentTurnCancellationResult { + session_id: request.session_id, + turn_id: None, + requested: true, + }, + }) + } RuntimeIpcOperation::UndoSession { request } | RuntimeIpcOperation::RedoSession { request } => { Ok(RuntimeIpcOperationResult::SessionReverted { @@ -415,6 +447,88 @@ async fn first_party_timeout_reports_unknown_outcome_and_releases_the_lease() { server.finish().await; } +#[tokio::test] +async fn cancellation_supersedes_a_slow_lineage_read_on_the_same_client() { + let handler = Arc::new(FakeHandler { + lineage_read_delay: Some(Duration::from_secs(2)), + ..FakeHandler::default() + }); + let server = TestServer::start(server_config(), handler.clone()).await; + let client = RuntimeIpcClient::connect( + server.runtime_root.path(), + &server.discovery, + "lineage-controller", + "0.1.0", + Duration::from_secs(2), + Duration::from_secs(3), + ) + .await + .expect("connect first-party client"); + client + .request(restore_operation(server.workspace.path(), "session-a")) + .await + .expect("restore root session"); + + let inspect_client = client.clone(); + let workspace_path = server.workspace.path().to_string_lossy().to_string(); + let inspect_workspace_path = workspace_path.clone(); + let inspect = tokio::spawn(async move { + inspect_client + .request(RuntimeIpcOperation::InspectLineageSession { + request: AgentSessionLineageTranscriptRequest { + workspace_path: inspect_workspace_path, + root_session_id: "session-a".to_string(), + session_id: "session-child".to_string(), + required_settled_turn_ids: Vec::new(), + remote_connection_id: None, + remote_ssh_host: None, + }, + }) + .await + }); + wait_for_calls(&handler, |calls| { + calls + .iter() + .any(|call| matches!(call, RuntimeIpcOperation::InspectLineageSession { .. })) + }) + .await; + + let cancellation = tokio::time::timeout( + Duration::from_millis(300), + client.request(RuntimeIpcOperation::CancelLineageSession { + request: AgentSessionLineageCancellationRequest { + workspace_path, + root_session_id: "session-a".to_string(), + session_id: "session-child".to_string(), + expected_active_turn_id: Some("turn-child".to_string()), + source: None, + reason: None, + wait_timeout_ms: None, + remote_connection_id: None, + remote_ssh_host: None, + }, + }), + ) + .await + .expect("lineage cancellation must not wait for transcript I/O") + .expect("cancel response"); + assert!(matches!( + cancellation, + RuntimeIpcOperationResult::TurnCancelled { cancellation } + if cancellation.requested && cancellation.session_id == "session-child" + )); + assert!(matches!( + inspect.await.expect("inspect task"), + Err(RuntimeIpcClientError::Remote(RuntimeIpcError { + code: RuntimeIpcErrorCode::Unavailable, + .. + })) + )); + + drop(client); + server.finish().await; +} + #[tokio::test] async fn generated_session_is_claimed_before_another_connection_can_restore_it() { let create_started = Arc::new(Notify::new()); diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index 6e61a04658..144611923e 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -788,7 +788,6 @@ struct ActiveSubagentExecution { subagent_session_id: String, subagent_dialog_turn_id: String, cancel_token: CancellationToken, - abort_handle: tokio::task::AbortHandle, } #[derive(Clone)] @@ -1010,6 +1009,36 @@ fn lineage_session_is_settling_without_active_state( active_turn_id.is_none() && in_flight_execution_count > 0 } +pub(crate) fn validate_required_lineage_turns_settled( + turns: &[DialogTurnData], + required_settled_turn_ids: &[String], +) -> bitfun_runtime_ports::PortResult<()> { + for required_turn_id in required_settled_turn_ids { + let settled = turns + .iter() + .any(|turn| turn.turn_id == *required_turn_id && turn.status != TurnStatus::InProgress); + if !settled { + return Err(bitfun_runtime_ports::PortError::new( + bitfun_runtime_ports::PortErrorKind::OutcomeUnknown, + format!( + "Required terminal Turn is not yet durable in the authoritative transcript: turn_id={required_turn_id}" + ), + )); + } + } + Ok(()) +} + +fn lineage_post_admission_cancellation_error( + error: BitFunError, + session_id: &str, + turn_id: &str, +) -> BitFunError { + BitFunError::OutcomeUnknown(format!( + "Subagent cancellation was admitted, but its final outcome was not confirmed: session_id={session_id}, turn_id={turn_id}; {error}" + )) +} + /// Conversation coordinator pub struct ConversationCoordinator { session_manager: Arc, @@ -5844,7 +5873,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet ))) } - async fn cancel_active_subagents_for_parent_turn( + fn cancel_active_subagents_for_parent_turn( &self, parent_session_id: &str, parent_dialog_turn_id: &str, @@ -5871,14 +5900,13 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet ); for active in active_subagents { - self.stop_active_subagent_execution(&active, "Parent dialog turn cancelled") - .await; + self.signal_active_subagent_cancellation(&active, "Parent dialog turn cancelled"); } } - async fn stop_active_subagent_execution(&self, active: &ActiveSubagentExecution, reason: &str) { + fn signal_active_subagent_cancellation(&self, active: &ActiveSubagentExecution, reason: &str) { debug!( - "Stopping active subagent execution: subagent_session_id={}, subagent_dialog_turn_id={}, parent_session_id={}, parent_dialog_turn_id={}, reason={}", + "Signalling active subagent cancellation: subagent_session_id={}, subagent_dialog_turn_id={}, parent_session_id={}, parent_dialog_turn_id={}, reason={}", active.subagent_session_id, active.subagent_dialog_turn_id, active.parent_session_id, @@ -5886,48 +5914,12 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet reason ); + // The outer subagent execution task is the sole terminal persistence + // owner. It observes this token, cancels the engine/tools, waits for + // the inner task, and writes exactly one Cancelled outcome. Aborting + // and persisting here races that owner and can turn cancellation into + // a JoinError/Failed outcome or emit duplicate terminal events. active.cancel_token.cancel(); - active.abort_handle.abort(); - - if let Err(error) = self - .execution_engine - .cancel_dialog_turn(&active.subagent_dialog_turn_id) - .await - { - warn!( - "Failed to cancel active subagent dialog turn: subagent_session_id={}, subagent_dialog_turn_id={}, error={}", - active.subagent_session_id, active.subagent_dialog_turn_id, error - ); - } - - if let Err(error) = self - .tool_pipeline - .cancel_dialog_turn_tools(&active.subagent_dialog_turn_id) - .await - { - warn!( - "Failed to cancel active subagent tools: subagent_session_id={}, subagent_dialog_turn_id={}, error={}", - active.subagent_session_id, active.subagent_dialog_turn_id, error - ); - } - - Self::persist_cancelled_dialog_turn( - self.event_queue.as_ref(), - self.session_manager.as_ref(), - None, - &active.subagent_session_id, - &active.subagent_dialog_turn_id, - true, - ) - .await; - - self.session_manager.reset_session_state_if_processing( - &active.subagent_session_id, - &active.subagent_dialog_turn_id, - ); - - self.active_subagent_executions - .remove(&active.subagent_session_id); } /// Cancel dialog turn execution @@ -5937,8 +5929,13 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet session_id: &str, dialog_turn_id: &str, ) -> BitFunResult<()> { - self.cancel_dialog_turn_with_descendant_policy(session_id, dialog_turn_id, true) - .await + self.cancel_dialog_turn_with_descendant_policy( + session_id, + dialog_turn_id, + true, + Duration::from_millis(1500), + ) + .await } async fn cancel_dialog_turn_with_descendant_policy( @@ -5946,6 +5943,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet session_id: &str, dialog_turn_id: &str, cancel_descendants: bool, + drain_timeout: Duration, ) -> BitFunResult<()> { info!( "Received cancel request: dialog_turn_id={}, session_id={}, cancel_descendants={}", @@ -5975,14 +5973,35 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet // cancellation still targets the currently processing turn. A delayed // cancel request for an older turn must not clear a newer turn. debug!("Conditionally updating session state to Idle for cancelled turn"); - let state_updated = self + let state_update_result = self .session_manager .update_session_state_for_turn_if_processing( session_id, dialog_turn_id, SessionState::Idle, ) - .await?; + .await; + + // A persistence failure can occur after SessionManager has already + // changed the in-memory state. Cancellation has been admitted at that + // point, so it must still reach the engine, tools, and descendants. + // Preserve the error for the caller, but never return before sending + // those signals. + let (state_updated, state_update_error) = match state_update_result { + Ok(state_updated) => (state_updated, None), + Err(error) => { + let updated_in_memory = self + .session_manager + .get_session(session_id) + .map(|session| matches!(session.state, SessionState::Idle)) + .unwrap_or(false); + warn!( + "Failed to persist cancelled Session state; cancellation signals will still be delivered: session_id={}, dialog_turn_id={}, error={}", + session_id, dialog_turn_id, error + ); + (updated_in_memory, Some(error)) + } + }; let new_state = self .session_manager @@ -6029,20 +6048,18 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet } if cancel_descendants { - self.cancel_active_subagents_for_parent_turn(session_id, dialog_turn_id) - .await; + self.cancel_active_subagents_for_parent_turn(session_id, dialog_turn_id); } // Step 4: Wait briefly for the spawn task that owns this turn to drain // its in-memory message writes before returning. Capped so the RPC // never blocks longer than ~1.5s — beyond that we let the new turn // proceed and rely on the cancellation token already being signalled. - let pending = self - .wait_session_drained(session_id, Duration::from_millis(1500)) - .await; + let pending = self.wait_session_drained(session_id, drain_timeout).await; if pending > 0 { warn!( - "Cancelled turn did not fully drain within 1500ms: session_id={}, dialog_turn_id={}, pending={}", + "Cancelled turn did not fully drain within {}ms: session_id={}, dialog_turn_id={}, pending={}", + drain_timeout.as_millis(), session_id, dialog_turn_id, pending ); } else { @@ -6052,6 +6069,10 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet ); } + if let Some(error) = state_update_error { + return Err(error); + } + Ok(()) } @@ -6084,16 +6105,22 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet return Ok(None); }; + let deadline = Instant::now() + wait_timeout; + let drain_timeout = std::cmp::min( + Duration::from_millis(1500), + deadline.saturating_duration_since(Instant::now()), + ); self.cancel_dialog_turn_with_descendant_policy( session_id, ¤t_turn_id, cancel_descendants, + drain_timeout, ) .await?; - let deadline = Instant::now() + wait_timeout; while self.execution_engine.has_active_turn(¤t_turn_id) { - if Instant::now() >= deadline { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { warn!( "Timed out waiting for active turn cancellation: session_id={}, dialog_turn_id={}, timeout_ms={}", session_id, @@ -6105,7 +6132,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet wait_timeout.as_millis() ))); } - sleep(Duration::from_millis(50)).await; + sleep(std::cmp::min(Duration::from_millis(50), remaining)).await; } Ok(Some(current_turn_id)) @@ -6115,24 +6142,64 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet &self, storage_path: &Path, session_id: &str, + expected_active_turn_id: Option<&str>, wait_timeout: Duration, ) -> BitFunResult> { - let _mutation_guard = self - .session_manager - .acquire_session_mutation(session_id) - .await?; + let deadline = Instant::now() + wait_timeout; + let _mutation_guard = tokio::time::timeout( + wait_timeout, + self.session_manager.acquire_session_mutation(session_id), + ) + .await + .map_err(|_| { + BitFunError::Timeout(format!( + "Timed out acquiring the Session lifecycle lease before lineage cancellation: session_id={session_id}" + )) + })??; if !self .session_manager .is_session_loaded_from_storage_path(storage_path, session_id)? { return Ok(None); } + let active_turn_id = self + .session_manager + .get_session(session_id) + .and_then(|session| match session.state { + SessionState::Processing { + current_turn_id, .. + } => Some(current_turn_id), + _ => None, + }); + if active_turn_id.as_deref() != expected_active_turn_id { + return Err(BitFunError::OutcomeUnknown(format!( + "Subagent Session active Turn changed before cancellation: session_id={session_id}, expected_turn_id={}, active_turn_id={}", + expected_active_turn_id.unwrap_or("none"), + active_turn_id.as_deref().unwrap_or("none") + ))); + } + if active_turn_id.is_none() { + return Ok(None); + } // Match the Session abort semantics used by OpenCode: interrupting an // inspected subagent stops the execution subtree rooted at that Session. // A running Task is part of the selected Turn, so preserving its child // while cancelling the owning Tool would leave the parent Turn unsettled. - self.cancel_active_turn_for_session_with_descendant_policy(session_id, wait_timeout, true) - .await + self.cancel_active_turn_for_session_with_descendant_policy( + session_id, + deadline.saturating_duration_since(Instant::now()), + true, + ) + .await + .map_err(|error| { + lineage_post_admission_cancellation_error( + error, + session_id, + active_turn_id + .as_deref() + .expect("active turn was checked before cancellation"), + ) + }) } /// Delete session @@ -7983,7 +8050,6 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet subagent_session_id: session_id.clone(), subagent_dialog_turn_id: dialog_turn_id.clone(), cancel_token: subagent_cancel_token.clone(), - abort_handle: abort_handle.clone(), }, ); } @@ -12005,6 +12071,7 @@ impl ConversationCoordinator { &self, request: bitfun_runtime_ports::SessionTranscriptRequest, status_turn_id: Option<&str>, + required_settled_turn_ids: &[String], ) -> bitfun_runtime_ports::PortResult<( bitfun_runtime_ports::SessionTranscript, Option, @@ -12015,29 +12082,40 @@ impl ConversationCoordinator { .await .map_err(runtime_port_error_preserving_message)? { - Some(turns) => ( - runtime_transcript_messages_from_turns(&turns, request.turn_id.as_deref()), - status_turn_id.and_then(|turn_id| { - turns - .iter() - .find(|turn| turn.turn_id == turn_id) - .map(|turn| turn.status.clone()) - }), - ), - None => ( - self.session_manager - .get_context_messages(&request.session_id) - .await - .map_err(runtime_port_error_preserving_message)? - .into_iter() - .filter(|message| match request.turn_id.as_ref() { - Some(turn_id) => message.metadata.turn_id.as_ref() == Some(turn_id), - None => true, - }) - .map(runtime_transcript_message_from_message) - .collect(), - None, - ), + Some(turns) => { + validate_required_lineage_turns_settled(&turns, required_settled_turn_ids)?; + ( + runtime_transcript_messages_from_turns(&turns, request.turn_id.as_deref()), + status_turn_id.and_then(|turn_id| { + turns + .iter() + .find(|turn| turn.turn_id == turn_id) + .map(|turn| turn.status.clone()) + }), + ) + } + None => { + if !required_settled_turn_ids.is_empty() { + return Err(bitfun_runtime_ports::PortError::new( + bitfun_runtime_ports::PortErrorKind::OutcomeUnknown, + "Required terminal Turns are not yet durable in the authoritative transcript", + )); + } + ( + self.session_manager + .get_context_messages(&request.session_id) + .await + .map_err(runtime_port_error_preserving_message)? + .into_iter() + .filter(|message| match request.turn_id.as_ref() { + Some(turn_id) => message.metadata.turn_id.as_ref() == Some(turn_id), + None => true, + }) + .map(runtime_transcript_message_from_message) + .collect(), + None, + ) + } }; Ok(( @@ -12053,7 +12131,7 @@ impl ConversationCoordinator { &self, request: bitfun_runtime_ports::SessionTranscriptRequest, ) -> bitfun_runtime_ports::PortResult { - self.read_session_transcript_with_turn_status_locked(request, None) + self.read_session_transcript_with_turn_status_locked(request, None, &[]) .await .map(|(transcript, _)| transcript) } @@ -12062,6 +12140,7 @@ impl ConversationCoordinator { &self, storage_path: &Path, request: bitfun_runtime_ports::SessionTranscriptRequest, + required_settled_turn_ids: &[String], ) -> bitfun_runtime_ports::PortResult> { let _mutation_guard = self @@ -12104,6 +12183,16 @@ impl ConversationCoordinator { .get(&request.session_id) .map(|counter| counter.load(Ordering::SeqCst)) .unwrap_or(0); + if candidate_active_turn_id.as_deref().is_some_and(|turn_id| { + required_settled_turn_ids + .iter() + .any(|observed| observed == turn_id) + }) { + return Err(bitfun_runtime_ports::PortError::new( + bitfun_runtime_ports::PortErrorKind::OutcomeUnknown, + "Session still reports an observed terminal turn as active; retry the inspection", + )); + } if lineage_session_is_settling_without_active_state( candidate_active_turn_id.as_deref(), in_flight_execution_count, @@ -12117,6 +12206,7 @@ impl ConversationCoordinator { .read_session_transcript_with_turn_status_locked( request.clone(), candidate_active_turn_id.as_deref(), + required_settled_turn_ids, ) .await?; let current_active_turn_id = self @@ -12133,6 +12223,7 @@ impl ConversationCoordinator { .read_session_transcript_with_turn_status_locked( request, candidate_active_turn_id.as_deref(), + required_settled_turn_ids, ) .await?; if settled_turn_status @@ -12311,16 +12402,18 @@ fn merge_prepended_messages_for_turn( mod tests { use super::{ btw_session_memory_mode, build_subagent_session_relationship, - lineage_active_turn_after_transcript, lineage_session_is_settling_without_active_state, - logical_subagent_type_or_runtime, merge_prepended_messages_for_turn, - normalize_subagent_max_concurrency, resolve_agent_session_create_created_by, - resolve_agent_submission_turn_id, resolve_subagent_model_selection, - runtime_port_error_preserving_message, runtime_session_summary, - runtime_tool_restrictions_for_session_lifetime, runtime_transcript_messages_from_turns, - session_storage_workspace_locator, turn_review_manifest_for_agent, - BackgroundSubagentWaitMode, ContextCompactionOutcome, ConversationCoordinator, - ManualCompactionCommitGate, SessionMemoryMode, SessionReferenceLocator, - SessionRelationshipKind, SubagentExecutionRequest, TEST_AGENT_MODEL_DEFAULTS, + lineage_active_turn_after_transcript, lineage_post_admission_cancellation_error, + lineage_session_is_settling_without_active_state, logical_subagent_type_or_runtime, + merge_prepended_messages_for_turn, normalize_subagent_max_concurrency, + resolve_agent_session_create_created_by, resolve_agent_submission_turn_id, + resolve_subagent_model_selection, runtime_port_error_preserving_message, + runtime_session_summary, runtime_tool_restrictions_for_session_lifetime, + runtime_transcript_messages_from_turns, session_storage_workspace_locator, + turn_review_manifest_for_agent, validate_required_lineage_turns_settled, + ActiveSubagentExecution, BackgroundSubagentWaitMode, ContextCompactionOutcome, + ConversationCoordinator, ManualCompactionCommitGate, SessionMemoryMode, + SessionReferenceLocator, SessionRelationshipKind, SubagentExecutionRequest, + TEST_AGENT_MODEL_DEFAULTS, }; use crate::agentic::agents::ExternalSubagentModelBinding; use crate::agentic::coordination::coordination_store::{ @@ -12328,8 +12421,8 @@ mod tests { }; use crate::agentic::core::{ InternalReminderKind, Message, MessageContent, MessageRole, MessageSemanticKind, - SessionConfig, SessionContinuationPolicy, SessionKind, SessionModelBindingPolicy, - SessionState, TurnStats, + ProcessingPhase, SessionConfig, SessionContinuationPolicy, SessionKind, + SessionModelBindingPolicy, SessionState, ToolCall, TurnStats, }; use crate::agentic::events::{AgenticEvent, EventQueue, EventQueueConfig, EventRouter}; use crate::agentic::execution::{ @@ -12346,7 +12439,9 @@ mod tests { use crate::agentic::tools::framework::{ PermissionIntent, Tool, ToolResult, ToolUseContext, ValidationResult, }; - use crate::agentic::tools::pipeline::SubagentParentInfo; + use crate::agentic::tools::pipeline::{ + SubagentParentInfo, ToolExecutionContext, ToolExecutionOptions, ToolTask, + }; use crate::agentic::tools::registry::ToolRegistry; use crate::agentic::tools::{ToolPipeline, ToolStateManager}; use crate::agentic::TurnSkillAgentSnapshot; @@ -12386,6 +12481,172 @@ mod tests { assert!(!lineage_session_is_settling_without_active_state(None, 0)); } + #[test] + fn lineage_read_barrier_requires_each_turn_to_be_durably_terminal() { + let turn = |turn_id: &str, status| { + let mut turn = DialogTurnData::new( + turn_id.to_string(), + 0, + "session-1".to_string(), + UserMessageData { + id: format!("{turn_id}-user"), + content: "question".to_string(), + timestamp: 1, + metadata: None, + }, + ); + turn.status = status; + turn + }; + let turns = vec![ + turn("turn-settled", TurnStatus::Cancelled), + turn("turn-active", TurnStatus::InProgress), + ]; + + validate_required_lineage_turns_settled(&turns, &["turn-settled".to_string()]) + .expect("terminal turn should satisfy the barrier"); + for required in ["turn-active", "turn-missing"] { + let error = validate_required_lineage_turns_settled(&turns, &[required.to_string()]) + .expect_err("non-terminal or absent turns must keep the read uncertain"); + assert_eq!( + error.kind, + bitfun_runtime_ports::PortErrorKind::OutcomeUnknown + ); + } + } + + #[test] + fn post_admission_cancellation_errors_are_outcome_unknown() { + for source_error in [ + crate::util::errors::BitFunError::Timeout("drain deadline".to_string()), + crate::util::errors::BitFunError::Session("state persistence failed".to_string()), + ] { + let error = + lineage_post_admission_cancellation_error(source_error, "session-1", "turn-1"); + + assert!(matches!( + error, + crate::util::errors::BitFunError::OutcomeUnknown(message) + if message.contains("session_id=session-1") + && message.contains("turn_id=turn-1") + )); + } + } + + #[tokio::test] + async fn post_admission_state_write_failure_still_delivers_all_cancellation_signals() { + let (coordinator, session_manager) = test_persistent_coordinator(); + let workspace = tempfile::tempdir().expect("workspace"); + let session_id = format!("lineage-cancel-{}", uuid::Uuid::new_v4()); + let turn_id = format!("turn-{}", uuid::Uuid::new_v4()); + session_manager + .create_session_with_id( + Some(session_id.clone()), + "Cancellation fault".to_string(), + "agentic".to_string(), + SessionConfig { + workspace_path: Some(workspace.path().to_string_lossy().into_owned()), + ..Default::default() + }, + ) + .await + .expect("create persistent session"); + session_manager + .update_session_state( + &session_id, + SessionState::Processing { + current_turn_id: turn_id.clone(), + phase: ProcessingPhase::ToolCalling, + }, + ) + .await + .expect("mark turn active"); + let storage_path = session_manager + .effective_session_storage_path(&session_id) + .await + .expect("session storage path"); + + let engine_token = CancellationToken::new(); + coordinator + .execution_engine + .register_cancel_token(&turn_id, engine_token.clone()); + + let tool_id = format!("tool-{}", uuid::Uuid::new_v4()); + coordinator + .tool_pipeline + .insert_tool_task_for_test(ToolTask::new( + ToolCall { + tool_id: tool_id.clone(), + tool_name: "Read".to_string(), + arguments: serde_json::json!({}), + ..Default::default() + }, + ToolExecutionContext { + session_id: session_id.clone(), + dialog_turn_id: turn_id.clone(), + round_id: "round-1".to_string(), + attempt_id: None, + attempt_index: None, + agent_type: "agentic".to_string(), + workspace: None, + primary_model_facts: Default::default(), + context_vars: HashMap::new(), + subagent_parent_info: None, + permission_delegation: None, + delegation_policy: DelegationPolicy::top_level(), + deferred_tools: Vec::new(), + loaded_deferred_tool_specs: Vec::new(), + allowed_tools: Vec::new(), + runtime_tool_restrictions: Default::default(), + steering_interrupt: None, + workspace_services: None, + terminal_port: None, + remote_exec_port: None, + }, + ToolExecutionOptions::default(), + )) + .await; + + let descendant_token = CancellationToken::new(); + coordinator.active_subagent_executions.insert( + "child-session".to_string(), + ActiveSubagentExecution { + parent_session_id: session_id.clone(), + parent_dialog_turn_id: turn_id.clone(), + subagent_session_id: "child-session".to_string(), + subagent_dialog_turn_id: "child-turn".to_string(), + cancel_token: descendant_token.clone(), + }, + ); + session_manager + .persistence_manager() + .fail_next_session_state_write_for_test(&session_id); + + let error = coordinator + .cancel_loaded_lineage_session_in_storage( + &storage_path, + &session_id, + Some(&turn_id), + Duration::from_secs(1), + ) + .await + .expect_err("admitted persistence failure must remain outcome-unknown"); + + assert!(matches!( + error, + crate::util::errors::BitFunError::OutcomeUnknown(message) + if message.contains("Injected session state write failure") + )); + assert!(engine_token.is_cancelled()); + assert!( + coordinator + .tool_pipeline + .tool_task_is_cancelled_for_test(&tool_id), + "tool cancellation must run before the state write error is returned" + ); + assert!(descendant_token.is_cancelled()); + } + #[test] fn runtime_session_list_preserves_the_runtime_owned_model_selector() { let summary = runtime_session_summary(bitfun_agent_runtime::session::SessionSummary { diff --git a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs index 6e9850848d..1e05b03ca1 100644 --- a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs +++ b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs @@ -1295,10 +1295,15 @@ impl DialogScheduler { &self, storage_path: &Path, request: SessionTranscriptRequest, + required_settled_turn_ids: &[String], ) -> PortResult> { let _operation_guard = self.lock_session_operation(&request.session_id).await; self.coordinator - .inspect_loaded_lineage_session_in_storage(storage_path, request) + .inspect_loaded_lineage_session_in_storage( + storage_path, + request, + required_settled_turn_ids, + ) .await } @@ -1306,12 +1311,27 @@ impl DialogScheduler { &self, storage_path: &Path, session_id: &str, + expected_active_turn_id: Option<&str>, wait_timeout: Duration, ) -> BitFunResult> { - let _operation_guard = self.lock_session_operation(session_id).await; - abort_thread_goal_continuation_for_session(session_id); + let deadline = Instant::now() + wait_timeout; + let _operation_guard = tokio::time::timeout( + wait_timeout, + self.lock_session_operation(session_id), + ) + .await + .map_err(|_| { + BitFunError::Timeout(format!( + "Timed out acquiring the Session operation lock before lineage cancellation: session_id={session_id}" + )) + })?; self.coordinator - .cancel_loaded_lineage_session_in_storage(storage_path, session_id, wait_timeout) + .cancel_loaded_lineage_session_in_storage( + storage_path, + session_id, + expected_active_turn_id, + deadline.saturating_duration_since(Instant::now()), + ) .await } diff --git a/src/crates/assembly/core/src/agentic/persistence/manager.rs b/src/crates/assembly/core/src/agentic/persistence/manager.rs index 8fc2bbab2a..53fcad12ad 100644 --- a/src/crates/assembly/core/src/agentic/persistence/manager.rs +++ b/src/crates/assembly/core/src/agentic/persistence/manager.rs @@ -1532,6 +1532,17 @@ impl PersistenceManager { session_id: &str, state: &StoredSessionStateFile, ) -> BitFunResult<()> { + #[cfg(test)] + { + let mut fault = self + .fail_next_session_state_write + .lock() + .expect("session state fault lock"); + if fault.as_deref() == Some(session_id) { + *fault = None; + return Err(BitFunError::io("Injected session state write failure")); + } + } self.write_json_atomic(&self.state_path(workspace_path, session_id), state) .await } @@ -2113,17 +2124,6 @@ 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 } diff --git a/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs b/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs index ada90de691..4719a5a7aa 100644 --- a/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs +++ b/src/crates/assembly/core/src/agentic/tools/pipeline/tool_pipeline.rs @@ -2425,6 +2425,18 @@ impl ToolPipeline { ); Ok(()) } + + #[cfg(test)] + pub(crate) async fn insert_tool_task_for_test(&self, task: ToolTask) { + self.state_manager.create_task(task).await; + } + + #[cfg(test)] + pub(crate) fn tool_task_is_cancelled_for_test(&self, tool_id: &str) -> bool { + self.state_manager + .get_task(tool_id) + .is_some_and(|task| matches!(task.state, ToolExecutionState::Cancelled { .. })) + } } #[cfg(test)] diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index 924e817de1..3b01ef44da 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -33,11 +33,12 @@ use bitfun_runtime_services::RuntimeServices; use bitfun_services_core::permission_store::ProjectPermissionSqliteStore; use bitfun_services_core::session::{ build_session_lineage_snapshot, normalized_session_relationship, SessionBranchBoundary, + SessionRelationshipKind, }; use crate::agentic::coordination::{ - runtime_transcript_messages_from_turns, ConversationCoordinator, DialogScheduler, - SessionMaintenancePermit, + runtime_transcript_messages_from_turns, validate_required_lineage_turns_settled, + ConversationCoordinator, DialogScheduler, SessionMaintenancePermit, }; use crate::agentic::core::Session; use crate::agentic::events::EventQueue; @@ -1329,24 +1330,13 @@ impl CoreSessionOperationsPort { root_session_id: &str, session_id: &str, ) -> PortResult<()> { - let metadata = self - .persistence - .list_session_metadata_including_internal(storage_path) - .await - .map_err(runtime_port_error)?; - let snapshot = build_session_lineage_snapshot(metadata, session_id).ok_or_else(|| { - PortError::new( - PortErrorKind::NotFound, - format!("Session lineage target was not found: {session_id}"), - ) - })?; - if snapshot.root_session_id != root_session_id || session_id == root_session_id { - return Err(PortError::new( - PortErrorKind::InvalidRequest, - "Lineage target is not a descendant of the requested root", - )); - } - Ok(()) + validate_persisted_lineage_descendant( + self.persistence.as_ref(), + storage_path, + root_session_id, + session_id, + ) + .await } async fn fork_at_persisted_turn( @@ -1471,6 +1461,77 @@ impl CoreSessionOperationsPort { } } +async fn validate_persisted_lineage_descendant( + persistence: &PersistenceManager, + storage_path: &Path, + root_session_id: &str, + session_id: &str, +) -> PortResult<()> { + if session_id == root_session_id { + return Err(PortError::new( + PortErrorKind::InvalidRequest, + "Lineage target is not a descendant of the requested root", + )); + } + + let mut current_session_id = session_id.to_string(); + let mut visited = std::collections::HashSet::new(); + loop { + if !visited.insert(current_session_id.clone()) { + return Err(PortError::new( + PortErrorKind::InvalidRequest, + "Session lineage contains a parent cycle", + )); + } + let metadata = persistence + .load_session_metadata(storage_path, ¤t_session_id) + .await + .map_err(runtime_port_error)? + .ok_or_else(|| { + PortError::new( + if current_session_id == session_id { + PortErrorKind::NotFound + } else { + PortErrorKind::InvalidRequest + }, + format!("Session lineage entry was not found: {current_session_id}"), + ) + })?; + let parent_session_id = normalized_session_relationship(&metadata) + .filter(|relationship| relationship.kind == Some(SessionRelationshipKind::Subagent)) + .and_then(|relationship| relationship.parent_session_id) + .map(|parent_session_id| parent_session_id.trim().to_string()) + .filter(|parent_session_id| !parent_session_id.is_empty()); + if current_session_id == root_session_id { + let Some(parent_session_id) = parent_session_id else { + return Ok(()); + }; + // Match `build_session_lineage_snapshot`: a broken parent link + // makes the current entry the effective root, while an existing + // parent proves that the requested root is only an intermediate + // descendant. + if persistence + .load_session_metadata(storage_path, &parent_session_id) + .await + .map_err(runtime_port_error)? + .is_none() + { + return Ok(()); + } + return Err(PortError::new( + PortErrorKind::InvalidRequest, + "Lineage target is not a descendant of the requested root", + )); + } + current_session_id = parent_session_id.ok_or_else(|| { + PortError::new( + PortErrorKind::InvalidRequest, + "Lineage target is not a descendant of the requested root", + ) + })?; + } +} + fn runtime_port_error(error: BitFunError) -> PortError { let kind = match &error { BitFunError::Validation(_) => PortErrorKind::InvalidRequest, @@ -1572,6 +1633,7 @@ impl AgentSessionLineagePort for CoreSessionOperationsPort { session_id: request.session_id.clone(), turn_id: None, }, + &request.required_settled_turn_ids, ) .await? { @@ -1585,6 +1647,7 @@ impl AgentSessionLineagePort for CoreSessionOperationsPort { ) .await .map_err(runtime_port_error)?; + validate_required_lineage_turns_settled(&turns, &request.required_settled_turn_ids)?; Ok(AgentSessionLineageInspection { transcript: SessionTranscript { session_id: request.session_id, @@ -1598,29 +1661,45 @@ impl AgentSessionLineagePort for CoreSessionOperationsPort { &self, request: AgentSessionLineageCancellationRequest, ) -> PortResult { - validate_persisted_session_id(&request.root_session_id).map_err(runtime_port_error)?; - validate_persisted_session_id(&request.session_id).map_err(runtime_port_error)?; - let storage_path = self - .resolve_session_storage_path( - request.workspace_path, - request.remote_connection_id, - request.remote_ssh_host, - ) - .await?; - self.validate_lineage_descendant( - &storage_path, - &request.root_session_id, - &request.session_id, - ) - .await?; - - let session_id = request.session_id; + let AgentSessionLineageCancellationRequest { + workspace_path, + root_session_id, + session_id, + expected_active_turn_id, + wait_timeout_ms, + remote_connection_id, + remote_ssh_host, + .. + } = request; + validate_persisted_session_id(&root_session_id).map_err(runtime_port_error)?; + validate_persisted_session_id(&session_id).map_err(runtime_port_error)?; + let wait_timeout = Duration::from_millis(wait_timeout_ms.unwrap_or(1500)); + let deadline = Instant::now() + wait_timeout; + let storage_path = match tokio::time::timeout(wait_timeout, async { + let storage_path = self + .resolve_session_storage_path(workspace_path, remote_connection_id, remote_ssh_host) + .await?; + self.validate_lineage_descendant(&storage_path, &root_session_id, &session_id) + .await?; + Ok(storage_path) + }) + .await + { + Ok(result) => result?, + Err(_) => { + return Err(PortError::new( + PortErrorKind::Timeout, + "Subagent Session cancellation validation exceeded its deadline", + )) + } + }; let cancelled_turn_id = self .scheduler .cancel_lineage_session_in_storage( &storage_path, &session_id, - Duration::from_millis(request.wait_timeout_ms.unwrap_or(1500)), + expected_active_turn_id.as_deref(), + deadline.saturating_duration_since(Instant::now()), ) .await .map_err(runtime_port_error)?; @@ -2100,16 +2179,131 @@ mod tests { assert!(lineage_impl.contains("cancel_lineage_session_in_storage")); assert!(lineage_impl.contains("active_turn_id_in_storage_path")); assert!(!lineage_impl.contains("SessionTranscriptReader::read_session_transcript")); + let cancellation = lineage_impl + .split("async fn cancel_lineage_session") + .nth(1) + .expect("lineage cancellation implementation"); + let (cancellation_preparation, admitted_cancellation) = cancellation + .split_once("let cancelled_turn_id") + .expect("lineage cancellation admission boundary"); + assert!(cancellation_preparation.contains("tokio::time::timeout(wait_timeout")); + assert!(admitted_cancellation.contains("cancel_lineage_session_in_storage")); + assert!( + admitted_cancellation.contains("deadline.saturating_duration_since(Instant::now())") + ); + assert!(!admitted_cancellation.contains("tokio::time::timeout")); + assert!( + cancellation_preparation + .find("tokio::time::timeout") + .unwrap() + < cancellation_preparation + .find("validate_lineage_descendant") + .unwrap(), + "the caller wait budget must include lineage validation" + ); let product_runtime_source = product_source .split("#[cfg(test)]") .next() .expect("production product runtime source"); + let lineage_validation = product_runtime_source + .split("async fn validate_persisted_lineage_descendant") + .nth(1) + .and_then(|source| source.split("#[async_trait::async_trait]").next()) + .expect("targeted persisted lineage validation"); + assert!(lineage_validation.contains("load_session_metadata")); + assert!(lineage_validation.contains("normalized_session_relationship")); + assert!(!lineage_validation.contains("list_session_metadata_including_internal")); assert!(!product_runtime_source.contains("fn lineage_active_turn_id(")); assert!(coordinator_source.contains("TurnStatus::InProgress")); assert!(coordinator_source .contains("Session turn settlement changed while its transcript was being inspected")); assert!(coordinator_source .contains("Session turn is still settling after its active state changed")); + + let inspect_body = coordinator_source + .split("fn inspect_loaded_lineage_session_in_storage") + .nth(1) + .and_then(|source| source.split("\n }").next()) + .expect("loaded lineage inspection body"); + let required_settlement_gate = inspect_body + .find("required_settled_turn_ids") + .expect("required settlement consistency gate"); + let transcript_read = inspect_body + .find("read_session_transcript_with_turn_status_locked") + .expect("authoritative transcript read"); + assert!( + required_settlement_gate < transcript_read, + "unsettled required Turns must return before transcript I/O" + ); + + let signal_body = coordinator_source + .split("fn signal_active_subagent_cancellation") + .nth(1) + .and_then(|source| source.split("\n }").next()) + .expect("subagent cancellation signal body"); + assert!(signal_body.contains("cancel_token.cancel()")); + assert!(!signal_body.contains("abort_handle.abort()")); + assert!(!signal_body.contains("persist_cancelled_dialog_turn")); + } + + #[tokio::test] + async fn targeted_lineage_validation_rejects_an_intermediate_root() { + let workspace = TestWorkspace::new(); + let persistence = + PersistenceManager::new(workspace.path_manager()).expect("persistence manager"); + let storage_path = workspace.path().join("sessions"); + std::fs::create_dir_all(&storage_path).expect("session storage"); + + let root = SessionMetadata::new( + "root".to_string(), + "Root".to_string(), + "agentic".to_string(), + "model".to_string(), + ); + let mut child = SessionMetadata::new( + "child".to_string(), + "Child".to_string(), + "explore".to_string(), + "model".to_string(), + ); + child.custom_metadata = Some(serde_json::json!({ + "kind": "subagent", + "parentSessionId": "root" + })); + let mut grandchild = SessionMetadata::new( + "grandchild".to_string(), + "Grandchild".to_string(), + "explore".to_string(), + "model".to_string(), + ); + grandchild.custom_metadata = Some(serde_json::json!({ + "kind": "subagent", + "parentSessionId": "child" + })); + for metadata in [&root, &child, &grandchild] { + persistence + .save_session_metadata(&storage_path, metadata) + .await + .expect("save lineage metadata"); + } + + super::validate_persisted_lineage_descendant( + &persistence, + &storage_path, + "root", + "grandchild", + ) + .await + .expect("actual root should validate"); + let error = super::validate_persisted_lineage_descendant( + &persistence, + &storage_path, + "child", + "grandchild", + ) + .await + .expect_err("intermediate root must be rejected"); + assert_eq!(error.kind, PortErrorKind::InvalidRequest); } #[tokio::test] diff --git a/src/crates/contracts/runtime-ports/src/lib.rs b/src/crates/contracts/runtime-ports/src/lib.rs index 200e1c6d10..83d82a21bd 100644 --- a/src/crates/contracts/runtime-ports/src/lib.rs +++ b/src/crates/contracts/runtime-ports/src/lib.rs @@ -2344,6 +2344,11 @@ pub struct AgentSessionLineageTranscriptRequest { pub workspace_path: String, pub root_session_id: String, pub session_id: String, + /// Terminal Turns that an authoritative read must prove settled. This is a + /// read-consistency precondition; settlement and transcript construction + /// remain Runtime-owned. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub required_settled_turn_ids: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] pub remote_connection_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -2364,6 +2369,10 @@ pub struct AgentSessionLineageCancellationRequest { pub workspace_path: String, pub root_session_id: String, pub session_id: String, + /// Exact Turn shown as active when the user requested interruption. The + /// Runtime rejects a changed target instead of cancelling a newer Turn. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub expected_active_turn_id: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub source: Option, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -4003,6 +4012,7 @@ mod tests { workspace_path: "/workspace/project".to_string(), root_session_id: "root_1".to_string(), session_id: "child_1".to_string(), + required_settled_turn_ids: vec!["turn-terminal".to_string()], remote_connection_id: None, remote_ssh_host: None, }; @@ -4010,6 +4020,7 @@ mod tests { workspace_path: "/workspace/project".to_string(), root_session_id: "root_1".to_string(), session_id: "child_1".to_string(), + expected_active_turn_id: Some("turn-live".to_string()), source: Some(AgentSubmissionSource::Cli), reason: Some("user_cancelled".to_string()), wait_timeout_ms: Some(5_000), @@ -4044,8 +4055,29 @@ mod tests { ); assert_eq!(transcript_json["sessionId"], "child_1"); assert_eq!(transcript_json["rootSessionId"], "root_1"); + assert_eq!( + transcript_json["requiredSettledTurnIds"], + serde_json::json!(["turn-terminal"]) + ); + let legacy_transcript = + serde_json::from_value::(serde_json::json!({ + "workspacePath": "/workspace/project", + "rootSessionId": "root_1", + "sessionId": "child_1" + })) + .expect("deserialize precondition-free lineage transcript request"); + assert!(legacy_transcript.required_settled_turn_ids.is_empty()); assert_eq!(cancellation_json["source"], "cli"); assert_eq!(cancellation_json["waitTimeoutMs"], 5_000); + assert_eq!(cancellation_json["expectedActiveTurnId"], "turn-live"); + let legacy_cancellation = + serde_json::from_value::(serde_json::json!({ + "workspacePath": "/workspace/project", + "rootSessionId": "root_1", + "sessionId": "child_1" + })) + .expect("deserialize pre-exact-target lineage cancellation request"); + assert!(legacy_cancellation.expected_active_turn_id.is_none()); assert_eq!(inspection_json["transcript"]["sessionId"], "child_1"); assert_eq!(inspection_json["activeTurnId"], "turn-live"); } diff --git a/src/crates/execution/agent-runtime/src/runtime.rs b/src/crates/execution/agent-runtime/src/runtime.rs index f5280c7384..65c36be6e4 100644 --- a/src/crates/execution/agent-runtime/src/runtime.rs +++ b/src/crates/execution/agent-runtime/src/runtime.rs @@ -2108,7 +2108,7 @@ mod tests { self, AgentTurnCancellationRequest { session_id: request.session_id, - turn_id: None, + turn_id: request.expected_active_turn_id, source: request.source, requester_session_id: None, reason: request.reason, @@ -2980,6 +2980,7 @@ mod tests { workspace_path: "/workspace/project".to_string(), root_session_id: "root_1".to_string(), session_id: "child_1".to_string(), + required_settled_turn_ids: Vec::new(), remote_connection_id: None, remote_ssh_host: None, }) @@ -3007,6 +3008,7 @@ mod tests { workspace_path: "/workspace/project".to_string(), root_session_id: "root_1".to_string(), session_id: "child_1".to_string(), + expected_active_turn_id: Some("child_turn".to_string()), source: Some(AgentSubmissionSource::Cli), reason: Some("user_cancelled".to_string()), wait_timeout_ms: Some(5_000), @@ -3028,6 +3030,7 @@ mod tests { workspace_path: "/workspace/project".to_string(), root_session_id: "root_1".to_string(), session_id: "outside".to_string(), + expected_active_turn_id: Some("outside_turn".to_string()), source: None, reason: None, wait_timeout_ms: None, diff --git a/src/crates/services/services-integrations/src/git/managed_worktree.rs b/src/crates/services/services-integrations/src/git/managed_worktree.rs index 0e7d8b927c..fbe33b1df7 100644 --- a/src/crates/services/services-integrations/src/git/managed_worktree.rs +++ b/src/crates/services/services-integrations/src/git/managed_worktree.rs @@ -588,8 +588,11 @@ mod tests { assert!(summary.untracked.contains(&"untracked.txt".to_string())); assert!(summary.included_ignored.contains(&"secret.env".to_string())); assert_eq!( - fs::read_to_string(target.join("shared.txt")).unwrap(), - "staged\nunstaged\n" + fs::read_to_string(target.join("shared.txt")) + .unwrap() + .lines() + .collect::>(), + vec!["staged", "unstaged"] ); assert_eq!( fs::read(target.join("binary.dat")).unwrap(), diff --git a/src/crates/services/services-integrations/src/review_platform.rs b/src/crates/services/services-integrations/src/review_platform.rs index d811609006..dcd87661cb 100644 --- a/src/crates/services/services-integrations/src/review_platform.rs +++ b/src/crates/services/services-integrations/src/review_platform.rs @@ -8041,10 +8041,14 @@ mod tests { .await .expect("git init should succeed for repository root test"); - let expected = normalize_repository_root( - root.to_str() - .expect("temporary repository path should be valid UTF-8"), - ); + let root_text = root + .to_str() + .expect("temporary repository path should be valid UTF-8"); + // Git for Windows reports a normal drive path, while + // std::fs::canonicalize uses the equivalent verbatim path form. + #[cfg(windows)] + let root_text = root_text.strip_prefix(r"\\?\").unwrap_or(root_text); + let expected = normalize_repository_root(root_text); let nested_root = get_repository_root( nested .to_str()