From 2c4f561b8e2c166fb3ee51d5c54e220e3b63f324 Mon Sep 17 00:00:00 2001 From: kev1n77 Date: Fri, 16 Oct 2026 19:19:26 +0800 Subject: [PATCH 1/6] perf(sessions): reuse external history reads and workspace ownership --- .../external-ai-work-sources-design.md | 5 +- .../core/src/external_session_import.rs | 41 +- .../external_session_import/auto_discovery.rs | 202 +++++++- .../src/external_session_import/deferred.rs | 9 +- .../deferred/legacy.rs | 8 +- .../deferred/test_io.rs | 43 ++ .../deferred/tracking.rs | 199 ++++++-- .../deferred/tracking/refresh_tests.rs | 437 ++++++++++++++++++ 8 files changed, 874 insertions(+), 70 deletions(-) create mode 100644 src/crates/assembly/core/src/external_session_import/deferred/test_io.rs create mode 100644 src/crates/assembly/core/src/external_session_import/deferred/tracking/refresh_tests.rs diff --git a/docs/architecture/extensions/external-ai-work-sources-design.md b/docs/architecture/extensions/external-ai-work-sources-design.md index 6f301da6cb..b27dcd1949 100644 --- a/docs/architecture/extensions/external-ai-work-sources-design.md +++ b/docs/architecture/extensions/external-ai-work-sources-design.md @@ -97,7 +97,8 @@ Zcode 会话读取不意味着其 ACP 运行时、设置、Skill、MCP 或 Hook 工作区归属按来源目录匹配最近的已登记本地工作区。后台任务不依赖控制端在线；会话元数据分页 通过可选的 `discoveryPending` / `discoveryWarnings` 投影进度和非致命错误，已有原生会话始终可读。 服务主机合并同一批工作区打开请求的来源目录扫描，Codex 读取器按文件大小和修改时间复用已解析的 -有界头部，不重复读取未变化文件。启动仅将会话标题、来源和归属写入集中会话目录的 +有界头部，不重复读取未变化文件。同批来源目录按不同路径计算并共享工作区归属；工作区登记根变化或 +下一次目录扫描时重新计算，保留最近本地工作区及同根登记顺序的匹配规则。启动仅将会话标题、来源和归属写入集中会话目录的 `.external-session-index.json`；该私有索引由现有持久化入口合并到元数据分页，旧主机忽略此文件， 既有原生 Session 格式不变。改名、归档和删除只更新索引；删除保留抑制记录，重新扫描不复活条目。 源会话的移除动作显示为“从 OpenBitFun 移除”，不修改外部原始记录或已有原生分支。生态兼容页 @@ -114,6 +115,8 @@ Zcode 会话读取不意味着其 ACP 运行时、设置、Skill、MCP 或 Hook 来源追加、回退或重写生成新版本。缓存缺失时来源失败明确报错；已有缓存则保留并标记来源不可用， 不发布空历史、不删除条目。元数据及工作区正文搜索不读取来源正文；正文搜索排除跟随视图并返回说明。 Web UI 每次打开来源会话时读取一次来源；停留、输入、窗口恢复可见及查看原生分支都不触发刷新。 +正文版本相同时复用已经校验的快照，刷新结果直接交给本次读取；索引条目未变化时不写盘或通知目录变更。 +旧原生副本仍执行本地编辑校验，来源不可用、标题变化以及并发改名、归档和移除仍按既有规则处理。 首次发送使用现有分叉入口复制已显示版本并切换到原生分支，之后使用主机当前模型与权限配置。 分支以 `externalBranchOf` 保留来源及版本，来源仍可独立更新，后续内容不自动混入分支。 编辑或回退来源历史须先显式分叉；共享持久化及运行时入口拒绝直接修改跟随视图。 diff --git a/src/crates/assembly/core/src/external_session_import.rs b/src/crates/assembly/core/src/external_session_import.rs index 4882f54af9..dc73bd6016 100644 --- a/src/crates/assembly/core/src/external_session_import.rs +++ b/src/crates/assembly/core/src/external_session_import.rs @@ -193,6 +193,33 @@ fn convert_document( include_tool_history: bool, include_thinking: bool, ) -> ExternalSourceOperationResult { + let mut metadata = document_metadata( + session_id, + workspace, + project, + agent_id, + session_name, + document, + )?; + let turns = convert_document_turns( + session_id, + agent_id, + document, + include_tool_history, + include_thinking, + ); + finish_document_metadata(&mut metadata, &turns); + Ok(OfflineSessionBundle { metadata, turns }) +} + +fn document_metadata( + session_id: &str, + workspace: &crate::service::workspace::WorkspaceInfo, + project: &crate::service::workspace::WorkspaceInfo, + agent_id: &str, + session_name: &str, + document: &ExternalSessionDocument, +) -> ExternalSourceOperationResult { // The source model is retained on the imported rounds and provenance // metadata, while continuation uses the target host's configured primary // model instead of assuming the source model exists locally. @@ -223,7 +250,16 @@ fn convert_document( "sourceModel": document.candidate.model_name, "continuationMode": "openbitfun_local" })); + Ok(metadata) +} +fn convert_document_turns( + session_id: &str, + agent_id: &str, + document: &ExternalSessionDocument, + include_tool_history: bool, + include_thinking: bool, +) -> Vec { let mut turns = Vec::with_capacity(document.turns.len()); for (turn_index, source) in document.turns.iter().enumerate() { let turn_id = format!("{session_id}-turn-{turn_index}"); @@ -390,6 +426,10 @@ fn convert_document( }; turns.push(turn); } + turns +} + +fn finish_document_metadata(metadata: &mut SessionMetadata, turns: &[DialogTurnData]) { metadata.turn_count = turns.len(); metadata.message_count = turns.len() + turns @@ -407,7 +447,6 @@ fn convert_document( .and_then(|turn| turn.end_time) .unwrap_or_else(current_time_ms); metadata.last_finished_at = Some(metadata.last_active_at); - Ok(OfflineSessionBundle { metadata, turns }) } fn current_time_ms() -> u64 { diff --git a/src/crates/assembly/core/src/external_session_import/auto_discovery.rs b/src/crates/assembly/core/src/external_session_import/auto_discovery.rs index dd6881e57b..1f05a7ffca 100644 --- a/src/crates/assembly/core/src/external_session_import/auto_discovery.rs +++ b/src/crates/assembly/core/src/external_session_import/auto_discovery.rs @@ -262,7 +262,105 @@ async fn discover_enabled( warnings } -type SourceCatalogs = tokio::sync::Mutex)>>; +#[derive(Debug)] +struct SourceCatalog { + catalog: ExternalSessionCatalog, + ownership: Mutex>, + #[cfg(test)] + ownership_resolutions: std::sync::atomic::AtomicUsize, +} + +#[derive(Debug)] +struct CatalogOwnership { + roots: Vec<(String, PathBuf)>, + candidates: HashMap>, +} + +impl SourceCatalog { + fn new(catalog: ExternalSessionCatalog) -> Self { + Self { + catalog, + ownership: Mutex::new(None), + #[cfg(test)] + ownership_resolutions: std::sync::atomic::AtomicUsize::new(0), + } + } + + fn workspace_candidates( + &self, + workspace: &WorkspaceInfo, + workspaces: &[WorkspaceInfo], + mut normalize: impl FnMut(&Path) -> String, + ) -> Vec { + if workspace.workspace_kind == WorkspaceKind::Remote { + return Vec::new(); + } + // Registry order is significant when two registered roots are equal: + // matching_workspace selects the last equally specific owner. + let roots = workspaces + .iter() + .filter(|workspace| workspace.workspace_kind != WorkspaceKind::Remote) + .map(|workspace| (workspace.id.clone(), workspace.root_path.clone())) + .collect::>(); + let mut ownership = self + .ownership + .lock() + .unwrap_or_else(|error| error.into_inner()); + if ownership + .as_ref() + .is_none_or(|cached| cached.roots != roots) + { + let mut normalized = HashMap::::new(); + let mut normalize_once = |path: &Path| { + normalized + .entry(path.to_path_buf()) + .or_insert_with(|| normalize(path)) + .clone() + }; + let normalized_roots = roots + .iter() + .map(|(id, root)| (id, normalize_once(root))) + .collect::>(); + let mut owners = HashMap::<&str, Option<&String>>::new(); + let mut candidates = HashMap::>::new(); + for (index, candidate) in self.catalog.candidates.iter().enumerate() { + let Some(cwd) = candidate.cwd.as_deref() else { + continue; + }; + let owner = owners.entry(cwd).or_insert_with(|| { + #[cfg(test)] + self.ownership_resolutions.fetch_add(1, Ordering::Relaxed); + let source = normalize_once(Path::new(cwd)); + normalized_roots + .iter() + .filter(|(_, root)| { + source == *root + || source + .strip_prefix(root.as_str()) + .is_some_and(|suffix| suffix.starts_with('/')) + }) + .max_by_key(|(_, root)| root.len()) + .map(|(id, _)| *id) + }); + if let Some(owner) = owner { + candidates.entry((*owner).clone()).or_default().push(index); + } + } + *ownership = Some(CatalogOwnership { roots, candidates }); + } + ownership + .as_ref() + .unwrap() + .candidates + .get(&workspace.id) + .into_iter() + .flatten() + .map(|index| self.catalog.candidates[*index].clone()) + .collect() + } +} + +type SourceCatalogs = tokio::sync::Mutex)>>; static SOURCE_CATALOGS: LazyLock = LazyLock::new(Default::default); async fn shared_catalog( @@ -270,7 +368,7 @@ async fn shared_catalog( ecosystem: &str, requested_at: Instant, read: impl FnOnce() -> Result + Send + 'static, -) -> Result, String> { +) -> Result, String> { // Serialize source scans across workspaces. Requests already queued when a // scan finishes share that snapshot; a later workspace activation refreshes it. let mut cache = cache.lock().await; @@ -279,11 +377,11 @@ async fn shared_catalog( return Ok(catalog.clone()); } } - let catalog = Arc::new( + let catalog = Arc::new(SourceCatalog::new( tokio::task::spawn_blocking(read) .await .map_err(|error| error.to_string())??, - ); + )); cache.insert(ecosystem.into(), (Instant::now(), catalog.clone())); Ok(catalog) } @@ -293,25 +391,20 @@ async fn register_catalog( workspace: &WorkspaceInfo, project: &WorkspaceInfo, workspaces: &[WorkspaceInfo], - catalog: &ExternalSessionCatalog, + catalog: &SourceCatalog, cancelled: &AtomicBool, ) -> Vec { if cancelled.load(Ordering::Acquire) { return Vec::new(); } let mut warnings = Vec::new(); - if !catalog.diagnostics.is_empty() { + if !catalog.catalog.diagnostics.is_empty() { warnings.push(format!( "Some {} source records could not be read", - catalog.ecosystem_id + catalog.catalog.ecosystem_id )); } - let candidates = catalog - .candidates - .iter() - .filter(|candidate| belongs_to_workspace(candidate, workspace, workspaces)) - .cloned() - .collect(); + let candidates = catalog.workspace_candidates(workspace, workspaces, normalized_path); if let Err(error) = super::deferred::register(store.sessions_root(), workspace, project, candidates).await { @@ -547,7 +640,7 @@ pub(super) mod tests { selected[0], &project, &workspaces, - &discovered, + &SourceCatalog::new(discovered), &AtomicBool::new(false), ) .await; @@ -673,7 +766,7 @@ pub(super) mod tests { &workspace, &workspace, &[workspace.clone()], - &catalog("codex", vec![candidate("codex", "a", temp.path())]), + &SourceCatalog::new(catalog("codex", vec![candidate("codex", "a", temp.path())])), &AtomicBool::new(true), ) .await; @@ -702,4 +795,83 @@ pub(super) mod tests { &sources )); } + + #[test] + fn catalog_ownership_normalizes_unique_paths_once_for_all_workspaces() { + let temp = tempfile::tempdir().unwrap(); + let root = workspace("root", temp.path(), WorkspaceKind::Normal); + let nested = workspace("nested", &temp.path().join("nested"), WorkspaceKind::Normal); + let remote = workspace("remote", &nested.root_path, WorkspaceKind::Remote); + let workspaces = vec![root.clone(), nested.clone(), remote.clone()]; + let mut candidates = Vec::new(); + for index in 0..200 { + let cwd = if index % 2 == 0 { + root.root_path.join("src") + } else { + nested.root_path.join("src") + }; + candidates.push(candidate("codex", &index.to_string(), &cwd)); + } + candidates.push(candidate( + "codex", + "sibling", + &temp.path().with_extension("sibling"), + )); + let mut missing = candidate("codex", "missing", temp.path()); + missing.cwd = None; + candidates.push(missing); + let source = SourceCatalog::new(catalog("codex", candidates)); + let mut normalized = Vec::new(); + for workspace in &workspaces { + let actual = source.workspace_candidates(workspace, &workspaces, |path| { + normalized.push(path.to_path_buf()); + normalized_path(path) + }); + let expected = source + .catalog + .candidates + .iter() + .filter(|candidate| belongs_to_workspace(candidate, workspace, &workspaces)) + .map(|candidate| &candidate.candidate_id) + .collect::>(); + assert_eq!( + actual + .iter() + .map(|candidate| &candidate.candidate_id) + .collect::>(), + expected + ); + } + assert_eq!(normalized.len(), 5, "Two local roots and three distinct source directories, shared across all workspace readers"); + assert_eq!( + normalized + .iter() + .collect::>() + .len(), + 5 + ); + + assert_eq!( + source.ownership_resolutions.load(Ordering::Relaxed), + 3, + "Resolve each distinct source directory once, not once per session or workspace" + ); + let replacement = workspace("replacement", &nested.root_path, WorkspaceKind::Normal); + let changed_registry = vec![root, nested, remote, replacement.clone()]; + let actual = source.workspace_candidates(&replacement, &changed_registry, |path| { + normalized.push(path.to_path_buf()); + normalized_path(path) + }); + assert_eq!( + actual.len(), + 100, + "Registry changes invalidate ownership; the last equal root wins" + ); + assert_eq!( + normalized.len(), + 10, + "A new registry snapshot resolves paths afresh, still once per unique path" + ); + assert_eq!(source.ownership_resolutions.load(Ordering::Relaxed), 6); + } } diff --git a/src/crates/assembly/core/src/external_session_import/deferred.rs b/src/crates/assembly/core/src/external_session_import/deferred.rs index a9658398ea..c8d64b9966 100644 --- a/src/crates/assembly/core/src/external_session_import/deferred.rs +++ b/src/crates/assembly/core/src/external_session_import/deferred.rs @@ -2,6 +2,8 @@ //! reads a source snapshot; navigation never copies transcript bodies. mod legacy; +#[cfg(test)] +mod test_io; pub(crate) mod tracking; pub(crate) use legacy::has_native_view; @@ -154,7 +156,11 @@ async fn update_index( .await .map_err(index_error)?; CACHE.lock().await.remove(&path); + #[cfg(test)] + test_io::record(root, |counts| counts.index_writes += 1); SessionMetadataStore::notify_catalog_changed(); + #[cfg(test)] + test_io::record(root, |counts| counts.catalog_notifications += 1); } Ok(result) } @@ -816,7 +822,8 @@ async fn materialize_reference( return Box::pin(tracking::refresh_reference( root, entry, workspace, project, workspaces, read, )) - .await; + .await + .map(|_| ()); } let store = OfflineSessionImportStore::new(root); let before = store diff --git a/src/crates/assembly/core/src/external_session_import/deferred/legacy.rs b/src/crates/assembly/core/src/external_session_import/deferred/legacy.rs index cf63444fd7..9ec54e0dae 100644 --- a/src/crates/assembly/core/src/external_session_import/deferred/legacy.rs +++ b/src/crates/assembly/core/src/external_session_import/deferred/legacy.rs @@ -202,10 +202,14 @@ pub(super) async fn validate_native( let Some(current) = index.entries.get_mut(&entry.metadata.session_id) else { return ((), false); }; - current + let custom = current .metadata .custom_metadata - .get_or_insert_with(|| json!({}))[HASH] = json!(hash); + .get_or_insert_with(|| json!({})); + if custom[HASH] == hash { + return ((), false); + } + custom[HASH] = json!(hash); ((), true) }) .await?; diff --git a/src/crates/assembly/core/src/external_session_import/deferred/test_io.rs b/src/crates/assembly/core/src/external_session_import/deferred/test_io.rs new file mode 100644 index 0000000000..0612c0de70 --- /dev/null +++ b/src/crates/assembly/core/src/external_session_import/deferred/test_io.rs @@ -0,0 +1,43 @@ +//! Per-store operation probes; concurrent fixtures never share counters. +use super::*; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub(super) struct Counts { + pub source_reads: usize, + pub snapshot_reads: usize, + pub snapshot_writes: usize, + pub conversions: usize, + pub index_writes: usize, + pub catalog_notifications: usize, +} + +static COUNTERS: LazyLock>> = LazyLock::new(Default::default); + +pub(super) fn record(root: &Path, update: impl FnOnce(&mut Counts)) { + if let Some(counts) = COUNTERS.lock().unwrap().get_mut(root) { + update(counts); + } +} + +pub(super) struct Probe(PathBuf); + +impl Probe { + pub(super) fn new(root: &Path) -> Self { + assert!(COUNTERS + .lock() + .unwrap() + .insert(root.to_path_buf(), Counts::default()) + .is_none()); + Self(root.to_path_buf()) + } + + pub(super) fn take(&self) -> Counts { + std::mem::take(COUNTERS.lock().unwrap().get_mut(&self.0).unwrap()) + } +} + +impl Drop for Probe { + fn drop(&mut self) { + COUNTERS.lock().unwrap().remove(&self.0); + } +} diff --git a/src/crates/assembly/core/src/external_session_import/deferred/tracking.rs b/src/crates/assembly/core/src/external_session_import/deferred/tracking.rs index 3e5137cb40..4a0fe83ec0 100644 --- a/src/crates/assembly/core/src/external_session_import/deferred/tracking.rs +++ b/src/crates/assembly/core/src/external_session_import/deferred/tracking.rs @@ -2,6 +2,9 @@ use super::*; +#[cfg(test)] +mod refresh_tests; + const MODE: &str = "externalHistoryMode"; const VERSION: &str = "externalContentVersion"; const SNAPSHOTS: &str = ".external-history"; @@ -78,6 +81,8 @@ async fn read_snapshot( entry: &Reference, version: &str, ) -> OpenBitFunResult { + #[cfg(test)] + super::test_io::record(root, |counts| counts.snapshot_reads += 1); let snapshot: Snapshot = JsonFileStore .read_optional(&snapshot_path(root, &entry.metadata.session_id, version)?) .await @@ -100,6 +105,11 @@ async fn read_snapshot( } snapshot.bundle.validate().map_err(index_error)?; let mut bundle = snapshot.bundle; + overlay_view_metadata(&mut bundle, entry); + Ok(bundle) +} + +fn overlay_view_metadata(bundle: &mut OfflineSessionBundle, entry: &Reference) { // Names and archive state belong to the local view, not to a frozen version. bundle.metadata.session_name = entry.metadata.session_name.clone(); bundle.metadata.tags = entry.metadata.tags.clone(); @@ -118,7 +128,6 @@ async fn read_snapshot( } } } - Ok(bundle) } /// A turn carries its immutable view version. Existing fork requests can pin it @@ -154,18 +163,48 @@ pub(crate) async fn snapshot( let Some(entry) = reference(root, id).await? else { return Ok(None); }; - if refresh || content_version(&entry.metadata).is_none() { - Box::pin(refresh_source(root, entry)).await?; - } + let refreshed = if refresh || content_version(&entry.metadata).is_none() { + Box::pin(refresh_source(root, entry)).await? + } else { + None + }; + load_current_snapshot(root, id, refreshed).await +} + +async fn load_current_snapshot( + root: &Path, + id: &str, + refreshed: Option, +) -> OpenBitFunResult> { let Some(entry) = reference(root, id).await? else { return Ok(None); }; let version = content_version(&entry.metadata) .ok_or_else(|| index_error("The original source has no readable history snapshot"))?; + if let Some(mut bundle) = refreshed.filter(|bundle| { + content_version(&bundle.metadata) == Some(version) + && bundle.metadata.session_id == entry.metadata.session_id + && bundle + .metadata + .custom_metadata + .as_ref() + .is_some_and(|custom| { + custom["importedFrom"] == entry.ecosystem_id + && custom["sourceSessionId"] == entry.source_session_id + }) + }) { + // Recheck the pointer after refreshing: a concurrent remove, rename or + // newer refresh must still win, as it did when this always reread disk. + overlay_view_metadata(&mut bundle, &entry); + return Ok(Some(bundle)); + } Ok(Some(read_snapshot(root, &entry, version).await?)) } -async fn refresh_source(root: &Path, entry: Reference) -> OpenBitFunResult<()> { +async fn refresh_source( + root: &Path, + entry: Reference, +) -> OpenBitFunResult> { let service = crate::service::workspace::get_global_workspace_service() .ok_or_else(|| index_error("Workspace service is unavailable"))?; let workspace_id = entry @@ -248,6 +287,8 @@ async fn cache_unavailable_legacy_history( ) .await .map_err(index_error)?; + #[cfg(test)] + super::test_io::record(root, |counts| counts.snapshot_writes += 1); update_index(root, |index| { let Some(current) = index.entries.get_mut(id).filter(|entry| !entry.dismissed) else { return ((), false); @@ -270,7 +311,7 @@ pub(super) async fn refresh_reference( project: WorkspaceInfo, workspaces: Vec, read: impl FnOnce(ImportSource, &str) -> Result + Send + 'static, -) -> OpenBitFunResult<()> { +) -> OpenBitFunResult> { let id = entry.metadata.session_id.clone(); openbitfun_core_types::validate_session_id(&id).map_err(OpenBitFunError::Validation)?; // The source lock is distinct from native Session writers. A snapshot is @@ -285,6 +326,8 @@ pub(super) async fn refresh_reference( let source = ImportSource::from_ecosystem(&entry.ecosystem_id) .map_err(|error| index_error(error.detail))?; let candidate = entry.candidate_id.clone(); + #[cfg(test)] + super::test_io::record(root, |counts| counts.source_reads += 1); let document = match tokio::task::spawn_blocking(move || read(source, &candidate)) .await .map_err(index_error)? @@ -297,19 +340,21 @@ pub(super) async fn refresh_reference( .metadata .custom_metadata .get_or_insert_with(|| json!({})); + let changed = custom["externalSourceStatus"] != "unavailable" + || custom["externalSourceError"] != error; custom["externalSourceStatus"] = json!("unavailable"); custom["externalSourceError"] = json!(error); - ((), true) + ((), changed) } else { ((), false) } }) .await?; if content_version(&entry.metadata).is_some() { - return Ok(()); + return Ok(None); } if cache_unavailable_legacy_history(root, &entry).await? { - return Ok(()); + return Ok(None); } return Err(index_error(format!( "The original {} history is unavailable. Restore the source and retry. {error}", @@ -332,19 +377,64 @@ pub(super) async fn refresh_reference( let version = hex::encode(Sha256::digest( serde_json::to_vec(&(source.revision(), &document.turns)).map_err(index_error)?, )); - let mut bundle = convert_document( - &id, - &workspace, - &project, - &entry.metadata.agent_type, - &entry.metadata.session_name, - &document, - true, - true, - ) - .map_err(|error| index_error(error.detail))?; + let path = snapshot_path(root, &id, &version)?; + // A legacy native copy still needs its unversioned conversion for the + // local-edit comparison. Never let snapshot reuse bypass that check. + let cached = if !super::legacy::follows_native(&entry.metadata) + && tokio::fs::try_exists(&path).await.map_err(index_error)? + { + Some(read_snapshot(root, &entry, &version).await?) + } else { + None + }; + let mut cached_metadata = cached.as_ref().map(|bundle| bundle.metadata.clone()); + let mut bundle = if let Some(mut cached) = cached { + cached.metadata = document_metadata( + &id, + &workspace, + &project, + &entry.metadata.agent_type, + &entry.metadata.session_name, + &document, + ) + .map_err(|error| index_error(error.detail))?; + finish_document_metadata(&mut cached.metadata, &cached.turns); + cached + } else { + #[cfg(test)] + super::test_io::record(root, |counts| counts.conversions += 1); + convert_document( + &id, + &workspace, + &project, + &entry.metadata.agent_type, + &entry.metadata.session_name, + &document, + true, + true, + ) + .map_err(|error| index_error(error.detail))? + }; if !super::legacy::validate_native(root, &entry, &bundle).await? { - return Ok(()); + return Ok(None); + } + if cached_metadata.is_none() && tokio::fs::try_exists(&path).await.map_err(index_error)? { + // Legacy comparison above must use freshly converted, unversioned + // turns. Once it succeeds, use the frozen turns for their synthesized + // timestamps too; missing source timestamps must not cause index churn. + let cached = read_snapshot(root, &entry, &version).await?; + finish_document_metadata(&mut bundle.metadata, &cached.turns); + bundle.turns = cached.turns; + cached_metadata = Some(cached.metadata); + } + if content_version(&entry.metadata) == Some(version.as_str()) { + // Conversion timestamps are not source changes. Keep the existing + // version's creation time (and activity time for an empty history). + bundle.metadata.created_at = entry.metadata.created_at; + if bundle.turns.is_empty() { + bundle.metadata.last_active_at = entry.metadata.last_active_at; + bundle.metadata.last_finished_at = entry.metadata.last_finished_at; + } } bundle.metadata.tags = entry.metadata.tags.clone(); bundle.metadata.status = entry.metadata.status.clone(); @@ -382,10 +472,12 @@ pub(super) async fn refresh_reference( json!(document.turns.last().is_some_and(|turn| turn.status != openbitfun_product_domains::session_import::ExternalSessionStatus::Complete)); custom["continuationMode"] = json!("fork_before_continue"); - for turn in &mut bundle.turns { - turn.turn_id = format!("{id}-v{version}-turn-{}", turn.turn_index); - if let Some(metadata) = turn.user_message.metadata.as_mut() { - metadata[VERSION] = json!(version); + if cached_metadata.is_none() { + for turn in &mut bundle.turns { + turn.turn_id = format!("{id}-v{version}-turn-{}", turn.turn_index); + if let Some(metadata) = turn.user_message.metadata.as_mut() { + metadata[VERSION] = json!(version); + } } } let last_active = bundle.metadata.last_active_at; @@ -399,33 +491,38 @@ pub(super) async fn refresh_reference( last.recovery_pending = Some(false); } bundle.validate().map_err(index_error)?; - let path = snapshot_path(root, &id, &version)?; - if !tokio::fs::try_exists(&path).await.map_err(index_error)? { + let mut metadata = bundle.metadata.clone(); + let published = if let Some(frozen_metadata) = cached_metadata { + bundle.metadata = frozen_metadata; + bundle + } else if !tokio::fs::try_exists(&path).await.map_err(index_error)? { + let snapshot = Snapshot { + schema_version: 1, + ecosystem_id: entry.ecosystem_id.clone(), + source_session_id: entry.source_session_id.clone(), + version, + bundle, + }; JsonFileStore - .write_atomic_strict( - &path, - &Snapshot { - schema_version: 1, - ecosystem_id: entry.ecosystem_id.clone(), - source_session_id: entry.source_session_id.clone(), - version, - bundle: bundle.clone(), - }, - ) + .write_atomic_strict(&path, &snapshot) .await .map_err(index_error)?; + #[cfg(test)] + super::test_io::record(root, |counts| counts.snapshot_writes += 1); + snapshot.bundle } else { // Never trust an interrupted or malformed cache merely because it exists. - read_snapshot(root, &entry, &version).await?; - } + read_snapshot(root, &entry, &version).await? + }; update_index(root, |index| { let Some(current) = index.entries.get_mut(&id).filter(|entry| !entry.dismissed) else { return ((), false); }; + let before = serde_json::to_value(&*current).ok(); // Preserve a concurrent rename/archive operation and a newer observation. - bundle.metadata.session_name = current.metadata.session_name.clone(); - bundle.metadata.tags = current.metadata.tags.clone(); - bundle.metadata.status = current.metadata.status.clone(); + metadata.session_name = current.metadata.session_name.clone(); + metadata.tags = current.metadata.tags.clone(); + metadata.status = current.metadata.status.clone(); if let Some(custom) = current.metadata.custom_metadata.as_ref() { for key in [ "externalSessionTitleOverride", @@ -433,7 +530,7 @@ pub(super) async fn refresh_reference( "externalLegacyNativeHash", ] { if let Some(value) = custom.get(key) { - bundle.metadata.custom_metadata.as_mut().unwrap()[key] = value.clone(); + metadata.custom_metadata.as_mut().unwrap()[key] = value.clone(); } } } @@ -448,19 +545,21 @@ pub(super) async fn refresh_reference( current.source_title.as_deref(), current.fallback_title.as_deref(), ); - bundle.metadata.session_name = current.metadata.session_name.clone(); + metadata.session_name = current.metadata.session_name.clone(); if let Some(title) = current .metadata .custom_metadata .as_ref() .and_then(|custom| custom.get(SOURCE_TITLE_KEY)) { - bundle.metadata.custom_metadata.as_mut().unwrap()[SOURCE_TITLE_KEY] = title.clone(); + metadata.custom_metadata.as_mut().unwrap()[SOURCE_TITLE_KEY] = title.clone(); } - current.metadata = bundle.metadata; - ((), true) + current.metadata = metadata; + let changed = before.is_none() || before != serde_json::to_value(&*current).ok(); + ((), changed) }) - .await + .await?; + Ok(Some(published)) } pub(crate) fn branch_metadata(source: &SessionMetadata, target: &mut SessionMetadata) { @@ -491,7 +590,7 @@ mod tests { use crate::external_session_import::auto_discovery::tests::{candidate, document}; use openbitfun_services_core::session::SessionBranchBoundary; - async fn create_source(root: &Path, workspace: &WorkspaceInfo) -> Reference { + pub(super) async fn create_source(root: &Path, workspace: &WorkspaceInfo) -> Reference { register( root, workspace, @@ -510,7 +609,7 @@ mod tests { .clone() } - async fn publish( + pub(super) async fn publish( root: &Path, workspace: &WorkspaceInfo, entry: &Reference, diff --git a/src/crates/assembly/core/src/external_session_import/deferred/tracking/refresh_tests.rs b/src/crates/assembly/core/src/external_session_import/deferred/tracking/refresh_tests.rs new file mode 100644 index 0000000000..ec14ee4b63 --- /dev/null +++ b/src/crates/assembly/core/src/external_session_import/deferred/tracking/refresh_tests.rs @@ -0,0 +1,437 @@ +use super::super::test_io::{Counts, Probe}; +use super::super::tests::fixture; +use super::tests::{create_source, publish}; +use super::*; +use crate::external_session_import::auto_discovery::tests::{candidate, document}; +use openbitfun_product_domains::session_import::ExternalSessionItem; + +async fn open_document( + root: &Path, + workspace: &WorkspaceInfo, + entry: &Reference, + document: ExternalSessionDocument, +) -> OpenBitFunResult> { + let refreshed = refresh_reference( + root, + entry.clone(), + workspace.clone(), + workspace.clone(), + vec![workspace.clone()], + move |_, _| Ok(document), + ) + .await?; + load_current_snapshot(root, &entry.metadata.session_id, refreshed).await +} + +#[tokio::test] +async fn unchanged_opens_read_one_snapshot_and_do_not_convert_write_or_notify() { + let (_temp, workspace, root, _manager) = fixture(); + let entry = create_source(&root, &workspace).await; + let mut document = document(candidate("codex", "source-1", &workspace.root_path)); + document.candidate.model_provider = Some("source-provider".into()); + document.candidate.model_name = Some("source-model".into()); + document.turns[0].items.extend([ + ExternalSessionItem::Thinking { + text: "Saved reasoning".into(), + }, + ExternalSessionItem::Tool { + id: "saved-call".into(), + name: "exec_command".into(), + input: json!({"cmd":"echo saved"}), + result: Some(json!({"stdout":"saved"})), + success: Some(true), + }, + ]); + let probe = Probe::new(&root); + let first = open_document(&root, &workspace, &entry, document.clone()) + .await + .unwrap() + .unwrap(); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + conversions: 1, + snapshot_writes: 1, + index_writes: 1, + catalog_notifications: 1, + ..Counts::default() + } + ); + let index = std::fs::read(root.join(INDEX_FILE)).unwrap(); + let path = snapshot_path( + &root, + &entry.metadata.session_id, + content_version(&first.metadata).unwrap(), + ) + .unwrap(); + let frozen = std::fs::read(&path).unwrap(); + // Wall-clock changes must not turn conversion-generated timestamps into + // source changes. The persisted version remains the same on every open. + tokio::time::sleep(Duration::from_millis(3)).await; + for _ in 0..5 { + let reopened = open_document(&root, &workspace, &entry, document.clone()) + .await + .unwrap() + .unwrap(); + assert_eq!( + serde_json::to_value(reopened).unwrap(), + serde_json::to_value(&first).unwrap() + ); + } + assert_eq!( + probe.take(), + Counts { + source_reads: 5, + snapshot_reads: 5, + ..Counts::default() + } + ); + assert_eq!(std::fs::read(root.join(INDEX_FILE)).unwrap(), index); + assert_eq!(std::fs::read(path).unwrap(), frozen); +} + +#[tokio::test] +async fn empty_and_undated_versions_are_stable_on_reopen() { + for empty in [true, false] { + let (_temp, workspace, root, _manager) = fixture(); + let entry = create_source(&root, &workspace).await; + let mut document = document(candidate("codex", "source-1", &workspace.root_path)); + if empty { + document.turns.clear(); + } else { + document.turns[0].timestamp = 0; + } + let first = open_document(&root, &workspace, &entry, document.clone()) + .await + .unwrap() + .unwrap(); + tokio::time::sleep(Duration::from_millis(3)).await; + let probe = Probe::new(&root); + let reopened = open_document(&root, &workspace, &entry, document) + .await + .unwrap() + .unwrap(); + assert_eq!( + serde_json::to_value(reopened).unwrap(), + serde_json::to_value(first).unwrap() + ); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + snapshot_reads: 1, + ..Counts::default() + } + ); + } +} + +#[tokio::test] +async fn fresh_source_changes_are_detected_even_when_discovery_fingerprint_is_unchanged() { + let (_temp, workspace, root, _manager) = fixture(); + let entry = create_source(&root, &workspace).await; + let original = document(candidate("codex", "source-1", &workspace.root_path)); + let first = open_document(&root, &workspace, &entry, original.clone()) + .await + .unwrap() + .unwrap(); + let mut changed = original; + changed.turns[0].user_text = "An externally rewritten question".into(); + let probe = Probe::new(&root); + let latest = open_document(&root, &workspace, &entry, changed) + .await + .unwrap() + .unwrap(); + assert_eq!( + latest.turns[0].user_message.content, + "An externally rewritten question" + ); + assert_ne!( + content_version(&latest.metadata), + content_version(&first.metadata) + ); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + conversions: 1, + snapshot_writes: 1, + index_writes: 1, + catalog_notifications: 1, + ..Counts::default() + } + ); + let pinned = snapshot_at_turn(&root, &entry.metadata.session_id, &first.turns[0].turn_id) + .await + .unwrap() + .unwrap(); + assert_eq!( + serde_json::to_value(pinned.turns).unwrap(), + serde_json::to_value(first.turns).unwrap() + ); +} + +#[tokio::test] +async fn unchanged_content_still_updates_source_title_and_availability() { + let (_temp, workspace, root, _manager) = fixture(); + let entry = create_source(&root, &workspace).await; + let original = document(candidate("codex", "source-1", &workspace.root_path)); + publish(&root, &workspace, &entry, original.clone()).await; + let mut renamed = original; + renamed.candidate.title = Some("Renamed in source".into()); + renamed.candidate.title_is_source_name = true; + let probe = Probe::new(&root); + let opened = open_document(&root, &workspace, &entry, renamed.clone()) + .await + .unwrap() + .unwrap(); + assert_eq!(opened.metadata.session_name, "Renamed in source"); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + snapshot_reads: 1, + index_writes: 1, + catalog_notifications: 1, + ..Counts::default() + } + ); + for expected_writes in [1, 0] { + let refreshed = refresh_reference( + &root, + entry.clone(), + workspace.clone(), + workspace.clone(), + vec![workspace.clone()], + |_, _| Err("Source is offline".into()), + ) + .await + .unwrap(); + let offline = load_current_snapshot(&root, &entry.metadata.session_id, refreshed) + .await + .unwrap() + .unwrap(); + assert_eq!( + offline.metadata.custom_metadata.unwrap()["externalSourceStatus"], + "unavailable" + ); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + snapshot_reads: 1, + index_writes: expected_writes, + catalog_notifications: expected_writes, + ..Counts::default() + } + ); + } + let restored = open_document(&root, &workspace, &entry, renamed) + .await + .unwrap() + .unwrap(); + assert_eq!( + restored.metadata.custom_metadata.unwrap()["externalSourceStatus"], + "available" + ); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + snapshot_reads: 1, + index_writes: 1, + catalog_notifications: 1, + ..Counts::default() + } + ); +} + +#[tokio::test] +async fn malformed_existing_snapshot_is_not_trusted_or_overwritten() { + let (_temp, workspace, root, _manager) = fixture(); + let entry = create_source(&root, &workspace).await; + let document = document(candidate("codex", "source-1", &workspace.root_path)); + let first = open_document(&root, &workspace, &entry, document.clone()) + .await + .unwrap() + .unwrap(); + let path = snapshot_path( + &root, + &entry.metadata.session_id, + content_version(&first.metadata).unwrap(), + ) + .unwrap(); + std::fs::write(&path, b"{interrupted snapshot").unwrap(); + let index = std::fs::read(root.join(INDEX_FILE)).unwrap(); + let probe = Probe::new(&root); + assert!(open_document(&root, &workspace, &entry, document) + .await + .is_err()); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + snapshot_reads: 1, + ..Counts::default() + } + ); + assert_eq!(std::fs::read(path).unwrap(), b"{interrupted snapshot"); + assert_eq!(std::fs::read(root.join(INDEX_FILE)).unwrap(), index); +} + +#[tokio::test] +async fn concurrent_local_changes_and_removal_win_during_an_unchanged_refresh() { + for remove in [false, true] { + let (_temp, workspace, root, _manager) = fixture(); + let entry = create_source(&root, &workspace).await; + let original = document(candidate("codex", "source-1", &workspace.root_path)); + publish(&root, &workspace, &entry, original.clone()).await; + let (started_tx, started_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = std::sync::mpsc::channel(); + let refresh_root = root.clone(); + let refresh_workspace = workspace.clone(); + let refresh_entry = entry.clone(); + let task = tokio::spawn(async move { + refresh_reference( + &refresh_root, + refresh_entry, + refresh_workspace.clone(), + refresh_workspace.clone(), + vec![refresh_workspace], + move |_, _| { + started_tx.send(()).unwrap(); + release_rx.recv_timeout(Duration::from_secs(5)).unwrap(); + Ok(original) + }, + ) + .await + }); + started_rx.await.unwrap(); + if remove { + dismiss(&root, &entry.metadata.session_id).await.unwrap(); + } else { + let mut local = metadata(&root, &entry.metadata.session_id) + .await + .unwrap() + .unwrap(); + local.session_name = "Local name".into(); + local.status = SessionStatus::Archived; + local.tags = vec!["keep-local-tag".into()]; + save_metadata(&root, &local).await.unwrap(); + } + let probe = Probe::new(&root); + release_tx.send(()).unwrap(); + let refreshed = task.await.unwrap().unwrap(); + let current = load_current_snapshot(&root, &entry.metadata.session_id, refreshed) + .await + .unwrap(); + if remove { + assert!(current.is_none()); + } else { + let current = current.unwrap(); + assert_eq!(current.metadata.session_name, "Local name"); + assert_eq!(current.metadata.status, SessionStatus::Archived); + assert_eq!(current.metadata.tags, vec!["keep-local-tag"]); + } + assert_eq!( + probe.take(), + Counts { + snapshot_reads: 1, + ..Counts::default() + } + ); + } +} + +#[tokio::test] +async fn legacy_reopens_skip_unchanged_index_writes_but_still_detect_native_edits() { + let (_temp, workspace, root, _manager) = fixture(); + let mut original = document(candidate("codex", "source-1", &workspace.root_path)); + original.turns[0].timestamp = 0; + let id = crate::external_session_import::auto_discovery::automatic_session_id( + &workspace.id, + "codex", + "source-1", + ); + let mut native = convert_document( + &id, + &workspace, + &workspace, + "primary", + "Old import", + &original, + true, + true, + ) + .unwrap(); + let store = OfflineSessionImportStore::new(&root); + store.write_bundle_atomic(&native).await.unwrap(); + register( + &root, + &workspace, + &workspace, + vec![original.candidate.clone()], + ) + .await + .unwrap(); + let entry = reference(&root, &id).await.unwrap().unwrap(); + publish(&root, &workspace, &entry, original.clone()).await; + tokio::time::sleep(Duration::from_millis(3)).await; + let probe = Probe::new(&root); + open_document(&root, &workspace, &entry, original.clone()) + .await + .unwrap() + .unwrap(); + assert_eq!( + probe.take(), + Counts { + source_reads: 1, + snapshot_reads: 1, + conversions: 1, + ..Counts::default() + } + ); + native.turns[0].user_message.content = "A local edit by an older host".into(); + store.write_bundle(&native).await.unwrap(); + assert!(open_document(&root, &workspace, &entry, original) + .await + .unwrap() + .is_none()); + let saved = store.load_bundle(&id).await.unwrap().unwrap(); + assert_eq!( + saved.turns[0].user_message.content, + "A local edit by an older host" + ); + assert!( + metadata(&root, &id).await.unwrap().is_none(), + "The locally edited copy must leave the followed-view metadata path" + ); +} + +#[tokio::test] +async fn a_newer_published_version_wins_before_the_refresh_result_is_delivered() { + let (_temp, workspace, root, _manager) = fixture(); + let entry = create_source(&root, &workspace).await; + let original = document(candidate("codex", "source-1", &workspace.root_path)); + let first = open_document(&root, &workspace, &entry, original.clone()) + .await + .unwrap() + .unwrap(); + let mut next = original; + next.turns[0].user_text = "Newer refresh".into(); + publish(&root, &workspace, &entry, next).await; + let probe = Probe::new(&root); + let current = load_current_snapshot(&root, &entry.metadata.session_id, Some(first)) + .await + .unwrap() + .unwrap(); + assert_eq!(current.turns[0].user_message.content, "Newer refresh"); + assert_eq!( + probe.take(), + Counts { + snapshot_reads: 1, + ..Counts::default() + } + ); +} From 8fbbe57560cb2390be2ecc2eb02f4e820aade8be Mon Sep 17 00:00:00 2001 From: kev1n77 Date: Fri, 16 Oct 2026 21:08:23 +0800 Subject: [PATCH 2/6] perf(sessions): page workspace catalogs and reuse versioned snapshots --- .../external-ai-work-sources-design.md | 11 +- .../cli/src/peer_host/commands/session.rs | 54 +- src/apps/desktop/AGENTS.md | 3 + src/apps/desktop/src/api/session_api.rs | 24 + .../src/runtime/session_application.rs | 4 +- src/crates/assembly/core/AGENTS.md | 4 + .../core/src/agentic/persistence/manager.rs | 214 +++++- .../persistence/manager/session_catalog.rs | 213 ++++++ .../src/external_session_import/deferred.rs | 166 ++++- .../assembly/core/src/product_runtime.rs | 35 +- src/crates/services/services-core/AGENTS.md | 1 + .../services/services-core/src/json_store.rs | 48 ++ .../src/session/metadata_store.rs | 109 ++- .../services-core/src/session/page.rs | 650 ++++++++++++++++-- src/web-ui/AGENTS.md | 4 + .../sections/sessions/SessionsSection.tsx | 56 +- .../sessions/sessionWorktreeRow.test.ts | 6 +- .../useWorkspaceSessionCatalog.test.tsx | 144 +++- .../sessions/useWorkspaceSessionCatalog.ts | 122 +++- .../workspaceSessionCatalogQuery.test.ts | 65 ++ .../sessions/workspaceSessionCatalogQuery.ts | 57 ++ .../src/flow_chat/store/FlowChatStore.test.ts | 16 + .../src/flow_chat/store/FlowChatStore.ts | 12 +- .../api/service-api/SessionAPI.test.ts | 11 + .../api/service-api/SessionAPI.ts | 14 + 25 files changed, 1882 insertions(+), 161 deletions(-) create mode 100644 src/crates/assembly/core/src/agentic/persistence/manager/session_catalog.rs create mode 100644 src/web-ui/src/app/components/NavPanel/sections/sessions/workspaceSessionCatalogQuery.test.ts create mode 100644 src/web-ui/src/app/components/NavPanel/sections/sessions/workspaceSessionCatalogQuery.ts diff --git a/docs/architecture/extensions/external-ai-work-sources-design.md b/docs/architecture/extensions/external-ai-work-sources-design.md index b27dcd1949..3121c9112f 100644 --- a/docs/architecture/extensions/external-ai-work-sources-design.md +++ b/docs/architecture/extensions/external-ai-work-sources-design.md @@ -101,6 +101,15 @@ Zcode 会话读取不意味着其 ACP 运行时、设置、Skill、MCP 或 Hook 下一次目录扫描时重新计算，保留最近本地工作区及同根登记顺序的匹配规则。启动仅将会话标题、来源和归属写入集中会话目录的 `.external-session-index.json`；该私有索引由现有持久化入口合并到元数据分页，旧主机忽略此文件， 既有原生 Session 格式不变。改名、归档和删除只更新索引；删除保留抑制记录，重新扫描不复活条目。 +会话元数据分页通过可选 `query` 声明工作区 ID、创建/最近活动排序、来源及已观察到的活动时间， +主机完成归属筛选、父子分组和来源筛选后返回分页与精确计数，并以 `catalogQuery` 回显已应用条件。 +Web UI 默认仅读取可见页，展开时续读，复用启动首屏；相同时间的边界条目一起返回，保留客户端的本地化排序。 +名称/状态排序、未读/关注及其他组合筛选继续使用完整目录投影。没有查询回显的旧主机、无法确定工作区 ID 的 +旧记录也保留原分页遍历，不能把共享存储根计数当成工作区计数。Desktop 与 CLI Peer 使用同一查询实现； +切换设备、工作区、来源或发现刷新会隔离读取。主机按存储目录、查询和原生/外部索引文件修订复用合并、排序后的 +有界内存快照；命中后只校验索引修订和本页原生文件存在性，不再解析、合并整个目录。索引的修改、替换或删除 +会使缓存失效，包括其他进程写入；游标绑定查询及修订，版本变化通过 `catalogReset` 返回新目录首段，前端重置累计数量。 +缓存淘汰不改变持久化格式或删除会话；过大目录仍按页返回，但不常驻缓存。分页不加载会话正文。 源会话的移除动作显示为“从 OpenBitFun 移除”，不修改外部原始记录或已有原生分支。生态兼容页 列出当前主机上已移除的来源，用户主动重新添加时按工作区 ID 和来源身份校验，恢复同一个会话 ID、 自定义名称和缓存；恢复操作幂等，迟到的重试不会撤销之后的归档。旧重复项恢复为隐藏别名。 @@ -125,7 +134,7 @@ Web UI 每次打开来源会话时读取一次来源；停留、输入、窗口 通过核对后记录本地历史摘要，后续来源改写不会误判为本地编辑；旧主机再次修改本地历史时重新核对。 跟随条目保留旧 `externalHistoryPending` 标记，旧版主机仍能物化副本；手动导入的原生副本始终独立。 来源输入区采用当前主机默认权限，用户选择先存于本地输入状态，创建分支后才写入分支，不恢复来源运行时。私有索引新字段可选， -不改变现有原生会话 invoke 请求形状；不理解跟随视图的客户端直接提交时收到明确的只读/需要分叉错误。 +保留现有原生会话 invoke 的旧请求兼容性；不理解跟随视图的客户端直接提交时收到明确的只读/需要分叉错误。 改名、归档、删除意图独立于缓存更新，既有完整副本及历史版本不自动清理。 来源标题由生态 adapter 读取：Codex 优先使用 `state_*.sqlite` 的 `threads.name`，随后是 `session_index.jsonl` 最新名称、数据库 `title`、rollout 头部名称及有效用户输入摘要；旧版 diff --git a/src/apps/cli/src/peer_host/commands/session.rs b/src/apps/cli/src/peer_host/commands/session.rs index 79eafa5348..c0a7a16589 100644 --- a/src/apps/cli/src/peer_host/commands/session.rs +++ b/src/apps/cli/src/peer_host/commands/session.rs @@ -317,20 +317,39 @@ pub(crate) async fn list_persisted_sessions_page( .map(|value| serde_json::from_value::>(value.clone())) .transpose() .map_err(|error| format!("Invalid session activity ids: {error}"))?; + let query = session_catalog_query(request)?; let page = state .compatibility - .list_persisted_sessions_page_with_activity( + .list_persisted_sessions_page_with_activity_and_query( &state.agent_runtime, &workspace_path, cursor.as_deref(), limit, session_ids.as_deref(), + query.as_ref(), ) .await .map_err(|e| format!("Failed to list persisted session page: {e}"))?; serde_json::to_value(page).map_err(|e| format!("serialize session page: {e}")) } +fn session_catalog_query( + request: &Value, +) -> Result, String> { + let query = request + .get("query") + .filter(|value| !value.is_null()) + .map(|value| { + serde_json::from_value::( + value.clone(), + ) + }) + .transpose() + .map_err(|error| format!("Invalid session catalog query: {error}"))?; + let workspace_id = optional_string(request, "workspaceId"); + Ok(query.filter(|query| Some(query.workspace_id.as_str()) == workspace_id.as_deref())) +} + pub(crate) async fn list_persisted_sessions_count( state: &PeerHostState, args: &Value, @@ -999,6 +1018,39 @@ mod tests { use openbitfun_core::util::errors::OpenBitFunError; use openbitfun_events::AgenticEvent; + #[test] + fn session_catalog_query_keeps_legacy_peers_and_workspace_identity_compatible() { + let query = + serde_json::json!({"workspaceId": "main", "ordering": "updated", "source": "codex"}); + for field in ["workspace_id", "workspaceId"] { + let mut request = serde_json::json!({"query": query}); + request[field] = serde_json::json!("main"); + assert_eq!( + super::session_catalog_query(&request) + .unwrap() + .unwrap() + .workspace_id, + "main" + ); + request[field] = serde_json::json!("another-workspace"); + assert!(super::session_catalog_query(&request).unwrap().is_none()); + } + assert!( + super::session_catalog_query(&serde_json::json!({"workspace_path": "/old/peer"})) + .unwrap() + .is_none() + ); + assert!(super::session_catalog_query( + &serde_json::json!({"workspace_path": "/old/peer", "query": query}) + ) + .unwrap() + .is_none()); + assert!( + super::session_catalog_query(&serde_json::json!({"query": {"source": "codex"}})) + .is_err() + ); + } + #[test] fn peer_writer_conflicts_keep_the_stable_transport_code() { let core_error = peer_core_session_error( diff --git a/src/apps/desktop/AGENTS.md b/src/apps/desktop/AGENTS.md index c8bdfca0e8..fe364b10de 100644 --- a/src/apps/desktop/AGENTS.md +++ b/src/apps/desktop/AGENTS.md @@ -128,6 +128,9 @@ For whole-session copy routing, run `cargo test --locked -p openbitfun-desktop --lib whole_session_copy_uses_the_latest_persisted_turn_fork`. After command registration changes, also run the `remote_workspace_policy` filter. +For optional workspace catalog query payload compatibility, run +`cargo test --locked -p openbitfun-desktop --lib api::session_api::catalog_request_tests`. + For model selector request compatibility and configuration effect routing: ```bash diff --git a/src/apps/desktop/src/api/session_api.rs b/src/apps/desktop/src/api/session_api.rs index 0ce00bc46a..18826dfff4 100644 --- a/src/apps/desktop/src/api/session_api.rs +++ b/src/apps/desktop/src/api/session_api.rs @@ -64,6 +64,8 @@ pub struct ListPersistedSessionsPageRequest { #[serde(default, alias = "sessionIds")] pub session_ids: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] + pub query: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] pub cursor: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub remote_connection_id: Option, @@ -458,6 +460,7 @@ pub async fn list_persisted_sessions_page( request.cursor.as_deref(), request.limit, request.session_ids.as_deref(), + request.query.as_ref(), ) .await .map_err(|error| { @@ -910,3 +913,24 @@ pub async fn delete_all_archived_sessions( Ok(deleted_count) } + +#[cfg(test)] +mod catalog_request_tests { + use super::ListPersistedSessionsPageRequest; + + #[test] + fn session_catalog_request_preserves_legacy_payload_and_optional_query() { + let legacy = serde_json::json!({"workspace_id": "main", "workspace_path": "", "limit": 5, "session_ids": null}); + let request: ListPersistedSessionsPageRequest = + serde_json::from_value(legacy.clone()).unwrap(); + assert!(request.query.is_none()); + assert_eq!(serde_json::to_value(request).unwrap(), legacy); + let mut current = legacy; + current["query"] = + serde_json::json!({"workspaceId": "main", "ordering": "updated", "source": "codex"}); + let request: ListPersistedSessionsPageRequest = + serde_json::from_value(current.clone()).unwrap(); + assert_eq!(request.query.as_ref().unwrap().workspace_id, "main"); + assert_eq!(serde_json::to_value(request).unwrap(), current); + } +} diff --git a/src/apps/desktop/src/runtime/session_application.rs b/src/apps/desktop/src/runtime/session_application.rs index 00e8e4a37e..31f466ed21 100644 --- a/src/apps/desktop/src/runtime/session_application.rs +++ b/src/apps/desktop/src/runtime/session_application.rs @@ -392,17 +392,19 @@ impl DesktopSessionApplication { cursor: Option<&str>, limit: usize, session_ids: Option<&[String]>, + query: Option<&openbitfun_services_core::session::page::SessionCatalogQuery>, ) -> DesktopSessionApplicationResult { let scope = self.resolved_scope(request).await?; let storage_path = self.storage_path(&scope); let mut page = self .compatibility - .list_persisted_sessions_page_with_activity( + .list_persisted_sessions_page_with_activity_and_query( &self.agent_runtime, &storage_path, cursor, limit, session_ids, + query.filter(|query| query.workspace_id == scope.workspace_id), ) .await .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))?; diff --git a/src/crates/assembly/core/AGENTS.md b/src/crates/assembly/core/AGENTS.md index 224cbebb72..f0a9675810 100644 --- a/src/crates/assembly/core/AGENTS.md +++ b/src/crates/assembly/core/AGENTS.md @@ -249,6 +249,7 @@ they do not exercise live SSH, relay, peer, or dispatch connections. For disk-backed history paging and legacy sessions without a catalog: `cargo test --locked -p openbitfun-core --no-default-features --features remote-connect,git --lib history_page_`. + Also run the `staged_revert_catalog_projection` and `load_relay_session_turns_` filters for the same target when changing visibility. Paging must not parse unrelated turn bodies or rewrite history. To compare real-file first-page work @@ -256,6 +257,9 @@ against full materialization locally, use the same target with `history_page_benchmark -- --ignored --nocapture`; it checks content equivalence and reports timings without asserting a machine-dependent latency in CI. +For workspace metadata catalog paging, cache reuse and cross-process invalidation: +`cargo test --locked -p openbitfun-core --no-default-features --features external-sources,git,product-search --lib -- catalog_snapshot_ catalog_query_pages_`. + For host-stream history reads and abandoned execution after a runtime restart: `cargo test --locked -p openbitfun-core --no-default-features --features agent-runtime,git --lib load_relay_session_turns_`. The observer must preserve terminal history and another process's writer lease; diff --git a/src/crates/assembly/core/src/agentic/persistence/manager.rs b/src/crates/assembly/core/src/agentic/persistence/manager.rs index cabc768c3f..ddb4657409 100644 --- a/src/crates/assembly/core/src/agentic/persistence/manager.rs +++ b/src/crates/assembly/core/src/agentic/persistence/manager.rs @@ -4,6 +4,7 @@ #[cfg(feature = "external-sources")] mod external_history; +mod session_catalog; use crate::agentic::core::{ sanitize_persisted_session_state, CompressionState, Message, MessageContent, @@ -49,7 +50,7 @@ use openbitfun_services_core::session_search::SessionSearchSqliteIndex; use openbitfun_services_core::{ json_store::{JsonFileStore, JsonFileStoreError}, session::{ - build_session_metadata as build_persisted_session_metadata, empty_session_metadata_page, + build_session_metadata as build_persisted_session_metadata, refresh_session_metadata_from_turns, try_refresh_session_metadata_for_saved_turn, DialogTurnKind, SessionLastTurn, SessionMemoryMode, SessionMetadataBuildFacts, SessionMetadataStore, SessionMetadataStoreError, SessionStorageLayout, SessionWriteLock, @@ -568,6 +569,7 @@ struct ParsedTranscriptTurnSelector { pub struct PersistenceManager { path_manager: Arc, runtime_service: Arc, + catalog_cache: Mutex, #[cfg(test)] fail_next_session_state_write: std::sync::Mutex>, #[cfg(test)] @@ -585,6 +587,7 @@ impl PersistenceManager { Ok(Self { runtime_service: Arc::new(WorkspaceRuntimeService::new(path_manager.clone())), path_manager, + catalog_cache: Mutex::new(session_catalog::CatalogCache::default()), #[cfg(test)] fail_next_session_state_write: std::sync::Mutex::new(None), #[cfg(test)] @@ -1489,14 +1492,55 @@ impl PersistenceManager { cursor: Option<&str>, limit: usize, ) -> OpenBitFunResult { + self.list_session_metadata_page_with_query(workspace_path, cursor, limit, None) + .await + } + + pub async fn list_session_metadata_page_with_query( + &self, + workspace_path: &Path, + cursor: Option<&str>, + limit: usize, + query: Option<&openbitfun_services_core::session::page::SessionCatalogQuery>, + ) -> OpenBitFunResult { + use openbitfun_services_core::session::page::build_session_metadata_page_with_query; if !workspace_path.exists() { - return Ok(empty_session_metadata_page()); + return Ok(build_session_metadata_page_with_query( + Vec::new(), + cursor, + limit, + query, + )); } if self.existing_project_sessions_dir(workspace_path).is_none() { - return Ok(empty_session_metadata_page()); + return Ok(build_session_metadata_page_with_query( + Vec::new(), + cursor, + limit, + query, + )); + } + + if let Some(query) = query { + return self + .list_cached_session_catalog_page(workspace_path, cursor, limit, query) + .await; } + self.list_uncached_session_metadata_page(workspace_path, cursor, limit, None) + .await + } + + async fn list_uncached_session_metadata_page( + &self, + workspace_path: &Path, + cursor: Option<&str>, + limit: usize, + query: Option<&openbitfun_services_core::session::page::SessionCatalogQuery>, + ) -> OpenBitFunResult { + use openbitfun_services_core::session::page::build_session_metadata_page_with_query; + #[cfg(feature = "external-sources")] { let root = self.project_sessions_dir(workspace_path); @@ -1509,23 +1553,20 @@ impl PersistenceManager { .map_err(Self::session_metadata_store_error)?; if let Err(error) = references { warn!("External session references could not be paged: {error}"); - let mut page = openbitfun_services_core::session::build_session_metadata_page( - native, cursor, limit, - ); + let mut page = + build_session_metadata_page_with_query(native, cursor, limit, query); page.discovery_warnings.push("External session index could not be read; existing native sessions remain available".into()); return Ok(page); } let metadata = crate::external_session_import::deferred::merge_metadata(&root, native).await?; - return Ok( - openbitfun_services_core::session::build_session_metadata_page( - metadata, cursor, limit, - ), - ); + return Ok(build_session_metadata_page_with_query( + metadata, cursor, limit, query, + )); } } self.session_metadata_store(workspace_path) - .list_metadata_page(cursor, limit) + .list_metadata_page_with_query(cursor, limit, query) .await .map_err(Self::session_metadata_store_error) } @@ -8283,6 +8324,155 @@ mod tests { ); } + #[test] + #[ignore = "invoked by the catalog cross-process writer test"] + fn catalog_cross_process_writer_child() { + let root = std::env::var_os("OPENBITFUN_CATALOG_WRITER_ROOT").expect("writer root"); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(async { + let store = openbitfun_services_core::session::SessionMetadataStore::new( + PathBuf::from(root), + ); + let mut metadata = store.load_metadata("catalog-11").await.unwrap().unwrap(); + metadata.session_name = "Renamed by another process".into(); + store.save_metadata(&metadata).await.unwrap(); + }); + } + + #[tokio::test] + async fn catalog_snapshot_reuses_pages_and_invalidates_after_another_process_writes() { + use openbitfun_services_core::session::page::{ + SessionCatalogOrdering, SessionCatalogQuery, + }; + let workspace = TestWorkspace::new(); + let manager = PersistenceManager::new(workspace.path_manager()).unwrap(); + for index in 0..12 { + let mut metadata = SessionMetadata::new( + format!("catalog-{index}"), + format!("Entry {index}"), + "agent".into(), + "model".into(), + ); + metadata.workspace_id = Some("main".into()); + metadata.created_at = index; + manager + .save_session_metadata(workspace.path(), &metadata) + .await + .unwrap(); + } + let query = SessionCatalogQuery { + workspace_id: "main".into(), + ordering: SessionCatalogOrdering::Created, + source: "all".into(), + sort_timestamps: Default::default(), + }; + let first = manager + .list_session_metadata_page_with_query(workspace.path(), None, 5, Some(&query)) + .await + .unwrap(); + let second = manager + .list_session_metadata_page_with_query( + workspace.path(), + first.next_cursor.as_deref(), + 5, + Some(&query), + ) + .await + .unwrap(); + assert_eq!(second.sessions[0].session_id, "catalog-6"); + assert_eq!(second.catalog_revision, first.catalog_revision); + assert_eq!(manager.session_catalog_build_count_for_test().await, 1); + + let executable = std::env::current_exe().unwrap(); + let mut child = openbitfun_services_core::process_manager::create_command( + executable.to_string_lossy().as_ref(), + ); + let output = child + .args([ + "--exact", + "agentic::persistence::manager::tests::catalog_cross_process_writer_child", + "--ignored", + "--nocapture", + ]) + .env( + "OPENBITFUN_CATALOG_WRITER_ROOT", + manager.project_sessions_dir(workspace.path()), + ) + .output() + .unwrap(); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let refreshed = manager + .list_session_metadata_page_with_query( + workspace.path(), + second.next_cursor.as_deref(), + 5, + Some(&query), + ) + .await + .unwrap(); + assert!(refreshed.catalog_reset); + assert_eq!( + refreshed.sessions[0].session_name, + "Renamed by another process" + ); + assert_ne!(refreshed.catalog_revision, first.catalog_revision); + assert_eq!(manager.session_catalog_build_count_for_test().await, 2); + let next = manager + .list_session_metadata_page_with_query( + workspace.path(), + refreshed.next_cursor.as_deref(), + 5, + Some(&query), + ) + .await + .unwrap(); + assert!(!next.catalog_reset); + assert_eq!(next.sessions[0].session_id, "catalog-6"); + assert_eq!(manager.session_catalog_build_count_for_test().await, 2); + + let mut changed_query = query.clone(); + changed_query.source = "codex".into(); + let filtered = manager + .list_session_metadata_page_with_query( + workspace.path(), + next.next_cursor.as_deref(), + 5, + Some(&changed_query), + ) + .await + .unwrap(); + assert!(filtered.catalog_reset); + assert_eq!(filtered.total_top_level_count, 0); + let replay = manager + .list_session_metadata_page_with_query(workspace.path(), None, 5, Some(&query)) + .await + .unwrap(); + assert_eq!(replay.catalog_revision, refreshed.catalog_revision); + assert_eq!(manager.session_catalog_build_count_for_test().await, 3); + + // A deleted selected entry still triggers the existing index repair. + std::fs::remove_file( + manager + .session_layout(workspace.path()) + .metadata_path("catalog-11"), + ) + .unwrap(); + let repaired = manager + .list_session_metadata_page_with_query(workspace.path(), None, 5, Some(&query)) + .await + .unwrap(); + assert_eq!(repaired.total_top_level_count, 11); + assert_eq!(repaired.sessions[0].session_id, "catalog-10"); + assert_eq!(manager.session_catalog_build_count_for_test().await, 4); + } + #[tokio::test] async fn list_session_metadata_page_rebuilds_stale_visible_page_entry() { let workspace = TestWorkspace::new(); diff --git a/src/crates/assembly/core/src/agentic/persistence/manager/session_catalog.rs b/src/crates/assembly/core/src/agentic/persistence/manager/session_catalog.rs new file mode 100644 index 0000000000..7ace50f6f4 --- /dev/null +++ b/src/crates/assembly/core/src/agentic/persistence/manager/session_catalog.rs @@ -0,0 +1,213 @@ +//! Host-owned, bounded navigation snapshots. Persisted indexes and their +//! filesystem revisions remain authoritative, including writes by other hosts. + +use super::*; +use openbitfun_services_core::json_store::JsonFileRevision; +use openbitfun_services_core::session::page::{PreparedSessionCatalog, SessionCatalogQuery}; +use std::collections::{HashSet, VecDeque}; + +const MAX_CATALOGS: usize = 8; +const MAX_CATALOG_BYTES: u64 = 32 * 1024 * 1024; + +#[derive(Default)] +pub(super) struct CatalogCache { + entries: VecDeque, + #[cfg(test)] + builds: usize, +} + +#[derive(Debug, PartialEq, Eq)] +struct CatalogRevision { + native: Option, + #[cfg(feature = "external-sources")] + external: Option, +} + +impl CatalogRevision { + async fn read(root: &Path) -> OpenBitFunResult { + Ok(Self { + native: JsonFileStore + .revision(&SessionStorageLayout::new(root).index_path()) + .await + .map_err(|error| OpenBitFunError::io(error.to_string()))?, + #[cfg(feature = "external-sources")] + external: crate::external_session_import::deferred::catalog_revision(root).await?, + }) + } + + fn cursor_revision(&self, root: &Path, query: &SessionCatalogQuery) -> String { + let mut sort_timestamps = query.sort_timestamps.iter().collect::>(); + sort_timestamps.sort_unstable_by(|left, right| left.0.cmp(right.0)); + let mut hash = Sha256::new(); + hash.update(format!("{root:?}\n{self:?}\n")); + hash.update( + serde_json::to_vec(&( + &query.workspace_id, + query.ordering, + &query.source, + sort_timestamps, + )) + .expect("catalog query is serializable"), + ); + format!("{:x}", hash.finalize()) + } +} + +struct CachedCatalog { + root: PathBuf, + query: SessionCatalogQuery, + revision: CatalogRevision, + prepared: PreparedSessionCatalog, + native_ids: HashSet, + bytes: u64, +} + +impl CachedCatalog { + fn has_missing_native_entry(&self, page: &SessionMetadataPage) -> bool { + let layout = SessionStorageLayout::new(&self.root); + page.sessions.iter().any(|metadata| { + self.native_ids.contains(&metadata.session_id) + && !layout.metadata_path(&metadata.session_id).exists() + }) + } +} + +impl PersistenceManager { + #[cfg(test)] + pub(crate) async fn session_catalog_build_count_for_test(&self) -> usize { + self.catalog_cache.lock().await.builds + } + + pub(super) async fn list_cached_session_catalog_page( + &self, + workspace: &Path, + cursor: Option<&str>, + limit: usize, + query: &SessionCatalogQuery, + ) -> OpenBitFunResult { + let root = self.project_sessions_dir(workspace); + // Serialize builds so simultaneous startup/expansion requests do not + // parse and sort the same version twice. + let mut cache = self.catalog_cache.lock().await; + let mut before = match CatalogRevision::read(&root).await { + Ok(revision) => revision, + Err(_) => { + // Withdraw the query acknowledgement so clients restart their + // legacy traversal instead of mixing an unversioned page in. + return self + .list_uncached_session_metadata_page(workspace, cursor, limit, None) + .await; + } + }; + let mut rebuild = false; + if let Some(index) = cache + .entries + .iter() + .position(|entry| entry.root == root && entry.query == *query) + { + let entry = cache.entries.remove(index).expect("located catalog"); + if entry.revision == before { + let page = entry.prepared.page(cursor, limit); + if !entry.has_missing_native_entry(&page) { + cache.entries.push_back(entry); + return Ok(page); + } + rebuild = true; + } + } + // Superseded versions cannot be reused by another query either. + cache + .entries + .retain(|entry| entry.root != root || entry.revision == before); + let store = self.session_metadata_store(workspace); + if rebuild { + store + .rebuild_index() + .await + .map_err(Self::session_metadata_store_error)?; + } + let (native, native_revision) = store + .indexed_metadata_snapshot() + .await + .map_err(Self::session_metadata_store_error)?; + before.native = native_revision; + let native_ids = native + .iter() + .map(|entry| entry.session_id.clone()) + .collect(); + #[cfg(feature = "external-sources")] + let metadata = { + match crate::external_session_import::deferred::has_references(&root).await { + Ok(true) => { + crate::external_session_import::deferred::merge_metadata(&root, native).await? + } + Ok(false) => native, + // Preserve the established warning and independently readable + // native rows instead of caching an incomplete merged catalog. + Err(_) => { + return self + .list_uncached_session_metadata_page(workspace, cursor, limit, None) + .await + } + } + }; + #[cfg(not(feature = "external-sources"))] + let metadata = native; + // Account for owned metadata plus lookup tables; oversized projections + // can still serve a page without occupying the persistent cache. + let bytes = serde_json::to_vec(&metadata) + .map(|bytes| bytes.len() as u64 * 2) + .unwrap_or(MAX_CATALOG_BYTES + 1) + .saturating_add( + (metadata.len() * (std::mem::size_of::() + 128)) as u64, + ); + let prepared = PreparedSessionCatalog::new(metadata, Some(query)) + .with_revision(before.cursor_revision(&root, query)); + #[cfg(test)] + { + cache.builds += 1; + } + let entry = CachedCatalog { + root, + query: query.clone(), + revision: before, + prepared, + native_ids, + bytes, + }; + let page = entry.prepared.page(cursor, limit); + if entry.has_missing_native_entry(&page) { + // The index may have been stale before this cache existed. Repair + // through its IO owner, then let the next request cache that version. + store + .rebuild_index() + .await + .map_err(Self::session_metadata_store_error)?; + let mut repaired = self + .list_uncached_session_metadata_page(workspace, None, limit, Some(query)) + .await?; + repaired.catalog_reset = cursor.is_some(); + return Ok(repaired); + } + if page.catalog_query.is_some() + && bytes <= MAX_CATALOG_BYTES + && CatalogRevision::read(&entry.root) + .await + .is_ok_and(|after| after == entry.revision) + { + while cache.entries.len() >= MAX_CATALOGS + || cache + .entries + .iter() + .map(|entry| entry.bytes) + .sum::() + .saturating_add(bytes) + > MAX_CATALOG_BYTES + { + cache.entries.pop_front(); + } + cache.entries.push_back(entry); + } + Ok(page) + } +} diff --git a/src/crates/assembly/core/src/external_session_import/deferred.rs b/src/crates/assembly/core/src/external_session_import/deferred.rs index c8d64b9966..1c8b9f7ec6 100644 --- a/src/crates/assembly/core/src/external_session_import/deferred.rs +++ b/src/crates/assembly/core/src/external_session_import/deferred.rs @@ -14,7 +14,6 @@ use openbitfun_services_core::{json_store::JsonFileStore, session::SessionMetada use serde::{Deserialize, Serialize}; use serde_json::Value; use std::collections::{BTreeMap, HashSet}; -use std::time::SystemTime; const INDEX_FILE: &str = ".external-session-index.json"; const PENDING_KEY: &str = "externalHistoryPending"; @@ -89,24 +88,27 @@ impl Default for Index { } } -type Stamp = (Option, u64); +type Stamp = openbitfun_services_core::json_store::JsonFileRevision; type IndexCache = HashMap)>; static CACHE: LazyLock> = LazyLock::new(Default::default); +pub(crate) async fn catalog_revision(root: &Path) -> OpenBitFunResult> { + JsonFileStore + .revision(&root.join(INDEX_FILE)) + .await + .map_err(index_error) +} + fn index_error(error: impl std::fmt::Display) -> OpenBitFunError { OpenBitFunError::io(format!("External session index is unavailable: {error}")) } async fn read_index(root: &Path) -> OpenBitFunResult> { let path = root.join(INDEX_FILE); - let stat = match tokio::fs::metadata(&path).await { - Ok(stat) => stat, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => { - return Ok(Arc::new(Index::default())) - } - Err(error) => return Err(index_error(error)), + let Some(stamp) = catalog_revision(root).await? else { + CACHE.lock().await.remove(&path); + return Ok(Arc::new(Index::default())); }; - let stamp = (stat.modified().ok(), stat.len()); let mut cache = CACHE.lock().await; if let Some((previous, index)) = cache.get(&path) { if *previous == stamp { @@ -1553,6 +1555,135 @@ mod tests { ); } + #[tokio::test] + async fn catalog_query_pages_native_and_external_metadata_without_materializing_history() { + use openbitfun_services_core::session::page::{ + SessionCatalogOrdering, SessionCatalogQuery, + }; + let (_temp, workspace, root, manager) = fixture(); + let candidates = (0..60) + .map(|index| { + let mut value = candidate( + if index % 2 == 0 { "codex" } else { "zcode" }, + &index.to_string(), + &workspace.root_path, + ); + value.created_at = Some(index + 1); + value.updated_at = Some(1000 - index); + value + }) + .collect(); + register(&root, &workspace, &workspace, candidates) + .await + .unwrap(); + let mut other = workspace.clone(); + other.id = "another-workspace".into(); + register( + &root, + &other, + &workspace, + vec![candidate("codex", "other", &workspace.root_path)], + ) + .await + .unwrap(); + let mut native = SessionMetadata::new( + "native".into(), + "Native".into(), + "agent".into(), + "model".into(), + ); + native.workspace_id = Some(workspace.id.clone()); + native.created_at = 100; + SessionMetadataStore::new(&root) + .save_metadata(&native) + .await + .unwrap(); + let before = tokio::fs::read(root.join(INDEX_FILE)).await.unwrap(); + let mut query = SessionCatalogQuery { + workspace_id: workspace.id.clone(), + ordering: SessionCatalogOrdering::Updated, + source: "codex".into(), + sort_timestamps: HashMap::new(), + }; + let first = manager + .list_session_metadata_page_with_query(&root, None, 5, Some(&query)) + .await + .unwrap(); + assert_eq!(first.catalog_query, Some(query.clone())); + assert_eq!(first.total_top_level_count, 30); + assert_eq!(first.sessions.len(), 5); + assert_eq!(first.sessions[0].session_name, "58"); + let second = manager + .list_session_metadata_page_with_query( + &root, + first.next_cursor.as_deref(), + 5, + Some(&query), + ) + .await + .unwrap(); + assert_eq!(second.sessions[0].session_name, "48"); + assert_eq!(second.catalog_revision, first.catalog_revision); + assert_eq!(manager.session_catalog_build_count_for_test().await, 1); + assert!(first + .sessions + .iter() + .chain(&second.sessions) + .all(|record| !root.join(&record.session_id).exists())); + query.source = "all".into(); + let all = manager + .list_session_metadata_page_with_query(&root, None, 5, Some(&query)) + .await + .unwrap(); + assert_eq!(all.total_top_level_count, 61); + assert_eq!(all.sessions[0].session_id, "native"); + assert_eq!( + tokio::fs::read(root.join(INDEX_FILE)).await.unwrap(), + before + ); + assert_eq!(manager.session_catalog_build_count_for_test().await, 2); + + // Bypass the process-local notification/cache as a foreign writer does. + let mut foreign: Index = serde_json::from_slice(&before).unwrap(); + let renamed_id = first.sessions[0].session_id.clone(); + let renamed = foreign.entries.get_mut(&renamed_id).unwrap(); + renamed.metadata.session_name = "Renamed outside this process".into(); + renamed.metadata.custom_metadata.as_mut().unwrap()["externalSessionTitleOverride"] = + Value::Bool(true); + JsonFileStore + .write_atomic_strict(&root.join(INDEX_FILE), &foreign) + .await + .unwrap(); + query.source = "codex".into(); + let refreshed = manager + .list_session_metadata_page_with_query( + &root, + first.next_cursor.as_deref(), + 5, + Some(&query), + ) + .await + .unwrap(); + assert!(refreshed.catalog_reset); + assert_eq!( + refreshed.sessions[0].session_name, + "Renamed outside this process" + ); + assert_eq!(manager.session_catalog_build_count_for_test().await, 3); + let continued = manager + .list_session_metadata_page_with_query( + &root, + refreshed.next_cursor.as_deref(), + 5, + Some(&query), + ) + .await + .unwrap(); + assert!(!continued.catalog_reset); + assert_eq!(continued.sessions[0].session_name, "48"); + assert_eq!(manager.session_catalog_build_count_for_test().await, 3); + } + #[tokio::test] async fn rename_archive_and_delete_remain_index_only_and_rediscovery_preserves_intent() { let (_temp, workspace, root, manager) = fixture(); @@ -1917,6 +2048,23 @@ mod tests { assert_eq!(page.sessions.len(), 1); assert_eq!(page.sessions[0].session_id, "native"); assert_eq!(page.discovery_warnings.len(), 1); + let query = openbitfun_services_core::session::page::SessionCatalogQuery { + workspace_id: workspace.id.clone(), + ordering: openbitfun_services_core::session::page::SessionCatalogOrdering::Updated, + source: "all".into(), + sort_timestamps: HashMap::new(), + }; + let fallback = manager + .list_session_metadata_page_with_query(&root, Some("old-cursor"), 5, Some(&query)) + .await + .unwrap(); + assert_eq!(fallback.sessions[0].session_id, "native"); + assert_eq!(fallback.discovery_warnings.len(), 1); + assert!( + fallback.catalog_query.is_none(), + "an unavailable revision must restart legacy traversal" + ); + assert_eq!(manager.session_catalog_build_count_for_test().await, 0); assert_eq!(manager.list_session_metadata(&root).await.unwrap().len(), 1); assert_eq!( manager diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs index bc91780522..21d2712ecb 100644 --- a/src/crates/assembly/core/src/product_runtime.rs +++ b/src/crates/assembly/core/src/product_runtime.rs @@ -1270,6 +1270,17 @@ impl CoreAgentRuntimeCompatibility { workspace_path: &Path, cursor: Option<&str>, limit: usize, + ) -> OpenBitFunResult { + self.list_persisted_session_catalog_page(workspace_path, cursor, limit, None) + .await + } + + async fn list_persisted_session_catalog_page( + &self, + workspace_path: &Path, + cursor: Option<&str>, + limit: usize, + query: Option<&openbitfun_services_core::session::page::SessionCatalogQuery>, ) -> OpenBitFunResult { #[cfg(feature = "external-sources")] let warnings = crate::external_session_import::auto_discovery::discover_for_storage_path( @@ -1279,7 +1290,7 @@ impl CoreAgentRuntimeCompatibility { #[allow(unused_mut)] let mut page = self .persistence - .list_session_metadata_page(workspace_path, cursor, limit) + .list_session_metadata_page_with_query(workspace_path, cursor, limit, query) .await?; #[cfg(feature = "external-sources")] { @@ -1299,6 +1310,26 @@ impl CoreAgentRuntimeCompatibility { cursor: Option<&str>, limit: usize, session_ids: Option<&[String]>, + ) -> OpenBitFunResult { + self.list_persisted_sessions_page_with_activity_and_query( + runtime, + workspace_path, + cursor, + limit, + session_ids, + None, + ) + .await + } + + pub async fn list_persisted_sessions_page_with_activity_and_query( + &self, + runtime: &AgentRuntime, + workspace_path: &Path, + cursor: Option<&str>, + limit: usize, + session_ids: Option<&[String]>, + query: Option<&openbitfun_services_core::session::page::SessionCatalogQuery>, ) -> OpenBitFunResult { use futures::StreamExt; use openbitfun_agent_runtime::session_state::SessionState; @@ -1319,7 +1350,7 @@ impl CoreAgentRuntimeCompatibility { let mut page = if session_ids.is_some() { empty_session_metadata_page() } else { - self.list_persisted_sessions_page(workspace_path, cursor, limit) + self.list_persisted_session_catalog_page(workspace_path, cursor, limit, query) .await? }; let selected = match session_ids { diff --git a/src/crates/services/services-core/AGENTS.md b/src/crates/services/services-core/AGENTS.md index d7a3dd7f26..3354db6f5c 100644 --- a/src/crates/services/services-core/AGENTS.md +++ b/src/crates/services/services-core/AGENTS.md @@ -95,6 +95,7 @@ cargo test -p openbitfun-services-core --no-default-features --features workspac cargo test -p openbitfun-services-core --no-default-features --features workspace-runtime --lib workspace::tests:: cargo test -p openbitfun-services-core --no-default-features --features local-storage --test session_contracts session_metadata_contracts:: cargo test -p openbitfun-services-core --no-default-features --features local-storage --lib session::metadata +cargo test -p openbitfun-services-core --no-default-features --features local-storage --lib session::page::tests cargo test -p openbitfun-services-core --no-default-features --features local-storage --lib session::offline_import cargo test -p openbitfun-services-core --no-default-features --features local-storage --test session_write_lock_contracts cargo test -p openbitfun-services-core --no-default-features --features local-storage --test exclusive_file_lease_contracts diff --git a/src/crates/services/services-core/src/json_store.rs b/src/crates/services/services-core/src/json_store.rs index 2ec82fb962..442eb33be3 100644 --- a/src/crates/services/services-core/src/json_store.rs +++ b/src/crates/services/services-core/src/json_store.rs @@ -108,6 +108,54 @@ impl JsonFileStoreError { #[derive(Debug, Default, Clone, Copy)] pub struct JsonFileStore; +/// Filesystem revision for rebuildable index caches. Replacement creation time +/// and (on Unix) inode/change time also detect replacement with preserved mtime. +/// This is not a content identity for externally edited history bodies. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct JsonFileRevision { + pub len: u64, + modified: SystemTime, + created: Option, + #[cfg(unix)] + identity: (u64, u64, i64, i64), +} + +impl JsonFileStore { + pub async fn revision( + &self, + path: &Path, + ) -> Result, JsonFileStoreError> { + let stat = match fs::metadata(path).await { + Ok(stat) => stat, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None), + Err(source) => { + return Err(JsonFileStoreError::ReadMetadata { + path: path.to_owned(), + source, + }) + } + }; + let modified = stat + .modified() + .map_err(|source| JsonFileStoreError::ReadMetadata { + path: path.to_owned(), + source, + })?; + #[cfg(unix)] + let identity = { + use std::os::unix::fs::MetadataExt; + (stat.dev(), stat.ino(), stat.ctime(), stat.ctime_nsec()) + }; + Ok(Some(JsonFileRevision { + len: stat.len(), + modified, + created: stat.created().ok(), + #[cfg(unix)] + identity, + })) + } +} + /// Held OS advisory lock for a JSON-owned transaction. /// /// Callers may use this guard when a read/modify/write transaction includes diff --git a/src/crates/services/services-core/src/session/metadata_store.rs b/src/crates/services/services-core/src/session/metadata_store.rs index 619fdd9764..de57a75847 100644 --- a/src/crates/services/services-core/src/session/metadata_store.rs +++ b/src/crates/services/services-core/src/session/metadata_store.rs @@ -9,7 +9,7 @@ use super::metadata::{ build_session_index_snapshot, remove_session_index_entry, upsert_session_index_entry, }; use super::ordinal; -use super::page::{build_session_metadata_page, empty_session_metadata_page}; +use super::page::{build_session_metadata_page_with_query, SessionCatalogQuery}; use super::types::{SessionMetadata, StoredSessionIndexFile, StoredSessionMetadataFile}; use super::SessionMetadataPage; use crate::file_lock::{FileLock, FileLockError, FileLockMode}; @@ -349,14 +349,36 @@ impl SessionMetadataStore { pub async fn list_indexed_metadata( &self, ) -> Result, SessionMetadataStoreError> { + self.indexed_metadata_snapshot() + .await + .map(|(metadata, _)| metadata) + } + + /// Capture the index revision under the same cross-process lock as its + /// contents, including a missing/corrupt index rebuilt during this read. + pub async fn indexed_metadata_snapshot( + &self, + ) -> Result< + ( + Vec, + Option, + ), + SessionMetadataStoreError, + > { if !self.sessions_root().exists() { - return Ok(Vec::new()); + return Ok((Vec::new(), None)); } let lock = self.get_index_lock().await; let _guard = lock.lock().await; let _file_guard = self.lock_index_file().await?; let (index, _) = self.read_or_rebuild_index_locked().await?; - Ok(compatible_metadata_list(index.sessions)) + let metadata = if index.metadata_file_count < index.sessions.len() { + self.rebuild_index_locked().await? + } else { + index.sessions + }; + let revision = self.json_store.revision(&self.index_path()).await?; + Ok((compatible_metadata_list(metadata), revision)) } pub async fn list_metadata(&self) -> Result, SessionMetadataStoreError> { @@ -428,9 +450,24 @@ impl SessionMetadataStore { &self, cursor: Option<&str>, limit: usize, + ) -> Result { + self.list_metadata_page_with_query(cursor, limit, None) + .await + } + + pub async fn list_metadata_page_with_query( + &self, + cursor: Option<&str>, + limit: usize, + query: Option<&SessionCatalogQuery>, ) -> Result { if !self.sessions_root().exists() { - return Ok(empty_session_metadata_page()); + return Ok(build_session_metadata_page_with_query( + Vec::new(), + cursor, + limit, + query, + )); } let limit = limit.max(1); @@ -451,7 +488,8 @@ impl SessionMetadataStore { index.sessions }; - let mut page = build_session_metadata_page(indexed_sessions, cursor, limit); + let mut page = + build_session_metadata_page_with_query(indexed_sessions, cursor, limit, query); let has_stale_page_entry = page .sessions .iter() @@ -466,7 +504,8 @@ impl SessionMetadataStore { index_path.display() ); let rebuilt_sessions = self.rebuild_index_locked().await?; - let mut page = build_session_metadata_page(rebuilt_sessions, cursor, limit); + let mut page = + build_session_metadata_page_with_query(rebuilt_sessions, cursor, limit, query); page.sessions = compatible_metadata_list(page.sessions); Ok(page) } @@ -642,6 +681,64 @@ mod tests { use super::*; use crate::session::{SessionStatus, StoredSessionIndexFile}; + #[tokio::test] + async fn catalog_first_page_does_not_open_unselected_metadata_or_history() { + let root = tempfile::tempdir().unwrap(); + let store = SessionMetadataStore::new(root.path()); + let sessions = (0..655) + .map(|index| { + let mut record = metadata(&format!("session-{index:04}"), 10_000 - index); + record.workspace_id = Some("main".into()); + record.created_at = index; + record + }) + .collect::>(); + // Only the selected five metadata paths exist. Reading or validating + // every directory would rebuild the index and lose the expected total. + for record in &sessions[650..] { + store.ensure_session_dir(&record.session_id).await.unwrap(); + store + .write_json_atomic( + &store.metadata_path(&record.session_id), + &StoredSessionMetadataFile::new(record.clone()), + ) + .await + .unwrap(); + fs::write( + store.session_dir(&record.session_id).join("turns.json"), + b"unreadable history", + ) + .await + .unwrap(); + } + let index = StoredSessionIndexFile { + schema_version: super::super::types::SESSION_STORAGE_SCHEMA_VERSION, + metadata_file_count: 655, + updated_at: 1, + sessions, + }; + store + .write_json_atomic(&store.index_path(), &index) + .await + .unwrap(); + let before = fs::read(store.index_path()).await.unwrap(); + let query = SessionCatalogQuery { + workspace_id: "main".into(), + ordering: super::super::page::SessionCatalogOrdering::Updated, + source: "all".into(), + sort_timestamps: HashMap::new(), + }; + let page = store + .list_metadata_page_with_query(None, 5, Some(&query)) + .await + .unwrap(); + assert_eq!(page.catalog_query, Some(query)); + assert_eq!(page.total_top_level_count, 655); + assert_eq!(page.sessions.len(), 5); + assert_eq!(page.sessions[0].session_id, "session-0654"); + assert_eq!(fs::read(store.index_path()).await.unwrap(), before); + } + #[tokio::test] async fn legacy_auto_metadata_reads_are_compatible_without_rewriting_files() { let root = tempfile::tempdir().unwrap(); diff --git a/src/crates/services/services-core/src/session/page.rs b/src/crates/services/services-core/src/session/page.rs index ec15477cdc..862a992fd3 100644 --- a/src/crates/services/services-core/src/session/page.rs +++ b/src/crates/services/services-core/src/session/page.rs @@ -4,6 +4,27 @@ use super::types::{SessionMetadata, SessionStatus}; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; +/// Read-only navigation query. A host echoes it only after applying every +/// field; clients without that acknowledgement must retain legacy traversal. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct SessionCatalogQuery { + pub workspace_id: String, + pub ordering: SessionCatalogOrdering, + pub source: String, + /// The view freezes recency on hydration and advances it on a new run. + /// Preserve those already-observed timestamps when selecting older rows. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub sort_timestamps: HashMap, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SessionCatalogOrdering { + Created, + Updated, +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct SessionMetadataPage { @@ -22,6 +43,12 @@ pub struct SessionMetadataPage { pub discovery_warnings: Vec, #[serde(default, skip_serializing_if = "is_false")] pub discovery_pending: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub catalog_query: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub catalog_revision: Option, + #[serde(default, skip_serializing_if = "is_false")] + pub catalog_reset: bool, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -46,6 +73,8 @@ pub struct SessionActivitySummary { struct SessionMetadataPageCursor { last_active_at: u64, session_id: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + catalog_revision: Option, } pub fn empty_session_metadata_page() -> SessionMetadataPage { @@ -58,6 +87,9 @@ pub fn empty_session_metadata_page() -> SessionMetadataPage { activities: None, discovery_warnings: Vec::new(), discovery_pending: false, + catalog_query: None, + catalog_revision: None, + catalog_reset: false, } } @@ -66,73 +98,304 @@ pub fn build_session_metadata_page( cursor: Option<&str>, limit: usize, ) -> SessionMetadataPage { - let visible_sessions = indexed_sessions - .into_iter() - .filter(|metadata| { - !metadata.should_hide_from_user_lists() && metadata.status != SessionStatus::Archived - }) - .collect::>(); - let visible_ids = visible_sessions - .iter() - .map(|metadata| metadata.session_id.clone()) - .collect::>(); - - let mut top_level_sessions = Vec::new(); - let mut children_by_parent: HashMap> = HashMap::new(); - for metadata in visible_sessions { - if let Some(parent_id) = session_parent_id(&metadata) { - if visible_ids.contains(&parent_id) { - children_by_parent - .entry(parent_id) - .or_default() - .push(metadata); - continue; + build_session_metadata_page_with_query(indexed_sessions, cursor, limit, None) +} + +pub fn build_session_metadata_page_with_query( + indexed_sessions: Vec, + cursor: Option<&str>, + limit: usize, + query: Option<&SessionCatalogQuery>, +) -> SessionMetadataPage { + PreparedSessionCatalog::new(indexed_sessions, query).page(cursor, limit) +} + +/// Immutable navigation projection. Hosts can reuse it for a stable index +/// revision; each page clones only its selected parents and their children. +pub struct PreparedSessionCatalog { + top_level_sessions: Vec, + children_by_parent: HashMap>, + offsets: HashMap, + query: Option, + revision: Option, +} + +impl PreparedSessionCatalog { + pub fn new( + indexed_sessions: Vec, + query: Option<&SessionCatalogQuery>, + ) -> Self { + // Older records can require the client's workspace registry to resolve + // ownership. Do not guess from the shared storage directory. + let query = query.filter(|query| { + !query.workspace_id.is_empty() + && indexed_sessions.iter().all(|metadata| { + metadata.should_hide_from_user_lists() + || metadata.status == SessionStatus::Archived + || catalog_hides_subagent(metadata) + || catalog_workspace_id(metadata).is_some() + }) + }); + let visible_sessions = indexed_sessions + .into_iter() + .filter(|metadata| { + !metadata.should_hide_from_user_lists() + && metadata.status != SessionStatus::Archived + && query.is_none_or(|query| { + catalog_workspace_id(metadata) == Some(query.workspace_id.as_str()) + && !catalog_hides_subagent(metadata) + }) + }) + .collect::>(); + let visible_ids = visible_sessions + .iter() + .map(|metadata| metadata.session_id.clone()) + .collect::>(); + + let mut top_level_sessions = Vec::new(); + let mut children_by_parent: HashMap> = HashMap::new(); + for metadata in visible_sessions { + let parent_id = if query.is_some() { + catalog_parent_id(&metadata) + } else { + session_parent_id(&metadata) + }; + if let Some(parent_id) = parent_id { + if visible_ids.contains(&parent_id) { + children_by_parent + .entry(parent_id) + .or_default() + .push(metadata); + continue; + } } + + top_level_sessions.push(metadata); + } + + if let Some(query) = query { + top_level_sessions.retain(|metadata| catalog_source_matches(metadata, &query.source)); + top_level_sessions.sort_by(|left, right| catalog_order(left, right, query)); } - top_level_sessions.push(metadata); - } - - let total_top_level_count = top_level_sessions.len(); - let limit = limit.max(1); - let offset = session_metadata_page_offset(cursor, &top_level_sessions); - let offset = offset.min(total_top_level_count); - let next_offset = offset.saturating_add(limit).min(total_top_level_count); - let selected_top_level = top_level_sessions - .iter() - .skip(offset) - .take(limit) - .cloned() - .collect::>(); - let loaded_top_level_count = selected_top_level.len(); - let has_more = next_offset < total_top_level_count; - let next_cursor = has_more - .then(|| selected_top_level.last().map(session_metadata_page_cursor)) - .flatten(); - - let mut sessions = Vec::new(); - for metadata in selected_top_level { - let session_id = metadata.session_id.clone(); - sessions.push(metadata); - - if let Some(mut children) = children_by_parent.remove(&session_id) { - children.sort_by_key(|metadata| std::cmp::Reverse(metadata.last_active_at)); - sessions.extend(children); + for children in children_by_parent.values_mut() { + if let Some(query) = query { + children.retain(|metadata| catalog_source_matches(metadata, &query.source)); + children.sort_by(|left, right| catalog_order(left, right, query)); + } else { + children.sort_by_key(|metadata| std::cmp::Reverse(metadata.last_active_at)); + } + } + let offsets = top_level_sessions + .iter() + .enumerate() + .map(|(index, metadata)| (metadata.session_id.clone(), index)) + .collect(); + Self { + top_level_sessions, + children_by_parent, + offsets, + query: query.cloned(), + revision: None, } } - SessionMetadataPage { - sessions, - total_top_level_count, - loaded_top_level_count, - next_cursor, - has_more, - activities: None, - discovery_warnings: Vec::new(), - discovery_pending: false, + pub fn with_revision(mut self, revision: String) -> Self { + if self.query.is_some() { + self.revision = Some(revision); + } + self + } + + pub fn page(&self, cursor: Option<&str>, limit: usize) -> SessionMetadataPage { + let query = self.query.as_ref(); + let top_level_sessions = &self.top_level_sessions; + let catalog_reset = self.revision.as_ref().is_some_and(|revision| { + cursor.is_some_and(|cursor| { + serde_json::from_str::(cursor) + .ok() + .and_then(|cursor| cursor.catalog_revision) + .as_ref() + != Some(revision) + }) + }); + let cursor = cursor.filter(|_| !catalog_reset); + + let total_top_level_count = top_level_sessions.len(); + let limit = limit.max(1); + let offset = session_metadata_page_offset(cursor, &self.offsets); + let offset = offset.min(total_top_level_count); + let mut next_offset = offset.saturating_add(limit).min(total_top_level_count); + if let Some(query) = query { + // The client uses locale-aware ID ordering for equal timestamps. Keep + // the entire boundary tie together so it can preserve that order. + while next_offset > offset + && next_offset < total_top_level_count + && catalog_sort_key(&top_level_sessions[next_offset - 1], query) + == catalog_sort_key(&top_level_sessions[next_offset], query) + { + next_offset += 1; + } + } + let selected_top_level = top_level_sessions + .iter() + .skip(offset) + .take(next_offset - offset) + .cloned() + .collect::>(); + let loaded_top_level_count = selected_top_level.len(); + let has_more = next_offset < total_top_level_count; + let next_cursor = has_more + .then(|| { + selected_top_level + .last() + .map(|metadata| session_metadata_page_cursor(metadata, self.revision.clone())) + }) + .flatten(); + + let mut sessions = Vec::new(); + for metadata in selected_top_level { + let session_id = metadata.session_id.clone(); + sessions.push(metadata); + + if let Some(children) = self.children_by_parent.get(&session_id) { + sessions.extend(children.iter().cloned()); + } + } + + SessionMetadataPage { + sessions, + total_top_level_count, + loaded_top_level_count, + next_cursor, + has_more, + activities: None, + discovery_warnings: Vec::new(), + discovery_pending: false, + catalog_query: query.cloned(), + catalog_revision: self.revision.clone(), + catalog_reset, + } + } +} + +fn catalog_workspace_id(metadata: &SessionMetadata) -> Option<&str> { + use openbitfun_core_types::SessionExecutionTargetKind; + let isolated = metadata + .execution_target + .as_ref() + .is_some_and(|target| target.kind != SessionExecutionTargetKind::Local); + if isolated { + // Missing project identity requires the legacy workspace resolver. + metadata.project_workspace_id.as_deref() + } else { + metadata.workspace_id.as_deref() } } +fn catalog_source_matches(metadata: &SessionMetadata, source: &str) -> bool { + source == "all" + || metadata + .custom_metadata + .as_ref() + .and_then(|custom| custom.get("importedFrom")) + .and_then(serde_json::Value::as_str) + .filter(|value| !value.is_empty()) + .unwrap_or("openbitfun") + == source +} + +fn catalog_parent_id(metadata: &SessionMetadata) -> Option { + use super::types::SessionRelationshipKind; + // Match the existing UI projection: structured Btw and Review sessions are + // children; legacy unstructured Btw and MiniApp records are standalone. + if let Some(kind) = metadata + .relationship + .as_ref() + .and_then(|relationship| relationship.kind.as_ref()) + { + if !matches!( + kind, + SessionRelationshipKind::Btw + | SessionRelationshipKind::Review + | SessionRelationshipKind::DeepReview + ) { + return None; + } + return metadata + .relationship + .as_ref() + .and_then(|relationship| relationship.parent_session_id.as_deref()) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned); + } + let kind = metadata + .custom_metadata + .as_ref() + .and_then(|custom| custom.get("kind")) + .and_then(serde_json::Value::as_str); + matches!(kind, Some("review" | "deep_review")) + .then(|| { + metadata + .custom_metadata + .as_ref() + .and_then(|custom| custom.get("parentSessionId")) + .and_then(serde_json::Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + .map(str::to_owned) + }) + .flatten() +} + +fn catalog_hides_subagent(metadata: &SessionMetadata) -> bool { + if let Some(kind) = metadata + .relationship + .as_ref() + .and_then(|relationship| relationship.kind.as_ref()) + { + return *kind == super::types::SessionRelationshipKind::Subagent; + } + metadata + .custom_metadata + .as_ref() + .and_then(|custom| custom.get("kind")) + .and_then(serde_json::Value::as_str) + == Some("subagent") +} + +fn catalog_sort_key(metadata: &SessionMetadata, query: &SessionCatalogQuery) -> (u64, u64) { + let timestamp = if query.ordering == SessionCatalogOrdering::Created { + metadata.created_at + } else { + query + .sort_timestamps + .get(&metadata.session_id) + .copied() + .or(metadata.last_finished_at) + .or_else(|| { + metadata + .custom_metadata + .as_ref() + .and_then(|custom| custom.get("lastFinishedAt")) + .and_then(serde_json::Value::as_u64) + }) + .unwrap_or(metadata.created_at) + }; + (timestamp, metadata.created_at) +} + +fn catalog_order( + left: &SessionMetadata, + right: &SessionMetadata, + query: &SessionCatalogQuery, +) -> std::cmp::Ordering { + catalog_sort_key(right, query) + .cmp(&catalog_sort_key(left, query)) + .then(left.session_id.cmp(&right.session_id)) +} + fn session_parent_id(metadata: &SessionMetadata) -> Option { if let Some(parent_id) = metadata .relationship @@ -162,26 +425,13 @@ fn is_false(value: &bool) -> bool { !value } -fn session_metadata_page_offset( - cursor: Option<&str>, - top_level_sessions: &[SessionMetadata], -) -> usize { +fn session_metadata_page_offset(cursor: Option<&str>, offsets: &HashMap) -> usize { let Some(cursor) = cursor else { return 0; }; if let Ok(parsed) = serde_json::from_str::(cursor) { - if let Some(index) = top_level_sessions.iter().position(|metadata| { - metadata.session_id == parsed.session_id - && metadata.last_active_at == parsed.last_active_at - }) { - return index + 1; - } - - if let Some(index) = top_level_sessions - .iter() - .position(|metadata| metadata.session_id == parsed.session_id) - { + if let Some(index) = offsets.get(&parsed.session_id) { return index + 1; } } @@ -189,10 +439,11 @@ fn session_metadata_page_offset( cursor.parse::().unwrap_or(0) } -fn session_metadata_page_cursor(metadata: &SessionMetadata) -> String { +fn session_metadata_page_cursor(metadata: &SessionMetadata, revision: Option) -> String { serde_json::to_string(&SessionMetadataPageCursor { last_active_at: metadata.last_active_at, session_id: metadata.session_id.clone(), + catalog_revision: revision, }) .unwrap_or_else(|_| metadata.session_id.clone()) } @@ -208,4 +459,255 @@ mod tests { assert!(page.discovery_warnings.is_empty()); assert_eq!(serde_json::to_value(page).unwrap(), legacy); } + + fn entry(id: &str, workspace: &str, created: u64) -> SessionMetadata { + let mut metadata = + SessionMetadata::new(id.into(), id.into(), "agent".into(), "model".into()); + metadata.workspace_id = Some(workspace.into()); + metadata.created_at = created; + metadata.last_active_at = 10_000 - created; + metadata + } + + fn query(source: &str) -> SessionCatalogQuery { + SessionCatalogQuery { + workspace_id: "main".into(), + ordering: SessionCatalogOrdering::Updated, + source: source.into(), + sort_timestamps: HashMap::new(), + } + } + + #[test] + fn catalog_pages_scope_and_sort_before_selecting_rows() { + let entries = (0..655) + .flat_map(|index| { + [ + entry(&format!("main-{index:04}"), "main", index), + entry(&format!("other-{index:04}"), "other", 1000 + index), + ] + }) + .collect::>(); + let query = query("all"); + let first = build_session_metadata_page_with_query(entries.clone(), None, 5, Some(&query)); + assert_eq!(first.sessions.len(), 5); + assert_eq!(first.total_top_level_count, 655); + assert_eq!(first.loaded_top_level_count, 5); + assert_eq!(first.sessions[0].session_id, "main-0654"); + assert_eq!(first.sessions[4].session_id, "main-0650"); + assert_eq!(first.catalog_query, Some(query.clone())); + let second = build_session_metadata_page_with_query( + entries, + first.next_cursor.as_deref(), + 5, + Some(&query), + ); + assert_eq!(second.sessions.len(), 5); + assert_eq!(second.sessions[0].session_id, "main-0649"); + assert_eq!(second.sessions[4].session_id, "main-0645"); + } + + #[test] + fn catalog_recency_preserves_legacy_finish_times_and_observed_run_order() { + let mut finished = entry("finished", "main", 1); + finished.custom_metadata = Some(serde_json::json!({"lastFinishedAt": 20})); + let mut explicit = entry("explicit", "main", 2); + explicit.last_finished_at = Some(30); + let entries = vec![ + entry("created", "main", 10), + finished, + explicit, + entry("running", "main", 0), + ]; + let mut query = query("all"); + query.sort_timestamps.insert("running".into(), 40); + let page = build_session_metadata_page_with_query(entries.clone(), None, 5, Some(&query)); + assert_eq!( + page.sessions + .iter() + .map(|value| value.session_id.as_str()) + .collect::>(), + ["running", "explicit", "finished", "created"] + ); + query.ordering = SessionCatalogOrdering::Created; + let page = build_session_metadata_page_with_query(entries, None, 5, Some(&query)); + assert_eq!( + page.sessions + .iter() + .map(|value| value.session_id.as_str()) + .collect::>(), + ["created", "explicit", "finished", "running"] + ); + } + + #[test] + fn catalog_source_counts_keep_children_with_the_selected_parent() { + let mut parent = entry("codex", "main", 1); + parent.custom_metadata = Some(serde_json::json!({"importedFrom": "codex"})); + let mut child = entry("review", "main", 9); + child.custom_metadata = Some( + serde_json::json!({"kind": "review", "parentSessionId": "codex", "importedFrom": "codex"}), + ); + let mut native_child = entry("native-review", "main", 10); + native_child.custom_metadata = + Some(serde_json::json!({"kind": "review", "parentSessionId": "codex"})); + let mut archived = entry("archived", "main", 100); + archived.status = SessionStatus::Archived; + let entries = vec![ + parent, + child, + native_child, + archived, + entry("native", "main", 2), + ]; + let page = + build_session_metadata_page_with_query(entries.clone(), None, 1, Some(&query("codex"))); + assert_eq!(page.total_top_level_count, 1); + assert_eq!(page.loaded_top_level_count, 1); + assert_eq!( + page.sessions + .iter() + .map(|value| value.session_id.as_str()) + .collect::>(), + ["codex", "review"] + ); + let page = + build_session_metadata_page_with_query(entries, None, 1, Some(&query("openbitfun"))); + assert_eq!(page.total_top_level_count, 1); + assert_eq!(page.sessions[0].session_id, "native"); + } + + #[test] + fn catalog_worktrees_use_execution_kind_and_legacy_ownership_remains_unscoped() { + let mut isolated = entry("isolated", "worktree", 1); + isolated.project_workspace_id = Some("main".into()); + isolated.execution_target = Some( + serde_json::from_value(serde_json::json!({ + "kind": "managedWorktree", "rootPath": "/remote/project/worktree" + })) + .unwrap(), + ); + let mut linked = entry("linked", "worktree", 2); + linked.project_workspace_id = Some("main".into()); + let entries = vec![isolated, linked]; + let page = + build_session_metadata_page_with_query(entries.clone(), None, 5, Some(&query("all"))); + assert_eq!(page.sessions.len(), 1); + assert_eq!(page.sessions[0].session_id, "isolated"); + let mut legacy = entries; + legacy[0].project_workspace_id = None; + let page = + build_session_metadata_page_with_query(legacy.clone(), None, 5, Some(&query("all"))); + assert!(page.catalog_query.is_none()); + assert_eq!( + serde_json::to_value(page).unwrap(), + serde_json::to_value(build_session_metadata_page(legacy, None, 5)).unwrap() + ); + } + + #[test] + fn catalog_empty_query_is_acknowledged_and_query_round_trips() { + let query = query("codex"); + let page = build_session_metadata_page_with_query(Vec::new(), None, 5, Some(&query)); + assert_eq!(page.total_top_level_count, 0); + assert!(!page.has_more); + let wire = serde_json::to_value(page).unwrap(); + let restored: SessionMetadataPage = serde_json::from_value(wire).unwrap(); + assert_eq!(restored.catalog_query, Some(query)); + } + + #[test] + fn catalog_keeps_timestamp_ties_together_for_client_locale_ordering() { + let entries = vec![ + entry("Z", "main", 9), + entry("_a", "main", 9), + entry("-a", "main", 9), + entry("a", "main", 8), + ]; + let first = + build_session_metadata_page_with_query(entries.clone(), None, 1, Some(&query("all"))); + assert_eq!(first.loaded_top_level_count, 3); + assert!(first.sessions.iter().all(|record| record.created_at == 9)); + let next = build_session_metadata_page_with_query( + entries, + first.next_cursor.as_deref(), + 1, + Some(&query("all")), + ); + assert_eq!(next.sessions[0].session_id, "a"); + assert!(!next.has_more); + } + + #[test] + fn catalog_cursor_restarts_on_revision_change_and_old_pages_remain_compatible() { + let entries = (0..12) + .map(|index| entry(&index.to_string(), "main", index)) + .collect::>(); + let first_catalog = PreparedSessionCatalog::new(entries.clone(), Some(&query("all"))) + .with_revision("query-a-version-1".into()); + let first = first_catalog.page(None, 5); + let next = first_catalog.page(first.next_cursor.as_deref(), 5); + assert_eq!(next.sessions[0].session_id, "6"); + assert!(!next.catalog_reset); + let changed = PreparedSessionCatalog::new(entries.clone(), Some(&query("all"))) + .with_revision("query-a-version-2".into()) + .page(first.next_cursor.as_deref(), 5); + assert!(changed.catalog_reset); + assert_eq!(changed.sessions[0].session_id, "11"); + let wire = serde_json::to_value(&changed).unwrap(); + let restored: SessionMetadataPage = serde_json::from_value(wire).unwrap(); + assert_eq!( + restored.catalog_revision.as_deref(), + Some("query-a-version-2") + ); + assert!(restored.catalog_reset); + // Hosts that ignore optional query/revision fields can still read the + // legacy ID anchor embedded in the new opaque cursor. + let legacy = build_session_metadata_page(entries, first.next_cursor.as_deref(), 5); + assert!(legacy.catalog_revision.is_none()); + assert!(!legacy.catalog_reset); + assert_eq!(legacy.sessions[0].session_id, "8"); + } + + #[test] + fn catalog_preserves_structured_and_legacy_child_visibility() { + let parent = entry("parent", "main", 1); + let mut legacy_btw = entry("legacy-btw", "main", 2); + legacy_btw.custom_metadata = + Some(serde_json::json!({"kind": "btw", "parentSessionId": "parent"})); + let mut structured_btw = legacy_btw.clone(); + structured_btw.session_id = "structured-btw".into(); + structured_btw.relationship = Some( + serde_json::from_value(serde_json::json!({"kind": "btw", "parentSessionId": "parent"})) + .unwrap(), + ); + let mut subagent = entry("legacy-subagent", "main", 3); + subagent.custom_metadata = + Some(serde_json::json!({"kind": "subagent", "parentSessionId": "parent"})); + subagent.workspace_id = None; + let mut miniapp = entry("miniapp", "main", 4); + miniapp.custom_metadata = + Some(serde_json::json!({"kind": "review", "parentSessionId": "parent"})); + miniapp.relationship = Some( + serde_json::from_value( + serde_json::json!({"kind": "miniapp", "parentSessionId": "parent"}), + ) + .unwrap(), + ); + let page = build_session_metadata_page_with_query( + vec![parent, legacy_btw, structured_btw, subagent, miniapp], + None, + 5, + Some(&query("all")), + ); + assert!(page.catalog_query.is_some()); + assert_eq!(page.total_top_level_count, 3); + assert_eq!( + page.sessions + .iter() + .map(|record| record.session_id.as_str()) + .collect::>(), + ["miniapp", "legacy-btw", "parent", "structured-btw"] + ); + } } diff --git a/src/web-ui/AGENTS.md b/src/web-ui/AGENTS.md index 5e1f7598c5..8d8c77eec6 100644 --- a/src/web-ui/AGENTS.md +++ b/src/web-ui/AGENTS.md @@ -147,6 +147,10 @@ pnpm --dir src/web-ui run test:run src/flow_chat/components/ExternalHistoryNotic These local store/DOM fixtures cover source-version pinning, device switches and explicit Peer gating; they do not validate native WebView or remote-host behavior. +For workspace catalog pagination, source filters, count/expansion behavior and +old-host fallback, run `pnpm --dir src/web-ui run test:run src/app/components/NavPanel/sections/sessions src/app/components/NavPanel/workspaceSessionView.test.ts src/flow_chat/store/FlowChatStore.test.ts src/infrastructure/api/service-api/SessionAPI.test.ts`. +These fixtures verify bounded page requests and transport payloads, not live remote connections. + For application update discovery, skip persistence, notification timing, and download/install transitions, run the focused behavior tests (these do not establish visual fidelity): diff --git a/src/web-ui/src/app/components/NavPanel/sections/sessions/SessionsSection.tsx b/src/web-ui/src/app/components/NavPanel/sections/sessions/SessionsSection.tsx index 31b92c8fee..f05e4deaaa 100644 --- a/src/web-ui/src/app/components/NavPanel/sections/sessions/SessionsSection.tsx +++ b/src/web-ui/src/app/components/NavPanel/sections/sessions/SessionsSection.tsx @@ -21,7 +21,7 @@ import { flowChatManager } from '../../../../../flow_chat/services/FlowChatManag import type { FlowChatState, Session } from '../../../../../flow_chat/types/flow-chat'; import { useSceneStore } from '../../../../stores/sceneStore'; import { useWorkspaceContext } from '@/infrastructure/contexts/WorkspaceContext'; -import { sessionAPI } from '@/infrastructure/api/service-api/SessionAPI'; +import { sessionAPI, type SessionCatalogQuery, type SessionMetadataPage } from '@/infrastructure/api/service-api/SessionAPI'; import { createLogger } from '@/shared/utils/logger'; import { isSamePath } from '@/shared/utils/pathUtils'; import { getAppearanceOverlayHost } from '@/infrastructure/appearance/runtime/AppearanceOverlayHost'; @@ -61,6 +61,7 @@ import { useSessionComposerStore } from '@/flow_chat/store/sessionComposerStore' import { getActiveSurfaceId, getActiveSurfaceScope, isSurfaceChangedError, surfaceScopedKey } from '@/infrastructure/peer-device/deviceSurface'; import { getWorkspaceSessionHierarchy } from './sessionNavHierarchy'; import { useWorkspaceSessionCatalog } from './useWorkspaceSessionCatalog'; +import { countPersistedWorkspaceCatalogSessions, workspaceSessionCatalogQuery, workspaceSessionCatalogTotal } from './workspaceSessionCatalogQuery'; import { useWorkspaceSessionDiscovery } from './useWorkspaceSessionDiscovery'; import { useSessionNavExpansion } from './useSessionNavExpansion'; import { isSessionNavRowActive } from './sessionNavSelection'; @@ -274,6 +275,7 @@ const SessionsSection: React.FC = ({ loadedScope: string | null; isLoading: boolean; loadError: boolean; + firstPage?: SessionMetadataPage; }>({ loadedScope: null, isLoading: false, @@ -298,12 +300,6 @@ const SessionsSection: React.FC = ({ onDiscoveryCompleted, ); const observeDiscovery = discovery.observe; - const catalog = useWorkspaceSessionCatalog({ - workspaceId, - enabled: isVisible && !workspaceScopes?.length && metadataPageState.loadedScope === initialMetadataKey, - refreshVersion: catalogRefreshVersion, - onPage: observeDiscovery, - }); const [openMenuSessionId, setOpenMenuSessionId] = useState(null); /** Second level of the session menu: pick what a Markdown export includes. */ const [isExportScopeMenu, setIsExportScopeMenu] = useState(false); @@ -320,6 +316,27 @@ const SessionsSection: React.FC = ({ void orderingRevision; return new Set([...flowChatState.sessions.keys()].filter(sessionNavStatusService.isRunning)); }, [flowChatState.sessions, orderingRevision]); + const catalogQueryJson = useMemo(() => { + void orderingRevision; + return JSON.stringify(workspaceSessionCatalogQuery(workspaceId, sessionOrdering, sessionShow, + sessionFilters, sourceAgent, flowChatState.sessions.values(), sessionNavStatusService.getSortTimestamp) ?? null); + }, [workspaceId, sessionOrdering, sessionShow, sessionFilters, sourceAgent, flowChatState.sessions, orderingRevision]); + // Streaming changes Session objects without changing the catalog query. + // Keep the startup scheduler and page reader stable across those updates. + const catalogQuery = useMemo(() => (JSON.parse(catalogQueryJson) ?? undefined) as SessionCatalogQuery | undefined, [catalogQueryJson]); + const getKnownCatalogCount = useCallback(() => countPersistedWorkspaceCatalogSessions( + flowChatStore.getState().sessions.values(), workspaceId, sourceAgent, + ), [workspaceId, sourceAgent]); + const catalog = useWorkspaceSessionCatalog({ + workspaceId, + enabled: isVisible && !workspaceScopes?.length && metadataPageState.loadedScope === initialMetadataKey, + refreshVersion: catalogRefreshVersion, + onPage: observeDiscovery, + query: catalogQuery, + targetCount: expandLevel === 2 ? level2DisplayCount : expandLevel === 1 ? SESSIONS_LEVEL_1 : SESSIONS_LEVEL_0, + firstPage: metadataPageState.loadedScope === initialMetadataKey ? metadataPageState.firstPage : undefined, + getKnownCount: getKnownCatalogCount, + }); const [scheduledJobsSessionId, setScheduledJobsSessionId] = useState(null); const cronJobCountsRevision = useSyncExternalStore( subscribeCronJobCounts, @@ -499,7 +516,7 @@ const SessionsSection: React.FC = ({ try { const page = await flowChatStore.loadSessionMetadataPage( workspaceId, limit, cursor, - source + source, { query: catalogQuery }, ); if (metadataLoadRequestIdRef.current === requestId && surface.isCurrent()) { observeDiscovery(workspaceId, page); @@ -507,6 +524,7 @@ const SessionsSection: React.FC = ({ loadedScope: initialMetadataKey, isLoading: false, loadError: false, + firstPage: page, }); } return page; @@ -522,7 +540,7 @@ const SessionsSection: React.FC = ({ return null; } }, - [initialMetadataKey, observeDiscovery, surface, workspaceId] + [catalogQuery, initialMetadataKey, observeDiscovery, surface, workspaceId] ); const loadInitialMetadataPage = useCallback( @@ -747,9 +765,16 @@ const SessionsSection: React.FC = ({ }); }, [topLevelSessions.length, expandLevel, level2DisplayCount, showAllWithoutLimit]); - // A storage root can include several workspace rows, including the main - // checkout. Its page total is never the total for this filtered list. - const totalTopLevelSessionCount = topLevelSessions.length; + // Only an acknowledged workspace query provides a scoped total. Include + // newly-created local rows while their metadata publication is in flight. + const totalTopLevelSessionCount = workspaceSessionCatalogTotal(topLevelSessions.length, catalog); + const knownCatalogCount = getKnownCatalogCount(); + useEffect(() => { + // Page hydration updates the baseline inside the reader. A later local + // creation/removal/archive needs a new bounded first page to refill rows. + if (isVisible && catalog.paged && catalog.complete && !catalog.loading + && catalog.knownCount !== knownCatalogCount) setCatalogRefreshVersion(value => value + 1); + }, [catalog.complete, catalog.knownCount, catalog.loading, catalog.paged, isVisible, knownCatalogCount]); const isSessionCountComplete = !workspaceId || Boolean(workspaceScopes?.length) || (catalog.complete && !discovery.pending && !discovery.failed); const expandToggleState = getSessionExpandToggleState(totalTopLevelSessionCount, expandLevel); @@ -1950,22 +1975,23 @@ const SessionsSection: React.FC = ({ {aggregateLoadStatus} - {!showAllWithoutLimit && expandLevel === 2 && topLevelSessions.length > sessionDisplayLimit && ( + {!showAllWithoutLimit && expandLevel === 2 && totalTopLevelSessionCount > sessionDisplayLimit && (