From 035c74ed363210ebe7f410d674f5f57db0a1e4c9 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Sat, 7 Mar 2026 15:27:30 -0500 Subject: [PATCH] feat(models): Add priority field to Ticket dataclass and update serialization --- conductor/tracks.md | 2 +- .../tracks/ticket_queue_mgmt_20260306/plan.md | 4 +- src/models.py | 375 +++++++++--------- tests/test_ticket_queue.py | 37 ++ 4 files changed, 229 insertions(+), 189 deletions(-) create mode 100644 tests/test_ticket_queue.py diff --git a/conductor/tracks.md b/conductor/tracks.md index 292f4fc..fceaa78 100644 --- a/conductor/tracks.md +++ b/conductor/tracks.md @@ -66,7 +66,7 @@ This file tracks all major tracks for the project. Each track has its own detail ### Manual UX Controls -14. [ ] **Track: Manual Ticket Queue Management** +14. [~] **Track: Manual Ticket Queue Management** *Link: [./tracks/ticket_queue_mgmt_20260306/](./tracks/ticket_queue_mgmt_20260306/)* 15. [ ] **Track: Kill/Abort Running Workers** diff --git a/conductor/tracks/ticket_queue_mgmt_20260306/plan.md b/conductor/tracks/ticket_queue_mgmt_20260306/plan.md index 00c8775..22bef4b 100644 --- a/conductor/tracks/ticket_queue_mgmt_20260306/plan.md +++ b/conductor/tracks/ticket_queue_mgmt_20260306/plan.md @@ -5,10 +5,10 @@ ## Phase 1: Priority Field Focus: Add priority to Ticket model -- [ ] Task 1.1: Initialize MMA Environment +- [x] Task 1.1: Initialize MMA Environment - Run `activate_skill mma-orchestrator` before starting -- [ ] Task 1.2: Add priority field to Ticket +- [~] Task 1.2: Add priority field to Ticket - WHERE: `src/models.py` `Ticket` dataclass - WHAT: Add `priority: str = "medium"` field - HOW: diff --git a/src/models.py b/src/models.py index 95dbc10..693e9d8 100644 --- a/src/models.py +++ b/src/models.py @@ -9,230 +9,233 @@ from src.paths import get_config_path CONFIG_PATH = get_config_path() def load_config() -> dict[str, Any]: - with open(CONFIG_PATH, "rb") as f: - return tomllib.load(f) + with open(CONFIG_PATH, "rb") as f: + return tomllib.load(f) def save_config(config: dict[str, Any]) -> None: - import tomli_w - with open(CONFIG_PATH, "wb") as f: - tomli_w.dump(config, f) + import tomli_w + with open(CONFIG_PATH, "wb") as f: + tomli_w.dump(config, f) AGENT_TOOL_NAMES = [ - "run_powershell", - "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" + "run_powershell", + "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" ] def parse_history_entries(history_strings: list[str], roles: list[str]) -> list[dict[str, Any]]: - import re - entries = [] - for raw in history_strings: - ts = "" - rest = raw - if rest.startswith("@"): - nl = rest.find("\n") - if nl != -1: - ts = rest[1:nl] - rest = rest[nl + 1:] - known = roles or ["User", "AI", "Vendor API", "System"] - role_pat = re.compile(r"^(" + "|".join(re.escape(r) for r in known) + r"):", re.IGNORECASE) - match = role_pat.match(rest) - role = match.group(1) if match else "User" - if match: - content = rest[match.end():].strip() - else: - content = rest - entries.append({"role": role, "content": content, "collapsed": True, "ts": ts}) - return entries + import re + entries = [] + for raw in history_strings: + ts = "" + rest = raw + if rest.startswith("@"): + nl = rest.find("\n") + if nl != -1: + ts = rest[1:nl] + rest = rest[nl + 1:] + known = roles or ["User", "AI", "Vendor API", "System"] + role_pat = re.compile(r"^(" + "|".join(re.escape(r) for r in known) + r"):", re.IGNORECASE) + match = role_pat.match(rest) + role = match.group(1) if match else "User" + if match: + content = rest[match.end():].strip() + else: + content = rest + entries.append({"role": role, "content": content, "collapsed": True, "ts": ts}) + return entries @dataclass class Ticket: - 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 + id: str + description: str + status: str = "todo" + assigned_to: str = "unassigned" + priority: str = "medium" + 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: - self.status = "blocked" - self.blocked_reason = reason + def mark_blocked(self, reason: str) -> None: + self.status = "blocked" + self.blocked_reason = reason - def mark_complete(self) -> None: - self.status = "completed" + def mark_complete(self) -> None: + self.status = "completed" - def get(self, key: str, default: Any = None) -> Any: - return getattr(self, key, default) + def get(self, key: str, default: Any = None) -> Any: + 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, - } + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "description": self.description, + "status": self.status, + "assigned_to": self.assigned_to, + "priority": self.priority, + "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), - ) + @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", "unassigned"), + priority=data.get("priority", "medium"), + 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: - id: str - description: str - tickets: List[Ticket] = field(default_factory=list) + id: str + description: str + tickets: List[Ticket] = field(default_factory=list) - def get_executable_tickets(self) -> List[Ticket]: - from src.dag_engine import TrackDAG - dag = TrackDAG(self.tickets) - return dag.get_ready_tasks() + def get_executable_tickets(self) -> List[Ticket]: + from src.dag_engine import TrackDAG + dag = TrackDAG(self.tickets) + return dag.get_ready_tasks() - def to_dict(self) -> Dict[str, Any]: - return { - "id": self.id, - "description": self.description, - "tickets": [t.to_dict() for t in self.tickets], - } + 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", [])], - ) + @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: - ticket_id: str - model_name: str - messages: List[Dict[str, Any]] = field(default_factory=list) + ticket_id: str + model_name: str + messages: List[Dict[str, Any]] = field(default_factory=list) @dataclass class Metadata: - id: str - name: str - status: Optional[str] = None - created_at: Optional[datetime.datetime] = None - updated_at: Optional[datetime.datetime] = None + id: str + name: str + status: Optional[str] = None + created_at: Optional[datetime.datetime] = None + updated_at: Optional[datetime.datetime] = None - def to_dict(self) -> Dict[str, Any]: - return { - "id": self.id, - "name": self.name, - "status": self.status, - "created_at": self.created_at.isoformat() if self.created_at else None, - "updated_at": self.updated_at.isoformat() if self.updated_at else None, - } + def to_dict(self) -> Dict[str, Any]: + return { + "id": self.id, + "name": self.name, + "status": self.status, + "created_at": self.created_at.isoformat() if self.created_at else None, + "updated_at": self.updated_at.isoformat() if self.updated_at else None, + } - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "Metadata": - created = data.get("created_at") - updated = data.get("updated_at") - if isinstance(created, str): - try: - created = datetime.datetime.fromisoformat(created) - except ValueError: - created = None - if isinstance(updated, str): - try: - updated = datetime.datetime.fromisoformat(updated) - except ValueError: - updated = None - return cls( - id=data["id"], - name=data.get("name", ""), - status=data.get("status"), - created_at=created, - updated_at=updated, - ) + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "Metadata": + created = data.get("created_at") + updated = data.get("updated_at") + if isinstance(created, str): + try: + created = datetime.datetime.fromisoformat(created) + except ValueError: + created = None + if isinstance(updated, str): + try: + updated = datetime.datetime.fromisoformat(updated) + except ValueError: + updated = None + return cls( + id=data["id"], + name=data.get("name", ""), + status=data.get("status"), + created_at=created, + updated_at=updated, + ) @dataclass class TrackState: - metadata: Metadata - discussion: List[str] = field(default_factory=list) - tasks: List[Ticket] = field(default_factory=list) + metadata: Metadata + discussion: List[str] = field(default_factory=list) + tasks: List[Ticket] = field(default_factory=list) - def to_dict(self) -> Dict[str, Any]: - serialized_discussion = [] - for item in self.discussion: - if isinstance(item, dict): - new_item = dict(item) - if "ts" in new_item and isinstance(new_item["ts"], datetime.datetime): - new_item["ts"] = new_item["ts"].isoformat() - serialized_discussion.append(new_item) - else: - serialized_discussion.append(item) - return { - "metadata": self.metadata.to_dict(), - "discussion": serialized_discussion, - "tasks": [t.to_dict() for t in self.tasks], - } + def to_dict(self) -> Dict[str, Any]: + serialized_discussion = [] + for item in self.discussion: + if isinstance(item, dict): + new_item = dict(item) + if "ts" in new_item and isinstance(new_item["ts"], datetime.datetime): + new_item["ts"] = new_item["ts"].isoformat() + serialized_discussion.append(new_item) + else: + serialized_discussion.append(item) + return { + "metadata": self.metadata.to_dict(), + "discussion": serialized_discussion, + "tasks": [t.to_dict() for t in self.tasks], + } - @classmethod - def from_dict(cls, data: Dict[str, Any]) -> "TrackState": - discussion = data.get("discussion", []) - parsed_discussion = [] - for item in discussion: - if isinstance(item, dict): - new_item = dict(item) - ts = new_item.get("ts") - if isinstance(ts, str): - try: - new_item["ts"] = datetime.datetime.fromisoformat(ts) - except ValueError: - pass - parsed_discussion.append(new_item) - else: - parsed_discussion.append(item) - return cls( - metadata=Metadata.from_dict(data["metadata"]), - discussion=parsed_discussion, - tasks=[Ticket.from_dict(t) for t in data.get("tasks", [])], - ) + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "TrackState": + discussion = data.get("discussion", []) + parsed_discussion = [] + for item in discussion: + if isinstance(item, dict): + new_item = dict(item) + ts = new_item.get("ts") + if isinstance(ts, str): + try: + new_item["ts"] = datetime.datetime.fromisoformat(ts) + except ValueError: + pass + parsed_discussion.append(new_item) + else: + parsed_discussion.append(item) + return cls( + metadata=Metadata.from_dict(data["metadata"]), + discussion=parsed_discussion, + tasks=[Ticket.from_dict(t) for t in data.get("tasks", [])], + ) @dataclass class FileItem: diff --git a/tests/test_ticket_queue.py b/tests/test_ticket_queue.py new file mode 100644 index 0000000..6752c44 --- /dev/null +++ b/tests/test_ticket_queue.py @@ -0,0 +1,37 @@ +import pytest +from src.models import Ticket + +def test_ticket_priority_default(): + ticket = Ticket(id="T1", description="Test ticket") + assert ticket.priority == "medium" + +def test_ticket_priority_custom(): + ticket_high = Ticket(id="T2", description="High priority", priority="high") + assert ticket_high.priority == "high" + + ticket_low = Ticket(id="T3", description="Low priority", priority="low") + assert ticket_low.priority == "low" + +def test_ticket_to_dict_priority(): + ticket = Ticket(id="T4", description="To dict test", priority="high") + d = ticket.to_dict() + assert "priority" in d + assert d["priority"] == "high" + +def test_ticket_from_dict_priority(): + data = { + "id": "T5", + "description": "From dict test", + "priority": "low", + "status": "todo" + } + ticket = Ticket.from_dict(data) + assert ticket.priority == "low" + +def test_ticket_from_dict_default_priority(): + data = { + "id": "T6", + "description": "No priority in dict" + } + ticket = Ticket.from_dict(data) + assert ticket.priority == "medium"