From df57dd757e9cf42d96128e32a077e6cf90cceadc Mon Sep 17 00:00:00 2001 From: kev1n77 Date: Sat, 17 Oct 2026 17:35:30 +0800 Subject: [PATCH] fix(sessions): restore valid modes for imported history branches --- .../external-ai-work-sources-design.md | 6 +- .../src/agentic/coordination/coordinator.rs | 104 ++++++++++++++++++ .../src/external_session_import/deferred.rs | 8 +- .../deferred/legacy.rs | 55 +++++++++ .../deferred/tracking.rs | 84 ++++++++++++++ .../deferred/tracking/history_cache/tests.rs | 30 +++-- .../src/session/metadata_store.rs | 104 +++++++++++++++++- .../services-core/src/session/types.rs | 35 ++++++ .../flow-chat-manager/MessageModule.test.ts | 49 +++++++++ .../flow-chat-manager/MessageModule.ts | 15 ++- .../flow-chat-manager/SessionModule.test.ts | 29 +++++ .../flow-chat-manager/SessionModule.ts | 4 +- 12 files changed, 509 insertions(+), 14 deletions(-) diff --git a/docs/architecture/extensions/external-ai-work-sources-design.md b/docs/architecture/extensions/external-ai-work-sources-design.md index e90bb10432..78c0f3ade3 100644 --- a/docs/architecture/extensions/external-ai-work-sources-design.md +++ b/docs/architecture/extensions/external-ai-work-sources-design.md @@ -85,7 +85,11 @@ MCP 副本的连接状态不借用外部来源 activation，页面明确“尚 不支持的消息角色、工具状态、损坏记录和超出读取预算的历史明确失败，不以空历史伪装成功。 产品装配按来源选择 Codex 或 Zcode reader，复用工作区映射、原子写入及 Session owner； -续聊分支使用宿主当前配置的主模型。来源生态与 Session ID 共同参与匹配，避免跨生态混认。 +续聊分支使用宿主当前配置的主模型，自动发现的来源视图以 `Standard` 作为原生续聊模式；模型选择值 +`primary` 不作为会话模式。旧索引、快照及已有续聊分支中带 Codex/Zcode 来源身份的 `primary` 模式 +在读取时兼容为 `Standard`，不重写原始历史；其他自定义模式和外部运行时绑定保持原样。 +分叉后发送使用主机返回的分支模式，并保留用户另选的模式；旧失败队列仍携带 `primary` 时，重试使用 +已修正分支的当前模式。来源生态与 Session ID 共同参与匹配，避免跨生态混认。 本功能开发阶段的手动列表、计划、执行和取消命令已随旧界面移除；当前入口使用原生会话列表、 历史读取和分叉协议。已有手动导入副本、旧索引及历史修复路径继续兼容，不删除用户数据。 Zcode 会话读取不意味着其 ACP 运行时、设置、Skill、MCP 或 Hook 已接入，其余类别仍显示真实支持状态。 diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs index eda2bb0782..edb8f6222b 100644 --- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs +++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs @@ -16526,6 +16526,110 @@ mod tests { assert!(error.to_string().contains("candidate_unavailable")); } + #[tokio::test] + async fn legacy_imported_primary_branches_resolve_real_turn_modes() { + use crate::agentic::persistence::SessionBranchRequest; + use openbitfun_services_core::session::{ + DialogTurnData, DialogTurnKind, OfflineSessionBundle, OfflineSessionImportStore, + SessionBranchBoundary, SessionMetadata, TurnStatus, UserMessageData, + }; + + let temp = tempfile::tempdir().unwrap(); + let project = temp.path().join("project"); + std::fs::create_dir_all(&project).unwrap(); + let workspace = + crate::service::workspace::legacy_compat::register_local_fixture_blocking(&project); + let paths = Arc::new(PathManager::with_user_root_for_tests( + temp.path().join("user"), + )); + let root = paths.project_sessions_dir(&workspace.root_path); + let persistence = PersistenceManager::new(paths).unwrap(); + let store = OfflineSessionImportStore::new(&root); + for ecosystem in ["codex", "zcode"] { + let id = format!("legacy-{ecosystem}-branch"); + let turn_id = format!("{id}-turn-0"); + let metadata: SessionMetadata = serde_json::from_value(serde_json::json!({ + "session_id": id, "session_name": "Existing native branch", "agent_type": "primary", "model_name": "primary", + "created_at": 1, "last_active_at": 2, "turn_count": 1, "message_count": 1, "tool_call_count": 0, "status": "completed", + "workspace_id": workspace.id, "workspace_path": workspace.root_path, + "project_workspace_id": workspace.id, "project_workspace_path": workspace.root_path, + "custom_metadata": {"externalBranchOf": {"sessionId": "source-view", "ecosystemId": ecosystem, "sourceSessionId": "source", "contentVersion": "pinned"}} + })).unwrap(); + let mut turn = DialogTurnData::new_with_kind( + DialogTurnKind::UserDialog, + turn_id.clone(), + 0, + id.clone(), + Some("primary".into()), + UserMessageData { + id: format!("{id}-user"), + content: "Preserved imported question".into(), + timestamp: 1, + metadata: None, + }, + ); + turn.status = TurnStatus::Completed; + store + .write_bundle_atomic(&OfflineSessionBundle { + metadata, + turns: vec![turn], + }) + .await + .unwrap(); + let metadata_path = root.join(&id).join("metadata.json"); + let before = std::fs::read(&metadata_path).unwrap(); + let restored = persistence + .load_session(&workspace.root_path, &id) + .await + .unwrap(); + + let admitted = ConversationCoordinator::resolve_session_agent_for_turn( + &restored, + &restored.agent_type, + &None, + ) + .await + .expect("legacy native branches must pass the same mode admission as a message send"); + assert_eq!(admitted.runtime_agent_key, "Standard"); + assert_eq!(restored.config.model_id.as_deref(), Some("primary")); + assert_eq!(std::fs::read(&metadata_path).unwrap(), before); + + let branch = persistence + .branch_session( + &workspace.root_path, + &SessionBranchRequest { + source_session_id: id.clone(), + source_turn_id: turn_id, + boundary: SessionBranchBoundary::ThroughTurn, + }, + ) + .await + .unwrap(); + let continued = persistence + .load_session(&workspace.root_path, &branch.session_id) + .await + .unwrap(); + let admitted = ConversationCoordinator::resolve_session_agent_for_turn( + &continued, + &continued.agent_type, + &None, + ) + .await + .expect("forking an existing legacy branch must retain a valid runtime mode"); + assert_eq!(admitted.runtime_agent_key, "Standard"); + let copied = store + .load_bundle(&branch.session_id) + .await + .unwrap() + .unwrap(); + assert_eq!( + copied.turns[0].user_message.content, + "Preserved imported question" + ); + assert_eq!(copied.turns[0].agent_type.as_deref(), Some("primary")); + } + } + #[tokio::test] async fn subagent_conversation_resolves_legacy_children_on_local_and_remote_workspaces() { for agent_type in ["GeneralPurpose", "Explore", "Standard"] { 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 1c8b9f7ec6..3f76252227 100644 --- a/src/crates/assembly/core/src/external_session_import/deferred.rs +++ b/src/crates/assembly/core/src/external_session_import/deferred.rs @@ -115,7 +115,7 @@ async fn read_index(root: &Path) -> OpenBitFunResult> { return Ok(index.clone()); } } - let index: Index = JsonFileStore + let mut index: Index = JsonFileStore .read_optional(&path) .await .map_err(index_error)? @@ -125,6 +125,10 @@ async fn read_index(root: &Path) -> OpenBitFunResult> { "This external history index requires a newer OpenBitFun host", )); } + // Read compatibility must not rewrite the user's index or frozen history. + for entry in index.entries.values_mut() { + entry.metadata.normalize_legacy_imported_agent_type(); + } let index = Arc::new(index); cache.insert(path, (stamp, index.clone())); Ok(index) @@ -443,7 +447,7 @@ pub(crate) async fn register( .title .clone() .unwrap_or_else(|| "Imported session".into()), - "primary".into(), + "Standard".into(), "primary".into(), ); metadata.workspace_id = Some(workspace.id.clone()); 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 9ec54e0dae..d7b2bee928 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 @@ -135,8 +135,18 @@ fn unedited_prefix( if before.metadata.status == SessionStatus::Active || before.turns.len() > after.turns.len() { return Ok(false); } + let mut compatible = before.metadata.clone(); + compatible.normalize_legacy_imported_agent_type(); for (old, new) in before.turns.iter().zip(&after.turns) { let mut old = old.clone(); + // The retired imported placeholder and its corrected native mode are + // equivalent for comparison only. Keep the persisted history intact, + // and still detect any other mode or content edit as a local change. + if old.agent_type.as_deref() == Some(before.metadata.agent_type.as_str()) + && new.agent_type.as_deref() == Some(compatible.agent_type.as_str()) + { + old.agent_type = new.agent_type.clone(); + } // A downgraded host may retain a view's versioned turn identity. if old.turn_id == format!( @@ -172,6 +182,51 @@ fn unedited_prefix( Ok(true) } +#[cfg(test)] +mod tests { + use super::super::tests::fixture; + use super::*; + use crate::external_session_import::auto_discovery::tests::{candidate, document}; + + #[tokio::test] + async fn legacy_mode_comparison_preserves_history_and_detects_real_edits() { + let (_temp, workspace, _root, _manager) = fixture(); + for ecosystem in ["codex", "zcode"] { + let source = document(candidate(ecosystem, "source-1", &workspace.root_path)); + let convert = |mode| { + convert_document( + "legacy", + &workspace, + &workspace, + mode, + "Old import", + &source, + true, + true, + ) + .unwrap() + }; + let before = convert("primary"); + let after = convert("Standard"); + let original = serde_json::to_value(&before).unwrap(); + assert!(unedited_prefix(&before, &after).unwrap()); + assert_eq!(serde_json::to_value(&before).unwrap(), original); + + let mut edited = before.clone(); + edited.turns[0].agent_type = Some("Creative".into()); + assert!(!unedited_prefix(&edited, &after).unwrap()); + + let mut edited = before.clone(); + edited.turns[0].user_message.content = "Local prompt edit".into(); + assert!(!unedited_prefix(&edited, &after).unwrap()); + + let mut unknown = before; + unknown.metadata.custom_metadata = None; + assert!(!unedited_prefix(&unknown, &after).unwrap()); + } + } +} + /// Called only when opening one source, never while discovering/listing it. pub(super) async fn validate_native( root: &Path, 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 246fea8fa6..e21ed9eacb 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 @@ -99,6 +99,7 @@ fn validate_snapshot( } fn overlay_view_metadata(metadata: &mut SessionMetadata, entry: &Reference) { + metadata.normalize_legacy_imported_agent_type(); // Names and archive state belong to the local view, not to a frozen version. metadata.session_name = entry.metadata.session_name.clone(); metadata.tags = entry.metadata.tags.clone(); @@ -689,6 +690,89 @@ mod tests { assert!(!root.join(&entry.metadata.session_id).exists()); } + #[tokio::test] + async fn legacy_primary_source_snapshots_fork_with_a_valid_mode_without_rewriting_history() { + let (_temp, workspace, root, manager) = fixture(); + let entry = create_source(&root, &workspace).await; + assert_eq!(entry.metadata.agent_type, "Standard"); + assert_eq!(entry.metadata.model_name, "primary"); + let id = &entry.metadata.session_id; + publish( + &root, + &workspace, + &entry, + document(candidate("codex", "source-1", &workspace.root_path)), + ) + .await; + let current = reference(&root, id).await.unwrap().unwrap(); + let version = content_version(¤t.metadata).unwrap(); + let path = snapshot_path(&root, id, version).unwrap(); + let mut frozen: Snapshot = JsonFileStore.read_optional(&path).await.unwrap().unwrap(); + // A previous host wrote the model selector into both metadata and turns. + frozen.bundle.metadata.agent_type = "primary".into(); + frozen.bundle.turns[0].agent_type = Some("primary".into()); + JsonFileStore + .write_atomic_strict(&path, &frozen) + .await + .unwrap(); + update_index(&root, |index| { + index.entries.get_mut(id).unwrap().metadata.agent_type = "primary".into(); + ((), true) + }) + .await + .unwrap(); + let original_history = std::fs::read(&path).unwrap(); + let original_index = std::fs::read(root.join(INDEX_FILE)).unwrap(); + + assert_eq!( + metadata(&root, id).await.unwrap().unwrap().agent_type, + "Standard" + ); + assert_eq!( + merge_metadata(&root, vec![]).await.unwrap()[0].agent_type, + "Standard" + ); + let viewed = snapshot(&root, id, false).await.unwrap().unwrap(); + assert_eq!(viewed.metadata.agent_type, "Standard"); + assert_eq!(viewed.metadata.model_name, "primary"); + assert_eq!(viewed.turns[0].agent_type.as_deref(), Some("primary")); + let branch_source = manager + .external_history_branch_source( + &workspace.root_path, + id, + Some(&viewed.turns[0].turn_id), + ) + .await + .unwrap() + .unwrap(); + let branch = manager + .branch_transient_session( + &workspace.root_path, + &SessionBranchRequest { + source_session_id: id.clone(), + source_turn_id: viewed.turns[0].turn_id.clone(), + boundary: SessionBranchBoundary::ThroughTurn, + }, + branch_source, + ) + .await + .unwrap(); + assert_eq!(branch.agent_type, "Standard"); + let native = OfflineSessionImportStore::new(&root) + .load_bundle(&branch.session_id) + .await + .unwrap() + .unwrap(); + assert_eq!(native.metadata.agent_type, "Standard"); + assert_eq!(native.turns[0].agent_type.as_deref(), Some("primary")); + assert_eq!(native.turns[0].user_message.content, "Saved question"); + assert_eq!(std::fs::read(&path).unwrap(), original_history); + assert_eq!( + std::fs::read(root.join(INDEX_FILE)).unwrap(), + original_index + ); + } + #[tokio::test] async fn append_rewrite_and_source_loss_preserve_viewed_versions_without_native_copies() { let (_temp, workspace, root, manager) = fixture(); diff --git a/src/crates/assembly/core/src/external_session_import/deferred/tracking/history_cache/tests.rs b/src/crates/assembly/core/src/external_session_import/deferred/tracking/history_cache/tests.rs index e3c5619e66..e83227f9f6 100644 --- a/src/crates/assembly/core/src/external_session_import/deferred/tracking/history_cache/tests.rs +++ b/src/crates/assembly/core/src/external_session_import/deferred/tracking/history_cache/tests.rs @@ -148,19 +148,32 @@ async fn history_cache_keeps_legacy_snapshot_and_window_shapes_readable() { .await .unwrap(); update_index(&root, |index| { - index - .entries - .get_mut(id) - .unwrap() - .metadata - .custom_metadata - .as_mut() - .unwrap()[VERSION] = json!(version); + let metadata = &mut index.entries.get_mut(id).unwrap().metadata; + metadata.agent_type = "primary".into(); + metadata.custom_metadata.as_mut().unwrap()[VERSION] = json!(version); ((), true) }) .await .unwrap(); let bytes = std::fs::read(&path).unwrap(); + let index_bytes = std::fs::read(root.join(INDEX_FILE)).unwrap(); + assert_eq!( + metadata(&root, id).await.unwrap().unwrap().agent_type, + "Standard" + ); + assert_eq!( + merge_metadata(&root, vec![]).await.unwrap()[0].agent_type, + "Standard" + ); + let probe = probes::Probe::new(&root); + let cold = snapshot_shared(&root, id, false).await.unwrap().unwrap(); + let warm = snapshot_shared(&root, id, false).await.unwrap().unwrap(); + for snapshot in [&cold, &warm] { + assert_eq!(snapshot.metadata.agent_type, "Standard"); + assert_eq!(snapshot.metadata.model_name, "primary"); + } + assert!(Arc::ptr_eq(&cold.history, &warm.history)); + assert_eq!(probe.take().snapshot_reads, 1); let request = SessionTurnWindowRequest { workspace_path: workspace.root_path.clone(), session_id: id.clone(), @@ -217,6 +230,7 @@ async fn history_cache_keeps_legacy_snapshot_and_window_shapes_readable() { assert!( matches!(missing, SessionTurnWindowResponse::NotFound { catalog } if catalog.total_turn_count == 1) ); + assert_eq!(std::fs::read(root.join(INDEX_FILE)).unwrap(), index_bytes); } #[tokio::test] 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 de57a75847..129b68478c 100644 --- a/src/crates/services/services-core/src/session/metadata_store.rs +++ b/src/crates/services/services-core/src/session/metadata_store.rs @@ -537,7 +537,8 @@ impl SessionMetadataStore { let _guard = lock.lock().await; let _file_guard = self.lock_index_file().await?; let metadata_file_created = !metadata_path.exists(); - let mut metadata = compatible_metadata(metadata.clone()); + // Read projections must not mutate an offline bundle or repair journal. + let mut metadata = compatible_stored_metadata(metadata.clone()); self.assign_workspace_session_number_locked(&mut metadata) .await?; let file = StoredSessionMetadataFile::new(metadata.clone()); @@ -656,6 +657,11 @@ impl SessionMetadataStore { } fn compatible_metadata(mut metadata: SessionMetadata) -> SessionMetadata { + metadata.normalize_legacy_imported_agent_type(); + compatible_stored_metadata(metadata) +} + +fn compatible_stored_metadata(mut metadata: SessionMetadata) -> SessionMetadata { metadata.normalize_legacy_model_selector(); super::naming::migrate_legacy_default_title(&mut metadata); metadata @@ -663,6 +669,7 @@ fn compatible_metadata(mut metadata: SessionMetadata) -> SessionMetadata { fn compatible_metadata_list(mut metadata: Vec) -> Vec { for entry in &mut metadata { + entry.normalize_legacy_imported_agent_type(); entry.normalize_legacy_model_selector(); super::naming::migrate_legacy_default_title(entry); } @@ -787,6 +794,101 @@ mod tests { } } + #[tokio::test] + async fn legacy_imported_primary_mode_reads_preserve_files_and_unrelated_agents() { + for (id, agent, custom, expected) in [ + ( + "codex", + "primary", + serde_json::json!({"importedFrom": "codex", "sourceSessionId": "source", "externalHistoryMode": "follow"}), + "Standard", + ), + ( + "zcode", + "primary", + serde_json::json!({"importedFrom": "zcode", "sourceSessionId": "source", "continuationMode": "openbitfun_local"}), + "Standard", + ), + ( + "branch", + "primary", + serde_json::json!({"externalBranchOf": {"ecosystemId": "codex", "sourceSessionId": "source", "contentVersion": "retained"}}), + "Standard", + ), + ( + "zcode-branch", + "primary", + serde_json::json!({"externalBranchOf": {"ecosystemId": "zcode", "sourceSessionId": "source"}}), + "Standard", + ), + ("native", "primary", serde_json::json!({}), "primary"), + ( + "other-source", + "primary", + serde_json::json!({"importedFrom": "custom", "sourceSessionId": "source"}), + "primary", + ), + ( + "missing-identity", + "primary", + serde_json::json!({"importedFrom": "codex"}), + "primary", + ), + ( + "external-provider", + "primary", + serde_json::json!({"provider": "acp", "importedFrom": "codex", "sourceSessionId": "source"}), + "primary", + ), + ( + "selected-mode", + "Creative", + serde_json::json!({"externalBranchOf": {"ecosystemId": "codex", "sourceSessionId": "source"}}), + "Creative", + ), + ( + "custom-mode", + "custom-primary", + serde_json::json!({"importedFrom": "codex", "sourceSessionId": "source"}), + "custom-primary", + ), + ] { + let root = tempfile::tempdir().unwrap(); + let store = SessionMetadataStore::new(root.path()); + // Old snake-case payload with absent optional fields must remain readable. + let record: SessionMetadata = serde_json::from_value(serde_json::json!({ + "session_id": id, "session_name": "Kept title", "agent_type": agent, + "model_name": "primary", "created_at": 1, "last_active_at": 2, + "turn_count": 0, "message_count": 0, "tool_call_count": 0, + "status": "completed", "workspace_id": "main", "custom_metadata": custom, + })) + .unwrap(); + store.save_metadata(&record).await.unwrap(); + let original = fs::read(store.metadata_path(id)).await.unwrap(); + let stored: StoredSessionMetadataFile = serde_json::from_slice(&original).unwrap(); + assert_eq!(stored.metadata.agent_type, agent); + let index = fs::read(store.index_path()).await.unwrap(); + for records in [ + vec![store.load_metadata(id).await.unwrap().unwrap()], + store.list_metadata().await.unwrap(), + store.list_metadata_page(None, 5).await.unwrap().sessions, + store.metadata_by_ids(&[id.to_string()]).await.unwrap(), + store.list_metadata_including_internal().await.unwrap(), + ] { + let returned = records.iter().find(|entry| entry.session_id == id).unwrap(); + assert_eq!(returned.agent_type, expected, "{id}"); + assert_eq!(returned.model_name, "primary"); + assert_eq!(returned.custom_metadata, record.custom_metadata); + let round_trip: SessionMetadata = + serde_json::from_slice(&serde_json::to_vec(returned).unwrap()).unwrap(); + assert_eq!(round_trip.agent_type, expected); + assert_eq!(round_trip.custom_metadata, record.custom_metadata); + } + assert_eq!(fs::read(store.metadata_path(id)).await.unwrap(), original); + assert_eq!(fs::read(store.index_path()).await.unwrap(), index); + } + } + #[tokio::test] async fn catalog_watch_tracks_committed_create_rename_and_delete() { let root = tempfile::tempdir().unwrap(); diff --git a/src/crates/services/services-core/src/session/types.rs b/src/crates/services/services-core/src/session/types.rs index c23495a7b1..57032cbe02 100644 --- a/src/crates/services/services-core/src/session/types.rs +++ b/src/crates/services/services-core/src/session/types.rs @@ -1142,6 +1142,41 @@ pub struct SessionLastTurn { } impl SessionMetadata { + /// Early external-history indexes used the model selector as the native + /// session mode. Project only that known imported shape; `primary` is not + /// a general alias for custom agents or external runtime bindings. + pub fn normalize_legacy_imported_agent_type(&mut self) { + if self.agent_type != "primary" || self.session_kind != SessionKind::Standard { + return; + } + let Some(custom) = self.custom_metadata.as_ref() else { + return; + }; + if custom + .get(openbitfun_core_types::SESSION_PROVIDER_METADATA_KEY) + .and_then(serde_json::Value::as_str) + .is_some_and(|provider| !provider.is_empty()) + { + return; + } + let has_import_identity = |value: &serde_json::Value, ecosystem_key: &str| { + matches!( + value.get(ecosystem_key).and_then(serde_json::Value::as_str), + Some("codex" | "zcode") + ) && value + .get("sourceSessionId") + .and_then(serde_json::Value::as_str) + .is_some_and(|id| !id.is_empty()) + }; + if has_import_identity(custom, "importedFrom") + || custom + .get("externalBranchOf") + .is_some_and(|source| has_import_identity(source, "ecosystemId")) + { + self.agent_type = "Standard".to_string(); + } + } + /// Translate the retired native selector without changing external agents' IDs. pub fn compatible_model_selector<'a>(&self, selector: &'a str) -> &'a str { let external_provider = self diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts index addfa8bbfa..4e6420a0c8 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.test.ts @@ -294,6 +294,55 @@ describe('MessageModule session writer conflict', () => { expect(context.flowChatStore.addDialogTurn).not.toHaveBeenCalled(); }); + it.each([ + ['primary', 'primary', 'Standard'], + ['primary', undefined, 'Standard'], + ['primary', 'Creative', 'Creative'], + ['Creative', 'Creative', 'Creative'], + ])('submits the native fork mode instead of an inherited source placeholder (%s, %s)', async (sourceMode, requestedMode, expectedMode) => { + const { context, session } = conflictContext('legacy-external-source'); + Object.assign(session, { + mode: sourceMode, + historyState: 'ready', externalHistory: { mode: 'follow' }, + dialogTurns: [{ id: 'viewed-turn', userMessage: { metadata: { externalContentVersion: 'viewed-version' } } }], + }); + const branch = { ...session, sessionId: 'native-branch', mode: 'Standard', externalHistory: undefined, dialogTurns: [...session.dialogTurns] }; + const sessions = new Map([[session.sessionId, session], [branch.sessionId, branch]]); + context.flowChatStore.getState = () => ({ sessions, activeSessionId: branch.sessionId }); + context.flowChatStore.addDialogTurn.mockImplementation((id: string, turn: any) => sessions.get(id)!.dialogTurns.push(turn)); + mockForkChatSession.mockResolvedValue(branch.sessionId); + mockEnsureBackendSession.mockResolvedValue(undefined); + mockGetConfigs.mockResolvedValue({}); + mockStartDialogTurn.mockImplementation(async ({ agentType }: { agentType: string }) => { + if (agentType === 'primary') throw new Error('Unknown session mode: primary'); + }); + + await sendMessage(context, 'Continue here', session.sessionId, undefined, requestedMode); + + expect(mockStartDialogTurn).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + sessionId: branch.sessionId, agentType: expectedMode, userInput: 'Continue here', + })); + expect(session.mode).toBe(sourceMode); + expect(session.dialogTurns).toHaveLength(1); + }); + + it.each(['codex', 'zcode', undefined])('retries an old queued mode using the repaired native branch (%s)', async ecosystemId => { + const { context, session } = conflictContext('existing-branch'); + if (ecosystemId) Object.assign(session, { + externalBranchOf: { sessionId: 'source', ecosystemId, sourceSessionId: 'original', contentVersion: 'pinned' }, + }); + mockEnsureBackendSession.mockResolvedValue(undefined); + mockGetConfigs.mockResolvedValue({}); + + // Failed queue entries retain the mode from before the metadata migration. + await sendMessage(context, 'Retry retained draft', session.sessionId, undefined, 'primary', undefined, { bypassPendingQueue: true }); + + expect(mockStartDialogTurn).toHaveBeenCalledExactlyOnceWith(expect.objectContaining({ + sessionId: session.sessionId, agentType: ecosystemId ? 'Standard' : 'primary', userInput: 'Retry retained draft', + })); + expect(mockForkChatSession).not.toHaveBeenCalled(); + }); + it('does not fork an archived source before rejecting submission', async () => { const { context, session } = conflictContext('archived-source'); Object.assign(session, { diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts index c13eec4cc4..97078c0e48 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/MessageModule.ts @@ -277,11 +277,24 @@ async function sendMessageInternal( const branchId = await fork; surfaceScopeAtSend.assertCurrent('continue external history in native branch'); options?.onSessionForked?.(sessionId, branchId); - return await sendMessageInternal(context, message, branchId, displayMessage, agentType, switchToMode, options); + // Do not replay the retired source placeholder over the host's corrected + // branch mode. Keep valid composer selections, even when stored locally. + const continuationAgentType = agentType?.trim() === 'primary' + && (session.mode || session.config.agentType)?.trim() === 'primary' + ? undefined : agentType; + return await sendMessageInternal(context, message, branchId, displayMessage, continuationAgentType, switchToMode, options); } finally { if (externalForks.get(key) === fork) externalForks.delete(key); } } + // A failed queue entry can retain the old source placeholder after the host + // has repaired its native branch. Retry with the branch's current mode. + const importedSource = session.externalBranchOf?.ecosystemId ?? session.importedSource; + if ((importedSource === 'codex' || importedSource === 'zcode') + && (!session.sessionKind || session.sessionKind === 'normal') + && agentType?.trim() === 'primary' && session.mode && session.mode !== 'primary') { + agentType = session.mode; + } if (session.sessionKind === 'subagent' && getSessionConversationCapability(sessionId, session).reason === 'metadata_pending') { const parent = session.parentSessionId diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.test.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.test.ts index a964589ffc..f5112b2242 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.test.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.test.ts @@ -590,6 +590,35 @@ describe('forkChatSession', () => { vi.clearAllMocks(); }); + it('uses the host fork mode when a legacy external source still carries primary', async () => { + const source = createSession({ + sessionId: 'legacy-source', mode: 'primary', config: { agentType: 'primary', modelName: 'primary' }, + externalHistory: { mode: 'follow' }, + }); + const { context, flowChatStore } = createContext(source); + sessionApiMocks.forkSession.mockResolvedValueOnce({ sessionId: 'branch', sessionName: 'Branch', agentType: 'Standard' }); + + await forkChatSession(context, source.sessionId, 'viewed-turn'); + + expect(flowChatStore.getState().sessions.get('branch')).toMatchObject({ + mode: 'Standard', config: { agentType: 'Standard' }, + }); + expect(flowChatStore.createSession.mock.calls[0][1]).toMatchObject({ agentType: 'Standard', modelName: 'primary' }); + expect(source.mode).toBe('primary'); + }); + + it('retains the selected source mode when an older host omits the fork mode', async () => { + const source = createSession({ sessionId: 'source', mode: 'Minimal', config: { agentType: 'Minimal' } }); + const { context, flowChatStore } = createContext(source); + sessionApiMocks.forkSession.mockResolvedValueOnce({ sessionId: 'branch', sessionName: 'Branch' }); + + await forkChatSession(context, source.sessionId, 'viewed-turn'); + + expect(flowChatStore.getState().sessions.get('branch')).toMatchObject({ + mode: 'Minimal', config: { agentType: 'Minimal' }, + }); + }); + it('applies a source composer permission selection only to the created native branch', async () => { const source = createSession({ sessionId: 'source', externalHistory: { mode: 'follow' }, externalContinuationPermissionMode: 'auto_approve' }); const { context } = createContext(source); diff --git a/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts b/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts index 0760bef641..6089de5b64 100644 --- a/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts +++ b/src/web-ui/src/flow_chat/services/flow-chat-manager/SessionModule.ts @@ -840,10 +840,12 @@ export async function forkChatSession( const currentState = context.flowChatStore.getState(); if (!currentState.sessions.has(response.sessionId)) { + const branchMode = response.agentType?.trim() || sourceSession.mode || sourceSession.config.agentType || 'Standard'; context.flowChatStore.createSession( response.sessionId, { ...sourceSession.config, + agentType: branchMode, workspacePath: executionWorkspacePath, projectWorkspacePath, workspaceId: sourceSession.workspaceId, @@ -853,7 +855,7 @@ export async function forkChatSession( undefined, response.sessionName, sourceSession.maxContextTokens, - sourceSession.mode, + branchMode, executionWorkspacePath, sourceSession.remoteConnectionId, sourceSession.remoteSshHost,