rune/shared/dependency_tracker.py

#!/usr/bin/env python3
"""dependency_tracker.py — Auto-propagation system for umbrella dependencies.

When a component changes, automatically updates dependent components.
Maps the dependency graph between umbrella systems and fires
propagation rules when changes are detected.

Usage:
    dependency_tracker.py check <changed_file>
    dependency_tracker.py propagate <changed_file>
    dependency_tracker.py map
"""

import json
import os
import sys
from datetime import datetime, timezone
from typing import Dict, List, Set

SCRIPTS_DIR = os.path.expanduser("~/.config/opencode/scripts")
DEPENDENCY_MAP_FILE = os.path.expanduser("~/.config/opencode/memory/dependency_map.json")

# ═══════════════════════════════════════════════════════════════════════════════
# DEPENDENCY GRAPH
# ═══════════════════════════════════════════════════════════════════════════════

# Maps file → list of files that depend on it
DEPENDENCY_GRAPH = {
    # MEMORY umbrella
    "memory_store.py": [
        "warm_router.py", "curator.py", "memory_consolidation.py",
        "drift_core.py", "drift_cache.py", "tier_manager.py", "shard_manager.py",
        "query_memory.py", "growth_simulator.py"
    ],
    "memory_consolidation.py": [
        "memory_store.py", "drift_compression.py", "curator.py"
    ],
    "warm_router.py": [
        "capability_broker.py", "memory_consolidation.py", "memory_store.py"
    ],
    "graph_memory.py": [
        "warm_router.py", "memory_consolidation.py", "shard_manager.py"
    ],
    "shard_manager.py": [
        "memory_store.py", "tier_manager.py"
    ],
    "tier_manager.py": [
        "memory_store.py", "drift_cache.py"
    ],
    "query_memory.py": [
        "memory_store.py", "growth_simulator.py"
    ],

    # GOVERNANCE umbrella
    "drift_core.py": [
        "curator.py", "audit_ledger.py", "memory_store.py",
        "drift_observability.py", "drift_cache.py"
    ],
    "curator.py": [
        "memory_store.py", "drift_core.py", "audit_ledger.py"
    ],
    "drift_compression.py": [
        "memory_consolidation.py", "curator.py"
    ],
    "drift_observability.py": [
        "drift_core.py", "audit_ledger.py"
    ],
    "audit_ledger.py": [
        "curator.py", "drift_core.py"
    ],

    # IDENTITY umbrella
    "persona_definition.py": [
        "persona_check.py", "identity_anchors.py", "drift_observability.py"
    ],
    "identity_anchors.py": [
        "persona_definition.py", "drift_core.py"
    ],
    "voice_hybrid.py": [
        "delivery_adapter.py", "persona_definition.py"
    ],

    # GROWTH umbrella
    "growth_delta.py": [
        "memory_store.py", "curator.py", "lessons.py"
    ],
    "growth_simulator.py": [
        "memory_store.py", "query_memory.py", "drift_core.py"
    ],

    # PERCEPTION umbrella
    "warm_router.py": [
        "capability_broker.py", "memory_consolidation.py"
    ],

    # INFRASTRUCTURE
    "microkernel.py": [
        "hook_engine.py", "event_bus.py"
    ],
}

# ═══════════════════════════════════════════════════════════════════════════════
# PROPAGATION ACTIONS
# ═══════════════════════════════════════════════════════════════════════════════

def get_dependents(changed_file: str) -> List[str]:
    """Get files that depend on the changed file."""
    basename = os.path.basename(changed_file)
    return DEPENDENCY_GRAPH.get(basename, [])


def check_propagation(changed_file: str) -> dict:
    """Check what needs to be updated when a file changes."""
    dependents = get_dependents(changed_file)
    return {
        "changed_file": os.path.basename(changed_file),
        "dependents": dependents,
        "count": len(dependents),
    }


def propagate(changed_file: str, dry_run: bool = False) -> dict:
    """Propagate changes to dependent files.

    Returns dict with propagation results.
    """
    dependents = get_dependents(changed_file)
    results = {
        "changed_file": os.path.basename(changed_file),
        "propagated": [],
        "skipped": [],
        "errors": [],
    }

    for dep in dependents:
        dep_path = os.path.join(SCRIPTS_DIR, dep)
        if not os.path.exists(dep_path):
            results["skipped"].append({"file": dep, "reason": "not found"})
            continue

        if dry_run:
            results["propagated"].append({"file": dep, "action": "would update"})
            continue

        # Check if the dependent file imports from the changed file
        try:
            with open(dep_path) as f:
                content = f.read()

            changed_basename = os.path.basename(changed_file).replace(".py", "")
            if changed_basename in content:
                results["propagated"].append({
                    "file": dep,
                    "action": "references changed file",
                    "detail": f"imports or references {changed_basename}"
                })
            else:
                results["skipped"].append({"file": dep, "reason": "no direct reference"})
        except Exception as e:
            results["errors"].append({"file": dep, "error": str(e)})

    return results


def get_full_map() -> dict:
    """Get the full dependency map."""
    return DEPENDENCY_GRAPH


# ═══════════════════════════════════════════════════════════════════════════════
# CLI
# ═══════════════════════════════════════════════════════════════════════════════

def main():
    import argparse
    parser = argparse.ArgumentParser(description="Dependency tracker for umbrella systems")
    sub = parser.add_subparsers(dest="cmd")

    p_check = sub.add_parser("check", help="Check what depends on a file")
    p_check.add_argument("file")

    p_prop = sub.add_parser("propagate", help="Propagate changes to dependents")
    p_prop.add_argument("file")
    p_prop.add_argument("--dry-run", action="store_true")

    sub.add_parser("map", help="Show full dependency map")

    args = parser.parse_args()

    if not args.cmd:
        parser.print_help()
        sys.exit(1)

    if args.cmd == "check":
        result = check_propagation(args.file)
        print(f"File: {result['changed_file']}")
        print(f"Dependents ({result['count']}):")
        for dep in result['dependents']:
            print(f"  → {dep}")

    elif args.cmd == "propagate":
        result = propagate(args.file, dry_run=args.dry_run)
        print(f"Propagation for: {result['changed_file']}")
        print(f"Propagated ({len(result['propagated'])}):")
        for p in result['propagated']:
            print(f"  ✓ {p['file']}: {p['action']}")
        print(f"Skipped ({len(result['skipped'])}):")
        for s in result['skipped']:
            print(f"  ⊘ {s['file']}: {s['reason']}")
        if result['errors']:
            print(f"Errors ({len(result['errors'])}):")
            for e in result['errors']:
                print(f"  ✗ {e['file']}: {e['error']}")

    elif args.cmd == "map":
        dep_map = get_full_map()
        print("Dependency Map:")
        for file, deps in dep_map.items():
            print(f"  {file} → {', '.join(deps)}")


if __name__ == "__main__":
    main()