feat(orchestrator): add native_orchestrator.py with plan/metadata CRUD and NativeOrchestrator class

src/native_orchestrator.py

@@ -0,0 +1,88 @@
+from pathlib import Path
+from typing import Optional
+import json
+import re
+
+def read_plan(track_id: str, base_dir: str = ".") -> str:
+ plan_path = Path(base_dir) / "conductor" / "tracks" / track_id / "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:
+ plan_path = Path(base_dir) / "conductor" / "tracks" / track_id / "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]]:
+ 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:
+ meta_path = Path(base_dir) / "conductor" / "tracks" / track_id / "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:
+ meta_path = Path(base_dir) / "conductor" / "tracks" / track_id / "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:
+ return Path(base_dir) / "conductor" / "tracks" / track_id
+
+def get_archive_dir(base_dir: str = ".") -> Path:
+ return Path(base_dir) / "conductor" / "archive"
+
+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)