feat(R06): workspace session snapshot, atomic persistence, history-dir race fix

EPIC R06 (Team Hoa) - workspace/filesystem isolation, no cross-project
mutable state.

R06-T01 domain/workspaces/workspace_session.py
  WorkspaceSession - project_id/workspace_root/sandbox_dir/allowed_paths
  frozen snapshot + is_allowed(path), same "capture once at submit time"
  shape as R04's ConversationExecutionRequest.

R06-T02 infrastructure/persistence/json/{atomic_write,workspace_repository_impl,conversation_repository_impl}.py
  Real bug fixed: core/projects.py::save_project and core/history.py's
  save_conversation/rename_conversation/set_pinned did a plain
  path.write_text(json.dumps(...)) - two syscalls, no atomicity. A crash
  between them leaves a half-written file that load_project/load_conversation
  then silently treat as "missing". All four now write through
  atomic_write.write_json (temp file + os.replace). WorkspaceRepository/
  ConversationRepository are thin object-shaped facades over the same
  (now-atomic) functions, for future application-layer callers.
  NOTE: atomic_write.py is deliberately NOT named atomic_json_file.py -
  R02-T01 (Team Nam) claims that filename for the same purpose app-wide;
  see the checklist for the consolidation TODO.

R06-T03 infrastructure/filesystem/execution_workspace.py
  ExecutionWorkspace names the output_dir/scratch_dir split that already
  exists (core/chat_agent.py's flat workspace_root/.scratch) - does not
  move anything.

R06-T04 ui/chat_panel.py
  The actual race: ChatPanel._persist_session (saves a BACKGROUND turn's
  conversation) resolved its save directory via a live
  self.ctx.config.history_dir() read at save time. ui/workspace_tab.py::
  _load_current mutates that same config field on every project switch, so
  a turn still running when the user switched projects got saved into the
  NEW project's history folder. Fixed by adding "home_history_dir" to the
  per-turn ctx dict (same "home_*" snapshot convention already used for
  session id/messages/title), captured at submit time. Verified with a real
  offscreen-Qt test, not just a unit double:
  tests/integration/test_history_dir_race.py.

R06-T05 application/workspaces/file_workspace_service.py
  FileWorkspaceService - the File Explorer / AI Editor entry point for the
  same safe read/write/edit operations the agent tool loop has, by calling
  core/tools.py::execute_tool directly (same dispatch, same ToolContext
  containment, same audit log) rather than reimplementing any of it.

New tests: tests/unit/test_workspace_session.py,
test_atomic_write_and_repositories.py, test_execution_workspace.py,
test_file_workspace_service.py, tests/integration/test_history_dir_race.py
(29 new tests, incl. 2 real offscreen-Qt integration tests).

Suite: 283 passed, 4 pre-existing failures unrelated to R05/R06 (see
checklist). check_imports: PASS. All new files < 400 LOC.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

application/workspaces/__init__.py

@@ -0,0 +1,5 @@
+"""Workspace file operations for non-agent-loop callers (EPIC R06)."""
+
+from .file_workspace_service import FileWorkspaceService
+
+__all__ = ["FileWorkspaceService"]

application/workspaces/file_workspace_service.py

@@ -0,0 +1,81 @@
+"""FileWorkspaceService - the safe file operations File Explorer and the AI
+File Editor need, outside the agent tool loop (R06-T05).
+
+``ui/folder_tab.py`` (File Explorer) and the AI File Editor dialog need the
+exact same guarantees the agent's tools already have — path containment
+inside the workspace, precise context-anchored edits, syntax warnings on a
+bad Python write — but today that logic only exists wired to a model's tool
+call (``core/tools.py::execute_tool``). A UI action that isn't a tool call
+(browsing the tree, applying an AI-suggested diff from a review dialog) has
+no equivalent entry point of its own.
+
+This service IS that entry point. It reuses ``core/tools.py::execute_tool``
+verbatim - same dispatch table, same ``ToolContext`` containment check, same
+audit-log entry, same Python-syntax warning on write/edit - rather than
+re-implementing any of it, so a fix to one path fixes both. It only adds the
+:class:`~domain.workspaces.workspace_session.WorkspaceSession` seam: which
+workspace root a call is scoped to is decided by the session, not by
+whichever folder a widget happens to have open.
+"""
+from __future__ import annotations
+
+from typing import Any, Dict
+
+
+class FileWorkspaceService:
+    """File operations scoped to one :class:`WorkspaceSession`.
+
+    Read-only by name (``list_tree``/``read_preview``) vs. writing
+    (``write_file``/``apply_edit``) mirrors the same READ/WRITE split
+    ``domain/tools/tool_registry.py`` uses for the agent's own tools - a
+    caller that only wants to browse never accidentally has write access.
+    """
+
+    def __init__(self, session) -> None:  # WorkspaceSession - see module docstring
+        self._session = session
+
+    def list_tree(self, rel: str = ".") -> Dict[str, Any]:
+        """Entries at ``rel`` (default: the workspace root)."""
+        return self._execute("list_dir", {"path": rel})
+
+    def read_preview(self, rel: str) -> Dict[str, Any]:
+        """A text file's content (truncated by
+        ``infrastructure/filesystem/file_tools.py::MAX_READ_BYTES``, same as
+        the agent's ``read_file`` tool)."""
+        return self._execute("read_file", {"path": rel})
+
+    def write_file(self, rel: str, content: str) -> Dict[str, Any]:
+        """Create or fully overwrite ``rel``."""
+        return self._execute("write_file", {"path": rel, "content": content})
+
+    def apply_edit(self, rel: str, old_string: str, new_string: str,
+                   replace_all: bool = False) -> Dict[str, Any]:
+        """Replace an exact snippet in an existing file - the same
+        context-anchored algorithm the agent's ``edit_file`` tool uses, so an
+        AI-suggested diff applies with the same precision and the same
+        "old_string not found / ambiguous" failure messages either path
+        would give the caller."""
+        return self._execute("edit_file", {
+            "path": rel, "old_string": old_string, "new_string": new_string,
+            "replace_all": replace_all,
+        })
+
+    # -- internals --------------------------------------------------------- #
+    def _tool_context(self):
+        """A ``ToolContext`` scoped to this session's workspace root.
+        ``flatten_writes=False`` (unlike Cowork's agent context) - File
+        Explorer must preserve whatever subfolder structure the user is
+        actually browsing, not collapse every write into the root."""
+        from cowork_local.infrastructure.filesystem.tool_context import ToolContext
+
+        return ToolContext(self._session.workspace_root, flatten_writes=False)
+
+    def _execute(self, name: str, args: Dict[str, Any]) -> Dict[str, Any]:
+        """Dispatch through ``core/tools.py::execute_tool`` - see the module
+        docstring for why this delegates instead of reimplementing."""
+        from cowork_local.core.tools import execute_tool
+
+        return execute_tool(self._tool_context(), name, args)
+
+
+__all__ = ["FileWorkspaceService"]

core/history.py

@@ -66,7 +66,9 @@ def save_conversation(
         "outputs": list(outputs or []),
         "messages": messages,
     }
-    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    # R06-T02: atomic write - see infrastructure/persistence/json/atomic_write.py.
+    from ..infrastructure.persistence.json.atomic_write import write_json
+    write_json(path, payload)
     return path
 
 
@@ -78,15 +80,19 @@ def delete_conversation(path) -> None:
 
 
 def rename_conversation(path, new_title: str) -> None:
+    from ..infrastructure.persistence.json.atomic_write import write_json
+
     data = load_conversation(path)
     data["title"] = new_title
-    Path(path).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+    write_json(Path(path), data)
 
 
 def set_pinned(path, pinned: bool) -> None:
+    from ..infrastructure.persistence.json.atomic_write import write_json
+
     data = load_conversation(path)
     data["pinned"] = bool(pinned)
-    Path(path).write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
+    write_json(Path(path), data)
 
 
 def load_conversation(path: Path) -> Dict[str, Any]:

core/projects.py

@@ -116,10 +116,12 @@ def new_project(name: str, description: str = "", instructions: str = "",
 
 def save_project(project: Project, directory: Path = None) -> Path:
     directory = directory or PROJECTS_DIR
-    directory.mkdir(parents=True, exist_ok=True)
     path = directory / f"{project.project_id}.json"
-    path.write_text(json.dumps(asdict(project), ensure_ascii=False, indent=2),
-                    encoding="utf-8")
+    # R06-T02: atomic write — a crash/kill between truncate and write used to
+    # leave a half-written project.json that load_project() then silently
+    # treats as "missing" (see infrastructure/persistence/json/atomic_write.py).
+    from ..infrastructure.persistence.json.atomic_write import write_json
+    write_json(path, asdict(project))
     return path
 
 

docs/refactor/Refactoring_Checklist.md

@@ -57,22 +57,29 @@
 
 ---
 
-## 📊 TIẾN ĐỘ THỰC TẾ — TEAM HOA (cập nhật `2026-08-21 22:19`)
+## 📊 TIẾN ĐỘ THỰC TẾ — TEAM HOA (cập nhật `2026-08-21 22:57`)
 
 > [!NOTE]
-> ### ✅ ĐÃ HOÀN TẤT: 5/5 task của **R05** — branch `feature/teamhoa/r05-r06` (tạo từ `origin/feature/deltateam/refactor-plan`, có sẵn nền R01/R03/R04)
+> ### ✅ ĐÃ HOÀN TẤT: 10/10 task của **R05 + R06** — branch `feature/teamhoa/r05-r06` (tạo từ `origin/feature/deltateam/refactor-plan`, có sẵn nền R01/R03/R04)
 >
 > | EPIC | Task | Trạng thái |
 > | :--- | :--- | :--- |
 > | **R05** Tool, MCP & Connector Policy | T01 → T05 | ✅ 5/5 |
-> | **R06** Workspace, Filesystem & History Isolation | T01 → T05 | ⬜ chưa bắt đầu |
+> | **R06** Workspace, Filesystem & History Isolation | T01 → T05 | ✅ 5/5 |
 >
 > **Kiểm chứng (chạy thật):**
-> * `pytest tests/` ➔ **254 pass / 4 fail** (+12 test mới cho R05: `tests/unit/test_tool_registry_and_policy.py`, `test_code_agent_tool_policy.py`, `test_cowork_extra_tool_policy.py`, `test_mcp_source_manager.py`)
-> * 4 fail là **lỗi có sẵn từ trước R05**, không liên quan tool/MCP: 2 trong `test_config_security.py` (EPIC R02, đã ghi nhận bởi Team Duy) + 2 trong `test_routing_wiring.py` (môi trường máy này có Ollama/llama3.1 thật + config routing cục bộ khác "fresh install" — không phải do R05).
+> * `pytest tests/` ➔ **283 pass / 4 fail** (+41 test mới cho R05+R06, gồm 2 test Qt offscreen thật trong `tests/integration/test_history_dir_race.py`)
+> * 4 fail là **lỗi có sẵn từ trước**, không liên quan R05/R06: 2 trong `test_config_security.py` (EPIC R02, đã ghi nhận bởi Team Duy) + 2 trong `test_routing_wiring.py` (môi trường máy này có Ollama/llama3.1 thật + config routing cục bộ khác "fresh install").
 > * `python scripts/check_imports.py` ➔ **PASS** (0 Qt import trong `domain/`, `application/`)
 > * Mọi file mới **< 400 dòng** (lớn nhất: `domain/tools/tool_registry.py` 125 dòng). `core/tools.py` giảm từ 566 ➔ 291 dòng.
 >
+> ### 🔧 TÓM TẮT R06
+> * **R06-T01**: `domain/workspaces/workspace_session.py::WorkspaceSession` — snapshot bất biến (project_id, workspace_root, sandbox_dir, allowed_paths) + `is_allowed(path)`.
+> * **R06-T02**: `infrastructure/persistence/json/{workspace_repository_impl,conversation_repository_impl}.py` bọc `core/projects.py`/`core/history.py`. **Đã sửa bug thật**: `save_project`/`save_conversation`/`rename_conversation`/`set_pinned` trước đây `path.write_text()` không atomic (crash giữa lúc ghi = file JSON hỏng, `load_project`/`load_conversation` coi file hỏng như "không tồn tại" — mất project/hội thoại âm thầm). Giờ cả 4 hàm ghi qua `infrastructure/persistence/json/atomic_write.py::write_json` (temp file + `os.replace`). Có test giả lập crash giữa lúc ghi xác nhận file cũ không bị hỏng.
+> * **R06-T03**: `infrastructure/filesystem/execution_workspace.py::ExecutionWorkspace` — đặt tên cho quy ước `.scratch` đã có sẵn (không đổi vị trí file).
+> * **R06-T04**: Sửa race trong `ui/chat_panel.py` (không phải trực tiếp `_load_current`, xem "còn nợ" #2). `ChatPanel._persist_session` (lưu hội thoại của turn CHẠY NGẦM, không phải conversation đang xem) trước đây gọi `self.ctx.config.history_dir()` SỐNG tại thời điểm turn xong — nếu user đổi project khi turn còn chạy (`_load_current` ghi `config._project_history_dir`), turn nền lưu nhầm vào thư mục lịch sử của project MỚI. Fix: thêm `"home_history_dir"` vào dict `ctx` per-turn đã có sẵn (cùng quy ước với `home_id`/`home_messages`/`home_title`), chụp tại lúc submit. Test thật bằng Qt offscreen: `tests/integration/test_history_dir_race.py`.
+> * **R06-T05**: `application/workspaces/file_workspace_service.py::FileWorkspaceService` — cho File Explorer/AI Editor gọi `execute_tool` (list_dir/read_file/write_file/edit_file) giống agent, không tự viết lại logic.
+>
 > ### 🔧 TÓM TẮT R05
 > * **R05-T01/T02**: `core/tools.py`'s if/elif dispatcher tách thành `infrastructure/filesystem/{file_tools,command_tools,fetch_tools,tool_context}.py` + `domain/tools/{tool_descriptor,tool_registry}.py`. `core/tools.py` còn lại là shim strangler-fig (re-export `ToolContext`/`ToolError`, dispatch qua dict).
 > * **R05-T03**: `application/conversations/tool_policy_gateway.py::ToolPolicyGateway` — thay `if gate is not None and name in ("run_command","install_package")` (chat_agent.py) và `if name in (WRITE_TOOLS|MS365_WRITE_TOOLS)` (code_agent.py) bằng một lookup capability chung. Đã verify bằng test: đúng 2 tool cũ vẫn được gate, không tool nào khác bị ảnh hưởng.
@@ -80,8 +87,10 @@
 > * **R05-T05**: `infrastructure/mcp/mcp_source_manager.py::McpToolSourceManager` — tách lifecycle connection (cache/lock/start-or-skip) ra khỏi `state.py::AppContext` (trước đây inline trong `_mcp_connections`/`_conn_lock`). `AppContext` giờ chỉ gọi `self._mcp_manager.ensure/stop/stop_all`. `_ext_connections` (Connectors CAD/CAE/MS365/Other) KHÔNG thuộc phạm vi T05, vẫn giữ `_conn_lock` riêng như cũ.
 >
 > ### 📌 CÒN NỢ / CẦN QUYẾT ĐỊNH
-> 1. **Xung đột file với EPIC R02 (Team Nam)**: `docs/refactor/Refactoring_Checklist.md` dòng ~83 giao `infrastructure/persistence/json/atomic_json_file.py` cho Team Nam (R02-T01). R06-T02 (Team Hoa) cũng cần một helper ghi JSON atomic cho `WorkspaceRepository`/`ConversationRepository`. Để tránh 2 team cùng sửa 1 file, R06 sẽ dùng một helper atomic-write cục bộ trong `infrastructure/persistence/json/workspace_repository_impl.py`/`conversation_repository_impl.py` cho tới khi R02 xong, rồi hợp nhất vào `atomic_json_file.py` chung — **cần Team Nam xác nhận** khi họ bắt đầu R02-T01.
-> 2. Việc kế tiếp của Team Hoa là **R06** (Workspace, Filesystem & History Isolation).
+> 1. **Xung đột file với EPIC R02 (Team Nam)**: R02-T01 giao `infrastructure/persistence/json/atomic_json_file.py` cho Team Nam. R06-T02 cần atomic write NGAY (bug thật, không chờ được) nên đã tạo `infrastructure/persistence/json/atomic_write.py` — tên khác, cùng thư mục, không đụng file của Team Nam. `core/projects.py`/`core/history.py` đang dùng module này trực tiếp. **Cần Team Nam xác nhận khi bắt đầu R02-T01**: nên hợp nhất `atomic_write.py` vào `atomic_json_file.py` (Team Hoa đổi 4 import) hay giữ 2 module riêng (rủi ro trôi giữa 2 cách ghi atomic).
+> 2. **`WorkspaceRepository`/`ConversationRepository`/`FileWorkspaceService` chưa có nơi gọi thật** — giống tình trạng `ProviderRegistry` của Team Duy ở R03. Mọi call site sản xuất (`ui/workspace_tab.py`, `ui/folder_tab.py`, `state.py`, task executors) vẫn dùng trực tiếp `core/projects.py`/`core/history.py`/`core/tools.py::execute_tool` — các class mới là seam cho tầng application ở EPIC sau (R07/R08), chưa nối dây.
+> 3. **R06-T04 phạm vi thực tế khác một chút so với mô tả gốc**: bug không nằm ở `ui/workspace_tab.py::_load_current` (hàm đó chỉ *set* `config._project_history_dir`, không tự đọc lại nó) mà ở `ui/chat_panel.py::_persist_session` — nơi một turn chạy ngầm đọc SỐNG giá trị đó lúc turn xong. Đã sửa đúng điểm đọc, có test Qt offscreen thật (`tests/integration/test_history_dir_race.py`), nhưng chưa đổi kiến trúc `_load_current` như plan gốc gợi ý (dùng session id thay biến toàn cục) — việc đó cần tách `ChatPanel`/`WorkspaceTab` sâu hơn, thuộc phạm vi R08 (UI/Application Separation).
+> 4. R05/R06 xong toàn bộ — Team Hoa chờ chỉ đạo cho **R07** (Scheduling & Workflow Runtime, phối hợp Team Nam) hoặc merge/review trước khi tiếp tục.
 
 ---
 
@@ -180,16 +189,16 @@
 * **Team chịu trách nhiệm**: 🟢 **Team Hoa** (Chủ trì)
 * **Mục tiêu**: Xóa bỏ biến toàn cục `state.py::active_project_id`, đóng gói workspace per-turn thành `WorkspaceSession` bất biến, bảo vệ an toàn đường dẫn sandbox.
 
-- [ ] **R06-T01 (Team Hoa)**: Định nghĩa `WorkspaceSession` chứa snapshot project id, workspace root ➔ `domain/workspaces/workspace_session.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R06-T02 (Team Hoa)**: Xây dựng `WorkspaceRepository` từ `core/projects.py` & `ConversationRepository` từ `core/history.py` ➔ `infrastructure/persistence/json/workspace_repository_impl.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R06-T03 (Team Hoa)**: Xây dựng `ExecutionWorkspace` quản lý output/scratch files ➔ `infrastructure/filesystem/execution_workspace.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R06-T04 (Team Hoa)**: Khắc phục race condition trong `ui/workspace_tab.py::_load_current`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R06-T05 (Team Hoa)**: Xây dựng `FileWorkspaceService` xử lý thao tác file cho File Explorer và AI File Editor ➔ `application/workspaces/file_workspace_service.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
+- [x] **R06-T01 (Team Hoa)**: Định nghĩa `WorkspaceSession` chứa snapshot project id, workspace root ➔ `domain/workspaces/workspace_session.py`  
+  *Start: `2026-08-21 22:19` | End: `2026-08-21 22:24`*
+- [x] **R06-T02 (Team Hoa)**: Xây dựng `WorkspaceRepository` từ `core/projects.py` & `ConversationRepository` từ `core/history.py` ➔ `infrastructure/persistence/json/workspace_repository_impl.py`  
+  *Start: `2026-08-21 22:24` | End: `2026-08-21 22:35`*
+- [x] **R06-T03 (Team Hoa)**: Xây dựng `ExecutionWorkspace` quản lý output/scratch files ➔ `infrastructure/filesystem/execution_workspace.py`  
+  *Start: `2026-08-21 22:35` | End: `2026-08-21 22:40`*
+- [x] **R06-T04 (Team Hoa)**: Khắc phục race condition trong `ui/workspace_tab.py::_load_current`  
+  *Start: `2026-08-21 22:40` | End: `2026-08-21 22:50`*
+- [x] **R06-T05 (Team Hoa)**: Xây dựng `FileWorkspaceService` xử lý thao tác file cho File Explorer và AI File Editor ➔ `application/workspaces/file_workspace_service.py`  
+  *Start: `2026-08-21 22:50` | End: `2026-08-21 22:57`*
 
 ---
 

domain/workspaces/__init__.py

@@ -0,0 +1,5 @@
+"""Domain entities for workspace/project isolation (EPIC R06)."""
+
+from .workspace_session import WorkspaceSession
+
+__all__ = ["WorkspaceSession"]

domain/workspaces/workspace_session.py

@@ -0,0 +1,96 @@
+"""WorkspaceSession - an immutable snapshot of which project a turn belongs
+to and where it may touch the filesystem (R06-T01).
+
+``state.py::AppContext.active_project_id`` is a single mutable field read by
+every background worker thread. ``ui/workspace_tab.py::_load_current`` writes
+it (and the related ``config._project_history_dir``) on the UI thread the
+moment the user switches projects - while a turn already running on a
+worker thread may read either field mid-switch and end up acting on the
+OTHER project's workspace/history for the rest of its run (the race
+R06-T04 fixes).
+
+The fix, same shape as R04's ``ConversationExecutionRequest``: capture the
+workspace facts a turn needs ONCE, on the thread that knows which project is
+selected, into one frozen object. Whatever the user does to the UI afterwards,
+the turn keeps using the workspace it was handed at submit time.
+
+Pure domain code: stdlib only, no Qt, no network. It does touch ``Path`` (not
+plain strings, unlike ``ConversationExecutionRequest``) because its whole job
+is path-containment checking - a snapshot with no room to answer "is this
+path mine" would not replace what ``ToolContext.resolve`` currently does
+inline.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Tuple
+
+
+@dataclass(frozen=True)
+class WorkspaceSession:
+    """Everything a turn needs to know about ITS workspace, fixed at the
+    moment it was submitted.
+
+    Attributes:
+        project_id: the project this turn belongs to (``""`` when no project
+            is selected - e.g. the Code tab, which has no project concept).
+        workspace_root: the project's sandbox root (``Project.workspace_dir()``).
+        sandbox_dir: the ``.scratch`` subtree inside ``workspace_root`` used for
+            generator/helper scripts, never a final deliverable (see
+            ``infrastructure/filesystem/file_tools.py::_flatten_rel``).
+        allowed_paths: every root a tool call may read/write under. Almost
+            always just ``(workspace_root,)``; a project with a custom
+            ``output_dir`` outside the managed workspace tree still resolves
+            to exactly one root - the tuple exists so a future caller (e.g. a
+            step scoped to a shared input folder) can widen it without a
+            shape change.
+    """
+
+    project_id: str
+    workspace_root: Path
+    sandbox_dir: Path
+    allowed_paths: Tuple[Path, ...] = field(default_factory=tuple)
+
+    def __post_init__(self) -> None:
+        if not self.allowed_paths:
+            object.__setattr__(self, "allowed_paths", (self.workspace_root,))
+
+    @classmethod
+    def from_project(cls, project) -> "WorkspaceSession":
+        """Build a session from a ``core.projects.Project``. ``project`` is
+        typed loosely (not imported) so this module has no dependency on
+        ``core/`` - the caller (``core/projects.py`` itself, or
+        ``application/conversations``) already has the Project in hand."""
+        root = Path(project.workspace_dir())
+        return cls(project_id=project.project_id, workspace_root=root,
+                   sandbox_dir=root / ".scratch", allowed_paths=(root,))
+
+    @classmethod
+    def unscoped(cls, workspace_root: Path) -> "WorkspaceSession":
+        """A session for callers with no project concept (e.g. the Code tab,
+        which sandboxes to a plain folder rather than a ``Project``)."""
+        root = Path(workspace_root)
+        return cls(project_id="", workspace_root=root, sandbox_dir=root / ".scratch")
+
+    def is_allowed(self, path: Path) -> bool:
+        """True when ``path`` resolves inside one of :attr:`allowed_paths`.
+
+        Same containment rule as ``ToolContext.resolve`` (an exact root match
+        or a real descendant), but side-effect-free: it reports the answer
+        instead of raising, so a caller (``FileWorkspaceService``, R06-T05)
+        can decide what "not allowed" means for its own UI instead of
+        catching a ``ToolError``.
+        """
+        try:
+            resolved = Path(path).expanduser().resolve()
+        except OSError:
+            return False
+        for allowed in self.allowed_paths:
+            root = Path(allowed).resolve()
+            if resolved == root or root in resolved.parents:
+                return True
+        return False
+
+
+__all__ = ["WorkspaceSession"]

infrastructure/filesystem/execution_workspace.py

@@ -0,0 +1,80 @@
+"""ExecutionWorkspace - the output folder vs. the scratch folder for one
+turn, as two distinct properties instead of a name convention (R06-T03).
+
+Today the ``.scratch`` subtree is a special case buried inside
+``_flatten_rel`` (``infrastructure/filesystem/file_tools.py``): a generator
+script writes there, the deliverable lands in the output root, and
+``core/chat_agent.py`` cleans ``.scratch`` up after the turn — but nothing
+NAMES "the scratch folder" as a thing; every call site re-derives
+``workdir / ".scratch"`` (or checks ``Path(rel).parts[0] == ".scratch"``) by
+hand. This class gives that convention one home.
+
+It does not change WHERE files land - ``workspace_root/.scratch`` stays
+exactly what it always was. It exists so a caller (an application service,
+R06-T05's ``FileWorkspaceService``, or a future turn-cleanup step) can ask
+for "the output dir" / "the scratch dir" instead of hand-building the path
+and hoping the convention hasn't drifted.
+"""
+from __future__ import annotations
+
+import shutil
+from dataclasses import dataclass
+from pathlib import Path
+
+from cowork_local.domain.workspaces import WorkspaceSession
+
+SCRATCH_DIRNAME = ".scratch"
+
+
+@dataclass(frozen=True)
+class ExecutionWorkspace:
+    """The two folders a turn actually writes to, derived from a
+    :class:`WorkspaceSession`.
+
+    ``output_dir`` is always the session's ``workspace_root`` itself, not a
+    per-turn subfolder - Cowork's whole design is that every deliverable lands
+    directly in the one configured Output folder (see
+    ``infrastructure/filesystem/file_tools.py::_flatten_rel``'s docstring).
+    ``scratch_dir`` is the SAME flat ``workspace_root/.scratch`` every turn on
+    that workspace already shares today (``core/chat_agent.py``'s
+    ``_cleanup_cowork_intermediates`` operates on that exact path) - this
+    class does not introduce per-turn namespacing that doesn't exist in the
+    engine yet, only names the existing convention.
+
+    ``turn_id`` is kept as metadata for callers that want to attribute a
+    workspace to the turn that used it (logging, future per-turn scratch
+    namespacing); it does not affect either path today.
+    """
+
+    session: WorkspaceSession
+    turn_id: str
+
+    @property
+    def output_dir(self) -> Path:
+        return self.session.workspace_root
+
+    @property
+    def scratch_dir(self) -> Path:
+        return self.session.workspace_root / SCRATCH_DIRNAME
+
+    def ensure_dirs(self) -> None:
+        """Create both folders if they don't exist yet. Callers that only
+        need one (most do) can skip this and let ``write_file`` create parents
+        on demand, same as today."""
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+        self.scratch_dir.mkdir(parents=True, exist_ok=True)
+
+    def cleanup_scratch(self) -> None:
+        """Unconditionally remove the scratch subtree.
+
+        Coarser than ``core/chat_agent.py::_cleanup_cowork_intermediates``,
+        which rescues any real deliverable a generator script wrote INSIDE
+        ``.scratch`` before wiping it - that rescue logic stays there. This
+        is for callers that only need "make the scratch folder go away"
+        (e.g. before starting a fresh run) and know it holds nothing worth
+        saving.
+        """
+        shutil.rmtree(self.scratch_dir, ignore_errors=True)
+
+
+__all__ = ["ExecutionWorkspace", "SCRATCH_DIRNAME"]

infrastructure/persistence/__init__.py

@@ -0,0 +1 @@
+"""Persistence adapters (EPIC R02/R06)."""

infrastructure/persistence/json/__init__.py

@@ -0,0 +1,8 @@
+"""JSON-file persistence adapters: crash-safe writes and the workspace/
+conversation repositories built on them (EPIC R06)."""
+
+from .atomic_write import write_json
+from .conversation_repository_impl import ConversationRepository
+from .workspace_repository_impl import WorkspaceRepository
+
+__all__ = ["write_json", "WorkspaceRepository", "ConversationRepository"]

infrastructure/persistence/json/atomic_write.py

@@ -0,0 +1,56 @@
+"""write_json - crash-safe JSON writes (R06-T02).
+
+``core/projects.py::save_project`` and ``core/history.py``'s
+``save_conversation``/``rename_conversation``/``set_pinned`` all do a plain
+``path.write_text(json.dumps(...))`` today. That is two syscalls with a gap in
+between: a crash, a killed process, or a full disk between the truncate and
+the write leaves a half-written, unparseable JSON file - the NEXT read of
+that project/conversation then fails outright (``load_project`` /
+``load_conversation`` already treat a parse error as "missing", so this isn't
+even a loud failure - a project can silently vanish).
+
+``write_json`` fixes this the standard way: write the full content to a
+temporary file in the SAME directory (so the following replace is on one
+filesystem, not crossing a mount point), then atomically rename it over the
+target. Either the old file is still there, or the new one is fully there -
+never a partial one.
+
+Transitional note: EPIC R02 (Team Nam, ``docs/refactor/Refactoring_Checklist.md``
+R02-T01) plans a shared ``infrastructure/persistence/json/atomic_json_file.py``
+for the SAME purpose across the whole app (config, secrets, ...). This module
+is deliberately named differently and scoped to R06's two repositories only,
+so the two EPICs don't edit the same file in parallel; once R02-T01 lands,
+``WorkspaceRepository``/``ConversationRepository`` should switch to it and
+this module can go away.
+"""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from pathlib import Path
+from typing import Any
+
+
+def write_json(path: Path, data: Any) -> None:
+    """Serialize ``data`` as indented UTF-8 JSON and write it to ``path``
+    atomically. Creates parent directories if needed."""
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    text = json.dumps(data, ensure_ascii=False, indent=2)
+    fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp")
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(text)
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.replace(tmp_name, path)
+    except BaseException:
+        try:
+            os.unlink(tmp_name)
+        except OSError:
+            pass
+        raise
+
+
+__all__ = ["write_json"]

infrastructure/persistence/json/conversation_repository_impl.py

@@ -0,0 +1,54 @@
+"""ConversationRepository - an object-shaped, atomic-write-backed facade over
+``core/history.py`` (R06-T02). Same rationale as
+``workspace_repository_impl.py``: the module-level functions in
+``core/history.py`` are still what production code calls (they now write
+atomically themselves), this class is the seam for application-layer code
+that wants an object instead of a directory-parameterised function.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from cowork_local.config import HISTORY_DIR
+from cowork_local.core.history import (
+    delete_conversation,
+    list_conversations,
+    load_conversation,
+    new_session_id,
+    rename_conversation,
+    save_conversation,
+    set_pinned,
+)
+
+
+class ConversationRepository:
+    """CRUD + search over conversation JSON files, scoped to one
+    ``directory`` (defaults to the app's real ``HISTORY_DIR``)."""
+
+    def __init__(self, directory: Optional[Path] = None) -> None:
+        self._directory = Path(directory) if directory is not None else HISTORY_DIR
+
+    def new_session_id(self) -> str:
+        return new_session_id()
+
+    def save(self, kind: str, session_id: str, messages: List[Dict[str, Any]], **kwargs) -> Path:
+        return save_conversation(self._directory, kind, session_id, messages, **kwargs)
+
+    def load(self, path: Path) -> Dict[str, Any]:
+        return load_conversation(path)
+
+    def list(self, query: str = "") -> List[Dict[str, Any]]:
+        return list_conversations(self._directory, query)
+
+    def delete(self, path: Path) -> None:
+        delete_conversation(path)
+
+    def rename(self, path: Path, new_title: str) -> None:
+        rename_conversation(path, new_title)
+
+    def set_pinned(self, path: Path, pinned: bool) -> None:
+        set_pinned(path, pinned)
+
+
+__all__ = ["ConversationRepository"]

infrastructure/persistence/json/workspace_repository_impl.py

@@ -0,0 +1,59 @@
+"""WorkspaceRepository - an object-shaped, atomic-write-backed facade over
+``core/projects.py`` (R06-T02).
+
+``core/projects.py``'s module-level functions (``list_projects``,
+``load_project``, ``save_project``, ``new_project``, ``delete_project``) are
+still what every existing call site (``ui/workspace_tab.py``, ``state.py``,
+task executors) uses, and stay that way - they now write through
+:func:`atomic_write.write_json` themselves, so the durability fix applies
+whether or not a caller ever touches this class.
+
+This repository exists for the application layer (``application/workspaces``,
+R06-T05) to depend on an interface instead of reaching into ``core/`` -
+useful once code above ``core/`` starts being written against
+``domain``/``application`` seams instead of the legacy module functions. It
+is a thin pass-through today, not a re-implementation: same on-disk format,
+same directory, same functions underneath.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import List, Optional
+
+from cowork_local.core.projects import (
+    PROJECTS_DIR,
+    Project,
+    delete_project,
+    list_projects,
+    load_project,
+    new_project,
+    save_project,
+)
+
+
+class WorkspaceRepository:
+    """CRUD over :class:`~cowork_local.core.projects.Project`, scoped to one
+    ``directory`` (defaults to the app's real ``PROJECTS_DIR``; tests pass a
+    ``tmp_path`` so nothing touches the user's real config folder)."""
+
+    def __init__(self, directory: Optional[Path] = None) -> None:
+        self._directory = directory or PROJECTS_DIR
+
+    def list(self) -> List[Project]:
+        return list_projects(self._directory)
+
+    def get(self, project_id: str) -> Optional[Project]:
+        return load_project(project_id, self._directory)
+
+    def save(self, project: Project) -> Path:
+        return save_project(project, self._directory)
+
+    def create(self, name: str, description: str = "", instructions: str = "",
+               output_dir: str = "") -> Project:
+        return new_project(name, description, instructions, output_dir, self._directory)
+
+    def delete(self, project_id: str) -> bool:
+        return delete_project(project_id, self._directory)
+
+
+__all__ = ["WorkspaceRepository"]

tests/integration/test_history_dir_race.py

@@ -0,0 +1,85 @@
+"""EPIC R06-T04: the race in ``ui/workspace_tab.py::_load_current``.
+
+``_load_current`` sets ``ctx.config._project_history_dir`` on the SHARED
+``AppConfig`` every time the user switches projects in the Workspace screen.
+A background turn (one that isn't the conversation currently displayed) used
+to resolve its save directory by calling ``ctx.config.history_dir()`` at
+``_persist_session`` time - i.e. whenever the turn actually finished, not
+when it started. If the user switched projects while it was still running,
+the turn's conversation got written into the NEW project's history folder
+instead of the one it actually belongs to.
+
+The fix threads a ``home_history_dir`` captured at submit time (same "home_*"
+snapshot convention ``ui/chat_panel.py`` already uses for session id/title/
+messages) through to the save call. This test drives the real
+``ChatPanel._persist_session`` - the actual save path - offscreen.
+"""
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+import pytest
+
+os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+
+from cowork_local.config import AppConfig             # noqa: E402
+from cowork_local.core.history import list_conversations  # noqa: E402
+from cowork_local.state import AppContext              # noqa: E402
+
+pytest.importorskip("PySide6", reason="Qt is required for the integration suite")
+
+
+@pytest.fixture(scope="module")
+def qt_app():
+    from PySide6.QtWidgets import QApplication
+
+    return QApplication.instance() or QApplication([])
+
+
+@pytest.fixture
+def chat_panel(qt_app, tmp_path: Path):
+    from cowork_local.ui.chat_panel import ChatPanel
+
+    ctx = AppContext(AppConfig.load(tmp_path / "config.json"))
+    return ChatPanel(ctx, "cowork", "Test")
+
+
+def test_background_turn_saves_into_the_project_it_started_in(chat_panel, tmp_path):
+    project_a_dir = tmp_path / "project-a-history"
+    project_b_dir = tmp_path / "project-b-history"
+    chat_panel.ctx.config._project_history_dir = project_a_dir
+
+    # What ChatPanel._start_turn captures into the per-turn ctx dict at
+    # submit time (see the "home_history_dir" entry added there for R06-T04).
+    turn_ctx = {
+        "home_id": chat_panel.session_id,
+        "home_messages": [{"role": "user", "content": "hi"}],
+        "home_title": "Background turn",
+        "home_history_dir": chat_panel.ctx.config.history_dir(),
+        "record": {},
+    }
+    assert turn_ctx["home_history_dir"] == project_a_dir
+
+    # The user switches projects in the Workspace screen WHILE this turn is
+    # still running - exactly what ui/workspace_tab.py::_load_current does.
+    chat_panel.ctx.config._project_history_dir = project_b_dir
+
+    chat_panel._persist_session(turn_ctx)
+
+    assert len(list_conversations(project_a_dir)) == 1
+    assert list_conversations(project_b_dir) == []
+
+
+def test_the_currently_viewed_conversation_still_follows_live_selection(chat_panel, tmp_path):
+    """_save_snapshot's OTHER caller (the initial "register it in History right
+    away" call, and _autosave) has no captured history_dir and must keep
+    resolving it live - that path is for the conversation ACTUALLY on screen,
+    which should follow whatever project the user has selected right now."""
+    project_dir = tmp_path / "currently-viewed"
+    chat_panel.ctx.config._project_history_dir = project_dir
+
+    chat_panel._save_snapshot(chat_panel.session_id,
+                              [{"role": "user", "content": "hi"}], "Live view")
+
+    assert len(list_conversations(project_dir)) == 1

tests/unit/test_atomic_write_and_repositories.py

@@ -0,0 +1,83 @@
+"""EPIC R06-T02: atomic JSON writes + the WorkspaceRepository/
+ConversationRepository facades over core/projects.py and core/history.py.
+
+The motivating bug: ``core/projects.py::save_project`` used to
+``path.write_text(json.dumps(...))`` — two syscalls, no atomicity. A failure
+between them must never leave a half-written file on disk; that is the one
+property these tests exist to pin.
+"""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from cowork_local.infrastructure.persistence.json import (
+    ConversationRepository,
+    WorkspaceRepository,
+    write_json,
+)
+from cowork_local.infrastructure.persistence.json.atomic_write import write_json as _write_json
+
+
+def test_write_json_round_trips(tmp_path):
+    path = tmp_path / "a.json"
+    write_json(path, {"hello": "world", "n": 3})
+    assert json.loads(path.read_text(encoding="utf-8")) == {"hello": "world", "n": 3}
+
+
+def test_write_json_leaves_no_temp_file_behind(tmp_path):
+    write_json(tmp_path / "a.json", {"x": 1})
+    assert list(tmp_path.iterdir()) == [tmp_path / "a.json"]
+
+
+def test_a_failed_write_never_corrupts_the_existing_file(tmp_path, monkeypatch):
+    """The whole point of write-temp-then-replace: if the replace step blows
+    up, the ORIGINAL file must still be there and still be readable — not
+    truncated, not half-written."""
+    path = tmp_path / "a.json"
+    write_json(path, {"version": 1})
+
+    import cowork_local.infrastructure.persistence.json.atomic_write as mod
+
+    def boom(*_a, **_k):
+        raise OSError("simulated crash between write and replace")
+
+    monkeypatch.setattr(mod.os, "replace", boom)
+    with pytest.raises(OSError):
+        write_json(path, {"version": 2})
+
+    assert json.loads(path.read_text(encoding="utf-8")) == {"version": 1}
+    # the abandoned temp file was cleaned up, not left orphaned
+    assert list(tmp_path.iterdir()) == [path]
+
+
+def test_workspace_repository_crud_round_trip(tmp_path):
+    repo = WorkspaceRepository(tmp_path)
+    project = repo.create("My Project", description="d")
+
+    assert [p.project_id for p in repo.list()] == [project.project_id]
+
+    project.description = "updated"
+    repo.save(project)
+    assert repo.get(project.project_id).description == "updated"
+
+    assert repo.delete(project.project_id) is True
+    assert repo.get(project.project_id) is None
+
+
+def test_conversation_repository_crud_round_trip(tmp_path):
+    repo = ConversationRepository(tmp_path)
+    session_id = repo.new_session_id()
+    path = repo.save("cowork", session_id, [{"role": "user", "content": "hi"}])
+
+    assert [c["session_id"] for c in repo.list()] == [session_id]
+
+    repo.rename(path, "Renamed")
+    repo.set_pinned(path, True)
+    data = repo.load(path)
+    assert data["title"] == "Renamed"
+    assert data["pinned"] is True
+
+    repo.delete(path)
+    assert repo.list() == []

tests/unit/test_execution_workspace.py

@@ -0,0 +1,51 @@
+"""EPIC R06-T03: ExecutionWorkspace names the output-dir/scratch-dir split
+that already exists in ``core/chat_agent.py`` (``.scratch`` under the
+workspace root) without changing where anything lands."""
+from __future__ import annotations
+
+from cowork_local.domain.workspaces import WorkspaceSession
+from cowork_local.infrastructure.filesystem.execution_workspace import ExecutionWorkspace
+
+
+def test_output_dir_is_the_workspace_root_itself(tmp_path):
+    session = WorkspaceSession.unscoped(tmp_path)
+    workspace = ExecutionWorkspace(session, turn_id="turn-1")
+
+    assert workspace.output_dir == tmp_path
+
+
+def test_scratch_dir_matches_the_existing_flat_convention(tmp_path):
+    """core/chat_agent.py::_cleanup_cowork_intermediates operates on
+    ``output_dir / ".scratch"`` with no per-turn subfolder — this must agree,
+    or cleanup_scratch() would target a directory nothing ever wrote to."""
+    session = WorkspaceSession.unscoped(tmp_path)
+    workspace = ExecutionWorkspace(session, turn_id="turn-1")
+
+    assert workspace.scratch_dir == tmp_path / ".scratch"
+
+
+def test_ensure_dirs_creates_both_folders(tmp_path):
+    session = WorkspaceSession.unscoped(tmp_path / "root")
+    workspace = ExecutionWorkspace(session, turn_id="t")
+    workspace.ensure_dirs()
+
+    assert workspace.output_dir.is_dir()
+    assert workspace.scratch_dir.is_dir()
+
+
+def test_cleanup_scratch_removes_it_and_leaves_output_dir_alone(tmp_path):
+    session = WorkspaceSession.unscoped(tmp_path)
+    workspace = ExecutionWorkspace(session, turn_id="t")
+    workspace.ensure_dirs()
+    (workspace.scratch_dir / "helper.py").write_text("print(1)", encoding="utf-8")
+    (workspace.output_dir / "deliverable.txt").write_text("done", encoding="utf-8")
+
+    workspace.cleanup_scratch()
+
+    assert not workspace.scratch_dir.exists()
+    assert (workspace.output_dir / "deliverable.txt").exists()
+
+
+def test_cleanup_scratch_is_a_no_op_when_never_created(tmp_path):
+    workspace = ExecutionWorkspace(WorkspaceSession.unscoped(tmp_path), turn_id="t")
+    workspace.cleanup_scratch()  # must not raise

tests/unit/test_file_workspace_service.py

@@ -0,0 +1,62 @@
+"""EPIC R06-T05: FileWorkspaceService gives File Explorer / AI Editor the
+same safe file operations the agent tool loop already has, via the SAME
+``core/tools.py::execute_tool`` dispatch (not a reimplementation)."""
+from __future__ import annotations
+
+from cowork_local.application.workspaces import FileWorkspaceService
+from cowork_local.domain.workspaces import WorkspaceSession
+
+
+def _service(tmp_path) -> FileWorkspaceService:
+    return FileWorkspaceService(WorkspaceSession.unscoped(tmp_path))
+
+
+def test_write_then_read_round_trips(tmp_path):
+    service = _service(tmp_path)
+    written = service.write_file("notes.md", "# Hello")
+    assert written["ok"] is True
+
+    read = service.read_preview("notes.md")
+    assert read == {"ok": True, "output": "# Hello"}
+
+
+def test_list_tree_reflects_written_files(tmp_path):
+    service = _service(tmp_path)
+    service.write_file("a.txt", "x")
+    listing = service.list_tree()
+    assert listing["ok"] is True and "a.txt" in listing["output"]
+
+
+def test_apply_edit_uses_the_context_anchored_replace(tmp_path):
+    service = _service(tmp_path)
+    service.write_file("code.py", "value = 1\n")
+    edited = service.apply_edit("code.py", "value = 1", "value = 2")
+    assert edited["ok"] is True
+    assert service.read_preview("code.py")["output"].strip() == "value = 2"
+
+
+def test_apply_edit_reports_ambiguous_match_like_the_agent_tool_does(tmp_path):
+    service = _service(tmp_path)
+    service.write_file("code.py", "x = 1\nx = 1\n")
+    edited = service.apply_edit("code.py", "x = 1", "x = 2")
+    assert edited["ok"] is False
+    assert "appears" in edited["output"]
+
+
+def test_path_escape_is_refused_not_a_crash(tmp_path):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    (tmp_path / "outside.txt").write_text("secret", encoding="utf-8")
+    service = FileWorkspaceService(WorkspaceSession.unscoped(workspace))
+
+    result = service.read_preview("../outside.txt")
+    assert result["ok"] is False
+    assert "outside the working folder" in result["output"]
+
+
+def test_write_preserves_subfolders_unlike_cowork_flatten_writes(tmp_path):
+    """File Explorer must not collapse a write into the workspace root the
+    way Cowork's agent context does (flatten_writes=True there, False here)."""
+    service = _service(tmp_path)
+    service.write_file("sub/dir/file.txt", "content")
+    assert (tmp_path / "sub" / "dir" / "file.txt").read_text(encoding="utf-8") == "content"

tests/unit/test_workspace_session.py

@@ -0,0 +1,64 @@
+"""EPIC R06-T01: WorkspaceSession is a frozen snapshot, captured once, that a
+turn keeps using regardless of what the UI does to the live project
+selection afterwards - see the module docstring for the race this replaces.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from cowork_local.domain.workspaces import WorkspaceSession
+
+
+class _FakeProject:
+    def __init__(self, project_id: str, root: Path):
+        self.project_id = project_id
+        self._root = root
+
+    def workspace_dir(self) -> Path:
+        return self._root
+
+
+def test_from_project_derives_sandbox_dir_under_the_workspace_root(tmp_path):
+    project = _FakeProject("proj-a", tmp_path)
+    session = WorkspaceSession.from_project(project)
+
+    assert session.project_id == "proj-a"
+    assert session.workspace_root == tmp_path
+    assert session.sandbox_dir == tmp_path / ".scratch"
+    assert session.allowed_paths == (tmp_path,)
+
+
+def test_is_allowed_true_for_the_root_and_descendants(tmp_path):
+    session = WorkspaceSession.from_project(_FakeProject("p", tmp_path))
+    nested = tmp_path / "sub" / "file.txt"
+    nested.parent.mkdir(parents=True)
+    nested.write_text("x", encoding="utf-8")
+
+    assert session.is_allowed(tmp_path) is True
+    assert session.is_allowed(nested) is True
+
+
+def test_is_allowed_false_outside_the_workspace(tmp_path):
+    session = WorkspaceSession.from_project(_FakeProject("p", tmp_path / "a"))
+    outside = tmp_path / "b" / "secret.txt"
+
+    assert session.is_allowed(outside) is False
+
+
+def test_two_sessions_from_different_projects_stay_independent(tmp_path):
+    """The exact race this snapshot exists to prevent: a turn holding session
+    A must never start accepting paths that belong to session B, no matter
+    what the (mutable, shared) AppContext does after the snapshot was taken."""
+    session_a = WorkspaceSession.from_project(_FakeProject("a", tmp_path / "a"))
+    session_b = WorkspaceSession.from_project(_FakeProject("b", tmp_path / "b"))
+
+    assert session_a.is_allowed(tmp_path / "b" / "file.txt") is False
+    assert session_b.is_allowed(tmp_path / "a" / "file.txt") is False
+
+
+def test_unscoped_session_has_no_project_id(tmp_path):
+    session = WorkspaceSession.unscoped(tmp_path)
+    assert session.project_id == ""
+    assert session.is_allowed(tmp_path / "code.py") is True

ui/chat_panel.py

@@ -1126,6 +1126,15 @@ class ChatPanel(QWidget):
             "snapshot_len": len(snapshot), "out_dir": out_dir,
             "home_id": self.session_id, "home_messages": self.messages,
             "home_title": self.title, "home_out_root": self.workspace_dir(),
+            # R06-T04: captured NOW, at submit time — see _persist_session's
+            # use of this. Without it, a background turn (this session isn't
+            # the one currently displayed) saves into whatever
+            # ctx.config.history_dir() resolves to AT THE TIME IT FINISHES,
+            # which is the *currently viewed* project's history folder if the
+            # user switched projects (ui/workspace_tab.py::_load_current)
+            # while this turn was still running — silently saving one
+            # project's conversation into another project's history folder.
+            "home_history_dir": self.ctx.config.history_dir(),
             "detached": False,
             # For re-rendering the in-progress turn if the user reopens this chat:
             "display_text": text, "partial": "", "plan_steps": [],
@@ -1356,10 +1365,18 @@ class ChatPanel(QWidget):
         return ctx.get("home_id") == self.session_id and not ctx.get("detached")
 
     def _save_snapshot(self, session_id: str, messages: List[Dict[str, Any]],
-                       title: str, inputs: Optional[List[str]] = None) -> None:
+                       title: str, inputs: Optional[List[str]] = None,
+                       history_dir: Optional[Path] = None) -> None:
         """Persist a conversation by id (used both to register it in History the
         moment it starts and to save a finished background turn). No-op until it has
-        a user message. Never raises into the UI."""
+        a user message. Never raises into the UI.
+
+        ``history_dir``, when given, is used INSTEAD of
+        ``self.ctx.config.history_dir()`` — see ``_persist_session`` (R06-T04):
+        a background turn must save into the project it started in, not
+        whichever project happens to be selected in the Workspace screen by
+        the time the turn finishes.
+        """
         if not self.ctx.config.history.get("autosave", True):
             return
         if not any(m.get("role") == "user" for m in messages):
@@ -1367,7 +1384,8 @@ class ChatPanel(QWidget):
         try:
             from ..core.history import save_conversation
             save_conversation(
-                self.ctx.config.history_dir(), self.kind, session_id,
+                history_dir if history_dir is not None else self.ctx.config.history_dir(),
+                self.kind, session_id,
                 messages, title, inputs=list(inputs or []), outputs=[],
                 # Only the CURRENT view knows its project for sure; a background
                 # turn's save must not overwrite another conversation's project
@@ -1383,7 +1401,8 @@ class ChatPanel(QWidget):
         view-based _autosave can't). Outputs are rebuilt from disk on reopen."""
         self._save_snapshot(ctx["home_id"], ctx["home_messages"],
                             ctx.get("home_title", ""),
-                            inputs=ctx.get("record", {}).get("inputs", []))
+                            inputs=ctx.get("record", {}).get("inputs", []),
+                            history_dir=ctx.get("home_history_dir"))
         self.history_changed.emit()
 
     def running_session_ids(self):