# -*- coding: utf-8 -*- """ pbl_agent_runtime M4a — 对外统一门面（facade） 装配： from pbl_agent_runtime.m4a import load_m4a, M4aApi api = load_m4a(tenant_id="T001") # 幂等：建表 + seed_agents + seed_tools 对外契约（modules/pbl_agent_runtime.md §5）： Designer : designer_generate / designer_modify / designer_clarify Critic : critic_review / get_critic_report 工具裁决 : invoke_tool / adjudicate / register_tool / list_tools / set_tool_status / seed_tools 轨迹 : start_trace / append_trace / get_trace / list_traces / trace_completeness 审批 : request_approval / decide_approval / get_approval / list_pending_approvals / mandatory_approval_matrix """ from .m4a_kernel import ( PblError, TenantContext, get_store, set_store, reset_store, fail, MemoryStore, AppendOnlyGuard, canonical_json, now_ts, gen_no, ADJUDICATION_STEPS, TRACE_ELEMENTS, E_TENANT_MISSING, E_FORBIDDEN, E_STATE_ILLEGAL, E_NOT_FOUND, E_VALIDATION, E_DUPLICATE, E_APPEND_ONLY, E_MODEL_UNAVAILABLE, E_BACKEND_UNAVAILABLE, E_TEMPLATE_NONE, ACTOR_USER, ACTOR_AGENT, ACTOR_SYSTEM, ) from .m4a_tables import TABLES, APPEND_ONLY_TABLES, to_sql, all_sql, ensure_tables from .m4a_registry import ( AGENT_DEFS, TOOL_SEED, ENABLED_TOOL_CODES, DISABLED_TOOL_CODES, APPROVAL_REQUIRED_TOOLS, APPROVAL_TYPES, DESIGNER, CRITIC, PERM_AUTHORING, PERM_AGENT_TOOLS, PERM_PUBLISHING, PERM_KDB, PERM_PLATFORM_ADMIN, seed_agents, seed_tools, get_agent_def, list_agents, register_tool, get_tool, list_tools, set_tool_status, registry_stats, ) from .m4a_adjudicate import ( Verdict, adjudicate, invoke_tool, validate_args, adjudicate_report, explain_chain, ) from .m4a_trace import ( start_trace, append_trace, finish_trace, get_trace, list_traces, list_trace_stages, trace_completeness, try_mutate_trace, ) from .m4a_approval import ( MANDATORY_APPROVAL_TYPES, APPROVAL_PUBLISH, APPROVAL_COMPILE, APPROVAL_BLUEPRINT, APPROVAL_REGISTRY, request_approval, decide_approval, get_approval, list_pending_approvals, list_approvals, find_approved, mandatory_approval_matrix, normalize_action, ) from .m4a_designer import ( designer_generate, designer_modify, designer_clarify, parse_intent, instruction_to_changes, build_clarifications, set_model_runner, CLARIFY_FIELDS, MAX_CLARIFY_ROUNDS, ) from .m4a_critic import ( critic_review, get_critic_report, assert_suggestion, critic_write_attempt, set_critic_runner, CRITIC_READONLY_TOOLS, SUGGESTION_FIELDS, ) from .m4a_backend import execute as execute_tool_backend, backend_availability # --------------------------------------------------------------------------- # 上下文构造辅助 # --------------------------------------------------------------------------- def make_agent_ctx(tenant_id, agent_code=DESIGNER, permissions=None, org_id=None): """构造 Agent 调用上下文（actor_type=agent，不可审批）。""" perms = set(permissions or ()) if not perms: # Agent 默认权限域：authoring + agent_tools + kdb（只读）+ publishing。 # 注意：持有 publishing 只代表"可提案"发布（S5 通过）， # 真正执行仍被 S7 人工审批门禁拦住（14.2 Publish 必须人工审批）。 perms = {PERM_AUTHORING, PERM_AGENT_TOOLS, PERM_KDB, PERM_PUBLISHING} return TenantContext(tenant_id=tenant_id, actor_type=ACTOR_AGENT, actor_id=agent_code, permissions=perms, org_id=org_id) def make_user_ctx(tenant_id, user_id, permissions=None, org_id=None): """构造人类用户上下文（actor_type=user，可审批 / 可治理注册表）。""" perms = set(permissions or ()) if not perms: perms = {PERM_AUTHORING, PERM_PUBLISHING, PERM_AGENT_TOOLS} return TenantContext(tenant_id=tenant_id, actor_type=ACTOR_USER, actor_id=user_id, permissions=perms, org_id=org_id) def make_admin_ctx(tenant_id, user_id="admin"): """Platform Admin（工具注册表治理 + 全量轨迹可读）。""" return TenantContext(tenant_id=tenant_id, actor_type=ACTOR_USER, actor_id=user_id, permissions={PERM_PLATFORM_ADMIN, PERM_AUTHORING, PERM_PUBLISHING, PERM_AGENT_TOOLS, PERM_KDB}) # --------------------------------------------------------------------------- # 装配 # --------------------------------------------------------------------------- def load_m4a(tenant_id=None, store=None, force_memory=False): """ 幂等装配：建表 → seed_agents（designer/critic）→ seed_tools（13+9）。 返回 M4aApi 实例。build.sh / 应用 init() 调用。 """ st = store or get_store(force_memory=force_memory) ensure_tables(st) ctx = TenantContext(tenant_id=tenant_id or "__platform__", actor_type=ACTOR_SYSTEM, actor_id="seed", permissions={PERM_PLATFORM_ADMIN}) agents = seed_agents(ctx=ctx, store=st) tools = seed_tools(ctx=ctx, store=st) return M4aApi(tenant_id=tenant_id or "__platform__", store=st, seed_result={"agents": agents, "tools": tools}) class M4aApi(object): """按租户绑定的 M4a 门面（避免每个调用都传 store/tenant）。""" def __init__(self, tenant_id, store=None, seed_result=None): self.tenant_id = tenant_id self.store = store or get_store() self.seed_result = seed_result or {} # --- 上下文 --- def agent_ctx(self, agent_code=DESIGNER, permissions=None): return make_agent_ctx(self.tenant_id, agent_code, permissions) def user_ctx(self, user_id, permissions=None): return make_user_ctx(self.tenant_id, user_id, permissions) def admin_ctx(self, user_id="admin"): return make_admin_ctx(self.tenant_id, user_id) # --- Designer --- def designer_generate(self, intent_text, owner_teacher_id=None, class_id=None, session_no=None, model_runner=None): return designer_generate(intent_text, owner_teacher_id=owner_teacher_id, class_id=class_id, session_no=session_no, ctx=self.agent_ctx(DESIGNER), store=self.store, model_runner=model_runner) def designer_modify(self, blueprint_id, instruction, session_no=None, version_no=None): return designer_modify(blueprint_id, instruction, session_no=session_no, ctx=self.agent_ctx(DESIGNER), store=self.store, version_no=version_no) def designer_clarify(self, blueprint_id, missing_fields, session_no=None, round_no=None): return designer_clarify(blueprint_id, missing_fields, ctx=self.agent_ctx(DESIGNER), store=self.store, session_no=session_no, round_no=round_no) # --- Critic --- def critic_review(self, blueprint_id, version_no=None, session_no=None): return critic_review(blueprint_id, version_no=version_no, ctx=self.agent_ctx(CRITIC), store=self.store, session_no=session_no) def get_critic_report(self, trace_no): return get_critic_report(trace_no, ctx=self.admin_ctx(), store=self.store) # --- 工具裁决 --- def invoke_tool(self, agent_code, tool_code, args=None, trace_no=None, session_no=None, approval_no=None): return invoke_tool(self.agent_ctx(agent_code), agent_code, tool_code, args=args, trace_no=trace_no, session_no=session_no, approval_no=approval_no, store=self.store) def adjudicate(self, agent_code, tool_code, args=None): return adjudicate_report(self.agent_ctx(agent_code), agent_code, tool_code, args=args, store=self.store) def list_tools(self, status=None, group=None): return list_tools(ctx=self.admin_ctx(), status=status, group=group, store=self.store) def registry_stats(self): return registry_stats(ctx=self.admin_ctx(), store=self.store) def set_tool_status(self, tool_code, status, reason=None, admin_id="admin", approval_no=None): return set_tool_status(self.admin_ctx(admin_id), tool_code, status, reason=reason or approval_no, store=self.store) # --- 轨迹 --- def get_trace(self, trace_no): return get_trace(trace_no, ctx=self.admin_ctx(), store=self.store) def list_traces(self, filters=None, page=1, size=20): return list_traces(filters, page=page, size=size, ctx=self.admin_ctx(), store=self.store) def trace_completeness(self, trace_no): return trace_completeness(trace_no, ctx=self.admin_ctx(), store=self.store) # --- 审批 --- def request_approval(self, action_type, object_id=None, agent_code=DESIGNER, tool_code=None, object_type=None, action_payload=None, approver_id=None, trace_no=None, trace_id=None): return request_approval(self.agent_ctx(agent_code), trace_id=trace_id, action_type=action_type, agent_code=agent_code, tool_code=tool_code, object_type=object_type, object_id=object_id, action_payload=action_payload, approver_id=approver_id, trace_no=trace_no, store=self.store) def decide_approval(self, approval_no, status, user_id, comment=None): return decide_approval(self.user_ctx(user_id), approval_no, status, comment=comment, store=self.store) def list_pending_approvals(self, approver_id): return list_pending_approvals(self.user_ctx(approver_id), approver_id=approver_id, store=self.store) # --- 自检 --- def self_check(self): """M4a 装配自检：Agent 定义 / 工具计数 / 审批矩阵 / 裁决链步数。""" agents = list_agents(ctx=self.admin_ctx(), store=self.store) stats = self.registry_stats() return { "tenant_id": self.tenant_id, "agents": [{"agent_code": a["agent_code"], "write_allowed": a.get("write_allowed"), "status": a.get("status"), "allowed_tools": a.get("allowed_tools")} for a in agents], "agent_count": len(agents), "critic_write_allowed": [a["agent_code"] for a in agents if a["agent_code"] == CRITIC and a.get("write_allowed") in (1, True)], "tools": stats, "adjudication_steps": [s[0] for s in ADJUDICATION_STEPS], "adjudication_step_count": len(ADJUDICATION_STEPS), "mandatory_approvals": [m["action_type"] for m in mandatory_approval_matrix()], "trace_elements": list(TRACE_ELEMENTS), "append_only_tables": list(APPEND_ONLY_TABLES), "backend": backend_availability(), } __all__ = [ "load_m4a", "M4aApi", "make_agent_ctx", "make_user_ctx", "make_admin_ctx", "TenantContext", "PblError", "get_store", "set_store", "reset_store", "MemoryStore", "AppendOnlyGuard", "ensure_tables", "TABLES", "to_sql", "all_sql", "APPEND_ONLY_TABLES", "seed_agents", "seed_tools", "get_agent_def", "list_agents", "register_tool", "get_tool", "list_tools", "set_tool_status", "registry_stats", "AGENT_DEFS", "TOOL_SEED", "ENABLED_TOOL_CODES", "DISABLED_TOOL_CODES", "APPROVAL_REQUIRED_TOOLS", "APPROVAL_TYPES", "adjudicate", "adjudicate_report", "invoke_tool", "validate_args", "explain_chain", "Verdict", "ADJUDICATION_STEPS", "start_trace", "append_trace", "finish_trace", "get_trace", "list_traces", "list_trace_stages", "trace_completeness", "try_mutate_trace", "TRACE_ELEMENTS", "request_approval", "decide_approval", "get_approval", "list_pending_approvals", "list_approvals", "find_approved", "mandatory_approval_matrix", "MANDATORY_APPROVAL_TYPES", "designer_generate", "designer_modify", "designer_clarify", "parse_intent", "instruction_to_changes", "set_model_runner", "critic_review", "get_critic_report", "assert_suggestion", "critic_write_attempt", "set_critic_runner", "backend_availability", "DESIGNER", "CRITIC", ]