ss-tools/migrate_hierarchical.py

# #region MigrateHierarchical [C:4] [TYPE Module] [SEMANTICS migration,hierarchical,ids]
# @defgroup Migration Script to upgrade flat contract IDs to hierarchical Domain.Name format.
# @BRIEF Dry-run first, then batch-apply hierarchical ID migration for all contracts in the project.
# @RATIONALE 97.9% of contracts have flat IDs (e.g. UserListModel) that die at HCA 128×.
#   Hierarchical IDs (Users.ListModel) survive as statistical signatures.
#   Script infers domain prefix from parent contract → @SEMANTICS → file path → fallback.
# @SIDE_EFFECT Modifies #region/#endregion anchors and @RELATION edges across all source files.
#   Creates axiom checkpoints before each batch.
# @INVARIANT Anchor pairs remain matched. @RELATION edges updated. NO business logic changed.

import re
import sys
from collections import defaultdict
from pathlib import Path

# ── Domain prefix inference ──────────────────────────────────────

# Path-based inference (ordered: most specific first)
PATH_TO_DOMAIN = [
    ("frontend/src/lib/models/",            "Models"),
    ("frontend/src/lib/components/ui/",      "UI"),
    ("frontend/src/lib/components/translate/", "Translate"),
    ("frontend/src/lib/components/dashboard/", "Dashboard"),
    ("frontend/src/lib/components/llm/",      "LLM"),
    ("frontend/src/lib/components/dataset-review/", "DatasetReview"),
    ("frontend/src/lib/components/settings/", "Settings"),
    ("frontend/src/lib/components/layout/",   "Layout"),
    ("frontend/src/lib/components/assistant/","Assistant"),
    ("frontend/src/lib/components/tasks/",    "Tasks"),
    ("frontend/src/lib/components/",          "Components"),
    ("frontend/src/lib/ui/",                 "UI"),
    ("frontend/src/lib/stores/",             "Stores"),
    ("frontend/src/lib/api/",                "Api"),
    ("frontend/src/routes/",                 "Routes"),
    ("frontend/src/types/",                  "Types"),
    ("backend/src/api/routes/assistant/",    "AssistantApi"),
    ("backend/src/api/routes/",              "Api"),
    ("backend/src/core/task_manager/",        "TaskManager"),
    ("backend/src/core/auth/",               "Auth"),
    ("backend/src/core/migration/",          "Migration"),
    ("backend/src/core/plugins/",            "Plugin"),
    ("backend/src/core/",                    "Core"),
    ("backend/src/services/dataset_review/", "DatasetReview"),
    ("backend/src/services/",                "Services"),
    ("backend/src/models/",                  "Models"),
    ("backend/src/schemas/",                 "Schemas"),
    ("backend/src/plugins/translate/",       "Translate"),
    ("backend/src/plugins/llm_analysis/",    "LLMAnalysis"),
    ("backend/src/plugins/git/",             "Git"),
    ("backend/src/plugins/storage/",         "Storage"),
    ("backend/src/plugins/",                 "Plugin"),
    ("backend/tests/",                       "Tests"),
]

# @SEMANTICS keyword → domain prefix
SEMANTICS_TO_DOMAIN = {
    "auth": "Auth",
    "login": "Auth",
    "token": "Auth",
    "migration": "Migration",
    "dashboard": "Dashboard",
    "dataset": "Dataset",
    "translation": "Translate",
    "translate": "Translate",
    "llm": "LLM",
    "git": "Git",
    "storage": "Storage",
    "plugin": "Plugin",
    "task": "Tasks",
    "test": "Tests",
    "api": "Api",
    "ui": "UI",
    "users": "Users",
    "roles": "Roles",
    "settings": "Settings",
    "profile": "Profile",
    "reports": "Reports",
    "validation": "Validation",
    "backup": "Backup",
    "mapper": "Mapper",
    "debug": "Debug",
    "assistant": "Assistant",
    "admin": "Admin",
    "notification": "Notifications",
    "schedule": "Schedule",
    "logging": "Logging",
    "indexing": "Index",
    "curation": "Curator",
    "semantic": "Semantic",
    "screenshot": "Screenshot",
    "health": "Health",
}


# ── File discovery ───────────────────────────────────────────────

