182 lines
5.2 KiB
Python
182 lines
5.2 KiB
Python
from __future__ import annotations
|
|
import tomllib
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional, Dict, Any, Union
|
|
from pathlib import Path
|
|
|
|
CONFIG_PATH = Path("config.toml")
|
|
|
|
def load_config() -> dict[str, Any]:
|
|
with open(CONFIG_PATH, "rb") as f:
|
|
return tomllib.load(f)
|
|
|
|
# Global constants for agent tools
|
|
AGENT_TOOL_NAMES = [
|
|
"read_file",
|
|
"list_directory",
|
|
"search_files",
|
|
"web_search",
|
|
"fetch_url",
|
|
"get_file_summary",
|
|
"py_get_skeleton",
|
|
"py_get_code_outline",
|
|
"py_get_definition",
|
|
"py_get_signature",
|
|
"py_get_class_summary",
|
|
"py_get_var_declaration",
|
|
"py_get_docstring",
|
|
"py_find_usages",
|
|
"py_get_imports",
|
|
"py_check_syntax",
|
|
"py_get_hierarchy"
|
|
]
|
|
|
|
@dataclass
|
|
class Ticket:
|
|
"""
|
|
Represents a discrete unit of work within a track.
|
|
"""
|
|
|
|
id: str
|
|
description: str
|
|
status: str = "todo"
|
|
assigned_to: str = "unassigned"
|
|
target_file: Optional[str] = None
|
|
target_symbols: List[str] = field(default_factory=list)
|
|
context_requirements: List[str] = field(default_factory=list)
|
|
depends_on: List[str] = field(default_factory=list)
|
|
blocked_reason: Optional[str] = None
|
|
step_mode: bool = False
|
|
retry_count: int = 0
|
|
|
|
def mark_blocked(self, reason: str) -> None:
|
|
"""Sets the ticket status to 'blocked' and records the reason."""
|
|
self.status = "blocked"
|
|
self.blocked_reason = reason
|
|
|
|
def mark_complete(self) -> None:
|
|
"""Sets the ticket status to 'completed'."""
|
|
self.status = "completed"
|
|
|
|
def get(self, key: str, default: Any = None) -> Any:
|
|
"""Helper to provide dictionary-like access to dataclass fields."""
|
|
return getattr(self, key, default)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"description": self.description,
|
|
"status": self.status,
|
|
"assigned_to": self.assigned_to,
|
|
"target_file": self.target_file,
|
|
"target_symbols": self.target_symbols,
|
|
"context_requirements": self.context_requirements,
|
|
"depends_on": self.depends_on,
|
|
"blocked_reason": self.blocked_reason,
|
|
"step_mode": self.step_mode,
|
|
"retry_count": self.retry_count,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "Ticket":
|
|
return cls(
|
|
id=data["id"],
|
|
description=data.get("description", ""),
|
|
status=data.get("status", "todo"),
|
|
assigned_to=data.get("assigned_to", ""),
|
|
target_file=data.get("target_file"),
|
|
target_symbols=data.get("target_symbols", []),
|
|
context_requirements=data.get("context_requirements", []),
|
|
depends_on=data.get("depends_on", []),
|
|
blocked_reason=data.get("blocked_reason"),
|
|
step_mode=data.get("step_mode", False),
|
|
retry_count=data.get("retry_count", 0),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Track:
|
|
"""
|
|
Represents a collection of tickets that together form an architectural track or epic.
|
|
"""
|
|
|
|
id: str
|
|
description: str
|
|
tickets: List[Ticket] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"description": self.description,
|
|
"tickets": [t.to_dict() for t in self.tickets],
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "Track":
|
|
return cls(
|
|
id=data["id"],
|
|
description=data.get("description", ""),
|
|
tickets=[Ticket.from_dict(t) for t in data.get("tickets", [])],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class WorkerContext:
|
|
"""
|
|
State preserved for a specific worker throughout its ticket lifecycle.
|
|
"""
|
|
|
|
ticket_id: str
|
|
model_name: str
|
|
messages: List[Dict[str, Any]] = field(default_factory=list)
|
|
|
|
|
|
@dataclass
|
|
class Metadata:
|
|
id: str
|
|
name: str
|
|
status: str
|
|
created_at: Union[str, Any]
|
|
updated_at: Union[str, Any]
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"id": self.id,
|
|
"name": self.name,
|
|
"status": self.status,
|
|
"created_at": str(self.created_at),
|
|
"updated_at": str(self.updated_at),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "Metadata":
|
|
return cls(
|
|
id=data["id"],
|
|
name=data.get("name", ""),
|
|
status=data.get("status", "todo"),
|
|
created_at=data.get("created_at"),
|
|
updated_at=data.get("updated_at"),
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class TrackState:
|
|
metadata: Metadata
|
|
discussion: List[str] = field(default_factory=list)
|
|
tasks: List[Ticket] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> Dict[str, Any]:
|
|
return {
|
|
"metadata": self.metadata.to_dict(),
|
|
"discussion": self.discussion,
|
|
"tasks": [t.to_dict() for t in self.tasks],
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "TrackState":
|
|
return cls(
|
|
metadata=Metadata.from_dict(data["metadata"]),
|
|
discussion=data.get("discussion", []),
|
|
tasks=[Ticket.from_dict(t) for t in data.get("tasks", [])],
|
|
)
|