rose-ash/shared/sx/deps.py

"""
Component dependency analysis.

Thin host wrapper over bootstrapped deps module from shared/sx/ref/deps.sx.
The canonical logic lives in the spec; this module provides Python-typed
entry points for the rest of the codebase.
"""
from __future__ import annotations

import os
from typing import Any

from .types import Component, Island, Macro, Symbol


def _use_ref() -> bool:
    return False  # sx_ref.py removed — always use fallback


# ---------------------------------------------------------------------------
# Hand-written fallback (used when SX_USE_REF != 1)
# ---------------------------------------------------------------------------

def _scan_ast(node: Any) -> set[str]:
    refs: set[str] = set()
    _walk(node, refs)
    return refs


def _walk(node: Any, refs: set[str]) -> None:
    if isinstance(node, Symbol):
        if node.name.startswith("~"):
            refs.add(node.name)
        return
    if isinstance(node, list):
        for item in node:
            _walk(item, refs)
        return
    if isinstance(node, dict):
        for v in node.values():
            _walk(v, refs)
        return


def _transitive_deps_fallback(name: str, env: dict[str, Any]) -> set[str]:
    seen: set[str] = set()

    def walk(n: str) -> None:
        if n in seen:
            return
        seen.add(n)
        val = env.get(n)
        if isinstance(val, (Component, Island)):
            for dep in _scan_ast(val.body):
                walk(dep)
        elif isinstance(val, Macro):
            for dep in _scan_ast(val.body):
                walk(dep)

    key = name if name.startswith("~") else f"~{name}"
    walk(key)
    return seen - {key}


def _compute_all_deps_fallback(env: dict[str, Any]) -> None:
    for key, val in env.items():
        if isinstance(val, (Component, Island)):
            val.deps = _transitive_deps_fallback(key, env)


def _scan_io_refs_fallback(node: Any, io_names: set[str]) -> set[str]:
    """Scan an AST node for references to IO primitive names."""
    refs: set[str] = set()
    _walk_io(node, io_names, refs)
    return refs


def _walk_io(node: Any, io_names: set[str], refs: set[str]) -> None:
    if isinstance(node, Symbol):
        if node.name in io_names:
            refs.add(node.name)
        return
    if isinstance(node, list):
        for item in node:
            _walk_io(item, io_names, refs)
        return
    if isinstance(node, dict):
        for v in node.values():
            _walk_io(v, io_names, refs)
        return


def _transitive_io_refs_fallback(
    name: str, env: dict[str, Any], io_names: set[str]
) -> set[str]:
    """Compute transitive IO primitive references for a component."""
    all_refs: set[str] = set()
    seen: set[str] = set()

    def walk(n: str) -> None:
        if n in seen:
            return
        seen.add(n)
        val = env.get(n)
        if isinstance(val, (Component, Island)):
            all_refs.update(_scan_io_refs_fallback(val.body, io_names))
            for dep in _scan_ast(val.body):
                walk(dep)
        elif isinstance(val, Macro):
            all_refs.update(_scan_io_refs_fallback(val.body, io_names))
            for dep in _scan_ast(val.body):
                walk(dep)

    key = name if name.startswith("~") else f"~{name}"
    walk(key)
    return all_refs


def _compute_all_io_refs_fallback(
    env: dict[str, Any], io_names: set[str]
) -> None:
    for key, val in env.items():
        if isinstance(val, (Component, Island)):
            val.io_refs = _transitive_io_refs_fallback(key, env, io_names)


def _scan_components_from_sx_fallback(source: str) -> set[str]:
    import re
    return {f"~{m}" for m in re.findall(r'\(~([a-zA-Z_][a-zA-Z0-9_\-:/]*)', source)}


def _components_needed_fallback(page_sx: str, env: dict[str, Any]) -> set[str]:
    direct = _scan_components_from_sx_fallback(page_sx)
    all_needed: set[str] = set()
    for name in direct:
        all_needed.add(name)
        val = env.get(name)
        if isinstance(val, (Component, Island)) and val.deps:
            all_needed.update(val.deps)
        else:
            all_needed.update(_transitive_deps_fallback(name, env))
    return all_needed


# ---------------------------------------------------------------------------
# Public API — dispatches to bootstrapped or fallback
# ---------------------------------------------------------------------------

def transitive_deps(name: str, env: dict[str, Any]) -> set[str]:
    """Compute transitive component dependencies for *name*.

    Returns the set of all component names (with ~ prefix) that
    *name* can transitively render, NOT including *name* itself.
    """
    return _transitive_deps_fallback(name, env)


def compute_all_deps(env: dict[str, Any]) -> None:
    """Compute and cache deps for all Component entries in *env*."""
    _compute_all_deps_fallback(env)


def scan_components_from_sx(source: str) -> set[str]:
    """Extract component names referenced in SX source text.

    Returns names with ~ prefix, e.g. {"~card", "~shared:layout/nav-link"}.
    """
    return _scan_components_from_sx_fallback(source)


def components_needed(page_sx: str, env: dict[str, Any]) -> set[str]:
    """Compute the full set of component names needed for a page.

    Returns names with ~ prefix.
    """
    return _components_needed_fallback(page_sx, env)


def compute_all_io_refs(env: dict[str, Any], io_names: set[str]) -> None:
    """Compute and cache transitive IO refs for all Component entries in *env*."""
    _compute_all_io_refs_fallback(env, io_names)


def page_render_plan(page_sx: str, env: dict[str, Any], io_names: set[str] | None = None) -> dict[str, Any]:
    """Compute the render plan for a page.

    Returns dict with:
      - "components": {name: "server"|"client", ...}
      - "server": [names rendered server-side]
      - "client": [names rendered client-side]
      - "io-deps": [IO primitive names needed by server components]
    """
    if io_names is None:
        io_names = get_all_io_names()
    # Use fallback implementation (sx_ref.py removed)
    needed = _components_needed_fallback(page_sx, env)
    server, client, io_deps = [], [], []
    for name in needed:
        comp = env.get(name)
        if comp and hasattr(comp, 'io_refs') and comp.io_refs:
            client.append(name)
        else:
            server.append(name)
    return {"components": {n: ("server" if n in server else "client") for n in needed},
            "server": server, "client": client, "io-deps": io_deps}


def get_all_io_names() -> set[str]:
    """Build the complete set of IO primitive names from all boundary tiers.

    Includes: core IO (primitives_io.py handlers), plus all page helper names
    from every service boundary.
    """
    from .primitives_io import IO_PRIMITIVES
    from .boundary import declared_helpers

    names = set(IO_PRIMITIVES)
    for _svc, helper_names in declared_helpers().items():
        names.update(helper_names)
    return names