From bbc09f628af9c8ae91b362a925ad693b32db84bb Mon Sep 17 00:00:00 2001 From: Anh Tran Nguyen Minh Date: Fri, 21 Aug 2026 10:05:50 +0900 Subject: [PATCH] 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) --- .../ADR-001-layered-architecture.md | 156 ++++++++++ docs/architecture/dormant-code.md | 85 ++++++ docs/refactor/Refactoring_Checklist.md | 20 +- scripts/check_imports.py | 239 +++++++++++++++ tests/characterization/__init__.py | 11 + tests/characterization/test_run_cowork.py | 288 ++++++++++++++++++ tests/conftest.py | 63 ++++ tests/fakes/__init__.py | 16 + tests/fakes/fake_provider.py | 213 +++++++++++++ tests/fakes/fake_tool_executor.py | 99 ++++++ tests/unit/__init__.py | 5 + tests/unit/test_check_imports.py | 194 ++++++++++++ 12 files changed, 1379 insertions(+), 10 deletions(-) create mode 100644 docs/architecture/ADR-001-layered-architecture.md create mode 100644 docs/architecture/dormant-code.md create mode 100644 scripts/check_imports.py create mode 100644 tests/characterization/__init__.py create mode 100644 tests/characterization/test_run_cowork.py create mode 100644 tests/conftest.py create mode 100644 tests/fakes/__init__.py create mode 100644 tests/fakes/fake_provider.py create mode 100644 tests/fakes/fake_tool_executor.py create mode 100644 tests/unit/__init__.py create mode 100644 tests/unit/test_check_imports.py diff --git a/docs/architecture/ADR-001-layered-architecture.md b/docs/architecture/ADR-001-layered-architecture.md new file mode 100644 index 0000000..fff0470 --- /dev/null +++ b/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) diff --git a/docs/architecture/dormant-code.md b/docs/architecture/dormant-code.md new file mode 100644 index 0000000..8001e97 --- /dev/null +++ b/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). +``` diff --git a/docs/refactor/Refactoring_Checklist.md b/docs/refactor/Refactoring_Checklist.md index c7e8c7b..789cdd3 100644 --- a/docs/refactor/Refactoring_Checklist.md +++ b/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`* --- diff --git a/scripts/check_imports.py b/scripts/check_imports.py new file mode 100644 index 0000000..25cf99e --- /dev/null +++ b/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, "", + 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()) diff --git a/tests/characterization/__init__.py b/tests/characterization/__init__.py new file mode 100644 index 0000000..ed78547 --- /dev/null +++ b/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. +""" diff --git a/tests/characterization/test_run_cowork.py b/tests/characterization/test_run_cowork.py new file mode 100644 index 0000000..a706b66 --- /dev/null +++ b/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"} diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..2a2e9ca --- /dev/null +++ b/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 + +# ...//tests/conftest.py -> .../ +_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() diff --git a/tests/fakes/__init__.py b/tests/fakes/__init__.py new file mode 100644 index 0000000..5d83af8 --- /dev/null +++ b/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"] diff --git a/tests/fakes/fake_provider.py b/tests/fakes/fake_provider.py new file mode 100644 index 0000000..ddc35fd --- /dev/null +++ b/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 ```` 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"] diff --git a/tests/fakes/fake_tool_executor.py b/tests/fakes/fake_tool_executor.py new file mode 100644 index 0000000..6b8e42a --- /dev/null +++ b/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"] diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..8f6fe98 --- /dev/null +++ b/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. +""" diff --git a/tests/unit/test_check_imports.py b/tests/unit/test_check_imports.py new file mode 100644 index 0000000..f9970b4 --- /dev/null +++ b/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 == "" + + +# --------------------------------------------------------------------------- # +# 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