Skip to main content

#!/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

  1. Each track file is SSOT for that domain
  2. Include task ID in all tool calls: Bash(description="A.9.1.3: action")
  3. Check dependencies before cross-track work

Track Ownership

TrackDomainPrimary Agent
{track_table}

Workflow

  1. Find track: ls TRACK-*.md
  2. Read track: Understand current state
  3. Find task: Look for [ ] or status: pending
  4. Execute: Include task ID in tool calls
  5. Update: Mark [x] or status: complete
  6. 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

TrackDomainProgressFile
{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


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()