85 lines
4.2 KiB
Python
85 lines
4.2 KiB
Python
# ============================================================
|
||
# 统一工具运行时 v1（moduleId: core-tool-runtime, 可重生 ✅）
|
||
# plan.md §3.3 + 矩阵「LLM 只有提议权」：所有 LLM 工具调用必须经本入口执行。
|
||
# - LLM 提议的 intent 必须先登记（_POWER_MAP 或只读意图白名单），未登记 → 拒绝 + 审计；
|
||
# - 已登记意图委托 handle_intent（P0/P1 直通，P2/P3 经 harness 确认卡，绝不绕过门禁）；
|
||
# - 每次执行记录 TOOL 审计事件（actor/source/power），溯源 LLM 提议。
|
||
# 不能通过未登记 API 或直接 store 写入绕过：本模块是 gateway 聊天的唯一执行入口。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
from server.agent_core import harness
|
||
from server.agent_core.audit import write_audit
|
||
from server.contracts import AgentReply, IntentResult
|
||
|
||
# 只读意图白名单：不写世界状态，但允许 LLM 提议直接执行（分析/查询/生成建议）
|
||
_READONLY_INTENTS = frozenset({
|
||
"help", "guidance.next", "master.query", "plan.buckets", "plan.rccp",
|
||
"plan.feasibility", "plan.inventory", "plan.leveling", "plan.supply",
|
||
"plan.trace", "data.analyze", "knowledge.query", "scenario.compare",
|
||
"report.generate", "order.pool", "conflict.list", "flex.capacity",
|
||
"flex.compare", "flex.simulate_due", "skill.list",
|
||
})
|
||
|
||
|
||
def is_registered(intent_name: str) -> bool:
|
||
"""intent 是否已登记：写入类必须在 _POWER_MAP；只读类在白名单。"""
|
||
if intent_name in _READONLY_INTENTS:
|
||
return True
|
||
return harness.power_of(intent_name) != "P3" or intent_name in harness._POWER_MAP
|
||
|
||
|
||
def check_tool(store, intent: IntentResult, actor: str = "planner") -> AgentReply | None:
|
||
"""统一工具运行时校验：LLM 提议意图执行前的强制登记检查（矩阵 111 行）。
|
||
|
||
- 未登记 intent → 写 TOOL 审计（tool.denied/P3）并返回拒绝回复（绝不执行）；
|
||
- 已登记 intent → 写 TOOL 审计（tool.run）并返回 None（gateway 继续调用 handle_intent）。
|
||
返回 None 表示放行；返回 AgentReply 表示已拒绝，调用方应直接返回它。
|
||
"""
|
||
name = intent.intent
|
||
source = str(getattr(intent, "source", "") or "unknown")
|
||
if not is_registered(name):
|
||
write_audit(
|
||
store.data, store.next_id, actor=actor, category="TOOL",
|
||
action="tool.denied", target={"type": "INTENT", "id": name},
|
||
power="P3", rationale={"source": source, "reason": "unregistered-intent"},
|
||
result="DENIED",
|
||
)
|
||
store.save()
|
||
return AgentReply(
|
||
text=f"「{name}」不是已登记的操作（LLM 只有提议权，未登记意图不执行）。"
|
||
"可换一种说法，或先查看帮助。")
|
||
|
||
power = harness.power_of(name)
|
||
write_audit(
|
||
store.data, store.next_id, actor=actor, category="TOOL",
|
||
action="tool.run", target={"type": "INTENT", "id": name},
|
||
power=power, rationale={"source": source, "intent": name},
|
||
)
|
||
store.save()
|
||
return None
|
||
|
||
|
||
async def run_tool_async(store, session_id: str, intent: IntentResult, actor: str = "planner") -> AgentReply:
|
||
"""统一工具运行时：LLM 提议意图的异步执行入口（矩阵 111 行，供 gateway 直连端点复用）。
|
||
|
||
gateway 直连端点（如 /api/actions/scenario/apply）与 chat 共用本入口：
|
||
先 check_tool（未登记 → tool.denied 拒绝 + TOOL 审计；已登记 → tool.run 审计），
|
||
放行后委托 handle_intent（P0/P1 直通，P2/P3 经 harness 确认卡，绝不绕过门禁）。
|
||
"""
|
||
denied = check_tool(store, intent, actor=actor)
|
||
if denied is not None:
|
||
return denied
|
||
from server.aps_domain.workflow import handle_intent
|
||
return await handle_intent(store, session_id, intent, actor=actor)
|
||
|
||
|
||
def run_tool(store, session_id: str, intent: IntentResult, actor: str = "planner") -> AgentReply:
|
||
"""统一工具运行时：LLM 提议意图的执行入口（矩阵 111 行，同步便捷包装）。
|
||
|
||
委托 run_tool_async（校验放行后 handle_intent）；测试/非 async 直连场景使用。
|
||
"""
|
||
return asyncio.run(run_tool_async(store, session_id, intent, actor=actor))
|