From b92b6a3409bf9da4ace2656fcf49fdbdb9df2e9e Mon Sep 17 00:00:00 2001 From: limityan Date: Wed, 29 Jul 2026 17:27:18 +0800 Subject: [PATCH] fix(session): surface cross-process writer conflicts --- .../agent-runtime-deployment-design.md | 12 +- docs/architecture/cli-product-line-design.md | 6 +- src/apps/cli/README.md | 8 +- src/apps/cli/src/agent/runtime_client.rs | 20 +- src/apps/cli/src/diagnostics.rs | 74 +++++ src/apps/cli/src/modes/exec/lifecycle.rs | 37 ++- src/apps/cli/src/modes/exec/tests.rs | 29 +- .../cli/src/peer_host/commands/session.rs | 77 ++++- .../src/runtime/session_application.rs | 35 ++- .../src/flow_chat/components/ChatInput.tsx | 151 ++++++++-- .../components/ChatInputDraftRecovery.test.ts | 55 ++++ .../components/chatInputDraftRecovery.ts | 38 +++ .../src/flow_chat/hooks/useMessageSender.ts | 39 ++- .../src/flow_chat/services/FlowChatManager.ts | 2 + .../flow-chat-manager/MessageModule.test.ts | 283 +++++++++++++++++- .../flow-chat-manager/MessageModule.ts | 126 +++++++- .../flow-chat-manager/SessionModule.test.ts | 19 ++ .../flow-chat-manager/SessionModule.ts | 4 + .../api/adapters/peer-device-adapter.test.ts | 18 ++ .../api/errors/TauriCommandError.test.ts | 35 +++ .../api/errors/TauriCommandError.ts | 35 ++- src/web-ui/src/locales/en-US/flow-chat.json | 5 +- src/web-ui/src/locales/zh-CN/flow-chat.json | 5 +- src/web-ui/src/locales/zh-TW/flow-chat.json | 5 +- .../components/NotificationItem.test.tsx | 59 ++++ .../components/NotificationItem.tsx | 7 +- 26 files changed, 1107 insertions(+), 77 deletions(-) create mode 100644 src/web-ui/src/flow_chat/components/ChatInputDraftRecovery.test.ts create mode 100644 src/web-ui/src/flow_chat/components/chatInputDraftRecovery.ts create mode 100644 src/web-ui/src/infrastructure/api/errors/TauriCommandError.test.ts create mode 100644 src/web-ui/src/shared/notification-system/components/NotificationItem.test.tsx diff --git a/docs/architecture/agent-runtime-deployment-design.md b/docs/architecture/agent-runtime-deployment-design.md index 2c2c47ed4e..2011848128 100644 --- a/docs/architecture/agent-runtime-deployment-design.md +++ b/docs/architecture/agent-runtime-deployment-design.md @@ -152,7 +152,17 @@ flowchart LR View -.->|"只读"| B ``` -BitFun Runtime Session 只有 `SessionManager` 决定何时开始和结束写入；底层持久化方法复用同一文件锁，不再实现第二套判断。Agent SDK、BitFun ACP adapter 和 Shared TUI 保留结构化的 `session_in_use` 分类；SDK Host 将其映射为可重试并建议 retry 的结构化 `action_required`。GUI、Embedded TUI 和 Headless CLI 当前只显示明确的冲突消息，尚未承诺结构化错误字段，自动化调用不能依赖该文案。Desktop 作为 ACP client 管理的外部 agent Session 不经过该 Runtime owner，不在本节的 Session 单写范围内。 +BitFun Runtime Session 只有 `SessionManager` 决定何时开始和结束写入；底层持久化方法复用同一文件锁，不再实现第二套判断。各产品入口只投影同一个 `session_in_use` 事实，不重新判断锁状态： + +| 入口 | 冲突呈现 | 恢复方式 | +|---|---|---| +| Agent SDK / BitFun ACP | 结构化 `session_in_use`；SDK Host 映射为可重试的 `action_required` | 调用方在原实例关闭 Session 后重试 | +| Embedded / Shared TUI | 明确提示 Session 已在另一实例打开；切换失败时保留当前 Session | 用户关闭另一实例后再次选择；不自动等待或切换 | +| Desktop / Peer GUI | 历史视图保持只读可见；首次写入显示持久提示和显式“重试”操作 | 用户关闭另一实例后点击重试；不自动提交消息 | +| Headless `json` | 失败结果带 `error_code=session_in_use`，详细说明进入结果和 stderr | 调用方依据稳定码决定是否重试 | +| Headless `stream-json` | 复用已有 `SystemError`，`error=session_in_use`、`recoverable=true` | 调用方结束本次非零退出后重新执行 | + +Desktop 作为 ACP client 管理的外部 agent Session 不经过该 Runtime owner，不在本节的 Session 单写范围内。`recoverable` 只表示关闭现有 writer 后可以重新调用，不表示自动等待、自动抢占或恢复当前调用。 | 场景 | 行为 | |---|---| diff --git a/docs/architecture/cli-product-line-design.md b/docs/architecture/cli-product-line-design.md index d3131a42c1..bdd4fbd368 100644 --- a/docs/architecture/cli-product-line-design.md +++ b/docs/architecture/cli-product-line-design.md @@ -228,14 +228,14 @@ CLI-P1 应保证： | 模式 | 当前约束 | |---|---| | `text` | 最终助手文本写 stdout；进度、思考、工具状态、日志和诊断写 stderr。显式 `--output-patch -` 是用户选择的额外 stdout 内容。 | -| `json` | stdout 只写一个结果对象，包含 `type=result`、`subtype`、`is_error`、`result`，以及已建立时的 `session_id`/`turn_id`、本 turn 累计 `usage` 和可用的 `patch`。 | -| `stream-json` | 每行直接序列化一个现有 Agent 事件对象；不增加 `schema_version`、`sequence` 或第二套 CLI 事件分类。 | +| `json` | stdout 只写一个结果对象，包含 `type=result`、`subtype`、`is_error`、`result`，以及已建立时的 `session_id`/`turn_id`、本 turn 累计 `usage` 和可用的 `patch`。准备 Session 时若命中跨进程单写冲突，额外返回稳定的 `error_code=session_in_use`；其他错误不猜测分类。 | +| `stream-json` | 每行直接序列化一个现有 Agent 事件对象；不增加 `schema_version`、`sequence` 或第二套 CLI 事件分类。准备 Session 时若命中单写冲突，复用 `SystemError`，令 `error=session_in_use`、`recoverable=true`。 | | 最终状态 | 精确结算和 Patch 交付完成后只发布一次。优先级是：结算失败、Patch 失败、Turn 结果；前两类统一替换为 `SystemError`。一次执行最多发布一个最终事件和一条 `BITFUN_EXIT` 分类。 | | 事件范围 | 只输出本次 session/turn 的事件，以及与其明确关联的 subagent link/tool 事件；同 session 的其他并发 turn 不得混入。 | | Patch | `json` 可把 `--output-patch -` 放入最终对象；`stream-json` 要求显式文件路径。Patch 是写出显式 Patch 文件前捕获的仓库 `HEAD` 相对工作区快照，包含 staged、unstaged、untracked 及命令启动前已有改动，不包含输出 artifact 本身，也不表达改动归因。 | | 权限 | 非交互默认拒绝并返回权限失败；`--auto` 只改变当前提交策略，不修改持久化配置。 | | 人工输入 | 非交互 `exec` 不暴露 `AskUserQuestion`；调用方必须在初始输入中提供完整上下文。该事实沿 Task、SessionMessage 及其自动回复链传播，避免子 Agent 或后续 turn 等待不存在的 stdin 处理器。 | -| 终止 | 最终事件的 `success=false` 不能映射为成功。`Ctrl+C` 只请求取消；若取消与完成/失败竞争，以实际观察结果为准。到期限仍无最终事件时发布 `SystemError` 并非零退出；只有实际取消使用 `BITFUN_EXIT: cancelled:`。当前不公开 Agent Turn 总时限参数。 | +| 终止 | 最终事件的 `success=false` 不能映射为成功。`Ctrl+C` 只请求取消；若取消与完成/失败竞争，以实际观察结果为准。到期限仍无最终事件时发布 `SystemError` 并非零退出；只有实际取消使用 `BITFUN_EXIT: cancelled:`。`session_in_use` 同样非零退出，`recoverable` 仅表示关闭另一 writer 后可重新执行，不触发自动重试。当前不公开 Agent Turn 总时限参数。 | CLI 不提供 `--output-schema v1`。Codex/Claude 同类参数表达的是调用方提供的 JSON Schema，用于约束最终模型 响应，不是协议版本选择；如未来支持，应复用该语义并独立设计，不能借此重定义事件对象。 diff --git a/src/apps/cli/README.md b/src/apps/cli/README.md index 9d36ac15a1..292ab1150b 100644 --- a/src/apps/cli/README.md +++ b/src/apps/cli/README.md @@ -93,8 +93,8 @@ This command is TUI-only and does not change the non-interactive `exec` contract | Format | stdout contract | |---|---| | `text` | Assistant text. Progress, tool status, logs, and diagnostics use stderr. | -| `json` | One final result object with status and result, plus session/turn identity once established, turn-accumulated usage, and available Patch facts. | -| `stream-json` | JSONL containing existing Agent event values; no separate CLI event schema. | +| `json` | One final result object with status and result, plus session/turn identity once established, turn-accumulated usage, and available Patch facts. A Session writer conflict adds `error_code: "session_in_use"`. | +| `stream-json` | JSONL containing existing Agent event values; no separate CLI event schema. A Session writer conflict reuses `SystemError` with `error: "session_in_use"` and `recoverable: true`. | Select a format with `--output-format text|json|stream-json`. When `--output-patch -` is used with `json`, the Patch is included in the final object. For `stream-json`, write the Patch to an explicit @@ -108,6 +108,10 @@ returning. Cancellation, an unsuccessful completion event, and a requested Patch that cannot be generated or written are error outcomes. An explicit Patch file is created even when the diff is empty. +If another BitFun process is writing the requested Session, `exec` exits non-zero without waiting +or taking over. Close that Session in the other instance and run the command again. `recoverable` +describes that later retry; it does not mean the current command retries automatically. + `doctor` and `health` validate product assembly and required capability registrations. They are not live probes for Network, Git, or MCP integrations that are currently represented by compatibility registrations. diff --git a/src/apps/cli/src/agent/runtime_client.rs b/src/apps/cli/src/agent/runtime_client.rs index eb2f73cd44..c9e350e385 100644 --- a/src/apps/cli/src/agent/runtime_client.rs +++ b/src/apps/cli/src/agent/runtime_client.rs @@ -31,18 +31,20 @@ use bitfun_runtime_ports::{ }; use crate::actions::SHARED_TUI_EMBEDDED_HANDOFF; +use crate::diagnostics::with_session_conflict_help; use crate::runtime::approval::{approval_metadata, CliApprovalPolicy}; use crate::runtime::CliRuntimeContext; fn shared_restore_error(error: RuntimeIpcClientError) -> anyhow::Error { - if matches!(&error, RuntimeIpcClientError::Remote(remote) if remote.code == RuntimeIpcErrorCode::FrameTooLarge) + let error = if matches!(&error, RuntimeIpcClientError::Remote(remote) if remote.code == RuntimeIpcErrorCode::FrameTooLarge) { anyhow::anyhow!( "Session history is too large for Shared TUI. {SHARED_TUI_EMBEDDED_HANDOFF}." ) } else { - error.into() - } + anyhow::Error::new(error) + }; + with_session_conflict_help(error) } fn validated_session_summary( @@ -413,7 +415,8 @@ impl CliAgentRuntimeClient { }) .await .map(|restored| restored.session) - .map_err(|error| anyhow::anyhow!(error.into_message()))?; + .map_err(anyhow::Error::new) + .map_err(with_session_conflict_help)?; let transcript = runtime .read_session_transcript(SessionTranscriptRequest { session_id: session_id.to_string(), @@ -650,7 +653,8 @@ impl CliAgentRuntimeClient { }, ) .await - .map_err(|error| anyhow::anyhow!(error.into_message()))?; + .map_err(anyhow::Error::new) + .map_err(with_session_conflict_help)?; tracing::info!("Recreated backend session with existing id: {}", session_id); Ok(()) @@ -677,7 +681,6 @@ impl CliAgentRuntimeClient { } Err(error) => { let session_not_found = Self::is_session_not_found_error(&error); - let message = error.into_message(); if session_not_found { tracing::warn!( "Session is unavailable, recreating backend session: {}", @@ -685,7 +688,7 @@ impl CliAgentRuntimeClient { ); self.recreate_session_with_id(session_id, agent_type).await } else { - Err(anyhow::anyhow!(message)) + Err(with_session_conflict_help(anyhow::Error::new(error))) } } } @@ -718,7 +721,8 @@ impl CliAgentRuntimeClient { }, ) .await - .map_err(|error| anyhow::anyhow!(error.into_message()))?; + .map_err(anyhow::Error::new) + .map_err(with_session_conflict_help)?; let id = session.session_id.clone(); *session_id_guard = Some(id.clone()); diff --git a/src/apps/cli/src/diagnostics.rs b/src/apps/cli/src/diagnostics.rs index ddd27cd356..a29d770c07 100644 --- a/src/apps/cli/src/diagnostics.rs +++ b/src/apps/cli/src/diagnostics.rs @@ -2,8 +2,14 @@ use std::path::Path; +use bitfun_agent_runtime::sdk::{PortErrorKind, RuntimeError}; +use bitfun_agent_runtime_ipc::{RuntimeIpcClientError, RuntimeIpcErrorCode}; + pub(crate) const EXIT_LINE_PREFIX: &str = "BITFUN_EXIT: "; pub(crate) const DETAIL_MAX_LEN: usize = 500; +pub(crate) const SESSION_IN_USE_ERROR_CODE: &str = "session_in_use"; +pub(crate) const SESSION_IN_USE_USER_MESSAGE: &str = + "This session is open in another BitFun instance. Close it there and retry."; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum ExitKind { @@ -64,6 +70,36 @@ pub(crate) fn format_exit_line(kind: ExitKind, detail: &str) -> String { ) } +pub(crate) fn cli_error_code(error: &anyhow::Error) -> Option<&'static str> { + let session_in_use = error.chain().any(|cause| { + matches!( + cause.downcast_ref::(), + Some(RuntimeError::Port(port_error)) + if port_error.kind == PortErrorKind::SessionInUse + ) || matches!( + cause.downcast_ref::(), + Some(RuntimeIpcClientError::Remote(remote)) + if remote.code == RuntimeIpcErrorCode::SessionInUse + ) + }); + session_in_use.then_some(SESSION_IN_USE_ERROR_CODE) +} + +pub(crate) fn user_facing_error_message(error: &anyhow::Error) -> String { + match cli_error_code(error) { + Some(SESSION_IN_USE_ERROR_CODE) => SESSION_IN_USE_USER_MESSAGE.to_string(), + _ => error.to_string(), + } +} + +pub(crate) fn with_session_conflict_help(error: anyhow::Error) -> anyhow::Error { + if cli_error_code(&error).is_some() { + error.context(SESSION_IN_USE_USER_MESSAGE) + } else { + error + } +} + pub(crate) fn emit_exit_diagnostic(kind: ExitKind, detail: &str, ctx: &ExitContext<'_>) { eprintln!("{}", format_exit_line(kind, detail)); tracing::error!( @@ -80,6 +116,8 @@ pub(crate) fn emit_exit_diagnostic(kind: ExitKind, detail: &str, ctx: &ExitConte #[cfg(test)] mod tests { use super::*; + use bitfun_agent_runtime::sdk::{PortError, PortErrorKind, RuntimeError}; + use bitfun_agent_runtime_ipc::{RuntimeIpcClientError, RuntimeIpcError, RuntimeIpcErrorCode}; #[test] fn format_exit_line_uses_stable_prefix_and_kind() { @@ -103,4 +141,40 @@ mod tests { assert!(sanitized.ends_with("...")); assert!(sanitized.chars().count() <= DETAIL_MAX_LEN + 3); } + + #[test] + fn embedded_session_conflict_keeps_a_stable_code_and_actionable_message() { + let error = anyhow::Error::new(RuntimeError::Port(PortError::new( + PortErrorKind::SessionInUse, + "Session is already open for writing: session-1", + ))); + + assert_eq!(cli_error_code(&error), Some(SESSION_IN_USE_ERROR_CODE)); + assert_eq!( + user_facing_error_message(&error), + SESSION_IN_USE_USER_MESSAGE + ); + } + + #[test] + fn shared_session_conflict_uses_the_same_cli_projection() { + let error = anyhow::Error::new(RuntimeIpcClientError::Remote(RuntimeIpcError { + code: RuntimeIpcErrorCode::SessionInUse, + message: "Session is already open for writing: session-1".to_string(), + })); + + assert_eq!(cli_error_code(&error), Some(SESSION_IN_USE_ERROR_CODE)); + assert_eq!( + user_facing_error_message(&error), + SESSION_IN_USE_USER_MESSAGE + ); + } + + #[test] + fn unrelated_errors_keep_their_original_message() { + let error = anyhow::anyhow!("provider unavailable"); + + assert_eq!(cli_error_code(&error), None); + assert_eq!(user_facing_error_message(&error), "provider unavailable"); + } } diff --git a/src/apps/cli/src/modes/exec/lifecycle.rs b/src/apps/cli/src/modes/exec/lifecycle.rs index 28c65b8149..afb4110251 100644 --- a/src/apps/cli/src/modes/exec/lifecycle.rs +++ b/src/apps/cli/src/modes/exec/lifecycle.rs @@ -21,7 +21,10 @@ use tokio::time::Instant; use crate::agent::runtime_client::CliAgentRuntimeClient; use crate::config::CliConfig; -use crate::diagnostics::{emit_exit_diagnostic, ExitContext, ExitKind}; +use crate::diagnostics::{ + cli_error_code, emit_exit_diagnostic, user_facing_error_message, ExitContext, ExitKind, + SESSION_IN_USE_ERROR_CODE, +}; use crate::runtime::CliRuntimeContext; pub(super) const TOOL_START_INPUT_PREVIEW_CHARS: usize = 4_000; @@ -118,6 +121,8 @@ pub(super) struct ExecJsonResult { is_error: bool, result: String, #[serde(skip_serializing_if = "Option::is_none")] + error_code: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] session_id: Option, #[serde(skip_serializing_if = "Option::is_none")] turn_id: Option, @@ -398,6 +403,7 @@ impl ExecJsonResult { subtype, is_error, result: result.into(), + error_code: None, session_id, turn_id, usage, @@ -411,6 +417,11 @@ impl ExecJsonResult { self } + pub(super) fn with_error_code(mut self, error_code: &'static str) -> Self { + self.error_code = Some(error_code); + self + } + fn with_verification( mut self, verification: Option, @@ -437,6 +448,17 @@ pub(super) fn serialize_stream_envelope( Ok(serde_json::to_string(envelope)?) } +pub(super) fn session_in_use_stream_envelope() -> bitfun_events::AgenticEventEnvelope { + bitfun_events::AgenticEventEnvelope::new( + AgenticEvent::SystemError { + session_id: None, + error: SESSION_IN_USE_ERROR_CODE.to_string(), + recoverable: true, + }, + bitfun_events::AgenticEventPriority::Critical, + ) +} + #[derive(Debug, Clone, Default)] pub(crate) struct ExecSessionOptions { pub resume: Option, @@ -604,14 +626,23 @@ impl ExecMode { let session_id = match self.prepare_session().await { Ok(session_id) => session_id, Err(error) => { + let error_code = cli_error_code(&error); + let detail = user_facing_error_message(&error); emit_exit_diagnostic( ExitKind::SessionCreateFailed, - &error.to_string(), + &detail, &self.exit_context(None, None), ); if self.output_format == ExecOutputFormat::Json { - let result = ExecJsonResult::preflight_error(error.to_string()); + let mut result = ExecJsonResult::preflight_error(detail); + if let Some(error_code) = error_code { + result = result.with_error_code(error_code); + } println!("{}", serde_json::to_string_pretty(&result)?); + } else if self.output_format == ExecOutputFormat::StreamJson + && error_code == Some(SESSION_IN_USE_ERROR_CODE) + { + self.emit_stream_envelope(&session_in_use_stream_envelope())?; } return Err(error); } diff --git a/src/apps/cli/src/modes/exec/tests.rs b/src/apps/cli/src/modes/exec/tests.rs index ef40b057da..0bd28b43ac 100644 --- a/src/apps/cli/src/modes/exec/tests.rs +++ b/src/apps/cli/src/modes/exec/tests.rs @@ -4,8 +4,9 @@ use super::lifecycle::{ completed_turn_failure, drain_interrupted_turn_events, effective_event_invocation, event_belongs_to_exec_turn, event_turn_id, is_exec_terminal, permission_action_required_message, resolve_cancelled_turn_observation, - serialize_stream_envelope, settlement_failure, should_reject_permission_request, - ExecApprovalMode, ExecJsonResult, ExecMode, ExecTokenUsage, TOOL_START_INPUT_PREVIEW_CHARS, + serialize_stream_envelope, session_in_use_stream_envelope, settlement_failure, + should_reject_permission_request, ExecApprovalMode, ExecJsonResult, ExecMode, ExecTokenUsage, + TOOL_START_INPUT_PREVIEW_CHARS, }; use super::patch::write_patch_to_path; use super::patch::{git_diff_base, untracked_files}; @@ -376,6 +377,17 @@ fn preflight_json_error_omits_unknown_runtime_ids() { assert_eq!(value["is_error"], true); assert!(value.get("session_id").is_none()); assert!(value.get("turn_id").is_none()); + assert!(value.get("error_code").is_none()); +} + +#[test] +fn session_conflict_json_error_exposes_the_existing_runtime_code() { + let result = ExecJsonResult::preflight_error("close the other instance") + .with_error_code("session_in_use"); + let value = serde_json::to_value(result).expect("serialize result"); + + assert_eq!(value["error_code"], "session_in_use"); + assert_eq!(value["result"], "close the other instance"); } #[test] @@ -406,6 +418,19 @@ fn stream_json_reuses_the_existing_agentic_envelope() { assert!(value.get("sequence").is_none()); } +#[test] +fn session_conflict_stream_json_reuses_system_error_without_a_new_event_schema() { + let envelope = session_in_use_stream_envelope(); + let encoded = serialize_stream_envelope(&envelope).expect("serialize envelope"); + let value: serde_json::Value = serde_json::from_str(&encoded).expect("JSONL record"); + + assert_eq!(value["event"]["type"], "SystemError"); + assert_eq!(value["event"]["error"], "session_in_use"); + assert_eq!(value["event"]["recoverable"], true); + assert!(value["event"]["session_id"].is_null()); + assert!(value.get("error_code").is_none()); +} + #[test] fn unsuccessful_completed_turn_is_an_error_outcome() { assert_eq!( diff --git a/src/apps/cli/src/peer_host/commands/session.rs b/src/apps/cli/src/peer_host/commands/session.rs index 4f8426d717..2a8201ca49 100644 --- a/src/apps/cli/src/peer_host/commands/session.rs +++ b/src/apps/cli/src/peer_host/commands/session.rs @@ -5,15 +5,19 @@ use std::time::{SystemTime, UNIX_EPOCH}; use serde_json::{json, Value}; -use bitfun_agent_runtime::sdk::{AgentSessionRestoreRequest, AgentSessionRestoreResult}; +use bitfun_agent_runtime::sdk::{ + AgentSessionRestoreRequest, AgentSessionRestoreResult, PortErrorKind, RuntimeError, +}; use bitfun_core::agentic::core::Session; use bitfun_core::agentic::get_agent_registry; +use bitfun_core::util::errors::BitFunError; use bitfun_runtime_ports::{ AgentSessionArchiveRequest, AgentSessionCreateRequest, AgentSessionDeleteRequest, AgentSessionModelUpdateRequest, AgentSessionRenameRequest, AgentThreadGoalGetRequest, SessionStoragePathRequest, }; +use crate::diagnostics::SESSION_IN_USE_ERROR_CODE; use crate::peer_host::args::{get_string, optional_bool, optional_string, request_value}; use crate::peer_host::state::PeerHostState; @@ -112,6 +116,24 @@ fn restored_session_to_json(restored: AgentSessionRestoreResult) -> Value { }) } +fn peer_core_session_error(operation: &str, error: BitFunError) -> String { + match error { + BitFunError::SessionInUse { session_id } => format!( + "{SESSION_IN_USE_ERROR_CODE}: Session is already open for writing: {session_id}" + ), + error => format!("{operation}: {error}"), + } +} + +fn peer_runtime_session_error(operation: &str, error: RuntimeError) -> String { + match error { + RuntimeError::Port(port_error) if port_error.kind == PortErrorKind::SessionInUse => { + format!("{SESSION_IN_USE_ERROR_CODE}: {}", port_error.message) + } + error => format!("{operation}: {}", error.into_message()), + } +} + pub(crate) async fn list_persisted_sessions( state: &PeerHostState, args: &Value, @@ -238,7 +260,7 @@ pub(crate) async fn restore_session_with_turns( .compatibility .restore_session_with_turns_for_workspace(storage_request, &session_id, include_internal) .await - .map_err(|e| format!("Failed to restore session with turns: {e}"))?; + .map_err(|error| peer_core_session_error("Failed to restore session with turns", error))?; let turn_count = turns.len(); Ok(json!({ @@ -266,7 +288,7 @@ pub(crate) async fn restore_session(state: &PeerHostState, args: &Value) -> Resu remote_ssh_host: storage_request.remote_ssh_host, }) .await - .map_err(|error| format!("Failed to restore session: {}", error.into_message()))?; + .map_err(|error| peer_runtime_session_error("Failed to restore session", error))?; Ok(restored_session_to_json(restored)) } @@ -315,7 +337,7 @@ pub(crate) async fn create_session(state: &PeerHostState, args: &Value) -> Resul } None => state.agent_runtime.create_session(create_request).await, } - .map_err(|error| format!("Failed to create session: {}", error.into_message()))?; + .map_err(|error| peer_runtime_session_error("Failed to create session", error))?; Ok(json!({ "sessionId": session.session_id, @@ -471,7 +493,7 @@ pub(crate) async fn ensure_coordinator_session( .ensure_session_loaded_from_storage_path(&storage, &session_id, include_internal) .await .map(|_| Value::Null) - .map_err(|e| e.to_string()) + .map_err(|error| peer_core_session_error("Failed to ensure session", error)) } pub(crate) async fn get_available_modes() -> Result { @@ -572,12 +594,53 @@ pub(crate) async fn save_session_turn( #[cfg(test)] mod tests { use super::{ - overlay_live_session_state, restored_session_to_json, session_stats_validation_error, + overlay_live_session_state, peer_core_session_error, peer_runtime_session_error, + restored_session_to_json, session_stats_validation_error, + }; + use bitfun_agent_runtime::sdk::{ + AgentSessionRestoreResult, AgentSessionSummary, PortError, PortErrorKind, RuntimeError, + SessionState, }; - use bitfun_agent_runtime::sdk::{AgentSessionRestoreResult, AgentSessionSummary, SessionState}; use bitfun_core::agentic::core::{ ProcessingPhase, Session as CoreSession, SessionConfig, SessionState as CoreSessionState, }; + use bitfun_core::util::errors::BitFunError; + + #[test] + fn peer_writer_conflicts_keep_the_stable_transport_code() { + let core_error = peer_core_session_error( + "Failed to restore session with turns", + BitFunError::SessionInUse { + session_id: "session-1".to_string(), + }, + ); + let runtime_error = peer_runtime_session_error( + "Failed to restore session", + RuntimeError::Port(PortError::new( + PortErrorKind::SessionInUse, + "Session is already open for writing: session-1", + )), + ); + + assert_eq!( + core_error, + "session_in_use: Session is already open for writing: session-1" + ); + assert_eq!(runtime_error, core_error); + } + + #[test] + fn peer_writer_errors_keep_operation_context_when_they_are_not_conflicts() { + let error = peer_runtime_session_error( + "Failed to restore session", + RuntimeError::MissingSessionRestorePort, + ); + + assert_eq!( + error, + "Failed to restore session: agent session restore port is not registered" + ); + } #[test] fn peer_attach_and_raw_mutations_reuse_core_runtime_ownership() { diff --git a/src/apps/desktop/src/runtime/session_application.rs b/src/apps/desktop/src/runtime/session_application.rs index 18298f7199..903a974520 100644 --- a/src/apps/desktop/src/runtime/session_application.rs +++ b/src/apps/desktop/src/runtime/session_application.rs @@ -24,6 +24,7 @@ use bitfun_core::service::session::{DialogTurnData, SessionMetadata, SessionStat use bitfun_core::service::session_usage::SessionUsageReport; use bitfun_core::service::token_usage::TokenUsageService; use bitfun_core::service::workspace::WorkspaceService; +use bitfun_core::util::errors::BitFunError; use serde::{Deserialize, Serialize}; use tokio::sync::RwLock; @@ -58,10 +59,21 @@ pub(crate) enum DesktopSessionApplicationError { Runtime(String), #[error("{0}")] RestoreBeforeRename(String), + #[error("session_in_use: {0}")] + SessionInUse(String), } pub(crate) type DesktopSessionApplicationResult = Result; +fn desktop_core_session_error(error: BitFunError) -> DesktopSessionApplicationError { + match error { + BitFunError::SessionInUse { session_id } => DesktopSessionApplicationError::SessionInUse( + format!("Session is already open for writing: {session_id}"), + ), + error => DesktopSessionApplicationError::Core(error.to_string()), + } +} + #[derive(Debug)] pub(crate) struct DesktopSessionViewRestore { pub session: Session, @@ -541,7 +553,7 @@ impl DesktopSessionApplication { self.compatibility .ensure_session_loaded_from_storage_path(&storage_path, session_id, include_internal) .await - .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + .map_err(desktop_core_session_error) } pub(crate) async fn restore_session( @@ -556,7 +568,7 @@ impl DesktopSessionApplication { self.compatibility .restore_session_from_storage_path(&storage_path, session_id, include_internal) .await - .map_err(|error| DesktopSessionApplicationError::Core(error.to_string())) + .map_err(desktop_core_session_error) } pub(crate) async fn restore_session_view( @@ -625,7 +637,7 @@ impl DesktopSessionApplication { include_internal, ) .await - .map_err(|error| DesktopSessionApplicationError::Core(error.to_string()))?; + .map_err(desktop_core_session_error)?; Ok(DesktopSessionWithTurnsRestore { session, turns }) } } @@ -707,6 +719,23 @@ mod tests { use serde_json::json; use std::sync::Mutex; + #[test] + fn session_writer_conflict_keeps_a_stable_desktop_transport_code() { + let error = + desktop_core_session_error(bitfun_core::util::errors::BitFunError::SessionInUse { + session_id: "session-1".to_string(), + }); + + assert!(matches!( + error, + DesktopSessionApplicationError::SessionInUse(_) + )); + assert_eq!( + error.to_string(), + "session_in_use: Session is already open for writing: session-1" + ); + } + struct RecordingDeletePort { events: Arc>>, workspace_path: Arc>>, diff --git a/src/web-ui/src/flow_chat/components/ChatInput.tsx b/src/web-ui/src/flow_chat/components/ChatInput.tsx index 4a54f015e8..d12c58f2de 100644 --- a/src/web-ui/src/flow_chat/components/ChatInput.tsx +++ b/src/web-ui/src/flow_chat/components/ChatInput.tsx @@ -58,6 +58,11 @@ import { sessionComposerStore, type PendingLargePasteMap, } from '../store/sessionComposerStore'; +import { + failedSubmissionRecoveryTarget, + shouldRecordContextMutation, + successfulRetryCleanupTarget, +} from './chatInputDraftRecovery'; import { startBtwThread } from '../services/BtwThreadService'; import { runUsageReportCommand } from '../services/usageReportService'; import { buildImagePayload } from '../utils/imagePayload'; @@ -401,6 +406,9 @@ export const ChatInput: React.FC = ({ // Ref so the queuedInput sync effect can read the latest value without it being a dep const inputValueRef = useRef(''); const pendingLargePastesRef = useRef({}); + const composerMutationRevisionsRef = useRef(new Map()); + const isRestoringSessionDraftRef = useRef(false); + const sessionConflictRetryBaselinesRef = useRef(new Map()); const reviewLaunchPendingRef = useRef(false); const largePasteCountersRef = useRef>({}); const undoImageStackRef = useRef([]); @@ -449,6 +457,17 @@ export const ChatInput: React.FC = ({ const effectiveTargetSessionIdRef = useRef(effectiveTargetSessionId); effectiveTargetSessionIdRef.current = effectiveTargetSessionId; + const markComposerMutation = useCallback(() => { + const sessionId = effectiveTargetSessionIdRef.current; + if (!sessionId) return; + const revisions = composerMutationRevisionsRef.current; + revisions.set(sessionId, (revisions.get(sessionId) ?? 0) + 1); + }, []); + const composerMutationRevision = useCallback( + (sessionId: string) => composerMutationRevisionsRef.current.get(sessionId) ?? 0, + [], + ); + useComposerDefaultFocus({ editorRef: richTextInputRef, sessionId: effectiveTargetSessionId, @@ -460,6 +479,7 @@ export const ChatInput: React.FC = ({ || (action.type === 'CLEAR_VALUE' && inputValueRef.current !== ''); if (changesValue) { nativePromptModeSelectionGenerationRef.current += 1; + markComposerMutation(); } dispatchLocalInput(action); @@ -475,7 +495,7 @@ export const ChatInput: React.FC = ({ inputValueRef.current = ''; sessionComposerStore.getState().setValue(sessionId, ''); } - }, []); + }, [markComposerMutation]); const effectiveTargetSession = effectiveTargetSessionId ? flowChatState.sessions.get(effectiveTargetSessionId) : undefined; @@ -1093,23 +1113,6 @@ export const ChatInput: React.FC = ({ setHistoryIndex(-1); }, [effectiveTargetSessionId]); - const { sendMessage } = useMessageSender({ - currentSessionId: effectiveTargetSessionId || undefined, - contexts, - onClearContexts: clearContexts, - onSuccess: onSendMessage, - currentAgentType: resolveChatInputSendAgentType({ - isSubagentTarget: isSubagentInputTarget, - subagentType: effectiveTargetSession?.subagentType, - sessionMode: effectiveTargetSession?.mode, - acpTargetAgentType, - // Composer mode is authoritative for normal sessions (synced from session - // on switch, updated in applyModeChange). Subagent continuations keep the - // child session's own agent type instead of inheriting the parent composer. - composerMode: modeState.current, - }), - }); - const modeInfoById = useMemo( () => new Map(modeState.available.map(mode => [mode.id, mode])), [modeState.available], @@ -1557,7 +1560,12 @@ export const ChatInput: React.FC = ({ dispatchLocalInput({ type: 'SET_VALUE', payload: nextValue }); inputValueRef.current = nextValue; pendingLargePastesRef.current = { ...nextPendingLargePastes }; - replaceContexts(nextContexts); + isRestoringSessionDraftRef.current = true; + try { + replaceContexts(nextContexts); + } finally { + isRestoringSessionDraftRef.current = false; + } setHistoryIndex(-1); setSavedDraft(''); setMentionState({ isActive: false, query: '', startOffset: 0 }); @@ -1576,7 +1584,15 @@ export const ChatInput: React.FC = ({ }, [effectiveTargetSessionId, replaceContexts]); useEffect(() => { + let previousContexts = useContextStore.getState().contexts; const unsubscribe = useContextStore.subscribe((state) => { + if (shouldRecordContextMutation( + state.contexts !== previousContexts, + isRestoringSessionDraftRef.current, + )) { + markComposerMutation(); + } + previousContexts = state.contexts; const sessionId = effectiveTargetSessionIdRef.current; if (sessionId) { sessionComposerStore.getState().setContexts(sessionId, state.contexts); @@ -1593,22 +1609,88 @@ export const ChatInput: React.FC = ({ } unsubscribe(); }; - }, []); + }, [markComposerMutation]); const replacePendingLargePastes = useCallback((pendingLargePastes: PendingLargePasteMap) => { const nextPendingLargePastes = { ...pendingLargePastes }; + const previousPendingLargePastes = pendingLargePastesRef.current; + const previousKeys = Object.keys(previousPendingLargePastes); + const nextKeys = Object.keys(nextPendingLargePastes); + if ( + previousKeys.length !== nextKeys.length || + nextKeys.some(key => previousPendingLargePastes[key] !== nextPendingLargePastes[key]) + ) { + markComposerMutation(); + } pendingLargePastesRef.current = nextPendingLargePastes; const sessionId = effectiveTargetSessionIdRef.current; if (sessionId) { sessionComposerStore.getState().setPendingLargePastes(sessionId, nextPendingLargePastes); } - }, []); + }, [markComposerMutation]); const clearPendingLargePastes = useCallback(() => { replacePendingLargePastes({}); }, [replacePendingLargePastes]); + const { sendMessage } = useMessageSender({ + currentSessionId: effectiveTargetSessionId || undefined, + contexts, + onClearContexts: clearContexts, + onSuccess: onSendMessage, + onSessionConflictRetryStart: ({ sessionId }) => { + sessionConflictRetryBaselinesRef.current.set( + sessionId, + composerMutationRevision(sessionId), + ); + }, + onSessionConflictRetrySuccess: ({ sessionId, message, contextIds }) => { + const baselineRevision = sessionConflictRetryBaselinesRef.current.get(sessionId); + sessionConflictRetryBaselinesRef.current.delete(sessionId); + const isCurrentSession = effectiveTargetSessionIdRef.current === sessionId; + const draft = isCurrentSession + ? { + value: inputValueRef.current, + contexts: contextsRef.current, + } + : sessionComposerStore.getState().getDraft(sessionId); + const cleanupTarget = baselineRevision !== undefined + ? successfulRetryCleanupTarget( + sessionId, + effectiveTargetSessionIdRef.current, + baselineRevision, + composerMutationRevision(sessionId), + draft.value, + draft.contexts.map(context => context.id), + message, + contextIds, + ) + : 'none'; + + if (cleanupTarget === 'current') { + clearContexts(); + clearPendingLargePastes(); + dispatchInput({ type: 'CLEAR_VALUE' }); + setQueuedInput(null); + dispatchInput({ type: 'DEACTIVATE' }); + } else if (cleanupTarget === 'stored') { + sessionComposerStore.getState().clearDraft(sessionId); + } + onSendMessage?.(message); + }, + currentAgentType: resolveChatInputSendAgentType({ + isSubagentTarget: isSubagentInputTarget, + subagentType: effectiveTargetSession?.subagentType, + sessionMode: effectiveTargetSession?.mode, + acpTargetAgentType, + // Composer mode is authoritative for normal sessions (synced from session + // on switch, updated in applyModeChange). Subagent continuations keep the + // child session's own agent type instead of inheriting the parent composer. + composerMode: modeState.current, + }), + }); + const consumedRegisteredDraftRef = useRef<{ registrationId?: string; draftId: number; @@ -3689,6 +3771,7 @@ export const ChatInput: React.FC = ({ if (!draftTrimmed) return; const originalMessage = draftTrimmed; + const submissionSessionId = effectiveTargetSessionId; const composerPresentation = messageOverride === undefined ? richTextInputRef.current?.getComposerPresentation?.() ?? null : null; @@ -3837,6 +3920,9 @@ export const ChatInput: React.FC = ({ clearPendingLargePastes(); // Clear machine queue too; otherwise the queuedInput→input sync effect puts the text back after send. setQueuedInput(null); + const clearedComposerRevision = submissionSessionId + ? composerMutationRevision(submissionSessionId) + : 0; try { const transport = await submitThroughChatInputRegistration( @@ -3864,11 +3950,23 @@ export const ChatInput: React.FC = ({ dispatchInput({ type: 'DEACTIVATE' }); } catch (error) { log.error('Failed to send message', { error }); - replacePendingLargePastes(originalPendingLargePastes); - dispatchInput({ type: 'ACTIVATE' }); - dispatchInput({ type: 'SET_VALUE', payload: originalMessage }); - if (derivedState?.isProcessing) { - setQueuedInput(originalMessage); + const recoveryTarget = failedSubmissionRecoveryTarget( + submissionSessionId, + effectiveTargetSessionIdRef.current, + clearedComposerRevision, + submissionSessionId ? composerMutationRevision(submissionSessionId) : 0, + ); + if (recoveryTarget === 'current') { + replacePendingLargePastes(originalPendingLargePastes); + dispatchInput({ type: 'ACTIVATE' }); + dispatchInput({ type: 'SET_VALUE', payload: originalMessage }); + if (derivedState?.isProcessing) { + setQueuedInput(originalMessage); + } + } else if (recoveryTarget === 'stored' && submissionSessionId) { + const composer = sessionComposerStore.getState(); + composer.setValue(submissionSessionId, originalMessage); + composer.setPendingLargePastes(submissionSessionId, originalPendingLargePastes); } } }, [ @@ -3907,6 +4005,7 @@ export const ChatInput: React.FC = ({ resolveTypedMcpPromptCommand, submitExternalPromptCommandFromInput, usesDispatchTransport, + composerMutationRevision, ]); const getFilteredIncrementalModes = useCallback(() => { diff --git a/src/web-ui/src/flow_chat/components/ChatInputDraftRecovery.test.ts b/src/web-ui/src/flow_chat/components/ChatInputDraftRecovery.test.ts new file mode 100644 index 0000000000..74168fac69 --- /dev/null +++ b/src/web-ui/src/flow_chat/components/ChatInputDraftRecovery.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it } from 'vitest'; +import { + failedSubmissionRecoveryTarget, + shouldRecordContextMutation, + successfulRetryCleanupTarget, +} from './chatInputDraftRecovery'; + +describe('ChatInput failed submission recovery', () => { + it('restores the visible composer only when the same session is still unchanged', () => { + expect(failedSubmissionRecoveryTarget('session-a', 'session-a', 4, 4)).toBe('current'); + }); + + it('does not overwrite a newer draft in the same session', () => { + expect(failedSubmissionRecoveryTarget('session-a', 'session-a', 4, 5)).toBe('none'); + }); + + it('restores the failed draft to its session store after the user switches sessions', () => { + expect(failedSubmissionRecoveryTarget('session-a', 'session-b', 4, 4)).toBe('stored'); + }); + + it('does not overwrite a newer draft in the original session after switching away', () => { + expect(failedSubmissionRecoveryTarget('session-a', 'session-b', 4, 5)).toBe('none'); + }); + + it('does not count A to B to A draft restoration as a user context mutation', () => { + expect(shouldRecordContextMutation(true, true)).toBe(false); + expect(failedSubmissionRecoveryTarget('session-a', 'session-a', 4, 4)).toBe('current'); + }); + + it('clears the stored draft after retry succeeds while another session is visible', () => { + expect(successfulRetryCleanupTarget( + 'session-a', + 'session-b', + 4, + 4, + 'retry this', + ['context-a'], + 'retry this', + ['context-a'], + )).toBe('stored'); + }); + + it('does not clear a stored draft changed before or during retry', () => { + expect(successfulRetryCleanupTarget( + 'session-a', + 'session-b', + 4, + 5, + 'newer draft', + ['context-a'], + 'retry this', + ['context-a'], + )).toBe('none'); + }); +}); diff --git a/src/web-ui/src/flow_chat/components/chatInputDraftRecovery.ts b/src/web-ui/src/flow_chat/components/chatInputDraftRecovery.ts new file mode 100644 index 0000000000..33aa4e0448 --- /dev/null +++ b/src/web-ui/src/flow_chat/components/chatInputDraftRecovery.ts @@ -0,0 +1,38 @@ +export function failedSubmissionRecoveryTarget( + submissionSessionId: string | null, + currentSessionId: string | null, + submissionRevision: number, + currentRevision: number, +): 'current' | 'stored' | 'none' { + if (submissionSessionId === currentSessionId) { + return submissionRevision === currentRevision ? 'current' : 'none'; + } + return submissionSessionId && submissionRevision === currentRevision ? 'stored' : 'none'; +} + +export function shouldRecordContextMutation( + contextsChanged: boolean, + isRestoringSessionDraft: boolean, +): boolean { + return contextsChanged && !isRestoringSessionDraft; +} + +export function successfulRetryCleanupTarget( + submissionSessionId: string, + currentSessionId: string | null, + baselineRevision: number, + currentRevision: number, + currentMessage: string, + currentContextIds: string[], + submittedMessage: string, + submittedContextIds: string[], +): 'current' | 'stored' | 'none' { + const composerIsUnchanged = + baselineRevision === currentRevision + && currentMessage.trim() === submittedMessage.trim() + && currentContextIds.length === submittedContextIds.length + && currentContextIds.every((id, index) => id === submittedContextIds[index]); + + if (!composerIsUnchanged) return 'none'; + return currentSessionId === submissionSessionId ? 'current' : 'stored'; +} diff --git a/src/web-ui/src/flow_chat/hooks/useMessageSender.ts b/src/web-ui/src/flow_chat/hooks/useMessageSender.ts index c8312170d5..ecba543b41 100644 --- a/src/web-ui/src/flow_chat/hooks/useMessageSender.ts +++ b/src/web-ui/src/flow_chat/hooks/useMessageSender.ts @@ -40,6 +40,18 @@ interface UseMessageSenderProps { onExitTemplateMode?: () => void; /** Selected agent type (mode) */ currentAgentType?: string; + /** Reconcile the composer after an explicit session-conflict retry succeeds. */ + onSessionConflictRetrySuccess?: (submission: { + sessionId: string; + message: string; + contextIds: string[]; + }) => void; + /** Capture composer state when the user explicitly starts a conflict retry. */ + onSessionConflictRetryStart?: (submission: { + sessionId: string; + message: string; + contextIds: string[]; + }) => void; } interface UseMessageSenderReturn { @@ -65,6 +77,8 @@ export function useMessageSender(props: UseMessageSenderProps): UseMessageSender onSuccess, onExitTemplateMode, currentAgentType, + onSessionConflictRetryStart, + onSessionConflictRetrySuccess, } = props; const sendMessage = useCallback(async ( @@ -188,6 +202,20 @@ export function useMessageSender(props: UseMessageSenderProps): UseMessageSender ...(imagePayload ?? {}), ...(userMessageMetadata ? { userMessageMetadata } : {}), ...(options?.dispatchAutoConfirmed ? { dispatchAutoConfirmed: true } : {}), + onSessionConflictRetryStart: () => { + onSessionConflictRetryStart?.({ + sessionId: sessionId!, + message: displayMessage, + contextIds: contexts.map(context => context.id), + }); + }, + onSessionConflictRetrySuccess: () => { + onSessionConflictRetrySuccess?.({ + sessionId: sessionId!, + message: displayMessage, + contextIds: contexts.map(context => context.id), + }); + }, } ); @@ -211,7 +239,16 @@ export function useMessageSender(props: UseMessageSenderProps): UseMessageSender }); throw error; } - }, [currentSessionId, contexts, onClearContexts, onSuccess, onExitTemplateMode, currentAgentType]); + }, [ + currentSessionId, + contexts, + onClearContexts, + onSuccess, + onExitTemplateMode, + currentAgentType, + onSessionConflictRetryStart, + onSessionConflictRetrySuccess, + ]); return { sendMessage, diff --git a/src/web-ui/src/flow_chat/services/FlowChatManager.ts b/src/web-ui/src/flow_chat/services/FlowChatManager.ts index de7aacd062..2fef896f61 100644 --- a/src/web-ui/src/flow_chat/services/FlowChatManager.ts +++ b/src/web-ui/src/flow_chat/services/FlowChatManager.ts @@ -661,6 +661,8 @@ export class FlowChatManager { preserveTurnOnStartError?: boolean; /** One-shot UI confirmation for unattended auto approval. */ dispatchAutoConfirmed?: boolean; + onSessionConflictRetryStart?: () => void; + onSessionConflictRetrySuccess?: () => void; } ): Promise { const targetSessionId = sessionId || this.context.flowChatStore.getState().activeSessionId; 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 53dbc5f97a..c0bf442742 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 @@ -3,14 +3,20 @@ import { cancelSessionTask, sendMessage, syncSessionModelSelection } from './Mes import { SessionExecutionEvent } from '../../state-machine/types'; const mockTransition = vi.fn(); +const mockGetCurrentState = vi.fn(() => 'processing'); +const mockGetStateMachine = vi.fn(() => null); const mockUpdateSessionModel = vi.fn(); +const mockStartDialogTurn = vi.fn(); const mockGetConfigs = vi.fn(); -const mockGetCurrentState = vi.fn(() => 'processing'); const mockDispatchSubmit = vi.fn(); const mockDispatchProgress = vi.fn(); const mockDispatchRefresh = vi.fn(); -const mockStartDialogTurn = vi.fn(); const mockBindSession = vi.fn(); +const mockEnsureBackendSession = vi.fn(); +const mockNotificationError = vi.fn(); +const mockNotificationDismiss = vi.fn(); +const mockPendingList = vi.fn((): unknown[] => []); +const mockPendingEnqueue = vi.fn(); vi.mock('../../state-machine', () => ({ SessionExecutionEvent: { @@ -21,7 +27,8 @@ vi.mock('../../state-machine', () => ({ PROCESSING: 'processing', }, stateMachineManager: { - getCurrentState: () => mockGetCurrentState(), + getCurrentState: (...args: unknown[]) => mockGetCurrentState(...args), + get: (...args: unknown[]) => mockGetStateMachine(...args), transition: (...args: any[]) => mockTransition(...args), }, })); @@ -57,13 +64,6 @@ vi.mock('@/features/dispatch/DispatchJobObserver', () => ({ requestDispatchJobRefresh: (...args: unknown[]) => mockDispatchRefresh(...args), })); -vi.mock('./PendingQueueModule', () => ({ - pendingQueueManager: { - list: () => [], - enqueue: vi.fn(), - }, -})); - vi.mock('@/infrastructure/api/service-api/ACPClientAPI', () => ({ ACPClientAPI: {}, })); @@ -76,10 +76,271 @@ vi.mock('@/infrastructure/config/services/ConfigManager', () => ({ vi.mock('../../../shared/notification-system', () => ({ notificationService: { - error: vi.fn(), + error: (...args: unknown[]) => mockNotificationError(...args), + dismiss: (...args: unknown[]) => mockNotificationDismiss(...args), }, })); +vi.mock('./SessionModule', () => ({ + ensureBackendSession: (...args: unknown[]) => mockEnsureBackendSession(...args), + getModelMaxTokens: vi.fn(async (modelId: string) => modelId === 'auto' ? 32000 : 64000), + retryCreateBackendSession: vi.fn(), +})); + +vi.mock('./PendingQueueModule', () => ({ + pendingQueueManager: { + list: (...args: unknown[]) => mockPendingList(...args), + enqueue: (...args: unknown[]) => mockPendingEnqueue(...args), + }, +})); + +vi.mock('@/infrastructure/i18n', () => ({ + i18nService: { + t: (key: string) => key, + }, +})); + +describe('MessageModule session writer conflict', () => { + function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((resolvePromise, rejectPromise) => { + resolve = resolvePromise; + reject = rejectPromise; + }); + return { promise, resolve, reject }; + } + + function conflictContext(sessionId: string) { + const session = { + sessionId, + mode: 'agentic', + dialogTurns: [] as any[], + config: { modelName: 'auto' }, + titleStatus: 'generated', + maxContextTokens: 32000, + }; + return { + session, + context: { + flowChatStore: { + getState: () => ({ sessions: new Map([[sessionId, session]]) }), + addDialogTurn: vi.fn((_id: string, turn: any) => session.dialogTurns.push(turn)), + deleteDialogTurn: vi.fn((_id: string, turnId: string) => { + session.dialogTurns = session.dialogTurns.filter(turn => turn.id !== turnId); + }), + updateSessionLastSubmittedMode: vi.fn(), + updateSessionMode: vi.fn(), + updateSessionModelName: vi.fn(), + updateSessionMaxContextTokens: vi.fn(), + }, + processingManager: { + registerStatus: vi.fn(), + clearSessionStatus: vi.fn(), + }, + pendingHistoryLoads: new Map(), + contentBuffers: new Map(), + activeTextItems: new Map(), + } as any, + }; + } + + beforeEach(() => { + vi.resetAllMocks(); + mockGetCurrentState.mockReturnValue('idle'); + mockGetStateMachine.mockReturnValue(null); + mockTransition.mockResolvedValue(true); + mockUpdateSessionModel.mockResolvedValue(undefined); + mockStartDialogTurn.mockResolvedValue(undefined); + mockPendingList.mockReturnValue([]); + mockPendingEnqueue.mockReturnValue({ id: 'queued-message' }); + mockEnsureBackendSession.mockRejectedValue( + new Error('session_in_use: Session is already open for writing: session-1'), + ); + mockNotificationError + .mockReturnValueOnce('notification-1') + .mockReturnValueOnce('notification-2') + .mockReturnValue('notification-3'); + }); + + it('keeps only the latest explicit retry for one conflicted session', async () => { + const session = { + sessionId: 'session-1', + mode: 'agentic', + dialogTurns: [], + config: {}, + titleStatus: 'generated', + }; + const context: any = { + flowChatStore: { + getState: () => ({ sessions: new Map([['session-1', session]]) }), + deleteDialogTurn: vi.fn(), + }, + pendingHistoryLoads: new Map(), + }; + + await expect(sendMessage(context, 'hello', 'session-1')).rejects.toThrow( + 'session_in_use', + ); + + expect(context.flowChatStore.deleteDialogTurn).not.toHaveBeenCalled(); + expect(mockEnsureBackendSession).toHaveBeenCalledTimes(1); + expect(mockNotificationError).toHaveBeenCalledTimes(1); + const [message, options] = mockNotificationError.mock.calls[0]; + expect(message).toBe('flow-chat:session.inUseMessage'); + expect(options).toMatchObject({ + title: 'flow-chat:session.inUseTitle', + duration: 0, + }); + expect(options.actions).toHaveLength(1); + expect(options.actions[0].label).toBe('flow-chat:session.retry'); + + await expect(sendMessage(context, 'hello', 'session-1')).rejects.toThrow( + 'session_in_use', + ); + expect(mockNotificationDismiss).toHaveBeenCalledWith('notification-1'); + const staleAction = options.actions[0]; + const latestAction = mockNotificationError.mock.calls[1][1].actions[0]; + + staleAction.onClick(); + expect(mockEnsureBackendSession).toHaveBeenCalledTimes(2); + + latestAction.onClick(); + latestAction.onClick(); + await vi.waitFor(() => { + expect(mockEnsureBackendSession).toHaveBeenCalledTimes(3); + }); + }); + + it('does not let an older retry failure replace a newer conflict', async () => { + const sessionId = 'session-race-failure'; + const { context } = conflictContext(sessionId); + const conflict = new Error(`session_in_use: Session is already open for writing: ${sessionId}`); + + await expect(sendMessage(context, 'older', sessionId)).rejects.toThrow('session_in_use'); + const retryAction = mockNotificationError.mock.calls[0][1].actions[0]; + const olderRetry = deferred(); + const newerSend = deferred(); + mockEnsureBackendSession + .mockImplementationOnce(() => olderRetry.promise) + .mockImplementationOnce(() => newerSend.promise); + + retryAction.onClick(); + await vi.waitFor(() => expect(mockEnsureBackendSession).toHaveBeenCalledTimes(2)); + const newerResult = sendMessage(context, 'newer', sessionId); + await vi.waitFor(() => expect(mockEnsureBackendSession).toHaveBeenCalledTimes(3)); + + newerSend.reject(conflict); + await expect(newerResult).rejects.toThrow('session_in_use'); + expect(mockNotificationError).toHaveBeenCalledTimes(2); + + olderRetry.reject(conflict); + await vi.waitFor(() => expect(mockEnsureBackendSession).toHaveBeenCalledTimes(3)); + await Promise.resolve(); + + expect(mockNotificationError).toHaveBeenCalledTimes(2); + expect(mockNotificationDismiss).not.toHaveBeenCalledWith('notification-2'); + }); + + it('does not let an older retry success dismiss a newer conflict', async () => { + const sessionId = 'session-race-success'; + const { context } = conflictContext(sessionId); + const conflict = new Error(`session_in_use: Session is already open for writing: ${sessionId}`); + const retryStart = vi.fn(); + const retrySuccess = vi.fn(); + + await expect(sendMessage(context, 'older', sessionId, undefined, undefined, undefined, { + onSessionConflictRetryStart: retryStart, + onSessionConflictRetrySuccess: retrySuccess, + })).rejects.toThrow('session_in_use'); + const retryAction = mockNotificationError.mock.calls[0][1].actions[0]; + const olderRetry = deferred(); + const newerSend = deferred(); + mockEnsureBackendSession + .mockImplementationOnce(() => olderRetry.promise) + .mockImplementationOnce(() => newerSend.promise); + + retryAction.onClick(); + expect(retryStart).toHaveBeenCalledTimes(1); + await vi.waitFor(() => expect(mockEnsureBackendSession).toHaveBeenCalledTimes(2)); + const newerResult = sendMessage(context, 'newer', sessionId); + await vi.waitFor(() => expect(mockEnsureBackendSession).toHaveBeenCalledTimes(3)); + + newerSend.reject(conflict); + await expect(newerResult).rejects.toThrow('session_in_use'); + olderRetry.resolve(undefined); + await vi.waitFor(() => expect(mockStartDialogTurn).toHaveBeenCalledTimes(1)); + + expect(mockNotificationError).toHaveBeenCalledTimes(2); + expect(mockNotificationDismiss).not.toHaveBeenCalledWith('notification-2'); + expect(retrySuccess).not.toHaveBeenCalled(); + }); + + it('does not let an older retry failure reset a newer processing turn', async () => { + const sessionId = 'session-processing-race'; + const { context } = conflictContext(sessionId); + const conflict = new Error(`session_in_use: Session is already open for writing: ${sessionId}`); + + await expect(sendMessage(context, 'older', sessionId)).rejects.toThrow('session_in_use'); + const retryAction = mockNotificationError.mock.calls[0][1].actions[0]; + const olderRetry = deferred(); + const newerTurn = deferred(); + let currentState = 'idle'; + let currentDialogTurnId: string | null = null; + mockGetCurrentState.mockImplementation(() => currentState); + mockGetStateMachine.mockReturnValue({ + getContext: () => ({ currentDialogTurnId }), + } as any); + mockTransition.mockImplementation(async (_id, event, payload) => { + if (event === 'start') { + currentState = 'processing'; + currentDialogTurnId = payload.dialogTurnId; + } + return true; + }); + mockEnsureBackendSession + .mockImplementationOnce(() => olderRetry.promise) + .mockResolvedValueOnce(undefined); + mockStartDialogTurn.mockImplementationOnce(() => newerTurn.promise); + + retryAction.onClick(); + await vi.waitFor(() => expect(mockEnsureBackendSession).toHaveBeenCalledTimes(2)); + const newerResult = sendMessage(context, 'newer', sessionId); + await vi.waitFor(() => expect(mockStartDialogTurn).toHaveBeenCalledTimes(1)); + + olderRetry.reject(conflict); + await new Promise(resolve => setTimeout(resolve, 0)); + expect(mockTransition).not.toHaveBeenCalledWith( + sessionId, + 'error_occurred', + expect.anything(), + ); + expect(mockTransition).not.toHaveBeenCalledWith(sessionId, 'reset'); + + newerTurn.resolve(undefined); + await expect(newerResult).resolves.toBeUndefined(); + }); + + it('invalidates an old retry when a newer message is queued', async () => { + const sessionId = 'session-queue-success'; + const { context } = conflictContext(sessionId); + + await expect(sendMessage(context, 'older', sessionId)).rejects.toThrow('session_in_use'); + const retryAction = mockNotificationError.mock.calls[0][1].actions[0]; + mockGetCurrentState.mockReturnValue('processing'); + + await expect(sendMessage(context, 'newer', sessionId)).resolves.toBeUndefined(); + expect(mockPendingEnqueue).toHaveBeenCalledWith(expect.objectContaining({ + sessionId, + content: 'newer', + })); + expect(mockNotificationDismiss).toHaveBeenCalledWith('notification-1'); + + retryAction.onClick(); + expect(mockEnsureBackendSession).toHaveBeenCalledTimes(1); + }); +}); + describe('MessageModule cancellation', () => { beforeEach(() => { vi.clearAllMocks(); 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 4b218172b2..e62bb80591 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 @@ -29,9 +29,46 @@ import { dispatchApi } from '@/features/dispatch/dispatchApi'; import { dispatchJobStore } from '@/features/dispatch/dispatchJobStore'; import { requestDispatchJobRefresh } from '@/features/dispatch/DispatchJobObserver'; import { isNonLocalDispatchTarget } from '@/features/dispatch/types'; +import { isSessionInUseError } from '@/infrastructure/api/errors/TauriCommandError'; +import { i18nService } from '@/infrastructure/i18n'; const log = createLogger('MessageModule'); +interface SessionConflictRetry { + notificationId: string; + active: boolean; + inFlight: boolean; +} + +const sessionConflictRetries = new Map(); +const latestSendBySession = new Map(); + +function clearSessionConflictRetry(sessionId: string): void { + const current = sessionConflictRetries.get(sessionId); + if (!current) return; + current.active = false; + sessionConflictRetries.delete(sessionId); + notificationService.dismiss(current.notificationId); +} + +function beginSessionSend(sessionId: string): symbol { + const attempt = Symbol(sessionId); + latestSendBySession.set(sessionId, attempt); + clearSessionConflictRetry(sessionId); + return attempt; +} + +function completeSessionSend( + sessionId: string, + attempt: symbol, + retrySuccess?: () => void, +): void { + if (latestSendBySession.get(sessionId) !== attempt) return; + latestSendBySession.delete(sessionId); + clearSessionConflictRetry(sessionId); + retrySuccess?.(); +} + function acpClientIdFromMode(mode: string | undefined): string | null { const value = mode?.trim(); if (!value?.startsWith('acp:')) return null; @@ -146,12 +183,16 @@ export async function sendMessage( preserveTurnOnStartError?: boolean; /** One-shot UI confirmation for unattended auto approval. Never persist this flag. */ dispatchAutoConfirmed?: boolean; + onSessionConflictRetryStart?: () => void; + onSessionConflictRetrySuccess?: () => void; + fromSessionConflictRetry?: boolean; } ): Promise { const session = context.flowChatStore.getState().sessions.get(sessionId); if (!session) { throw new Error(`Session does not exist: ${sessionId}`); } + const sendAttempt = beginSessionSend(sessionId); if (!options?.bypassPendingQueue) { const machineState = stateMachineManager.getCurrentState(sessionId); @@ -186,6 +227,13 @@ export async function sendMessage( }); throw error; } + completeSessionSend( + sessionId, + sendAttempt, + options?.fromSessionConflictRetry + ? options.onSessionConflictRetrySuccess + : undefined, + ); return; } } @@ -281,6 +329,13 @@ export async function sendMessage( }); context.flowChatStore.updateSessionLastSubmittedMode(sessionId, currentAgentType); requestDispatchJobRefresh(jobId); + completeSessionSend( + sessionId, + sendAttempt, + options?.fromSessionConflictRetry + ? options.onSessionConflictRetrySuccess + : undefined, + ); return; } @@ -454,6 +509,13 @@ export async function sendMessage( if (sessionStateMachine) { sessionStateMachine.getContext().taskId = sessionId; } + completeSessionSend( + sessionId, + sendAttempt, + options?.fromSessionConflictRetry + ? options.onSessionConflictRetrySuccess + : undefined, + ); } catch (error) { log.error('Failed to send message', { sessionId: sessionId, error }); @@ -461,7 +523,13 @@ export async function sendMessage( const errorMessage = error instanceof Error ? error.message : 'Failed to send message'; const currentState = stateMachineManager.getCurrentState(sessionId); - if (currentState === SessionExecutionState.PROCESSING) { + const activeDialogTurnId = stateMachineManager + .get(sessionId) + ?.getContext().currentDialogTurnId; + const ownsProcessingTurn = + createdLocalTurnId !== null && + activeDialogTurnId === createdLocalTurnId; + if (currentState === SessionExecutionState.PROCESSING && ownsProcessingTurn) { await stateMachineManager.transition(sessionId, SessionExecutionEvent.ERROR_OCCURRED, { error: errorMessage }); @@ -475,10 +543,58 @@ export async function sendMessage( } if (!options?.preserveTurnOnStartError) { - notificationService.error(errorMessage, { - title: 'Thinking process error', - duration: 5000 - }); + if (isSessionInUseError(error)) { + if (latestSendBySession.get(sessionId) !== sendAttempt) { + throw error; + } + clearSessionConflictRetry(sessionId); + const retry: SessionConflictRetry = { + notificationId: '', + active: true, + inFlight: false, + }; + retry.notificationId = notificationService.error( + i18nService.t('flow-chat:session.inUseMessage'), { + title: i18nService.t('flow-chat:session.inUseTitle'), + duration: 0, + actions: [{ + label: i18nService.t('flow-chat:session.retry'), + variant: 'primary', + onClick: () => { + if ( + !retry.active || + retry.inFlight || + sessionConflictRetries.get(sessionId) !== retry + ) { + return; + } + retry.inFlight = true; + options?.onSessionConflictRetryStart?.(); + void sendMessage( + context, + message, + sessionId, + displayMessage, + agentType, + switchToMode, + { ...options, fromSessionConflictRetry: true }, + ) + .catch(() => undefined); + }, + }], + }); + sessionConflictRetries.set(sessionId, retry); + } else { + if (latestSendBySession.get(sessionId) === sendAttempt) { + latestSendBySession.delete(sessionId); + notificationService.error(errorMessage, { + title: 'Thinking process error', + duration: 5000 + }); + } + } + } else if (latestSendBySession.get(sessionId) === sendAttempt) { + latestSendBySession.delete(sessionId); } throw error; 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 20e1d9d229..b8c0e77f02 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 @@ -1225,6 +1225,25 @@ describe('SessionModule historical session coordination', () => { }); }); + it('does not recreate a session that another BitFun instance is writing', async () => { + const { context } = createContext(createSession({ + isHistorical: false, + historyState: 'ready', + contextRestoreState: 'pending', + dialogTurns: [], + } as any)); + agentApiMocks.ensureCoordinatorSession.mockRejectedValueOnce( + new Error('session_in_use: Session is already open for writing: history-1') + ); + + await expect(ensureBackendSession(context, 'history-1')).rejects.toThrow( + 'session_in_use:', + ); + + expect(agentApiMocks.ensureCoordinatorSession).toHaveBeenCalledTimes(1); + expect(agentApiMocks.createSession).not.toHaveBeenCalled(); + }); + it('keeps recreate fallback for empty pending context sessions', async () => { const { context } = createContext(createSession({ isHistorical: false, 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 3bd4e7801f..11c31bac7a 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 @@ -6,6 +6,7 @@ import { agentAPI } from '@/infrastructure/api/service-api/AgentAPI'; import { configAPI } from '@/infrastructure/api/service-api/ConfigAPI'; import { sessionAPI } from '@/infrastructure/api/service-api/SessionAPI'; +import { isSessionInUseError } from '@/infrastructure/api/errors/TauriCommandError'; import { notificationService } from '../../../shared/notification-system'; import { createLogger } from '@/shared/utils/logger'; import { isRemoteTraceContext, startupTrace } from '@/shared/utils/startupTrace'; @@ -1312,6 +1313,9 @@ export async function ensureBackendSession( await ensureCoordinator(); } catch (e: any) { + if (isSessionInUseError(e)) { + throw e; + } if (!allowRecreateOnCoordinatorFailure) { const raw = typeof e?.message === 'string' ? e.message : String(e); const hint = diff --git a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts b/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts index 8f3f9e5746..5cb81154bb 100644 --- a/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts +++ b/src/web-ui/src/infrastructure/api/adapters/peer-device-adapter.test.ts @@ -5,6 +5,7 @@ import { PEER_READ_REQUEST_TIMEOUT_MS, PEER_RETRY_BASE_DELAY_MS, PeerDeviceTransportAdapter, + PeerProductCommandError, isPeerLocalOnlyCommand, isPeerRetryableIdempotentMutation, isPeerRetryableReadCommand, @@ -278,6 +279,23 @@ describe('PeerDeviceTransportAdapter queue', () => { ); }); + it('preserves a session conflict returned by the Peer Host without replaying it', async () => { + const deviceRpc = vi.fn().mockResolvedValue(JSON.stringify({ + resp: 'host_invoke_result', + ok: false, + error: 'session_in_use: Session is already open for writing: session-1', + })); + const adapter = new PeerDeviceTransportAdapter('peer-1', deviceRpc); + + await expect(adapter.request('ensure_coordinator_session', { + request: { sessionId: 'session-1', workspacePath: '/repo' }, + })).rejects.toEqual(expect.objectContaining>({ + name: 'PeerProductCommandError', + message: 'session_in_use: Session is already open for writing: session-1', + })); + expect(deviceRpc).toHaveBeenCalledTimes(1); + }); + it('recovers an idempotent dialog submission after a transient Relay failure', async () => { vi.useFakeTimers(); try { diff --git a/src/web-ui/src/infrastructure/api/errors/TauriCommandError.test.ts b/src/web-ui/src/infrastructure/api/errors/TauriCommandError.test.ts new file mode 100644 index 0000000000..f221528a26 --- /dev/null +++ b/src/web-ui/src/infrastructure/api/errors/TauriCommandError.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from 'vitest'; +import { isSessionInUseError, TauriCommandError } from './TauriCommandError'; + +describe('isSessionInUseError', () => { + it('recognizes local Tauri command errors without parsing human prose', () => { + const error = new TauriCommandError('Command failed', { + command: 'ensure_coordinator_session', + originalError: new Error( + 'session_in_use: Session is already open for writing: session-1', + ), + }); + + expect(isSessionInUseError(error)).toBe(true); + }); + + it('recognizes the same stable prefix through Peer error wrapping', () => { + const error = { + message: 'Host command failed', + details: { + originalError: + 'session_in_use: Session is already open for writing: session-1', + }, + }; + + expect(isSessionInUseError(error)).toBe(true); + }); + + it('does not classify similar human prose as the stable error', () => { + expect( + isSessionInUseError( + new Error('This session seems to be in use by another process'), + ), + ).toBe(false); + }); +}); diff --git a/src/web-ui/src/infrastructure/api/errors/TauriCommandError.ts b/src/web-ui/src/infrastructure/api/errors/TauriCommandError.ts index 6e48e9c0a7..b46ae2f73b 100644 --- a/src/web-ui/src/infrastructure/api/errors/TauriCommandError.ts +++ b/src/web-ui/src/infrastructure/api/errors/TauriCommandError.ts @@ -98,4 +98,37 @@ export function createTauriCommandError( export function isTauriCommandError(error: any): error is TauriCommandError { return error && error.isTauriCommandError === true; -} \ No newline at end of file +} + +const SESSION_IN_USE_PREFIX = 'session_in_use:'; + +/** Recognizes the stable Desktop/Peer error code without parsing localized prose. */ +export function isSessionInUseError(error: unknown): boolean { + const pending: unknown[] = [error]; + const seen = new Set(); + + for (let inspected = 0; pending.length > 0 && inspected < 12; inspected += 1) { + const current = pending.shift(); + if (typeof current === 'string') { + if (current.trimStart().startsWith(SESSION_IN_USE_PREFIX)) return true; + continue; + } + if (!current || typeof current !== 'object' || seen.has(current)) continue; + seen.add(current); + + const candidate = current as { + message?: unknown; + originalError?: unknown; + context?: { originalError?: unknown }; + details?: { originalError?: unknown }; + }; + pending.push( + candidate.message, + candidate.originalError, + candidate.context?.originalError, + candidate.details?.originalError, + ); + } + + return false; +} diff --git a/src/web-ui/src/locales/en-US/flow-chat.json b/src/web-ui/src/locales/en-US/flow-chat.json index 2f3df82811..60ccda7595 100644 --- a/src/web-ui/src/locales/en-US/flow-chat.json +++ b/src/web-ui/src/locales/en-US/flow-chat.json @@ -449,7 +449,10 @@ "switchSession": "Switch Session", "create": "Create New Session", "more": "More Sessions", - "restoreMain": "Restore Main Window" + "restoreMain": "Restore Main Window", + "inUseTitle": "Session open elsewhere", + "inUseMessage": "This session is open in another BitFun instance. Close it there, then retry.", + "retry": "Retry" }, "scroll": { "toBottom": "Back to bottom", diff --git a/src/web-ui/src/locales/zh-CN/flow-chat.json b/src/web-ui/src/locales/zh-CN/flow-chat.json index fc37bb3c14..dbcb5c0ede 100644 --- a/src/web-ui/src/locales/zh-CN/flow-chat.json +++ b/src/web-ui/src/locales/zh-CN/flow-chat.json @@ -449,7 +449,10 @@ "switchSession": "切换会话", "create": "创建新会话", "more": "更多会话", - "restoreMain": "恢复主窗口" + "restoreMain": "恢复主窗口", + "inUseTitle": "会话已在其他实例中打开", + "inUseMessage": "该会话已在另一个 BitFun 实例中打开。请先在那里关闭，然后重试。", + "retry": "重试" }, "scroll": { "toBottom": "回到底部", diff --git a/src/web-ui/src/locales/zh-TW/flow-chat.json b/src/web-ui/src/locales/zh-TW/flow-chat.json index 5ce855fba6..5b961a8a1a 100644 --- a/src/web-ui/src/locales/zh-TW/flow-chat.json +++ b/src/web-ui/src/locales/zh-TW/flow-chat.json @@ -449,7 +449,10 @@ "switchSession": "切換會話", "create": "建立新會話", "more": "更多會話", - "restoreMain": "恢復主視窗" + "restoreMain": "恢復主視窗", + "inUseTitle": "會話已在其他執行個體中開啟", + "inUseMessage": "此會話已在另一個 BitFun 執行個體中開啟。請先在該處關閉，然後重試。", + "retry": "重試" }, "scroll": { "toBottom": "回到底部", diff --git a/src/web-ui/src/shared/notification-system/components/NotificationItem.test.tsx b/src/web-ui/src/shared/notification-system/components/NotificationItem.test.tsx new file mode 100644 index 0000000000..5a576774db --- /dev/null +++ b/src/web-ui/src/shared/notification-system/components/NotificationItem.test.tsx @@ -0,0 +1,59 @@ +import React, { act } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import { JSDOM } from 'jsdom'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import type { Notification } from '../types'; +import { NotificationItem } from './NotificationItem'; + +vi.mock('@/infrastructure/i18n', () => ({ + useI18n: () => ({ t: (key: string) => key }), +})); + +vi.mock('../services/NotificationService', () => ({ + notificationService: { dismiss: vi.fn() }, +})); + +globalThis.IS_REACT_ACT_ENVIRONMENT = true; + +describe('NotificationItem accessibility', () => { + let dom: JSDOM; + let container: HTMLDivElement; + let root: Root; + + beforeEach(() => { + dom = new JSDOM('
'); + globalThis.window = dom.window as unknown as Window & typeof globalThis; + globalThis.document = dom.window.document; + container = document.getElementById('root') as HTMLDivElement; + root = createRoot(container); + }); + + afterEach(() => { + act(() => root.unmount()); + dom.window.close(); + }); + + it('announces an actionable error while focus remains in the composer', () => { + const notification: Notification = { + id: 'session-conflict', + type: 'error', + variant: 'toast', + title: 'Session is in use', + message: 'Close the other instance and retry.', + timestamp: 1, + duration: 0, + closable: true, + actions: [{ label: 'Retry', onClick: vi.fn() }], + status: 'active', + }; + + act(() => root.render()); + + const item = container.querySelector('.notification-item'); + expect(item?.getAttribute('role')).toBe('alert'); + expect(item?.getAttribute('aria-live')).toBe('assertive'); + expect(item?.getAttribute('aria-atomic')).toBe('true'); + expect(container.querySelector('button.notification-item__action')?.textContent).toBe('Retry'); + }); +}); diff --git a/src/web-ui/src/shared/notification-system/components/NotificationItem.tsx b/src/web-ui/src/shared/notification-system/components/NotificationItem.tsx index 4f9df83103..149f2bd9cc 100644 --- a/src/web-ui/src/shared/notification-system/components/NotificationItem.tsx +++ b/src/web-ui/src/shared/notification-system/components/NotificationItem.tsx @@ -45,7 +45,12 @@ export const NotificationItem: React.FC = ({ notification }; return ( -
+
{getIcon()}