fix: Windows兼容性修复 + P1-24弱模型优化 + 3.2.9b禁止降级方案 + AirRvr三层审查放行标准

- lock.py: 跨平台进程锁（Unix fcntl / Windows msvcrt / O_CREAT|O_EXCL降级） - eng_mode.py/eng_orchestrator.py: hasattr(os, "getloadavg") Windows防护 - arc_mode.py: 路径分隔符 replace("\\", "/") Windows兼容 - deploy_runtime.py: 修复语法错误（清理 import tempfile 残留） - P1-24(3.2.18): AMBIGUOUS_VERBS歧义词检测 + SAFE_VERBS安全动词 + validate_task_description() - TaskNode.keep_constraints 保留约束字段 + JSON序列化 - _build_graph_from_todo 返回歧义警告 + Arc自检集成 - 3.2.9b: FORBIDDEN_DEGRADATION_PATTERNS + check_forbidden_degradation() - AirRvr三层审查放行标准: ReviewVerdict + evaluate_review_pass() + is_forbidden_pass_reason() - commands/arc.md: 弱模型安全重写（操作类型拆分+保留约束+自检） - commands/do.md/eng.md/rvr.md: 禁止降级方案 + 三层审查标准 - 测试: 7个新测试 + 74全量通过 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-15 09:58:29 +08:00
parent 6130478c96
commit a60d1a0c04
13 changed files with 486 additions and 42 deletions
--- a/commands/arc.md
+++ b/commands/arc.md
@@ -56,14 +56,35 @@ ArcPhaseGate 控制 execution-plan.json 写入权限。phase 默认 `discussing`

 **推进方式**：用户说"确认"/"可以"/"同意"后，`ArcPhaseGate.confirm_architecture()` 自动推进。不丢失（已修复）。

-## 弱模型安全（INV-16）
+## 弱模型安全（INV-16）+ 任务描述弱模型优化（P1-24 / 3.2.18）

-Do Worker 可能是廉价模型/本地小模型，字面理解任务无推断能力。产出每个任务时必须：
+Do Worker 可能是廉价模型/本地小模型，字面理解任务无推断能力。真实案例中"清理旧产品实现"被弱模型理解为"删除整个 src/"。产出每个任务时必须：

-1. **禁止歧义词** — 不用"清理"、"重构"、"优化"等宽泛动词，指明具体改什么
-2. **否定约束显式化** — 写明**不做什么**（如"不删除 src/ 下现有模块"）
-3. **文件范围精确化** — `files_dirs` 精确到文件级，不写 `src/` 目录级
-4. **完成标准可验证** — `done_when` 能用 `grep`/`diff`/`cmake --build` 客观验证
+### 操作类型拆分（按动词分类）
+
+**禁止歧义词**：不使用"清理"、"优化"、"整理"、"更新"等宽泛动词。必须用具体动词：
+- `重构` — 修改实现但保持外部接口不变
+- `新增` — 添加新功能，不修改现有代码
+- `删除` — 移除指定文件或函数（必须列出具体目标）
+- `修改` — 修改指定文件的具体部分（必须指明改什么）
+- `保留` — 明确标记为不可修改的文件/目录
+
+### 保留约束机制
+
+每个任务必须包含：
+1. **操作指令**: 用具体动词描述要做什么（重构/新增/删除/修改）
+2. **保留约束**: 明确列出不可修改的文件、目录或函数
+3. **变更边界**: 精确到文件级别，每个文件标注"新建|修改|删除|保留"
+4. **完成标准**: 可验证的条件，避免主观判断
+
+### 禁止的写法
+
+- "清理旧实现" → 改为 "重构 CMakeLists.txt 去掉 sipclient 依赖，保留 src/ 下所有现有模块"
+- "优化模块结构" → 改为 "将 auth/login.py 中的 validate() 函数提取到 auth/validator.py"
+
+### Arc 自检
+
+生成任务后执行 `validate_task_description()` 自检。发现歧义词时自动拆分任务或补充保留约束，不将歧义任务传递给 Eng。

 ## Logging Standard

--- a/commands/do.md
+++ b/commands/do.md
@@ -13,6 +13,11 @@ AirDo 是任务执行器。运行单个任务，强制执行全专家插件路
 1. **UI Task Handling (P1-20)** — 当任务涉及 UI/前端/界面时，检测并确保 frontend-design Skill 可用。不可用时阻止执行
 2. **Evidence Gate (INV-4)** — GUI 任务需截图，网络任务需抓包，代码任务需静态分析/测试
 3. **全专家插件强制路由** — finish 时按任务类型和状态强制路由到对应专家插件，不可跳过
