feat: improve skill import and runtime delivery

resources/skills/product-marketing/SKILL.md

@@ -0,0 +1,200 @@
+---
+name: product-marketing
+version: 1.0.0
+description: |
+  编排产品事实、功能列表、招标参数、产品 PPT、技术方案、一页纸、白皮书、招标响应、
+  演示套件、客户案例和竞品定位等多个独立技能。用于一次请求需要选择或组合多种
+  产品市场产物，并确保它们共享同一事实版本、术语和承诺边界。
+allowed-tools:
+  - Read
+  - Grep
+  - Glob
+  - Execute
+compatibility: 可独立规划；执行节点需要对应子技能可用
+---
+
+# 产品市场总编排
+
+本技能只负责需求澄清、路由、依赖、门禁和跨产物一致性。各产物的方法和模板由
+对应独立技能负责，不能在总技能中重新实现一份简化版本。
+
+## 子技能
+
+| 技能 | 唯一职责 |
+|---|---|
+| `product-evidence` | 冻结产品事实、状态、证据和限制 |
+| `product-feature-catalog` | 功能目录和版本矩阵 |
+| `tender-technical-spec` | 可采购、可验收的招标技术规格 |
+| `product-presentation` | 产品介绍 PPT 与讲稿 |
+| `technical-proposal` | 客户或投标技术方案 |
+| `product-one-pager` | 一页纸产品概览和彩页文案 |
+| `solution-whitepaper` | 原理、架构、证据和边界白皮书 |
+| `tender-response-matrix` | 招标要求逐条响应、缺口和偏离 |
+| `sales-demo-kit` | 可执行演示套件、回退和演练材料 |
+| `customer-case-study` | 经授权且可复核的客户案例 |
+| `competitive-positioning` | 有来源的竞品矩阵和定位 |
+
+`deai-writing` 和 `longdoc-docx` 是可选质量与导出技能，不承担产品事实判断。
+
+## 第一步：形成任务简报
+
+必须确认：
+
+- 产品和版本。
+- 目标受众、决策目标和发布渠道。
+- 需要的产物、格式、篇幅、语言和截止时间。
+- 公开级别、客户信息和竞品信息的使用权限。
+- 原始事实材料、招标文件、客户需求和品牌资产。
+- 最终审批人。
+
+信息不足时把缺口写入计划，不默认补齐。
+
+## 第二步：生成路由计划
+
+复制 `templates/route-plan.example.json`，只选择完成请求所需节点：
+
+先探测可用的 Python 3 解释器：Windows 优先使用 `python`，macOS/Linux
+优先使用 `python3`。下文 `<python>` 表示探测成功的解释器命令。
+
+```bash
+<python> "<skill-dir>/scripts/validate_route_plan.py" ./route-plan.json
+```
+
+所有产物必须直接或间接依赖唯一的 `product-evidence` 节点。节点缺少对应技能时
+明确报告缺失，不允许由总技能静默生成低质量替代物。
+
+`depends_on` 表示硬依赖：被依赖节点未 `completed` 时，本节点不能进入 `running`
+或 `completed`。可选输入（例如尚无授权的客户案例）不要写进 `depends_on`，而是
+在 `reason` 中说明“可用则引用，不可用则不提及”。
+
+## 推荐路由
+
+### 投标响应（我方应标）
+
+```text
+product-evidence
+  → tender-response-matrix（提取要求与缺口）
+  → technical-proposal
+  → tender-response-matrix（回填方案章节和证明材料）
+```
+
+### 招标文件编制（采购方）
+
+```text
+product-evidence → product-feature-catalog → tender-technical-spec
+```
+
+`tender-technical-spec` 面向采购参数编写，不参与我方符合性判定，两条路由不要
+混用。
+
+### 客户方案
+
+```text
+product-evidence
+  → product-feature-catalog
+  → technical-proposal
+  → [product-presentation, product-one-pager]
+```
+
+PPT 和一页纸从已批准方案摘要派生，避免重新解释范围。
+
+### 产品发布与销售
+
+```text
+product-evidence
+  → product-feature-catalog
+  → [competitive-positioning, customer-case-study]
+  → product-one-pager
+  → product-presentation
+  → sales-demo-kit
+```
+
+客户案例没有授权时把该节点标记 `skipped` 且 `required` 为 false，其他节点可
+继续，但不得引用该案例。
+
+### 白皮书
+
+```text
+product-evidence → product-feature-catalog → solution-whitepaper
+```
+
+## 第三步：执行门禁
+
+每个节点开始前检查依赖是否通过；失败节点的下游必须阻断，不能标记成功。执行
+过程中更新 `status`，交付前用 final 阶段复核：
+
+```bash
+<python> "<skill-dir>/scripts/validate_route_plan.py" ./route-plan.json \
+  --phase final --output-root .
+```
+
+final 阶段要求必需节点全部 `completed`、依赖链无未完成节点，并核验每个声明产物
+文件存在且非空。
+
+门禁分两类：
+
+- `evidence-validation`、`cross-artifact-consistency` 由本套件自行判定，必须
+  `passed`，不能豁免。
+- `confidentiality-review`、`human-approval` 取决于组织流程。需要评审时记为
+  `passed`；组织不要求时记为 `waived` 并在 `waiver` 写明责任人与理由，校验通过
+  但会输出警告，保留豁免记录。
+
+最终统一核对：
+
+- 产品名称、版本、功能状态和术语一致。
+- 相同参数的数值、单位、条件和统计口径一致。
+- `planned`、`beta`、`released` 没有跨产物变形。
+- 客户案例、Logo、引语和竞品结论权限一致。
+- 技术方案、PPT、一页纸和演示套件使用相同架构与工作流。
+- 所有对外主张可追溯到同一版 `product-evidence.json`。
+- 不含密钥、私有地址、内部证据路径和未批准承诺。
+
+## 第四步：质量与导出
+
+中文叙事产物可调用 `deai-writing`；长文可调用 `longdoc-docx`；PPT 使用
+`product-presentation` 自带构建器。质量工具只能修正表达和排版，不能更改事实、
+成熟度、参数或授权边界。
+
+### 交付物形态
+
+Markdown 与 JSON 是中间产物，不是交付物。除非用户另有指定，交付物为：
+
+| 产物 | 交付格式 |
+| --- | --- |
+| 文档类（技术方案、白皮书、功能列表、响应矩阵、招标规格、一页纸、演示套件、竞争定位） | DOCX |
+| 产品 PPT | PPTX |
+
+PDF 只作为排版核验中间件，核验后删除，不放入交付目录；用户明确要求时才交付。
+
+### 目录与密级
+
+中间产物与交付物必须分离，交付目录按各产物自身声明的密级分区：
+
+```text
+build/          # 事实清单、路由计划、Markdown、构建配置
+  check/        # 核验用 PDF、核验报告、页面 PNG
+deliverables/
+  public/       # 可公开
+  restricted/   # 受控客户交流与投标
+  internal/     # 仅内部
+```
+
+各子技能的 `output` 一律指向 `deliverables/<密级>/`，核验产物一律写入
+`build/check/`。交付目录内只允许出现 DOCX 与 PPTX。
+
+单文件产物直接放 `build/<产物名>.md`。技术方案、白皮书等需要分章节的长文，改用
+`build/<产物名>/` 子目录，内部按 `longdoc-docx` 的 `chapters/`、`assets/`、
+`drafts/` 分层，避免多个长文的章节文件在 `build/` 根目录互相混淆。
+
+密级以产物自身标注为准，不得由编排者主观下调。分区后须扫描公开级产物，确认
+未夹带受控结论、内部路径与凭据。
+
+路由计划节点的 `outputs` 必须声明真实交付物路径，Markdown 与 JSON 记入
+`intermediates`。若 `outputs` 指向中间产物，交付门禁将只校验中间产物而放行缺失
+的真实交付物。
+
+## 完成标准
+
+路由计划通过校验；所有必需节点完成；无失败依赖被忽略；各产物共享同一事实版本；
+跨产物一致性、保密审查和人工审批全部通过，并保留选择、跳过和阻断理由；交付目录
+中每个产物都以约定格式实际存在，且已按密级分区。

