test: record pre-migration workspace evidence

docs/evidence/foundation-migration-preflight.json

@@ -0,0 +1,54 @@
+[
+  {
+    "path": "D:\\桌面\\财富项目\\group_fqcd_jr",
+    "branch": "develop",
+    "head": "c6be99078e2be5d00397449c826fa049b60910ba",
+    "status": [
+      "?? \"docs/15-\\347\\237\\245\\350\\257\\206\\346\\243\\200\\347\\264\\242\\346\\216\\245\\345\\205\\245\\346\\226\\271\\346\\241\\210.md\"",
+      "?? docs/superpowers/plans/2026-09-10-customer-service-rag-plan-a.md"
+    ]
+  },
+  {
+    "path": "D:\\桌面\\财富项目\\group_fqcd_jr\\.worktrees\\customer-service-rag",
+    "branch": "feature/customer-service-rag",
+    "head": "ad7172367a0d69903e8bb92916f5d927084d8970",
+    "status": [
+      " M app/core/config.py",
+      " M app/core/contracts.py",
+      " M app/main.py",
+      " M app/service/agent/bootstrap.py",
+      " M app/service/agent/factory.py",
+      " M app/service/agent/governance.py",
+      " M app/service/agent_run_application_service.py",
+      " M app/service/model_gateway.py",
+      " M app/worker/runtime.py",
+      " M tests/conftest.py",
+      " M tests/integration/test_agent_run_acceptance.py",
+      " M tests/integration/test_worker_runtime_mysql.py",
+      " M tests/unit/service/test_model_gateway.py",
+      "?? app/core/knowledge_contracts.py",
+      "?? app/infrastructure/milvus_knowledge_adapter.py",
+      "?? app/model/knowledge.py",
+      "?? app/service/agent/customer_service_agent.py",
+      "?? app/service/agent/customer_service_routing.py",
+      "?? app/service/knowledge_authority.py",
+      "?? app/service/knowledge_config.py",
+      "?? app/service/knowledge_retrieval_service.py",
+      "?? app/service/knowledge_tool_service.py",
+      "?? app/static/",
+      "?? tests/unit/api/test_customer_service_test_page.py",
+      "?? tests/unit/core/test_knowledge_contracts.py",
+      "?? tests/unit/infrastructure/test_milvus_knowledge_adapter.py",
+      "?? tests/unit/service/test_customer_service_agent.py",
+      "?? tests/unit/service/test_knowledge_authority.py",
+      "?? tests/unit/service/test_knowledge_config.py",
+      "?? tests/unit/service/test_knowledge_retrieval.py"
+    ]
+  },
+  {
+    "path": "D:\\桌面\\财富项目\\111\\qyqy_develop",
+    "branch": "qyqy_develop",
+    "head": "6516ccb385024f7c9fe9e2207de69cd924e26262",
+    "status": []
+  }
+]

tests/unit/tools/test_foundation_migration_preflight.py

