muse-agent-example/harness/test_skill_harness.py
zizi 091b66a9bb 重构: 收敛 Agent/Skill 运行时与创作质量闭环
将角色与 Skill 从 .claude 迁入 .agent,移除 Claude CLI 运行时并接入固定 Opus 角色 profile、完整 schema、预算 deadline、raw 与回执证据链。

同步拆分 Skill 职责、复利 lesson、Gate 回放、Dashboard 人审入口、数据库登记和机械门禁;候选设计正文不包含在本提交中。
2026-08-22 02:12:32 +08:00

817 lines
31 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""skill_harness 的纯标准库离线回归测试。
所有夹具都在临时目录中构造,不读取当前仓库的 Skill,也不依赖数据库、网络或模型。
"""
from __future__ import annotations
import io
import json
import tempfile
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from typing import Optional
try:
from .skill_harness import (
DESCRIPTION_BUDGET,
REFERENCE_FILE_BUDGET_BYTES,
audit_skills,
main,
)
except ImportError: # 允许直接执行 `.venv/bin/python harness/test_skill_harness.py`
from skill_harness import (
DESCRIPTION_BUDGET,
REFERENCE_FILE_BUDGET_BYTES,
audit_skills,
main,
)
class SkillHarnessTests(unittest.TestCase):
def manifest_entry(
self,
directory: str,
*,
lifecycle: str = "platform",
compounding: str = "none",
invocation: str = "model_routed",
side_effects: Optional[list[str]] = None,
) -> dict[str, object]:
return {
"name": directory,
"lifecycle": lifecycle,
"compounding": compounding,
"invocation": invocation,
"side_effects": list(side_effects or ["none"]),
"contract_owner": "测试夹具",
"collaborates_with": [],
"skill_path": (
Path(".agent") / "skills" / directory / "SKILL.md"
).as_posix(),
}
def make_skill(
self,
root: Path,
directory: str,
*,
name: Optional[str] = None,
body: str = "# 合同\n\n只描述运行时行为。\n",
description: str = "离线夹具",
orchestrated: bool = False,
) -> Path:
skill_dir = root / ".agent" / "skills" / directory
skill_dir.mkdir(parents=True, exist_ok=True)
skill_path = skill_dir / "SKILL.md"
frontmatter_name = directory if name is None else name
flag = "disable-model-invocation: true\n" if orchestrated else ""
skill_path.write_text(
f"---\nname: {frontmatter_name}\ndescription: {description}\n{flag}---\n{body}",
encoding="utf-8",
)
self.write_manifest(root)
return skill_path
def write_manifest(
self,
root: Path,
entries: Optional[list[dict[str, object]]] = None,
*,
path: Optional[Path] = None,
) -> Path:
if entries is None:
entries = []
skills_root = root / ".agent" / "skills"
if skills_root.exists():
for directory in sorted(skills_root.iterdir(), key=lambda item: item.name):
if directory.is_dir():
entries.append(
{
"name": directory.name,
"lifecycle": "platform",
"compounding": "none",
"invocation": "model_routed",
"side_effects": ["none"],
"contract_owner": "测试夹具",
"collaborates_with": [],
"skill_path": (
Path(".agent")
/ "skills"
/ directory.name
/ "SKILL.md"
).as_posix(),
}
)
manifest_path = path or root / "harness" / "manifests" / "skills.json"
manifest_path.parent.mkdir(parents=True, exist_ok=True)
manifest_path.write_text(
json.dumps(
{"schema_version": 1, "skills": entries},
ensure_ascii=False,
indent=2,
)
+ "\n",
encoding="utf-8",
)
return manifest_path
def test_passing_project_is_independent_of_current_repository(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha")
report = audit_skills(root)
self.assertTrue(report["ok"])
self.assertEqual(report["status"], "passed")
self.assertEqual(report["skills_scanned"], 1)
self.assertEqual(report["issues"], [])
def test_pollution_is_reported_for_each_obvious_category(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"polluted",
body=(
"## 自测\n"
"pytest -q\n"
"import unittest\n"
"run test_sample.py and sample_test.py\n"
"覆盖率达到 100%。\n"
"测试通过。\n"
),
)
report = audit_skills(root)
codes = {issue["code"] for issue in report["issues"]}
self.assertFalse(report["ok"])
self.assertTrue(
{
"development_test_heading",
"pytest_reference",
"unittest_reference",
"test_file_reference",
"coverage_reference",
"test_pass_declaration",
}.issubset(codes)
)
def test_development_headings_support_multiple_levels(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"nested-development-sections",
body=(
"### 自测\n"
"这里不能登记测试命令。\n"
"## 离线验证\n"
"这里也不能登记测试命令。\n"
),
)
report = audit_skills(root)
headings = [
issue
for issue in report["issues"]
if issue["code"] == "development_test_heading"
]
self.assertFalse(report["ok"])
self.assertEqual(len(headings), 2)
self.assertEqual([issue["line"] for issue in headings], [5, 7])
def test_check_contract_command_requires_development_section_context(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"contextual-contract-command",
body=(
"业务合同字段名可以写 check_contract.py。\n"
"## 离线验证\n"
"python check_contract.py\n"
"## 业务说明\n"
"普通业务 check_contract.py 不是测试入口。\n"
),
)
report = audit_skills(root)
command_issues = [
issue
for issue in report["issues"]
if issue["code"] == "development_test_command_reference"
]
self.assertFalse(report["ok"])
self.assertEqual(len(command_issues), 1)
self.assertEqual(command_issues[0]["line"], 7)
def test_business_coverage_terms_are_not_development_pollution(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"business-coverage",
body=(
"细纲覆盖率达到 100%。\n"
"事件覆盖率为 80%。\n"
"硬约束覆盖率 100%。\n"
),
)
report = audit_skills(root)
self.assertTrue(report["ok"])
self.assertNotIn(
"coverage_reference",
{issue["code"] for issue in report["issues"]},
)
def test_development_coverage_terms_are_reported(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"development-coverage",
body=(
"coverage.py\n"
"pytest-cov\n"
"pytest --cov=harness\n"
"coverage report\n"
"测试覆盖率达到 90%。\n"
"覆盖率达到 80%。\n"
),
)
report = audit_skills(root)
self.assertFalse(report["ok"])
self.assertIn(
"coverage_reference",
{issue["code"] for issue in report["issues"]},
)
def test_frontmatter_name_must_match_skill_directory(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", name="beta")
report = audit_skills(root)
self.assertFalse(report["ok"])
self.assertIn(
"name_directory_mismatch",
{issue["code"] for issue in report["issues"]},
)
def test_duplicate_frontmatter_names_are_reported(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", name="shared")
self.make_skill(root, "beta", name="shared")
report = audit_skills(root)
duplicates = [
issue for issue in report["issues"] if issue["code"] == "duplicate_name"
]
self.assertFalse(report["ok"])
self.assertEqual(len(duplicates), 1)
self.assertEqual(duplicates[0]["name"], "shared")
self.assertEqual(len(duplicates[0]["paths"]), 2)
def test_clean_manifest_can_be_selected_explicitly(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha")
manifest_path = self.write_manifest(
root,
path=root / "clean-manifest.json",
)
output = io.StringIO()
with redirect_stdout(output):
return_code = main(
[
"--root",
str(root),
"--manifest",
str(manifest_path),
"--json",
"--quiet",
]
)
payload = json.loads(output.getvalue())
self.assertEqual(return_code, 0)
self.assertTrue(payload["ok"])
self.assertTrue(payload["manifest"]["loaded"])
self.assertEqual(payload["manifest"]["path"], "clean-manifest.json")
def test_error_manifest_fails_with_structured_issues(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha")
self.make_skill(root, "beta")
broken_manifest = self.write_manifest(
root,
entries=[
{
"name": "alpha",
"lifecycle": "platform",
"compounding": "none",
"invocation": "model_routed",
"side_effects": ["none"],
"contract_owner": "",
"collaborates_with": ["上下文与知识检索", 3],
"skill_path": ".agent/skills/beta/SKILL.md",
},
{
"name": "alpha",
"lifecycle": "platform",
"compounding": "none",
"invocation": "model_routed",
"side_effects": ["none"],
"contract_owner": "测试夹具",
"collaborates_with": [],
"skill_path": ".agent/skills/beta/SKILL.md",
},
{
"name": "extra",
"lifecycle": "platform",
"compounding": "none",
"invocation": "model_routed",
"side_effects": ["none"],
"contract_owner": "测试夹具",
"collaborates_with": [],
"skill_path": ".agent/skills/extra/SKILL.md",
},
],
path=root / "broken-manifest.json",
)
output = io.StringIO()
with redirect_stdout(output):
return_code = main(
[
"--root",
str(root),
"--manifest",
str(broken_manifest),
"--json",
]
)
payload = json.loads(output.getvalue())
codes = {issue["code"] for issue in payload["issues"]}
self.assertEqual(return_code, 1)
self.assertFalse(payload["ok"])
self.assertTrue(all(isinstance(issue, dict) for issue in payload["issues"]))
self.assertTrue(
{
"manifest_contract_owner_invalid",
"manifest_collaborates_with_invalid",
"manifest_duplicate_skill_path",
"manifest_duplicate_name",
"manifest_skill_missing",
"manifest_skill_extra",
"manifest_skill_path_mismatch",
"manifest_name_directory_mismatch",
"manifest_skill_path_missing",
}.issubset(codes)
)
def test_missing_manifest_fails_closed(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha")
(root / "harness" / "manifests" / "skills.json").unlink()
report = audit_skills(root)
self.assertFalse(report["ok"])
self.assertIn(
"manifest_missing",
{issue["code"] for issue in report["issues"]},
)
def test_missing_skills_directory_fails_closed(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
report = audit_skills(root)
self.assertFalse(report["ok"])
self.assertIn(
"skills_directory_missing",
{issue["code"] for issue in report["issues"]},
)
def test_skill_directory_without_skill_file_fails(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
(root / ".agent" / "skills" / "missing").mkdir(parents=True)
report = audit_skills(root)
self.assertFalse(report["ok"])
self.assertIn(
"missing_skill_file",
{issue["code"] for issue in report["issues"]},
)
def test_empty_skill_file_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
skill_path = self.make_skill(root, "empty")
skill_path.write_text("", encoding="utf-8")
report = audit_skills(root)
self.assertFalse(report["ok"])
self.assertIn(
"empty_skill_file",
{issue["code"] for issue in report["issues"]},
)
def test_classification_fields_are_validated(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha")
self.write_manifest(
root,
entries=[
{
"name": "alpha",
"lifecycle": "不存在的周期段",
"invocation": "auto",
"side_effects": ["none", "db_write"],
"compounding": "someday",
"collaborates_with": [],
"skill_path": ".agent/skills/alpha/SKILL.md",
}
],
)
report = audit_skills(root)
codes = {issue["code"] for issue in report["issues"]}
self.assertFalse(report["ok"])
self.assertTrue(
{
"manifest_lifecycle_invalid",
"manifest_invocation_invalid",
"manifest_side_effects_invalid",
"manifest_compounding_invalid",
}.issubset(codes),
codes,
)
def test_orchestration_claimed_in_manifest_must_exist_in_frontmatter(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", orchestrated=False)
self.write_manifest(
root,
entries=[
{
"name": "alpha",
"lifecycle": "platform",
"compounding": "none",
"invocation": "orchestrated",
"side_effects": ["none"],
"contract_owner": "测试夹具",
"collaborates_with": [],
"skill_path": ".agent/skills/alpha/SKILL.md",
}
],
)
report = audit_skills(root)
self.assertFalse(report["ok"])
self.assertIn(
"invocation_frontmatter_mismatch",
{issue["code"] for issue in report["issues"]},
)
def test_quality_advisories_only_block_under_strict(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
# 正文缺必备节、description 超预算,属严重级质量发现而非阻断问题。
self.make_skill(
root,
"alpha",
body="# 合同\n\n只描述运行时行为。\n",
description="离" * (DESCRIPTION_BUDGET + 1),
)
report = audit_skills(root)
advisory_codes = {item["code"] for item in report["advisories"]}
self.assertTrue(report["ok"])
self.assertEqual(report["issues"], [])
self.assertTrue(
{"required_section_missing", "description_budget_exceeded"}.issubset(
advisory_codes
),
advisory_codes,
)
self.assertTrue(
all(item["severity"] in {"major", "minor"} for item in report["advisories"])
)
self.assertFalse(audit_skills(root, strict=True)["ok"])
def test_compounding_gap_is_reported_except_for_platform(self) -> None:
body = "## 输入\n\n入参。\n\n## 输出\n\n产物。\n\n## 红线\n\n禁止。\n"
for lifecycle, expected in (("writing", True), ("platform", False)):
with self.subTest(lifecycle=lifecycle):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", body=body)
self.write_manifest(
root,
[
{
"name": "alpha",
"lifecycle": lifecycle,
"invocation": "model_routed",
"side_effects": ["none"],
"compounding": "none",
"contract_owner": "测试夹具",
"collaborates_with": [],
"skill_path": ".agent/skills/alpha/SKILL.md",
}
],
)
report = audit_skills(root)
codes = {item["code"] for item in report["advisories"]}
self.assertTrue(report["ok"])
self.assertEqual("compounding_not_wired" in codes, expected, codes)
def test_compounding_wired_requires_contract_section(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"alpha",
body="## 输入\n\n入参。\n\n## 输出\n\n产物。\n\n## 红线\n\n禁止。\n",
)
self.write_manifest(
root,
[
{
"name": "alpha",
"lifecycle": "writing",
"invocation": "model_routed",
"side_effects": ["none"],
"compounding": "closed_loop",
"contract_owner": "测试夹具",
"collaborates_with": [],
"skill_path": ".agent/skills/alpha/SKILL.md",
}
],
)
report = audit_skills(root)
missing = {
item.get("section")
for item in report["advisories"]
if item["code"] == "required_section_missing"
}
self.assertIn("复利合同", missing)
def test_scripts_directory_without_implementation_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"alpha",
body="## 输入\n\n入参。\n\n## 输出\n\n产物。\n\n## 红线\n\n禁止。\n",
)
scripts = root / ".agent" / "skills" / "alpha" / "scripts"
scripts.mkdir()
(scripts / "test_alpha.py").write_text("# 夹具\n", encoding="utf-8")
report = audit_skills(root)
advisory_codes = {item["code"] for item in report["advisories"]}
self.assertTrue(report["ok"])
self.assertTrue(
{"scripts_test_file", "scripts_without_implementation"}.issubset(
advisory_codes
),
advisory_codes,
)
def test_dangling_internal_codename_is_reported_unless_target_exists(self) -> None:
for create_target, expected in ((False, True), (True, False)):
with self.subTest(create_target=create_target):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", body="旧路由见 MERGE_PLAN §4。\n")
if create_target:
(root / "MERGE_PLAN.md").write_text("# 计划\n", encoding="utf-8")
report = audit_skills(root)
codes = {item["code"] for item in report["issues"]}
self.assertEqual("dangling_internal_codename" in codes, expected, codes)
self.assertEqual(report["ok"], not expected)
def test_reference_target_missing_ignores_external_links(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"alpha",
body=(
"见 [在的](references/present.md)。\n"
"见 [不在的](references/absent.md)。\n"
"见 [外链](https://example.com/x.md)。\n"
"见 [锚点](#红线)。\n"
),
)
references = root / ".agent" / "skills" / "alpha" / "references"
references.mkdir()
(references / "present.md").write_text("# 在\n", encoding="utf-8")
report = audit_skills(root)
missing = [
item
for item in report["issues"]
if item["code"] == "reference_target_missing"
]
self.assertFalse(report["ok"])
self.assertEqual(len(missing), 1)
self.assertEqual(missing[0]["target"], "references/absent.md")
def test_orphan_reference_file_exempts_underscore_bookkeeping(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", body="见 cited.md。\n")
references = root / ".agent" / "skills" / "alpha" / "references"
references.mkdir()
(references / "cited.md").write_text("# 被引用\n", encoding="utf-8")
(references / "orphan.md").write_text("# 没人引用\n", encoding="utf-8")
(references / "_coverage.md").write_text("# 记账\n", encoding="utf-8")
report = audit_skills(root)
orphans = {
item["path"]
for item in report["issues"]
if item["code"] == "orphan_reference_file"
}
self.assertEqual(orphans, {".agent/skills/alpha/references/orphan.md"})
def test_trigger_collision_requires_same_lifecycle_and_no_handoff(self) -> None:
cases = (
("writing", "写作方法,激活场景:太抽象。", True),
("planning", "写作方法,激活场景:太抽象。", False),
)
for peer_lifecycle, peer_description, expected in cases:
with self.subTest(peer_lifecycle=peer_lifecycle):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root, "alpha", description="写作方法,激活场景:太抽象。"
)
self.make_skill(root, "beta", description=peer_description)
self.write_manifest(
root,
[
self.manifest_entry("alpha", lifecycle="writing"),
self.manifest_entry("beta", lifecycle=peer_lifecycle),
],
)
report = audit_skills(root)
codes = {item["code"] for item in report["advisories"]}
self.assertEqual(
"trigger_collision_without_handoff" in codes, expected, codes
)
def test_trigger_collision_is_cleared_by_naming_the_peer(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"alpha",
description="写作方法,激活场景:太抽象。不适用于:句法节奏(见 beta)。",
)
self.make_skill(root, "beta", description="写作方法,激活场景:太抽象。")
self.write_manifest(
root,
[
self.manifest_entry("alpha", lifecycle="writing"),
self.manifest_entry("beta", lifecycle="writing"),
],
)
report = audit_skills(root)
codes = {item["code"] for item in report["advisories"]}
self.assertNotIn("trigger_collision_without_handoff", codes)
def test_description_names_unknown_skill_exempts_own_package_documents(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(
root,
"alpha",
description="见 boundaries-and-pitfalls 与 set-aside-the-manuscript。",
)
references = root / ".agent" / "skills" / "alpha" / "references"
references.mkdir()
(references / "boundaries-and-pitfalls.md").write_text("# 边界\n", encoding="utf-8")
report = audit_skills(root)
tokens = {
item["token"]
for item in report["issues"]
if item["code"] == "description_names_unknown_skill"
}
self.assertEqual(tokens, {"set-aside-the-manuscript"})
def test_shared_exemplar_heading_only_fires_across_skills(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", body="见 shared.md 与 solo.md。\n")
self.make_skill(root, "beta", body="见 shared.md。\n")
for directory, filename, heading in (
("alpha", "shared.md", "### 案例 1:蒲松龄《促织》成名"),
("alpha", "solo.md", "### 案例 2:只有这里有的范本"),
("beta", "shared.md", "### 案例 3:蒲松龄《促织》成名"),
):
references = root / ".agent" / "skills" / directory / "references"
references.mkdir(exist_ok=True)
(references / filename).write_text(f"{heading}\n\n正文。\n", encoding="utf-8")
report = audit_skills(root)
exemplars = {
item["exemplar"]
for item in report["advisories"]
if item["code"] == "shared_exemplar_heading"
}
self.assertEqual(exemplars, {"蒲松龄《促织》成名"})
def test_oversized_reference_file_is_reported(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha", body="见 huge.md。\n")
references = root / ".agent" / "skills" / "alpha" / "references"
references.mkdir()
(references / "huge.md").write_text(
"字" * (REFERENCE_FILE_BUDGET_BYTES), encoding="utf-8"
)
report = audit_skills(root)
codes = {item["code"] for item in report["advisories"]}
self.assertIn("reference_file_too_large", codes)
def test_json_cli_and_exit_code(self) -> None:
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.make_skill(root, "alpha")
output = io.StringIO()
with redirect_stdout(output):
return_code = main(["--root", str(root), "--json", "--quiet"])
payload = json.loads(output.getvalue())
self.assertEqual(return_code, 0)
self.assertTrue(payload["ok"])
self.assertEqual(payload["skills_scanned"], 1)
(root / ".agent" / "skills" / "alpha" / "SKILL.md").write_text(
"---\nname: alpha\n---\n## 测试\n",
encoding="utf-8",
)
output = io.StringIO()
with redirect_stdout(output):
return_code = main(["--root", str(root), "--json"])
failed_payload = json.loads(output.getvalue())
self.assertEqual(return_code, 1)
self.assertFalse(failed_payload["ok"])
if __name__ == "__main__":
unittest.main(verbosity=2)