121 lines
4.3 KiB
Python
121 lines
4.3 KiB
Python
"""
|
|
Conductor Tech Lead - Tier 2 ticket generation for MMA orchestration.
|
|
|
|
This module implements the Tier 2 (Tech Lead) function for generating implementation tickets from track briefs.
|
|
It uses the LLM to analyze the track requirements and produce structured ticket definitions.
|
|
|
|
Architecture:
|
|
- Uses ai_client.send() for LLM communication
|
|
- Uses mma_prompts.PROMPTS["tier2_sprint_planning"] for system prompt
|
|
- Returns JSON array of ticket definitions
|
|
|
|
Ticket Format:
|
|
Each ticket is a dict with:
|
|
- id: Unique identifier
|
|
- description: Task description
|
|
- depends_on: List of dependency ticket IDs
|
|
- step_mode: Whether to pause for approval between steps
|
|
|
|
Dependencies:
|
|
- Uses TrackDAG from dag_engine.py for topological sorting
|
|
- Uses Ticket from models.py for validation
|
|
|
|
Error Handling:
|
|
- Retries JSON parsing errors up to 3 times
|
|
- Raises RuntimeError if all retries fail
|
|
|
|
Thread Safety:
|
|
- NOT thread-safe. Should only be called from the main GUI thread.
|
|
- Modifies ai_client state (custom_system_prompt, current_tier)
|
|
|
|
See Also:
|
|
- docs/guide_mma.md for MMA orchestration documentation
|
|
- src/mma_prompts.py for Tier-specific prompts
|
|
- src/dag_engine.py for TrackDAG
|
|
"""
|
|
import json
|
|
from src import ai_client
|
|
from src import mma_prompts
|
|
import re
|
|
from typing import Any
|
|
|
|
def generate_tickets(track_brief: str, module_skeletons: str) -> list[dict[str, Any]]:
|
|
"""
|
|
Tier 2 (Tech Lead) call.
|
|
Breaks down a Track Brief and module skeletons into discrete Tier 3 Tickets.
|
|
"""
|
|
# 1. Set Tier 2 Model (Tech Lead - Flash)
|
|
# 2. Construct Prompt
|
|
system_prompt = mma_prompts.PROMPTS.get("tier2_sprint_planning")
|
|
user_message = (
|
|
f"### TRACK BRIEF:\n{track_brief}\n\n"
|
|
f"### MODULE SKELETONS:\n{module_skeletons}\n\n"
|
|
"Please generate the implementation tickets for this track."
|
|
)
|
|
# Set custom system prompt for this call
|
|
old_system_prompt = ai_client._custom_system_prompt
|
|
ai_client.set_custom_system_prompt(system_prompt or "")
|
|
ai_client.set_current_tier("Tier 2")
|
|
last_error = None
|
|
try:
|
|
for _ in range(3):
|
|
try:
|
|
# 3. Call Tier 2 Model
|
|
response = ai_client.send(
|
|
md_content="",
|
|
user_message=user_message
|
|
)
|
|
# 4. Parse JSON Output
|
|
# Extract JSON array from markdown code blocks if present
|
|
json_match = response.strip()
|
|
if "```json" in json_match:
|
|
json_match = json_match.split("```json")[1].split("```")[0].strip()
|
|
elif "```" in json_match:
|
|
json_match = json_match.split("```")[1].split("```")[0].strip()
|
|
# If it's still not valid JSON, try to find a [ ... ] block
|
|
if not (json_match.startswith('[') and json_match.endswith(']')):
|
|
match = re.search(r'\[\s*\{.*\}\s*\]', json_match, re.DOTALL)
|
|
if match:
|
|
json_match = match.group(0)
|
|
tickets: list[dict[str, Any]] = json.loads(json_match)
|
|
return tickets
|
|
except json.JSONDecodeError as e:
|
|
last_error = e
|
|
correction = f"\n\nYour previous output failed to parse as JSON: {e}. Here was your raw output:\n{json_match[:500]}\n\nPlease fix the formatting and output ONLY valid JSON array."
|
|
user_message += correction
|
|
print(f"JSON parsing error, retrying... ({_ + 1}/3)")
|
|
raise RuntimeError(f"Failed to generate valid JSON tickets after 3 attempts. Last error: {last_error}")
|
|
finally:
|
|
# Restore old system prompt and clear tier tag
|
|
ai_client.set_custom_system_prompt(old_system_prompt or "")
|
|
ai_client.set_current_tier(None)
|
|
|
|
from src.dag_engine import TrackDAG
|
|
from src.models import Ticket
|
|
|
|
def topological_sort(tickets: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""
|
|
Sorts a list of tickets based on their 'depends_on' field.
|
|
Raises ValueError if a circular dependency or missing internal dependency is detected.
|
|
"""
|
|
# 1. Convert to Ticket objects for TrackDAG
|
|
ticket_objs = []
|
|
for t_data in tickets:
|
|
ticket_objs.append(Ticket.from_dict(t_data))
|
|
# 2. Use TrackDAG for validation and sorting
|
|
dag = TrackDAG(ticket_objs)
|
|
try:
|
|
sorted_ids = dag.topological_sort()
|
|
except ValueError as e:
|
|
raise ValueError(f"DAG Validation Error: {e}")
|
|
# 3. Return sorted dictionaries
|
|
ticket_map = {t['id']: t for t in tickets}
|
|
return [ticket_map[tid] for tid in sorted_ids]
|
|
|
|
if __name__ == "__main__":
|
|
# Quick test if run directly
|
|
test_brief = "Implement a new feature."
|
|
test_skeletons = "class NewFeature: pass"
|
|
tickets = generate_tickets(test_brief, test_skeletons)
|
|
print(json.dumps(tickets, indent=2))
|