#!/usr/bin/env python3 """ CODITECT Ralph Loop PILOT Integration (H.8.6.4)
Bridges the PILOT plan TRACK task system with Ralph Wiggum autonomous loops. Reads pending tasks from TRACK files, groups by section, and generates LoopConfig parameters for autonomous execution.
Usage: from scripts.core.ralph_wiggum.pilot_integration import ( PilotTaskExtractor, TaskGroup, generate_loop_goal, )
extractor = PilotTaskExtractor()
groups = extractor.get_pending_groups("H")
for group in groups:
goal = generate_loop_goal(group)
# Start Ralph loop with goal
Author: CODITECT Framework Version: 1.0.0 Created: 2026-02-17 Task Reference: H.8.6.4 ADR References: ADR-108, ADR-110, ADR-111, ADR-054 """
import logging import os import re import sys from dataclasses import dataclass, field from pathlib import Path from typing import Dict, List, Optional, Tuple
Handle imports for both module and direct execution
try: from scripts.core.paths import get_tracks_dir except ModuleNotFoundError: _script_dir = Path(file).resolve().parent _core_root = _script_dir.parent.parent.parent if str(_core_root) not in sys.path: sys.path.insert(0, str(_core_root)) from scripts.core.paths import get_tracks_dir
logger = logging.getLogger("ralph-pilot-integration")
TRACK file location (ADR-213: coditect-documentation is SSOT)
TRACKS_DIR = Path(os.environ.get("CODITECT_TRACKS", "")) if os.environ.get("CODITECT_TRACKS") else get_tracks_dir()
@dataclass class PilotTask: """A single task from a TRACK file.""" task_id: str # e.g., "H.8.6.4" description: str # e.g., "Integrate with PILOT plan task execution" hours: int = 0 # Estimated hours completed: bool = False agent_hint: str = "" # Suggested agent from TRACK file
@dataclass class TaskGroup: """A group of related tasks (same section).""" track: str # e.g., "H" section: str # e.g., "H.8.6" section_name: str # e.g., "Loop Orchestration" tasks: List[PilotTask] = field(default_factory=list) total_hours: int = 0 pending_count: int = 0 completed_count: int = 0
@property
def completion_pct(self) -> int:
total = self.pending_count + self.completed_count
if total == 0:
return 0
return int(self.completed_count / total * 100)
class PilotTaskExtractor: """ Extract pending tasks from PILOT TRACK files.
Reads TRACK-{X}-*.md files and parses checkbox items
in the format: - [ ] X.n.n.n: Description - Nh
"""
TASK_RE = re.compile(
r"^- \[( |x)\] ([\w.]+):\s+(.+?)(?:\s*-\s*(\d+)h)?$"
)
SECTION_RE = re.compile(
r"^#{3,4}\s+([\w.]+):\s+(.+?)(?:\s*-\s*~?\d+h)?$"
)
AGENT_RE = re.compile(
r'\*\*Agent:\*\*\s+`/agent\s+(\S+)'
)
def __init__(self, tracks_dir: Optional[Path] = None):
self.tracks_dir = tracks_dir or TRACKS_DIR
def find_track_file(self, track: str) -> Optional[Path]:
"""Find the TRACK file for a given track letter."""
pattern = f"TRACK-{track.upper()}-*.md"
matches = list(self.tracks_dir.glob(pattern))
if matches:
return matches[0]
return None
def parse_track(self, track: str) -> List[PilotTask]:
"""Parse all tasks from a TRACK file."""
track_file = self.find_track_file(track)
if not track_file or not track_file.exists():
logger.warning(f"Track file not found for track {track}")
return []
tasks = []
lines = track_file.read_text().splitlines()
for i, line in enumerate(lines):
match = self.TASK_RE.match(line.strip())
if match:
completed = match.group(1) == "x"
task_id = match.group(2)
description = match.group(3).strip()
hours = int(match.group(4)) if match.group(4) else 0
# Look for agent hint in next few lines
agent_hint = ""
for j in range(i + 1, min(i + 3, len(lines))):
agent_match = self.AGENT_RE.search(lines[j])
if agent_match:
agent_hint = agent_match.group(1)
break
tasks.append(PilotTask(
task_id=task_id,
description=description,
hours=hours,
completed=completed,
agent_hint=agent_hint,
))
return tasks
def get_pending_tasks(self, track: str) -> List[PilotTask]:
"""Get only pending (unchecked) tasks for a track."""
return [t for t in self.parse_track(track) if not t.completed]
def get_pending_groups(
self, track: str, min_section_depth: int = 3
) -> List[TaskGroup]:
"""
Group pending tasks by section.
Returns TaskGroups where each group contains tasks from the same
section (e.g., H.8.6 groups H.8.6.1, H.8.6.2, etc.)
"""
all_tasks = self.parse_track(track)
sections: Dict[str, TaskGroup] = {}
for task in all_tasks:
parts = task.task_id.split(".")
if len(parts) >= min_section_depth:
section_id = ".".join(parts[:min_section_depth])
else:
section_id = task.task_id
if section_id not in sections:
sections[section_id] = TaskGroup(
track=track.upper(),
section=section_id,
section_name="",
tasks=[],
)
group = sections[section_id]
if task.completed:
group.completed_count += 1
else:
group.pending_count += 1
group.tasks.append(task)
group.total_hours += task.hours
# Filter to groups with pending tasks
return [g for g in sections.values() if g.pending_count > 0]
def get_next_batch(
self, track: str, max_hours: int = 40, max_tasks: int = 10
) -> List[PilotTask]:
"""
Get the next batch of tasks suitable for a Ralph loop.
Selects tasks up to max_hours/max_tasks, preferring same-section
tasks for coherent execution.
"""
groups = self.get_pending_groups(track)
if not groups:
return []
# Pick the first group with pending tasks
batch = []
total_hours = 0
for group in groups:
for task in group.tasks:
if len(batch) >= max_tasks:
break
if total_hours + task.hours > max_hours and batch:
break
batch.append(task)
total_hours += task.hours
if batch:
break
return batch
def generate_loop_goal(group: TaskGroup) -> str: """Generate a goal string for a Ralph loop from a task group.""" task_list = "\n".join( f" - {t.task_id}: {t.description}" for t in group.tasks ) return ( f"Execute {group.section} tasks ({group.pending_count} remaining):\n" f"{task_list}\n" f"Mark each task [x] in TRACK-{group.track} after completion." )
def generate_loop_config( batch: List[PilotTask], agent_type: str = "senior-architect", model: str = "claude-opus-4-6", max_cost: float = 50.0, ) -> dict: """ Generate LoopConfig parameters from a batch of tasks.
Returns a dict suitable for LoopConfig(**result).
"""
total_hours = sum(t.hours for t in batch)
# Estimate iterations: ~1 per 2-4 hours of work
est_iterations = max(3, min(15, total_hours // 2))
# Use agent hint from first task if available
if batch and batch[0].agent_hint:
agent_type = batch[0].agent_hint
task_ids = [t.task_id for t in batch]
task_range = f"{task_ids[0]}-{task_ids[-1]}" if len(task_ids) > 1 else task_ids[0]
return {
"max_iterations": est_iterations,
"max_cost": max_cost,
"max_duration_minutes": max(60, total_hours * 15),
"agent_type": agent_type,
"model": model,
"metadata": {
"task_ids": task_ids,
"task_range": task_range,
"estimated_hours": total_hours,
"source": "pilot-track",
},
}
def format_batch_summary(batch: List[PilotTask]) -> str: """Format a batch of tasks for display.""" if not batch: return "No pending tasks found."
lines = [
"PILOT TASK BATCH FOR RALPH LOOP",
"=" * 40,
"",
]
total_hours = 0
for task in batch:
lines.append(f" {task.task_id}: {task.description} ({task.hours}h)")
if task.agent_hint:
lines.append(f" Agent: {task.agent_hint}")
total_hours += task.hours
lines.extend([
"",
f"Total: {len(batch)} tasks, ~{total_hours}h estimated",
"",
"Suggested invocation:",
f" /ralph-loop start --task {batch[0].task_id.rsplit('.', 1)[0]}",
f' --goal "Execute {len(batch)} PILOT tasks"',
f" --max-iterations {max(3, total_hours // 2)}",
f" --max-cost {min(100, total_hours * 3):.0f}",
])
if batch[0].agent_hint:
lines.append(f" --agent {batch[0].agent_hint}")
return "\n".join(lines)