slots

def absorb_cache_rebind(
    node: Procedure[CGType],
    memo: tuple[object, ...] | None,
    value: object,
    slot: str,
) -> None:
    """Swallow the ``|=`` / ``&=`` rebind at ``node``'s slot ``memo``; reject the rest.

    ``path.x |= v`` desugars to ``setattr(path, 'x', path.x.__ior__(v))`` (and
    ``path[k] |= v`` to the ``__setitem__`` twin); ``__ior__`` already recorded the
    cache and returns it, so the rebind is a no-op — but only when the value is
    *this* slot's own cache. Every other write, including a plain ``=``, raises.
    """
    from capturegraph.procedures.nodes.destination.cache import (
        CacheProcedure,
    )

    child = node._children.get(memo) if memo is not None else None
    if child is not None and isinstance(value, CacheProcedure) and value.destination is child:
        return
    raise TypeError(
        f"Procedures are immutable: cache at a path with '|=' (skip if exists) or "
        f"'&=' (overwrite), and use set_label() for labels (tried to set {slot})"
    )

def catch_superseded_key(key: object) -> None:
    """Mark a key node consumed when the slot it names was already built.

    ``map[cg.const("k")]`` a second time returns the memoized entry, leaving that
    call's freshly built key node with no consumer; the navigation is its consumer.
    """
    from capturegraph.procedures.procedure import (
        Procedure,
    )
    from capturegraph.procedures.procedure.authoring.contexts.base import (
        ProcedureContext,
    )

    context = ProcedureContext.current_or_none()
    if isinstance(key, Procedure) and context is not None:
        context._catch(key)

def memoized_child(
    parent: Procedure[CGType],
    key: tuple[object, ...],
    factory: Callable[[], Procedure[CGType]],
) -> Procedure[CGType]:
    """The ``parent``'s navigation child for ``key``, built once.

    Dedup is node-local: each node remembers its own children, so repeated
    navigation to the same slot returns the same node — which is what carries
    ``_assigned_value`` — with no central bookkeeping. A non-empty ``_children``
    also marks a *path* slot a container, so saving a value there is rejected.
    """
    existing = parent._children.get(key)
    if existing is not None:
        return existing
    child = factory()
    parent._children[key] = child
    return child

def record_cache[T: CGType](
    position: Procedure[Path[T]],
    value: Procedure[T],
    *,
    skip_if_exists: bool,
) -> CacheProcedure[T]:
    """Record a cache at ``position`` and return it, typed as its value ``T``.

    Builds the ``CacheProcedure``, wraps it in a ``RequiredProcedure`` (so the
    stored value is a ``Void`` step the context keeps, with the cache as its
    caught child), and records the assignment on the slot for read-back. A slot
    that has been navigated into (non-empty ``_children``) is a container, and one
    already assigned cannot be assigned again.
    """
    from capturegraph.procedures.nodes.destination.cache import (
        CacheProcedure,
    )
    from capturegraph.procedures.nodes.process.structural import (
        RequiredProcedure,
    )
    from capturegraph.procedures.procedure.authoring.bridge import (
        coerce_input,
    )
    from capturegraph.procedures.procedure.authoring.errors import (
        AlreadyAssignedError,
        AuthoringError,
    )
    from capturegraph.procedures.procedure.authoring.paths import (
        render_path,
    )

    if position._children:
        raise AuthoringError(
            f"cannot save to navigated directory {render_path(position)} "
            "(a slot is a value or a container, not both)"
        )
    if position._assigned_value is not None:
        raise AlreadyAssignedError(f"{render_path(position)} is assigned more than once")

    cache = CacheProcedure(
        destination=position,
        value=coerce_input(value),
        skip_if_exists=skip_if_exists,
    )
    RequiredProcedure(procedure=cache)  # = cg.do(cache): the kept Void step; catches the cache
    object.__setattr__(position, "_assigned_value", cache)
    return cache

