fix(session): surface cross-process writer conflicts

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 后可以重新调用，不表示自动等待、自动抢占或恢复当前调用。
 
 | 场景 | 行为 |
 |---|---|

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，用于约束最终模型
 响应，不是协议版本选择；如未来支持，应复用该语义并独立设计，不能借此重定义事件对象。

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.

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,15 +681,14 @@ 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: {}",
                         session_id
                     );
                     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());

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::<RuntimeError>(),
+            Some(RuntimeError::Port(port_error))
+                if port_error.kind == PortErrorKind::SessionInUse
+        ) || matches!(
+            cause.downcast_ref::<RuntimeIpcClientError>(),
+            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");
+    }
 }

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<String>,
     #[serde(skip_serializing_if = "Option::is_none")]
     turn_id: Option<String>,
@@ -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<super::verification::VerifyOutcome>,
@@ -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<String>,
@@ -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);
             }

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!(