+4. **禁止降级实现 (3.2.9b)** — 必须严格遵循 AirArc 设计方案，禁止以下行为和语言模式：
+   - 禁止使用"兜底方案"、"先这样做"、"以后再补"、"临时方案"等降级语言
+   - 禁止自行发明替代方案绕过设计约束
+   - 禁止以"能跑就行"为标准降低实现质量
+   - 设计方案无法执行时：报 blocked 并说明原因，等待 Eng 协调

 ## 子命令

--- a/commands/eng.md
+++ b/commands/eng.md
@@ -21,6 +21,7 @@ AirEng 是调度引擎，不是执行器。它派发隔离 Worker、监控、合
   - 验证失败但非关键 → 记录问题继续下一任务
   - 波次间衔接 → 自行启动下一波次
 7. **仅以下情况才询问用户**：修复预算耗尽且任务仍 blocked；需求歧义无法继续；系统资源耗尽；用户显式暂停。
+8. **禁止降级方案 (3.2.9b)** — 派发任务时不得建议 Worker 使用降级方案。任务 blocked 时分析根因并协调解决，不得让 Worker "先这样跑通"。禁止使用"兜底方案"、"先这样做"、"以后再改"、"临时方案"、"quick fix"等降级语言。必须遵循 AirArc 产出方案，无法推进时报 blocked 等待重规划，不自行发明替代方案。

 ## 子命令

--- a/commands/rvr.md
+++ b/commands/rvr.md
@@ -12,8 +12,15 @@ AirRvr 是需求审查器。基于原始需求文档对已完成任务进行独

 1. **code-to-design 对照** — 每次审查必须逐行对照代码实现与设计文档（ADR、C4、需求）
 2. **高风险审计** — 终审必须包含 lifecycle/nullPointer/danglingPointer/exceptionSafety/concurrency 五项检查
-3. **verdict 控制合并** — fail 阻止合并，conditional-pass 记录遗留项，pass 正常合并
-4. **deliveryVerdict = block-release** — eng dispatch 阻止所有后续派发
+3. **审查放行标准（三层，不可降级）**：
+   - **第一层（最高优先级）**: Code-to-Design 逐行对照 — codeToDesignTable 中不得有 status=divergent 或 status=missing 的条目，任何 divergent/missing 必须修复后才能放行
+   - **第二层（必选）**: 静态分析通过 — AirSDB 报告无 critical/high severity finding，生命周期/空指针/悬垂指针/异常安全专项审计通过
+   - **第三层（必选）**: 测试通过 — 单元测试全部通过，接口测试全部通过，GUI 任务需 AirXDB 截图证据与设计稿一致
+   - 三层全部通过 → verdict=pass；第一层有 divergent/missing → verdict=fail（即使测试全绿）；第一层通过但第二/三层有问题 → verdict=conditional-pass
+4. **禁止表面理由判定 pass** — 以下理由不得单独作为 pass 依据："测试 pass"/"测试全绿"、"实现存在"/"函数存在"、"编译通过"/"无报错"、"能跑通"/"功能可用"。以上仅为必要不充分条件。
+5. **verdict 控制合并** — fail 阻止合并，conditional-pass 记录遗留项，pass 正常合并
+6. **deliveryVerdict = block-release** — eng dispatch 阻止所有后续派发
+7. **降级语言检测 (3.2.9b)** — 在 code-to-design 审查中检测 Worker 是否使用了降级语言（"兜底方案"、"先这样做"、"临时方案"等），检测到则 verdict 直接判定 fail，要求按原始设计重新实现

 ## 审查模式

--- a/lib/air_runtime/deploy_runtime.py
+++ b/lib/air_runtime/deploy_runtime.py
@@ -7,7 +7,8 @@ from __future__ import annotations

 import hashlib
 import subprocess
-from dataclasses import dataclass
+import tempfile
+from dataclasses import dataclass, field
 from pathlib import Path

 from air_runtime.contracts import DeploymentRecord, now_iso
@@ -20,7 +21,7 @@ class DeployTarget:
    host: str
    user: str = "root"
    port: int = 22
-    build_dir: str = "/tmp/airdep-build"
+    build_dir: str = tempfile.gettempdir() + "/airdep-build"
    deploy_dir: str = "/opt/app"


--- a/lib/air_runtime/lock.py
+++ b/lib/air_runtime/lock.py
@@ -1,43 +1,123 @@
 """
-文件级并发控制 — 解决 V1 P0-4 零并发控制问题。
-基于 fcntl.flock 的进程级文件锁，超时自动释放。
+文件级并发控制 — 跨平台进程级文件锁。
+Unix: fcntl.flock (POSIX)
+Windows: msvcrt.locking (Win32)
+均不支持时: 原子文件创建 (O_CREAT | O_EXCL)
 """

 from __future__ import annotations

