592 lines
30 KiB
Python
592 lines
30 KiB
Python
# -*- 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())
|