68 lines
2.0 KiB
Python
68 lines
2.0 KiB
Python
from dataclasses import dataclass, field
|
|
from typing import List, Optional
|
|
|
|
@dataclass
|
|
class Ticket:
|
|
"""
|
|
Represents a discrete unit of work within a track.
|
|
"""
|
|
id: str
|
|
description: str
|
|
status: str
|
|
assigned_to: str
|
|
depends_on: List[str] = field(default_factory=list)
|
|
blocked_reason: Optional[str] = None
|
|
step_mode: bool = False
|
|
|
|
def mark_blocked(self, reason: str):
|
|
"""Sets the ticket status to 'blocked' and records the reason."""
|
|
self.status = "blocked"
|
|
self.blocked_reason = reason
|
|
|
|
def mark_complete(self):
|
|
"""Sets the ticket status to 'completed'."""
|
|
self.status = "completed"
|
|
|
|
@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 get_executable_tickets(self) -> List[Ticket]:
|
|
"""
|
|
Returns all 'todo' tickets whose dependencies are all 'completed'.
|
|
"""
|
|
# Map ticket IDs to their current status for efficient lookup
|
|
status_map = {t.id: t.status for t in self.tickets}
|
|
|
|
executable = []
|
|
for ticket in self.tickets:
|
|
if ticket.status != "todo":
|
|
continue
|
|
|
|
# Check if all dependencies are completed
|
|
all_deps_completed = True
|
|
for dep_id in ticket.depends_on:
|
|
# If a dependency is missing from the track, we treat it as not completed (or we could raise an error)
|
|
if status_map.get(dep_id) != "completed":
|
|
all_deps_completed = False
|
|
break
|
|
|
|
if all_deps_completed:
|
|
executable.append(ticket)
|
|
|
|
return executable
|
|
|
|
@dataclass
|
|
class WorkerContext:
|
|
"""
|
|
Represents the context provided to a Tier 3 Worker for a specific ticket.
|
|
"""
|
|
ticket_id: str
|
|
model_name: str
|
|
messages: List[dict]
|