strukto-ai--mirage/python/mirage/workspace/executor/fanout.py

# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2026 @ Strukto.AI All Rights Reserved. =========

from mirage.commands.builtin.find_eval import FindEntry, keep
from mirage.commands.builtin.find_parse import parse_find_expression
from mirage.commands.errors import FindParseError
from mirage.commands.safeguard import resolve_across_mounts
from mirage.io import IOResult
from mirage.io.stream import materialize
from mirage.io.types import ByteSource
from mirage.types import PathSpec
from mirage.workspace.executor.find_action_dispatch import _apply_find_actions
from mirage.workspace.mount import MountRegistry
from mirage.workspace.types import ExecutionNode

_TRAVERSAL_CMDS = frozenset({"find", "tree", "du"})


def _path_segments(path: str) -> list[str]:
    return [s for s in path.strip("/").split("/") if s]


def _should_fan_out(
    cmd_name: str,
    paths: list[PathSpec],
    flag_kwargs: dict,
    registry: MountRegistry,
) -> bool:
    """Whether `cmd` on this path should run across multiple mounts.

    True when the command is in the traversal whitelist (find/tree/du)
    and the path has at least one descendant mount; or for grep with
    -r/-R; or for ls -R. Returns False when there's no descendant
    mount under the path (single-mount dispatch is correct).
    """
    if not paths:
        return False
    target = paths[0].virtual
    if not registry.descendant_mounts(target):
        return False
    if cmd_name in _TRAVERSAL_CMDS:
        return True
    if cmd_name == "grep":
        return (flag_kwargs.get("r") is True or flag_kwargs.get("R") is True
                or flag_kwargs.get("recursive") is True)
    if cmd_name == "ls":
        return flag_kwargs.get("R") is True
    return False


def _adjust_depth_flags(
    flag_kwargs: dict,
    parent_path: str,
    mount_prefix: str,
) -> dict | None:
    """Adjust find's -maxdepth/-mindepth for a fan-out into a child mount.

    Returns the new kwargs dict, or None if the child mount falls
    outside the depth budget (caller should skip it).
    """
    parent_depth = len(_path_segments(parent_path))
    mount_depth = len(_path_segments(mount_prefix))
    delta = mount_depth - parent_depth
    new = dict(flag_kwargs)
    if "maxdepth" in new:
        raw_md = new["maxdepth"]
        if isinstance(raw_md, list):
            raw_md = raw_md[0] if raw_md else None
        try:
            md = int(raw_md) - delta
        except (TypeError, ValueError):
            md = None
        if md is not None:
            if md < 0:
                return None
            new["maxdepth"] = str(md)
    if "mindepth" in new:
        raw_mn = new["mindepth"]
        if isinstance(raw_mn, list):
            raw_mn = raw_mn[0] if raw_mn else None
        try:
            mn = max(0, int(raw_mn) - delta)
            new["mindepth"] = str(mn)
        except (TypeError, ValueError):
            pass
    return new


def _adjust_depth_texts(
    texts: list[str],
    parent_path: str,
    mount_prefix: str,
) -> list[str]:
    """Adjust -maxdepth/-mindepth values inside a find expression.

    The generic find parses depth from the expression tokens (`texts`),
    not from flag kwargs, so a fan-out into a deeper child mount must
    rewrite the depth values by the parent-to-mount delta. Mirrors
    `_adjust_depth_flags`.

    Args:
        texts (list[str]): the find expression tokens.
        parent_path (str): the find start path the fan-out runs from.
        mount_prefix (str): the child mount prefix being descended into.
    """
    delta = len(_path_segments(mount_prefix)) - len(
        _path_segments(parent_path))
    if delta == 0:
        return list(texts)
    out = list(texts)
    i = 0
    while i < len(out) - 1:
        tok = out[i]
        if tok in ("-maxdepth", "-mindepth"):
            try:
                val = int(out[i + 1])
            except ValueError:
                i += 2
                continue
            if tok == "-maxdepth":
                out[i + 1] = str(val - delta)
            else:
                out[i + 1] = str(max(0, val - delta))
            i += 2
            continue
        i += 1
    return out


def _synthesize_find_mount_entries(
    target_path: str,
    descendants: list,
    texts: list[str],
) -> str:
    """Return synthetic find lines for descendant mount roots.

    `find /` and friends should list mount prefixes as directory
    entries even though no per-mount find emits its own root. The find
    expression is parsed into a predicate tree and evaluated per mount
    root (kind "d"), mirroring the per-backend cores, so -not / -o /
    -path / -type and the -maxdepth / -mindepth window all apply.

    Args:
        target_path (str): the find start path the fan-out runs from.
        descendants (list): descendant mounts to inject as entries.
        texts (list[str]): the find expression tokens.
    """
    try:
        expr = parse_find_expression(list(texts))
    except FindParseError:
        return ""
    tree = expr.tree
    max_depth = expr.maxdepth
    min_depth = expr.mindepth if expr.mindepth is not None else 0
    parent_depth = len(_path_segments(target_path))
    out: list[str] = []
    for m in descendants:
        prefix_no_slash = m.prefix.rstrip("/")
        depth = len(_path_segments(prefix_no_slash)) - parent_depth
        if max_depth is not None and depth > max_depth:
            continue
        base = prefix_no_slash.rsplit("/", 1)[-1] or prefix_no_slash
        entry = FindEntry(key=prefix_no_slash,
                          name=base,
                          kind="d",
                          depth=depth)
        if not keep(entry, tree, min_depth):
            continue
        out.append(prefix_no_slash)
    return "\n".join(out)