@@ -0,0 +1,95 @@
+"""验证迁移前状态证据采集器只读取 Git 元数据。"""
+
+# 导入测试所需的标准库 JSON、路径和子进程结果类型。
+import json
+from pathlib import Path
+from subprocess import CompletedProcess
+
+# 导入 pytest 补丁类型以及待验证的预检工具接口。
+from pytest import MonkeyPatch
+
+from tools.foundation_migration_preflight import (
+    collect_workspace_state,
+    run_git,
+    write_preflight_report,
+)
+
+
+# 验证未跟踪客服文件会被记录，且采集结果不含环境变量名称或值。
+def test_collect_workspace_state_records_untracked_paths_without_environment_values(
+    tmp_path: Path,
+) -> None:
+    # 构造确定性的 Git 命令替身，不调用真实 Git 或环境变量。
+    def fake_git_runner(_worktree: Path, *args: str) -> str:
+        # 为分支查询返回客服功能分支名称。
+        if args == ("branch", "--show-current"):
+            return "feature/customer-service-rag\n"
+        # 为提交查询返回固定的非敏感提交标识。
+        if args == ("rev-parse", "HEAD"):
+            return "abc123\n"
+        # 为状态查询返回一个未跟踪客服文件。
+        if args == ("status", "--short"):
+            return "?? app/service/agent/customer_service_agent.py\n"
+        # 防止测试静默接受未定义的 Git 查询。
+        raise AssertionError(f"unexpected git arguments: {args}")
+
+    # 使用替身采集临时工作区的只读状态。
+    state = collect_workspace_state(tmp_path, runner=fake_git_runner)
+
+    # 断言采集到预期的分支名称。
+    assert state["branch"] == "feature/customer-service-rag"
+    # 断言未跟踪客服文件完整保留在状态清单中。
+    assert state["status"] == ["?? app/service/agent/customer_service_agent.py"]
+    # 断言序列化结果不包含任何环境变量敏感字段。
+    assert "MYSQL_PASSWORD" not in json.dumps(state)
+
+
+# 验证报告写入器只输出传入的非敏感 Git 状态结构。
+def test_write_preflight_report_persists_utf8_json(tmp_path: Path) -> None:
+    # 指定临时报告文件，避免写入任何真实工作目录。
+    report_path = tmp_path / "preflight.json"
+    # 构造只含安全 Git 元数据的状态条目。
+    states = [{"path": "D:/workspace", "branch": "develop", "head": "abc123", "status": []}]
+
+    # 写入迁移前报告。
+    write_preflight_report(report_path, states)
+
+    # 以 UTF-8 读取并解析报告正文。
+    report = json.loads(report_path.read_text(encoding="utf-8"))
+    # 断言报告保留原始状态条目。
+    assert report == states
+
+
+# 验证每次 Git 查询只信任当前显式工作区，而不写入全局 Git 配置。
+def test_run_git_scopes_safe_directory_to_the_requested_worktree(
+    tmp_path: Path,
+    monkeypatch: MonkeyPatch,
+) -> None:
+    # 保存被测函数交给子进程层的命令参数。
+    captured_commands: list[list[str]] = []
+
+    # 构造返回固定分支名的无副作用子进程替身。
+    def fake_run(command: list[str], **_kwargs: object) -> CompletedProcess[str]:
+        # 记录命令以便后续断言安全目录范围。
+        captured_commands.append(command)
+        # 返回模拟的 Git 成功结果。
+        return CompletedProcess(command, 0, "develop\n", "")
+
+    # 让测试不执行真实 Git。
+    monkeypatch.setattr("tools.foundation_migration_preflight.subprocess.run", fake_run)
+
+    # 执行一个受限 Git 分支查询。
+    output = run_git(tmp_path, "branch", "--show-current")
+
+    # 断言调用仍返回 Git 输出。
+    assert output == "develop\n"
+    # 断言命令仅为该临时工作区附加安全目录。
+    assert captured_commands == [[
+        "git",
+        "-c",
+        f"safe.directory={tmp_path.resolve().as_posix()}",
+        "-C",
+        str(tmp_path.resolve()),
+        "branch",
+        "--show-current",
+    ]]

tools/foundation_migration_preflight.py

