feat(R01): architecture foundation, offline fakes and characterization net

EPIC R01 (Team Duy) - safety net before the parallel refactor starts.

R01-T01 docs/architecture/ADR-001-layered-architecture.md
  4-tier boundaries, allowed dependency directions, invariants I1-I6 and
  the strangler-fig migration strategy.
R01-T02 tests/fakes/{fake_provider,fake_tool_executor}.py
  Scripted, offline Provider and extra-tool executor doubles.
R01-T03 scripts/check_imports.py
  AST-based Clean Architecture Guard (CASAN Check 3). Also covers relative
  imports and function-local imports; ASCII-only output for cp932 consoles.
R01-T04 tests/characterization/test_run_cowork.py
  13 snapshot tests pinning run_cowork's current observable contract before
  EPIC R04 moves its orchestration into application/.
R01-T05 docs/architecture/dormant-code.md
  Import-graph scan: 43 unimported modules verified down to 6 genuinely
  dormant items (~1887 LOC); the rest run via subprocess/CLI entry points.

tests/conftest.py binds `cowork_local` to THIS checkout by absolute path -
previously sys.path discovery could import a sibling checkout and the suite
would silently test the wrong code.

Suite: 104 passed, 1.08s (2 pre-existing failures in test_config_security.py
remain - config.py still ships a hardcoded default password, EPIC R02/Team Nam).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

docs/architecture/ADR-001-layered-architecture.md

@@ -0,0 +1,156 @@
+# ADR-001: Kiến Trúc 4 Tầng (Layered / Clean Architecture)
+
+* **Status**: Accepted
+* **Date**: 2026-08-21
+* **EPIC / Task**: R01-T01
+* **Owner**: 🔵 Team Duy (Tech Lead)
+* **Áp dụng cho**: toàn bộ mã nguồn mới của `cowork_local` (3 team)
+
+---
+
+## 1. Context (Bối cảnh)
+
+`cowork_local` hiện là một ứng dụng PySide6 desktop local-first ~55.000 dòng Python,
+được phát triển nhanh theo hướng feature-first. Hệ quả đo được tại thời điểm viết ADR:
+
+| Vấn đề | Bằng chứng cụ thể trong repo |
+| :--- | :--- |
+| **God widget** | `ui/co4e_tab.py` 2.089 dòng, `ui/chat_panel.py` 1.795 dòng, `ui/folder_tab.py` 1.590 dòng |
+| **Business logic nằm trong widget** | Vòng đời turn chat, quyết định routing, ghép prompt đều nằm trong `ui/chat_panel.py` |
+| **Logic trùng lặp 3 nơi** | `ui/chat_panel.py::_apply_routing`, `ui/co4e_tab.py::_apply_co4e_routing`, `ui/folder_tab.py::_ai_apply_routing` là ba bản sao gần như y hệt của cùng một thuật toán |
+| **Không test được nếu không có Qt** | Muốn test một quyết định routing phải dựng widget → không chạy được headless, không chạy được nhanh |
+| **Side-effect ẩn trong tầng hạ tầng** | Provider tự gọi `core.usage_tracker.record()` ngay trong vòng lặp stream (`providers/openai_compat.py::_record_usage`) |
+
+Ba team (Duy / Nam / Hoa) sẽ sửa song song trên cùng codebase trong 10 ngày. Nếu
+không có một ranh giới phụ thuộc được **kiểm chứng tự động**, các thay đổi song song
+sẽ hội tụ về đúng cấu trúc rối như cũ.
+
+## 2. Decision (Quyết định)
+
+Mã nguồn mới được tổ chức thành **4 tầng**, với **chiều phụ thuộc một chiều** như sau:
+
+```text
+┌─────────────────────────────────────────────────────────────┐
+│  presentation/            PySide6 widgets, Qt signals/slots  │
+│  (chat, co4e, workspace…) Chỉ dựng UI và phát/nhận signal    │
+└───────────────────────────┬─────────────────────────────────┘
+                            │ gọi xuống (được phép)
+┌───────────────────────────▼─────────────────────────────────┐
+│  application/             Pure Python orchestration          │
+│  (conversations,          Điều phối use-case, không biết Qt  │
+│   model_routing…)         và không biết HTTP/đĩa cụ thể      │
+└───────────────────────────┬─────────────────────────────────┘
+                            │ gọi xuống (được phép)
+┌───────────────────────────▼─────────────────────────────────┐
+│  domain/                  Pure Python entities & events      │
+│  (agents, models…)        Frozen dataclass, enum, quy tắc    │
+│                           nghiệp vụ thuần. KHÔNG import gì   │
+│                           từ 3 tầng còn lại.                 │
+└───────────────────────────▲─────────────────────────────────┘
+                            │ implement interface của domain
+┌───────────────────────────┴─────────────────────────────────┐
+│  infrastructure/          Adapters: network, keyring, đĩa,   │
+│  (providers, telemetry…)  process, Qt-free I/O               │
+└─────────────────────────────────────────────────────────────┘
+```
+
+### 2.1 Quy tắc bất biến (Invariants)
+
+| # | Quy tắc | Được kiểm bởi |
+| :--- | :--- | :--- |
+| **I1** | `domain/` và `application/` là **100% pure Python** — cấm import `PySide6`, `PyQt5`, `PyQt6`, `shiboken6` | `scripts/check_imports.py` (R01-T03) |
+| **I2** | `domain/` **không import** `application/`, `infrastructure/`, `presentation/`, `ui/` | `scripts/check_imports.py` |
+| **I3** | `application/` **không import** `presentation/` hay `ui/` | `scripts/check_imports.py` |
+| **I4** | Không file production nào vượt **400 dòng** | `scripts/check_loc.py` (R10-T02) |
+| **I5** | `presentation/` **không** gọi thẳng provider/HTTP/đĩa — phải đi qua một application service | Code review + I1–I3 |
+| **I6** | Mọi input của một use-case được đóng gói thành **snapshot bất biến** (`frozen dataclass`) trước khi rời UI thread | Code review + unit test |
+
+### 2.2 Chiều phụ thuộc được phép
+
+| Từ tầng | Được import | Bị cấm |
+| :--- | :--- | :--- |
+| `presentation/` | `application/`, `domain/`, PySide6 | — (nên tránh gọi thẳng `infrastructure/`) |
+| `application/` | `domain/`, interface do `domain/` định nghĩa | `presentation/`, `ui/`, PySide6 |
+| `domain/` | chỉ stdlib | tất cả các tầng khác, PySide6 |
+| `infrastructure/` | `domain/`, thư viện ngoài (requests, keyring…) | `presentation/`, `ui/`, PySide6 |
+
+### 2.3 Cách tầng dưới "nói chuyện ngược" lên UI
+
+`application/` **không được** giữ tham chiếu tới widget. Việc trao đổi ngược chiều
+đi qua **callback thuần Python nhận một `AgentEvent` có kiểu**
+(`domain/agents/agent_event.py`, R04-T02):
+
+```python
+# application layer — pure Python, không biết Qt tồn tại
+service.run_turn(request, on_event=my_callback)
+
+# presentation layer — chuyển event sang Qt signal ở ranh giới duy nhất này
+def my_callback(event: AgentEvent) -> None:
+    self.agent_event.emit(event)   # Qt signal → cập nhật UI trên main thread
+```
+
+Đây là **seam** duy nhất giữa hai thế giới: dưới seam là Python thuần test được
+offline, trên seam là Qt. Mọi cập nhật UI phải xảy ra qua Qt signal/slot, không
+bao giờ gọi trực tiếp từ worker thread.
+
+## 3. Vị trí sở hữu theo team
+
+| Tầng / thư mục | Team | EPIC |
+| :--- | :--- | :--- |
+| `presentation/chat/`, `application/conversations/`, `application/model_routing/`, `domain/agents/`, `domain/models/`, `infrastructure/providers/`, `infrastructure/telemetry/`, `tests/`, `scripts/` | 🔵 Duy | R01, R03, R04, R08, R10 |
+| `presentation/co4e/`, `monitoring/`, `settings/`, `shell/`, `application/workflows/`, `infrastructure/config/`, `secrets/`, `sandbox/` | 🟣 Nam | R02, R08, R09 |
+| `presentation/workspace/`, `folder/`, `scheduling/`, `application/workspaces/`, `scheduling/`, `domain/tools/`, `domain/tasks/`, `infrastructure/filesystem/`, `mcp/`, `persistence/` | 🟢 Hoa | R05, R06, R07, R08 |
+
+## 4. Chiến lược di trú (Strangler Fig, không big-bang)
+
+Code cũ trong `core/`, `ui/`, `providers/` **không bị xoá ngay**. Ta bọc dần:
+
+1. **Tạo seam mới** ở tầng đúng (ví dụ `RoutingApplicationService`).
+2. **Chuyển call site** cũ sang gọi seam mới (`ui/*.py` chỉ còn vài dòng adapter).
+3. **Giữ module cũ làm implementation detail** phía sau seam (ví dụ
+   `application/model_routing/` vẫn gọi xuống `core/routing/` để dùng lại
+   scorer/selector đã có test).
+4. Chỉ khi mọi call site đã đi qua seam mới → cân nhắc gỡ code cũ.
+
+Nhờ vậy `pytest` luôn xanh giữa các bước, và một team có thể merge mà không chờ
+team khác refactor xong.
+
+## 5. Consequences (Hệ quả)
+
+### Tích cực
+
+* Test một quyết định routing / một vòng đời turn chat **không cần Qt, không cần mạng** → suite unit chạy < 1 giây.
+* Ba bản sao logic routing hội tụ về một nơi duy nhất → sửa một lần, cả 3 màn hình cùng đúng.
+* Người mới có thể thêm một provider mà chỉ chạm `infrastructure/providers/` + `domain/models/`.
+* Vi phạm kiến trúc bị chặn ở CI thay vì phát hiện lúc review.
+
+### Tiêu cực / chi phí phải chấp nhận
+
+* Nhiều file nhỏ hơn thay vì vài file lớn → tăng số lần "nhảy file" khi đọc code.
+* Tồn tại **hai đường** trong giai đoạn di trú (code cũ + seam mới) cho tới khi call site cuối cùng chuyển xong.
+* Phải viết DTO/snapshot rõ ràng thay vì truyền thẳng `self` của widget — tốn thêm code, đổi lại được thread-safety.
+
+## 6. Alternatives considered (Phương án đã cân nhắc)
+
+| Phương án | Lý do loại |
+| :--- | :--- |
+| **Giữ nguyên, chỉ tách file cho ngắn** | Giải quyết được I4 (LOC) nhưng không giải quyết được nguyên nhân gốc: logic vẫn dính Qt nên vẫn không test được offline. |
+| **MVVM/MVP thuần Qt** | Vẫn buộc business logic phụ thuộc vòng đời Qt object; không chạy được trong scheduler headless và trong task nền. |
+| **Hexagonal đầy đủ (port/adapter cho mọi thứ)** | Đúng về lý thuyết nhưng quá tốn cho 10 ngày và cho một app desktop 1 process; 4 tầng là điểm cân bằng. |
+| **Big-bang rewrite** | Rủi ro hồi quy quá cao khi 3 team sửa song song và không có bộ test bảo vệ đầy đủ. |
+
+## 7. Enforcement (Thực thi)
+
+```bash
+python scripts/check_imports.py      # I1, I2, I3 — quét AST
+python scripts/check_loc.py          # I4 — giới hạn 400 dòng
+python scripts/run_quality_gate.py   # chạy toàn bộ CASAN Gate + pytest
+```
+
+CASAN Verification Gate phải PASS trước khi merge bất kỳ PR nào vào `main`.
+
+## 8. Tài liệu liên quan
+
+* `docs/refactor/Feature_Architecture_Proposal.md` — thiết kế tổng thể 10 EPIC
+* `docs/refactor/Refactoring_Checklist.md` — bảng tiến độ theo task
+* `docs/architecture/dormant-code.md` — danh mục code không còn hoạt động (R01-T05)

