109 lines
3.3 KiB
Python
109 lines
3.3 KiB
Python
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import List, Optional, Dict, Any
|
|
|
|
|
|
class TaskStatus(Enum):
|
|
"""Task status enum"""
|
|
PENDING = "pending"
|
|
READY = "ready"
|
|
RUNNING = "running"
|
|
BLOCK = "block"
|
|
TERMINATED = "terminated"
|
|
|
|
|
|
class StepStatus(Enum):
|
|
"""Step status enum"""
|
|
PENDING = "pending"
|
|
RUNNING = "running"
|
|
COMPLETED = "completed"
|
|
FAILED = "failed"
|
|
SKIPPED = "skipped"
|
|
|
|
|
|
@dataclass
|
|
class Step:
|
|
"""Task step"""
|
|
id: str
|
|
name: str
|
|
description: str = ""
|
|
status: StepStatus = StepStatus.PENDING
|
|
result: Optional[Dict[str, Any]] = None
|
|
created_at: datetime = field(default_factory=datetime.now)
|
|
updated_at: datetime = field(default_factory=datetime.now)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary"""
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"status": self.status.value,
|
|
"result": self.result,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Task:
|
|
"""Task with nested subtasks"""
|
|
id: str
|
|
name: str
|
|
description: str = ""
|
|
goal: str = ""
|
|
status: TaskStatus = TaskStatus.PENDING
|
|
steps: List[Step] = field(default_factory=list)
|
|
subtasks: List["Task"] = field(default_factory=list)
|
|
result: Optional[Dict[str, Any]] = None
|
|
created_at: datetime = field(default_factory=datetime.now)
|
|
updated_at: datetime = field(default_factory=datetime.now)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary"""
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"goal": self.goal,
|
|
"status": self.status.value,
|
|
"steps": [s.to_dict() for s in self.steps],
|
|
"subtasks": [t.to_dict() for t in self.subtasks],
|
|
"result": self.result,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Agent:
|
|
"""Agent"""
|
|
id: str
|
|
name: str
|
|
description: str = ""
|
|
instructions: str = ""
|
|
tools: List[str] = field(default_factory=list)
|
|
current_task: Optional[Task] = None
|
|
status: str = "idle"
|
|
result: Optional[Dict[str, Any]] = None
|
|
error: Optional[str] = None
|
|
created_at: datetime = field(default_factory=datetime.now)
|
|
updated_at: datetime = field(default_factory=datetime.now)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
"""Convert to dictionary"""
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"description": self.description,
|
|
"instructions": self.instructions,
|
|
"tools": self.tools,
|
|
"current_task": self.current_task.to_dict() if self.current_task else None,
|
|
"status": self.status,
|
|
"result": self.result,
|
|
"error": self.error,
|
|
"created_at": self.created_at.isoformat() if self.created_at else None,
|
|
"updated_at": self.updated_at.isoformat() if self.updated_at else None
|
|
}
|