@@ -0,0 +1,117 @@
+"""为无损底座迁移保存并校验工作区的只读 Git 状态证据。"""
+
+# 导入命令行参数解析器以支持生成和校验报告。
+import argparse
+# 导入 JSON 序列化工具以写入 UTF-8 状态证据。
+import json
+# 导入子进程工具以直接调用 Git 而不经过 shell。
+import subprocess
+# 导入可调用协议和类型别名支持。
+from collections.abc import Callable
+# 导入路径类型以约束工作区和报告位置。
+from pathlib import Path
+# 导入任意 JSON 对象的静态类型。
+from typing import Any
+
+# 定义只允许执行的 Git 只读子命令首参数集合。
+READ_ONLY_GIT_COMMANDS = frozenset({"branch", "rev-parse", "status"})
+# 定义便于测试注入的 Git 调用函数类型。
+GitRunner = Callable[..., str]
+
+
+# 执行受限的 Git 只读命令并返回标准输出文本。
+def run_git(worktree: Path, *args: str) -> str:
+    # 解析目标工作区以确保 Git 和安全目录使用同一个绝对路径。
+    resolved = worktree.resolve()
+    # 拒绝空命令，避免形成未约束的 Git 调用。
+    if not args:
+        raise ValueError("git command is required")
+    # 拒绝任何不在白名单中的 Git 子命令。
+    if args[0] not in READ_ONLY_GIT_COMMANDS:
+        raise ValueError("git command is not read-only")
+    # 将安全目录限定为当前查询工作区，避免写入全局 Git 配置。
+    safe_directory = resolved.as_posix()
+    # 以参数数组执行 Git，禁止 shell 解释路径或输入内容。
+    result = subprocess.run(
+        ["git", "-c", f"safe.directory={safe_directory}", "-C", str(resolved), *args],
+        check=True,
+        capture_output=True,
+        encoding="utf-8",
+        errors="replace",
+    )
+    # 返回 Git 的标准输出，供调用方以确定性方式解析。
+    return result.stdout
+
+
+# 收集一个工作区的分支、提交与简短状态，不读取环境变量或业务数据。
+def collect_workspace_state(worktree: Path, runner: GitRunner = run_git) -> dict[str, object]:
+    # 解析绝对路径，防止报告中出现随当前目录变化的相对路径。
+    resolved = worktree.resolve()
+    # 依次执行已白名单化的 Git 查询并构造安全状态字典。
+    return {
+        "path": str(resolved),
+        "branch": runner(resolved, "branch", "--show-current").strip(),
+        "head": runner(resolved, "rev-parse", "HEAD").strip(),
+        "status": runner(resolved, "status", "--short").splitlines(),
+    }
+
+
+# 将已收集的状态以 UTF-8 JSON 写入调用者指定的报告文件。
+def write_preflight_report(target: Path, states: list[dict[str, object]]) -> None:
+    # 确保报告父目录存在，但不创建或改动任何工作区内容。
+    target.parent.mkdir(parents=True, exist_ok=True)
+    # 使用稳定缩进和 UTF-8 编码写入仅由调用方提供的状态数据。
+    target.write_text(json.dumps(states, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+# 从磁盘读取先前报告并验证指定工作区状态完全一致。
+def verify_preflight_report(target: Path, worktrees: list[Path]) -> list[str]:
+    # 以 UTF-8 读取 JSON 报告，避免系统默认编码影响比较。
+    expected_value: Any = json.loads(target.read_text(encoding="utf-8"))
+    # 拒绝非列表报告，防止错误文件被误当作迁移证据。
+    if not isinstance(expected_value, list):
+        raise ValueError("preflight report must be a list")
+    # 为每个当前工作区重新采集只读 Git 状态。
+    current = [collect_workspace_state(worktree) for worktree in worktrees]
+    # 返回 JSON 表示不同的工作区路径，空列表代表完全一致。
+    return [
+        str(item["path"])
+        for item, expected in zip(current, expected_value, strict=True)
+        if item != expected
+    ]
+
+
+# 解析 CLI 工作区参数并执行报告生成或校验。
+def main() -> int:
+    # 创建命令行解析器并限定所有输入为显式路径。
+    parser = argparse.ArgumentParser(description=__doc__)
+    # 要求调用者指定报告文件路径。
+    parser.add_argument("--report", type=Path, required=True)
+    # 允许多次提供要采集或校验的工作区路径。
+    parser.add_argument("--worktree", type=Path, action="append", required=True)
+    # 启用校验模式时不覆盖报告。
+    parser.add_argument("--verify", action="store_true")
+    # 解析用户传入的参数。
+    arguments = parser.parse_args()
+    # 在校验模式下输出差异并返回非零状态。
+    if arguments.verify:
+        # 比较当前状态和报告状态。
+        differences = verify_preflight_report(arguments.report, arguments.worktree)
+        # 输出机器和人工都可识别的校验结果。
+        print("UNCHANGED" if not differences else f"CHANGED: {', '.join(differences)}")
+        # 有任何差异时返回失败状态。
+        return 0 if not differences else 1
+    # 收集调用方明确列出的工作区状态。
+    states = [collect_workspace_state(worktree) for worktree in arguments.worktree]
+    # 写入新的迁移前证据报告。
+    write_preflight_report(arguments.report, states)
+    # 输出报告保存位置，避免输出任何敏感运行配置。
+    print(f"WROTE: {arguments.report}")
+    # 报告创建成功时返回零状态。
+    return 0
+
+
+# 仅在脚本直接执行时运行命令行入口。
+if __name__ == "__main__":
+    # 用 main 的返回值作为进程退出码。
+    raise SystemExit(main())