From f4eb39fbbfc0ae0748a54f321d31abf69b365e2c Mon Sep 17 00:00:00 2001 From: hh <2587203630@qq.com> Date: Mon, 20 Jul 2026 10:06:59 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E6=99=BA=E8=83=BD?= =?UTF-8?q?=E6=9E=B6=E6=9E=84=E6=96=87=E6=A1=A3=E5=90=8C=E6=AD=A5=E8=83=BD?= =?UTF-8?q?=E5=8A=9B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../skills/sync-architecture-docs/SKILL.md | 12 + .../skills/sync-architecture-docs/SKILL.md | 12 + core/backend/ARCHITECTURE.md | 3 +- core/frontend/ARCHITECTURE.md | 3 +- tools/architecture-sync/WORKFLOW.md | 194 ++ tools/architecture-sync/scan_architecture.py | 2130 +++++++++++++++++ 6 files changed, 2352 insertions(+), 2 deletions(-) create mode 100644 .agents/skills/sync-architecture-docs/SKILL.md create mode 100644 .claude/skills/sync-architecture-docs/SKILL.md create mode 100644 tools/architecture-sync/WORKFLOW.md create mode 100644 tools/architecture-sync/scan_architecture.py diff --git a/.agents/skills/sync-architecture-docs/SKILL.md b/.agents/skills/sync-architecture-docs/SKILL.md new file mode 100644 index 0000000..807aff4 --- /dev/null +++ b/.agents/skills/sync-architecture-docs/SKILL.md @@ -0,0 +1,12 @@ +--- +name: sync-architecture-docs +description: "分析 AirShelf 前后端代码与架构文档的差异，并且只有在用户明确确认后才更新前后端 ARCHITECTURE.md。用户提出“帮我更新项目文档”“同步架构文档”“检查架构文档”，或对待确认的架构文档更新回复“确认”时使用。" +--- + +# 同步架构文档 + +1. 所有相对路径均以当前 Skill 目录为基准解析。 +2. 执行任何任务操作前，完整读取 `../../../tools/architecture-sync/WORKFLOW.md`。 +3. 严格遵循公共工作流，包括只读规划、确认门禁、写入白名单、计划漂移检查和结果验证。 +4. 如果公共工作流或其要求的扫描器不存在，立即停止并说明功能尚未安装完整。 +5. 禁止用不受约束的 AI 扫描代替扫描器；获得有效确认前，不得修改任何文件。 diff --git a/.claude/skills/sync-architecture-docs/SKILL.md b/.claude/skills/sync-architecture-docs/SKILL.md new file mode 100644 index 0000000..807aff4 --- /dev/null +++ b/.claude/skills/sync-architecture-docs/SKILL.md @@ -0,0 +1,12 @@ +--- +name: sync-architecture-docs +description: "分析 AirShelf 前后端代码与架构文档的差异，并且只有在用户明确确认后才更新前后端 ARCHITECTURE.md。用户提出“帮我更新项目文档”“同步架构文档”“检查架构文档”，或对待确认的架构文档更新回复“确认”时使用。" +--- + +# 同步架构文档 + +1. 所有相对路径均以当前 Skill 目录为基准解析。 +2. 执行任何任务操作前，完整读取 `../../../tools/architecture-sync/WORKFLOW.md`。 +3. 严格遵循公共工作流，包括只读规划、确认门禁、写入白名单、计划漂移检查和结果验证。 +4. 如果公共工作流或其要求的扫描器不存在，立即停止并说明功能尚未安装完整。 +5. 禁止用不受约束的 AI 扫描代替扫描器；获得有效确认前，不得修改任何文件。 diff --git a/core/backend/ARCHITECTURE.md b/core/backend/ARCHITECTURE.md index 1b2ba57..2b9e33d 100644 --- a/core/backend/ARCHITECTURE.md +++ b/core/backend/ARCHITECTURE.md @@ -1,7 +1,8 @@ # AirShelf 后端架构 > 适用目录：`core/backend`。 -> 最后核对：2026-07-17。 +> 最后核对：2026-07-18。 + > 本文描述当前代码结构；系统级拓扑见 [core/ARCHITECTURE.md](../ARCHITECTURE.md)。 ## 1. 定位 diff --git a/core/frontend/ARCHITECTURE.md b/core/frontend/ARCHITECTURE.md index a6c5cbb..959100a 100644 --- a/core/frontend/ARCHITECTURE.md +++ b/core/frontend/ARCHITECTURE.md @@ -1,7 +1,8 @@ # AirShelf 前端架构 > 适用目录：`core/frontend`。 -> 最后核对：2026-07-17。 +> 最后核对：2026-07-18。 + > 本文描述当前代码结构；系统级拓扑见 [core/ARCHITECTURE.md](../ARCHITECTURE.md)。 ## 1. 定位 diff --git a/tools/architecture-sync/WORKFLOW.md b/tools/architecture-sync/WORKFLOW.md new file mode 100644 index 0000000..27c8148 --- /dev/null +++ b/tools/architecture-sync/WORKFLOW.md @@ -0,0 +1,194 @@ +# 架构文档同步工作流 + +## 1. 目标与边界 + +当用户要求“更新项目文档”或“同步架构文档”时，先只读分析影响，得到明确确认后，再局部同步当前实现文档。 + +运行时只允许修改： + +- `core/backend/ARCHITECTURE.md` +- `core/frontend/ARCHITECTURE.md` + +禁止修改： + +- `core/ARCHITECTURE.md` +- 任何业务代码、配置、迁移、测试和部署文件 +- Git 暂存区、commit、push 和远端状态 +- 本工作流、扫描脚本和 Skill 自身 + +不得访问或输出 `.env`、`account.md`、密钥、数据库文件、用户上传文件及其他敏感信息。 + +## 2. 基本原则 + +- 默认只读；没有有效确认时不得写文件。 +- 使用扫描脚本提取事实，不用 AI 全量阅读项目代替扫描。 +- 只读取扫描结果列出的证据文件和目标章节。 +- 只做局部补丁，不整份重写文档。 +- 保留原章节、历史决策和规划内容。 +- 明确区分“当前实现”与“目标架构”。 +- 无法从代码证实时，跳过并报告，不推测补写。 +- 不覆盖或整理与本次任务无关的用户改动。 +- 每次写入后必须验证；验证失败时停止，不扩大范围补救。 + +## 3. 状态机 + +```text +IDLE + → ANALYZE_READ_ONLY + → WAIT_CONFIRMATION + → REVALIDATE_PLAN + → UPDATE_ALLOWED_DOCS + → VERIFY + → DONE +``` + +任何异常进入 `STOPPED`，报告原因并等待用户决定。 + +## 4. 阶段 A：只读分析 + +1. 使用 Git 确定仓库根目录。 +2. 确认 `tools/architecture-sync/scan_architecture.py` 存在；不存在则报告安装未完成并停止，不回退为全量 AI 扫描。 +3. 运行扫描器的 `plan` 模式；默认使用 `--scope all`，用户明确只处理后端或前端时分别使用 `--scope backend` 或 `--scope frontend`。 +4. 解析扫描器 JSON；不得自行扩大 `changed_files`、`evidence` 或目标文档范围。 +5. 若 `analysis_ready` 不为 `true`，说明扫描器的影响分析能力尚未安装完整并停止，不得进入确认或写入阶段。 +6. 若 `requires_update=false`，说明没有架构性变化并结束，不修改文件。 +7. 若存在影响，将扫描器返回的 `plan_id`、`scope`、`snapshot_id`、目标文档指纹和 `planned_updates` 作为当前对话唯一待确认计划，按“影响预览”格式展示，然后强制停止并等待确认；不得预先修改文件。 + +影响预览必须包含： + +- `plan_id`。 +- 发现的架构变化。 +- 每项变化的证据文件。 +- 准备修改的文档和章节。 +- 明确不会修改的范围。 +- 扫描警告和无法确认的事项。 + +使用以下输出结构： + +```text +架构文档更新计划： +更新范围： + +发现的架构变化： +1. <变化> + - 依据：<证据文件或符号> + - 影响：<目标文档> · <章节> + +准备修改： +- <目标文档> · <章节> + +不会修改： +- core/ARCHITECTURE.md +- 业务代码、配置、迁移、测试和部署文件 +- Git 暂存、commit、push + +警告或待确认： +- <没有则写“无”> + +是否确认更新？请回复：确认更新项目文档 +``` + +## 5. 确认规则 + +有效确认必须同时满足： + +- 当前对话存在尚未执行的影响预览。 +- 用户确认的必须是当前唯一待确认 `plan_id`，旧计划、已执行计划和被新请求替代的计划均无效。 +- 用户在预览后回复 `确认更新项目文档`；同一连续对话中也可接受明确的 `确认`。 +- 用户没有改变目标、排除项或更新范围。 +- 待确认计划没有被新的请求替代。 + +新对话中的单独“确认”、含糊回复、提前确认或针对其他任务的确认均无效。 + +用户拒绝、改变范围或提出新约束时，立即丢弃旧 `plan_id`，使用新的 `scope` 或约束返回阶段 A 重新生成并展示计划；不得沿用旧确认。 + +## 6. 阶段 B：确认后重新校验 + +写文件前必须使用与待确认计划完全相同的 `--scope` 重新运行 `plan`： + +- 新 `plan_id` 与已确认值一致，才可继续。 +- 证据文件、目标章节或目标文档发生变化时，展示新计划并重新确认。 +- 目标文档自计划生成后被其他操作修改时，停止并重新计划。 +- 扫描器返回警告升级、敏感文件或越界路径时，停止。 + +确认只授权当前 `plan_id`，不授权未来变化。 + +## 7. 局部更新规则 + +阶段 B 重校验通过后，锁定扫描器返回的 `update_context`，不得自行扩大范围： + +1. 校验 `update_context.plan_id` 等于已确认 `plan_id`，并校验每个 `document_preconditions.before_sha256` 仍与目标文档一致；任一不一致立即停止并重新规划。 +2. 将 `evidence_allowlist` 视为最大读取白名单，只选取证明当前影响所必需的文件；禁止因方便而遍历其他代码。 +3. 目标文档只读取 `document_preconditions.sections` 指定章节及定位标题所需的相邻行；不把整篇文档交给 AI 重写。 +4. 根据证据判断这是当前实现变化、规划变化还是普通业务修复；普通业务修复没有改变架构边界时，跳过该影响。 +5. 使用 `apply_patch` 对一个文档、一个局部事实逐项修改；禁止生成整篇替换内容或用脚本重写 Markdown。 +6. 保留标题层级、术语、相对链接、风险说明、演进原则、历史决策和原有写作风格。 +7. 当前实现与历史设计或目标架构并存时，保留原内容并明确标注“当前实现”与“目标架构”，不得用当前代码覆盖历史决策。 +8. 正文事实确实发生变化时，将文首“最后核对”更新为执行当日日期，并将同步标记设置为 `update_context.sync_metadata.marker`；标记必须使用完整小写 HEAD SHA，且每份文档只能存在一条。 +9. 对缺少同步标记的 `baseline_review`，若证据核对后正文已经准确，允许只新增首次同步标记并更新“最后核对”日期；不得借首次基线核对改写无关正文。 +10. 若 `has_uncommitted_evidence=true`，同步标记仍只表示当前 HEAD；结果中必须说明包含了工作区证据，后续代码提交后需要再次扫描。 +11. 修改完成后直接进入阶段 C；不得执行 `git add`、`git commit`、`git push`，也不得生成或保存临时计划文件。 + +禁止： + +- 删除未受影响章节。 +- 为了“精简”覆盖整个文件。 +- 把规划内容改写成已实现事实。 +- 把暂时未使用的历史决策直接删除。 +- 根据文件名猜测实现细节。 +- 顺手修复代码、格式化其他文件或清理工作区。 +- 在没有有效确认时执行以上任何写入步骤。 + +## 8. 阶段 C：验证 + +更新后必须使用原计划令牌运行： + +```text +python tools/architecture-sync/scan_architecture.py verify --scope <原 scope> --guard +``` + +不得省略、修改或跨计划复用校验令牌。只有 `verification.status=passed` 才能完成流程。验证器检查： + +- 本流程新增的文件修改是否仅落在两个允许文档。 +- 计划外的目标章节是否保持不变。 +- 原有一级、二级标题是否被误删。 +- 文档引用的目录、文件和关键符号是否存在。 +- 相对链接是否有效。 +- 是否引入本机绝对路径、敏感信息或裸密钥。 +- 是否出现异常大范围删除或整篇覆盖。 +- Markdown 格式是否有效。 +- 计划生成后 HEAD、相关代码指纹、证据文件和同步控制文件是否发生漂移。 + +验证失败： + +- 停止流程并列出失败项。 +- 按验证器返回的 `code`、`message` 和 `suggestion` 向用户说明原因；不得绕过失败项。 +- 不自动修改业务代码来迁就文档。 +- 不自动扩大写入白名单。 +- 不暂存、不提交、不推送。 + +验证通过后输出： + +```text +架构文档已更新： +- <文档> · <章节>：<一句话变化> + +验证：通过 +未执行：业务代码修改、git add、commit、push +``` + +## 9. 工作区保护 + +- 阶段 A 记录目标文档摘要和现有工作区变化，阶段 B 用于漂移检查。 +- 已存在的用户改动属于用户，不得覆盖、回滚、暂存或提交。 +- 如果现有改动与目标章节重叠且不能安全合并，停止并请求用户处理。 +- 不使用 `git reset`、`git checkout --` 或任何破坏性恢复命令。 + +## 10. 安装与能力缺失 + +缺少扫描脚本、Python、Git、读取权限或写入权限时： + +- 明确指出缺少的能力。 +- 保持只读并停止。 +- 不改用无约束的全项目 AI 扫描。 +- 不建议扩大业务服务权限或让线上服务访问 Git 仓库。 diff --git a/tools/architecture-sync/scan_architecture.py b/tools/architecture-sync/scan_architecture.py new file mode 100644 index 0000000..31a3bb7 --- /dev/null +++ b/tools/architecture-sync/scan_architecture.py @@ -0,0 +1,2130 @@ +#!/usr/bin/env python3 +"""Read-only architecture inventory for AirShelf. + +The scanner intentionally has no apply/write mode. It emits deterministic JSON to +stdout and limits all content reads to architecture-relevant, non-sensitive files. +""" + +from __future__ import annotations + +import argparse +import ast +import base64 +import fnmatch +import hashlib +import json +import os +import re +import subprocess +import sys +import zlib +from pathlib import Path, PurePosixPath +from typing import Any, Callable, Iterable, Sequence +from urllib.parse import unquote + + +SCHEMA_VERSION = 1 +MAX_TEXT_BYTES = 1_000_000 +MAX_HASH_BYTES = 10_000_000 +TARGET_DOCUMENTS = ( + "core/backend/ARCHITECTURE.md", + "core/frontend/ARCHITECTURE.md", +) +GIT_SCOPES = ("core/backend", "core/frontend", "k8s/core") +RELEVANT_PATTERNS = ( + "core/backend/airshelf/settings/*.py", + "core/backend/airshelf/urls.py", + "core/backend/apps/*/apps.py", + "core/backend/apps/*/models.py", + "core/backend/apps/*/urls.py", + "core/backend/apps/*/tasks.py", + "core/backend/apps/ai/providers/*.py", + "core/backend/requirements.txt", + "core/backend/Dockerfile", + "core/backend/docker-entrypoint.sh", + "k8s/core/*", + "k8s/core/**/*", + "core/frontend/package.json", + "core/frontend/src/App.tsx", + "core/frontend/src/api.ts", + "core/frontend/src/types.ts", + "core/frontend/src/routes/*", + "core/frontend/src/routes/**/*", + "core/frontend/src/components/*", + "core/frontend/src/components/**/*", + "core/frontend/vite.config.ts", + "core/frontend/Dockerfile", + "core/frontend/nginx.conf", +) +EXCLUDED_PARTS = frozenset( + { + ".git", + ".venv", + "venv", + "node_modules", + "dist", + "build", + ".next", + "coverage", + "__pycache__", + ".pytest_cache", + ".mypy_cache", + ".ruff_cache", + ".cache", + "cache", + "media", + "uploads", + "user_uploads", + "staticfiles", + "ab_results", + } +) +SENSITIVE_NAMES = frozenset( + { + "account.md", + "credentials.json", + "service-account.json", + "secrets.json", + "id_rsa", + "id_ed25519", + } +) +SENSITIVE_SUFFIXES = ( + ".pem", + ".key", + ".p12", + ".pfx", + ".sqlite", + ".sqlite3", + ".db", +) +SYNC_MARKER_RE = re.compile( + r"" +) +HEADING_RE = re.compile(r"^(#{1,6})\s+(.+?)\s*$", re.MULTILINE) +TS_EXPORT_RE = re.compile( + r"\bexport\s+(?:default\s+)?(?:async\s+)?" + r"(?:const|let|var|function|class|type|interface|enum)\s+([A-Za-z_$][\w$]*)" +) +QUOTED_PATH_RE = re.compile(r"[\"'`](/[^\"'`\s]*)[\"'`]") +MARKDOWN_LINK_RE = re.compile(r"!?\[[^\]]*\]\(([^)]+)\)") +MARKDOWN_REFERENCE_LINK_RE = re.compile(r"(?m)^\s*\[[^\]]+\]:\s*(\S+)") +ABSOLUTE_LOCAL_PATH_RE = re.compile( + r"(?i)(?:\b[A-Z]:[\\/]|file://|/(?:Users|home|var/folders|private/tmp)/)" +) +SECRET_CONTENT_PATTERNS = ( + re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"), + re.compile(r"\bsk-[A-Za-z0-9_-]{20,}\b"), + re.compile( + r"(?i)\b(?:api[_-]?key|secret|password)\b\s*[:=]\s*[\"']?[A-Za-z0-9_./+=-]{12,}" + ), +) +BACKEND_DOCUMENT = "core/backend/ARCHITECTURE.md" +FRONTEND_DOCUMENT = "core/frontend/ARCHITECTURE.md" +BACKEND_APP_SECTIONS = { + "common": "4.1 `common`", + "accounts": "4.2 `accounts`", + "products": "4.3 `products`", + "assets": "4.4 `assets`", + "projects": "4.5 `projects`", + "ai": "4.6 `ai`", + "billing": "4.7 `billing`", + "ops": "4.8 `ops`", + "adminpanel": "4.9 `adminpanel`", +} + + +class ScanError(RuntimeError): + """A safe, user-reportable scanner failure.""" + + +def normalize_path(value: str) -> str: + path = value.replace("\\", "/").removeprefix("./") + pure = PurePosixPath(path) + if not path or pure.is_absolute() or ".." in pure.parts: + return "" + return pure.as_posix() + + +def is_forbidden(path: str) -> bool: + normalized = normalize_path(path) + if not normalized: + return True + parts = tuple(part.lower() for part in PurePosixPath(normalized).parts) + name = parts[-1] + if any(part in EXCLUDED_PARTS for part in parts): + return True + if name == ".env" or name.startswith(".env."): + return True + if name in SENSITIVE_NAMES or name.endswith(SENSITIVE_SUFFIXES): + return True + return False + + +def is_relevant(path: str) -> bool: + normalized = normalize_path(path) + return bool(normalized) and not is_forbidden(normalized) and any( + fnmatch.fnmatchcase(normalized, pattern) for pattern in RELEVANT_PATTERNS + ) + + +def selected_documents(scope: str) -> tuple[str, ...]: + if scope == "backend": + return (BACKEND_DOCUMENT,) + if scope == "frontend": + return (FRONTEND_DOCUMENT,) + return TARGET_DOCUMENTS + + +def path_in_scope(path: str, scope: str) -> bool: + if scope == "backend": + return path.startswith("core/backend/") or path.startswith("k8s/core/") + if scope == "frontend": + return path.startswith("core/frontend/") + return path.startswith(("core/backend/", "core/frontend/", "k8s/core/")) + + +def filter_entries(entries: Iterable[dict[str, str]], scope: str) -> list[dict[str, str]]: + return [entry for entry in entries if path_in_scope(entry["path"], scope)] + + +def safe_candidate(root: Path, relative: str) -> Path | None: + normalized = normalize_path(relative) + if not normalized or is_forbidden(normalized): + return None + candidate = root.joinpath(*PurePosixPath(normalized).parts) + try: + resolved_root = root.resolve(strict=True) + resolved = candidate.resolve(strict=False) + resolved.relative_to(resolved_root) + except (OSError, ValueError): + return None + current = candidate + while current != root: + if current.is_symlink(): + return None + current = current.parent + return candidate + + +def read_bytes(root: Path, relative: str, warnings: set[str]) -> bytes | None: + candidate = safe_candidate(root, relative) + if candidate is None or not candidate.is_file(): + return None + try: + size = candidate.stat().st_size + if size > MAX_TEXT_BYTES: + warnings.add(f"file_too_large:{normalize_path(relative)}") + return None + return candidate.read_bytes() + except OSError: + warnings.add(f"file_unreadable:{normalize_path(relative)}") + return None + + +def read_text(root: Path, relative: str, warnings: set[str]) -> str | None: + data = read_bytes(root, relative, warnings) + if data is None: + return None + try: + return data.decode("utf-8-sig") + except UnicodeDecodeError: + warnings.add(f"file_not_utf8:{normalize_path(relative)}") + return None + + +def file_fingerprint(root: Path, relative: str, warnings: set[str]) -> str | None: + candidate = safe_candidate(root, relative) + if candidate is None or not candidate.is_file(): + return None + try: + if candidate.stat().st_size > MAX_HASH_BYTES: + warnings.add(f"fingerprint_file_too_large:{normalize_path(relative)}") + return None + digest = hashlib.sha256() + with candidate.open("rb") as stream: + for chunk in iter(lambda: stream.read(131_072), b""): + digest.update(chunk) + return digest.hexdigest() + except OSError: + warnings.add(f"fingerprint_unavailable:{normalize_path(relative)}") + return None + + +def git(root: Path, arguments: Sequence[str], *, required: bool = True) -> bytes: + environment = os.environ.copy() + environment["GIT_OPTIONAL_LOCKS"] = "0" + try: + result = subprocess.run( + ["git", "-c", "core.quotepath=false", *arguments], + cwd=root, + env=environment, + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + except OSError as error: + raise ScanError("git_unavailable") from error + if result.returncode and required: + raise ScanError(f"git_command_failed:{arguments[0]}") + return result.stdout if result.returncode == 0 else b"" + + +def locate_repository(start: Path) -> Path: + output = git(start, ("rev-parse", "--show-toplevel")) + try: + root = Path(output.decode("utf-8").strip()).resolve(strict=True) + except (UnicodeDecodeError, OSError) as error: + raise ScanError("repository_root_unavailable") from error + if not root.is_dir(): + raise ScanError("repository_root_unavailable") + return root + + +def decode_git_token(token: bytes) -> str: + return token.decode("utf-8", errors="replace") + + +def parse_name_status( + data: bytes, path_filter: Callable[[str], bool] = is_relevant +) -> list[dict[str, str]]: + tokens = [decode_git_token(token) for token in data.split(b"\0") if token] + entries: list[dict[str, str]] = [] + index = 0 + while index < len(tokens): + status = tokens[index] + index += 1 + if index >= len(tokens): + break + if status.startswith(("R", "C")): + old_path = normalize_path(tokens[index]) + index += 1 + if index >= len(tokens): + break + new_path = normalize_path(tokens[index]) + index += 1 + if path_filter(new_path): + entry = {"path": new_path, "status": status} + if path_filter(old_path): + entry["old_path"] = old_path + entries.append(entry) + elif path_filter(old_path): + entries.append({"path": old_path, "status": "D"}) + continue + path = normalize_path(tokens[index]) + index += 1 + if path_filter(path): + entries.append({"path": path, "status": status}) + return sorted(entries, key=lambda item: (item["path"], item["status"])) + + +def diff_entries(root: Path, arguments: Sequence[str]) -> list[dict[str, str]]: + output = git( + root, + (*arguments, "--name-status", "--find-renames", "-z", "--", *GIT_SCOPES), + ) + return parse_name_status(output) + + +def untracked_entries(root: Path) -> list[dict[str, str]]: + output = git( + root, + ("ls-files", "--others", "--exclude-standard", "-z", "--", *GIT_SCOPES), + ) + entries = [] + for token in output.split(b"\0"): + if not token: + continue + path = normalize_path(decode_git_token(token)) + if is_relevant(path): + entries.append({"path": path, "status": "??"}) + return sorted(entries, key=lambda item: item["path"]) + + +def all_diff_entries(root: Path, arguments: Sequence[str]) -> list[dict[str, str]]: + output = git( + root, + (*arguments, "--name-status", "--find-renames", "-z", "--"), + ) + return parse_name_status(output, lambda path: bool(normalize_path(path))) + + +def all_untracked_entries(root: Path) -> list[dict[str, str]]: + output = git(root, ("ls-files", "--others", "--exclude-standard", "-z", "--")) + entries = [] + for token in output.split(b"\0"): + if not token: + continue + path = normalize_path(decode_git_token(token)) + if path: + entries.append({"path": path, "status": "??"}) + return sorted(entries, key=lambda item: item["path"]) + + +def worktree_guard(root: Path) -> dict[str, Any]: + groups = ( + ("staged", all_diff_entries(root, ("diff", "--cached"))), + ("unstaged", all_diff_entries(root, ("diff",))), + ("untracked", all_untracked_entries(root)), + ) + states: dict[str, set[str]] = {} + forbidden_paths: set[str] = set() + for source, entries in groups: + for entry in entries: + paths = [entry["path"]] + if entry.get("old_path"): + paths.append(entry["old_path"]) + for path in paths: + if is_forbidden(path): + forbidden_paths.add(path) + continue + states.setdefault(path, set()).add(f"{source}:{entry['status']}") + return { + "files": [ + {"path": path, "states": sorted(values)} + for path, values in sorted(states.items()) + ], + "forbidden_change_count": len(forbidden_paths), + } + + +def verification_error( + code: str, message: str, suggestion: str, **details: Any +) -> dict[str, Any]: + return { + "code": code, + "message": message, + "suggestion": suggestion, + **details, + } + + +def compare_worktree_guards( + before: dict[str, Any], current: dict[str, Any], allowed_documents: set[str] +) -> list[dict[str, Any]]: + errors: list[dict[str, Any]] = [] + before_map = {item["path"]: item["states"] for item in before.get("files", [])} + current_map = {item["path"]: item["states"] for item in current.get("files", [])} + for path in sorted(set(before_map) | set(current_map)): + if path in allowed_documents: + continue + if before_map.get(path) != current_map.get(path): + errors.append( + verification_error( + "write_outside_allowlist", + "检测到计划外文件状态发生变化。", + "撤销本流程对该文件的修改，或重新生成包含正确范围的新计划。", + path=path, + ) + ) + if before.get("forbidden_change_count", 0) != current.get( + "forbidden_change_count", 0 + ): + errors.append( + verification_error( + "sensitive_or_excluded_path_changed", + "检测到敏感或排除路径的工作区状态发生变化。", + "停止文档同步并由用户单独检查敏感或排除文件。", + ) + ) + return errors + + +def current_head(root: Path) -> str | None: + output = git(root, ("rev-parse", "--verify", "HEAD"), required=False) + value = output.decode("ascii", errors="ignore").strip().lower() + return value if re.fullmatch(r"[0-9a-f]{40}", value) else None + + +def marker_is_usable(root: Path, marker: str, head: str | None) -> bool: + if head is None: + return False + exists = subprocess.run( + ["git", "cat-file", "-e", f"{marker}^{{commit}}"], + cwd=root, + env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ).returncode == 0 + if not exists: + return False + return subprocess.run( + ["git", "merge-base", "--is-ancestor", marker, head], + cwd=root, + env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + check=False, + ).returncode == 0 + + +def document_states( + root: Path, warnings: set[str], documents: Sequence[str] = TARGET_DOCUMENTS +) -> tuple[dict[str, dict[str, Any]], dict[str, str | None]]: + states: dict[str, dict[str, Any]] = {} + markers: dict[str, str | None] = {} + for path in documents: + data = read_bytes(root, path, warnings) + if data is None: + states[path] = {"exists": False, "headings": [], "sha256": None} + markers[path] = None + warnings.add(f"target_document_missing:{path}") + continue + try: + text = data.decode("utf-8-sig") + except UnicodeDecodeError: + text = "" + warnings.add(f"target_document_not_utf8:{path}") + marker_match = SYNC_MARKER_RE.search(text) + marker = marker_match.group(1).lower() if marker_match else None + markers[path] = marker + headings = [ + {"level": len(match.group(1)), "title": match.group(2).strip()} + for match in HEADING_RE.finditer(text) + ] + states[path] = { + "exists": True, + "headings": headings, + "sha256": hashlib.sha256(data).hexdigest(), + "sync_commit": marker, + } + return states, markers + + +def walk_relevant_files(root: Path) -> list[str]: + roots = ( + "core/backend/airshelf", + "core/backend/apps", + "core/backend/requirements.txt", + "core/backend/Dockerfile", + "core/backend/docker-entrypoint.sh", + "core/frontend/package.json", + "core/frontend/src", + "core/frontend/vite.config.ts", + "core/frontend/Dockerfile", + "core/frontend/nginx.conf", + "k8s/core", + ) + paths: set[str] = set() + for relative in roots: + candidate = safe_candidate(root, relative) + if candidate is None or not candidate.exists(): + continue + candidates: Iterable[Path] = (candidate,) if candidate.is_file() else candidate.rglob("*") + for item in candidates: + if not item.is_file() or item.is_symlink(): + continue + try: + rel = item.relative_to(root).as_posix() + except ValueError: + continue + if is_relevant(rel): + paths.add(rel) + return sorted(paths) + + +def scope_content_fingerprint( + root: Path, scope: str, warnings: set[str] +) -> tuple[str, int]: + paths = [path for path in walk_relevant_files(root) if path_in_scope(path, scope)] + fingerprints = {path: file_fingerprint(root, path, warnings) for path in paths} + canonical = json.dumps( + fingerprints, ensure_ascii=True, sort_keys=True, separators=(",", ":") + ) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20], len(paths) + + +def parse_python(root: Path, relative: str, warnings: set[str]) -> ast.Module | None: + text = read_text(root, relative, warnings) + if text is None: + return None + try: + return ast.parse(text, filename=relative) + except SyntaxError: + warnings.add(f"python_parse_failed:{relative}") + return None + + +def dotted_name(node: ast.AST) -> str: + if isinstance(node, ast.Name): + return node.id + if isinstance(node, ast.Attribute): + parent = dotted_name(node.value) + return f"{parent}.{node.attr}" if parent else node.attr + return "" + + +def model_classes(root: Path, relative: str, warnings: set[str]) -> list[str]: + tree = parse_python(root, relative, warnings) + if tree is None: + return [] + classes = [] + for node in tree.body: + if not isinstance(node, ast.ClassDef): + continue + bases = [dotted_name(base).split(".")[-1] for base in node.bases] + if any(base.lower().endswith("model") or base == "AbstractUser" for base in bases): + classes.append(node.name) + return sorted(classes) + + +def django_routes(root: Path, relative: str, warnings: set[str]) -> list[dict[str, str | None]]: + tree = parse_python(root, relative, warnings) + if tree is None: + return [] + routes: list[dict[str, str | None]] = [] + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not node.args: + continue + if dotted_name(node.func).split(".")[-1] not in {"path", "re_path"}: + continue + first = node.args[0] + if not isinstance(first, ast.Constant) or not isinstance(first.value, str): + continue + included: str | None = None + if len(node.args) > 1 and isinstance(node.args[1], ast.Call): + call = node.args[1] + if dotted_name(call.func).split(".")[-1] == "include" and call.args: + value = call.args[0] + if isinstance(value, ast.Constant) and isinstance(value.value, str): + included = value.value + routes.append({"include": included, "path": first.value}) + return sorted(routes, key=lambda item: (item["path"], item["include"] or "")) + + +def celery_tasks(root: Path, relative: str, warnings: set[str]) -> list[str]: + tree = parse_python(root, relative, warnings) + if tree is None: + return [] + tasks = [] + for node in tree.body: + if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + decorator_names = [] + for decorator in node.decorator_list: + target = decorator.func if isinstance(decorator, ast.Call) else decorator + decorator_names.append(dotted_name(target).split(".")[-1]) + if any(name in {"shared_task", "task", "periodic_task"} for name in decorator_names): + tasks.append(node.name) + return sorted(tasks) + + +def top_level_classes(root: Path, relative: str, warnings: set[str]) -> list[str]: + tree = parse_python(root, relative, warnings) + if tree is None: + return [] + return sorted(node.name for node in tree.body if isinstance(node, ast.ClassDef)) + + +def assignment_names(tree: ast.Module) -> set[str]: + names: set[str] = set() + for node in tree.body: + targets: list[ast.AST] = [] + if isinstance(node, ast.Assign): + targets.extend(node.targets) + elif isinstance(node, ast.AnnAssign): + targets.append(node.target) + for target in targets: + if isinstance(target, ast.Name): + names.add(target.id) + return names + + +def requirement_names(root: Path, warnings: set[str]) -> list[str]: + text = read_text(root, "core/backend/requirements.txt", warnings) + if text is None: + return [] + names = set() + for line in text.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith(("#", "-")): + continue + match = re.match(r"([A-Za-z0-9_.-]+)", stripped) + if match: + names.add(match.group(1).lower()) + return sorted(names) + + +def collect_backend(root: Path, warnings: set[str]) -> dict[str, Any]: + apps_root = root / "core" / "backend" / "apps" + apps = [] + if apps_root.is_dir() and not apps_root.is_symlink(): + apps = sorted( + item.name + for item in apps_root.iterdir() + if item.is_dir() + and not item.is_symlink() + and not is_forbidden(f"core/backend/apps/{item.name}") + and ((item / "apps.py").is_file() or (item / "__init__.py").is_file()) + ) + + models: dict[str, list[str]] = {} + app_routes: dict[str, list[dict[str, str | None]]] = {} + tasks: dict[str, list[str]] = {} + for app in apps: + model_path = f"core/backend/apps/{app}/models.py" + route_path = f"core/backend/apps/{app}/urls.py" + task_path = f"core/backend/apps/{app}/tasks.py" + if safe_candidate(root, model_path) and (root / model_path).is_file(): + models[app] = model_classes(root, model_path, warnings) + if safe_candidate(root, route_path) and (root / route_path).is_file(): + app_routes[app] = django_routes(root, route_path, warnings) + if safe_candidate(root, task_path) and (root / task_path).is_file(): + tasks[app] = celery_tasks(root, task_path, warnings) + + provider_root = root / "core" / "backend" / "apps" / "ai" / "providers" + providers = [] + if provider_root.is_dir() and not provider_root.is_symlink(): + for file in sorted(provider_root.glob("*.py"), key=lambda item: item.name): + if file.name == "__init__.py" or file.is_symlink(): + continue + relative = file.relative_to(root).as_posix() + providers.append( + {"classes": top_level_classes(root, relative, warnings), "path": relative} + ) + + settings = {} + settings_root = root / "core" / "backend" / "airshelf" / "settings" + if settings_root.is_dir() and not settings_root.is_symlink(): + important = { + "AUTHENTICATION_BACKENDS", + "AUTH_USER_MODEL", + "CACHES", + "DATABASES", + "INSTALLED_APPS", + "MIDDLEWARE", + "REST_FRAMEWORK", + "STORAGES", + } + prefixes = ("AI_", "ARK_", "CELERY_", "MODEL_", "REDIS_", "TOKENSSR_", "TOS_", "YUNQI_") + for file in sorted(settings_root.glob("*.py"), key=lambda item: item.name): + if file.name == "__init__.py" or file.is_symlink(): + continue + relative = file.relative_to(root).as_posix() + tree = parse_python(root, relative, warnings) + symbols = [] if tree is None else sorted( + name for name in assignment_names(tree) if name in important or name.startswith(prefixes) + ) + settings[file.stem] = symbols + + base_settings = read_text(root, "core/backend/airshelf/settings/base.py", warnings) or "" + configured_apps = sorted(set(re.findall(r"[\"']apps\.([a-zA-Z0-9_]+)", base_settings))) + deployment_candidates = ( + "core/backend/Dockerfile", + "core/backend/docker-entrypoint.sh", + ) + deployment_files = [ + path for path in deployment_candidates if (safe_candidate(root, path) or Path()).is_file() + ] + k8s_files = [path for path in walk_relevant_files(root) if path.startswith("k8s/core/")] + return { + "api_routes": django_routes(root, "core/backend/airshelf/urls.py", warnings), + "app_routes": app_routes, + "apps": apps, + "celery_tasks": tasks, + "configured_apps": configured_apps, + "deployment_files": deployment_files + k8s_files, + "models": models, + "providers": providers, + "requirements": requirement_names(root, warnings), + "settings_symbols": settings, + } + + +def relative_source_files(root: Path, base: str, suffixes: tuple[str, ...]) -> list[str]: + directory = safe_candidate(root, base) + if directory is None or not directory.is_dir() or directory.is_symlink(): + return [] + files = [] + for item in directory.rglob("*"): + if not item.is_file() or item.is_symlink() or item.suffix.lower() not in suffixes: + continue + relative = item.relative_to(root).as_posix() + if not is_forbidden(relative): + files.append(relative) + return sorted(files) + + +def typescript_exports(root: Path, relative: str, warnings: set[str]) -> list[str]: + text = read_text(root, relative, warnings) + return [] if text is None else sorted(set(TS_EXPORT_RE.findall(text))) + + +def quoted_paths(root: Path, relative: str, warnings: set[str]) -> list[str]: + text = read_text(root, relative, warnings) + return [] if text is None else sorted(set(QUOTED_PATH_RE.findall(text))) + + +def stage_order(root: Path, warnings: set[str]) -> list[str]: + text = read_text(root, "core/frontend/src/routes/stage-config.ts", warnings) + if text is None: + return [] + match = re.search(r"\bstageOrder\s*=\s*\[([^\]]*)\]", text, re.DOTALL) + if not match: + return [] + return re.findall(r"[\"']([A-Za-z0-9_-]+)[\"']", match.group(1)) + + +def frontend_dependencies(root: Path, warnings: set[str]) -> dict[str, list[str]]: + text = read_text(root, "core/frontend/package.json", warnings) + if text is None: + return {"dependencies": [], "devDependencies": []} + try: + package = json.loads(text) + except json.JSONDecodeError: + warnings.add("json_parse_failed:core/frontend/package.json") + return {"dependencies": [], "devDependencies": []} + return { + group: sorted(str(name) for name in package.get(group, {}).keys()) + for group in ("dependencies", "devDependencies") + } + + +def collect_frontend(root: Path, warnings: set[str]) -> dict[str, Any]: + deployment_candidates = ( + "core/frontend/Dockerfile", + "core/frontend/nginx.conf", + "core/frontend/vite.config.ts", + ) + deployment_files = [ + path for path in deployment_candidates if (safe_candidate(root, path) or Path()).is_file() + ] + return { + "api_endpoints": quoted_paths(root, "core/frontend/src/api.ts", warnings), + "api_exports": typescript_exports(root, "core/frontend/src/api.ts", warnings), + "component_files": relative_source_files( + root, "core/frontend/src/components", (".ts", ".tsx") + ), + "dependencies": frontend_dependencies(root, warnings), + "deployment_files": deployment_files, + "route_files": relative_source_files(root, "core/frontend/src/routes", (".ts", ".tsx")), + "route_paths": quoted_paths(root, "core/frontend/src/routes/route-config.ts", warnings), + "stage_order": stage_order(root, warnings), + "type_exports": typescript_exports(root, "core/frontend/src/types.ts", warnings), + } + + +def changed_line_text(root: Path, base: str, path: str, warnings: set[str]) -> list[str]: + if not is_relevant(path): + return [] + output = git( + root, + ("diff", "--no-ext-diff", "--unified=0", base, "--", path), + required=False, + ) + if not output: + current = read_text(root, path, warnings) + return [] if current is None else [f"+{line}" for line in current.splitlines()] + if len(output) > 2_000_000: + warnings.add(f"diff_too_large:{path}") + output = output[:2_000_000] + text = output.decode("utf-8", errors="replace") + return [ + line + for line in text.splitlines() + if line.startswith(("+", "-")) and not line.startswith(("+++", "---")) + ] + + +def text_at_commit(root: Path, commit: str, path: str, warnings: set[str]) -> str | None: + if not is_relevant(path): + return None + output = git(root, ("show", f"{commit}:{path}"), required=False) + if not output: + return None + if len(output) > MAX_TEXT_BYTES: + warnings.add(f"base_file_too_large:{path}") + return None + try: + return output.decode("utf-8-sig") + except UnicodeDecodeError: + warnings.add(f"base_file_not_utf8:{path}") + return None + + +def change_index(entries: Iterable[dict[str, str]]) -> dict[str, set[str]]: + indexed: dict[str, set[str]] = {} + for entry in entries: + status = entry["status"] + code = "A" if status == "??" else status[:1] + path = entry["path"] + if path: + indexed.setdefault(path, set()).add(code) + old_path = entry.get("old_path") + if old_path: + indexed.setdefault(old_path, set()).add("D") + indexed[path].add("A") + return indexed + + +def dependency_names_from_package(text: str | None) -> set[str] | None: + if text is None: + return set() + try: + package = json.loads(text) + except json.JSONDecodeError: + return None + names: set[str] = set() + for group in ("dependencies", "devDependencies"): + values = package.get(group, {}) + if isinstance(values, dict): + names.update(str(name) for name in values) + return names + + +def dependency_names_from_requirements(text: str | None) -> set[str]: + if text is None: + return set() + names = set() + for line in text.splitlines(): + stripped = line.strip() + if not stripped or stripped.startswith(("#", "-")): + continue + match = re.match(r"([A-Za-z0-9_.-]+)", stripped) + if match: + names.add(match.group(1).lower()) + return names + + +def dependency_set_changed( + root: Path, base: str, path: str, warnings: set[str] +) -> bool: + before = text_at_commit(root, base, path, warnings) + after = read_text(root, path, warnings) + if path.endswith("package.json"): + before_names = dependency_names_from_package(before) + after_names = dependency_names_from_package(after) + if before_names is None or after_names is None: + warnings.add(f"dependency_parse_failed:{path}") + return True + return before_names != after_names + return dependency_names_from_requirements(before) != dependency_names_from_requirements(after) + + +def stage_order_from_text(text: str | None) -> list[str]: + if text is None: + return [] + match = re.search(r"\bstageOrder\s*=\s*\[([^\]]*)\]", text, re.DOTALL) + if not match: + return [] + return re.findall(r"[\"']([A-Za-z0-9_-]+)[\"']", match.group(1)) + + +def stage_order_changed(root: Path, base: str, warnings: set[str]) -> bool: + path = "core/frontend/src/routes/stage-config.ts" + before = text_at_commit(root, base, path, warnings) + after = read_text(root, path, warnings) + return stage_order_from_text(before) != stage_order_from_text(after) + + +def frontend_page_section(path: str) -> str: + if "/admin/" in path: + return "8.7 平台超管后台" + stem = PurePosixPath(path).stem + sections = { + "products": "8.1 商品", + "product-detail": "8.1 商品", + "product-create": "8.1 商品", + "projects": "8.2 视频项目", + "pipeline": "8.3 生产管线", + "ai-tools": "8.4 图片创作", + "models": "8.4 图片创作", + "free-create": "8.5 自由视频创作", + "library": "8.6 资产库与垃圾桶", + "trash": "8.6 资产库与垃圾桶", + } + return sections.get(stem, "8. 页面模块") + + +def impact_spec( + path: str, + statuses: set[str], + changed_lines: Sequence[str], + *, + dependency_changed: bool = False, + stages_changed: bool = False, +) -> dict[str, str] | None: + changed = "\n".join(changed_lines) + added_or_deleted = bool(statuses & {"A", "D", "R", "C"}) + + if path.startswith("core/backend/") or path.startswith("k8s/core/"): + document = BACKEND_DOCUMENT + app_match = re.match(r"core/backend/apps/([^/]+)/([^/]+\.py)$", path) + app = app_match.group(1) if app_match else "" + filename = app_match.group(2) if app_match else "" + app_section = BACKEND_APP_SECTIONS.get(app, "2. 目录结构、4. 应用模块") + + if path.startswith("core/backend/airshelf/settings/") and re.search( + r"(?i)(INSTALLED_APPS|MIDDLEWARE|DATABASE|DB_ENGINE|\bENGINE\b|CACHE|REDIS|CELERY|" + r"REST_FRAMEWORK|AUTH_|Authentication|Permission|STORAGES|TOS_|ARK_|" + r"YUNQI_|TOKENSSR_|MODEL_|AI_|apps\.)", + changed, + ): + return { + "change_type": "backend.settings_boundary", + "classification": "architecture", + "document": document, + "reason": "后端框架、基础设施或鉴权配置边界发生变化", + "section": "3.1 Settings、3.3 鉴权", + } + if path == "core/backend/airshelf/urls.py" and ( + added_or_deleted or re.search(r"\b(path|re_path|include)\s*\(", changed) + ): + return { + "change_type": "backend.root_routes", + "classification": "architecture", + "document": document, + "reason": "后端一级路由入口发生变化", + "section": "3.2 路由", + } + if filename == "apps.py" and ( + added_or_deleted or re.search(r"(?m)^[+-].*\bname\s*=", changed) + ): + return { + "change_type": "backend.django_apps", + "classification": "architecture", + "document": document, + "reason": f"Django 应用 `{app}` 的注册结构发生变化", + "section": "2. 目录结构、4. 应用模块", + } + if filename == "models.py" and ( + added_or_deleted + or re.search(r"(?m)^[+-]\s*class\s+\w+\s*\([^\n]*(?:Model|AbstractUser)", changed) + ): + return { + "change_type": "backend.model_classes", + "classification": "architecture", + "document": document, + "reason": f"`{app}` 核心模型类型发生增删或继承变化", + "section": app_section, + } + if filename == "urls.py" and ( + added_or_deleted or re.search(r"\b(path|re_path|include)\s*\(", changed) + ): + return { + "change_type": "backend.app_routes", + "classification": "architecture", + "document": document, + "reason": f"`{app}` 应用接口边界发生变化", + "section": app_section, + } + if filename == "tasks.py" and ( + added_or_deleted or re.search(r"(shared_task|periodic_task|\.task\s*\()", changed) + ): + return { + "change_type": "backend.celery_tasks", + "classification": "architecture", + "document": document, + "reason": f"`{app}` 的 Celery 任务入口发生变化", + "section": "7.1 Celery", + } + if path.startswith("core/backend/apps/ai/providers/") and ( + added_or_deleted + or re.search(r"(?m)^[+-]\s*class\s+", changed) + or ( + PurePosixPath(path).name == "__init__.py" + and re.search(r"(?m)^[+-]\s*(?:from|import)\s+", changed) + ) + ): + return { + "change_type": "backend.ai_providers", + "classification": "architecture", + "document": document, + "reason": "AI Provider 类型或适配层结构发生变化", + "section": "6. AI Provider 架构", + } + if path == "core/backend/requirements.txt" and dependency_changed: + return { + "change_type": "backend.dependencies", + "classification": "architecture", + "document": document, + "reason": "后端依赖集合发生变化", + "section": "10. 部署", + } + if path in {"core/backend/Dockerfile", "core/backend/docker-entrypoint.sh"} or path.startswith( + "k8s/core/" + ): + if changed_lines or added_or_deleted: + return { + "change_type": "backend.deployment", + "classification": "architecture", + "document": document, + "reason": "后端容器或部署拓扑发生变化", + "section": "10. 部署", + } + return None + + if path.startswith("core/frontend/"): + document = FRONTEND_DOCUMENT + if path == "core/frontend/package.json" and dependency_changed: + return { + "change_type": "frontend.dependencies", + "classification": "architecture", + "document": document, + "reason": "前端依赖集合发生变化", + "section": "2. 技术栈", + } + if path == "core/frontend/src/routes/route-config.ts" and ( + added_or_deleted + or re.search(r"(?m)^[+-].*(?:[\"'`]/|type\s+Page|mainNav|supplementNav|pathForPage)", changed) + ): + return { + "change_type": "frontend.routing", + "classification": "architecture", + "document": document, + "reason": "前端页面路由或导航边界发生变化", + "section": "5. 路由架构", + } + if path == "core/frontend/src/routes/stage-config.ts" and stages_changed: + return { + "change_type": "frontend.pipeline_stages", + "classification": "architecture", + "document": document, + "reason": "五阶段顺序或阶段标识发生变化", + "section": "8.3 生产管线", + } + if path.startswith("core/frontend/src/routes/") and PurePosixPath(path).suffix in { + ".ts", + ".tsx", + } and added_or_deleted: + return { + "change_type": "frontend.pages", + "classification": "architecture", + "document": document, + "reason": "正式业务页面模块发生增删或移动", + "section": frontend_page_section(path), + } + if path == "core/frontend/src/App.tsx" and re.search( + r"(?i)(createContext|Context\.Provider|useReducer|redux|zustand|RouterProvider|BrowserRouter)", + changed, + ): + return { + "change_type": "frontend.app_controller", + "classification": "architecture", + "document": document, + "reason": "应用总控制器或全局状态边界发生变化", + "section": "4.2 `App.tsx`", + } + if path == "core/frontend/src/api.ts" and re.search( + r"(?i)(VITE_API_BASE_URL|Authorization|localStorage|sessionStorage|fetch\s*\(|" + r"axios|(?:const|function)\s+request\b|export\s+const\s+\w*Api\b)", + changed, + ): + return { + "change_type": "frontend.api_boundary", + "classification": "architecture", + "document": document, + "reason": "前端请求、鉴权或 API SDK 边界发生变化", + "section": "6. API 层、7. 登录、存储与权限", + } + if path.startswith("core/frontend/src/components/") and added_or_deleted and re.search( + r"(?i)(shell|layout|router|provider|store|context)", PurePosixPath(path).stem + ): + return { + "change_type": "frontend.shared_infrastructure", + "classification": "architecture", + "document": document, + "reason": "前端共享基础组件发生增删或移动", + "section": "3. 目录结构、4. 应用入口", + } + if path in { + "core/frontend/Dockerfile", + "core/frontend/nginx.conf", + "core/frontend/vite.config.ts", + } and (changed_lines or added_or_deleted): + return { + "change_type": "frontend.deployment", + "classification": "architecture", + "document": document, + "reason": "前端构建、代理或容器部署边界发生变化", + "section": "11. 构建与部署", + } + return None + + +def baseline_impacts(document: str, baseline_files: Sequence[str]) -> list[dict[str, Any]]: + if document == BACKEND_DOCUMENT: + groups = ( + ( + "backend.baseline.apps", + "2. 目录结构、4. 应用模块", + "缺少同步标记，需要核对 Django 应用与核心模型基线", + lambda path: "/apps/" in path and path.endswith(("apps.py", "models.py", "urls.py")), + ), + ( + "backend.baseline.entry", + "3. Django 工程入口", + "缺少同步标记，需要核对 Settings 与一级路由基线", + lambda path: "/airshelf/settings/" in path or path.endswith("airshelf/urls.py"), + ), + ( + "backend.baseline.execution", + "6. AI Provider 架构、7. 任务执行模型", + "缺少同步标记，需要核对 Provider 与 Celery 任务基线", + lambda path: "/providers/" in path or path.endswith("tasks.py"), + ), + ( + "backend.baseline.deployment", + "10. 部署", + "缺少同步标记，需要核对依赖、容器与部署基线", + lambda path: path.endswith(("requirements.txt", "Dockerfile", "docker-entrypoint.sh")) + or path.startswith("k8s/core/"), + ), + ) + else: + groups = ( + ( + "frontend.baseline.stack", + "2. 技术栈", + "缺少同步标记，需要核对前端技术依赖基线", + lambda path: path.endswith("package.json"), + ), + ( + "frontend.baseline.structure", + "3. 目录结构、5. 路由架构、8. 页面模块", + "缺少同步标记，需要核对页面、路由与共享组件基线", + lambda path: "/src/routes/" in path or "/src/components/" in path, + ), + ( + "frontend.baseline.boundaries", + "4. 应用入口、6. API 层", + "缺少同步标记，需要核对应用入口与 API SDK 基线", + lambda path: path.endswith(("src/App.tsx", "src/api.ts", "src/types.ts")), + ), + ( + "frontend.baseline.deployment", + "11. 构建与部署", + "缺少同步标记，需要核对构建、代理与容器基线", + lambda path: path.endswith(("Dockerfile", "nginx.conf", "vite.config.ts")), + ), + ) + impacts = [] + for change_type, section, reason, matches in groups: + evidence = sorted(path for path in baseline_files if matches(path))[:24] + if evidence: + impacts.append( + { + "change_type": change_type, + "classification": "baseline_review", + "document": document, + "evidence": evidence, + "reason": reason, + "section": section, + } + ) + return impacts + + +def map_impacts( + root: Path, + head: str | None, + documents: Sequence[str], + usable_markers: dict[str, str], + baseline_files: Sequence[str], + staged: Sequence[dict[str, str]], + unstaged: Sequence[dict[str, str]], + untracked: Sequence[dict[str, str]], + warnings: set[str], +) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + impacts: list[dict[str, Any]] = [] + ordinary: dict[str, str] = {} + for document in documents: + base = usable_markers.get(document) + if base is None or head is None: + impacts.extend(baseline_impacts(document, baseline_files)) + continue + committed = diff_entries(root, ("diff", f"{base}..{head}")) + indexed = change_index([*committed, *staged, *unstaged, *untracked]) + if document == BACKEND_DOCUMENT: + domain_paths = sorted( + path + for path in indexed + if path.startswith("core/backend/") or path.startswith("k8s/core/") + ) + else: + domain_paths = sorted( + path for path in indexed if path.startswith("core/frontend/") + ) + for path in domain_paths: + lines = changed_line_text(root, base, path, warnings) + dependency_changed = path in { + "core/backend/requirements.txt", + "core/frontend/package.json", + } and dependency_set_changed(root, base, path, warnings) + stages_changed = path == "core/frontend/src/routes/stage-config.ts" and stage_order_changed( + root, base, warnings + ) + spec = impact_spec( + path, + indexed[path], + lines, + dependency_changed=dependency_changed, + stages_changed=stages_changed, + ) + if spec is None: + ordinary[path] = "未发现会改变架构文档事实的结构信号" + continue + impacts.append({**spec, "evidence": [path]}) + + buckets: dict[tuple[str, str, str, str, str], set[str]] = {} + for impact in impacts: + key = ( + impact["document"], + impact["section"], + impact["change_type"], + impact["classification"], + impact["reason"], + ) + buckets.setdefault(key, set()).update(impact["evidence"]) + merged = [ + { + "change_type": key[2], + "classification": key[3], + "document": key[0], + "evidence": sorted(evidence), + "reason": key[4], + "section": key[1], + } + for key, evidence in buckets.items() + ] + merged.sort(key=lambda item: (item["document"], item["section"], item["change_type"])) + non_architecture = [ + {"path": path, "reason": reason} for path, reason in sorted(ordinary.items()) + ] + return merged, non_architecture + + +def encode_guard(payload: dict[str, Any]) -> str: + raw = json.dumps(payload, ensure_ascii=True, sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) + encoded = base64.urlsafe_b64encode(zlib.compress(raw, level=9)).decode("ascii").rstrip("=") + checksum = hashlib.sha256(raw).hexdigest()[:16] + return f"{encoded}.{checksum}" + + +def decode_guard(token: str) -> dict[str, Any]: + try: + encoded, checksum = token.rsplit(".", 1) + padding = "=" * (-len(encoded) % 4) + raw = zlib.decompress(base64.urlsafe_b64decode(encoded + padding)) + if hashlib.sha256(raw).hexdigest()[:16] != checksum: + raise ValueError("checksum") + payload = json.loads(raw.decode("utf-8")) + except (ValueError, UnicodeDecodeError, json.JSONDecodeError, zlib.error) as error: + raise ScanError("verification_guard_invalid") from error + if not isinstance(payload, dict) or payload.get("guard_version") != 1: + raise ScanError("verification_guard_invalid") + return payload + + +def headings_from_text(text: str, maximum_level: int = 3) -> list[dict[str, Any]]: + return [ + {"level": len(match.group(1)), "title": match.group(2).strip()} + for match in HEADING_RE.finditer(text) + if len(match.group(1)) <= maximum_level + ] + + +def section_records(text: str) -> list[dict[str, Any]]: + matches = [ + match for match in HEADING_RE.finditer(text) if len(match.group(1)) <= 3 + ] + records: list[dict[str, Any]] = [] + stack: list[tuple[int, str]] = [] + key_counts: dict[str, int] = {} + for index, match in enumerate(matches): + level = len(match.group(1)) + title = match.group(2).strip() + while stack and stack[-1][0] >= level: + stack.pop() + ancestors = [item[1] for item in stack] + stack.append((level, title)) + end = len(text) + for candidate in matches[index + 1 :]: + if len(candidate.group(1)) <= level: + end = candidate.start() + break + base_key = f"{level}:{title}" + key_counts[base_key] = key_counts.get(base_key, 0) + 1 + key = ( + base_key + if key_counts[base_key] == 1 + else f"{base_key}#{key_counts[base_key]}" + ) + records.append( + { + "ancestors": ancestors, + "end": end, + "hash": hashlib.sha256(text[match.start() : end].encode("utf-8")).hexdigest(), + "key": key, + "level": level, + "start": match.start(), + "title": title, + } + ) + return records + + +def protected_section_hashes(text: str, planned_sections: Sequence[str]) -> dict[str, str]: + records = section_records(text) + planned = set(planned_sections) + protected: dict[str, str] = {} + for record in records: + descendant_planned = any( + other["title"] in planned + and other["start"] > record["start"] + and other["start"] < record["end"] + for other in records + ) + affected = ( + record["title"] in planned + or any(ancestor in planned for ancestor in record["ancestors"]) + or descendant_planned + ) + if not affected: + protected[record["key"]] = record["hash"] + return protected + + +def document_header_hash(text: str) -> str: + second_level = re.search(r"(?m)^##\s+", text) + header = text[: second_level.start()] if second_level else text + kept_lines = [] + for line in header.splitlines(): + stripped = line.strip() + if re.fullmatch(r">\s*最后核对：\d{4}-\d{2}-\d{2}。", stripped): + continue + if SYNC_MARKER_RE.fullmatch(stripped): + continue + kept_lines.append(line.rstrip()) + normalized = "\n".join(kept_lines).strip() + return hashlib.sha256(normalized.encode("utf-8")).hexdigest() + + +def is_ordered_subsequence(before: Sequence[str], current: Sequence[str]) -> bool: + iterator = iter(current) + return all(any(candidate == expected for candidate in iterator) for expected in before) + + +def validate_headings( + document: str, + before: Sequence[dict[str, Any]], + current: Sequence[dict[str, Any]], +) -> list[dict[str, Any]]: + errors: list[dict[str, Any]] = [] + before_primary = [item["title"] for item in before if item["level"] <= 2] + current_primary = [item["title"] for item in current if item["level"] <= 2] + if before_primary != current_primary: + errors.append( + verification_error( + "primary_headings_changed", + "原有一级或二级标题被删除、改名、增加或调整顺序。", + "恢复原有一级和二级标题；内容变化应限制在已批准章节正文内。", + document=document, + ) + ) + before_tertiary = [item["title"] for item in before if item["level"] == 3] + current_tertiary = [item["title"] for item in current if item["level"] == 3] + if not is_ordered_subsequence(before_tertiary, current_tertiary): + errors.append( + verification_error( + "existing_subheadings_removed_or_reordered", + "原有三级标题被删除或调整顺序。", + "恢复原有三级标题及顺序；如需新增标题，只能追加到已批准章节。", + document=document, + ) + ) + return errors + + +def markdown_targets(text: str) -> list[str]: + values = MARKDOWN_LINK_RE.findall(text) + MARKDOWN_REFERENCE_LINK_RE.findall(text) + targets = [] + for value in values: + target = value.strip().strip("<>") + if " " in target and not target.startswith(("http://", "https://")): + target = target.split(" ", 1)[0] + targets.append(target) + return targets + + +def validate_relative_links(root: Path, document: str, text: str) -> list[dict[str, Any]]: + errors: list[dict[str, Any]] = [] + document_path = root.joinpath(*PurePosixPath(document).parts) + for raw_target in markdown_targets(text): + if not raw_target or raw_target.startswith("#"): + continue + if raw_target.lower().startswith("file://"): + errors.append( + verification_error( + "absolute_link_forbidden", + "文档包含本机绝对链接。", + "改为仓库内相对链接。", + document=document, + ) + ) + continue + target = unquote(raw_target.split("#", 1)[0].split("?", 1)[0]) + if not target: + continue + if target.startswith(("/", "\\")) or re.match(r"^[A-Za-z]:[\\/]", target): + errors.append( + verification_error( + "absolute_link_forbidden", + "文档包含本机或根路径链接。", + "改为仓库内相对链接。", + document=document, + ) + ) + continue + if re.match(r"^[A-Za-z][A-Za-z0-9+.-]*:", raw_target): + continue + candidate = (document_path.parent / target).resolve(strict=False) + try: + relative = candidate.relative_to(root.resolve(strict=True)).as_posix() + except (OSError, ValueError): + errors.append( + verification_error( + "link_outside_repository", + "相对链接解析到了仓库外部。", + "将链接改为仓库内部存在的相对路径。", + document=document, + ) + ) + continue + if is_forbidden(relative) or not candidate.exists(): + errors.append( + verification_error( + "relative_link_invalid", + "文档包含不存在或不允许引用的相对链接。", + "修正链接目标，或移除无法由仓库内容证明的链接。", + document=document, + target=raw_target, + ) + ) + return errors + + +def validate_document_safety(document: str, text: str) -> list[dict[str, Any]]: + errors: list[dict[str, Any]] = [] + if ABSOLUTE_LOCAL_PATH_RE.search(text): + errors.append( + verification_error( + "absolute_local_path_detected", + "文档包含本机绝对路径。", + "删除本机路径并改用仓库相对路径。", + document=document, + ) + ) + if any(pattern.search(text) for pattern in SECRET_CONTENT_PATTERNS): + errors.append( + verification_error( + "sensitive_content_detected", + "文档包含疑似密钥、密码或私钥内容。", + "立即删除敏感值，仅保留环境变量名称或安全配置说明。", + document=document, + ) + ) + return errors + + +def referenced_code_paths(document: str, text: str) -> list[str]: + candidates = set() + for value in re.findall(r"`([^`\n]+)`", text): + value = value.strip().replace("\\", "/") + if not value or any(char in value for char in "*?{}<>") or " " in value: + continue + if value.startswith("core/"): + candidate = value.rstrip("/.,;:") + elif document == BACKEND_DOCUMENT and value.startswith(("apps/", "airshelf/", "skills/")): + candidate = f"core/backend/{value}".rstrip("/.,;:") + elif document == FRONTEND_DOCUMENT and value.startswith(("src/", "public/")): + candidate = f"core/frontend/{value}".rstrip("/.,;:") + else: + continue + candidates.add(normalize_path(candidate)) + return sorted(path for path in candidates if path) + + +def validate_code_references( + root: Path, document: str, text: str +) -> tuple[list[dict[str, Any]], list[str]]: + errors: list[dict[str, Any]] = [] + checked: list[str] = [] + for path in referenced_code_paths(document, text): + if is_forbidden(path): + errors.append( + verification_error( + "forbidden_code_reference", + "文档引用了敏感或排除路径。", + "移除该引用，仅保留允许公开的架构事实。", + document=document, + ) + ) + continue + candidate = safe_candidate(root, path) + if candidate is None or not candidate.exists(): + errors.append( + verification_error( + "code_reference_missing", + "文档中的代码路径无法在仓库中验证。", + "修正路径，或删除没有代码依据的描述。", + document=document, + path=path, + ) + ) + continue + checked.append(path) + return errors, checked + + +def expand_sections(section: str) -> list[str]: + return [part.strip() for part in section.split("、") if part.strip()] + + +def planned_updates(impacts: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + sections_by_document: dict[str, set[str]] = {} + for impact in impacts: + sections_by_document.setdefault(impact["document"], set()).update( + expand_sections(impact["section"]) + ) + return [ + {"document": document, "sections": sorted(sections)} + for document, sections in sorted(sections_by_document.items()) + ] + + +def build_snapshot(root: Path, mode: str, scope: str = "all") -> dict[str, Any]: + warnings: set[str] = set() + targets = selected_documents(scope) + head = current_head(root) + if head is None: + warnings.add("head_commit_unavailable") + + documents, markers = document_states(root, warnings, targets) + usable_markers: dict[str, str] = {} + for document, marker in markers.items(): + if marker is None: + warnings.add(f"sync_marker_missing:{document}") + elif marker_is_usable(root, marker, head): + usable_markers[document] = marker + else: + warnings.add(f"sync_marker_unusable:{document}") + + baseline_required = len(usable_markers) != len(targets) + committed = [] + for marker in sorted(set(usable_markers.values())): + committed.append( + { + "base_commit": marker, + "files": filter_entries( + diff_entries(root, ("diff", f"{marker}..{head}")), scope + ), + } + ) + baseline_files = ( + [path for path in walk_relevant_files(root) if path_in_scope(path, scope)] + if baseline_required + else [] + ) + staged = filter_entries(diff_entries(root, ("diff", "--cached")), scope) + unstaged = filter_entries(diff_entries(root, ("diff",)), scope) + untracked = filter_entries(untracked_entries(root), scope) + + changed_paths = set(baseline_files) + for group in committed: + changed_paths.update(item["path"] for item in group["files"]) + for group in (staged, unstaged, untracked): + changed_paths.update(item["path"] for item in group) + + common_marker: str | None = None + marker_values = set(usable_markers.values()) + if len(usable_markers) == len(targets) and len(marker_values) == 1: + common_marker = next(iter(marker_values)) + + impacts, non_architecture = map_impacts( + root, + head, + targets, + usable_markers, + baseline_files, + staged, + unstaged, + untracked, + warnings, + ) + architecture: dict[str, Any] = {} + if BACKEND_DOCUMENT in targets: + architecture["backend"] = collect_backend(root, warnings) + if FRONTEND_DOCUMENT in targets: + architecture["frontend"] = collect_frontend(root, warnings) + content_fingerprint, fingerprinted_file_count = scope_content_fingerprint( + root, scope, warnings + ) + snapshot_core = { + "architecture": architecture, + "changed_files": sorted(changed_paths), + "changes": { + "baseline": baseline_files, + "committed": committed, + "staged": staged, + "unstaged": unstaged, + "untracked": untracked, + }, + "content_fingerprint": content_fingerprint, + "documents": documents, + "fingerprinted_file_count": fingerprinted_file_count, + "repository": { + "base_commit": common_marker, + "baseline_documents": sorted(set(targets) - set(usable_markers)), + "baseline_required": baseline_required, + "current_commit": head, + "sync_commits": markers, + }, + "scope": scope, + } + canonical = json.dumps(snapshot_core, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + snapshot_id = hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:20] + warnings_list = sorted(warnings) + update_plan = planned_updates(impacts) + requires_update = bool(impacts) + plan_material = { + "impacts": impacts, + "non_architecture_changes": non_architecture, + "planned_updates": update_plan, + "scope": scope, + "snapshot_id": snapshot_id, + "warnings": warnings_list, + } + plan_canonical = json.dumps( + plan_material, ensure_ascii=True, sort_keys=True, separators=(",", ":") + ) + plan_id = f"arch-{hashlib.sha256(plan_canonical.encode('utf-8')).hexdigest()[:20]}" + evidence_allowlist = sorted( + {path for impact in impacts for path in impact.get("evidence", [])} + ) + document_preconditions = [ + { + "document": item["document"], + "before_sha256": documents[item["document"]]["sha256"], + "sections": item["sections"], + } + for item in update_plan + ] + working_tree_evidence = sorted( + { + entry["path"] + for entries in (staged, unstaged, untracked) + for entry in entries + } + ) + control_paths = ( + "tools/architecture-sync/WORKFLOW.md", + "tools/architecture-sync/scan_architecture.py", + ".agents/skills/sync-architecture-docs/SKILL.md", + ".claude/skills/sync-architecture-docs/SKILL.md", + ) + guard_documents: dict[str, Any] = {} + for item in document_preconditions: + document_text = read_text(root, item["document"], warnings) or "" + guard_documents[item["document"]] = { + "before_sha256": item["before_sha256"], + "header_hash": document_header_hash(document_text), + "headings": [ + heading + for heading in documents[item["document"]]["headings"] + if heading["level"] <= 3 + ], + "protected_sections": protected_section_hashes( + document_text, item["sections"] + ), + "sections": item["sections"], + } + guard_payload = { + "allowed_documents": [item["document"] for item in update_plan], + "content_fingerprint": content_fingerprint, + "current_commit": head, + "documents": guard_documents, + "evidence_sample": [ + { + "exists": bool( + (safe_candidate(root, path) or Path("__missing__")).exists() + ), + "path": path, + } + for path in evidence_allowlist[:24] + ], + "guard_version": 1, + "initial_worktree": worktree_guard(root), + "plan_id": plan_id, + "protected_control_files": { + path: file_fingerprint(root, path, warnings) for path in control_paths + }, + "scope": scope, + } + verification_guard = encode_guard(guard_payload) + return { + "analysis_ready": True, + **snapshot_core, + "confirmation": { + "accepted_phrase": "确认更新项目文档", + "plan_id": plan_id, + "required": requires_update, + "scanner_write_capability": False, + "short_phrase": "确认", + "short_phrase_requires_same_pending_context": True, + }, + "impact_mapping_version": 1, + "impacts": impacts, + "mode": mode, + "next_action": "await_confirmation" if requires_update else "none", + "non_architecture_changes": non_architecture, + "plan_id": plan_id, + "planned_updates": update_plan, + "requires_update": requires_update, + "schema_version": SCHEMA_VERSION, + "snapshot_id": snapshot_id, + "update_context": { + "document_preconditions": document_preconditions, + "evidence_allowlist": evidence_allowlist, + "git_actions_allowed": False, + "patch_method": "apply_patch", + "plan_id": plan_id, + "read_policy": "minimum_needed_from_allowlist", + "sync_metadata": { + "checked_date_format": "YYYY-MM-DD", + "commit": head, + "has_uncommitted_evidence": bool(working_tree_evidence), + "marker": ( + f"" if head else None + ), + }, + "verification_guard": verification_guard, + "working_tree_evidence": working_tree_evidence, + }, + "warnings": warnings_list, + "write_policy": { + "current_plan_documents": list(targets), + "never_modify": [ + "core/ARCHITECTURE.md", + "business_code", + "configuration", + "migrations", + "tests", + "deployment_files", + "git_index", + "git_commits", + "git_remote", + ], + "write_authorized": False, + }, + } + + +def verify_snapshot(root: Path, token: str, scope: str) -> dict[str, Any]: + errors: list[dict[str, Any]] = [] + warnings: list[str] = [] + checked_code_paths: set[str] = set() + guard = decode_guard(token) + plan_id = str(guard.get("plan_id", "")) + guard_scope = guard.get("scope") + if guard_scope != scope: + errors.append( + verification_error( + "scope_mismatch", + "校验范围与已确认计划不一致。", + "使用原计划相同的 --scope 重新执行 verify。", + ) + ) + allowed_documents = set(guard.get("allowed_documents", [])) + if not allowed_documents or not allowed_documents.issubset(set(TARGET_DOCUMENTS)): + errors.append( + verification_error( + "allowed_documents_invalid", + "校验令牌没有有效的目标文档白名单。", + "丢弃当前令牌并重新运行 plan。", + ) + ) + + head = current_head(root) + if head != guard.get("current_commit"): + errors.append( + verification_error( + "head_commit_changed", + "计划生成后 HEAD 已发生变化。", + "停止写入并重新运行 plan 获取新确认。", + ) + ) + + fingerprint_warnings: set[str] = set() + current_fingerprint, fingerprinted_count = scope_content_fingerprint( + root, str(guard_scope), fingerprint_warnings + ) + if current_fingerprint != guard.get("content_fingerprint"): + errors.append( + verification_error( + "code_state_changed", + "计划生成后相关代码事实发生变化。", + "停止当前更新，重新扫描并重新确认。", + ) + ) + warnings.extend(sorted(fingerprint_warnings)) + + for path, expected in guard.get("protected_control_files", {}).items(): + current = file_fingerprint(root, path, set()) + if current != expected: + errors.append( + verification_error( + "control_file_changed", + "同步工作流、扫描器或 Skill 在执行期间发生变化。", + "停止验证，恢复控制文件或重新生成计划。", + path=path, + ) + ) + + errors.extend( + compare_worktree_guards( + guard.get("initial_worktree", {}), worktree_guard(root), allowed_documents + ) + ) + + document_results = [] + for document in sorted(allowed_documents): + guard_document = guard.get("documents", {}).get(document) + if not isinstance(guard_document, dict): + errors.append( + verification_error( + "document_guard_missing", + "目标文档缺少修改前保护信息。", + "丢弃当前令牌并重新运行 plan。", + document=document, + ) + ) + continue + read_warnings: set[str] = set() + data = read_bytes(root, document, read_warnings) + if data is None: + errors.append( + verification_error( + "target_document_missing", + "目标架构文档不存在或无法读取。", + "恢复目标文档后重新生成计划。", + document=document, + ) + ) + continue + warnings.extend(sorted(read_warnings)) + try: + text = data.decode("utf-8-sig") + except UnicodeDecodeError: + errors.append( + verification_error( + "target_document_not_utf8", + "目标架构文档不是有效 UTF-8。", + "将文档恢复为 UTF-8 编码后重新验证。", + document=document, + ) + ) + continue + current_sha = hashlib.sha256(data).hexdigest() + changed = current_sha != guard_document.get("before_sha256") + if not changed: + warnings.append(f"planned_document_unchanged:{document}") + current_headings = headings_from_text(text) + errors.extend( + validate_headings(document, guard_document.get("headings", []), current_headings) + ) + if document_header_hash(text) != guard_document.get("header_hash"): + errors.append( + verification_error( + "document_header_changed", + "文档头部除核对日期和同步标记外发生了计划外变化。", + "恢复文档标题和说明，只保留允许更新的日期与同步标记。", + document=document, + ) + ) + current_sections = {record["key"]: record["hash"] for record in section_records(text)} + changed_protected_sections = [ + key + for key, expected_hash in guard_document.get("protected_sections", {}).items() + if current_sections.get(key) != expected_hash + ] + if changed_protected_sections: + errors.append( + verification_error( + "unplanned_section_changed", + "计划外章节正文发生变化。", + "恢复计划外章节，只保留 planned_updates 指定范围内的局部补丁。", + document=document, + sections=changed_protected_sections, + ) + ) + errors.extend(validate_relative_links(root, document, text)) + errors.extend(validate_document_safety(document, text)) + reference_errors, checked = validate_code_references(root, document, text) + errors.extend(reference_errors) + checked_code_paths.update(checked) + + if changed: + markers = [value.lower() for value in SYNC_MARKER_RE.findall(text)] + if len(markers) != 1 or markers[0] != guard.get("current_commit"): + errors.append( + verification_error( + "sync_marker_invalid", + "已修改文档缺少唯一且正确的同步 commit 标记。", + "在文首保留一条指向计划 HEAD 的 architecture-sync-commit 标记。", + document=document, + ) + ) + if not re.search(r"(?m)^>\s*最后核对：\d{4}-\d{2}-\d{2}。\s*$", text): + errors.append( + verification_error( + "checked_date_missing", + "已修改文档缺少格式正确的最后核对日期。", + "在文首使用“> 最后核对：YYYY-MM-DD。”格式更新日期。", + document=document, + ) + ) + document_results.append( + { + "changed": changed, + "document": document, + "headings_checked": len(current_headings), + } + ) + + evidence_checked = [] + for item in guard.get("evidence_sample", []): + path = item.get("path", "") + if not path or is_forbidden(path): + errors.append( + verification_error( + "evidence_path_invalid", + "校验令牌包含越界或敏感证据路径。", + "丢弃当前令牌并重新运行 plan。", + ) + ) + continue + current_exists = bool((safe_candidate(root, path) or Path("__missing__")).exists()) + if current_exists != bool(item.get("exists")): + errors.append( + verification_error( + "evidence_presence_changed", + "计划证据文件的存在状态发生变化。", + "停止当前更新，重新扫描并重新确认。", + path=path, + ) + ) + elif current_exists: + evidence_checked.append(path) + + status = "passed" if not errors else "failed" + return { + "mode": "verify", + "plan_id": plan_id, + "schema_version": SCHEMA_VERSION, + "verification": { + "checks": { + "allowed_documents": sorted(allowed_documents), + "code_references_checked": len(checked_code_paths), + "evidence_files_checked": len(evidence_checked), + "fingerprinted_files_checked": fingerprinted_count, + "head_commit": head, + "scope": guard_scope, + }, + "documents": document_results, + "errors": errors, + "status": status, + "warnings": sorted(set(warnings)), + }, + } + + +def parse_arguments(arguments: Sequence[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Read-only AirShelf architecture scanner") + parser.add_argument("mode", choices=("plan", "verify")) + parser.add_argument( + "--scope", + choices=("all", "backend", "frontend"), + default="all", + help="Limit the current plan to both documents or one architecture domain", + ) + parser.add_argument( + "--guard", + help="Verification guard emitted by plan mode; required by verify mode", + ) + parser.add_argument( + "--repo-root", + type=Path, + default=Path.cwd(), + help="Any path inside the Git repository (default: current directory)", + ) + return parser.parse_args(arguments) + + +def main(arguments: Sequence[str] | None = None) -> int: + args = parse_arguments(arguments) + try: + root = locate_repository(args.repo_root) + if args.mode == "verify": + if not args.guard: + payload = { + "mode": "verify", + "schema_version": SCHEMA_VERSION, + "verification": { + "errors": [ + verification_error( + "verification_guard_required", + "verify 模式缺少 plan 生成的校验令牌。", + "先运行 plan，确认后传入 update_context.verification_guard。", + ) + ], + "status": "failed", + "warnings": [], + }, + } + print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2)) + return 3 + payload = verify_snapshot(root, args.guard, args.scope) + else: + payload = build_snapshot(root, args.mode, args.scope) + except ScanError as error: + if args.mode == "verify": + payload = { + "mode": "verify", + "schema_version": SCHEMA_VERSION, + "verification": { + "errors": [ + verification_error( + str(error), + "校验令牌无效或只读校验无法继续。", + "丢弃当前令牌，重新运行 plan 并重新确认。", + ) + ], + "status": "failed", + "warnings": [], + }, + } + else: + payload = { + "error": str(error), + "mode": args.mode, + "schema_version": SCHEMA_VERSION, + } + print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2)) + return 2 + print(json.dumps(payload, ensure_ascii=True, sort_keys=True, indent=2)) + if args.mode == "verify" and payload["verification"]["status"] != "passed": + return 3 + return 0 + + +if __name__ == "__main__": + if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8", errors="backslashreplace") + raise SystemExit(main())