448 lines
24 KiB
Python
448 lines
24 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
M4a 自测脚本（离线可跑，MemoryStore）—— 覆盖本任务四个验收锚点：
|
||
A. Designer/Critic 定义（Critic 零写权限）
|
||
B. 13 启用 / 9 禁用工具清单注册（含 pbl.publish 类自主发布禁用）
|
||
C. fail-closed 8 步裁决顺序（default-deny，逐步拒绝码正确）
|
||
D. 四类强制人工审批（publish / compile_execute / blueprint_approve /
|
||
tool_registry_change）
|
||
|
||
运行：python3 scripts/m4a_selftest.py
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from pbl_agent_runtime import m4a as M # noqa: E402
|
||
from pbl_agent_runtime import m4a_critic # noqa: E402
|
||
from pbl_agent_runtime.m4a_kernel import ( # noqa: E402
|
||
MemoryStore, PblError, ADJUDICATION_STEPS, TRACE_ELEMENTS,
|
||
E_FORBIDDEN, E_STATE_ILLEGAL, E_TENANT_MISSING, E_VALIDATION,
|
||
E_APPEND_ONLY, E_DUPLICATE,
|
||
)
|
||
|
||
PASS, FAIL = [], []
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
(PASS if cond else FAIL).append(name)
|
||
print("%s %s%s" % ("[PASS]" if cond else "[FAIL]", name,
|
||
(" -> " + str(detail)) if detail and not cond else ""))
|
||
|
||
|
||
def expect_error(name, code, fn):
|
||
try:
|
||
fn()
|
||
except PblError as e:
|
||
check(name, e.code == code, "期望 %s 实得 %s(%s)" % (code, e.code, e.message))
|
||
return e
|
||
except Exception as e:
|
||
check(name, False, "非 PblError: %r" % e)
|
||
return None
|
||
check(name, False, "未抛错（期望 %s）" % code)
|
||
return None
|
||
|
||
|
||
def main():
|
||
store = M.set_store(MemoryStore())
|
||
api = M.load_m4a(tenant_id="T001", store=store)
|
||
sc = api.self_check()
|
||
|
||
# ---------------- A. Agent 定义 ----------------
|
||
check("A1 仅 designer/critic 两个 Agent", sc["agent_count"] == 2,
|
||
sc["agents"])
|
||
codes = sorted(a["agent_code"] for a in sc["agents"])
|
||
check("A2 agent_code = [critic, designer]", codes == ["critic", "designer"], codes)
|
||
critic = [a for a in sc["agents"] if a["agent_code"] == "critic"][0]
|
||
designer = [a for a in sc["agents"] if a["agent_code"] == "designer"][0]
|
||
check("A3 Critic write_allowed=0（零写权限）", critic["write_allowed"] == 0,
|
||
critic)
|
||
check("A4 Designer write_allowed=1", designer["write_allowed"] == 1)
|
||
check("A5 Critic 白名单仅只读工具",
|
||
set(critic["allowed_tools"]) == set(M.CRITIC_READONLY_TOOLS),
|
||
critic["allowed_tools"])
|
||
check("A6 sc.critic_write_allowed 为空", sc["critic_write_allowed"] == [])
|
||
check("A7 registry 无 critic 可写工具",
|
||
sc["tools"]["critic_writable_tools"] == [],
|
||
sc["tools"]["critic_writable_tools"])
|
||
|
||
# ---------------- B. 工具注册表 13+9 ----------------
|
||
check("B1 工具总数 22", sc["tools"]["total"] == 22, sc["tools"]["total"])
|
||
check("B2 启用 13", sc["tools"]["enabled"] == 13, sc["tools"]["enabled"])
|
||
check("B3 禁用 9", sc["tools"]["disabled"] == 9, sc["tools"]["disabled"])
|
||
check("B4 禁用项 disable_reason 齐备",
|
||
sc["tools"]["disabled_missing_reason"] == [],
|
||
sc["tools"]["disabled_missing_reason"])
|
||
check("B5 require_approval 工具 = compile.trigger + publish.request",
|
||
sorted(sc["tools"]["approval_required"]) ==
|
||
["compile.trigger", "publish.request"],
|
||
sc["tools"]["approval_required"])
|
||
check("B6 自主发布工具 blueprint.publish_auto 已注册且禁用",
|
||
any(t["tool_code"] == "blueprint.publish_auto" and
|
||
t["status"] == "disabled" for t in api.list_tools()))
|
||
dis = {t["tool_code"]: t for t in api.list_tools(status="disabled")}
|
||
expect9 = ["blueprint.publish_auto", "curriculum.modify_auto",
|
||
"marketplace.create_listing", "kdb.write", "research.collect",
|
||
"experiment.ab_run", "agent.mentor_invoke", "world.edit_3d",
|
||
"billing.charge"]
|
||
check("B7 9 个禁用工具编码完全匹配", sorted(dis) == sorted(expect9),
|
||
sorted(dis))
|
||
check("B8 kdb.write 禁用（Q5 KDB 只读）",
|
||
"KDB 只读" in (dis["kdb.write"]["disable_reason"] or ""))
|
||
check("B9 research.collect 禁用（Q6 禁采集学生数据）",
|
||
"research" in (dis["research.collect"]["disable_reason"] or "").lower())
|
||
en = {t["tool_code"] for t in api.list_tools(status="enabled")}
|
||
check("B10 13 启用工具编码匹配", en == set(M.ENABLED_TOOL_CODES) and len(en) == 13,
|
||
sorted(en))
|
||
# seed 幂等
|
||
r1 = M.seed_tools(ctx=api.admin_ctx(), store=store)
|
||
check("B11 seed_tools 幂等（第二次 inserted=0/skipped=22）",
|
||
r1["inserted"] == 0 and r1["skipped"] == 22, r1)
|
||
|
||
# ---------------- C. fail-closed 8 步裁决 ----------------
|
||
check("C1 裁决链 8 步", len(ADJUDICATION_STEPS) == 8,
|
||
[s[0] for s in ADJUDICATION_STEPS])
|
||
check("C2 裁决顺序 = S1..S8",
|
||
[s[0] for s in ADJUDICATION_STEPS] ==
|
||
["S1", "S2", "S3", "S4", "S5", "S6", "S7", "S8"],
|
||
M.explain_chain())
|
||
|
||
dctx = api.agent_ctx("designer")
|
||
# S1 租户缺失
|
||
noctx = M.TenantContext(tenant_id=None, actor_type="agent", actor_id="designer",
|
||
permissions={M.PERM_AUTHORING})
|
||
v = M.adjudicate(noctx, "designer", "blueprint.get", {"blueprint_id": 1},
|
||
store=store)
|
||
check("C3 S1 租户缺失 → PBL_E_TENANT_MISSING",
|
||
not v.allowed and v.step == "S1" and v.reason_code == E_TENANT_MISSING,
|
||
v.to_dict())
|
||
# S2 Agent 未注册
|
||
v = M.adjudicate(dctx, "mentor", "blueprint.get", {"blueprint_id": 1}, store=store)
|
||
check("C4 S2 未注册 Agent(mentor) → PBL_E_FORBIDDEN",
|
||
not v.allowed and v.step == "S2" and v.reason_code == E_FORBIDDEN,
|
||
v.to_dict())
|
||
# S3 工具未注册（default-deny）
|
||
v = M.adjudicate(dctx, "designer", "blueprint.delete", {}, store=store)
|
||
check("C5 S3 白名单外工具 → PBL_E_FORBIDDEN(default-deny)",
|
||
not v.allowed and v.step == "S3" and v.reason_code == E_FORBIDDEN,
|
||
v.to_dict())
|
||
# S4 禁用工具
|
||
v = M.adjudicate(dctx, "designer", "blueprint.publish_auto", {}, store=store)
|
||
check("C6 S4 禁用工具 → PBL_E_FORBIDDEN + disable_reason",
|
||
not v.allowed and v.step == "S4" and v.reason_code == E_FORBIDDEN
|
||
and v.tool.get("disable_reason"), v.to_dict())
|
||
v = M.adjudicate(dctx, "designer", "kdb.write", {}, store=store)
|
||
check("C7 S4 kdb.write 禁用拒绝", not v.allowed and v.step == "S4")
|
||
v = M.adjudicate(dctx, "designer", "world.edit_3d", {}, store=store)
|
||
check("C8 S4 world.edit_3d 禁用拒绝（Q1 无 3D 编辑器）",
|
||
not v.allowed and v.step == "S4")
|
||
# S5 越权
|
||
narrow = M.make_agent_ctx("T001", "designer", permissions={"agent_tools"})
|
||
v = M.adjudicate(narrow, "designer", "blueprint.create",
|
||
{"intent_text": "x", "title": "y"}, store=store)
|
||
check("C9 S5 缺 pbl_authoring 权限域 → PBL_E_FORBIDDEN",
|
||
not v.allowed and v.step == "S5" and v.reason_code == E_FORBIDDEN,
|
||
v.to_dict())
|
||
# S6 Critic 零写权限
|
||
cctx = api.agent_ctx("critic")
|
||
for tc in ("blueprint.create", "blueprint.update", "template.copy",
|
||
"compile.trigger", "publish.request"):
|
||
v = M.adjudicate(cctx, "critic", tc, {"blueprint_id": 1}, store=store)
|
||
check("C10.%s Critic 调写工具被 S6 拒绝" % tc,
|
||
not v.allowed and v.step == "S6" and v.reason_code == E_FORBIDDEN,
|
||
v.to_dict())
|
||
v = M.adjudicate(cctx, "critic", "blueprint.get", {"blueprint_id": 1},
|
||
store=store)
|
||
check("C11 Critic 调只读 blueprint.get 通过 S6",
|
||
v.allowed or v.step in ("S8",), v.to_dict())
|
||
# S6 designer 越界调 critic.review
|
||
v = M.adjudicate(dctx, "designer", "critic.review", {"blueprint_id": 1},
|
||
store=store)
|
||
check("C12 S6 designer 调 critic.review 被拒（allowed_agents 限定）",
|
||
not v.allowed and v.step == "S6", v.to_dict())
|
||
# S7 需审批未批
|
||
v = M.adjudicate(dctx, "designer", "publish.request",
|
||
{"blueprint_id": 7, "visibility": "org"}, store=store)
|
||
check("C13 S7 publish.request 无审批 → PBL_E_STATE_ILLEGAL",
|
||
not v.allowed and v.step == "S7" and v.reason_code == E_STATE_ILLEGAL,
|
||
v.to_dict())
|
||
v = M.adjudicate(dctx, "designer", "compile.trigger",
|
||
{"blueprint_id": 7, "version_no": 1,
|
||
"compiler_version": "v1"}, store=store)
|
||
check("C14 S7 compile.trigger 无审批 → PBL_E_STATE_ILLEGAL",
|
||
not v.allowed and v.step == "S7" and v.reason_code == E_STATE_ILLEGAL,
|
||
v.to_dict())
|
||
# S8 入参契约
|
||
v = M.adjudicate(dctx, "designer", "blueprint.create", {"intent_text": "x"},
|
||
store=store)
|
||
check("C15 S8 缺 title → PBL_E_VALIDATION",
|
||
not v.allowed and v.step == "S8" and v.reason_code == E_VALIDATION,
|
||
v.to_dict())
|
||
v = M.adjudicate(dctx, "designer", "blueprint.create",
|
||
{"intent_text": "x", "title": "y",
|
||
"generation_source": "bogus"}, store=store)
|
||
check("C16 S8 枚举外取值 → PBL_E_VALIDATION",
|
||
not v.allowed and v.step == "S8" and v.reason_code == E_VALIDATION,
|
||
v.to_dict())
|
||
v = M.adjudicate(dctx, "designer", "publish.request",
|
||
{"blueprint_id": 7, "visibility": "galaxy"}, store=store)
|
||
check("C16b S7 先于 S8（审批门禁优先，fail-closed）",
|
||
not v.allowed and v.step == "S7"
|
||
and v.reason_code == E_STATE_ILLEGAL, v.to_dict())
|
||
v = M.adjudicate(dctx, "designer", "blueprint.create",
|
||
{"intent_text": "x", "title": "y"}, store=store)
|
||
check("C17 全通过 → allowed=True（step=S8）",
|
||
v.allowed and v.step == "S8", v.to_dict())
|
||
# kdb.search 空结果不报错（Q5/US-24）
|
||
r = api.invoke_tool("designer", "kdb.search", {"query": "海洋保护"})
|
||
check("C18 kdb.search 桩返回空集不报错",
|
||
r.get("stub") is True and r.get("total") == 0 and r.get("items") == [], r)
|
||
# 被拒调用也落轨迹（US-20）
|
||
try:
|
||
api.invoke_tool("designer", "blueprint.publish_auto", {})
|
||
except PblError:
|
||
pass
|
||
traces = api.list_traces({"agent_code": "designer"}, size=50)
|
||
deny_traces = [t for t in traces["items"] if t["status"] == "deny"]
|
||
check("C19 被拒调用写入 deny 轨迹", len(deny_traces) >= 1,
|
||
traces["total"])
|
||
dt = deny_traces[0]
|
||
check("C20 deny 轨迹含 proposed_action + result.deny",
|
||
bool(dt.get("proposed_action")) and
|
||
(dt.get("result") or {}).get("deny") is True, dt.get("result"))
|
||
check("C21 deny 轨迹记录 step_reached=S4",
|
||
dt.get("step_reached") == "S4", dt.get("step_reached"))
|
||
|
||
# ---------------- D. 四类强制人工审批 ----------------
|
||
check("D1 四类审批类型齐备",
|
||
sorted(M.MANDATORY_APPROVAL_TYPES) ==
|
||
sorted(["publish", "compile_execute", "blueprint_approve",
|
||
"tool_registry_change"]),
|
||
M.MANDATORY_APPROVAL_TYPES)
|
||
mx = M.mandatory_approval_matrix()
|
||
check("D2 审批矩阵 4 条", len(mx) == 4, len(mx))
|
||
check("D3 矩阵含 publish 触发工具 publish.request",
|
||
any(m["action_type"] == "publish" and
|
||
m["trigger_tool"] == "publish.request" for m in mx))
|
||
|
||
# Agent 自批被拒
|
||
ap = api.request_approval("publish", object_id=7, tool_code="publish.request",
|
||
object_type="blueprint",
|
||
action_payload={"visibility": "org"})
|
||
check("D4 发起审批返回 pending", ap["status"] == "pending", ap)
|
||
expect_error("D5 Agent 自批被拒（PBL_E_FORBIDDEN）", E_FORBIDDEN,
|
||
lambda: M.decide_approval(api.agent_ctx("designer"),
|
||
ap["approval_no"], "approved",
|
||
store=store))
|
||
# 人类批准
|
||
d = api.decide_approval(ap["approval_no"], "approved", "teacher_01",
|
||
comment="同意发布到机构")
|
||
check("D6 人类批准后 status=approved", d["status"] == "approved", d)
|
||
check("D7 approver_type=user", d.get("approver_type") == "user", d)
|
||
# 已决不可再决
|
||
expect_error("D8 已决审批单不可重复决定", E_STATE_ILLEGAL,
|
||
lambda: api.decide_approval(ap["approval_no"], "rejected",
|
||
"teacher_02"))
|
||
# 批准后 S7 通过
|
||
v = M.adjudicate(dctx, "designer", "publish.request",
|
||
{"blueprint_id": 7, "visibility": "org"},
|
||
approval_no=ap["approval_no"], store=store)
|
||
check("D9 有 approved 记录后 publish.request 通过 S7",
|
||
v.allowed and v.approval and
|
||
v.approval["approval_no"] == ap["approval_no"], v.to_dict())
|
||
# compile_execute 未批仍拒
|
||
v = M.adjudicate(dctx, "designer", "compile.trigger",
|
||
{"blueprint_id": 7, "version_no": 1,
|
||
"compiler_version": "v1"}, store=store)
|
||
check("D10 publish 审批不能顶替 compile_execute 审批",
|
||
not v.allowed and v.step == "S7", v.to_dict())
|
||
# tool_registry_change：禁用→启用须审批
|
||
expect_error("D11 无审批启用禁用工具 → PBL_E_STATE_ILLEGAL", E_STATE_ILLEGAL,
|
||
lambda: api.set_tool_status("kdb.write", "enabled",
|
||
admin_id="admin"))
|
||
ap2 = api.request_approval("tool_registry_change", object_id="kdb.search",
|
||
tool_code="kdb.search", object_type="tool",
|
||
action_payload={"to": "enabled"})
|
||
api.decide_approval(ap2["approval_no"], "approved", "admin")
|
||
# 非 admin 不能改注册表
|
||
expect_error("D12 非 Platform Admin 改注册表 → PBL_E_FORBIDDEN", E_FORBIDDEN,
|
||
lambda: M.set_tool_status(api.user_ctx("teacher_01"),
|
||
"kdb.search", "disabled",
|
||
reason="test", store=store))
|
||
# Agent 不能改注册表
|
||
expect_error("D13 Agent 改注册表 → PBL_E_FORBIDDEN", E_FORBIDDEN,
|
||
lambda: M.set_tool_status(api.agent_ctx("designer"),
|
||
"kdb.search", "disabled",
|
||
reason="test", store=store))
|
||
r = api.set_tool_status("kdb.search", "disabled", reason="临时下线（测试）",
|
||
admin_id="admin")
|
||
check("D14 Platform Admin 可禁用工具", r["status"] == "disabled", r)
|
||
v = M.adjudicate(dctx, "designer", "kdb.search", {"query": "x"}, store=store)
|
||
check("D15 禁用后 S4 立即拒绝", not v.allowed and v.step == "S4", v.to_dict())
|
||
ap3 = api.request_approval("tool_registry_change", object_id="kdb.search",
|
||
tool_code="kdb.search", object_type="tool")
|
||
api.decide_approval(ap3["approval_no"], "approved", "admin")
|
||
r = api.set_tool_status("kdb.search", "enabled",
|
||
approval_no=ap3["approval_no"], admin_id="admin")
|
||
check("D16 审批通过后启用成功", r["status"] == "enabled", r)
|
||
# register_tool 重复
|
||
expect_error("D17 重复注册 → PBL_E_DUPLICATE", E_DUPLICATE,
|
||
lambda: M.register_tool(api.admin_ctx("admin"), "kdb.search",
|
||
"x", "kdb", "enabled", "kdb",
|
||
store=store))
|
||
expect_error("D18 禁用注册缺 disable_reason → PBL_E_VALIDATION", E_VALIDATION,
|
||
lambda: M.register_tool(api.admin_ctx("admin"), "foo.bar",
|
||
"x", "foo", "disabled", "pbl_authoring",
|
||
store=store))
|
||
|
||
# ---------------- E. 轨迹 append-only + 7 要素 ----------------
|
||
check("E1 7 要素定义齐备", list(TRACE_ELEMENTS) ==
|
||
["input_context", "retrieved_knowledge", "tool_calls",
|
||
"proposed_action", "result", "approval", "final_output"],
|
||
TRACE_ELEMENTS)
|
||
tn = M.start_trace("designer", session_no="S1",
|
||
input_context={"api": "unit"}, ctx=dctx, store=store)
|
||
for e in ("retrieved_knowledge", "tool_calls", "proposed_action", "result",
|
||
"approval", "final_output"):
|
||
M.append_trace(tn, e, {"k": e}, ctx=dctx, store=store)
|
||
comp = api.trace_completeness(tn)
|
||
check("E2 7 要素全部写入 complete=True", comp["complete"] is True, comp["filled"])
|
||
check("E3 要素流水 7 条", comp["stage_log_count"] == 7, comp["stage_log_count"])
|
||
M.finish_trace(tn, status="done", step_reached="EXECUTED", ctx=dctx, store=store)
|
||
expect_error("E4 已完成轨迹追加 → PBL_E_APPEND_ONLY", E_APPEND_ONLY,
|
||
lambda: M.append_trace(tn, "result", {"hack": 1}, ctx=dctx,
|
||
store=store))
|
||
expect_error("E5 直接 update 轨迹被 append-only 守卫拒绝", E_APPEND_ONLY,
|
||
lambda: M.try_mutate_trace(tn, {"status": "open"}, ctx=dctx,
|
||
store=store))
|
||
expect_error("E6 非法 stage → PBL_E_VALIDATION", E_VALIDATION,
|
||
lambda: M.append_trace(tn, "not_a_stage", {}, ctx=dctx,
|
||
store=store))
|
||
# 读权限
|
||
other = M.make_agent_ctx("T001", "critic")
|
||
t2 = M.start_trace("designer", session_no="S2", input_context={},
|
||
ctx=dctx, store=store)
|
||
expect_error("E7 他方 Agent 读轨迹 → PBL_E_FORBIDDEN", E_FORBIDDEN,
|
||
lambda: M.get_trace(t2, ctx=other, store=store))
|
||
check("E8 Platform Admin 可读全量轨迹",
|
||
M.get_trace(t2, ctx=api.admin_ctx(), store=store)["trace_no"] == t2)
|
||
# 跨租户隔离
|
||
api2 = M.load_m4a(tenant_id="T002", store=store)
|
||
expect_error("E9 跨租户读轨迹 → PBL_E_NOT_FOUND", M.E_NOT_FOUND,
|
||
lambda: M.get_trace(t2, ctx=api2.agent_ctx("designer"),
|
||
store=store))
|
||
|
||
# ---------------- F. Designer / Critic 运行时 ----------------
|
||
slots, missing = M.parse_intent("为12岁学生设计一个海洋保护主题的PBL，"
|
||
"4人一组，共8课时，产出一段视频，预算5万")
|
||
check("F1 意图解析 age=12", slots["age"] == 12, slots)
|
||
check("F2 意图解析 teamSize=4", slots["teamSize"] == 4, slots)
|
||
check("F3 意图解析 duration=360", slots["duration"] == 360, slots)
|
||
check("F4 意图解析 artifact=video", slots["artifact"] == "video", slots)
|
||
check("F5 意图解析 budget=50000", slots["budget"] == 50000, slots)
|
||
check("F6 四要素齐备 → missing 为空", missing == [], missing)
|
||
_, miss2 = M.parse_intent("做一个关于火星的项目")
|
||
check("F7 信息不足 → missing 含四实质字段",
|
||
set(miss2) == set(M.CLARIFY_FIELDS), miss2)
|
||
|
||
ch, _ = M.instruction_to_changes("把预算增加到500万")
|
||
check("F8 指令解析为结构化变更 budget=5000000",
|
||
len(ch) == 1 and ch[0]["slot"] == "budget"
|
||
and ch[0]["value"] == 5000000, ch)
|
||
ch2, _ = M.instruction_to_changes("把团队规模改成6人")
|
||
check("F9 指令解析 teamSize=6",
|
||
any(c["slot"] == "teamSize" and c["value"] == 6 for c in ch2), ch2)
|
||
ch3, _ = M.instruction_to_changes("写得更有诗意一些")
|
||
check("F10 无法结构化 → 空变更（禁止整篇重写 US-03）", ch3 == [], ch3)
|
||
|
||
q = M.build_clarifications(["age", "duration"], round_no=1)
|
||
check("F11 clarify 只问实质缺失字段", len(q) == 2 and
|
||
{x["field"] for x in q} == {"age", "duration"}, q)
|
||
q4 = M.build_clarifications(["age"], round_no=4)
|
||
check("F12 轮次>3 不再追问，给默认假设",
|
||
q4 and q4[0].get("exceeded_max_rounds") is True and
|
||
q4[0]["question"] is None, q4)
|
||
q_ign = M.designer_clarify(None, ["age", "favorite_color"], ctx=dctx,
|
||
store=store, session_no="S9", round_no=1)
|
||
check("F13 非实质字段被忽略",
|
||
q_ign["ignored_fields"] == ["favorite_color"] and
|
||
len(q_ign["questions"]) == 1, q_ign)
|
||
|
||
# Designer 生成（后端未挂载 → 离线兜底 template_fallback，非错误）
|
||
res = api.designer_generate("为12岁学生设计海洋保护PBL，4人一组，8课时，产出视频",
|
||
owner_teacher_id="teacher_01",
|
||
class_id="C01", session_no="SG1")
|
||
check("F14 designer_generate 返回 trace_no", bool(res.get("trace_no")), res)
|
||
check("F15 后端未挂载时自动兜底 template_fallback（US-05 不抛错）",
|
||
res.get("generation_source") == "template_fallback", res)
|
||
check("F16 兜底提案落 pending_backend（Agent 不直连 DB）",
|
||
res.get("pending_backend") is True or res.get("blueprint_id") is not None
|
||
or bool(res.get("code")), res)
|
||
comp2 = api.trace_completeness(res["trace_no"])
|
||
check("F17 Designer 轨迹含 input_context/retrieved_knowledge/final_output",
|
||
comp2["filled"]["input_context"] and
|
||
comp2["filled"]["retrieved_knowledge"] and
|
||
comp2["filled"]["final_output"], comp2["filled"])
|
||
|
||
# Critic：零写权限 + 四要素
|
||
sug = M.assert_suggestion({"recommendation": "补充时长", "reason": "缺 duration",
|
||
"evidence": {"field": "duration"}, "confidence": 0.8})
|
||
check("F18 四要素齐备建议通过校验", sug["confidence"] == 0.8, sug)
|
||
expect_error("F19 缺 evidence → PBL_E_VALIDATION（14.3）", E_VALIDATION,
|
||
lambda: M.assert_suggestion({"recommendation": "a",
|
||
"reason": "b", "confidence": 0.5}))
|
||
expect_error("F20 confidence 越界 → PBL_E_VALIDATION", E_VALIDATION,
|
||
lambda: M.assert_suggestion({"recommendation": "a", "reason": "b",
|
||
"evidence": {}, "confidence": 1.5}))
|
||
wa = M.critic_write_attempt(api.agent_ctx("critic"), "blueprint.update",
|
||
{"blueprint_id": 1, "instruction": "x",
|
||
"target_changes": [{"object_type": "scene",
|
||
"field": "budget",
|
||
"value": 1}]},
|
||
store=store)
|
||
check("F21 Critic 写蓝图尝试被拒（14.1）",
|
||
wa["allowed"] is False and wa["verdict"] == "PASS"
|
||
and wa["step_reached"] == "S6", wa)
|
||
wa2 = M.critic_write_attempt(api.agent_ctx("critic"), "publish.request",
|
||
{"blueprint_id": 1, "visibility": "org"},
|
||
store=store)
|
||
check("F22 Critic 发布尝试被拒", wa2["allowed"] is False, wa2)
|
||
rules = m4a_critic._rule_suggestions(
|
||
{"slots": {"age": 12}},
|
||
{"dimensions": [{"dimension_code": "D01", "result": "fail",
|
||
"message": "缺学习目标"}],
|
||
"alerts": ["时长与阶段数不匹配"], "run": {"run_no": "R1"}})
|
||
check("F23 规则式建议覆盖 fail 维度 + 告警 + 缺失槽位",
|
||
len(rules) >= 3 and all(
|
||
all(k in r for k in M.SUGGESTION_FIELDS) for r in rules), rules)
|
||
|
||
# ---------------- G. DDL / 表定义 ----------------
|
||
sql = M.all_sql()
|
||
check("G1 DDL 含 4 张主表 + 流水表",
|
||
all(t in sql for t in ("pbl_agent_def", "pbl_agent_tool",
|
||
"pbl_agent_trace", "pbl_agent_approval",
|
||
"pbl_agent_trace_stage")))
|
||
check("G2 append-only 表带标记",
|
||
"[APPEND-ONLY" in sql and
|
||
sorted(M.APPEND_ONLY_TABLES) ==
|
||
["pbl_agent_trace", "pbl_agent_trace_stage"], M.APPEND_ONLY_TABLES)
|
||
check("G3 每表 tenant_id 打头（首个业务列）",
|
||
all(t["fields"][1]["name"] == "tenant_id" for t in M.TABLES))
|
||
check("G4 联合唯一索引以 tenant_id 起首",
|
||
all(idx["cols"][0] == "tenant_id"
|
||
for t in M.TABLES for idx in t["indexes"] if idx.get("unique")))
|
||
|
||
print("\n===== M4a 自测汇总：PASS=%d FAIL=%d =====" % (len(PASS), len(FAIL)))
|
||
if FAIL:
|
||
for f in FAIL:
|
||
print(" FAILED: %s" % f)
|
||
return 1
|
||
print("全部通过（%d 项断言）" % len(PASS))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|