"""Deterministic post-approval tool executor for the MAF roundtable. After a human approves an external-write action, this executor calls the approved DeerFlow tool **directly** with the **decrypted original params** — no model free-will, no re-decision, no raw params in any prompt or checkpoint. It is invoked by ``WorkflowRunner.resume()`` instead of re-running an LLM seat turn (which would let the model pick different arguments or a different tool). Design — direct tool invocation ------------------------------- 1. **Atomic occupy** (``approved → running``): only the caller whose conditional UPDATE affects one row wins the right to execute. Concurrent ``resume()`` or post-restart replay finds the row already ``running`` or terminal and bails out — at-most-once execution is guaranteed at the DB level, not by locking. 2. **Direct tool call**: decrypts approved params (server-only), looks up the tool in DeerFlow's registry, builds a minimal ToolRuntime from the thread's persisted state (sandbox/thread_data), and calls ``tool.func()`` or ``tool.ainvoke()`` directly. The LLM is never involved. 3. **Authority**: the approved ``action_id`` is carried in the runtime context so the WriteGuard middleware can validate both tool name AND action_id as defense-in-depth (though the LLM is bypassed, so this is purely a secondary check). 4. **Outcome recording**: ``succeeded`` / ``failed`` / ``unknown`` are recorded via the state-machine-constrained ``record_terminal_result`` (only ``running → terminal``). ``unknown`` prohibits auto-retry. Security: the decrypted params are read server-side from the encrypted secret and never leave this process. They do NOT appear in prompts, checkpoints, events, or API responses. """ from __future__ import annotations import logging from typing import Any from app.maf_roundtable.action_coordinator import ActionCoordinator from app.maf_roundtable.sanitize import sanitize logger = logging.getLogger(__name__) class ToolExecutor: """Executes an approved external-write action deterministically. Constructed per-resume from the ``WorkflowRunner``; takes the already-built adapter + coordinator + runtime client so it reuses identity / sandbox / middleware wiring. """ def __init__( self, *, adapter: Any, action_coordinator: ActionCoordinator, user_id: str, ) -> None: self._adapter = adapter self._coordinator = action_coordinator self._user_id = user_id async def execute( self, *, action_id: str, session_id: str, run_id: str, agent_id: str, thread_id: str, approved_tool: str, capability_snapshot: dict[str, Any] | None = None, write_policy: str = "confirm_high_risk", model_name: str | None = None, excluded_tools: list[str] | None = None, ) -> dict[str, Any]: """Execute the approved tool deterministically and record the outcome. Returns:: {"status": "succeeded"|"failed"|"skipped"|"unknown", ...} ``skipped`` means another caller already occupied/finished the action (at-most-once) — the caller should return the existing result. The approved params are recovered **server-side** from the encrypted action secret *after* the atomic occupy. If the secret is missing or cannot be decrypted, ``ActionSecretUnavailableError`` is raised by the coordinator and turned into a terminal ``failed`` outcome here — the tool is NEVER called with empty params (no ``params or {}`` fallback), so the external side effect happens exactly zero times. The actual call is delegated to the harness ``execute_approved_tool`` (real ``ToolNode`` / ``ToolRuntime`` / middleware chain), rebuilt from the seat's frozen capability snapshot / model / ``excluded_tools``. The stored action record is passed so the harness can verify run/session/ agent/tool consistency before executing. """ # 1. Atomic occupy — only the winner ("claimed") proceeds to execute. # A lost claim reflects the action's ACTUAL state (in_progress / # terminal) rather than a blanket "skip", so the caller can keep the # parent run running while another executor finishes it. from deerflow.persistence.maf_roundtable.actions import ( MafActionStateMachineError, ) try: claim = await self._coordinator.occupy_for_execution(action_id) except MafActionStateMachineError as exc: logger.error( "ToolExecutor: action %s cannot be claimed: %s", action_id, exc, ) return {"status": "failed", "error": f"action state machine error: {exc}"} if claim == "target_busy": logger.info("ToolExecutor: target lease is held for action %s", action_id) return {"status": "waiting_target_lease", "error": "target lease is held by another action"} if claim == "target_reconciliation_required": logger.warning("ToolExecutor: expired target lease requires reconciliation for %s", action_id) return { "status": "unknown", "error": "target lease expired during another action; reconciliation required", } if claim != "claimed": existing = await self._coordinator._store.actions.get_action(action_id) # noqa: SLF001 if claim == "in_progress": logger.info( "ToolExecutor: action %s already running elsewhere (in_progress)", action_id, ) return {"status": "in_progress", "action": existing} logger.info( "ToolExecutor: action %s already terminal (%s) — at-most-once", action_id, claim, ) return {"status": claim, "action": existing} if not approved_tool: await self._record_result_or_log( action_id, status="failed", result={"error": "no approved tool name — cannot execute"} ) return {"status": "failed", "error": "no approved tool name"} # 2. Recover the approved params from the server-only encrypted secret. # A missing/undecryptable/invalid secret is a HARD failure: record # the action failed and never invoke the tool (execution count == 0). from app.maf_roundtable.action_coordinator import ActionSecretUnavailableError try: params = await self._coordinator.get_approved_action_params(action_id) except ActionSecretUnavailableError as exc: safe_msg = str(sanitize(exc.detail))[:500] logger.error( "ToolExecutor: action %s params unavailable (%s): %s", action_id, exc.reason, safe_msg, ) persisted = await self._record_result_or_log( action_id, status="failed", result={"error": f"action secret unavailable: {exc.reason}"}, ) return { "status": "failed" if persisted else "unknown", "error": safe_msg if persisted else "action result persistence failed", } # 3. Read the stored action record (authoritative turn_index + the # run/session/agent/tool the harness will verify against). action_record = await self._coordinator._store.actions.get_action(action_id) # noqa: SLF001 turn_index = int((action_record or {}).get("turn_index") or 0) # 4. Execute via the real DeerFlow tool path — no LLM, no prompt # injection. The raw params are read from the encrypted secret and # passed straight to the tool; they never enter any model context. from app.maf_roundtable.runtime_client import ExecutionPrincipal principal = ExecutionPrincipal( user_id=self._user_id, session_id=session_id, run_id=run_id, ) client = self._adapter._client # noqa: SLF001 — DeerFlowRuntimeClient try: outcome = await client.execute_tool_directly( thread_id=thread_id, agent_id=agent_id, tool_name=approved_tool, params=params, principal=principal, turn_index=turn_index, capability_snapshot=capability_snapshot, write_policy=write_policy, model_name=model_name, excluded_tools=excluded_tools, approved_action_id=action_id, action_record=action_record, ) except Exception as exc: safe_msg = str(sanitize(str(exc)))[:500] logger.error( "ToolExecutor: direct tool call '%s' raised: %s", approved_tool, safe_msg, ) # An exception can arrive after the remote request was accepted # (network disconnect / timeout), so fail closed as ``unknown``. persisted = await self._record_result_or_log( action_id, status="unknown", result={"error": "tool_transport_exception"} ) return { "status": "unknown", "error": safe_msg if persisted else "action result persistence failed", } # 5. Record terminal result (state-machine constrained: only # running → terminal; a lost race is a no-op). outcome_status = outcome.get("status") if outcome_status == "succeeded": tool_result = outcome.get("result") external_object_id = ( str(tool_result.get("external_object_id")) if isinstance(tool_result, dict) and tool_result.get("external_object_id") else None ) persisted = await self._record_result_or_log( action_id, status="succeeded", result={"tool_result": tool_result}, external_object_id=external_object_id, ) if not persisted: # The external side effect succeeded but its local terminal # record did not. Never retry; the retained running lease is # later reconciled to unknown by recovery. return {"status": "unknown", "error": "action result persistence failed"} return {"status": "succeeded", "tool_result": tool_result} elif outcome_status == "unknown": # Inconclusive external result (timeout / disconnect / gateway): # record ``unknown`` — never auto-retry, a human must reconcile. error = outcome.get("error", "inconclusive external result") persisted = await self._record_result_or_log( action_id, status="unknown", result={"error": error}, ) return {"status": "unknown", "error": error if persisted else "action result persistence failed"} else: error = outcome.get("error", "tool execution failed") persisted = await self._record_result_or_log( action_id, status="failed", result={"error": error}, ) return { "status": "failed" if persisted else "unknown", "error": error if persisted else "action result persistence failed", } async def _record_result_or_log( self, action_id: str, *, status: str, result: dict[str, Any], external_object_id: str | None = None, ) -> bool: """Persist a terminal outcome without opening a retry path after I/O. If this write fails after a real tool reports success, returning an ``unknown`` outcome lets the parent run stop for reconciliation while the still-held lease prevents a replay of the external call. """ try: await self._coordinator.record_result( action_id, status=status, result=result, external_object_id=external_object_id, ) except Exception: # noqa: BLE001 logger.exception( "ToolExecutor: could not persist terminal result for action %s", action_id ) return False return True async def record_unknown(self, action_id: str, *, reason: str) -> None: """Record an ``unknown`` outcome (network timeout / inconclusive). ``unknown`` prohibits auto-retry — the action stays in ``unknown`` until a human reconciles it. """ logger.warning( "ToolExecutor: action %s outcome unknown — %s (no auto-retry)", action_id, reason, ) await self._record_result_or_log( action_id, status="unknown", result={"reason": reason} )