-import fcntl
 import os
 import time
+import sys
 from pathlib import Path


+def _lock_file_unix(fd: int) -> None:
+    import fcntl
+    fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+
+def _unlock_file_unix(fd: int) -> None:
+    import fcntl
+    fcntl.flock(fd, fcntl.LOCK_UN)
+
+
+def _lock_file_windows(fd: int) -> None:
+    import msvcrt
+    msvcrt.locking(fd, msvcrt.LK_NBLCK, 1)
+
+
+def _unlock_file_windows(fd: int) -> None:
+    import msvcrt
+    msvcrt.locking(fd, msvcrt.LK_UNLCK, 1)
+
+
+if sys.platform == "win32":
+    try:
+        import msvcrt
+        _lock_fn = _lock_file_windows
+        _unlock_fn = _unlock_file_windows
+    except ImportError:
+        _lock_fn = None
+        _unlock_fn = None
+else:
+    try:
+        import fcntl  # noqa: F401
+        _lock_fn = _lock_file_unix
+        _unlock_fn = _unlock_file_unix
+    except ImportError:
+        _lock_fn = None
+        _unlock_fn = None
+
+
 class FileLock:
-    """基于 fcntl.flock(LOCK_EX | LOCK_NB) 的进程级文件锁"""
+    """跨平台进程级文件锁。
+
+    Unix: fcntl.flock(LOCK_EX | LOCK_NB)
+    Windows: msvcrt.locking(LK_NBLCK)
+    降级方案: 原子文件创建 (O_CREAT | O_EXCL)
+    """

    def __init__(self, path: Path, timeout: float = 10.0):
-        self._path = path.with_suffix(path.suffix + ".lock") if not path.suffix.endswith(".lock") else path
+        lock_suffix = ".lock" if not path.suffix.endswith(".lock") else ""
+        self._path = path.with_suffix(path.suffix + lock_suffix) if lock_suffix else path
        self._timeout = timeout
        self._fd: int | None = None
+        self._fallback = _lock_fn is None

-    def __enter__(self) -> FileLock:
-        self._fd = os.open(self._path, os.O_CREAT | os.O_RDWR)
+    def __enter__(self) -> "FileLock":
+        if self._fallback:
+            self._acquire_fallback()
+        else:
+            self._acquire_native()
+        return self
+
+    def __exit__(self, *exc) -> None:
+        if self._fallback:
+            self._release_fallback()
+        elif self._fd is not None:
+            _unlock_fn(self._fd)
+            os.close(self._fd)
+            self._fd = None
+
+    def _acquire_native(self) -> None:
+        self._fd = os.open(str(self._path), os.O_CREAT | os.O_RDWR)
        deadline = time.monotonic() + self._timeout
        while True:
            try:
-                fcntl.flock(self._fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-                return self
-            except OSError:
+                _lock_fn(self._fd)
+                return
+            except (OSError, IOError):
                if time.monotonic() >= deadline:
                    os.close(self._fd)
                    self._fd = None
                    raise TimeoutError(f"lock timeout after {self._timeout}s: {self._path}")
                time.sleep(0.1)

-    def __exit__(self, *exc) -> None:
+    def _acquire_fallback(self) -> None:
+        self._path.parent.mkdir(parents=True, exist_ok=True)
+        deadline = time.monotonic() + self._timeout
+        while True:
+            try:
+                self._fd = os.open(
+                    str(self._path),
+                    os.O_CREAT | os.O_EXCL | os.O_WRONLY,
+                )
+                return
+            except FileExistsError:
+                if time.monotonic() >= deadline:
+                    raise TimeoutError(f"lock timeout after {self._timeout}s: {self._path}")
+                time.sleep(0.1)
+
+    def _release_fallback(self) -> None:
        if self._fd is not None:
-            fcntl.flock(self._fd, fcntl.LOCK_UN)
            os.close(self._fd)
            self._fd = None
+        try:
+            self._path.unlink(missing_ok=True)
+        except OSError:
+            pass

    @property
    def path(self) -> Path:
--- a/lib/air_runtime/modes/arc_mode.py
+++ b/lib/air_runtime/modes/arc_mode.py
@@ -107,19 +107,22 @@ def _export_task_graph_json(graph: TaskGraph, path: Path) -> None:
                        "filesDirs": n.files_dirs, "doneWhen": n.done_when,
                        "inDegree": n.in_degree, "outEdges": n.out_edges,
                        "writeSet": n.write_set, "testRequired": n.test_required,
-                        "adrRefs": n.adr_refs}  # P1-21
+                        "adrRefs": n.adr_refs,  # P1-21
+                        "keepConstraints": n.keep_constraints}  # P1-24
                  for nid, n in graph.nodes.items()},
        "edges": [{"source": e.source, "target": e.target, "kind": e.kind} for e in graph.edges],
    }
    atomic_json_write(path, data)


