diff --git a/.gitea/workflows/ci.yaml b/.gitea/workflows/ci.yaml index beb45be..dce0038 100644 --- a/.gitea/workflows/ci.yaml +++ b/.gitea/workflows/ci.yaml @@ -12,16 +12,29 @@ jobs: test: runs-on: ubuntu-latest timeout-minutes: 15 + defaults: + run: + working-directory: cowork_local + env: + # Tiến trình con của test import `cowork_local` qua đường này. + PYTHONPATH: ${{ github.workspace }} steps: + # Checkout PHẢI nằm trong thư mục tên đúng `cowork_local`. + # Nhiều test characterization sinh tiến trình con chạy + # `python -c "from cowork_local... import ..."`; tiến trình con đó chỉ + # import được khi trên sys.path có một thư mục mang đúng tên gói. Checkout + # vào thư mục tên khác làm 73 test đỏ vì lý do không liên quan tới mã. - name: Check out source uses: actions/checkout@v4 + with: + path: cowork_local - name: Set up Python uses: actions/setup-python@v5 with: python-version: "3.11" cache: pip - cache-dependency-path: requirements-test.txt + cache-dependency-path: cowork_local/requirements-test.txt - name: Install test dependencies run: python -m pip install --disable-pip-version-check -r requirements-test.txt @@ -68,3 +81,10 @@ jobs: else echo "scripts/check_imports.py chưa có — Team Duy viết, hạn 30/08. Bỏ qua." fi + + # Cổng O bổ sung sau đợt đối chiếu AS-IS/TO-BE: ba check trên đều không + # bắt được mã chết (file không ai import vẫn đúng chiều phụ thuộc, vẫn + # sạch credential, vẫn dưới 400 dòng). Đợt đó tìm ra 1.400 dòng mã trùng + # lặp chết lọt qua đúng theo cách này. + - name: "CASAN Check O — module production phải có nơi import" + run: python scripts/check_orphan_modules.py diff --git a/scripts/check_loc.py b/scripts/check_loc.py index 4bb8a03..03eebbd 100644 --- a/scripts/check_loc.py +++ b/scripts/check_loc.py @@ -2,13 +2,27 @@ """Lines-of-Code (LOC) Quality Guard (EPIC R10 - CASAN Gate S). Enforces the Single Responsibility Principle by ensuring that no production -Python file in Clean Architecture packages exceeds the configured limit (400 LOC). +Python file exceeds the configured limit (400 LOC). + +Phạm vi quét là TOÀN BỘ cây mã production, không chỉ bốn gói Clean +Architecture: ``ui/``, ``core/``, ``providers/``, ``security/``, +``mcp_servers/`` và các module nằm thẳng ở thư mục gốc đều được tính. Trước +đợt mở rộng này, 18 file dài hơn 400 dòng (dài nhất 944) vẫn qua cổng chỉ vì +chúng nằm ngoài bốn gói kia. + +18 file đó không thể sửa hết trong một lần, nên chúng nằm trong +``LEGACY_ALLOWANCE`` với trần riêng bằng đúng số dòng hiện tại — một bánh cóc +chỉ quay một chiều: nợ cũ được giữ nguyên nhưng không được phình thêm, và mỗi +lần file co bớt thì cổng in ra lời nhắc hạ con số xuống. """ from __future__ import annotations import argparse +import ast +import io import os import sys +import tokenize from pathlib import Path from typing import List, Tuple @@ -19,10 +33,52 @@ if hasattr(sys.stdout, "reconfigure"): except Exception: pass -# Default target directories strictly subjected to the 400 LOC constraint -DEFAULT_TARGET_DIRS = ["domain", "application", "infrastructure", "presentation"] +# Toàn bộ cây mã production. Bốn gói Clean Architecture là phần cổng này canh +# từ đầu; ``ui``/``core``/``providers``/``security``/``mcp_servers`` được đưa +# vào sau đợt đối chiếu AS-IS/TO-BE — trước đó chúng nằm ngoài tầm quét, nên +# một file 944 dòng vẫn qua cổng chỉ vì nó không nằm trong bốn gói kia. +DEFAULT_TARGET_DIRS = [ + "domain", "application", "infrastructure", "presentation", + "ui", "core", "providers", "security", "mcp_servers", +] DEFAULT_MAX_LINES = 400 +#: Các file ``.py`` nằm thẳng ở thư mục gốc cũng được quét (không đệ quy) — +#: ``app.py``, ``state.py``, ``theme*.py``… đều là mã production. +SCAN_ROOT_MODULES = True + +#: Nợ cũ: file đã dài hơn 400 dòng TỪ TRƯỚC khi cổng mở rộng sang ``ui``/ +#: ``core``/``providers``. Giá trị là số dòng tại thời điểm ghi nhận và đóng +#: vai trò trần riêng của từng file — đây là bánh cóc CHỈ QUAY MỘT CHIỀU: +#: +#: * file vượt quá trần riêng -> cổng đỏ (đang làm nợ cũ tệ thêm) +#: * file co xuống dưới trần -> [INFO] nhắc hạ con số xuống +#: * file co xuống <= 400 dòng -> [INFO] nhắc gỡ hẳn khỏi danh sách +#: +#: Không bao giờ thêm mục mới vào đây để làm cổng xanh trở lại: file mới viết +#: phải dưới 400 dòng ngay từ đầu. Nới một con số cũng vậy — cách duy nhất +#: đúng là tách file. +LEGACY_ALLOWANCE = { + "ui/workspace_tab.py": 566, + "ui/widgets.py": 505, + "ui/task_editor_dialog.py": 627, + "ui/accounts_tab.py": 559, + "core/skills.py": 405, + "core/chat_agent.py": 419, + "ui/flow_dialog.py": 483, + "core/tasks.py": 339, + "core/co4e.py": 330, + "core/task_executors.py": 347, + "ui/help_agent_widget.py": 313, + "core/structure_graph.py": 346, + "ui/cowork_tab.py": 255, + "providers/base.py": 224, + "ui/co4e_tab.py": 180, + "providers/openai_compat.py": 279, + "core/doc_extract.py": 283, + "ui/connectors_panel.py": 281, +} + def count_file_lines(file_path: Path) -> int: """Read a python file and return total physical line count.""" @@ -34,43 +90,149 @@ def count_file_lines(file_path: Path) -> int: return 0 -def scan_directories( - root_dir: Path, target_dirs: List[str], max_lines: int, verbose: bool = False -) -> Tuple[int, List[Tuple[str, int]]]: - """Recursively scan target packages for files exceeding the maximum LOC limit. +def iter_source_files(root_dir: Path, target_dirs: List[str]): + """Sinh (đường dẫn tương đối, số dòng) cho mọi file production cần quét. - Returns: - A tuple of (total_files_scanned, list_of_violations_as_(relative_path, line_count)) + Ngoài các gói trong ``target_dirs``, quét thêm các ``.py`` nằm thẳng ở thư + mục gốc (``app.py``, ``state.py``, ``theme*.py``…) — chúng cũng là mã chạy + thật nhưng không thuộc gói nào, nên trước đây không ai canh. """ - total_files = 0 - violations: List[Tuple[str, int]] = [] - for target in target_dirs: dir_path = root_dir / target if not dir_path.is_dir(): - if verbose: - print(f"[INFO] Skipping missing directory: {target}") continue + for current_root, dirnames, files in os.walk(dir_path): + dirnames[:] = [d for d in dirnames if d != "__pycache__"] + for file_name in sorted(files): + if file_name.endswith(".py"): + full = Path(current_root) / file_name + yield full.relative_to(root_dir).as_posix(), count_file_lines(full) - for current_root, _, files in os.walk(dir_path): - for file_name in files: - if not file_name.endswith(".py"): - continue + if SCAN_ROOT_MODULES: + for full in sorted(root_dir.glob("*.py")): + yield full.name, count_file_lines(full) - full_path = Path(current_root) / file_name - rel_path = full_path.relative_to(root_dir).as_posix() - lines = count_file_lines(full_path) - total_files += 1 - if verbose: - print(f" {rel_path}: {lines} lines") +def iter_code_sizes(root_dir: Path, target_dirs: List[str]): + """Như :func:`iter_source_files` nhưng đếm DÒNG MÃ — dành cho bánh cóc.""" + for target in target_dirs: + dir_path = root_dir / target + if not dir_path.is_dir(): + continue + for current_root, dirnames, files in os.walk(dir_path): + dirnames[:] = [d for d in dirnames if d != "__pycache__"] + for file_name in sorted(files): + if file_name.endswith(".py"): + full = Path(current_root) / file_name + yield full.relative_to(root_dir).as_posix(), count_code_lines(full) - if lines > max_lines: - violations.append((rel_path, lines)) + if SCAN_ROOT_MODULES: + for full in sorted(root_dir.glob("*.py")): + yield full.name, count_code_lines(full) + + +def count_code_lines(file_path: Path) -> int: + """Số dòng MÃ của một file: bỏ docstring, chú thích và dòng trống. + + Dùng riêng cho bánh cóc ``LEGACY_ALLOWANCE``, không dùng cho trần 400 dòng. + Lý do: bánh cóc có một câu hỏi duy nhất — "file này có đang ĐỂ THÊM + VIỆC vào không?" — mà viết thêm một docstring thì không. Đếm dòng vật lý + ở đây biến cổng thành thứ phạt người viết tài liệu, và cách dễ nhất để làm + nó xanh lại sẽ là xoá bớt chú thích — đúng thứ không ai muốn. + + Trần 400 dòng thì VẪN đếm dòng vật lý: đó là hợp đồng đã chốt của cổng + S từ đầu, đổi cách đo là âm thầm nới nó ra cho mọi file. + """ + try: + src = file_path.read_text(encoding="utf-8", errors="ignore") + except OSError as exc: + print(f"[WARN] Failed to read {file_path}: {exc}", file=sys.stderr) + return 0 + + skip: set = set() + try: + tree = ast.parse(src) + except SyntaxError: + return len(src.splitlines()) + for node in ast.walk(tree): + if not isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, + ast.AsyncFunctionDef)): + continue + body = getattr(node, "body", None) + if body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) \ + and isinstance(body[0].value.value, str): + skip.update(range(body[0].lineno, body[0].end_lineno + 1)) + + try: + for tok in tokenize.generate_tokens(io.StringIO(src).readline): + if tok.type == tokenize.COMMENT: + skip.add(tok.start[0]) + except (tokenize.TokenError, IndentationError): + pass + + return sum(1 for i, line in enumerate(src.splitlines(), 1) + if i not in skip and line.strip()) + + +def scan_directories( + root_dir: Path, target_dirs: List[str], max_lines: int, verbose: bool = False +) -> Tuple[int, List[Tuple[str, int]]]: + """Quét cây mã production, đối chiếu với trần chung và trần riêng của nợ cũ. + + Trả về ``(số file đã quét, danh sách vi phạm)``. Một file bị tính là vi + phạm khi nó vượt trần chung VÀ không có trong ``LEGACY_ALLOWANCE``, hoặc + khi nó có trong danh sách nợ cũ nhưng đã phình quá con số ghi ở đó. + """ + total_files = 0 + violations: List[Tuple[str, int]] = [] + code_sizes = dict(iter_code_sizes(root_dir, target_dirs)) + + for rel_path, lines in iter_source_files(root_dir, target_dirs): + total_files += 1 + if verbose: + print(f" {rel_path}: {lines} lines") + + allowance = LEGACY_ALLOWANCE.get(rel_path) + if allowance is None: + if lines > max_lines: + violations.append((rel_path, lines)) + else: + # Nợ cũ: đo bằng dòng mã, không đo bằng dòng vật lý. + code = code_sizes.get(rel_path, lines) + if code > allowance: + violations.append((rel_path, code)) return total_files, violations +def audit_legacy(root_dir: Path, target_dirs: List[str], max_lines: int) -> List[str]: + """Các dòng nhắc về ``LEGACY_ALLOWANCE`` — chỉ để báo, không làm cổng đỏ. + + Bánh cóc chỉ có nghĩa khi con số được siết lại mỗi lần file co bớt; nếu + không ai nhắc thì nó đứng yên mãi ở mức của lần ghi đầu tiên. + + Hai thước đo, mỗi thước trả lời một câu khác nhau: + + * **Gỡ hẳn khỏi danh sách** chỉ đúng khi file đã xuống dưới trần đo bằng + DÒNG VẬT LÝ — vì đó mới là thước của trần 400. Nhắc gỡ một file 950 + dòng chỉ vì phần mã của nó dưới 400 là lời khuyên sai: gỡ xong cổng đỏ + ngay. + * **Hạ con số xuống** đo bằng DÒNG MÃ, cùng thước với chính bánh cóc. + """ + code = dict(iter_code_sizes(root_dir, target_dirs)) + physical = dict(iter_source_files(root_dir, target_dirs)) + notes: List[str] = [] + for rel_path, allowance in sorted(LEGACY_ALLOWANCE.items()): + lines = code.get(rel_path) + if lines is None: + notes.append(f"{rel_path}: file khong con ton tai - go khoi LEGACY_ALLOWANCE") + elif physical.get(rel_path, lines) <= max_lines: + notes.append(f"{rel_path}: nay chi {physical[rel_path]} dong - go khoi LEGACY_ALLOWANCE") + elif lines < allowance: + notes.append(f"{rel_path}: {lines} dong ma (tran dang ghi {allowance}) - ha con so xuong {lines}") + return notes + + def main() -> int: """CLI entry point for the LOC guard script.""" parser = argparse.ArgumentParser( @@ -115,14 +277,27 @@ def main() -> int: verbose=args.verbose, ) + notes = audit_legacy(root_dir, args.dirs, args.max_lines) + if notes: + print(f"\n[INFO] {len(notes)} muc trong LEGACY_ALLOWANCE co the siet lai:") + for note in notes: + print(f" - {note}") + if violations: print(f"\n[FAIL] Found {len(violations)} oversized file(s) (> {args.max_lines} LOC):") for file_path, lines in sorted(violations, key=lambda x: x[1], reverse=True): - print(f" ❌ {file_path}: {lines} lines (exceeds limit by {lines - args.max_lines})") + allowance = LEGACY_ALLOWANCE.get(file_path) + if allowance: + print(f" ❌ {file_path}: {lines} dong ma - no cu ghi la {allowance}, " + f"nay phinh them {lines - allowance}. Tach bot, dung noi con so.") + else: + print(f" ❌ {file_path}: {lines} lines (exceeds limit by {lines - args.max_lines})") print("\nAction Required: Refactor oversized files into smaller single-responsibility modules.") return 1 - print(f"\n[PASS] All {total_files} production files in {args.dirs} satisfy <= {args.max_lines} LOC limit.") + legacy = len(LEGACY_ALLOWANCE) + print(f"\n[PASS] All {total_files} production files in {args.dirs} satisfy <= {args.max_lines} LOC limit " + f"({legacy} file no cu duoc mien tru, khong file nao phinh them).") return 0 diff --git a/scripts/check_orphan_modules.py b/scripts/check_orphan_modules.py new file mode 100644 index 0000000..c87d522 --- /dev/null +++ b/scripts/check_orphan_modules.py @@ -0,0 +1,369 @@ +#!/usr/bin/env python3 +"""Cổng chất lượng: không để module production nào không ai import (CASAN Gate O). + +Vì sao cần cổng này +------------------- +Ba cổng đang có (Clean Architecture / Secrets / LOC) đều **không bắt được mã +chết**: một file không ai import vẫn qua cả ba, vì nó đúng chiều phụ thuộc, +không chứa credential, và ngắn hơn 400 dòng. Đợt kiểm chứng ngày 29/08 tìm ra +**1.400 dòng mã trùng lặp chết** lọt qua đúng theo cách đó — hai bản tách song +song của cùng một god-file cùng được giữ lại sau một lần merge, trong đó hai +file còn không import nổi (``GraphQaMixin`` không tồn tại, +``ui.calendar_view`` đã bị xoá). Không ai phát hiện vì không có gì đi tìm. + +Cách làm +-------- +Dựng đồ thị import tĩnh bằng AST, bắt đầu từ ``__init__`` / ``__main__`` / +``app``, đi theo cả import tuyệt đối lẫn tương đối, kể cả import nằm trong thân +hàm (mã này dùng import muộn rất nhiều). Import một module con cũng chạy +``__init__.py`` của mọi gói cha, nên các gói cha đó cũng được coi là tới được. + +Module không tới được mà KHÔNG nằm trong ``ALLOWLIST`` thì cổng đỏ. Hai +ngoại lệ tự động, để danh sách miễn trừ không phải chép lại cùng một lý do +nhiều lần: ``__init__.py`` của một gói mà mọi thành viên đều dormant, và +module chỉ được chính mã dormant đã miễn trừ import. + +Seam chưa nối dây +------------------ +Một phần ``ALLOWLIST`` là *seam*: hợp đồng dựng trước để hai nhóm làm song +song, chờ bên kia nối vào. Những file đó mang nhãn ``SEAM · dựng `` +trong docstring đầu file, và cổng này đếm tuổi của chúng. Quá +``SEAM_MAX_AGE_DAYS`` thì in [WARN] — chỉ nhắc, không làm cổng đỏ. + +Danh sách miễn trừ +------------------ +``ALLOWLIST`` là mã dormant đã có TỪ TRƯỚC đợt refactor (xem +``docs/architecture/dormant-code.md``) cộng các seam đã dựng nhưng chưa nối +dây. Đây là danh sách **chỉ được co lại**: xoá hoặc nối dây một mục thì gỡ nó +khỏi đây, đừng bao giờ thêm mục mới để làm cổng xanh trở lại. + +Chạy: python scripts/check_orphan_modules.py +""" +from __future__ import annotations + +import argparse +import ast +import os +import re +import sys +from datetime import date +from pathlib import Path +from typing import Dict, List, Optional, Set, Tuple + +if hasattr(sys.stdout, "reconfigure"): + try: + sys.stdout.reconfigure(encoding="utf-8", errors="replace") + except Exception: # noqa: BLE001 + pass + +REPO_ROOT = Path(__file__).resolve().parent.parent +PACKAGE = "cowork_local" + +#: Không phải mã production — không quét. +SKIP_TOP = { + ".git", ".gitea", ".vibeflow-preview", "__pycache__", "assets", "config", + "docs", "scripts", "skill_library", "slides", "tests", "tools", +} + +#: Điểm bắt đầu: mọi thứ tới được từ đây là mã đang sống. +ROOTS = (PACKAGE, f"{PACKAGE}.__main__", f"{PACKAGE}.app") + +#: Bao nhiêu ngày thì một seam chưa nối dây đáng được nhắc lại. +#: +#: Seam là hợp đồng dựng trước để hai nhóm làm song song — hợp lý trong vài +#: tuần, nhưng quá lâu thì nó không còn là hợp đồng nữa mà thành mã chết có lời +#: biện hộ. 30 ngày là một chu kỳ epic của dự án này: qua một chu kỳ mà vẫn +#: chưa ai nối thì phải quyết — nối, hoặc xoá. +#: +#: Chỉ nhắc, KHÔNG làm cổng đỏ: để nó đỏ thì CI sẽ đỏ vào một buổi sáng mà +#: không ai sửa gì cả, và cách nhanh nhất để xanh lại là sửa ngày. +SEAM_MAX_AGE_DAYS = 30 + +#: Nhận dạng nhãn seam trong docstring đầu file. +_SEAM_RE = re.compile(r"SEAM \u00b7 d\u1ef1ng (\d{4})-(\d{2})-(\d{2})") + +#: Mã dormant được chấp nhận, kèm lý do. CHỈ ĐƯỢC CO LẠI. +ALLOWLIST: Dict[str, str] = { + # --- dormant từ trước refactor (docs/architecture/dormant-code.md) ----- + "core/account_excel.py": "quản lý tài khoản — chưa bật, dormant từ trước R01", + "core/accounts.py": "quản lý tài khoản — chưa bật, dormant từ trước R01", + "core/codebase_memory_ui.py": "phần UI của codebase-memory — chưa bật", + "core/custom_agents.py": "bản agent tự tạo cũ, đã thay bằng core/co4e.py", + "core/graph_server.py": "máy chủ HTTP phục vụ đồ thị D3 — chỉ dùng khi bật cờ", + "core/groups.py": "nhóm người dùng — chưa bật, dormant từ trước R01", + "security/action_validator.py": "lớp bọc mỏng quanh command_risk_classifier", + "security/attachment_validator.py": "lớp bọc mỏng quanh command_risk_classifier", + "security/prompt_validator.py": "lớp bọc mỏng quanh command_risk_classifier", + "ui/accounts_tab.py": "màn quản lý tài khoản — chưa bật", + "ui/agent_manager_tab.py": "đã thay bằng presentation/monitoring/tabs/agents_admin_tab.py", + "ui/flow_dialog.py": "trình sửa luồng cũ, đã thay bằng Co4E Studio", + "ui/login_dialog.py": "bản này không có lớp đăng nhập", + "ui/mcp_servers_dialog.py": "đã thay bằng ui/connectors_panel.py", + "ui/skill_manager_tab.py": "đã thay bằng ui/skills_dialog.py", + # --- chạy bằng tiến trình con, không ai import ------------------------ + "mcp_servers/ms365_server.py": "chạy bằng subprocess (state.py)", + "mcp_servers/project_context_server.py": "chạy bằng subprocess", + "mcp_servers/project_context/server.py": "chạy bằng subprocess", + "mcp_servers/project_context/foundation.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/registry.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/runtime.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/providers/change.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/providers/issue.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/providers/knowledge.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/tools/change_context.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/tools/issue_context.py": "gói MCP chạy bằng subprocess", + "mcp_servers/project_context/tools/knowledge_search.py": "gói MCP chạy bằng subprocess", + # --- seam đã dựng, chưa nối dây (xem F-05 trong báo cáo đối chiếu) ---- + "application/workflows/co4e_workflow_service.py": "R07-T06 — bootstrap chưa gọi build_co4e_tab", + "presentation/co4e/co4e_tab.py": "factory chờ bootstrap.py nối, hạn đã ghi trong file", + "domain/security/tool_policy.py": "hình dạng dữ liệu, chờ nối vào ToolPolicyGateway", + "domain/workflows/run_record.py": "DTO, dùng khi Co4EWorkflowService được nối dây", + "domain/agents/agent_event_codec.py": "shim tạm, bỏ khi chat_panel dùng event có kiểu", + "infrastructure/filesystem/execution_workspace.py": "R06-T03 — chưa có call site", + "infrastructure/sandbox/sandbox_capabilities.py": "ma trận năng lực sandbox, chưa nối", + "infrastructure/config/settings_facade.py": "R02-T03 — chưa có call site", + "infrastructure/config/config_repository.py": "Protocol, chỉ dùng làm chú thích kiểu", + # --- vỏ chuyển tiếp (strangler-fig) ------------------------------------ + # Vỏ chuyển tiếp tồn tại để giữ ĐƯỜNG IMPORT CŨ chạy được, nên việc mã mới + # không import nó là trạng thái ĐÚNG chứ không phải thiếu sót. Ba vỏ còn + # lại (ui/chat_panel.py, ui/monitoring_tab.py, core/tools.py) hiện vẫn tới + # được vì presentation/ đang nhập ngược qua chúng — đi vòng như thế tạo chu + # trình import, và khi nào gỡ nốt thì chúng cũng xuống đây. + "ui/composer.py": "vỏ chuyển tiếp R08-T02, giữ đường import cũ", + # --- hạ tầng test / công cụ ------------------------------------------- + "conftest.py": "pytest tự nạp, không ai import", +} + + +def _discover() -> Dict[str, str]: + """{tên module đầy đủ: đường dẫn tương đối} cho mọi file production.""" + out: Dict[str, str] = {} + for dirpath, dirnames, filenames in os.walk(REPO_ROOT): + rel_dir = os.path.relpath(dirpath, REPO_ROOT).replace("\\", "/") + if rel_dir == ".": + rel_dir = "" + dirnames[:] = [d for d in dirnames if d not in SKIP_TOP] + else: + dirnames[:] = [d for d in dirnames if d != "__pycache__"] + for name in filenames: + if not name.endswith(".py"): + continue + rel = f"{rel_dir}/{name}" if rel_dir else name + parts = rel[:-3].split("/") + if parts[-1] == "__init__": + parts = parts[:-1] + out[".".join([PACKAGE] + parts) if parts else PACKAGE] = rel + return out + + +def _targets(module: str, node: ast.AST, modules: Dict[str, str]) -> List[str]: + """Các module trong gói mà một câu lệnh import trỏ tới.""" + names: List[str] = [] + if isinstance(node, ast.Import): + names = [a.name for a in node.names] + elif isinstance(node, ast.ImportFrom): + if node.level: + parts = module.split(".") + # gói (__init__) tự nó là tiền tố; module thường thì lấy gói cha + if modules.get(module, "").endswith("__init__.py"): + pkg = parts + else: + pkg = parts[:-1] + up = node.level - 1 + if up: + pkg = pkg[: len(pkg) - up] + base = ".".join(pkg) + head = f"{base}.{node.module}" if node.module else base + names = [head] + [f"{head}.{a.name}" for a in node.names] + elif node.module: + names = [node.module] + [f"{node.module}.{a.name}" for a in node.names] + resolved = [] + for n in names: + if n in modules: + resolved.append(n) + elif n.rsplit(".", 1)[0] in modules: + resolved.append(n.rsplit(".", 1)[0]) + return resolved + + +def _reach(modules: Dict[str, str], edges: Dict[str, Set[str]], + roots) -> Set[str]: + """Tập module tới được từ ``roots`` theo đồ thị import.""" + seen: Set[str] = set() + stack = [r for r in roots if r in modules] + while stack: + module = stack.pop() + if module in seen: + continue + seen.add(module) + stack.extend(edges.get(module, ())) + # Import một module con cũng chạy __init__.py của mọi gói cha. + parts = module.split(".") + for i in range(1, len(parts)): + parent = ".".join(parts[:i]) + if parent in modules and parent not in seen: + stack.append(parent) + return seen + + +def find_orphans() -> List[str]: + """Đường dẫn các module production không tới được từ điểm bắt đầu.""" + modules = _discover() + edges: Dict[str, Set[str]] = {} + for module, rel in modules.items(): + try: + tree = ast.parse((REPO_ROOT / rel).read_text(encoding="utf-8", errors="replace")) + except SyntaxError: + edges[module] = set() + continue + found: Set[str] = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Import, ast.ImportFrom)): + found.update(_targets(module, node, modules)) + edges[module] = found + + seen = _reach(modules, edges, ROOTS) + return sorted(modules[m] for m in set(modules) - seen) + + +def find_dormant_reachable() -> Set[str]: + """Module chỉ tới được từ một mục trong ``ALLOWLIST``. + + Một module mà người import duy nhất là mã dormant đã được miễn trừ thì + dormant vì ĐÚNG LÝ DO ẤY — bắt nó phải có dòng miễn trừ riêng chỉ nhân đôi + cùng một thông tin, và tệ hơn là khiến người ta ngại tách file trong vùng + dormant. Đổi lại, khi mục dormant kia được nối dây hoặc bị xoá, cả nhánh + này tự động theo — sống theo hoặc bị báo lên, không có dòng miễn trừ cũ + nào ở lại che mắt. + """ + modules = _discover() + rel_to_mod = {rel: mod for mod, rel in modules.items()} + edges: Dict[str, Set[str]] = {} + for module, rel in modules.items(): + try: + tree = ast.parse((REPO_ROOT / rel).read_text(encoding="utf-8", errors="replace")) + except SyntaxError: + edges[module] = set() + continue + found: Set[str] = set() + for node in ast.walk(tree): + if isinstance(node, (ast.Import, ast.ImportFrom)): + found.update(_targets(module, node, modules)) + edges[module] = found + + dormant_roots = [rel_to_mod[rel] for rel in ALLOWLIST if rel in rel_to_mod] + from_real = _reach(modules, edges, ROOTS) + from_dormant = _reach(modules, edges, dormant_roots) + # Bỏ chính các mục ALLOWLIST — chúng đã có dòng miễn trừ riêng, kể lại + # ở đây chỉ làm nhiễu. + return {modules[m] for m in from_dormant - from_real} - set(ALLOWLIST) + + +def seam_ages(today: Optional[date] = None) -> List[Tuple[str, int, bool]]: + """(đường dẫn, số ngày dormant, có nhãn không) cho mọi seam trong ALLOWLIST. + + Seam được nhận ra bằng chính nhãn ``SEAM · dựng `` trong docstring + đầu file chứ không bằng một danh sách thứ hai ở đây: hai danh sách là hai + chỗ phải nhớ cập nhật, và chỗ thứ hai bao giờ cũng là chỗ bị quên. + + Ngày trong nhãn là ngày file được thêm vào repo (lấy từ git lúc đặt nhãn), + không phải một hạn ai đó tự đặt. + """ + today = today or date.today() + out: List[Tuple[str, int, bool]] = [] + for rel in sorted(ALLOWLIST): + path = REPO_ROOT / rel + if not path.is_file(): + continue + head = path.read_text(encoding="utf-8", errors="replace")[:4000] + m = _SEAM_RE.search(head) + if m is None: + continue + made = date(int(m.group(1)), int(m.group(2)), int(m.group(3))) + age = (today - made).days + out.append((rel, age, age > SEAM_MAX_AGE_DAYS)) + return out + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--list-allowed", action="store_true", + help="In danh sách miễn trừ rồi thoát") + args = parser.parse_args() + + if args.list_allowed: + for rel, why in sorted(ALLOWLIST.items()): + print(f" {rel:62} {why}") + return 0 + + print("=" * 70) + print("CASAN Guard 'O' (Orphan): module production phải có ít nhất 1 nơi import") + print("=" * 70) + + orphans = find_orphans() + orphan_set = set(orphans) + + def _dormant_package(rel: str) -> bool: + """``__init__.py`` của một gói mà KHÔNG module nào bên trong tới được. + + Gói kiểu đó dormant vì thành viên của nó dormant — báo riêng nó ra chỉ + nhân đôi cùng một phát hiện. Gói còn dù chỉ một module đang sống thì + ``__init__.py`` cũng phải sống theo (import module con chạy nó), nên + luật này không che được mã chết thật. + """ + if not rel.endswith("__init__.py"): + return False + pkg = rel[: -len("__init__.py")] + members = [m for m in _discover().values() + if m.startswith(pkg) and m != rel] + return bool(members) and all( + m in orphan_set or m in ALLOWLIST for m in members) + + via_dormant = find_dormant_reachable() + unexpected = [o for o in orphans + if o not in ALLOWLIST + and o not in via_dormant + and not _dormant_package(o)] + stale = sorted(set(ALLOWLIST) - orphan_set) + + if stale: + print(f"\n[INFO] {len(stale)} mục trong ALLOWLIST nay đã có nơi import — gỡ khỏi danh sách:") + for rel in stale: + print(f" - {rel}") + + if unexpected: + print(f"\n[FAIL] {len(unexpected)} module production không ai import:") + for rel in unexpected: + print(f" x {rel}") + print("\nXoá chúng, hoặc nối dây, hoặc thêm vào ALLOWLIST kèm lý do") + print("(scripts/check_orphan_modules.py) nếu đó là mã dormant có chủ ý.") + return 1 + + seams = seam_ages() + overdue = [(rel, age) for rel, age, late in seams if late] + if overdue: + print(f"\n[WARN] {len(overdue)}/{len(seams)} seam đã dựng quá {SEAM_MAX_AGE_DAYS} " + f"ngày mà chưa nối dây — nối, hoặc xoá:") + for rel, age in sorted(overdue, key=lambda x: -x[1]): + print(f" ! {rel}: {age} ngày") + elif seams: + oldest = max(age for _, age, _ in seams) + print(f"\n[INFO] {len(seams)} seam chưa nối dây, cái lâu nhất {oldest} ngày " + f"(nhắc khi quá {SEAM_MAX_AGE_DAYS}).") + + total = len(_discover()) + # __init__.py của gói toàn thành viên dormant đã có luật riêng ở trên — + # không kể lại lần nữa. + implicit = sorted(r for r in via_dormant if not _dormant_package(r)) + if implicit: + print(f"\n[INFO] {len(implicit)} module chỉ được mã dormant import — " + f"dormant theo cùng lý do, không cần dòng miễn trừ riêng:") + for rel in implicit: + print(f" - {rel}") + print(f"\n[PASS] {total - len(orphans)}/{total} module production đều có nơi import " + f"({len(orphans)} mục dormant đã được miễn trừ).") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_quality_gate.py b/scripts/run_quality_gate.py index dfe0d12..513b3b4 100644 --- a/scripts/run_quality_gate.py +++ b/scripts/run_quality_gate.py @@ -8,7 +8,8 @@ Verification Stages (CASAN): 1. [C] Clean Architecture Guard (scripts/check_imports.py) 2. [A] Atomic & Secrets Audit (scripts/audit_security.py) 3. [S] Single Responsibility / LOC Guard (scripts/check_loc.py) - 4. [A/N] Automated Tests & No-Regression Suite (pytest) + 4. [O] Orphan Module Guard (scripts/check_orphan_modules.py) + 5. [A/N] Automated Tests & No-Regression Suite (pytest) """ from __future__ import annotations @@ -90,6 +91,13 @@ def main() -> int: "S - Single Responsibility LOC Limit (<= 400 LOC)", [sys.executable, str(REPO_ROOT / "scripts" / "check_loc.py"), "--max-lines", "400"], ), + # Ba cổng trên không cổng nào bắt được mã chết: một file không ai import + # vẫn đúng chiều phụ thuộc, vẫn không có credential, vẫn dưới 400 dòng. + # Cổng O đi tìm đúng khoảng trống đó. + ( + "O - Orphan Module Guard (moi module phai co noi import)", + [sys.executable, str(REPO_ROOT / "scripts" / "check_orphan_modules.py")], + ), ] if not args.skip_tests: