From 5da2ae01a0f695f57f9755de9a3f3ffbfc9d3bee Mon Sep 17 00:00:00 2001 From: "Gan, Jimmy" Date: Fri, 3 Apr 2026 22:30:51 +0800 Subject: [PATCH] fix: complete OPC system implementation - Add database connection retry logic (3 attempts, 5s delay) - Add missing database functions: get_task_by_id, get_agent_execution, approve_agent_execution, update_agent_config, update_execution_status - Complete agent execution approval workflow - Fix agent executor to handle approved executions - Fix inefficient task queries (use direct ID lookup) - Add error action handling in agent executor - Improve LiteLLM error handling with specific exception types - Reduce LiteLLM timeout from 60s to 30s - Fix execution status update to only set started_at once - Consolidate WebSocket datetime serialization with helper function - Implement update agent endpoint --- dashboard/backend/db/opc_db.py | 231 +++++++++++++++++- dashboard/backend/routers/opc_agents.py | 34 ++- dashboard/backend/routers/opc_ws.py | 41 ++-- dashboard/backend/services/agent_executor.py | 68 +++++- .../backend/services/agents/base_agent.py | 15 +- 5 files changed, 333 insertions(+), 56 deletions(-) diff --git a/dashboard/backend/db/opc_db.py b/dashboard/backend/db/opc_db.py index 8aef275..14a98ae 100644 --- a/dashboard/backend/db/opc_db.py +++ b/dashboard/backend/db/opc_db.py @@ -19,18 +19,38 @@ _pool: Optional[asyncpg.Pool] = None async def get_pool() -> asyncpg.Pool: - """Get or create database connection pool""" + """Get or create database connection pool with retry logic""" global _pool if _pool is None: - _pool = await asyncpg.create_pool( - host=DB_HOST, - port=DB_PORT, - database=DB_NAME, - user=DB_USER, - password=DB_PASSWORD, - min_size=2, - max_size=10, - ) + import asyncio + import logging + logger = logging.getLogger(__name__) + + max_retries = 3 + retry_delay = 5 + + for attempt in range(max_retries): + try: + logger.info(f"Attempting to connect to OPC database (attempt {attempt + 1}/{max_retries})") + _pool = await asyncpg.create_pool( + host=DB_HOST, + port=DB_PORT, + database=DB_NAME, + user=DB_USER, + password=DB_PASSWORD, + min_size=2, + max_size=10, + ) + logger.info("Successfully connected to OPC database") + break + except Exception as e: + logger.error(f"Failed to connect to OPC database (attempt {attempt + 1}/{max_retries}): {e}") + if attempt < max_retries - 1: + logger.info(f"Retrying in {retry_delay} seconds...") + await asyncio.sleep(retry_delay) + else: + logger.error("All connection attempts failed") + raise Exception(f"Failed to connect to OPC database after {max_retries} attempts: {e}") return _pool @@ -606,3 +626,194 @@ async def get_agent_executions( execution[json_field] = json.loads(execution[json_field]) executions.append(execution) return executions + + +async def get_task_by_id(task_id: int) -> Optional[Dict[str, Any]]: + """Get a single task by ID""" + pool = await get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT * FROM tasks WHERE id = $1", task_id) + if not row: + return None + + task = dict(row) + task["tags"] = json.loads(task["tags"]) if task["tags"] else [] + return task + + +async def get_agent_execution(execution_id: int) -> Optional[Dict[str, Any]]: + """Get a single agent execution by ID""" + pool = await get_pool() + async with pool.acquire() as conn: + row = await conn.fetchrow("SELECT * FROM agent_executions WHERE id = $1", execution_id) + if not row: + return None + + execution = dict(row) + for json_field in ["input_context", "output_result", "actions_proposed", "actions_executed"]: + if execution[json_field]: + execution[json_field] = json.loads(execution[json_field]) + return execution + + +async def approve_agent_execution( + execution_id: int, + approved: bool, + approved_by: str, + feedback: Optional[str] = None +) -> Dict[str, Any]: + """Approve or reject an agent execution""" + pool = await get_pool() + async with pool.acquire() as conn: + # Update execution with approval status + row = await conn.fetchrow(""" + UPDATE agent_executions + SET approved_by = $1, + approved_at = CURRENT_TIMESTAMP, + status = CASE WHEN $2 THEN 'approved' ELSE 'rejected' END + WHERE id = $3 + RETURNING * + """, approved_by, approved, execution_id) + + if not row: + raise ValueError(f"Execution {execution_id} not found") + + execution = dict(row) + for json_field in ["input_context", "output_result", "actions_proposed", "actions_executed"]: + if execution[json_field]: + execution[json_field] = json.loads(execution[json_field]) + + # If feedback provided, add to output_result + if feedback: + if not execution["output_result"]: + execution["output_result"] = {} + execution["output_result"]["approval_feedback"] = feedback + + await conn.execute(""" + UPDATE agent_executions + SET output_result = $1 + WHERE id = $2 + """, json.dumps(execution["output_result"]), execution_id) + + return execution + + +async def update_agent_config( + agent_id: str, + name: Optional[str] = None, + description: Optional[str] = None, + system_prompt: Optional[str] = None, + config: Optional[Dict[str, Any]] = None, + enabled: Optional[bool] = None +) -> Dict[str, Any]: + """Update agent configuration""" + pool = await get_pool() + async with pool.acquire() as conn: + # Build update query dynamically + updates = [] + params = [] + param_count = 0 + + if name is not None: + param_count += 1 + updates.append(f"name = ${param_count}") + params.append(name) + + if description is not None: + param_count += 1 + updates.append(f"description = ${param_count}") + params.append(description) + + if system_prompt is not None: + param_count += 1 + updates.append(f"system_prompt = ${param_count}") + params.append(system_prompt) + + if config is not None: + param_count += 1 + updates.append(f"config = ${param_count}") + params.append(json.dumps(config)) + + if enabled is not None: + param_count += 1 + updates.append(f"enabled = ${param_count}") + params.append(enabled) + + if not updates: + # No updates provided, just return current agent + return await get_agent(agent_id) + + param_count += 1 + params.append(agent_id) + + query = f"UPDATE agents SET {', '.join(updates)} WHERE id = ${param_count} RETURNING *" + row = await conn.fetchrow(query, *params) + + if not row: + raise ValueError(f"Agent {agent_id} not found") + + agent = dict(row) + agent["capabilities"] = json.loads(agent["capabilities"]) if agent["capabilities"] else [] + agent["config"] = json.loads(agent["config"]) if agent["config"] else {} + return agent + + +async def update_execution_status( + execution_id: int, + status: str, + output_result: Optional[Dict[str, Any]] = None, + actions_proposed: Optional[List[Dict[str, Any]]] = None, + actions_executed: Optional[List[Dict[str, Any]]] = None, + error_message: Optional[str] = None +) -> Dict[str, Any]: + """Update agent execution status and results""" + pool = await get_pool() + async with pool.acquire() as conn: + # Check if started_at is already set + current = await conn.fetchrow("SELECT started_at FROM agent_executions WHERE id = $1", execution_id) + + updates = ["status = $1"] + params = [status] + param_count = 1 + + # Only set started_at if transitioning to running and not already set + if status == "running" and (not current or not current["started_at"]): + updates.append("started_at = CURRENT_TIMESTAMP") + + if status in ["completed", "failed", "rejected"]: + updates.append("completed_at = CURRENT_TIMESTAMP") + + if output_result is not None: + param_count += 1 + updates.append(f"output_result = ${param_count}") + params.append(json.dumps(output_result)) + + if actions_proposed is not None: + param_count += 1 + updates.append(f"actions_proposed = ${param_count}") + params.append(json.dumps(actions_proposed)) + + if actions_executed is not None: + param_count += 1 + updates.append(f"actions_executed = ${param_count}") + params.append(json.dumps(actions_executed)) + + if error_message is not None: + param_count += 1 + updates.append(f"error_message = ${param_count}") + params.append(error_message) + + param_count += 1 + params.append(execution_id) + + query = f"UPDATE agent_executions SET {', '.join(updates)} WHERE id = ${param_count} RETURNING *" + row = await conn.fetchrow(query, *params) + + if not row: + raise ValueError(f"Execution {execution_id} not found") + + execution = dict(row) + for json_field in ["input_context", "output_result", "actions_proposed", "actions_executed"]: + if execution[json_field]: + execution[json_field] = json.loads(execution[json_field]) + return execution diff --git a/dashboard/backend/routers/opc_agents.py b/dashboard/backend/routers/opc_agents.py index 9fbb794..0ffe591 100644 --- a/dashboard/backend/routers/opc_agents.py +++ b/dashboard/backend/routers/opc_agents.py @@ -19,6 +19,7 @@ class AgentUpdate(BaseModel): class ExecutionApproval(BaseModel): approved: bool + feedback: Optional[str] = None @router.get("/agents") @@ -53,8 +54,17 @@ async def update_agent( ): """Update agent configuration (admin only)""" user: User = request.state.user - # This would need an update_agent function in opc_db - raise HTTPException(status_code=501, detail="Not implemented yet") + try: + agent = await opc_db.update_agent_config( + agent_id=agent_id, + name=updates.name, + description=updates.description, + system_prompt=updates.system_prompt, + enabled=updates.enabled + ) + return agent + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) @router.post("/agents/{agent_id}/execute") @@ -99,8 +109,7 @@ async def get_execution( ): """Get execution details""" user: User = request.state.user - executions = await opc_db.get_agent_executions(limit=1000) - execution = next((e for e in executions if e["id"] == execution_id), None) + execution = await opc_db.get_agent_execution(execution_id) if not execution: raise HTTPException(status_code=404, detail="Execution not found") return execution @@ -114,10 +123,13 @@ async def approve_execution( ): """Approve or reject agent execution (for CxO level actions)""" user: User = request.state.user - # This would need an approve_execution function in opc_db - # For now, return placeholder - return { - "execution_id": execution_id, - "approved": approval.approved, - "approved_by": user.username - } + try: + execution = await opc_db.approve_agent_execution( + execution_id=execution_id, + approved=approval.approved, + approved_by=user.username, + feedback=approval.feedback + ) + return execution + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) diff --git a/dashboard/backend/routers/opc_ws.py b/dashboard/backend/routers/opc_ws.py index 6f7afca..93f8b9d 100644 --- a/dashboard/backend/routers/opc_ws.py +++ b/dashboard/backend/routers/opc_ws.py @@ -2,7 +2,8 @@ OPC WebSocket Router - Real-time updates for tasks and agent executions """ from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends -from typing import Set +from typing import Set, Any, Dict +from datetime import datetime import json import asyncio import logging @@ -16,6 +17,18 @@ router = APIRouter() active_connections: Set[WebSocket] = set() +def serialize_datetime(obj: Any) -> Any: + """Recursively serialize datetime objects to ISO format strings""" + if isinstance(obj, datetime): + return obj.isoformat() + elif isinstance(obj, dict): + return {key: serialize_datetime(value) for key, value in obj.items()} + elif isinstance(obj, list): + return [serialize_datetime(item) for item in obj] + else: + return obj + + class ConnectionManager: """Manages WebSocket connections""" @@ -74,44 +87,30 @@ async def websocket_endpoint(websocket: WebSocket): async def broadcast_task_update(task_id: int, action: str, task_data: dict): """Broadcast task update to all connected clients""" - # Serialize datetime objects to ISO format - serialized_data = task_data.copy() - for key in ["created_at", "updated_at", "completed_at", "due_date"]: - if serialized_data.get(key): - serialized_data[key] = serialized_data[key].isoformat() if hasattr(serialized_data[key], 'isoformat') else serialized_data[key] - - timestamp = task_data.get("updated_at") - if timestamp and hasattr(timestamp, 'isoformat'): - timestamp = timestamp.isoformat() + # Serialize all datetime objects recursively + serialized_data = serialize_datetime(task_data) message = { "type": "task_update", "task_id": task_id, "action": action, # created, updated, moved, deleted "data": serialized_data, - "timestamp": timestamp + "timestamp": datetime.utcnow().isoformat() } await manager.broadcast(message) async def broadcast_agent_execution(execution_id: int, status: str, execution_data: dict): """Broadcast agent execution update""" - # Serialize datetime objects to ISO format - serialized_data = execution_data.copy() - for key in ["started_at", "completed_at"]: - if serialized_data.get(key): - serialized_data[key] = serialized_data[key].isoformat() if hasattr(serialized_data[key], 'isoformat') else serialized_data[key] - - timestamp = execution_data.get("started_at") or execution_data.get("completed_at") - if timestamp and hasattr(timestamp, 'isoformat'): - timestamp = timestamp.isoformat() + # Serialize all datetime objects recursively + serialized_data = serialize_datetime(execution_data) message = { "type": "agent_execution", "execution_id": execution_id, "status": status, # pending, running, completed, failed, pending_approval "data": serialized_data, - "timestamp": timestamp + "timestamp": datetime.utcnow().isoformat() } await manager.broadcast(message) diff --git a/dashboard/backend/services/agent_executor.py b/dashboard/backend/services/agent_executor.py index bf860be..c5e51e5 100644 --- a/dashboard/backend/services/agent_executor.py +++ b/dashboard/backend/services/agent_executor.py @@ -65,13 +65,23 @@ class AgentExecutor: async def _process_pending_executions(self): """Process pending agent executions""" - executions = await opc_db.get_agent_executions(status="pending", limit=10) - print(f"[Executor] Found {len(executions)} pending executions") + # Process both pending and approved executions + pending_executions = await opc_db.get_agent_executions(status="pending", limit=10) + approved_executions = await opc_db.get_agent_executions(status="approved", limit=10) + + executions = pending_executions + approved_executions + print(f"[Executor] Found {len(pending_executions)} pending and {len(approved_executions)} approved executions") for execution in executions: try: - print(f"[Executor] Processing execution {execution['id']} for agent {execution['agent_id']}") - await self._execute_agent(execution) + print(f"[Executor] Processing execution {execution['id']} for agent {execution['agent_id']} (status: {execution['status']})") + + if execution['status'] == 'approved': + # Resume approved execution + await self._resume_approved_execution(execution) + else: + # Execute new pending execution + await self._execute_agent(execution) except Exception as e: print(f"[Executor] Failed to execute agent {execution['agent_id']}: {e}") logger.error(f"Failed to execute agent {execution['agent_id']}: {e}") @@ -91,8 +101,7 @@ class AgentExecutor: await self._update_execution_status(execution["id"], "running") # Get task and context - tasks = await opc_db.get_tasks(limit=1000) - task = next((t for t in tasks if t["id"] == task_id), None) + task = await opc_db.get_task_by_id(task_id) if not task: raise ValueError(f"Task {task_id} not found") @@ -118,6 +127,39 @@ class AgentExecutor: # Send completion notification await self._send_completion_notification(agent, task, result) + async def _resume_approved_execution(self, execution: Dict[str, Any]): + """Resume an approved execution and execute its actions""" + agent_id = execution["agent_id"] + task_id = execution["task_id"] + + # Get agent + agent = self.agents.get(agent_id) + if not agent: + raise ValueError(f"Agent {agent_id} not found") + + # Get task + task = await opc_db.get_task_by_id(task_id) + if not task: + raise ValueError(f"Task {task_id} not found") + + # Mark as running + await self._update_execution_status(execution["id"], "running") + + # Get proposed actions from execution + result = execution.get("output_result", {}) + actions = execution.get("actions_proposed", []) + + logger.info(f"Resuming approved execution for agent {agent.name} on task {task['title']}") + + # Execute approved actions + await self._execute_actions(task, actions) + + # Mark as completed + await self._mark_execution_completed(execution["id"], result) + + # Send completion notification + await self._send_completion_notification(agent, task, result) + async def _build_context(self, task: Dict[str, Any]) -> Dict[str, Any]: """Build context for agent execution""" context = { @@ -150,10 +192,16 @@ class AgentExecutor: await self._action_send_notification(action) elif action_type == "add_comment": await self._action_add_comment(task, action) + elif action_type == "error": + # Handle error action - log and fail execution + error_msg = action.get("message", "Unknown error") + logger.error(f"Agent reported error: {error_msg}") + raise Exception(error_msg) else: logger.warning(f"Unknown action type: {action_type}") except Exception as e: logger.error(f"Failed to execute action {action_type}: {e}") + raise async def _action_create_subtask(self, parent_task: Dict[str, Any], action: Dict[str, Any]): """Create a subtask""" @@ -199,13 +247,7 @@ class AgentExecutor: async def _update_execution_status(self, execution_id: int, status: str): """Update execution status""" - pool = await opc_db.get_pool() - async with pool.acquire() as conn: - await conn.execute(""" - UPDATE agent_executions - SET status = $1, started_at = CURRENT_TIMESTAMP - WHERE id = $2 - """, status, execution_id) + await opc_db.update_execution_status(execution_id, status) async def _mark_execution_completed(self, execution_id: int, result: Dict[str, Any]): """Mark execution as completed""" diff --git a/dashboard/backend/services/agents/base_agent.py b/dashboard/backend/services/agents/base_agent.py index b832e84..524aec5 100644 --- a/dashboard/backend/services/agents/base_agent.py +++ b/dashboard/backend/services/agents/base_agent.py @@ -76,6 +76,9 @@ Only propose actions you can actually execute. Be specific and actionable.""" async def _call_llm(self, prompt: str) -> str: """Call LiteLLM proxy""" + import logging + logger = logging.getLogger(__name__) + model = self.config.get("model", "claude-sonnet-4-6") temperature = self.config.get("temperature", 0.7) @@ -85,7 +88,7 @@ Only propose actions you can actually execute. Be specific and actionable.""" if self.litellm_api_key and self.litellm_api_key.strip(): headers["Authorization"] = f"Bearer {self.litellm_api_key}" - async with httpx.AsyncClient(timeout=60.0) as client: + async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( f"{self.litellm_url}/chat/completions", headers=headers, @@ -102,7 +105,17 @@ Only propose actions you can actually execute. Be specific and actionable.""" response.raise_for_status() data = response.json() return data["choices"][0]["message"]["content"] + except httpx.ConnectError as e: + logger.error(f"LiteLLM connection failed: {e}") + raise Exception(f"LiteLLM service unavailable: {str(e)}") + except httpx.TimeoutException as e: + logger.error(f"LiteLLM request timeout: {e}") + raise Exception(f"LiteLLM request timeout: {str(e)}") + except httpx.HTTPStatusError as e: + logger.error(f"LiteLLM HTTP error: {e.response.status_code} - {e.response.text}") + raise Exception(f"LiteLLM HTTP error {e.response.status_code}: {e.response.text}") except Exception as e: + logger.error(f"LiteLLM call failed: {e}") raise Exception(f"LLM call failed: {str(e)}") def _parse_response(self, response: str) -> Dict[str, Any]: