[BUG] FlyteRemote.sync_node_execution: unbounded recursion (no depth check, no cycle detection) #7338

"""Reproducer: FlyteRemote.sync_node_execution unbounded recursion."""
from __future__ import annotations

import inspect
import sys
from unittest.mock import MagicMock, patch

import pytest

from flytekit.configuration import Config
from flytekit.remote.remote import FlyteRemote


@pytest.fixture
def remote():
    with patch("flytekit.clients.friendly.SynchronousFlyteClient"):
        flyte_remote = FlyteRemote(
            config=Config.auto(),
            default_project="p1",
            default_domain="d1",
        )
        flyte_remote._client_initialized = True
        flyte_remote._client = MagicMock()
        return flyte_remote


def test_sync_node_execution_lacks_depth_or_visited_check(remote):
    """Static-analysis assertion: the recursive call sites lack any
    depth/visited mechanism. Source-level evidence of the gap."""
    sync_node_src = inspect.getsource(remote.sync_node_execution)
    sync_exec_src = inspect.getsource(remote.sync_execution)

    # Confirms the recursive call sites exist
    assert "self.sync_execution(launched_exec" in sync_node_src
    assert "self.sync_node_execution(" in sync_node_src

    # No depth parameter or visited set
    assert "depth" not in sync_node_src.lower() or "depth=" not in sync_node_src
    assert "visited" not in sync_node_src.lower()
    assert "depth" not in sync_exec_src.lower() or "max_depth" not in sync_exec_src


def test_recursion_limit_demonstration():
    """Synthetic mirror of the sync_node_execution call shape; crashes
    with RecursionError when given a deeply-nested input graph at low
    recursion limit. Shows the consequence of the missing bound."""
    def synthetic_sync_node_execution(node, depth_observed=None):
        if depth_observed is None:
            depth_observed = []
        depth_observed.append(len(depth_observed))
        if node.get("launched_exec_id"):
            for child in node["launched_exec_id"].get("nodes", []):
                synthetic_sync_node_execution(child, depth_observed)
        return depth_observed

    def chain(n):
        # ITERATIVE builder so we don't hit the recursion limit at
        # construction time
        node = {"launched_exec_id": None}
        for _ in range(n):
            node = {"launched_exec_id": {"nodes": [node]}}
        return node

    deep_node = chain(60)
    original = sys.getrecursionlimit()
    try:
        sys.setrecursionlimit(50)
        with pytest.raises(RecursionError):
            synthetic_sync_node_execution(deep_node)
    finally:
        sys.setrecursionlimit(original)

def sync_execution(
    self,
    execution: FlyteWorkflowExecution,
    entity_definition: typing.Union[FlyteWorkflow, FlyteTask] = None,
    sync_nodes: bool = False,
+   _depth: int = 0,
+   _max_depth: int = 50,
) -> FlyteWorkflowExecution:
+   if _depth > _max_depth:
+       raise FlyteAssertion(
+           f"Workflow execution {execution.id} has nesting deeper than "
+           f"_max_depth={_max_depth}; refusing to recurse further to "
+           f"avoid RecursionError."
+       )
    ...
    if sync_nodes:
        node_execs[n.id.node_id] = self.sync_node_execution(
-           n, node_mapping
+           n, node_mapping, _depth=_depth + 1, _max_depth=_max_depth,
        )