def find_source_files(root: Path) -> list[Path]:
    """Find all source files with contracts."""
    patterns = ["**/*.py", "**/*.svelte", "**/*.svelte.ts"]
    source_dirs = [
        root / "backend/src",
        root / "frontend/src",
        root / "backend/tests",
    ]
    files = []
    for d in source_dirs:
        if not d.exists():
            continue
        for pat in patterns:
            files.extend(d.rglob(pat.split("/")[-1]))
    return sorted(set(files))


# ── Contract parsing ─────────────────────────────────────────────

ANCHOR_RE = re.compile(
    r'^(?P<prefix>\s*(?:#|//|<!--)\s*)#region\s+(?P<id>[^\s\[\(]+)\s*'
    r'(\[C:(?P<complexity>\d)\]\s*)?'
    r'(\[TYPE\s+(?P<type>[^\]]+)\]\s*)?'
    r'(\[SEMANTICS\s+(?P<semantics>[^\]]+)\]\s*)?',
    re.MULTILINE
)

ENDANCHOR_RE = re.compile(
    r'^(?P<prefix>\s*(?:#|//|<!--)\s*)#endregion\s+(?P<id>[^\s]+)',
    re.MULTILINE
)

RELATION_RE = re.compile(
    r'@RELATION\s+(?P<predicate>\w+)\s*->\s*\[(?P<target>[^\]]+)\]'
)


def parse_contracts(content: str) -> list[dict]:
    """Extract all contracts with their metadata."""
    contracts = []
    for m in ANCHOR_RE.finditer(content):
        cid = m.group("id")
        end_pattern = re.compile(
            rf'^{m.group("prefix")}#endregion\s+{re.escape(cid)}',
            re.MULTILINE
        )
        end_m = end_pattern.search(content, m.end())
        if not end_m:
            continue

        body = content[m.start():end_m.end()]
        # Extract @SEMANTICS keywords
        sem_match = re.search(r'@SEMANTICS\s+([^\n]+)', body)
        semantics = sem_match.group(1).strip().split(",") if sem_match else []
        semantics = [s.strip() for s in semantics]

        # Check if already hierarchical
        is_hierarchical = "." in cid

        contracts.append({
            "id": cid,
            "complexity": int(m.group("complexity")) if m.group("complexity") else 1,
            "type": m.group("type") or "Unknown",
            "semantics": semantics,
            "is_hierarchical": is_hierarchical,
            "start": m.start(),
            "end": end_m.end(),
            "body": body,
            "anchor_prefix": m.group("prefix"),
        })
    return contracts


def parse_parent_chain(contracts: list[dict]) -> None:
    """Annotate each contract with its parent contract ID."""
    # Sort by position
    sorted_contracts = sorted(contracts, key=lambda c: c["start"])
    stack = []
    for c in sorted_contracts:
        # Pop parents that end before this contract starts
        while stack and stack[-1]["end"] < c["start"]:
            stack.pop()
        c["parent"] = stack[-1]["id"] if stack else None
        c["parent_obj"] = stack[-1] if stack else None
        stack.append(c)


# ── Domain inference ─────────────────────────────────────────────

def infer_domain(contract: dict, file_path: Path) -> str:
    """Infer domain prefix for a flat contract ID."""
    # Strategy 1: Parent contract's domain
    parent = contract.get("parent_obj")
    if parent and parent.get("is_hierarchical"):
        # Parent is already hierarchical — use its domain prefix
        parts = parent["id"].split(".")
        if len(parts) >= 2:
            return parts[0]  # Core from Core.Auth

    # Strategy 2: @SEMANTICS first keyword
    semantics = contract.get("semantics", [])
    for kw in semantics:
        kw_clean = kw.strip().lower()
        if kw_clean in SEMANTICS_TO_DOMAIN:
            return SEMANTICS_TO_DOMAIN[kw_clean]

    # Strategy 3: File path
    fpath = str(file_path)
    for path_prefix, domain in PATH_TO_DOMAIN:
        if path_prefix in fpath:
            return domain

    # Strategy 4: Contract type
    ctype = contract.get("type", "")
    if ctype == "Model":
        return "Models"
    if ctype == "Component":
        return "Components"
    if ctype == "Function":
        return "Core"

    return "Module"


# ── Migration ────────────────────────────────────────────────────

