#!/usr/bin/env python3 """ CODITECT Pilot Plan Migration Script
Losslessly migrates monolithic PILOT-PARALLEL-EXECUTION-PLAN.md to track-based structure per ADR-116.
Usage: python3 scripts/migrate-pilot-to-tracks.py [--dry-run] [--output-dir DIR]
Examples: # Preview migration (no changes) python3 scripts/migrate-pilot-to-tracks.py --dry-run
# Migrate to default location
python3 scripts/migrate-pilot-to-tracks.py
# Migrate to custom location
python3 scripts/migrate-pilot-to-tracks.py --output-dir internal/project/plans/tracks/
"""
import argparse import re import sys from datetime import datetime, timezone from pathlib import Path from typing import Optional
Track metadata
TRACK_INFO = { 'A': {'name': 'Backend Completion', 'domain': 'Backend API', 'agent': 'senior-architect'}, 'B': {'name': 'Frontend Build', 'domain': 'Frontend UI', 'agent': 'frontend-react-typescript-expert'}, 'C': {'name': 'DevOps & Deployment', 'domain': 'DevOps/Infra', 'agent': 'devops-engineer'}, 'D': {'name': 'Security Hardening', 'domain': 'Security', 'agent': 'security-specialist'}, 'E': {'name': 'Integration Testing', 'domain': 'Testing/QA', 'agent': 'testing-specialist'}, 'F': {'name': 'Documentation & Support', 'domain': 'Documentation', 'agent': 'codi-documentation-writer'}, 'G': {'name': 'DMS Integration', 'domain': 'DMS Product', 'agent': 'prompt-analyzer-specialist'}, 'H': {'name': 'Framework Autonomy', 'domain': 'Framework', 'agent': 'senior-architect'}, 'I': {'name': 'UI Components Library', 'domain': 'UI Components', 'agent': 'frontend-react-typescript-expert'}, 'J': {'name': 'Memory Intelligence', 'domain': 'Memory', 'agent': 'senior-architect'}, 'K': {'name': 'Workflow Automation', 'domain': 'Workflow', 'agent': 'devops-engineer'}, 'L': {'name': 'Extended Testing', 'domain': 'Extended Testing', 'agent': 'testing-specialist'}, 'M': {'name': 'Extended Security', 'domain': 'Extended Security', 'agent': 'security-specialist'}, 'N': {'name': 'GTM & Launch', 'domain': 'GTM/Launch', 'agent': 'senior-architect'}, 'T': {'name': 'External Tools', 'domain': 'Tools', 'agent': 'adk-orchestrator'}, }
Track pattern: ## Track X: Name
TRACK_PATTERN = re.compile(r'^## Track ([A-Z]): (.+)$', re.MULTILINE)
def generate_frontmatter(track_letter: str, track_name: str) -> str: """Generate YAML frontmatter for track file.""" info = TRACK_INFO.get(track_letter, { 'name': track_name, 'domain': track_name, 'agent': 'senior-architect' })
now = datetime.now(timezone.utc).strftime('%Y-%m-%d')
return f'''---
type: project-plan component_type: track-plan track: {track_letter} track_name: {info['domain']} status: active version: 1.0.0 created: '2026-01-13' updated: '{now}'
governance: adrs: - ADR-054: Track Nomenclature - ADR-115: Task Specification - ADR-116: Track-Based Architecture standards: - CODITECT-STANDARD-TASK-SPECIFICATION.md
agent: primary: {info['agent']}
moe_confidence: 0.900 moe_classified: {now}
'''
def extract_tracks(pilot_content: str) -> dict[str, str]: """Extract individual tracks from PILOT plan content.""" tracks = {}
# Find all track headers
matches = list(TRACK_PATTERN.finditer(pilot_content))
if not matches:
print("WARNING: No track headers found in PILOT plan", file=sys.stderr)
return tracks
for i, match in enumerate(matches):
track_letter = match.group(1)
track_name = match.group(2)
start_pos = match.start()
# End position is either next track header or end of file
if i + 1 < len(matches):
end_pos = matches[i + 1].start()
else:
end_pos = len(pilot_content)
track_content = pilot_content[start_pos:end_pos].strip()
tracks[track_letter] = {
'name': track_name,
'content': track_content
}
return tracks
def calculate_progress(content: str) -> tuple[int, int]: """Calculate task progress from content.""" completed = len(re.findall(r'[x]', content, re.IGNORECASE)) pending = len(re.findall(r'[ ]', content)) return completed, completed + pending
def generate_track_file(track_letter: str, track_data: dict) -> str: """Generate complete track file content.""" frontmatter = generate_frontmatter(track_letter, track_data['name'])
# Add progress summary after frontmatter
completed, total = calculate_progress(track_data['content'])
if total > 0:
progress_pct = int((completed / total) * 100)
progress_section = f'''
Status Summary
Progress: {progress_pct}% ({completed}/{total} tasks)
''' else: progress_section = ''
return frontmatter + progress_section + track_data['content']
def get_track_filename(track_letter: str, track_name: str) -> str: """Generate filename for track.""" # Normalize name: replace spaces with dashes, remove special chars normalized = re.sub(r'[^a-zA-Z0-9-]', '', track_name.replace(' ', '-')) normalized = re.sub(r'-+', '-', normalized).strip('-').upper() return f'TRACK-{track_letter}-{normalized}.md'
def migrate_pilot_to_tracks( pilot_path: Path, output_dir: Path, dry_run: bool = False ) -> dict: """ Migrate monolithic PILOT plan to track-based structure.
Returns:
dict with migration results
"""
results = {
'tracks_found': 0,
'tracks_migrated': 0,
'files_created': [],
'errors': []
}
# Read PILOT plan
if not pilot_path.exists():
results['errors'].append(f"PILOT plan not found: {pilot_path}")
return results
pilot_content = pilot_path.read_text()
print(f"Read PILOT plan: {len(pilot_content):,} characters")
# Extract tracks
tracks = extract_tracks(pilot_content)
results['tracks_found'] = len(tracks)
print(f"Found {len(tracks)} tracks: {', '.join(sorted(tracks.keys()))}")
if not tracks:
results['errors'].append("No tracks found in PILOT plan")
return results
# Create output directory
if not dry_run:
output_dir.mkdir(parents=True, exist_ok=True)
# Process each track
for track_letter, track_data in sorted(tracks.items()):
filename = get_track_filename(track_letter, track_data['name'])
filepath = output_dir / filename
track_content = generate_track_file(track_letter, track_data)
completed, total = calculate_progress(track_data['content'])
print(f" Track {track_letter}: {track_data['name']}")
print(f" → {filename} ({len(track_content):,} chars, {completed}/{total} tasks)")
if dry_run:
print(f" [DRY RUN] Would create: {filepath}")
else:
filepath.write_text(track_content)
print(f" ✓ Created: {filepath}")
results['tracks_migrated'] += 1
results['files_created'].append(str(filepath))
# Generate summary files
if not dry_run:
# Create CLAUDE.md for tracks directory
claude_md = generate_tracks_claude_md(tracks)
(output_dir / 'CLAUDE.md').write_text(claude_md)
print(f" ✓ Created: {output_dir / 'CLAUDE.md'}")
# Create README.md for tracks directory
readme_md = generate_tracks_readme(tracks)
(output_dir / 'README.md').write_text(readme_md)
print(f" ✓ Created: {output_dir / 'README.md'}")
return results
def generate_tracks_claude_md(tracks: dict) -> str: """Generate CLAUDE.md for tracks directory.""" now = datetime.now(timezone.utc).strftime('%Y-%m-%d')
track_table = ""
for letter in sorted(tracks.keys()):
info = TRACK_INFO.get(letter, {'domain': tracks[letter]['name'], 'agent': 'senior-architect'})
track_table += f"| **{letter}** | {info['domain']} | `{info['agent']}` |\n"
return f'''---
title: Project Plan Tracks - AI Agent Context type: reference component_type: claude-md version: 1.0.0 created: '{now}' updated: '{now}'
Project Plan Tracks - AI Agent Context
Status: Active Migrated From: PILOT-PARALLEL-EXECUTION-PLAN.md Migration Date: {now}
Critical Rules
- Each track file is SSOT for that domain
- Include task ID in all tool calls:
Bash(description="A.9.1.3: action") - Check dependencies before cross-track work
Track Ownership
| Track | Domain | Primary Agent |
|---|---|---|
| {track_table} |
Workflow
- Find track:
ls TRACK-*.md - Read track: Understand current state
- Find task: Look for
[ ]orstatus: pending - Execute: Include task ID in tool calls
- Update: Mark
[x]orstatus: complete - Commit: Changes to track file only
Governing ADRs
- ADR-054: Track Nomenclature
- ADR-115: Task Specification
- ADR-116: Track-Based Architecture
- ADR-117: Plan Location Strategy
Updated: {now} '''
def generate_tracks_readme(tracks: dict) -> str: """Generate README.md for tracks directory.""" now = datetime.now(timezone.utc).strftime('%Y-%m-%d')
track_status = ""
for letter in sorted(tracks.keys()):
data = tracks[letter]
completed, total = calculate_progress(data['content'])
progress = int((completed / total) * 100) if total > 0 else 0
status = "✅" if progress == 100 else "🟡" if progress >= 80 else "🔴"
filename = get_track_filename(letter, data['name'])
track_status += f"| {letter} | {data['name']} | {status} {progress}% | [{filename}]({filename}) |\n"
return f'''---
title: CODITECT Pilot Tracks Index type: reference component_type: project-plan version: 1.0.0 created: '{now}' updated: '{now}'
CODITECT Pilot Tracks Index
Migrated From: PILOT-PARALLEL-EXECUTION-PLAN.md Migration Date: {now} Total Tracks: {len(tracks)}
Track Status
| Track | Domain | Progress | File |
|---|---|---|---|
| {track_status} |
Quick Reference
# Find pending tasks in a track
grep '\\[ \\]' TRACK-A-*.md
# Count pending tasks per track
for f in TRACK-*.md; do echo "$(basename $f): $(grep -c '\\[ \\]' $f 2>/dev/null || echo 0)"; done
Governance
- ADR-054: Track Nomenclature
- ADR-115: Task Specification
- ADR-116: Track-Based Architecture
- ADR-117: Plan Location Strategy
For AI agents: See CLAUDE.md '''
def main(): parser = argparse.ArgumentParser( description="Migrate PILOT plan to track-based structure (ADR-116)", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=doc ) parser.add_argument( '--dry-run', '-n', action='store_true', help='Preview migration without creating files' ) parser.add_argument( '--output-dir', '-o', type=Path, default=None, help='Output directory for track files' ) parser.add_argument( '--pilot-file', '-p', type=Path, default=None, help='Path to PILOT plan file' )
args = parser.parse_args()
# Determine paths
script_dir = Path(__file__).parent
repo_root = script_dir.parent
pilot_path = args.pilot_file or (
repo_root / 'internal' / 'project' / 'plans' / 'PILOT-PARALLEL-EXECUTION-PLAN.md'
)
output_dir = args.output_dir or (
repo_root / 'internal' / 'project' / 'plans' / 'tracks'
)
print("=" * 60)
print("CODITECT Pilot Plan Migration (ADR-116)")
print("=" * 60)
print(f"Source: {pilot_path}")
print(f"Output: {output_dir}")
print(f"Mode: {'DRY RUN' if args.dry_run else 'LIVE'}")
print("=" * 60)
results = migrate_pilot_to_tracks(pilot_path, output_dir, args.dry_run)
print("")
print("=" * 60)
print("MIGRATION SUMMARY")
print("=" * 60)
print(f"Tracks found: {results['tracks_found']}")
print(f"Tracks migrated: {results['tracks_migrated']}")
print(f"Files created: {len(results['files_created'])}")
if results['errors']:
print(f"Errors: {len(results['errors'])}")
for error in results['errors']:
print(f" ❌ {error}")
sys.exit(1)
else:
print("✅ Migration complete!")
if args.dry_run:
print("\nTo perform actual migration, run without --dry-run")
if name == 'main': main()