diff --git a/docs/architecture/agent-runtime-deployment-design.md b/docs/architecture/agent-runtime-deployment-design.md
index d7fb2c3eef..2c2c47ed4e 100644
--- a/docs/architecture/agent-runtime-deployment-design.md
+++ b/docs/architecture/agent-runtime-deployment-design.md
@@ -39,8 +39,9 @@ flowchart TB
| Embedded TUI/Headless CLI/Peer Host | Session、Turn、Permission 和事件订阅统一通过同一个 Rust Runtime SDK（当前 preview）；CLI crate 只保留第一方 adapter 和各形态自己的展示/断流策略 |
| ACP/SDK Host | 使用同一个 Runtime 事件入口的 session-scoped 订阅；各自协议和进程生命周期保持独立 |
| Runtime ownership | Desktop、CLI、ACP、SDK Host 和现有 Server agent bootstrap 共用 Core owner；Embedded 取得共享锁，Shared TUI 取得独占锁，同一 workspace 上两种 deployment 互斥 |
+| Session 写入 | BitFun Runtime 的持久化 Session 由 `SessionManager` 管理；同一存储位置中的同一 Session 同时只允许一个本机进程写入，list/view 等只读操作不受影响 |
| 当前 HTTP Server | 只提供 health/info/WebSocket 外壳，未装配 Agent Runtime，因此不取得 workspace ownership；`bootstrap.rs` 仅保持 agent-enabled composition 的一致边界，不由当前入口启动 |
-| Shared local IPC | 未发布的本机协议已有 discovery、实例锁、严格握手、Session 控制租约、有界事件流和 cleanup；唯一 consumer 是第一方交互式 TUI adapter |
+| Shared local IPC | 未发布的本机协议已有 discovery、实例锁、严格握手、Session 控制权、有界事件流和 cleanup；唯一 consumer 是第一方交互式 TUI adapter |
| Shared TUI | `bitfun --shared` / `bitfun chat --shared` 可列出、创建、恢复 Session，读取 transcript，提交/取消 Turn，处理 Permission 和 UserInput；默认仍是 Embedded |
| Shared GUI/Headless/ACP/SDK Host/Remote | 未交付，也不会由 `--shared` 隐式启用；Replay、Observer、Controller transfer、Session delete/fork 同样不在当前协议中 |
@@ -106,7 +107,7 @@ ownership 分成“产品决策”和“文件锁原语”两层；入口不再
```mermaid
flowchart TB
Entrypoints["Desktop · CLI · ACP · SDK Host · Server bootstrap"]
- Entrypoints --> Core["CoreRuntimeOwnership
deployment · product identity · process leases"]
+ Entrypoints --> Core["CoreRuntimeOwnership
deployment · product identity · process-held lock"]
Core --> Primitive["services-core::runtime_ownership
canonical key · RAII file lock"]
Primitive --> E["Embedded · shared lock"]
Primitive --> S["Shared · exclusive lock"]
@@ -119,21 +120,58 @@ flowchart TD
Read -->|"no · attach/mutate/turn"| Remote{"structured remote facts?"}
Remote -->|"yes"| RemoteHost["由目标 execution host 负责"]
Remote -->|"no"| Gate["Coordinator → CoreRuntimeOwnership"]
- Gate --> Lease["按 canonical workspace 保留进程期 lease"]
+ Gate --> Lock["按 canonical workspace 持有文件锁"]
```
| 场景 | 行为 | 原因 |
|---|---|---|
| 多个 Embedded 进程访问同一 workspace | 共享锁允许并存 | 保持单实例、CI 和隔离测试的既有成本模型 |
| Shared 与任一 Embedded 访问同一 workspace | 后启动者返回稳定错误码和启动建议 | 防止同一 workspace 同时存在两种 Runtime deployment |
-| Desktop 打开多个 workspace | 首次 attach/write 时逐个取得并保留 lease | 不把窗口数、Session 数等同于 Runtime 进程数 |
+| Desktop 打开多个 workspace | 首次 attach/write 时逐个取得并持有文件锁 | 不把窗口数、Session 数等同于 Runtime 进程数 |
| 只读 list/view | 不加锁 | ownership 只管理 Runtime deployment，不扩大成读取权限 |
| 已解析且带有效 `connection_id` 的 remote workspace | 本机不加锁 | 与 Session storage 的远端判据一致；`host` 提示本身不能绕过本地锁 |
| 当前只读 HTTP Server | 不创建 Core owner | 没有 Agent Runtime 就没有 ownership 可声明 |
-`CoreRuntimeOwnership` 只选择 deployment、产品 identity 并保留进程期 lease；`services-core` 只负责 canonical key 和跨进程锁。二者都不选择 workspace、不启动 Runtime，也不替代 Session 单写、数据库事务、文件冲突控制或安全沙箱。
+`CoreRuntimeOwnership` 只选择 deployment、产品 identity 并在进程存活期间持有锁；`services-core` 只负责 canonical key 和跨进程锁。二者都不选择 workspace、不启动 Runtime，也不替代 Session 单写、数据库事务、文件冲突控制或安全沙箱。
-### 4.2 私有本机 IPC
+### 4.2 Session 单写
+
+workspace 可以被多个 Embedded 进程同时打开，但持久化 Session 不能被多个进程同时写入。保护粒度是“实际 Session 存储位置 + Session ID”，不是窗口、TUI 实例或 workspace。
+
+```mermaid
+flowchart LR
+ subgraph W["同一 workspace"]
+ A["Session A"]
+ B["Session B"]
+ end
+
+ GUI["GUI 进程"] -->|"写入"| A
+ TUI["TUI 进程"] -->|"写入"| B
+ CLI["另一个 CLI 进程"] -.->|"写入 A：session_in_use"| A
+ View["任意入口的 list / view"] -.->|"只读"| A
+ 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 单写范围内。
+
+| 场景 | 行为 |
+|---|---|
+| 同一进程重复 restore 同一 Session | 返回已加载的 Session，不重复取得或释放写入权 |
+| 另一个进程打开同一存储位置中的同一 Session | 立即返回 `session_in_use`；不等待、不自动抢占 |
+| 多个进程打开同一 workspace 中的不同 Session | 允许，各 Session 独立写入 |
+| 多个进程更新同一 Session 列表索引 | 按存储位置串行更新共享索引，不影响不同 Session 文件并行写入 |
+| `.`、`..`、符号链接或 Windows 路径大小写指向同一存储位置 | 视为同一个 Session 存储位置 |
+| 相同 Session ID 位于不同存储位置 | 文件锁相互独立；同一 `SessionManager` 仍按 Session ID 保持唯一绑定，不能同时加载 |
+| Session 存储路径无法解析或错误地指向文件系统根目录 | 在发布内存状态前返回错误，不创建可写 Session |
+| create/restore 在发布到内存前失败、取消或超时 | 临时文件锁随操作释放；后续进程可以重试 |
+| save、cleanup 或 unload 失败 | 已加载 Session 继续持有写入权，避免另一个进程接手不完整状态 |
+| unload 或 delete 成功 | 释放写入权 |
+| 进程崩溃或被强制结束 | 操作系统释放文件锁；残留锁文件本身不代表 Session 仍在使用 |
+| Remote workspace | 在实际 Session 存储所在机器执行同一检查；控制端不得用本机路径替代 |
+
+该机制不增加后台进程、轮询、连接或常驻线程，也不改变 Shared TUI 的连接控制规则。临时 Session 不写入磁盘，因此不参与此检查。
+
+### 4.3 私有本机 IPC
```mermaid
sequenceDiagram
@@ -147,7 +185,7 @@ sequenceDiagram
alt valid
S-->>C: initialized(health + interactive_tui)
C->>S: create or restore Session
- S-->>C: controller lease + Session facts
+ S-->>C: Session control + Session facts
C->>S: submit/cancel Turn or answer Permission/UserInput
S-->>C: Session-filtered authoritative events
else invalid
@@ -177,7 +215,7 @@ sequenceDiagram
- 未认证连接也计入有界 connection budget，单个客户端不能无限制造 server task；
- 未知字段、未知 operation、错误身份和不兼容版本 fail closed；
- 一个连接最多控制一个 Session、同时最多提交一个活动 Turn；一个 Session 同时只有一个 controller。create/restore 在完整结果通过大小检查后才原子切换控制权，失败时保留原 Session。活动 Turn 期间不能切换 Session。
-- Submit 使用调用方已有的 `turn_id` 标识不确定结果；若提交超时，返回 `outcome_unknown`、关闭连接并按该 ID 取消。断连取消只有得到确认后才释放 Session 租约；无法确认时租约保持隔离，直到 Runtime 进程退出。
+- Submit 使用调用方已有的 `turn_id` 标识不确定结果；若提交超时，返回 `outcome_unknown`、关闭连接并按该 ID 取消。断连取消只有得到确认后才释放 Session 控制权；无法确认时继续隔离该 Session，直到 Runtime 进程退出。
- Agent 事件流 lag/closed 后 fail closed；Permission lag 先从 Runtime 权威 pending 集合重建，重建失败或流关闭时取消当前 Turn 并退出。路由到父 Session 的嵌套 Permission 与 AskUserQuestion 复用现有 TUI 交互，不新增第二套 UI 状态。
- Windows Shared Runtime 在初始化前把自身放入 kill-on-close Job；Unix 仅在应用内优雅退出路径中通过受管子进程组回收后代。Runtime 被 `SIGTERM`、`SIGKILL` 或崩溃直接终止后的 Unix 后代回收不在当前保证内。两者都只负责生命周期，不是安全沙箱。
- 最后一个连接离开后等待 30 秒再退出；新连接会取消 idle 退出。退出只删除自己发布的 discovery；Unix 下继任 owner 会在持有实例锁后清理同一 identity 的陈旧 socket。
@@ -223,9 +261,9 @@ flowchart TB
| canonical workspace + product | 防止 Embedded 与 Shared 同时拥有同一工作区 Runtime |
| workspace + product + release channel + user + protocol | 定位兼容的本机 Shared instance |
| stable local endpoint + bearer token + owner id | endpoint 定位同一 instance；随机 token 认证本轮 server；owner id 防止旧实例误删新 discovery |
-| Session identity | 未来 Runtime 内的持久化和写入隔离；不由 IPC foundation 定义 |
+| 实际 Session 存储位置 + Session ID | 限制持久化 Session 的跨进程并发写入；不由 IPC 协议定义 |
-当前 Shared TUI 只有 controller，没有 observer 或 detached Query：一个 Client 关闭不会删除 Session；它会取消仍拥有的活动 Turn，只有取消得到确认才释放 Session 控制租约，否则该租约隔离到 Runtime 退出。最后一个 Client 关闭后，Runtime 进入 30 秒空闲期；期间重连可继续使用，超时后 Runtime 正常关闭。若未来增加后台任务、observer 或 Remote 引用，必须先扩展 Runtime-aware drain，不能把这些引用塞进当前简单连接计数。
+当前 Shared TUI 只有 controller，没有 observer 或 detached Query：一个 Client 关闭不会删除 Session；它会取消仍拥有的活动 Turn，只有取消得到确认才释放 Session 控制权，否则继续隔离该 Session，直到 Runtime 退出。最后一个 Client 关闭后，Runtime 进入 30 秒空闲期；期间重连可继续使用，超时后 Runtime 正常关闭。若未来增加后台任务、observer 或 Remote 引用，必须先扩展 Runtime-aware drain，不能把这些引用塞进当前简单连接计数。
对普通单实例用户，未显式启用 Shared deployment 时不增加后台进程、连接、发现扫描或常驻内存。
diff --git a/src/apps/cli/src/shared_runtime.rs b/src/apps/cli/src/shared_runtime.rs
index fba5fcbf99..4c3ed98c11 100644
--- a/src/apps/cli/src/shared_runtime.rs
+++ b/src/apps/cli/src/shared_runtime.rs
@@ -2,7 +2,8 @@ use anyhow::{anyhow, Context, Result};
use async_trait::async_trait;
use bitfun_agent_runtime::sdk::{
AgentRuntime, AgentSessionRestoreRequest, AgentUserAnswersRequest, DialogSubmitOutcome,
- PermissionRequest, PermissionRequestEvent, RuntimeError, SessionTranscriptRequest,
+ PermissionRequest, PermissionRequestEvent, PortErrorKind, RuntimeError,
+ SessionTranscriptRequest,
};
use bitfun_agent_runtime_ipc::{
DiscoveryStore, RuntimeInstanceIdentity, RuntimeIpcClient, RuntimeIpcError,
@@ -914,8 +915,14 @@ fn runtime_error_message(error: RuntimeError) -> anyhow::Error {
}
fn runtime_ipc_error(error: RuntimeError) -> RuntimeIpcError {
+ let code = match &error {
+ RuntimeError::Port(port_error) if port_error.kind == PortErrorKind::SessionInUse => {
+ RuntimeIpcErrorCode::SessionInUse
+ }
+ _ => RuntimeIpcErrorCode::Unavailable,
+ };
RuntimeIpcError {
- code: RuntimeIpcErrorCode::Unavailable,
+ code,
message: error.into_message(),
}
}
@@ -925,12 +932,13 @@ mod tests {
use super::{
await_permission_route, connect_existing, index_user_question, invalidate_event_stream,
permission_event_session, permission_targets_session, project_subagent_link_route,
- project_user_question_route, publish_event, route_agent_event, subscribe_session_events,
- SessionEventSenders, EVENT_BUFFER,
+ project_user_question_route, publish_event, route_agent_event, runtime_ipc_error,
+ subscribe_session_events, SessionEventSenders, EVENT_BUFFER,
};
use bitfun_agent_runtime::sdk::{
PermissionDelegationContext, PermissionReplySource, PermissionRequest,
- PermissionRequestEvent, PermissionRequestSource, PermissionRequestSourceKind,
+ PermissionRequestEvent, PermissionRequestSource, PermissionRequestSourceKind, PortError,
+ PortErrorKind, RuntimeError,
};
use bitfun_events::{AgenticEvent, ToolEventData, ToolEventIdentity};
use std::collections::HashMap;
@@ -938,6 +946,19 @@ mod tests {
use std::time::Duration;
use tokio::sync::{watch, Notify};
+ #[test]
+ fn session_writer_conflict_reuses_the_existing_ipc_error() {
+ let error = runtime_ipc_error(RuntimeError::Port(PortError::new(
+ PortErrorKind::SessionInUse,
+ "Session is already open for writing: session-1",
+ )));
+
+ assert_eq!(
+ error.code,
+ bitfun_agent_runtime_ipc::RuntimeIpcErrorCode::SessionInUse
+ );
+ }
+
#[tokio::test]
async fn existing_runtime_connection_errors_are_not_hidden_as_absence() {
let root = tempfile::tempdir().unwrap();
diff --git a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs
index 30a0c11701..2b7d086270 100644
--- a/src/crates/assembly/core/src/agentic/coordination/coordinator.rs
+++ b/src/crates/assembly/core/src/agentic/coordination/coordinator.rs
@@ -8930,6 +8930,10 @@ fn runtime_port_error_from_bitfun(error: BitFunError) -> bitfun_runtime_ports::P
(bitfun_runtime_ports::PortErrorKind::Cancelled, message)
}
BitFunError::Timeout(message) => (bitfun_runtime_ports::PortErrorKind::Timeout, message),
+ BitFunError::SessionInUse { session_id } => (
+ bitfun_runtime_ports::PortErrorKind::SessionInUse,
+ format!("Session is already open for writing: {session_id}"),
+ ),
BitFunError::NotImplemented(message) => {
(bitfun_runtime_ports::PortErrorKind::NotAvailable, message)
}
@@ -11778,7 +11782,7 @@ mod tests {
.session_storage_dir();
assert!(!session_manager
.persistence_manager()
- .session_storage_exists(&storage_path, &transient.session_id)
+ .session_storage_exists(&workspace_path, &transient.session_id)
.expect("persistence probe should succeed"));
let transient_child = session_manager
diff --git a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs
index 2ca87e7591..3adc9c5f79 100644
--- a/src/crates/assembly/core/src/agentic/coordination/scheduler.rs
+++ b/src/crates/assembly/core/src/agentic/coordination/scheduler.rs
@@ -134,6 +134,10 @@ impl SchedulerSubmitError {
Self::Core(BitFunError::Timeout(message)) => {
PortError::new(PortErrorKind::Timeout, message)
}
+ Self::Core(BitFunError::SessionInUse { session_id }) => PortError::new(
+ PortErrorKind::SessionInUse,
+ format!("Session is already open for writing: {session_id}"),
+ ),
Self::Core(BitFunError::NotImplemented(message)) => {
PortError::new(PortErrorKind::NotAvailable, message)
}
@@ -2390,6 +2394,16 @@ mod tests {
use bitfun_runtime_ports::{AgentDialogPrependedReminder, AgentInputAttachment, PortErrorKind};
use tokio::sync::RwLock as TokioRwLock;
+ #[test]
+ fn scheduler_preserves_session_writer_conflicts() {
+ let error = SchedulerSubmitError::Core(BitFunError::SessionInUse {
+ session_id: "session-1".to_string(),
+ })
+ .into_port_error();
+
+ assert_eq!(error.kind, PortErrorKind::SessionInUse);
+ }
+
fn test_scheduler() -> (
Arc,
Arc,
diff --git a/src/crates/assembly/core/src/agentic/persistence/manager.rs b/src/crates/assembly/core/src/agentic/persistence/manager.rs
index 1073da085e..2d79c6cb73 100644
--- a/src/crates/assembly/core/src/agentic/persistence/manager.rs
+++ b/src/crates/assembly/core/src/agentic/persistence/manager.rs
@@ -36,7 +36,7 @@ use bitfun_services_core::{
build_session_metadata as build_persisted_session_metadata, empty_session_metadata_page,
refresh_session_metadata_from_turns, try_refresh_session_metadata_for_saved_turn,
SessionMemoryMode, SessionMetadataBuildFacts, SessionMetadataStore,
- SessionMetadataStoreError, SessionStorageLayout,
+ SessionMetadataStoreError, SessionStorageLayout, SessionWriteLock, SessionWriteLockError,
},
};
use futures::{stream, StreamExt};
@@ -66,6 +66,39 @@ static SESSION_PERSISTENCE_LOCKS: OnceLock
static SESSION_BRANCH_ALLOCATION_LOCKS: OnceLock>>>> =
OnceLock::new();
+struct PendingSessionDirectory {
+ path: PathBuf,
+ committed: bool,
+}
+
+impl PendingSessionDirectory {
+ fn new(path: PathBuf) -> Self {
+ Self {
+ path,
+ committed: false,
+ }
+ }
+
+ fn commit(mut self) {
+ self.committed = true;
+ }
+}
+
+impl Drop for PendingSessionDirectory {
+ fn drop(&mut self) {
+ if self.committed || !self.path.exists() {
+ return;
+ }
+ if let Err(error) = std::fs::remove_dir_all(&self.path) {
+ warn!(
+ "Failed to remove an unfinished Session directory: path={}, error={}",
+ self.path.display(),
+ error
+ );
+ }
+ }
+}
+
async fn memory_pollution_guard_enabled() -> bool {
match get_global_config_service().await {
Ok(service) => {
@@ -373,6 +406,40 @@ impl PersistenceManager {
self.path_manager.project_sessions_dir(workspace_path)
}
+ /// Hold this across a multi-step Session write that is not already owned by
+ /// a loaded Session runtime.
+ pub(crate) fn lock_session_writes(
+ &self,
+ workspace_path: &Path,
+ session_id: &str,
+ ) -> BitFunResult {
+ let sessions_dir = self.project_sessions_dir(workspace_path);
+ SessionWriteLock::try_acquire(&sessions_dir, session_id)
+ .map_err(|error| Self::session_write_lock_error(session_id, error))
+ }
+
+ pub(super) fn lock_session_write_operation(
+ &self,
+ workspace_path: &Path,
+ session_id: &str,
+ ) -> BitFunResult {
+ let sessions_dir = self.project_sessions_dir(workspace_path);
+ SessionWriteLock::try_acquire_for_operation(&sessions_dir, session_id)
+ .map_err(|error| Self::session_write_lock_error(session_id, error))
+ }
+
+ fn session_write_lock_error(session_id: &str, error: SessionWriteLockError) -> BitFunError {
+ match error {
+ SessionWriteLockError::InUse => BitFunError::SessionInUse {
+ session_id: session_id.to_string(),
+ },
+ other => BitFunError::Session(format!(
+ "Failed to protect Session writes: session_id={session_id}, code={}, error={other}",
+ other.code()
+ )),
+ }
+ }
+
pub(crate) fn is_resolved_sessions_dir(&self, path: &Path) -> bool {
CoreSessionStorePort::resolved_sessions_dir_kind(self.path_manager.as_ref(), path).is_some()
}
@@ -1002,6 +1069,8 @@ impl PersistenceManager {
workspace_path: &Path,
metadata: &SessionMetadata,
) -> BitFunResult<()> {
+ let _session_write =
+ self.lock_session_write_operation(workspace_path, &metadata.session_id)?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, &metadata.session_id)
.await;
@@ -1040,6 +1109,8 @@ impl PersistenceManager {
metadata: &SessionMetadata,
) -> BitFunResult {
Self::validate_session_id(&metadata.session_id)?;
+ let _session_write =
+ self.lock_session_write_operation(workspace_path, &metadata.session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, &metadata.session_id)
@@ -1086,6 +1157,7 @@ impl PersistenceManager {
update: impl FnOnce(&mut SessionMetadata) -> BitFunResult<()>,
) -> BitFunResult {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
@@ -1120,6 +1192,7 @@ impl PersistenceManager {
mode: SessionMemoryMode,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
.await;
@@ -1141,6 +1214,7 @@ impl PersistenceManager {
session_id: &str,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
.await;
@@ -1235,6 +1309,7 @@ impl PersistenceManager {
cache: &SessionPromptCache,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
self.ensure_session_dir(workspace_path, session_id).await?;
@@ -1254,6 +1329,7 @@ impl PersistenceManager {
session_id: &str,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
match fs::remove_file(self.prompt_cache_path(workspace_path, session_id)).await {
Ok(()) => Ok(()),
Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
@@ -1285,6 +1361,7 @@ impl PersistenceManager {
anchors: &[TokenAnchor],
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
self.ensure_session_dir(workspace_path, session_id).await?;
@@ -1305,6 +1382,7 @@ impl PersistenceManager {
session_id: &str,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
match fs::remove_file(self.token_anchors_path(workspace_path, session_id)).await {
Ok(()) => Ok(()),
Err(error) if error.kind() == ErrorKind::NotFound => Ok(()),
@@ -1325,6 +1403,7 @@ impl PersistenceManager {
messages: &[Message],
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
self.ensure_snapshots_dir(workspace_path, session_id)
.await?;
@@ -1445,6 +1524,7 @@ impl PersistenceManager {
snapshot: &TurnSkillAgentSnapshot,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
self.ensure_snapshots_dir(workspace_path, session_id)
.await?;
@@ -1483,6 +1563,7 @@ impl PersistenceManager {
turn_index: usize,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let dir = self.snapshots_dir(workspace_path, session_id);
if !dir.exists() {
return Ok(());
@@ -1524,6 +1605,7 @@ impl PersistenceManager {
snapshot: &TurnSkillAgentSnapshot,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
self.ensure_snapshots_dir(workspace_path, session_id)
.await?;
@@ -1560,6 +1642,7 @@ impl PersistenceManager {
turn_index: usize,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let dir = self.snapshots_dir(workspace_path, session_id);
if !dir.exists() {
return Ok(());
@@ -1611,6 +1694,8 @@ impl PersistenceManager {
session: &Session,
) -> BitFunResult<()> {
Self::validate_session_id(&session.session_id)?;
+ let _session_write =
+ self.lock_session_write_operation(workspace_path, &session.session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
let sessions_dir = self.project_sessions_dir(workspace_path);
@@ -1644,6 +1729,10 @@ impl PersistenceManager {
)));
}
}
+ // Dropping an interrupted future removes the directory while the
+ // cross-process writer is still held. A stale index entry is repaired
+ // by the existing index validation on the next read.
+ let pending_session_dir = PendingSessionDirectory::new(session_dir);
if let Err(error) = self
.save_session_files_locked(workspace_path, session)
@@ -1664,15 +1753,19 @@ impl PersistenceManager {
cleanup_error: cleanup_error.to_string(),
});
}
+ pending_session_dir.commit();
return Err(error);
}
+ pending_session_dir.commit();
Ok(())
}
/// Save session
pub async fn save_session(&self, workspace_path: &Path, session: &Session) -> BitFunResult<()> {
Self::validate_session_id(&session.session_id)?;
+ let _session_write =
+ self.lock_session_write_operation(workspace_path, &session.session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, &session.session_id)
@@ -2041,6 +2134,7 @@ impl PersistenceManager {
state: &SessionState,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
self.ensure_runtime_for_write(workspace_path).await?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
@@ -2074,6 +2168,7 @@ impl PersistenceManager {
session_id: &str,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
.await;
@@ -2123,6 +2218,7 @@ impl PersistenceManager {
turn: &DialogTurnData,
) -> BitFunResult<()> {
Self::validate_session_id(&turn.session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, &turn.session_id)?;
let save_started_at = Instant::now();
self.ensure_runtime_for_write(workspace_path).await?;
let persistence_lock = self
@@ -2464,6 +2560,7 @@ impl PersistenceManager {
turn_index: usize,
) -> BitFunResult<()> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
.await;
@@ -2541,6 +2638,7 @@ impl PersistenceManager {
trigger: &str,
) -> BitFunResult> {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let all_turns = self.load_session_turns(workspace_path, session_id).await?;
let selected_indices = all_turns
.iter()
@@ -2690,6 +2788,7 @@ impl PersistenceManager {
start_turn_index: usize,
) -> BitFunResult {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let dir = self.compression_transcripts_dir(workspace_path, session_id);
if !dir.exists() {
return Ok(0);
@@ -2735,6 +2834,8 @@ impl PersistenceManager {
) -> BitFunResult {
Self::validate_session_id(source_session_id)?;
Self::validate_session_id(target_session_id)?;
+ let _session_write =
+ self.lock_session_write_operation(workspace_path, target_session_id)?;
let source_dir = self.compression_transcripts_dir(workspace_path, source_session_id);
if !source_dir.exists() {
return Ok(0);
@@ -2789,6 +2890,7 @@ impl PersistenceManager {
options: &SessionTranscriptExportOptions,
) -> BitFunResult {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
if self
.load_session_metadata(workspace_path, session_id)
.await?
@@ -2907,6 +3009,8 @@ impl PersistenceManager {
Self::validate_session_id(source_session_id)?;
Self::validate_session_id(reference_session_id)?;
Self::validate_session_id(reference_artifact_stem)?;
+ let _session_write =
+ self.lock_session_write_operation(source_workspace_path, source_session_id)?;
if self
.load_session_metadata(reference_workspace_path, reference_session_id)
@@ -2975,6 +3079,7 @@ impl PersistenceManager {
turn_index: usize,
) -> BitFunResult {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
.await;
@@ -3028,6 +3133,7 @@ impl PersistenceManager {
turn_index: usize,
) -> BitFunResult {
Self::validate_session_id(session_id)?;
+ let _session_write = self.lock_session_write_operation(workspace_path, session_id)?;
let persistence_lock = self
.get_session_persistence_lock(workspace_path, session_id)
.await;
@@ -3087,8 +3193,8 @@ impl PersistenceManager {
#[cfg(test)]
mod tests {
use super::{
- context_snapshot_payload_stats, current_unix_secs, PersistenceManager,
- StoredDialogTurnFile, SESSION_REFERENCE_TRANSCRIPT_CHAR_LIMIT,
+ context_snapshot_payload_stats, current_unix_secs, PendingSessionDirectory,
+ PersistenceManager, StoredDialogTurnFile, SESSION_REFERENCE_TRANSCRIPT_CHAR_LIMIT,
};
use crate::agentic::core::{Message, Session, SessionConfig, SessionKind, ToolResult};
use crate::agentic::memories::db::{MemoryDatabase, MemoryRow, MEMORY_PHASE2_GLOBAL_JOB_KEY};
@@ -3137,6 +3243,28 @@ mod tests {
}
}
+ #[test]
+ fn unfinished_session_directory_is_removed_when_creation_is_cancelled() {
+ let workspace = TestWorkspace::new();
+ let session_dir = workspace.path().join("cancelled-session");
+ std::fs::create_dir(&session_dir).expect("claimed session directory");
+
+ drop(PendingSessionDirectory::new(session_dir.clone()));
+
+ assert!(!session_dir.exists());
+ }
+
+ #[test]
+ fn completed_session_directory_is_kept() {
+ let workspace = TestWorkspace::new();
+ let session_dir = workspace.path().join("completed-session");
+ std::fs::create_dir(&session_dir).expect("claimed session directory");
+
+ PendingSessionDirectory::new(session_dir.clone()).commit();
+
+ assert!(session_dir.exists());
+ }
+
#[tokio::test]
async fn unsafe_session_ids_are_rejected_before_turn_path_resolution() {
let workspace = TestWorkspace::new();
diff --git a/src/crates/assembly/core/src/agentic/persistence/session_branch.rs b/src/crates/assembly/core/src/agentic/persistence/session_branch.rs
index bf7b224ced..52490c387e 100644
--- a/src/crates/assembly/core/src/agentic/persistence/session_branch.rs
+++ b/src/crates/assembly/core/src/agentic/persistence/session_branch.rs
@@ -79,6 +79,8 @@ impl PersistenceManager {
target_session.snapshot_session_id = None;
target_session.compression_state = source_session.compression_state.clone();
let target_session_id = target_session.session_id.clone();
+ let _target_session_write =
+ self.lock_session_write_operation(workspace_path, &target_session_id)?;
self.save_session(workspace_path, &target_session).await?;
diff --git a/src/crates/assembly/core/src/agentic/session/session_manager.rs b/src/crates/assembly/core/src/agentic/session/session_manager.rs
index 4f16d27f29..1a0fe306b3 100644
--- a/src/crates/assembly/core/src/agentic/session/session_manager.rs
+++ b/src/crates/assembly/core/src/agentic/session/session_manager.rs
@@ -49,7 +49,7 @@ use bitfun_services_core::session::{
apply_session_lineage, collect_hidden_subagent_cascade as collect_hidden_subagent_cascade_ids,
merge_session_custom_metadata as merge_session_custom_metadata_value,
set_deep_review_run_manifest, set_review_target_evidence, set_session_relationship,
- SessionStorageLayout,
+ SessionStorageLayout, SessionWriteLock,
};
use dashmap::{mapref::entry::Entry, DashMap};
use log::{debug, error, info, warn};
@@ -217,6 +217,10 @@ pub struct SessionManager {
/// concurrent operation has already made active.
session_mutation_locks: KeyedAsyncLock,
+ /// Cross-process writers for durable Sessions currently loaded by this manager.
+ /// The Session lifecycle remains the only owner of acquisition and release.
+ session_write_locks: Arc>,
+
/// Sub-components
context_store: Arc,
prompt_cache_store: Arc,
@@ -314,11 +318,36 @@ impl SessionManager {
self.active_session_permits.remove(session_id);
}
+ fn try_acquire_session_write_lock(
+ &self,
+ session_storage_path: &Path,
+ session_id: &str,
+ ) -> BitFunResult {
+ self.persistence_manager
+ .lock_session_writes(session_storage_path, session_id)
+ }
+
+ fn commit_session_write_lock(&self, session_id: &str, write_lock: SessionWriteLock) {
+ match self.session_write_locks.entry(session_id.to_string()) {
+ Entry::Vacant(entry) => {
+ entry.insert(write_lock);
+ }
+ Entry::Occupied(_) => {
+ debug_assert!(false, "Session write lock already existed");
+ }
+ }
+ }
+
+ fn release_session_write_lock(&self, session_id: &str) {
+ self.session_write_locks.remove(session_id);
+ }
+
#[cfg(test)]
pub(crate) fn evict_loaded_session_for_test(&self, session_id: &str) {
self.sessions.remove(session_id);
self.transient_session_ids.remove(session_id);
self.release_active_session_reservation(session_id);
+ self.release_session_write_lock(session_id);
}
#[cfg(test)]
@@ -1752,6 +1781,7 @@ impl SessionManager {
active_session_permits: Arc::new(DashMap::new()),
session_storage_path_index: Arc::new(DashMap::new()),
session_mutation_locks: KeyedAsyncLock::default(),
+ session_write_locks: Arc::new(DashMap::new()),
context_store,
prompt_cache_store: Arc::new(SessionPromptCacheStore::new()),
token_anchor_store: Arc::new(TokenAnchorStore::new()),
@@ -1959,6 +1989,7 @@ impl SessionManager {
let active_session_permits = self.active_session_permits.clone();
let session_storage_path_index = self.session_storage_path_index.clone();
let session_mutation_locks = self.session_mutation_locks.clone();
+ let session_write_locks = self.session_write_locks.clone();
let context_store = self.context_store.clone();
let prompt_cache_store = self.prompt_cache_store.clone();
let token_anchor_store = self.token_anchor_store.clone();
@@ -1990,6 +2021,7 @@ impl SessionManager {
active_session_permits,
session_storage_path_index,
session_mutation_locks,
+ session_write_locks,
context_store,
prompt_cache_store,
token_anchor_store,
@@ -2163,8 +2195,21 @@ impl SessionManager {
};
session.created_by = created_by;
session.kind = kind;
+ let persist = self.config.enable_persistence
+ && !transient
+ && Self::should_persist_session_kind(session.kind);
let session_id = session.session_id.clone();
let _mutation_guard = self.lock_session_mutation(&session_id).await;
+ if self.sessions.contains_key(&session_id) {
+ return Err(BitFunError::Validation(format!(
+ "Session ID already exists: {session_id}"
+ )));
+ }
+ let session_write_lock = if persist {
+ Some(self.try_acquire_session_write_lock(&session_storage_path, &session_id)?)
+ } else {
+ None
+ };
// Claim both the runtime session ID and its workspace storage identity before
// exposing the session. Persistent sessions must never reuse an on-disk ID:
@@ -2186,20 +2231,30 @@ impl SessionManager {
let active_session_permit = self.reserve_active_session()?;
let storage_claim =
self.claim_session_storage_path(&session_id, &session_storage_path, true)?;
- if transient {
- self.transient_session_ids.insert(session_id.clone(), ());
+
+ // Persist before publishing runtime state. Cancellation or timeout while
+ // this await is in progress cannot leave a writable in-memory Session.
+ if persist {
+ if let Err(error) = self
+ .persistence_manager
+ .create_session_if_absent(&session_storage_path, &session)
+ .await
+ {
+ self.release_failed_session_storage_path_claim(
+ &session_id,
+ &session_storage_path,
+ storage_claim,
+ );
+ return Err(error);
+ }
}
- // 1. Add to memory
+ // Publication is synchronous after all fallible persistence work.
match self.sessions.entry(session_id.clone()) {
Entry::Vacant(entry) => {
entry.insert(session.clone());
}
- Entry::Occupied(entry) => {
- drop(entry);
- if transient {
- self.transient_session_ids.remove(&session_id);
- }
+ Entry::Occupied(_) => {
self.release_failed_session_storage_path_claim(
&session_id,
&session_storage_path,
@@ -2210,42 +2265,19 @@ impl SessionManager {
)));
}
}
- // 2. Initialize the in-memory context cache.
+ if transient {
+ self.transient_session_ids.insert(session_id.clone(), ());
+ }
self.context_store.create_session(&session_id);
self.token_anchor_store.create_session(&session_id);
self.turn_skill_agent_snapshot_store
.create_session(&session_id);
self.file_read_state_store.create_session(&session_id);
-
- // 3. Persist to local path (handles remote workspaces correctly)
- // Use the local `session` directly -- no need to re-fetch from DashMap,
- // which would hold a Ref guard across the async save_session call.
- if self.config.enable_persistence && self.should_persist_session(&session) {
- if let Err(error) = self
- .persistence_manager
- .create_session_if_absent(&session_storage_path, &session)
- .await
- {
- self.sessions.remove(&session_id);
- self.context_store.delete_session(&session_id);
- self.token_anchor_store.delete_session(&session_id);
- self.turn_skill_agent_snapshot_store
- .delete_session(&session_id);
- self.file_read_state_store.delete_session(&session_id);
- self.evidence_ledger.delete_session(&session_id);
- if transient {
- self.transient_session_ids.remove(&session_id);
- }
- self.release_failed_session_storage_path_claim(
- &session_id,
- &session_storage_path,
- storage_claim,
- );
- return Err(error);
- }
- }
self.commit_session_storage_path_claim(&session_id, &session_storage_path, storage_claim);
self.commit_active_session_reservation(&session_id, active_session_permit);
+ if let Some(write_lock) = session_write_lock {
+ self.commit_session_write_lock(&session_id, write_lock);
+ }
info!("Session created: session_name={}", session.session_name);
@@ -3949,6 +3981,7 @@ impl SessionManager {
self.file_read_state_store.as_ref(),
self.evidence_ledger.as_ref(),
);
+ self.release_session_write_lock(session_id);
Ok(true)
}
@@ -4023,6 +4056,14 @@ impl SessionManager {
session_id: &str,
) -> BitFunResult<()> {
let delete_started_at = Instant::now();
+ let _temporary_write_lock = if self.config.enable_persistence
+ && !self.is_transient_session(session_id)
+ && !self.session_write_locks.contains_key(session_id)
+ {
+ Some(self.try_acquire_session_write_lock(session_storage_path, session_id)?)
+ } else {
+ None
+ };
debug!(
"Session deletion started: session_id={}, cleanup_workspace_path={}, session_storage_path={}, persistence_enabled={}",
session_id,
@@ -4072,6 +4113,7 @@ impl SessionManager {
elapsed_ms_u64(memory_stage_started_at)
);
self.session_storage_path_index.remove(session_id);
+ self.release_session_write_lock(session_id);
info!(
"Session deletion completed: session_id={}, cleanup_workspace_path={}, session_storage_path={}, duration_ms={}",
@@ -4630,6 +4672,11 @@ impl SessionManager {
return Ok((session, turns));
}
+ let session_write_lock = if self.config.enable_persistence {
+ Some(self.try_acquire_session_write_lock(session_storage_path, session_id)?)
+ } else {
+ None
+ };
let claimed = self.claim_session_storage_path(session_id, session_storage_path, true)?;
let result = self
.restore_session_with_turns_from_claimed_storage_path_internal(
@@ -4644,6 +4691,8 @@ impl SessionManager {
session_storage_path,
claimed,
);
+ } else if let Some(write_lock) = session_write_lock {
+ self.commit_session_write_lock(session_id, write_lock);
}
result
}
@@ -4908,6 +4957,22 @@ impl SessionManager {
.await?;
}
+ // Finish async notifications before publishing runtime state. If restore is
+ // cancelled or times out before publication, the temporary write lock drops
+ // together with this future and no writable in-memory Session remains.
+ if let Some(previous_model_id) = auto_migrated_model_id {
+ if let Some(coordinator) = crate::agentic::coordination::get_global_coordinator() {
+ coordinator
+ .emit_session_model_auto_migrated(
+ session_id,
+ &previous_model_id,
+ "auto",
+ "model_unavailable_on_restore",
+ )
+ .await;
+ }
+ }
+
// 3. Publish the recovered runtime context only after migrations are durable.
if session_already_in_memory {
clear_session_runtime_stores(
@@ -4956,18 +5021,6 @@ impl SessionManager {
}
self.bind_session_storage_path_committed(session_id, session_storage_path.to_path_buf());
- if let Some(previous_model_id) = auto_migrated_model_id {
- if let Some(coordinator) = crate::agentic::coordination::get_global_coordinator() {
- coordinator
- .emit_session_model_auto_migrated(
- session_id,
- &previous_model_id,
- "auto",
- "model_unavailable_on_restore",
- )
- .await;
- }
- }
if let Some(state) = restored_edit_constraint_state {
self.edit_constraints_store
.insert(session_id.to_string(), state);
@@ -6747,6 +6800,7 @@ impl SessionManager {
let persistence = self.persistence_manager.clone();
let enable_persistence = self.config.enable_persistence;
let session_mutation_locks = self.session_mutation_locks.clone();
+ let session_write_locks = self.session_write_locks.clone();
let context_store = self.context_store.clone();
let prompt_cache_store = self.prompt_cache_store.clone();
let token_anchor_store = self.token_anchor_store.clone();
@@ -6788,6 +6842,7 @@ impl SessionManager {
continue;
};
+ let mut can_remove = true;
if enable_persistence
&& Self::should_persist_session_with_transient_ids(
&session,
@@ -6809,11 +6864,25 @@ impl SessionManager {
)
.is_some()
{
- let _ = persistence.save_session(&workspace_path, &session).await;
+ if let Err(error) =
+ persistence.save_session(&workspace_path, &session).await
+ {
+ error!(
+ "Failed to save Session before idle eviction: session_id={}, error={}",
+ candidate.session_id, error
+ );
+ can_remove = false;
+ }
}
+ } else {
+ can_remove = false;
}
}
+ if !can_remove {
+ continue;
+ }
+
let removal_now = SystemTime::now();
if sessions
.remove_if(&candidate.session_id, |_, session| {
@@ -6827,6 +6896,7 @@ impl SessionManager {
.is_some()
{
active_session_permits.remove(&candidate.session_id);
+ session_write_locks.remove(&candidate.session_id);
clear_session_runtime_stores(
&candidate.session_id,
context_store.as_ref(),
@@ -7236,6 +7306,312 @@ mod tests {
assert_ne!(first.session_id, second.session_id);
}
+ #[tokio::test]
+ async fn a_persisted_session_has_one_writer_across_managers() {
+ let workspace = TestWorkspace::new();
+ let path_manager = workspace.path_manager();
+ let first = test_manager(Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("first persistence manager"),
+ ));
+ let second = test_manager(Arc::new(
+ PersistenceManager::new(path_manager).expect("second persistence manager"),
+ ));
+ let session = first
+ .create_session(
+ "Single writer".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect("first writer should create the session");
+
+ let duplicate = first
+ .create_session_with_id(
+ Some(session.session_id.clone()),
+ "Duplicate".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect_err("the current manager must reject an already-loaded Session ID");
+ assert!(matches!(
+ duplicate,
+ crate::util::errors::BitFunError::Validation(ref message)
+ if message.contains("already exists")
+ ));
+
+ let error = second
+ .restore_session(workspace.path(), &session.session_id)
+ .await
+ .expect_err("second writer must fail immediately");
+ assert!(matches!(
+ error,
+ crate::util::errors::BitFunError::SessionInUse { ref session_id }
+ if session_id == &session.session_id
+ ));
+
+ let (view, _) = second
+ .restore_session_view(workspace.path(), &session.session_id)
+ .await
+ .expect("read-only view must remain available");
+ assert_eq!(view.session_id, session.session_id);
+
+ assert!(first
+ .unload_session_from_memory(&session.session_id)
+ .await
+ .expect("first writer should unload"));
+ second
+ .restore_session(workspace.path(), &session.session_id)
+ .await
+ .expect("writer should transfer after successful unload");
+ }
+
+ #[tokio::test]
+ async fn different_sessions_in_the_same_workspace_can_have_different_writers() {
+ let workspace = TestWorkspace::new();
+ let path_manager = workspace.path_manager();
+ let persistence = Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("fixture persistence manager"),
+ );
+ let config = SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ };
+ let first_session = Session::new_with_id(
+ "first-workspace-session".to_string(),
+ "First".to_string(),
+ "agentic".to_string(),
+ config.clone(),
+ );
+ let second_session = Session::new_with_id(
+ "second-workspace-session".to_string(),
+ "Second".to_string(),
+ "agentic".to_string(),
+ config,
+ );
+ persistence
+ .save_session(workspace.path(), &first_session)
+ .await
+ .expect("first fixture");
+ persistence
+ .save_session(workspace.path(), &second_session)
+ .await
+ .expect("second fixture");
+ let first = test_manager(Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("first persistence manager"),
+ ));
+ let second = test_manager(Arc::new(
+ PersistenceManager::new(path_manager).expect("second persistence manager"),
+ ));
+
+ first
+ .restore_session(workspace.path(), &first_session.session_id)
+ .await
+ .expect("first Session writer");
+ second
+ .restore_session(workspace.path(), &second_session.session_id)
+ .await
+ .expect("second Session writer in the same workspace");
+ }
+
+ #[tokio::test]
+ async fn workspace_path_aliases_cannot_bypass_session_single_writer() {
+ let workspace = TestWorkspace::new();
+ let path_manager = workspace.path_manager();
+ let first = test_manager(Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("first persistence manager"),
+ ));
+ let second = test_manager(Arc::new(
+ PersistenceManager::new(path_manager).expect("second persistence manager"),
+ ));
+ let session = first
+ .create_session(
+ "Aliased workspace".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect("first writer");
+
+ let error = second
+ .restore_session(&workspace.path().join("."), &session.session_id)
+ .await
+ .expect_err("workspace alias must identify the same Session");
+ assert!(matches!(
+ error,
+ crate::util::errors::BitFunError::SessionInUse { .. }
+ ));
+ }
+
+ #[tokio::test]
+ async fn a_failed_restore_does_not_keep_the_session_write_lock() {
+ let workspace = TestWorkspace::new();
+ let path_manager = workspace.path_manager();
+ let first_persistence = Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("first persistence manager"),
+ );
+ let first = test_manager(first_persistence.clone());
+ let session_id = "restore-after-failure";
+
+ first
+ .restore_session(workspace.path(), session_id)
+ .await
+ .expect_err("missing Session restore should fail");
+
+ let fixture = Session::new_with_id(
+ session_id.to_string(),
+ "Recovered fixture".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ );
+ first_persistence
+ .save_session(workspace.path(), &fixture)
+ .await
+ .expect("persist recovered fixture");
+ let second = test_manager(Arc::new(
+ PersistenceManager::new(path_manager).expect("second persistence manager"),
+ ));
+ second
+ .restore_session(workspace.path(), session_id)
+ .await
+ .expect("failed restore must release the temporary writer lock");
+ }
+
+ #[tokio::test]
+ async fn a_failed_create_does_not_keep_the_session_write_lock() {
+ let workspace = TestWorkspace::new();
+ let path_manager = workspace.path_manager();
+ let first_persistence = Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("first persistence manager"),
+ );
+ let first = test_manager(first_persistence.clone());
+ let session_id = "create-after-failure";
+ first_persistence.fail_next_session_state_write_for_test(session_id);
+
+ first
+ .create_session_with_id(
+ Some(session_id.to_string()),
+ "Failed fixture".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect_err("injected persistence failure");
+
+ let second = test_manager(Arc::new(
+ PersistenceManager::new(path_manager).expect("second persistence manager"),
+ ));
+ second
+ .create_session_with_id(
+ Some(session_id.to_string()),
+ "Recovered fixture".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect("failed create must release the temporary writer lock");
+ }
+
+ #[tokio::test]
+ async fn failed_unload_save_keeps_the_session_write_lock() {
+ let workspace = TestWorkspace::new();
+ let path_manager = workspace.path_manager();
+ let first_persistence = Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("first persistence manager"),
+ );
+ let first = test_manager(first_persistence.clone());
+ let second = test_manager(Arc::new(
+ PersistenceManager::new(path_manager).expect("second persistence manager"),
+ ));
+ let session = first
+ .create_session(
+ "Unload failure".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect("first writer");
+ first_persistence.fail_next_session_state_write_for_test(&session.session_id);
+
+ first
+ .unload_session_from_memory(&session.session_id)
+ .await
+ .expect_err("injected unload save failure");
+ let error = second
+ .restore_session(workspace.path(), &session.session_id)
+ .await
+ .expect_err("failed unload save must retain writer ownership");
+ assert!(matches!(
+ error,
+ crate::util::errors::BitFunError::SessionInUse { .. }
+ ));
+ }
+
+ #[tokio::test]
+ async fn rejected_unload_keeps_the_session_write_lock() {
+ let workspace = TestWorkspace::new();
+ let path_manager = workspace.path_manager();
+ let first = test_manager(Arc::new(
+ PersistenceManager::new(path_manager.clone()).expect("first persistence manager"),
+ ));
+ let second = test_manager(Arc::new(
+ PersistenceManager::new(path_manager).expect("second persistence manager"),
+ ));
+ let session = first
+ .create_session(
+ "Processing".to_string(),
+ "agentic".to_string(),
+ SessionConfig {
+ workspace_path: Some(workspace.path().to_string_lossy().to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect("first writer");
+ first
+ .sessions
+ .get_mut(&session.session_id)
+ .expect("loaded Session")
+ .state = SessionState::Processing {
+ current_turn_id: "active-turn".to_string(),
+ phase: ProcessingPhase::Thinking,
+ };
+
+ first
+ .unload_session_from_memory(&session.session_id)
+ .await
+ .expect_err("processing Session must not unload");
+ let error = second
+ .restore_session(workspace.path(), &session.session_id)
+ .await
+ .expect_err("failed unload must retain writer ownership");
+ assert!(matches!(
+ error,
+ crate::util::errors::BitFunError::SessionInUse { .. }
+ ));
+ }
+
#[tokio::test]
async fn transient_session_cannot_bypass_owned_discard_through_unload() {
let workspace = TestWorkspace::new();
@@ -10239,10 +10615,7 @@ mod tests {
let persistence_manager = Arc::new(
PersistenceManager::new(workspace.path_manager()).expect("persistence manager"),
);
- let expected_storage_path = persistence_manager
- .path_manager()
- .project_sessions_dir(workspace.path());
- let manager = test_manager(persistence_manager);
+ let manager = test_manager(persistence_manager.clone());
let session = manager
.create_session(
"Cached session".to_string(),
@@ -10254,6 +10627,12 @@ mod tests {
)
.await
.expect("session should create");
+ let session_storage_dir = persistence_manager
+ .path_manager()
+ .project_sessions_dir(workspace.path());
+ assert!(session_storage_dir.exists());
+ let expected_storage_path =
+ SessionManager::normalize_session_storage_path(&session_storage_dir);
assert_eq!(
manager
@@ -10565,6 +10944,7 @@ mod tests {
)
.await;
+ manager.evict_loaded_session_for_test(&session.session_id);
let restored_manager = test_manager(persistence_manager);
restored_manager
.restore_session(workspace.path(), &session.session_id)
@@ -10629,6 +11009,7 @@ mod tests {
Some(baseline.clone())
);
+ manager.evict_loaded_session_for_test(&session.session_id);
let restored_manager = test_manager(persistence_manager);
restored_manager
.restore_session(workspace.path(), &session.session_id)
@@ -10843,6 +11224,7 @@ mod tests {
.and_then(|value| value.get(EDIT_CONSTRAINT_METADATA_KEY))
.is_some());
+ manager.evict_loaded_session_for_test(&session.session_id);
let restored_manager = test_manager(persistence_manager);
restored_manager
.restore_session(workspace.path(), &session.session_id)
@@ -10946,6 +11328,7 @@ mod tests {
Some(latest_baseline.clone())
);
+ manager.evict_loaded_session_for_test(&child.session_id);
let restored_manager = test_manager(persistence_manager);
restored_manager
.restore_session(workspace.path(), &child.session_id)
@@ -11007,6 +11390,7 @@ mod tests {
.invalidate_prompt_cache(&session.session_id, PromptCacheScope::All, "test")
.await;
+ manager.evict_loaded_session_for_test(&session.session_id);
let restored_manager = test_manager(persistence_manager.clone());
restored_manager
.restore_session(workspace.path(), &session.session_id)
@@ -11177,6 +11561,7 @@ mod tests {
Some("cached user context".to_string())
);
+ manager.evict_loaded_session_for_test(&session.session_id);
let restored_manager = test_manager_with_config(
persistence_manager.clone(),
SessionManagerConfig {
diff --git a/src/crates/assembly/core/src/agentic/tools/implementations/thread_goal_tools.rs b/src/crates/assembly/core/src/agentic/tools/implementations/thread_goal_tools.rs
index d592b407b1..4634f13503 100644
--- a/src/crates/assembly/core/src/agentic/tools/implementations/thread_goal_tools.rs
+++ b/src/crates/assembly/core/src/agentic/tools/implementations/thread_goal_tools.rs
@@ -58,6 +58,7 @@ fn thread_goal_port_error(port_error: PortError) -> BitFunError {
user_facing_thread_goal_error(BitFunError::NotImplemented(port_error.message))
}
PortErrorKind::PermissionDenied
+ | PortErrorKind::SessionInUse
| PortErrorKind::CleanupRequired
| PortErrorKind::Backend => {
user_facing_thread_goal_error(BitFunError::Tool(port_error.message))
diff --git a/src/crates/assembly/core/src/product_runtime.rs b/src/crates/assembly/core/src/product_runtime.rs
index d22586eb14..99bb6482a5 100644
--- a/src/crates/assembly/core/src/product_runtime.rs
+++ b/src/crates/assembly/core/src/product_runtime.rs
@@ -1026,6 +1026,7 @@ fn runtime_port_error(error: BitFunError) -> PortError {
BitFunError::NotFound(_) => PortErrorKind::NotFound,
BitFunError::Timeout(_) => PortErrorKind::Timeout,
BitFunError::Cancelled(_) => PortErrorKind::Cancelled,
+ BitFunError::SessionInUse { .. } => PortErrorKind::SessionInUse,
BitFunError::SessionCreateCleanupRequired { .. } => PortErrorKind::CleanupRequired,
_ => PortErrorKind::Backend,
};
@@ -1488,6 +1489,16 @@ mod tests {
assert!(error.message.contains("session-1"), "{error}");
}
+ #[test]
+ fn session_writer_conflict_remains_typed_across_the_runtime_port() {
+ let error = runtime_port_error(BitFunError::SessionInUse {
+ session_id: "session-1".to_string(),
+ });
+
+ assert_eq!(error.kind, PortErrorKind::SessionInUse);
+ assert!(error.message.contains("session-1"), "{error}");
+ }
+
#[test]
fn local_session_fork_uses_latest_persisted_turn_and_preserves_empty_error() {
let turns = [
diff --git a/src/crates/assembly/core/src/service_agent_runtime.rs b/src/crates/assembly/core/src/service_agent_runtime.rs
index 0b9312097b..095cf4dd4d 100644
--- a/src/crates/assembly/core/src/service_agent_runtime.rs
+++ b/src/crates/assembly/core/src/service_agent_runtime.rs
@@ -531,6 +531,9 @@ impl AgentSessionManagementPort for ScheduledSessionManagementPort {
crate::util::errors::BitFunError::Cancelled(_) => {
bitfun_runtime_ports::PortErrorKind::Cancelled
}
+ crate::util::errors::BitFunError::SessionInUse { .. } => {
+ bitfun_runtime_ports::PortErrorKind::SessionInUse
+ }
_ => bitfun_runtime_ports::PortErrorKind::Backend,
};
bitfun_runtime_ports::PortError::new(kind, error.to_string())
@@ -650,6 +653,9 @@ fn map_session_close_error(
crate::util::errors::BitFunError::Cancelled(_) => {
bitfun_runtime_ports::PortErrorKind::Cancelled
}
+ crate::util::errors::BitFunError::SessionInUse { .. } => {
+ bitfun_runtime_ports::PortErrorKind::SessionInUse
+ }
_ => bitfun_runtime_ports::PortErrorKind::Backend,
};
bitfun_runtime_ports::PortError::new(kind, error.to_string())
@@ -1888,6 +1894,19 @@ mod tests {
DialogTurnData, DialogTurnKind, ModelRoundData, TextItemData, ThinkingItemData,
ToolCallData, ToolItemData, TurnStatus, UserMessageData,
};
+ use crate::BitFunError;
+
+ #[test]
+ fn session_close_preserves_writer_conflicts() {
+ let error = map_session_close_error(BitFunError::SessionInUse {
+ session_id: "session-1".to_string(),
+ });
+
+ assert_eq!(
+ error.kind,
+ bitfun_runtime_ports::PortErrorKind::SessionInUse
+ );
+ }
#[test]
fn core_service_agent_runtime_owner_keeps_coordinator_port_contracts() {
diff --git a/src/crates/assembly/core/src/util/errors.rs b/src/crates/assembly/core/src/util/errors.rs
index 14fa351bbb..bb14b5ee96 100644
--- a/src/crates/assembly/core/src/util/errors.rs
+++ b/src/crates/assembly/core/src/util/errors.rs
@@ -36,6 +36,9 @@ pub enum BitFunError {
#[error("Session error: {0}")]
Session(String),
+ #[error("Session is already open for writing: {session_id}")]
+ SessionInUse { session_id: String },
+
#[error(
"Session creation persistence failed and rollback did not complete: session_id={session_id}, error={error}, cleanup_error={cleanup_error}"
)]
diff --git a/src/crates/contracts/runtime-ports/src/lib.rs b/src/crates/contracts/runtime-ports/src/lib.rs
index a2148dbdfc..68c4ed2242 100644
--- a/src/crates/contracts/runtime-ports/src/lib.rs
+++ b/src/crates/contracts/runtime-ports/src/lib.rs
@@ -74,6 +74,7 @@ pub enum PortErrorKind {
PermissionDenied,
Cancelled,
Timeout,
+ SessionInUse,
CleanupRequired,
Backend,
}
diff --git a/src/crates/interfaces/acp/src/runtime.rs b/src/crates/interfaces/acp/src/runtime.rs
index 4fd400e7a0..92b890a70f 100644
--- a/src/crates/interfaces/acp/src/runtime.rs
+++ b/src/crates/interfaces/acp/src/runtime.rs
@@ -146,6 +146,7 @@ impl BitfunAcpRuntime {
RuntimeError::Port(error) => match error.kind {
PortErrorKind::InvalidRequest => Error::invalid_params().data(error.message),
PortErrorKind::NotFound => Error::resource_not_found(None),
+ PortErrorKind::SessionInUse => Self::session_in_use_error(error.message),
_ => Self::internal_error(error.message),
},
other => Self::internal_error(other.into_message()),
@@ -165,10 +166,21 @@ impl BitfunAcpRuntime {
match error {
BitFunError::NotFound(_) => Error::resource_not_found(Some(session_id.to_string())),
BitFunError::Validation(message) => Error::invalid_params().data(message),
+ BitFunError::SessionInUse { session_id } => Self::session_in_use_error(format!(
+ "Session is already open for writing: {session_id}"
+ )),
other => Self::internal_error(other),
}
}
+ fn session_in_use_error(message: String) -> Error {
+ Error::invalid_params().data(serde_json::json!({
+ "state": "session_in_use",
+ "message": message,
+ "retryable": true
+ }))
+ }
+
pub(crate) async fn lock_active_session(
&self,
session_id: &str,
@@ -276,6 +288,7 @@ impl AcpRuntime for BitfunAcpRuntime {
mod tests {
use agent_client_protocol::schema::ErrorCode;
use bitfun_agent_runtime::sdk::{PortError, PortErrorKind, RuntimeError};
+ use bitfun_core::util::errors::BitFunError;
use dashmap::DashMap;
use super::{AcpSessionTransition, BitfunAcpRuntime};
@@ -302,6 +315,36 @@ mod tests {
assert_eq!(error.data, None);
}
+ #[test]
+ fn session_writer_conflict_is_actionable_at_the_protocol_boundary() {
+ let error = BitfunAcpRuntime::runtime_error(RuntimeError::Port(PortError::new(
+ PortErrorKind::SessionInUse,
+ "Session is already open for writing: session-1",
+ )));
+
+ assert_eq!(error.code, ErrorCode::InvalidParams);
+ assert_eq!(
+ error.data.as_ref().and_then(|data| data.get("state")),
+ Some(&serde_json::json!("session_in_use"))
+ );
+ }
+
+ #[test]
+ fn compatibility_restore_preserves_session_writer_conflicts() {
+ let error = BitfunAcpRuntime::session_core_error(
+ "session-1",
+ BitFunError::SessionInUse {
+ session_id: "session-1".to_string(),
+ },
+ );
+
+ assert_eq!(error.code, ErrorCode::InvalidParams);
+ assert_eq!(
+ error.data.as_ref().and_then(|data| data.get("state")),
+ Some(&serde_json::json!("session_in_use"))
+ );
+ }
+
#[test]
fn session_runtime_error_uses_the_requested_resource_id() {
let error = BitfunAcpRuntime::session_runtime_error(
diff --git a/src/crates/interfaces/sdk-host/src/host.rs b/src/crates/interfaces/sdk-host/src/host.rs
index 3df96d4154..829a9c585e 100644
--- a/src/crates/interfaces/sdk-host/src/host.rs
+++ b/src/crates/interfaces/sdk-host/src/host.rs
@@ -2153,6 +2153,9 @@ fn runtime_error_facts(error: &RuntimeError) -> (ErrorCode, bool, Option (ErrorCode::PermissionDenied, false, None),
PortErrorKind::Cancelled => (ErrorCode::Cancelled, false, None),
PortErrorKind::Timeout => (ErrorCode::Timeout, true, Some(RecoveryAction::Retry)),
+ PortErrorKind::SessionInUse => {
+ (ErrorCode::ActionRequired, true, Some(RecoveryAction::Retry))
+ }
PortErrorKind::CleanupRequired => (
ErrorCode::CleanupRequired,
false,
@@ -2188,6 +2191,7 @@ fn runtime_error_kind(error: &RuntimeError) -> &'static str {
PortErrorKind::PermissionDenied => "permission_denied",
PortErrorKind::Cancelled => "cancelled",
PortErrorKind::Timeout => "timeout",
+ PortErrorKind::SessionInUse => "session_in_use",
PortErrorKind::CleanupRequired => "cleanup_required",
PortErrorKind::Backend => "backend",
},
@@ -2206,3 +2210,24 @@ fn runtime_error_kind(error: &RuntimeError) -> &'static str {
RuntimeError::PermissionRequest(_) => "permission_request",
}
}
+
+#[cfg(test)]
+mod session_conflict_tests {
+ use super::{runtime_error_facts, runtime_error_kind};
+ use crate::protocol::{ErrorCode, RecoveryAction};
+ use bitfun_agent_runtime::sdk::{PortError, PortErrorKind, RuntimeError};
+
+ #[test]
+ fn session_writer_conflict_uses_existing_action_required_response() {
+ let error = RuntimeError::Port(PortError::new(
+ PortErrorKind::SessionInUse,
+ "Session is already open for writing: session-1",
+ ));
+
+ assert_eq!(
+ runtime_error_facts(&error),
+ (ErrorCode::ActionRequired, true, Some(RecoveryAction::Retry))
+ );
+ assert_eq!(runtime_error_kind(&error), "session_in_use");
+ }
+}
diff --git a/src/crates/services/services-core/Cargo.toml b/src/crates/services/services-core/Cargo.toml
index ceb4415820..6547ad51e6 100644
--- a/src/crates/services/services-core/Cargo.toml
+++ b/src/crates/services/services-core/Cargo.toml
@@ -75,5 +75,8 @@ required-features = ["lsp"]
name = "runtime_ownership_contracts"
required-features = ["runtime-ownership"]
+[[test]]
+name = "session_write_lock_contracts"
+
[lints]
workspace = true
diff --git a/src/crates/services/services-core/src/file_lock.rs b/src/crates/services/services-core/src/file_lock.rs
new file mode 100644
index 0000000000..e51a7f3548
--- /dev/null
+++ b/src/crates/services/services-core/src/file_lock.rs
@@ -0,0 +1,59 @@
+use fs2::FileExt;
+use std::fs::{File, OpenOptions};
+use std::path::Path;
+
+pub(crate) enum FileLockMode {
+ #[cfg(feature = "runtime-ownership")]
+ Shared,
+ Exclusive,
+}
+
+pub(crate) struct FileLock {
+ file: File,
+}
+
+#[derive(Debug)]
+pub(crate) enum FileLockError {
+ Open(std::io::Error),
+ Unavailable(std::io::Error),
+}
+
+impl FileLock {
+ pub(crate) fn acquire(path: &Path, mode: FileLockMode) -> Result {
+ let file = open_lock_file(path)?;
+ match mode {
+ #[cfg(feature = "runtime-ownership")]
+ FileLockMode::Shared => FileExt::lock_shared(&file),
+ FileLockMode::Exclusive => FileExt::lock_exclusive(&file),
+ }
+ .map_err(FileLockError::Unavailable)?;
+ Ok(Self { file })
+ }
+
+ pub(crate) fn try_acquire(path: &Path, mode: FileLockMode) -> Result {
+ let file = open_lock_file(path)?;
+ match mode {
+ #[cfg(feature = "runtime-ownership")]
+ FileLockMode::Shared => FileExt::try_lock_shared(&file),
+ FileLockMode::Exclusive => FileExt::try_lock_exclusive(&file),
+ }
+ .map_err(FileLockError::Unavailable)?;
+ Ok(Self { file })
+ }
+}
+
+fn open_lock_file(path: &Path) -> Result {
+ OpenOptions::new()
+ .create(true)
+ .truncate(false)
+ .read(true)
+ .write(true)
+ .open(path)
+ .map_err(FileLockError::Open)
+}
+
+impl Drop for FileLock {
+ fn drop(&mut self) {
+ let _ = FileExt::unlock(&self.file);
+ }
+}
diff --git a/src/crates/services/services-core/src/lib.rs b/src/crates/services/services-core/src/lib.rs
index 3853cd327c..01c6cdbcd5 100644
--- a/src/crates/services/services-core/src/lib.rs
+++ b/src/crates/services/services-core/src/lib.rs
@@ -5,6 +5,7 @@
pub mod diagnostics;
pub mod diff;
+mod file_lock;
pub mod filesystem;
pub mod json_store;
#[cfg(feature = "workspace-runtime")]
diff --git a/src/crates/services/services-core/src/runtime_ownership.rs b/src/crates/services/services-core/src/runtime_ownership.rs
index 25c37b4033..52bc3a229a 100644
--- a/src/crates/services/services-core/src/runtime_ownership.rs
+++ b/src/crates/services/services-core/src/runtime_ownership.rs
@@ -4,10 +4,9 @@
//! assembly may use the shared/exclusive lock to prevent an embedded runtime
//! and a future shared runtime from owning the same workspace simultaneously.
-use fs2::FileExt;
+use crate::file_lock::{FileLock, FileLockError, FileLockMode};
use sha2::{Digest, Sha256};
use std::fmt::{self, Write as _};
-use std::fs::{File, OpenOptions};
use std::path::{Path, PathBuf};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -65,7 +64,7 @@ impl RuntimeOwnershipKey {
pub struct WorkspaceRuntimeOwnership {
deployment: RuntimeDeployment,
- file: File,
+ _lock: FileLock,
}
impl WorkspaceRuntimeOwnership {
@@ -81,25 +80,24 @@ impl WorkspaceRuntimeOwnership {
}
})?;
let path = key.lock_path(ownership_root);
- let file = OpenOptions::new()
- .create(true)
- .truncate(false)
- .read(true)
- .write(true)
- .open(&path)
- .map_err(|source| RuntimeOwnershipError::OpenLockFile {
+ let mode = match deployment {
+ RuntimeDeployment::Embedded => FileLockMode::Shared,
+ RuntimeDeployment::Shared => FileLockMode::Exclusive,
+ };
+ let lock = FileLock::try_acquire(&path, mode).map_err(|error| match error {
+ FileLockError::Open(source) => RuntimeOwnershipError::OpenLockFile {
path: path.clone(),
source,
- })?;
-
- let lock_result = match deployment {
- RuntimeDeployment::Embedded => FileExt::try_lock_shared(&file),
- RuntimeDeployment::Shared => FileExt::try_lock_exclusive(&file),
- };
- lock_result
- .map_err(|source| RuntimeOwnershipError::OwnershipUnavailable { deployment, source })?;
+ },
+ FileLockError::Unavailable(source) => {
+ RuntimeOwnershipError::OwnershipUnavailable { deployment, source }
+ }
+ })?;
- Ok(Self { deployment, file })
+ Ok(Self {
+ deployment,
+ _lock: lock,
+ })
}
pub fn deployment(&self) -> RuntimeDeployment {
@@ -107,12 +105,6 @@ impl WorkspaceRuntimeOwnership {
}
}
-impl Drop for WorkspaceRuntimeOwnership {
- fn drop(&mut self) {
- let _ = FileExt::unlock(&self.file);
- }
-}
-
impl fmt::Debug for WorkspaceRuntimeOwnership {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
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 6b28033904..e19e8f1fe5 100644
--- a/src/crates/services/services-core/src/session/metadata_store.rs
+++ b/src/crates/services/services-core/src/session/metadata_store.rs
@@ -11,6 +11,7 @@ use super::metadata::{
use super::page::{build_session_metadata_page, empty_session_metadata_page};
use super::types::{SessionMetadata, StoredSessionIndexFile, StoredSessionMetadataFile};
use super::SessionMetadataPage;
+use crate::file_lock::{FileLock, FileLockError, FileLockMode};
use crate::json_store::{JsonFileStore, JsonFileStoreError};
use bitfun_core_types::validate_session_id;
use log::warn;
@@ -48,6 +49,12 @@ pub enum SessionMetadataStoreError {
#[source]
source: std::io::Error,
},
+ #[error("Failed to lock Session index {path}: {source}")]
+ LockSessionIndex {
+ path: PathBuf,
+ #[source]
+ source: std::io::Error,
+ },
#[error("Failed to delete session directory: {source}")]
DeleteSessionDir {
#[source]
@@ -115,6 +122,26 @@ impl SessionMetadataStore {
.clone()
}
+ async fn lock_index_file(&self) -> Result {
+ fs::create_dir_all(self.sessions_root())
+ .await
+ .map_err(|source| SessionMetadataStoreError::CreateSessionDir { source })?;
+ let lock_path = self.sessions_root().join(".index.lock");
+ let task_path = lock_path.clone();
+ tokio::task::spawn_blocking(move || FileLock::acquire(&task_path, FileLockMode::Exclusive))
+ .await
+ .map_err(|error| SessionMetadataStoreError::LockSessionIndex {
+ path: lock_path.clone(),
+ source: std::io::Error::other(error),
+ })?
+ .map_err(|error| SessionMetadataStoreError::LockSessionIndex {
+ path: lock_path,
+ source: match error {
+ FileLockError::Open(source) | FileLockError::Unavailable(source) => source,
+ },
+ })
+ }
+
async fn read_json_optional(
&self,
path: &Path,
@@ -269,6 +296,7 @@ impl SessionMetadataStore {
let lock = self.get_index_lock().await;
let _guard = lock.lock().await;
+ let _file_guard = self.lock_index_file().await?;
let index_path = self.index_path();
if let Some(index) = self
.read_json_optional::(&index_path)
@@ -315,6 +343,7 @@ impl SessionMetadataStore {
let limit = limit.max(1);
let lock = self.get_index_lock().await;
let _guard = lock.lock().await;
+ let _file_guard = self.lock_index_file().await?;
let index_path = self.index_path();
let indexed_sessions = if let Some(index) = self
.read_json_optional::(&index_path)
@@ -361,6 +390,7 @@ impl SessionMetadataStore {
pub async fn rebuild_index(&self) -> Result, SessionMetadataStoreError> {
let lock = self.get_index_lock().await;
let _guard = lock.lock().await;
+ let _file_guard = self.lock_index_file().await?;
self.rebuild_index_locked().await
}
@@ -376,6 +406,7 @@ impl SessionMetadataStore {
let lock = self.get_index_lock().await;
let _guard = lock.lock().await;
+ let _file_guard = self.lock_index_file().await?;
let metadata_file_created = !metadata_path.exists();
self.write_json_atomic(&metadata_path, &file).await?;
if !metadata.should_hide_from_user_lists() {
@@ -409,6 +440,7 @@ impl SessionMetadataStore {
validate_session_id(session_id).map_err(SessionMetadataStoreError::InvalidSessionId)?;
let lock = self.get_index_lock().await;
let _guard = lock.lock().await;
+ let _file_guard = self.lock_index_file().await?;
let dir = self.session_dir(session_id);
let metadata_file_removed = self.metadata_path(session_id).exists();
if dir.exists() {
@@ -466,6 +498,74 @@ mod tests {
use crate::session::{SessionStatus, StoredSessionIndexFile};
use tempfile::tempdir;
+ #[test]
+ fn index_lock_child_holds_the_cross_process_guard() {
+ if std::env::var_os("BITFUN_SESSION_INDEX_LOCK_CHILD").is_none() {
+ return;
+ }
+ let sessions_root =
+ PathBuf::from(std::env::var_os("BITFUN_SESSION_INDEX_ROOT").expect("index lock root"));
+ let ready_path = PathBuf::from(
+ std::env::var_os("BITFUN_SESSION_INDEX_READY").expect("index lock ready path"),
+ );
+ let release_path = PathBuf::from(
+ std::env::var_os("BITFUN_SESSION_INDEX_RELEASE").expect("index lock release path"),
+ );
+ std::fs::create_dir_all(&sessions_root).expect("sessions root");
+ let _guard = FileLock::acquire(&sessions_root.join(".index.lock"), FileLockMode::Exclusive)
+ .expect("child index lock");
+ std::fs::write(&ready_path, b"ready").expect("publish child readiness");
+ while !release_path.exists() {
+ std::thread::sleep(std::time::Duration::from_millis(10));
+ }
+ }
+
+ #[tokio::test]
+ async fn metadata_save_waits_for_a_cross_process_index_writer() {
+ use std::process::{Command, Stdio};
+ use std::time::{Duration, Instant};
+
+ let dir = tempdir().expect("tempdir");
+ let ready_path = dir.path().join("child-ready");
+ let release_path = dir.path().join("child-release");
+ let mut child = Command::new(std::env::current_exe().expect("test executable"))
+ .arg("--exact")
+ .arg("session::metadata_store::tests::index_lock_child_holds_the_cross_process_guard")
+ .arg("--nocapture")
+ .env("BITFUN_SESSION_INDEX_LOCK_CHILD", "1")
+ .env("BITFUN_SESSION_INDEX_ROOT", dir.path())
+ .env("BITFUN_SESSION_INDEX_READY", &ready_path)
+ .env("BITFUN_SESSION_INDEX_RELEASE", &release_path)
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn index lock child");
+ let deadline = Instant::now() + Duration::from_secs(5);
+ while !ready_path.exists() && Instant::now() < deadline {
+ std::thread::sleep(Duration::from_millis(10));
+ }
+ if !ready_path.exists() {
+ let _ = child.kill();
+ let _ = child.wait();
+ panic!("index lock child did not become ready");
+ }
+
+ let store = SessionMetadataStore::new(dir.path());
+ let mut save =
+ tokio::spawn(async move { store.save_metadata(&metadata("session-a", 10)).await });
+ let blocked = tokio::time::timeout(Duration::from_millis(50), &mut save)
+ .await
+ .is_err();
+
+ std::fs::write(&release_path, b"release").expect("release child index lock");
+ save.await.expect("save task").expect("metadata save");
+ assert!(child.wait().expect("index lock child").success());
+ assert!(
+ blocked,
+ "metadata save must wait while another process owns the index"
+ );
+ }
+
fn metadata(session_id: &str, last_active_at: u64) -> SessionMetadata {
let mut metadata = SessionMetadata::new(
session_id.to_string(),
diff --git a/src/crates/services/services-core/src/session/mod.rs b/src/crates/services/services-core/src/session/mod.rs
index a59fd413d9..bb9f4e3088 100644
--- a/src/crates/services/services-core/src/session/mod.rs
+++ b/src/crates/services/services-core/src/session/mod.rs
@@ -6,6 +6,7 @@ mod metadata_store;
mod migration;
pub mod page;
pub mod types;
+mod write_lock;
pub use bitfun_core_types::SessionKind;
pub use layout::SessionStorageLayout;
@@ -33,3 +34,4 @@ pub use migration::{
};
pub use page::{build_session_metadata_page, empty_session_metadata_page, SessionMetadataPage};
pub use types::*;
+pub use write_lock::{SessionWriteLock, SessionWriteLockError};
diff --git a/src/crates/services/services-core/src/session/write_lock.rs b/src/crates/services/services-core/src/session/write_lock.rs
new file mode 100644
index 0000000000..91f4fb3cde
--- /dev/null
+++ b/src/crates/services/services-core/src/session/write_lock.rs
@@ -0,0 +1,268 @@
+use crate::file_lock::{FileLock, FileLockError, FileLockMode};
+use sha2::{Digest, Sha256};
+use std::collections::HashMap;
+use std::fmt::Write as _;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex, OnceLock, Weak};
+
+/// Keeps one persisted Session writable by only one local process at a time.
+///
+/// The file may remain after this guard is dropped; ownership is represented
+/// by the OS lock, not by file existence.
+pub struct SessionWriteLock {
+ inner: Arc,
+}
+
+struct SessionWriteLockInner {
+ lock: Option,
+ lock_path: PathBuf,
+}
+
+impl SessionWriteLock {
+ pub fn try_acquire(
+ session_storage_root: &Path,
+ session_id: &str,
+ ) -> Result {
+ Self::acquire(session_storage_root, session_id, false)
+ }
+
+ /// Reuses a lock already held by this process for one persistence operation.
+ /// A new process-level writer still fails through [`Self::try_acquire`].
+ #[doc(hidden)]
+ pub fn try_acquire_for_operation(
+ session_storage_root: &Path,
+ session_id: &str,
+ ) -> Result {
+ Self::acquire(session_storage_root, session_id, true)
+ }
+
+ fn acquire(
+ session_storage_root: &Path,
+ session_id: &str,
+ reuse_process_writer: bool,
+ ) -> Result {
+ bitfun_core_types::validate_session_id(session_id)
+ .map_err(|_| SessionWriteLockError::InvalidSessionId)?;
+ std::fs::create_dir_all(session_storage_root).map_err(|source| {
+ SessionWriteLockError::CreateStorageDirectory {
+ path: session_storage_root.to_path_buf(),
+ source,
+ }
+ })?;
+ let canonical_storage_root =
+ std::fs::canonicalize(session_storage_root).map_err(|source| {
+ SessionWriteLockError::CanonicalizeStorage {
+ path: session_storage_root.to_path_buf(),
+ source,
+ }
+ })?;
+ let lock_root = session_lock_root(&canonical_storage_root)?;
+ std::fs::create_dir_all(&lock_root).map_err(|source| {
+ SessionWriteLockError::CreateLockDirectory {
+ path: lock_root.clone(),
+ source,
+ }
+ })?;
+ let canonical_lock_root = std::fs::canonicalize(&lock_root).map_err(|source| {
+ SessionWriteLockError::CreateLockDirectory {
+ path: lock_root.clone(),
+ source,
+ }
+ })?;
+ let lock_path = canonical_lock_root.join(format!(
+ "{}.lock",
+ lock_key(&canonical_storage_root, session_id)
+ ));
+ let mut process_locks = process_locks()
+ .lock()
+ .expect("Session write lock registry poisoned");
+ if let Some(existing) = process_locks.get(&lock_path).and_then(Weak::upgrade) {
+ if reuse_process_writer {
+ return Ok(Self { inner: existing });
+ }
+ return Err(SessionWriteLockError::InUse);
+ }
+ let lock =
+ FileLock::try_acquire(&lock_path, FileLockMode::Exclusive).map_err(
+ |error| match error {
+ FileLockError::Open(source) => SessionWriteLockError::OpenLockFile {
+ path: lock_path.clone(),
+ source,
+ },
+ FileLockError::Unavailable(source) if is_contention(&source) => {
+ SessionWriteLockError::InUse
+ }
+ FileLockError::Unavailable(source) => {
+ SessionWriteLockError::LockFailed { source }
+ }
+ },
+ )?;
+ let inner = Arc::new(SessionWriteLockInner {
+ lock: Some(lock),
+ lock_path: lock_path.clone(),
+ });
+ process_locks.insert(lock_path, Arc::downgrade(&inner));
+ Ok(Self { inner })
+ }
+}
+
+impl Clone for SessionWriteLock {
+ fn clone(&self) -> Self {
+ Self {
+ inner: self.inner.clone(),
+ }
+ }
+}
+
+impl Drop for SessionWriteLockInner {
+ fn drop(&mut self) {
+ if let Ok(mut process_locks) = process_locks().lock() {
+ // Keep the in-process registry authoritative until the OS lock is
+ // released so an immediate reacquire cannot observe a false gap.
+ drop(self.lock.take());
+ process_locks.remove(&self.lock_path);
+ }
+ }
+}
+
+impl std::fmt::Debug for SessionWriteLock {
+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ formatter
+ .debug_struct("SessionWriteLock")
+ .field("strong_count", &Arc::strong_count(&self.inner))
+ .finish_non_exhaustive()
+ }
+}
+
+fn process_locks() -> &'static Mutex>> {
+ static PROCESS_LOCKS: OnceLock>>> =
+ OnceLock::new();
+ PROCESS_LOCKS.get_or_init(|| Mutex::new(HashMap::new()))
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SessionWriteLockError {
+ #[error("session ID is invalid")]
+ InvalidSessionId,
+ #[error("failed to resolve Session storage path {path}")]
+ CanonicalizeStorage {
+ path: PathBuf,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("Session storage path cannot be a filesystem root: {path}")]
+ InvalidStorageRoot { path: PathBuf },
+ #[error("failed to create Session storage directory {path}")]
+ CreateStorageDirectory {
+ path: PathBuf,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("failed to create Session lock directory {path}")]
+ CreateLockDirectory {
+ path: PathBuf,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("failed to open Session lock file {path}")]
+ OpenLockFile {
+ path: PathBuf,
+ #[source]
+ source: std::io::Error,
+ },
+ #[error("session is already open for writing")]
+ InUse,
+ #[error("failed to lock Session for writing")]
+ LockFailed {
+ #[source]
+ source: std::io::Error,
+ },
+}
+
+impl SessionWriteLockError {
+ pub fn code(&self) -> &'static str {
+ match self {
+ Self::InvalidSessionId => "invalid_session_id",
+ Self::CanonicalizeStorage { .. } => "session_storage_resolve_failed",
+ Self::InvalidStorageRoot { .. } => "session_storage_root_invalid",
+ Self::CreateStorageDirectory { .. } => "session_storage_create_failed",
+ Self::CreateLockDirectory { .. } => "session_lock_root_create_failed",
+ Self::OpenLockFile { .. } => "session_lock_open_failed",
+ Self::InUse => "session_in_use",
+ Self::LockFailed { .. } => "session_lock_failed",
+ }
+ }
+}
+
+fn session_lock_root(canonical_storage_root: &Path) -> Result {
+ canonical_storage_root
+ .parent()
+ .map(|parent| parent.join(".session-write-locks"))
+ .ok_or_else(|| SessionWriteLockError::InvalidStorageRoot {
+ path: canonical_storage_root.to_path_buf(),
+ })
+}
+
+fn lock_key(canonical_storage_root: &Path, session_id: &str) -> String {
+ let mut hasher = Sha256::new();
+ hasher.update(b"bitfun-session-write-v1\0");
+ hash_path(&mut hasher, canonical_storage_root);
+ hasher.update(b"\0");
+ hasher.update(session_id.as_bytes());
+ let digest = hasher.finalize();
+ let mut encoded = String::with_capacity(digest.len() * 2);
+ for byte in digest {
+ write!(&mut encoded, "{byte:02x}").expect("writing to String cannot fail");
+ }
+ encoded
+}
+
+fn hash_path(hasher: &mut Sha256, path: &Path) {
+ #[cfg(unix)]
+ {
+ use std::os::unix::ffi::OsStrExt;
+ hasher.update(path.as_os_str().as_bytes());
+ }
+ #[cfg(windows)]
+ {
+ use std::os::windows::ffi::OsStrExt;
+ for unit in path.as_os_str().encode_wide() {
+ hasher.update(unit.to_le_bytes());
+ }
+ }
+}
+
+fn is_contention(error: &std::io::Error) -> bool {
+ if error.kind() == std::io::ErrorKind::WouldBlock {
+ return true;
+ }
+ #[cfg(windows)]
+ {
+ error.raw_os_error() == Some(33)
+ }
+ #[cfg(unix)]
+ {
+ matches!(error.raw_os_error(), Some(libc::EAGAIN))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::{session_lock_root, SessionWriteLockError};
+ use std::path::Path;
+
+ #[test]
+ fn filesystem_root_is_rejected_without_panicking() {
+ #[cfg(unix)]
+ let root = Path::new("/");
+ #[cfg(windows)]
+ let root = Path::new(r"C:\");
+
+ let error = session_lock_root(root).expect_err("filesystem root must be rejected");
+ assert!(matches!(
+ error,
+ SessionWriteLockError::InvalidStorageRoot { .. }
+ ));
+ assert_eq!(error.code(), "session_storage_root_invalid");
+ }
+}
diff --git a/src/crates/services/services-core/tests/session_write_lock_contracts.rs b/src/crates/services/services-core/tests/session_write_lock_contracts.rs
new file mode 100644
index 0000000000..3051c96fc9
--- /dev/null
+++ b/src/crates/services/services-core/tests/session_write_lock_contracts.rs
@@ -0,0 +1,194 @@
+use bitfun_services_core::session::{SessionWriteLock, SessionWriteLockError};
+use std::process::{Command, Stdio};
+use std::time::{Duration, Instant};
+use tempfile::tempdir;
+
+#[test]
+fn one_session_can_have_only_one_writer() {
+ let storage_root = tempdir().expect("session storage root");
+
+ let first =
+ SessionWriteLock::try_acquire(storage_root.path(), "shared-session").expect("first writer");
+ let error = SessionWriteLock::try_acquire(storage_root.path(), "shared-session")
+ .expect_err("second writer must fail immediately");
+
+ assert!(matches!(error, SessionWriteLockError::InUse));
+ assert_eq!(error.code(), "session_in_use");
+
+ drop(first);
+ SessionWriteLock::try_acquire(storage_root.path(), "shared-session")
+ .expect("writer after release");
+}
+
+#[test]
+fn different_sessions_in_one_workspace_are_independent() {
+ let storage_root = tempdir().expect("session storage root");
+
+ let _first = SessionWriteLock::try_acquire(storage_root.path(), "first-session")
+ .expect("first session writer");
+ let _second = SessionWriteLock::try_acquire(storage_root.path(), "second-session")
+ .expect("second session writer");
+}
+
+#[test]
+fn the_same_session_id_in_different_storage_roots_is_independent() {
+ let first_storage_root = tempdir().expect("first session storage root");
+ let second_storage_root = tempdir().expect("second session storage root");
+
+ let _first = SessionWriteLock::try_acquire(first_storage_root.path(), "shared-session")
+ .expect("first storage writer");
+ let _second = SessionWriteLock::try_acquire(second_storage_root.path(), "shared-session")
+ .expect("second storage writer");
+}
+
+#[test]
+fn storage_path_aliases_resolve_to_the_same_writer() {
+ let storage_root = tempdir().expect("session storage root");
+ let alias = storage_root.path().join(".");
+
+ let _first =
+ SessionWriteLock::try_acquire(storage_root.path(), "shared-session").expect("first writer");
+ let error = SessionWriteLock::try_acquire(&alias, "shared-session")
+ .expect_err("path alias must identify the same session");
+
+ assert_eq!(error.code(), "session_in_use");
+}
+
+#[test]
+fn a_stale_lock_file_does_not_block_a_new_writer() {
+ let project_root = tempdir().expect("project runtime root");
+ let storage_root = project_root.path().join("sessions");
+
+ let first =
+ SessionWriteLock::try_acquire(&storage_root, "shared-session").expect("first writer");
+ let lock_root = project_root.path().join(".session-write-locks");
+ assert_eq!(
+ std::fs::read_dir(&lock_root)
+ .expect("read lock root")
+ .count(),
+ 1
+ );
+ drop(first);
+
+ assert_eq!(
+ std::fs::read_dir(&lock_root)
+ .expect("read lock root")
+ .count(),
+ 1,
+ "the lock file may remain after the OS lock is released"
+ );
+ SessionWriteLock::try_acquire(&storage_root, "shared-session")
+ .expect("stale file must not imply ownership");
+}
+
+#[test]
+fn persistence_operations_reuse_the_process_writer_without_releasing_it() {
+ let storage_root = tempdir().expect("session storage root");
+ let writer = SessionWriteLock::try_acquire(storage_root.path(), "shared-session")
+ .expect("Session writer");
+
+ let operation =
+ SessionWriteLock::try_acquire_for_operation(storage_root.path(), "shared-session")
+ .expect("same-process persistence operation");
+ drop(operation);
+
+ assert!(matches!(
+ SessionWriteLock::try_acquire(storage_root.path(), "shared-session"),
+ Err(SessionWriteLockError::InUse)
+ ));
+ drop(writer);
+ SessionWriteLock::try_acquire(storage_root.path(), "shared-session")
+ .expect("writer after the process writer releases");
+}
+
+#[test]
+fn invalid_session_ids_fail_before_touching_the_lock_root() {
+ let project_root = tempdir().expect("project runtime root");
+ let storage_root = project_root.path().join("sessions");
+
+ let error = SessionWriteLock::try_acquire(&storage_root, "bad\nid").expect_err("invalid id");
+
+ assert_eq!(error.code(), "invalid_session_id");
+ assert_eq!(
+ std::fs::read_dir(project_root.path())
+ .expect("read project runtime root")
+ .count(),
+ 0
+ );
+}
+
+#[tokio::test]
+async fn a_timed_out_write_operation_releases_its_writer() {
+ let storage_root = tempdir().expect("session storage root");
+
+ let timed_out = tokio::time::timeout(Duration::from_millis(20), async {
+ let _writer = SessionWriteLock::try_acquire(storage_root.path(), "timed-out-session")
+ .expect("temporary writer");
+ std::future::pending::<()>().await;
+ })
+ .await;
+ assert!(timed_out.is_err());
+
+ SessionWriteLock::try_acquire(storage_root.path(), "timed-out-session")
+ .expect("writer after timeout cancellation");
+}
+
+#[test]
+fn abnormal_process_exit_releases_the_writer() {
+ let project_root = tempdir().expect("project runtime root");
+ let storage_root = project_root.path().join("sessions");
+ let ready_path = project_root.path().join("child-ready");
+ let mut child = Command::new(std::env::current_exe().expect("current test executable"))
+ .arg("--exact")
+ .arg("abnormal_exit_child_holds_writer")
+ .arg("--nocapture")
+ .env("BITFUN_SESSION_WRITE_LOCK_CHILD", "1")
+ .env("BITFUN_SESSION_WRITE_STORAGE_ROOT", &storage_root)
+ .env("BITFUN_SESSION_WRITE_READY_PATH", &ready_path)
+ .stdout(Stdio::null())
+ .stderr(Stdio::null())
+ .spawn()
+ .expect("spawn writer child");
+
+ let deadline = Instant::now() + Duration::from_secs(5);
+ while !ready_path.exists() && Instant::now() < deadline {
+ if let Some(status) = child.try_wait().expect("poll writer child") {
+ panic!("writer child exited before acquiring the lock: {status}");
+ }
+ std::thread::sleep(Duration::from_millis(10));
+ }
+ if !ready_path.exists() {
+ let _ = child.kill();
+ let _ = child.wait();
+ panic!("writer child did not become ready");
+ }
+ let was_blocked = matches!(
+ SessionWriteLock::try_acquire(&storage_root, "abnormal-exit-session"),
+ Err(SessionWriteLockError::InUse)
+ );
+
+ child.kill().expect("terminate writer child");
+ child.wait().expect("reap writer child");
+ assert!(was_blocked, "child process must own the Session writer");
+ SessionWriteLock::try_acquire(&storage_root, "abnormal-exit-session")
+ .expect("writer after abnormal process exit");
+}
+
+#[test]
+fn abnormal_exit_child_holds_writer() {
+ if std::env::var_os("BITFUN_SESSION_WRITE_LOCK_CHILD").is_none() {
+ return;
+ }
+ let storage_root = std::path::PathBuf::from(
+ std::env::var_os("BITFUN_SESSION_WRITE_STORAGE_ROOT").expect("child storage root"),
+ );
+ let ready_path = std::path::PathBuf::from(
+ std::env::var_os("BITFUN_SESSION_WRITE_READY_PATH").expect("child ready path"),
+ );
+ let _writer = SessionWriteLock::try_acquire(&storage_root, "abnormal-exit-session")
+ .expect("child writer");
+ std::fs::write(ready_path, b"ready").expect("publish child readiness");
+ loop {
+ std::thread::park();
+ }
+}