90 lines
4.0 KiB
Python
90 lines
4.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
M4a seed 脚本（build.sh / 部署时调用，幂等）
|
||
|
||
用法：
|
||
python3 scripts/seed_m4a.py # 平台级 seed（tenant_id=__platform__）
|
||
python3 scripts/seed_m4a.py --tenant T001 # 指定租户 seed
|
||
python3 scripts/seed_m4a.py --memory # 离线内存库演练（不连 DB）
|
||
|
||
动作：建表（4 主表 + 1 append-only 流水表）→ seed_agents（designer/critic）
|
||
→ seed_tools（13 enabled + 9 disabled）→ 打印自检结果
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from pbl_agent_runtime.m4a_init import load_pbl_agent_runtime_m4a # noqa: E402
|
||
from pbl_agent_runtime.m4a_kernel import get_store, MemoryStore, set_store # noqa: E402
|
||
from pbl_agent_runtime.m4a import M4aApi, make_admin_ctx # noqa: E402
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--tenant", default=None, help="租户ID（缺省=平台级 __platform__）")
|
||
ap.add_argument("--memory", action="store_true", help="使用内存库演练（不连 DB）")
|
||
args = ap.parse_args()
|
||
|
||
store = None
|
||
if args.memory:
|
||
store = set_store(MemoryStore())
|
||
else:
|
||
store = get_store()
|
||
|
||
ret = load_pbl_agent_runtime_m4a(tenant_id=args.tenant, store=store)
|
||
api = M4aApi(tenant_id=args.tenant or "__platform__", store=store)
|
||
sc = api.self_check()
|
||
|
||
print("=" * 72)
|
||
print("pbl_agent_runtime M4a seed 结果")
|
||
print("=" * 72)
|
||
print("module : %s (%s)" % (ret["module"], ret["part"]))
|
||
print("dbname : %s" % ret["dbname"])
|
||
print("tables : %s" % ", ".join(t for t in ret["tables"]["created"]))
|
||
print("append_only : %s" % ", ".join(ret["append_only_tables"]))
|
||
print("agents seed : %s" % json.dumps(ret["seed"]["agents"],
|
||
ensure_ascii=False))
|
||
print("tools seed : inserted=%s skipped=%s enabled=%s disabled=%s total=%s"
|
||
% (ret["seed"]["tools"]["inserted"], ret["seed"]["tools"]["skipped"],
|
||
ret["seed"]["tools"]["enabled"], ret["seed"]["tools"]["disabled"],
|
||
ret["seed"]["tools"]["total"]))
|
||
print("enabled (13) : %s" % ", ".join(ret["enabled_tools"]))
|
||
print("disabled (9) : %s" % ", ".join(ret["disabled_tools"]))
|
||
print("approval needed : %s" % ", ".join(ret["approval_required_tools"]))
|
||
print("-" * 72)
|
||
print("self_check.agents : %s"
|
||
% json.dumps([{"code": a["agent_code"], "write": a["write_allowed"]}
|
||
for a in sc["agents"]], ensure_ascii=False))
|
||
print("self_check.critic_write : %s (须为空=Critic 零写权限)"
|
||
% sc["critic_write_allowed"])
|
||
print("self_check.tools : %s" % json.dumps(
|
||
{k: v for k, v in sc["tools"].items()
|
||
if k in ("total", "enabled", "disabled", "approval_required",
|
||
"disabled_missing_reason", "critic_writable_tools")},
|
||
ensure_ascii=False))
|
||
print("self_check.adjudication : %s (%d 步)"
|
||
% ("→".join(sc["adjudication_steps"]), sc["adjudication_step_count"]))
|
||
print("self_check.approvals : %s" % ", ".join(sc["mandatory_approvals"]))
|
||
print("self_check.trace_elements: %s" % ", ".join(sc["trace_elements"]))
|
||
print("-" * 72)
|
||
ok = (sc["agent_count"] == 2 and not sc["critic_write_allowed"]
|
||
and sc["tools"]["total"] == 22 and sc["tools"]["enabled"] == 13
|
||
and sc["tools"]["disabled"] == 9
|
||
and not sc["tools"]["disabled_missing_reason"]
|
||
and not sc["tools"]["critic_writable_tools"]
|
||
and sc["adjudication_step_count"] == 8
|
||
and len(sc["mandatory_approvals"]) == 4
|
||
and len(sc["trace_elements"]) == 7)
|
||
print("M4a seed 门禁: %s" % ("PASS" if ok else "FAIL"))
|
||
print("backend availability: %s" % json.dumps(
|
||
{k: v["available"] for k, v in sc["backend"].items()}, ensure_ascii=False))
|
||
return 0 if ok else 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|