""" 将 Python 子进程的逐次调用接回当前 Agent 的真实工具链。 """ import json import uuid from collections.abc import Awaitable, Callable from dataclasses import replace from typing import Any, cast from langchain.agents.middleware.types import AgentMiddleware, ToolCallRequest from langchain_core.messages import AIMessage, ToolMessage from app.agent.code.authority import CURRENT_CELL, RPC_RESULT_KEY, CellAuthority, allowed_tools from app.agent.middleware.policy import AgentPolicyMiddleware from app.agent.policy.contracts import ExecutionOutcome from app.agent.terminal.ownership import TerminalScope, current_terminal_scope from app.agent.tools.impl.execute_code import ExecuteCodeTool from app.agent.tools.result import inspect_tool_result class CodeExecutionMiddleware(AgentMiddleware): """ 外层绑定执行窗口;内层调用保留策略、Skill、日志和原始结果归档行为。 """ def __init__(self, policy: AgentPolicyMiddleware) -> None: """ 共享同一个宿主策略实例,不复制权限或跳过当前身份事实源。 """ self.policy = policy self.selections: dict[TerminalScope, dict[str, dict[str, Any]]] = {} async def awrap_tool_call(self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[Any]]) -> Any: """ 只为模型真实请求中已启用的原生 execute_code 绑定当前窗口,默认直接执行。 """ if type(request.tool) is not ExecuteCodeTool: return await handler(request) scope = current_terminal_scope() selections = self.selections.get(scope, {}) if scope is not None else {} tools = selections.pop(str(request.tool_call['id']), None) if scope is None or tools is None: return ToolMessage(content='当前 Python 请求缺少宿主工具快照,未执行代码。', name='execute_code', tool_call_id=request.tool_call['id'], status='error') async def dispatch(tool: Any, arguments: dict[str, Any]) -> Any: """ 重入当前 handler,逐次经过 Skill 和工具门禁,而不是直接调用工具方法。 """ return await self._dispatch(request, handler, tool, arguments) authority = CellAuthority(scope, self.policy.context, tools, dispatch) token = CURRENT_CELL.set(authority) try: return await handler(request) finally: CURRENT_CELL.reset(token) await authority.retire() if not selections: self.selections.pop(scope, None) async def _dispatch(self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[Any]], tool: Any, arguments: dict[str, Any]) -> Any: """ 每次内部调用使用独立回执 ID;原文只交给 Python,外层模型只接收程序输出。 """ identifier = uuid.uuid4().hex raw: dict[str, str] = {} state = {**request.state, RPC_RESULT_KEY: raw} runtime = replace(request.runtime, tool_call_id=identifier, state=state) nested = ToolCallRequest(tool=tool, tool_call={'name': tool.name, 'args': arguments, 'id': identifier, 'type': 'tool_call'}, state=state, runtime=runtime) allowed, result = await self.policy.execute_tool_call( tool=tool, arguments=arguments, invocation_id=identifier, handler=lambda: handler(nested), enforce_decision=True, ) if not allowed: return {'success': False, 'error': str(result)} if not isinstance(result, ToolMessage): return {'success': False, 'error': '代码内调用未返回可读取的工具回执。'} content = raw.get(tool.name, result.content) if not isinstance(content, str): return {'success': False, 'error': '代码内调用只支持文本或 JSON 结果。'} if len(content.encode('utf-8')) > 16 * 1024 * 1024: return {'success': False, 'error': '单次工具结果超过16MB,请使用工具的分页或过滤参数缩小结果。'} try: payload = json.loads(content) except ValueError: payload = content if inspect_tool_result(result) is not ExecutionOutcome.SUCCEEDED: return {'success': False, 'error': payload} return payload class CodeCaptureMiddleware(AgentMiddleware): """ 筛选后、压缩前生成真实 helper 合同,新增说明必须计入最终请求预算。 """ def __init__(self, execution: CodeExecutionMiddleware) -> None: """ 捕获器与工具包装器共享短生命周期快照,不增加图节点或注册重复工具。 """ self.execution = execution async def awrap_model_call(self, request: Any, handler: Callable[[Any], Awaitable[Any]]) -> Any: """ 只为本次模型实际发出的代码调用保存快照,下一次请求替换同 owner 的旧快照。 """ scope = current_terminal_scope() if scope is not None: self.execution.selections.pop(scope, None) tools = allowed_tools(request.tools) signatures = [] for name, tool in tools.items(): schema = tool.args_schema.model_json_schema() parameters = [field if field in schema.get('required', []) else f'{field}=...' for field in schema.get('properties', {})] signatures.append(f'{name}({", ".join(parameters)})') description = '\nCurrent Python helpers: ' + ('; '.join(signatures) or 'none; only local Python computation is available.') request = request.override(tools=[tool.model_copy(update={'description': tool.description + description}) if type(tool) is ExecuteCodeTool else tool for tool in request.tools]) result = await handler(request) if scope is None or not any(type(tool) is ExecuteCodeTool for tool in request.tools): return result selections = {cast(str, call['id']): tools for message in result.result if isinstance(message, AIMessage) for call in message.tool_calls if call['name'] == 'execute_code'} if selections: self.execution.selections[scope] = selections return result