-def _build_graph_from_todo(todo_path: Path) -> tuple[TaskGraph, list[str]]:
-    """从 todo.md 构建初始 DAG，返回图和未满足 Done When 条件的任务列表。"""
+def _build_graph_from_todo(todo_path: Path) -> tuple[TaskGraph, list[str], list[str]]:
+    """从 todo.md 构建初始 DAG，返回图、Done When 违规列表、歧义词警告列表。"""
    from air_runtime.todo_parser import parse_tasks
+    from air_runtime.review import validate_task_description
    tasks = parse_tasks(todo_path)
    graph = TaskGraph()
    violations = []  # P1-19.1: 记录 Done When 不含"测试通过"的任务
+    ambiguity_warnings = []  # P1-24: 弱模型优化 — 歧义词警告

    for t in tasks:
        # P1-21: 从 todo.md ADR 列提取 adr_refs
@@ -137,10 +140,13 @@ def _build_graph_from_todo(todo_path: Path) -> tuple[TaskGraph, list[str]]:
        if t.done_when and "测试通过" not in t.done_when:
            violations.append(t.task_id)

+        # P1-24: 弱模型优化 — 检测歧义词
+        ambiguity_warnings.extend(validate_task_description(t.task_id, t.task))
+
    # P1-19.1: 注入边界测试任务
    _inject_boundary_tests(graph, tasks)

-    return graph, violations
+    return graph, violations, ambiguity_warnings


 def _inject_boundary_tests(graph: TaskGraph, tasks: list) -> None:
@@ -159,8 +165,8 @@ def _inject_boundary_tests(graph: TaskGraph, tasks: list) -> None:
            for fd in t.files_dirs.split(","):
                fd = fd.strip()
                if fd:
-                    # 取第一级目录作为模块名
-                    parts = fd.split("/")
+                    # 取第一级目录作为模块名 (跨平台路径解析)
+                    parts = fd.replace("\\", "/").split("/")
                    if len(parts) > 1:
                        module = parts[0]
                    else:
@@ -238,7 +244,7 @@ def parallel_review_mode(project_root: Path, todo_path: Path) -> dict:
    review_md_path.write_text(render_review_markdown(review), encoding="utf-8")

    # 构建 DAG（包含边界测试任务注入）
-    graph, done_when_violations = _build_graph_from_todo(todo_path)
+    graph, done_when_violations, ambiguity_warnings = _build_graph_from_todo(todo_path)
    for edge_info in review.to_dict().get("edges", []):
        graph.add_edge(Edge(source=edge_info["source"], target=edge_info["target"],
                           kind=edge_info.get("kind", "dependency")))
@@ -264,6 +270,7 @@ def parallel_review_mode(project_root: Path, todo_path: Path) -> dict:
        "conflicts": [c.to_dict() for c in review.conflicts],
        "serializationPoints": review.serialization_points,
        "doneWhenViolations": done_when_violations,  # P1-19.1: Done When 不含"测试通过"的任务
+        "ambiguityWarnings": ambiguity_warnings,  # P1-24: 弱模型优化 — 歧义词警告
        "boundaryTestTasks": [n.id for n in graph.nodes.values() if n.test_required],
        "safetyWarnings": safety_warnings,  # INV-16: 弱模型安全警告
    }
@@ -292,6 +299,7 @@ def parallel_review_mode(project_root: Path, todo_path: Path) -> dict:
            "parallel_group_count": len(review.parallel_groups),
            "conflict_count": len(review.conflicts),
            "done_when_violations": done_when_violations,
+            "ambiguity_warnings": ambiguity_warnings,
            "boundary_test_task_count": len([n for n in graph.nodes.values() if n.test_required])}