async def _filter_under_prefixes(
    stdout: ByteSource,
    descendant_prefixes: list[str],
) -> bytes:
    """Drop lines whose path falls under any descendant mount prefix.

    Path is taken from the start of the line up to the first tab,
    colon, or whitespace (handles find / du / grep output formats).
    Lines that do not start with `/` are passed through.
    """
    data = await materialize(stdout)
    text = data.decode("utf-8", errors="replace")
    out_lines: list[str] = []
    for line in text.split("\n"):
        if line == "":
            continue
        path = line
        for sep in ("\t", ":"):
            if sep in path:
                path = path.split(sep, 1)[0]
                break
        if path.startswith("/"):
            shadowed = False
            for pre in descendant_prefixes:
                if path == pre or path.startswith(pre + "/"):
                    shadowed = True
                    break
            if shadowed:
                continue
        out_lines.append(line)
    return ("\n".join(out_lines) + "\n").encode("utf-8") if out_lines else b""


async def _drop_mount_root_line(stdout: ByteSource, mount_root: str) -> bytes:
    """Drop a descendant find's own mount-root line.

    A per-mount find now emits its start directory, so a descendant
    mount's find yields its mount-root path. The mount-point entry is
    instead synthesized centrally (with the mount's display name and the
    full predicate tree applied), so the raw root line is dropped here to
    avoid a duplicate that skips the name filter.

    Args:
        stdout (ByteSource): descendant find output.
        mount_root (str): the descendant mount root path.
    """
    data = await materialize(stdout)
    text = data.decode("utf-8", errors="replace")
    out_lines = [
        line for line in text.split("\n") if line != "" and line != mount_root
    ]
    return ("\n".join(out_lines) + "\n").encode("utf-8") if out_lines else b""


async def _fan_out_traversal(
    cmd_name: str,
    paths: list[PathSpec],
    texts: list[str],
    flag_kwargs: dict,
    registry: MountRegistry,
    primary_mount: object,
    cwd: str,
    cmd_str: str,
    stdin: ByteSource | None,
) -> tuple[ByteSource | None, IOResult, ExecutionNode]:
    """Run a traversal command across the parent mount + descendant mounts.

    Each mount runs the command with its own root as the path argument
    (depth flags adjusted for find/tree). Outputs are concatenated in
    mount-prefix-sorted order. The parent mount's output is filtered to
    drop lines that fall under any descendant mount (avoids duplicates
    when the parent's resource has shadowed keys).

    For `find`, mount-prefix paths themselves are injected as synthetic
    directory entries (subject to depth and -type filters) because
    mirage's per-mount find doesn't emit the path argument itself.
    """
    target_path = paths[0].virtual
    descendants = registry.descendant_mounts(target_path)
    descendant_prefixes = [m.prefix.rstrip("/") for m in descendants]

    all_stdout: list[bytes] = []
    merged_io = IOResult()
    final_exit = 0
    success_seen = False

    for mount in [primary_mount] + list(descendants):
        if mount is primary_mount:
            sub_paths = list(paths)
            sub_flags = dict(flag_kwargs)
            sub_texts = list(texts)
        else:
            mount_root = mount.prefix.rstrip("/") or "/"
            sub_flags = _adjust_depth_flags(flag_kwargs, target_path,
                                            mount.prefix)
            if sub_flags is None:
                continue
            sub_texts = _adjust_depth_texts(texts, target_path, mount.prefix)
            sub_paths = [
                PathSpec(virtual=mount_root,
                         directory=mount_root,
                         resource_path="",
                         resolved=True)
            ]
        try:
            stdout, io = await mount.execute_cmd(cmd_name,
                                                 sub_paths,
                                                 sub_texts,
                                                 sub_flags,
                                                 stdin=stdin,
                                                 cwd=cwd)
        except FindParseError:
            # A bad numeric/size/mtime argument is a usage error that
            # applies to every mount identically; fail the whole command
            # instead of silently skipping mounts and exiting 0.
            raise
        except Exception:
            continue

        if mount is primary_mount and descendant_prefixes and stdout:
            stdout = await _filter_under_prefixes(stdout, descendant_prefixes)
        elif mount is not primary_mount and cmd_name == "find" and stdout:
            stdout = await _drop_mount_root_line(stdout, mount_root)

        if stdout is not None:
            data = await materialize(stdout)
            if data:
                all_stdout.append(data)
        if io.exit_code == 0:
            success_seen = True
        elif io.exit_code != 0 and final_exit == 0:
            final_exit = io.exit_code
        merged_io = await merged_io.merge(io)

    if cmd_name == "find":
        synthetic = _synthesize_find_mount_entries(target_path, descendants,
                                                   texts)
        if synthetic:
            all_stdout.append(synthetic.encode("utf-8"))

    combined: ByteSource | None
    if all_stdout:
        combined = b"\n".join(b.rstrip(b"\n") for b in all_stdout) + b"\n"
    else:
        combined = None
    final_io_exit = 0 if success_seen else final_exit

    if cmd_name == "find":
        combined, action_err = await _apply_find_actions(
            combined, flag_kwargs, registry, cwd)
        if action_err:
            existing = (await materialize(merged_io.stderr)
                        if merged_io.stderr else b"")
            merged_io.stderr = existing + action_err
            if final_io_exit == 0:
                final_io_exit = 1

    merged_io.exit_code = final_io_exit
    merged_io.safeguard = resolve_across_mounts(cmd_name,
                                                [primary_mount, *descendants])
    exec_node = ExecutionNode(command=cmd_str,
                              exit_code=final_io_exit,
                              stderr=merged_io.stderr)
    return combined, merged_io, exec_node