scripts-analyze-pilot-hierarchy

#!/usr/bin/env python3 """

title: "Database path" component_type: script version: "1.0.0" audience: contributor status: stable summary: "Analyze Pilot Launch Work Item Hierarchy Validates F010.2 structure, parent relationships, an..." keywords: ['analysis', 'analyze', 'database', 'git', 'hierarchy'] tokens: ~500 created: 2025-12-22 updated: 2025-12-22 script_name: "analyze_pilot_hierarchy.py" language: python executable: true usage: "python3 scripts/analyze_pilot_hierarchy.py [options]" python_version: "3.10+" dependencies: [] modifies_files: false network_access: false requires_auth: false

Analyze Pilot Launch Work Item Hierarchy Validates F010.2 structure, parent relationships, and naming conventions """

import sqlite3 import json from datetime import datetime from pathlib import Path from collections import defaultdict

ADR-114 & ADR-118: Use centralized path discovery

work_items table is in sessions.db (Tier 3: regenerable task data)

SCRIPT_DIR = Path(file).resolve().parent import sys sys.path.insert(0, str(SCRIPT_DIR / "core")) try: from paths import get_sessions_db_path, SESSIONS_DB DB_PATH = SESSIONS_DB except ImportError: _user_data = Path.home() / "PROJECTS" / ".coditect-data" / "context-storage" if _user_data.exists(): DB_PATH = _user_data / "sessions.db" else: DB_PATH = SCRIPT_DIR.parent / "context-storage" / "sessions.db"

def analyze_hierarchy(): """Analyze pilot launch work item hierarchy"""

conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()

print("=" * 80)
print("PILOT LAUNCH WORK ITEM HIERARCHY ANALYSIS")
print("=" * 80)
print()

# 1. Verify F010.2 exists
print("1. FEATURE F010.2 VERIFICATION")
print("-" * 80)

cursor.execute("""
    SELECT id, title, work_item_type, status, priority, due_date, parent_id
    FROM work_items
    WHERE id = 'F010.2'
""")

feature = cursor.fetchone()
if feature:
    print(f"✓ Found: {feature['id']} - {feature['title']}")
    print(f"  Type: {feature['work_item_type']}")
    print(f"  Status: {feature['status']}")
    print(f"  Priority: {feature['priority']}")
    print(f"  Due Date: {feature['due_date']}")
    print(f"  Parent: {feature['parent_id']}")
else:
    print("✗ ERROR: F010.2 not found!")
    return

print()

# 2. Get all tasks (T010.2.XXX)
print("2. TASK INVENTORY (T010.2.XXX)")
print("-" * 80)

cursor.execute("""
    SELECT id, title, parent_id, work_item_type, status, due_date, priority
    FROM work_items
    WHERE id LIKE 'T010.2.%' AND id NOT LIKE 'T010.2.%.%'
    ORDER BY id
""")

tasks = cursor.fetchall()
print(f"Total Tasks: {len(tasks)}\n")

task_issues = []
task_by_id = {}

for task in tasks:
    task_by_id[task['id']] = dict(task)

    # Validate parent_id
    if task['parent_id'] != 'F010.2':
        task_issues.append(f"✗ {task['id']}: Invalid parent_id '{task['parent_id']}' (should be 'F010.2')")

    # Validate work_item_type
    if task['work_item_type'] != 'task':
        task_issues.append(f"✗ {task['id']}: Invalid type '{task['work_item_type']}' (should be 'task')")

    # Validate ID format (T010.2.XXX)
    parts = task['id'].split('.')
    if len(parts) != 3 or not parts[2].isdigit():
        task_issues.append(f"✗ {task['id']}: Invalid ID format (should be T010.2.XXX)")

if task_issues:
    print("ISSUES FOUND:")
    for issue in task_issues:
        print(f"  {issue}")
else:
    print("✓ All tasks have correct parent_id and type")

print()

# 3. Get all subtasks (ST010.2.XXX.YY)
print("3. SUBTASK INVENTORY (ST010.2.XXX.YY)")
print("-" * 80)

cursor.execute("""
    SELECT id, title, parent_id, work_item_type, status, due_date, priority
    FROM work_items
    WHERE id LIKE 'ST010.2.%'
    ORDER BY id
""")

subtasks = cursor.fetchall()
print(f"Total Subtasks: {len(subtasks)}\n")

subtask_issues = []
subtask_by_parent = defaultdict(list)

for subtask in subtasks:
    subtask_by_parent[subtask['parent_id']].append(dict(subtask))

    # Validate parent_id exists in tasks
    if subtask['parent_id'] not in task_by_id:
        subtask_issues.append(f"✗ {subtask['id']}: Parent '{subtask['parent_id']}' not found (orphaned)")

    # Validate work_item_type
    if subtask['work_item_type'] != 'subtask':
        subtask_issues.append(f"✗ {subtask['id']}: Invalid type '{subtask['work_item_type']}' (should be 'subtask')")

    # Validate ID format (ST010.2.XXX.YY)
    parts = subtask['id'].split('.')
    if len(parts) != 4 or not parts[2].isdigit() or not parts[3].isdigit():
        subtask_issues.append(f"✗ {subtask['id']}: Invalid ID format (should be ST010.2.XXX.YY)")

    # Validate parent relationship matches ID pattern
    expected_parent = f"T{subtask['id'][2:]}"  # ST010.2.001.01 -> T010.2.001
    expected_parent = '.'.join(expected_parent.split('.')[:-1])
    if subtask['parent_id'] != expected_parent:
        subtask_issues.append(f"✗ {subtask['id']}: Parent mismatch (has '{subtask['parent_id']}', expected '{expected_parent}')")

if subtask_issues:
    print("ISSUES FOUND:")
    for issue in subtask_issues:
        print(f"  {issue}")
else:
    print("✓ All subtasks have correct parent relationships")

print()

# 4. Status validation
print("4. STATUS DISTRIBUTION")
print("-" * 80)

cursor.execute("""
    SELECT status, COUNT(*) as count
    FROM work_items
    WHERE id LIKE 'F010.2%' OR id LIKE 'T010.2%' OR id LIKE 'ST010.2%'
    GROUP BY status
    ORDER BY count DESC