@@ -301,7 +309,7 @@ def incremental_replan_mode(project_root: Path, todo_path: Path, previous_graph_
    _ensure_dirs(paths)

    review = build_parallel_review(todo_path)
-    new_graph, _ = _build_graph_from_todo(todo_path)
+    new_graph, _, _ = _build_graph_from_todo(todo_path)
    for edge_info in review.to_dict().get("edges", []):
        new_graph.add_edge(Edge(source=edge_info["source"], target=edge_info["target"],
                                kind=edge_info.get("kind", "dependency")))
--- a/lib/air_runtime/modes/eng_mode.py
+++ b/lib/air_runtime/modes/eng_mode.py
@@ -370,13 +370,15 @@ def monitor_engine(project_root: Path) -> dict:
        else:
            ready_to_merge += 1 if worker.get("status") == "done" else 0

-    # 资源压力检测
+    # 资源压力检测 (Unix only; Windows 上不可用)
+    resource_pressure = False
+    if hasattr(os, "getloadavg"):
        try:
            load = os.getloadavg()[0]
            cpu_count = os.cpu_count() or 4
            resource_pressure = load > cpu_count * 2
        except OSError:
-        resource_pressure = False
+            pass

    # P1-21: ADR 变更自动检测
    adr_changes = _detect_adr_changes(project_root, state)
--- a/lib/air_runtime/modes/eng_orchestrator.py
+++ b/lib/air_runtime/modes/eng_orchestrator.py
@@ -51,6 +51,8 @@ class AdaptivePoller:


 def _resource_pressure() -> bool:
+    if not hasattr(os, "getloadavg"):
+        return False
    try:
        load = os.getloadavg()[0]
        cpu = os.cpu_count() or 4
--- a/lib/air_runtime/review.py
+++ b/lib/air_runtime/review.py
@@ -1,6 +1,7 @@
 """
 并行审查模块 — V2 从 V1 迁移。
 分析任务依赖、写集冲突、产出并行组和串行点。
+包含三层审查放行标准 (P1-24/P1-25) 和弱模型任务描述优化 (3.2.18)。
 """

 from __future__ import annotations
@@ -10,6 +11,41 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from air_runtime.todo_parser import parse_tasks

+# 3.2.18: 弱模型优化 — 歧义动词检测
+AMBIGUOUS_VERBS = {
+    "清理": "歧义——可能是删除、重构、或移除依赖",
+    "优化": "歧义——可能是性能优化、代码重构、或简化逻辑",
+    "整理": "歧义——可能是格式化、重命名、或删除",
+    "更新": "歧义——可能是修改现有代码、或替换为新实现",
+}
+
+SAFE_VERBS = {
+    "重构": "修改实现但保持外部接口不变",
+    "新增": "添加新功能，不修改现有代码",
+    "删除": "移除指定文件或函数（必须列出具体目标）",
+    "修改": "修改指定文件的具体部分（必须指明改什么）",
+    "保留": "明确标记为不可修改的文件/目录",
+}
+
+# 3.2.9b: 禁止使用的降级语言模式
+FORBIDDEN_DEGRADATION_PATTERNS = [
+    "兜底方案", "fallback",
+    "先这样做", "先这样跑通", "先这样实现",
+    "以后再删", "以后再补", "以后再改", "以后再优化",
+    "先回退",
+    "临时方案", "temporary workaround",
+    "hack 一下", "quick fix",
+    "MVP 先上", "先 ship 再迭代",
+]
+
+# AirRvr 审查放行标准 — 禁止的表面理由
+FORBIDDEN_PASS_REASONS = [
+    "测试 pass", "测试全绿", "all tests passed",
+    "实现存在", "函数存在", "文件已创建",
+    "编译通过", "无报错",
+    "能跑通", "功能可用",
+]
+

@dataclass
 class ParallelGroup:
@@ -124,3 +160,125 @@ def render_review_markdown(review: ReviewResult) -> str:
    for c in review.conflicts:
        lines.append(f"- {c.task_a} <-> {c.task_b}: {c.reason}")
    return "\n".join(lines)
+
+
+# ── 3.2.18: 弱模型任务描述优化 ──
+
+def validate_task_description(task_id: str, task_text: str) -> list[str]:
+    """检测任务描述中的歧义词并建议替换。返回警告列表。"""
+    warnings = []
+    for verb, explanation in AMBIGUOUS_VERBS.items():
+        if verb in task_text:
+            safe_suggestions = "、".join(SAFE_VERBS.keys())
+            warnings.append(
+                f"[{task_id}] 任务描述包含歧义词「{verb}」({explanation})，"
+                f"请拆分为具体操作（{safe_suggestions}）"
+            )
+    return warnings
+
+
+def check_forbidden_degradation(text: str) -> list[str]:
+    """检测文本中的降级语言模式，返回匹配到的模式列表。
+    检测时归一化空白字符以处理中文/英文间距变化。"""
+    import re
+    normalized = re.sub(r'\s+', '', text.lower())
+    found = []
+    for pattern in FORBIDDEN_DEGRADATION_PATTERNS:
+        normalized_pattern = re.sub(r'\s+', '', pattern.lower())
+        if normalized_pattern in normalized:
+            found.append(pattern)
+    return found
+
+
+# ── AirRvr 三层审查放行标准 ──
+
+@dataclass
+class ReviewVerdict:
+    """三层审查结果。"""
+    verdict: str  # pass | fail | conditional-pass
+    layer1_pass: bool = False   # Code-to-Design
+    layer2_pass: bool = False   # 静态分析
+    layer3_pass: bool = False   # 测试
+    divergent_entries: list[str] = field(default_factory=list)
+    missing_entries: list[str] = field(default_factory=list)
+    static_findings: list[str] = field(default_factory=list)
+    test_failures: list[str] = field(default_factory=list)
+    pass_reason: str = ""
+
+
+def evaluate_review_pass(
+    code_to_design_table: list[dict],
+    sdb_report: dict | None = None,
+    test_results: dict | None = None,
+    xdb_evidence: dict | None = None,
+    design_spec: dict | None = None,
+) -> ReviewVerdict:
+    """三层审查放行评估（不可降级）。
+
+    Layer 1 (最高优先级): Code-to-Design 逐行对照
+    Layer 2: 静态分析通过
+    Layer 3: 测试通过
+    """
+    divergent = []
+    missing = []
+    for entry in code_to_design_table:
+        status = entry.get("status", "")
+        if status == "divergent":
+            divergent.append(entry.get("designPoint", entry.get("requirement", "?")))
+        elif status == "missing":
+            missing.append(entry.get("designPoint", entry.get("requirement", "?")))
+
+    layer1_pass = len(divergent) == 0 and len(missing) == 0
+
+    # Layer 2: 静态分析
+    static_findings = []
+    if sdb_report:
+        for finding in sdb_report.get("findings", []):
+            if finding.get("severity") in ("critical", "high"):
+                static_findings.append(finding.get("description", str(finding)))
+    layer2_pass = len(static_findings) == 0
+
+    # Layer 3: 测试
+    test_failures = []
+    if test_results:
+        for suite in test_results.get("suites", []):
+            if suite.get("status") != "pass":
+                test_failures.append(suite.get("name", "unknown"))
+    layer3_pass = len(test_failures) == 0
+
+    # GUI 任务额外检查
+    if xdb_evidence and design_spec:
+        if not xdb_evidence.get("matches_design", True):
+            layer3_pass = False
+            test_failures.append("xdb screenshot does not match design spec")
+
+    # 判定
+    if not layer1_pass:
+        verdict = "fail"
+    elif layer2_pass and layer3_pass:
+        verdict = "pass"
+    else:
+        verdict = "conditional-pass"
+
+    return ReviewVerdict(
+        verdict=verdict,
+        layer1_pass=layer1_pass,
+        layer2_pass=layer2_pass,
+        layer3_pass=layer3_pass,
+        divergent_entries=divergent,
+        missing_entries=missing,
+        static_findings=static_findings,
+        test_failures=test_failures,
+    )
+
+
+def is_forbidden_pass_reason(reason: str) -> bool:
+    """检查审查通过理由是否使用了禁止的表面原因。
+    检测时归一化空白字符以处理间距变化。"""
+    import re
+    normalized_reason = re.sub(r'\s+', '', reason.lower())
+    for forbidden in FORBIDDEN_PASS_REASONS:
+        normalized_forbidden = re.sub(r'\s+', '', forbidden.lower())
+        if normalized_forbidden in normalized_reason:
+            return True
+    return False
--- a/lib/air_runtime/task_graph.py
+++ b/lib/air_runtime/task_graph.py
@@ -22,6 +22,7 @@ class TaskNode:
    meta: dict[str, Any] = field(default_factory=dict)
    test_required: bool = False  # P1-19.1: 边界测试强制标记
    adr_refs: list[str] = field(default_factory=list)  # P1-21: ADR→任务溯源链
+    keep_constraints: list[str] = field(default_factory=list)  # P1-24: 保留约束（不可修改的文件/目录）


@dataclass
@@ -168,6 +169,7 @@ class TaskGraph:
                write_set=list(nd.get("writeSet", [])),
                test_required=nd.get("testRequired", False),
                adr_refs=list(nd.get("adrRefs", [])),
+                keep_constraints=list(nd.get("keepConstraints", [])),
            )
        for ed in data.get("edges", []):
            graph.edges.append(Edge(
--- a/test_arc.py
+++ b/test_arc.py
@@ -54,7 +54,7 @@ def test_arc_build_graph():
 | [T-003] 任务3 | DONE | src/c/ | 完成 |
 """)

-        graph, violations = _build_graph_from_todo(todo_path)
+        graph, violations, ambiguity_warnings = _build_graph_from_todo(todo_path)

        # 检查节点数量
        assert len(graph.nodes) >= 3, f"应有至少3个节点，实际: {len(graph.nodes)}"
@@ -66,7 +66,7 @@ def test_arc_build_graph():
        # 检查 Done When 违规检测
        assert len(violations) > 0, "应该有 Done When 不含'测试通过'的违规"

-        print(f"✓ DAG 构建测试通过，节点数: {len(graph.nodes)}, 测试任务: {len(test_nodes)}, 违规: {len(violations)}")
+        print(f"✓ DAG 构建测试通过，节点数: {len(graph.nodes)}, 测试任务: {len(test_nodes)}, 违规: {len(violations)}, 歧义警告: {len(ambiguity_warnings)}")
    return True


@@ -107,6 +107,153 @@ def test_arc_parallel_review():
    return True


+def test_validate_task_description():
+    """P1-24: 测试任务描述歧义词检测"""
+    from air_runtime.review import validate_task_description
+
+    # 包含歧义词
+    warnings = validate_task_description("T-001", "清理旧产品实现并重建新CMake骨架")
+    assert len(warnings) > 0, "应该检测到歧义词'清理'"
+    assert "清理" in warnings[0], f"警告应包含'清理': {warnings[0]}"
+
+    # 不包含歧义词
+    clean_warnings = validate_task_description("T-002", "重构CMakeLists.txt去掉sipclient依赖")
+    assert len(clean_warnings) == 0, f"不应该有歧义词警告: {clean_warnings}"
+
+    # 多个歧义词
+    multi_warnings = validate_task_description("T-003", "清理旧代码并优化模块结构，更新API接口")
+    assert len(multi_warnings) >= 3, f"应该检测到至少3个歧义词: {len(multi_warnings)}"
+
+    print(f"✓ validate_task_description 测试通过，歧义词检测: {len(warnings)}/{len(clean_warnings)}/{len(multi_warnings)}")
+    return True
+
+
+def test_ambiguity_warnings_in_graph():
+    """P1-24: 测试 _build_graph_from_todo 产出歧义警告"""
+    from air_runtime.modes.arc_mode import _build_graph_from_todo
+
+    with tempfile.TemporaryDirectory() as tmpdir:
+        todo_path = Path(tmpdir) / "todo.md"
+        todo_path.write_text("""
+| Task | Status | Files/Dirs | Done When |
+|------|--------|------------|-----------|
+| [T-001] 清理旧模块 | TODO | src/module.cpp | 编译通过 |
+| [T-002] 优化数据库查询性能 | TODO | db/query.cpp | 测试通过 |
+| [T-003] 新增日志模块 | TODO | log/spdlog.cpp | 测试通过 |
+""")
+        graph, violations, ambiguity_warnings = _build_graph_from_todo(todo_path)
+
+        # T-001 "清理" 和 T-002 "优化" 应该触发歧义警告
+        assert len(ambiguity_warnings) >= 2, f"应有至少2个歧义警告，实际: {len(ambiguity_warnings)}"
+        # T-003 "新增" 应该没有歧义
+        ambiguous_task_ids = set()
+        for w in ambiguity_warnings:
+            tid = w.split("]")[0].replace("[", "") if "]" in w else ""
+            ambiguous_task_ids.add(tid)
+        assert "T-003" not in ambiguous_task_ids, f"'新增'不应触发歧义警告: {ambiguity_warnings}"
+
+        print(f"✓ 歧义警告产出测试通过，警告: {len(ambiguity_warnings)}，违规: {len(violations)}")
+    return True
+
+
+def test_forbidden_degradation():
+    """3.2.9b: 测试降级语言检测"""
+    from air_runtime.review import check_forbidden_degradation
+
+    # 包含降级语言
+    found = check_forbidden_degradation("我们先用兜底方案实现，以后再优化")
+    assert len(found) > 0, "应该检测到降级语言"
+    assert "兜底方案" in found or "以后再优化" in found, f"应检测到具体模式: {found}"
+
+    # 不包含降级语言
+    clean = check_forbidden_degradation("按照AirArc设计方案实现，使用RAII管理资源")
+    assert len(clean) == 0, f"不应有降级语言: {clean}"
+
+    # 多种降级语言
+    multi = check_forbidden_degradation("先用临时方案实现，以后再删掉这个兜底方案")
+    assert len(multi) >= 3, f"应检测到多个降级模式: {len(multi)}"
+
+    print(f"✓ 降级语言检测测试通过: {len(found)}/{len(clean)}/{len(multi)}")
+    return True
+
+
+def test_evaluate_review_pass():
+    """AirRvr: 测试三层审查放行标准"""
+    from air_runtime.review import evaluate_review_pass, is_forbidden_pass_reason
+
+    # 全部通过
+    result = evaluate_review_pass(
+        code_to_design_table=[
+            {"designPoint": "使用RAII管理资源", "status": "aligned"},
+            {"designPoint": "接口使用智能指针", "status": "aligned"},
+        ],
+        sdb_report={"findings": []},
+        test_results={"suites": [{"name": "unit", "status": "pass"}]},
+    )
+    assert result.verdict == "pass", f"全部通过应为 pass，实际: {result.verdict}"
+    assert result.layer1_pass and result.layer2_pass and result.layer3_pass
+
+    # 第一层有 divergent → fail
+    result2 = evaluate_review_pass(
+        code_to_design_table=[
+            {"designPoint": "使用RAII管理资源", "status": "divergent"},
+        ],
+        sdb_report={"findings": []},
+        test_results={"suites": [{"name": "unit", "status": "pass"}]},
+    )
+    assert result2.verdict == "fail", f"第一层有divergent应为fail，实际: {result2.verdict}"
+    assert not result2.layer1_pass
+
+    # 第一层通过，第二层有 critical finding → conditional-pass
+    result3 = evaluate_review_pass(
+        code_to_design_table=[
+            {"designPoint": "使用RAII管理资源", "status": "aligned"},
+        ],
+        sdb_report={"findings": [{"severity": "critical", "description": "use-after-free"}]},
+        test_results={"suites": [{"name": "unit", "status": "pass"}]},
+    )
+    assert result3.verdict == "conditional-pass", f"应为conditional-pass，实际: {result3.verdict}"
+
+    # 第一层有 missing → fail
+    result4 = evaluate_review_pass(
+        code_to_design_table=[
+            {"designPoint": "使用RAII管理资源", "status": "aligned"},
+            {"designPoint": "AES加密", "status": "missing"},
+        ],
+        sdb_report={"findings": []},
+        test_results={"suites": []},
+    )
+    assert result4.verdict == "fail", f"第一层有missing应为fail，实际: {result4.verdict}"
+
+    # 禁止的表面原因
+    assert is_forbidden_pass_reason("测试全绿") == True
+    assert is_forbidden_pass_reason("编译通过无报错") == True
+    assert is_forbidden_pass_reason("代码实现正确，三层审查均通过") == False
+
+    print(f"✓ 三层审查放行标准测试通过: pass/fail/cond-pass/forbidden")
+    return True
+
+
+def test_keep_constraints_in_tasknode():
+    """P1-24: 测试 TaskNode.keep_constraints 字段"""
+    from air_runtime.task_graph import TaskNode
+
+    node = TaskNode(
+        id="T-001",
+        task="重构 CMakeLists.txt 去掉 sipclient 依赖",
+        keep_constraints=["src/ 目录下所有现有源文件不得删除或修改"],
+    )
+    assert node.keep_constraints == ["src/ 目录下所有现有源文件不得删除或修改"]
+    assert len(node.keep_constraints) == 1
+
+    # 空约束
+    node2 = TaskNode(id="T-002", task="新增日志模块")
+    assert node2.keep_constraints == []
+
+    print("✓ TaskNode.keep_constraints 字段测试通过")
+    return True
+
+
 def main():
    print("=" * 50)
    print("Arc Mode 功能测试")
@@ -116,6 +263,11 @@ def main():
        ("三阶段门控", test_arc_phase_gate),
        ("DAG 构建", test_arc_build_graph),
        ("并行审查", test_arc_parallel_review),
+        ("P1-24 歧义词检测", test_validate_task_description),
+        ("P1-24 歧义警告产出", test_ambiguity_warnings_in_graph),
+        ("3.2.9b 降级语言检测", test_forbidden_degradation),
+        ("三层审查放行标准", test_evaluate_review_pass),
+        ("P1-24 keep_constraints", test_keep_constraints_in_tasknode),
    ]

    passed = 0
--- a/test_t_121_122.py
+++ b/test_t_121_122.py
@@ -1,5 +1,10 @@
 #!/usr/bin/env python3
-"""T-1.21 + T-1.22 端到端测试：arc → eng → do → merge → task-graph sync 全链路"""
+"""T-1.22 + T-1.23 端到端测试：arc → eng dispatch → do → merge → task-graph sync 全链路
+
+T-1.21 (弱模型优化) 测试见 test_arc.py (test_validate_task_description 等)
+T-1.22: Dispatch → Worker 桥接（spawn_workers + 全专家插件路由）
+T-1.23: Merge → TaskGraph 状态同步（merge 后更新 task-graph.json 节点 status）
+"""

 import sys
 import tempfile
@@ -254,7 +259,7 @@ def test_dispatch_no_workers_means_no_dispatch():

 def main():
    print("=" * 50)
-    print("T-1.21 + T-1.22 端到端全链路测试")
+    print("T-1.22 + T-1.23 端到端全链路测试")
    print("=" * 50)

    tests = [