"""FlowDeck — AgentEngine: ReAct orchestrator (v4.10.0). `objective → comprehension → context → reasoning ↔ action → result`. The engine drives the LLM (which only emits tool intentions), gates each call through PermissionManager, executes it via ToolRegistry, journals every action to `agent_actions` with an undo snapshot, and yields a stream of SSE events so the UI can render reasoning + actions live. """ from __future__ import annotations import asyncio import json import logging from datetime import datetime from app.config import settings from app.db import get_conn from app.services.llm_client import LLMClient from app.services.permission_manager import PermissionManager from app.services.context_builder import ContextBuilder from app.services.tool_registry import ToolRegistry logger = logging.getLogger(__name__) MAX_ITERATIONS = 12 class AgentEngine: def __init__(self, user_id: int, workspace_id: int | None = None, llm: LLMClient | None = None): self.user_id = user_id self.workspace_id = workspace_id self.llm = llm or LLMClient() self.tools = ToolRegistry() self.ctx = ContextBuilder(user_id, workspace_id) self.perms = PermissionManager(user_id) self._tokens = 0 # ── Helpers ── def _load_agent(self, conversation_id: int) -> dict: with get_conn() as conn: row = conn.execute( "SELECT a.* FROM agents a JOIN agent_conversations c ON c.agent_id=a.id WHERE c.id=?", (conversation_id,), ).fetchone() if not row: row = {"id": None, "name": "FlowDeck Agent", "icon": "🤖", "agent_type": "personal", "system_instructions": "", "model": settings.llm_model, "scope_json": "{}", "approval_mode": "auto"} return dict(row) def _build_system_prompt(self, agent: dict, skill: dict | None = None) -> str: lines = [ "Tu es FlowDeck Agent, un agent IA qui réalise des tâches dans le workspace FlowDeck.", "Tu réfléchis (reasoning) puis agis en appelant les outils disponibles.", "Appelle UN ou PLUSIEURS outils pour atteindre l'objectif, puis conclus avec une réponse finale.", "N'invente jamais d'IDs : utilise ceux fournis dans le contexte.", f"Workspace courant : {self.workspace_id}.", ] if agent.get("system_instructions"): lines.append(f"\nInstructions de l'agent {agent.get('name','')}:\n{agent['system_instructions']}") if skill: lines.append(f"\nSkill appliquée « {skill.get('name','')} »:\n{skill.get('prompt_template','')}") return "\n".join(lines) # ── Main run (async generator of SSE events) ── async def run(self, conversation_id: int, objective: str, *, model: str | None = None, mentions: list[str] | None = None, files: list[dict] | None = None, skill_id: int | None = None): agent = self._load_agent(conversation_id) scope = json.loads(agent.get("scope_json") or "{}") approval_mode = agent.get("approval_mode") or "auto" model = model or agent.get("model") or settings.llm_model skill = self._load_skill(skill_id, scope) if skill_id else None system = self._build_system_prompt(agent, skill) context = self.ctx.build(mentions=mentions, files=files) messages = [ {"role": "system", "content": system}, {"role": "user", "content": f"{objective}\n\n# Contexte\n{context}"}, ] self._persist_message(conversation_id, "user", objective) self._update_conversation(conversation_id, status="running") tool_schema = self.tools.schema(scope) final_text = None try: for step in range(settings.agent_max_iterations or MAX_ITERATIONS): if self._tokens >= settings.agent_max_tokens_budget: yield self._event("error", {"message": "Budget de tokens dépassé"}) break response = await asyncio.wait_for( self.llm.complete(messages, model=model, tools=tool_schema, stream=True), timeout=settings.agent_run_timeout_seconds, ) self._tokens += response.usage.get("total_tokens", 0) or 0 if response.text and response.text.strip(): yield self._event("reasoning", {"content": response.text}) messages.append({"role": "assistant", "content": response.text}) if not response.tool_calls: final_text = response.text or self._no_tool_message(response) yield self._event("final", {"content": final_text}) break for call in response.tool_calls: tool, args = call["name"], call.get("arguments") or {} try: self.perms.assert_can(tool, args, self.workspace_id, approval_mode) except Exception as exc: # permission / approval guard detail = self._exc_detail(exc) yield self._event("action", {"tool": tool, "status": "error", "detail": detail}) self._log_action(conversation_id, tool, args, {}, "error", detail=detail) continue result = await self.tools.execute(tool, args, user_id=self.user_id) if result.status == "success": yield self._event("action", { "tool": tool, "status": result.status, "target_type": result.target_type, "target_id": result.target_id, "message": result.message, }) self._log_action(conversation_id, tool, args, result.data, "success", target_type=result.target_type, target_id=result.target_id, undo=result.undo) messages.append({ "role": "tool", "name": tool, "content": json.dumps({"status": "ok", "result": result.data, "target_id": result.target_id}, ensure_ascii=False), }) else: yield self._event("action", {"tool": tool, "status": "error", "detail": result.message}) self._log_action(conversation_id, tool, args, {}, "error", detail=result.message) messages.append({ "role": "tool", "name": tool, "content": json.dumps({"status": "error", "message": result.message}, ensure_ascii=False), }) if final_text is None: final_text = "Objectif traité. Consultez le journal des actions pour le détail." yield self._event("final", {"content": final_text}) self._persist_message(conversation_id, "assistant", final_text, model=model, tokens=self._tokens) except Exception as exc: # noqa: BLE001 logger.exception("AgentEngine run failed") yield self._event("error", {"message": f"Erreur interne: {exc}"}) finally: self._update_conversation(conversation_id, status="idle") # ── Skills ── def _load_skill(self, skill_id: int, scope: dict | None) -> dict | None: with get_conn() as conn: row = conn.execute("SELECT * FROM agent_skills WHERE id=?", (skill_id,)).fetchone() if not row: return None skill = dict(row) allowed = json.loads(skill.get("allowed_tools_json") or "[]") if allowed: skill["prompt_template"] = (skill.get("prompt_template") or "") + \ "\nOutils autorisés: " + ", ".join(allowed) return skill # ── Audit & persistence ── def _log_action(self, conversation_id, tool, args, result, status, *, target_type="", target_id=None, undo=None, detail=""): with get_conn() as conn: conn.execute( """INSERT INTO agent_actions (conversation_id, tool_name, target_type, target_id, payload_json, result_json, status, undo_snapshot_json, executed_by) VALUES (?,?,?,?,?,?,?,?,?)""", (conversation_id, tool, target_type, str(target_id) if target_id is not None else None, json.dumps(args, ensure_ascii=False), json.dumps(result, ensure_ascii=False, default=str), status, json.dumps(undo or {}, ensure_ascii=False), self.user_id), ) conn.commit() def _persist_message(self, conversation_id, role, content, *, model="", tokens=0): with get_conn() as conn: conn.execute( "INSERT INTO agent_messages (conversation_id, role, content, model, tokens_used) VALUES (?,?,?,?,?)", (conversation_id, role, content, model, tokens), ) conn.commit() def _update_conversation(self, conversation_id, *, status=None, title=None): with get_conn() as conn: sets, params = ["updated_at=CURRENT_TIMESTAMP"], [] if status: sets.append("status=?") params.append(status) if title: sets.append("title=?") params.append(title) params.append(conversation_id) conn.execute(f"UPDATE agent_conversations SET {', '.join(sets)} WHERE id=?", params) conn.commit() # ── Misc ── @staticmethod def _event(etype: str, data: dict) -> dict: return {"type": etype, **data} @staticmethod def _no_tool_message(response) -> str: return "Je n'ai pas d'action à proposer pour cet objectif. Posez-moi une question plus précise ou demandez-moi de créer un élément." @staticmethod def _exc_detail(exc: Exception) -> str: detail = getattr(exc, "detail", None) return detail if isinstance(detail, str) else str(exc) def undo_action(action_id: int) -> bool: """Reverse a single agent action using its stored undo snapshot. Returns True on success. Marks the action row `reverted`. """ with get_conn() as conn: action = conn.execute("SELECT * FROM agent_actions WHERE id=?", (action_id,)).fetchone() if not action: raise ValueError(f"Action #{action_id} introuvable") undo = json.loads(action["undo_snapshot_json"] or "{}") op, table, rid = undo.get("action"), undo.get("table"), undo.get("id") if not op or not table or rid is None: raise ValueError(f"Action #{action_id} n'a pas de snapshot annulable") if op == "delete": conn.execute(f"DELETE FROM {table} WHERE id=?", (rid,)) elif op == "insert": snapshot = undo.get("snapshot") if not snapshot: raise ValueError("Snapshot manquant pour insert") cols = ", ".join(snapshot.keys()) ph = ", ".join("?" for _ in snapshot) conn.execute(f"INSERT INTO {table} ({cols}) VALUES ({ph})", list(snapshot.values())) elif op == "update": snapshot = undo.get("snapshot") if table == "collection_pages": conn.execute( "UPDATE collection_pages SET property_values_json=?, title=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (json.dumps(snapshot, ensure_ascii=False), undo.get("title", ""), rid), ) elif table == "pages": conn.execute("UPDATE pages SET content=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (snapshot, rid)) else: raise ValueError(f"Table non gérée pour rollback: {table}") else: raise ValueError(f"Opération de rollback inconnue: {op}") conn.execute("UPDATE agent_actions SET status='reverted' WHERE id=?", (action_id,)) conn.commit() return True