197 lines
10 KiB
Python
197 lines
10 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Agent 工具注册表（M4a）—— 22 工具 = 13 启用 + 9 禁用，fail-closed。
|
||
|
||
对齐 docs/01-design/agent-tool-contract.md：
|
||
* 未注册工具 → 一律 DENY（PBL-TOOL-0001）
|
||
* 注册但 enabled=False → 一律 DENY（PBL-TOOL-0002）
|
||
* 写保护域（rbac/world/scene/entity/scense/scense_runtime/script_engine 基表）
|
||
出现 enabled 的写工具 → self_check 直接失败（启动即拒）
|
||
|
||
工具契约字段：
|
||
name / kind(read|write) / domain / enabled / role(最低角色) /
|
||
required(必填参数) / schema(参数类型与枚举) / handler(点分路径) / desc
|
||
"""
|
||
|
||
WRITE_PROTECTED_DOMAINS = (
|
||
"rbac", "world", "scene", "entity",
|
||
"scense", "scense_runtime", "script_engine",
|
||
)
|
||
|
||
POLICY_VERSION = "v1"
|
||
|
||
# ---- 13 个启用工具（Agent 可调用）
|
||
ENABLED_TOOLS = [
|
||
{"name": "blueprint.get", "kind": "read", "domain": "pbl_blueprint", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "blueprint_id"],
|
||
"schema": {"tenant_id": "int", "blueprint_id": "int"},
|
||
"handler": "pbl_blueprint.api:get_blueprint",
|
||
"desc": "读取蓝图聚合根（租户隔离）"},
|
||
{"name": "blueprint.tree", "kind": "read", "domain": "pbl_blueprint", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "blueprint_id"],
|
||
"schema": {"tenant_id": "int", "blueprint_id": "int", "depth": "int"},
|
||
"handler": "pbl_blueprint.api:get_tree",
|
||
"desc": "读取蓝图子对象树"},
|
||
{"name": "blueprint.patch", "kind": "write", "domain": "pbl_blueprint", "enabled": True,
|
||
"role": "teacher.pbls", "required": ["tenant_id", "blueprint_id", "obj_type", "payload"],
|
||
"schema": {"tenant_id": "int", "blueprint_id": "int", "payload": "dict",
|
||
"obj_type": {"enum": ["goal", "role", "task", "artifact", "rubric",
|
||
"resource", "flow"]}},
|
||
"handler": "pbl_blueprint.api:patch_subobject",
|
||
"desc": "泛化写入 7 类子对象"},
|
||
{"name": "validation.run", "kind": "read", "domain": "pbl_validation", "enabled": True,
|
||
"role": "teacher.pbls", "required": ["tenant_id", "blueprint_id"],
|
||
"schema": {"tenant_id": "int", "blueprint_id": "int", "ruleset_code": "str"},
|
||
"handler": "pbl_validation.api:run_validation",
|
||
"desc": "14 维校验 + 5 级质量状态"},
|
||
{"name": "validation.findings", "kind": "read", "domain": "pbl_validation", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "run_id"],
|
||
"schema": {"tenant_id": "int", "run_id": "int", "severity": {"enum": ["info", "warn", "error", "block"]}},
|
||
"handler": "pbl_validation.api:list_findings",
|
||
"desc": "读取校验发现项"},
|
||
{"name": "compiler.compile", "kind": "write", "domain": "pbl_compiler", "enabled": True,
|
||
"role": "teacher.pbls", "required": ["tenant_id", "blueprint_id"],
|
||
"schema": {"tenant_id": "int", "blueprint_id": "int", "target": {"enum": ["game_definition", "runtime_bundle"]}},
|
||
"handler": "pbl_compiler.api:compile_blueprint",
|
||
"desc": "确定性编译（同输入同输出）"},
|
||
{"name": "compiler.artifact", "kind": "read", "domain": "pbl_compiler", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "job_id"],
|
||
"schema": {"tenant_id": "int", "job_id": "int"},
|
||
"handler": "pbl_compiler.api:get_artifact",
|
||
"desc": "读取编译产物"},
|
||
{"name": "evidence.collect", "kind": "write", "domain": "pbl_evidence", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "run_id", "idem_key", "payload"],
|
||
"schema": {"tenant_id": "int", "run_id": "int", "idem_key": "str", "payload": "dict"},
|
||
"handler": "pbl_evidence.api:collect_evidence",
|
||
"desc": "幂等采集产出物证据"},
|
||
{"name": "evidence.list", "kind": "read", "domain": "pbl_evidence", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "run_id"],
|
||
"schema": {"tenant_id": "int", "run_id": "int", "kind": "str"},
|
||
"handler": "pbl_evidence.api:list_evidence",
|
||
"desc": "列出证据"},
|
||
{"name": "assessment.score", "kind": "write", "domain": "pbl_assessment", "enabled": True,
|
||
"role": "teacher.pbls", "required": ["tenant_id", "blueprint_id", "team_id"],
|
||
"schema": {"tenant_id": "int", "blueprint_id": "int", "team_id": "int"},
|
||
"handler": "pbl_assessment.api:score_rubric",
|
||
"desc": "Rubric 加权评估"},
|
||
{"name": "assessment.report", "kind": "read", "domain": "pbl_assessment", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "report_id"],
|
||
"schema": {"tenant_id": "int", "report_id": "int"},
|
||
"handler": "pbl_assessment.api:get_report",
|
||
"desc": "读取评估报告"},
|
||
{"name": "runtime.event_append", "kind": "write", "domain": "pbl_runtime_ext", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "world_id", "event_type", "payload"],
|
||
"schema": {"tenant_id": "int", "world_id": "int", "event_type": "str", "payload": "dict"},
|
||
"handler": "pbl_runtime_ext.api:append_event",
|
||
"desc": "单事务事件+状态写入（薄扩展，不改基表）"},
|
||
{"name": "kdb.query", "kind": "read", "domain": "pbl_kdb_ext", "enabled": True,
|
||
"role": "student.pbls", "required": ["tenant_id", "query_code"],
|
||
"schema": {"tenant_id": "int", "query_code": "str", "anon": "bool"},
|
||
"handler": "pbl_kdb_ext.api:kdb_query",
|
||
"desc": "KDB 只读桩 + 匿名聚合（零写入）"},
|
||
]
|
||
|
||
# ---- 9 个禁用工具（注册但 enabled=False，调用一律 DENY）
|
||
DISABLED_TOOLS = [
|
||
{"name": "rbac.grant_role", "kind": "write", "domain": "rbac", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "user_id", "role_code"],
|
||
"schema": {"tenant_id": "int", "user_id": "int", "role_code": "str"},
|
||
"handler": "", "desc": "写保护域：禁止 Agent 授权（Phase 0/1 禁用）"},
|
||
{"name": "rbac.revoke_role", "kind": "write", "domain": "rbac", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "user_id", "role_code"],
|
||
"schema": {"tenant_id": "int", "user_id": "int", "role_code": "str"},
|
||
"handler": "", "desc": "写保护域：禁止 Agent 撤权"},
|
||
{"name": "world.delete", "kind": "write", "domain": "world", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "world_id"],
|
||
"schema": {"tenant_id": "int", "world_id": "int"},
|
||
"handler": "", "desc": "写保护域：禁止删除世界基表"},
|
||
{"name": "scene.delete", "kind": "write", "domain": "scene", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "scene_id"],
|
||
"schema": {"tenant_id": "int", "scene_id": "int"},
|
||
"handler": "", "desc": "写保护域：禁止删除场景基表"},
|
||
{"name": "entity.delete", "kind": "write", "domain": "entity", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "entity_id"],
|
||
"schema": {"tenant_id": "int", "entity_id": "int"},
|
||
"handler": "", "desc": "写保护域：禁止删除实体基表"},
|
||
{"name": "script_engine.execute_raw", "kind": "write", "domain": "script_engine", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "script_text"],
|
||
"schema": {"tenant_id": "int", "script_text": "str"},
|
||
"handler": "", "desc": "写保护域：禁止执行任意脚本"},
|
||
{"name": "scense.reset_session", "kind": "write", "domain": "scense", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "session_id"],
|
||
"schema": {"tenant_id": "int", "session_id": "int"},
|
||
"handler": "", "desc": "写保护域：禁止重置游戏会话"},
|
||
{"name": "db.ddl_exec", "kind": "write", "domain": "platform", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id", "ddl"],
|
||
"schema": {"tenant_id": "int", "ddl": "str"},
|
||
"handler": "", "desc": "禁止 Agent 执行 DDL"},
|
||
{"name": "tenant.switch", "kind": "write", "domain": "platform", "enabled": False,
|
||
"role": "owner.pbls", "required": ["tenant_id"],
|
||
"schema": {"tenant_id": "int"},
|
||
"handler": "", "desc": "禁止 Agent 切换租户上下文（防跨租户）"},
|
||
]
|
||
|
||
EXPECTED_ENABLED = 13
|
||
EXPECTED_DISABLED = 9
|
||
|
||
|
||
def all_tools():
|
||
"""返回全部注册工具（启用 + 禁用），22 项。"""
|
||
return list(ENABLED_TOOLS) + list(DISABLED_TOOLS)
|
||
|
||
|
||
def registry():
|
||
"""返回 name → tool 字典。"""
|
||
return {t["name"]: t for t in all_tools()}
|
||
|
||
|
||
def get_tool(name):
|
||
"""按名取工具契约；未注册返回 None（调用方必须 fail-closed DENY）。"""
|
||
return registry().get(name)
|
||
|
||
|
||
def is_registered(name):
|
||
return name in registry()
|
||
|
||
|
||
def is_enabled(name):
|
||
t = get_tool(name)
|
||
return bool(t and t.get("enabled"))
|
||
|
||
|
||
def self_check():
|
||
"""数量契约 + 写保护域校验。返回 (all_ok, msgs)。
|
||
|
||
硬约束：
|
||
* len(ENABLED_TOOLS) == 13 且 len(DISABLED_TOOLS) == 9（总 22）
|
||
* 工具名全局唯一
|
||
* 写保护域内不得存在 enabled=True 的 write 工具
|
||
* 每个 enabled 工具必须有非空 handler
|
||
"""
|
||
msgs = []
|
||
all_ok = True
|
||
if len(ENABLED_TOOLS) != EXPECTED_ENABLED:
|
||
all_ok = False
|
||
msgs.append("启用工具数=%d 应为 %d" % (len(ENABLED_TOOLS), EXPECTED_ENABLED))
|
||
if len(DISABLED_TOOLS) != EXPECTED_DISABLED:
|
||
all_ok = False
|
||
msgs.append("禁用工具数=%d 应为 %d" % (len(DISABLED_TOOLS), EXPECTED_DISABLED))
|
||
names = [t["name"] for t in all_tools()]
|
||
dup = sorted(set([n for n in names if names.count(n) > 1]))
|
||
if dup:
|
||
all_ok = False
|
||
msgs.append("工具名重复：%s" % dup)
|
||
for t in all_tools():
|
||
if t.get("domain") in WRITE_PROTECTED_DOMAINS and t.get("enabled") and t.get("kind") == "write":
|
||
all_ok = False
|
||
msgs.append("写保护域 %s 存在启用写工具 %s" % (t["domain"], t["name"]))
|
||
if t.get("enabled") and not t.get("handler"):
|
||
all_ok = False
|
||
msgs.append("启用工具 %s 缺 handler" % t["name"])
|
||
if not t.get("required") or "tenant_id" not in t["required"]:
|
||
all_ok = False
|
||
msgs.append("工具 %s required 未含 tenant_id" % t["name"])
|
||
if all_ok:
|
||
msgs.append("SELF_CHECK pbl_agent_runtime.tool_registry: PASS %d/%d (enabled=%d disabled=%d)"
|
||
% (len(names), len(names), len(ENABLED_TOOLS), len(DISABLED_TOOLS)))
|
||
return all_ok, msgs
|