"""Core ReAct (Reasoning + Acting) loop for AI agents. Implements a true ReAct loop that alternates between LLM reasoning and tool execution. Each step records the thought (LLM content), action (tool name), action_input (tool arguments), and observation (tool result). The loop terminates when: - The LLM returns a final response without tool calls (completed) - max_steps is reached (stopped_max_steps) - timeout is exceeded (stopped_timeout) - A permanent error occurs (stopped_error) Error handling uses ``ErrorCategory`` from ``app.core.error_codes``: - TRANSIENT → retry the LLM call (up to 3 retries per step) - PERMANENT → stop the loop immediately - PARTIAL → continue with partial results Usage:: from app.ai.agent_loop import run_react_loop result = await run_react_loop( agent_definition=agent, messages=[{"role": "user", "content": "Summarize recent emails"}], tools=tool_schemas, tool_registry=registry, db=db_session, tenant_id=tenant_id, user_id=user_id, ) print(result.final_content, result.total_cost_usd, result.steps_taken) """ from __future__ import annotations import asyncio import json import logging import time import uuid from dataclasses import dataclass, field from datetime import UTC, datetime from typing import TYPE_CHECKING, Any from app.ai.llm_client import llm_complete from app.core.error_codes import ErrorCategory, classify_exception if TYPE_CHECKING: from collections.abc import Callable from sqlalchemy.ext.asyncio import AsyncSession from app.ai.tool_registry import ToolRegistry logger = logging.getLogger(__name__) # Maximum retries for transient errors per LLM step _MAX_TRANSIENT_RETRIES = 3 # ────────────────────────────────────────────────────────────────────────── # Data structures # ────────────────────────────────────────────────────────────────────────── @dataclass class ReActStep: """A single step in the ReAct loop (Thought → Action → Observation).""" step_number: int thought: str # LLM content before tool calls action: str | None # Tool name (None if final response) action_input: dict[str, Any] | None # Tool arguments observation: str | None # Tool result cost_usd: float timestamp: str # ISO format @dataclass class ReActResult: """Final result of the ReAct loop.""" final_content: str steps: list[ReActStep] = field(default_factory=list) total_cost_usd: float = 0.0 steps_taken: int = 0 status: str = "completed" # completed | stopped_max_steps | stopped_timeout | stopped_error error: str | None = None # ────────────────────────────────────────────────────────────────────────── # Core loop # ────────────────────────────────────────────────────────────────────────── def _extract_tool_calls(raw_response: Any) -> list[dict[str, Any]]: """Extract tool calls from a LiteLLM raw response. Returns a list of dicts with keys: ``id``, ``name``, ``arguments``. """ tool_calls: list[dict[str, Any]] = [] try: msg = raw_response.choices[0].message if hasattr(msg, "tool_calls") and msg.tool_calls: for tc in msg.tool_calls: tool_calls.append({ "id": tc.id or "", "name": tc.function.name if tc.function else "", "arguments": tc.function.arguments if tc.function and tc.function.arguments else "{}", }) except (AttributeError, IndexError, TypeError) as exc: logger.debug("Failed to extract tool calls from response: %s", exc) return tool_calls async def _execute_tool( tool_registry: ToolRegistry, tool_name: str, arguments: dict[str, Any], context: dict[str, Any], ) -> str: """Execute a single tool call via the registry. Returns the tool result as a string, or an error message. Note: access control (allowlist + required_permission) is enforced by ``_check_tool_access`` in ``run_react_loop`` BEFORE any execution path (dry-run, approval, execute) is reached. """ tool = tool_registry.get(tool_name) if tool is None: return f"Error: Tool '{tool_name}' not found" try: result = await tool.handler(arguments=arguments, context=context) return result if isinstance(result, str) else json.dumps(result) except Exception as exc: logger.exception("Tool '%s' execution failed", tool_name) return f"Error: {exc}" def _filter_observation(observation: str) -> str: """F14 (Astra P1): sanitize a tool observation before it re-enters the LLM conversation. Tool responses are raw data (CRM records, mail payloads, settings) and may contain sensitive fields (smtp_password, api keys, ...). The data policy runs BEFORE the loop — observations arise INSIDE it and used to reach the provider verbatim. Parse JSON observations and strip sensitive fields (same SENSITIVE_FIELDS set as the data policy). """ stripped = observation.strip() if not stripped.startswith(("{", "[")): return observation try: import json parsed = json.loads(stripped) except (json.JSONDecodeError, ValueError): return observation from app.core.sensitive_data import SENSITIVE_FIELDS sensitive_names: set[str] = set() for fields in SENSITIVE_FIELDS.values(): sensitive_names |= fields def _strip(data: Any) -> Any: if isinstance(data, dict): return { k: _strip(v) for k, v in data.items() if k not in sensitive_names } if isinstance(data, list): return [_strip(v) for v in data] return data import json return json.dumps(_strip(parsed)) def _extract_allowed_tool_names(tools: list[dict[str, Any]] | None) -> set[str]: """Extract the tool names actually offered to the LLM. Always returns a set (possibly empty) — empty means no tools were offered, so the caller fails closed on ANY tool call. """ if not tools: return set() names: set[str] = set() for t in tools: fn = (t or {}).get("function") or {} name = fn.get("name") if name: names.add(str(name)) return names def _normalize_permissions(ctx: dict[str, Any] | None) -> dict[str, Any]: """Normalize a user/agent context into the resolved-permissions shape expected by ``check_permission`` (permissions / denied / is_system_admin). Session user contexts carry ``denied_permissions`` while resolved permission dicts use ``denied`` — both are accepted here. """ if not isinstance(ctx, dict): return {"permissions": [], "denied": [], "is_system_admin": False} return { "permissions": list(ctx.get("permissions", []) or []), "denied": list(ctx.get("denied", ctx.get("denied_permissions", [])) or []), "is_system_admin": bool(ctx.get("is_system_admin", False)), } def _check_tool_access( tool_registry: ToolRegistry, tool_name: str, allowed_tools: set[str] | None, user_permissions: dict[str, Any] | None, ) -> str | None: """F01 guard: enforce allowlist + required_permission before execution. Must run before EVERY execution path (dry-run, approval, execute). Returns None when access is granted, otherwise an error observation. Checks (fail-closed): 1. Allowlist — the tool must be among the schemas actually offered to the LLM. A hallucinated/injected tool name never reaches a handler. 2. required_permission — when the tool declares one, the acting user's CURRENT permissions must grant it. Without a permission context the call is rejected (deny list first, system admin bypass). """ tool = tool_registry.get(tool_name) if tool is None: return None # "not found" is handled by _execute_tool # 1. Allowlist: only tools offered to the LLM may run. if allowed_tools is not None and tool_name not in allowed_tools: logger.warning( "F01 guard: tool '%s' is registered but NOT offered to this agent — rejected", tool_name, ) return f"Error: Tool '{tool_name}' is not available to this agent" # 2. required_permission: enforce against the user's CURRENT permissions. required = getattr(tool, "required_permission", None) if isinstance(required, str) and required: resolved = _normalize_permissions(user_permissions) from app.core.permissions import check_permission if not check_permission(resolved, required): logger.warning( "F01 guard: tool '%s' requires '%s' which the acting user lacks — rejected", tool_name, required, ) return ( f"Error: Permission '{required}' required for tool '{tool_name}' " "and not granted to the acting user" ) return None async def run_react_loop( agent_definition: Any, # AgentDefinition from automation models messages: list[dict[str, Any]], tools: list[dict[str, Any]], tool_registry: ToolRegistry, db: AsyncSession, tenant_id: uuid.UUID, user_id: uuid.UUID, agent_run_id: uuid.UUID | None = None, max_steps: int = 20, timeout_seconds: int = 300, trace_id: str | None = None, on_step: Callable | None = None, dry_run: bool = False, require_approval: bool = False, approval_tools: list[str] | None = None, user_permissions: dict[str, Any] | None = None, ) -> ReActResult: """Execute a ReAct loop: LLM reasoning → tool execution → repeat. Args: agent_definition: AgentDefinition with llm_model, system_prompt, etc. messages: Initial chat messages (without system prompt). tools: OpenAI-format tool schemas for function calling. tool_registry: ToolRegistry instance for tool execution. db: Async DB session. tenant_id: Tenant ID for multi-tenancy. user_id: User ID for permission context. agent_run_id: Optional AgentRun ID for step persistence. max_steps: Maximum loop iterations (default 20). timeout_seconds: Overall timeout (default 300). trace_id: Optional trace ID for correlation. on_step: Optional async callback fired after each step. dry_run: When True, tool execution is simulated — tool handlers are NOT called. A mock result is returned instead and steps are still logged with real LLM cost. Returns: ReActResult with final content, steps, cost, and status. """ from app.core.hooks import do_action result = ReActResult(final_content="", status="completed") start_time = time.monotonic() # Build LLM parameters from agent definition litellm_model = getattr(agent_definition, "llm_model", None) or "gpt-4o" system_prompt = getattr(agent_definition, "system_prompt", "") or "You are a helpful AI assistant." api_key = getattr(agent_definition, "api_key", None) api_base = getattr(agent_definition, "api_base", None) provider = getattr(agent_definition, "provider", None) max_tokens = getattr(agent_definition, "max_tokens", None) or 1000 # Build the full message list with system prompt prepended full_messages: list[dict[str, Any]] = [ {"role": "system", "content": system_prompt}, *messages, ] tool_context: dict[str, Any] = { "tenant_id": str(tenant_id), "user_id": str(user_id), "db": db, "agent_name": getattr(agent_definition, "name", "Agent"), } # F01 (Astra P0): allowlist — only tools actually offered to the LLM # may ever execute. Empty set = no tools offered = every call rejected. allowed_tool_names: set[str] = _extract_allowed_tool_names(tools) # Audit helper — records every tool call in the audit log. async def _audit_tool_call( step_number: int, tool_name: str, arguments: dict[str, Any], result: str, cost_usd: float, ) -> None: """Create an audit log entry for a single tool call.""" try: from app.core.audit import log_audit await log_audit( db=db, tenant_id=tenant_id, user_id=user_id, action="agent.tool_call", entity_type="agent_run", entity_id=agent_run_id, details={ "agent_run_id": str(agent_run_id) if agent_run_id else None, "step_number": step_number, "tool_name": tool_name, "arguments": arguments, "result": result[:2000], "cost_usd": cost_usd, "dry_run": dry_run, }, ) except Exception: logger.exception("Failed to audit tool call '%s'", tool_name) for step_num in range(1, max_steps + 1): # ── Timeout check ── elapsed = time.monotonic() - start_time if elapsed >= timeout_seconds: result.status = "stopped_timeout" result.error = f"Timeout after {elapsed:.1f}s (limit {timeout_seconds}s)" logger.warning("ReAct loop timed out at step %d: %s", step_num, result.error) break # ── LLM call with transient retry ── llm_result: dict[str, Any] | None = None last_error: str | None = None for retry in range(_MAX_TRANSIENT_RETRIES + 1): try: llm_result = await llm_complete( model=litellm_model, messages=full_messages, tools=tools if tools else None, temperature=0.3, max_tokens=max_tokens, api_key=api_key, api_base=api_base, provider=provider, trace_id=trace_id, tenant_id=tenant_id, db=db, ) break except Exception as exc: last_error = str(exc) category = classify_exception(exc) if category == ErrorCategory.PERMANENT: result.status = "stopped_error" result.error = f"Permanent error at step {step_num}: {exc}" logger.error("ReAct loop permanent error: %s", result.error) return result if category == ErrorCategory.TRANSIENT and retry < _MAX_TRANSIENT_RETRIES: backoff = 2 ** retry logger.warning( "Transient error at step %d (retry %d/%d): %s — retrying in %ds", step_num, retry + 1, _MAX_TRANSIENT_RETRIES, exc, backoff, ) await asyncio.sleep(backoff) continue # PARTIAL or exhausted retries if category == ErrorCategory.PARTIAL: logger.warning("Partial error at step %d: %s — continuing", step_num, exc) last_error = str(exc) break # Exhausted transient retries result.status = "stopped_error" result.error = f"Error after {retry + 1} retries at step {step_num}: {exc}" logger.error("ReAct loop error: %s", result.error) return result if llm_result is None: result.status = "stopped_error" result.error = f"LLM call failed at step {step_num}: {last_error}" return result # ── Extract response data ── content = llm_result.get("content", "") cost_usd = llm_result.get("cost_usd", 0.0) result.total_cost_usd += cost_usd tool_calls = _extract_tool_calls(llm_result.get("raw_response")) # ── No tool calls → final response ── if not tool_calls: step = ReActStep( step_number=step_num, thought=content, action=None, action_input=None, observation=None, cost_usd=cost_usd, timestamp=datetime.now(UTC).isoformat(), ) result.steps.append(step) result.final_content = content result.steps_taken = step_num # Fire hook await do_action( "agent.step", agent_id=str(getattr(agent_definition, "id", "")), step_number=step_num, thought=content, action=None, observation=None, cost_usd=cost_usd, agent_run_id=str(agent_run_id) if agent_run_id else None, trace_id=trace_id, ) # Callback if on_step: try: await on_step(step) except Exception: logger.debug("on_step callback failed", exc_info=True) break # ── Execute tool calls ── # Append assistant message with tool calls to conversation full_messages.append({ "role": "assistant", "content": content, "tool_calls": [ { "id": tc["id"], "type": "function", "function": {"name": tc["name"], "arguments": tc["arguments"]}, } for tc in tool_calls ], }) # Execute each tool call and collect observations observations: list[str] = [] for tc in tool_calls: tool_name = tc["name"] try: args = json.loads(tc["arguments"]) if tc["arguments"] else {} except json.JSONDecodeError: args = {} logger.warning("Invalid JSON arguments for tool '%s': %s", tool_name, tc["arguments"]) # F01 (Astra P0): enforce allowlist + required_permission before # every execution path (dry-run, approval, execute). Fail-closed: # a hallucinated or injected tool name never reaches a handler. guard_error = _check_tool_access( tool_registry, tool_name, allowed_tool_names, user_permissions ) if guard_error is not None: observation = guard_error elif dry_run: observation = json.dumps( { "dry_run": True, "would_execute": tool_name, "arguments": args, } ) elif require_approval and (approval_tools is None or tool_name in (approval_tools or [])): # I-APPR-LOOP: Human-in-the-Loop Approval # Create an ApprovalRequest and pause the loop try: from app.core.approval import create_approval_request pass # agent_workstream removed approval = await create_approval_request( db=db, tenant_id=tenant_id, entity_type="agent_run", entity_id=agent_run_id or uuid.uuid4(), action=f"tool:{tool_name}", requested_by=user_id, requested_by_type="agent", ) # Post approval request to Communication (I-WORK-HANDOFF) if agent_run_id: try: from app.plugins.builtins.contracts import get_contract_registry komm = get_contract_registry().get_contract("kommunikation") if komm: agent_id = getattr(agent_definition, "id", uuid.uuid4()) room_title = f"Agent: {getattr(agent_definition, 'name', 'Agent')}" conv_id = await komm.find_locked_room_id( db=db, tenant_id=tenant_id, plugin_name="automation", title=room_title, ) if conv_id: await komm.send_message( db=db, tenant_id=tenant_id, conversation_id=conv_id, sender_id=agent_id, sender_type="agent", content=f"Approval required for tool '{tool_name}'", content_format="text", blocks=[ { "block_type": "approval_request", "block_data": { "title": f"Approval: {tool_name}", "description": f"Agent wants to execute tool '{tool_name}' with arguments: {json.dumps(args)[:300]}", "approval_id": str(approval.id), "status": "pending", }, "sort_order": 0, } ], metadata={"approval_id": str(approval.id), "agent_run_id": str(agent_run_id)}, ) except Exception: logger.warning("Failed to post approval request to communication", exc_info=True) # Pause the loop — return with waiting_for_approval status result.status = "waiting_for_approval" result.error = f"Tool '{tool_name}' requires human approval (request_id: {approval.id})" result.steps_taken = step_num result.final_content = f"I need approval to execute tool '{tool_name}'. Approval request {approval.id} has been created." logger.info("Agent loop paused for approval on tool '%s' (request: %s)", tool_name, approval.id) return result except Exception as e: logger.warning("Failed to create approval request for tool '%s': %s", tool_name, e) observation = json.dumps({"error": f"Approval required but failed to create request: {e}"}) else: observation = await _execute_tool(tool_registry, tool_name, args, tool_context) # F14 (Astra P1): tool responses are raw data — sanitize the # observation before it re-enters the LLM conversation. observation = _filter_observation(observation) observations.append(observation) # Audit every tool call (real or simulated) await _audit_tool_call( step_number=step_num, tool_name=tool_name, arguments=args, result=observation, cost_usd=cost_usd / len(tool_calls) if tool_calls else cost_usd, ) # Feed tool result back into conversation full_messages.append({ "role": "tool", "tool_call_id": tc["id"], "content": observation, }) # Record step step = ReActStep( step_number=step_num, thought=content, action=tool_name, action_input=args, observation=observation, cost_usd=cost_usd / len(tool_calls) if tool_calls else cost_usd, timestamp=datetime.now(UTC).isoformat(), ) result.steps.append(step) # Fire hook await do_action( "agent.step", agent_id=str(getattr(agent_definition, "id", "")), step_number=step_num, thought=content, action=tool_name, observation=observation, cost_usd=cost_usd, agent_run_id=str(agent_run_id) if agent_run_id else None, trace_id=trace_id, ) # Callback if on_step: try: await on_step(step) except Exception: logger.debug("on_step callback failed", exc_info=True) result.steps_taken = step_num # If this was the last allowed step, stop gracefully if step_num >= max_steps: result.status = "stopped_max_steps" result.error = f"Reached max_steps limit ({max_steps})" result.final_content = content logger.warning("ReAct loop stopped at max_steps=%d", max_steps) break # If loop completed without a final response (e.g. all steps had tool calls) if not result.final_content and result.steps: result.final_content = result.steps[-1].thought or "" if result.status == "completed" and not result.final_content: result.final_content = "" return result