manual_slop/src/orchestrator_pm.py

import json

from pathlib import Path
from typing import Any, Optional

from src import aggregate
from src import ai_client
from src import mma_prompts
from src import paths
from src import summarize
from src.models import FileItem
from src.result_types import Result, ErrorInfo, ErrorKind
from src.type_aliases import Metadata


def get_track_history_summary() -> Result[str]:
 """
 Scans conductor/archive/ and conductor/tracks/ to build a summary of past work.
 [C: tests/test_orchestrator_pm_history.py:TestOrchestratorPMHistory.test_get_track_history_summary, tests/test_orchestrator_pm_history.py:TestOrchestratorPMHistory.test_get_track_history_summary_missing_files]
 """
 summary_parts = []
 scan_errors: list[ErrorInfo] = []
 archive_path  = paths.get_archive_dir()
 tracks_path   = paths.get_tracks_dir()
 paths_to_scan = []
 if archive_path.exists(): paths_to_scan.extend(list(archive_path.iterdir()))
 if tracks_path.exists():  paths_to_scan.extend(list(tracks_path.iterdir()))
 for track_dir in paths_to_scan:
  if not track_dir.is_dir(): continue
  metadata_file = track_dir / "metadata.json"
  spec_file     = track_dir / "spec.md"
  title         = track_dir.name
  status        = "unknown"
  overview      = "No overview available."
  if metadata_file.exists():
   try:
    with open(metadata_file, "r", encoding="utf-8") as f:
     meta   = json.load(f)
     title  = meta.get("title", title)
     status = meta.get("status", status)
   except (OSError, json.JSONDecodeError, UnicodeDecodeError) as e:
    scan_errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"orchestrator_pm.get_track_history_summary[{track_dir.name}].metadata", original=e))
  if spec_file.exists():
   try:
    with open(spec_file, "r", encoding="utf-8") as f:
     content = f.read()
     # Basic extraction of Overview section if it exists
     if "## Overview" in content:
      overview = content.split("## Overview")[1].split("##")[0].strip()
     else:
     # Just take a snippet of the beginning
      overview = content[:200] + "..."
   except (OSError, UnicodeDecodeError) as e:
    scan_errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"orchestrator_pm.get_track_history_summary[{track_dir.name}].spec", original=e))
  summary_parts.append(f"Track: {title}\nStatus: {status}\nOverview: {overview}\n---")
 if not summary_parts:
  return Result(data="No previous tracks found.", errors=scan_errors)
 return Result(data="\n".join(summary_parts), errors=scan_errors)

def generate_tracks(user_request: str, project_config: Metadata, file_items: list[FileItem], history_summary: str = "") -> list[Metadata]:
 """
 Tier 1 (Strategic PM) call. 
 Analyzes the project state and user request to generate a list of Tracks.
 [C: tests/test_orchestration_logic.py:test_generate_tracks, tests/test_orchestrator_pm.py:TestOrchestratorPM.test_generate_tracks_malformed_json, tests/test_orchestrator_pm.py:TestOrchestratorPM.test_generate_tracks_markdown_wrapped, tests/test_orchestrator_pm.py:TestOrchestratorPM.test_generate_tracks_success, tests/test_orchestrator_pm_history.py:TestOrchestratorPMHistory.test_generate_tracks_with_history]
 """
 # 1. Build Repository Map (Summary View)
 repo_map = summarize.build_summary_markdown(file_items)
 # 2. Construct Prompt
 system_prompt = mma_prompts.PROMPTS.get("tier1_epic_init")
 user_message_parts = [
  f"### USER REQUEST:\n{user_request}\n",
  f"### REPOSITORY MAP:\n{repo_map}\n"
 ]
 if history_summary:
  user_message_parts.append(f"### TRACK HISTORY:\n{history_summary}\n")
 user_message_parts.append("Please generate the implementation tracks for this request.")
 user_message = "\n".join(user_message_parts)
 # Set custom system prompt for this call
 old_system_prompt = ai_client._custom_system_prompt
 ai_client.set_custom_system_prompt(system_prompt or "")
 # Ensure we use the current provider from ai_client state
 # Import ai_client module-level to access globals
 import src.ai_client as ai_client_module
 current_provider = ai_client_module._provider
 current_model = ai_client_module._model
 ai_client.set_provider(current_provider, current_model)
 try:
 # 3. Call Tier 1 Model (Strategic - Pro)
 # Note: We use gemini-1.5-pro or similar high-reasoning model for Tier 1
  result = ai_client.send(
   md_content="", # We pass everything in user_message for clarity
   user_message=user_message,
   enable_tools=False,
  )
  if not result.ok:
   _err = result.errors[0] if result.errors else None
   _msg = _err.ui_message() if _err else "unknown error"
   print(f"[orchestrator_pm] send failed: {_msg}")
   return []
  response = result.data
  # 4. Parse JSON Output
  try:
  # The prompt asks for a JSON array. We need to extract it if the AI added markdown blocks.
   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()
   tracks: list[dict[str, Any]] = json.loads(json_match)
   # Ensure each track has a 'title' for the GUI
   for t in tracks:
    if "title" not in t:
     t["title"] = t.get("goal", "Untitled Track")[:50]
   return tracks
  except Exception as e:
   _parse_err = Result(data=None, errors=[ErrorInfo(kind=ErrorKind.INVALID_INPUT, message=f"Error parsing Tier 1 response: {e}", source="orchestrator_pm.generate_tracks", original=e)])
   print(f"Error parsing Tier 1 response: {e}")
   print(f"Raw response: {response}")
   return []
 finally:
 # Restore old system prompt
  ai_client.set_custom_system_prompt(old_system_prompt or "")

if __name__ == "__main__":
# Quick CLI test
 import project_manager
 from src import aggregate
 test_project = Path("manual_slop.toml")
 if not test_project.exists():
  print(f"Error: {test_project} not found for testing.")
 else:
  proj = project_manager.load_project(str(test_project))
  flat = project_manager.flat_config(proj)
  file_items = aggregate.build_file_items(Path("."), flat.get("files", {}).get("paths", []))
  print("Testing Tier 1 Track Generation...")
  history = get_track_history_summary()
  tracks = generate_tracks("Implement a basic unit test for the ai_client.py module.", flat, file_items, history_summary=history)
  print(json.dumps(tracks, indent=2))