resources/skills/product-marketing/scripts/validate_route_plan.py

@@ -0,0 +1,422 @@
+#!/usr/bin/env python3
+"""Validate product-marketing orchestration route plans."""
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path, PurePosixPath
+
+
+SKILL_ARTIFACT = {
+    "product-evidence": "evidence",
+    "product-feature-catalog": "feature-catalog",
+    "tender-technical-spec": "technical-spec",
+    "product-presentation": "presentation",
+    "technical-proposal": "technical-proposal",
+    "product-one-pager": "one-pager",
+    "solution-whitepaper": "whitepaper",
+    "tender-response-matrix": "tender-response",
+    "sales-demo-kit": "sales-demo",
+    "customer-case-study": "case-study",
+    "competitive-positioning": "competitive-positioning",
+}
+STATUSES = {"pending", "running", "completed", "failed", "skipped"}
+GATE_STATUSES = {"pending", "passed", "failed", "waived"}
+CONFIDENTIALITY = {"public", "restricted", "internal"}
+# Gates this suite can evaluate itself; they must actually pass.
+CORE_GATES = {"evidence-validation", "cross-artifact-consistency"}
+# Gates that depend on organizational policy; they may be waived with a reason.
+POLICY_GATES = {"confidentiality-review", "human-approval"}
+REQUIRED_GATES = CORE_GATES | POLICY_GATES
+SECRET_PATTERN = re.compile(r"(?i)(?:token|api[_-]?key|secret|password)=")
+
+
+def issue(path, message):
+    return {"path": path, "message": message}
+
+
+def nonempty(value):
+    return isinstance(value, str) and bool(value.strip())
+
+
+def validate_output_path(value, path, errors):
+    if not nonempty(value):
+        errors.append(issue(path, "必须是非空字符串"))
+        return
+    normalized = value.replace("\\", "/")
+    pure = PurePosixPath(normalized)
+    if pure.is_absolute() or ".." in pure.parts:
+        errors.append(issue(path, "必须是工作目录内的相对路径"))
+    if SECRET_PATTERN.search(value):
+        errors.append(issue(path, "路径疑似包含密钥或令牌"))
+
+
+def known_dependencies(node, nodes_by_id):
+    return [
+        dependency
+        for dependency in node.get("depends_on", [])
+        if isinstance(dependency, str) and dependency in nodes_by_id
+    ]
+
+
+def topological_order(nodes_by_id):
+    """Return (order, cycle). Kahn's algorithm keeps deep graphs iterative."""
+    dependents = {node_id: [] for node_id in nodes_by_id}
+    remaining = {}
+    for node_id, node in nodes_by_id.items():
+        dependencies = set(known_dependencies(node, nodes_by_id)) - {node_id}
+        remaining[node_id] = len(dependencies)
+        for dependency in dependencies:
+            dependents[dependency].append(node_id)
+
+    queue = [node_id for node_id, count in remaining.items() if count == 0]
+    order = []
+    while queue:
+        node_id = queue.pop()
+        order.append(node_id)
+        for dependent in dependents[node_id]:
+            remaining[dependent] -= 1
+            if remaining[dependent] == 0:
+                queue.append(dependent)
+
+    if len(order) == len(nodes_by_id):
+        return order, None
+    return order, sorted(node_id for node_id in nodes_by_id if remaining[node_id] > 0)
+
+
+def evidence_reachability(order, nodes_by_id, evidence_id):
+    reaches = {}
+    for node_id in order:
+        dependencies = known_dependencies(nodes_by_id[node_id], nodes_by_id)
+        reaches[node_id] = any(
+            dependency == evidence_id or reaches.get(dependency, False)
+            for dependency in dependencies
+        )
+    return reaches
+
+
+def validate_plan(data, phase="plan", output_root=None):
+    errors = []
+    warnings = []
+    if not isinstance(data, dict):
+        return [issue("$", "根节点必须是对象")], warnings
+    if data.get("schema_version") != 1:
+        errors.append(issue("schema_version", "当前仅支持整数 1"))
+
+    request = data.get("request")
+    if not isinstance(request, dict):
+        errors.append(issue("request", "必须是对象"))
+        request = {}
+    for field in ("id", "objective", "channel", "evidence_ref", "product_version"):
+        if not nonempty(request.get(field)):
+            errors.append(issue(f"request.{field}", "必须是非空字符串"))
+    if request.get("confidentiality") not in CONFIDENTIALITY:
+        errors.append(
+            issue(
+                "request.confidentiality",
+                f"必须是 {sorted(CONFIDENTIALITY)} 之一",
+            )
+        )
+    if nonempty(request.get("evidence_ref")):
+        validate_output_path(
+            request["evidence_ref"],
+            "request.evidence_ref",
+            errors,
+        )
+
+    nodes = data.get("nodes")
+    if not isinstance(nodes, list) or not nodes:
+        errors.append(issue("nodes", "必须是非空数组"))
+        return errors, warnings
+
+    nodes_by_id = {}
+    all_outputs = {}
+    evidence_nodes = []
+    for index, node in enumerate(nodes):
+        path = f"nodes[{index}]"
+        if not isinstance(node, dict):
+            errors.append(issue(path, "必须是对象"))
+            continue
+        node_id = node.get("id")
+        if not nonempty(node_id):
+            errors.append(issue(f"{path}.id", "必须是非空字符串"))
+            continue
+        if node_id in nodes_by_id:
+            errors.append(issue(f"{path}.id", f"节点 ID 重复：{node_id}"))
+            continue
+        nodes_by_id[node_id] = node
+        skill = node.get("skill")
+        if skill not in SKILL_ARTIFACT:
+            errors.append(issue(f"{path}.skill", f"未知技能：{skill!r}"))
+        elif node.get("artifact_type") != SKILL_ARTIFACT[skill]:
+            errors.append(
+                issue(
+                    f"{path}.artifact_type",
+                    f"{skill} 应生成 {SKILL_ARTIFACT[skill]}",
+                )
+            )
+        if skill == "product-evidence":
+            evidence_nodes.append(node_id)
+        dependencies = node.get("depends_on")
+        if not isinstance(dependencies, list):
+            errors.append(issue(f"{path}.depends_on", "必须是数组"))
+        elif len(dependencies) != len(set(dependencies)):
+            errors.append(issue(f"{path}.depends_on", "存在重复依赖"))
+        if not isinstance(node.get("required"), bool):
+            errors.append(issue(f"{path}.required", "必须是布尔值"))
+        if node.get("status") not in STATUSES:
+            errors.append(
+                issue(
+                    f"{path}.status",
+                    f"必须是 {sorted(STATUSES)} 之一",
+                )
+            )
+        if node.get("required") is True and node.get("status") == "skipped":
+            errors.append(issue(f"{path}.status", "必需节点不能标记 skipped"))
+        if not nonempty(node.get("reason")):
+            errors.append(issue(f"{path}.reason", "必须说明选择理由"))
+        outputs = node.get("outputs")
+        if not isinstance(outputs, list) or not outputs:
+            errors.append(issue(f"{path}.outputs", "必须是非空数组"))
+        else:
+            for output_index, output in enumerate(outputs):
+                output_path = f"{path}.outputs[{output_index}]"
+                validate_output_path(output, output_path, errors)
+                if output in all_outputs:
+                    errors.append(
+                        issue(
+                            output_path,
+                            f"输出与 {all_outputs[output]} 重复：{output}",
+                        )
+                    )
+                else:
+                    all_outputs[output] = node_id
+
+    for node_id, node in nodes_by_id.items():
+        for dependency in node.get("depends_on", []):
+            if dependency == node_id:
+                errors.append(
+                    issue(f"nodes.{node_id}.depends_on", "不能依赖自身")
+                )
+            elif dependency not in nodes_by_id:
+                errors.append(
+                    issue(
+                        f"nodes.{node_id}.depends_on",
+                        f"依赖节点不存在：{dependency}",
+                    )
+                )
+
+    order, cycle = topological_order(nodes_by_id)
+    if cycle:
+        errors.append(issue("nodes", f"依赖存在环：{'、'.join(cycle)}"))
+
+    if len(evidence_nodes) != 1:
+        errors.append(issue("nodes", "必须且只能有一个 product-evidence 节点"))
+    elif not cycle:
+        evidence_id = evidence_nodes[0]
+        reaches = evidence_reachability(order, nodes_by_id, evidence_id)
+        for node_id in nodes_by_id:
+            if node_id != evidence_id and not reaches.get(node_id, False):
+                errors.append(
+                    issue(
+                        f"nodes.{node_id}.depends_on",
+                        "所有产物必须直接或间接依赖 product-evidence",
+                    )
+                )
+
+    validate_execution(nodes_by_id, errors)
+    validate_gates(data.get("final_gates"), errors, warnings, phase)
+    if phase == "final":
+        validate_final_phase(nodes_by_id, output_root, errors)
+
+    if len(nodes_by_id) == 1:
+        warnings.append(issue("nodes", "路由计划没有任何最终产物节点"))
+    return errors, warnings
+
+
+def validate_execution(nodes_by_id, errors):
+    for node_id, node in nodes_by_id.items():
+        status = node.get("status")
+        if status not in ("running", "completed"):
+            continue
+        for dependency in known_dependencies(node, nodes_by_id):
+            dependency_node = nodes_by_id[dependency]
+            dependency_status = dependency_node.get("status")
+            if dependency_status == "completed":
+                continue
+            if (
+                dependency_status == "skipped"
+                and dependency_node.get("required") is False
+            ):
+                continue
+            errors.append(
+                issue(
+                    f"nodes.{node_id}.status",
+                    f"依赖 {dependency} 状态为 {dependency_status}，"
+                    f"不能标记 {status}",
+                )
+            )
+
+
+def validate_gates(gates, errors, warnings, phase):
+    if not isinstance(gates, list):
+        errors.append(issue("final_gates", "必须是数组"))
+        return {}
+    resolved = {}
+    for index, gate in enumerate(gates):
+        path = f"final_gates[{index}]"
+        if isinstance(gate, str):
+            if not nonempty(gate):
+                errors.append(issue(path, "门禁名称不能为空"))
+                continue
+            resolved[gate] = {"status": None, "waiver": None}
+        elif isinstance(gate, dict):
+            name = gate.get("name")
+            if not nonempty(name):
+                errors.append(issue(f"{path}.name", "门禁名称不能为空"))
+                continue
+            status = gate.get("status")
+            if status not in GATE_STATUSES:
+                errors.append(
+                    issue(
+                        f"{path}.status",
+                        f"必须是 {sorted(GATE_STATUSES)} 之一",
+                    )
+                )
+                continue
+            waiver = gate.get("waiver")
+            if status == "waived":
+                if name not in POLICY_GATES:
+                    errors.append(
+                        issue(
+                            f"{path}.status",
+                            f"{name} 由本套件自行判定，不能豁免",
+                        )
+                    )
+                    continue
+                if not nonempty(waiver):
+                    errors.append(
+                        issue(f"{path}.waiver", "豁免必须写明责任人与理由")
+                    )
+                    continue
+            resolved[name] = {"status": status, "waiver": waiver}
+        else:
+            errors.append(issue(path, "必须是字符串或对象"))
+    missing = REQUIRED_GATES - set(resolved)
+    if missing:
+        errors.append(issue("final_gates", f"缺少门禁：{sorted(missing)}"))
+    if phase == "final":
+        for name in sorted(REQUIRED_GATES & set(resolved)):
+            status = resolved[name]["status"]
+            if status == "passed":
+                continue
+            if status == "waived":
+                warnings.append(
+                    issue(
+                        "final_gates",
+                        f"{name} 已豁免：{resolved[name]['waiver']}",
+                    )
+                )
+                continue
+            errors.append(
+                issue(
+                    "final_gates",
+                    f"交付前门禁 {name} 必须记录 passed，当前为 "
+                    f"{status or '未记录'}",
+                )
+            )
+    return resolved
+
+
+def validate_final_phase(nodes_by_id, output_root, errors):
+    for node_id, node in nodes_by_id.items():
+        status = node.get("status")
+        if node.get("required") is True and status != "completed":
+            errors.append(
+                issue(
+                    f"nodes.{node_id}.status",
+                    f"交付前必需节点必须 completed，当前为 {status}",
+                )
+            )
+        elif status not in ("completed", "skipped"):
+            errors.append(
+                issue(
+                    f"nodes.{node_id}.status",
+                    f"交付前可选节点必须 completed 或 skipped，当前为 {status}",
+                )
+            )
+        if status != "completed" or output_root is None:
+            continue
+        outputs = node.get("outputs")
+        if not isinstance(outputs, list):
+            continue
+        for output in outputs:
+            if not nonempty(output):
+                continue
+            artifact = output_root / output
+            if not artifact.is_file():
+                errors.append(
+                    issue(
+                        f"nodes.{node_id}.outputs",
+                        f"声明的产物不存在：{output}",
+                    )
+                )
+            elif artifact.stat().st_size == 0:
+                errors.append(
+                    issue(
+                        f"nodes.{node_id}.outputs",
+                        f"产物为空文件：{output}",
+                    )
+                )
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(description="验证产品市场技能路由计划")
+    parser.add_argument("plan", help="route-plan.json")
+    parser.add_argument(
+        "--phase",
+        choices=("plan", "final"),
+        default="plan",
+        help="plan 校验结构，final 追加交付前状态与产物核验",
+    )
+    parser.add_argument(
+        "--output-root",
+        help="final 阶段解析 outputs 相对路径的根目录",
+    )
+    parser.add_argument("--json", action="store_true")
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+    path = Path(args.plan).expanduser().resolve()
+    output_root = (
+        Path(args.output_root).expanduser().resolve() if args.output_root else None
+    )
+    with path.open(encoding="utf-8") as handle:
+        data = json.load(handle)
+    errors, warnings = validate_plan(data, phase=args.phase, output_root=output_root)
+    report = {
+        "file": str(path),
+        "phase": args.phase,
+        "valid": not errors,
+        "errors": errors,
+        "warnings": warnings,
+        "nodes": len(data.get("nodes", [])) if isinstance(data, dict) else 0,
+    }
+    if args.json:
+        print(json.dumps(report, ensure_ascii=False, indent=2))
+    else:
+        for level, items in (("错误", errors), ("警告", warnings)):
+            for item in items:
+                print(f"[{level}] {item['path']}：{item['message']}")
+        print(
+            f"路由核验（{args.phase}）：错误 {len(errors)}，警告 {len(warnings)}，"
+            f"状态 {'通过' if report['valid'] else '失败'}"
+        )
+    return 0 if report["valid"] else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

