# -*- 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())