def migrate_file(file_path: Path, dry_run: bool = True) -> dict:
    """Migrate flat IDs to hierarchical in a single file."""
    content = file_path.read_text(encoding="utf-8")
    contracts = parse_contracts(content)
    parse_parent_chain(contracts)

    # Skip rules
    def should_skip(c: dict, file_path: Path) -> bool:
        """Skip contracts that don't need hierarchical migration."""
        cid = c["id"]
        # Skip already hierarchical
        if c["is_hierarchical"]:
            return True
        # Skip private helpers (C1/C2)
        if cid.startswith("_"):
            return True
        # Skip C1 — single-level IDs acceptable per ATTN_2
        if c.get("complexity", 1) <= 1:
            return True
        parent = c.get("parent_obj")
        if parent and parent.get("is_hierarchical") and c.get("complexity", 1) <= 2:
            return True
        # Skip test files — they use Test.Domain.Name convention separately
        if "tests" in str(file_path).lower() or "__tests__" in str(file_path):
            return True
        return False

    targets = [c for c in contracts if not should_skip(c, file_path)]
    
    if not targets:
        return {"file": str(file_path), "migrated": 0, "changes": []}

    new_content = content
    renames = {}  # old_id → new_id
    changes = []

    # Process bottom-up (children before parents)
    for c in reversed(targets):
        domain = infer_domain(c, file_path)

        # Smart prefix: if parent already has the domain, use parent's full ID as prefix
        parent = c.get("parent_obj")
        if parent and parent.get("is_hierarchical"):
            parent_domain = parent["id"].split(".")[0]
            if domain == parent_domain:
                domain = parent["id"]  # Core.Auth → Core.Auth.Login, not Core.Login

        # Avoid double-prefixing
        if c["id"].startswith(domain + "."):
            continue
        new_id = f"{domain}.{c['id']}"
        renames[c["id"]] = new_id

    # Apply renames: #region IDs and #endregion IDs
    for old_id, new_id in renames.items():
        # Find the exact anchor pair for this contract
        anchor_start = rf'(#region\s+){re.escape(old_id)}(\s*\[)'
        anchor_end = rf'(#endregion\s+){re.escape(old_id)}(\s*$|\n)'

        new_content = re.sub(anchor_start, rf'\1{new_id}\2', new_content)
        new_content = re.sub(anchor_end, rf'\1{new_id}\2', new_content)

        changes.append(f"  {old_id} → {new_id}")

    # Update @RELATION edges referencing renamed contracts
    for old_id, new_id in renames.items():
        relation_pat = rf'(@RELATION\s+\w+\s*->\s*\[){re.escape(old_id)}(\])'
        new_content = re.sub(relation_pat, rf'\1{new_id}\2', new_content)

    if not dry_run and new_content != content:
        # Create backup
        backup = file_path.with_suffix(file_path.suffix + ".bak")
        file_path.rename(backup)
        file_path.write_text(new_content, encoding="utf-8")
        # Remove backup on success
        backup.unlink()

    return {
        "file": str(file_path),
        "migrated": len(changes),
        "changes": changes,
    }


# ── Main ─────────────────────────────────────────────────────────

def main():
    root = Path.cwd()
    dry_run = True
    for arg in sys.argv[1:]:
        if arg == "--apply":
            dry_run = False
        elif arg.startswith("--root="):
            root = Path(arg.split("=", 1)[1])
        elif not arg.startswith("--"):
            root = Path(arg)
    mode = "DRY RUN" if dry_run else "APPLY"

    print(f"=== Hierarchical ID Migration — {mode} ===")
    print(f"Root: {root}")
    print()

    files = find_source_files(root)
    print(f"Found {len(files)} source files")
    print()

    total_contracts = 0
    total_flat = 0
    total_migrated = 0
    file_results = []

    for file_path in files:
        result = migrate_file(file_path, dry_run=dry_run)
        if result["migrated"] > 0:
            file_results.append(result)
            total_migrated += result["migrated"]

    # Summary
    print(f"Files with flat IDs: {len(file_results)}")
    print(f"Contracts to migrate: {total_migrated}")
    print()

    if file_results:
        print("Top changes:")
        for r in file_results[:40]:
            print(f"  {Path(r['file']).name}: {r['migrated']} IDs")
            for c in r["changes"][:5]:
                print(c)
            if len(r["changes"]) > 5:
                print(f"  ... and {len(r['changes']) - 5} more")
            print()

    if dry_run:
        print("Run with --apply to execute the migration.")
    else:
        print("Migration complete. Run 'axiom_semantic_index rebuild rebuild_mode=full' to reindex.")


if __name__ == "__main__":
    main()

# #endregion MigrateHierarchical