98 lines
3.6 KiB
Python
98 lines
3.6 KiB
Python
from pathlib import Path
|
|
from typing import Optional
|
|
import json
|
|
import re
|
|
from src import paths
|
|
|
|
def read_plan(track_id: str, base_dir: str = ".") -> str:
|
|
"""Reads the implementation plan (plan.md) for a track."""
|
|
plan_path = paths.get_track_state_dir(track_id, base_dir) / "plan.md"
|
|
if not plan_path.exists():
|
|
return ""
|
|
return plan_path.read_text(encoding="utf-8")
|
|
|
|
def write_plan(track_id: str, content: str, base_dir: str = ".") -> None:
|
|
"""Writes the implementation plan (plan.md) for a track."""
|
|
plan_path = paths.get_track_state_dir(track_id, base_dir) / "plan.md"
|
|
plan_path.parent.mkdir(parents=True, exist_ok=True)
|
|
plan_path.write_text(content, encoding="utf-8")
|
|
|
|
def parse_plan_tasks(content: str) -> list[dict[str, str]]:
|
|
"""Parses the tasks from a plan.md file."""
|
|
tasks = []
|
|
for line in content.split("\n"):
|
|
stripped = line.strip()
|
|
if stripped.startswith("- ["):
|
|
checked = "[x]" in stripped
|
|
text = stripped[4:] if len(stripped) > 4 else ""
|
|
tasks.append({"text": text, "completed": checked})
|
|
return tasks
|
|
|
|
def read_metadata(track_id: str, base_dir: str = ".") -> dict:
|
|
"""Reads the metadata (metadata.json) for a track."""
|
|
meta_path = paths.get_track_state_dir(track_id, base_dir) / "metadata.json"
|
|
if not meta_path.exists():
|
|
return {}
|
|
return json.loads(meta_path.read_text(encoding="utf-8"))
|
|
|
|
def write_metadata(track_id: str, data: dict, base_dir: str = ".") -> None:
|
|
"""Writes the metadata (metadata.json) for a track."""
|
|
meta_path = paths.get_track_state_dir(track_id, base_dir) / "metadata.json"
|
|
meta_path.parent.mkdir(parents=True, exist_ok=True)
|
|
meta_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
|
|
|
|
def get_track_dir(track_id: str, base_dir: str = ".") -> Path:
|
|
"""Returns the state directory for a specific track."""
|
|
return paths.get_track_state_dir(track_id, base_dir)
|
|
|
|
def get_archive_dir(base_dir: str = ".") -> Path:
|
|
"""Returns the central archive directory for completed tracks."""
|
|
return paths.get_archive_dir(base_dir)
|
|
|
|
class NativeOrchestrator:
|
|
def __init__(self, base_dir: str = "."):
|
|
self.base_dir = Path(base_dir)
|
|
self._conductor = None
|
|
|
|
def load_track(self, track_id: str) -> dict:
|
|
"""Load track from metadata.json"""
|
|
return read_metadata(track_id, str(self.base_dir))
|
|
|
|
def save_track(self, track_id: str, data: dict) -> None:
|
|
"""Persist track metadata"""
|
|
write_metadata(track_id, data, str(self.base_dir))
|
|
|
|
def load_plan(self, track_id: str) -> str:
|
|
"""Load plan.md content"""
|
|
return read_plan(track_id, str(self.base_dir))
|
|
|
|
def save_plan(self, track_id: str, content: str) -> None:
|
|
"""Persist plan.md content"""
|
|
write_plan(track_id, content, str(self.base_dir))
|
|
|
|
def get_track_tasks(self, track_id: str) -> list[dict]:
|
|
"""Get parsed task list from plan.md"""
|
|
content = self.load_plan(track_id)
|
|
return parse_plan_tasks(content)
|
|
|
|
def generate_tickets(self, brief: str, module_skeletons: str = "") -> list[dict]:
|
|
"""Tier 2: Generate tickets from brief"""
|
|
from src import conductor_tech_lead
|
|
return conductor_tech_lead.generate_tickets(brief, module_skeletons)
|
|
|
|
def execute_ticket(self, ticket: dict, context: str) -> str:
|
|
"""Tier 3: Execute single ticket"""
|
|
from src import ai_client
|
|
return ai_client.send(context, ticket.get("description", ""), str(self.base_dir))
|
|
|
|
def analyze_error(self, error: str) -> str:
|
|
"""Tier 4: Analyze error"""
|
|
from src import ai_client
|
|
return ai_client.run_tier4_analysis(error)
|
|
|
|
def run_tier4_patch(self, error: str, file_context: str) -> str:
|
|
"""Tier 4: Generate patch for error"""
|
|
from src import ai_client
|
|
return ai_client.run_tier4_patch_generation(error, file_context)
|
|
|