resources/skills/product-marketing/templates/route-plan.example.json

@@ -0,0 +1,77 @@
+{
+  "schema_version": 1,
+  "request": {
+    "id": "PM-001",
+    "objective": "为目标客户准备产品介绍与技术方案",
+    "audiences": [
+      "技术负责人",
+      "业务负责人"
+    ],
+    "channel": "受控客户交流",
+    "confidentiality": "restricted",
+    "evidence_ref": "product-evidence.json",
+    "product_version": "1.0"
+  },
+  "nodes": [
+    {
+      "id": "evidence",
+      "skill": "product-evidence",
+      "artifact_type": "evidence",
+      "depends_on": [],
+      "required": true,
+      "status": "pending",
+      "outputs": [
+        "product-evidence.json"
+      ],
+      "reason": "所有产物共享同一事实版本"
+    },
+    {
+      "id": "catalog",
+      "skill": "product-feature-catalog",
+      "artifact_type": "feature-catalog",
+      "depends_on": [
+        "evidence"
+      ],
+      "required": true,
+      "status": "pending",
+      "outputs": [
+        "feature-catalog.md"
+      ],
+      "reason": "先固定产品能力范围"
+    },
+    {
+      "id": "proposal",
+      "skill": "technical-proposal",
+      "artifact_type": "technical-proposal",
+      "depends_on": [
+        "catalog"
+      ],
+      "required": true,
+      "status": "pending",
+      "outputs": [
+        "technical-proposal.md"
+      ],
+      "reason": "客户方案依赖已核验能力目录"
+    },
+    {
+      "id": "presentation",
+      "skill": "product-presentation",
+      "artifact_type": "presentation",
+      "depends_on": [
+        "proposal"
+      ],
+      "required": true,
+      "status": "pending",
+      "outputs": [
+        "product-presentation.pptx"
+      ],
+      "reason": "PPT 从已批准方案摘要派生"
+    }
+  ],
+  "final_gates": [
+    { "name": "evidence-validation", "status": "pending" },
+    { "name": "cross-artifact-consistency", "status": "pending" },
+    { "name": "confidentiality-review", "status": "pending" },
+    { "name": "human-approval", "status": "pending" }
+  ]
+}