""")

status_counts = cursor.fetchall()
for row in status_counts:
    print(f"  {row['status']}: {row['count']}")

print()

# 5. Due date validation (Dec 17-24)
print("5. DUE DATE VALIDATION (Dec 17-24, 2025)")
print("-" * 80)

cursor.execute("""
    SELECT id, title, due_date, status
    FROM work_items
    WHERE (id LIKE 'T010.2%' OR id LIKE 'ST010.2%')
    AND (due_date < '2025-12-17' OR due_date > '2025-12-24')
    ORDER BY due_date
""")

date_issues = cursor.fetchall()
if date_issues:
    print(f"✗ {len(date_issues)} items with due dates outside Dec 17-24:")
    for item in date_issues[:10]:  # Show first 10
        print(f"  {item['id']}: {item['due_date']} - {item['title'][:50]}")
    if len(date_issues) > 10:
        print(f"  ... and {len(date_issues) - 10} more")
else:
    print("✓ All items have due dates within Dec 17-24 range")

print()

# 6. Hierarchy summary
print("6. HIERARCHY SUMMARY")
print("-" * 80)

total_items = 1 + len(tasks) + len(subtasks)  # Feature + tasks + subtasks
print(f"Feature F010.2: 1")
print(f"Tasks (T010.2.XXX): {len(tasks)}")
print(f"Subtasks (ST010.2.XXX.YY): {len(subtasks)}")
print(f"Total Work Items: {total_items}")
print()

# Expected: 1 feature + 16 tasks + 95 subtasks = 112 items
expected_tasks = 16
expected_subtasks = 95
expected_total = 1 + expected_tasks + expected_subtasks

print(f"Expected: {expected_total} (1 feature + {expected_tasks} tasks + {expected_subtasks} subtasks)")

if total_items != expected_total:
    print(f"✗ WARNING: Count mismatch! Found {total_items}, expected {expected_total}")
else:
    print("✓ Total count matches expected")

print()

# 7. Orphaned items check
print("7. ORPHANED ITEMS CHECK")
print("-" * 80)

cursor.execute("""
    SELECT w1.id, w1.title, w1.parent_id, w1.work_item_type
    FROM work_items w1
    WHERE (w1.id LIKE 'T010.2%' OR w1.id LIKE 'ST010.2%')
    AND w1.parent_id IS NOT NULL
    AND NOT EXISTS (
        SELECT 1 FROM work_items w2 WHERE w2.id = w1.parent_id
    )
    ORDER BY w1.id
""")

orphaned = cursor.fetchall()
if orphaned:
    print(f"✗ Found {len(orphaned)} orphaned items:")
    for item in orphaned:
        print(f"  {item['id']}: parent '{item['parent_id']}' not found")
else:
    print("✓ No orphaned items found")

print()

# 8. Task tree structure
print("8. TASK TREE STRUCTURE (Sample)")
print("-" * 80)

# Show first 3 tasks with their subtasks
for i, task in enumerate(tasks[:3]):
    print(f"\n{task['id']}: {task['title']}")
    print(f"  Status: {task['status']} | Due: {task['due_date']}")

    children = subtask_by_parent.get(task['id'], [])
    if children:
        print(f"  Subtasks ({len(children)}):")
        for subtask in children[:3]:  # Show first 3 subtasks
            print(f"    └─ {subtask['id']}: {subtask['title'][:60]}")
        if len(children) > 3:
            print(f"    └─ ... and {len(children) - 3} more")
    else:
        print("  No subtasks")

print()

# 9. Final recommendations
print("9. RECOMMENDATIONS")
print("-" * 80)

recommendations = []

if task_issues:
    recommendations.append(f"Fix {len(task_issues)} task-level issues (see section 2)")

if subtask_issues:
    recommendations.append(f"Fix {len(subtask_issues)} subtask-level issues (see section 3)")

if date_issues:
    recommendations.append(f"Review {len(date_issues)} items with dates outside Dec 17-24 (see section 5)")

if orphaned:
    recommendations.append(f"Fix {len(orphaned)} orphaned items (see section 7)")

if total_items != expected_total:
    recommendations.append(f"Investigate count mismatch: found {total_items}, expected {expected_total}")

if not recommendations:
    print("✓ Hierarchy is valid and complete!")
    print("\nAll checks passed:")
    print("  • F010.2 feature exists with correct attributes")
    print("  • All tasks have correct parent_id (F010.2)")
    print("  • All subtasks have valid parent tasks")
    print("  • ID naming follows V2 convention")
    print("  • No orphaned work items")
    print("  • Due dates within pilot timeline")
else:
    print("⚠ Issues found that need attention:\n")
    for i, rec in enumerate(recommendations, 1):
        print(f"  {i}. {rec}")

print()
print("=" * 80)

conn.close()

if name == "main": analyze_hierarchy()