259 lines
7.4 KiB
Python
259 lines
7.4 KiB
Python
"""Base classes and utilities for VoIdea agents."""
|
|
|
|
import hashlib
|
|
import inspect
|
|
import re
|
|
from abc import ABC, abstractmethod
|
|
from datetime import datetime, timezone
|
|
from enum import Enum
|
|
from pathlib import Path
|
|
from typing import Any, Generic, TypeVar
|
|
from uuid import UUID, uuid4
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
from app.core.base import CoreModel
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class AgentStatus(str, Enum):
|
|
"""Agent status enumeration."""
|
|
|
|
IDLE = "idle"
|
|
RUNNING = "running"
|
|
ERROR = "error"
|
|
OFFLINE = "offline"
|
|
|
|
|
|
class AgentTrigger(str, Enum):
|
|
"""Agent trigger types."""
|
|
|
|
MANUAL = "manual"
|
|
PRE_COMMIT = "pre_commit"
|
|
PUSH = "push"
|
|
TAG_CREATION = "tag_creation"
|
|
CRON = "cron"
|
|
API = "api"
|
|
EVENT = "event"
|
|
|
|
|
|
class AgentResult(CoreModel):
|
|
"""Result of agent execution."""
|
|
|
|
success: bool
|
|
message: str = ""
|
|
data: dict[str, Any] = Field(default_factory=dict)
|
|
errors: list[str] = Field(default_factory=list)
|
|
duration_ms: int = 0
|
|
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
|
|
|
|
|
|
class AgentMetrics(CoreModel):
|
|
"""Agent performance metrics."""
|
|
|
|
agent_id: str
|
|
version: str = "1.0.0"
|
|
requests_total: int = 0
|
|
requests_success: int = 0
|
|
requests_failed: int = 0
|
|
average_duration_ms: float = 0.0
|
|
last_run: datetime | None = None
|
|
|
|
|
|
class BaseAgent(ABC):
|
|
"""Abstract base class for all agents.
|
|
|
|
All agents must inherit from this class and implement required methods.
|
|
"""
|
|
|
|
name: str = ""
|
|
version: str = "1.0.0"
|
|
description: str = ""
|
|
triggers: list[AgentTrigger] = [AgentTrigger.MANUAL]
|
|
changelog_dir: Path = Path("CHANGELOG") / "agents"
|
|
|
|
def __init__(self):
|
|
self._status = AgentStatus.IDLE
|
|
self._last_run: datetime | None = None
|
|
self._current_task: str | None = None
|
|
self._changelog_path = self.changelog_dir / f"{self.name}.md"
|
|
|
|
@property
|
|
def status(self) -> AgentStatus:
|
|
"""Get current agent status."""
|
|
return self._status
|
|
|
|
@property
|
|
def last_run(self) -> datetime | None:
|
|
"""Get last run timestamp."""
|
|
return self._last_run
|
|
|
|
@abstractmethod
|
|
async def run(self, context: dict[str, Any] | None = None) -> AgentResult:
|
|
"""Execute agent task.
|
|
|
|
Args:
|
|
context: Optional context data for the agent
|
|
|
|
Returns:
|
|
AgentResult with execution outcome
|
|
"""
|
|
pass
|
|
|
|
@abstractmethod
|
|
async def health_check(self) -> bool:
|
|
"""Check if agent is healthy and operational.
|
|
|
|
Returns:
|
|
True if agent can execute tasks
|
|
"""
|
|
pass
|
|
|
|
async def get_status(self) -> AgentStatus:
|
|
"""Get current agent status.
|
|
|
|
Returns:
|
|
Current status from status property
|
|
"""
|
|
return self.status
|
|
|
|
async def get_metrics(self) -> AgentMetrics:
|
|
"""Get agent performance metrics.
|
|
|
|
Returns:
|
|
AgentMetrics instance
|
|
"""
|
|
return AgentMetrics(
|
|
agent_id=self.name,
|
|
version=self.version,
|
|
last_run=self.last_run,
|
|
)
|
|
|
|
async def set_running(self, task: str) -> None:
|
|
"""Set agent to running state."""
|
|
self._status = AgentStatus.RUNNING
|
|
self._current_task = task
|
|
self._last_run = datetime.now(timezone.utc)
|
|
|
|
async def set_idle(self) -> None:
|
|
"""Set agent to idle state."""
|
|
self._status = AgentStatus.IDLE
|
|
self._current_task = None
|
|
|
|
async def set_error(self, error: str) -> None:
|
|
"""Set agent to error state."""
|
|
self._status = AgentStatus.ERROR
|
|
self._current_task = None
|
|
|
|
async def set_offline(self) -> None:
|
|
"""Set agent to offline state."""
|
|
self._status = AgentStatus.OFFLINE
|
|
|
|
def compute_checksum(self) -> str:
|
|
"""Compute SHA256 checksum of this agent's source file.
|
|
|
|
Returns:
|
|
Hex digest of the file content.
|
|
"""
|
|
file_path = inspect.getfile(self.__class__)
|
|
content = Path(file_path).read_bytes()
|
|
return hashlib.sha256(content).hexdigest()
|
|
|
|
def _read_changelog_checksum(self) -> str | None:
|
|
"""Read stored checksum from changelog file.
|
|
|
|
Returns:
|
|
Stored checksum or None if file doesn't exist.
|
|
"""
|
|
if not self._changelog_path.exists():
|
|
return None
|
|
content = self._changelog_path.read_text(encoding="utf-8")
|
|
match = re.search(r"<!--\s*checksum:\s*([a-f0-9]{64})\s*-->", content)
|
|
return match.group(1) if match else None
|
|
|
|
def bump_version(self, version_type: str = "patch") -> str:
|
|
"""Bump agent version (major.minor.patch).
|
|
|
|
Args:
|
|
version_type: "major", "minor", or "patch"
|
|
|
|
Returns:
|
|
New version string.
|
|
"""
|
|
major, minor, patch = map(int, self.version.split("."))
|
|
if version_type == "major":
|
|
major += 1
|
|
minor = 0
|
|
patch = 0
|
|
elif version_type == "minor":
|
|
minor += 1
|
|
patch = 0
|
|
else:
|
|
patch += 1
|
|
self.version = f"{major}.{minor}.{patch}"
|
|
return self.version
|
|
|
|
def _write_changelog_entry(self, version: str, entries: list[str]) -> None:
|
|
"""Write a changelog entry for this agent.
|
|
|
|
Args:
|
|
version: New version string.
|
|
entries: List of change descriptions.
|
|
"""
|
|
self.changelog_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
|
checksum = self.compute_checksum()
|
|
header = f"# {self.name} Changelog\n<!-- checksum: {checksum} -->\n"
|
|
entry = f"\n## {version} ({today})\n"
|
|
for line in entries:
|
|
entry += f"- {line}\n"
|
|
|
|
if self._changelog_path.exists():
|
|
old = self._changelog_path.read_text(encoding="utf-8")
|
|
new = header + entry + old.split("\n", 2)[-1] if "\n" in old else old
|
|
self._changelog_path.write_text(new, encoding="utf-8")
|
|
else:
|
|
self._changelog_path.write_text(header + entry, encoding="utf-8")
|
|
|
|
async def _check_version(self, changelog_entries: list[str] | None = None) -> str | None:
|
|
"""Check if agent changed and bump version if needed.
|
|
|
|
Should be called after run(). Compares current file checksum
|
|
with stored checksum in changelog. On mismatch bumps patch
|
|
and writes changelog entry.
|
|
|
|
Args:
|
|
changelog_entries: Optional list of change descriptions.
|
|
If None, auto-generated from git diff summary.
|
|
|
|
Returns:
|
|
New version string if bumped, None if unchanged.
|
|
"""
|
|
current_checksum = self.compute_checksum()
|
|
stored_checksum = self._read_changelog_checksum()
|
|
|
|
if current_checksum == stored_checksum:
|
|
return None
|
|
|
|
old_version = self.version
|
|
new_version = self.bump_version("patch")
|
|
entries = changelog_entries or ["Auto-detected code changes"]
|
|
self._write_changelog_entry(new_version, entries)
|
|
|
|
return new_version
|
|
|
|
def __repr__(self) -> str:
|
|
return f"<{self.__class__.__name__}(name={self.name}, status={self.status.value})>"
|
|
|
|
|
|
class AgentResponse(CoreModel):
|
|
"""Standardized agent response."""
|
|
|
|
agent: str
|
|
status: str
|
|
message: str
|
|
data: dict[str, Any] = Field(default_factory=dict)
|
|
timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) |