# -*- coding: utf-8 -*- """M4a 契约自测（离线可跑，逐条覆盖 QC 退回 #1~#7）。 用法： python3 scripts/m4a_contract_selftest.py # 全量自测 python3 scripts/m4a_contract_selftest.py --smoke # 只跑冒烟测试兼容签名核验 退出码 0 = 全部通过；非 0 = 有失败项（fail-closed，不允许带病交付）。 """ import os import sys import tempfile import traceback ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) WORKSPACE_ROOT = os.path.dirname(os.path.dirname(ROOT)) # 与 check_contract_sync 同基准 sys.path.insert(0, ROOT) sys.path.insert(0, os.path.dirname(ROOT)) # 独立临时库，避免污染工作空间 _TMP_DB = os.path.join(tempfile.gettempdir(), "pbl_m4a_selftest.sqlite3") os.environ["PBL_AGENT_RUNTIME_SQLITE"] = _TMP_DB os.environ.setdefault("PBL_TENANT_ID", "t_selftest") RESULTS = [] def case(name): def deco(fn): def runner(): try: fn() RESULTS.append((name, True, "")) print("PASS %s" % name) except AssertionError as exc: RESULTS.append((name, False, str(exc))) print("FAIL %s :: %s" % (name, exc)) except Exception as exc: # noqa: BLE001 RESULTS.append((name, False, "%s: %s" % (type(exc).__name__, exc))) print("ERROR %s :: %s" % (name, exc)) traceback.print_exc() runner.__name__ = fn.__name__ return runner return deco from pbl_agent_runtime import api as AR # noqa: E402 —— QC #1：这行必须不 ImportError from pbl_agent_runtime import ( # noqa: E402 CONTRACT_FUNCTIONS, MANDATORY_APPROVAL_TYPES, PblContractError, load_pbl_agent_runtime, self_check, ) from pbl_agent_runtime.m4a_store import APPEND_ONLY_TABLES, get_store # noqa: E402 TENANT = "t_selftest" # --------------------------------------------------------------------------- # QC #1：__init__ 导出 api；引擎 import 闭包核验（冒烟测试第 267 行写法） # --------------------------------------------------------------------------- @case("QC#1 from pbl_agent_runtime import api as AR 不 ImportError") def t_qc1_import(): assert AR is not None for name in CONTRACT_FUNCTIONS: assert callable(getattr(AR, name)), "api.%s not callable" % name @case("QC#1 api() 工厂写法与属性写法指向同一契约集合") def t_qc1_dual_form(): obj = AR() assert callable(obj.pbl_tool_adjudicate) assert obj["pbl_tool_adjudicate"] is not None assert sorted(obj.names()) == sorted(CONTRACT_FUNCTIONS) assert len(CONTRACT_FUNCTIONS) == 10, CONTRACT_FUNCTIONS # --------------------------------------------------------------------------- # QC #2：冒烟测试签名兼容（tool_key / params_json / approval_type / enabled） # --------------------------------------------------------------------------- @case("QC#2 AR.pbl_tool_registry_save(tool_key=, enabled=) 签名可用") def t_qc2_registry_save_signature(): # 无审批 → fail-closed 拒绝并自动补建 pending 审批 r = AR.pbl_tool_registry_save(tenant_id=TENANT, tool_key="blueprint.read", enabled=False, operator="dev.bot") assert r["success"] is False, r assert r["error_code"] == "PBL.APPROVAL.REQUIRED", r assert r.get("approval_id"), r assert r.get("need_approval") is True, r # 人工审批通过后 → 保存成功 apr = AR.pbl_approval_create(tenant_id=TENANT, approval_type="tool_registry_change", title="关闭 blueprint.read", requested_by="dev.bot") assert apr["success"] is True, apr decided = AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr["approval_id"], decision="approve", decided_by="human.pm", comment="同意临时下线") assert decided["success"] is True and decided["status"] == "approved", decided saved = AR.pbl_tool_registry_save(tenant_id=TENANT, tool_key="blueprint.read", enabled=False, operator="dev.bot", approval_id=apr["approval_id"]) assert saved["success"] is True, saved assert saved["tool"]["status"] == "disabled", saved # 复原 apr2 = AR.pbl_approval_create(tenant_id=TENANT, approval_type="tool_registry_change", title="恢复 blueprint.read", requested_by="dev.bot") AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr2["approval_id"], decision="approve", decided_by="human.pm") back = AR.pbl_tool_registry_save(tenant_id=TENANT, tool_key="blueprint.read", enabled=True, operator="dev.bot", approval_id=apr2["approval_id"]) assert back["success"] is True and back["tool"]["status"] == "enabled", back @case("QC#2 AR.pbl_tool_adjudicate(agent_code=, tool_key=, params_json=) 签名可用") def t_qc2_adjudicate_signature(): r = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="blueprint.read", params_json="{}") assert r["success"] is True and r["allowed"] is True, r assert r["decision"] == "ALLOW", r assert len(r["steps"]) == 8, r["steps"] @case("QC#2 AR.pbl_approval_create(approval_type=) 签名可用 + 非法类型拒绝") def t_qc2_approval_signature(): for kind in MANDATORY_APPROVAL_TYPES: r = AR.pbl_approval_create(tenant_id=TENANT, approval_type=kind, title="%s 审批" % kind) assert r["success"] is True and r["status"] == "pending", (kind, r) bad = AR.pbl_approval_create(tenant_id=TENANT, approval_type="whatever") assert bad["success"] is False and bad["error_code"] == "PBL.APPROVAL.TYPE_INVALID", bad assert len(MANDATORY_APPROVAL_TYPES) == 4, MANDATORY_APPROVAL_TYPES @case("QC#2 params_json 传 JSON 字符串与 dict 等价") def t_qc2_params_json_forms(): a = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="evidence.collect", params_json='{"a": 1}') b = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="evidence.collect", params_json={"a": 1}) assert a["allowed"] is True and b["allowed"] is True, (a, b) bad = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="evidence.collect", params_json="{not json") assert bad["allowed"] is False and bad["denied_at_step"] == "params_valid", bad # --------------------------------------------------------------------------- # QC #3：10 个契约端点齐全 + 6 张表 # --------------------------------------------------------------------------- @case("QC#3 wwwroot/api 下 10 个 .dspy 端点齐全（含 registry_list/save）") def t_qc3_endpoints(): api_dir = os.path.join(ROOT, "wwwroot", "api") files = sorted(f[:-5] for f in os.listdir(api_dir) if f.endswith(".dspy")) missing = sorted(set(CONTRACT_FUNCTIONS) - set(files)) extra = sorted(set(files) - set(CONTRACT_FUNCTIONS)) assert not missing, "missing endpoints: %s" % missing assert not extra, "unregistered endpoints: %s" % extra assert len(files) == 10, files for must in ("pbl_tool_registry_list.dspy", "pbl_tool_registry_save.dspy"): assert os.path.isfile(os.path.join(api_dir, must)), must @case("QC#3/#7 6 张表齐全（含 pbl_tool_registry 与 pbl_llm_call_log）") def t_qc3_tables(): from pbl_agent_runtime.m4a_store import ALL_TABLES, DDL_SQLITE assert len(ALL_TABLES) == 6, ALL_TABLES for must in ("pbl_agent_def", "pbl_tool_registry", "pbl_agent_trace", "pbl_agent_tool_call", "pbl_approval", "pbl_llm_call_log"): assert must in ALL_TABLES, must assert ("CREATE TABLE IF NOT EXISTS %s" % must) in DDL_SQLITE, must addon = os.path.join(ROOT, "sql", "m4a_contract_addon.sql") assert os.path.isfile(addon), "sql/m4a_contract_addon.sql missing" with open(addon, encoding="utf-8") as handle: text = handle.read() for must in ("pbl_tool_registry", "pbl_llm_call_log"): assert ("CREATE TABLE IF NOT EXISTS %s" % must) in text, must @case("QC#3 pbl_tool_registry_list 返回 13 启用 / 9 禁用") def t_qc3_registry_counts(): r = AR.pbl_tool_registry_list(tenant_id=TENANT) assert r["success"] is True, r items = r["items"] assert len(items) == 22, len(items) enabled = [i for i in items if i["status"] == "enabled"] disabled = [i for i in items if i["status"] == "disabled"] assert len(enabled) == 13, [i["tool_key"] for i in enabled] assert len(disabled) == 9, [i["tool_key"] for i in disabled] keys = {i["tool_key"]: i for i in items} assert "pbl.publish" in keys and keys["pbl.publish"]["status"] == "disabled", keys.get("pbl.publish") assert keys["pbl.publish"]["approval_type"] == "publish" # 过滤参数生效 only_disabled = AR.pbl_tool_registry_list(tenant_id=TENANT, status="disabled") assert len(only_disabled["items"]) == 9, only_disabled["total"] # --------------------------------------------------------------------------- # QC #4：load_path.py 与 wwwroot/api 实际文件一一对应 # --------------------------------------------------------------------------- @case("QC#4 load_path.PATHS 与磁盘文件一一对应（无悬空路径）") def t_qc4_load_path(): sys.path.insert(0, os.path.join(ROOT, "scripts")) import importlib lp = importlib.import_module("load_path") importlib.reload(lp) dspy_paths = [] for path, role in lp.PATHS: rel = path.lstrip("/").split("/", 1)[1] full = os.path.join(ROOT, "wwwroot", rel) assert os.path.isfile(full), "PATHS 指向不存在文件: %s" % path assert role in ("any", "user", "admin"), (path, role) if path.endswith(".dspy"): dspy_paths.append(path) assert len(dspy_paths) == 10, dspy_paths names = sorted(p.split("/")[-1][:-5] for p in dspy_paths) assert names == sorted(CONTRACT_FUNCTIONS), names # 反向：磁盘上每个 dspy 都在 PATHS 里 on_disk = sorted(f[:-5] for f in os.listdir(os.path.join(ROOT, "wwwroot", "api")) if f.endswith(".dspy")) assert on_disk == names, (on_disk, names) # --------------------------------------------------------------------------- # QC #5：交付清单真实性（只列磁盘存在的文件） # --------------------------------------------------------------------------- @case("QC#5 交付清单文件全部真实存在（无幽灵文件）") def t_qc5_inventory(): manifest = os.path.join(ROOT, "scripts", "m4a_manifest.json") assert os.path.isfile(manifest), "先运行 check_contract_sync.py --json 生成清单" import json with open(manifest, encoding="utf-8") as handle: data = json.load(handle) assert data["success"] is True, data["problems"] ghost = [] for item in data["inventory"]: full = os.path.join(WORKSPACE_ROOT, item["path"]) if not os.path.isfile(full): ghost.append(item["path"]) else: real = os.path.getsize(full) assert real == item["bytes"], "%s bytes %d != %d" % (item["path"], real, item["bytes"]) assert not ghost, "幽灵文件: %s" % ghost paths = [i["path"] for i in data["inventory"]] assert not any("/son/" in p for p in paths), "笔误路径 son/ 仍在清单" for must in ("modules/pbl_agent_runtime/pbl_agent_runtime/init.py", "modules/pbl_agent_runtime/pbl_agent_runtime/__init__.py", "modules/pbl_agent_runtime/sql/m4a_contract_addon.sql", "modules/pbl_agent_runtime/scripts/load_path.py", "modules/pbl_agent_runtime/skill/SKILL.md", "modules/pbl_agent_runtime/pyproject.toml"): assert must in paths, must # --------------------------------------------------------------------------- # QC #6：dspy debug 必须是 f-string # --------------------------------------------------------------------------- @case("QC#6 所有 .dspy 的 debug 行为 f-string 且输出真实 params_kw") def t_qc6_debug_fstring(): api_dir = os.path.join(ROOT, "wwwroot", "api") for fname in sorted(os.listdir(api_dir)): if not fname.endswith(".dspy"): continue with open(os.path.join(api_dir, fname), encoding="utf-8") as handle: lines = handle.read().splitlines() debug_lines = [ln for ln in lines if ln.strip().startswith("debug(")] assert debug_lines, "%s 无 debug 行" % fname for ln in debug_lines: assert "debug(f'" in ln or 'debug(f"' in ln, "%s debug 非 f-string: %s" % (fname, ln) assert "params_kw" in ln or "result" in ln, "%s debug 无真实参数: %s" % (fname, ln) assert "{dict(params_kw)}" not in ln.replace("debug(f'", "").replace('debug(f"', "") or True # 字面量占位检测：非 f-string 时 {dict(params_kw)} 会原样打印 for ln in debug_lines: body = ln.strip() assert body.startswith("debug(f"), "%s debug 未用 f-string 前缀: %s" % (fname, body) for bad in ("import ", "print(", "uuid"): for idx, ln in enumerate(lines, 1): assert bad not in ln, "%s:%d 含禁项 %r" % (fname, idx, bad) assert any(ln.strip().startswith("return ") for ln in lines), "%s 缺显式 return" % fname # --------------------------------------------------------------------------- # QC #7：pbl_llm_call_log 表 + Designer/Critic 运行时留痕 # --------------------------------------------------------------------------- @case("QC#7 designer_run/critic_run 写入 pbl_llm_call_log 留痕") def t_qc7_llm_log(): before = get_store().query_one( "SELECT COUNT(1) AS c FROM pbl_llm_call_log WHERE tenant_id = ?", (TENANT,))["c"] r = AR.pbl_agent_designer_run(tenant_id=TENANT, task="生成第 3 关蓝图", prompt="设计一个电路关卡") assert r["success"] is True, r assert r["agent_code"] == "designer" and r["can_write"] is True, r assert r.get("output"), r c = AR.pbl_agent_critic_run(tenant_id=TENANT, task="评审第 3 关蓝图") assert c["success"] is True and c["can_write"] is False, c after = get_store().query_one( "SELECT COUNT(1) AS c FROM pbl_llm_call_log WHERE tenant_id = ?", (TENANT,))["c"] assert after >= before + 2, (before, after) rows = get_store().query( "SELECT * FROM pbl_llm_call_log WHERE tenant_id = ? ORDER BY created_at DESC LIMIT 5", (TENANT,)) assert rows and rows[0]["prompt_hash"] and len(rows[0]["prompt_hash"]) == 64, rows[:1] assert rows[0]["status"] in ("ok", "timeout", "error", "rate_limited", "fallback"), rows[0] for col in ("latency_ms", "timeout_ms", "retry_count", "rate_limited", "fallback_used"): assert col in rows[0], col @case("QC#7 LLM 失败路径留痕：retry/限流/兜底字段可写且 append-only") def t_qc7_llm_log_failure_paths(): from pbl_agent_runtime.m4a_contract import log_llm_call log_llm_call(tenant_id=TENANT, agent_code="designer", run_id="run_x", purpose="designer_run", status="rate_limited", retry_count=2, rate_limited=True, latency_ms=5100, timeout_ms=30000, prompt="p", error_code="PBL.LLM.RATE_LIMIT", error_msg="429 too many requests") log_llm_call(tenant_id=TENANT, agent_code="designer", run_id="run_x", purpose="designer_run", status="fallback", retry_count=2, fallback_used=True, prompt="p", error_code="PBL.LLM.FALLBACK", error_msg="upstream timeout 510s") rows = get_store().query( "SELECT * FROM pbl_llm_call_log WHERE tenant_id = ? AND run_id = 'run_x'", (TENANT,)) assert len(rows) == 2, rows assert {r["status"] for r in rows} == {"rate_limited", "fallback"}, rows assert any(r["fallback_used"] == 1 for r in rows) assert any(r["rate_limited"] == 1 for r in rows) # append-only：留痕表禁 UPDATE/DELETE for table in APPEND_ONLY_TABLES: try: get_store().execute("DELETE FROM %s WHERE tenant_id = ?" % table, (TENANT,)) raise AssertionError("%s append-only 守卫失效" % table) except PblContractError as exc: assert exc.code == "PBL.STORE.APPEND_ONLY", exc.code # --------------------------------------------------------------------------- # 需求主链路：Designer/Critic、8 步裁决、四类审批、Critic 零写 # --------------------------------------------------------------------------- @case("需求 Agent 定义：designer 可写 / critic 零写（write_scope=none）") def t_agent_defs(): rows = get_store().query( "SELECT * FROM pbl_agent_def WHERE tenant_id = ? ORDER BY agent_code", (TENANT,)) m = {r["agent_code"]: r for r in rows} assert set(m) == {"critic", "designer"}, m.keys() assert m["designer"]["can_write"] == 1 and m["designer"]["write_scope"] != "none" assert m["critic"]["can_write"] == 0 and m["critic"]["write_scope"] == "none" @case("需求 Critic 零写：写类工具在 S3 被 DENY") def t_critic_zero_write(): for tool in ("blueprint.write", "evidence.collect", "runtime.event", "assessment.score", "compiler.execute", "blueprint.approve", "approval.create"): r = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="critic", tool_key=tool, params_json="{}") assert r["allowed"] is False, (tool, r) assert r["decision"] == "DENY", (tool, r) assert r["denied_at_step"] == "agent_write_scope", (tool, r["denied_at_step"]) assert r["error_code"] == "PBL.AGENT.WRITE_DENIED", (tool, r["error_code"]) # critic 只读工具放行 for tool in ("blueprint.read", "kdb.query", "blueprint.validate", "tool_registry.read"): r = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="critic", tool_key=tool, params_json="{}") assert r["allowed"] is True, (tool, r) @case("需求 critic_run 入口即拒写类工具请求") def t_critic_run_rejects_write_tools(): r = AR.pbl_agent_critic_run(tenant_id=TENANT, task="评审", tools=["blueprint.write"]) assert r["success"] is False and r["error_code"] == "PBL.AGENT.WRITE_DENIED", r assert r["denied_tools"] == ["blueprint.write"], r ok_run = AR.pbl_agent_critic_run(tenant_id=TENANT, task="评审", tools=["blueprint.read"]) assert ok_run["success"] is True and ok_run["can_write"] is False, ok_run @case("需求 8 步裁决顺序固定 S1~S8 且逐步留痕") def t_eight_steps(): from pbl_agent_runtime.m4a_contract import ADJUDICATION_STEPS assert [s[0] for s in ADJUDICATION_STEPS] == ["S%d" % i for i in range(1, 9)] assert [s[1] for s in ADJUDICATION_STEPS] == [ "tenant_context", "agent_registered", "agent_write_scope", "tool_registered", "tool_enabled", "permission_granted", "params_valid", "approval_granted"] r = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="blueprint.read", params_json="{}") assert [s["step"] for s in r["steps"]] == ["S%d" % i for i in range(1, 9)], r["steps"] @case("需求 fail-closed：未注册工具/未注册 Agent/缺租户 一律 DENY") def t_fail_closed(): unknown_tool = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="no.such.tool", params_json="{}") assert unknown_tool["allowed"] is False and unknown_tool["decision"] == "DENY" assert unknown_tool["denied_at_step"] == "tool_registered", unknown_tool unknown_agent = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="ghost", tool_key="blueprint.read", params_json="{}") assert unknown_agent["allowed"] is False assert unknown_agent["denied_at_step"] == "agent_registered", unknown_agent no_tenant = AR.pbl_tool_adjudicate(tenant_id="", agent_code="designer", tool_key="blueprint.read", params_json="{}") assert no_tenant["success"] is False, no_tenant assert no_tenant["error_code"] == "PBL.TENANT.MISSING", no_tenant empty_key = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="", params_json="{}") assert empty_key["allowed"] is False, empty_key @case("需求 pbl.publish 等 9 个禁用工具在 S5 被 DENY") def t_disabled_tools(): disabled = ("pbl.publish", "blueprint.publish_auto", "compiler.autorun", "kdb.write", "tenant.switch", "rbac.grant", "approval.auto_decide", "trace.delete", "llm.raw_exec") assert len(disabled) == 9 for tool in disabled: r = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key=tool, params_json="{}") assert r["allowed"] is False, (tool, r) assert r["denied_at_step"] in ("tool_enabled", "permission_granted"), (tool, r) publish = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="pbl.publish", params_json="{}") assert publish["denied_at_step"] == "tool_enabled", publish assert publish["error_code"] == "PBL.TOOL.DISABLED", publish @case("需求 禁用工具不可通过 registry_save 翻转为 enabled") def t_forbidden_cannot_enable(): apr = AR.pbl_approval_create(tenant_id=TENANT, approval_type="tool_registry_change", title="试图启用 pbl.publish", requested_by="dev.bot") AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr["approval_id"], decision="approve", decided_by="human.pm") r = AR.pbl_tool_registry_save(tenant_id=TENANT, tool_key="pbl.publish", enabled=True, operator="dev.bot", approval_id=apr["approval_id"]) assert r["success"] is False and r["error_code"] == "PBL.REGISTRY.FORBIDDEN_TOOL", r still = AR.pbl_tool_registry_list(tenant_id=TENANT, tool_key="pbl.publish") assert still["items"][0]["status"] == "disabled", still["items"][0] @case("需求 四类强制审批：无 approved 审批单即 DENY 且自动补建 pending") def t_four_approvals(): pairs = (("blueprint.approve", "blueprint_approve"), ("compiler.execute", "compile_execute")) for tool, kind in pairs: r = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key=tool, params_json="{}") assert r["allowed"] is False, (tool, r) assert r["denied_at_step"] == "approval_granted", (tool, r) assert r["error_code"] == "PBL.APPROVAL.REQUIRED", (tool, r) assert r["approval_type"] == kind, (tool, r) assert r.get("approval_id"), (tool, r) pending = AR.pbl_approval_list(tenant_id=TENANT, approval_id=None, status="pending") assert pending["success"] is True # 人工批准后放行 decided = AR.pbl_approval_decide(tenant_id=TENANT, approval_id=r["approval_id"], decision="approve", decided_by="human.pm", comment="评审通过") assert decided["status"] == "approved", decided again = AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key=tool, params_json="{}", approval_id=r["approval_id"]) assert again["allowed"] is True, (tool, again) @case("需求 审批无绕过路径：Agent 不能自批、不能重复批") def t_approval_no_bypass(): apr = AR.pbl_approval_create(tenant_id=TENANT, approval_type="publish", title="发布上线") for bot in ("designer", "critic", "agent", "system"): r = AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr["approval_id"], decision="approve", decided_by=bot) assert r["success"] is False and r["error_code"] == "PBL.APPROVAL.NO_BYPASS", (bot, r) no_decider = AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr["approval_id"], decision="approve") assert no_decider["success"] is False, no_decider bad_verdict = AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr["approval_id"], decision="maybe", decided_by="human.pm") assert bad_verdict["success"] is False and bad_verdict["error_code"] == "PBL.APPROVAL.DECISION_INVALID" ok_reject = AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr["approval_id"], decision="reject", decided_by="human.pm", comment="证据不足") assert ok_reject["status"] == "rejected", ok_reject twice = AR.pbl_approval_decide(tenant_id=TENANT, approval_id=apr["approval_id"], decision="approve", decided_by="human.pm") assert twice["success"] is False and twice["error_code"] == "PBL.APPROVAL.ALREADY_DECIDED", twice missing = AR.pbl_approval_decide(tenant_id=TENANT, approval_id="apr_nope", decision="approve", decided_by="human.pm") assert missing["success"] is False, missing @case("需求 轨迹 7 要素强制 + append-only + 可查询") def t_trace(): from pbl_agent_runtime.m4a_contract import TRACE_ELEMENTS assert list(TRACE_ELEMENTS) == ["who", "occurred_at", "what", "why", "how", "result", "evidence_ref"] bad = AR.pbl_agent_trace_write(tenant_id=TENANT, who="designer", what="改了蓝图") assert bad["success"] is False and bad["error_code"] == "PBL.TRACE.ELEMENTS_REQUIRED", bad good = AR.pbl_agent_trace_write( tenant_id=TENANT, who="designer", what="生成第 3 关蓝图", why="任务要求", how="compiler.compile 确定性编译", result="ok", evidence_ref="blueprint:bp_3") assert good["success"] is True and good["trace_id"], good listed = AR.pbl_agent_trace_list(tenant_id=TENANT, trace_id=good["trace_id"]) assert listed["success"] is True and listed["total"] >= 1, listed rows = AR.pbl_agent_trace_list(tenant_id=TENANT, limit=5) assert rows["success"] is True and len(rows["items"]) >= 1 @case("需求 裁决留痕落 pbl_agent_tool_call（ALLOW/DENY 均留痕）") def t_tool_call_audit(): AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="designer", tool_key="blueprint.read", params_json="{}", trace_id="trc_audit_1") AR.pbl_tool_adjudicate(tenant_id=TENANT, agent_code="critic", tool_key="blueprint.write", params_json="{}", trace_id="trc_audit_2") rows = get_store().query( "SELECT * FROM pbl_agent_tool_call WHERE tenant_id = ? AND trace_id IN " "('trc_audit_1','trc_audit_2') ORDER BY trace_id", (TENANT,)) assert len(rows) == 2, rows m = {r["trace_id"]: r for r in rows} assert m["trc_audit_1"]["decision"] == "ALLOW", m["trc_audit_1"] assert m["trc_audit_2"]["decision"] == "DENY", m["trc_audit_2"] assert m["trc_audit_2"]["denied_at_step"] == "agent_write_scope" assert m["trc_audit_2"]["reason_code"] == "PBL.AGENT.WRITE_DENIED" @case("挂载 load_pbl_agent_runtime：注册 10 契约 + 自检通过") def t_load_module(): report = load_pbl_agent_runtime(tenant_id=TENANT, strict=True) assert report["success"] is True, report["problems"] assert report["registered_count"] >= 10, report["registered_count"] for name in CONTRACT_FUNCTIONS: assert name in report["registered"], name assert len(report["api_routes"]) == 10, report["api_routes"] assert report["tables"] == 6, report["tables"] assert report["seed"]["enabled"] == 13 and report["seed"]["disabled"] == 9, report["seed"] # 幂等：重复挂载不报错、数量不变 again = load_pbl_agent_runtime(tenant_id=TENANT, strict=True) assert again["success"] is True listing = AR.pbl_tool_registry_list(tenant_id=TENANT) assert len(listing["items"]) == 22, listing["total"] @case("挂载自检 self_check(strict=True) 全绿") def t_self_check(): r = self_check(tenant_id=TENANT, strict=True) assert r["success"] is True, r["problems"] assert r["contract_count"] == 10, r["contract_count"] assert r["tables"] == 6, r["tables"] @case("库名不硬编码：get_module_dbname 优先") def t_dbname(): from pbl_agent_runtime.m4a_store import module_dbname os.environ["PBL_AGENT_RUNTIME_DBNAME"] = "pbl_runtime_db" try: assert module_dbname() in ("pbl_runtime_db", "pbl_agent_runtime"), module_dbname() finally: del os.environ["PBL_AGENT_RUNTIME_DBNAME"] src_files = [] for base, dirs, files in os.walk(os.path.join(ROOT, "pbl_agent_runtime")): dirs[:] = [d for d in dirs if d != "__pycache__"] src_files += [os.path.join(base, f) for f in files if f.endswith(".py")] for full in src_files: with open(full, encoding="utf-8") as handle: text = handle.read() assert "DBNAME = 'pbl" not in text and 'DBNAME = "pbl' not in text, full def main(): if os.path.isfile(_TMP_DB): os.remove(_TMP_DB) # 用例按名排序执行，先做一次幂等 seed，保证每个用例彼此独立可单跑 get_store().ensure_schema() from pbl_agent_runtime.m4a_contract import seed_registry as _seed _seed(TENANT) only_smoke = "--smoke" in sys.argv runners = [v for k, v in sorted(globals().items()) if k.startswith("t_") and callable(v)] if only_smoke: runners = [t_qc1_import, t_qc1_dual_form, t_qc2_registry_save_signature, t_qc2_adjudicate_signature, t_qc2_approval_signature] for runner in runners: runner() passed = len([r for r in RESULTS if r[1]]) total = len(RESULTS) print("\n==== M4a contract selftest: %d/%d passed ====" % (passed, total)) for name, ok_flag, msg in RESULTS: if not ok_flag: print(" FAILED: %s :: %s" % (name, msg)) return 0 if passed == total else 1 if __name__ == "__main__": sys.exit(main())