def require_map_position(node: Procedure[Path], op: str) -> None:
    """Guard ``.keys()`` / ``.values()``: both enumerate a ``Path[Map]`` position.

    A captured map *value* supports ``[key]`` indexing only.
    """
    from capturegraph.procedures.procedure.authoring.dispatch import (
        navigable,
    )

    nav = navigable(node)
    if nav is None or not nav.is_map:
        raise TypeError(f"{node} has no {op} (it is not a map)")
    if not nav.is_path:
        raise TypeError(
            f"{node} returns a map value; .{op}() enumerates a Path[Map] position — "
            "index a captured map by key instead (map[key])"
        )
    require_no_saved_value(node)

def require_no_saved_value(node: Procedure[Path]) -> None:
    """Guard descending into a *path* position.

    A slot is a value or a container, not both, so a slot with a saved value has no
    children.
    """
    if node._assigned_value is not None:
        from capturegraph.procedures.procedure.authoring.errors import (
            AuthoringError,
        )
        from capturegraph.procedures.procedure.authoring.paths import (
            render_path,
        )

        raise AuthoringError(
            f"cannot navigate into {render_path(node)}: a value was saved there "
            "(a slot is a value or a container, not both)"
        )

def string_literal(key: object) -> str | None:
    """A constant string key as a plain ``str``, or ``None`` if it can't be one.

    Accepts a raw ``str`` or a ``String`` value; returns ``None`` when the key is
    computed and so can't be checked statically.
    """
    from capturegraph.types.scalars.primitives.string import (
        String,
    )

    if isinstance(key, str):
        return key
    if isinstance(key, String):
        return key.value
    return None

def subscript_child(
    node: Procedure[Path],
    nav: Navigable,
    key: int | Procedure[CGType] | JSONScalar | bool | float | str,
) -> Procedure[CGType]:
    """Build (once) ``node[key]`` — an array element (int) or a map entry (key value)."""
    memo = subscript_memo(nav, key)
    if memo is None:
        raise TypeError(f"array index must be an int literal, got {key!r}")

    existing = node._children.get(memo)
    if existing is not None:
        catch_superseded_key(key)
        return existing
    if nav.is_array:
        index = cast("int", key)
        return memoized_child(node, memo, lambda: nav.nodes.index(source=node, index=index))

    from capturegraph.procedures.procedure.authoring.bridge import (
        coerce_input,
    )

    if not nav.is_path:
        warn_unknown_map_key(node, key)
    return memoized_child(
        node,
        memo,
        lambda: nav.nodes.key(source=node, key=coerce_input(key)),
    )

def subscript_memo(
    nav: Navigable,
    key: int | Procedure[CGType] | JSONScalar | bool | float | str,
) -> tuple[object, ...] | None:
    """``[key]``'s slot identity on ``nav``, or ``None`` if ``key`` cannot index it.

    An array slot is its integer index. A map slot is a *literal* key's value —
    raw or already lifted, so ``m["a"]``, ``m[cg.String(value="a")]`` and
    ``m[cg.const("a")]`` all name one slot — and a computed key's node otherwise.
    """
    if nav.is_array:
        return ("index", key) if isinstance(key, int) else None

    from capturegraph.procedures.nodes.process.constant import (
        Constant,
        as_scalar,
    )
    from capturegraph.procedures.procedure import (
        Procedure,
    )

    if isinstance(key, Procedure):
        if isinstance(key, Constant):
            return ("key", key.return_type.__name__, repr(key.value))
        return ("key", key.uuid)
    scalar = as_scalar(key)
    return ("key", type(scalar).__name__, repr(type(scalar).encode(scalar)))

def warn_unknown_map_key(source: Procedure[CGType], key: object) -> None:
    """Warn when a value-level ``map[key]`` names a key the source map provably lacks.

    Only a node that advertises its keys (``static_map_keys``) with a literal key is
    checked; a dynamic source or computed key is left alone. The access still
    builds — at runtime it yields no value, and the client reports the unknown key.
    """
    import warnings

    known = source.static_map_keys()
    literal = string_literal(key)
    if known is None or literal is None or literal in known:
        return
    warnings.warn(
        f"{source} has no key {literal!r}; its keys are {sorted(known)}. "
        f"This access produces no value at runtime.",
        stacklevel=4,  # warn ← here ← subscript_child ← __getitem__ ← user code
    )