resources/skills/product-marketing/tests/test_validate_route_plan.py

@@ -0,0 +1,157 @@
+import copy
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+
+SKILL_DIR = Path(__file__).resolve().parents[1]
+SCRIPT = SKILL_DIR / "scripts" / "validate_route_plan.py"
+SPEC = importlib.util.spec_from_file_location("validate_route_plan", SCRIPT)
+validate_route_plan = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(validate_route_plan)
+
+
+class RoutePlanTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        template = SKILL_DIR / "templates" / "route-plan.example.json"
+        cls.plan = json.loads(template.read_text(encoding="utf-8"))
+
+    def test_example_plan_is_valid(self):
+        errors, warnings = validate_route_plan.validate_plan(self.plan)
+        self.assertEqual(errors, [])
+        self.assertEqual(warnings, [])
+
+    def test_rejects_cycle(self):
+        plan = copy.deepcopy(self.plan)
+        plan["nodes"][0]["depends_on"] = ["presentation"]
+        errors, _ = validate_route_plan.validate_plan(plan)
+        self.assertTrue(any("依赖存在环" in item["message"] for item in errors))
+
+    def test_rejects_artifact_without_evidence_dependency(self):
+        plan = copy.deepcopy(self.plan)
+        plan["nodes"][1]["depends_on"] = []
+        errors, _ = validate_route_plan.validate_plan(plan)
+        self.assertTrue(
+            any("product-evidence" in item["message"] for item in errors)
+        )
+
+    def test_rejects_duplicate_output(self):
+        plan = copy.deepcopy(self.plan)
+        plan["nodes"][2]["outputs"] = ["feature-catalog.md"]
+        errors, _ = validate_route_plan.validate_plan(plan)
+        self.assertTrue(any("输出与" in item["message"] for item in errors))
+
+    def test_rejects_completed_node_with_unfinished_dependency(self):
+        plan = copy.deepcopy(self.plan)
+        plan["nodes"][1]["status"] = "completed"
+        errors, _ = validate_route_plan.validate_plan(plan)
+        self.assertTrue(
+            any("不能标记 completed" in item["message"] for item in errors)
+        )
+
+    def test_final_phase_requires_completion_and_artifacts(self):
+        plan = copy.deepcopy(self.plan)
+        errors, _ = validate_route_plan.validate_plan(plan, phase="final")
+        self.assertTrue(
+            any("必须 completed" in item["message"] for item in errors)
+        )
+        self.assertTrue(
+            any("必须记录 passed" in item["message"] for item in errors)
+        )
+
+    def test_policy_gate_can_be_waived_with_reason(self):
+        plan = copy.deepcopy(self.plan)
+        for node in plan["nodes"]:
+            node["status"] = "completed"
+        for gate in plan["final_gates"]:
+            gate["status"] = "passed"
+        approval = next(
+            g for g in plan["final_gates"] if g["name"] == "human-approval"
+        )
+        approval["status"] = "waived"
+        errors, _ = validate_route_plan.validate_plan(plan, phase="final")
+        self.assertTrue(any("豁免必须写明" in i["message"] for i in errors))
+
+        approval["waiver"] = "内部草稿，责任人张三，2026-08-01"
+        errors, warnings = validate_route_plan.validate_plan(plan, phase="final")
+        self.assertEqual(errors, [])
+        self.assertTrue(any("已豁免" in i["message"] for i in warnings))
+
+    def test_core_gate_cannot_be_waived(self):
+        plan = copy.deepcopy(self.plan)
+        for node in plan["nodes"]:
+            node["status"] = "completed"
+        for gate in plan["final_gates"]:
+            gate["status"] = "passed"
+        core = next(
+            g for g in plan["final_gates"] if g["name"] == "evidence-validation"
+        )
+        core["status"] = "waived"
+        core["waiver"] = "跳过"
+        errors, _ = validate_route_plan.validate_plan(plan, phase="final")
+        self.assertTrue(any("不能豁免" in i["message"] for i in errors))
+
+    def test_final_phase_accepts_completed_plan_with_outputs(self):
+        plan = copy.deepcopy(self.plan)
+        for node in plan["nodes"]:
+            node["status"] = "completed"
+        for gate in plan["final_gates"]:
+            gate["status"] = "passed"
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            for node in plan["nodes"]:
+                for output in node["outputs"]:
+                    (root / output).write_text("content", encoding="utf-8")
+            errors, _ = validate_route_plan.validate_plan(
+                plan, phase="final", output_root=root
+            )
+            self.assertEqual(errors, [])
+            missing = copy.deepcopy(plan)
+            (root / missing["nodes"][-1]["outputs"][0]).unlink()
+            errors, _ = validate_route_plan.validate_plan(
+                missing, phase="final", output_root=root
+            )
+            self.assertTrue(
+                any("产物不存在" in item["message"] for item in errors)
+            )
+
+    def test_deep_chain_does_not_recurse(self):
+        nodes = [
+            {
+                "id": "evidence",
+                "skill": "product-evidence",
+                "artifact_type": "evidence",
+                "depends_on": [],
+                "required": True,
+                "status": "pending",
+                "outputs": ["product-evidence.json"],
+                "reason": "基线",
+            }
+        ]
+        previous = "evidence"
+        for index in range(1500):
+            node_id = f"catalog-{index}"
+            nodes.append(
+                {
+                    "id": node_id,
+                    "skill": "product-feature-catalog",
+                    "artifact_type": "feature-catalog",
+                    "depends_on": [previous],
+                    "required": True,
+                    "status": "pending",
+                    "outputs": [f"feature-catalog-{index}.md"],
+                    "reason": "链式依赖",
+                }
+            )
+            previous = node_id
+        plan = copy.deepcopy(self.plan)
+        plan["nodes"] = list(reversed(nodes))
+        errors, _ = validate_route_plan.validate_plan(plan)
+        self.assertEqual(errors, [])
+
+
+if __name__ == "__main__":
+    unittest.main()