multi-agent-bidding/app/execution_monitor.py

"""
执行监控器 - 监控执行状态、超时处理、结果收集
"""

import asyncio
import time
import threading
from typing import Dict, List, Optional, Any, Callable
from .models import Task, Execution, AgentProfile, Bid, TaskStatus


class ExecutionMonitor:
    """执行监控器"""
    
    def __init__(
        self,
        task: Task,
        agent: AgentProfile,
        bid: Bid,
        on_complete: Callable = None,
        on_timeout: Callable = None,
        on_error: Callable = None
    ):
        self.task = task
        self.agent = agent
        self.bid = bid
        self.timeout = task.timeout
        
        self.callbacks = {
            'complete': on_complete,
            'timeout': on_timeout,
            'error': on_error
        }
        
        self.execution: Optional[Execution] = None
        self.backup_agents: List[AgentProfile] = []
        self.active_executions: Dict[str, Any] = {}
        
        self._start_time: float = 0
        self._is_running: bool = False
        self._result: Optional[Any] = None
        self._error: Optional[str] = None
    
    def add_backup_agent(self, agent: AgentProfile):
        """添加备选Agent"""
        self.backup_agents.append(agent)
    
    def start(self, execute_func: Callable) -> Execution:
        """
        启动执行
        
        Args:
            execute_func: 执行函数
        
        Returns:
            Execution对象
        """
        self.execution = Execution(
            task_id=self.task.id,
            agent_id=self.agent.id,
            bid_id=self.bid.id,
            status='running'
        )
        self._start_time = time.time()
        self._is_running = True
        
        # 启动执行线程
        thread = threading.Thread(target=self._run_execution, args=(execute_func,))
        thread.start()
        
        # 启动超时监控线程
        timeout_thread = threading.Thread(target=self._watch_timeout)
        timeout_thread.start()
        
        return self.execution
    
    def _run_execution(self, execute_func: Callable):
        """执行任务"""
        try:
            result = execute_func(self.task, self.agent, self.bid)
            
            if self._is_running:
                self._result = result
                self.execution.status = 'completed'
                self.execution.end_time = time.time()
                self.execution.result = result
                self._is_running = False
                
                if self.callbacks['complete']:
                    self.callbacks['complete'](self.execution)
        
        except Exception as e:
            if self._is_running:
                self._error = str(e)
                self.execution.status = 'failed'
                self.execution.end_time = time.time()
                self.execution.error = str(e)
                self._is_running = False
                
                if self.callbacks['error']:
                    self.callbacks['error'](self.execution, e)
    
    def _watch_timeout(self):
        """超时监控"""
        while self._is_running:
            elapsed = time.time() - self._start_time
            
            if elapsed >= self.timeout:
                # 超时
                if self._is_running:
                    self.execution.status = 'timeout'
                    self.execution.end_time = time.time()
                    self.execution.error = f"执行超时 ({self.timeout}秒)"
                    self._is_running = False
                    
                    if self.callbacks['timeout']:
                        self.callbacks['timeout'](self.execution)
                    
                    # 启动备选Agent
                    if self.backup_agents:
                        self._start_backup()
                break
            
            time.sleep(1)
    
    def _start_backup(self):
        """启动备选Agent"""
        # 这里简化处理，实际应该启动备选执行
        pass
    
    def get_result(self) -> Optional[Any]:
        """获取结果"""
        return self._result
    
    def is_complete(self) -> bool:
        """是否完成"""
        return not self._is_running
    
    def get_duration(self) -> float:
        """获取执行时长"""
        if self.execution and self.execution.end_time:
            return self.execution.end_time - self.execution.start_time
        return time.time() - self._start_time


class ExecutionMonitorManager:
    """执行监控管理器"""
    
    def __init__(self):
        self.monitors: Dict[str, ExecutionMonitor] = {}  # execution_id -> monitor
    
    def create_monitor(
        self,
        task: Task,
        agent: AgentProfile,
        bid: Bid,
        callbacks: Dict[str, Callable] = None
    ) -> ExecutionMonitor:
        """创建监控器"""
        monitor = ExecutionMonitor(
            task=task,
            agent=agent,
            bid=bid,
            on_complete=callbacks.get('complete'),
            on_timeout=callbacks.get('timeout'),
            on_error=callbacks.get('error')
        )
        
        # 保存到管理器
        if monitor.execution:
            self.monitors[monitor.execution.id] = monitor
        
        return monitor
    
    def get_monitor(self, execution_id: str) -> Optional[ExecutionMonitor]:
        """获取监控器"""
        return self.monitors.get(execution_id)
    
    def cleanup_completed(self):
        """清理已完成的监控器"""
        completed = [
            eid for eid, m in self.monitors.items()
            if m.is_complete()
        ]
        for eid in completed:
            self.monitors.pop(eid, None)
    
    def get_active_count(self) -> int:
        """获取活跃执行数"""
        return len([m for m in self.monitors.values() if not m.is_complete()])