docs/architecture/dormant-code.md

@@ -0,0 +1,85 @@
+# Dormant / Dead Code Inventory (R01-T05)
+
+* **Task**: R01-T05 — Phân loại và cô lập mã nguồn cũ
+* **Owner**: 🔵 Team Duy
+* **Ngày quét**: 2026-08-21
+* **Phạm vi quét**: toàn bộ `*.py` production (loại trừ `tests/`, `assets/`, `docs/`, `.git/`)
+
+---
+
+## 1. Mục đích
+
+Trước khi 3 team refactor song song, cần biết **file nào thật sự đang chạy**. Refactor
+một module đã chết là lãng phí; xoá nhầm một module chỉ được gọi động là gây sự cố
+runtime. Tài liệu này phân loại từng ứng viên, kèm **bằng chứng** và **hành động đề xuất**.
+
+## 2. Phương pháp
+
+Quét AST toàn repo, dựng đồ thị import, tìm module **không có module nào khác import**.
+Kết quả thô: **43 module**. Sau đó xác minh thủ công từng ứng viên, vì phân tích tĩnh
+không thấy 3 kiểu tham chiếu:
+
+| Kiểu tham chiếu ẩn | Ví dụ thật trong repo |
+| :--- | :--- |
+| Chạy như subprocess | `state.py:285` gọi `python -m cowork_local.mcp_servers.ms365_server` |
+| Entry point của gói | `__main__.py` (chạy bằng `python -m cowork_local`) |
+| Script chạy tay | `tools/check_*.py`, `scripts/*.py` |
+
+> ⚠️ **Kết luận quan trọng**: 43 module "không ai import" **KHÔNG** đồng nghĩa 43 module chết.
+> Sau xác minh, chỉ còn **6 hạng mục (~1.887 dòng)** là dormant thật.
+
+## 3. Phân loại kết quả
+
+### 🟥 A. DORMANT THẬT — không có đường nào chạy tới (ứng viên xoá)
+
+| Module | LOC | Bằng chứng | Rủi ro khi xoá | Hành động |
+| :--- | ---: | :--- | :--- | :--- |
+| `ui/accounts_tab.py` | 700 | Chỉ xuất hiện trong comment của `i18n.py:92`; không widget nào khởi tạo `AccountsTab` | Thấp — panel Monitoring → Accounts hiện không có đường vào | Cô lập, chờ xác nhận PO rồi xoá |
+| `ui/flow_dialog.py` | 596 | Chỉ được nhắc trong docstring `ui/agent_manager_tab.py:4` và comment `i18n.py:2124` | Trung bình — Flow Manager có thể là tính năng tạm ẩn | **Hỏi PO trước**, chưa xoá |
+| `security/` (cả package) | 296 | `prompt_validator`, `action_validator`, `attachment_validator`, `audit_logger`, `command_risk_classifier` — không file nào ngoài package tự import. Chức năng **trùng** `core/agent_security.py` + `core/security_rules.py` (đang chạy thật) | Trung bình — dễ nhầm đây là lớp bảo mật đang hoạt động | ⚠️ Ưu tiên cao: xoá hoặc hợp nhất trong **R09 (Team Nam)** |
+| `core/codebase_memory_ui.py` | 123 | Không nơi nào import; `core/codebase_memory.py` (bản không-UI) mới là bản đang dùng | Thấp | Xoá |
+| `core/graph_server.py` | 115 | Docstring nói phục vụ build không có QtWebEngine, nhưng **không có call site nào**; `ui/structure_graph_view.py` không gọi | Trung bình — có thể là fallback cho bản .exe chưa nối dây | Xác minh với bản đóng gói PyInstaller trước khi xoá |
+| `ui/mcp_servers_dialog.py` | 57 | Không import; MCP settings hiện nằm trong `ui/settings_dialog.py` | Thấp | Xoá |
+
+**Tổng: ~1.887 dòng (≈ 3,4% codebase).**
+
+### 🟨 B. KHÔNG CHẾT — chạy qua đường ẩn (giữ nguyên)
+
+| Module | Vì sao phân tích tĩnh báo nhầm |
+| :--- | :--- |
+| `__main__.py` | Entry point `python -m cowork_local` |
+| `mcp_servers/ms365_server.py` | Chạy như tiến trình con — `state.py:285` |
+| `core/routing/__init__.py` | Được import qua đường dẫn con (`from .routing.service import RoutingService`), heuristic theo tên lá không thấy |
+| `tools/check_*.py` (34 file, 6.608 dòng) | Bộ smoke-test UI chạy tay: `python tools/check_nav.py`. Là **dev tooling**, không phải code chết |
+| `scripts/bootstrap_gitea_repo.py`, `scripts/check_imports.py` | Script CLI chạy tay / chạy trong CI |
+
+### 🟩 C. CODE SỐNG NHƯNG "ĐÓNG BĂNG" — đụng vào phải cẩn thận
+
+| Module | LOC | Ghi chú cho người refactor |
+| :--- | ---: | :--- |
+| `core/chat_agent.py::run_cowork` | 580 | Đang có **characterization test** (`tests/characterization/test_run_cowork.py`, R01-T04). Mọi thay đổi hành vi phải làm cùng lúc với cập nhật snapshot |
+| `providers/base.py` | 401 | Là contract chung của mọi provider; đổi chữ ký = vỡ cả 3 team. Đã có contract test (R03-T01) |
+| `core/routing/*` | 2.263 | Đã có 79 test đang xanh. R03 **bọc** chứ không viết lại: `application/model_routing/` gọi xuống đây |
+
+## 4. Quy tắc xử lý (bắt buộc)
+
+1. **Không xoá trong cùng PR với refactor.** Xoá code chết là một commit riêng, để `git revert` được độc lập khi có sự cố.
+2. **Cô lập trước, xoá sau.** Đánh dấu module bằng docstring cảnh báo, chạy 1 vòng release; không ai báo lỗi mới xoá.
+3. **Hạng mục 🟥 A cần một người xác nhận** (PO hoặc chủ tính năng) trước khi xoá — trừ khi rõ ràng là bản trùng lặp (`codebase_memory_ui`, `mcp_servers_dialog`).
+4. **Không refactor code trong nhóm 🟥 A.** Nếu một file trong danh sách này >400 dòng, nó **không** tính vào CASAN Check 2 — vì đường đi đúng là xoá, không phải tách nhỏ.
+
+## 5. Việc cần bàn giao
+
+| Hạng mục | Team nhận | EPIC |
+| :--- | :--- | :--- |
+| `security/` trùng lặp với `core/agent_security.py` | 🟣 Nam | R09 |
+| `ui/accounts_tab.py`, `ui/flow_dialog.py`, `ui/mcp_servers_dialog.py` | 🟣 Nam (sở hữu `presentation/shell/`, `settings/`) | R08 |
+| `core/graph_server.py`, `core/codebase_memory_ui.py` | 🟢 Hoa (sở hữu `presentation/graph/`) | R06 |
+
+## 6. Cách chạy lại lần quét này
+
+```bash
+python scripts/check_imports.py       # ranh giới kiến trúc (R01-T03)
+# Bản quét đồ thị import dùng cho tài liệu này sẽ được đóng gói thành
+# scripts/find_dormant.py trong R10-T02 (Testing & Governance tooling).
+```

docs/refactor/Refactoring_Checklist.md

@@ -26,16 +26,16 @@
 * **Team chịu trách nhiệm**: 🔵 **Team Duy** (Chủ trì ADR & Test Doubles) + Phối hợp cả 3 team
 * **Mục tiêu**: Khóa DTO, dựng fakes/test doubles chạy offline không phụ thuộc Qt/mạng, thiết lập script chặn vi phạm kiến trúc.
 
-- [ ] **R01-T01 (Team Duy)**: Viết Architecture ADR định rõ ranh giới các tầng ➔ `docs/architecture/ADR-001-layered-architecture.md`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R01-T02 (Team Duy)**: Xây dựng `FakeProvider` và `FakeToolExecutor` chạy offline từ `providers/base.py` ➔ `tests/fakes/fake_provider.py` & `tests/fakes/fake_tool_executor.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R01-T03 (Team Duy)**: Viết script quét tĩnh chặn code mới trong `domain/` và `application/` import `PySide6` ➔ `scripts/check_imports.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R01-T04 (Team Duy)**: Viết Characterization Tests cho `core/chat_agent.py::run_cowork` ➔ `tests/characterization/test_run_cowork.py`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
-- [ ] **R01-T05 (Team Duy)**: Lập danh mục và phân loại mã nguồn dormant/dead code ➔ `docs/architecture/dormant-code.md`  
-  *Start: `____-__-__ __:__` | End: `____-__-__ __:__`*
+- [x] **R01-T01 (Team Duy)**: Viết Architecture ADR định rõ ranh giới các tầng ➔ `docs/architecture/ADR-001-layered-architecture.md`  
+  *Start: `2026-08-21 09:56` | End: `2026-08-21 10:00`*
+- [x] **R01-T02 (Team Duy)**: Xây dựng `FakeProvider` và `FakeToolExecutor` chạy offline từ `providers/base.py` ➔ `tests/fakes/fake_provider.py` & `tests/fakes/fake_tool_executor.py`  
+  *Start: `2026-08-21 10:00` | End: `2026-08-21 10:02`*
+- [x] **R01-T03 (Team Duy)**: Viết script quét tĩnh chặn code mới trong `domain/` và `application/` import `PySide6` ➔ `scripts/check_imports.py`  
+  *Start: `2026-08-21 09:58` | End: `2026-08-21 10:05`*
+- [x] **R01-T04 (Team Duy)**: Viết Characterization Tests cho `core/chat_agent.py::run_cowork` ➔ `tests/characterization/test_run_cowork.py`  
+  *Start: `2026-08-21 10:02` | End: `2026-08-21 10:04`*
+- [x] **R01-T05 (Team Duy)**: Lập danh mục và phân loại mã nguồn dormant/dead code ➔ `docs/architecture/dormant-code.md`  
+  *Start: `2026-08-21 10:04` | End: `2026-08-21 10:05`*
 
 ---
 

scripts/check_imports.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+"""CASAN Check 3 — Clean Architecture Guard (R01-T03).
+
+Statically walks the AST of every Python file in the pure-Python layers and
+fails when a file imports something the layer is not allowed to depend on.
+
+Why AST instead of ``grep``: a regex over source text cannot tell an import
+apart from the same words appearing inside a docstring, a comment or a string
+literal (this repo has several docstrings that legitimately mention
+``PySide6``). ``ast`` sees only real ``import`` / ``from … import`` nodes, so
+the check has no false positives and needs no ``# noqa`` escape hatches.
+
+Rules enforced (see docs/architecture/ADR-001-layered-architecture.md):
+
+* **I1** ``domain/`` and ``application/`` must be 100% pure Python — no Qt.
+* **I2** ``domain/`` must not import ``application/``, ``infrastructure/``,
+  ``presentation/`` or the legacy ``ui/``.
+* **I3** ``application/`` must not import ``presentation/`` or ``ui/``.
+
+Usage::
+
+    python scripts/check_imports.py            # scan the whole repo
+    python scripts/check_imports.py domain     # scan one layer only
+
+Exit code is 0 when clean and 1 when at least one violation is found, so it
+can be wired straight into CI / ``scripts/run_quality_gate.py`` (R10-T02).
+"""
+from __future__ import annotations
+
+import argparse
+import ast
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Dict, Iterable, List, Sequence, Tuple
+
+# Repository root = parent of this scripts/ folder. Everything below is resolved
+# relative to it so the checker works no matter what the checkout folder is
+# named or which directory the developer runs it from.
+REPO_ROOT = Path(__file__).resolve().parents[1]
+
+# The distribution package name. Absolute imports may be written either as
+# ``from cowork_local.ui import x`` or ``from ui import x`` depending on how the
+# module was reached; we normalise the prefix away so both spellings are caught.
+PACKAGE_NAME = "cowork_local"
+
+# Any import whose first dotted segment is one of these is a GUI toolkit.
+QT_ROOTS = frozenset({"PySide6", "PySide2", "PyQt5", "PyQt6", "shiboken6", "shiboken2"})
+
+# Per-layer rules: layer directory -> top-level package names it may not import.
+# Kept as a plain table so adding a layer later is a one-line change and the
+# rules stay readable next to the ADR they implement.
+LAYER_RULES: Dict[str, frozenset] = {
+    # I1 + I2: domain is the innermost layer and depends on nothing but stdlib.
+    "domain": frozenset({"application", "infrastructure", "presentation", "ui", "core"}),
+    # I1 + I3: application may use domain, but never anything that draws pixels.
+    "application": frozenset({"presentation", "ui"}),
+}
+
+# Directories that are never production code and therefore never scanned.
+SKIP_DIRS = frozenset({".git", "__pycache__", ".pytest_cache", "tests", "build", "dist"})
+
+
+@dataclass(frozen=True)
+class Violation:
+    """One forbidden import, carrying enough context to fix it without grepping."""
+
+    path: Path
+    line: int
+    imported: str
+    rule: str
+
+    def render(self) -> str:
+        """Format as ``file:line: message`` — the shape editors turn into a
+        clickable link, so a CI failure lands the developer on the exact line."""
+        rel = self.path.relative_to(REPO_ROOT).as_posix()
+        # ASCII-only on purpose: this line is printed to a console that may run a
+        # legacy code page (cp932 on the team's Windows boxes), where a non-ASCII
+        # dash raises UnicodeEncodeError and would crash the gate on the very
+        # failure path it exists to report.
+        return f"{rel}:{self.line}: imports '{self.imported}' - {self.rule}"
+
+
+def iter_python_files(layer_dir: Path) -> Iterable[Path]:
+    """Yield every production ``.py`` file under ``layer_dir``.
+
+    Test files are excluded on purpose: a test for a pure-Python service is
+    allowed to import Qt (an integration test may need a headless widget), and
+    holding tests to the production rule would push people to disable the gate.
+    """
+    if not layer_dir.is_dir():
+        return
+    for path in sorted(layer_dir.rglob("*.py")):
+        # Reject a path as soon as ANY of its parent folder names is skippable,
+        # which also covers nested __pycache__ inside a sub-package.
+        if any(part in SKIP_DIRS for part in path.parts):
+            continue
+        yield path
+
+
+def module_parts(path: Path) -> List[str]:
+    """Dotted package path of ``path`` relative to the repo root, as a list.
+
+    ``domain/agents/agent_event.py``   -> ``["domain", "agents", "agent_event"]``
+    ``domain/agents/__init__.py``      -> ``["domain", "agents"]``
+
+    Needed to resolve *relative* imports: ``from ..models import X`` inside
+    ``domain/agents/foo.py`` really means ``domain.models``, and only the file's
+    own position tells us that.
+    """
+    rel = path.relative_to(REPO_ROOT)
+    parts = list(rel.parts)
+    if parts[-1] == "__init__.py":
+        parts.pop()
+    else:
+        parts[-1] = parts[-1][: -len(".py")]
+    return parts
+
+
+def resolve_relative(parts: Sequence[str], level: int, module: str) -> str:
+    """Turn a relative import into the absolute top-level package it points at.
+
+    ``level`` is the number of leading dots. Level 1 means "the package this
+    module lives in", so we drop the module's own name plus ``level - 1``
+    further parents. Returns the FIRST segment of the resolved path, because
+    the rules are expressed in terms of top-level layers.
+
+    Walking off the top of the tree (more dots than there are parents) yields
+    an empty string, which simply never matches a rule — a malformed import
+    like that is a syntax/packaging problem, not an architecture violation.
+    """
+    base = list(parts[:-1])  # the package containing this module
+    if level > 1:
+        drop = level - 1
+        if drop > len(base):
+            return ""
+        base = base[: len(base) - drop]
+    tail = module.split(".") if module else []
+    resolved = base + tail
+    return resolved[0] if resolved else ""
+
+
+def top_level(name: str) -> str:
+    """First dotted segment of an absolute import, with the distribution package
+    prefix stripped so ``cowork_local.ui.chat_panel`` and ``ui.chat_panel`` are
+    treated as the same dependency."""
+    segments = name.split(".")
+    if segments and segments[0] == PACKAGE_NAME:
+        segments = segments[1:]
+    return segments[0] if segments else ""
+
+
+def imported_roots(tree: ast.AST, parts: Sequence[str]) -> Iterable[Tuple[str, int, str]]:
+    """Yield ``(top_level_package, line_number, as_written)`` for every import.
+
+    ``as_written`` is kept so the error message shows what the developer
+    actually typed rather than the normalised root, which makes the violation
+    obvious at a glance.
+
+    ``ast.walk`` (not just the module body) is deliberate: this repo defers many
+    heavy imports into function bodies to keep app start-up fast, and a
+    function-local ``from PySide6 import QtWidgets`` breaks the layer exactly
+    the same way a top-level one does.
+    """
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                yield top_level(alias.name), node.lineno, alias.name
+        elif isinstance(node, ast.ImportFrom):
+            if node.level:
+                written = "." * node.level + (node.module or "")
+                yield resolve_relative(parts, node.level, node.module or ""), node.lineno, written
+            else:
+                module = node.module or ""
+                yield top_level(module), node.lineno, module
+
+
+def check_file(path: Path, layer: str, banned: frozenset) -> List[Violation]:
+    """Collect every rule violation in one file.
+
+    A file that cannot be parsed is reported as a violation rather than skipped:
+    silently passing a file the checker could not read would make the gate lie.
+    """
+    try:
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    except (SyntaxError, UnicodeDecodeError) as exc:
+        return [Violation(path, getattr(exc, "lineno", 0) or 0, "<unparseable>",
+                          f"cannot be parsed by the architecture guard ({exc})")]
+
+    parts = module_parts(path)
+    out: List[Violation] = []
+    for root, lineno, written in imported_roots(tree, parts):
+        if root in QT_ROOTS:
+            out.append(Violation(path, lineno, written,
+                                 f"'{layer}/' must be 100% pure Python (ADR-001 I1)"))
+        elif root in banned:
+            out.append(Violation(path, lineno, written,
+                                 f"'{layer}/' must not depend on '{root}/' (ADR-001 I2/I3)"))
+    return out
+
+
+def run(layers: Sequence[str]) -> List[Violation]:
+    """Scan the requested layers and return every violation found, in file order."""
+    found: List[Violation] = []
+    for layer in layers:
+        banned = LAYER_RULES[layer]
+        for path in iter_python_files(REPO_ROOT / layer):
+            found.extend(check_file(path, layer, banned))
+    return found
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="CASAN Check 3 - Clean Architecture Guard (see ADR-001).")
+    parser.add_argument(
+        "layers", nargs="*", choices=sorted(LAYER_RULES) or None, default=None,
+        help="Layers to scan (default: every layer with a rule).",
+    )
+    args = parser.parse_args(argv)
+    layers = args.layers or sorted(LAYER_RULES)
+
+    violations = run(layers)
+    scanned = sum(1 for layer in layers for _ in iter_python_files(REPO_ROOT / layer))
+
+    if violations:
+        print(f"FAIL - {len(violations)} architecture violation(s) in {scanned} file(s):\n")
+        for v in violations:
+            print("  " + v.render())
+        # Point at the rationale instead of just the rule id, so someone hitting
+        # this for the first time knows where the decision was made.
+        print("\nSee docs/architecture/ADR-001-layered-architecture.md")
+        return 1
+
+    print(f"PASS - 0 Qt imports in {', '.join(layers)} ({scanned} file(s) scanned)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/characterization/__init__.py

@@ -0,0 +1,11 @@
+"""Characterization tests: pin the CURRENT behaviour of legacy code (R01-T04).
+
+These are not specifications of what the code *should* do - they are a snapshot
+of what it *does* today, written before the refactor so that any behavioural
+drift introduced while moving logic into ``application/`` shows up as a failing
+test rather than as a bug report from a user.
+
+Rule for this folder: when a test here fails during the refactor, do not "fix"
+the test first. Decide deliberately whether the behaviour change is intended,
+and only then update the snapshot in the same commit as the change.
+"""

tests/characterization/test_run_cowork.py

@@ -0,0 +1,288 @@
+"""Characterization snapshot of ``core.chat_agent.run_cowork`` (R01-T04).
+
+``run_cowork`` is the turn engine every Cowork surface funnels through (chat tab,
+Co4E flow steps, Schedule Task runs). EPIC R04 moves its orchestration into
+``application/conversations/conversation_application_service.py``; these tests
+lock down the observable contract BEFORE that move so the new service can be
+proven equivalent:
+
+* which system prompt ends up in ``messages``
+* which tools are advertised to the provider
+* the exact ``emit`` event sequence for a plain turn and for a tool turn
+* that ``save_file`` produces a real file in the turn's output folder
+* that ``cancel`` stops the loop without calling the provider
+
+Everything runs offline: :class:`FakeProvider` replaces the network and the two
+disk-backed prompt sources (skills, security rules) are stubbed to empty so the
+snapshot does not depend on the developer's own ``~/.cowork_local`` contents.
+"""
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any, Dict, List
+
+import pytest
+
+from cowork_local.core import chat_agent
+from tests.fakes import FakeProvider, FakeToolExecutor, ScriptedTurn
+
+
+@pytest.fixture
+def isolated_agent(monkeypatch, tmp_path: Path):
+    """Neutralise every ambient input ``run_cowork`` reads from the machine.
+
+    Without this the snapshot would silently depend on whichever skills and
+    security rules the developer happens to have enabled locally, and on the
+    real audit log under ``~/.cowork_local`` - the test would then pass on one
+    laptop and fail on another for reasons unrelated to the code under test.
+    """
+    monkeypatch.setattr(chat_agent, "active_skills_text", lambda: "")
+    monkeypatch.setattr(chat_agent, "load_rules", lambda: "")
+    # audit_log is imported lazily inside run_cowork, so patch the module's own
+    # target directory rather than the name chat_agent sees.
+    from cowork_local.core import audit_log
+
+    monkeypatch.setattr(audit_log, "AUDIT_DIR", tmp_path / "audit")
+    return tmp_path
+
+
+def _run(provider, messages, out_dir: Path, **kwargs):
+    """Run one turn and return ``(returned_messages, emitted_events)``."""
+    events: List[Dict[str, Any]] = []
+    result = chat_agent.run_cowork(provider, messages, out_dir, events.append, **kwargs)
+    return result, events
+
+
+def _types(events: List[Dict[str, Any]]) -> List[str]:
+    """Event ``type`` values in order - the shape assertions read on."""
+    return [e.get("type") for e in events]
+
+
+# --------------------------------------------------------------------------- #
+# A plain answer with no tool calls
+# --------------------------------------------------------------------------- #
+def test_plain_turn_streams_text_and_appends_assistant_message(isolated_agent):
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([ScriptedTurn(text="Hello there.")])
+    messages: List[Dict[str, Any]] = [{"role": "user", "content": "hi"}]
+
+    result, events = _run(provider, messages, out_dir)
+
+    # The loop ends as soon as the model stops calling tools: exactly one call.
+    assert provider.call_count == 1
+    # run_cowork mutates and returns the SAME list the caller passed in - callers
+    # (ui/cowork_tab.py::build_job) rely on this to persist conversation history.
+    assert result is messages
+    assert result[-1]["role"] == "assistant"
+    assert result[-1]["content"] == "Hello there."
+    assert _types(events) == ["text", "assistant_done"]
+    assert events[0]["delta"] == "Hello there."
+    assert events[-1]["content"] == "Hello there."
+
+
+def test_system_prompt_is_inserted_once_at_the_front(isolated_agent):
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([ScriptedTurn(text="ok")])
+    messages: List[Dict[str, Any]] = [{"role": "user", "content": "hi"}]
+
+    result, _ = _run(provider, messages, out_dir)
+
+    assert result[0]["role"] == "system"
+    assert result[0]["content"].startswith("You are Cowork Local")
+    # Exactly one system message: a second turn on the same conversation must not
+    # stack another copy of the prompt (that would grow the context every turn).
+    assert sum(1 for m in result if m.get("role") == "system") == 1
+
+
+def test_caller_supplied_system_prompt_is_preserved(isolated_agent):
+    """A caller that already put a system message first keeps its own prompt.
+
+    Co4E flow steps depend on this to give a step its own persona instead of the
+    generic Cowork prompt.
+    """
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([ScriptedTurn(text="ok")])
+    messages: List[Dict[str, Any]] = [
+        {"role": "system", "content": "CUSTOM PERSONA"},
+        {"role": "user", "content": "hi"},
+    ]
+
+    result, _ = _run(provider, messages, out_dir)
+
+    assert result[0]["content"] == "CUSTOM PERSONA"
+
+
+def test_reasoning_is_emitted_separately_and_never_joins_the_answer(isolated_agent):
+    """Reasoning drives the "Thinking" indicator only - it must not become part
+    of the assistant's content, otherwise a reasoning model's private chain of
+    thought would be persisted into conversation history."""
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([ScriptedTurn(text="42", reasoning="let me think...")])
+
+    result, events = _run(provider, [{"role": "user", "content": "q"}], out_dir)
+
+    assert _types(events) == ["reasoning", "text", "assistant_done"]
+    assert result[-1]["content"] == "42"
+    assert "let me think" not in result[-1]["content"]
+
+
+def test_reasoning_only_reply_gets_a_placeholder_answer(isolated_agent):
+    """A model that returns only reasoning must not end the turn on a blank
+    bubble - headless callers (Schedule Task) read this content back as the
+    run's final answer and would otherwise write "(no output)"."""
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([ScriptedTurn(text="", reasoning="thinking")])
+
+    result, events = _run(provider, [{"role": "user", "content": "q"}], out_dir)
+
+    assert result[-1]["content"].startswith("*(model returned only its reasoning")
+    assert "text" in _types(events)
+
+
+# --------------------------------------------------------------------------- #
+# Tool advertising
+# --------------------------------------------------------------------------- #
+def test_save_file_and_update_plan_are_always_advertised(isolated_agent):
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([ScriptedTurn(text="ok")])
+
+    _run(provider, [{"role": "user", "content": "hi"}], out_dir)
+
+    advertised = provider.calls[0].tool_names
+    assert "save_file" in advertised
+    assert "update_plan" in advertised
+
+
+def test_allowed_tools_scopes_the_catalogue_but_keeps_update_plan(isolated_agent):
+    """``allowed_tools`` is the permission scope Co4E steps use: a read-only step
+    must literally not be offered a writing tool. ``update_plan`` survives the
+    filter because it has no side effects."""
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([ScriptedTurn(text="ok")])
+
+    _run(provider, [{"role": "user", "content": "hi"}], out_dir,
+         allowed_tools=["read_file"])
+
+    advertised = set(provider.calls[0].tool_names)
+    assert "save_file" not in advertised
+    assert "update_plan" in advertised
+
+
+def test_extra_tools_are_advertised_alongside_built_ins(isolated_agent):
+    out_dir = isolated_agent / "out"
+    executor = FakeToolExecutor(results={"ms365_send_mail": {"output": "sent"}})
+    provider = FakeProvider([ScriptedTurn(text="ok")])
+
+    _run(provider, [{"role": "user", "content": "hi"}], out_dir,
+         extra_tools=executor.specs(), extra_executor=executor)
+
+    assert "ms365_send_mail" in provider.calls[0].tool_names
+
+
+# --------------------------------------------------------------------------- #
+# Tool execution
+# --------------------------------------------------------------------------- #
+def test_save_file_writes_a_real_file_and_reports_it(isolated_agent):
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([
+        ScriptedTurn(tool_calls=[("save_file", {"filename": "note.md",
+                                                "content": "# Result\n"})]),
+        ScriptedTurn(text="Done."),
+    ])
+
+    result, events = _run(provider, [{"role": "user", "content": "make a note"}], out_dir)
+
+    written = [p for p in out_dir.iterdir() if p.is_file()]
+    assert len(written) == 1
+    assert written[0].read_text(encoding="utf-8") == "# Result\n"
+
+    assert _types(events) == [
+        "assistant_done",       # first turn: tool call only, no visible text
+        "tool_proposed",        # the diff preview shown in the chat
+        "tool_result",
+        "text",                 # second turn's answer
+        "assistant_done",
+    ]
+    assert events[2]["ok"] is True
+
+    # The tool result is fed back as a `tool` message so the model can react to it.
+    roles = [m["role"] for m in result]
+    assert roles == ["system", "user", "assistant", "tool", "assistant"]
+    assert result[3]["name"] == "save_file"
+
+
+def test_extra_tool_calls_are_routed_to_the_extra_executor(isolated_agent):
+    """MCP / Microsoft 365 tools bypass the built-in file+command handlers and go
+    to the caller-supplied executor instead."""
+    out_dir = isolated_agent / "out"
+    executor = FakeToolExecutor(results={"ms365_send_mail": {"ok": True, "output": "sent"}})
+    provider = FakeProvider([
+        ScriptedTurn(tool_calls=[("ms365_send_mail", {"to": "a@b.c"})]),
+        ScriptedTurn(text="Mail sent."),
+    ])
+
+    result, events = _run(provider, [{"role": "user", "content": "mail them"}], out_dir,
+                          extra_tools=executor.specs(), extra_executor=executor)
+
+    assert executor.call_names == ["ms365_send_mail"]
+    assert executor.args_for("ms365_send_mail") == [{"to": "a@b.c"}]
+    assert [e for e in events if e["type"] == "tool_result"][0]["output"] == "sent"
+    assert result[3] == {"role": "tool", "tool_call_id": result[3]["tool_call_id"],
+                         "name": "ms365_send_mail", "content": "sent"}
+
+
+def test_update_plan_drives_the_plan_panel_without_producing_a_file(isolated_agent):
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([
+        ScriptedTurn(tool_calls=[("update_plan", {"steps": [{"title": "step one"}]})]),
+        ScriptedTurn(text="Planned."),
+    ])
+
+    result, events = _run(provider, [{"role": "user", "content": "plan it"}], out_dir)
+
+    plan_events = [e for e in events if e["type"] == "plan_set"]
+    assert len(plan_events) == 1
+    assert plan_events[0]["steps"]
+    # No tool_proposed/tool_result bubbles for a plan update, and no file on disk.
+    assert "tool_proposed" not in _types(events)
+    assert list(out_dir.iterdir()) == []
+    assert result[3]["content"] == "Plan updated."
+
+
+# --------------------------------------------------------------------------- #
+# Cancellation
+# --------------------------------------------------------------------------- #
+def test_cancel_before_the_first_step_never_calls_the_provider(isolated_agent):
+    """Stop pressed before the loop starts must cost zero tokens."""
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([], strict=True)
+
+    result, events = _run(provider, [{"role": "user", "content": "hi"}], out_dir,
+                          cancel=lambda: True)
+
+    assert provider.call_count == 0
+    assert _types(events) == []
+    # The system prompt is still installed, so the conversation stays well-formed
+    # for a later retry on the same message list.
+    assert result[0]["role"] == "system"
+
+
+def test_cancel_between_steps_stops_before_the_next_provider_call(isolated_agent):
+    """After a tool call runs, a Stop must end the turn instead of paying for
+    another round trip."""
+    out_dir = isolated_agent / "out"
+    provider = FakeProvider([
+        ScriptedTurn(tool_calls=[("save_file", {"filename": "a.md", "content": "x"})]),
+    ])
+    calls = {"n": 0}
+
+    def cancel() -> bool:
+        # False on the first check (loop entry), True afterwards - i.e. the user
+        # pressed Stop while the first step was running.
+        calls["n"] += 1
+        return calls["n"] > 1
+
+    result, _ = _run(provider, [{"role": "user", "content": "hi"}], out_dir, cancel=cancel)
+
+    assert provider.call_count == 1
+    assert result[-1]["role"] in {"assistant", "tool"}

tests/conftest.py

@@ -0,0 +1,63 @@
+"""Root pytest configuration: bind ``cowork_local`` to THIS checkout (R01-T02).
+
+Why this file exists
+--------------------
+The package directory is itself the distribution package (``__init__.py`` sits
+at the repo root), so ``import cowork_local`` only resolves when the checkout
+folder happens to be named exactly ``cowork_local``. It frequently is not — this
+one is checked out as ``cowork_local_gitea``, and developers keep several dated
+copies side by side (``cowork_local``, ``cowork_local_20260722``, ...).
+
+Left alone, ``sys.path``-based discovery would import whichever *sibling* folder
+is named ``cowork_local`` and the whole suite would silently test a DIFFERENT
+checkout: green here, broken in the branch under review. That is the worst kind
+of test failure, because it fails to fail.
+
+So instead of relying on the folder name, we load ``__init__.py`` by absolute
+path and register the result in ``sys.modules`` under the canonical name before
+any test imports it. Submodules (``cowork_local.providers.base``, ...) then
+resolve through this package's own ``__path__``, i.e. always this checkout.
+"""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+# .../<checkout>/tests/conftest.py -> .../<checkout>
+_PKG_DIR = Path(__file__).resolve().parents[1]
+_PKG_NAME = "cowork_local"
+
+
+def _bind_package_to_this_checkout() -> None:
+    """Make ``import cowork_local`` mean this directory, whatever it is named.
+
+    A no-op when the correct package object is already bound, so running the
+    suite from a folder that IS named ``cowork_local`` costs nothing and the
+    hook stays idempotent across repeated conftest collection.
+    """
+    existing = sys.modules.get(_PKG_NAME)
+    existing_file = getattr(existing, "__file__", None)
+    if existing_file and Path(existing_file).resolve().parent == _PKG_DIR:
+        return  # already the right one
+
+    spec = importlib.util.spec_from_file_location(
+        _PKG_NAME,
+        _PKG_DIR / "__init__.py",
+        # Setting the search locations is what makes dotted submodule imports
+        # (cowork_local.core.*, cowork_local.providers.*) resolve inside THIS
+        # directory rather than through sys.path.
+        submodule_search_locations=[str(_PKG_DIR)],
+    )
+    if spec is None or spec.loader is None:  # pragma: no cover - packaging error
+        raise RuntimeError(f"cannot load {_PKG_NAME} from {_PKG_DIR}")
+
+    module = importlib.util.module_from_spec(spec)
+    # Registered BEFORE exec_module so that a self-referential import inside
+    # __init__.py would find the partially-initialised module instead of
+    # recursing - the same protocol CPython's own import machinery follows.
+    sys.modules[_PKG_NAME] = module
+    spec.loader.exec_module(module)
+
+
+_bind_package_to_this_checkout()

tests/fakes/__init__.py

@@ -0,0 +1,16 @@
+"""Offline test doubles for the refactoring safety net (R01-T02).
+
+Every double here is deliberately Qt-free, network-free and disk-free so the
+unit/contract suites run in well under a second and give the same answer on a
+laptop, in CI and on a machine with no API keys configured.
+
+* :class:`~tests.fakes.fake_provider.FakeProvider` - a scripted
+  ``providers.base.Provider`` that streams canned text/tool calls.
+* :class:`~tests.fakes.fake_tool_executor.FakeToolExecutor` - a scripted stand-in
+  for the ``extra_executor`` callable that ``core.chat_agent.run_cowork`` routes
+  MCP/connector tool calls to.
+"""
+from .fake_provider import FakeProvider, ScriptedTurn
+from .fake_tool_executor import FakeToolExecutor, ToolInvocation
+
+__all__ = ["FakeProvider", "ScriptedTurn", "FakeToolExecutor", "ToolInvocation"]

tests/fakes/fake_provider.py

@@ -0,0 +1,213 @@
+"""FakeProvider - a scripted, offline stand-in for a real LLM provider (R01-T02).
+
+The real providers (``providers/openai_compat.py``, ``providers/anthropic.py``)
+open HTTP connections, need API keys and stream at the mercy of the network, so
+nothing above them could be tested deterministically. This double implements the
+same :class:`providers.base.Provider` contract from a list of scripted turns:
+
+    provider = FakeProvider([
+        ScriptedTurn(tool_calls=[("save_file", {"filename": "a.md", "content": "hi"})]),
+        ScriptedTurn(text="Saved it."),
+    ])
+
+Turn 1 asks the agent loop to call a tool, turn 2 ends the loop with plain text -
+exactly the two-step shape ``run_cowork`` exercises, with zero I/O.
+
+It records every call it received (:attr:`FakeProvider.calls`) so a test can
+assert on what the layer above actually sent (message list, tool catalogue),
+which is how the characterization and contract suites pin current behaviour.
+"""
+from __future__ import annotations
+
+import itertools
+from dataclasses import dataclass
+from typing import Any, Dict, List, Optional, Sequence, Tuple
+
+from cowork_local.providers.base import (
+    CancelFn,
+    Provider,
+    ProviderError,
+    TextCallback,
+    ToolSpec,
+)
+
+# One scripted tool call: (name, arguments). Ids are generated by the provider so
+# a test never has to invent them, mirroring what a real gateway does.
+ToolCallScript = Tuple[str, Dict[str, Any]]
+
+
+@dataclass(frozen=True)
+class ScriptedTurn:
+    """What :class:`FakeProvider` should do for ONE ``chat()`` call.
+
+    ``text`` is streamed through ``on_text`` and returned as the assistant
+    message content. ``reasoning`` goes to ``on_reasoning`` only - it must never
+    leak into the answer, and asserting that is one of this double's jobs.
+
+    ``tool_calls`` makes the agent loop run tools and come back for another turn;
+    an empty tuple ends the loop.
+
+    ``error``, when set, raises :class:`ProviderError` instead of answering, so
+    error/recovery paths are testable without simulating a network fault.
+
+    ``chunk_size`` > 0 splits ``text`` into fixed-size pieces to exercise
+    chunk-boundary handling in stream consumers (the ``<think>`` splitter and the
+    UI's incremental markdown renderer both have boundary logic worth covering).
+    """
+
+    text: str = ""
+    reasoning: str = ""
+    tool_calls: Sequence[ToolCallScript] = ()
+    error: Optional[str] = None
+    chunk_size: int = 0
+
+
+@dataclass
+class RecordedCall:
+    """A snapshot of one ``chat()`` invocation, for assertions after the fact."""
+
+    messages: List[Dict[str, Any]]
+    tool_names: List[str]
+    cancelled: bool = False
+
+
+class FakeProvider(Provider):
+    """A ``Provider`` that replays :class:`ScriptedTurn` objects.
+
+    Args:
+        turns: the scripted turns, consumed in order.
+        model: the model id reported through ``describe()`` / usage records.
+        models: what :meth:`list_models` returns (Settings' "Load models").
+        strict: when True (default) running past the end of the script raises
+            ``AssertionError``. That is intentional noise: a silent extra turn
+            usually means the code under test looped more than the test author
+            expected, and hiding it behind an empty answer would turn a real
+            behaviour change into a passing test.
+    """
+
+    name = "fake"
+    # The double can accept image content blocks, so vision code paths are
+    # reachable in tests without a real vision-capable gateway.
+    supports_vision = True
+
+    def __init__(
+        self,
+        turns: Optional[Sequence[ScriptedTurn]] = None,
+        *,
+        model: str = "fake-model",
+        models: Optional[Sequence[str]] = None,
+        strict: bool = True,
+        conf: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        super().__init__(dict(conf or {}, model=model))
+        self._turns: List[ScriptedTurn] = list(turns or [])
+        self._models = list(models or [model])
+        self._strict = strict
+        self._ids = itertools.count(1)  # deterministic tool-call ids: call_1, call_2, ...
+        self.calls: List[RecordedCall] = []
+
+    # -- introspection helpers used by tests ---------------------------- #
+    @property
+    def call_count(self) -> int:
+        """How many times the layer above asked this provider to run a turn."""
+        return len(self.calls)
+
+    @property
+    def remaining_turns(self) -> int:
+        """Scripted turns not consumed yet - assert 0 to prove the script was
+        fully used (an unused turn means the code stopped earlier than intended)."""
+        return len(self._turns)
+
+    def last_messages(self) -> List[Dict[str, Any]]:
+        """The message list sent on the most recent call (empty if never called)."""
+        return self.calls[-1].messages if self.calls else []
+
+    # -- Provider contract ---------------------------------------------- #
+    def chat(
+        self,
+        messages: List[Dict[str, Any]],
+        tools: Optional[List[ToolSpec]] = None,
+        on_text: Optional[TextCallback] = None,
+        cancel: Optional[CancelFn] = None,
+        on_reasoning: Optional[TextCallback] = None,
+    ) -> Dict[str, Any]:
+        """Replay the next scripted turn, honouring cancel and both callbacks.
+
+        The message list is deep-ish copied into the recording because the agent
+        loop keeps appending to the SAME list object; without the copy every
+        recorded call would show the final state and assertions on "what was
+        sent at step 1" would be meaningless.
+        """
+        record = RecordedCall(
+            messages=[dict(m) for m in messages],
+            tool_names=[t.name for t in (tools or [])],
+        )
+        self.calls.append(record)
+
+        turn = self._next_turn()
+
+        # Checked before streaming anything: a provider that already knows the
+        # caller gave up must not spend callbacks on text nobody will render.
+        if self._is_cancelled(cancel):
+            record.cancelled = True
+            return {"role": "assistant", "content": "", "tool_calls": []}
+
+        if turn.error:
+            raise ProviderError(turn.error)
+
+        if turn.reasoning and on_reasoning:
+            on_reasoning(turn.reasoning)
+
+        for piece in self._stream_pieces(turn):
+            # Re-checked between chunks so a mid-stream Stop truncates the answer
+            # the same way a real streamed response does.
+            if self._is_cancelled(cancel):
+                record.cancelled = True
+                break
+            if on_text:
+                on_text(piece)
+
+        return {
+            "role": "assistant",
+            "content": turn.text,
+            "tool_calls": [
+                {"id": f"call_{next(self._ids)}", "name": name, "arguments": dict(args)}
+                for name, args in turn.tool_calls
+            ],
+        }
+
+    def list_models(self) -> List[str]:
+        """Configured model ids. Clears ``last_error`` so ``test_connection()``
+        reports success, matching how a healthy real provider behaves."""
+        self.last_error = ""
+        return list(self._models)
+
+    # -- internals ------------------------------------------------------- #
+    def _next_turn(self) -> ScriptedTurn:
+        """Pop the next scripted turn, or fail loudly when the script ran out."""
+        if self._turns:
+            return self._turns.pop(0)
+        if self._strict:
+            raise AssertionError(
+                f"FakeProvider script exhausted: chat() was called {len(self.calls)} "
+                "time(s) but fewer turns were scripted. Add a ScriptedTurn, or pass "
+                "strict=False if the extra call is genuinely expected."
+            )
+        return ScriptedTurn()
+
+    @staticmethod
+    def _stream_pieces(turn: ScriptedTurn) -> List[str]:
+        """Split a turn's answer into the fragments to stream.
+
+        ``chunk_size == 0`` streams the whole answer in one piece (the common
+        case); a positive size slices it so tests can drive chunk-boundary logic.
+        """
+        if not turn.text:
+            return []
+        if turn.chunk_size <= 0:
+            return [turn.text]
+        size = turn.chunk_size
+        return [turn.text[i:i + size] for i in range(0, len(turn.text), size)]
+
+
+__all__ = ["FakeProvider", "ScriptedTurn", "RecordedCall"]

tests/fakes/fake_tool_executor.py

@@ -0,0 +1,99 @@
+"""FakeToolExecutor - offline stand-in for the extra-tool executor (R01-T02).
+
+``core.chat_agent.run_cowork`` routes any tool call whose name appears in
+``extra_tools`` to ``extra_executor(name, args)`` and expects back::
+
+    {"ok": bool, "output": str}
+
+In production that callable reaches MCP servers, Microsoft 365 connectors and
+subprocesses. This double answers from a table instead, so the agent loop's tool
+branch is testable with no processes, no sockets and no credentials - and every
+invocation is recorded for assertions about what the agent actually asked for.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Callable, Dict, List, Optional, Union
+
+from cowork_local.providers.base import ToolSpec
+
+# A scripted answer is either the literal result dict, or a callable computing it
+# from the arguments (for tools whose output must depend on the input).
+ToolResult = Dict[str, Any]
+ScriptedResult = Union[ToolResult, Callable[[Dict[str, Any]], ToolResult]]
+
+
+@dataclass(frozen=True)
+class ToolInvocation:
+    """One recorded ``extra_executor(name, args)`` call."""
+
+    name: str
+    args: Dict[str, Any]
+
+
+@dataclass
+class FakeToolExecutor:
+    """Callable test double for ``run_cowork(extra_executor=...)``.
+
+    Args:
+        results: tool name -> scripted result (dict, or callable taking args).
+        default: what to answer for a tool with no scripted result. ``None``
+            (the default) answers with ``ok=False`` and an explicit message
+            rather than raising - the production executor also reports unknown
+            tools as a failed tool result, and matching that keeps the agent
+            loop on its real code path instead of an exception path it would
+            never take in production.
+    """
+
+    results: Dict[str, ScriptedResult] = field(default_factory=dict)
+    default: Optional[ScriptedResult] = None
+    calls: List[ToolInvocation] = field(default_factory=list)
+
+    def __call__(self, name: str, args: Dict[str, Any]) -> ToolResult:
+        """Record the invocation and return its scripted result."""
+        self.calls.append(ToolInvocation(name=name, args=dict(args or {})))
+        scripted = self.results.get(name, self.default)
+        if scripted is None:
+            return {"ok": False, "output": f"No fake result scripted for tool '{name}'."}
+        # A callable lets one entry serve many different arguments (e.g. echo the
+        # path it was asked to read) without scripting every combination.
+        resolved = scripted(dict(args or {})) if callable(scripted) else dict(scripted)
+        resolved.setdefault("ok", True)
+        resolved.setdefault("output", "")
+        return resolved
+
+    # -- introspection helpers used by tests ---------------------------- #
+    @property
+    def call_names(self) -> List[str]:
+        """Tool names in call order - the usual thing a test asserts on."""
+        return [c.name for c in self.calls]
+
+    def called(self, name: str) -> bool:
+        """True when ``name`` was invoked at least once."""
+        return any(c.name == name for c in self.calls)
+
+    def args_for(self, name: str) -> List[Dict[str, Any]]:
+        """Every argument dict this tool was called with, in order."""
+        return [c.args for c in self.calls if c.name == name]
+
+    def specs(self) -> List[ToolSpec]:
+        """``ToolSpec`` entries for the scripted tools, ready to pass as
+        ``run_cowork(extra_tools=...)``.
+
+        The agent loop dispatches to ``extra_executor`` only for names present in
+        ``extra_tools``; generating the specs from the same table removes the
+        chance of a test scripting a result the loop can never reach.
+        """
+        return [
+            ToolSpec(
+                name=name,
+                description=f"Fake tool '{name}' (test double).",
+                # Permissive schema on purpose: these specs exist to register the
+                # name with the agent loop, not to validate arguments.
+                parameters={"type": "object", "properties": {}, "additionalProperties": True},
+            )
+            for name in self.results
+        ]
+
+
+__all__ = ["FakeToolExecutor", "ToolInvocation"]

tests/unit/__init__.py

@@ -0,0 +1,5 @@
+"""Fast, isolated unit tests for the new 4-tier layers (R01/R03/R04, R10-T01).
+
+Everything in this folder must run offline, without Qt and without touching the
+real user config directory, so the whole folder stays well under one second.
+"""

tests/unit/test_check_imports.py

@@ -0,0 +1,194 @@
+"""Unit tests for the Clean Architecture Guard, ``scripts/check_imports.py`` (R01-T03).
+
+The guard is what makes ADR-001 enforceable rather than aspirational, so it needs
+its own tests: a guard that silently passes everything is worse than no guard,
+because the CASAN Gate would then report a green architecture that isn't.
+
+Both directions are covered - it must FLAG real violations (including the
+function-local and relative import spellings this codebase actually uses) and it
+must NOT flag legal code (Qt named only in a docstring, domain importing stdlib).
+"""
+from __future__ import annotations
+
+import importlib.util
+import sys
+from pathlib import Path
+
+import pytest
+
+_GUARD_PATH = Path(__file__).resolve().parents[2] / "scripts" / "check_imports.py"
+
+
+def _load_guard():
+    """Import ``scripts/check_imports.py`` by path.
+
+    ``scripts/`` is deliberately not a package (it holds standalone CLI tools),
+    so a normal import statement cannot reach it.
+    """
+    name = "_check_imports_under_test"
+    spec = importlib.util.spec_from_file_location(name, _GUARD_PATH)
+    module = importlib.util.module_from_spec(spec)
+    # Registered before exec_module because @dataclass resolves a class's own
+    # module out of sys.modules while processing annotations; without this the
+    # guard's Violation dataclass fails to build under a by-path import.
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+guard = _load_guard()
+
+
+@pytest.fixture
+def fake_repo(tmp_path: Path, monkeypatch):
+    """A throwaway repo root the guard scans instead of the real one.
+
+    Pointing ``REPO_ROOT`` at a tmp dir keeps these tests independent of the
+    actual state of ``domain/`` and ``application/`` - otherwise adding a real
+    module later could flip a guard test red for no reason.
+    """
+    monkeypatch.setattr(guard, "REPO_ROOT", tmp_path)
+    return tmp_path
+
+
+def _write(root: Path, rel: str, source: str) -> Path:
+    path = root / rel
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(source, encoding="utf-8")
+    return path
+
+
+# --------------------------------------------------------------------------- #
+# Violations that must be caught
+# --------------------------------------------------------------------------- #
+def test_top_level_qt_import_in_domain_is_flagged(fake_repo):
+    _write(fake_repo, "domain/agents/bad.py", "from PySide6 import QtWidgets\n")
+
+    violations = guard.run(["domain"])
+
+    assert len(violations) == 1
+    assert "PySide6" in violations[0].imported
+    assert "pure Python" in violations[0].rule
+
+
+def test_function_local_qt_import_is_flagged(fake_repo):
+    """This repo defers heavy imports into function bodies to speed up start-up,
+    so the guard walks the whole tree - a deferred Qt import breaks the layer
+    exactly as much as a top-level one."""
+    _write(fake_repo, "application/conversations/bad.py",
+           "def build():\n    import PySide6.QtCore\n    return PySide6\n")
+
+    violations = guard.run(["application"])
+
+    assert len(violations) == 1
+    assert violations[0].line == 2
+
+
+def test_application_importing_ui_is_flagged(fake_repo):
+    _write(fake_repo, "application/conversations/bad.py",
+           "from cowork_local.ui.chat_panel import ChatPanel\n")
+
+    violations = guard.run(["application"])
+
+    assert len(violations) == 1
+    assert "ui/" in violations[0].rule
+
+
+def test_relative_import_that_escapes_the_layer_is_flagged(fake_repo):
+    """``from ...ui import x`` inside ``domain/agents/`` resolves to the top-level
+    ``ui`` package. Only relative-import resolution catches this - the text
+    ``ui`` never appears as an absolute module name."""
+    _write(fake_repo, "domain/agents/bad.py", "from ...ui import widgets\n")
+
+    violations = guard.run(["domain"])
+
+    assert len(violations) == 1
+    assert violations[0].imported == "...ui"
+
+
+def test_domain_importing_core_is_flagged(fake_repo):
+    """``domain/`` is the innermost layer: it may not reach back into the legacy
+    ``core/`` package either, or the dependency arrow would point outward."""
+    _write(fake_repo, "domain/models/bad.py", "from cowork_local.core import history\n")
+
+    violations = guard.run(["domain"])
+
+    assert len(violations) == 1
+
+
+def test_unparseable_file_is_reported_rather_than_skipped(fake_repo):
+    """A file the guard cannot read must fail the gate. Skipping it would let a
+    broken file smuggle any import past the check."""
+    _write(fake_repo, "domain/agents/broken.py", "def oops(:\n")
+
+    violations = guard.run(["domain"])
+
+    assert len(violations) == 1
+    assert violations[0].imported == "<unparseable>"
+
+
+# --------------------------------------------------------------------------- #
+# Legal code that must NOT be flagged
+# --------------------------------------------------------------------------- #
+def test_qt_mentioned_only_in_a_docstring_is_not_flagged(fake_repo):
+    """The whole reason the guard parses an AST instead of grepping: several
+    real modules explain in prose that they must not import PySide6."""
+    _write(fake_repo, "domain/agents/ok.py",
+           '"""This layer must never import PySide6 or PyQt6."""\n'
+           'QT = "PySide6"  # a string, not an import\n')
+
+    assert guard.run(["domain"]) == []
+
+
+def test_stdlib_and_intra_layer_imports_are_allowed(fake_repo):
+    _write(fake_repo, "domain/agents/ok.py",
+           "import json\n"
+           "from dataclasses import dataclass\n"
+           "from ..models.provider_descriptor import ProviderDescriptor\n")
+
+    assert guard.run(["domain"]) == []
+
+
+def test_application_may_import_domain_and_infrastructure(fake_repo):
+    """Application orchestrates: reaching down to domain is the point, and
+    wiring an infrastructure adapter is allowed (only UI is forbidden)."""
+    _write(fake_repo, "application/model_routing/ok.py",
+           "from cowork_local.domain.models import provider_descriptor\n"
+           "from cowork_local.infrastructure.providers import provider_registry\n")
+
+    assert guard.run(["application"]) == []
+
+
+def test_tests_folder_inside_a_layer_is_not_scanned(fake_repo):
+    """A test living next to the code may legitimately import Qt; holding tests
+    to the production rule would only teach people to disable the gate."""
+    _write(fake_repo, "domain/tests/test_thing.py", "from PySide6 import QtWidgets\n")
+
+    assert guard.run(["domain"]) == []
+
+
+# --------------------------------------------------------------------------- #
+# Reporting / exit codes - what CI actually consumes
+# --------------------------------------------------------------------------- #
+def test_main_returns_nonzero_and_prints_ascii_only_on_failure(fake_repo, capsys):
+    """The team's Windows consoles run a legacy code page (cp932): a non-ASCII
+    character in the failure output would raise UnicodeEncodeError and crash the
+    gate on the very path it exists to report."""
+    _write(fake_repo, "domain/agents/bad.py", "from PySide6 import QtWidgets\n")
+
+    exit_code = guard.main(["domain"])
+    out = capsys.readouterr().out
+
+    assert exit_code == 1
+    assert "FAIL" in out
+    assert "domain/agents/bad.py:1" in out
+    out.encode("cp932")  # raises if any character is unprintable on the target console
+
+
+def test_main_returns_zero_on_a_clean_tree(fake_repo, capsys):
+    _write(fake_repo, "domain/agents/ok.py", "import json\n")
+
+    exit_code = guard.main(["domain"])
+
+    assert exit_code == 0
+    assert "PASS" in capsys.readouterr().out