dca4d45d4e
把 21 个内置 agent 的 system_prompt 从 specs.py 的 Python 常量外置为 prompts/<spec.name>.md，import 期由 load_prompt 确定性加载；建立 SPECS 名册 + SCHEMA_CATALOG（Pydantic 类型留 Python）+ 统一只读解析入口 SpecResolver。 纯重构、零功能/schema 变更，缓存断点前块字节级不变（不变量 #9）。 @llm packages/agents（步骤1-3） - spec_model.py：抽出 AgentSpec（frozen，字段不变） - prompt_loader.py：load_prompt = utf-8-sig 去BOM → LF 归一 → NFC → rstrip尾LF， 内存缓存 + fail-fast（PromptNotFoundError），import 期确定性 - schema_catalog.py：SCHEMA_CATALOG[name]→output type 唯一真相源（refiner=None） - prompts/*.md ×21：取常量「运行时值」程序化外迁（反斜杠折行已塌缩， 物理换行≡运行时换行）；文件名按 spec.name 连字符（style.md/character-gen.md 等） - specs.py：删 21 常量 + AgentSpec 类；system_prompt=load_prompt(name)、 output_schema=SCHEMA_CATALOG[name]；建 SPECS + REVIEW_RESERVED_NAMES； *_spec 兼容期保留且 SPECS[name] is *_spec（同一实例）。804→337 行 - __init__.py：显式 __all__ 重导出（避 F401） @backend packages/skills（步骤4-5） - SpecResolver：内置 SPECS（纯内存、零 DB）+ 用户 SkillRegistry 统一 get； 内置 name 永不触发 DB；output_schema_for 精确匹配 - skill_registry：保留命名空间守卫前移至入库校验，拒同名内置 → VALIDATION - toolbox_registry：GeneratorTool.spec 改走 SPECS[...]，删 12 个 *_spec 直接 import @devops repo-root - .gitattributes：prompts/*.md text eol=lf（修正：须用完整嵌套路径才匹配） - packages/agents/pyproject：hatchling artifacts 纳入 prompts/*.md 随 wheel/sdist 分发 - ci.yml：新增 build wheel → 裸装 → import ww_agents.SPECS 冒烟 TDD 全程 mock 网关；门禁绿：ruff/format clean · mypy 209 files · pytest 744 passed （含金标准 sha256 回归 / md↔spec↔catalog 一一对应 / fail-fast / BOM+NFC / 内置守卫 / 同一实例 / 编排器无回归 / apps/api import-smoke / 打包冒烟）
131 lines
4.1 KiB
Python
131 lines
4.1 KiB
Python
"""T5.5 表权限沙箱单测（ARCH §5.6：声明式权限 + apply 层白名单，非进程沙箱）。
|
||
|
||
纯函数、无 IO：
|
||
- `filter_reads`：注入时只把 skill 声明 `reads` 的表数据喂给它（越权读 → 丢弃，不喂）。
|
||
- `partition_writes`：写库时只应用 skill 声明 `writes` 的产出字段（越权写 → 丢弃 + 审计），
|
||
返回 `(allowed, rejected)` 两份，调用方据此落库 allowed、log rejected。
|
||
- `validate_declaration`：加载/注册 skill 时校验其声明的 `reads/writes` 全在已知表白名单内
|
||
（杜绝声明指向不存在的表）；越界 → AppError(VALIDATION)。
|
||
不变量 #3：自定义 skill 产出仍过验收 gate——本层只做白名单裁剪，不开写后门。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from ww_agents import AgentSpec
|
||
from ww_shared import AppError, ErrorCode
|
||
from ww_skills import (
|
||
KNOWN_TABLES,
|
||
filter_reads,
|
||
partition_writes,
|
||
validate_declaration,
|
||
)
|
||
|
||
|
||
def _spec(*, reads: list[str], writes: list[str], scope: str = "custom") -> AgentSpec:
|
||
return AgentSpec(
|
||
name="custom_skill",
|
||
tier="writer",
|
||
system_prompt="x",
|
||
reads=tuple(reads),
|
||
writes=tuple(writes),
|
||
scope=scope,
|
||
)
|
||
|
||
|
||
# ---- filter_reads ----
|
||
|
||
|
||
def test_filter_reads_keeps_only_declared_tables() -> None:
|
||
spec = _spec(reads=["world_entities"], writes=[])
|
||
available = {
|
||
"world_entities": [{"name": "灵根"}],
|
||
"characters": [{"name": "主角"}], # 未声明 → 必须丢弃
|
||
}
|
||
|
||
fed = filter_reads(spec, available)
|
||
|
||
assert fed == {"world_entities": [{"name": "灵根"}]}
|
||
|
||
|
||
def test_filter_reads_ignores_declared_table_absent_from_available() -> None:
|
||
spec = _spec(reads=["world_entities", "characters"], writes=[])
|
||
available = {"world_entities": [{"name": "灵根"}]}
|
||
|
||
fed = filter_reads(spec, available)
|
||
|
||
assert fed == {"world_entities": [{"name": "灵根"}]}
|
||
|
||
|
||
def test_filter_reads_empty_declaration_feeds_nothing() -> None:
|
||
spec = _spec(reads=[], writes=[])
|
||
available = {"world_entities": [{"name": "灵根"}]}
|
||
|
||
assert filter_reads(spec, available) == {}
|
||
|
||
|
||
# ---- partition_writes ----
|
||
|
||
|
||
def test_partition_writes_drops_over_permission_fields() -> None:
|
||
spec = _spec(reads=[], writes=["world_entities"])
|
||
produced = {
|
||
"world_entities": [{"name": "新势力"}],
|
||
"characters": [{"name": "越权角色"}], # 越权 → 丢弃 + 审计
|
||
}
|
||
|
||
allowed, rejected = partition_writes(spec, produced)
|
||
|
||
assert allowed == {"world_entities": [{"name": "新势力"}]}
|
||
assert rejected == ["characters"]
|
||
|
||
|
||
def test_partition_writes_all_allowed_yields_empty_rejected() -> None:
|
||
spec = _spec(reads=[], writes=["world_entities", "characters"])
|
||
produced: dict[str, object] = {"world_entities": [], "characters": []}
|
||
|
||
allowed, rejected = partition_writes(spec, produced)
|
||
|
||
assert set(allowed) == {"world_entities", "characters"}
|
||
assert rejected == []
|
||
|
||
|
||
def test_partition_writes_no_declared_writes_rejects_all() -> None:
|
||
spec = _spec(reads=[], writes=[])
|
||
produced = {"world_entities": [{"name": "x"}]}
|
||
|
||
allowed, rejected = partition_writes(spec, produced)
|
||
|
||
assert allowed == {}
|
||
assert rejected == ["world_entities"]
|
||
|
||
|
||
# ---- validate_declaration ----
|
||
|
||
|
||
def test_validate_declaration_passes_for_known_tables() -> None:
|
||
spec = _spec(reads=["world_entities"], writes=["world_entities"])
|
||
# 不抛即通过
|
||
validate_declaration(spec)
|
||
|
||
|
||
def test_validate_declaration_rejects_unknown_read_table() -> None:
|
||
spec = _spec(reads=["secret_table"], writes=[])
|
||
|
||
with pytest.raises(AppError) as exc:
|
||
validate_declaration(spec)
|
||
|
||
assert exc.value.code is ErrorCode.VALIDATION
|
||
assert "secret_table" in str(exc.value.details)
|
||
|
||
|
||
def test_validate_declaration_rejects_unknown_write_table() -> None:
|
||
# 系统表 users 不在创作表白名单 → 越权写声明被拒。
|
||
assert "users" not in KNOWN_TABLES
|
||
spec = _spec(reads=[], writes=["users"])
|
||
|
||
with pytest.raises(AppError) as exc:
|
||
validate_declaration(spec)
|
||
|
||
assert exc.value.code is ErrorCode.VALIDATION
|