manual_slop/src/conductor_tech_lead.py

"""
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
import re

from typing import Any

from src import ai_client
from src import mma_prompts


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.
 [C: tests/test_conductor_tech_lead.py:TestConductorTechLead.test_generate_tickets_retry_failure, tests/test_conductor_tech_lead.py:TestConductorTechLead.test_generate_tickets_retry_success, tests/test_conductor_tech_lead.py:TestConductorTechLead.test_generate_tickets_success, tests/test_orchestration_logic.py:test_generate_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
    result = ai_client.send(
     md_content   = "",
     user_message = user_message
    )
    if not result.ok:
     _err = result.errors[0] if result.errors else None
     _msg = _err.ui_message() if _err else "unknown error"
     print(f"[conductor_tech_lead] send failed: {_msg}")
     return None
    response = result.data
    # 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.mma import Ticket
from src.result_types import ErrorInfo, ErrorKind, Result

def topological_sort(tickets: list[Ticket]) -> list[Ticket]:
 """
 Sorts a list of Ticket objects based on their depends_on field.
 Raises ValueError if a circular dependency or missing internal dependency is detected.
 [C: tests/test_conductor_tech_lead.py:TestTopologicalSort.test_topological_sort_complex, tests/test_conductor_tech_lead.py:TestTopologicalSort.test_topological_sort_cycle, tests/test_conductor_tech_lead.py:TestTopologicalSort.test_topological_sort_empty, tests/test_conductor_tech_lead.py:TestTopologicalSort.test_topological_sort_linear, tests/test_conductor_tech_lead.py:TestTopologicalSort.test_topological_sort_missing_dependency, tests/test_conductor_tech_lead.py:test_topological_sort_vlog, tests/test_dag_engine.py:test_topological_sort, tests/test_dag_engine.py:test_topological_sort_cycle, tests/test_orchestration_logic.py:test_topological_sort, tests/test_orchestration_logic.py:test_topological_sort_circular, tests/test_perf_dag.py:test_dag_edge_cases, tests/test_perf_dag.py:test_dag_performance]
 """
 dag = TrackDAG(tickets)
 try:
  sorted_ids = dag.topological_sort()
 except ValueError as e:
  _dag_err = Result(data=None, errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"DAG Validation Error: {e}", source="conductor_tech_lead.topological_sort", original=e)])
  raise ValueError(f"DAG Validation Error: {e}")
 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))